Report bucket spaces statistics in distributor host info.

6 files changed, 234 insertions, 65 deletions

diff --git a/storage/src/tests/distributor/distributor_host_info_reporter_test.cpp b/storage/src/tests/distributor/distributor_host_info_reporter_test.cpp
index 2617c843912..14df6ee5016 100644
--- a/storage/src/tests/distributor/distributor_host_info_reporter_test.cpp
+++ b/storage/src/tests/distributor/distributor_host_info_reporter_test.cpp
@@ -9,10 +9,11 @@
 #include <vespa/vespalib/testkit/test_kit.h>
 #include <tests/common/hostreporter/util.h>
 #include <vespa/vespalib/stllike/asciistream.h>
+#include <vespa/storage/distributor/bucket_spaces_stats_provider.h>
 
-namespace storage {
-namespace distributor {
+namespace storage::distributor {
 
+using PerNodeBucketSpacesStats = BucketSpacesStatsProvider::PerNodeBucketSpacesStats;
 using End = vespalib::JsonStream::End;
 using File = vespalib::File;
 using Object = vespalib::JsonStream::Object;
@@ -24,17 +25,26 @@ class DistributorHostInfoReporterTest : public CppUnit::TestFixture
     CPPUNIT_TEST(hostInfoAllInfo);
     CPPUNIT_TEST(generateExampleJson);
     CPPUNIT_TEST(noReportGeneratedIfDisabled);
+    CPPUNIT_TEST(bucket_spaces_stats_are_reported);
     CPPUNIT_TEST_SUITE_END();
 
     void hostInfoWithPutLatenciesOnly();
     void hostInfoAllInfo();
-    void verifyReportedNodeLatencies(
-            const vespalib::Slime& root,
-            uint16_t node,
-            int64_t latencySum,
-            int64_t count);
+    void verifyReportedNodeLatencies(const vespalib::Slime& root,
+                                     uint16_t nodeIndex,
+                                     int64_t latencySum,
+                                     int64_t count);
+    void verifyBucketSpaceStats(const vespalib::Slime& root,
+                                uint16_t nodeIndex,
+                                const vespalib::string& bucketSpaceName,
+                                size_t bucketsTotal,
+                                size_t bucketsPending);
+    void verifyBucketSpaceStats(const vespalib::Slime& root,
+                                uint16_t nodeIndex,
+                                const vespalib::string& bucketSpaceName);
     void generateExampleJson();
     void noReportGeneratedIfDisabled();
+    void bucket_spaces_stats_are_reported();
 };
 
 CPPUNIT_TEST_SUITE_REGISTRATION(DistributorHostInfoReporterTest);
@@ -71,6 +81,14 @@ struct MockedMinReplicaProvider : MinReplicaProvider
     }
 };
 
+struct MockedBucketSpacesStatsProvider : public BucketSpacesStatsProvider {
+    PerNodeBucketSpacesStats stats;
+
+    PerNodeBucketSpacesStats getBucketSpacesStats() const override {
+        return stats;
+    }
+};
+
 const vespalib::slime::Inspector&
 getNode(const vespalib::Slime& root, uint16_t nodeIndex)
 {
@@ -97,37 +115,80 @@ getLatenciesForNode(const vespalib::Slime& root, uint16_t nodeIndex)
     return getNode(root, nodeIndex)["ops-latency"];
 }
 
-} // anon ns
+const vespalib::slime::Inspector&
+getBucketSpaceStats(const vespalib::Slime& root, uint16_t nodeIndex, const vespalib::string& bucketSpaceName)
+{
+    const auto& bucketSpaces = getNode(root, nodeIndex)["bucket-spaces"];
+    for (size_t i = 0; i < bucketSpaces.entries(); ++i) {
+        if (bucketSpaces[i]["name"].asString().make_stringref() == bucketSpaceName) {
+            return bucketSpaces[i];
+        }
+    }
+    throw std::runtime_error("No bucket space found with name " + bucketSpaceName);
+}
+
+}
 
 void
