update ceph source to reef 18.2.1

[ceph.git] / ceph / src / pybind / mgr / telemetry / module.py

"""
Telemetry module for ceph-mgr

Collect statistics from Ceph cluster and send this back to the Ceph project
when user has opted-in
"""
import logging
import numbers
import enum
import errno
import hashlib
import json
import rbd
import requests
import uuid
import time
from datetime import datetime, timedelta
from prettytable import PrettyTable
from threading import Event, Lock
from collections import defaultdict
from typing import cast, Any, DefaultDict, Dict, List, Optional, Tuple, TypeVar, TYPE_CHECKING, Union

from mgr_module import CLICommand, CLIReadCommand, MgrModule, Option, OptionValue, ServiceInfoT


ALL_CHANNELS = ['basic', 'ident', 'crash', 'device', 'perf']

LICENSE = 'sharing-1-0'
LICENSE_NAME = 'Community Data License Agreement - Sharing - Version 1.0'
LICENSE_URL = 'https://cdla.io/sharing-1-0/'
NO_SALT_CNT = 0

# Latest revision of the telemetry report.  Bump this each time we make
# *any* change.
REVISION = 3

# History of revisions
# --------------------
#
# Version 1:
#   Mimic and/or nautilus are lumped together here, since
#   we didn't track revisions yet.
#
# Version 2:
#   - added revision tracking, nagging, etc.
#   - added config option changes
#   - added channels
#   - added explicit license acknowledgement to the opt-in process
#
# Version 3:
#   - added device health metrics (i.e., SMART data, minus serial number)
#   - remove crush_rule
#   - added CephFS metadata (how many MDSs, fs features, how many data pools,
#     how much metadata is cached, rfiles, rbytes, rsnapshots)
#   - added more pool metadata (rep vs ec, cache tiering mode, ec profile)
#   - added host count, and counts for hosts with each of (mon, osd, mds, mgr)
#   - whether an OSD cluster network is in use
#   - rbd pool and image count, and rbd mirror mode (pool-level)
#   - rgw daemons, zones, zonegroups; which rgw frontends
#   - crush map stats

class Collection(str, enum.Enum):
    basic_base = 'basic_base'
    device_base = 'device_base'
    crash_base = 'crash_base'
    ident_base = 'ident_base'
    perf_perf = 'perf_perf'
    basic_mds_metadata = 'basic_mds_metadata'
    basic_pool_usage = 'basic_pool_usage'
    basic_usage_by_class = 'basic_usage_by_class'
    basic_rook_v01 = 'basic_rook_v01'
    perf_memory_metrics = 'perf_memory_metrics'
    basic_pool_options_bluestore = 'basic_pool_options_bluestore'

MODULE_COLLECTION : List[Dict] = [
    {
        "name": Collection.basic_base,
        "description": "Basic information about the cluster (capacity, number and type of daemons, version, etc.)",
        "channel": "basic",
        "nag": False
    },
    {
        "name": Collection.device_base,
        "description": "Information about device health metrics",
        "channel": "device",
        "nag": False
    },
    {
        "name": Collection.crash_base,
        "description": "Information about daemon crashes (daemon type and version, backtrace, etc.)",
        "channel": "crash",
        "nag": False
    },
    {
        "name": Collection.ident_base,
        "description": "User-provided identifying information about the cluster",
        "channel": "ident",
        "nag": False
    },
    {
        "name": Collection.perf_perf,
        "description": "Information about performance counters of the cluster",
        "channel": "perf",
        "nag": True
    },
    {
        "name": Collection.basic_mds_metadata,
        "description": "MDS metadata",
        "channel": "basic",
        "nag": False
    },
    {
        "name": Collection.basic_pool_usage,
        "description": "Default pool application and usage statistics",
        "channel": "basic",
        "nag": False
    },
    {
        "name": Collection.basic_usage_by_class,
        "description": "Default device class usage statistics",
        "channel": "basic",
        "nag": False
    },
    {
        "name": Collection.basic_rook_v01,
        "description": "Basic Rook deployment data",
        "channel": "basic",
        "nag": True
    },
    {
        "name": Collection.perf_memory_metrics,
        "description": "Heap stats and mempools for mon and mds",
        "channel": "perf",
        "nag": False
    },
    {
        "name": Collection.basic_pool_options_bluestore,
        "description": "Per-pool bluestore config options",
        "channel": "basic",
        "nag": False
    },
]

ROOK_KEYS_BY_COLLECTION : List[Tuple[str, Collection]] = [
        # Note: a key cannot be both a node and a leaf, e.g.
        # "rook/a/b"
        # "rook/a/b/c"
        ("rook/version", Collection.basic_rook_v01),
        ("rook/kubernetes/version", Collection.basic_rook_v01),
        ("rook/csi/version", Collection.basic_rook_v01),
        ("rook/node/count/kubernetes-total", Collection.basic_rook_v01),
        ("rook/node/count/with-ceph-daemons", Collection.basic_rook_v01),
        ("rook/node/count/with-csi-rbd-plugin", Collection.basic_rook_v01),
        ("rook/node/count/with-csi-cephfs-plugin", Collection.basic_rook_v01),
        ("rook/node/count/with-csi-nfs-plugin", Collection.basic_rook_v01),
        ("rook/usage/storage-class/count/total", Collection.basic_rook_v01),
        ("rook/usage/storage-class/count/rbd", Collection.basic_rook_v01),
        ("rook/usage/storage-class/count/cephfs", Collection.basic_rook_v01),
        ("rook/usage/storage-class/count/nfs", Collection.basic_rook_v01),
        ("rook/usage/storage-class/count/bucket", Collection.basic_rook_v01),
        ("rook/cluster/storage/device-set/count/total", Collection.basic_rook_v01),
        ("rook/cluster/storage/device-set/count/portable", Collection.basic_rook_v01),
        ("rook/cluster/storage/device-set/count/non-portable", Collection.basic_rook_v01),
        ("rook/cluster/mon/count", Collection.basic_rook_v01),
        ("rook/cluster/mon/allow-multiple-per-node", Collection.basic_rook_v01),
        ("rook/cluster/mon/max-id", Collection.basic_rook_v01),
        ("rook/cluster/mon/pvc/enabled", Collection.basic_rook_v01),
        ("rook/cluster/mon/stretch/enabled", Collection.basic_rook_v01),
        ("rook/cluster/network/provider", Collection.basic_rook_v01),
        ("rook/cluster/external-mode", Collection.basic_rook_v01),
]

