Include deferred activation flag with cluster state bundles

Bundles including this flag from the cluster controller indicate to
receiver nodes that an explicit activation RPC will follow. When it
is not present, nodes must activate the cluster state at their own
leisure as they have done historically.

7 files changed, 114 insertions, 18 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 76177e8f1c1..35fe32f21c9 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
@@ -19,16 +19,23 @@ import java.util.stream.Collectors;
  *
  * The baseline state is identical to the legacy, global cluster state that the
  * cluster controller has historically produced as its only output.
+ *
+ * The bundle also contains an additional "deferred activation" flag which tells
+ * the recipient if the cluster state transition should complete immediately or
+ * await an explicit activation RPC from the cluster controller.
  */
 public class ClusterStateBundle {
 
     private final AnnotatedClusterState baselineState;
     private final Map<String, AnnotatedClusterState> derivedBucketSpaceStates;
+    private final boolean deferredActivation;
 
     public static class Builder {
         private final AnnotatedClusterState baselineState;
+        private Map<String, AnnotatedClusterState> explicitDerivedStates;
         private ClusterStateDeriver stateDeriver;
         private Set<String> bucketSpaces;
+        private boolean deferredActivation = true;
 
         public Builder(AnnotatedClusterState baselineState) {
             this.baselineState = baselineState;
@@ -40,30 +47,59 @@ public class ClusterStateBundle {
         }
 
         public Builder bucketSpaces(Set<String> bucketSpaces) {
+            if (this.explicitDerivedStates != null) {
+                throw new IllegalStateException("Cannot set bucket spaces on Builder that already " +
+                                                "has explicit derived states set");
+            }
             this.bucketSpaces = bucketSpaces;
             return this;
         }
 
         public Builder bucketSpaces(String... bucketSpaces) {
-            this.bucketSpaces = new TreeSet<>(Arrays.asList(bucketSpaces));
+            return bucketSpaces(new TreeSet<>(Arrays.asList(bucketSpaces)));
+        }
+
+        public Builder explicitDerivedStates(Map<String, AnnotatedClusterState> derivedStates) {
+            if (this.bucketSpaces != null || this.stateDeriver != null) {
+                throw new IllegalStateException("Cannot set explicitly derived states on Builder " +
+                                                "that already has bucket spaces or deriver set");
+            }
+            this.explicitDerivedStates = derivedStates;
+            return this;
+        }
+
+        public Builder deferredActivation(boolean deferred) {
+            this.deferredActivation = deferred;
             return this;
         }
 
         public ClusterStateBundle deriveAndBuild() {
-            if (stateDeriver == null || bucketSpaces == null || bucketSpaces.isEmpty()) {
+            if ((stateDeriver == null || bucketSpaces == null || bucketSpaces.isEmpty()) && explicitDerivedStates == null) {
                 return ClusterStateBundle.ofBaselineOnly(baselineState);
             }
-            Map<String, AnnotatedClusterState> derived = bucketSpaces.stream()
-                    .collect(Collectors.toMap(
-                            Function.identity(),
-                            s -> stateDeriver.derivedFrom(baselineState, s)));
-            return new ClusterStateBundle(baselineState, derived);
+            Map<String, AnnotatedClusterState> derived;
+            if (explicitDerivedStates != null) {
+                derived = explicitDerivedStates;
+            } else {
+                derived = bucketSpaces.stream()
+                        .collect(Collectors.toMap(
+                                Function.identity(),
+                                s -> stateDeriver.derivedFrom(baselineState, s)));
+            }
+            return new ClusterStateBundle(baselineState, derived, deferredActivation);
         }
     }
 
     private ClusterStateBundle(AnnotatedClusterState baselineState, Map<String, AnnotatedClusterState> derivedBucketSpaceStates) {
+        this(baselineState, derivedBucketSpaceStates, true);
+    }
+
+    private ClusterStateBundle(AnnotatedClusterState baselineState, Map<String,
+                               AnnotatedClusterState> derivedBucketSpaceStates,
+                               boolean deferredActivation) {
         this.baselineState = baselineState;
         this.derivedBucketSpaceStates = Collections.unmodifiableMap(derivedBucketSpaceStates);
+        this.deferredActivation = deferredActivation;
     }
 
     public static Builder builder(AnnotatedClusterState baselineState) {
@@ -74,6 +110,16 @@ public class ClusterStateBundle {
         return new ClusterStateBundle(baselineState, derivedBucketSpaceStates);
     }
 
+    public static ClusterStateBundle of(AnnotatedClusterState baselineState,
+                                        Map<String, AnnotatedClusterState> derivedBucketSpaceStates,
+                                        boolean deferredActivation) {
+        return new ClusterStateBundle(baselineState, derivedBucketSpaceStates, deferredActivation);
+    }
+
+    public static ClusterStateBundle ofBaselineOnly(AnnotatedClusterState baselineState, boolean deferredActivation) {
+        return new ClusterStateBundle(baselineState, Collections.emptyMap(), deferredActivation);
+    }
+
     public static ClusterStateBundle ofBaselineOnly(AnnotatedClusterState baselineState) {
         return new ClusterStateBundle(baselineState, Collections.emptyMap());
     }
@@ -94,6 +140,8 @@ public class ClusterStateBundle {
         return derivedBucketSpaceStates;
     }
 
+    public boolean deferredActivation() { return this.deferredActivation; }
+
     public ClusterStateBundle cloneWithMapper(Function<ClusterState, ClusterState> mapper) {
         AnnotatedClusterState clonedBaseline = baselineState.cloneWithClusterState(
                 mapper.apply(baselineState.getClusterState().clone()));
@@ -140,13 +188,13 @@ public class ClusterStateBundle {
         if (this == o) return true;
         if (o == null || getClass() != o.getClass()) return false;
         ClusterStateBundle that = (ClusterStateBundle) o;
-        return Objects.equals(baselineState, that.baselineState) &&
+        return deferredActivation == that.deferredActivation &&
+                Objects.equals(baselineState, that.baselineState) &&
                 Objects.equals(derivedBucketSpaceStates, that.derivedBucketSpaceStates);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(baselineState, derivedBucketSpaceStates);
+        return Objects.hash(baselineState, derivedBucketSpaceStates, deferredActivation);
     }
-
 }
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetController.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetController.java
index 005bf7971a5..9d833f366e5 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetController.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetController.java
@@ -822,6 +822,7 @@ public class FleetController implements NodeStateOrHostInfoChangeHandler, NodeAd
             final ClusterStateBundle candidateBundle = ClusterStateBundle.builder(candidate)
                     .bucketSpaces(configuredBucketSpaces)
                     .stateDeriver(createBucketSpaceStateDeriver())
+                    .deferredActivation(true)
                     .deriveAndBuild();
             stateVersionTracker.updateLatestCandidateStateBundle(candidateBundle);
             invokeCandidateStateListeners(candidateBundle);
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/SystemStateBroadcaster.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/SystemStateBroadcaster.java
index 629800fb13c..3683fe342bc 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/SystemStateBroadcaster.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/SystemStateBroadcaster.java
@@ -98,8 +98,8 @@ public class SystemStateBroadcaster {
             return false; // Can't set state on nodes we don't know where are
         }
         if (node.getReportedState().getState() == State.MAINTENANCE ||
-                node.getReportedState().getState() == State.DOWN ||
-                node.getReportedState().getState() == State.STOPPING)
+            node.getReportedState().getState() == State.DOWN ||
+            node.getReportedState().getState() == State.STOPPING)
         {
             return false; // No point in sending system state to nodes that can't receive messages or don't want them
         }
@@ -123,8 +123,8 @@ public class SystemStateBroadcaster {
      * object and updates the broadcaster's last known in-sync cluster state version.
      */
     void checkIfClusterStateIsAckedByAllDistributors(DatabaseHandler database,
-                                                        DatabaseHandler.Context dbContext,
-                                                        FleetController fleetController) throws InterruptedException {
+                                                     DatabaseHandler.Context dbContext,
+                                                     FleetController fleetController) throws InterruptedException {
         if ((clusterStateBundle == null) || (lastClusterStateInSync == clusterStateBundle.getVersion())) {
             return; // Nothing to do for the current state
         }
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 1c391f9aacf..cb76f67038c 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
@@ -30,6 +30,9 @@ public class SlimeClusterStateBundleCodec implements ClusterStateBundleCodec, En
     public EncodedClusterStateBundle encode(ClusterStateBundle stateBundle) {
         Slime slime = new Slime();
         Cursor root = slime.setObject();
+        if (stateBundle.deferredActivation()) {
+            root.setBool("deferred-activation", stateBundle.deferredActivation());
+        }
         Cursor states = root.setObject("states");
         // TODO add another function that is not toString for this..!
         states.setString("baseline", stateBundle.getBaselineClusterState().toString());
@@ -55,8 +58,9 @@ public class SlimeClusterStateBundleCodec implements ClusterStateBundleCodec, En
         spaces.traverse(((ObjectTraverser)(key, value) -> {
             derivedStates.put(key, AnnotatedClusterState.withoutAnnotations(ClusterState.stateFromString(value.asString())));
         }));
+        boolean deferredActivation = root.field("deferred-activation").asBool(); // defaults to false if not present
 
-        return ClusterStateBundle.of(AnnotatedClusterState.withoutAnnotations(baseline), derivedStates);
+        return ClusterStateBundle.of(AnnotatedClusterState.withoutAnnotations(baseline), derivedStates, 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 7dccae988df..fa802f7fd71 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
@@ -19,7 +19,7 @@ public class ClusterStateBundleTest {
         return AnnotatedClusterState.withoutAnnotations(stateOf(state));
     }
 
-    private static ClusterStateBundle createTestBundle(boolean modifyDefaultSpace) {
+    private static ClusterStateBundle.Builder createTestBundleBuilder(boolean modifyDefaultSpace) {
         return ClusterStateBundle
                 .builder(annotatedStateOf("distributor:2 storage:2"))
                 .bucketSpaces("default", "global", "narnia")
@@ -33,8 +33,11 @@ public class ClusterStateBundleTest {
                                 .setNodeState(Node.ofDistributor(0), new NodeState(NodeType.DISTRIBUTOR, State.DOWN));
                     }
                     return derived;
-                })
-                .deriveAndBuild();
+                });
+    }
+
+    private static ClusterStateBundle createTestBundle(boolean modifyDefaultSpace) {
+        return createTestBundleBuilder(modifyDefaultSpace).deriveAndBuild();
     }
 
     private static ClusterStateBundle createTestBundle() {
@@ -96,4 +99,22 @@ public class ClusterStateBundleTest {
                 "narnia 'distributor:2 .0.s:d storage:2')"));
     }
 
+    @Test
+    public void deferred_activation_is_enabled_by_default() {
+        ClusterStateBundle bundle = createTestBundle();
+        assertTrue(bundle.deferredActivation());
+    }
+
+    @Test
+    public void can_build_bundle_with_deferred_activation_enabled() {
+        var bundle = createTestBundleBuilder(false).deferredActivation(true).deriveAndBuild();
+        assertTrue(bundle.deferredActivation());
+    }
+
+    @Test
+    public void can_build_bundle_with_deferred_activation_disabled() {
+        var bundle = createTestBundleBuilder(false).deferredActivation(false).deriveAndBuild();
+        assertFalse(bundle.deferredActivation());
+    }
+
 }
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStateBundleUtil.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStateBundleUtil.java
index 00c2194205d..cceb6d6f03f 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStateBundleUtil.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStateBundleUtil.java
@@ -12,6 +12,12 @@ import java.util.stream.Stream;
  */
 public class ClusterStateBundleUtil {
 
+    public static ClusterStateBundle.Builder makeBundleBuilder(String baselineState, StateMapping... bucketSpaceStates) {
+        return ClusterStateBundle.builder(AnnotatedClusterState.withoutAnnotations(ClusterState.stateFromString(baselineState)))
+                .explicitDerivedStates(Stream.of(bucketSpaceStates).collect(Collectors.toMap(sm -> sm.bucketSpace,
+                        sm -> AnnotatedClusterState.withoutAnnotations(sm.state))));
+    }
+
     public static ClusterStateBundle makeBundle(String baselineState, StateMapping... bucketSpaceStates) {
         return ClusterStateBundle.of(AnnotatedClusterState.withoutAnnotations(ClusterState.stateFromString(baselineState)),
                 Stream.of(bucketSpaceStates).collect(Collectors.toMap(sm -> sm.bucketSpace,
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 b19b1d780bf..3dce1153685 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
@@ -72,4 +72,20 @@ public class SlimeClusterStateBundleCodecTest {
         assertThat(roundtripEncodeWithEnvelope(stateBundle), equalTo(stateBundle));
     }
 
+    @Test
+    public void can_roundtrip_encode_bundle_with_deferred_activation_enabled() {
+        var stateBundle = ClusterStateBundleUtil.makeBundleBuilder("distributor:2 storage:2")
+                .deferredActivation(true)
+                .deriveAndBuild();
+        assertThat(roundtripEncode(stateBundle), equalTo(stateBundle));
+    }
+
+    @Test
+    public void can_roundtrip_encode_bundle_with_deferred_activation_disabled() {
+        var stateBundle = ClusterStateBundleUtil.makeBundleBuilder("distributor:2 storage:2")
+                .deferredActivation(false)
+                .deriveAndBuild();
+        assertThat(roundtripEncode(stateBundle), equalTo(stateBundle));
+    }
+
 }