[ceph.git] / ceph / src / pybind / mgr / rbd_support / perf.py

import errno
import json
import rados
import rbd
import time
import traceback

from datetime import datetime, timedelta
from threading import Condition, Lock, Thread
from typing import cast, Any, Callable, Dict, List, Optional, Set, Tuple, Union

from .common import (GLOBAL_POOL_KEY, authorize_request, extract_pool_key,
                     get_rbd_pools, PoolKeyT)

QUERY_POOL_ID = "pool_id"
QUERY_POOL_ID_MAP = "pool_id_map"
QUERY_IDS = "query_ids"
QUERY_SUM_POOL_COUNTERS = "pool_counters"
QUERY_RAW_POOL_COUNTERS = "raw_pool_counters"
QUERY_LAST_REQUEST = "last_request"

OSD_PERF_QUERY_REGEX_MATCH_ALL = '^(.*)$'
OSD_PERF_QUERY_COUNTERS = ['write_ops',
                           'read_ops',
                           'write_bytes',
                           'read_bytes',
                           'write_latency',
                           'read_latency']
OSD_PERF_QUERY_COUNTERS_INDICES = {
    OSD_PERF_QUERY_COUNTERS[i]: i for i in range(len(OSD_PERF_QUERY_COUNTERS))}

OSD_PERF_QUERY_LATENCY_COUNTER_INDICES = [4, 5]
OSD_PERF_QUERY_MAX_RESULTS = 256

POOL_REFRESH_INTERVAL = timedelta(minutes=5)
QUERY_EXPIRE_INTERVAL = timedelta(minutes=1)
STATS_RATE_INTERVAL = timedelta(minutes=1)

REPORT_MAX_RESULTS = 64


# {(pool_id, namespace)...}
ResolveImageNamesT = Set[Tuple[int, str]]

# (time, [value,...])
PerfCounterT = Tuple[int, List[int]]
# current, previous
RawImageCounterT = Tuple[PerfCounterT, Optional[PerfCounterT]]
# image_id => perf_counter
RawImagesCounterT = Dict[str, RawImageCounterT]
# namespace_counters => raw_images
RawNamespacesCountersT = Dict[str, RawImagesCounterT]
# pool_id => namespaces_counters
RawPoolCountersT = Dict[int, RawNamespacesCountersT]

SumImageCounterT = List[int]
# image_id => sum_image
SumImagesCounterT = Dict[str, SumImageCounterT]
# namespace => sum_images
SumNamespacesCountersT = Dict[str, SumImagesCounterT]
# pool_id, sum_namespaces
SumPoolCountersT = Dict[int, SumNamespacesCountersT]

ExtractDataFuncT = Callable[[int, Optional[RawImageCounterT], SumImageCounterT], float]


