Add metrics for active operations on service layer.

12 files changed, 447 insertions, 5 deletions

diff --git a/storage/src/tests/persistence/CMakeLists.txt b/storage/src/tests/persistence/CMakeLists.txt
index f0deec90aae..7b165e11b66 100644
--- a/storage/src/tests/persistence/CMakeLists.txt
+++ b/storage/src/tests/persistence/CMakeLists.txt
@@ -2,6 +2,7 @@
 
 vespa_add_executable(storage_persistence_gtest_runner_app TEST
     SOURCES
+    active_operations_stats_test.cpp
     apply_bucket_diff_state_test.cpp
     bucketownershipnotifiertest.cpp
     has_mask_remapper_test.cpp
diff --git a/storage/src/tests/persistence/active_operations_stats_test.cpp b/storage/src/tests/persistence/active_operations_stats_test.cpp
new file mode 100644
index 00000000000..a5dd3d929db
--- /dev/null
+++ b/storage/src/tests/persistence/active_operations_stats_test.cpp
@@ -0,0 +1,150 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include <vespa/persistence/dummyimpl/dummypersistence.h>
+#include <tests/persistence/common/filestortestfixture.h>
+#include <tests/persistence/filestorage/forwardingmessagesender.h>
+#include <vespa/storage/persistence/filestorage/filestormetrics.h>
+#include <vespa/document/test/make_document_bucket.h>
+#include <vespa/document/fieldset/fieldsets.h>
+#include <vespa/metrics/updatehook.h>
+
+using document::test::makeDocumentBucket;
+
+namespace storage {
+
+class ActiveOperationsStatsTest : public FileStorTestFixture
+{
+protected:
+    DummyStorageLink top;
+    std::unique_ptr<DummyStorageLink> dummyManager;
+    ForwardingMessageSender messageSender;
+    FileStorMetrics metrics;
+    std::unique_ptr<FileStorHandler> filestorHandler;
+    uint32_t stripeId;
+
+public:
+    ActiveOperationsStatsTest();
+    ~ActiveOperationsStatsTest() override;
+    std::shared_ptr<api::StorageMessage> createPut(uint64_t bucket, uint64_t docIdx);
+    std::shared_ptr<api::StorageMessage> createGet(uint64_t bucket) const;
+
+    void assert_active_operations_stats(const ActiveOperationsStats &stats, uint32_t exp_active_size, uint32_t exp_size_samples, uint32_t exp_latency_samples);
+    void update_metrics();
+    void test_active_operations_stats();
+};
+
+ActiveOperationsStatsTest::ActiveOperationsStatsTest()
+    : FileStorTestFixture(),
+      top(),
+      dummyManager(std::make_unique<DummyStorageLink>()),
+      messageSender(*dummyManager),
+      metrics(),
+      stripeId(0)
+{
+    setupPersistenceThreads(1);
+    _node->setPersistenceProvider(std::make_unique<spi::dummy::DummyPersistence>(_node->getTypeRepo()));
+    top.push_back(std::move(dummyManager));
+    top.open();
+    metrics.initDiskMetrics(1, 1);
+    filestorHandler = std::make_unique<FileStorHandlerImpl>(messageSender, metrics,
+                                                            _node->getComponentRegister());
+    filestorHandler->setGetNextMessageTimeout(20ms);
+}
+
+ActiveOperationsStatsTest::~ActiveOperationsStatsTest() = default;
+
+std::shared_ptr<api::StorageMessage>
+ActiveOperationsStatsTest::createPut(uint64_t bucket, uint64_t docIdx)
+{
+    auto doc = _node->getTestDocMan().createDocument(
+            "foobar", vespalib::make_string("id:foo:testdoctype1:n=%" PRIu64 ":%" PRIu64, bucket, docIdx));
+    auto cmd = std::make_shared<api::PutCommand>(makeDocumentBucket(document::BucketId(16, bucket)), std::move(doc), 1234);
+    cmd->setAddress(makeSelfAddress());
+    return cmd;
+}
+
+std::shared_ptr<api::StorageMessage>
+ActiveOperationsStatsTest::createGet(uint64_t bucket) const
+{
+    auto cmd = std::make_shared<api::GetCommand>(
+            makeDocumentBucket(document::BucketId(16, bucket)),
+            document::DocumentId(vespalib::make_string("id:foo:testdoctype1:n=%" PRIu64 ":0", bucket)), document::AllFields::NAME);
+    cmd->setAddress(makeSelfAddress());
+    return cmd;
+}
+
+void
+ActiveOperationsStatsTest::assert_active_operations_stats(const ActiveOperationsStats &stats, uint32_t exp_active_size, uint32_t exp_size_samples, uint32_t exp_latency_samples)
+{
+    EXPECT_EQ(exp_active_size, stats.get_active_size());
+    EXPECT_EQ(exp_size_samples, stats.get_size_samples());
+    EXPECT_EQ(exp_latency_samples, stats.get_latency_samples());
+}
+
+void
+ActiveOperationsStatsTest::update_metrics()
+{
+    std::mutex dummy_lock;
+    auto &impl = dynamic_cast<FileStorHandlerImpl&>(*filestorHandler);
+    auto& hook = impl.get_metric_update_hook_for_testing();
+    hook.updateMetrics(metrics::MetricLockGuard(dummy_lock));
+}
+
+void
+ActiveOperationsStatsTest::test_active_operations_stats()
+{
+    auto lock0 = filestorHandler->getNextMessage(stripeId);
+    auto lock1 = filestorHandler->getNextMessage(stripeId);
+    auto lock2 = filestorHandler->getNextMessage(stripeId);
+    ASSERT_TRUE(lock0.first);
+    ASSERT_TRUE(lock1.first);
+    ASSERT_FALSE(lock2.first);
+    auto stats = filestorHandler->get_active_operations_stats(false);
+    {
+        SCOPED_TRACE("during");
+        assert_active_operations_stats(stats, 2, 2, 0);
+    }
+    EXPECT_EQ(3, stats.get_total_size());
+    lock0.first.reset();
+    lock1.first.reset();
+    stats = filestorHandler->get_active_operations_stats(false);
+    {
+        SCOPED_TRACE("after");
+        assert_active_operations_stats(stats, 0, 4, 2);
+    }
+    EXPECT_EQ(4, stats.get_total_size());
+    EXPECT_LT(0.0, stats.get_total_latency());
+    update_metrics();
+    auto &ao_metrics = metrics.disk->active_operations;
+    EXPECT_DOUBLE_EQ(1.0, ao_metrics.size.getAverage());
+    EXPECT_DOUBLE_EQ(0.0, ao_metrics.size.getMinimum());
+    EXPECT_DOUBLE_EQ(2.0, ao_metrics.size.getMaximum());
+    EXPECT_DOUBLE_EQ(4.0, ao_metrics.size.getCount());
+    EXPECT_LT(0.0, ao_metrics.latency.getAverage());
+    EXPECT_LT(0.0, ao_metrics.latency.getMinimum());
+    EXPECT_LT(0.0, ao_metrics.latency.getMaximum());
+    EXPECT_DOUBLE_EQ(2.0, ao_metrics.latency.getCount());
+}
+
+TEST_F(ActiveOperationsStatsTest, empty_stats)
+{
+    auto stats = filestorHandler->get_active_operations_stats(false);
+    assert_active_operations_stats(stats, 0, 0, 0);
+}
+
+TEST_F(ActiveOperationsStatsTest, exclusive_lock_active_operations_stats)
+{
+    filestorHandler->schedule(createPut(1234, 0));
+    filestorHandler->schedule(createPut(1234, 1));
+    filestorHandler->schedule(createPut(5432, 0));
+    test_active_operations_stats();
+}
+
+TEST_F(ActiveOperationsStatsTest, shared_lock_active_operations_stats)
+{
+    filestorHandler->schedule(createGet(1234));
+    filestorHandler->schedule(createGet(1234));
+    test_active_operations_stats();
+}
+
+}
diff --git a/storage/src/vespa/storage/persistence/filestorage/CMakeLists.txt b/storage/src/vespa/storage/persistence/filestorage/CMakeLists.txt
index b23ec142448..62d1a80501a 100644
--- a/storage/src/vespa/storage/persistence/filestorage/CMakeLists.txt
+++ b/storage/src/vespa/storage/persistence/filestorage/CMakeLists.txt
@@ -1,6 +1,8 @@
 # Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 vespa_add_library(storage_filestorpersistence OBJECT
     SOURCES
+    active_operations_metrics.cpp
+    active_operations_stats.cpp
     filestorhandlerimpl.cpp
     filestormanager.cpp
     filestormetrics.cpp
diff --git a/storage/src/vespa/storage/persistence/filestorage/active_operations_metrics.cpp b/storage/src/vespa/storage/persistence/filestorage/active_operations_metrics.cpp
new file mode 100644
index 00000000000..e2259b98b3f
--- /dev/null
+++ b/storage/src/vespa/storage/persistence/filestorage/active_operations_metrics.cpp
@@ -0,0 +1,16 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "active_operations_metrics.h"
+
+namespace storage {
+
+ActiveOperationsMetrics::ActiveOperationsMetrics(metrics::MetricSet* parent)
+    : MetricSet("active_operations", {}, "metrics for active operations at service layer", parent),
+      size("size", {}, "Number of concurrent active operations", this),
+      latency("latency", {}, "Latency for active operations", this)
+{
+}
+
+ActiveOperationsMetrics::~ActiveOperationsMetrics() = default;
+
+}
diff --git a/storage/src/vespa/storage/persistence/filestorage/active_operations_metrics.h b/storage/src/vespa/storage/persistence/filestorage/active_operations_metrics.h
new file mode 100644
index 00000000000..94856d70f9e
--- /dev/null
+++ b/storage/src/vespa/storage/persistence/filestorage/active_operations_metrics.h
@@ -0,0 +1,22 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include <vespa/metrics/metricset.h>
+#include <vespa/metrics/valuemetric.h>
+
+namespace storage {
+
+/*
+ * Metrics for active operations with bucket lock at service layer.
+ */
+struct ActiveOperationsMetrics : public metrics::MetricSet
+{
+    metrics::DoubleAverageMetric size;
+    metrics::DoubleAverageMetric latency;
+
+    ActiveOperationsMetrics(metrics::MetricSet* parent);
+    ~ActiveOperationsMetrics() override;
+};
+
+}
diff --git a/storage/src/vespa/storage/persistence/filestorage/active_operations_stats.cpp b/storage/src/vespa/storage/persistence/filestorage/active_operations_stats.cpp
new file mode 100644
index 00000000000..bd7468971d4
--- /dev/null
+++ b/storage/src/vespa/storage/persistence/filestorage/active_operations_stats.cpp
@@ -0,0 +1,133 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "active_operations_stats.h"
+
+namespace storage {
+
+namespace {
+
+template <typename T>
+void update_min_max(T value, std::optional<T>& min, std::optional<T>& max)
+{
+    if (!min.has_value() || value < min.value()) {
+        min = value;
+    }
+    if (!max.has_value() || value > max.value()) {
+        max = value;
+    }
+}
+
+template <typename T>
+void merge_min(std::optional<T>& min, const std::optional<T>& rhs_min)
+{
+    if (!rhs_min.has_value()) {
+        return;
+    }
+    if (min.has_value() && !(rhs_min.value() < min.value())) {
+        return;
+    }
+    min = rhs_min;
+}
+
+template <typename T>
+void merge_max(std::optional<T>& max, const std::optional<T>& rhs_max)
+{
+    if (!rhs_max.has_value()) {
+        return;
+    }
+    if (max.has_value() && !(rhs_max.value() > max.value())) {
+        return;
+    }
+    max = rhs_max;
+}
+
+template <typename T>
+void merge_min_max_sum(std::optional<T>& lhs, const std::optional<T>& rhs)
+{
+    if (!rhs.has_value()) {
+        return;
+    }
+    if (lhs.has_value()) {
+        lhs = lhs.value() + rhs.value();
+        return;
+    }
+    lhs = rhs;
+}
+
+}
+
+ActiveOperationsStats::ActiveOperationsStats() noexcept
+    : _size_samples(0u),
+      _total_size(0u),
+      _active_size(0u),
+      _min_size(),
+      _max_size(),
+      _latency_samples(0u),
+      _total_latency(0.0),
+      _min_latency(),
+      _max_latency()
+{
+}
+
+ActiveOperationsStats::~ActiveOperationsStats() = default;
+
+
+void
+ActiveOperationsStats::update_size() noexcept
+{
+    ++_size_samples;
+    _total_size += _active_size;
+    update_min_max(_active_size, _min_size, _max_size);
+}
+
+ActiveOperationsStats&
+ActiveOperationsStats::operator-=(const ActiveOperationsStats& rhs) noexcept
+{
+    _size_samples -= rhs._size_samples;
+    _total_size -= rhs._total_size;
+    _latency_samples -= rhs._latency_samples;
+    _total_latency -= rhs._total_latency;
+    return *this;
+}
+
+void
+ActiveOperationsStats::merge(const ActiveOperationsStats& rhs) noexcept
+{
+    _size_samples += rhs._size_samples;
+    _total_size += rhs._total_size;
+    _active_size += rhs._active_size;
+    merge_min_max_sum(_min_size, rhs._min_size);
+    merge_min_max_sum(_max_size, rhs._max_size);
+    _latency_samples += rhs._latency_samples;
+    _total_latency += rhs._total_latency;
+    merge_min(_min_latency, rhs._min_latency);
+    merge_max(_max_latency, rhs._max_latency);
+}
+
+void
+ActiveOperationsStats::operation_started() noexcept
+{
+    ++_active_size;
+    update_size();
+}
+
+void
+ActiveOperationsStats::operation_done(double latency) noexcept
+{
+    --_active_size;
+    update_size();
+    ++_latency_samples;
+    _total_latency += latency;
+    update_min_max(latency, _min_latency, _max_latency);
+}
+
+void
+ActiveOperationsStats::reset_min_max() noexcept
+{
+    _min_size.reset();
+    _max_size.reset();
+    _min_latency.reset();
+    _max_latency.reset();
+}
+
+}
diff --git a/storage/src/vespa/storage/persistence/filestorage/active_operations_stats.h b/storage/src/vespa/storage/persistence/filestorage/active_operations_stats.h
new file mode 100644
index 00000000000..bdf4e87b1f5
--- /dev/null
+++ b/storage/src/vespa/storage/persistence/filestorage/active_operations_stats.h
@@ -0,0 +1,45 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include <cstdint>
+#include <optional>
+
+namespace storage {
+
+/*
+ * Stats for active operations at service layer
+ */
+class ActiveOperationsStats
+{
+    uint64_t                _size_samples;
+    uint64_t                _total_size;
+    uint32_t                _active_size;
+    std::optional<uint32_t> _min_size;
+    std::optional<uint32_t> _max_size;
+    uint64_t                _latency_samples;
+    double                  _total_latency;
+    std::optional<double>   _min_latency;
+    std::optional<double>   _max_latency;
+
+    void update_size() noexcept;
+public:
+    ActiveOperationsStats() noexcept;
+    ~ActiveOperationsStats();
+    ActiveOperationsStats& operator-=(const ActiveOperationsStats& rhs) noexcept;
+    void merge(const ActiveOperationsStats& rhs) noexcept;
+    void operation_started() noexcept;
+    void operation_done(double latency) noexcept;
+    void reset_min_max() noexcept;
+    uint64_t get_size_samples() const noexcept { return _size_samples; }
+    uint64_t get_latency_samples() const noexcept { return _latency_samples; }
+    uint64_t get_total_size() const noexcept { return _total_size; }
+    uint32_t get_active_size() const noexcept { return _active_size; }
+    double get_total_latency() const noexcept { return _total_latency; }
+    const std::optional<uint32_t>& get_min_size() const noexcept { return _min_size; }
+    const std::optional<uint32_t>& get_max_size() const noexcept { return _max_size; }
+    const std::optional<double>& get_min_latency() const noexcept { return _min_latency; }
+    const std::optional<double>& get_max_latency() const noexcept { return _max_latency; }
+};
+
+}
diff --git a/storage/src/vespa/storage/persistence/filestorage/filestorhandler.h b/storage/src/vespa/storage/persistence/filestorage/filestorhandler.h
index 70ed9845cb0..0d05fd21ce2 100644
--- a/storage/src/vespa/storage/persistence/filestorage/filestorhandler.h
+++ b/storage/src/vespa/storage/persistence/filestorage/filestorhandler.h
@@ -29,6 +29,7 @@ namespace framework {
     class HttpUrlPath;
 }
 
+class ActiveOperationsStats;
 class FileStorHandlerImpl;
 struct FileStorMetrics;
 struct MessageSender;
@@ -233,6 +234,7 @@ public:
 
     virtual std::string dumpQueue() const = 0;
 
+    virtual ActiveOperationsStats get_active_operations_stats(bool reset_min_max) const = 0;
 };
 
 } // storage
