Merge pull request ceph#56782 from joscollin/wip-B65171-support-replication-start-end-notifications

cephfs_mirror: provide metrics for last successful snapshot sync

Reviewed-by: Venky Shankar <[email protected]>

Author: vshankar

doc/cephfs/cephfs-mirroring.rst

@@ -341,7 +341,8 @@ command is of format `filesystem-name@filesystem-id peer-uuid`::
             "id": 120,
             "name": "snap1",
             "sync_duration": 0.079997898999999997,
-            "sync_time_stamp": "274900.558797s"
+            "sync_time_stamp": "274900.558797s",
+            "sync_bytes": 52428800
         },
         "snaps_synced": 2,
         "snaps_deleted": 0,
@@ -359,6 +360,32 @@ A directory can be in one of the following states::
   - `syncing`: The directory is currently being synchronized
   - `failed`: The directory has hit upper limit of consecutive failures
 
+When a directory is currently being synchronized, the mirror daemon marks it as `syncing` and
+`fs mirror peer status` shows the snapshot being synchronized under the `current_syncing_snap`::
+
+  $ ceph --admin-daemon /var/run/ceph/cephfs-mirror.asok fs mirror peer status cephfs@360 a2dc7784-e7a1-4723-b103-03ee8d8768f8
+  {
+    "/d0": {
+        "state": "syncing",
+        "current_syncing_snap": {
+            "id": 121,
+            "name": "snap2"
+        },
+        "last_synced_snap": {
+            "id": 120,
+            "name": "snap1",
+            "sync_duration": 0.079997898999999997,
+            "sync_time_stamp": "274900.558797s",
+            "sync_bytes": 52428800
+        },
+        "snaps_synced": 2,
+        "snaps_deleted": 0,
+        "snaps_renamed": 0
+    }
+  }
+
+The mirror daemon marks it back to `idle`, when the syncing completes.
+
 When a directory experiences a configured number of consecutive synchronization failures, the
 mirror daemon marks it as `failed`. Synchronization for these directories is retried.
 By default, the number of consecutive failures before a directory is marked as failed
@@ -374,12 +401,13 @@ E.g., adding a regular file for synchronization would result in failed status::
     "/d0": {
         "state": "idle",
         "last_synced_snap": {
-            "id": 120,
-            "name": "snap1",
-            "sync_duration": 0.079997898999999997,
-            "sync_time_stamp": "274900.558797s"
+            "id": 121,
+            "name": "snap2",
+            "sync_duration": 300,
+            "sync_time_stamp": "500900.600797s",
+            "sync_bytes": 78643200
         },
-        "snaps_synced": 2,
+        "snaps_synced": 3,
         "snaps_deleted": 0,
         "snaps_renamed": 0
     },
@@ -398,6 +426,69 @@ created, the mirror daemon will clear the failed state upon successful synchroni
 When mirroring is disabled, the respective `fs mirror status` command for the file system
 will not show up in command help.
 
+Metrics
+-------
+
+CephFS exports mirroring metrics as :ref:`Labeled Perf Counters` which will be consumed by the OCP/ODF Dashboard to provide monitoring of the Geo Replication. These metrics can be used to measure the progress of cephfs_mirror syncing and thus provide the monitoring capability. CephFS exports the following mirroring metrics, which are displayed using the ``counter dump`` command.
+
+.. list-table:: Mirror Status Metrics
+   :widths: 25 25 75
+   :header-rows: 1
+
+   * - Name
+     - Type
+     - Description
+   * - mirroring_peers
+     - Gauge
+     - The number of peers involved in mirroring
+   * - directory_count
+     - Gauge
+     - The total number of directories being synchronized
+   * - mirrored_filesystems
+     - Gauge
+     - The total number of filesystems which are mirrored
+   * - mirror_enable_failures
+     - Counter
+     - Enable mirroring failures
+
+.. list-table:: Replication Metrics
+   :widths: 25 25 75
+   :header-rows: 1
+
+   * - Name
+     - Type
+     - Description
+   * - snaps_synced
+     - Counter
+     - The total number of snapshots successfully synchronized
+   * - sync_bytes
+     - Counter
+     - The total bytes being synchronized
+   * - sync_failures
+     - Counter
+     - The total number of failed snapshot synchronizations
+   * - snaps_deleted
+     - Counter
+     - The total number of snapshots deleted
+   * - snaps_renamed
+     - Counter
+     - The total number of snapshots renamed
+   * - avg_sync_time
+     - Gauge
+     - The average time (ms) taken by all snapshot synchronizations
+   * - last_synced_start
+     - Gauge
+     - The sync start time (ms) of the last synced snapshot
+   * - last_synced_end
+     - Gauge
+     - The sync end time (ms) of the last synced snapshot
+   * - last_synced_duration
+     - Gauge
+     - The time duration (ms) of the last synchronization
+   * - last_synced_bytes
+     - counter
+     - The total bytes being synchronized for the last synced snapshot
+
 Configuration Options
 ---------------------
 
