Fix[MQB]: collect queue stats on replicas (#659)

Author: 678098

src/groups/mqb/mqbblp/mqbblp_localqueue.cpp

@@ -162,18 +162,18 @@ int LocalQueue::configure(bsl::ostream& errorDescription, bool isReconfigure)
                                 d_allocator_p);
     }
 
+    // Inform the storage about the queue.
+    d_state_p->storageManager()->setQueueRaw(queue,
+                                             d_state_p->uri(),
+                                             d_state_p->partitionId());
+
     rc = d_queueEngine_mp->configure(errorDescription, isReconfigure);
     if (rc != 0) {
         return 10 * rc + rc_QUEUE_ENGINE_CFG_FAILURE;  // RETURN
     }
 
     d_haveStrongConsistency = domainCfg.consistency().isStrongValue();
 
-    // Inform the storage about the queue.
-    d_state_p->storageManager()->setQueueRaw(queue,
-                                             d_state_p->uri(),
-                                             d_state_p->partitionId());
-
     d_state_p->stats()
         ->onEvent<mqbstat::QueueStatsDomain::EventType::e_CHANGE_ROLE>(
             mqbstat::QueueStatsDomain::Role::e_PRIMARY);

src/groups/mqb/mqbblp/mqbblp_pushstream.t.cpp

@@ -16,6 +16,10 @@
 // mqbblp_pushstream.t.cpp                                            -*-C++-*-
 #include <mqbblp_pushstream.h>
 
+// MQB
+#include <mqbmock_cluster.h>
+#include <mqbmock_domain.h>
+
 // BMQ
 #include <bmqp_messageguidgenerator.h>
 
@@ -133,12 +137,16 @@ static void test2_iterations()
         0,
         bmqtst::TestHelperUtil::allocator());
     bmqt::Uri        dummyUri("dummy", bmqtst::TestHelperUtil::allocator());
-    mqbconfm::Domain dummyDomain(bmqtst::TestHelperUtil::allocator());
+    mqbmock::Cluster dummyCluster(bmqtst::TestHelperUtil::allocator());
+    mqbmock::Domain  dummyDomain(&dummyCluster,
+                                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());
 

src/groups/mqb/mqbblp/mqbblp_queueconsumptionmonitor.t.cpp

