Feat[MQB]: Enhance queue consumption monitor alarm log with additional details

src/groups/mqb/mqbblp/mqbblp_queueconsumptionmonitor.cpp

@@ -19,20 +19,16 @@
 #include <mqbscm_version.h>
 // MBQ
 #include <mqbblp_queuehandlecatalog.h>
-#include <mqbcmd_humanprinter.h>
 #include <mqbcmd_messages.h>
 #include <mqbi_queueengine.h>
 #include <mqbi_storage.h>
-#include <mqbs_storageprintutil.h>
-#include <mqbu_capacitymeter.h>
 
 // BMQ
 #include <bmqp_ctrlmsg_messages.h>
 #include <bmqt_queueflags.h>
 #include <bmqt_uri.h>
 
 // MWC
-#include <mwctsk_alarmlog.h>
 #include <mwcu_memoutstream.h>
 #include <mwcu_printutil.h>
 
@@ -150,14 +146,17 @@ QueueConsumptionMonitor::SubStreamInfo::SubStreamInfo(
 
 // CREATORS
 QueueConsumptionMonitor::QueueConsumptionMonitor(QueueState*       queueState,
+                                                 const LoggingCb&  loggingCb,
                                                  bslma::Allocator* allocator)
 : d_queueState_p(queueState)
 , d_maxIdleTime(0)
 , d_currentTimer(0)
 , d_subStreamInfos(allocator)
+, d_loggingCb(loggingCb)
 {
     // PRECONDITIONS
     BSLS_ASSERT_SAFE(d_queueState_p);
+    BSLS_ASSERT_SAFE(d_loggingCb);
 }
 
 // MANIPULATORS
@@ -331,118 +330,12 @@ void QueueConsumptionMonitor::onTransitionToIdle(
     // PRECONDITIONS
     BSLS_ASSERT_SAFE(d_queueState_p->queue()->dispatcher()->inDispatcherThread(
         d_queueState_p->queue()));
+    BSLS_ASSERT_SAFE(d_loggingCb);
 
     subStreamInfo->d_state = State::e_IDLE;
 
-    bdlma::LocalSequentialAllocator<2048> localAllocator(0);
-    bsl::vector<mqbi::QueueHandle*>       handles(&localAllocator);
-    d_queueState_p->handleCatalog().loadHandles(&handles);
-
-    bmqt::UriBuilder uriBuilder(d_queueState_p->uri(), &localAllocator);
-    bsl::string      appId;
-
-    if (appKey.isNull()) {
-        appId = bmqp::ProtocolUtil::k_DEFAULT_APP_ID;
-    }
-    else if (d_queueState_p->storage()->hasVirtualStorage(appKey, &appId)) {
-        uriBuilder.setId(appId);
-    }
-
-    bmqt::Uri uri(&localAllocator);
-    uriBuilder.uri(&uri);
-
-    mwcu::MemOutStream ss(&localAllocator);
-
-    int idx          = 1;
-    int numConsumers = 0;
-
-    const bool isFanoutValue =
-        d_queueState_p->queue()->hasMultipleSubStreams();
-
-    for (bsl::vector<mqbi::QueueHandle*>::const_iterator it = handles.begin(),
-                                                         last = handles.end();
-         it != last;
-         ++it) {
-        const mqbi::QueueHandle::SubStreams& subStreamInfos =
-            (*it)->subStreamInfos();
-
-        for (mqbi::QueueHandle::SubStreams::const_iterator infoCiter =
-                 subStreamInfos.begin();
-             infoCiter != subStreamInfos.end();
-             ++infoCiter) {
-            const bsl::string& itemAppId = infoCiter->first;
-
-            bool isReader = !isFanoutValue &&
-                            bmqt::QueueFlagsUtil::isReader(
-                                (*it)->handleParameters().flags());
-            // Non-fanout mode consumer in the default subStream ?
-            isReader |= isFanoutValue && !itemAppId.empty();
-
-            if (!isReader) {
-                continue;  // CONTINUE
-            }
-
-            if (itemAppId != appId) {
-                continue;  // CONTINUE
-            }
-
-            numConsumers += infoCiter->second.d_counts.d_readCount;
-
-            const int level = 2, spacesPerLevel = 2;
-
-            ss << "\n  " << idx++ << ". " << (*it)->client()->description()
-               << mwcu::PrintUtil::newlineAndIndent(level, spacesPerLevel)
-               << "Handle Parameters .....: " << (*it)->handleParameters()
-               << mwcu::PrintUtil::newlineAndIndent(level, spacesPerLevel)
-               << "UnconfirmedMonitors ....:";
-
-            const bsl::vector<const mqbu::ResourceUsageMonitor*> monitors =
-                (*it)->unconfirmedMonitors(appId);
-            for (size_t i = 0; i < monitors.size(); ++i) {
-                ss << "\n  " << monitors[i];
-            }
-        }
-    }
-
-    mwcu::MemOutStream out;
-    out << "Queue '" << uri << "' ";
-    d_queueState_p->storage()->capacityMeter()->printShortSummary(out);
-    out << ", max idle time "
-        << mwcu::PrintUtil::prettyTimeInterval(d_maxIdleTime)
-        << " appears to be stuck. It currently has " << numConsumers
-        << " consumers." << ss.str() << "\n";
-
-    // Print the 10 oldest messages in the queue
-    static const int k_NUM_MSGS = 10;
-    const int        level = 0, spacesPerLevel = 2;
-
-    out << mwcu::PrintUtil::newlineAndIndent(level, spacesPerLevel)
-        << k_NUM_MSGS << " oldest messages in the queue:\n";
-
-    mqbcmd::Result result;
-    mqbs::StoragePrintUtil::listMessages(&result.makeQueueContents(),
-                                         appId,
-                                         0,
-                                         k_NUM_MSGS,
-                                         d_queueState_p->storage());
-    mqbcmd::HumanPrinter::print(out, result);
-
-    if (!head) {
-        return;  // RETURN
-    }
-
-    // Print the current head of the queue
-    mqbi::Storage* const storage = d_queueState_p->storage();
-    out << mwcu::PrintUtil::newlineAndIndent(level, spacesPerLevel)
-        << "Current head of the queue:\n";
-
-    mqbs::StoragePrintUtil::listMessage(&result.makeMessage(), storage, *head);
-
-    mqbcmd::HumanPrinter::print(out, result);
-    out << "\n";
-
-    MWCTSK_ALARMLOG_ALARM("QUEUE_CONSUMER_MONITOR")
-        << out.str() << MWCTSK_ALARMLOG_END;
+    // Call logging callback to log alarm info.
+    d_loggingCb(appKey, head);
 }
 
 }  // close package namespace