@@ -410,6 +501,7 @@ Configuration Options
 .. confval:: cephfs_mirror_retry_failed_directories_interval
 .. confval:: cephfs_mirror_restart_mirror_on_failure_interval
 .. confval:: cephfs_mirror_mount_timeout
+.. confval:: cephfs_mirror_perf_stats_prio
 
 Re-adding Peers
 ---------------

qa/tasks/cephfs/test_mirroring.py

@@ -549,7 +549,8 @@ def test_cephfs_mirror_stats(self):
 
         # create a bunch of files in a directory to snap
         self.mount_a.run_shell(["mkdir", "d0"])
-        self.mount_a.create_n_files('d0/file', 50, sync=True)
+        for i in range(50):
+            self.mount_a.write_n_mb(os.path.join('d0', f'file.{i}'), 1)
 
         self.enable_mirroring(self.primary_fs_name, self.primary_fs_id)
         self.add_directory(self.primary_fs_name, self.primary_fs_id, '/d0')
@@ -567,17 +568,20 @@ def test_cephfs_mirror_stats(self):
                                "client.mirror_remote@ceph", '/d0', 'snap0', 1)
         self.verify_snapshot('d0', 'snap0')
 
-        # check snaps_synced
+        # check perf counters
         res = self.mirror_daemon_command(f'counter dump for fs: {self.primary_fs_name}', 'counter', 'dump')
         second = res[TestMirroring.PERF_COUNTER_KEY_NAME_CEPHFS_MIRROR_PEER][0]
         self.assertGreater(second["counters"]["snaps_synced"], first["counters"]["snaps_synced"])
+        self.assertGreater(second["counters"]["last_synced_start"], first["counters"]["last_synced_start"])
+        self.assertGreater(second["counters"]["last_synced_end"], second["counters"]["last_synced_start"])
+        self.assertGreater(second["counters"]["last_synced_duration"], 0)
+        self.assertEquals(second["counters"]["last_synced_bytes"], 52428800) # last_synced_bytes = 50 files of 1MB size each
 
         # some more IO
-        self.mount_a.run_shell(["mkdir", "d0/d00"])
-        self.mount_a.run_shell(["mkdir", "d0/d01"])
+        for i in range(75):
+            self.mount_a.write_n_mb(os.path.join('d0', f'more_file.{i}'), 1)
 
-        self.mount_a.create_n_files('d0/d00/more_file', 20, sync=True)
-        self.mount_a.create_n_files('d0/d01/some_more_file', 75, sync=True)
+        time.sleep(60)
 
         # take another snapshot
         self.mount_a.run_shell(["mkdir", "d0/.snap/snap1"])
@@ -587,10 +591,14 @@ def test_cephfs_mirror_stats(self):
                                "client.mirror_remote@ceph", '/d0', 'snap1', 2)
         self.verify_snapshot('d0', 'snap1')
 
-        # check snaps_synced
+        # check perf counters
         res = self.mirror_daemon_command(f'counter dump for fs: {self.primary_fs_name}', 'counter', 'dump')
         third = res[TestMirroring.PERF_COUNTER_KEY_NAME_CEPHFS_MIRROR_PEER][0]
         self.assertGreater(third["counters"]["snaps_synced"], second["counters"]["snaps_synced"])