diff --git a/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.cpp b/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.cpp
index e395a7df9e0..99ff3eb1bcb 100644
--- a/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.cpp
+++ b/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.cpp
@@ -55,7 +55,8 @@ FileStorHandlerImpl::FileStorHandlerImpl(uint32_t numThreads, uint32_t numStripe
       _bucketIdFactory(_component.getBucketIdFactory()),
       _getNextMessageTimeout(100ms),
       _max_active_merges_per_stripe(per_stripe_merge_limit(numThreads, numStripes)),
-      _paused(false)
+      _paused(false),
+      _last_active_operations_stats()
 {
     assert(numStripes > 0);
     _stripes.reserve(numStripes);
@@ -296,6 +297,34 @@ FileStorHandlerImpl::abortQueuedOperations(const AbortBucketOperationsCommand& c
     }
 }
 
+namespace {
+
+void
+update_active_operations_metrics(ActiveOperationsMetrics& metrics, ActiveOperationsStats stats, std::optional<ActiveOperationsStats>& last_stats)
+{
+    auto delta_stats = stats;
+    if (last_stats.has_value()) {
+        delta_stats -= last_stats.value();
+    }
+    last_stats = stats;
+    uint32_t size_samples = delta_stats.get_size_samples();
+    if (size_samples != 0) {
+        double min_size = delta_stats.get_min_size().value_or(0);
+        double max_size = delta_stats.get_max_size().value_or(0);
+        double avg_size = ((double) delta_stats.get_total_size()) / size_samples;
+        metrics.size.addValueBatch(avg_size, size_samples, min_size, max_size);
+    }
+    uint32_t latency_samples = delta_stats.get_latency_samples();
+    if (latency_samples != 0) {
+        double min_latency = delta_stats.get_min_latency().value_or(0.0);
+        double max_latency = delta_stats.get_max_latency().value_or(0.0);
+        double avg_latency = delta_stats.get_total_latency() / latency_samples;
+        metrics.latency.addValueBatch(avg_latency, latency_samples, min_latency, max_latency);
+    }
+}
+
+}
+
 void
 FileStorHandlerImpl::updateMetrics(const MetricLockGuard &)
 {
@@ -307,6 +336,8 @@ FileStorHandlerImpl::updateMetrics(const MetricLockGuard &)
         const auto & m = stripe->averageQueueWaitingTime;
         _metrics->averageQueueWaitingTime.addTotalValueWithCount(m.getTotal(), m.getCount());
     }
+    auto active_operations_stats = get_active_operations_stats(true);
+    update_active_operations_metrics(_metrics->active_operations, active_operations_stats, _last_active_operations_stats);
 }
 
 bool
