diff options
Diffstat (limited to 'src/pybind/mgr/stats/fs/perf_stats.py')
| -rw-r--r-- | src/pybind/mgr/stats/fs/perf_stats.py | 567 |
1 files changed, 567 insertions, 0 deletions
diff --git a/src/pybind/mgr/stats/fs/perf_stats.py b/src/pybind/mgr/stats/fs/perf_stats.py new file mode 100644 index 000000000..596b3bc08 --- /dev/null +++ b/src/pybind/mgr/stats/fs/perf_stats.py @@ -0,0 +1,567 @@ +import re +import json +import time +import uuid +import errno +import traceback +import logging +from collections import OrderedDict +from typing import List, Dict, Set + +from mgr_module import CommandResult + +from datetime import datetime, timedelta +from threading import Lock, Condition, Thread, Timer +from ipaddress import ip_address + +PERF_STATS_VERSION = 2 + +QUERY_IDS = "query_ids" +GLOBAL_QUERY_ID = "global_query_id" +QUERY_LAST_REQUEST = "last_time_stamp" +QUERY_RAW_COUNTERS = "query_raw_counters" +QUERY_RAW_COUNTERS_GLOBAL = "query_raw_counters_global" + +MDS_RANK_ALL = (-1,) +CLIENT_ID_ALL = "\d*" +CLIENT_IP_ALL = ".*" + +fs_list = [] # type: List[str] + +MDS_PERF_QUERY_REGEX_MATCH_ALL_RANKS = '^(.*)$' +MDS_PERF_QUERY_REGEX_MATCH_CLIENTS = '^(client.{0}\s+{1}):.*' +MDS_PERF_QUERY_COUNTERS_MAP = OrderedDict({'cap_hit': 0, + 'read_latency': 1, + 'write_latency': 2, + 'metadata_latency': 3, + 'dentry_lease': 4, + 'opened_files': 5, + 'pinned_icaps': 6, + 'opened_inodes': 7, + 'read_io_sizes': 8, + 'write_io_sizes': 9, + 'avg_read_latency': 10, + 'stdev_read_latency': 11, + 'avg_write_latency': 12, + 'stdev_write_latency': 13, + 'avg_metadata_latency': 14, + 'stdev_metadata_latency': 15}) +MDS_PERF_QUERY_COUNTERS = [] # type: List[str] +MDS_GLOBAL_PERF_QUERY_COUNTERS = list(MDS_PERF_QUERY_COUNTERS_MAP.keys()) + +QUERY_EXPIRE_INTERVAL = timedelta(minutes=1) +REREGISTER_TIMER_INTERVAL = 1 + +CLIENT_METADATA_KEY = "client_metadata" +CLIENT_METADATA_SUBKEYS = ["hostname", "root"] +CLIENT_METADATA_SUBKEYS_OPTIONAL = ["mount_point"] + +NON_EXISTENT_KEY_STR = "N/A" + +logger = logging.getLogger(__name__) + +class FilterSpec(object): + """ + query filters encapsulated and used as key for query map + """ + def __init__(self, mds_ranks, client_id, client_ip): + self.mds_ranks = mds_ranks + self.client_id = client_id + self.client_ip = client_ip + + def __hash__(self): + return hash((self.mds_ranks, self.client_id, self.client_ip)) + + def __eq__(self, other): + return (self.mds_ranks, self.client_id, self.client_ip) == (other.mds_ranks, other.client_id, self.client_ip) + + def __ne__(self, other): + return not(self == other) + +def extract_mds_ranks_from_spec(mds_rank_spec): + if not mds_rank_spec: + return MDS_RANK_ALL + match = re.match(r'^\d+(,\d+)*$', mds_rank_spec) + if not match: + raise ValueError("invalid mds filter spec: {}".format(mds_rank_spec)) + return tuple(int(mds_rank) for mds_rank in match.group(0).split(',')) + +def extract_client_id_from_spec(client_id_spec): + if not client_id_spec: + return CLIENT_ID_ALL + # the client id is the spec itself since it'll be a part + # of client filter regex. + if not client_id_spec.isdigit(): + raise ValueError('invalid client_id filter spec: {}'.format(client_id_spec)) + return client_id_spec + +def extract_client_ip_from_spec(client_ip_spec): + if not client_ip_spec: + return CLIENT_IP_ALL + + client_ip = client_ip_spec + if client_ip.startswith('v1:'): + client_ip = client_ip.replace('v1:', '') + elif client_ip.startswith('v2:'): + client_ip = client_ip.replace('v2:', '') + + try: + ip_address(client_ip) + return client_ip_spec + except ValueError: + raise ValueError('invalid client_ip filter spec: {}'.format(client_ip_spec)) + +def extract_mds_ranks_from_report(mds_ranks_str): + if not mds_ranks_str: + return [] + return [int(x) for x in mds_ranks_str.split(',')] + +def extract_client_id_and_ip(client): + match = re.match(r'^(client\.\d+)\s(.*)', client) + if match: + return match.group(1), match.group(2) + return None, None + +class FSPerfStats(object): + lock = Lock() + q_cv = Condition(lock) + r_cv = Condition(lock) + + user_queries = {} # type: Dict[str, Dict] + + meta_lock = Lock() + rqtimer = None + client_metadata = { + 'metadata' : {}, + 'to_purge' : set(), + 'in_progress' : {}, + } # type: Dict + + def __init__(self, module): + self.module = module + self.log = module.log + self.prev_rank0_gid = None + # report processor thread + self.report_processor = Thread(target=self.run) + self.report_processor.start() + + def set_client_metadata(self, fs_name, client_id, key, meta): + result = (self.client_metadata['metadata'].setdefault( + fs_name, {})).setdefault(client_id, {}) + if not key in result or not result[key] == meta: + result[key] = meta + + def notify_cmd(self, cmdtag): + self.log.debug("cmdtag={0}".format(cmdtag)) + with self.meta_lock: + try: + result = self.client_metadata['in_progress'].pop(cmdtag) + except KeyError: + self.log.warn(f"cmdtag {cmdtag} not found in client metadata") + return + fs_name = result[0] + client_meta = result[2].wait() + if client_meta[0] != 0: + self.log.warn("failed to fetch client metadata from gid {0}, err={1}".format( + result[1], client_meta[2])) + return + self.log.debug("notify: client metadata={0}".format(json.loads(client_meta[1]))) + for metadata in json.loads(client_meta[1]): + client_id = "client.{0}".format(metadata['id']) + result = (self.client_metadata['metadata'].setdefault(fs_name, {})).setdefault(client_id, {}) + for subkey in CLIENT_METADATA_SUBKEYS: + self.set_client_metadata(fs_name, client_id, subkey, metadata[CLIENT_METADATA_KEY][subkey]) + for subkey in CLIENT_METADATA_SUBKEYS_OPTIONAL: + self.set_client_metadata(fs_name, client_id, subkey, + metadata[CLIENT_METADATA_KEY].get(subkey, NON_EXISTENT_KEY_STR)) + metric_features = int(metadata[CLIENT_METADATA_KEY]["metric_spec"]["metric_flags"]["feature_bits"], 16) + supported_metrics = [metric for metric, bit in MDS_PERF_QUERY_COUNTERS_MAP.items() if metric_features & (1 << bit)] + self.set_client_metadata(fs_name, client_id, "valid_metrics", supported_metrics) + kver = metadata[CLIENT_METADATA_KEY].get("kernel_version", None) + if kver: + self.set_client_metadata(fs_name, client_id, "kernel_version", kver) + # when all async requests are done, purge clients metadata if any. + if not self.client_metadata['in_progress']: + global fs_list + for fs_name in fs_list: + for client in self.client_metadata['to_purge']: + try: + if client in self.client_metadata['metadata'][fs_name]: + self.log.info("purge client metadata for {0}".format(client)) + self.client_metadata['metadata'][fs_name].pop(client) + except: + pass + if fs_name in self.client_metadata['metadata'] and not bool(self.client_metadata['metadata'][fs_name]): + self.client_metadata['metadata'].pop(fs_name) + self.client_metadata['to_purge'].clear() + self.log.debug("client_metadata={0}, to_purge={1}".format( + self.client_metadata['metadata'], self.client_metadata['to_purge'])) + + def notify_fsmap(self): + #Reregister the user queries when there is a new rank0 mds + with self.lock: + gid_state = FSPerfStats.get_rank0_mds_gid_state(self.module.get('fs_map')) + if not gid_state: + return + for value in gid_state: + rank0_gid, state = value + if (rank0_gid and rank0_gid != self.prev_rank0_gid and state == 'up:active'): + #the new rank0 MDS is up:active + ua_last_updated = time.monotonic() + if (self.rqtimer and self.rqtimer.is_alive()): + self.rqtimer.cancel() + self.rqtimer = Timer(REREGISTER_TIMER_INTERVAL, + self.re_register_queries, + args=(rank0_gid, ua_last_updated,)) + self.rqtimer.start() + + def re_register_queries(self, rank0_gid, ua_last_updated): + #reregister queries if the metrics are the latest. Otherwise reschedule the timer and + #wait for the empty metrics + with self.lock: + if self.mx_last_updated >= ua_last_updated: + self.log.debug("reregistering queries...") + self.module.reregister_mds_perf_queries() + self.prev_rank0_gid = rank0_gid + else: + #reschedule the timer + self.rqtimer = Timer(REREGISTER_TIMER_INTERVAL, + self.re_register_queries, args=(rank0_gid, ua_last_updated,)) + self.rqtimer.start() + + @staticmethod + def get_rank0_mds_gid_state(fsmap): + gid_state = [] + for fs in fsmap['filesystems']: + mds_map = fs['mdsmap'] + if mds_map is not None: + for mds_id, mds_status in mds_map['info'].items(): + if mds_status['rank'] == 0: + gid_state.append([mds_status['gid'], mds_status['state']]) + if gid_state: + return gid_state + logger.warn("No rank0 mds in the fsmap") + + def update_client_meta(self): + new_updates = {} + pending_updates = [v[0] for v in self.client_metadata['in_progress'].values()] + global fs_list + fs_list.clear() + with self.meta_lock: + fsmap = self.module.get('fs_map') + for fs in fsmap['filesystems']: + mds_map = fs['mdsmap'] + if mds_map is not None: + fsname = mds_map['fs_name'] + for mds_id, mds_status in mds_map['info'].items(): + if mds_status['rank'] == 0: + fs_list.append(fsname) + rank0_gid = mds_status['gid'] + tag = str(uuid.uuid4()) + result = CommandResult(tag) + new_updates[tag] = (fsname, rank0_gid, result) + self.client_metadata['in_progress'].update(new_updates) + + self.log.debug(f"updating client metadata from {new_updates}") + + cmd_dict = {'prefix': 'client ls'} + for tag,val in new_updates.items(): + self.module.send_command(val[2], "mds", str(val[1]), json.dumps(cmd_dict), tag) + + def run(self): + try: + self.log.info("FSPerfStats::report_processor starting...") + while True: + with self.lock: + self.scrub_expired_queries() + self.process_mds_reports() + self.r_cv.notify() + + stats_period = int(self.module.get_ceph_option("mgr_stats_period")) + self.q_cv.wait(stats_period) + self.log.debug("FSPerfStats::tick") + except Exception as e: + self.log.fatal("fatal error: {}".format(traceback.format_exc())) + + def cull_mds_entries(self, raw_perf_counters, incoming_metrics, missing_clients): + # this is pretty straight forward -- find what MDSs are missing from + # what is tracked vs what we received in incoming report and purge + # the whole bunch. + tracked_ranks = raw_perf_counters.keys() + available_ranks = [int(counter['k'][0][0]) for counter in incoming_metrics] + for rank in set(tracked_ranks) - set(available_ranks): + culled = raw_perf_counters.pop(rank) + self.log.info("culled {0} client entries from rank {1} (laggy: {2})".format( + len(culled[1]), rank, "yes" if culled[0] else "no")) + missing_clients.update(list(culled[1].keys())) + + def cull_client_entries(self, raw_perf_counters, incoming_metrics, missing_clients): + # this is a bit more involed -- for each rank figure out what clients + # are missing in incoming report and purge them from our tracked map. + # but, if this is invoked after cull_mds_entries(), the rank set + # is same, so we can loop based on that assumption. + ranks = raw_perf_counters.keys() + for rank in ranks: + tracked_clients = raw_perf_counters[rank][1].keys() + available_clients = [extract_client_id_and_ip(counter['k'][1][0]) for counter in incoming_metrics] + for client in set(tracked_clients) - set([c[0] for c in available_clients if c[0] is not None]): + raw_perf_counters[rank][1].pop(client) + self.log.info("culled {0} from rank {1}".format(client, rank)) + missing_clients.add(client) + + def cull_missing_entries(self, raw_perf_counters, incoming_metrics): + missing_clients = set() # type: Set[str] + self.cull_mds_entries(raw_perf_counters, incoming_metrics, missing_clients) + self.cull_client_entries(raw_perf_counters, incoming_metrics, missing_clients) + + self.log.debug("missing_clients={0}".format(missing_clients)) + with self.meta_lock: + if self.client_metadata['in_progress']: + self.client_metadata['to_purge'].update(missing_clients) + self.log.info("deferring client metadata purge (now {0} client(s))".format( + len(self.client_metadata['to_purge']))) + else: + global fs_list + for fs_name in fs_list: + for client in missing_clients: + try: + self.log.info("purge client metadata for {0}".format(client)) + if client in self.client_metadata['metadata'][fs_name]: + self.client_metadata['metadata'][fs_name].pop(client) + except KeyError: + pass + self.log.debug("client_metadata={0}".format(self.client_metadata['metadata'])) + + def cull_global_metrics(self, raw_perf_counters, incoming_metrics): + tracked_clients = raw_perf_counters.keys() + available_clients = [counter['k'][0][0] for counter in incoming_metrics] + for client in set(tracked_clients) - set(available_clients): + raw_perf_counters.pop(client) + + def get_raw_perf_counters(self, query): + raw_perf_counters = query.setdefault(QUERY_RAW_COUNTERS, {}) + + for query_id in query[QUERY_IDS]: + result = self.module.get_mds_perf_counters(query_id) + self.log.debug("raw_perf_counters={}".format(raw_perf_counters)) + self.log.debug("get_raw_perf_counters={}".format(result)) + + # extract passed in delayed ranks. metrics for delayed ranks are tagged + # as stale. + delayed_ranks = extract_mds_ranks_from_report(result['metrics'][0][0]) + + # what's received from MDS + incoming_metrics = result['metrics'][1] + + # metrics updated (monotonic) time + self.mx_last_updated = result['metrics'][2][0] + + # cull missing MDSs and clients + self.cull_missing_entries(raw_perf_counters, incoming_metrics) + + # iterate over metrics list and update our copy (note that we have + # already culled the differences). + global fs_list + for fs_name in fs_list: + for counter in incoming_metrics: + mds_rank = int(counter['k'][0][0]) + client_id, client_ip = extract_client_id_and_ip(counter['k'][1][0]) + if self.client_metadata['metadata'].get(fs_name): + if (client_id is not None or not client_ip) and\ + self.client_metadata["metadata"][fs_name].get(client_id): # client_id _could_ be 0 + with self.meta_lock: + self.set_client_metadata(fs_name, client_id, "IP", client_ip) + else: + self.log.warn(f"client metadata for client_id={client_id} might be unavailable") + else: + self.log.warn(f"client metadata for filesystem={fs_name} might be unavailable") + + raw_counters = raw_perf_counters.setdefault(mds_rank, [False, {}]) + raw_counters[0] = True if mds_rank in delayed_ranks else False + raw_client_counters = raw_counters[1].setdefault(client_id, []) + + del raw_client_counters[:] + raw_client_counters.extend(counter['c']) + # send an asynchronous client metadata refresh + self.update_client_meta() + + def get_raw_perf_counters_global(self, query): + raw_perf_counters = query.setdefault(QUERY_RAW_COUNTERS_GLOBAL, {}) + result = self.module.get_mds_perf_counters(query[GLOBAL_QUERY_ID]) + + self.log.debug("raw_perf_counters_global={}".format(raw_perf_counters)) + self.log.debug("get_raw_perf_counters_global={}".format(result)) + + global_metrics = result['metrics'][1] + self.cull_global_metrics(raw_perf_counters, global_metrics) + for counter in global_metrics: + client_id, _ = extract_client_id_and_ip(counter['k'][0][0]) + raw_client_counters = raw_perf_counters.setdefault(client_id, []) + del raw_client_counters[:] + raw_client_counters.extend(counter['c']) + + def process_mds_reports(self): + for query in self.user_queries.values(): + self.get_raw_perf_counters(query) + self.get_raw_perf_counters_global(query) + + def scrub_expired_queries(self): + expire_time = datetime.now() - QUERY_EXPIRE_INTERVAL + for filter_spec in list(self.user_queries.keys()): + user_query = self.user_queries[filter_spec] + self.log.debug("scrubbing query={}".format(user_query)) + if user_query[QUERY_LAST_REQUEST] < expire_time: + expired_query_ids = user_query[QUERY_IDS].copy() + expired_query_ids.append(user_query[GLOBAL_QUERY_ID]) + self.log.debug("unregistering query={} ids={}".format(user_query, expired_query_ids)) + self.unregister_mds_perf_queries(filter_spec, expired_query_ids) + del self.user_queries[filter_spec] + + def prepare_mds_perf_query(self, rank, client_id, client_ip): + mds_rank_regex = MDS_PERF_QUERY_REGEX_MATCH_ALL_RANKS + if not rank == -1: + mds_rank_regex = '^({})$'.format(rank) + client_regex = MDS_PERF_QUERY_REGEX_MATCH_CLIENTS.format(client_id, client_ip) + return { + 'key_descriptor' : [ + {'type' : 'mds_rank', 'regex' : mds_rank_regex}, + {'type' : 'client_id', 'regex' : client_regex}, + ], + 'performance_counter_descriptors' : MDS_PERF_QUERY_COUNTERS, + } + + def prepare_global_perf_query(self, client_id, client_ip): + client_regex = MDS_PERF_QUERY_REGEX_MATCH_CLIENTS.format(client_id, client_ip) + return { + 'key_descriptor' : [ + {'type' : 'client_id', 'regex' : client_regex}, + ], + 'performance_counter_descriptors' : MDS_GLOBAL_PERF_QUERY_COUNTERS, + } + + def unregister_mds_perf_queries(self, filter_spec, query_ids): + self.log.info("unregister_mds_perf_queries: filter_spec={0}, query_id={1}".format( + filter_spec, query_ids)) + for query_id in query_ids: + self.module.remove_mds_perf_query(query_id) + + def register_mds_perf_query(self, filter_spec): + mds_ranks = filter_spec.mds_ranks + client_id = filter_spec.client_id + client_ip = filter_spec.client_ip + + query_ids = [] + try: + # register per-mds perf query + for rank in mds_ranks: + query = self.prepare_mds_perf_query(rank, client_id, client_ip) + self.log.info("register_mds_perf_query: {}".format(query)) + + query_id = self.module.add_mds_perf_query(query) + if query_id is None: # query id can be 0 + raise RuntimeError("failed to add MDS perf query: {}".format(query)) + query_ids.append(query_id) + except Exception: + for query_id in query_ids: + self.module.remove_mds_perf_query(query_id) + raise + return query_ids + + def register_global_perf_query(self, filter_spec): + client_id = filter_spec.client_id + client_ip = filter_spec.client_ip + + # register a global perf query for metrics + query = self.prepare_global_perf_query(client_id, client_ip) + self.log.info("register_global_perf_query: {}".format(query)) + + query_id = self.module.add_mds_perf_query(query) + if query_id is None: # query id can be 0 + raise RuntimeError("failed to add global perf query: {}".format(query)) + return query_id + + def register_query(self, filter_spec): + user_query = self.user_queries.get(filter_spec, None) + if not user_query: + user_query = { + QUERY_IDS : self.register_mds_perf_query(filter_spec), + GLOBAL_QUERY_ID : self.register_global_perf_query(filter_spec), + QUERY_LAST_REQUEST : datetime.now(), + } + self.user_queries[filter_spec] = user_query + + self.q_cv.notify() + self.r_cv.wait(5) + else: + user_query[QUERY_LAST_REQUEST] = datetime.now() + return user_query + + def generate_report(self, user_query): + result = {} # type: Dict + global fs_list + # start with counter info -- metrics that are global and per mds + result["version"] = PERF_STATS_VERSION + result["global_counters"] = MDS_GLOBAL_PERF_QUERY_COUNTERS + result["counters"] = MDS_PERF_QUERY_COUNTERS + + # fill in client metadata + raw_perfs_global = user_query.setdefault(QUERY_RAW_COUNTERS_GLOBAL, {}) + raw_perfs = user_query.setdefault(QUERY_RAW_COUNTERS, {}) + with self.meta_lock: + raw_counters_clients = [] + for val in raw_perfs.values(): + raw_counters_clients.extend(list(val[1])) + result_meta = result.setdefault("client_metadata", {}) + for fs_name in fs_list: + meta = self.client_metadata["metadata"] + if fs_name in meta and len(meta[fs_name]): + for client_id in raw_perfs_global.keys(): + if client_id in meta[fs_name] and client_id in raw_counters_clients: + client_meta = (result_meta.setdefault(fs_name, {})).setdefault(client_id, {}) + client_meta.update(meta[fs_name][client_id]) + + # start populating global perf metrics w/ client metadata + metrics = result.setdefault("global_metrics", {}) + for fs_name in fs_list: + if fs_name in meta and len(meta[fs_name]): + for client_id, counters in raw_perfs_global.items(): + if client_id in meta[fs_name] and client_id in raw_counters_clients: + global_client_metrics = (metrics.setdefault(fs_name, {})).setdefault(client_id, []) + del global_client_metrics[:] + global_client_metrics.extend(counters) + + # and, now per-mds metrics keyed by mds rank along with delayed ranks + metrics = result.setdefault("metrics", {}) + + metrics["delayed_ranks"] = [rank for rank, counters in raw_perfs.items() if counters[0]] + for rank, counters in raw_perfs.items(): + mds_key = "mds.{}".format(rank) + mds_metrics = metrics.setdefault(mds_key, {}) + mds_metrics.update(counters[1]) + return result + + def extract_query_filters(self, cmd): + mds_rank_spec = cmd.get('mds_rank', None) + client_id_spec = cmd.get('client_id', None) + client_ip_spec = cmd.get('client_ip', None) + + self.log.debug("mds_rank_spec={0}, client_id_spec={1}, client_ip_spec={2}".format( + mds_rank_spec, client_id_spec, client_ip_spec)) + + mds_ranks = extract_mds_ranks_from_spec(mds_rank_spec) + client_id = extract_client_id_from_spec(client_id_spec) + client_ip = extract_client_ip_from_spec(client_ip_spec) + + return FilterSpec(mds_ranks, client_id, client_ip) + + def get_perf_data(self, cmd): + try: + filter_spec = self.extract_query_filters(cmd) + except ValueError as e: + return -errno.EINVAL, "", str(e) + + counters = {} + with self.lock: + user_query = self.register_query(filter_spec) + result = self.generate_report(user_query) + return 0, json.dumps(result), "" |