class PerfHandler:
    user_queries: Dict[PoolKeyT, Dict[str, Any]] = {}
    image_cache: Dict[str, str] = {}

    lock = Lock()
    query_condition = Condition(lock)
    refresh_condition = Condition(lock)
    thread = None

    image_name_cache: Dict[Tuple[int, str], Dict[str, str]] = {}
    image_name_refresh_time = datetime.fromtimestamp(0)

    @classmethod
    def prepare_regex(cls, value: Any) -> str:
        return '^({})$'.format(value)

    @classmethod
    def prepare_osd_perf_query(cls,
                               pool_id: Optional[int],
                               namespace: Optional[str],
                               counter_type: str) -> Dict[str, Any]:
        pool_id_regex = OSD_PERF_QUERY_REGEX_MATCH_ALL
        namespace_regex = OSD_PERF_QUERY_REGEX_MATCH_ALL
        if pool_id:
            pool_id_regex = cls.prepare_regex(pool_id)
            if namespace:
                namespace_regex = cls.prepare_regex(namespace)

        return {
            'key_descriptor': [
                {'type': 'pool_id', 'regex': pool_id_regex},
                {'type': 'namespace', 'regex': namespace_regex},
                {'type': 'object_name',
                 'regex': '^(?:rbd|journal)_data\\.(?:([0-9]+)\\.)?([^.]+)\\.'},
            ],
            'performance_counter_descriptors': OSD_PERF_QUERY_COUNTERS,
            'limit': {'order_by': counter_type,
                      'max_count': OSD_PERF_QUERY_MAX_RESULTS},
        }

    @classmethod
    def pool_spec_search_keys(cls, pool_key: str) -> List[str]:
        return [pool_key[0:len(pool_key) - x]
                for x in range(0, len(pool_key) + 1)]

    @classmethod
    def submatch_pool_key(cls, pool_key: PoolKeyT, search_key: str) -> bool:
        return ((pool_key[1] == search_key[1] or not search_key[1])
                and (pool_key[0] == search_key[0] or not search_key[0]))

    def __init__(self, module: Any) -> None:
        self.module = module
        self.log = module.log

        self.thread = Thread(target=self.run)
        self.thread.start()

    def run(self) -> None:
        try:
            self.log.info("PerfHandler: starting")
            while True:
                with self.lock:
                    self.scrub_expired_queries()
                    self.process_raw_osd_perf_counters()
                    self.refresh_condition.notify()

                    stats_period = self.module.get_ceph_option("mgr_stats_period")
                    self.query_condition.wait(stats_period)

                self.log.debug("PerfHandler: tick")

        except Exception as ex:
            self.log.fatal("Fatal runtime error: {}\n{}".format(
                ex, traceback.format_exc()))

    def merge_raw_osd_perf_counters(self,
                                    pool_key: PoolKeyT,
                                    query: Dict[str, Any],
                                    now_ts: int,
                                    resolve_image_names: ResolveImageNamesT) -> RawPoolCountersT:
        pool_id_map = query[QUERY_POOL_ID_MAP]

        # collect and combine the raw counters from all sort orders
        raw_pool_counters: Dict[int, Dict[str, Dict[str, Any]]] = query.setdefault(QUERY_RAW_POOL_COUNTERS, {})
        for query_id in query[QUERY_IDS]:
            res = self.module.get_osd_perf_counters(query_id)
            for counter in res['counters']:
                # replace pool id from object name if it exists
                k = counter['k']
                pool_id = int(k[2][0]) if k[2][0] else int(k[0][0])
                namespace = k[1][0]
                image_id = k[2][1]

                # ignore metrics from non-matching pools/namespaces
                if pool_id not in pool_id_map:
                    continue
                if pool_key[1] is not None and pool_key[1] != namespace:
                    continue

                # flag the pool (and namespace) for refresh if we cannot find
                # image name in the cache
                resolve_image_key = (pool_id, namespace)
                if image_id not in self.image_name_cache.get(resolve_image_key, {}):
                    resolve_image_names.add(resolve_image_key)

                # copy the 'sum' counter values for each image (ignore count)
                # if we haven't already processed it for this round
                raw_namespaces = raw_pool_counters.setdefault(pool_id, {})
                raw_images = raw_namespaces.setdefault(namespace, {})
                raw_image = raw_images.get(image_id)
                # save the last two perf counters for each image
                new_current = (now_ts, [int(x[0]) for x in counter['c']])
                if raw_image:
                    old_current, _ = raw_image
                    if old_current[0] < now_ts:
                        raw_images[image_id] = (new_current, old_current)
                else:
                    raw_images[image_id] = (new_current, None)

        self.log.debug("merge_raw_osd_perf_counters: {}".format(raw_pool_counters))
        return raw_pool_counters

    def sum_osd_perf_counters(self,
                              query: Dict[str, dict],
                              raw_pool_counters: RawPoolCountersT,
                              now_ts: int) -> SumPoolCountersT:
        # update the cumulative counters for each image
        sum_pool_counters = query.setdefault(QUERY_SUM_POOL_COUNTERS, {})
        for pool_id, raw_namespaces in raw_pool_counters.items():
            sum_namespaces = sum_pool_counters.setdefault(pool_id, {})
            for namespace, raw_images in raw_namespaces.items():
                sum_namespace = sum_namespaces.setdefault(namespace, {})
                for image_id, raw_image in raw_images.items():
                    # zero-out non-updated raw counters
                    if not raw_image[0]:
                        continue
                    old_current, _ = raw_image
                    if old_current[0] < now_ts:
                        new_current = (now_ts, [0] * len(old_current[1]))
                        raw_images[image_id] = (new_current, old_current)
                        continue

                    counters = old_current[1]

                    # copy raw counters if this is a newly discovered image or
                    # increment existing counters
                    sum_image = sum_namespace.setdefault(image_id, None)
                    if sum_image:
                        for i in range(len(counters)):
                            sum_image[i] += counters[i]
                    else:
                        sum_namespace[image_id] = [x for x in counters]

        self.log.debug("sum_osd_perf_counters: {}".format(sum_pool_counters))
        return sum_pool_counters

    def refresh_image_names(self, resolve_image_names: ResolveImageNamesT) -> None:
        for pool_id, namespace in resolve_image_names:
            image_key = (pool_id, namespace)
            images = self.image_name_cache.setdefault(image_key, {})
            with self.module.rados.open_ioctx2(int(pool_id)) as ioctx:
                ioctx.set_namespace(namespace)
                for image_meta in rbd.RBD().list2(ioctx):
                    images[image_meta['id']] = image_meta['name']
            self.log.debug("resolve_image_names: {}={}".format(image_key, images))

    def scrub_missing_images(self) -> None:
        for pool_key, query in self.user_queries.items():
            raw_pool_counters = query.get(QUERY_RAW_POOL_COUNTERS, {})
            sum_pool_counters = query.get(QUERY_SUM_POOL_COUNTERS, {})
            for pool_id, sum_namespaces in sum_pool_counters.items():
                raw_namespaces = raw_pool_counters.get(pool_id, {})
                for namespace, sum_images in sum_namespaces.items():
                    raw_images = raw_namespaces.get(namespace, {})

                    image_key = (pool_id, namespace)
                    image_names = self.image_name_cache.get(image_key, {})
                    for image_id in list(sum_images.keys()):
                        # scrub image counters if we failed to resolve image name
                        if image_id not in image_names:
                            self.log.debug("scrub_missing_images: dropping {}/{}".format(
                                image_key, image_id))
                            del sum_images[image_id]
                            if image_id in raw_images:
                                del raw_images[image_id]

    def process_raw_osd_perf_counters(self) -> None:
        now = datetime.now()
        now_ts = int(now.strftime("%s"))

        # clear the image name cache if we need to refresh all active pools
        if self.image_name_cache and \
                self.image_name_refresh_time + POOL_REFRESH_INTERVAL < now:
            self.log.debug("process_raw_osd_perf_counters: expiring image name cache")
            self.image_name_cache = {}

        resolve_image_names: Set[Tuple[int, str]] = set()
        for pool_key, query in self.user_queries.items():
            if not query[QUERY_IDS]:
                continue

            raw_pool_counters = self.merge_raw_osd_perf_counters(
                pool_key, query, now_ts, resolve_image_names)
            self.sum_osd_perf_counters(query, raw_pool_counters, now_ts)

        if resolve_image_names:
            self.image_name_refresh_time = now
            self.refresh_image_names(resolve_image_names)
            self.scrub_missing_images()
        elif not self.image_name_cache:
            self.scrub_missing_images()

    def resolve_pool_id(self, pool_name: str) -> int:
        pool_id = self.module.rados.pool_lookup(pool_name)
        if not pool_id:
            raise rados.ObjectNotFound("Pool '{}' not found".format(pool_name),
                                       errno.ENOENT)
        return pool_id

    def scrub_expired_queries(self) -> None:
        # perf counters need to be periodically refreshed to continue
        # to be registered
        expire_time = datetime.now() - QUERY_EXPIRE_INTERVAL
        for pool_key in list(self.user_queries.keys()):
            user_query = self.user_queries[pool_key]
            if user_query[QUERY_LAST_REQUEST] < expire_time:
                self.unregister_osd_perf_queries(pool_key, user_query[QUERY_IDS])
                del self.user_queries[pool_key]

    def register_osd_perf_queries(self,
                                  pool_id: Optional[int],
                                  namespace: Optional[str]) -> List[int]:
        query_ids = []
        try:
            for counter in OSD_PERF_QUERY_COUNTERS:
                query = self.prepare_osd_perf_query(pool_id, namespace, counter)
                self.log.debug("register_osd_perf_queries: {}".format(query))

                query_id = self.module.add_osd_perf_query(query)
                if query_id is None:
                    raise RuntimeError('Failed to add OSD perf query: {}'.format(query))
                query_ids.append(query_id)

        except Exception:
            for query_id in query_ids:
                self.module.remove_osd_perf_query(query_id)
            raise

        return query_ids

    def unregister_osd_perf_queries(self, pool_key: PoolKeyT, query_ids: List[int]) -> None:
        self.log.info("unregister_osd_perf_queries: pool_key={}, query_ids={}".format(
            pool_key, query_ids))
        for query_id in query_ids:
            self.module.remove_osd_perf_query(query_id)
        query_ids[:] = []

    def register_query(self, pool_key: PoolKeyT) -> Dict[str, Any]:
        if pool_key not in self.user_queries:
            pool_name, namespace = pool_key
            pool_id = None
            if pool_name:
                pool_id = self.resolve_pool_id(cast(str, pool_name))

            user_query = {
                QUERY_POOL_ID: pool_id,
                QUERY_POOL_ID_MAP: {pool_id: pool_name},
                QUERY_IDS: self.register_osd_perf_queries(pool_id, namespace),
                QUERY_LAST_REQUEST: datetime.now()
            }

            self.user_queries[pool_key] = user_query

            # force an immediate stat pull if this is a new query
            self.query_condition.notify()
            self.refresh_condition.wait(5)

        else:
            user_query = self.user_queries[pool_key]

            # ensure query doesn't expire
            user_query[QUERY_LAST_REQUEST] = datetime.now()

            if pool_key == GLOBAL_POOL_KEY:
                # refresh the global pool id -> name map upon each
                # processing period
                user_query[QUERY_POOL_ID_MAP] = {
                    pool_id: pool_name for pool_id, pool_name
                    in get_rbd_pools(self.module).items()}

        self.log.debug("register_query: pool_key={}, query_ids={}".format(
            pool_key, user_query[QUERY_IDS]))

        return user_query

    def extract_stat(self,
                     index: int,
                     raw_image: Optional[RawImageCounterT],
                     sum_image: Any) -> float:
        # require two raw counters between a fixed time window
        if not raw_image or not raw_image[0] or not raw_image[1]:
            return 0

        current_counter, previous_counter = cast(Tuple[PerfCounterT, PerfCounterT], raw_image)
        current_time = current_counter[0]
        previous_time = previous_counter[0]
        if current_time <= previous_time or \
                current_time - previous_time > STATS_RATE_INTERVAL.total_seconds():
            return 0

        current_value = current_counter[1][index]
        instant_rate = float(current_value) / (current_time - previous_time)

        # convert latencies from sum to average per op
        ops_index = None
        if OSD_PERF_QUERY_COUNTERS[index] == 'write_latency':
            ops_index = OSD_PERF_QUERY_COUNTERS_INDICES['write_ops']
        elif OSD_PERF_QUERY_COUNTERS[index] == 'read_latency':
            ops_index = OSD_PERF_QUERY_COUNTERS_INDICES['read_ops']

        if ops_index is not None:
            ops = max(1, self.extract_stat(ops_index, raw_image, sum_image))
            instant_rate /= ops

        return instant_rate

    def extract_counter(self,
                        index: int,
                        raw_image: Optional[RawImageCounterT],
                        sum_image: List[int]) -> int:
        if sum_image:
            return sum_image[index]
        return 0

    def generate_report(self,
                        query: Dict[str, Union[Dict[str, str],
                                               Dict[int, Dict[str, dict]]]],
                        sort_by: str,
                        extract_data: ExtractDataFuncT) -> Tuple[Dict[int, str],
                                                                 List[Dict[str, List[float]]]]:
        pool_id_map = cast(Dict[int, str], query[QUERY_POOL_ID_MAP])
        sum_pool_counters = cast(SumPoolCountersT,
                                 query.setdefault(QUERY_SUM_POOL_COUNTERS,
                                                  cast(SumPoolCountersT, {})))
        # pool_id => {namespace => {image_id => [counter..] }
        raw_pool_counters = cast(RawPoolCountersT,
                                 query.setdefault(QUERY_RAW_POOL_COUNTERS,
                                                  cast(RawPoolCountersT, {})))

        sort_by_index = OSD_PERF_QUERY_COUNTERS.index(sort_by)

        # pre-sort and limit the response
        results = []
        for pool_id, sum_namespaces in sum_pool_counters.items():
            if pool_id not in pool_id_map:
                continue
            raw_namespaces: RawNamespacesCountersT = raw_pool_counters.get(pool_id, {})
            for namespace, sum_images in sum_namespaces.items():
                raw_images = raw_namespaces.get(namespace, {})
                for image_id, sum_image in sum_images.items():
                    raw_image = raw_images.get(image_id)

                    # always sort by recent IO activity
                    results.append(((pool_id, namespace, image_id),
                                    self.extract_stat(sort_by_index, raw_image,
                                                      sum_image)))
        results = sorted(results, key=lambda x: x[1], reverse=True)[:REPORT_MAX_RESULTS]

        # build the report in sorted order
        pool_descriptors: Dict[str, int] = {}
        counters = []
        for key, _ in results:
            pool_id = key[0]
            pool_name = pool_id_map[pool_id]

            namespace = key[1]
            image_id = key[2]
            image_names = self.image_name_cache.get((pool_id, namespace), {})
            image_name = image_names[image_id]

            raw_namespaces = raw_pool_counters.get(pool_id, {})
            raw_images = raw_namespaces.get(namespace, {})
            raw_image = raw_images.get(image_id)

            sum_namespaces = sum_pool_counters[pool_id]
            sum_images = sum_namespaces[namespace]
            sum_image = sum_images.get(image_id, [])

            pool_descriptor = pool_name
            if namespace:
                pool_descriptor += "/{}".format(namespace)
            pool_index = pool_descriptors.setdefault(pool_descriptor,
                                                     len(pool_descriptors))
            image_descriptor = "{}/{}".format(pool_index, image_name)
            data = [extract_data(i, raw_image, sum_image)
                    for i in range(len(OSD_PERF_QUERY_COUNTERS))]

            # skip if no data to report
            if data == [0 for i in range(len(OSD_PERF_QUERY_COUNTERS))]:
                continue

            counters.append({image_descriptor: data})

        return {idx: descriptor for descriptor, idx
                in pool_descriptors.items()}, \
            counters

    def get_perf_data(self,
                      report: str,
                      pool_spec: Optional[str],
                      sort_by: str,
                      extract_data: ExtractDataFuncT) -> Tuple[int, str, str]:
        self.log.debug("get_perf_{}s: pool_spec={}, sort_by={}".format(
            report, pool_spec, sort_by))
        self.scrub_expired_queries()

        pool_key = extract_pool_key(pool_spec)
        authorize_request(self.module, pool_key[0], pool_key[1])
        user_query = self.register_query(pool_key)

        now = datetime.now()
        pool_descriptors, counters = self.generate_report(
            user_query, sort_by, extract_data)

        report = {
            'timestamp': time.mktime(now.timetuple()),
            '{}_descriptors'.format(report): OSD_PERF_QUERY_COUNTERS,
            'pool_descriptors': pool_descriptors,
            '{}s'.format(report): counters
        }

        return 0, json.dumps(report), ""

    def get_perf_stats(self,
                       pool_spec: Optional[str],
                       sort_by: str) -> Tuple[int, str, str]:
        return self.get_perf_data(
            "stat", pool_spec, sort_by, self.extract_stat)

    def get_perf_counters(self,
                          pool_spec: Optional[str],
                          sort_by: str) -> Tuple[int, str, str]:
        return self.get_perf_data(
            "counter", pool_spec, sort_by, self.extract_counter)