@@ -852,7 +883,8 @@ FileStorHandlerImpl::Stripe::Stripe(const FileStorHandlerImpl & owner, MessageSe
       _cond(std::make_unique<std::condition_variable>()),
       _queue(std::make_unique<PriorityQueue>()),
       _lockedBuckets(),
-      _active_merges(0)
+      _active_merges(0),
+      _active_operations_stats()
 {}
 
 FileStorHandler::LockedMessage
@@ -1030,6 +1062,7 @@ FileStorHandlerImpl::Stripe::release(const document::Bucket & bucket,
     auto iter = _lockedBuckets.find(bucket);
     assert(iter != _lockedBuckets.end());
     auto& entry = iter->second;
+    Clock::time_point start_time;
 
     if (reqOfReleasedLock == api::LockingRequirements::Exclusive) {
         assert(entry._exclusiveLock);
@@ -1038,14 +1071,18 @@ FileStorHandlerImpl::Stripe::release(const document::Bucket & bucket,
             assert(_active_merges > 0);
             --_active_merges;
         }
+        start_time = entry._exclusiveLock.value().timestamp;
         entry._exclusiveLock.reset();
     } else {
         assert(!entry._exclusiveLock);
         auto shared_iter = entry._sharedLocks.find(lockMsgId);
         assert(shared_iter != entry._sharedLocks.end());
+        start_time = shared_iter->second.timestamp;
         entry._sharedLocks.erase(shared_iter);
     }
-
+    Clock::time_point now_ts = Clock::now();
+    double latency = std::chrono::duration<double>(now_ts - start_time).count();
+    _active_operations_stats.operation_done(latency);
     if (!entry._exclusiveLock && entry._sharedLocks.empty()) {
         _lockedBuckets.erase(iter); // No more locks held
     }
@@ -1070,6 +1107,7 @@ FileStorHandlerImpl::Stripe::lock(const monitor_guard &, const document::Bucket
         (void) inserted;
         assert(inserted.second);
     }
+    _active_operations_stats.operation_started();
 }
 
 bool
@@ -1104,6 +1142,17 @@ FileStorHandlerImpl::Stripe::operationIsInhibited(const monitor_guard & guard, c
     return isLocked(guard, bucket, msg.lockingRequirements());
 }
 
+ActiveOperationsStats
+FileStorHandlerImpl::Stripe::get_active_operations_stats(bool reset_min_max) const
+{
+    std::lock_guard<std::mutex> guard(*_lock);
+    auto result = _active_operations_stats;
+    if (reset_min_max) {
+        _active_operations_stats.reset_min_max();
+    }
+    return result;
+}
+
 FileStorHandlerImpl::BucketLock::BucketLock(const monitor_guard & guard, Stripe& stripe,
                                             const document::Bucket &bucket, uint8_t priority,
                                             api::MessageType::Id msgType, api::StorageMessage::Id msgId,
@@ -1243,6 +1292,17 @@ FileStorHandlerImpl::getStatus(std::ostream& out, const framework::HttpUrlPath&
     }
 }
 
+ActiveOperationsStats
+FileStorHandlerImpl::get_active_operations_stats(bool reset_min_max) const
+{
+    ActiveOperationsStats result;
+    for (const auto & stripe : _stripes) {
+        auto stats = stripe.get_active_operations_stats(reset_min_max);
+        result.merge(stats);
+    }
+    return result;
+}
+
 void
 FileStorHandlerImpl::waitUntilNoLocks()
 {
diff --git a/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.h b/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.h
index 5f212b18a7f..5f0bc2d9939 100644
--- a/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.h
+++ b/storage/src/vespa/storage/persistence/filestorage/filestorhandlerimpl.h
@@ -16,6 +16,7 @@
 #pragma once
 
 #include "filestorhandler.h"
+#include "active_operations_stats.h"
 #include <vespa/document/bucket/bucketid.h>
 #include <vespa/metrics/metrictimer.h>
 #include <vespa/storage/common/servicelayercomponent.h>
@@ -136,6 +137,7 @@ public:
         PriorityQueue & exposeQueue() { return *_queue; }
         BucketIdx & exposeBucketIdx() { return bmi::get<2>(*_queue); }
         void setMetrics(FileStorStripeMetrics * metrics) { _metrics = metrics; }
+        ActiveOperationsStats get_active_operations_stats(bool reset_min_max) const;
     private:
         bool hasActive(monitor_guard & monitor, const AbortBucketOperationsCommand& cmd) const;
         FileStorHandler::LockedMessage get_next_async_message(monitor_guard& guard);
@@ -153,6 +155,7 @@ public:
         std::unique_ptr<PriorityQueue>  _queue;
         LockedBuckets                   _lockedBuckets;
         uint32_t                        _active_merges;
+        mutable ActiveOperationsStats   _active_operations_stats;
     };
 
     class BucketLock : public FileStorHandler::BucketLockInterface {
@@ -232,6 +235,9 @@ public:
     // Implements ResumeGuard::Callback
     void resume() override;
 
+    // Use only for testing
+    framework::MetricUpdateHook& get_metric_update_hook_for_testing() { return *this; }
+
 private:
     ServiceLayerComponent   _component;
     std::atomic<DiskState>  _state;
@@ -246,6 +252,7 @@ private:
     mutable std::mutex              _pauseMonitor;
     mutable std::condition_variable _pauseCond;
     std::atomic<bool>               _paused;
+    std::optional<ActiveOperationsStats> _last_active_operations_stats;
 
     // Returns the index in the targets array we are sending to, or -1 if none of them match.
     int calculateTargetBasedOnDocId(const api::StorageMessage& msg, std::vector<RemapInfo*>& targets);
@@ -322,6 +329,7 @@ private:
         return _stripes[stripeId].getNextMessage(timeout);
     }
 
+    ActiveOperationsStats get_active_operations_stats(bool reset_min_max) const override;
 };
 
 } // storage
