[#28910] xCluster: Add safe time to UI

Summary:
Display the xCluster safe time in yb-master xCluster UI page to improve debugging capabilities.

New UI output

{image uri=https://github.com/user-attachments/assets/bcd96f4f-fa5a-44d5-9b50-2d828c43d277, width=500px}

Fixes #28910
Fixes #28806
Jira: DB-18633

Test Plan: Manually tested

Reviewers: jhe

Reviewed By: jhe

Subscribers: ybase

Differential Revision: https://phorge.dev.yugabyte.com/D45287

Author: hari90

src/yb/master/master-path-handlers.cc

@@ -3130,8 +3130,7 @@ void MasterPathHandlers::HandleXCluster(
 
     output << "<pre class=\"prettyprint\">"
            << "state: " << inbound_replication_group.state
-           << "\ndisable_stream: " << BoolToString(inbound_replication_group.disable_stream)
-           << "\ntype: "
+           << (inbound_replication_group.disable_stream ? " (DISABLED)" : "") << "\ntype: "
            << xcluster::ShortReplicationType(inbound_replication_group.replication_type)
            << "\nmaster_addrs: " << inbound_replication_group.master_addrs;
     if (!inbound_replication_group.db_scoped_info.empty()) {

src/yb/master/xcluster/add_table_to_xcluster_target_task.cc

@@ -210,7 +210,7 @@ AddTableToXClusterTargetTask::GetXClusterSafeTimeWithoutDdlQueue() {
   const auto namespace_id = table_info_->namespace_id();
 
   auto safe_time_res = xcluster_manager_.GetXClusterSafeTimeForNamespace(
-      epoch_, namespace_id, XClusterSafeTimeFilter::DDL_QUEUE);
+      namespace_id, XClusterSafeTimeFilter::DDL_QUEUE);
   if (!safe_time_res) {
     if (!safe_time_res.status().IsNotFound()) {
       return safe_time_res.status();

src/yb/master/xcluster/xcluster_manager.cc

@@ -344,9 +344,8 @@ Status XClusterManager::GetXClusterSafeTimeForNamespace(
 }
 
 Result<HybridTime> XClusterManager::GetXClusterSafeTimeForNamespace(
-    const LeaderEpoch& epoch, const NamespaceId& namespace_id,
-    const XClusterSafeTimeFilter& filter) {
-  return XClusterTargetManager::GetXClusterSafeTimeForNamespace(epoch, namespace_id, filter);
+    const NamespaceId& namespace_id, const XClusterSafeTimeFilter& filter) const {
+  return XClusterTargetManager::GetXClusterSafeTimeForNamespace(namespace_id, filter);
 }
 
 Status XClusterManager::RefreshXClusterSafeTimeMap(const LeaderEpoch& epoch) {

src/yb/master/xcluster/xcluster_manager.h

@@ -109,8 +109,7 @@ class XClusterManager : public XClusterManagerIf,
       GetXClusterSafeTimeForNamespaceResponsePB* resp, rpc::RpcContext* rpc,
       const LeaderEpoch& epoch);
   Result<HybridTime> GetXClusterSafeTimeForNamespace(
-      const LeaderEpoch& epoch, const NamespaceId& namespace_id,
-      const XClusterSafeTimeFilter& filter) override;
+      const NamespaceId& namespace_id, const XClusterSafeTimeFilter& filter) const override;
 
   Status RefreshXClusterSafeTimeMap(const LeaderEpoch& epoch) override;
 

src/yb/master/xcluster/xcluster_manager_if.h

@@ -51,8 +51,7 @@ class XClusterManagerIf {
   virtual Status SetXClusterNamespaceToSafeTimeMap(
       const int64_t leader_term, const XClusterNamespaceToSafeTimeMap& safe_time_map) = 0;
   virtual Result<HybridTime> GetXClusterSafeTimeForNamespace(
-      const LeaderEpoch& epoch, const NamespaceId& namespace_id,
-      const XClusterSafeTimeFilter& filter) = 0;
+      const NamespaceId& namespace_id, const XClusterSafeTimeFilter& filter) const = 0;
   virtual Status MarkIndexBackfillCompleted(
       const std::unordered_set<TableId>& index_ids, const LeaderEpoch& epoch) = 0;
 

src/yb/master/xcluster/xcluster_safe_time_service-test.cc

@@ -58,7 +58,7 @@ class XClusterSafeTimeServiceMocked : public XClusterSafeTimeService {
     return OK();
   }
 
-  XClusterNamespaceToSafeTimeMap GetXClusterNamespaceToSafeTimeMap() override {
+  XClusterNamespaceToSafeTimeMap GetXClusterNamespaceToSafeTimeMap() const override {
     return safe_time_map_;
   }
 
@@ -133,13 +133,13 @@ class XClusterSafeTimeServiceTest : public YBTest {
   Result<HybridTime> GetXClusterSafeTimeWithNoFilter(
       XClusterSafeTimeServiceMocked& safe_time_service, const NamespaceId& namespace_id) {
     return safe_time_service.GetXClusterSafeTimeForNamespace(
-        dummy_leader_term, namespace_id, XClusterSafeTimeFilter::NONE);
+        namespace_id, XClusterSafeTimeFilter::NONE);
   }
 
   Result<HybridTime> GetXClusterSafeTimeFilterOutDdlQueue(
       XClusterSafeTimeServiceMocked& safe_time_service, const NamespaceId& namespace_id) {
     return safe_time_service.GetXClusterSafeTimeForNamespace(
-        dummy_leader_term, namespace_id, XClusterSafeTimeFilter::DDL_QUEUE);
+        namespace_id, XClusterSafeTimeFilter::DDL_QUEUE);
   }
 };
 

src/yb/master/xcluster/xcluster_safe_time_service.cc

@@ -133,10 +133,11 @@ void XClusterSafeTimeService::ProcessTaskPeriodically() {
   }
 }
 
-Status XClusterSafeTimeService::GetXClusterSafeTimeInfoFromMap(
-    const LeaderEpoch& epoch, GetXClusterSafeTimeResponsePB* resp) {
+Status XClusterSafeTimeService::ComputeAndGetXClusterSafeTimeInfo(
+    const LeaderEpoch& epoch, GetXClusterSafeTimeResponsePB& resp) {
   // Recompute safe times again before fetching maps.
   RETURN_NOT_OK(ComputeSafeTime(epoch.leader_term));
+
   const auto& current_safe_time_map = GetXClusterNamespaceToSafeTimeMap();
   XClusterNamespaceToSafeTimeMap max_safe_time_map;
   {
@@ -147,7 +148,7 @@ Status XClusterSafeTimeService::GetXClusterSafeTimeInfoFromMap(
 
   for (const auto& [namespace_id, safe_time] : current_safe_time_map) {
     // First set all the current safe time values.
-    auto entry = resp->add_namespace_safe_times();
+    auto entry = resp.add_namespace_safe_times();
     entry->set_namespace_id(namespace_id);
     entry->set_safe_time_ht(safe_time.ToUint64());
     // Safe time lag is calculated as (current time - current safe time).
@@ -177,7 +178,7 @@ Status XClusterSafeTimeService::GetXClusterSafeTimeInfoFromMap(
 }
 
 Result<XClusterNamespaceToSafeTimeMap> XClusterSafeTimeService::GetFilteredXClusterSafeTimeMap(
-    const XClusterSafeTimeFilter& filter) {
+    const XClusterSafeTimeFilter& filter) const {
   switch (filter) {
     case XClusterSafeTimeFilter::NONE:
       return GetXClusterNamespaceToSafeTimeMap();
@@ -189,8 +190,7 @@ Result<XClusterNamespaceToSafeTimeMap> XClusterSafeTimeService::GetFilteredXClus
 
 // If the filter removes all tables, then this returns master leader's safe time.
 Result<HybridTime> XClusterSafeTimeService::GetXClusterSafeTimeForNamespace(
-    const int64_t leader_term, const NamespaceId& namespace_id,
-    const XClusterSafeTimeFilter& filter) {
+    const NamespaceId& namespace_id, const XClusterSafeTimeFilter& filter) const {
   SharedLock lock(mutex_);
   SCHECK(safe_time_table_ready_, IllegalState, "Safe time table is not ready yet.");
 
@@ -603,7 +603,7 @@ Result<bool> XClusterSafeTimeService::CreateTableRequired() {
   return !producer_tablet_namespace_map_.empty();
 }
 
-XClusterNamespaceToSafeTimeMap XClusterSafeTimeService::GetXClusterNamespaceToSafeTimeMap() {
+XClusterNamespaceToSafeTimeMap XClusterSafeTimeService::GetXClusterNamespaceToSafeTimeMap() const {
   return master_->xcluster_manager()->GetXClusterNamespaceToSafeTimeMap();
 }
 

src/yb/master/xcluster/xcluster_safe_time_service.h

@@ -73,12 +73,12 @@ class XClusterSafeTimeService {
   Result<std::unordered_map<NamespaceId, uint64_t>> GetEstimatedDataLossMicroSec(
       const LeaderEpoch& epoch);
 
-  Status GetXClusterSafeTimeInfoFromMap(
-      const LeaderEpoch& epoch, GetXClusterSafeTimeResponsePB* resp);
+  // Computes the safe time map and fills it in the response.
+  Status ComputeAndGetXClusterSafeTimeInfo(
+      const LeaderEpoch& epoch, GetXClusterSafeTimeResponsePB& resp);
 
   Result<HybridTime> GetXClusterSafeTimeForNamespace(
-      const int64_t leader_term, const NamespaceId& namespace_id,
-      const XClusterSafeTimeFilter& filter);
+      const NamespaceId& namespace_id, const XClusterSafeTimeFilter& filter) const;
 
   xcluster::XClusterConsumerClusterMetrics* TEST_GetMetricsForNamespace(
       const NamespaceId& namespace_id);
@@ -106,7 +106,7 @@ class XClusterSafeTimeService {
 
   virtual Result<bool> CreateTableRequired() REQUIRES(mutex_);
 
-  virtual XClusterNamespaceToSafeTimeMap GetXClusterNamespaceToSafeTimeMap();
+  virtual XClusterNamespaceToSafeTimeMap GetXClusterNamespaceToSafeTimeMap() const;
 
   virtual Status SetXClusterSafeTime(
       const int64_t leader_term, const XClusterNamespaceToSafeTimeMap& new_safe_time_map);
@@ -125,15 +125,15 @@ class XClusterSafeTimeService {
   void EnterIdleMode(const std::string& reason);
 
   Result<XClusterNamespaceToSafeTimeMap> GetFilteredXClusterSafeTimeMap(
-      const XClusterSafeTimeFilter& filter) REQUIRES_SHARED(mutex_);
+      const XClusterSafeTimeFilter& filter) const REQUIRES_SHARED(mutex_);
 
   Master* const master_;
   CatalogManager* const catalog_manager_;
 
   rpc::Poller poller_;
   std::optional<boost::asio::io_context::strand> poll_strand_;
 
-  std::shared_mutex mutex_;
+  mutable std::shared_mutex mutex_;
   bool safe_time_table_ready_ GUARDED_BY(mutex_) = false;
 
   std::unique_ptr<client::TableHandle> safe_time_table_;

src/yb/master/xcluster/xcluster_target_manager.cc

@@ -155,7 +155,7 @@ void XClusterTargetManager::CreateXClusterSafeTimeTableAndStartService() {
 Status XClusterTargetManager::GetXClusterSafeTime(
     GetXClusterSafeTimeResponsePB* resp, const LeaderEpoch& epoch) {
   RETURN_NOT_OK_SET_CODE(
-      safe_time_service_->GetXClusterSafeTimeInfoFromMap(epoch, resp),
+      safe_time_service_->ComputeAndGetXClusterSafeTimeInfo(epoch, *resp),
       MasterError(MasterErrorPB::INTERNAL_ERROR));
 
   // Also fill out the namespace_name for each entry.
@@ -199,16 +199,14 @@ Status XClusterTargetManager::GetXClusterSafeTimeForNamespace(
 
   RETURN_NOT_OK(safe_time_service_->ComputeSafeTime(epoch.leader_term));
   auto safe_time_ht =
-      VERIFY_RESULT(GetXClusterSafeTimeForNamespace(epoch, req->namespace_id(), req->filter()));
+      VERIFY_RESULT(GetXClusterSafeTimeForNamespace(req->namespace_id(), req->filter()));
   resp->set_safe_time_ht(safe_time_ht.ToUint64());
   return Status::OK();
 }
 
 Result<HybridTime> XClusterTargetManager::GetXClusterSafeTimeForNamespace(
-    const LeaderEpoch& epoch, const NamespaceId& namespace_id,
-    const XClusterSafeTimeFilter& filter) {
-  return safe_time_service_->GetXClusterSafeTimeForNamespace(
-      epoch.leader_term, namespace_id, filter);
+    const NamespaceId& namespace_id, const XClusterSafeTimeFilter& filter) const {
+  return safe_time_service_->GetXClusterSafeTimeForNamespace(namespace_id, filter);
 }
 
 Status XClusterTargetManager::RefreshXClusterSafeTimeMap(const LeaderEpoch& epoch) {
@@ -429,9 +427,11 @@ XClusterTargetManager::GetXClusterStatus() const {
   const auto cluster_config_pb = VERIFY_RESULT(catalog_manager_.GetClusterConfig());
   const auto replication_infos = catalog_manager_.GetAllXClusterUniverseReplicationInfos();
 
+  const auto namespace_safe_time = GetXClusterNamespaceToSafeTimeMap();
+
   for (const auto& replication_info : replication_infos) {
-    auto replication_group_status =
-        VERIFY_RESULT(GetUniverseReplicationInfo(replication_info, cluster_config_pb));
+    auto replication_group_status = VERIFY_RESULT(
+        GetUniverseReplicationInfo(replication_info, cluster_config_pb, namespace_safe_time));
 
     for (auto& [_, tables] : replication_group_status.table_statuses_by_namespace) {
       for (auto& table : tables) {
@@ -454,7 +454,8 @@ XClusterTargetManager::GetXClusterStatus() const {
 
 Result<XClusterInboundReplicationGroupStatus> XClusterTargetManager::GetUniverseReplicationInfo(
     const SysUniverseReplicationEntryPB& replication_info_pb,
-    const SysClusterConfigEntryPB& cluster_config_pb) const {
+    const SysClusterConfigEntryPB& cluster_config_pb,
+    const XClusterNamespaceToSafeTimeMap& namespace_safe_time) const {
   XClusterInboundReplicationGroupStatus result;
   result.replication_group_id =
       xcluster::ReplicationGroupId(replication_info_pb.replication_group_id());
@@ -475,10 +476,18 @@ Result<XClusterInboundReplicationGroupStatus> XClusterTargetManager::GetUniverse
     for (const auto& namespace_info : replication_info_pb.db_scoped_info().namespace_infos()) {
       result.db_scope_namespace_id_map[namespace_info.consumer_namespace_id()] =
           namespace_info.producer_namespace_id();
+
       result.db_scoped_info += Format(
           "\n  namespace: $0\n    consumer_namespace_id: $1\n    producer_namespace_id: $2",
           catalog_manager_.GetNamespaceName(namespace_info.consumer_namespace_id()),
           namespace_info.consumer_namespace_id(), namespace_info.producer_namespace_id());
+
+      auto safe_time_info = FindOrNull(namespace_safe_time, namespace_info.consumer_namespace_id());
+      if (safe_time_info) {
+        result.db_scoped_info += Format(
+            "\n    safe_time: $0",
+            Timestamp(safe_time_info->GetPhysicalValueMicros()).ToHumanReadableTime());
+      }
     }
   }
 
@@ -566,6 +575,23 @@ Status XClusterTargetManager::PopulateXClusterStatusJson(JsonWriter& jw) const {
   }
   jw.EndArray();
 
+  const auto safe_time_map = GetXClusterNamespaceToSafeTimeMap();
+  if (!safe_time_map.empty()) {
+    jw.String("safe_time_map");
+    jw.StartArray();
+    for (const auto& [namespace_id, safe_time] : safe_time_map) {
+      jw.StartObject();
+      jw.String("consumer_namespace_id");
+      jw.String(namespace_id);
+
+      jw.String("safe_time");
+      jw.String(safe_time.ToString());
+
+      jw.EndObject();
+    }
+    jw.EndArray();
+  }
+
   jw.String("consumer_registry");
   jw.Protobuf(cluster_config.consumer_registry());
 
@@ -602,10 +628,12 @@ Result<XClusterInboundReplicationGroupStatus> XClusterTargetManager::GetUniverse
   auto replication_info = catalog_manager_.GetUniverseReplication(replication_group_id);
   SCHECK_FORMAT(replication_info, NotFound, "Replication group $0 not found", replication_group_id);
 
+  static const XClusterNamespaceToSafeTimeMap empty_namespace_safe_time;
+
   const auto cluster_config = VERIFY_RESULT(catalog_manager_.GetClusterConfig());
   // Make pb copy to avoid potential deadlock while calling GetUniverseReplicationInfo.
   auto pb = replication_info->LockForRead()->pb;
-  return GetUniverseReplicationInfo(pb, cluster_config);
+  return GetUniverseReplicationInfo(pb, cluster_config, empty_namespace_safe_time);
 }
 
 Status XClusterTargetManager::ClearXClusterFieldsAfterYsqlDDL(

src/yb/master/xcluster/xcluster_target_manager.h

@@ -28,11 +28,11 @@ class TSHeartbeatRequestPB;
 class TSHeartbeatResponsePB;
 
 class PostTabletCreateTaskBase;
-class XClusterInboundReplicationGroupSetupTaskIf;
 class UniverseReplicationInfo;
+class XClusterConsumerReplicationStatusPB;
+class XClusterInboundReplicationGroupSetupTaskIf;
 class XClusterSafeTimeService;
 struct XClusterInboundReplicationGroupStatus;
-class XClusterConsumerReplicationStatusPB;
 struct XClusterSetupUniverseReplicationData;
 
 class XClusterTargetManager {
@@ -54,8 +54,7 @@ class XClusterTargetManager {
       GetXClusterSafeTimeForNamespaceResponsePB* resp, const LeaderEpoch& epoch);
 
   Result<HybridTime> GetXClusterSafeTimeForNamespace(
-      const LeaderEpoch& epoch, const NamespaceId& namespace_id,
-      const XClusterSafeTimeFilter& filter);
+      const NamespaceId& namespace_id, const XClusterSafeTimeFilter& filter) const;
 
   Status RefreshXClusterSafeTimeMap(const LeaderEpoch& epoch);
 
@@ -230,12 +229,12 @@ class XClusterTargetManager {
   // table statuses.
   Result<XClusterInboundReplicationGroupStatus> GetUniverseReplicationInfo(
       const SysUniverseReplicationEntryPB& replication_info_pb,
-      const SysClusterConfigEntryPB& cluster_config_pb) const;
+      const SysClusterConfigEntryPB& cluster_config_pb,
+      const XClusterNamespaceToSafeTimeMap& namespace_safe_time) const;
 
   Status RefreshLocalAutoFlagConfig(const LeaderEpoch& epoch);
   Status DoRefreshLocalAutoFlagConfig(const LeaderEpoch& epoch);
 
-
   // Populate the response with the errors for the given replication group.
   Status PopulateReplicationGroupErrors(
       const xcluster::ReplicationGroupId& replication_group_id,