Feat[STATS]: Partition replication time (#909)

waldgange · web-flow · commit 3ec7eb495eaf · 2025-10-30T18:26:56.000+02:00
Signed-off-by: Anton Pryakhin &lt;apryakhin1@bloomberg.net&gt;
diff --git a/src/groups/mqb/mqbblp/mqbblp_localqueue.cpp b/src/groups/mqb/mqbblp/mqbblp_localqueue.cpp
@@ -484,24 +484,16 @@ void LocalQueue::postMessage(const bmqp::PutHeader&              putHeader,
 
     // Send acknowledgement if post failed or if ack was requested (both could
     // be true as well).
-    if (res != mqbi::StorageResult::e_SUCCESS || haveReceipt) {
-        // Calculate time delta between PUT and ACK
-        const bsls::Types::Int64 timeDelta =
-            bmqsys::Time::highResolutionTimer() - timePoint;
-        d_state_p->stats()
-            ->onEvent<mqbstat::QueueStatsDomain::EventType::e_ACK_TIME>(
-                timeDelta);
-        if (res != mqbi::StorageResult::e_SUCCESS || doAck) {
-            bmqp::AckMessage ackMessage;
-            ackMessage
-                .setStatus(bmqp::ProtocolUtil::ackResultToCode(
-                    mqbi::StorageResult::toAckResult(res)))
-                .setMessageGUID(putHeader.messageGUID());
-            // CorrelationId & QueueId are left unset as those fields will
-            // be filled downstream.
-
-            source->onAckMessage(ackMessage);
-        }
+    if (res != mqbi::StorageResult::e_SUCCESS || (haveReceipt && doAck)) {
+        bmqp::AckMessage ackMessage;
+        ackMessage
+            .setStatus(bmqp::ProtocolUtil::ackResultToCode(
+                mqbi::StorageResult::toAckResult(res)))
+            .setMessageGUID(putHeader.messageGUID());
+        // CorrelationId & QueueId are left unset as those fields will
+        // be filled downstream.
+
+        source->onAckMessage(ackMessage);
     }
 
     if (BSLS_PERFORMANCEHINT_PREDICT_LIKELY(res ==
@@ -547,17 +539,9 @@ void LocalQueue::onPushMessage(
                     "onPushMessage should not be called on LocalQueue");
 }
 
-void LocalQueue::onReceipt(const bmqt::MessageGUID&  msgGUID,
-                           mqbi::QueueHandle*        qH,
-                           const bsls::Types::Int64& arrivalTimepoint)
+void LocalQueue::onReceipt(const bmqt::MessageGUID& msgGUID,
+                           mqbi::QueueHandle*       qH)
 {
-    // Calculate time delta between PUT and ACK
-    const bsls::Types::Int64 timeDelta = bmqsys::Time::highResolutionTimer() -
-                                         arrivalTimepoint;
-
-    d_state_p->stats()
-        ->onEvent<mqbstat::QueueStatsDomain::EventType::e_ACK_TIME>(timeDelta);
-
     if (d_state_p->handleCatalog().hasHandle(qH)) {
         // Send acknowledgement
         bmqp::AckMessage ackMessage;
diff --git a/src/groups/mqb/mqbblp/mqbblp_localqueue.h b/src/groups/mqb/mqbblp/mqbblp_localqueue.h
@@ -180,13 +180,10 @@ class LocalQueue BSLS_CPP11_FINAL {
 
     /// Invoked by the Data Store when it receives quorum Receipts for the
     /// specified `msgGUID`.  Send ACK to the specified `qH` if it is
-    /// present in the queue handle catalog.  Update ACK time stats using
-    /// the specified `arrivalTimepoint`.
+    /// present in the queue handle catalog.
     ///
     /// THREAD: This method is called from the Storage dispatcher thread.
-    void onReceipt(const bmqt::MessageGUID&  msgGUID,
-                   mqbi::QueueHandle*        qH,
-                   const bsls::Types::Int64& arrivalTimepoint);
+    void onReceipt(const bmqt::MessageGUID& msgGUID, mqbi::QueueHandle* qH);
 
     /// Invoked by the Data Store when it removes (times out waiting for
     /// quorum Receipts for) a message with the specified `msgGUID`.  Send
diff --git a/src/groups/mqb/mqbblp/mqbblp_queue.cpp b/src/groups/mqb/mqbblp/mqbblp_queue.cpp
@@ -593,13 +593,12 @@ void Queue::onOpenUpstream(bsls::Types::Uint64 genCount,
     }
 }
 
-void Queue::onReceipt(const bmqt::MessageGUID&  msgGUID,
-                      mqbi::QueueHandle*        queueHandle,
-                      const bsls::Types::Int64& arrivalTimepoint)
+void Queue::onReceipt(const bmqt::MessageGUID& msgGUID,
+                      mqbi::QueueHandle*       queueHandle)
 {
     BSLS_ASSERT_SAFE(d_localQueue_mp);
 
-    d_localQueue_mp->onReceipt(msgGUID, queueHandle, arrivalTimepoint);
+    d_localQueue_mp->onReceipt(msgGUID, queueHandle);
 }
 
 void Queue::onRemoval(const bmqt::MessageGUID& msgGUID,
diff --git a/src/groups/mqb/mqbblp/mqbblp_queue.h b/src/groups/mqb/mqbblp/mqbblp_queue.h
@@ -328,14 +328,11 @@ class Queue BSLS_CPP11_FINAL : public mqbi::Queue {
 
     /// Invoked by the Data Store when it receives quorum Receipts for the
     /// specified `msgGUID`.  Send ACK to the specified `queueHandle` if it
-    /// is present in the queue handle catalog.  Update AVK time stats using
-    /// the specified `arrivalTimepoint`.
+    /// is present in the queue handle catalog.
     ///
     /// THREAD: This method is called from the Storage dispatcher thread.
-    void onReceipt(const bmqt::MessageGUID&  msgGUID,
-                   mqbi::QueueHandle*        queueHandle,
-                   const bsls::Types::Int64& arrivalTimepoint)
-        BSLS_KEYWORD_OVERRIDE;
+    void onReceipt(const bmqt::MessageGUID& msgGUID,
+                   mqbi::QueueHandle*       queueHandle) BSLS_KEYWORD_OVERRIDE;
 
     /// Invoked by the Data Store when it removes (times out waiting for
     /// quorum Receipts for) a message with the specified `msgGUID`.  Send
diff --git a/src/groups/mqb/mqbi/mqbi_cluster.h b/src/groups/mqb/mqbi/mqbi_cluster.h
@@ -288,8 +288,8 @@ class Cluster : public DispatcherClient {
     /// used by this cluster.
     virtual RequestManagerType& requestManager() = 0;
 
-    // Return a reference offering a modifiable access to the multi request
-    // manager used by this cluster.
+    /// Return a reference offering a modifiable access to the multi request
+    /// manager used by this cluster.
     virtual MultiRequestManagerType& multiRequestManager() = 0;
 
     /// Send the specified `request` with the specified `timeout` to the
diff --git a/src/groups/mqb/mqbi/mqbi_queue.h b/src/groups/mqb/mqbi/mqbi_queue.h
@@ -864,13 +864,11 @@ class Queue : public DispatcherClient {
 
     /// Invoked by the Data Store when it receives quorum Receipts for the
     /// specified `msgGUID`.  Send ACK to the specified `queueHandle` if it
-    /// is present in the queue handle catalog.  Update AVK time stats using
-    /// the specified `arrivalTimepoint`.
+    /// is present in the queue handle catalog.
     ///
     /// THREAD: This method is called from the Queue's dispatcher thread.
-    virtual void onReceipt(const bmqt::MessageGUID&  msgGUID,
-                           mqbi::QueueHandle*        queueHandle,
-                           const bsls::Types::Int64& arrivalTimepoint) = 0;
+    virtual void onReceipt(const bmqt::MessageGUID& msgGUID,
+                           mqbi::QueueHandle*       queueHandle) = 0;
 
     /// Invoked by the Data Store when it removes (times out waiting for
     /// quorum Receipts for) a message with the specified `msgGUID`.  Send
diff --git a/src/groups/mqb/mqbmock/mqbmock_queue.cpp b/src/groups/mqb/mqbmock/mqbmock_queue.cpp
@@ -363,8 +363,7 @@ void Queue::onOpenFailure(BSLA_UNUSED unsigned int subQueueId)
 }
 
 void Queue::onReceipt(BSLA_UNUSED const bmqt::MessageGUID& msgGUID,
-                      BSLA_UNUSED mqbi::QueueHandle* qH,
-                      BSLA_UNUSED const bsls::Types::Int64& arrivalTimepoint)
+                      BSLA_UNUSED mqbi::QueueHandle* qH)
 {
     // NOTHING
 }
diff --git a/src/groups/mqb/mqbmock/mqbmock_queue.h b/src/groups/mqb/mqbmock/mqbmock_queue.h
@@ -319,10 +319,8 @@ class Queue : public mqbi::Queue {
     /// Invoked by the Data Store when it receives quorum Receipts.
     ///
     /// THREAD: This method is called from the Queue's dispatcher thread.
-    void onReceipt(const bmqt::MessageGUID&  msgGUID,
-                   mqbi::QueueHandle*        qH,
-                   const bsls::Types::Int64& arrivalTimepoint)
-        BSLS_KEYWORD_OVERRIDE;
+    void onReceipt(const bmqt::MessageGUID& msgGUID,
+                   mqbi::QueueHandle*       qH) BSLS_KEYWORD_OVERRIDE;
 
     /// Invoked by the Data Store when it removes (times out waiting for
     /// quorum Receipts for) a message with the specified `msgGUID`.  Send
diff --git a/src/groups/mqb/mqbs/mqbs_filebackedstorage.cpp b/src/groups/mqb/mqbs/mqbs_filebackedstorage.cpp
@@ -374,7 +374,6 @@ FileBackedStorage::put(mqbi::StorageMessageAttributes*     attributes,
     queue()
         ->stats()
         ->onEvent<mqbstat::QueueStatsDomain::EventType::e_ADD_MESSAGE>(
-
             msgSize);
 
     d_isEmpty.storeRelaxed(0);
diff --git a/src/groups/mqb/mqbs/mqbs_filestore.cpp b/src/groups/mqb/mqbs/mqbs_filestore.cpp
@@ -4078,8 +4078,19 @@ void FileStore::processReceiptEvent(unsigned int         primaryLeaseId,
         }
         if (++(from->second.d_count) >= d_replicationFactor) {
             from->second.d_handle->second.d_hasReceipt = true;
-            // notify the queue
 
+            // Calculate time it took for the message to be stored and
+            // replicated.
+            const bsls::Types::Int64 timeDelta =
+                bmqsys::Time::highResolutionTimer() -
+                from->second.d_handle->second.d_arrivalTimepoint;
+            d_clusterStats_p->onPartitionEvent(
+                mqbstat::ClusterStats::PartitionEventType::
+                    e_PARTITION_REPLICATION,
+                d_config.partitionId(),
+                timeDelta);
+
+            // notify the queue
             const mqbu::StorageKey& queueKey  = from->second.d_queueKey;
             bool                    haveQueue = (queueKey == lastKey);
             if (!haveQueue) {
@@ -4089,16 +4100,12 @@ void FileStore::processReceiptEvent(unsigned int         primaryLeaseId,
                     lastKey   = queueKey;
                     lastQueue = sit->second->queue();
                     BSLS_ASSERT_SAFE(lastQueue);
-
                     affectedQueues.insert(lastQueue);
                 }
                 // else the queue and its storage are gone; ignore the receipt
             }
             if (haveQueue) {
-                lastQueue->onReceipt(
-                    from->second.d_guid,
-                    from->second.d_qH,
-                    from->second.d_handle->second.d_arrivalTimepoint);
+                lastQueue->onReceipt(from->second.d_guid, from->second.d_qH);
             }  // else the queue is gone
             from = d_unreceipted.erase(from);
         }
@@ -7276,8 +7283,19 @@ void FileStore::setReplicationFactor(int value)
     while (it != d_unreceipted.end()) {
         if (it->second.d_count >= d_replicationFactor) {
             it->second.d_handle->second.d_hasReceipt = true;
-            // notify the queue.
 
+            // Calculate time it took for the message to be stored and
+            // replicated.
+            const bsls::Types::Int64 timeDelta =
+                bmqsys::Time::highResolutionTimer() -
+                it->second.d_handle->second.d_arrivalTimepoint;
+            d_clusterStats_p->onPartitionEvent(
+                mqbstat::ClusterStats::PartitionEventType::
+                    e_PARTITION_REPLICATION,
+                d_config.partitionId(),
+                timeDelta);
+
+            // notify the queue.
             const mqbu::StorageKey& queueKey  = it->second.d_queueKey;
             bool                    haveQueue = (queueKey == lastKey);
             if (!haveQueue) {
@@ -7287,16 +7305,12 @@ void FileStore::setReplicationFactor(int value)
                     lastKey   = queueKey;
                     lastQueue = sit->second->queue();
                     BSLS_ASSERT_SAFE(lastQueue);
-
                     affectedQueues.insert(lastQueue);
                 }
                 // else the queue and its storage are gone; ignore the receipt
             }
             if (haveQueue) {
-                lastQueue->onReceipt(
-                    it->second.d_guid,
-                    it->second.d_qH,
-                    it->second.d_handle->second.d_arrivalTimepoint);
+                lastQueue->onReceipt(it->second.d_guid, it->second.d_qH);
             }  // else the queue is gone
             it = d_unreceipted.erase(it);
         }
diff --git a/src/groups/mqb/mqbstat/mqbstat_clusterstats.cpp b/src/groups/mqb/mqbstat/mqbstat_clusterstats.cpp
@@ -109,6 +109,10 @@ struct ClusterStatsIndex {
         ,
         e_PARTITION_JOURNAL_OFFSET_BYTES
         // Value: Offset bytes in the journal file of the partition.
+        ,
+        e_PARTITION_REPLICATION_TIME_NS
+        // Value: Time in nanoseconds it took for replication of a new entry
+        //        in journal file.
     };
 };
 
@@ -256,6 +260,18 @@ bsls::Types::Int64 ClusterStats::getValue(const bmqst::StatContext& context,
     case Stat::e_PARTITION_SEQUENCE_NUMBER: {
         return STAT_SINGLE(value, e_PARTITION_SEQUENCE_NUMBER);
     }
+    case Stat::e_PARTITION_REPLICATION_TIME_NS_AVG: {
+        const bsls::Types::Int64 value =
+            STAT_RANGE(averagePerEvent, e_PARTITION_REPLICATION_TIME_NS);
+        return value == bsl::numeric_limits<bsls::Types::Int64>::max() ? 0
+                                                                       : value;
+    }
+    case Stat::e_PARTITION_REPLICATION_TIME_NS_MAX: {
+        const bsls::Types::Int64 value =
+            STAT_RANGE(rangeMax, e_PARTITION_REPLICATION_TIME_NS);
+        return value == bsl::numeric_limits<bsls::Types::Int64>::min() ? 0
+                                                                       : value;
+    }
 
     default: {
         BSLS_ASSERT_SAFE(false && "Attempting to access an unknown stat");
@@ -329,6 +345,10 @@ void ClusterStats::onPartitionEvent(PartitionEventType::Enum type,
     case PartitionEventType::e_PARTITION_ROLLOVER: {
         sc->reportValue(ClusterStatsIndex::e_PARTITION_ROLLOVER_TIME, value);
     } break;
+    case PartitionEventType::e_PARTITION_REPLICATION: {
+        sc->reportValue(ClusterStatsIndex::e_PARTITION_REPLICATION_TIME_NS,
+                        value);
+    } break;
     default: {
         BSLS_ASSERT_SAFE(false && "Unknown event type");
     } break;
@@ -644,7 +664,8 @@ ClusterStatsUtil::initializeStatContextCluster(int               historySize,
         .value("partition.journal_bytes", bmqst::StatValue::e_DISCRETE)
         .value("partition.data_offset_bytes")
         .value("partition.journal_offset_bytes")
-        .value("partition.sequence_number");
+        .value("partition.sequence_number")
+        .value("partition.replication_time_ns", bmqst::StatValue::e_DISCRETE);
 
     // NOTE: For the clusters, the stat context will have two levels of
     //       children, first level is per cluster, and second level is per
diff --git a/src/groups/mqb/mqbstat/mqbstat_clusterstats.h b/src/groups/mqb/mqbstat/mqbstat_clusterstats.h
@@ -67,7 +67,10 @@ class ClusterStats {
         // TYPES
         enum Enum {
             /// Time in nanoseconds it took for the rollover operation.
-            e_PARTITION_ROLLOVER
+            e_PARTITION_ROLLOVER,
+            /// Time in nanoseconds it took for the replication of a new entry
+            /// in journal file.
+            e_PARTITION_REPLICATION
         };
     };
 
@@ -129,7 +132,15 @@ class ClusterStats {
             /// partition.
             e_PARTITION_JOURNAL_UTILIZATION_MAX,
             /// Currently observed sequence number of the partition.
-            e_PARTITION_SEQUENCE_NUMBER
+            e_PARTITION_SEQUENCE_NUMBER,
+            /// Average observed time in nanoseconds it took to store a message
+            /// record at primary and replicate it to a majority of nodes in
+            /// the cluster.
+            e_PARTITION_REPLICATION_TIME_NS_AVG,
+            /// Maximum observed time in nanoseconds it took to store a message
+            /// record at primary and replicate it to a majority of nodes in
+            /// the cluster.
+            e_PARTITION_REPLICATION_TIME_NS_MAX
         };
     };
 
diff --git a/src/integration-tests/test_admin_client.py b/src/integration-tests/test_admin_client.py
@@ -20,6 +20,7 @@
 
 import dataclasses
 import json
+import time
 from typing import Dict, Optional, Union
 
 import blazingmq.dev.it.testconstants as tc
@@ -375,6 +376,9 @@ def test_app_id_stats(multi_node: Cluster, domain_urls: tc.DomainUrls) -> None:
 
     _ = _change_leader(cluster)
 
+    # wait for at least one snapshot to initialize stats
+    time.sleep(1)
+
     admin.connect(*cluster.admin_endpoint)
 
     stats = extract_stats(admin.send_admin("encoding json_pretty stat show"))
diff --git a/src/plugins/bmqprometheus/bmqprometheus_prometheusstatconsumer.cpp b/src/plugins/bmqprometheus/bmqprometheus_prometheusstatconsumer.cpp

Original file line number	Diff line number	Diff line change
`@@ -593,13 +593,12 @@ void Queue::onOpenUpstream(bsls::Types::Uint64 genCount,`
`593`	`593`	`}`
`594`	`594`	`}`
`595`	`595`
`596`		`-void Queue::onReceipt(const bmqt::MessageGUID& msgGUID,`
`597`		`- mqbi::QueueHandle* queueHandle,`
`598`		`- const bsls::Types::Int64& arrivalTimepoint)`
	`596`	`+void Queue::onReceipt(const bmqt::MessageGUID& msgGUID,`
	`597`	`+ mqbi::QueueHandle* queueHandle)`
`599`	`598`	`{`
`600`	`599`	`BSLS_ASSERT_SAFE(d_localQueue_mp);`
`601`	`600`
`602`		`- d_localQueue_mp->onReceipt(msgGUID, queueHandle, arrivalTimepoint);`
	`601`	`+ d_localQueue_mp->onReceipt(msgGUID, queueHandle);`
`603`	`602`	`}`
`604`	`603`
`605`	`604`	`void Queue::onRemoval(const bmqt::MessageGUID& msgGUID,`
Original file line number	Diff line number	Diff line change
`@@ -363,8 +363,7 @@ void Queue::onOpenFailure(BSLA_UNUSED unsigned int subQueueId)`
`363`	`363`	`}`
`364`	`364`
`365`	`365`	`void Queue::onReceipt(BSLA_UNUSED const bmqt::MessageGUID& msgGUID,`
`366`		`- BSLA_UNUSED mqbi::QueueHandle* qH,`
`367`		`- BSLA_UNUSED const bsls::Types::Int64& arrivalTimepoint)`
	`366`	`+ BSLA_UNUSED mqbi::QueueHandle* qH)`
`368`	`367`	`{`
`369`	`368`	`// NOTHING`
`370`	`369`	`}`