diff --git a/storage/src/vespa/storage/persistence/filestorage/filestormetrics.cpp b/storage/src/vespa/storage/persistence/filestorage/filestormetrics.cpp
index cb7e141f5db..c119fdc4f69 100644
--- a/storage/src/vespa/storage/persistence/filestorage/filestormetrics.cpp
+++ b/storage/src/vespa/storage/persistence/filestorage/filestormetrics.cpp
@@ -206,7 +206,8 @@ FileStorDiskMetrics::FileStorDiskMetrics(const std::string& name, const std::str
       waitingForLockHitRate("waitingforlockrate", {},
               "Amount of times a filestor thread has needed to wait for "
               "lock to take next message in queue.", this),
-      lockWaitTime("lockwaittime", {}, "Amount of time waiting used waiting for lock.", this)
+      lockWaitTime("lockwaittime", {}, "Amount of time waiting used waiting for lock.", this),
+      active_operations(this)
 {
     pendingMerges.unsetOnZeroValue();
     waitingForLockHitRate.unsetOnZeroValue();
diff --git a/storage/src/vespa/storage/persistence/filestorage/filestormetrics.h b/storage/src/vespa/storage/persistence/filestorage/filestormetrics.h
index 8f7f79add00..7543e6e0771 100644
--- a/storage/src/vespa/storage/persistence/filestorage/filestormetrics.h
+++ b/storage/src/vespa/storage/persistence/filestorage/filestormetrics.h
@@ -11,6 +11,7 @@
 #pragma once
 
 #include "merge_handler_metrics.h"
+#include "active_operations_metrics.h"
 #include <vespa/metrics/metricset.h>
 #include <vespa/metrics/summetric.h>
 
@@ -147,7 +148,8 @@ public:
     metrics::LongAverageMetric queueSize;
     metrics::LongAverageMetric pendingMerges;
     metrics::DoubleAverageMetric waitingForLockHitRate;
-    metrics::DoubleAverageMetric lockWaitTime;
+    metrics::DoubleAverageMetric lockWaitTime; // unused
+    ActiveOperationsMetrics      active_operations;
 
     FileStorDiskMetrics(const std::string& name, const std::string& description, MetricSet* owner);
     ~FileStorDiskMetrics() override;