Feat[MQB]: collect queue stats on replicas

678098 · 678098 · commit 2cb2a9727675 · 2025-03-14T17:28:12.000Z
Signed-off-by: Evgeny Malygin &lt;emalygin@bloomberg.net&gt;
diff --git a/src/groups/mqb/mqbblp/mqbblp_pushstream.t.cpp b/src/groups/mqb/mqbblp/mqbblp_pushstream.t.cpp
@@ -16,6 +16,9 @@
 // mqbblp_pushstream.t.cpp                                            -*-C++-*-
 #include <mqbblp_pushstream.h>
 
+// MQB
+#include <mqbmock_domain.h>
+
 // BMQ
 #include <bmqp_messageguidgenerator.h>
 
@@ -122,12 +125,14 @@ static void test2_iterations()
         "dummy",
         bmqtst::TestHelperUtil::allocator());
     bmqt::Uri        dummyUri("dummy", bmqtst::TestHelperUtil::allocator());
-    mqbconfm::Domain dummyDomain(bmqtst::TestHelperUtil::allocator());
+    mqbmock::Domain  dummyDomain(0, bmqtst::TestHelperUtil::allocator());
+    mqbconfm::Domain dummyDomainConfig(bmqtst::TestHelperUtil::allocator());
 
     mqbs::InMemoryStorage dummyStorage(dummyUri,
                                        mqbu::StorageKey::k_NULL_KEY,
+                                       &dummyDomain,
                                        mqbs::DataStore::k_INVALID_PARTITION_ID,
-                                       dummyDomain,
+                                       dummyDomainConfig,
                                        &dummyCapacityMeter,
                                        bmqtst::TestHelperUtil::allocator());
 