src/groups/mqb/mqbblp/mqbblp_queueconsumptionmonitor.h

@@ -213,6 +213,11 @@ class QueueConsumptionMonitor {
     typedef bsl::function<bslma::ManagedPtr<mqbi::StorageIterator>(void)>
         HeadCb;
 
+    typedef bsl::function<void(
+        const mqbu::StorageKey&                         appKey,
+        const bslma::ManagedPtr<mqbi::StorageIterator>& head)>
+        LoggingCb;
+
   private:
     // PRIVATE TYPES
 
@@ -265,6 +270,9 @@ class QueueConsumptionMonitor {
 
     SubStreamInfoMap d_subStreamInfos;
 
+    LoggingCb d_loggingCb;
+    // Callback to log alarm info.
+
     // NOT IMPLEMENTED
     QueueConsumptionMonitor(const QueueConsumptionMonitor&) BSLS_CPP11_DELETED;
     QueueConsumptionMonitor&
@@ -306,6 +314,7 @@ class QueueConsumptionMonitor {
     /// `basicAllocator` to supply memory.  If `basicAllocator` is 0, the
     /// currently installed default allocator is used.
     QueueConsumptionMonitor(QueueState*       queueState,
+                            const LoggingCb&  loggingCb,
                             bslma::Allocator* allocator);
 
     // MANIPULATORS

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

@@ -39,6 +39,7 @@
 #include <bmqt_queueflags.h>
 
 // MWC
+#include <mwctsk_alarmlog.h>
 #include <mwctst_scopedlogobserver.h>
 #include <mwctst_testhelper.h>
 #include <mwcu_memoutstream.h>
@@ -93,6 +94,17 @@ ClientContext::~ClientContext()
     // NOTHING
 }
 
+static void loggingCb(BSLS_ANNOTATION_UNUSED const mqbu::StorageKey& appKey,
+                      BSLS_ANNOTATION_UNUSED const
+                          bslma::ManagedPtr<mqbi::StorageIterator>& head)
+{
+    BALL_LOG_SET_CATEGORY("MQBBLP.QUEUECONSUMPTIONMONITORTEST");
+
+    mwcu::MemOutStream out(s_allocator_p);
+    out << "Test Alarm";
+    MWCTSK_ALARMLOG_ALARM("QUEUE_STUCK") << out.str() << MWCTSK_ALARMLOG_END;
+}
+
 struct Test : mwctst::Test {
     typedef bsl::vector<
         bsl::pair<mqbi::QueueHandle*, bmqp_ctrlmsg::QueueHandleParameters> >
@@ -152,7 +164,7 @@ Test::Test()
                d_partitionId,
                &d_domain,
                s_allocator_p)
-, d_monitor(&d_queueState, s_allocator_p)
+, d_monitor(&d_queueState, &loggingCb, s_allocator_p)
 , d_storage(d_queue.uri(),
             mqbu::StorageKey::k_NULL_KEY,
             mqbs::DataStore::k_INVALID_PARTITION_ID,
@@ -386,11 +398,7 @@ TEST_F(Test, logFormat)
     ASSERT_EQ(logObserver.records().size(), 1u);
     ASSERT(mwctst::ScopedLogObserverUtil::recordMessageMatch(
         logObserver.records().back(),
-        "ALARM \\[QUEUE_CONSUMER_MONITOR\\]",
-        s_allocator_p));
-    ASSERT(mwctst::ScopedLogObserverUtil::recordMessageMatch(
-        logObserver.records().back(),
-        "Queue '.*'",
+        "ALARM \\[QUEUE_STUCK\\]",
         s_allocator_p));
 }
 
@@ -440,11 +448,7 @@ TEST_F(Test, putAliveIdleSendAlive)
     ASSERT_EQ(logObserver.records().size(), ++expectedLogRecords);
     ASSERT(mwctst::ScopedLogObserverUtil::recordMessageMatch(
         logObserver.records().back(),
-        "ALARM \\[QUEUE_CONSUMER_MONITOR\\]",
-        s_allocator_p));
-    ASSERT(mwctst::ScopedLogObserverUtil::recordMessageMatch(
-        logObserver.records().back(),
-        "0 consumers",
+        "ALARM \\[QUEUE_STUCK\\]",
         s_allocator_p));
 
     d_monitor.onTimer(2 * k_MAX_IDLE_TIME + 2);
@@ -503,19 +507,7 @@ TEST_F(Test, putAliveIdleWithConsumer)
     ASSERT_EQ(logObserver.records().size(), ++expectedLogRecords);
     ASSERT(mwctst::ScopedLogObserverUtil::recordMessageMatch(
         logObserver.records().back(),
-        "ALARM \\[QUEUE_CONSUMER_MONITOR\\].*It currently has 2 consumers",
-        s_allocator_p));
-    ASSERT(mwctst::ScopedLogObserverUtil::recordMessageMatch(
-        logObserver.records().back(),
-        "test consumer 1",
-        s_allocator_p));
-    ASSERT(mwctst::ScopedLogObserverUtil::recordMessageMatch(
-        logObserver.records().back(),
-        "test consumer 2",
-        s_allocator_p));
-    ASSERT(!mwctst::ScopedLogObserverUtil::recordMessageMatch(
-        logObserver.records().back(),
-        "test producer",
+        "ALARM \\[QUEUE_STUCK\\]",
         s_allocator_p));
 }
 
@@ -689,11 +681,7 @@ TEST_F(Test, putAliveIdleSendAliveTwoSubstreams)
     for (int i = 0; i < 2; ++i) {
         ASSERT(mwctst::ScopedLogObserverUtil::recordMessageMatch(
             logObserver.records().rbegin()[i],
-            "ALARM \\[QUEUE_CONSUMER_MONITOR\\]",
-            s_allocator_p));
-        ASSERT(mwctst::ScopedLogObserverUtil::recordMessageMatch(
-            logObserver.records().rbegin()[i],
-            "0 consumers",
+            "ALARM \\[QUEUE_STUCK\\]",
             s_allocator_p));
     }
 
@@ -797,21 +785,8 @@ TEST_F(Test, putAliveIdleSendAliveTwoSubstreamsTwoConsumers)
          ++iter) {
         ASSERT(mwctst::ScopedLogObserverUtil::recordMessageMatch(
             *iter,
-            "ALARM \\[QUEUE_CONSUMER_MONITOR\\] Queue "
-            "'bmq://bmq.test.local/test_queue\\?id=app\\d'",
-            s_allocator_p));
-        ASSERT(
-            mwctst::ScopedLogObserverUtil::recordMessageMatch(*iter,
-                                                              "1 consumers",
-                                                              s_allocator_p));
-        ASSERT(mwctst::ScopedLogObserverUtil::recordMessageMatch(
-            *iter,
-            "test consumer \\d",
+            "ALARM \\[QUEUE_STUCK\\]",
             s_allocator_p));
-        ASSERT(
-            !mwctst::ScopedLogObserverUtil::recordMessageMatch(*iter,
-                                                               "test producer",
-                                                               s_allocator_p));
     }
 
     d_monitor.onTimer(2 * k_MAX_IDLE_TIME + 2);

src/groups/mqb/mqbblp/mqbblp_queueengineutil.h

@@ -514,6 +514,8 @@ struct QueueEngineUtil_AppState {
     // ACCESSORS
     size_t redeliveryListSize() const;
 
+    size_t putAsideListSize() const;
+
     Routers::Consumer* findQueueHandleContext(mqbi::QueueHandle* handle);
 
     unsigned int upstreamSubQueueId() const;
@@ -695,6 +697,11 @@ inline size_t QueueEngineUtil_AppState::redeliveryListSize() const
     return d_redeliveryList.size();
 }
 
+inline size_t QueueEngineUtil_AppState::putAsideListSize() const
+{
+    return d_putAsideList.size();
+}
+
 inline Routers::Consumer*
 QueueEngineUtil_AppState::findQueueHandleContext(mqbi::QueueHandle* handle)
 {