From: Kotresh HR Date: Wed, 29 Apr 2026 09:19:15 +0000 (+0530) Subject: mgr/mirroring: Add new interface to expose mirroring metrics X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=33c572242ba78924df3e65f9fec23f2aefe81c66;p=ceph.git mgr/mirroring: Add new interface to expose mirroring metrics Add the following new interface to expose mirroring metrics ceph fs snapshot mirror status [] [--peer_uuid=] The cmd loads the persisted directory sync metrics from the cephfs_mirror object's omap. Metrics are grouped by mirrored directory and peer. When --peer_uuid is specified, only metrics for that peer are returned. peer_uuid is a named CLI argument (_end_positional_) so peer-only filtering does not require a mirrored directory path. Fixes: https://tracker.ceph.com/issues/76686 Signed-off-by: Kotresh HR --- diff --git a/src/pybind/mgr/mirroring/fs/metrics/__init__.py b/src/pybind/mgr/mirroring/fs/metrics/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/src/pybind/mgr/mirroring/fs/metrics/load.py b/src/pybind/mgr/mirroring/fs/metrics/load.py new file mode 100644 index 00000000000..8eff6a5c819 --- /dev/null +++ b/src/pybind/mgr/mirroring/fs/metrics/load.py @@ -0,0 +1,96 @@ +import errno +import json +import logging +from typing import Any, Dict + +import rados + +from ..dir_map.load import MAX_RETURN +from ..exception import MirrorException +from ..utils import ( + MIRROR_OBJECT_NAME, + SYNC_STAT_KEY_PREFIX, + decode_sync_stat_val, + get_metadata_pool, + parse_sync_stat_omap_key, +) + +log = logging.getLogger(__name__) + + +def nest_sync_stat_in_metrics(metrics, dir_path, peer_uuid, stat): + metrics.setdefault(dir_path, {}).setdefault('peer', {})[peer_uuid] = stat + + +def load_sync_stat_by_keys(ioctx, keys): + stats: Dict[str, Any] = {} + if not keys: + return stats + try: + with rados.ReadOpCtx() as read_op: + it, ret = ioctx.get_omap_vals_by_keys(read_op, keys) + if ret != 0: + log.error('failed to read sync stat omap keys') + raise MirrorException(-errno.EINVAL, + 'failed to read sync stat omap keys') + ioctx.operate_read_op(read_op, MIRROR_OBJECT_NAME) + for key, val in it: + try: + stats[key] = decode_sync_stat_val(val) + except (json.JSONDecodeError, UnicodeDecodeError) as e: + log.warning(f'failed to decode sync stat for key {key}: {e}') + except rados.Error as e: + log.error(f'failed to read sync stat omap keys: {e}') + raise MirrorException(-e.errno, 'failed to read sync stat omap keys') + return stats + + +def open_metadata_ioctx(rados_inst, fs_map, filesystem): + metadata_pool_id = get_metadata_pool(filesystem, fs_map) + if not metadata_pool_id: + raise MirrorException(-errno.EINVAL, + f'cannot find metadata pool for filesystem {filesystem}') + try: + return rados_inst.open_ioctx2(metadata_pool_id) + except rados.Error as e: + log.error(f'failed to open metadata pool for {filesystem}: {e}') + raise MirrorException(-e.errno, + f'failed to open metadata pool for {filesystem}') + + +def load_sync_stat_metrics(ioctx, filesystem, peer_uuid=None): + metrics: Dict[str, Any] = {} + prefix = f'{SYNC_STAT_KEY_PREFIX}/{filesystem}/' + if peer_uuid: + prefix += f'{peer_uuid}/' + try: + with rados.ReadOpCtx() as read_op: + start = "" + while True: + it, ret = ioctx.get_omap_vals(read_op, start, prefix, MAX_RETURN) + if ret != 0: + log.error('failed to read sync stat omap') + raise MirrorException(-errno.EINVAL, + 'failed to read sync stat omap') + ioctx.operate_read_op(read_op, MIRROR_OBJECT_NAME) + omap_vals = dict(it) + if not omap_vals: + break + for key, val in omap_vals.items(): + parsed = parse_sync_stat_omap_key(key, filesystem) + if not parsed: + continue + peer, dir_path = parsed + if peer_uuid and peer != peer_uuid: + continue + try: + stat = decode_sync_stat_val(val) + except (json.JSONDecodeError, UnicodeDecodeError) as e: + log.warning(f'failed to decode sync stat for key {key}: {e}') + continue + nest_sync_stat_in_metrics(metrics, dir_path, peer, stat) + start = omap_vals.popitem()[0] + except rados.Error as e: + log.error(f'failed to read sync stat omap: {e}') + raise MirrorException(-e.errno, 'failed to read sync stat omap') + return metrics diff --git a/src/pybind/mgr/mirroring/fs/snapshot_mirror.py b/src/pybind/mgr/mirroring/fs/snapshot_mirror.py index 237b12a945a..e25a0b19458 100644 --- a/src/pybind/mgr/mirroring/fs/snapshot_mirror.py +++ b/src/pybind/mgr/mirroring/fs/snapshot_mirror.py @@ -19,7 +19,10 @@ from mgr_module import NotifyType from .blocklist import blocklist from .notify import Notifier, InstanceWatcher from .utils import INSTANCE_ID_PREFIX, MIRROR_OBJECT_NAME, Finisher, \ - AsyncOpTracker, connect_to_filesystem, disconnect_from_filesystem + AsyncOpTracker, get_metadata_pool, norm_path, connect_to_filesystem, \ + disconnect_from_filesystem, sync_stat_omap_key +from .metrics import load as metrics_load +from .metrics.load import nest_sync_stat_in_metrics from .exception import MirrorException from .dir_map.create import create_mirror_object from .dir_map.load import load_dir_map, load_instances @@ -313,13 +316,6 @@ class FSSnapshotMirror: except ValueError: raise MirrorException(-errno.EINVAL, f'invalid cluster spec {spec}') - @staticmethod - def get_metadata_pool(filesystem, fs_map): - for fs in fs_map['filesystems']: - if fs['mdsmap']['fs_name'] == filesystem: - return fs['mdsmap']['metadata_pool'] - return None - @staticmethod def get_filesystem_id(filesystem, fs_map): for fs in fs_map['filesystems']: @@ -483,7 +479,7 @@ class FSSnapshotMirror: disconnect_from_filesystem(cluster_name, remote_fs_name, remote_cluster, remote_fs) def init_pool_policy(self, filesystem): - metadata_pool_id = FSSnapshotMirror.get_metadata_pool(filesystem, self.fs_map) + metadata_pool_id = get_metadata_pool(filesystem, self.fs_map) if not metadata_pool_id: log.error(f'cannot find metadata pool-id for filesystem {filesystem}') raise Exception(-errno.EINVAL) @@ -522,7 +518,7 @@ class FSSnapshotMirror: log.info(f'enabling mirror for filesystem {filesystem}') with self.lock: try: - metadata_pool_id = FSSnapshotMirror.get_metadata_pool(filesystem, self.fs_map) + metadata_pool_id = get_metadata_pool(filesystem, self.fs_map) if not metadata_pool_id: log.error(f'cannot find metadata pool-id for filesystem {filesystem}') raise Exception(-errno.EINVAL) @@ -690,12 +686,6 @@ class FSSnapshotMirror: remote_cluster_spec = f'{client_name}@{cluster_name}' return self.peer_add(filesystem, remote_cluster_spec, remote_fs_name, token_dct) - @staticmethod - def norm_path(dir_path): - if not os.path.isabs(dir_path): - raise MirrorException(-errno.EINVAL, f'{dir_path} should be an absolute path') - return os.path.normpath(dir_path) - def add_dir(self, filesystem, dir_path): try: with self.lock: @@ -704,7 +694,7 @@ class FSSnapshotMirror: fspolicy = self.pool_policy.get(filesystem, None) if not fspolicy: raise MirrorException(-errno.EINVAL, f'filesystem {filesystem} is not mirrored') - dir_path = FSSnapshotMirror.norm_path(dir_path) + dir_path = norm_path(dir_path) log.debug(f'path normalized to {dir_path}') fspolicy.add_dir(dir_path) return 0, json.dumps({}), '' @@ -721,7 +711,7 @@ class FSSnapshotMirror: fspolicy = self.pool_policy.get(filesystem, None) if not fspolicy: raise MirrorException(-errno.EINVAL, f'filesystem {filesystem} is not mirrored') - dir_path = FSSnapshotMirror.norm_path(dir_path) + dir_path = norm_path(dir_path) fspolicy.remove_dir(dir_path) return 0, json.dumps({}), '' except MirrorException as me: @@ -751,7 +741,7 @@ class FSSnapshotMirror: fspolicy = self.pool_policy.get(filesystem, None) if not fspolicy: raise MirrorException(-errno.EINVAL, f'filesystem {filesystem} is not mirrored') - dir_path = FSSnapshotMirror.norm_path(dir_path) + dir_path = norm_path(dir_path) return fspolicy.status(dir_path) except MirrorException as me: return me.args[0], '', me.args[1] @@ -768,6 +758,53 @@ class FSSnapshotMirror: except MirrorException as me: return me.args[0], '', me.args[1] + def metrics_status(self, filesystem, mirrored_dir_path, peer_uuid): + """Return persisted mirror directory snapshot metrics as JSON""" + try: + with self.lock: + if not self.filesystem_exist(filesystem): + raise MirrorException(-errno.ENOENT, + f'filesystem {filesystem} does not exist') + fspolicy = self.pool_policy.get(filesystem, None) + if not fspolicy: + raise MirrorException(-errno.EINVAL, + f'filesystem {filesystem} is not mirrored') + if mirrored_dir_path: + dir_path = norm_path(mirrored_dir_path) + 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: + return 0, json.dumps({}), '' + + if peer_uuid: + if peer_uuid not in peers: + raise MirrorException(-errno.ENOENT, + f'peer {peer_uuid} not found for ' + f'filesystem {filesystem}') + peers = {peer_uuid: peers[peer_uuid]} + + 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: + nest_sync_stat_in_metrics(metrics, dir_path, peer, + omap_stats[omap_key]) + else: + metrics = metrics_load.load_sync_stat_metrics( + ioctx, filesystem, peer_uuid) + return 0, json.dumps(metrics, indent=4), '' + except MirrorException as me: + return me.args[0], '', me.args[1] + def daemon_status(self, format='json'): try: with self.lock: diff --git a/src/pybind/mgr/mirroring/fs/utils.py b/src/pybind/mgr/mirroring/fs/utils.py index 5e1d0537356..a67e6c06a7f 100644 --- a/src/pybind/mgr/mirroring/fs/utils.py +++ b/src/pybind/mgr/mirroring/fs/utils.py @@ -1,5 +1,7 @@ import errno +import json import logging +import os import threading import rados @@ -12,9 +14,43 @@ MIRROR_OBJECT_NAME = MIRROR_OBJECT_PREFIX INSTANCE_ID_PREFIX = "instance_" DIRECTORY_MAP_PREFIX = "dir_map_" +SYNC_STAT_KEY_PREFIX = "sync_stat" log = logging.getLogger(__name__) + +def get_metadata_pool(filesystem, fs_map): + for fs in fs_map['filesystems']: + if fs['mdsmap']['fs_name'] == filesystem: + return fs['mdsmap']['metadata_pool'] + return None + + +def norm_path(dir_path): + if not os.path.isabs(dir_path): + raise MirrorException(-errno.EINVAL, f'{dir_path} should be an absolute path') + return os.path.normpath(dir_path) + + +def sync_stat_omap_key(filesystem, peer_uuid, dir_path): + return (f'{SYNC_STAT_KEY_PREFIX}/{filesystem}/{peer_uuid}/' + f'{dir_path.lstrip("/")}') + + +def parse_sync_stat_omap_key(key, filesystem): + prefix = f'{SYNC_STAT_KEY_PREFIX}/{filesystem}/' + if not key.startswith(prefix): + return None + suffix = key[len(prefix):] + peer_uuid, sep, dir_rel = suffix.partition('/') + if not peer_uuid or not sep: + return None + return peer_uuid, os.path.normpath('/' + dir_rel) + + +def decode_sync_stat_val(val): + return json.loads(val.decode('utf-8')) + def connect_to_cluster(client_name, cluster_name, conf_dct, desc=''): try: log.debug(f'connecting to {desc} cluster: {client_name}/{cluster_name}') diff --git a/src/pybind/mgr/mirroring/module.py b/src/pybind/mgr/mirroring/module.py index 4a9813c6df9..1321e27f749 100644 --- a/src/pybind/mgr/mirroring/module.py +++ b/src/pybind/mgr/mirroring/module.py @@ -116,3 +116,13 @@ class Module(MgrModule): format: str = 'json'): """Get mirror daemon status""" return self.fs_snapshot_mirror.daemon_status(format) + + @MirroringCLICommand.Read('fs snapshot mirror status') + def snapshot_mirror_status(self, + fs_name: str, + mirrored_dir_path: Optional[str] = None, + _end_positional_: int = 0, + peer_uuid: Optional[str] = None): + """Get snapshot mirror metrics for a filesystem (optional mirrored directory and peer)""" + return self.fs_snapshot_mirror.metrics_status(fs_name, mirrored_dir_path, + peer_uuid)