Add count metric for number of documents garbage collected

New distributor metric available as:
```
vds.idealstate.garbage_collection.documents_removed
```
Add documents removed statistics to `RemoveLocation` responses,
which is what GC is currently built around. Could technically have
been implemented as a diff of before/after BucketInfo, but GC is
very low priority so many other mutating ops may have changed the
bucket document set in the time span between sending the GC ops
and receiving the replies.

This relates to issue #12139

19 files changed, 131 insertions, 56 deletions

diff --git a/storage/src/tests/distributor/garbagecollectiontest.cpp b/storage/src/tests/distributor/garbagecollectiontest.cpp
index 65c1ac726b5..776cfc14d84 100644
--- a/storage/src/tests/distributor/garbagecollectiontest.cpp
+++ b/storage/src/tests/distributor/garbagecollectiontest.cpp
@@ -3,6 +3,7 @@
 #include <vespa/storageapi/message/removelocation.h>
 #include <vespa/storage/distributor/operations/idealstate/garbagecollectionoperation.h>
 #include <vespa/storage/distributor/idealstatemanager.h>
+#include <vespa/storage/distributor/idealstatemetricsset.h>
 #include <tests/distributor/distributortestutil.h>
 #include <vespa/storage/distributor/distributor.h>
 #include <vespa/document/test/make_document_bucket.h>
@@ -35,11 +36,13 @@ struct GarbageCollectionOperationTest : Test, DistributorTestUtil {
     }
 
     // FIXME fragile to assume that send order == node index, but that's the way it currently works
