From: Kotresh HR Date: Tue, 16 Jun 2026 10:42:41 +0000 (+0530) Subject: mgr/mirroring: cache fs snapshot mirror status omap metrics X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=8647c2ee79649b002c23bfb88757955640f886ce;p=ceph.git mgr/mirroring: cache fs snapshot mirror status omap metrics Add a short-lived in-memory cache for metrics returned by "ceph fs snapshot mirror status", which reads persisted sync stats from the cephfs_mirror object omap. Omap walks are relatively expensive; caching reduces repeated reads when the CLI or multiple clients poll at short intervals. Two TTL-bucketed LRU caches (CACHE_TTL_SECS) are used instead of one unified cache. The complete cache always holds every mirrored directory and peer for a filesystem; the partial cache holds one directory. A single directory-granularity cache cannot prove it has all directories, so full-scan queries rely on the complete cache contract. Cache implementation (lru_cache_timeout decorator backed by _TimedLRUCache, not functools.lru_cache): - lru_cache has no TTL and no peek-on-hit API. Single-directory queries must read the complete cache without loading on miss; lru_cache.cache is also unavailable on Python 3.14+. - complete (sync_stat_complete_cache): full omap prefix scan via load_sync_stat_metrics. Cache key: (time_token, filesystem). COMPLETE_CACHE_MAX limits filesystem entries per TTL window. - partial (sync_stat_partial_cache): per-directory omap key load via fetch_sync_stat_metrics. Cache key: (time_token, filesystem, dir_path, peer_ids). PARTIAL_CACHE_MAX limits directory entries. Bind cache_peek/cache_info/cache_clear via a CachedMethod descriptor so TTL lookups receive (self, ...); otherwise single-dir status fails with 'str' object has no attribute 'mgr'. Serve logic (under the existing lock): - status [--peer_uuid=]: load complete cache on miss; filter by peer when requested. - status : peek complete cache (no load on miss); if the entry contains the directory and peers, serve from complete. Otherwise load partial cache (omap on miss). Fixes: https://tracker.ceph.com/issues/76686 Signed-off-by: Kotresh HR --- diff --git a/src/pybind/mgr/mirroring/fs/metrics/cache.py b/src/pybind/mgr/mirroring/fs/metrics/cache.py new file mode 100644 index 00000000000..0fcfdbd7e45 --- /dev/null +++ b/src/pybind/mgr/mirroring/fs/metrics/cache.py @@ -0,0 +1,202 @@ +from collections import OrderedDict, namedtuple +from functools import wraps +from typing import Any, Callable, Dict + +import time + +from ..utils import norm_path +from .format import format_and_order_sync_stat_for_display, format_peer_status_metrics + +# Two caches are used instead of one unified cache: +# +# - Complete cache holds a full omap snapshot (all dirs, all peers) per +# filesystem. A hit guarantees the entry is complete, so full-scan queries +# ('status ', 'status ') can be served without doubt. +# +# - Partial cache holds per-directory omap loads. Single-dir queries +# ('status ') peek the complete cache first; on miss they use +# partial cache to avoid a full omap scan. +# +# A single cache keyed at directory granularity cannot cleanly answer "do I +# have every directory?" — partial entries may be present while others are +# missing, so serving a full scan from cache would require extra bookkeeping +# to prove completeness. Splitting complete vs partial makes that contract +# explicit: complete = all dirs; partial = one dir. + +# LRU max entries for complete-cache lookups (one entry per filesystem per TTL +# window; each entry holds all dirs and all peers for that filesystem). +# Cache key: (time_token, filesystem). +COMPLETE_CACHE_MAX = 8 +# LRU max entries for partial-cache lookups (per-directory omap batch load). +# Cache key: (time_token, filesystem, dir_path, peer_ids). +PARTIAL_CACHE_MAX = 32 + +CacheInfo = namedtuple('CacheInfo', ['hits', 'misses', 'maxsize', 'currsize']) + + +# functools.lru_cache is not sufficient here for two reasons: +# +# 1. TTL: entries must expire after snapshot_mirror_metrics_cache_ttl. lru_cache +# has no TTL, so we bucket keys with a time_token derived from monotonic time. +# +# 2. cache_peek: single-directory status queries must read the complete cache +# without loading on miss (otherwise a cold complete cache would trigger a +# full omap scan). lru_cache has no peek API; calling the wrapper always +# invokes the loader on miss. Inspecting lru_cache.cache for a hit-only +# lookup is also not viable: that attribute is not part of the public API and +# is unavailable on Python 3.14+ (_lru_cache_wrapper has no .cache). +# +# _TimedLRUCache therefore implements TTL-bucketed LRU storage plus peek(). +class _TimedLRUCache: + def __init__(self, func, maxsize): + self.func = func + self.maxsize = maxsize + self.store = OrderedDict() + self.hits = 0 + self.misses = 0 + + def _key(self, time_token, args): + # time_token buckets entries by snapshot_mirror_metrics_cache_ttl; args are + # the decorated method's positional arguments (e.g. filesystem, or + # filesystem+dir+peers). + return (time_token,) + args + + def get(self, time_token, args, load=True): + key = self._key(time_token, args) + if key in self.store: + self.hits += 1 + self.store.move_to_end(key) + return self.store[key] + if not load: + return None + self.misses += 1 + val = self.func(*args) + self.store[key] = val + while len(self.store) > self.maxsize: + self.store.popitem(last=False) + return val + + def peek(self, time_token, args): + return self.get(time_token, args, load=False) + + def clear(self): + self.store.clear() + self.hits = 0 + self.misses = 0 + + def info(self): + return CacheInfo(self.hits, self.misses, self.maxsize, len(self.store)) + + +def _resolve_ttl(ttl_getter, args, kwargs): + try: + return ttl_getter(*args, **kwargs) + except TypeError: + return ttl_getter() + + +def lru_cache_timeout(ttl_getter: Callable[..., int], maxsize: int = 128): + def decorator(func): + cache = _TimedLRUCache(func, maxsize) + + # BoundCached is returned from CachedMethod.__get__ so cache_peek, + # cache_info, and cache_clear are bound to the instance. Attaching + # those helpers to a plain wrapper function would drop self (e.g. + # self.sync_stat_complete_cache.cache_peek(fs) would pass only fs), + # breaking ttl_getter lambdas that use self.mgr. + class BoundCached: + def __init__(self, instance): + self._instance = instance + + def _full_args(self, *args): + return (self._instance,) + args + + def __call__(self, *args, **kwargs): + full = self._full_args(*args) + ttl = _resolve_ttl(ttl_getter, full, kwargs) + if not ttl: + return func(*full, **kwargs) + time_token = int(time.monotonic() / ttl) + return cache.get(time_token, full) + + def cache_peek(self, *args, **kwargs): + full = self._full_args(*args) + ttl = _resolve_ttl(ttl_getter, full, kwargs) + if not ttl: + return None + time_token = int(time.monotonic() / ttl) + return cache.peek(time_token, full) + + def cache_clear(self): + cache.clear() + + def cache_info(self): + return cache.info() + + # Descriptor: self.method(...) and self.method.cache_peek(...) share + # the same BoundCached instance and cache keys. + class CachedMethod: + def __get__(self, obj, owner=None): + if obj is None: + return self + return BoundCached(obj) + + def __call__(self, *args, **kwargs): + ttl = _resolve_ttl(ttl_getter, args, kwargs) + if not ttl: + return func(*args, **kwargs) + time_token = int(time.monotonic() / ttl) + return cache.get(time_token, args) + + cached = CachedMethod() + wraps(func)(cached) + return cached + return decorator + + +def _peer_stats_for_dir(metrics, dir_path): + return metrics.get(dir_path, {}).get('peer', {}) + + +def _requested_peers_have_stats(metrics, peers, dir_path): + peer_stats = _peer_stats_for_dir(metrics, dir_path) + return all(peer in peer_stats for peer in peers) + + +def metrics_for_dir_and_peers(metrics, dir_path, peers): + result: Dict[str, Any] = {} + peer_stats = _peer_stats_for_dir(metrics, dir_path) + for peer in peers: + if peer in peer_stats: + format_peer_status_metrics( + result, dir_path, peer, + format_and_order_sync_stat_for_display(peer_stats[peer])) + return result + + +def metrics_for_peer(metrics, peer_uuid): + result: Dict[str, Any] = {} + for dir_path, dir_entry in metrics.items(): + peer_stats = dir_entry.get('peer', {}) + if peer_uuid in peer_stats: + format_peer_status_metrics( + result, dir_path, peer_uuid, + format_and_order_sync_stat_for_display(peer_stats[peer_uuid])) + return result + + +def try_get_from_complete(metrics, mirrored_dir_path, peer_uuid, peers): + """Filter complete cache (all dirs, all peers) for the CLI request. + + The complete cache always holds every peer and directory. peer_uuid and + peers select the subset to return; they do not affect what is loaded. + """ + if mirrored_dir_path: + dir_path = norm_path(mirrored_dir_path) + if _requested_peers_have_stats(metrics, peers, dir_path): + return metrics_for_dir_and_peers(metrics, dir_path, peers) + return None + + if peer_uuid is None: + return dict(metrics) + return metrics_for_peer(metrics, peer_uuid) diff --git a/src/pybind/mgr/mirroring/fs/metrics/load.py b/src/pybind/mgr/mirroring/fs/metrics/load.py index f5aed3a28a5..d157b36b7c2 100644 --- a/src/pybind/mgr/mirroring/fs/metrics/load.py +++ b/src/pybind/mgr/mirroring/fs/metrics/load.py @@ -12,7 +12,9 @@ from ..utils import ( SYNC_STAT_KEY_PREFIX, decode_sync_stat_val, get_metadata_pool, + norm_path, parse_sync_stat_omap_key, + sync_stat_omap_key, ) from .format import format_and_order_sync_stat_for_display, format_peer_status_metrics @@ -93,3 +95,22 @@ def load_sync_stat_metrics(ioctx, filesystem, peer_uuid=None): log.error(f'failed to read sync stat omap: {e}') raise MirrorException(-e.errno, 'failed to read sync stat omap') return metrics + + +def fetch_sync_stat_metrics(ioctx, filesystem, peers, mirrored_dir_path, peer_uuid): + if mirrored_dir_path: + dir_path = norm_path(mirrored_dir_path) + keys = [sync_stat_omap_key(filesystem, peer, dir_path) for peer in peers] + omap_stats = load_sync_stat_by_keys(ioctx, keys) + metrics: Dict[str, Any] = {} + for peer in peers: + omap_key = sync_stat_omap_key(filesystem, peer, dir_path) + stat = omap_stats.get(omap_key) + if stat is not None: + format_peer_status_metrics( + metrics, dir_path, peer, + format_and_order_sync_stat_for_display(stat)) + return metrics, False, dir_path + + metrics = load_sync_stat_metrics(ioctx, filesystem, peer_uuid) + return metrics, True, None diff --git a/src/pybind/mgr/mirroring/fs/snapshot_mirror.py b/src/pybind/mgr/mirroring/fs/snapshot_mirror.py index 1504b8831ef..2b1fa76425e 100644 --- a/src/pybind/mgr/mirroring/fs/snapshot_mirror.py +++ b/src/pybind/mgr/mirroring/fs/snapshot_mirror.py @@ -20,10 +20,11 @@ from .blocklist import blocklist from .notify import Notifier, InstanceWatcher from .utils import INSTANCE_ID_PREFIX, MIRROR_OBJECT_NAME, Finisher, \ AsyncOpTracker, get_metadata_pool, norm_path, connect_to_filesystem, \ - disconnect_from_filesystem, sync_stat_omap_key + disconnect_from_filesystem +from .metrics.cache import ( + CACHE_TTL_SECS, COMPLETE_CACHE_MAX, lru_cache_timeout, PARTIAL_CACHE_MAX, + metrics_for_dir_and_peers, try_get_from_complete) from .metrics import load as metrics_load -from .metrics.format import format_and_order_sync_stat_for_display, \ - format_peer_status_metrics from .exception import MirrorException from .dir_map.create import create_mirror_object from .dir_map.load import load_dir_map, load_instances @@ -759,8 +760,52 @@ class FSSnapshotMirror: except MirrorException as me: return me.args[0], '', me.args[1] + @lru_cache_timeout( + lambda self, *_args, **_kwargs: CACHE_TTL_SECS, + COMPLETE_CACHE_MAX) + def sync_stat_complete_cache(self, filesystem): + """Load all directories and all peers for a filesystem from omap. + + Cache key: (time_token, filesystem). Each entry stores the full omap + snapshot for that filesystem (all dirs, all peers). A hit means every + directory is present, so full-scan queries can be served safely. + Used for 'status ' and 'status --peer_uuid='; peer + filtering is applied when serving. + """ + log.debug('sync stat metrics for filesystem %s loaded from omap (complete)', + filesystem) + ioctx = metrics_load.open_metadata_ioctx( + self.rados, self.fs_map, filesystem) + return metrics_load.load_sync_stat_metrics(ioctx, filesystem) + + @lru_cache_timeout( + lambda self, *_args, **_kwargs: CACHE_TTL_SECS, + PARTIAL_CACHE_MAX) + def sync_stat_partial_cache(self, filesystem, dir_path, peer_ids): + """Load sync-stat omap keys for one directory and a peer set. + + Cache key: (time_token, filesystem, dir_path, peer_ids). Used as a + fallback for 'status ' when the complete cache is cold or + does not contain the requested directory. + """ + peer_scope = (next(iter(peer_ids)) if len(peer_ids) == 1 else '*') + log.debug('sync stat metrics for filesystem %s (dir=%s, peer=%s) ' + 'loaded from omap', + filesystem, dir_path, peer_scope) + peers = {peer_id: None for peer_id in peer_ids} + ioctx = metrics_load.open_metadata_ioctx( + self.rados, self.fs_map, filesystem) + metrics, _, _ = metrics_load.fetch_sync_stat_metrics( + ioctx, filesystem, peers, dir_path, None) + return metrics + def metrics_status(self, filesystem, mirrored_dir_path, peer_uuid): - """Return persisted mirror directory snapshot metrics as JSON""" + """Return persisted mirror directory snapshot metrics as JSON. + + Uses two caches (see metrics/cache.py): complete for full-scan queries + where a hit proves all dirs are present; partial for single-dir + queries when the complete cache is cold or lacks that directory. + """ try: with self.lock: if not self.filesystem_exist(filesystem): @@ -775,35 +820,70 @@ class FSSnapshotMirror: if not fspolicy.policy.lookup(dir_path): raise MirrorException(-errno.ENOENT, f'directory {dir_path} is not mirrored') - peers = self.get_filesystem_peers(filesystem) - if not peers: + all_peers = self.get_filesystem_peers(filesystem) + if not all_peers: return 0, json.dumps({'metrics': {}}, indent=4), '' if peer_uuid: - if peer_uuid not in peers: + if peer_uuid not in all_peers: raise MirrorException(-errno.ENOENT, f'peer {peer_uuid} not found for ' f'filesystem {filesystem}') - peers = {peer_uuid: peers[peer_uuid]} + requested_peers = {peer_uuid: all_peers[peer_uuid]} + else: + requested_peers = all_peers - ioctx = metrics_load.open_metadata_ioctx( - self.rados, self.fs_map, filesystem) if mirrored_dir_path: dir_path = norm_path(mirrored_dir_path) - keys = [sync_stat_omap_key(filesystem, peer, dir_path) - for peer in peers] - omap_stats = metrics_load.load_sync_stat_by_keys(ioctx, keys) - metrics = {} - for peer in peers: - omap_key = sync_stat_omap_key(filesystem, peer, dir_path) - if omap_key in omap_stats: - format_peer_status_metrics( - metrics, dir_path, peer, - format_and_order_sync_stat_for_display( - omap_stats[omap_key])) + # Single-dir query: peek complete cache (key: filesystem + # only); on miss fall back to partial cache (key: + # filesystem, dir_path, peer_ids). + complete_metrics = self.sync_stat_complete_cache.cache_peek( + filesystem) + if complete_metrics is not None: + metrics = try_get_from_complete( + complete_metrics, mirrored_dir_path, peer_uuid, + requested_peers) + if metrics is not None: + log.debug('sync stat metrics for filesystem %s (dir=%s, peer=%s) ' + 'served from complete cache', + filesystem, dir_path, peer_uuid or '*') + return 0, json.dumps({'metrics': metrics}, indent=4), '' + + partial_info_before = self.sync_stat_partial_cache.cache_info() + raw_metrics = self.sync_stat_partial_cache( + filesystem, dir_path, frozenset(requested_peers)) + if (self.sync_stat_partial_cache.cache_info().hits > + partial_info_before.hits): + log.debug('sync stat metrics for filesystem %s (dir=%s, peer=%s) ' + 'served from partial cache', + filesystem, dir_path, peer_uuid or '*') + else: + log.debug('sync stat metrics for filesystem %s (dir=%s, peer=%s) ' + 'loaded from omap', + filesystem, dir_path, peer_uuid or '*') + metrics = metrics_for_dir_and_peers( + raw_metrics, dir_path, requested_peers) + return 0, json.dumps({'metrics': metrics}, indent=4), '' + + # Full-scan query: load complete cache on miss (key: filesystem + # only); peer filtering is applied when serving. + complete_info_before = self.sync_stat_complete_cache.cache_info() + complete_metrics = self.sync_stat_complete_cache(filesystem) + complete_hit = (self.sync_stat_complete_cache.cache_info().hits > + complete_info_before.hits) + metrics = try_get_from_complete( + complete_metrics, mirrored_dir_path, peer_uuid, requested_peers) + if complete_hit: + log.debug('sync stat metrics for filesystem %s (dir=%s, peer=%s) ' + 'served from complete cache', + filesystem, mirrored_dir_path or '*', + peer_uuid or '*') else: - metrics = metrics_load.load_sync_stat_metrics( - ioctx, filesystem, peer_uuid) + log.debug('sync stat metrics for filesystem %s (dir=%s, peer=%s) ' + 'loaded from omap', + filesystem, mirrored_dir_path or '*', + peer_uuid or '*') return 0, json.dumps({'metrics': metrics}, indent=4), '' except MirrorException as me: return me.args[0], '', me.args[1]