-DistributorHostInfoReporterTest::verifyReportedNodeLatencies(
-        const vespalib::Slime& root,
-        uint16_t node,
-        int64_t latencySum,
-        int64_t count)
+DistributorHostInfoReporterTest::verifyReportedNodeLatencies(const vespalib::Slime& root,
+                                                             uint16_t nodeIndex,
+                                                             int64_t latencySum,
+                                                             int64_t count)
 {
-    auto& latencies = getLatenciesForNode(root, node);
+    auto& latencies = getLatenciesForNode(root, nodeIndex);
     CPPUNIT_ASSERT_EQUAL(latencySum,
                          latencies["put"]["latency-ms-sum"].asLong());
     CPPUNIT_ASSERT_EQUAL(count, latencies["put"]["count"].asLong());
 }
 
 void
-DistributorHostInfoReporterTest::hostInfoWithPutLatenciesOnly()
+DistributorHostInfoReporterTest::verifyBucketSpaceStats(const vespalib::Slime& root,
+                                                        uint16_t nodeIndex,
+                                                        const vespalib::string& bucketSpaceName,
+                                                        size_t bucketsTotal,
+                                                        size_t bucketsPending)
+{
+    const auto &stats = getBucketSpaceStats(root, nodeIndex, bucketSpaceName);
+    CPPUNIT_ASSERT_EQUAL(bucketsTotal, static_cast<size_t>(stats["total"].asLong()));
+    CPPUNIT_ASSERT_EQUAL(bucketsPending, static_cast<size_t>(stats["pending"].asLong()));
+}
+
+void
+DistributorHostInfoReporterTest::verifyBucketSpaceStats(const vespalib::Slime& root,
+                                                        uint16_t nodeIndex,
+                                                        const vespalib::string& bucketSpaceName)
 {
+    const auto &stats = getBucketSpaceStats(root, nodeIndex, bucketSpaceName);
+    CPPUNIT_ASSERT(!stats["total"].valid());
+    CPPUNIT_ASSERT(!stats["pending"].valid());
+}
+
+struct Fixture {
     MockedLatencyStatisticsProvider latencyStatsProvider;
     MockedMinReplicaProvider minReplicaProvider;
-    DistributorHostInfoReporter reporter(latencyStatsProvider,
-                                         minReplicaProvider);
+    MockedBucketSpacesStatsProvider bucketSpacesStatsProvider;
+    DistributorHostInfoReporter reporter;
+    Fixture()
+        : latencyStatsProvider(),
+          minReplicaProvider(),
+          bucketSpacesStatsProvider(),
+          reporter(latencyStatsProvider, minReplicaProvider, bucketSpacesStatsProvider)
+    {}
+    ~Fixture() {}
+};
 
+void
+DistributorHostInfoReporterTest::hostInfoWithPutLatenciesOnly()
+{
+    Fixture f;
     NodeStatsSnapshot snapshot;
     snapshot.nodeToStats[0] = { makeOpStats(ms(10000), 3) };
     snapshot.nodeToStats[5] = { makeOpStats(ms(25000), 7) };
 
-    latencyStatsProvider.returnedSnapshot = snapshot;
+    f.latencyStatsProvider.returnedSnapshot = snapshot;
 
     vespalib::Slime root;
-    util::reporterToSlime(reporter, root);
+    util::reporterToSlime(f.reporter, root);
     verifyReportedNodeLatencies(root, 0, 10000, 3);
     verifyReportedNodeLatencies(root, 5, 25000, 7);
 }