Commit	Line	Data
9f95a23c TL	1	import errno
	2	import json
	3	import rados
	4	import rbd
	5	import time
	6	import traceback
	7
	8	from datetime import datetime, timedelta
	9	from threading import Condition, Lock, Thread
20effc67	10	from typing import cast, Any, Callable, Dict, List, Optional, Set, Tuple, Union
9f95a23c TL	11
9f95a23c TL	12	from .common import (GLOBAL_POOL_KEY, authorize_request, extract_pool_key,
20effc67	13	get_rbd_pools, PoolKeyT)
9f95a23c TL	14
	15	QUERY_POOL_ID = "pool_id"
	16	QUERY_POOL_ID_MAP = "pool_id_map"
	17	QUERY_IDS = "query_ids"
	18	QUERY_SUM_POOL_COUNTERS = "pool_counters"
	19	QUERY_RAW_POOL_COUNTERS = "raw_pool_counters"
	20	QUERY_LAST_REQUEST = "last_request"
	21
	22	OSD_PERF_QUERY_REGEX_MATCH_ALL = '^(.*)$'
	23	OSD_PERF_QUERY_COUNTERS = ['write_ops',
	24	'read_ops',
	25	'write_bytes',
	26	'read_bytes',
	27	'write_latency',
	28	'read_latency']
	29	OSD_PERF_QUERY_COUNTERS_INDICES = {
	30	OSD_PERF_QUERY_COUNTERS[i]: i for i in range(len(OSD_PERF_QUERY_COUNTERS))}
	31
	32	OSD_PERF_QUERY_LATENCY_COUNTER_INDICES = [4, 5]
	33	OSD_PERF_QUERY_MAX_RESULTS = 256
	34
	35	POOL_REFRESH_INTERVAL = timedelta(minutes=5)
	36	QUERY_EXPIRE_INTERVAL = timedelta(minutes=1)
	37	STATS_RATE_INTERVAL = timedelta(minutes=1)
	38
	39	REPORT_MAX_RESULTS = 64
	40
	41
