Merge pull request #19015 from vespa-engine/geirst/pending-message-tracker-reporter-per-stripe

Use distributor stripe index when setting up reporter for PendingMess…

5 files changed, 17 insertions, 14 deletions

diff --git a/storage/src/tests/distributor/blockingoperationstartertest.cpp b/storage/src/tests/distributor/blockingoperationstartertest.cpp
index 861f8e72832..72cc0e1ba9f 100644
--- a/storage/src/tests/distributor/blockingoperationstartertest.cpp
+++ b/storage/src/tests/distributor/blockingoperationstartertest.cpp
@@ -122,7 +122,7 @@ BlockingOperationStarterTest::SetUp()
     _compReg = std::make_unique<StorageComponentRegisterImpl>();
     _compReg->setClock(_clock);
     _clock.setAbsoluteTimeInSeconds(1);
-    _messageTracker = std::make_unique<PendingMessageTracker>(*_compReg);
+    _messageTracker = std::make_unique<PendingMessageTracker>(*_compReg, 0);
     _fake_ctx = std::make_unique<FakeDistributorStripeOperationContext>(*_messageTracker);
     _operation_sequencer = std::make_unique<OperationSequencer>();
     _operationStarter = std::make_unique<BlockingOperationStarter>(*_fake_ctx, *_operation_sequencer, *_starterImpl);
diff --git a/storage/src/tests/distributor/pendingmessagetrackertest.cpp b/storage/src/tests/distributor/pendingmessagetrackertest.cpp
index 20ffd216e3d..293e5d02d11 100644
--- a/storage/src/tests/distributor/pendingmessagetrackertest.cpp
+++ b/storage/src/tests/distributor/pendingmessagetrackertest.cpp
@@ -149,7 +149,7 @@ Fixture::Fixture()
     _clock.setAbsoluteTimeInSeconds(1);
     // Have to set clock in compReg before constructing tracker, or it'll
     // flip out and die on an explicit nullptr check.
-    _tracker = std::make_unique<PendingMessageTracker>(_compReg);
+    _tracker = std::make_unique<PendingMessageTracker>(_compReg, 0);
 }
 Fixture::~Fixture() = default;
 