@@ -135,23 +196,19 @@ DistributorHostInfoReporterTest::hostInfoWithPutLatenciesOnly()
 void
 DistributorHostInfoReporterTest::hostInfoAllInfo()
 {
-    MockedLatencyStatisticsProvider latencyStatsProvider;
-    MockedMinReplicaProvider minReplicaProvider;
-    DistributorHostInfoReporter reporter(latencyStatsProvider,
-                                         minReplicaProvider);
-
+    Fixture f;
     NodeStatsSnapshot latencySnapshot;
     latencySnapshot.nodeToStats[0] = { makeOpStats(ms(10000), 3) };
     latencySnapshot.nodeToStats[5] = { makeOpStats(ms(25000), 7) };
-    latencyStatsProvider.returnedSnapshot = latencySnapshot;
+    f.latencyStatsProvider.returnedSnapshot = latencySnapshot;
 
     std::unordered_map<uint16_t, uint32_t> minReplica;
     minReplica[0] = 2;
     minReplica[5] = 9;
-    minReplicaProvider.minReplica = minReplica;
+    f.minReplicaProvider.minReplica = minReplica;
 
     vespalib::Slime root;
-    util::reporterToSlime(reporter, root);
+    util::reporterToSlime(f.reporter, root);
     verifyReportedNodeLatencies(root, 0, 10000, 3);
     verifyReportedNodeLatencies(root, 5, 25000, 7);
 
@@ -162,26 +219,28 @@ DistributorHostInfoReporterTest::hostInfoAllInfo()
 void
 DistributorHostInfoReporterTest::generateExampleJson()
 {
-    MockedLatencyStatisticsProvider latencyStatsProvider;
-    MockedMinReplicaProvider minReplicaProvider;
-    DistributorHostInfoReporter reporter(latencyStatsProvider,
-                                         minReplicaProvider);
-
+    Fixture f;
     NodeStatsSnapshot snapshot;
     snapshot.nodeToStats[0] = { makeOpStats(ms(10000), 3) };
     snapshot.nodeToStats[5] = { makeOpStats(ms(25000), 7) };
-    latencyStatsProvider.returnedSnapshot = snapshot;
+    f.latencyStatsProvider.returnedSnapshot = snapshot;
 
     std::unordered_map<uint16_t, uint32_t> minReplica;
     minReplica[0] = 2;
     minReplica[5] = 9;
-    minReplicaProvider.minReplica = minReplica;
+    f.minReplicaProvider.minReplica = minReplica;
+
+    PerNodeBucketSpacesStats stats;
+    stats[0]["default"] = BucketSpaceStats(11, 3);
+    stats[0]["global"] = BucketSpaceStats(13, 5);
+    stats[5]["default"] = BucketSpaceStats(17, 7);
+    f.bucketSpacesStatsProvider.stats = stats;
 
     vespalib::asciistream json;
     vespalib::JsonStream stream(json, true);
 
     stream << Object();
-    reporter.report(stream);
+    f.reporter.report(stream);
     stream << End();
     stream.finalize();
 
@@ -204,23 +263,46 @@ DistributorHostInfoReporterTest::generateExampleJson()
 void
 DistributorHostInfoReporterTest::noReportGeneratedIfDisabled()
 {
-    MockedLatencyStatisticsProvider latencyStatsProvider;
-    MockedMinReplicaProvider minReplicaProvider;
-    DistributorHostInfoReporter reporter(latencyStatsProvider,
-                                         minReplicaProvider);
-    reporter.enableReporting(false);
+    Fixture f;
+    f.reporter.enableReporting(false);
 
     NodeStatsSnapshot snapshot;
     snapshot.nodeToStats[0] = { makeOpStats(ms(10000), 3) };
     snapshot.nodeToStats[5] = { makeOpStats(ms(25000), 7) };
 
-    latencyStatsProvider.returnedSnapshot = snapshot;
+    f.latencyStatsProvider.returnedSnapshot = snapshot;
 
     vespalib::Slime root;
-    util::reporterToSlime(reporter, root);
+    util::reporterToSlime(f.reporter, root);
     CPPUNIT_ASSERT_EQUAL(size_t(0), root.get().children());
 }
 
-} // distributor
-} // storage
+void
+DistributorHostInfoReporterTest::bucket_spaces_stats_are_reported()
+{
+    Fixture f;
+    PerNodeBucketSpacesStats stats;
+    stats[1]["default"] = BucketSpaceStats(11, 3);
+    stats[1]["global"] = BucketSpaceStats(13, 5);
+    stats[2]["default"] = BucketSpaceStats(17, 7);
+    stats[2]["global"] = BucketSpaceStats();
+    stats[3]["default"] = BucketSpaceStats(19, 11);
+    f.bucketSpacesStatsProvider.stats = stats;
+
+    vespalib::Slime root;
+    util::reporterToSlime(f.reporter, root);
+    verifyBucketSpaceStats(root, 1, "default", 11, 3);
+    verifyBucketSpaceStats(root, 1, "global", 13, 5);
+    verifyBucketSpaceStats(root, 2, "default", 17, 7);
+    verifyBucketSpaceStats(root, 2, "global");
+    verifyBucketSpaceStats(root, 3, "default", 19, 11);
+    try {
+        verifyBucketSpaceStats(root, 3, "global");
+        CPPUNIT_ASSERT(false);
+    } catch (const std::runtime_error &ex) {
+        CPPUNIT_ASSERT("No bucket space found with name global" == vespalib::string(ex.what()));
+    }
+}
+
+}
 