20effc67 TL	42	# {(pool_id, namespace)...}
	43	ResolveImageNamesT = Set[Tuple[int, str]]
	44
	45	# (time, [value,...])
	46	PerfCounterT = Tuple[int, List[int]]
	47	# current, previous
	48	RawImageCounterT = Tuple[PerfCounterT, Optional[PerfCounterT]]
	49	# image_id => perf_counter
	50	RawImagesCounterT = Dict[str, RawImageCounterT]
	51	# namespace_counters => raw_images
	52	RawNamespacesCountersT = Dict[str, RawImagesCounterT]
	53	# pool_id => namespaces_counters
	54	RawPoolCountersT = Dict[int, RawNamespacesCountersT]
	55
	56	SumImageCounterT = List[int]
	57	# image_id => sum_image
	58	SumImagesCounterT = Dict[str, SumImageCounterT]
	59	# namespace => sum_images
	60	SumNamespacesCountersT = Dict[str, SumImagesCounterT]
	61	# pool_id, sum_namespaces
	62	SumPoolCountersT = Dict[int, SumNamespacesCountersT]
	63
	64	ExtractDataFuncT = Callable[[int, Optional[RawImageCounterT], SumImageCounterT], float]
	65
	66
9f95a23c	67	class PerfHandler:
20effc67 TL	68	user_queries: Dict[PoolKeyT, Dict[str, Any]] = {}
20effc67 TL	69	image_cache: Dict[str, str] = {}
9f95a23c TL	70
	71	lock = Lock()
	72	query_condition = Condition(lock)
	73	refresh_condition = Condition(lock)
	74	thread = None
	75
