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/server_family_test.py

@@ -7,7 +7,7 @@
 from redis.exceptions import ResponseError
 
 from . import dfly_args
-from .instance import DflyInstance
+from .instance import DflyInstance, DflyInstanceFactory
 from .utility import *
 
 
@@ -313,3 +313,25 @@ async def check_metrics():
         assert m.samples[0].value > 0
 
     await check_metrics()
+
+
+async def test_replication_metrics(df_factory: DflyInstanceFactory):
+    master = df_factory.create(proactor_threads=2)
+    replica = df_factory.create(proactor_threads=2)
+    df_factory.start_all([master, replica])
+
+    c_replica = replica.client()
+    await c_replica.execute_command("REPLICAOF", "localhost", master.port)
+    await wait_for_replicas_state(c_replica)
+
+    # Replica should expose replication metrics
+    metrics = await replica.metrics()
+    assert metrics["dragonfly_master_link_status"].samples[0].value == 1
+    assert metrics["dragonfly_master_sync_in_progress"].samples[0].value == 0
+    assert metrics["dragonfly_master_last_io_seconds_ago"].samples[0].value >= 0
+    assert "dragonfly_slave_repl_offset" in 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