diff --git a/storage/src/vespa/storage/distributor/bucket_spaces_stats_provider.h b/storage/src/vespa/storage/distributor/bucket_spaces_stats_provider.h
new file mode 100644
index 00000000000..aeba827a976
--- /dev/null
+++ b/storage/src/vespa/storage/distributor/bucket_spaces_stats_provider.h
@@ -0,0 +1,48 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+#pragma once
+
+#include <vespa/vespalib/stllike/string.h>
+#include <map>
+#include <unordered_map>
+
+namespace storage::distributor {
+
+/**
+ * Statistics for a single bucket space on a content node.
+ */
+class BucketSpaceStats {
+private:
+    bool   _valid;
+    size_t _bucketsTotal;
+    size_t _bucketsPending;
+public:
+    BucketSpaceStats(size_t bucketsTotal_, size_t bucketsPending_)
+        : _valid(true),
+          _bucketsTotal(bucketsTotal_),
+          _bucketsPending(bucketsPending_)
+    {}
+    BucketSpaceStats()
+        : _valid(false),
+          _bucketsTotal(0),
+          _bucketsPending(0)
+    {}
+    bool valid() const { return _valid; }
+    size_t bucketsTotal() const { return _bucketsTotal; }
+    size_t bucketsPending() const { return _bucketsPending; }
+};
+
+/**
+ * Interface that provides snapshots of bucket spaces statistics per content node.
+ */
+class BucketSpacesStatsProvider {
+public:
+    // Mapping from bucket space name to statistics for that bucket space.
+    using BucketSpacesStats = std::map<vespalib::string, BucketSpaceStats>;
+    // Mapping from content node index to statistics for all bucket spaces on that node.
+    using PerNodeBucketSpacesStats = std::unordered_map<uint16_t, BucketSpacesStats>;
+
+    virtual ~BucketSpacesStatsProvider() {}
+    virtual PerNodeBucketSpacesStats getBucketSpacesStats() const = 0;
+};
+
+}
diff --git a/storage/src/vespa/storage/distributor/distributor.cpp b/storage/src/vespa/storage/distributor/distributor.cpp
index a559ce2ad1a..917ec59cba4 100644
--- a/storage/src/vespa/storage/distributor/distributor.cpp
+++ b/storage/src/vespa/storage/distributor/distributor.cpp
@@ -94,7 +94,7 @@ Distributor::Distributor(DistributorComponentRegister& compReg,
       _metricLock(),
       _maintenanceStats(),
       _bucketDbStats(),
-      _hostInfoReporter(_pendingMessageTracker.getLatencyStatisticsProvider(), *this),
+      _hostInfoReporter(_pendingMessageTracker.getLatencyStatisticsProvider(), *this, *this),
       _ownershipSafeTimeCalc(
             std::make_unique<OwnershipTransferSafeTimePointCalculator>(
                 std::chrono::seconds(0))) // Set by config later
diff --git a/storage/src/vespa/storage/distributor/distributor.h b/storage/src/vespa/storage/distributor/distributor.h
index 39695b26415..1cf0a4f1866 100644
--- a/storage/src/vespa/storage/distributor/distributor.h
+++ b/storage/src/vespa/storage/distributor/distributor.h
@@ -2,27 +2,27 @@
 
 #pragma once
 
-#include "idealstatemanager.h"
+#include "bucket_spaces_stats_provider.h"
 #include "bucketdbupdater.h"
-#include "pendingmessagetracker.h"
+#include "distributor_host_info_reporter.h"
+#include "distributorinterface.h"
 #include "externaloperationhandler.h"
+#include "idealstatemanager.h"
 #include "min_replica_provider.h"
-#include "distributorinterface.h"
-
+#include "pendingmessagetracker.h"
 #include "statusreporterdelegate.h"
-#include "distributor_host_info_reporter.h"
-#include <vespa/storage/distributor/maintenance/maintenancescheduler.h>
-#include <vespa/storage/distributor/bucketdb/bucketdbmetricupdater.h>
+#include <vespa/config/config.h>
 #include <vespa/storage/common/distributorcomponent.h>
 #include <vespa/storage/common/doneinitializehandler.h>
 #include <vespa/storage/common/messagesender.h>
+#include <vespa/storage/distributor/bucketdb/bucketdbmetricupdater.h>
+#include <vespa/storage/distributor/maintenance/maintenancescheduler.h>
 #include <vespa/storageapi/message/state.h>
-#include <vespa/storageframework/generic/thread/tickingthread.h>
 #include <vespa/storageframework/generic/metric/metricupdatehook.h>
-#include <vespa/config/config.h>
+#include <vespa/storageframework/generic/thread/tickingthread.h>
 #include <vespa/vespalib/util/sync.h>
-#include <unordered_map>
 #include <queue>
+#include <unordered_map>
 
 namespace storage {
 
@@ -43,7 +43,8 @@ class Distributor : public StorageLink,
                     public StatusDelegator,
                     public framework::StatusReporter,
                     public framework::TickingThread,
-                    public MinReplicaProvider
+                    public MinReplicaProvider,
+                    public BucketSpacesStatsProvider
 {
 public:
     Distributor(DistributorComponentRegister&,
@@ -197,6 +198,11 @@ private:
      */
     std::unordered_map<uint16_t, uint32_t> getMinReplica() const override;
 
+    PerNodeBucketSpacesStats getBucketSpacesStats() const override {
+        // TODO: implement
+        return BucketSpacesStatsProvider::PerNodeBucketSpacesStats();
+    }
+
     /**
      * Atomically publish internal metrics to external ideal state metrics.
      * Takes metric lock.
diff --git a/storage/src/vespa/storage/distributor/distributor_host_info_reporter.cpp b/storage/src/vespa/storage/distributor/distributor_host_info_reporter.cpp
index 8ac471ee0b7..5929f02c04b 100644
--- a/storage/src/vespa/storage/distributor/distributor_host_info_reporter.cpp
+++ b/storage/src/vespa/storage/distributor/distributor_host_info_reporter.cpp
@@ -1,5 +1,6 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 
+#include "bucket_spaces_stats_provider.h"
 #include "distributor_host_info_reporter.h"
 #include "min_replica_provider.h"
 #include "pendingmessagetracker.h"
@@ -12,15 +13,19 @@ using std::unordered_map;
 namespace storage {
 namespace distributor {
 
+using BucketSpacesStats = BucketSpacesStatsProvider::BucketSpacesStats;
+using PerNodeBucketSpacesStats = BucketSpacesStatsProvider::PerNodeBucketSpacesStats;
 using Object = vespalib::JsonStream::Object;
 using Array = vespalib::JsonStream::Array;
 using End = vespalib::JsonStream::End;
 
 DistributorHostInfoReporter::DistributorHostInfoReporter(
         LatencyStatisticsProvider& latencyProvider,
-        MinReplicaProvider& minReplicaProvider)
+        MinReplicaProvider& minReplicaProvider,
+        BucketSpacesStatsProvider& bucketSpacesStatsProvider)
     : _latencyProvider(latencyProvider),
       _minReplicaProvider(minReplicaProvider),
+      _bucketSpacesStatsProvider(bucketSpacesStatsProvider),
       _enabled(true)
 {
 }
@@ -38,15 +43,33 @@ writeOperationStats(vespalib::JsonStream& stream,
 }
 
 void
+writeBucketSpacesStats(vespalib::JsonStream& stream,
+                       const BucketSpacesStats& stats)
+{
+    for (const auto& elem : stats) {
+        stream << Object() << "name" << elem.first;
+        if (elem.second.valid()) {
+            stream << "total" << elem.second.bucketsTotal()
+                   << "pending" << elem.second.bucketsPending();
+        }
+        stream << End();
+    }
+}
+
+void
 outputStorageNodes(vespalib::JsonStream& output,
                    const unordered_map<uint16_t, NodeStats>& nodeStats,
-                   const unordered_map<uint16_t, uint32_t>& minReplica)
+                   const unordered_map<uint16_t, uint32_t>& minReplica,
+                   const PerNodeBucketSpacesStats& bucketSpacesStats)
 {
     set<uint16_t> nodes;
-    for (auto& element : nodeStats) {
+    for (const auto& element : nodeStats) {
         nodes.insert(element.first);
     }
-    for (auto& element : minReplica) {
+    for (const auto& element : minReplica) {
+        nodes.insert(element.first);
+    }
+    for (const auto& element : bucketSpacesStats) {
         nodes.insert(element.first);
     }
     
@@ -69,6 +92,13 @@ outputStorageNodes(vespalib::JsonStream& output,
                 output << "min-current-replication-factor"
                        << minReplicaIt->second;
             }
+
+            auto bucketSpacesStatsIt = bucketSpacesStats.find(node);
+            if (bucketSpacesStatsIt != bucketSpacesStats.end()) {
+                output << "bucket-spaces" << Array();
+                writeBucketSpacesStats(output, bucketSpacesStatsIt->second);
+                output << End();
+            }
         }
         output << End();
     }
@@ -83,15 +113,15 @@ DistributorHostInfoReporter::report(vespalib::JsonStream& output)
         return;
     }
 
-    NodeStatsSnapshot nodeStats = _latencyProvider.getLatencyStatistics();
-    std::unordered_map<uint16_t, uint32_t> minReplica =
-        _minReplicaProvider.getMinReplica();
+    auto nodeStats = _latencyProvider.getLatencyStatistics();
+    auto minReplica = _minReplicaProvider.getMinReplica();
+    auto bucketSpacesStats = _bucketSpacesStatsProvider.getBucketSpacesStats();
 
     output << "distributor" << Object();
     {
         output << "storage-nodes" << Array();
 
-        outputStorageNodes(output, nodeStats.nodeToStats, minReplica);
+        outputStorageNodes(output, nodeStats.nodeToStats, minReplica, bucketSpacesStats);
 
         output << End();
     }
diff --git a/storage/src/vespa/storage/distributor/distributor_host_info_reporter.h b/storage/src/vespa/storage/distributor/distributor_host_info_reporter.h
index 3cb878fc75c..3e6a02120c2 100644
--- a/storage/src/vespa/storage/distributor/distributor_host_info_reporter.h
+++ b/storage/src/vespa/storage/distributor/distributor_host_info_reporter.h
@@ -7,6 +7,7 @@
 namespace storage {
 namespace distributor {
 
+class BucketSpacesStatsProvider;
 class LatencyStatisticsProvider;
 class MinReplicaProvider;
 struct OperationStats;
@@ -15,7 +16,8 @@ class DistributorHostInfoReporter : public HostReporter
 {
 public:
     DistributorHostInfoReporter(LatencyStatisticsProvider& latencyProvider,
-                                MinReplicaProvider& minReplicaProvider);
+                                MinReplicaProvider& minReplicaProvider,
+                                BucketSpacesStatsProvider& bucketSpacesStatsProvider);
 
     DistributorHostInfoReporter(const DistributorHostInfoReporter&) = delete;
     DistributorHostInfoReporter& operator=(
@@ -43,6 +45,7 @@ public:
 private:
     LatencyStatisticsProvider& _latencyProvider;
     MinReplicaProvider& _minReplicaProvider;
+    BucketSpacesStatsProvider& _bucketSpacesStatsProvider;
     std::atomic<bool> _enabled;
 };