Merge pull request #19637 from vespa-engine/vekterli/add-active-merge-window-size-metric

Add metric for active merge throttler window size

3 files changed, 16 insertions, 0 deletions

diff --git a/storage/src/tests/storageserver/mergethrottlertest.cpp b/storage/src/tests/storageserver/mergethrottlertest.cpp
index 568eee1e92c..5c192942521 100644
--- a/storage/src/tests/storageserver/mergethrottlertest.cpp
+++ b/storage/src/tests/storageserver/mergethrottlertest.cpp
@@ -1151,6 +1151,10 @@ TEST_F(MergeThrottlerTest, busy_returned_on_full_queue_for_merges_sent_from_dist
     size_t maxQueue = _throttlers[0]->getMaxQueueSize();
     ASSERT_EQ(20, maxQueue);
     ASSERT_LT(maxPending, 100);
+
+    EXPECT_EQ(_throttlers[0]->getMetrics().active_window_size.getLast(), 0);
+    EXPECT_EQ(_throttlers[0]->getMetrics().queueSize.getLast(), 0);
+
     for (size_t i = 0; i < maxPending + maxQueue; ++i) {
         std::vector<MergeBucketCommand::Node> nodes({{0}, {1}, {2}});
         // No chain set, i.e. merge command is freshly squeezed from a distributor.
@@ -1162,6 +1166,7 @@ TEST_F(MergeThrottlerTest, busy_returned_on_full_queue_for_merges_sent_from_dist
     // Wait till we have maxPending replies and maxQueue queued
     _topLinks[0]->waitForMessages(maxPending, _messageWaitTime);
     waitUntilMergeQueueIs(*_throttlers[0], maxQueue, _messageWaitTime);
+    EXPECT_EQ(_throttlers[0]->getMetrics().active_window_size.getLast(), maxPending);
     EXPECT_EQ(maxQueue, _throttlers[0]->getMetrics().queueSize.getMaximum());
 
     // Clear all forwarded merges
diff --git a/storage/src/vespa/storage/storageserver/mergethrottler.cpp b/storage/src/vespa/storage/storageserver/mergethrottler.cpp
index fa9ab22c1cb..a17c77f6ca4 100644
--- a/storage/src/vespa/storage/storageserver/mergethrottler.cpp
+++ b/storage/src/vespa/storage/storageserver/mergethrottler.cpp
@@ -72,6 +72,7 @@ MergeThrottler::Metrics::Metrics(metrics::MetricSet* owner)
     : metrics::MetricSet("mergethrottler", {}, "", owner),
       averageQueueWaitingTime("averagequeuewaitingtime", {}, "Average time a merge spends in the throttler queue", this),
       queueSize("queuesize", {}, "Length of merge queue", this),
+      active_window_size("active_window_size", {}, "Number of merges active within the pending window size", this),
       bounced_due_to_back_pressure("bounced_due_to_back_pressure", {}, "Number of merges bounced due to resource exhaustion back-pressure", this),
       chaining("mergechains", this),
       local("locallyexecutedmerges", this)
@@ -366,6 +367,7 @@ MergeThrottler::removeActiveMerge(ActiveMergeMap::iterator mergeIter)
     LOG(debug, "Removed merge for %s from internal state",
         mergeIter->first.toString().c_str());
     _merges.erase(mergeIter);
+    update_active_merge_window_size_metric();
 }
 
 api::StorageMessage::SP
@@ -815,6 +817,7 @@ MergeThrottler::processNewMergeCommand(
     // merge throttling window.
     assert(_merges.find(mergeCmd.getBucket()) == _merges.end());
     auto state = _merges.emplace(mergeCmd.getBucket(), ChainedMergeState(msg)).first;
+    update_active_merge_window_size_metric();
 
     LOG(debug, "Added merge %s to internal state",
         mergeCmd.toString().c_str());
@@ -1247,6 +1250,11 @@ MergeThrottler::set_disable_queue_limits_for_chained_merges(bool disable_limits)
 }
 
 void
+MergeThrottler::update_active_merge_window_size_metric() noexcept {
+    _metrics->active_window_size.set(static_cast<int64_t>(_merges.size()));
+}
+
+void
 MergeThrottler::print(std::ostream& out, bool /*verbose*/,
                       const std::string& /*indent*/) const
 {
diff --git a/storage/src/vespa/storage/storageserver/mergethrottler.h b/storage/src/vespa/storage/storageserver/mergethrottler.h
index 9b0fb125b2f..997477a4b70 100644
--- a/storage/src/vespa/storage/storageserver/mergethrottler.h
+++ b/storage/src/vespa/storage/storageserver/mergethrottler.h
@@ -64,6 +64,7 @@ public:
     public:
         metrics::DoubleAverageMetric averageQueueWaitingTime;
         metrics::LongValueMetric queueSize;
+        metrics::LongValueMetric active_window_size;
         metrics::LongCountMetric bounced_due_to_back_pressure;
         MergeOperationMetrics chaining;
         MergeOperationMetrics local;
@@ -388,6 +389,8 @@ private:
     void rejectOperationsInThreadQueue(MessageGuard&, uint32_t minimumStateVersion);
     void markActiveMergesAsAborted(uint32_t minimumStateVersion);
 
+    void update_active_merge_window_size_metric() noexcept;
+
     // const function, but metrics are mutable
     void updateOperationMetrics(
             const api::ReturnCode& result,