Function admin_socket

src/pybind/ceph_daemon.py:34–88

Send a daemon (--admin-daemon) command 'cmd'. asok_path is the path to the admin socket; cmd is a list of strings; format may be set to one of the formatted forms to get output in that form (daemon commands don't support 'plain' output).

(asok_path: str,
                 cmd: List[str],
                 format: Optional[str] = '')

def admin_socket(asok_path: str,
                 cmd: List[str],
                 format: Optional[str] = '') -> bytes:
    """
    Send a daemon (--admin-daemon) command 'cmd'.  asok_path is the
    path to the admin socket; cmd is a list of strings; format may be
    set to one of the formatted forms to get output in that form
    (daemon commands don't support 'plain' output).
    """

    def do_sockio(path: str, cmd_bytes: bytes) -> bytes:
        """ helper: do all the actual low-level stream I/O """
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        sock.connect(path)
        try:
            sock.sendall(cmd_bytes + b'\0')
            len_str = sock.recv(4)
            if len(len_str) < 4:
                raise RuntimeError("no data returned from admin socket")
            l, = struct.unpack(">I", len_str)
            sock_ret = b''

            got = 0
            while got < l:
                # recv() receives signed int, i.e max 2GB
                # workaround by capping READ_CHUNK_SIZE per call.
                want = min(l - got, READ_CHUNK_SIZE)
                bit = sock.recv(want)
                sock_ret += bit
                got += len(bit)

        except Exception as sock_e:
            raise RuntimeError('exception: ' + str(sock_e))
        return sock_ret

    try:
        cmd_json = do_sockio(asok_path,
                             b'{"prefix": "get_command_descriptions"}')
    except Exception as e:
        raise RuntimeError('exception getting command descriptions: ' + str(e))

    sigdict = parse_json_funcsigs(cmd_json.decode('utf-8'), 'cli')
    valid_dict = validate_command(sigdict, cmd)
    if not valid_dict:
        raise RuntimeError('invalid command')

    if format:
        valid_dict['format'] = format

    try:
        ret = do_sockio(asok_path, json.dumps(valid_dict).encode('utf-8'))
    except Exception as e:
        raise RuntimeError('exception: ' + str(e))

    return ret


class Termsize(object):