Fix ClusterStatsChangeTracker to keep previous 'may have merges pending' state per content node.

5 files changed, 104 insertions, 70 deletions

diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/AggregatedStatsMergePendingChecker.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/AggregatedStatsMergePendingChecker.java
index 88b7d0d877b..07586569bf5 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/AggregatedStatsMergePendingChecker.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/AggregatedStatsMergePendingChecker.java
@@ -31,16 +31,4 @@ public class AggregatedStatsMergePendingChecker implements MergePendingChecker {
         return true;
     }
 
-    public boolean mayHaveMergesPendingInGlobalSpace() {
-        if (!stats.hasUpdatesFromAllDistributors()) {
-            return true;
-        }
-        for (Iterator<ContentNodeStats> itr = stats.getStats().iterator(); itr.hasNext(); ) {
-            ContentNodeStats stats = itr.next();
-            if (mayHaveMergesPending(FixedBucketSpaces.globalSpace(), stats.getNodeIndex())) {
-                return true;
-            }
-        }
-        return false;
-    }
 }
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsChangeTracker.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsChangeTracker.java
index abab66cbde3..adfae7e36ea 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsChangeTracker.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsChangeTracker.java
@@ -1,8 +1,14 @@
 // Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.vespa.clustercontroller.core;
 
+import com.yahoo.document.FixedBucketSpaces;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+
 /**
- * Class tracking whether we have changes in current and previous cluster stats.
+ * Class tracking whether we have changes in current and previous aggregated cluster stats.
  *
  * The cluster stats are considered changed if the current and previous stats differs in whether
  * they may have buckets pending in the 'global' bucket space. This signals that the ClusterStateBundle should be recomputed.
@@ -11,11 +17,10 @@ public class ClusterStatsChangeTracker {
 
     private AggregatedClusterStats aggregatedStats;
     private AggregatedStatsMergePendingChecker checker;
-    private boolean prevMayHaveMergesPending;
+    private Map<Integer, Boolean> prevMayHaveMergesPending = null;
 
     public ClusterStatsChangeTracker(AggregatedClusterStats aggregatedStats) {
         setAggregatedStats(aggregatedStats);
-        prevMayHaveMergesPending = false;
     }
 
     private void setAggregatedStats(AggregatedClusterStats aggregatedStats) {
@@ -23,12 +28,16 @@ public class ClusterStatsChangeTracker {
         checker = new AggregatedStatsMergePendingChecker(this.aggregatedStats);
     }
 
-    public void syncBucketsPendingFlag() {
-        prevMayHaveMergesPending = checker.mayHaveMergesPendingInGlobalSpace();
+    public void syncAggregatedStats() {
+        prevMayHaveMergesPending = new HashMap<>();
+        for (Iterator<ContentNodeStats> itr = aggregatedStats.getStats().iterator(); itr.hasNext(); ) {
+            int nodeIndex = itr.next().getNodeIndex();
+            prevMayHaveMergesPending.put(nodeIndex, mayHaveMergesPendingInGlobalSpace(nodeIndex));
+        }
     }
 
     public void updateAggregatedStats(AggregatedClusterStats newAggregatedStats) {
-        syncBucketsPendingFlag();
+        syncAggregatedStats();
         setAggregatedStats(newAggregatedStats);
     }
 
@@ -36,10 +45,30 @@ public class ClusterStatsChangeTracker {
         if (!aggregatedStats.hasUpdatesFromAllDistributors()) {
             return false;
         }
-        if (prevMayHaveMergesPending != checker.mayHaveMergesPendingInGlobalSpace()) {
-            return true;
+        for (Iterator<ContentNodeStats> itr = aggregatedStats.getStats().iterator(); itr.hasNext(); ) {
+            int nodeIndex = itr.next().getNodeIndex();
+            boolean currValue = mayHaveMergesPendingInGlobalSpace(nodeIndex);
+            Boolean prevValue = prevMayHaveMergesPendingInGlobalSpace(nodeIndex);
+            if (prevValue != null) {
+                if (prevValue != currValue) {
+                    return true;
+                }
+            } else if (currValue) {
+                return true;
+            }
         }
         return false;
     }
 
+    private boolean mayHaveMergesPendingInGlobalSpace(int nodeIndex) {
+        return checker.mayHaveMergesPending(FixedBucketSpaces.globalSpace(), nodeIndex);
+    }
+
+    private Boolean prevMayHaveMergesPendingInGlobalSpace(int nodeIndex) {
+        if (prevMayHaveMergesPending != null) {
+            return prevMayHaveMergesPending.get(nodeIndex);
+        }
+        return null;
+    }
+
 }
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/StateVersionTracker.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/StateVersionTracker.java
index fcbc8ac1e21..d4cd8f902c6 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/StateVersionTracker.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/StateVersionTracker.java
@@ -89,7 +89,7 @@ public class StateVersionTracker {
     public void updateLatestCandidateStateBundle(final ClusterStateBundle candidateBundle) {
         assert(latestCandidateState.getBaselineClusterState().getVersion() == 0);
         latestCandidateState = candidateBundle;
-        clusterStatsChangeTracker.syncBucketsPendingFlag();
+        clusterStatsChangeTracker.syncAggregatedStats();
     }
 
     /**
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/AggregatedStatsMergePendingCheckerTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/AggregatedStatsMergePendingCheckerTest.java
index 15786dc6130..8ae6501d16b 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/AggregatedStatsMergePendingCheckerTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/AggregatedStatsMergePendingCheckerTest.java
@@ -39,10 +39,6 @@ public class AggregatedStatsMergePendingCheckerTest {
             return checker.mayHaveMergesPending(bucketSpace, contentNodeIndex);
         }
 
-        public boolean mayHaveMergesPendingInGlobalSpace() {
-            return checker.mayHaveMergesPendingInGlobalSpace();
-        }
-
     }
 
     @Test
@@ -81,27 +77,4 @@ public class AggregatedStatsMergePendingCheckerTest {
         assertTrue(f.mayHaveMergesPending("default", 1));
     }
 
-    @Test
-    public void cluster_without_updates_from_all_distributors_may_have_merges_pending_in_global_space() {
-        Fixture f = Fixture.fromIncompleteStats();
-        assertTrue(f.mayHaveMergesPendingInGlobalSpace());
-    }
-
-    @Test
-    public void cluster_may_have_merges_pending_in_global_space_if_one_node_has_buckets_pending() {
-        Fixture f = new Fixture(new ContentClusterStatsBuilder()
-                .add(1, "global", 10, 0)
-                .add(2, "global", 11, 1), true);
-        assertTrue(f.mayHaveMergesPendingInGlobalSpace());
-    }
-
-    @Test
-    public void cluster_does_not_have_merges_pending_in_global_space_if_no_nodes_have_buckets_pending() {
-        Fixture f = new Fixture(new ContentClusterStatsBuilder()
-                .add(3, "global", 10, 0)
-                .add(4, "global", 11, 0)
-                .add(4, "default", 12, 1), true);
-        assertFalse(f.mayHaveMergesPendingInGlobalSpace());
-    }
-
 }
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsChangeTrackerTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsChangeTrackerTest.java
index 0439532e02e..c3dcf1e1cfc 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsChangeTrackerTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsChangeTrackerTest.java
@@ -4,36 +4,64 @@ package com.yahoo.vespa.clustercontroller.core;
 import com.google.common.collect.Sets;
 import org.junit.Test;
 
+import java.util.Set;
+
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
 public class ClusterStatsChangeTrackerTest {
 
+    private static class StatsBuilder {
+        private final ContentClusterStatsBuilder builder = new ContentClusterStatsBuilder();
+
+        public StatsBuilder bucketsPending(int contentNodeIndex) {
+            builder.add(contentNodeIndex, "global", 5, 1);
+            return this;
+        }
+
+        public StatsBuilder inSync(int contentNodeIndex) {
+            builder.add(contentNodeIndex, "global", 5, 0);
+            return this;
+        }
+
+        public ContentClusterStats build() {
+            return builder.build();
+        }
+    }
+
+    private static StatsBuilder stats() {
+        return new StatsBuilder();
+    }
+
     private static class Fixture {
+        private final Set<Integer> contentNodeIndices;
         private ClusterStatsAggregator aggregator;
-        private ClusterStatsChangeTracker tracker;
+        private final ClusterStatsChangeTracker tracker;
 
-        public Fixture() {
-            aggregator = new ClusterStatsAggregator(Sets.newHashSet(1), Sets.newHashSet(2));
+        private Fixture(Integer... contentNodeIndices) {
+            this.contentNodeIndices = Sets.newHashSet(contentNodeIndices);
+            aggregator = new ClusterStatsAggregator(Sets.newHashSet(1), this.contentNodeIndices);
             tracker = new ClusterStatsChangeTracker(aggregator.getAggregatedStats());
         }
 
-        public void setBucketsPendingStats() {
-            updateStats(1);
+        public static Fixture empty() {
+            return new Fixture(0, 1);
         }
 
-        public void setInSyncStats() {
-            updateStats(0);
+        public static Fixture fromStats(StatsBuilder builder) {
+            Fixture result = new Fixture(0, 1);
+            result.updateStats(builder);
+            return result;
         }
 
-        public void updateStats(long bucketsPending) {
-            aggregator.updateForDistributor(1, new ContentClusterStatsBuilder()
-                    .add(2, "global", 5, bucketsPending).build());
+        public void newAggregatedStats(StatsBuilder builder) {
+            aggregator = new ClusterStatsAggregator(Sets.newHashSet(1), contentNodeIndices);
+            updateStats(builder);
+            tracker.updateAggregatedStats(aggregator.getAggregatedStats());
         }
 
-        public void updateAggregatedStats() {
-            aggregator = new ClusterStatsAggregator(Sets.newHashSet(1), Sets.newHashSet(2));
-            tracker.updateAggregatedStats(aggregator.getAggregatedStats());
+        private void updateStats(StatsBuilder builder) {
+            aggregator.updateForDistributor(1, builder.build());
         }
 
         public boolean statsHaveChanged() {
@@ -44,24 +72,40 @@ public class ClusterStatsChangeTrackerTest {
 
     @Test
     public void stats_have_not_changed_if_not_all_distributors_are_updated() {
-        Fixture f = new Fixture();
+        Fixture f = Fixture.empty();
         assertFalse(f.statsHaveChanged());
     }
 
     @Test
-    public void stats_have_changed_if_previous_buckets_pending_stats_are_different_from_current() {
-        Fixture f = new Fixture();
-
-        f.setInSyncStats();
+    public void stats_have_not_changed_if_all_nodes_in_sync_and_nothing_previous() {
+        Fixture f = Fixture.fromStats(stats().inSync(0).inSync(1));
         assertFalse(f.statsHaveChanged());
-        f.setBucketsPendingStats();
+    }
+
+    @Test
+    public void stats_have_changed_if_one_node_with_buckets_pending_and_nothing_previous() {
+        Fixture f = Fixture.fromStats(stats().inSync(0).bucketsPending(1));
         assertTrue(f.statsHaveChanged());
+    }
 
-        f.updateAggregatedStats(); // previous stats may now have buckets pending
+    @Test
+    public void stats_have_changed_if_one_node_has_in_sync_to_buckets_pending_transition() {
+        Fixture f = Fixture.fromStats(stats().bucketsPending(0).inSync(1));
+        f.newAggregatedStats(stats().bucketsPending(0).bucketsPending(1));
+        assertTrue(f.statsHaveChanged());
+    }
 
-        f.setInSyncStats();
+    @Test
+    public void stats_have_changed_if_one_node_has_buckets_pending_to_in_sync_transition() {
+        Fixture f = Fixture.fromStats(stats().bucketsPending(0).bucketsPending(1));
+        f.newAggregatedStats(stats().bucketsPending(0).inSync(1));
         assertTrue(f.statsHaveChanged());
-        f.setBucketsPendingStats();
+    }
+
+    @Test
+    public void stats_have_not_changed_if_no_nodes_have_changed_state() {
+        Fixture f = Fixture.fromStats(stats().bucketsPending(0).bucketsPending(1));
+        f.newAggregatedStats(stats().bucketsPending(0).bucketsPending(1));
         assertFalse(f.statsHaveChanged());
     }