Add feed block status to ClusterStateBundle in C++.

3 files changed, 111 insertions, 6 deletions

diff --git a/vdslib/src/tests/state/cluster_state_bundle_test.cpp b/vdslib/src/tests/state/cluster_state_bundle_test.cpp
index bfd83673442..75a37218913 100644
--- a/vdslib/src/tests/state/cluster_state_bundle_test.cpp
+++ b/vdslib/src/tests/state/cluster_state_bundle_test.cpp
@@ -35,13 +35,20 @@ TEST(ClusterStateBundleTest, baseline_state_is_returned_if_bucket_space_is_not_f
 }
 
 ClusterStateBundle
-makeBundle(const vespalib::string &baselineState, const std::map<BucketSpace, vespalib::string> &derivedStates)
+makeBundle(const vespalib::string &baselineState, const std::map<BucketSpace, vespalib::string> &derivedStates,
+           bool deferred_activation = false)
 {
     ClusterStateBundle::BucketSpaceStateMapping derivedBucketSpaceStates;
     for (const auto &entry : derivedStates) {
         derivedBucketSpaceStates[entry.first] = std::make_shared<const ClusterState>(entry.second);
     }
-    return ClusterStateBundle(ClusterState(baselineState), std::move(derivedBucketSpaceStates));
+    return ClusterStateBundle(ClusterState(baselineState), std::move(derivedBucketSpaceStates), deferred_activation);
+}
+
+ClusterStateBundle
+bundle_with_feed_block(const ClusterStateBundle::FeedBlock& feed_block)
+{
+    return ClusterStateBundle(ClusterState("storage:2"), {}, feed_block, false);
 }
 
 TEST(ClusterStateBundleTest, verify_equality_operator)
@@ -51,9 +58,40 @@ TEST(ClusterStateBundleTest, verify_equality_operator)
     EXPECT_NE(f.bundle, makeBundle("storage:2", {}));
     EXPECT_NE(f.bundle, makeBundle("storage:2", {{BucketSpace(1), "storage:2 .0.s:m"}}));
     EXPECT_NE(f.bundle, makeBundle("storage:2", {{BucketSpace(2), "storage:2 .1.s:m"}}));
+    EXPECT_NE(f.bundle, makeBundle("storage:2", {{BucketSpace(1), "storage:2 .1.s:m"}}, true));
 
     EXPECT_EQ(f.bundle, makeBundle("storage:2", {{BucketSpace(1), "storage:2 .1.s:m"}}));
 }
 
+TEST(ClusterStateBundleTest, feed_block_state_is_available)
+{
+    auto non_blocking = makeBundle("storage:2", {});
+    auto blocking = bundle_with_feed_block({true, "foo"});
+
+    EXPECT_FALSE(non_blocking.block_feed_in_cluster());
+    EXPECT_FALSE(non_blocking.feed_block().has_value());
+
+    EXPECT_TRUE(blocking.block_feed_in_cluster());
+    EXPECT_TRUE(blocking.feed_block().has_value());
+    EXPECT_TRUE(blocking.feed_block()->block_feed_in_cluster());
+    EXPECT_EQ("foo", blocking.feed_block()->description());
+}
+
+TEST(ClusterStateBundleTest, equality_operator_considers_feed_block)
+{
+    EXPECT_NE(bundle_with_feed_block({true, "foo"}), bundle_with_feed_block({false, "foo"}));
+    EXPECT_NE(bundle_with_feed_block({true, "foo"}), bundle_with_feed_block({true, "bar"}));
+    EXPECT_NE(makeBundle("storage:2", {}), bundle_with_feed_block({false, "bar"}));
+
+    EXPECT_EQ(bundle_with_feed_block({true, "foo"}), bundle_with_feed_block({true, "foo"}));
+    EXPECT_EQ(bundle_with_feed_block({false, "foo"}), bundle_with_feed_block({false, "foo"}));
+}
+
+TEST(ClusterStateBundleTest, toString_with_feed_block_includes_description)
+{
+    EXPECT_EQ("ClusterStateBundle('storage:2', feed blocked: 'full disk')",
+              bundle_with_feed_block({true, "full disk"}).toString());
+}
+
 }
 