20effc67	76	image_name_cache: Dict[Tuple[int, str], Dict[str, str]] = {}
9f95a23c TL	77	image_name_refresh_time = datetime.fromtimestamp(0)
	78
	79	@classmethod
20effc67	80	def prepare_regex(cls, value: Any) -> str:
9f95a23c TL	81	return '^({})$'.format(value)
	82
	83	@classmethod
20effc67 TL	84	def prepare_osd_perf_query(cls,
	85	pool_id: Optional[int],
	86	namespace: Optional[str],
	87	counter_type: str) -> Dict[str, Any]:
9f95a23c TL	88	pool_id_regex = OSD_PERF_QUERY_REGEX_MATCH_ALL
	89	namespace_regex = OSD_PERF_QUERY_REGEX_MATCH_ALL
	90	if pool_id:
	91	pool_id_regex = cls.prepare_regex(pool_id)
	92	if namespace:
	93	namespace_regex = cls.prepare_regex(namespace)
	94
	95	return {
	96	'key_descriptor': [
	97	{'type': 'pool_id', 'regex': pool_id_regex},
	98	{'type': 'namespace', 'regex': namespace_regex},
	99	{'type': 'object_name',
	100	'regex': '^(?:rbd\|journal)_data\\.(?:([0-9]+)\\.)?([^.]+)\\.'},
	101	],
	102	'performance_counter_descriptors': OSD_PERF_QUERY_COUNTERS,
	103	'limit': {'order_by': counter_type,
	104	'max_count': OSD_PERF_QUERY_MAX_RESULTS},
	105	}
	106
	107	@classmethod
20effc67	108	def pool_spec_search_keys(cls, pool_key: str) -> List[str]:
9f95a23c TL	109	return [pool_key[0:len(pool_key) - x]
	110	for x in range(0, len(pool_key) + 1)]
	111
	112	@classmethod
20effc67	113	def submatch_pool_key(cls, pool_key: PoolKeyT, search_key: str) -> bool:
9f95a23c TL	114	return ((pool_key[1] == search_key[1] or not search_key[1])
	115	and (pool_key[0] == search_key[0] or not search_key[0]))
	116
20effc67	117	def __init__(self, module: Any) -> None:
9f95a23c TL	118	self.module = module
	119	self.log = module.log
	120
	121	self.thread = Thread(target=self.run)
	122	self.thread.start()
	123
20effc67	124	def run(self) -> None:
9f95a23c TL	125	try:
	126	self.log.info("PerfHandler: starting")
	127	while True:
	128	with self.lock:
	129	self.scrub_expired_queries()
	130	self.process_raw_osd_perf_counters()
	131	self.refresh_condition.notify()
	132
	133	stats_period = self.module.get_ceph_option("mgr_stats_period")
	134	self.query_condition.wait(stats_period)
	135
	136	self.log.debug("PerfHandler: tick")
	137
	138	except Exception as ex:
	139	self.log.fatal("Fatal runtime error: {}\n{}".format(
	140	ex, traceback.format_exc()))
	141
20effc67 TL	142	def merge_raw_osd_perf_counters(self,
	143	pool_key: PoolKeyT,
	144	query: Dict[str, Any],
	145	now_ts: int,
	146	resolve_image_names: ResolveImageNamesT) -> RawPoolCountersT:
9f95a23c TL	147	pool_id_map = query[QUERY_POOL_ID_MAP]
	148
	149	# collect and combine the raw counters from all sort orders
