def handle_command(self, inbuf, cmd):
        self.log.error("handle_command")

        if cmd['prefix'] == 'device query-daemon-health-metrics':
            who = cmd.get('who', '')
            if who[0:4] != 'osd.':
                return (-errno.EINVAL, '', 'not a valid <osd.NNN> id')
            osd_id = who[4:]
            result = CommandResult('')
            self.send_command(result, 'osd', osd_id, json.dumps({
                'prefix': 'smart',
                'format': 'json',
            }), '')
            r, outb, outs = result.wait()
            return (r, outb, outs)
        elif cmd['prefix'] == 'device scrape-daemon-health-metrics':
            who = cmd.get('who', '')
            if who[0:4] != 'osd.':
                return (-errno.EINVAL, '', 'not a valid <osd.NNN> id')
            id = int(who[4:])
            return self.scrape_osd(id)
        elif cmd['prefix'] == 'device scrape-health-metrics':
            if 'devid' in cmd:
                return self.scrape_device(cmd['devid'])
            return self.scrape_all();
        elif cmd['prefix'] == 'device show-health-metrics':
            return self.show_device_metrics(cmd['devid'], cmd.get('sample'))
        else:
            # mgr should respect our self.COMMANDS and not call us for
            # any prefix we don't advertise
            raise NotImplementedError(cmd['prefix'])

    def _open_connection(self, pool_name='device_health_metrics'):
        pools = self.module.rados.list_pools()
        is_pool = False
        for pool in pools:
            if pool == pool_name:
                is_pool = True
                break
        if not is_pool:
            self.module.log.debug('create %s pool' % pool_name)
            # create pool
            result = CommandResult('')
            self.module.send_command(result, 'mon', '', json.dumps({
                'prefix': 'osd pool create',
                'format': 'json',
                'pool': pool_name,
                'pg_num': 1,
            }), '')
            r, outb, outs = result.wait()
            assert r == 0

            # set pool application
            result = CommandResult('')
            self.module.send_command(result, 'mon', '', json.dumps({
                'prefix': 'osd pool application enable',
                'format': 'json',
                'pool': pool_name,
                'app': 'mgr_devicehealth',
            }), '')
            r, outb, outs = result.wait()
            assert r == 0

        ioctx = self.module.rados.open_ioctx(pool_name)
        return ioctx

    def get_file_sd_config(self):
        servers = self.list_servers()
        targets = []
        for server in servers:
            hostname = server.get('hostname', '')
            for service in server.get('services', []):
                if service['type'] != 'mgr':
                    continue
                id_ = service['id']
                # get port for prometheus module at mgr with id_
                # TODO use get_config_prefix or get_config here once
                # https://github.com/ceph/ceph/pull/20458 is merged
                result = CommandResult("")
                global_instance().send_command(
                    result, "mon", '',
                    json.dumps({
                        "prefix": "config-key get",
                        'key': "config/mgr/mgr/prometheus/{}/server_port".format(id_),
                    }),
                    "")
                r, outb, outs = result.wait()
                if r != 0:
                    global_instance().log.error("Failed to retrieve port for mgr {}: {}".format(id_, outs))
                    targets.append('{}:{}'.format(hostname, DEFAULT_PORT))
                else:
                    port = json.loads(outb)
                    targets.append('{}:{}'.format(hostname, port))

        ret = [
            {
                "targets": targets,
                "labels": {}
            }
        ]
        return 0, json.dumps(ret), ""

            def _osd(self, osd_id):
                osd_id = int(osd_id)

                osd_map = global_instance().get("osd_map")

                osd = None
                for o in osd_map['osds']:
                    if o['osd'] == osd_id:
                        osd = o
                        break

                assert osd is not None  # TODO 400

                osd_spec = "{0}".format(osd_id)

                osd_metadata = global_instance().get_metadata(
                        "osd", osd_spec)

                result = CommandResult("")
                global_instance().send_command(result, "osd", osd_spec,
                       json.dumps({
                           "prefix": "perf histogram dump",
                           }),
                       "")
                r, outb, outs = result.wait()
                assert r == 0
                histogram = json.loads(outb)

                return {
                    "osd": osd,
                    "osd_metadata": osd_metadata,
                    "osd_histogram": histogram
                }

    def open_connection(self, create_if_missing=True):
        pools = self.rados.list_pools()
        is_pool = False
        for pool in pools:
            if pool == self.pool_name:
                is_pool = True
                break
        if not is_pool:
            if not create_if_missing:
                return None
            self.log.debug('create %s pool' % self.pool_name)
            # create pool
            result = CommandResult('')
            self.send_command(result, 'mon', '', json.dumps({
                'prefix': 'osd pool create',
                'format': 'json',
                'pool': self.pool_name,
                'pg_num': 1,
            }), '')
            r, outb, outs = result.wait()
            assert r == 0

            # set pool application
            result = CommandResult('')
            self.send_command(result, 'mon', '', json.dumps({
                'prefix': 'osd pool application enable',
                'format': 'json',
                'pool': self.pool_name,
                'app': 'mgr_devicehealth',
            }), '')
            r, outb, outs = result.wait()
            assert r == 0

        ioctx = self.rados.open_ioctx(self.pool_name)
        return ioctx

 def reset_device_life_expectancy(self, device_id):
     result = CommandResult('')
     self.module.send_command(result, 'mon', '', json.dumps({
         'prefix': 'device rm-life-expectancy',
         'devid': device_id
     }), '')
     ret, outb, outs = result.wait()
     if ret != 0:
         self.module.log.error(
             'failed to reset device life expectancy, %s' % outs)
     return ret

 def _get(self):
     mds_spec = "{0}:0".format(self.fscid)
     result = CommandResult("")
     self._module.send_command(result, "mds", mds_spec,
            json.dumps({
                "prefix": "session ls",
                }),
            "")
     r, outb, outs = result.wait()
     # TODO handle nonzero returns, e.g. when rank isn't active
     assert r == 0
     return json.loads(outb)

 def compat_weight_set_reweight(self, osd, new_weight):
     self.log.debug('ceph osd crush weight-set reweight-compat')
     result = CommandResult('')
     self.send_command(result, 'mon', '', json.dumps({
         'prefix': 'osd crush weight-set reweight-compat',
         'format': 'json',
         'item': 'osd.%d' % osd,
         'weight': [new_weight],
     }), '')
     r, outb, outs = result.wait()
     if r != 0:
         self.log.error('Error setting compat weight-set osd.%d to %f' %
         (osd, new_weight))
         return

    def get_compat_weight_set_weights(self, ms):
        if not CRUSHMap.have_default_choose_args(ms.crush_dump):
            # enable compat weight-set first
            self.log.debug('ceph osd crush weight-set create-compat')
            result = CommandResult('')
            self.send_command(result, 'mon', '', json.dumps({
                'prefix': 'osd crush weight-set create-compat',
                'format': 'json',
            }), '')
            r, outb, outs = result.wait()
            if r != 0:
                self.log.error('Error creating compat weight-set')
                return

            result = CommandResult('')
            self.send_command(result, 'mon', '', json.dumps({
                'prefix': 'osd crush dump',
                'format': 'json',
            }), '')
            r, outb, outs = result.wait()
            if r != 0:
                self.log.error('Error dumping crush map')
                return
            try:
                crushmap = json.loads(outb)
            except:
                raise RuntimeError('unable to parse crush map')
        else:
            crushmap = ms.crush_dump

        raw = CRUSHMap.get_default_choose_args(crushmap)
        weight_set = {}
        for b in raw:
            bucket = None
            for t in crushmap['buckets']:
                if t['id'] == b['bucket_id']:
                    bucket = t
                    break
            if not bucket:
                raise RuntimeError('could not find bucket %s' % b['bucket_id'])
            self.log.debug('bucket items %s' % bucket['items'])
            self.log.debug('weight set %s' % b['weight_set'][0])
            if len(bucket['items']) != len(b['weight_set'][0]):
                raise RuntimeError('weight-set size does not match bucket items')
            for pos in range(len(bucket['items'])):
                weight_set[bucket['items'][pos]['id']] = b['weight_set'][0][pos]

        self.log.debug('weight_set weights %s' % weight_set)
        return weight_set

    def _config_dump(self):
        """Report cluster configuration

        This report is the standard `config dump` report. It does not include
        configuration defaults; these can be inferred from the version number.
        """
        result = CommandResult("")
        args = dict(prefix = "config dump", format = "json")
        self.send_command(result, "mon", "", json.dumps(args), "")
        ret, outb, outs = result.wait()
        if ret == 0:
            return json.loads(outb), []
        else:
            self.log.warning("send_command 'config dump' failed. \
                    ret={}, outs=\"{}\"".format(ret, outs))
            return [], ["Failed to read monitor config dump"]

    def handle_command(self, cmd):
        self.log.error("handle_command")

        if cmd['prefix'] == 'osd smart get':
            result = CommandResult('')
            self.send_command(result, 'osd', cmd['osd_id'], json.dumps({
                'prefix': 'smart',
                'format': 'json',
            }), '')
            r, outb, outs = result.wait()
            return (r, outb, outs)

        else:
            # mgr should respect our self.COMMANDS and not call us for
            # any prefix we don't advertise
            raise NotImplementedError(cmd['prefix'])

    def do_scrape_osd(self, osd_id, ioctx, devid=''):
        self.log.debug('do_scrape_osd osd.%d' % osd_id)

        # scrape from osd
        result = CommandResult('')
        self.send_command(result, 'osd', str(osd_id), json.dumps({
            'prefix': 'smart',
            'format': 'json',
            'devid': devid,
        }), '')
        r, outb, outs = result.wait()

        try:
            return json.loads(outb)
        except:
            self.log.debug('Fail to parse JSON result from "%s"' % outb)

    def do_scrape_daemon(self, daemon_type, daemon_id, devid=''):
        """
        :return: a dict, or None if the scrape failed.
        """
        self.log.debug('do_scrape_daemon %s.%s' % (daemon_type, daemon_id))
        result = CommandResult('')
        self.send_command(result, daemon_type, daemon_id, json.dumps({
            'prefix': 'smart',
            'format': 'json',
            'devid': devid,
        }), '')
        r, outb, outs = result.wait()

        try:
            return json.loads(outb)
        except (IndexError, ValueError):
            self.log.error(
                "Fail to parse JSON result from daemon {0}.{1} ({2})".format(
                    daemon_type, daemon_id, outb))

    def handle_command(self, _, cmd):
        self.log.error("handle_command")

        if cmd['prefix'] == 'device query-daemon-health-metrics':
            who = cmd.get('who', '')
            if not self.is_valid_daemon_name(who):
                return -errno.EINVAL, '', 'not a valid mon or osd daemon name'
            (daemon_type, daemon_id) = cmd.get('who', '').split('.')
            result = CommandResult('')
            self.send_command(result, daemon_type, daemon_id, json.dumps({
                'prefix': 'smart',
                'format': 'json',
            }), '')
            r, outb, outs = result.wait()
            return r, outb, outs
        elif cmd['prefix'] == 'device scrape-daemon-health-metrics':
            who = cmd.get('who', '')
            if not self.is_valid_daemon_name(who):
                return -errno.EINVAL, '', 'not a valid mon or osd daemon name'
            (daemon_type, daemon_id) = cmd.get('who', '').split('.')
            return self.scrape_daemon(daemon_type, daemon_id)
        elif cmd['prefix'] == 'device scrape-health-metrics':
            if 'devid' in cmd:
                return self.scrape_device(cmd['devid'])
            return self.scrape_all()
        elif cmd['prefix'] == 'device get-health-metrics':
            return self.show_device_metrics(cmd['devid'], cmd.get('sample'))
        elif cmd['prefix'] == 'device check-health':
            return self.check_health()
        elif cmd['prefix'] == 'device monitoring on':
            self.set_module_option('enable_monitoring', True)
            self.event.set()
            return 0, '', ''
        elif cmd['prefix'] == 'device monitoring off':
            self.set_module_option('enable_monitoring', False)
            self.set_health_checks({})  # avoid stuck health alerts
            return 0, '', ''
        elif cmd['prefix'] == 'device predict-life-expectancy':
            return self.predict_lift_expectancy(cmd['devid'])
        else:
            # mgr should respect our self.COMMANDS and not call us for
            # any prefix we don't advertise
            raise NotImplementedError(cmd['prefix'])

 def get(self, svc_id):
     result = CommandResult('')
     mgr.send_command(result, 'osd', svc_id,
                      json.dumps({
                          'prefix': 'perf histogram dump',
                      }),
                      '')
     r, outb, outs = result.wait()
     if r != 0:
         logger.warning('Failed to load histogram for OSD %s', svc_id)
         logger.debug(outs)
         histogram = outs
     else:
         histogram = json.loads(outb)
     return {
         'osd_map': self.get_osd_map()[svc_id],
         'osd_metadata': mgr.get_metadata('osd', svc_id),
         'histogram': histogram,
     }

    def run(self, commands):
        """
        A static method that will execute the given list of commands in
        parallel and will return the list of command results.
        """

        # Gather the results (in parallel)
        results = []
        for index in range(len(commands)):
            tag = '%s:%d' % (str(self.id), index)

            # Store the result
            result = CommandResult(tag)
            result.command = common.humanify_command(commands[index])
            results.append(result)

            # Run the command
            context.instance.send_command(result, 'mon', '', json.dumps(commands[index]), tag)

        return results

 def get_object_pg_info(self, pool_name, object_name):
     result = CommandResult('')
     data_jaon = {}
     self.module.send_command(
         result, 'mon', '', json.dumps({
             'prefix': 'osd map',
             'format': 'json',
             'pool': pool_name,
             'object': object_name,
         }), '')
     ret, outb, outs = result.wait()
     try:
         if outb:
             data_jaon = json.loads(outb)
         else:
             self.module.log.error('unable to get %s pg info' % pool_name)
     except Exception as e:
         self.module.log.error(
             'unable to get %s pg, error: %s' % (pool_name, str(e)))
     return data_jaon

    def handle_command(self, _, cmd):
        self.log.error("handle_command")

        if cmd['prefix'] == 'device query-daemon-health-metrics':
            who = cmd.get('who', '')
            if who[0:4] != 'osd.':
                return -errno.EINVAL, '', 'not a valid <osd.NNN> id'
            osd_id = who[4:]
            result = CommandResult('')
            self.send_command(result, 'osd', osd_id, json.dumps({
                'prefix': 'smart',
                'format': 'json',
            }), '')
            r, outb, outs = result.wait()
            return r, outb, outs
        elif cmd['prefix'] == 'device scrape-daemon-health-metrics':
            who = cmd.get('who', '')
            if who[0:4] != 'osd.':
                return -errno.EINVAL, '', 'not a valid <osd.NNN> id'
            osd_id = int(who[4:])
            return self.scrape_osd(osd_id)
        elif cmd['prefix'] == 'device scrape-health-metrics':
            if 'devid' in cmd:
                return self.scrape_device(cmd['devid'])
            return self.scrape_all()
        elif cmd['prefix'] == 'device get-health-metrics':
            return self.show_device_metrics(cmd['devid'], cmd.get('sample'))
        elif cmd['prefix'] == 'device check-health':
            return self.check_health()
        elif cmd['prefix'] == 'device monitoring on':
            self.set_config('enable_monitoring', 'true')
            self.event.set()
            return 0, '', ''
        elif cmd['prefix'] == 'device monitoring off':
            self.set_config('enable_monitoring', 'false')
            self.set_health_checks({})  # avoid stuck health alerts
            return 0, '', ''
        else:
            # mgr should respect our self.COMMANDS and not call us for
            # any prefix we don't advertise
            raise NotImplementedError(cmd['prefix'])

    def set_device_life_expectancy(self, device_id, from_date, to_date=None):
        result = CommandResult('')

        if to_date is None:
            self.module.send_command(result, 'mon', '', json.dumps({
                'prefix': 'device set-life-expectancy',
                'devid': device_id,
                'from': from_date
            }), '')
        else:
            self.module.send_command(result, 'mon', '', json.dumps({
                'prefix': 'device set-life-expectancy',
                'devid': device_id,
                'from': from_date,
                'to': to_date
            }), '')
        ret, outb, outs = result.wait()
        if ret != 0:
            self.module.log.error(
                'failed to set device life expectancy, %s' % outs)
        return ret

 def load_buffer(buf, channel_name):
     result = CommandResult("")
     self.send_command(result, "mon", "", json.dumps({
         "prefix": "log last",
         "format": "json",
         "channel": channel_name,
         "num": LOG_BUFFER_SIZE
         }), "")
     r, outb, outs = result.wait()
     if r != 0:
         # Oh well.  We won't let this stop us though.
         self.log.error("Error fetching log history (r={0}, \"{1}\")".format(
             r, outs))
     else:
         try:
             lines = json.loads(outb)
         except ValueError:
             self.log.error("Error decoding log history")
         else:
             for l in lines:
                 buf.appendleft(l)