diff --git a/vdslib/src/vespa/vdslib/state/cluster_state_bundle.cpp b/vdslib/src/vespa/vdslib/state/cluster_state_bundle.cpp
index 198ef4e54d1..803a0e29018 100644
--- a/vdslib/src/vespa/vdslib/state/cluster_state_bundle.cpp
+++ b/vdslib/src/vespa/vdslib/state/cluster_state_bundle.cpp
@@ -8,8 +8,24 @@
 
 namespace storage::lib {
 
+ClusterStateBundle::FeedBlock::FeedBlock(bool block_feed_in_cluster_in,
+                                         const vespalib::string& description_in)
+    : _block_feed_in_cluster(block_feed_in_cluster_in),
+      _description(description_in)
+{
+}
+
+bool
+ClusterStateBundle::FeedBlock::operator==(const FeedBlock& rhs) const
+{
+    return (_block_feed_in_cluster == rhs._block_feed_in_cluster) &&
+            (_description == rhs._description);
+}
+
 ClusterStateBundle::ClusterStateBundle(const ClusterState &baselineClusterState)
     : _baselineClusterState(std::make_shared<const ClusterState>(baselineClusterState)),
+      _derivedBucketSpaceStates(),
+      _feed_block(),
       _deferredActivation(false)
 {
 }
@@ -18,6 +34,7 @@ ClusterStateBundle::ClusterStateBundle(const ClusterState& baselineClusterState,
                                        BucketSpaceStateMapping derivedBucketSpaceStates)
     : _baselineClusterState(std::make_shared<const ClusterState>(baselineClusterState)),
       _derivedBucketSpaceStates(std::move(derivedBucketSpaceStates)),
+      _feed_block(),
       _deferredActivation(false)
 {
 }
@@ -25,9 +42,21 @@ ClusterStateBundle::ClusterStateBundle(const ClusterState& baselineClusterState,
 ClusterStateBundle::ClusterStateBundle(const ClusterState& baselineClusterState,
                                        BucketSpaceStateMapping derivedBucketSpaceStates,
                                        bool deferredActivation)
-        : _baselineClusterState(std::make_shared<const ClusterState>(baselineClusterState)),
-          _derivedBucketSpaceStates(std::move(derivedBucketSpaceStates)),
-          _deferredActivation(deferredActivation)
+    : _baselineClusterState(std::make_shared<const ClusterState>(baselineClusterState)),
+      _derivedBucketSpaceStates(std::move(derivedBucketSpaceStates)),
+      _feed_block(),
+      _deferredActivation(deferredActivation)
+{
+}
+
+ClusterStateBundle::ClusterStateBundle(const ClusterState& baselineClusterState,
+                                       BucketSpaceStateMapping derivedBucketSpaceStates,
+                                       const FeedBlock& feed_block_in,
+                                       bool deferredActivation)
+    : _baselineClusterState(std::make_shared<const ClusterState>(baselineClusterState)),
+      _derivedBucketSpaceStates(std::move(derivedBucketSpaceStates)),
+      _feed_block(feed_block_in),
+      _deferredActivation(deferredActivation)
 {
 }
 
@@ -69,6 +98,9 @@ ClusterStateBundle::operator==(const ClusterStateBundle &rhs) const
     if (_derivedBucketSpaceStates.size() != rhs._derivedBucketSpaceStates.size()) {
         return false;
     }
+    if (_feed_block != rhs._feed_block) {
+        return false;
+    }
     if (_deferredActivation != rhs._deferredActivation) {
         return false;
     }
@@ -103,6 +135,9 @@ std::ostream& operator<<(std::ostream& os, const ClusterStateBundle& bundle) {
         }
     }
     os << '\'';
+    if (bundle.block_feed_in_cluster()) {
+        os << ", feed blocked: '" << bundle.feed_block()->description() << "'";
+    }
     if (bundle.deferredActivation()) {
         os << " (deferred activation)";
     }
diff --git a/vdslib/src/vespa/vdslib/state/cluster_state_bundle.h b/vdslib/src/vespa/vdslib/state/cluster_state_bundle.h
index 1502335f030..039dca2f455 100644
--- a/vdslib/src/vespa/vdslib/state/cluster_state_bundle.h
+++ b/vdslib/src/vespa/vdslib/state/cluster_state_bundle.h
@@ -3,9 +3,10 @@
 #pragma once
 
 #include <vespa/document/bucket/bucketspace.h>
-#include <unordered_map>
 #include <iosfwd>
+#include <optional>
 #include <string>
+#include <unordered_map>
 
 namespace storage::lib {
 
@@ -18,6 +19,28 @@ class ClusterState;
 class ClusterStateBundle
 {
 public:
+
+    /**
+     * Represents feed blocking status of the entire cluster.
+     *
+     * Feed blocking only applies to client feed, and is turned on
+     * when this object is present and _block_feed_in_cluster is true.
+     * Feed generated by internal maintenance operations (e.g. merging) is not affected.
+     */
+    class FeedBlock {
+    private:
+        bool _block_feed_in_cluster;
+        vespalib::string _description;
+
+    public:
+        FeedBlock(bool block_feed_in_cluster_in,
+                  const vespalib::string& description_in);
+        bool block_feed_in_cluster() const { return _block_feed_in_cluster; }
+        const vespalib::string& description() const { return _description; }
+        bool operator==(const FeedBlock& rhs) const;
+        bool operator!=(const FeedBlock& rhs) const { return !operator==(rhs); }
+    };
+
     using BucketSpaceStateMapping = std::unordered_map<
         document::BucketSpace,
         std::shared_ptr<const ClusterState>,
@@ -25,6 +48,7 @@ public:
     >;
     std::shared_ptr<const ClusterState> _baselineClusterState;
     BucketSpaceStateMapping _derivedBucketSpaceStates;
+    std::optional<FeedBlock> _feed_block;
     bool _deferredActivation;
 public:
     explicit ClusterStateBundle(const ClusterState &baselineClusterState);
@@ -33,6 +57,10 @@ public:
     ClusterStateBundle(const ClusterState& baselineClusterState,
                        BucketSpaceStateMapping derivedBucketSpaceStates,
                        bool deferredActivation);
+    ClusterStateBundle(const ClusterState& baselineClusterState,
+                       BucketSpaceStateMapping derivedBucketSpaceStates,
+                       const FeedBlock& feed_block_in,
+                       bool deferredActivation);
 
     ClusterStateBundle(const ClusterStateBundle&);
     ClusterStateBundle& operator=(const ClusterStateBundle&);
@@ -45,6 +73,10 @@ public:
     const BucketSpaceStateMapping& getDerivedClusterStates() const noexcept {
         return _derivedBucketSpaceStates;
     }
+    bool block_feed_in_cluster() const {
+        return _feed_block.has_value() && _feed_block->block_feed_in_cluster();
+    }
+    const std::optional<FeedBlock>& feed_block() const { return _feed_block; }
     uint32_t getVersion() const;
     bool deferredActivation() const noexcept { return _deferredActivation; }
     std::string toString() const;