--- /dev/null
+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
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
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']:
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)
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)
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:
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({}), ''
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:
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]
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:
import errno
+import json
import logging
+import os
import threading
import rados
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}')
projects / ceph.git / commitdiff