Add feed block propagation to ClusterStateBundle in Java

Fully forwards and backwards compatible. Currently only supports indicating
feed blocked status for the entire cluster, with one associated descriptive
message intended to be used by distributors.

4 files changed, 173 insertions, 19 deletions

diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStateBundle.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStateBundle.java
index fc06fef5b30..2f9f67a4b6b 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStateBundle.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStateBundle.java
@@ -28,14 +28,62 @@ public class ClusterStateBundle {
 
     private final AnnotatedClusterState baselineState;
     private final Map<String, AnnotatedClusterState> derivedBucketSpaceStates;
+    private final FeedBlock feedBlock;
     private final boolean deferredActivation;
 
+    /**
+     * Feed blocking status of the entire cluster that will be communicated to the nodes
+     * as part of the cluster state bundle. If not present, or if blockFeedInCluster is
+     * false, feed is not automatically blocked.
+     *
+     * Note that feed blocking only applies to client feed, not to feed generated by internal
+     * maintenance operations such as merging.
+     *
+     * Immutable, so may be safely passed around.
+     */
+    public static class FeedBlock {
+        private final boolean blockFeedInCluster;
+        private final String description;
+
+        public FeedBlock(boolean blockFeedInCluster, String description) {
+            this.blockFeedInCluster = blockFeedInCluster;
+            this.description = description;
+        }
+
+        public static FeedBlock blockedWithDescription(String desc) {
+            return new FeedBlock(true, desc);
+        }
+
+        public boolean blockFeedInCluster() {
+            return blockFeedInCluster;
+        }
+
+        public String getDescription() {
+            return description;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            FeedBlock feedBlock = (FeedBlock) o;
+            return (blockFeedInCluster == feedBlock.blockFeedInCluster &&
+                    Objects.equals(description, feedBlock.description));
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(blockFeedInCluster, description);
+        }
+    }
+
     public static class Builder {
         private final AnnotatedClusterState baselineState;
         private Map<String, AnnotatedClusterState> explicitDerivedStates;
         private ClusterStateDeriver stateDeriver;
         private Set<String> bucketSpaces;
         private boolean deferredActivation = false;
+        private FeedBlock feedBlock = null;
 
         public Builder(AnnotatedClusterState baselineState) {
             this.baselineState = baselineState;
@@ -73,9 +121,14 @@ public class ClusterStateBundle {
             return this;
         }
 
+        public Builder feedBlock(FeedBlock fb) {
+            this.feedBlock = fb;
+            return this;
+        }
+
         public ClusterStateBundle deriveAndBuild() {
             if ((stateDeriver == null || bucketSpaces == null || bucketSpaces.isEmpty()) && explicitDerivedStates == null) {
-                return ClusterStateBundle.ofBaselineOnly(baselineState, deferredActivation);
+                return ClusterStateBundle.ofBaselineOnly(baselineState, feedBlock, deferredActivation);
             }
             Map<String, AnnotatedClusterState> derived;
             if (explicitDerivedStates != null) {
@@ -86,19 +139,21 @@ public class ClusterStateBundle {
                                 Function.identity(),
                                 s -> stateDeriver.derivedFrom(baselineState, s)));
             }
-            return new ClusterStateBundle(baselineState, derived, deferredActivation);
+            return new ClusterStateBundle(baselineState, derived, feedBlock, deferredActivation);
         }
     }
 
     private ClusterStateBundle(AnnotatedClusterState baselineState, Map<String, AnnotatedClusterState> derivedBucketSpaceStates) {
-        this(baselineState, derivedBucketSpaceStates, false);
+        this(baselineState, derivedBucketSpaceStates, null, false);
     }
 
-    private ClusterStateBundle(AnnotatedClusterState baselineState, Map<String,
-                               AnnotatedClusterState> derivedBucketSpaceStates,
+    private ClusterStateBundle(AnnotatedClusterState baselineState,
+                               Map<String, AnnotatedClusterState> derivedBucketSpaceStates,
+                               FeedBlock feedBlock,
                                boolean deferredActivation) {
         this.baselineState = baselineState;
         this.derivedBucketSpaceStates = Collections.unmodifiableMap(derivedBucketSpaceStates);
+        this.feedBlock = feedBlock;
         this.deferredActivation = deferredActivation;
     }
 
@@ -113,11 +168,20 @@ public class ClusterStateBundle {
     public static ClusterStateBundle of(AnnotatedClusterState baselineState,
                                         Map<String, AnnotatedClusterState> derivedBucketSpaceStates,
                                         boolean deferredActivation) {
-        return new ClusterStateBundle(baselineState, derivedBucketSpaceStates, deferredActivation);
+        return new ClusterStateBundle(baselineState, derivedBucketSpaceStates, null, deferredActivation);
+    }
+
+    public static ClusterStateBundle of(AnnotatedClusterState baselineState,
+                                        Map<String, AnnotatedClusterState> derivedBucketSpaceStates,
+                                        FeedBlock feedBlock,
+                                        boolean deferredActivation) {
+        return new ClusterStateBundle(baselineState, derivedBucketSpaceStates, feedBlock, deferredActivation);
     }
 
-    public static ClusterStateBundle ofBaselineOnly(AnnotatedClusterState baselineState, boolean deferredActivation) {
-        return new ClusterStateBundle(baselineState, Collections.emptyMap(), deferredActivation);
+    public static ClusterStateBundle ofBaselineOnly(AnnotatedClusterState baselineState,
+                                                    FeedBlock feedBlock,
+                                                    boolean deferredActivation) {
+        return new ClusterStateBundle(baselineState, Collections.emptyMap(), feedBlock, deferredActivation);
     }
 
     public static ClusterStateBundle ofBaselineOnly(AnnotatedClusterState baselineState) {
@@ -148,7 +212,7 @@ public class ClusterStateBundle {
         Map<String, AnnotatedClusterState> clonedDerived = derivedBucketSpaceStates.entrySet().stream()
                 .collect(Collectors.toMap(e -> e.getKey(), e -> e.getValue().cloneWithClusterState(
                         mapper.apply(e.getValue().getClusterState().clone()))));
-        return new ClusterStateBundle(clonedBaseline, clonedDerived, deferredActivation);
+        return new ClusterStateBundle(clonedBaseline, clonedDerived, feedBlock, deferredActivation);
     }
 
     public ClusterStateBundle clonedWithVersionSet(int version) {
@@ -162,6 +226,9 @@ public class ClusterStateBundle {
         if (!baselineState.getClusterState().similarToIgnoringInitProgress(other.baselineState.getClusterState())) {
             return false;
         }
+        if (clusterFeedIsBlocked() != other.clusterFeedIsBlocked()) {
+            return false;
+        }
         // FIXME we currently treat mismatching bucket space sets as unchanged to avoid breaking some tests
         return derivedBucketSpaceStates.entrySet().stream()
                 .allMatch(entry -> other.derivedBucketSpaceStates.getOrDefault(entry.getKey(), entry.getValue())
@@ -172,17 +239,30 @@ public class ClusterStateBundle {
         return baselineState.getClusterState().getVersion();
     }
 
+    public Optional<FeedBlock> getFeedBlock() {
+        return Optional.ofNullable(feedBlock);
+    }
+
+    public boolean clusterFeedIsBlocked() {
+        return (feedBlock != null && feedBlock.blockFeedInCluster());
+    }
+
     @Override
     public String toString() {
+        String feedBlockedStr = clusterFeedIsBlocked()
+                ? String.format(", feed blocked: '%s'", feedBlock.description)
+                : "";
         if (derivedBucketSpaceStates.isEmpty()) {
-            return String.format("ClusterStateBundle('%s'%s)", baselineState,
-                    deferredActivation ? " (deferred activation)" : "");
+            return String.format("ClusterStateBundle('%s'%s%s)", baselineState,
+                    deferredActivation ? " (deferred activation)" : "",
+                    feedBlockedStr);
         }
         Map<String, AnnotatedClusterState> orderedStates = new TreeMap<>(derivedBucketSpaceStates);
-        return String.format("ClusterStateBundle('%s', %s%s)", baselineState, orderedStates.entrySet().stream()
+        return String.format("ClusterStateBundle('%s', %s%s%s)", baselineState, orderedStates.entrySet().stream()
                 .map(e -> String.format("%s '%s'", e.getKey(), e.getValue()))
                 .collect(Collectors.joining(", ")),
-                deferredActivation ? " (deferred activation)" : "");
+                deferredActivation ? " (deferred activation)" : "",
+                feedBlockedStr);
     }
 
     @Override
@@ -190,13 +270,14 @@ public class ClusterStateBundle {
         if (this == o) return true;
         if (o == null || getClass() != o.getClass()) return false;
         ClusterStateBundle that = (ClusterStateBundle) o;
-        return deferredActivation == that.deferredActivation &&
+        return (deferredActivation == that.deferredActivation &&
                 Objects.equals(baselineState, that.baselineState) &&
-                Objects.equals(derivedBucketSpaceStates, that.derivedBucketSpaceStates);
+                Objects.equals(derivedBucketSpaceStates, that.derivedBucketSpaceStates) &&
+                Objects.equals(feedBlock, that.feedBlock));
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(baselineState, derivedBucketSpaceStates, deferredActivation);
+        return Objects.hash(baselineState, derivedBucketSpaceStates, feedBlock, deferredActivation);
     }
 }
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/rpc/SlimeClusterStateBundleCodec.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/rpc/SlimeClusterStateBundleCodec.java
index cb76f67038c..1a3184955d5 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/rpc/SlimeClusterStateBundleCodec.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/rpc/SlimeClusterStateBundleCodec.java
@@ -40,6 +40,13 @@ public class SlimeClusterStateBundleCodec implements ClusterStateBundleCodec, En
         stateBundle.getDerivedBucketSpaceStates().entrySet()
                 .forEach(entry -> spaces.setString(entry.getKey(), entry.getValue().toString()));
 
+        // Only bother to encode feed block state if cluster is actually blocked
+        if (stateBundle.getFeedBlock().map(fb -> fb.blockFeedInCluster()).orElse(false)) {
+            Cursor feedBlock = root.setObject("feed-block");
+            feedBlock.setBool("block-feed-in-cluster", true);
+            feedBlock.setString("description", stateBundle.getFeedBlock().get().getDescription());
+        }
+
         byte[] serialized = BinaryFormat.encode(slime);
         Compressor.Compression compression = compressor.compress(serialized);
         return EncodedClusterStateBundle.fromCompressionBuffer(compression);
@@ -60,7 +67,14 @@ public class SlimeClusterStateBundleCodec implements ClusterStateBundleCodec, En
         }));
         boolean deferredActivation = root.field("deferred-activation").asBool(); // defaults to false if not present
 
-        return ClusterStateBundle.of(AnnotatedClusterState.withoutAnnotations(baseline), derivedStates, deferredActivation);
+        ClusterStateBundle.FeedBlock feedBlock = null;
+        Inspector fb = root.field("feed-block");
+        if (fb.valid() && fb.field("block-feed-in-cluster").asBool()) {
+            feedBlock = ClusterStateBundle.FeedBlock.blockedWithDescription(fb.field("description").asString());
+        }
+
+        return ClusterStateBundle.of(AnnotatedClusterState.withoutAnnotations(baseline), derivedStates,
+                                     feedBlock, deferredActivation);
     }
 
     // Technically the Slime enveloping could be its own class that is bundle codec independent, but
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStateBundleTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStateBundleTest.java
index 9ec3902fc1d..d2db47131bd 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStateBundleTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStateBundleTest.java
@@ -47,6 +47,12 @@ public class ClusterStateBundleTest {
         return createTestBundleBuilder(modifyDefaultSpace).deriveAndBuild();
     }
 
+    private static ClusterStateBundle createTestBundleWithFeedBlock(String description) {
+        return createTestBundleBuilder(false)
+                .feedBlock(ClusterStateBundle.FeedBlock.blockedWithDescription(description))
+                .deriveAndBuild();
+    }
+
     private static ClusterStateBundle createTestBundle() {
         return createTestBundle(true);
     }
@@ -91,6 +97,34 @@ public class ClusterStateBundleTest {
     }
 
     @Test
+    public void similarity_test_considers_cluster_feed_block_state() {
+        var nonBlockingBundle = createTestBundle(false);
+        var blockingBundle = createTestBundleWithFeedBlock("foo");
+        var blockingBundleWithOtherDesc = createTestBundleWithFeedBlock("bar");
+
+        assertFalse(nonBlockingBundle.similarTo(blockingBundle));
+        assertFalse(blockingBundle.similarTo(nonBlockingBundle));
+        assertTrue(blockingBundle.similarTo(blockingBundle));
+        // We currently consider different descriptions with same blocking status to be similar
+        assertTrue(blockingBundle.similarTo(blockingBundleWithOtherDesc));
+    }
+
+    @Test
+    public void feed_block_state_is_available() {
+        var nonBlockingBundle = createTestBundle(false);
+        var blockingBundle = createTestBundleWithFeedBlock("foo");
+
+        assertFalse(nonBlockingBundle.clusterFeedIsBlocked());
+        assertFalse(nonBlockingBundle.getFeedBlock().isPresent());
+
+        assertTrue(blockingBundle.clusterFeedIsBlocked());
+        var block = blockingBundle.getFeedBlock();
+        assertTrue(block.isPresent());
+        assertTrue(block.get().blockFeedInCluster());
+        assertEquals(block.get().getDescription(), "foo");
+    }
+
+    @Test
     public void toString_without_bucket_space_states_prints_only_baseline_state() {
         ClusterStateBundle bundle = ClusterStateBundle.ofBaselineOnly(
                 annotatedStateOf("distributor:2 storage:2"));
@@ -107,14 +141,24 @@ public class ClusterStateBundleTest {
     }
 
     @Test
+    public void toString_with_feed_blocked_includes_description() {
+        var blockingBundle = createTestBundleWithFeedBlock("bear sleeping on server rack");
+        assertThat(blockingBundle.toString(), equalTo("ClusterStateBundle('distributor:2 storage:2', " +
+                "default 'distributor:2 storage:2', " +
+                "global 'distributor:2 storage:2', " +
+                "narnia 'distributor:2 .0.s:d storage:2', " +
+                "feed blocked: 'bear sleeping on server rack')"));
+    }
+
+    @Test
     public void toString_without_derived_states_specifies_deferred_activation_iff_set() {
-        var bundle = ClusterStateBundle.ofBaselineOnly(annotatedStateOf("distributor:2 storage:2"), true);
+        var bundle = ClusterStateBundle.ofBaselineOnly(annotatedStateOf("distributor:2 storage:2"), null, true);
         assertThat(bundle.toString(), equalTo("ClusterStateBundle('distributor:2 storage:2' (deferred activation))"));
     }
 
     @Test
     public void toString_without_derived_states_does_not_specify_deferred_activation_iff_not_set() {
-        var bundle = ClusterStateBundle.ofBaselineOnly(annotatedStateOf("distributor:2 storage:2"), false);
+        var bundle = ClusterStateBundle.ofBaselineOnly(annotatedStateOf("distributor:2 storage:2"), null, false);
         assertThat(bundle.toString(), equalTo("ClusterStateBundle('distributor:2 storage:2')"));
     }
 
@@ -177,4 +221,11 @@ public class ClusterStateBundleTest {
         assertEquals(bundle, derived);
     }
 
+    @Test
+    public void cloning_preserves_feed_block_state() {
+        var bundle = createTestBundleWithFeedBlock("foo");;
+        var derived = bundle.cloneWithMapper(Function.identity());
+        assertEquals(bundle, derived);
+    }
+
 }
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/rpc/SlimeClusterStateBundleCodecTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/rpc/SlimeClusterStateBundleCodecTest.java
index 3dce1153685..5fdeb29d1c6 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/rpc/SlimeClusterStateBundleCodecTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/rpc/SlimeClusterStateBundleCodecTest.java
@@ -88,4 +88,12 @@ public class SlimeClusterStateBundleCodecTest {
         assertThat(roundtripEncode(stateBundle), equalTo(stateBundle));
     }
 
+    @Test
+    public void can_roundtrip_encode_bundle_with_feed_block_state() {
+        var stateBundle = ClusterStateBundleUtil.makeBundleBuilder("distributor:2 storage:2")
+                .feedBlock(ClusterStateBundle.FeedBlock.blockedWithDescription("more cake needed"))
+                .deriveAndBuild();
+        assertThat(roundtripEncode(stateBundle), equalTo(stateBundle));
+    }
+
 }