Feat[STATS]: Report csl replication time (#901)

Signed-off-by: Anton Pryakhin <[email protected]>

Author: waldgange

src/groups/mqb/mqbc/mqbc_incoreclusterstateledger.cpp

@@ -490,7 +490,10 @@ int IncoreClusterStateLedger::applyRecordInternalImpl(
         ClusterMessageInfo info;
         info.d_clusterMessage = clusterMessage;
         info.d_ackCount       = 0;
-        d_uncommittedAdvisories.insert(bsl::make_pair(sequenceNumber, info));
+        const AdvisoriesMap::iterator advIt =
+            d_uncommittedAdvisories
+                .insert(bsl::make_pair(sequenceNumber, info))
+                .first;
 
         if (isSelfLeader()) {
             // TBD: How/if to handle timeout for this message?
@@ -508,6 +511,10 @@ int IncoreClusterStateLedger::applyRecordInternalImpl(
             // record offset should be 0.
             BSLS_ASSERT_SAFE(recordOffset == 0);
 
+            // Save the replication start time
+            advIt->second.d_timestampNs =
+                bdlt::CurrentTime::now().totalNanoseconds();
+
             bsl::shared_ptr<bdlbb::Blob> advisoryEvent =
                 d_blobSpPool_p->getObject();
 
@@ -622,6 +629,11 @@ int IncoreClusterStateLedger::applyRecordInternalImpl(
         }
 
         if (isSelfLeader()) {
+            bsls::Types::Int64 replicationTimeNs =
+                bdlt::CurrentTime::now().totalNanoseconds() -
+                iter->second.d_timestampNs;
+            d_clusterData_p->stats().setCslReplicationTime(replicationTimeNs);
+
             bsl::shared_ptr<bdlbb::Blob> commitEvent =
                 d_blobSpPool_p->getObject();
 
@@ -644,12 +656,11 @@ int IncoreClusterStateLedger::applyRecordInternalImpl(
         //       the commit callback should be enqueued to be invoked from the
         //       cluster dispatcher thread.
         // TODO: In phase 2 of IncoreCSL, this can return to enqueueing on
-        // the
-        //       cluster dispatcher thread
+        //       the cluster dispatcher thread.
         d_commitCb(committedControlMessage,
                    ClusterStateLedgerCommitStatus::e_SUCCESS);
 
-        d_uncommittedAdvisories.erase(commit.sequenceNumberCommitted());
+        d_uncommittedAdvisories.erase(iter);
     } break;  // BREAK
     case (ClusterStateRecordType::e_ACK): {
         // PRECONDITIONS

src/groups/mqb/mqbc/mqbc_incoreclusterstateledger.h

@@ -90,6 +90,10 @@ struct IncoreClusterStateLedger_ClusterMessageInfo {
     ///   - `QueueUnAssignmentAdvisory`.
     bmqp_ctrlmsg::ClusterMessage d_clusterMessage;
 
+    /// Timestamp in nanoseconds at the moment when the replication of this
+    /// `ClusterMessage` started.
+    bsls::Types::Uint64 d_timestampNs;
+
     /// Number of ACKs received for this `ClusterMessage`.
     int d_ackCount;
 
@@ -418,6 +422,7 @@ class IncoreClusterStateLedger BSLS_KEYWORD_FINAL : public ClusterStateLedger {
 inline IncoreClusterStateLedger_ClusterMessageInfo ::
     IncoreClusterStateLedger_ClusterMessageInfo(bslma::Allocator* allocator)
 : d_clusterMessage(bslma::Default::allocator(allocator))
+, d_timestampNs(0)
 , d_ackCount(0)
 {
     // NOTHING
@@ -428,6 +433,7 @@ inline IncoreClusterStateLedger_ClusterMessageInfo ::
         const IncoreClusterStateLedger_ClusterMessageInfo& other,
         bslma::Allocator*                                  allocator)
 : d_clusterMessage(other.d_clusterMessage, allocator)
+, d_timestampNs(other.d_timestampNs)
 , d_ackCount(other.d_ackCount)
 {
     // NOTHING

src/groups/mqb/mqbstat/mqbstat_clusterstats.cpp

@@ -68,6 +68,10 @@ struct ClusterStatsIndex {
         // Value: Leader status of cluster, non-zero (1) implies leader
         //        and 0 implies follower
         ,
+        e_CSL_REPLICATION_TIME_NS
+        // Value: time in nanoseconds it took for replication of a new entry
+        //        in CSL file.
+        ,
         e_PARTITION_CFG_DATA_BYTES
         // Value: Configured size of partitions' data file
         ,
@@ -150,6 +154,18 @@ bsls::Types::Int64 ClusterStats::getValue(const bmqst::StatContext& context,
 
         return STAT_RANGE(rangeMax, e_LEADER_STATUS);
     }
+    case Stat::e_CSL_REPLICATION_TIME_NS_AVG: {
+        const bsls::Types::Int64 value = STAT_RANGE(averagePerEvent,
+                                                    e_CSL_REPLICATION_TIME_NS);
+        return value == bsl::numeric_limits<bsls::Types::Int64>::max() ? 0
+                                                                       : value;
+    }
+    case Stat::e_CSL_REPLICATION_TIME_NS_MAX: {
+        const bsls::Types::Int64 value = STAT_RANGE(rangeMax,
+                                                    e_CSL_REPLICATION_TIME_NS);
+        return value == bsl::numeric_limits<bsls::Types::Int64>::min() ? 0
+                                                                       : value;
+    }
     case Stat::e_PARTITION_CFG_JOURNAL_BYTES: {
         return STAT_SINGLE(value, e_PARTITION_CFG_JOURNAL_BYTES);
     }
@@ -335,6 +351,13 @@ ClusterStats& ClusterStats::setIsLeader(LeaderStatus::Enum value)
     return *this;
 }
 
+ClusterStats& ClusterStats::setCslReplicationTime(bsls::Types::Int64 value)
+{
+    d_statContext_mp->reportValue(ClusterStatsIndex::e_CSL_REPLICATION_TIME_NS,
+                                  value);
+    return *this;
+}
+
 ClusterStats&
 ClusterStats::setPartitionCfgBytes(bsls::Types::Int64 dataBytes,
                                    bsls::Types::Int64 journalBytes)
@@ -542,6 +565,7 @@ ClusterStatsUtil::initializeStatContextCluster(int               historySize,
         .storeExpiredSubcontextValues(true)
         .value("cluster_status")
         .value("cluster_leader")
+        .value("cluster_csl_replication_time_ns", bmqst::StatValue::e_DISCRETE)
         .value("cluster.partition.cfg_journal_bytes")
         .value("cluster.partition.cfg_data_bytes")
         .value("partition_status")

src/groups/mqb/mqbstat/mqbstat_clusterstats.h

@@ -81,6 +81,8 @@ class ClusterStats {
             e_CLUSTER_STATUS,
             e_ROLE,
             e_LEADER_STATUS,
+            e_CSL_REPLICATION_TIME_NS_AVG,
+            e_CSL_REPLICATION_TIME_NS_MAX,
             e_PARTITION_CFG_DATA_BYTES,
             e_PARTITION_CFG_JOURNAL_BYTES,
 
@@ -250,6 +252,10 @@ class ClusterStats {
     ClusterStats& setNodeRoleForPartition(int                 partitionId,
                                           PrimaryStatus::Enum value);
 
+    /// Set the csl replication time of the StatContext being referred
+    /// to by this object to be the specified `value`.
+    ClusterStats& setCslReplicationTime(bsls::Types::Int64 value);
+
     /// Set the partition outstanding bytes of the specified data and
     /// journal files for the specified `partitionId` to the corresponding
     /// specified `dataBytes` and `journalBytes` values.

src/plugins/bmqprometheus/bmqprometheus_prometheusstatconsumer.cpp

@@ -680,6 +680,12 @@ void PrometheusStatConsumer::captureClusterStats(const LeaderSet& leaders)
                 {"cluster_partition_cfg_data_bytes",
                  Stat::e_PARTITION_CFG_DATA_BYTES,
                  false},
+                {"cluster_csl_replication_time_ns_avg",
+                 Stat::e_CSL_REPLICATION_TIME_NS_AVG,
+                 false},
+                {"cluster_csl_replication_time_ns_max",
+                 Stat::e_CSL_REPLICATION_TIME_NS_MAX,
+                 false},
             };
 
             Tagger tagger;