--- /dev/null
+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 <fs>', 'status <fs> <peer>') can be served without doubt.
+#
+# - Partial cache holds per-directory omap loads. Single-dir queries
+# ('status <fs> <dir>') 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)
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
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
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
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 <fs>' and 'status <fs> --peer_uuid=<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 <fs> <dir>' 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):
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]
projects / ceph.git / commitdiff