+        self.assertGreater(third["counters"]["last_synced_start"], second["counters"]["last_synced_end"])
+        self.assertGreater(third["counters"]["last_synced_end"], third["counters"]["last_synced_start"])
+        self.assertGreater(third["counters"]["last_synced_duration"], 0)
+        self.assertEquals(third["counters"]["last_synced_bytes"], 78643200) # last_synced_bytes = 75 files of 1MB size each
 
         # delete a snapshot
         self.mount_a.run_shell(["rmdir", "d0/.snap/snap0"])

src/tools/cephfs_mirror/PeerReplayer.cc

@@ -38,6 +38,10 @@ enum {
   l_cephfs_mirror_peer_replayer_snap_sync_failures,
   l_cephfs_mirror_peer_replayer_avg_sync_time,
   l_cephfs_mirror_peer_replayer_sync_bytes,
+  l_cephfs_mirror_peer_replayer_last_synced_start,
+  l_cephfs_mirror_peer_replayer_last_synced_end,
+  l_cephfs_mirror_peer_replayer_last_synced_duration,
+  l_cephfs_mirror_peer_replayer_last_synced_bytes,
   l_cephfs_mirror_peer_replayer_last,
 };
 
@@ -203,6 +207,14 @@ PeerReplayer::PeerReplayer(CephContext *cct, FSMirror *fs_mirror,
 		   "avg_sync_time", "Average Sync Time", "asyn", prio);
   plb.add_u64_counter(l_cephfs_mirror_peer_replayer_sync_bytes,
 		      "sync_bytes", "Sync Bytes", "sbye", prio);
+  plb.add_time(l_cephfs_mirror_peer_replayer_last_synced_start,
+	       "last_synced_start", "Last Synced Start", "lsst", prio);
+  plb.add_time(l_cephfs_mirror_peer_replayer_last_synced_end,
+	       "last_synced_end", "Last Synced End", "lsen", prio);
+  plb.add_time(l_cephfs_mirror_peer_replayer_last_synced_duration,
+	       "last_synced_duration", "Last Synced Duration", "lsdn", prio);
+  plb.add_u64_counter(l_cephfs_mirror_peer_replayer_last_synced_bytes,
+		      "last_synced_bytes", "Last Synced Bytes", "lsbt", prio);
   m_perf_counters = plb.create_perf_counters();
   m_cct->get_perfcounters_collection()->add(m_perf_counters);
 }
@@ -751,8 +763,9 @@ int PeerReplayer::remote_file_op(const std::string &dir_root, const std::string
         return r;
       }
       if (m_perf_counters) {
-	m_perf_counters->inc(l_cephfs_mirror_peer_replayer_sync_bytes, stx.stx_size);
+        m_perf_counters->inc(l_cephfs_mirror_peer_replayer_sync_bytes, stx.stx_size);
       }