@@ -160,7 +160,7 @@ TEST_F(PendingMessageTrackerTest, simple) {
     framework::defaultimplementation::FakeClock clock;
     compReg.setClock(clock);
     clock.setAbsoluteTimeInSeconds(1);
-    PendingMessageTracker tracker(compReg);
+    PendingMessageTracker tracker(compReg, 0);
 
     auto remove = std::make_shared<api::RemoveCommand>(
                     makeDocumentBucket(document::BucketId(16, 1234)),
@@ -217,14 +217,14 @@ TEST_F(PendingMessageTrackerTest, start_page) {
     StorageComponentRegisterImpl compReg;
     framework::defaultimplementation::FakeClock clock;
     compReg.setClock(clock);
-    PendingMessageTracker tracker(compReg);
+    PendingMessageTracker tracker(compReg, 3);
 
     {
         std::ostringstream ost;
-        tracker.reportStatus(ost, framework::HttpUrlPath("/pendingmessages"));
+        tracker.reportStatus(ost, framework::HttpUrlPath("/pendingmessages3"));
 
         EXPECT_THAT(ost.str(), HasSubstr(
-                "<h1>Pending messages to storage nodes</h1>\n"
+                "<h1>Pending messages to storage nodes (stripe 3)</h1>\n"
                 "View:\n"
                 "<ul>\n"
                 "<li><a href=\"?order=bucket\">Group by bucket</a></li>"
@@ -237,7 +237,7 @@ TEST_F(PendingMessageTrackerTest, multiple_messages) {
     framework::defaultimplementation::FakeClock clock;
     compReg.setClock(clock);
     clock.setAbsoluteTimeInSeconds(1);
-    PendingMessageTracker tracker(compReg);
+    PendingMessageTracker tracker(compReg, 0);
 
     insertMessages(tracker);
 
@@ -332,7 +332,7 @@ TEST_F(PendingMessageTrackerTest, get_pending_message_types) {
     framework::defaultimplementation::FakeClock clock;
     compReg.setClock(clock);
     clock.setAbsoluteTimeInSeconds(1);
-    PendingMessageTracker tracker(compReg);
+    PendingMessageTracker tracker(compReg, 0);
     document::BucketId bid(16, 1234);
 
     auto remove = std::make_shared<api::RemoveCommand>(makeDocumentBucket(bid),
@@ -364,7 +364,7 @@ TEST_F(PendingMessageTrackerTest, has_pending_message) {
     framework::defaultimplementation::FakeClock clock;
     compReg.setClock(clock);
     clock.setAbsoluteTimeInSeconds(1);
-    PendingMessageTracker tracker(compReg);
+    PendingMessageTracker tracker(compReg, 0);
     document::BucketId bid(16, 1234);
 
     EXPECT_FALSE(tracker.hasPendingMessage(1, makeDocumentBucket(bid), api::MessageType::REMOVE_ID));
@@ -407,7 +407,7 @@ TEST_F(PendingMessageTrackerTest, get_all_messages_for_single_bucket) {
     framework::defaultimplementation::FakeClock clock;
     compReg.setClock(clock);
     clock.setAbsoluteTimeInSeconds(1);
-    PendingMessageTracker tracker(compReg);
+    PendingMessageTracker tracker(compReg, 0);
 
     insertMessages(tracker);
 
diff --git a/storage/src/vespa/storage/distributor/distributor_stripe.cpp b/storage/src/vespa/storage/distributor/distributor_stripe.cpp
index 1050f511ad6..8cb260455ed 100644
--- a/storage/src/vespa/storage/distributor/distributor_stripe.cpp
+++ b/storage/src/vespa/storage/distributor/distributor_stripe.cpp
@@ -55,7 +55,7 @@ DistributorStripe::DistributorStripe(DistributorComponentRegister& compReg,
       _operationOwner(*this, _component.getClock()),
       _maintenanceOperationOwner(*this, _component.getClock()),
       _operation_sequencer(std::make_unique<OperationSequencer>()),
-      _pendingMessageTracker(compReg),
+      _pendingMessageTracker(compReg, stripe_index),
       _bucketDBUpdater(_component, _component, *this, *this, use_legacy_mode),
       _distributorStatusDelegate(compReg, *this, *this),
       _bucketDBStatusDelegate(compReg, *this, _bucketDBUpdater),
diff --git a/storage/src/vespa/storage/distributor/pendingmessagetracker.cpp b/storage/src/vespa/storage/distributor/pendingmessagetracker.cpp
index cb76b6d50b2..e4554b793b8 100644
--- a/storage/src/vespa/storage/distributor/pendingmessagetracker.cpp
+++ b/storage/src/vespa/storage/distributor/pendingmessagetracker.cpp
@@ -1,6 +1,7 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 #include "pendingmessagetracker.h"
 #include <vespa/vespalib/stllike/asciistream.h>
+#include <vespa/vespalib/util/stringfmt.h>
 #include <map>
 #include <algorithm>
 
@@ -9,11 +10,13 @@ LOG_SETUP(".pendingmessages");
 
 namespace storage::distributor {
 
-PendingMessageTracker::PendingMessageTracker(framework::ComponentRegister& cr)
-    : framework::HtmlStatusReporter("pendingmessages", "Pending messages to storage nodes"),
+PendingMessageTracker::PendingMessageTracker(framework::ComponentRegister& cr, uint32_t stripe_index)
+    : framework::HtmlStatusReporter(vespalib::make_string("pendingmessages%u", stripe_index),
+                                    vespalib::make_string("Pending messages to storage nodes (stripe %u)", stripe_index)),
       _component(cr, "pendingmessagetracker"),
       _nodeInfo(_component.getClock()),
       _nodeBusyDuration(60),
+      _deferred_read_tasks(),
       _lock()
 {
     _component.registerStatusPage(*this);
diff --git a/storage/src/vespa/storage/distributor/pendingmessagetracker.h b/storage/src/vespa/storage/distributor/pendingmessagetracker.h
index 13d83157150..ac30304ca9c 100644
--- a/storage/src/vespa/storage/distributor/pendingmessagetracker.h
+++ b/storage/src/vespa/storage/distributor/pendingmessagetracker.h
@@ -77,7 +77,7 @@ public:
      */
     using TimePoint = std::chrono::milliseconds;
 
-    explicit PendingMessageTracker(framework::ComponentRegister&);
+    explicit PendingMessageTracker(framework::ComponentRegister&, uint32_t stripe_index);
     ~PendingMessageTracker() override;
 
     void insert(const std::shared_ptr<api::StorageMessage>&);