Method admin_socket

qa/tasks/ceph_manager.py:1937–1984

Remotely start up ceph specifying the admin socket :param command: a list of words to use as the command to the admin socket

(self, service_type, service_id,
                     command, check_status=True, timeout=0, stdout=None)

                          service_type, service_id)

    def admin_socket(self, service_type, service_id,
                     command, check_status=True, timeout=0, stdout=None):
        """
        Remotely start up ceph specifying the admin socket
        :param command: a list of words to use as the command
                        to the admin socket
        """
        if stdout is None:
            stdout = StringIO()

        remote = self.find_remote(service_type, service_id)

        if self.cephadm:
            return shell(
                self.ctx, self.cluster, remote,
                args=[
                    'ceph', 'daemon', '%s.%s' % (service_type, service_id),
                ] + command,
                stdout=stdout,
                wait=True,
                check_status=check_status,
            )
        if self.rook:
            assert False, 'not implemented'

        args = [
            'sudo',
            'adjust-ulimits',
            'ceph-coverage',
           f'{self.testdir}/archive/coverage',
            'timeout',
            str(timeout),
            'ceph',
            '--cluster',
            self.cluster,
            '--admin-daemon',
            '/var/run/ceph/{cluster}-{type}.{id}.asok'.format(
                cluster=self.cluster,
                type=service_type,
                id=service_id),
            ]
        args.extend(command)
        return remote.run(
            args=args,
            stdout=stdout,
            wait=True,
            check_status=check_status
            )

    def objectstore_tool(self, pool, options, args, **kwargs):
        return ObjectStoreTool(self, pool, **kwargs).run(options, args)