-    void reply_to_nth_request(GarbageCollectionOperation& op, size_t n, uint32_t bucket_info_checksum) {
+    void reply_to_nth_request(GarbageCollectionOperation& op, size_t n,
+                              uint32_t bucket_info_checksum, uint32_t n_docs_removed) {
         auto msg = _sender.command(n);
         assert(msg->getType() == api::MessageType::REMOVELOCATION);
         std::shared_ptr<api::StorageReply> reply(msg->makeReply());
         auto& gc_reply = dynamic_cast<api::RemoveLocationReply&>(*reply);
+        gc_reply.set_documents_removed(n_docs_removed);
         gc_reply.setBucketInfo(api::BucketInfo(bucket_info_checksum, 90, 500));
 
         op.receive(_sender, reply);
@@ -56,6 +59,13 @@ struct GarbageCollectionOperationTest : Test, DistributorTestUtil {
                     << entry->getNode(i)->getBucketInfo();
         }
     }
+
+    uint32_t gc_removed_documents_metric() {
+        auto metric_base = getIdealStateManager().getMetrics().operations[IdealStateOperation::GARBAGE_COLLECTION];
+        auto gc_metrics = std::dynamic_pointer_cast<GcMetricSet>(metric_base);
+        assert(gc_metrics);
+        return gc_metrics->documents_removed.getValue();
+    }
 };
 
 TEST_F(GarbageCollectionOperationTest, simple) {
@@ -63,29 +73,34 @@ TEST_F(GarbageCollectionOperationTest, simple) {
     op->start(_sender, framework::MilliSecTime(0));
 
     ASSERT_EQ(2, _sender.commands().size());
+    EXPECT_EQ(0u, gc_removed_documents_metric());
 
     for (uint32_t i = 0; i < 2; ++i) {
         std::shared_ptr<api::StorageCommand> msg = _sender.command(i);
         ASSERT_EQ(msg->getType(), api::MessageType::REMOVELOCATION);
         auto& tmp = dynamic_cast<api::RemoveLocationCommand&>(*msg);
         EXPECT_EQ("music.date < 34", tmp.getDocumentSelection());
-        reply_to_nth_request(*op, i, 777 + i);
+        reply_to_nth_request(*op, i, 777 + i, 50);
     }
     ASSERT_NO_FATAL_FAILURE(assert_bucket_db_contains({api::BucketInfo(777, 90, 500), api::BucketInfo(778, 90, 500)}, 34));
+    EXPECT_EQ(50u, gc_removed_documents_metric());
 }
 
 TEST_F(GarbageCollectionOperationTest, replica_bucket_info_not_added_to_db_until_all_replies_received) {
     auto op = create_op();
     op->start(_sender, framework::MilliSecTime(0));
     ASSERT_EQ(2, _sender.commands().size());
+    EXPECT_EQ(0u, gc_removed_documents_metric());
 
     // Respond to 1st request. Should _not_ cause bucket info to be merged into the database yet
-    reply_to_nth_request(*op, 0, 1234);
+    reply_to_nth_request(*op, 0, 1234, 70);
     ASSERT_NO_FATAL_FAILURE(assert_bucket_db_contains({api::BucketInfo(250, 50, 300), api::BucketInfo(250, 50, 300)}, 0));
 
     // Respond to 2nd request. This _should_ cause bucket info to be merged into the database.
-    reply_to_nth_request(*op, 1, 4567);
+    reply_to_nth_request(*op, 1, 4567, 60);
     ASSERT_NO_FATAL_FAILURE(assert_bucket_db_contains({api::BucketInfo(1234, 90, 500), api::BucketInfo(4567, 90, 500)}, 34));
+
+    EXPECT_EQ(70u, gc_removed_documents_metric()); // Use max of received metrics
 }
 
 TEST_F(GarbageCollectionOperationTest, gc_bucket_info_does_not_overwrite_later_sequenced_bucket_info_writes) {
@@ -93,10 +108,10 @@ TEST_F(GarbageCollectionOperationTest, gc_bucket_info_does_not_overwrite_later_s
     op->start(_sender, framework::MilliSecTime(0));
     ASSERT_EQ(2, _sender.commands().size());
 
-    reply_to_nth_request(*op, 0, 1234);
+    reply_to_nth_request(*op, 0, 1234, 0);
     // Change to replica on node 0 happens after GC op, but before GC info is merged into the DB. Must not be lost.
     insertBucketInfo(op->getBucketId(), 0, 7777, 100, 2000);
-    reply_to_nth_request(*op, 1, 4567);
+    reply_to_nth_request(*op, 1, 4567, 0);
     // Bucket info for node 0 is that of the later sequenced operation, _not_ from the earlier GC op.
     ASSERT_NO_FATAL_FAILURE(assert_bucket_db_contains({api::BucketInfo(7777, 100, 2000), api::BucketInfo(4567, 90, 500)}, 34));
 }
diff --git a/storage/src/tests/persistence/processalltest.cpp b/storage/src/tests/persistence/processalltest.cpp
index 8c0f8853d2d..83f243ed1b2 100644
--- a/storage/src/tests/persistence/processalltest.cpp
+++ b/storage/src/tests/persistence/processalltest.cpp
@@ -23,11 +23,15 @@ TEST_F(ProcessAllHandlerTest, remove_location) {
     api::RemoveLocationCommand removeLocation("id.user == 4", makeDocumentBucket(bucketId));
     ProcessAllHandler handler(getEnv(), getPersistenceProvider());
     spi::Context context(documentapi::LoadType::DEFAULT, 0, 0);
-    handler.handleRemoveLocation(removeLocation, context);
+    auto tracker = handler.handleRemoveLocation(removeLocation, context);
 
     EXPECT_EQ("DocEntry(1234, 1, id:mail:testdoctype1:n=4:3619.html)\n"
               "DocEntry(2345, 1, id:mail:testdoctype1:n=4:4008.html)\n",
               dumpBucket(bucketId));
+
+    auto reply = std::dynamic_pointer_cast<api::RemoveLocationReply>(tracker->getReply());
+    ASSERT_TRUE(reply.get() != nullptr);
+    EXPECT_EQ(2u, reply->documents_removed());
 }
 
 TEST_F(ProcessAllHandlerTest, remove_location_document_subset) {
@@ -44,7 +48,7 @@ TEST_F(ProcessAllHandlerTest, remove_location_document_subset) {
     api::RemoveLocationCommand
         removeLocation("testdoctype1.headerval % 2 == 0", makeDocumentBucket(bucketId));
     spi::Context context(documentapi::LoadType::DEFAULT, 0, 0);
-    handler.handleRemoveLocation(removeLocation, context);
+    auto tracker = handler.handleRemoveLocation(removeLocation, context);
 
     EXPECT_EQ("DocEntry(100, 1, id:mail:testdoctype1:n=4:3619.html)\n"
               "DocEntry(101, 0, Doc(id:mail:testdoctype1:n=4:33113.html))\n"
@@ -57,6 +61,10 @@ TEST_F(ProcessAllHandlerTest, remove_location_document_subset) {
               "DocEntry(108, 1, id:mail:testdoctype1:n=4:42967.html)\n"
               "DocEntry(109, 0, Doc(id:mail:testdoctype1:n=4:6925.html))\n",
               dumpBucket(bucketId));
+
+    auto reply = std::dynamic_pointer_cast<api::RemoveLocationReply>(tracker->getReply());
+    ASSERT_TRUE(reply.get() != nullptr);
+    EXPECT_EQ(5u, reply->documents_removed());
 }
 
 TEST_F(ProcessAllHandlerTest, remove_location_throws_exception_on_unknown_doc_type) {
diff --git a/storage/src/vespa/storage/distributor/bucketdb/bucketdbmetricupdater.cpp b/storage/src/vespa/storage/distributor/bucketdb/bucketdbmetricupdater.cpp
index 61e67b40f44..c211e775326 100644
--- a/storage/src/vespa/storage/distributor/bucketdb/bucketdbmetricupdater.cpp
+++ b/storage/src/vespa/storage/distributor/bucketdb/bucketdbmetricupdater.cpp
@@ -4,8 +4,7 @@
 #include <vespa/storage/distributor/distributormetricsset.h>
 #include <vespa/storage/distributor/idealstatemetricsset.h>
 
-namespace storage {
-namespace distributor {
+namespace storage::distributor {
 
 BucketDBMetricUpdater::Stats::Stats()
     : _docCount(0),
@@ -27,9 +26,7 @@ BucketDBMetricUpdater::BucketDBMetricUpdater()
 {
 }
 
-BucketDBMetricUpdater::~BucketDBMetricUpdater()
-{
-}
+BucketDBMetricUpdater::~BucketDBMetricUpdater() = default;
 
 void
 BucketDBMetricUpdater::resetStats()
@@ -148,5 +145,4 @@ BucketDBMetricUpdater::reset()
     resetStats();
 }
 
-} // distributor
-} // storage
+} // storage::distributor
diff --git a/storage/src/vespa/storage/distributor/bucketdb/bucketdbmetricupdater.h b/storage/src/vespa/storage/distributor/bucketdb/bucketdbmetricupdater.h
index 6e15ee03d12..7ef8479866f 100644
--- a/storage/src/vespa/storage/distributor/bucketdb/bucketdbmetricupdater.h
+++ b/storage/src/vespa/storage/distributor/bucketdb/bucketdbmetricupdater.h
@@ -7,12 +7,9 @@
 
 #include <unordered_map>
 
-namespace storage {
+namespace storage::distributor {
 
 class DistributorMetricSet;
-
-namespace distributor {
-
 class IdealStateMetricSet;
 
 class BucketDBMetricUpdater {
@@ -107,5 +104,4 @@ private:
     void resetStats();
 };
 
-} // distributor
-} // storage
+} // storage::distributor
diff --git a/storage/src/vespa/storage/distributor/distributorinterface.h b/storage/src/vespa/storage/distributor/distributorinterface.h
index aba58e112dc..b17bcd56d19 100644
--- a/storage/src/vespa/storage/distributor/distributorinterface.h
+++ b/storage/src/vespa/storage/distributor/distributorinterface.h
@@ -12,10 +12,10 @@ namespace storage::api { class MergeBucketReply; }
 namespace storage::lib { class ClusterStateBundle; }
 namespace storage {
    class DistributorConfiguration;
-   class DistributorMetricSet;
 }
 namespace storage::distributor {
 
+class DistributorMetricSet;
 class PendingMessageTracker;
 
 class DistributorInterface : public DistributorMessageSender
diff --git a/storage/src/vespa/storage/distributor/distributormetricsset.cpp b/storage/src/vespa/storage/distributor/distributormetricsset.cpp
index 244406ca6fb..98e96f9294f 100644
--- a/storage/src/vespa/storage/distributor/distributormetricsset.cpp
+++ b/storage/src/vespa/storage/distributor/distributormetricsset.cpp
@@ -3,7 +3,7 @@
 #include <vespa/metrics/loadmetric.hpp>
 #include <vespa/metrics/summetric.hpp>
 
-namespace storage {
+namespace storage::distributor {
 
 using metrics::MetricSet;
 
diff --git a/storage/src/vespa/storage/distributor/distributormetricsset.h b/storage/src/vespa/storage/distributor/distributormetricsset.h
index 1e4730b8de6..b5be72e8c14 100644
--- a/storage/src/vespa/storage/distributor/distributormetricsset.h
+++ b/storage/src/vespa/storage/distributor/distributormetricsset.h
@@ -7,7 +7,7 @@
 #include <vespa/metrics/metrics.h>
 #include <vespa/documentapi/loadtypes/loadtypeset.h>
 
-namespace storage {
+namespace storage::distributor {
 
 class DistributorMetricSet : public metrics::MetricSet
 {
diff --git a/storage/src/vespa/storage/distributor/externaloperationhandler.h b/storage/src/vespa/storage/distributor/externaloperationhandler.h
index 96875a3644a..60cad15a791 100644
--- a/storage/src/vespa/storage/distributor/externaloperationhandler.h
+++ b/storage/src/vespa/storage/distributor/externaloperationhandler.h
@@ -13,11 +13,11 @@
 
 namespace storage {
 
-class DistributorMetricSet;
 class PersistenceOperationMetricSet;
 
 namespace distributor {
 
+class DistributorMetricSet;
 class Distributor;
 class MaintenanceOperationGenerator;
 class DirectDispatchSender;
diff --git a/storage/src/vespa/storage/distributor/idealstatemetricsset.cpp b/storage/src/vespa/storage/distributor/idealstatemetricsset.cpp
index d72f4a80ef4..fd193ad6fd8 100644
--- a/storage/src/vespa/storage/distributor/idealstatemetricsset.cpp
+++ b/storage/src/vespa/storage/distributor/idealstatemetricsset.cpp
@@ -6,7 +6,7 @@ namespace storage {
 namespace distributor {
 
 OperationMetricSet::OperationMetricSet(const std::string& name, metrics::Metric::Tags tags, const std::string& description, MetricSet* owner)
-    : MetricSet(name, tags, description, owner),
+    : MetricSet(name, std::move(tags), description, owner),
       pending("pending",
               {{"logdefault"},{"yamasdefault"}},
               "The number of operations pending", this),
@@ -16,14 +16,25 @@ OperationMetricSet::OperationMetricSet(const std::string& name, metrics::Metric:
       failed("done_failed",
              {{"logdefault"},{"yamasdefault"}},
              "The number of operations that failed", this)
-{ }
+{}
 
-OperationMetricSet::~OperationMetricSet() { }
+OperationMetricSet::~OperationMetricSet() = default;
+
+GcMetricSet::GcMetricSet(const std::string& name, metrics::Metric::Tags tags, const std::string& description, MetricSet* owner)
+        : OperationMetricSet(name, std::move(tags), description, owner),
+          documents_removed("documents_removed",
+                           {{"logdefault"},{"yamasdefault"}},
+                           "Number of documents removed by GC operations", this)
+{}
+
+GcMetricSet::~GcMetricSet() = default;
 
 void
 IdealStateMetricSet::createOperationMetrics() {
     typedef IdealStateOperation ISO;
     operations.resize(ISO::OPERATION_COUNT);
+    // Note: naked new is used instead of make_shared due to the latter not being
+    // able to properly transitively deduce the types for the tag initializer lists.
     operations[ISO::DELETE_BUCKET] = std::shared_ptr<OperationMetricSet>(
             new OperationMetricSet("delete_bucket",
                                    {{"logdefault"},{"yamasdefault"}},
@@ -45,9 +56,9 @@ IdealStateMetricSet::createOperationMetrics() {
                                    {{"logdefault"},{"yamasdefault"}},
                                    "Operations to set active/ready state for bucket copies", this));
     operations[ISO::GARBAGE_COLLECTION] = std::shared_ptr<OperationMetricSet>(
-            new OperationMetricSet("garbage_collection",
-                                   {{"logdefault"},{"yamasdefault"}},
-                                   "Operations to garbage collect data from buckets", this));
+            new GcMetricSet("garbage_collection",
+                            {{"logdefault"},{"yamasdefault"}},
+                            "Operations to garbage collect data from buckets", this));
 }
 
 IdealStateMetricSet::IdealStateMetricSet()
@@ -81,7 +92,7 @@ IdealStateMetricSet::IdealStateMetricSet()
     createOperationMetrics();
 }
 
-IdealStateMetricSet::~IdealStateMetricSet() { }
+IdealStateMetricSet::~IdealStateMetricSet() = default;
 
 void IdealStateMetricSet::setPendingOperations(const std::vector<uint64_t>& newMetrics) {
     for (uint32_t i = 0; i < IdealStateOperation::OPERATION_COUNT; i++) {
diff --git a/storage/src/vespa/storage/distributor/idealstatemetricsset.h b/storage/src/vespa/storage/distributor/idealstatemetricsset.h
index 7bb472b4a2c..2679da17598 100644
--- a/storage/src/vespa/storage/distributor/idealstatemetricsset.h
+++ b/storage/src/vespa/storage/distributor/idealstatemetricsset.h
@@ -16,13 +16,21 @@ public:
     metrics::LongCountMetric failed;
 
     OperationMetricSet(const std::string& name, metrics::Metric::Tags tags, const std::string& description, MetricSet* owner);
-    ~OperationMetricSet();
+    ~OperationMetricSet() override;
+};
+
+struct GcMetricSet : OperationMetricSet {
+    metrics::LongCountMetric documents_removed;
+
+    GcMetricSet(const std::string& name, metrics::Metric::Tags tags,
+                const std::string& description, MetricSet* owner);
+    ~GcMetricSet() override;
 };
 
 class IdealStateMetricSet : public metrics::MetricSet
 {
 public:
-    std::vector<std::shared_ptr<OperationMetricSet> > operations;
+    std::vector<std::shared_ptr<OperationMetricSet>> operations;
     metrics::LongValueMetric idealstate_diff;
     metrics::LongValueMetric buckets_toofewcopies;
     metrics::LongValueMetric buckets_toomanycopies;
@@ -35,7 +43,7 @@ public:
     void createOperationMetrics();
 
     IdealStateMetricSet();
-    ~IdealStateMetricSet();
+    ~IdealStateMetricSet() override;
 
     void setPendingOperations(const std::vector<uint64_t>& newMetrics);
 };
diff --git a/storage/src/vespa/storage/distributor/operations/idealstate/garbagecollectionoperation.cpp b/storage/src/vespa/storage/distributor/operations/idealstate/garbagecollectionoperation.cpp
index c674add80f7..fc127c2e0eb 100644
--- a/storage/src/vespa/storage/distributor/operations/idealstate/garbagecollectionoperation.cpp
+++ b/storage/src/vespa/storage/distributor/operations/idealstate/garbagecollectionoperation.cpp
@@ -2,6 +2,7 @@
 
 #include "garbagecollectionoperation.h"
 #include <vespa/storage/distributor/idealstatemanager.h>
+#include <vespa/storage/distributor/idealstatemetricsset.h>
 #include <vespa/storage/distributor/distributor.h>
 #include <vespa/storage/distributor/distributor_bucket_space.h>
 #include <vespa/storageapi/message/removelocation.h>
@@ -9,19 +10,18 @@
 #include <vespa/log/log.h>
 LOG_SETUP(".distributor.operation.idealstate.remove");
 
-using namespace storage::distributor;
+namespace storage::distributor {
 
 GarbageCollectionOperation::GarbageCollectionOperation(const std::string& clusterName, const BucketAndNodes& nodes)
     : IdealStateOperation(nodes),
       _tracker(clusterName),
-      _replica_info()
+      _replica_info(),
+      _max_documents_removed(0)
 {}
 
 GarbageCollectionOperation::~GarbageCollectionOperation() = default;
 
-void
-GarbageCollectionOperation::onStart(DistributorMessageSender& sender)
-{
+void GarbageCollectionOperation::onStart(DistributorMessageSender& sender) {
     BucketDatabase::Entry entry = _bucketSpace->getBucketDatabase().get(getBucketId());
     std::vector<uint16_t> nodes = entry->getNodes();
 
@@ -43,7 +43,7 @@ GarbageCollectionOperation::onStart(DistributorMessageSender& sender)
 
 void
 GarbageCollectionOperation::onReceive(DistributorMessageSender&,
-                           const std::shared_ptr<api::StorageReply>& reply)
+                                      const std::shared_ptr<api::StorageReply>& reply)
 {
     auto* rep = dynamic_cast<api::RemoveLocationReply*>(reply.get());
     assert(rep != nullptr);
@@ -53,6 +53,7 @@ GarbageCollectionOperation::onReceive(DistributorMessageSender&,
     if (!rep->getResult().failed()) {
         _replica_info.emplace_back(_manager->getDistributorComponent().getUniqueTimestamp(),
                                    node, rep->getBucketInfo());
+        _max_documents_removed = std::max(_max_documents_removed, rep->documents_removed());
     } else {
         _ok = false;
     }
@@ -61,6 +62,7 @@ GarbageCollectionOperation::onReceive(DistributorMessageSender&,
         if (_ok) {
             merge_received_bucket_info_into_db();
         }
+        update_gc_metrics();
         done();
     }
 }
@@ -76,8 +78,16 @@ void GarbageCollectionOperation::merge_received_bucket_info_into_db() {
     }
 }
 
+void GarbageCollectionOperation::update_gc_metrics() {
+    auto metric_base = _manager->getMetrics().operations[IdealStateOperation::GARBAGE_COLLECTION];
+    auto gc_metrics = std::dynamic_pointer_cast<GcMetricSet>(metric_base);
+    assert(gc_metrics);
+    gc_metrics->documents_removed.inc(_max_documents_removed);
+}
+
 bool
-GarbageCollectionOperation::shouldBlockThisOperation(uint32_t, uint8_t) const
-{
+GarbageCollectionOperation::shouldBlockThisOperation(uint32_t, uint8_t) const {
     return true;
 }
+
+}
diff --git a/storage/src/vespa/storage/distributor/operations/idealstate/garbagecollectionoperation.h b/storage/src/vespa/storage/distributor/operations/idealstate/garbagecollectionoperation.h
index 47ea11bb328..28de9592a63 100644
--- a/storage/src/vespa/storage/distributor/operations/idealstate/garbagecollectionoperation.h
+++ b/storage/src/vespa/storage/distributor/operations/idealstate/garbagecollectionoperation.h
@@ -26,8 +26,10 @@ protected:
     MessageTracker _tracker;
 private:
     std::vector<BucketCopy> _replica_info;
+    uint32_t _max_documents_removed;
 
     void merge_received_bucket_info_into_db();
+    void update_gc_metrics();
 };
 
 }
diff --git a/storage/src/vespa/storage/distributor/persistence_operation_metric_set.h b/storage/src/vespa/storage/distributor/persistence_operation_metric_set.h
index d951d7ceba2..1299fdad2ad 100644
--- a/storage/src/vespa/storage/distributor/persistence_operation_metric_set.h
+++ b/storage/src/vespa/storage/distributor/persistence_operation_metric_set.h
@@ -16,7 +16,7 @@ class PersistenceFailuresMetricSet : public metrics::MetricSet
 {
 public:
     explicit PersistenceFailuresMetricSet(metrics::MetricSet* owner);
-    ~PersistenceFailuresMetricSet();
+    ~PersistenceFailuresMetricSet() override;
 
     metrics::SumMetric<metrics::LongCountMetric> sum;
     metrics::LongCountMetric notready;
@@ -44,7 +44,7 @@ public:
     PersistenceFailuresMetricSet failures;
 
     PersistenceOperationMetricSet(const std::string& name, metrics::MetricSet* owner = nullptr);
-    ~PersistenceOperationMetricSet();
+    ~PersistenceOperationMetricSet() override;
 
     MetricSet * clone(std::vector<Metric::UP>& ownerList, CopyType copyType,
                       metrics::MetricSet* owner, bool includeUnused) const override;
diff --git a/storage/src/vespa/storage/persistence/processallhandler.cpp b/storage/src/vespa/storage/persistence/processallhandler.cpp
index 8c951a9f50d..5b94a3da027 100644
--- a/storage/src/vespa/storage/persistence/processallhandler.cpp
+++ b/storage/src/vespa/storage/persistence/processallhandler.cpp
@@ -23,6 +23,7 @@ public:
     spi::PersistenceProvider& _provider;
     const spi::Bucket& _bucket;
     spi::Context& _context;
+    uint32_t _n_removed;
 
     UnrevertableRemoveEntryProcessor(
             spi::PersistenceProvider& provider,
@@ -30,7 +31,9 @@ public:
             spi::Context& context)
         : _provider(provider),
           _bucket(bucket),
-          _context(context) {}
+          _context(context),
+          _n_removed(0)
+    {}
 
     void process(spi::DocEntry& entry) override {
         spi::RemoveResult removeResult = _provider.remove(
@@ -45,13 +48,14 @@ public:
                << removeResult.getErrorMessage();
             throw std::runtime_error(ss.str());
         }
+        ++_n_removed;
     }
 };
 
 class StatEntryProcessor : public BucketProcessor::EntryProcessor {
 public:
     std::ostream& ost;
-    StatEntryProcessor(std::ostream& o)
+    explicit StatEntryProcessor(std::ostream& o)
         : ost(o) {};
 
     void process(spi::DocEntry& e) override {
@@ -97,7 +101,9 @@ ProcessAllHandler::handleRemoveLocation(api::RemoveLocationCommand& cmd,
                                 context);
     spi::Result result = _spi.flush(bucket, context);
     uint32_t code = _env.convertErrorCode(result);
-    if (code != 0) {
+    if (code == 0) {
+        tracker->setReply(std::make_shared<api::RemoveLocationReply>(cmd, processor._n_removed));
+    } else {
         tracker->fail(code, result.getErrorMessage());
     }
 
diff --git a/storageapi/src/tests/mbusprot/storageprotocoltest.cpp b/storageapi/src/tests/mbusprot/storageprotocoltest.cpp
index 2f959e40e2a..2e5eb115844 100644
--- a/storageapi/src/tests/mbusprot/storageprotocoltest.cpp
+++ b/storageapi/src/tests/mbusprot/storageprotocoltest.cpp
@@ -522,8 +522,15 @@ TEST_P(StorageProtocolTest, remove_location) {
     EXPECT_EQ("id.group == \"mygroup\"", cmd2->getDocumentSelection());
     EXPECT_EQ(_bucket, cmd2->getBucket());
 
-    auto reply = std::make_shared<RemoveLocationReply>(*cmd2);
+    uint32_t n_docs_removed = 12345;
+    auto reply = std::make_shared<RemoveLocationReply>(*cmd2, n_docs_removed);
     auto reply2 = copyReply(reply);
+    if (GetParam().getMajor() == 7) {
+        // Statistics are only available for protobuf-enabled version.
+        EXPECT_EQ(n_docs_removed, reply2->documents_removed());
+    } else {
+        EXPECT_EQ(0, reply2->documents_removed());
+    }
 }
 
 TEST_P(StorageProtocolTest, create_visitor) {
diff --git a/storageapi/src/vespa/storageapi/mbusprot/protobuf/feed.proto b/storageapi/src/vespa/storageapi/mbusprot/protobuf/feed.proto
index 810f88f588f..12dbaf59146 100644
--- a/storageapi/src/vespa/storageapi/mbusprot/protobuf/feed.proto
+++ b/storageapi/src/vespa/storageapi/mbusprot/protobuf/feed.proto
@@ -90,7 +90,12 @@ message RemoveLocationRequest {
     bytes  document_selection = 2;
 }
 
+message RemoveLocationStats {
+    uint32 documents_removed = 1;
+}
+
 message RemoveLocationResponse {
     BucketInfo bucket_info        = 1;
     BucketId   remapped_bucket_id = 2;
+    RemoveLocationStats stats     = 3;
 }
diff --git a/storageapi/src/vespa/storageapi/mbusprot/protocolserialization7.cpp b/storageapi/src/vespa/storageapi/mbusprot/protocolserialization7.cpp
index 9751fd1be98..90c8d1c7d2a 100644
--- a/storageapi/src/vespa/storageapi/mbusprot/protocolserialization7.cpp
+++ b/storageapi/src/vespa/storageapi/mbusprot/protocolserialization7.cpp
@@ -643,7 +643,9 @@ void ProtocolSerialization7::onEncode(GBBuf& buf, const api::RemoveLocationComma
 }
 
 void ProtocolSerialization7::onEncode(GBBuf& buf, const api::RemoveLocationReply& msg) const {
-    encode_bucket_info_response<protobuf::RemoveLocationResponse>(buf, msg, no_op_encode);
+    encode_bucket_info_response<protobuf::RemoveLocationResponse>(buf, msg, [&](auto& res) {
+        res.mutable_stats()->set_documents_removed(msg.documents_removed());
+    });
 }
 
 api::StorageCommand::UP ProtocolSerialization7::onDecodeRemoveLocationCommand(BBuf& buf) const {
@@ -653,8 +655,11 @@ api::StorageCommand::UP ProtocolSerialization7::onDecodeRemoveLocationCommand(BB
 }
 
 api::StorageReply::UP ProtocolSerialization7::onDecodeRemoveLocationReply(const SCmd& cmd, BBuf& buf) const {
-    return decode_bucket_info_response<protobuf::RemoveLocationResponse>(buf, [&]([[maybe_unused]] auto& res) {
-        return std::make_unique<api::RemoveLocationReply>(static_cast<const api::RemoveLocationCommand&>(cmd));
+    return decode_bucket_info_response<protobuf::RemoveLocationResponse>(buf, [&](auto& res) {
+        uint32_t documents_removed = (res.has_stats() ? res.stats().documents_removed() : 0u);
+        return std::make_unique<api::RemoveLocationReply>(
+                static_cast<const api::RemoveLocationCommand&>(cmd),
+                documents_removed);
     });
 }
 
diff --git a/storageapi/src/vespa/storageapi/message/removelocation.cpp b/storageapi/src/vespa/storageapi/message/removelocation.cpp
index b53584601ef..49c9d22f5ee 100644
--- a/storageapi/src/vespa/storageapi/message/removelocation.cpp
+++ b/storageapi/src/vespa/storageapi/message/removelocation.cpp
@@ -25,8 +25,9 @@ RemoveLocationCommand::print(std::ostream& out, bool verbose, const std::string&
     BucketInfoCommand::print(out, verbose, indent);
 }
 
-RemoveLocationReply::RemoveLocationReply(const RemoveLocationCommand& cmd)
-    : BucketInfoReply(cmd)
+RemoveLocationReply::RemoveLocationReply(const RemoveLocationCommand& cmd, uint32_t docs_removed)
+    : BucketInfoReply(cmd),
+      _documents_removed(docs_removed)
 {
 }
 
diff --git a/storageapi/src/vespa/storageapi/message/removelocation.h b/storageapi/src/vespa/storageapi/message/removelocation.h
index 46555497035..812cc8c413b 100644
--- a/storageapi/src/vespa/storageapi/message/removelocation.h
+++ b/storageapi/src/vespa/storageapi/message/removelocation.h
@@ -11,7 +11,7 @@ class RemoveLocationCommand : public BucketInfoCommand
 {
 public:
     RemoveLocationCommand(vespalib::stringref documentSelection, const document::Bucket &bucket);
-    ~RemoveLocationCommand();
+    ~RemoveLocationCommand() override;
 
     void print(std::ostream& out, bool verbose, const std::string& indent) const override;
     const vespalib::string& getDocumentSelection() const { return _documentSelection; }
@@ -22,8 +22,13 @@ private:
 
 class RemoveLocationReply : public BucketInfoReply
 {
+    uint32_t _documents_removed;
 public:
-    RemoveLocationReply(const RemoveLocationCommand& cmd);
+    explicit RemoveLocationReply(const RemoveLocationCommand& cmd, uint32_t docs_removed = 0);
+    void set_documents_removed(uint32_t docs_removed) noexcept {
+        _documents_removed = docs_removed;
+    }
+    uint32_t documents_removed() const noexcept { return _documents_removed; }
     DECLARE_STORAGEREPLY(RemoveLocationReply, onRemoveLocationReply)
 };