Merge pull request #19164 from vespa-engine/vekterli/aggregate-pending-operation-stats-across-stripes

Aggregate pending operation stats across all stripes in stripe guard

3 files changed, 30 insertions, 3 deletions

diff --git a/storage/src/tests/distributor/multi_thread_stripe_access_guard_test.cpp b/storage/src/tests/distributor/multi_thread_stripe_access_guard_test.cpp
index ba28396886f..07005aa898b 100644
--- a/storage/src/tests/distributor/multi_thread_stripe_access_guard_test.cpp
+++ b/storage/src/tests/distributor/multi_thread_stripe_access_guard_test.cpp
@@ -18,6 +18,7 @@ namespace storage::distributor {
 struct AggregationTestingMockTickableStripe : MockTickableStripe {
     PotentialDataLossReport report;
     std::vector<dbtransition::Entry> entries;
+    StripeAccessGuard::PendingOperationStats pending_stats{0, 0};
 
     PotentialDataLossReport remove_superfluous_buckets(document::BucketSpace, const lib::ClusterState&, bool) override {
         return report;
@@ -38,6 +39,10 @@ struct AggregationTestingMockTickableStripe : MockTickableStripe {
         return result;
     }
 
+    StripeAccessGuard::PendingOperationStats pending_operation_stats() const override {
+        return pending_stats;
+    }
+
     bool tick() override {
         return false;
     }
@@ -96,6 +101,21 @@ TEST_F(MultiThreadedStripeAccessGuardTest, remove_superfluous_buckets_aggregates
     EXPECT_EQ(report.documents, 680);
 }
 
+TEST_F(MultiThreadedStripeAccessGuardTest, pending_operation_stats_aggregates_stats_across_stripes) {
+    using Stats = StripeAccessGuard::PendingOperationStats;
+    _stripe0.pending_stats = Stats(20, 100);
+    _stripe1.pending_stats = Stats(5,  200);
+    _stripe2.pending_stats = Stats(7,  350);
+    _stripe3.pending_stats = Stats(3,  30);
+    start_pool_with_stripes();
+
+    auto guard = _accessor.rendezvous_and_hold_all();
+    auto pending_stats = guard->pending_operation_stats();
+
+    EXPECT_EQ(pending_stats.external_load_operations, 35);
+    EXPECT_EQ(pending_stats.maintenance_operations, 680);
+}
+
 TEST_F(MultiThreadedStripeAccessGuardTest, merge_entries_into_db_operates_across_all_stripes) {
     start_pool_with_stripes();
     // Note: The bucket key is calculated by reversing the bits of the raw bucket id.
diff --git a/storage/src/vespa/storage/distributor/multi_threaded_stripe_access_guard.cpp b/storage/src/vespa/storage/distributor/multi_threaded_stripe_access_guard.cpp
index 359621425b2..1ca985947a2 100644
--- a/storage/src/vespa/storage/distributor/multi_threaded_stripe_access_guard.cpp
+++ b/storage/src/vespa/storage/distributor/multi_threaded_stripe_access_guard.cpp
@@ -140,8 +140,11 @@ void MultiThreadedStripeAccessGuard::report_bucket_db_status(document::BucketSpa
 
 StripeAccessGuard::PendingOperationStats
 MultiThreadedStripeAccessGuard::pending_operation_stats() const {
-    // TODO STRIPE multiple stripes
-    return first_stripe().pending_operation_stats();
+    StripeAccessGuard::PendingOperationStats stats(0, 0);
+    for_each_stripe([&](const TickableStripe& stripe) {
+        stats.merge(stripe.pending_operation_stats());
+    });
+    return stats;
 }
 
 void MultiThreadedStripeAccessGuard::report_single_bucket_requests(vespalib::xml::XmlOutputStream& xos) const {
diff --git a/storage/src/vespa/storage/distributor/stripe_access_guard.h b/storage/src/vespa/storage/distributor/stripe_access_guard.h
index 70f3ce846ea..1d8a63d088c 100644
--- a/storage/src/vespa/storage/distributor/stripe_access_guard.h
+++ b/storage/src/vespa/storage/distributor/stripe_access_guard.h
@@ -57,7 +57,6 @@ public:
     virtual void update_read_snapshot_after_activation(const lib::ClusterStateBundle& activated_state) = 0;
     virtual void clear_read_only_bucket_repo_databases() = 0;
 
-    // TODO STRIPE: Add merge() function.
     struct PendingOperationStats {
         size_t external_load_operations;
         size_t maintenance_operations;
@@ -65,6 +64,11 @@ public:
                               size_t maintenance_operations_in)
             : external_load_operations(external_load_operations_in),
               maintenance_operations(maintenance_operations_in) {}
+
+        void merge(const PendingOperationStats& rhs) noexcept {
+            external_load_operations += rhs.external_load_operations;
+            maintenance_operations   += rhs.maintenance_operations;
+        }
     };
 
     // Functions used for state reporting