class Module(MgrModule):
    metadata_keys = [
        "arch",
        "ceph_version",
        "os",
        "cpu",
        "kernel_description",
        "kernel_version",
        "distro_description",
        "distro"
    ]

    MODULE_OPTIONS = [
        Option(name='url',
               type='str',
               default='https://telemetry.ceph.com/report'),
        Option(name='device_url',
               type='str',
               default='https://telemetry.ceph.com/device'),
        Option(name='enabled',
               type='bool',
               default=False),
        Option(name='last_opt_revision',
               type='int',
               default=1),
        Option(name='leaderboard',
               type='bool',
               default=False),
        Option(name='leaderboard_description',
               type='str',
               default=None),
        Option(name='description',
               type='str',
               default=None),
        Option(name='contact',
               type='str',
               default=None),
        Option(name='organization',
               type='str',
               default=None),
        Option(name='proxy',
               type='str',
               default=None),
        Option(name='interval',
               type='int',
               default=24,
               min=8),
        Option(name='channel_basic',
               type='bool',
               default=True,
               desc='Share basic cluster information (size, version)'),
        Option(name='channel_ident',
               type='bool',
               default=False,
               desc='Share a user-provided description and/or contact email for the cluster'),
        Option(name='channel_crash',
               type='bool',
               default=True,
               desc='Share metadata about Ceph daemon crashes (version, stack straces, etc)'),
        Option(name='channel_device',
               type='bool',
               default=True,
               desc=('Share device health metrics '
                     '(e.g., SMART data, minus potentially identifying info like serial numbers)')),
        Option(name='channel_perf',
               type='bool',
               default=False,
               desc='Share various performance metrics of a cluster'),
    ]

    @property
    def config_keys(self) -> Dict[str, OptionValue]:
        return dict((o['name'], o.get('default', None)) for o in self.MODULE_OPTIONS)

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super(Module, self).__init__(*args, **kwargs)
        self.event = Event()
        self.run = False
        self.db_collection: Optional[List[str]] = None
        self.last_opted_in_ceph_version: Optional[int] = None
        self.last_opted_out_ceph_version: Optional[int] = None
        self.last_upload: Optional[int] = None
        self.last_report: Dict[str, Any] = dict()
        self.report_id: Optional[str] = None
        self.salt: Optional[str] = None
        self.get_report_lock = Lock()
        self.config_update_module_option()
        # for mypy which does not run the code
        if TYPE_CHECKING:
            self.url = ''
            self.device_url = ''
            self.enabled = False
            self.last_opt_revision = 0
            self.leaderboard = ''
            self.leaderboard_description = ''
            self.interval = 0
            self.proxy = ''
            self.channel_basic = True
            self.channel_ident = False
            self.channel_crash = True
            self.channel_device = True
            self.channel_perf = False
            self.db_collection = ['basic_base', 'device_base']
            self.last_opted_in_ceph_version = 17
            self.last_opted_out_ceph_version = 0

    def config_update_module_option(self) -> None:
        for opt in self.MODULE_OPTIONS:
            setattr(self,
                    opt['name'],
                    self.get_module_option(opt['name']))
            self.log.debug(' %s = %s', opt['name'], getattr(self, opt['name']))

    def config_notify(self) -> None:
        self.config_update_module_option()
        # wake up serve() thread
        self.event.set()

    def load(self) -> None:
        last_upload = self.get_store('last_upload', None)
        if last_upload is None:
            self.last_upload = None
        else:
            self.last_upload = int(last_upload)

        report_id = self.get_store('report_id', None)
        if report_id is None:
            self.report_id = str(uuid.uuid4())
            self.set_store('report_id', self.report_id)
        else:
            self.report_id = report_id

        salt = self.get_store('salt', None)
        if salt is None:
            self.salt = str(uuid.uuid4())
            self.set_store('salt', self.salt)
        else:
            self.salt = salt

        self.init_collection()

        last_opted_in_ceph_version = self.get_store('last_opted_in_ceph_version', None)
        if last_opted_in_ceph_version is None:
            self.last_opted_in_ceph_version = None
        else:
            self.last_opted_in_ceph_version = int(last_opted_in_ceph_version)

        last_opted_out_ceph_version = self.get_store('last_opted_out_ceph_version', None)
        if last_opted_out_ceph_version is None:
            self.last_opted_out_ceph_version = None
        else:
            self.last_opted_out_ceph_version = int(last_opted_out_ceph_version)

    def gather_osd_metadata(self,
                            osd_map: Dict[str, List[Dict[str, int]]]) -> Dict[str, Dict[str, int]]:
        keys = ["osd_objectstore", "rotational"]
        keys += self.metadata_keys

        metadata: Dict[str, Dict[str, int]] = dict()
        for key in keys:
            metadata[key] = defaultdict(int)

        for osd in osd_map['osds']:
            res = self.get_metadata('osd', str(osd['osd']))
            if res is None:
                self.log.debug('Could not get metadata for osd.%s' % str(osd['osd']))
                continue
            for k, v in res.items():
                if k not in keys:
                    continue

                metadata[k][v] += 1

        return metadata

    def gather_mon_metadata(self,
                            mon_map: Dict[str, List[Dict[str, str]]]) -> Dict[str, Dict[str, int]]:
        keys = list()
        keys += self.metadata_keys

        metadata: Dict[str, Dict[str, int]] = dict()
        for key in keys:
            metadata[key] = defaultdict(int)

        for mon in mon_map['mons']:
            res = self.get_metadata('mon', mon['name'])
            if res is None:
                self.log.debug('Could not get metadata for mon.%s' % (mon['name']))
                continue
            for k, v in res.items():
                if k not in keys:
                    continue

                metadata[k][v] += 1

        return metadata

    def gather_mds_metadata(self) -> Dict[str, Dict[str, int]]:
        metadata: Dict[str, Dict[str, int]] = dict()

        res = self.get('mds_metadata')  # metadata of *all* mds daemons
        if res is None or not res:
            self.log.debug('Could not get metadata for mds daemons')
            return metadata

        keys = list()
        keys += self.metadata_keys

        for key in keys:
            metadata[key] = defaultdict(int)

        for mds in res.values():
            for k, v in mds.items():
                if k not in keys:
                    continue

                metadata[k][v] += 1

        return metadata

    def gather_crush_info(self) -> Dict[str, Union[int,
                                                   bool,
                                                   List[int],
                                                   Dict[str, int],
                                                   Dict[int, int]]]:
        osdmap = self.get_osdmap()
        crush_raw = osdmap.get_crush()
        crush = crush_raw.dump()

        BucketKeyT = TypeVar('BucketKeyT', int, str)

        def inc(d: Dict[BucketKeyT, int], k: BucketKeyT) -> None:
            if k in d:
                d[k] += 1
            else:
                d[k] = 1

        device_classes: Dict[str, int] = {}
        for dev in crush['devices']:
            inc(device_classes, dev.get('class', ''))

        bucket_algs: Dict[str, int] = {}
        bucket_types: Dict[str, int] = {}
        bucket_sizes: Dict[int, int] = {}
        for bucket in crush['buckets']:
            if '~' in bucket['name']:  # ignore shadow buckets
                continue
            inc(bucket_algs, bucket['alg'])
            inc(bucket_types, bucket['type_id'])
            inc(bucket_sizes, len(bucket['items']))

        return {
            'num_devices': len(crush['devices']),
            'num_types': len(crush['types']),
            'num_buckets': len(crush['buckets']),
            'num_rules': len(crush['rules']),
            'device_classes': list(device_classes.values()),
            'tunables': crush['tunables'],
            'compat_weight_set': '-1' in crush['choose_args'],
            'num_weight_sets': len(crush['choose_args']),
            'bucket_algs': bucket_algs,
            'bucket_sizes': bucket_sizes,
            'bucket_types': bucket_types,
        }

    def gather_configs(self) -> Dict[str, List[str]]:
        # cluster config options
        cluster = set()
        r, outb, outs = self.mon_command({
            'prefix': 'config dump',
            'format': 'json'
        })
        if r != 0:
            return {}
        try:
            dump = json.loads(outb)
        except json.decoder.JSONDecodeError:
            return {}
        for opt in dump:
            name = opt.get('name')
            if name:
                cluster.add(name)
        # daemon-reported options (which may include ceph.conf)
        active = set()
        ls = self.get("modified_config_options")
        for opt in ls.get('options', {}):
            active.add(opt)
        return {
            'cluster_changed': sorted(list(cluster)),
            'active_changed': sorted(list(active)),
        }

    def anonymize_entity_name(self, entity_name:str) -> str:
        if '.' not in entity_name:
            self.log.debug(f"Cannot split entity name ({entity_name}), no '.' is found")
            return entity_name

        (etype, eid) = entity_name.split('.', 1)
        m = hashlib.sha1()
        salt = ''
        if self.salt is not None:
            salt = self.salt
        # avoid asserting that salt exists
        if not self.salt:
            # do not set self.salt to a temp value
            salt = f"no_salt_found_{NO_SALT_CNT}"
            NO_SALT_CNT += 1
            self.log.debug(f"No salt found, created a temp one: {salt}")
        m.update(salt.encode('utf-8'))
        m.update(eid.encode('utf-8'))
        m.update(salt.encode('utf-8'))

        return  etype + '.' + m.hexdigest()

    def get_heap_stats(self) -> Dict[str, dict]:
        result: Dict[str, dict] = defaultdict(lambda: defaultdict(lambda: defaultdict(int)))
        anonymized_daemons = {}
        osd_map = self.get('osd_map')

        # Combine available daemons
        daemons = []
        for osd in osd_map['osds']:
            daemons.append('osd'+'.'+str(osd['osd']))
        # perf_memory_metrics collection (1/2)
        if self.is_enabled_collection(Collection.perf_memory_metrics):
            mon_map = self.get('mon_map')
            mds_metadata = self.get('mds_metadata')
            for mon in mon_map['mons']:
                daemons.append('mon'+'.'+mon['name'])
            for mds in mds_metadata:
                daemons.append('mds'+'.'+mds)

        # Grab output from the "daemon.x heap stats" command
        for daemon in daemons:
            daemon_type, daemon_id = daemon.split('.', 1)
            heap_stats = self.parse_heap_stats(daemon_type, daemon_id)
            if heap_stats:
                if (daemon_type != 'osd'):
                    # Anonymize mon and mds
                    anonymized_daemons[daemon] = self.anonymize_entity_name(daemon)
                    daemon = anonymized_daemons[daemon]
                result[daemon_type][daemon] = heap_stats
            else:
                continue

        if anonymized_daemons:
            # for debugging purposes only, this data is never reported
            self.log.debug('Anonymized daemon mapping for telemetry heap_stats (anonymized: real): {}'.format(anonymized_daemons))
        return result

    def parse_heap_stats(self, daemon_type: str, daemon_id: Any) -> Dict[str, int]:
        parsed_output = {}

        cmd_dict = {
            'prefix': 'heap',
            'heapcmd': 'stats'
        }
        r, outb, outs = self.tell_command(daemon_type, str(daemon_id), cmd_dict)

        if r != 0:
            self.log.error("Invalid command dictionary: {}".format(cmd_dict))
        else:
            if 'tcmalloc heap stats' in outb:
                values = [int(i) for i in outb.split() if i.isdigit()]
                # `categories` must be ordered this way for the correct output to be parsed
                categories = ['use_by_application',
                              'page_heap_freelist',
                              'central_cache_freelist',
                              'transfer_cache_freelist',
                              'thread_cache_freelists',
                              'malloc_metadata',
                              'actual_memory_used',
                              'released_to_os',
                              'virtual_address_space_used',
                              'spans_in_use',
                              'thread_heaps_in_use',
                              'tcmalloc_page_size']
                if len(values) != len(categories):
                    self.log.error('Received unexpected output from {}.{}; ' \
                                   'number of values should match the number' \
                                   'of expected categories:\n values: len={} {} '\
                                   '~ categories: len={} {} ~ outs: {}'.format(daemon_type, daemon_id, len(values), values, len(categories), categories, outs))
                else:
                    parsed_output = dict(zip(categories, values))
            else:
                self.log.error('No heap stats available on {}.{}: {}'.format(daemon_type, daemon_id, outs))
        
        return parsed_output

    def get_mempool(self, mode: str = 'separated') -> Dict[str, dict]:
        result: Dict[str, dict] = defaultdict(lambda: defaultdict(lambda: defaultdict(int)))
        anonymized_daemons = {}
        osd_map = self.get('osd_map')

        # Combine available daemons
        daemons = []
        for osd in osd_map['osds']:
            daemons.append('osd'+'.'+str(osd['osd']))
        # perf_memory_metrics collection (2/2)
        if self.is_enabled_collection(Collection.perf_memory_metrics):
            mon_map = self.get('mon_map')
            mds_metadata = self.get('mds_metadata')
            for mon in mon_map['mons']:
                daemons.append('mon'+'.'+mon['name'])
            for mds in mds_metadata:
                daemons.append('mds'+'.'+mds)

        # Grab output from the "dump_mempools" command
        for daemon in daemons:
            daemon_type, daemon_id = daemon.split('.', 1)
            cmd_dict = {
                'prefix': 'dump_mempools',
                'format': 'json'
            }
            r, outb, outs = self.tell_command(daemon_type, daemon_id, cmd_dict)
            if r != 0:
                self.log.error("Invalid command dictionary: {}".format(cmd_dict))
                continue
            else:
                try:
                    # This is where the mempool will land.
                    dump = json.loads(outb)
                    if mode == 'separated':
                        # Anonymize mon and mds
                        if daemon_type != 'osd':
                            anonymized_daemons[daemon] = self.anonymize_entity_name(daemon)
                            daemon = anonymized_daemons[daemon]
                        result[daemon_type][daemon] = dump['mempool']['by_pool']
                    elif mode == 'aggregated':
                        for mem_type in dump['mempool']['by_pool']:
                            result[daemon_type][mem_type]['bytes'] += dump['mempool']['by_pool'][mem_type]['bytes']
                            result[daemon_type][mem_type]['items'] += dump['mempool']['by_pool'][mem_type]['items']
                    else:
                        self.log.error("Incorrect mode specified in get_mempool: {}".format(mode))
                except (json.decoder.JSONDecodeError, KeyError) as e:
                    self.log.exception("Error caught on {}.{}: {}".format(daemon_type, daemon_id, e))
                    continue

        if anonymized_daemons:
            # for debugging purposes only, this data is never reported
            self.log.debug('Anonymized daemon mapping for telemetry mempool (anonymized: real): {}'.format(anonymized_daemons))

        return result

    def get_osd_histograms(self, mode: str = 'separated') -> List[Dict[str, dict]]:
        # Initialize result dict
        result: Dict[str, dict] = defaultdict(lambda: defaultdict(
                                              lambda: defaultdict(
                                              lambda: defaultdict(
                                              lambda: defaultdict(
                                              lambda: defaultdict(int))))))

        # Get list of osd ids from the metadata
        osd_metadata = self.get('osd_metadata')

        # Grab output from the "osd.x perf histogram dump" command
        for osd_id in osd_metadata:
            cmd_dict = {
                'prefix': 'perf histogram dump',
                'id': str(osd_id),
                'format': 'json'
            }
            r, outb, outs = self.osd_command(cmd_dict)
            # Check for invalid calls
            if r != 0:
                self.log.error("Invalid command dictionary: {}".format(cmd_dict))
                continue
            else:
                try:
                    # This is where the histograms will land if there are any.
                    dump = json.loads(outb)

                    for histogram in dump['osd']:
                        # Log axis information. There are two axes, each represented
                        # as a dictionary. Both dictionaries are contained inside a
                        # list called 'axes'.
                        axes = []
                        for axis in dump['osd'][histogram]['axes']:

                            # This is the dict that contains information for an individual
                            # axis. It will be appended to the 'axes' list at the end.
                            axis_dict: Dict[str, Any] = defaultdict()

                            # Collecting information for buckets, min, name, etc.
                            axis_dict['buckets'] = axis['buckets']
                            axis_dict['min'] = axis['min']
                            axis_dict['name'] = axis['name']
                            axis_dict['quant_size'] = axis['quant_size']
                            axis_dict['scale_type'] = axis['scale_type']

                            # Collecting ranges; placing them in lists to
                            # improve readability later on.
                            ranges = []
                            for _range in axis['ranges']:
                                _max, _min = None, None
                                if 'max' in _range:
                                    _max = _range['max']
                                if 'min' in _range:
                                    _min = _range['min']
                                ranges.append([_min, _max])
                            axis_dict['ranges'] = ranges

                            # Now that 'axis_dict' contains all the appropriate
                            # information for the current axis, append it to the 'axes' list.
                            # There will end up being two axes in the 'axes' list, since the
                            # histograms are 2D.
                            axes.append(axis_dict)

                        # Add the 'axes' list, containing both axes, to result.
                        # At this point, you will see that the name of the key is the string
                        # form of our axes list (str(axes)). This is there so that histograms
                        # with different axis configs will not be combined.
                        # These key names are later dropped when only the values are returned.
                        result[str(axes)][histogram]['axes'] = axes

                        # Collect current values and make sure they are in
                        # integer form.
                        values = []
                        for value_list in dump['osd'][histogram]['values']:
                            values.append([int(v) for v in value_list])

                        if mode == 'separated':
                            if 'osds' not in result[str(axes)][histogram]:
                                result[str(axes)][histogram]['osds'] = []
                            result[str(axes)][histogram]['osds'].append({'osd_id': int(osd_id), 'values': values})

                        elif mode == 'aggregated':
                            # Aggregate values. If 'values' have already been initialized,
                            # we can safely add.
                            if 'values' in result[str(axes)][histogram]:
                                for i in range (0, len(values)):
                                    for j in range (0, len(values[i])):
                                        values[i][j] += result[str(axes)][histogram]['values'][i][j]

                            # Add the values to result.
                            result[str(axes)][histogram]['values'] = values

                            # Update num_combined_osds
                            if 'num_combined_osds' not in result[str(axes)][histogram]:
                                result[str(axes)][histogram]['num_combined_osds'] = 1
                            else:
                                result[str(axes)][histogram]['num_combined_osds'] += 1
                        else:
                            self.log.error('Incorrect mode specified in get_osd_histograms: {}'.format(mode))
                            return list()

                # Sometimes, json errors occur if you give it an empty string.
                # I am also putting in a catch for a KeyError since it could
                # happen where the code is assuming that a key exists in the
                # schema when it doesn't. In either case, we'll handle that
                # by continuing and collecting what we can from other osds.
                except (json.decoder.JSONDecodeError, KeyError) as e:
                    self.log.exception("Error caught on osd.{}: {}".format(osd_id, e))
                    continue

        return list(result.values())

    def get_io_rate(self) -> dict:
        return self.get('io_rate')

    def get_stats_per_pool(self) -> dict:
        result = self.get('pg_dump')['pool_stats']

        # collect application metadata from osd_map
        osd_map = self.get('osd_map')
        application_metadata = {pool['pool']: pool['application_metadata'] for pool in osd_map['pools']}

        # add application to each pool from pg_dump
        for pool in result:
            pool['application'] = []
            # Only include default applications
            for application in application_metadata[pool['poolid']]:
                if application in ['cephfs', 'mgr', 'rbd', 'rgw']:
                    pool['application'].append(application)

        return result

    def get_stats_per_pg(self) -> dict:
        return self.get('pg_dump')['pg_stats']

    def get_rocksdb_stats(self) -> Dict[str, str]:
        # Initalizers
        result: Dict[str, str] = defaultdict()
        version = self.get_rocksdb_version()

        # Update result
        result['version'] = version

        return result

    def gather_crashinfo(self) -> List[Dict[str, str]]:
        crashlist: List[Dict[str, str]] = list()
        errno, crashids, err = self.remote('crash', 'ls')
        if errno:
            return crashlist
        for crashid in crashids.split():
            errno, crashinfo, err = self.remote('crash', 'do_info', crashid)
            if errno:
                continue
            c = json.loads(crashinfo)

            # redact hostname
            del c['utsname_hostname']

            # entity_name might have more than one '.', beware
            (etype, eid) = c.get('entity_name', '').split('.', 1)
            m = hashlib.sha1()
            assert self.salt
            m.update(self.salt.encode('utf-8'))
            m.update(eid.encode('utf-8'))
            m.update(self.salt.encode('utf-8'))
            c['entity_name'] = etype + '.' + m.hexdigest()

            # redact final line of python tracebacks, as the exception
            # payload may contain identifying information
            if 'mgr_module' in c and 'backtrace' in c:
                # backtrace might be empty
                if len(c['backtrace']) > 0:
                    c['backtrace'][-1] = '<redacted>'

            crashlist.append(c)
        return crashlist

    def gather_perf_counters(self, mode: str = 'separated') -> Dict[str, dict]:
        # Extract perf counter data with get_unlabeled_perf_counters(), a method
        # from mgr/mgr_module.py. This method returns a nested dictionary that
        # looks a lot like perf schema, except with some additional fields.
        #
        # Example of output, a snapshot of a mon daemon:
        #   "mon.b": {
        #       "bluestore.kv_flush_lat": {
        #           "count": 2431,
        #           "description": "Average kv_thread flush latency",
        #           "nick": "fl_l",
        #           "priority": 8,
        #           "type": 5,
        #           "units": 1,
        #           "value": 88814109
        #       },
        #   },
        perf_counters = self.get_unlabeled_perf_counters()

        # Initialize 'result' dict
        result: Dict[str, dict] = defaultdict(lambda: defaultdict(
            lambda: defaultdict(lambda: defaultdict(int))))

        # 'separated' mode
        anonymized_daemon_dict = {}

        for daemon, perf_counters_by_daemon in perf_counters.items():
            daemon_type = daemon[0:3] # i.e. 'mds', 'osd', 'rgw'

            if mode == 'separated':
                # anonymize individual daemon names except osds
                if (daemon_type != 'osd'):
                    anonymized_daemon = self.anonymize_entity_name(daemon)
                    anonymized_daemon_dict[anonymized_daemon] = daemon
                    daemon = anonymized_daemon

            # Calculate num combined daemon types if in aggregated mode
            if mode == 'aggregated':
                if 'num_combined_daemons' not in result[daemon_type]:
                    result[daemon_type]['num_combined_daemons'] = 1
                else:
                    result[daemon_type]['num_combined_daemons'] += 1

            for collection in perf_counters_by_daemon:
                # Split the collection to avoid redundancy in final report; i.e.:
                #   bluestore.kv_flush_lat, bluestore.kv_final_lat -->
                #   bluestore: kv_flush_lat, kv_final_lat
                col_0, col_1 = collection.split('.')

                # Debug log for empty keys. This initially was a problem for prioritycache
                # perf counters, where the col_0 was empty for certain mon counters:
                #
                # "mon.a": {                  instead of    "mon.a": {
                #      "": {                                     "prioritycache": {
                #        "cache_bytes": {...},                          "cache_bytes": {...},
                #
                # This log is here to detect any future instances of a similar issue.
                if (daemon == "") or (col_0 == "") or (col_1 == ""):
                    self.log.debug("Instance of an empty key: {}{}".format(daemon, collection))

                if mode == 'separated':
                    # Add value to result
                    result[daemon][col_0][col_1]['value'] = \
                            perf_counters_by_daemon[collection]['value']

                    # Check that 'count' exists, as not all counters have a count field.
                    if 'count' in perf_counters_by_daemon[collection]:
                        result[daemon][col_0][col_1]['count'] = \
                                perf_counters_by_daemon[collection]['count']
                elif mode == 'aggregated':
                    # Not every rgw daemon has the same schema. Specifically, each rgw daemon
                    # has a uniquely-named collection that starts off identically (i.e.
                    # "objecter-0x...") then diverges (i.e. "...55f4e778e140.op_rmw").
                    # This bit of code combines these unique counters all under one rgw instance.
                    # Without this check, the schema would remain separeted out in the final report.
                    if col_0[0:11] == "objecter-0x":
                        col_0 = "objecter-0x"

                    # Check that the value can be incremented. In some cases,
                    # the files are of type 'pair' (real-integer-pair, integer-integer pair).
                    # In those cases, the value is a dictionary, and not a number.
                    #   i.e. throttle-msgr_dispatch_throttler-hbserver["wait"]
                    if isinstance(perf_counters_by_daemon[collection]['value'], numbers.Number):
                        result[daemon_type][col_0][col_1]['value'] += \
                                perf_counters_by_daemon[collection]['value']

                    # Check that 'count' exists, as not all counters have a count field.
                    if 'count' in perf_counters_by_daemon[collection]:
                        result[daemon_type][col_0][col_1]['count'] += \
                                perf_counters_by_daemon[collection]['count']
                else:
                    self.log.error('Incorrect mode specified in gather_perf_counters: {}'.format(mode))
                    return {}

        if mode == 'separated':
            # for debugging purposes only, this data is never reported
            self.log.debug('Anonymized daemon mapping for telemetry perf_counters (anonymized: real): {}'.format(anonymized_daemon_dict))

        return result

    def get_active_channels(self) -> List[str]:
        r = []
        if self.channel_basic:
            r.append('basic')
        if self.channel_crash:
            r.append('crash')
        if self.channel_device:
            r.append('device')
        if self.channel_ident:
            r.append('ident')
        if self.channel_perf:
            r.append('perf')
        return r

    def gather_device_report(self) -> Dict[str, Dict[str, Dict[str, str]]]:
        try:
            time_format = self.remote('devicehealth', 'get_time_format')
        except Exception as e:
            self.log.debug('Unable to format time: {}'.format(e))
            return {}
        cutoff = datetime.utcnow() - timedelta(hours=self.interval * 2)
        min_sample = cutoff.strftime(time_format)

        devices = self.get('devices')['devices']
        if not devices:
            self.log.debug('Unable to get device info from the mgr.')
            return {}

        # anon-host-id -> anon-devid -> { timestamp -> record }
        res: Dict[str, Dict[str, Dict[str, str]]] = {}
        for d in devices:
            devid = d['devid']
            try:
                # this is a map of stamp -> {device info}
                m = self.remote('devicehealth', 'get_recent_device_metrics',
                                devid, min_sample)
            except Exception as e:
                self.log.error('Unable to get recent metrics from device with id "{}": {}'.format(devid, e))
                continue

            # anonymize host id
            try:
                host = d['location'][0]['host']
            except (KeyError, IndexError) as e:
                self.log.exception('Unable to get host from device with id "{}": {}'.format(devid, e))
                continue
            anon_host = self.get_store('host-id/%s' % host)
            if not anon_host:
                anon_host = str(uuid.uuid1())
                self.set_store('host-id/%s' % host, anon_host)
            serial = None
            for dev, rep in m.items():
                rep['host_id'] = anon_host
                if serial is None and 'serial_number' in rep:
                    serial = rep['serial_number']

            # anonymize device id
            anon_devid = self.get_store('devid-id/%s' % devid)
            if not anon_devid:
                # ideally devid is 'vendor_model_serial',
                # but can also be 'model_serial', 'serial'
                if '_' in devid:
                    anon_devid = f"{devid.rsplit('_', 1)[0]}_{uuid.uuid1()}"
                else:
                    anon_devid = str(uuid.uuid1())
                self.set_store('devid-id/%s' % devid, anon_devid)
            self.log.info('devid %s / %s, host %s / %s' % (devid, anon_devid,
                                                           host, anon_host))

            # anonymize the smartctl report itself
            if serial:
                m_str = json.dumps(m)
                m = json.loads(m_str.replace(serial, 'deleted'))

            if anon_host not in res:
                res[anon_host] = {}
            res[anon_host][anon_devid] = m
        return res

    def get_latest(self, daemon_type: str, daemon_name: str, stat: str) -> int:
        data = self.get_counter(daemon_type, daemon_name, stat)[stat]
        if data:
            return data[-1][1]
        else:
            return 0

    def compile_report(self, channels: Optional[List[str]] = None) -> Dict[str, Any]:
        if not channels:
            channels = self.get_active_channels()
        report = {
            'leaderboard': self.leaderboard,
            'leaderboard_description': self.leaderboard_description,
            'report_version': 1,
            'report_timestamp': datetime.utcnow().isoformat(),
            'report_id': self.report_id,
            'channels': channels,
            'channels_available': ALL_CHANNELS,
            'license': LICENSE,
            'collections_available': [c['name'].name for c in MODULE_COLLECTION],
            'collections_opted_in': [c['name'].name for c in MODULE_COLLECTION if self.is_enabled_collection(c['name'])],
        }

        if 'ident' in channels:
            for option in ['description', 'contact', 'organization']:
                report[option] = getattr(self, option)

        if 'basic' in channels:
            mon_map = self.get('mon_map')
            osd_map = self.get('osd_map')
            service_map = self.get('service_map')
            fs_map = self.get('fs_map')
            df = self.get('df')
            df_pools = {pool['id']: pool for pool in df['pools']}

            report['created'] = mon_map['created']

            # mons
            v1_mons = 0
            v2_mons = 0
            ipv4_mons = 0
            ipv6_mons = 0
            for mon in mon_map['mons']:
                for a in mon['public_addrs']['addrvec']:
                    if a['type'] == 'v2':
                        v2_mons += 1
                    elif a['type'] == 'v1':
                        v1_mons += 1
                    if a['addr'].startswith('['):
                        ipv6_mons += 1
                    else:
                        ipv4_mons += 1
            report['mon'] = {
                'count': len(mon_map['mons']),
                'features': mon_map['features'],
                'min_mon_release': mon_map['min_mon_release'],
                'v1_addr_mons': v1_mons,
                'v2_addr_mons': v2_mons,
                'ipv4_addr_mons': ipv4_mons,
                'ipv6_addr_mons': ipv6_mons,
            }

            report['config'] = self.gather_configs()

            # pools

            rbd_num_pools = 0
            rbd_num_images_by_pool = []
            rbd_mirroring_by_pool = []
            num_pg = 0
            report['pools'] = list()
            for pool in osd_map['pools']:
                num_pg += pool['pg_num']
                ec_profile = {}
                if pool['erasure_code_profile']:
                    orig = osd_map['erasure_code_profiles'].get(
                        pool['erasure_code_profile'], {})
                    ec_profile = {
                        k: orig[k] for k in orig.keys()
                        if k in ['k', 'm', 'plugin', 'technique',
                                 'crush-failure-domain', 'l']
                    }
                pool_data = {
                        'pool': pool['pool'],
                        'pg_num': pool['pg_num'],
                        'pgp_num': pool['pg_placement_num'],
                        'size': pool['size'],
                        'min_size': pool['min_size'],
                        'pg_autoscale_mode': pool['pg_autoscale_mode'],
                        'target_max_bytes': pool['target_max_bytes'],
                        'target_max_objects': pool['target_max_objects'],
                        'type': ['', 'replicated', '', 'erasure'][pool['type']],
                        'erasure_code_profile': ec_profile,
                        'cache_mode': pool['cache_mode'],
                    }

                # basic_pool_usage collection
                if self.is_enabled_collection(Collection.basic_pool_usage):
                    pool_data['application'] = []
                    for application in pool['application_metadata']:
                        # Only include default applications
                        if application in ['cephfs', 'mgr', 'rbd', 'rgw']:
                            pool_data['application'].append(application)
                    pool_stats = df_pools[pool['pool']]['stats']
                    pool_data['stats'] = { # filter out kb_used
                                            'avail_raw': pool_stats['avail_raw'],
                                            'bytes_used': pool_stats['bytes_used'],
                                            'compress_bytes_used': pool_stats['compress_bytes_used'],
                                            'compress_under_bytes': pool_stats['compress_under_bytes'],
                                            'data_bytes_used': pool_stats['data_bytes_used'],
                                            'dirty': pool_stats['dirty'],
                                            'max_avail': pool_stats['max_avail'],
                                            'objects': pool_stats['objects'],
                                            'omap_bytes_used': pool_stats['omap_bytes_used'],
                                            'percent_used': pool_stats['percent_used'],
                                            'quota_bytes': pool_stats['quota_bytes'],
                                            'quota_objects': pool_stats['quota_objects'],
                                            'rd': pool_stats['rd'],
                                            'rd_bytes': pool_stats['rd_bytes'],
                                            'stored': pool_stats['stored'],
                                            'stored_data': pool_stats['stored_data'],
                                            'stored_omap': pool_stats['stored_omap'],
                                            'stored_raw': pool_stats['stored_raw'],
                                            'wr': pool_stats['wr'],
                                            'wr_bytes': pool_stats['wr_bytes']
                        }
                    pool_data['options'] = {}
                    # basic_pool_options_bluestore collection
                    if self.is_enabled_collection(Collection.basic_pool_options_bluestore):
                        bluestore_options = ['compression_algorithm',
                                             'compression_mode',
                                             'compression_required_ratio',
                                             'compression_min_blob_size',
                                             'compression_max_blob_size']
                        for option in bluestore_options:
                            if option in pool['options']:
                                pool_data['options'][option] = pool['options'][option]
                cast(List[Dict[str, Any]], report['pools']).append(pool_data)
                if 'rbd' in pool['application_metadata']:
                    rbd_num_pools += 1
                    ioctx = self.rados.open_ioctx(pool['pool_name'])
                    rbd_num_images_by_pool.append(
                        sum(1 for _ in rbd.RBD().list2(ioctx)))
                    rbd_mirroring_by_pool.append(
                        rbd.RBD().mirror_mode_get(ioctx) != rbd.RBD_MIRROR_MODE_DISABLED)
            report['rbd'] = {
                'num_pools': rbd_num_pools,
                'num_images_by_pool': rbd_num_images_by_pool,
                'mirroring_by_pool': rbd_mirroring_by_pool}

            # osds
            cluster_network = False
            for osd in osd_map['osds']:
                if osd['up'] and not cluster_network:
                    front_ip = osd['public_addrs']['addrvec'][0]['addr'].split(':')[0]
                    back_ip = osd['cluster_addrs']['addrvec'][0]['addr'].split(':')[0]
                    if front_ip != back_ip:
                        cluster_network = True
            report['osd'] = {
                'count': len(osd_map['osds']),
                'require_osd_release': osd_map['require_osd_release'],
                'require_min_compat_client': osd_map['require_min_compat_client'],
                'cluster_network': cluster_network,
            }

            # crush
            report['crush'] = self.gather_crush_info()

            # cephfs
            report['fs'] = {
                'count': len(fs_map['filesystems']),
                'feature_flags': fs_map['feature_flags'],
                'num_standby_mds': len(fs_map['standbys']),
                'filesystems': [],
            }
            num_mds = len(fs_map['standbys'])
            for fsm in fs_map['filesystems']:
                fs = fsm['mdsmap']
                num_sessions = 0
                cached_ino = 0
                cached_dn = 0
                cached_cap = 0
                subtrees = 0
                rfiles = 0
                rbytes = 0
                rsnaps = 0
                for gid, mds in fs['info'].items():
                    num_sessions += self.get_latest('mds', mds['name'],
                                                    'mds_sessions.session_count')
                    cached_ino += self.get_latest('mds', mds['name'],
                                                  'mds_mem.ino')
                    cached_dn += self.get_latest('mds', mds['name'],
                                                 'mds_mem.dn')
                    cached_cap += self.get_latest('mds', mds['name'],
                                                  'mds_mem.cap')
                    subtrees += self.get_latest('mds', mds['name'],
                                                'mds.subtrees')
                    if mds['rank'] == 0:
                        rfiles = self.get_latest('mds', mds['name'],
                                                 'mds.root_rfiles')
                        rbytes = self.get_latest('mds', mds['name'],
                                                 'mds.root_rbytes')
                        rsnaps = self.get_latest('mds', mds['name'],
                                                 'mds.root_rsnaps')
                report['fs']['filesystems'].append({  # type: ignore
                    'max_mds': fs['max_mds'],
                    'ever_allowed_features': fs['ever_allowed_features'],
                    'explicitly_allowed_features': fs['explicitly_allowed_features'],
                    'num_in': len(fs['in']),
                    'num_up': len(fs['up']),
                    'num_standby_replay': len(
                        [mds for gid, mds in fs['info'].items()
                         if mds['state'] == 'up:standby-replay']),
                    'num_mds': len(fs['info']),
                    'num_sessions': num_sessions,
                    'cached_inos': cached_ino,
                    'cached_dns': cached_dn,
                    'cached_caps': cached_cap,
                    'cached_subtrees': subtrees,
                    'balancer_enabled': len(fs['balancer']) > 0,
                    'num_data_pools': len(fs['data_pools']),
                    'standby_count_wanted': fs['standby_count_wanted'],
                    'approx_ctime': fs['created'][0:7],
                    'files': rfiles,
                    'bytes': rbytes,
                    'snaps': rsnaps,
                })
                num_mds += len(fs['info'])
            report['fs']['total_num_mds'] = num_mds  # type: ignore

            # daemons
            report['metadata'] = dict(osd=self.gather_osd_metadata(osd_map),
                                      mon=self.gather_mon_metadata(mon_map))

            if self.is_enabled_collection(Collection.basic_mds_metadata):
                report['metadata']['mds'] = self.gather_mds_metadata()  # type: ignore

            # host counts
            servers = self.list_servers()
            self.log.debug('servers %s' % servers)
            hosts = {
                'num': len([h for h in servers if h['hostname']]),
            }
            for t in ['mon', 'mds', 'osd', 'mgr']:
                nr_services = sum(1 for host in servers if
                                  any(service for service in cast(List[ServiceInfoT],
                                                                  host['services'])
                                      if service['type'] == t))
                hosts['num_with_' + t] = nr_services
            report['hosts'] = hosts

            report['usage'] = {
                'pools': len(df['pools']),
                'pg_num': num_pg,
                'total_used_bytes': df['stats']['total_used_bytes'],
                'total_bytes': df['stats']['total_bytes'],
                'total_avail_bytes': df['stats']['total_avail_bytes']
            }
            # basic_usage_by_class collection
            if self.is_enabled_collection(Collection.basic_usage_by_class):
                report['usage']['stats_by_class'] = {} # type: ignore
                for device_class in df['stats_by_class']:
                    if device_class in ['hdd', 'ssd', 'nvme']:
                        report['usage']['stats_by_class'][device_class] = df['stats_by_class'][device_class] # type: ignore

            services: DefaultDict[str, int] = defaultdict(int)
            for key, value in service_map['services'].items():
                services[key] += 1
                if key == 'rgw':
                    rgw = {}
                    zones = set()
                    zonegroups = set()
                    frontends = set()
                    count = 0
                    d = value.get('daemons', dict())
                    for k, v in d.items():
                        if k == 'summary' and v:
                            rgw[k] = v
                        elif isinstance(v, dict) and 'metadata' in v:
                            count += 1
                            zones.add(v['metadata']['zone_id'])
                            zonegroups.add(v['metadata']['zonegroup_id'])
                            frontends.add(v['metadata']['frontend_type#0'])

                            # we could actually iterate over all the keys of
                            # the dict and check for how many frontends there
                            # are, but it is unlikely that one would be running
                            # more than 2 supported ones
                            f2 = v['metadata'].get('frontend_type#1', None)
                            if f2:
                                frontends.add(f2)

                    rgw['count'] = count
                    rgw['zones'] = len(zones)
                    rgw['zonegroups'] = len(zonegroups)
                    rgw['frontends'] = list(frontends)  # sets aren't json-serializable
                    report['rgw'] = rgw
            report['services'] = services

            try:
                report['balancer'] = self.remote('balancer', 'gather_telemetry')
            except ImportError:
                report['balancer'] = {
                    'active': False
                }

            # Rook
            self.get_rook_data(report)

        if 'crash' in channels:
            report['crashes'] = self.gather_crashinfo()

        if 'perf' in channels:
            if self.is_enabled_collection(Collection.perf_perf):
                report['perf_counters'] = self.gather_perf_counters('separated')
                report['stats_per_pool'] = self.get_stats_per_pool()
                report['stats_per_pg'] = self.get_stats_per_pg()
                report['io_rate'] = self.get_io_rate()
                report['osd_perf_histograms'] = self.get_osd_histograms('separated')
                report['mempool'] = self.get_mempool('separated')
                report['heap_stats'] = self.get_heap_stats()
                report['rocksdb_stats'] = self.get_rocksdb_stats()

        # NOTE: We do not include the 'device' channel in this report; it is
        # sent to a different endpoint.

        return report

    def get_rook_data(self, report: Dict[str, object]) -> None:
        r, outb, outs = self.mon_command({
            'prefix': 'config-key dump',
            'format': 'json'
        })
        if r != 0:
            return
        try:
            config_kv_dump = json.loads(outb)
        except json.decoder.JSONDecodeError:
            return

        for elem in ROOK_KEYS_BY_COLLECTION:
            # elem[0] is the full key path (e.g. "rook/node/count/with-csi-nfs-plugin")
            # elem[1] is the Collection this key belongs to
            if self.is_enabled_collection(elem[1]):
                self.add_kv_to_report(report, elem[0], config_kv_dump.get(elem[0]))

    def add_kv_to_report(self, report: Dict[str, object], key_path: str, value: Any) -> None:
        last_node = key_path.split('/')[-1]
        for node in key_path.split('/')[0:-1]:
            if node not in report:
                report[node] = {}
            report = report[node]  # type: ignore

            # sanity check of keys correctness
            if not isinstance(report, dict):
                self.log.error(f"'{key_path}' is an invalid key, expected type 'dict' but got {type(report)}")
                return

        if last_node in report:
            self.log.error(f"'{key_path}' is an invalid key, last part must not exist at this point")
            return

        report[last_node] = value

    def _try_post(self, what: str, url: str, report: Dict[str, Dict[str, str]]) -> Optional[str]:
        self.log.info('Sending %s to: %s' % (what, url))
        proxies = dict()
        if self.proxy:
            self.log.info('Send using HTTP(S) proxy: %s', self.proxy)
            proxies['http'] = self.proxy
            proxies['https'] = self.proxy
        try:
            resp = requests.put(url=url, json=report, proxies=proxies)
            resp.raise_for_status()
        except Exception as e:
            fail_reason = 'Failed to send %s to %s: %s' % (what, url, str(e))
            self.log.error(fail_reason)
            return fail_reason
        return None

    class EndPoint(enum.Enum):
        ceph = 'ceph'
        device = 'device'

    def collection_delta(self, channels: Optional[List[str]] = None) -> Optional[List[Collection]]:
        '''
        Find collections that are available in the module, but are not in the db
        '''
        if self.db_collection is None:
            return None

        if not channels:
            channels = ALL_CHANNELS
        else:
            for ch in channels:
                if ch not in ALL_CHANNELS:
                    self.log.debug(f"invalid channel name: {ch}")
                    return None

        new_collection : List[Collection] = []

        for c in MODULE_COLLECTION:
            if c['name'].name not in self.db_collection:
                if c['channel'] in channels:
                    new_collection.append(c['name'])

        return new_collection

    def is_major_upgrade(self) -> bool:
        '''
        Returns True only if the user last opted-in to an older major
        '''
        if self.last_opted_in_ceph_version is None or self.last_opted_in_ceph_version == 0:
            # we do not know what Ceph version was when the user last opted-in,
            # thus we do not wish to nag in case of a major upgrade
            return False

        mon_map = self.get('mon_map')
        mon_min = mon_map.get("min_mon_release", 0)

        if mon_min - self.last_opted_in_ceph_version > 0:
            self.log.debug(f"major upgrade: mon_min is: {mon_min} and user last opted-in in {self.last_opted_in_ceph_version}")
            return True

        return False

    def is_opted_in(self) -> bool:
        # If len is 0 it means that the user is either opted-out (never
        # opted-in, or invoked `telemetry off`), or they upgraded from a
        # telemetry revision 1 or 2, which required to re-opt in to revision 3,
        # regardless, hence is considered as opted-out
        if self.db_collection is None:
            return False
        return len(self.db_collection) > 0

    def should_nag(self) -> bool:
        # Find delta between opted-in collections and module collections;
        # nag only if module has a collection which is not in db, and nag == True.

        # We currently do not nag if the user is opted-out (or never opted-in).
        # If we wish to do this in the future, we need to have a tri-mode state
        # (opted in, opted out, no action yet), and it needs to be guarded by a
        # config option (so that nagging can be turned off via config).
        # We also need to add a last_opted_out_ceph_version variable, for the
        # major upgrade check.

        # check if there are collections the user is not opt-in to
        # that we should nag about
        if self.db_collection is not None:
            for c in MODULE_COLLECTION:
                if c['name'].name not in self.db_collection:
                    if c['nag'] == True:
                        self.log.debug(f"The collection: {c['name']} is not reported")
                        return True

        # user might be opted-in to the most recent collection, or there is no
        # new collection which requires nagging about; thus nag in case it's a
        # major upgrade and there are new collections
        # (which their own nag == False):
        new_collections = False
        col_delta = self.collection_delta()
        if col_delta is not None and len(col_delta) > 0:
            new_collections = True

        return self.is_major_upgrade() and new_collections

    def init_collection(self) -> None:
        # We fetch from db the collections the user had already opted-in to.
        # During the transition the results will be empty, but the user might
        # be opted-in to an older version (e.g. revision = 3)

        collection = self.get_store('collection')

        if collection is not None:
            self.db_collection = json.loads(collection)

        if self.db_collection is None:
            # happens once on upgrade
            if not self.enabled:
                # user is not opted-in
                self.set_store('collection', json.dumps([]))
                self.log.debug("user is not opted-in")
            else:
                # user is opted-in, verify the revision:
                if self.last_opt_revision == REVISION:
                    self.log.debug(f"telemetry revision is {REVISION}")
                    base_collection = [Collection.basic_base.name, Collection.device_base.name, Collection.crash_base.name, Collection.ident_base.name]
                    self.set_store('collection', json.dumps(base_collection))
                else:
                    # user is opted-in to an older version, meaning they need
                    # to re-opt in regardless
                    self.set_store('collection', json.dumps([]))
                    self.log.debug(f"user is opted-in but revision is old ({self.last_opt_revision}), needs to re-opt-in")

            # reload collection after setting
            collection = self.get_store('collection')
            if collection is not None:
                self.db_collection = json.loads(collection)
            else:
                raise RuntimeError('collection is None after initial setting')
        else:
            # user has already upgraded
            self.log.debug(f"user has upgraded already: collection: {self.db_collection}")

    def is_enabled_collection(self, collection: Collection) -> bool:
        if self.db_collection is None:
            return False
        return collection.name in self.db_collection

    def opt_in_all_collections(self) -> None:
        """
        Opt-in to all collections; Update db with the currently available collections in the module
        """
        if self.db_collection is None:
            raise RuntimeError('db_collection is None after initial setting')

        for c in MODULE_COLLECTION:
            if c['name'].name not in self.db_collection:
                self.db_collection.append(c['name'])

        self.set_store('collection', json.dumps(self.db_collection))

    def send(self,
             report: Dict[str, Dict[str, str]],
             endpoint: Optional[List[EndPoint]] = None) -> Tuple[int, str, str]:
        if not endpoint:
            endpoint = [self.EndPoint.ceph, self.EndPoint.device]
        failed = []
        success = []
        self.log.debug('Send endpoints %s' % endpoint)
        for e in endpoint:
            if e == self.EndPoint.ceph:
                fail_reason = self._try_post('ceph report', self.url, report)
                if fail_reason:
                    failed.append(fail_reason)
                else:
                    now = int(time.time())
                    self.last_upload = now
                    self.set_store('last_upload', str(now))
                    success.append('Ceph report sent to {0}'.format(self.url))
                    self.log.info('Sent report to {0}'.format(self.url))
            elif e == self.EndPoint.device:
                if 'device' in self.get_active_channels():
                    devices = self.gather_device_report()
                    if devices:
                        num_devs = 0
                        num_hosts = 0
                        for host, ls in devices.items():
                            self.log.debug('host %s devices %s' % (host, ls))
                            if not len(ls):
                                continue
                            fail_reason = self._try_post('devices', self.device_url,
                                                         ls)
                            if fail_reason:
                                failed.append(fail_reason)
                            else:
                                num_devs += len(ls)
                                num_hosts += 1
                        if num_devs:
                            success.append('Reported %d devices from %d hosts across a total of %d hosts' % (
                                num_devs, num_hosts, len(devices)))
                    else:
                        fail_reason = 'Unable to send device report: Device channel is on, but the generated report was empty.'
                        failed.append(fail_reason)
                        self.log.error(fail_reason)
        if failed:
            return 1, '', '\n'.join(success + failed)
        return 0, '', '\n'.join(success)

    def format_perf_histogram(self, report: Dict[str, Any]) -> None:
        # Formatting the perf histograms so they are human-readable. This will change the
        # ranges and values, which are currently in list form, into strings so that
        # they are displayed horizontally instead of vertically.
        if 'report' in report:
            report = report['report']
        try:
            # Formatting ranges and values in osd_perf_histograms
            mode = 'osd_perf_histograms'
            for config in report[mode]:
                for histogram in config:
                    # Adjust ranges by converting lists into strings
                    for axis in config[histogram]['axes']:
                        for i in range(0, len(axis['ranges'])):
                            axis['ranges'][i] = str(axis['ranges'][i])

                    for osd in config[histogram]['osds']:
                        for i in range(0, len(osd['values'])):
                            osd['values'][i] = str(osd['values'][i])
        except KeyError:
            # If the perf channel is not enabled, there should be a KeyError since
            # 'osd_perf_histograms' would not be present in the report. In that case,
            # the show function should pass as usual without trying to format the
            # histograms.
            pass

    def toggle_channel(self, action: str, channels: Optional[List[str]] = None) -> Tuple[int, str, str]:
        '''
        Enable or disable a list of channels
        '''
        if not self.enabled:
            # telemetry should be on for channels to be toggled
            msg = 'Telemetry is off. Please consider opting-in with `ceph telemetry on`.\n' \
                  'Preview sample reports with `ceph telemetry preview`.'
            return 0, msg, ''

        if channels is None:
            msg = f'Please provide a channel name. Available channels: {ALL_CHANNELS}.'
            return 0, msg, ''

        state = action == 'enable'
        msg = ''
        for c in channels:
            if c not in ALL_CHANNELS:
                msg = f"{msg}{c} is not a valid channel name. "\
                        f"Available channels: {ALL_CHANNELS}.\n"
            else:
                self.set_module_option(f"channel_{c}", state)
                setattr(self,
                        f"channel_{c}",
                        state)
                msg = f"{msg}channel_{c} is {action}d\n"

        return 0, msg, ''

    @CLIReadCommand('telemetry status')
    def status(self) -> Tuple[int, str, str]:
        '''
        Show current configuration
        '''
        r = {}
        for opt in self.MODULE_OPTIONS:
            r[opt['name']] = getattr(self, opt['name'])
        r['last_upload'] = (time.ctime(self.last_upload)
                            if self.last_upload else self.last_upload)
        return 0, json.dumps(r, indent=4, sort_keys=True), ''

    @CLIReadCommand('telemetry diff')
    def diff(self) -> Tuple[int, str, str]:
        '''
        Show the diff between opted-in collection and available collection
        '''
        diff = []
        keys = ['nag']

        for c in MODULE_COLLECTION:
            if not self.is_enabled_collection(c['name']):
                diff.append({key: val for key, val in c.items() if key not in keys})

        r = None
        if diff == []:
            r = "Telemetry is up to date"
        else:
            r = json.dumps(diff, indent=4, sort_keys=True)

        return 0, r, ''

    @CLICommand('telemetry on')
    def on(self, license: Optional[str] = None) -> Tuple[int, str, str]:
        '''
        Enable telemetry reports from this cluster
        '''
        if license != LICENSE:
            return -errno.EPERM, '', f'''Telemetry data is licensed under the {LICENSE_NAME} ({LICENSE_URL}).
To enable, add '--license {LICENSE}' to the 'ceph telemetry on' command.'''
        else:
            self.set_module_option('enabled', True)
            self.enabled = True
            self.opt_in_all_collections()

            # for major releases upgrade nagging
            mon_map = self.get('mon_map')
            mon_min = mon_map.get("min_mon_release", 0)
            self.set_store('last_opted_in_ceph_version', str(mon_min))
            self.last_opted_in_ceph_version = mon_min

            msg = 'Telemetry is on.'
            disabled_channels = ''
            active_channels = self.get_active_channels()
            for c in ALL_CHANNELS:
                if c not in active_channels and c != 'ident':
                    disabled_channels = f"{disabled_channels} {c}"

            if len(disabled_channels) > 0:
                msg = f"{msg}\nSome channels are disabled, please enable with:\n"\
                        f"`ceph telemetry enable channel{disabled_channels}`"

            # wake up serve() to reset health warning
            self.event.set()

            return 0, msg, ''

    @CLICommand('telemetry off')
    def off(self) -> Tuple[int, str, str]:
        '''
        Disable telemetry reports from this cluster
        '''
        if not self.enabled:
            # telemetry is already off
            msg = 'Telemetry is currently not enabled, nothing to turn off. '\
                    'Please consider opting-in with `ceph telemetry on`.\n' \
                  'Preview sample reports with `ceph telemetry preview`.'
            return 0, msg, ''

        self.set_module_option('enabled', False)
        self.enabled = False
        self.set_store('collection', json.dumps([]))
        self.db_collection = []

        # we might need this info in the future, in case
        # of nagging when user is opted-out
        mon_map = self.get('mon_map')
        mon_min = mon_map.get("min_mon_release", 0)
        self.set_store('last_opted_out_ceph_version', str(mon_min))
        self.last_opted_out_ceph_version = mon_min

        msg = 'Telemetry is now disabled.'
        return 0, msg, ''

    @CLIReadCommand('telemetry enable channel all')
    def enable_channel_all(self, channels: List[str] = ALL_CHANNELS) -> Tuple[int, str, str]:
        '''
        Enable all channels
        '''
        return self.toggle_channel('enable', channels)

    @CLIReadCommand('telemetry enable channel')
    def enable_channel(self, channels: Optional[List[str]] = None) -> Tuple[int, str, str]:
        '''
        Enable a list of channels
        '''
        return self.toggle_channel('enable', channels)

    @CLIReadCommand('telemetry disable channel all')
    def disable_channel_all(self, channels: List[str] = ALL_CHANNELS) -> Tuple[int, str, str]:
        '''
        Disable all channels
        '''
        return self.toggle_channel('disable', channels)

    @CLIReadCommand('telemetry disable channel')
    def disable_channel(self, channels: Optional[List[str]] = None) -> Tuple[int, str, str]:
        '''
        Disable a list of channels
        '''
        return self.toggle_channel('disable', channels)

    @CLIReadCommand('telemetry channel ls')
    def channel_ls(self) -> Tuple[int, str, str]:
        '''
        List all channels
        '''
        table = PrettyTable(
            [
                'NAME', 'ENABLED', 'DEFAULT', 'DESC',
            ],
            border=False)
        table.align['NAME'] = 'l'
        table.align['ENABLED'] = 'l'
        table.align['DEFAULT'] = 'l'
        table.align['DESC'] = 'l'
        table.left_padding_width = 0
        table.right_padding_width = 4

        for c in ALL_CHANNELS:
            enabled = "ON" if getattr(self, f"channel_{c}") else "OFF"
            for o in self.MODULE_OPTIONS:
                if o['name'] == f"channel_{c}":
                    default = "ON" if o.get('default', None) else "OFF"
                    desc = o.get('desc', None)

            table.add_row((
                c,
                enabled,
                default,
                desc,
            ))

        return 0, table.get_string(sortby="NAME"), ''

    @CLIReadCommand('telemetry collection ls')
    def collection_ls(self) -> Tuple[int, str, str]:
        '''
        List all collections
        '''
        col_delta = self.collection_delta()
        msg = ''
        if col_delta is not None and len(col_delta) > 0:
            msg = f"New collections are available:\n" \
                  f"{sorted([c.name for c in col_delta])}\n" \
                  f"Run `ceph telemetry on` to opt-in to these collections.\n"

        table = PrettyTable(
            [
                'NAME', 'STATUS', 'DESC',
            ],
            border=False)
        table.align['NAME'] = 'l'
        table.align['STATUS'] = 'l'
        table.align['DESC'] = 'l'
        table.left_padding_width = 0
        table.right_padding_width = 4

        for c in MODULE_COLLECTION:
            name = c['name']
            opted_in = self.is_enabled_collection(name)
            channel_enabled = getattr(self, f"channel_{c['channel']}")

            status = ''
            if channel_enabled and opted_in:
                status = "REPORTING"
            else:
                why = ''
                delimiter = ''

                if not opted_in:
                    why += "NOT OPTED-IN"
                    delimiter = ', '
                if not channel_enabled:
                    why += f"{delimiter}CHANNEL {c['channel']} IS OFF"

                status = f"NOT REPORTING: {why}"

            desc = c['description']

            table.add_row((
                name,
                status,
                desc,
            ))

        if len(msg):
            # add a new line between message and table output
            msg = f"{msg} \n"

        return 0, f'{msg}{table.get_string(sortby="NAME")}', ''

    @CLICommand('telemetry send')
    def do_send(self,
                endpoint: Optional[List[EndPoint]] = None,
                license: Optional[str] = None) -> Tuple[int, str, str]:
        '''
        Send a sample report
        '''
        if not self.is_opted_in() and license != LICENSE:
            self.log.debug(('A telemetry send attempt while opted-out. '
                            'Asking for license agreement'))
            return -errno.EPERM, '', f'''Telemetry data is licensed under the {LICENSE_NAME} ({LICENSE_URL}).
To manually send telemetry data, add '--license {LICENSE}' to the 'ceph telemetry send' command.
Please consider enabling the telemetry module with 'ceph telemetry on'.'''
        else:
            self.last_report = self.compile_report()
            return self.send(self.last_report, endpoint)

    @CLIReadCommand('telemetry show')
    def show(self, channels: Optional[List[str]] = None) -> Tuple[int, str, str]:
        '''
        Show a sample report of opted-in collections (except for 'device')
        '''
        if not self.enabled:
            # if telemetry is off, no report is being sent, hence nothing to show
            msg = 'Telemetry is off. Please consider opting-in with `ceph telemetry on`.\n' \
                  'Preview sample reports with `ceph telemetry preview`.'
            return 0, msg, ''

        report = self.get_report_locked(channels=channels)
        self.format_perf_histogram(report)
        report = json.dumps(report, indent=4, sort_keys=True)

        if self.channel_device:
            report += '''\nDevice report is generated separately. To see it run 'ceph telemetry show-device'.'''

        return 0, report, ''

    @CLIReadCommand('telemetry preview')
    def preview(self, channels: Optional[List[str]] = None) -> Tuple[int, str, str]:
        '''
        Preview a sample report of the most recent collections available (except for 'device')
        '''
        report = {}

        # We use a lock to prevent a scenario where the user wishes to preview
        # the report, and at the same time the module hits the interval of
        # sending a report with the opted-in collection, which has less data
        # than in the preview report.
        col_delta = self.collection_delta()
        with self.get_report_lock:
            if col_delta is not None and len(col_delta) == 0:
                # user is already opted-in to the most recent collection
                msg = 'Telemetry is up to date, see report with `ceph telemetry show`.'
                return 0, msg, ''
            else:
                # there are collections the user is not opted-in to
                next_collection = []

                for c in MODULE_COLLECTION:
                    next_collection.append(c['name'].name)

                opted_in_collection = self.db_collection
                self.db_collection = next_collection
                report = self.get_report(channels=channels)
                self.db_collection = opted_in_collection

        self.format_perf_histogram(report)
        report = json.dumps(report, indent=4, sort_keys=True)

        if self.channel_device:
            report += '''\nDevice report is generated separately. To see it run 'ceph telemetry preview-device'.'''

        return 0, report, ''

    @CLIReadCommand('telemetry show-device')
    def show_device(self) -> Tuple[int, str, str]:
        '''
        Show a sample device report
        '''
        if not self.enabled:
            # if telemetry is off, no report is being sent, hence nothing to show
            msg = 'Telemetry is off. Please consider opting-in with `ceph telemetry on`.\n' \
                  'Preview sample device reports with `ceph telemetry preview-device`.'
            return 0, msg, ''

        if not self.channel_device:
            # if device channel is off, device report is not being sent, hence nothing to show
            msg = 'device channel is off. Please enable with `ceph telemetry enable channel device`.\n' \
                  'Preview sample device reports with `ceph telemetry preview-device`.'
            return 0, msg, ''

        return 0, json.dumps(self.get_report_locked('device'), indent=4, sort_keys=True), ''

    @CLIReadCommand('telemetry preview-device')
    def preview_device(self) -> Tuple[int, str, str]:
        '''
        Preview a sample device report of the most recent device collection
        '''
        report = {}

        device_col_delta = self.collection_delta(['device'])
        with self.get_report_lock:
            if device_col_delta is not None and len(device_col_delta) == 0 and self.channel_device:
                # user is already opted-in to the most recent device collection,
                # and device channel is on, thus `show-device` should be called
                msg = 'device channel is on and up to date, see report with `ceph telemetry show-device`.'
                return 0, msg, ''

            # either the user is not opted-in at all, or there are collections
            # they are not opted-in to
            next_collection = []

            for c in MODULE_COLLECTION:
                next_collection.append(c['name'].name)

            opted_in_collection = self.db_collection
            self.db_collection = next_collection
            report = self.get_report('device')
            self.db_collection = opted_in_collection

        report = json.dumps(report, indent=4, sort_keys=True)
        return 0, report, ''

    @CLIReadCommand('telemetry show-all')
    def show_all(self) -> Tuple[int, str, str]:
        '''
        Show a sample report of all enabled channels (including 'device' channel)
        '''
        if not self.enabled:
            # if telemetry is off, no report is being sent, hence nothing to show
            msg = 'Telemetry is off. Please consider opting-in with `ceph telemetry on`.\n' \
                  'Preview sample reports with `ceph telemetry preview`.'
            return 0, msg, ''

        if not self.channel_device:
            # device channel is off, no need to display its report
            report = self.get_report_locked('default')
        else:
            # telemetry is on and device channel is enabled, show both
            report = self.get_report_locked('all')

        self.format_perf_histogram(report)
        return 0, json.dumps(report, indent=4, sort_keys=True), ''

    @CLIReadCommand('telemetry preview-all')
    def preview_all(self) -> Tuple[int, str, str]:
        '''
        Preview a sample report of the most recent collections available of all channels (including 'device')
        '''
        report = {}

        col_delta = self.collection_delta()
        with self.get_report_lock:
            if col_delta is not None and len(col_delta) == 0:
                # user is already opted-in to the most recent collection
                msg = 'Telemetry is up to date, see report with `ceph telemetry show`.'
                return 0, msg, ''

            # there are collections the user is not opted-in to
            next_collection = []

            for c in MODULE_COLLECTION:
                next_collection.append(c['name'].name)

            opted_in_collection = self.db_collection
            self.db_collection = next_collection
            report = self.get_report('all')
            self.db_collection = opted_in_collection

        self.format_perf_histogram(report)
        report = json.dumps(report, indent=4, sort_keys=True)

        return 0, report, ''

    def get_report_locked(self,
                          report_type: str = 'default',
                          channels: Optional[List[str]] = None) -> Dict[str, Any]:
        '''
        A wrapper around get_report to allow for compiling a report of the most recent module collections
        '''
        with self.get_report_lock:
            return self.get_report(report_type, channels)

    def get_report(self,
                   report_type: str = 'default',
                   channels: Optional[List[str]] = None) -> Dict[str, Any]:
        if report_type == 'default':
            return self.compile_report(channels=channels)
        elif report_type == 'device':
            return self.gather_device_report()
        elif report_type == 'all':
            return {'report': self.compile_report(channels=channels),
                    'device_report': self.gather_device_report()}
        return {}

    def self_test(self) -> None:
        self.opt_in_all_collections()
        report = self.compile_report(channels=ALL_CHANNELS)
        if len(report) == 0:
            raise RuntimeError('Report is empty')

        if 'report_id' not in report:
            raise RuntimeError('report_id not found in report')

    def shutdown(self) -> None:
        self.run = False
        self.event.set()

    def refresh_health_checks(self) -> None:
        health_checks = {}
        # TODO do we want to nag also in case the user is not opted-in?
        if self.enabled and self.should_nag():
            health_checks['TELEMETRY_CHANGED'] = {
                'severity': 'warning',
                'summary': 'Telemetry requires re-opt-in',
                'detail': [
                    'telemetry module includes new collections; please re-opt-in to new collections with `ceph telemetry on`'
                ]
            }
        self.set_health_checks(health_checks)

    def serve(self) -> None:
        self.load()
        self.run = True

        self.log.debug('Waiting for mgr to warm up')
        time.sleep(10)

        while self.run:
            self.event.clear()

            self.refresh_health_checks()

            if not self.is_opted_in():
                self.log.debug('Not sending report until user re-opts-in')
                self.event.wait(1800)
                continue
            if not self.enabled:
                self.log.debug('Not sending report until configured to do so')
                self.event.wait(1800)
                continue

            now = int(time.time())
            if not self.last_upload or \
               (now - self.last_upload) > self.interval * 3600:
                self.log.info('Compiling and sending report to %s',
                              self.url)

                try:
                    self.last_report = self.compile_report()
                except Exception:
                    self.log.exception('Exception while compiling report:')

                self.send(self.last_report)
            else:
                self.log.debug('Interval for sending new report has not expired')

            sleep = 3600
            self.log.debug('Sleeping for %d seconds', sleep)
            self.event.wait(sleep)

    @staticmethod
    def can_run() -> Tuple[bool, str]:
        return True, ''