--- /dev/null
+# Matches the format_time function used in peer_status
+def _format_time(total_seconds_d):
+ total_seconds = int(round(total_seconds_d))
+ days = total_seconds // 86400
+ total_seconds %= 86400
+ hours = total_seconds // 3600
+ total_seconds %= 3600
+ minutes = total_seconds // 60
+ seconds = total_seconds % 60
+ if days > 0:
+ return (f'{days}d {hours:02d}h {minutes:02d}m {seconds:02d}s')
+ if hours > 0:
+ return f'{hours}h {minutes:02d}m {seconds:02d}s'
+ if minutes > 0:
+ return f'{minutes}m {seconds:02d}s'
+ return f'{seconds}s'
+
+
+# Matches the format_bytes function used in peer_status
+def _format_bytes(bytes_val):
+ kib = 1024.0
+ mib = kib * 1024.0
+ gib = mib * 1024.0
+ tib = gib * 1024.0
+ pib = tib * 1024.0
+ if bytes_val >= pib:
+ return f'{bytes_val / pib:.2f} PiB'
+ if bytes_val >= tib:
+ return f'{bytes_val / tib:.2f} TiB'
+ if bytes_val >= gib:
+ return f'{bytes_val / gib:.2f} GiB'
+ if bytes_val >= mib:
+ return f'{bytes_val / mib:.2f} MiB'
+ if bytes_val >= kib:
+ return f'{bytes_val / kib:.2f} KiB'
+ return f'{bytes_val:.2f} B'
+
+
+def _order_dict(d, keys):
+ ordered = {}
+ for key in keys:
+ if key in d:
+ ordered[key] = d[key]
+ for key, val in d.items():
+ if key not in ordered:
+ ordered[key] = val
+ return ordered
+
+
+def _order_current_syncing_snap(snap):
+ if not isinstance(snap, dict):
+ return snap
+ snap = dict(snap)
+ crawl = snap.get('crawl')
+ if isinstance(crawl, dict):
+ snap['crawl'] = _order_dict(crawl, ('state', 'duration'))
+ dq_wait = snap.get('datasync_queue_wait')
+ if isinstance(dq_wait, dict):
+ snap['datasync_queue_wait'] = _order_dict(dq_wait, ('state', 'duration'))
+ bytes_obj = snap.get('bytes')
+ if isinstance(bytes_obj, dict):
+ snap['bytes'] = _order_dict(
+ bytes_obj, ('sync_bytes', 'total_bytes', 'sync_percent'))
+ files_obj = snap.get('files')
+ if isinstance(files_obj, dict):
+ snap['files'] = _order_dict(
+ files_obj, ('sync_files', 'total_files', 'sync_percent'))
+ return _order_dict(
+ snap, ('id', 'name', 'sync-mode', 'avg_read_throughput_bytes',
+ 'avg_write_throughput_bytes', 'crawl', 'datasync_queue_wait',
+ 'bytes', 'files', 'eta'))
+
+
+def format_peer_status_metrics(metrics, dir_path, peer_uuid, stat):
+ metrics.setdefault(dir_path, {}).setdefault('peer', {})[peer_uuid] = stat
+
+
+# to match the output of peer_status
+def format_and_order_sync_stat_for_display(stat):
+ if not isinstance(stat, dict):
+ return stat
+
+ out = dict(stat)
+ last_synced_snap = out.get('last_synced_snap')
+ if isinstance(last_synced_snap, dict):
+ snap = dict(last_synced_snap)
+ for key in ('crawl_duration', 'datasync_queue_wait_duration',
+ 'sync_duration'):
+ val = snap.get(key)
+ if isinstance(val, (int, float)):
+ snap[key] = _format_time(val)
+ sync_bytes = snap.get('sync_bytes')
+ if isinstance(sync_bytes, (int, float)):
+ snap['sync_bytes'] = _format_bytes(sync_bytes)
+ sync_time_stamp = snap.get('sync_time_stamp')
+ if isinstance(sync_time_stamp, (int, float)):
+ snap['sync_time_stamp'] = f'{sync_time_stamp:.6f}s'
+ out['last_synced_snap'] = _order_dict(
+ snap, ('id', 'name', 'crawl_duration',
+ 'datasync_queue_wait_duration', 'sync_duration',
+ 'sync_time_stamp', 'sync_bytes', 'sync_files'))
+
+ current_syncing_snap = out.get('current_syncing_snap')
+ if isinstance(current_syncing_snap, dict):
+ out['current_syncing_snap'] = _order_current_syncing_snap(
+ current_syncing_snap)
+
+ out.pop('_instance_id', None)
+
+ return _order_dict(
+ out, ('state', 'failure_reason', 'current_syncing_snap',
+ 'last_synced_snap', 'snaps_synced', 'snaps_deleted',
+ 'snaps_renamed', 'metrics_updated_at'))
get_metadata_pool,
parse_sync_stat_omap_key,
)
+from .format import format_and_order_sync_stat_for_display, format_peer_status_metrics
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:
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)
+ format_peer_status_metrics(
+ metrics, dir_path, peer,
+ format_and_order_sync_stat_for_display(stat))
start = omap_vals.popitem()[0]
except rados.Error as e:
log.error(f'failed to read sync stat omap: {e}')
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 .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
f'directory {dir_path} is not mirrored')
peers = self.get_filesystem_peers(filesystem)
if not peers:
- return 0, json.dumps({}), ''
+ return 0, json.dumps({'metrics': {}}, indent=4), ''
if peer_uuid:
if peer_uuid not in peers:
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])
+ format_peer_status_metrics(
+ metrics, dir_path, peer,
+ format_and_order_sync_stat_for_display(
+ omap_stats[omap_key]))
else:
metrics = metrics_load.load_sync_stat_metrics(
ioctx, filesystem, peer_uuid)
- return 0, json.dumps(metrics, indent=4), ''
+ return 0, json.dumps({'metrics': metrics}, indent=4), ''
except MirrorException as me:
return me.args[0], '', me.args[1]
projects / ceph.git / commitdiff