20effc67	150	raw_pool_counters: Dict[int, Dict[str, Dict[str, Any]]] = query.setdefault(QUERY_RAW_POOL_COUNTERS, {})
9f95a23c TL	151	for query_id in query[QUERY_IDS]:
	152	res = self.module.get_osd_perf_counters(query_id)
	153	for counter in res['counters']:
	154	# replace pool id from object name if it exists
	155	k = counter['k']
	156	pool_id = int(k[2][0]) if k[2][0] else int(k[0][0])
	157	namespace = k[1][0]
	158	image_id = k[2][1]
	159
	160	# ignore metrics from non-matching pools/namespaces
	161	if pool_id not in pool_id_map:
	162	continue
	163	if pool_key[1] is not None and pool_key[1] != namespace:
	164	continue
	165
	166	# flag the pool (and namespace) for refresh if we cannot find
	167	# image name in the cache
	168	resolve_image_key = (pool_id, namespace)
	169	if image_id not in self.image_name_cache.get(resolve_image_key, {}):
	170	resolve_image_names.add(resolve_image_key)
	171
	172	# copy the 'sum' counter values for each image (ignore count)
	173	# if we haven't already processed it for this round
	174	raw_namespaces = raw_pool_counters.setdefault(pool_id, {})
	175	raw_images = raw_namespaces.setdefault(namespace, {})
20effc67	176	raw_image = raw_images.get(image_id)
9f95a23c	177	# save the last two perf counters for each image
20effc67 TL	178	new_current = (now_ts, [int(x[0]) for x in counter['c']])
	179	if raw_image:
	180	old_current, _ = raw_image
	181	if old_current[0] < now_ts:
	182	raw_images[image_id] = (new_current, old_current)
	183	else:
	184	raw_images[image_id] = (new_current, None)
9f95a23c TL	185
	186	self.log.debug("merge_raw_osd_perf_counters: {}".format(raw_pool_counters))
	187	return raw_pool_counters
	188
20effc67 TL	189	def sum_osd_perf_counters(self,
	190	query: Dict[str, dict],
	191	raw_pool_counters: RawPoolCountersT,
	192	now_ts: int) -> SumPoolCountersT:
9f95a23c TL	193	# update the cumulative counters for each image
	194	sum_pool_counters = query.setdefault(QUERY_SUM_POOL_COUNTERS, {})
	195	for pool_id, raw_namespaces in raw_pool_counters.items():
	196	sum_namespaces = sum_pool_counters.setdefault(pool_id, {})
	197	for namespace, raw_images in raw_namespaces.items():
	198	sum_namespace = sum_namespaces.setdefault(namespace, {})
	199	for image_id, raw_image in raw_images.items():
	200	# zero-out non-updated raw counters
	201	if not raw_image[0]:
	202	continue
20effc67 TL	203	old_current, _ = raw_image
	204	if old_current[0] < now_ts:
	205	new_current = (now_ts, [0] * len(old_current[1]))
	206	raw_images[image_id] = (new_current, old_current)
9f95a23c TL	207	continue
9f95a23c TL	208
20effc67	209	counters = old_current[1]
9f95a23c TL	210
	211	# copy raw counters if this is a newly discovered image or
	212	# increment existing counters
	213	sum_image = sum_namespace.setdefault(image_id, None)
	214	if sum_image:
	215	for i in range(len(counters)):
	216	sum_image[i] += counters[i]
	217	else:
	218	sum_namespace[image_id] = [x for x in counters]
	219
	220	self.log.debug("sum_osd_perf_counters: {}".format(sum_pool_counters))
	221	return sum_pool_counters
	222
20effc67	223	def refresh_image_names(self, resolve_image_names: ResolveImageNamesT) -> None:
9f95a23c TL	224	for pool_id, namespace in resolve_image_names:
	225	image_key = (pool_id, namespace)
	226	images = self.image_name_cache.setdefault(image_key, {})
	227	with self.module.rados.open_ioctx2(int(pool_id)) as ioctx:
	228	ioctx.set_namespace(namespace)
	229	for image_meta in rbd.RBD().list2(ioctx):
	230	images[image_meta['id']] = image_meta['name']
	231	self.log.debug("resolve_image_names: {}={}".format(image_key, images))
	232
20effc67	233	def scrub_missing_images(self) -> None:
9f95a23c TL	234	for pool_key, query in self.user_queries.items():
	235	raw_pool_counters = query.get(QUERY_RAW_POOL_COUNTERS, {})
	236	sum_pool_counters = query.get(QUERY_SUM_POOL_COUNTERS, {})
	237	for pool_id, sum_namespaces in sum_pool_counters.items():
	238	raw_namespaces = raw_pool_counters.get(pool_id, {})
	239	for namespace, sum_images in sum_namespaces.items():
	240	raw_images = raw_namespaces.get(namespace, {})
	241
	242	image_key = (pool_id, namespace)
	243	image_names = self.image_name_cache.get(image_key, {})
	244	for image_id in list(sum_images.keys()):
	245	# scrub image counters if we failed to resolve image name
	246	if image_id not in image_names:
	247	self.log.debug("scrub_missing_images: dropping {}/{}".format(
	248	image_key, image_id))
	249	del sum_images[image_id]
	250	if image_id in raw_images:
	251	del raw_images[image_id]
	252
20effc67	253	def process_raw_osd_perf_counters(self) -> None:
9f95a23c TL	254	now = datetime.now()
	255	now_ts = int(now.strftime("%s"))
	256
	257	# clear the image name cache if we need to refresh all active pools
	258	if self.image_name_cache and \
	259	self.image_name_refresh_time + POOL_REFRESH_INTERVAL < now:
	260	self.log.debug("process_raw_osd_perf_counters: expiring image name cache")
	261	self.image_name_cache = {}
	262
20effc67	263	resolve_image_names: Set[Tuple[int, str]] = set()
9f95a23c TL	264	for pool_key, query in self.user_queries.items():
	265	if not query[QUERY_IDS]:
	266	continue
	267
	268	raw_pool_counters = self.merge_raw_osd_perf_counters(
	269	pool_key, query, now_ts, resolve_image_names)
	270	self.sum_osd_perf_counters(query, raw_pool_counters, now_ts)
	271
	272	if resolve_image_names:
	273	self.image_name_refresh_time = now
	274	self.refresh_image_names(resolve_image_names)
	275	self.scrub_missing_images()
	276	elif not self.image_name_cache:
	277	self.scrub_missing_images()
	278
