feat: add replication info to /metrics endpoint

Exposes 4 new fields to the `/metrics` endpoint relating to replication info:
- `master_link_status`
- `master_last_io_seconds_ago`
- `master_sync_in_progress`
- `slave_repl_offset`

This commit also adds integration testing to validate the new functionality.

Fixes #6183

Signed-off-by: Eric <hayter.eric@gmail.com>

Author: EricHayter

src/server/server_family.cc

@@ -2127,6 +2127,24 @@ void PrintPrometheusMetrics(uint64_t uptime, const Metrics& m, DflyCmd* dfly_cmd
                       &resp->body());
   }
 
+  // Replication Info
+  if (m.replica_side_info) {
+    const ReplicaSummary& rsummary = m.replica_side_info->summary;
+    AppendMetricWithoutLabels("master_link_status", "1 if up 0 if down",
+                              rsummary.master_link_established ? 1 : 0, MetricType::GAUGE,
+                              &resp->body());
+    AppendMetricWithoutLabels("master_last_io_seconds_ago", "Last Master IO Seconds Ago",
+                              rsummary.master_last_io_sec, MetricType::GAUGE, &resp->body());
+    AppendMetricWithoutLabels("master_sync_in_progress", "1 if true 0 if false",
+                              rsummary.full_sync_in_progress ? 1 : 0, MetricType::GAUGE,
+                              &resp->body());
+    // Print last known offset either during stable sync (online) or during disconnects when
+    // the full sync phase did not start yet.
+    if (rsummary.full_sync_done || (rsummary.passed_full_sync && !rsummary.master_link_established))
+      AppendMetricWithoutLabels("slave_repl_offset", "Slave Replication Offset",
+                                rsummary.repl_offset_sum, MetricType::GAUGE, &resp->body());
+  }
+
   // Stream access pattern metrics
   if (m.shard_stats.stream_sequential_accesses || m.shard_stats.stream_random_accesses ||
       m.shard_stats.stream_fetch_all_accesses) {

tests/dragonfly/replication_test.py

@@ -1147,6 +1147,18 @@ async def test_replication_info(df_factory: DflyInstanceFactory, df_seeder_facto
     await wait_available_async(c_replica)
     await assert_lag_condition(master, c_master, lambda lag: lag == 0)
 
+    # Replica should expose replication metrics
+    replica_metrics = await replica.metrics()
+    assert replica_metrics["dragonfly_master_link_status"].samples[0].value == 1
+    assert replica_metrics["dragonfly_master_sync_in_progress"].samples[0].value == 0
+    assert replica_metrics["dragonfly_master_last_io_seconds_ago"].samples[0].value >= 0
+    assert "dragonfly_slave_repl_offset" in replica_metrics
+
+    # Master should not expose replica-side metrics
+    master_metrics = await master.metrics()
+    assert "dragonfly_master_link_status" not in master_metrics
+    assert "dragonfly_slave_repl_offset" not in master_metrics
+
     await c_master.connection_pool.disconnect()
     await c_replica.connection_pool.disconnect()