@@ -232,6 +232,7 @@ Test::Test()
             d_allocator_p)
 , d_storage(d_queue.uri(),
             mqbu::StorageKey::k_NULL_KEY,
+            &d_domain,
             mqbs::DataStore::k_INVALID_PARTITION_ID,
             getDomainConfig(),
             d_domain.capacityMeter(),

src/groups/mqb/mqbblp/mqbblp_queueenginetester.cpp

@@ -534,6 +534,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(),

src/groups/mqb/mqbblp/mqbblp_remotequeue.cpp

@@ -116,6 +116,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(),

src/groups/mqb/mqbblp/mqbblp_rootqueueengine.cpp

@@ -462,6 +462,19 @@ int RootQueueEngine::initializeAppId(const bsl::string& appId,
 
         d_consumptionMonitor.registerSubStream(appId);
 
+        const bsls::Types::Int64 appNumMessages =
+            d_queueState_p->storage()->numMessages(appKey);
+        const bsls::Types::Int64 appNumBytes =
+            d_queueState_p->storage()->numBytes(appKey);
+
+        d_queueState_p->queue()->stats()->setOutstandingData(appNumMessages,
+                                                             appNumBytes,
+                                                             appId);
+        BALL_LOG_INFO << "Set outstanding data for appId[" << appId
+                      << "], queue [" << d_queueState_p->uri() << "]: ("
+                      << appNumMessages << " msgs, " << appNumBytes
+                      << " bytes)";
+
         BALL_LOG_INFO << "Found virtual storage for appId [" << appId
                       << "], queue [" << d_queueState_p->uri() << "], appKey ["
                       << appKey << "], ordinal [" << ordinal << "]";

src/groups/mqb/mqbblp/mqbblp_routers.t.cpp

@@ -26,6 +26,8 @@
 
 // MQB
 #include <mqbcfg_brokerconfig.h>
+#include <mqbmock_cluster.h>
+#include <mqbmock_domain.h>
 #include <mqbmock_queue.h>
 #include <mqbmock_queuehandle.h>
 #include <mqbs_inmemorystorage.h>
@@ -57,6 +59,8 @@ struct TestStorage {
     mqbconfm::Domain                         d_domainCfg;
     mqbu::CapacityMeter                      d_capacityMeter;
     mqbu::StorageKey                         d_storageKey;
+    mqbmock::Cluster                         d_cluster;
+    mqbmock::Domain                          d_domain;
     mqbs::InMemoryStorage                    d_storage;
     bslma::ManagedPtr<mqbi::StorageIterator> d_iterator;
     bdlbb::PooledBlobBufferFactory           d_bufferFactory;
@@ -69,8 +73,12 @@ struct TestStorage {
     , d_domainCfg(d_allocator_p)
     , d_capacityMeter(bsl::string("cm", d_allocator_p), 0, d_allocator_p)
     , d_storageKey(d_subQueueId)
+    , d_cluster(d_allocator_p)
+    , d_domain(&d_cluster,
+               d_allocator_p)  // Use domain only to hold mqbstat::StatContext
     , d_storage(bmqt::Uri("uri", d_allocator_p),
                 d_storageKey,
+                &d_domain,
                 1,
                 d_domainCfg,
                 &d_capacityMeter,

src/groups/mqb/mqbs/mqbs_filebackedstorage.cpp

@@ -89,9 +89,9 @@ void FileBackedStorage::purgeCommon(const mqbu::StorageKey& appKey,
         // Update stats
         d_capacityMeter.clear();
 
-        d_queueStats_sp
+        d_virtualStorageCatalog.stats()
             ->onEvent<mqbstat::QueueStatsDomain::EventType::e_PURGE>(0);
-        d_queueStats_sp
+        d_virtualStorageCatalog.stats()
             ->onEvent<mqbstat::QueueStatsDomain::EventType::e_UPDATE_HISTORY>(
                 d_handles.historySize());
     }
@@ -134,7 +134,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,12 +146,9 @@ FileBackedStorage::FileBackedStorage(
     // instance associated with it (instead of a 'mqbblp::Cluster' instance),
     // and domain instance will return a zero capacity meter when queries to be
     // passed to the 'FileBackedStorage' instance.
-
+    d_virtualStorageCatalog.stats()->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()
@@ -265,8 +261,6 @@ void FileBackedStorage::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(
@@ -500,7 +494,7 @@ FileBackedStorage::releaseRef(const bmqt::MessageGUID& guid, bool asPrimary)
             if (queue()) {
                 queue()->queueEngine()->beforeMessageRemoved(guid);
             }
-            d_queueStats_sp
+            d_virtualStorageCatalog.stats()
                 ->onEvent<mqbstat::QueueStatsDomain::EventType::e_DEL_MESSAGE>(
                     msgLen);
 
@@ -517,9 +511,10 @@ FileBackedStorage::releaseRef(const bmqt::MessageGUID& guid, bool asPrimary)
             d_capacityMeter.remove(1, msgLen);
             d_handles.erase(it);
 
-            d_queueStats_sp->onEvent<
-                mqbstat::QueueStatsDomain::EventType::e_UPDATE_HISTORY>(
-                d_handles.historySize());
+            d_virtualStorageCatalog.stats()
+                ->onEvent<
+                    mqbstat::QueueStatsDomain::EventType::e_UPDATE_HISTORY>(
+                    d_handles.historySize());
         }
 
         return mqbi::StorageResult::e_ZERO_REFERENCES;
@@ -622,7 +617,7 @@ FileBackedStorage::removeAll(const mqbu::StorageKey& appKey)
         }
     }
 
-    d_queueStats_sp
+    d_virtualStorageCatalog.stats()
         ->onEvent<mqbstat::QueueStatsDomain::EventType::e_UPDATE_HISTORY>(
             d_handles.historySize());
 
@@ -751,10 +746,12 @@ int FileBackedStorage::gcExpiredMessages(
     int                numMsgsUnreceipted = 0;
     bsls::Types::Int64 now   = bmqsys::Time::highResolutionTimer();
     int                limit = k_GC_MESSAGES_BATCH_SIZE;
-    bsls::Types::Int64 deduplicationTimeNs =
-        queue() ? queue()->domain()->config().deduplicationTimeMs() *
-                      bdlt::TimeUnitRatio::k_NANOSECONDS_PER_MILLISECOND
-                : 0;
+    bsls::Types::Int64 deduplicationTimeNs = 0;
+    if (queue() && queue()->domain()) {
+        deduplicationTimeNs =
+            queue()->domain()->config().deduplicationTimeMs() *
+            bdlt::TimeUnitRatio::k_NANOSECONDS_PER_MILLISECOND;
+    }
 
     for (RecordHandleMapIter next = d_handles.begin(), cit;
          next != d_handles.end() && --limit;) {
@@ -813,7 +810,7 @@ int FileBackedStorage::gcExpiredMessages(
         if (queue()) {
             queue()->queueEngine()->beforeMessageRemoved(cit->first);
         }
-        d_queueStats_sp
+        d_virtualStorageCatalog.stats()
             ->onEvent<mqbstat::QueueStatsDomain::EventType::e_DEL_MESSAGE>(
                 msgLen);
 
@@ -834,16 +831,17 @@ int FileBackedStorage::gcExpiredMessages(
 
     if (numMsgsDeleted > 0) {
         if (numMsgsDeleted > numMsgsUnreceipted) {
-            d_queueStats_sp
+            d_virtualStorageCatalog.stats()
                 ->onEvent<mqbstat::QueueStatsDomain::EventType::e_GC_MESSAGE>(
                     numMsgsDeleted - numMsgsUnreceipted);
         }
         if (numMsgsUnreceipted) {
-            d_queueStats_sp->onEvent<
-                mqbstat::QueueStatsDomain::EventType::e_NO_SC_MESSAGE>(
-                numMsgsUnreceipted);
+            d_virtualStorageCatalog.stats()
+                ->onEvent<
+                    mqbstat::QueueStatsDomain::EventType::e_NO_SC_MESSAGE>(
+                    numMsgsUnreceipted);
         }
-        d_queueStats_sp
+        d_virtualStorageCatalog.stats()
             ->onEvent<mqbstat::QueueStatsDomain::EventType::e_UPDATE_HISTORY>(
                 d_handles.historySize());
     }
@@ -859,7 +857,7 @@ int FileBackedStorage::gcHistory(bsls::Types::Int64 now)
 {
     const int rc = d_handles.gc(now, k_GC_MESSAGES_BATCH_SIZE);
     if (0 != rc) {
-        d_queueStats_sp
+        d_virtualStorageCatalog.stats()
             ->onEvent<mqbstat::QueueStatsDomain::EventType::e_UPDATE_HISTORY>(
                 d_handles.historySize());
     }
@@ -917,7 +915,7 @@ void FileBackedStorage::processMessageRecord(
         // Update the messages & bytes monitors, and the stats.
         d_capacityMeter.forceCommit(1, msgLen);  // Return value ignored.
 
-        d_queueStats_sp
+        d_virtualStorageCatalog.stats()
             ->onEvent<mqbstat::QueueStatsDomain::EventType::e_ADD_MESSAGE>(
                 msgLen);
 
@@ -1033,15 +1031,14 @@ void FileBackedStorage::processDeletionRecord(const bmqt::MessageGUID& guid)
     if (queue()) {
         queue()->queueEngine()->beforeMessageRemoved(guid);
     }
-    d_queueStats_sp
+    d_virtualStorageCatalog.stats()
         ->onEvent<mqbstat::QueueStatsDomain::EventType::e_DEL_MESSAGE>(msgLen);
 
-    // Delete 'guid' from all virtual storages, if any.  Note that 'guid'
-    // should have already been removed from each virtual storage when confirm
-    // records were received earlier for each appKey, but we remove the guid
-    // again, just in case.  When the code is mature enough, we could remove
-    // this.
-    d_virtualStorageCatalog.remove(guid);
+    // Delete 'guid' from all virtual storages, if any.
+    // Note that we call `gc`, not `remove`, because we want to update
+    // message/byte counters.  We don't replicate the last confirm and
+    // REPLICA needs to find appId that was implicitly confirmed and update it.
+    d_virtualStorageCatalog.gc(guid);
 
     d_capacityMeter.remove(1, msgLen, true /* silent mode; don't log */);
 
@@ -1060,7 +1057,7 @@ void FileBackedStorage::processDeletionRecord(const bmqt::MessageGUID& guid)
         d_isEmpty.storeRelaxed(1);
     }
 
-    d_queueStats_sp
+    d_virtualStorageCatalog.stats()
         ->onEvent<mqbstat::QueueStatsDomain::EventType::e_UPDATE_HISTORY>(
             d_handles.historySize());
 }
@@ -1133,7 +1130,7 @@ FileBackedStorage::autoConfirm(const mqbu::StorageKey& appKey,
 
 void FileBackedStorage::setPrimary()
 {
-    d_queueStats_sp
+    d_virtualStorageCatalog.stats()
         ->onEvent<mqbstat::QueueStatsDomain::EventType::e_CHANGE_ROLE>(
             mqbstat::QueueStatsDomain::Role::e_PRIMARY);
 }

src/groups/mqb/mqbs/mqbs_filebackedstorage.h

@@ -211,9 +211,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;

src/groups/mqb/mqbs/mqbs_filestore.cpp

@@ -5266,6 +5266,7 @@ void FileStore::createStorage(bsl::shared_ptr<ReplicatedStorage>* storageSp,
         storageSp->reset(new (*storageAlloc)
                              InMemoryStorage(queueUri,
                                              queueKey,
+                                             domain,
                                              config().partitionId(),
                                              domain->config(),
                                              domain->capacityMeter(),