diff --git a/src/groups/mqb/mqbblp/mqbblp_queueconsumptionmonitor.t.cpp b/src/groups/mqb/mqbblp/mqbblp_queueconsumptionmonitor.t.cpp
@@ -109,6 +109,7 @@ Test::Test()
             bmqtst::TestHelperUtil::allocator())
 , d_storage(d_queue.uri(),
             mqbu::StorageKey::k_NULL_KEY,
+            &d_domain,
             mqbs::DataStore::k_INVALID_PARTITION_ID,
             getDomainConfig(),
             d_domain.capacityMeter(),
diff --git a/src/groups/mqb/mqbblp/mqbblp_queueenginetester.cpp b/src/groups/mqb/mqbblp/mqbblp_queueenginetester.cpp
@@ -527,6 +527,7 @@ void QueueEngineTester::init(const mqbconfm::Domain& domainConfig,
     mqbi::Storage* storage_p = new (*d_allocator_p)
         mqbs::InMemoryStorage(d_mockQueue_sp->uri(),
                               k_NULL_QUEUE_KEY,
+                              d_mockDomain_mp.get(),
                               k_PARTITION_ID,
                               domainConfig,
                               d_mockDomain_mp->capacityMeter(),
diff --git a/src/groups/mqb/mqbblp/mqbblp_remotequeue.cpp b/src/groups/mqb/mqbblp/mqbblp_remotequeue.cpp
@@ -113,6 +113,7 @@ int RemoteQueue::configureAsProxy(bsl::ostream& errorDescription,
     storageSp.load(new (*d_allocator_p) mqbs::InMemoryStorage(
                        d_state_p->uri(),
                        d_state_p->key(),
+                       d_state_p->domain(),
                        mqbs::DataStore::k_INVALID_PARTITION_ID,
                        domainCfg,
                        d_state_p->domain()->capacityMeter(),
diff --git a/src/groups/mqb/mqbblp/mqbblp_routers.t.cpp b/src/groups/mqb/mqbblp/mqbblp_routers.t.cpp
@@ -26,6 +26,7 @@
 
 // MQB
 #include <mqbcfg_brokerconfig.h>
+#include <mqbmock_domain.h>
 #include <mqbmock_queue.h>
 #include <mqbmock_queuehandle.h>
 #include <mqbs_inmemorystorage.h>
@@ -54,6 +55,7 @@ struct TestStorage {
     mqbconfm::Domain                         d_domainCfg;
     mqbu::CapacityMeter                      d_capacityMeter;
     mqbu::StorageKey                         d_storageKey;
+    mqbmock::Domain                          d_domain;
     mqbs::InMemoryStorage                    d_storage;
     bslma::ManagedPtr<mqbi::StorageIterator> d_iterator;
     bdlbb::PooledBlobBufferFactory           d_bufferFactory;
@@ -67,8 +69,10 @@ struct TestStorage {
     , d_domainCfg(allocator)
     , d_capacityMeter("cm", allocator)
     , d_storageKey(d_subQueueId)
+    , d_domain(0, allocator)  // Use domain only to hold mqbstat::StatContext
     , d_storage(bmqt::Uri("uri", allocator),
                 d_storageKey,
+                &d_domain,
                 1,
                 d_domainCfg,
                 &d_capacityMeter,
diff --git a/src/groups/mqb/mqbs/mqbs_filebackedstorage.cpp b/src/groups/mqb/mqbs/mqbs_filebackedstorage.cpp
@@ -112,8 +112,10 @@ FileBackedStorage::FileBackedStorage(
 , d_queueKey(queueKey)
 , d_config()
 , d_queueUri(queueUri, allocator)
+, d_queueStats_sp(bsl::allocate_shared<mqbstat::QueueStatsDomain>(allocator))
 , d_virtualStorageCatalog(
       this,
+      d_queueStats_sp,
       allocatorStore ? allocatorStore->get("VirtualHandles") : d_allocator_p)
 , d_ttlSeconds(domain->config().messageTtl())
 , d_capacityMeter(
@@ -133,7 +135,6 @@ FileBackedStorage::FileBackedStorage(
 , d_hasReceipts(!domain->config().consistency().isStrongValue())
 , d_currentlyAutoConfirming()
 , d_autoConfirms(d_allocator_p)
-, d_queueStats_sp()
 {
     BSLS_ASSERT(d_store_p);
 
@@ -147,11 +148,9 @@ FileBackedStorage::FileBackedStorage(
     // and domain instance will return a zero capacity meter when queries to be
     // passed to the 'FileBackedStorage' instance.
 
+    d_queueStats_sp->initialize(queueUri, domain);
     d_virtualStorageCatalog.setDefaultRda(
         domain->config().maxDeliveryAttempts());
-
-    d_queueStats_sp.createInplace(d_allocator_p, d_allocator_p);
-    d_queueStats_sp->initialize(queueUri, domain);
 }
 
 FileBackedStorage::~FileBackedStorage()
diff --git a/src/groups/mqb/mqbs/mqbs_filebackedstorage.h b/src/groups/mqb/mqbs/mqbs_filebackedstorage.h
@@ -169,6 +169,9 @@ class FileBackedStorage BSLS_KEYWORD_FINAL : public ReplicatedStorage {
 
     bmqt::Uri d_queueUri;
 
+    /// Statistics of the queue associated to this storage.
+    bsl::shared_ptr<mqbstat::QueueStatsDomain> d_queueStats_sp;
+
     VirtualStorageCatalog d_virtualStorageCatalog;
 
     bsls::Types::Int64 d_ttlSeconds;
@@ -210,9 +213,6 @@ class FileBackedStorage BSLS_KEYWORD_FINAL : public ReplicatedStorage {
     AutoConfirms d_autoConfirms;
     // Auto CONFIRMs waiting for 'put' or 'processMessageRecord'
 
-    bsl::shared_ptr<mqbstat::QueueStatsDomain> d_queueStats_sp;
-    // Statistics of the queue associated to this storage.
-
   private:
     // NOT IMPLEMENTED
     FileBackedStorage(const FileBackedStorage&) BSLS_KEYWORD_DELETED;
diff --git a/src/groups/mqb/mqbs/mqbs_filestore.cpp b/src/groups/mqb/mqbs/mqbs_filestore.cpp
@@ -5457,6 +5457,7 @@ void FileStore::createStorage(bsl::shared_ptr<ReplicatedStorage>* storageSp,
         storageSp->reset(new (*storageAlloc)
                              InMemoryStorage(queueUri,
                                              queueKey,
+                                             domain,
                                              config().partitionId(),
                                              domain->config(),
                                              domain->capacityMeter(),
diff --git a/src/groups/mqb/mqbs/mqbs_inmemorystorage.cpp b/src/groups/mqb/mqbs/mqbs_inmemorystorage.cpp
@@ -50,6 +50,7 @@ const int k_GC_MESSAGES_BATCH_SIZE = 1000;  // how many to process in one run
 // CREATORS
 InMemoryStorage::InMemoryStorage(const bmqt::Uri&        uri,
                                  const mqbu::StorageKey& queueKey,
+                                 mqbi::Domain*           domain,
                                  int                     partitionId,
                                  const mqbconfm::Domain& config,
                                  mqbu::CapacityMeter*    parentCapacityMeter,
@@ -72,8 +73,11 @@ InMemoryStorage::InMemoryStorage(const bmqt::Uri&        uri,
               .addMilliseconds(config.deduplicationTimeMs())
               .totalNanoseconds(),
           allocatorStore ? allocatorStore->get("Handles") : d_allocator_p)
+, d_queueStats_sp(
+      bsl::allocate_shared<mqbstat::QueueStatsDomain>(d_allocator_p))
 , d_virtualStorageCatalog(
       this,
+      d_queueStats_sp,
       allocatorStore ? allocatorStore->get("VirtualHandles") : d_allocator_p)
 , d_ttlSeconds(config.messageTtl())
 , d_isEmpty(1)
@@ -82,6 +86,7 @@ InMemoryStorage::InMemoryStorage(const bmqt::Uri&        uri,
 {
     BSLS_ASSERT_SAFE(0 <= d_ttlSeconds);  // Broadcast queues can use 0 for TTL
 
+    d_queueStats_sp->initialize(d_uri, domain);
     d_virtualStorageCatalog.setDefaultRda(config.maxDeliveryAttempts());
 
     if (isProxy()) {
@@ -132,6 +137,8 @@ void InMemoryStorage::setQueue(mqbi::Queue* queue)
     // Update queue stats if a queue has been associated with the storage.
 
     if (queue) {
+        queue->setStats(d_queueStats_sp);
+
         const bsls::Types::Int64 numMessage = numMessages(
             mqbu::StorageKey::k_NULL_KEY);
         const bsls::Types::Int64 numByte = numBytes(
@@ -234,13 +241,10 @@ InMemoryStorage::put(mqbi::StorageMessageAttributes*     attributes,
 
         d_currentlyAutoConfirming = bmqt::MessageGUID();
 
-        if (queue()) {
-            queue()
-                ->stats()
-                ->onEvent<mqbstat::QueueStatsDomain::EventType::e_ADD_MESSAGE>(
+        d_queueStats_sp
+            ->onEvent<mqbstat::QueueStatsDomain::EventType::e_ADD_MESSAGE>(
 
-                    msgSize);
-        }
+                msgSize);
 
         d_isEmpty.storeRelaxed(0);
 
@@ -332,11 +336,10 @@ InMemoryStorage::releaseRef(const bmqt::MessageGUID& guid)
         d_capacityMeter.remove(1, msgLen);
         if (queue()) {
             queue()->queueEngine()->beforeMessageRemoved(guid);
-            queue()
-                ->stats()
-                ->onEvent<mqbstat::QueueStatsDomain::EventType::e_DEL_MESSAGE>(
-                    msgLen);
         }
+        d_queueStats_sp
+            ->onEvent<mqbstat::QueueStatsDomain::EventType::e_DEL_MESSAGE>(
+                msgLen);
 
         // There is not really a need to remove the guid from all virtual
         // storages, because we can be here only if guid doesn't exist in
@@ -347,13 +350,9 @@ InMemoryStorage::releaseRef(const bmqt::MessageGUID& guid)
 
         d_items.erase(it);
 
-        if (queue()) {
-            queue()
-                ->stats()
-                ->onEvent<
-                    mqbstat::QueueStatsDomain::EventType::e_UPDATE_HISTORY>(
-                    d_items.historySize());
-        }
+        d_queueStats_sp
+            ->onEvent<mqbstat::QueueStatsDomain::EventType::e_UPDATE_HISTORY>(
+                d_items.historySize());
 
         return mqbi::StorageResult::e_ZERO_REFERENCES;  // RETURN
     }
@@ -378,16 +377,11 @@ InMemoryStorage::remove(const bmqt::MessageGUID& msgGUID, int* msgSize)
     // Update resource usage
     d_capacityMeter.remove(1, msgLen);
 
-    if (queue()) {
-        queue()
-            ->stats()
-            ->onEvent<mqbstat::QueueStatsDomain::EventType::e_DEL_MESSAGE>(
-                msgLen);
-        queue()
-            ->stats()
-            ->onEvent<mqbstat::QueueStatsDomain::EventType::e_UPDATE_HISTORY>(
-                d_items.historySize());
-    }
+    d_queueStats_sp
+        ->onEvent<mqbstat::QueueStatsDomain::EventType::e_DEL_MESSAGE>(msgLen);
+    d_queueStats_sp
+        ->onEvent<mqbstat::QueueStatsDomain::EventType::e_UPDATE_HISTORY>(
+            d_items.historySize());
 
     if (msgSize) {
         *msgSize = msgLen;
@@ -410,11 +404,8 @@ InMemoryStorage::removeAll(const mqbu::StorageKey& appKey)
         d_items.clear();
         d_capacityMeter.clear();
 
-        if (queue()) {
-            queue()
-                ->stats()
-                ->onEvent<mqbstat::QueueStatsDomain::EventType::e_PURGE>(0);
-        }
+        d_queueStats_sp
+            ->onEvent<mqbstat::QueueStatsDomain::EventType::e_PURGE>(0);
 
         d_isEmpty.storeRelaxed(1);
 
@@ -440,12 +431,9 @@ InMemoryStorage::removeAll(const mqbu::StorageKey& appKey)
         d_isEmpty.storeRelaxed(1);
     }
 
-    if (queue()) {
-        queue()
-            ->stats()
-            ->onEvent<mqbstat::QueueStatsDomain::EventType::e_UPDATE_HISTORY>(
-                d_items.historySize());
-    }
+    d_queueStats_sp
+        ->onEvent<mqbstat::QueueStatsDomain::EventType::e_UPDATE_HISTORY>(
+            d_items.historySize());
 
     return mqbi::StorageResult::e_SUCCESS;
 }
@@ -483,11 +471,10 @@ int InMemoryStorage::gcExpiredMessages(
         d_capacityMeter.remove(1, msgLen);
         if (queue()) {
             queue()->queueEngine()->beforeMessageRemoved(cit->first);
-            queue()
-                ->stats()
-                ->onEvent<mqbstat::QueueStatsDomain::EventType::e_DEL_MESSAGE>(
-                    msgLen);
         }
+        d_queueStats_sp
+            ->onEvent<mqbstat::QueueStatsDomain::EventType::e_DEL_MESSAGE>(
+                msgLen);
 
         // Remove message from all virtual storages and the physical (this)
         // storage.
@@ -496,13 +483,11 @@ int InMemoryStorage::gcExpiredMessages(
         ++numMsgsDeleted;
     }
 
-    if (queue() && (numMsgsDeleted > 0)) {
-        queue()
-            ->stats()
+    if (numMsgsDeleted > 0) {
+        d_queueStats_sp
             ->onEvent<mqbstat::QueueStatsDomain::EventType::e_GC_MESSAGE>(
                 numMsgsDeleted);
-        queue()
-            ->stats()
+        d_queueStats_sp
             ->onEvent<mqbstat::QueueStatsDomain::EventType::e_UPDATE_HISTORY>(
                 d_items.historySize());
     }
@@ -519,12 +504,9 @@ bool InMemoryStorage::gcHistory()
     bool hasMoreToGc = d_items.gc(bmqsys::Time::highResolutionTimer(),
                                   k_GC_MESSAGES_BATCH_SIZE);
 
-    if (queue()) {
-        queue()
-            ->stats()
-            ->onEvent<mqbstat::QueueStatsDomain::EventType::e_UPDATE_HISTORY>(
-                d_items.historySize());
-    }
+    d_queueStats_sp
+        ->onEvent<mqbstat::QueueStatsDomain::EventType::e_UPDATE_HISTORY>(
+            d_items.historySize());
 
     return hasMoreToGc;
 }
diff --git a/src/groups/mqb/mqbs/mqbs_inmemorystorage.h b/src/groups/mqb/mqbs/mqbs_inmemorystorage.h
@@ -187,6 +187,9 @@ class InMemoryStorage BSLS_KEYWORD_FINAL : public ReplicatedStorage {
 
     ItemsMap d_items;
 
+    /// Statistics of the queue associated to this storage.
+    bsl::shared_ptr<mqbstat::QueueStatsDomain> d_queueStats_sp;
+
     VirtualStorageCatalog d_virtualStorageCatalog;
 
     RecordHandles d_queueOpRecordHandles;
@@ -245,6 +248,7 @@ class InMemoryStorage BSLS_KEYWORD_FINAL : public ReplicatedStorage {
     /// `allocator`.
     InMemoryStorage(const bmqt::Uri&               uri,
                     const mqbu::StorageKey&        queueKey,
+                    mqbi::Domain*                  domain,
                     int                            partitionId,
                     const mqbconfm::Domain&        config,
                     mqbu::CapacityMeter*           parentCapacityMeter,
diff --git a/src/groups/mqb/mqbs/mqbs_inmemorystorage.t.cpp b/src/groups/mqb/mqbs/mqbs_inmemorystorage.t.cpp
@@ -205,6 +205,7 @@ struct Tester {
             new (*d_allocator_p) mqbs::InMemoryStorage(
                 bmqt::Uri(uri, bmqtst::TestHelperUtil::allocator()),
                 queueKey,
+                &d_mockDomain,
                 partitionId,
                 domainCfg,
                 d_mockDomain.capacityMeter(),
diff --git a/src/groups/mqb/mqbs/mqbs_storageprintutil.t.cpp b/src/groups/mqb/mqbs/mqbs_storageprintutil.t.cpp
@@ -19,6 +19,7 @@
 // MQB
 #include <mqbcmd_messages.h>
 #include <mqbi_queueengine.h>
+#include <mqbmock_domain.h>
 #include <mqbs_inmemorystorage.h>
 #include <mqbu_capacitymeter.h>
 #include <mqbu_messageguidutil.h>
@@ -122,6 +123,7 @@ struct Tester {
     bdlbb::PooledBlobBufferFactory           d_bufferFactory;
     bsl::vector<bmqt::MessageGUID>           d_guids;
     mqbu::CapacityMeter                      d_capacityMeter;
+    mqbmock::Domain                          d_domain;
     bslma::ManagedPtr<mqbs::InMemoryStorage> d_storage_mp;
 
   public:
@@ -130,6 +132,7 @@ struct Tester {
     : d_bufferFactory(1024, bmqtst::TestHelperUtil::allocator())
     , d_guids(bmqtst::TestHelperUtil::allocator())
     , d_capacityMeter("test", bmqtst::TestHelperUtil::allocator())
+    , d_domain(0, bmqtst::TestHelperUtil::allocator())
     {
         d_capacityMeter.setLimits(k_INT64_MAX, k_INT64_MAX);
 
@@ -142,6 +145,7 @@ struct Tester {
             new (*bmqtst::TestHelperUtil::allocator()) mqbs::InMemoryStorage(
                 bmqt::Uri(uri, bmqtst::TestHelperUtil::allocator()),
                 k_QUEUE_KEY,
+                &d_domain,
                 0,
                 domainCfg,
                 &d_capacityMeter,
diff --git a/src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.cpp b/src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.cpp
diff --git a/src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.h b/src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.h
diff --git a/src/integration-tests/test_admin_client.py b/src/integration-tests/test_admin_client.py
diff --git a/src/python/blazingmq/dev/it/cluster.py b/src/python/blazingmq/dev/it/cluster.py
diff --git a/src/python/blazingmq/dev/it/data/data_metrics.py b/src/python/blazingmq/dev/it/data/data_metrics.py