+      inc_sync_bytes(dir_root, stx.stx_size);
     } else if (S_ISLNK(stx.stx_mode)) {
       // free the remote link before relinking
       r = ceph_unlinkat(m_remote_mount, fh.r_fd_dir_root, epath.c_str(), 0);
@@ -1668,9 +1681,17 @@ int PeerReplayer::do_sync_snaps(const std::string &dir_root) {
     "cephfs_mirror_max_snapshot_sync_per_cycle");
 
   dout(10) << ": synchronizing from snap-id=" << it->first << dendl;
+  double start = 0;
+  double end = 0;
+  double duration = 0;
   for (; it != local_snap_map.end(); ++it) {
+    if (m_perf_counters) {
+      start = std::chrono::duration_cast<std::chrono::milliseconds>(clock::now().time_since_epoch()).count();
+      utime_t t;
+      t.set_from_double(start);
+      m_perf_counters->tset(l_cephfs_mirror_peer_replayer_last_synced_start, t);
+    }
     set_current_syncing_snap(dir_root, it->first, it->second);
-    auto start = clock::now();
     boost::optional<Snapshot> prev = boost::none;
     if (last_snap_id != 0) {
       prev = std::make_pair(last_snap_name, last_snap_id);
@@ -1684,16 +1705,18 @@ int PeerReplayer::do_sync_snaps(const std::string &dir_root) {
     }
     if (m_perf_counters) {
       m_perf_counters->inc(l_cephfs_mirror_peer_replayer_snaps_synced);
-    }
-    std::chrono::duration<double> duration = clock::now() - start;
-
-    utime_t d;
-    d.set_from_double(duration.count());
-    if (m_perf_counters) {
-      m_perf_counters->tinc(l_cephfs_mirror_peer_replayer_avg_sync_time, d);
-    }
-
-    set_last_synced_stat(dir_root, it->first, it->second, duration.count());
+      end = std::chrono::duration_cast<std::chrono::milliseconds>(clock::now().time_since_epoch()).count();
+      utime_t t;
+      t.set_from_double(end);
+      m_perf_counters->tset(l_cephfs_mirror_peer_replayer_last_synced_end, t);
+      duration = end - start;
+      t.set_from_double(duration);
+      m_perf_counters->tinc(l_cephfs_mirror_peer_replayer_avg_sync_time, t);
+      m_perf_counters->tset(l_cephfs_mirror_peer_replayer_last_synced_duration, t);
+      m_perf_counters->set(l_cephfs_mirror_peer_replayer_last_synced_bytes, m_snap_sync_stats.at(dir_root).sync_bytes);
+    }
+
+    set_last_synced_stat(dir_root, it->first, it->second, duration);
     if (--snaps_per_cycle == 0) {
       break;
     }
@@ -1801,6 +1824,9 @@ void PeerReplayer::peer_status(Formatter *f) {
         f->dump_float("sync_duration", *sync_stat.last_sync_duration);
         f->dump_stream("sync_time_stamp") << sync_stat.last_synced;
       }
+      if (sync_stat.last_sync_bytes) {
+	f->dump_unsigned("sync_bytes", *sync_stat.last_sync_bytes);
+      }
       f->close_section();
     }
     f->dump_unsigned("snaps_synced", sync_stat.synced_snap_count);

src/tools/cephfs_mirror/PeerReplayer.h

@@ -149,6 +149,8 @@ class PeerReplayer {
     uint64_t renamed_snap_count = 0;
     monotime last_synced = clock::zero();
     boost::optional<double> last_sync_duration;
+    boost::optional<uint64_t> last_sync_bytes; //last sync bytes for display in status
+    uint64_t sync_bytes = 0; //sync bytes counter, independently for each directory sync.
   };
 
   void _inc_failed_count(const std::string &dir_root) {
@@ -187,6 +189,8 @@ class PeerReplayer {
                             const std::string &snap_name) {
     std::scoped_lock locker(m_lock);
     _set_last_synced_snap(dir_root, snap_id, snap_name);
+    auto &sync_stat = m_snap_sync_stats.at(dir_root);
+    sync_stat.sync_bytes = 0;
   }
   void set_current_syncing_snap(const std::string &dir_root, uint64_t snap_id,
                                 const std::string &snap_name) {
@@ -216,9 +220,14 @@ class PeerReplayer {
     auto &sync_stat = m_snap_sync_stats.at(dir_root);
     sync_stat.last_synced = clock::now();
     sync_stat.last_sync_duration = duration;
+    sync_stat.last_sync_bytes = sync_stat.sync_bytes;
     ++sync_stat.synced_snap_count;
   }
-
+  void inc_sync_bytes(const std::string &dir_root, const uint64_t& b) {
+    std::scoped_lock locker(m_lock);
+    auto &sync_stat = m_snap_sync_stats.at(dir_root);
+    sync_stat.sync_bytes += b;
+  }
   bool should_backoff(const std::string &dir_root, int *retval) {
     if (m_fs_mirror->is_blocklisted()) {
       *retval = -EBLOCKLISTED;