20effc67	279	def resolve_pool_id(self, pool_name: str) -> int:
9f95a23c TL	280	pool_id = self.module.rados.pool_lookup(pool_name)
	281	if not pool_id:
	282	raise rados.ObjectNotFound("Pool '{}' not found".format(pool_name),
	283	errno.ENOENT)
	284	return pool_id
	285
20effc67	286	def scrub_expired_queries(self) -> None:
9f95a23c TL	287	# perf counters need to be periodically refreshed to continue
	288	# to be registered
	289	expire_time = datetime.now() - QUERY_EXPIRE_INTERVAL
	290	for pool_key in list(self.user_queries.keys()):
	291	user_query = self.user_queries[pool_key]
	292	if user_query[QUERY_LAST_REQUEST] < expire_time:
	293	self.unregister_osd_perf_queries(pool_key, user_query[QUERY_IDS])
	294	del self.user_queries[pool_key]
	295
20effc67 TL	296	def register_osd_perf_queries(self,
	297	pool_id: Optional[int],
	298	namespace: Optional[str]) -> List[int]:
9f95a23c TL	299	query_ids = []
	300	try:
	301	for counter in OSD_PERF_QUERY_COUNTERS:
	302	query = self.prepare_osd_perf_query(pool_id, namespace, counter)
	303	self.log.debug("register_osd_perf_queries: {}".format(query))
	304
	305	query_id = self.module.add_osd_perf_query(query)
	306	if query_id is None:
	307	raise RuntimeError('Failed to add OSD perf query: {}'.format(query))
	308	query_ids.append(query_id)
	309
	310	except Exception:
	311	for query_id in query_ids:
	312	self.module.remove_osd_perf_query(query_id)
	313	raise
	314
	315	return query_ids
	316
20effc67	317	def unregister_osd_perf_queries(self, pool_key: PoolKeyT, query_ids: List[int]) -> None:
9f95a23c TL	318	self.log.info("unregister_osd_perf_queries: pool_key={}, query_ids={}".format(
	319	pool_key, query_ids))
	320	for query_id in query_ids:
	321	self.module.remove_osd_perf_query(query_id)
	322	query_ids[:] = []
	323
20effc67	324	def register_query(self, pool_key: PoolKeyT) -> Dict[str, Any]:
9f95a23c	325	if pool_key not in self.user_queries:
20effc67	326	pool_name, namespace = pool_key
9f95a23c	327	pool_id = None
20effc67 TL	328	if pool_name:
20effc67 TL	329	pool_id = self.resolve_pool_id(cast(str, pool_name))
9f95a23c TL	330
	331	user_query = {
	332	QUERY_POOL_ID: pool_id,
20effc67 TL	333	QUERY_POOL_ID_MAP: {pool_id: pool_name},
20effc67 TL	334	QUERY_IDS: self.register_osd_perf_queries(pool_id, namespace),
9f95a23c TL	335	QUERY_LAST_REQUEST: datetime.now()
	336	}
	337
	338	self.user_queries[pool_key] = user_query
	339
	340	# force an immediate stat pull if this is a new query
	341	self.query_condition.notify()
	342	self.refresh_condition.wait(5)
	343
	344	else:
	345	user_query = self.user_queries[pool_key]
	346
	347	# ensure query doesn't expire
	348	user_query[QUERY_LAST_REQUEST] = datetime.now()
	349
	350	if pool_key == GLOBAL_POOL_KEY:
	351	# refresh the global pool id -> name map upon each
	352	# processing period
	353	user_query[QUERY_POOL_ID_MAP] = {
	354	pool_id: pool_name for pool_id, pool_name
	355	in get_rbd_pools(self.module).items()}
	356
	357	self.log.debug("register_query: pool_key={}, query_ids={}".format(
	358	pool_key, user_query[QUERY_IDS]))
	359
	360	return user_query
	361
20effc67 TL	362	def extract_stat(self,
	363	index: int,
	364	raw_image: Optional[RawImageCounterT],
	365	sum_image: Any) -> float:
9f95a23c TL	366	# require two raw counters between a fixed time window
	367	if not raw_image or not raw_image[0] or not raw_image[1]:
	368	return 0
	369
20effc67 TL	370	current_counter, previous_counter = cast(Tuple[PerfCounterT, PerfCounterT], raw_image)
	371	current_time = current_counter[0]
	372	previous_time = previous_counter[0]
9f95a23c TL	373	if current_time <= previous_time or \
	374	current_time - previous_time > STATS_RATE_INTERVAL.total_seconds():
	375	return 0
	376
20effc67	377	current_value = current_counter[1][index]
9f95a23c TL	378	instant_rate = float(current_value) / (current_time - previous_time)
	379
	380	# convert latencies from sum to average per op
	381	ops_index = None
	382	if OSD_PERF_QUERY_COUNTERS[index] == 'write_latency':
	383	ops_index = OSD_PERF_QUERY_COUNTERS_INDICES['write_ops']
	384	elif OSD_PERF_QUERY_COUNTERS[index] == 'read_latency':
	385	ops_index = OSD_PERF_QUERY_COUNTERS_INDICES['read_ops']
	386
	387	if ops_index is not None:
	388	ops = max(1, self.extract_stat(ops_index, raw_image, sum_image))
	389	instant_rate /= ops
	390
	391	return instant_rate
	392
20effc67 TL	393	def extract_counter(self,
	394	index: int,
	395	raw_image: Optional[RawImageCounterT],
	396	sum_image: List[int]) -> int:
9f95a23c TL	397	if sum_image:
	398	return sum_image[index]
	399	return 0
	400
20effc67 TL	401	def generate_report(self,
	402	query: Dict[str, Union[Dict[str, str],
	403	Dict[int, Dict[str, dict]]]],
	404	sort_by: str,
	405	extract_data: ExtractDataFuncT) -> Tuple[Dict[int, str],
	406	List[Dict[str, List[float]]]]:
	407	pool_id_map = cast(Dict[int, str], query[QUERY_POOL_ID_MAP])
	408	sum_pool_counters = cast(SumPoolCountersT,
	409	query.setdefault(QUERY_SUM_POOL_COUNTERS,
	410	cast(SumPoolCountersT, {})))
	411	# pool_id => {namespace => {image_id => [counter..] }
	412	raw_pool_counters = cast(RawPoolCountersT,
	413	query.setdefault(QUERY_RAW_POOL_COUNTERS,
	414	cast(RawPoolCountersT, {})))
9f95a23c TL	415
	416	sort_by_index = OSD_PERF_QUERY_COUNTERS.index(sort_by)
	417
	418	# pre-sort and limit the response
	419	results = []
	420	for pool_id, sum_namespaces in sum_pool_counters.items():
	421	if pool_id not in pool_id_map:
	422	continue
20effc67	423	raw_namespaces: RawNamespacesCountersT = raw_pool_counters.get(pool_id, {})
9f95a23c TL	424	for namespace, sum_images in sum_namespaces.items():
	425	raw_images = raw_namespaces.get(namespace, {})
	426	for image_id, sum_image in sum_images.items():
20effc67	427	raw_image = raw_images.get(image_id)
9f95a23c TL	428
9f95a23c TL	429	# always sort by recent IO activity
20effc67	430	results.append(((pool_id, namespace, image_id),
9f95a23c	431	self.extract_stat(sort_by_index, raw_image,
20effc67	432	sum_image)))
9f95a23c TL	433	results = sorted(results, key=lambda x: x[1], reverse=True)[:REPORT_MAX_RESULTS]
	434
	435	# build the report in sorted order
20effc67	436	pool_descriptors: Dict[str, int] = {}
9f95a23c TL	437	counters = []
	438	for key, _ in results:
	439	pool_id = key[0]
	440	pool_name = pool_id_map[pool_id]
	441
	442	namespace = key[1]
	443	image_id = key[2]
	444	image_names = self.image_name_cache.get((pool_id, namespace), {})
	445	image_name = image_names[image_id]
	446
	447	raw_namespaces = raw_pool_counters.get(pool_id, {})
	448	raw_images = raw_namespaces.get(namespace, {})
20effc67	449	raw_image = raw_images.get(image_id)
9f95a23c TL	450
	451	sum_namespaces = sum_pool_counters[pool_id]
	452	sum_images = sum_namespaces[namespace]
	453	sum_image = sum_images.get(image_id, [])
	454
	455	pool_descriptor = pool_name
	456	if namespace:
	457	pool_descriptor += "/{}".format(namespace)
	458	pool_index = pool_descriptors.setdefault(pool_descriptor,
	459	len(pool_descriptors))
	460	image_descriptor = "{}/{}".format(pool_index, image_name)
	461	data = [extract_data(i, raw_image, sum_image)
	462	for i in range(len(OSD_PERF_QUERY_COUNTERS))]
	463
	464	# skip if no data to report
	465	if data == [0 for i in range(len(OSD_PERF_QUERY_COUNTERS))]:
	466	continue
	467
	468	counters.append({image_descriptor: data})
	469
	470	return {idx: descriptor for descriptor, idx
	471	in pool_descriptors.items()}, \
	472	counters
	473
20effc67 TL	474	def get_perf_data(self,
	475	report: str,
	476	pool_spec: Optional[str],
	477	sort_by: str,
	478	extract_data: ExtractDataFuncT) -> Tuple[int, str, str]:
9f95a23c TL	479	self.log.debug("get_perf_{}s: pool_spec={}, sort_by={}".format(
	480	report, pool_spec, sort_by))
	481	self.scrub_expired_queries()
	482
	483	pool_key = extract_pool_key(pool_spec)
	484	authorize_request(self.module, pool_key[0], pool_key[1])
9f95a23c TL	485	user_query = self.register_query(pool_key)
	486
	487	now = datetime.now()
	488	pool_descriptors, counters = self.generate_report(
	489	user_query, sort_by, extract_data)
	490
	491	report = {
	492	'timestamp': time.mktime(now.timetuple()),
	493	'{}_descriptors'.format(report): OSD_PERF_QUERY_COUNTERS,
	494	'pool_descriptors': pool_descriptors,
	495	'{}s'.format(report): counters
	496	}
	497
	498	return 0, json.dumps(report), ""
	499
20effc67 TL	500	def get_perf_stats(self,
	501	pool_spec: Optional[str],
	502	sort_by: str) -> Tuple[int, str, str]:
9f95a23c TL	503	return self.get_perf_data(
	504	"stat", pool_spec, sort_by, self.extract_stat)
	505
20effc67 TL	506	def get_perf_counters(self,
	507	pool_spec: Optional[str],
	508	sort_by: str) -> Tuple[int, str, str]:
9f95a23c TL	509	return self.get_perf_data(
9f95a23c TL	510	"counter", pool_spec, sort_by, self.extract_counter)