Add class tracking whether we have changes in buckets pending state in the 'global' bucket space.

4 files changed, 171 insertions, 3 deletions

diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregator.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregator.java
index 3f7cd129fc1..318a2080660 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregator.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregator.java
@@ -1,9 +1,9 @@
 // Copyright 2017 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 java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
+import com.yahoo.document.FixedBucketSpaces;
+
+import java.util.*;
 
 /**
  * Class that stores content cluster stats (with bucket space stats per node) for
@@ -27,6 +27,7 @@ import java.util.Set;
 public class ClusterStatsAggregator {
 
     private final Set<Integer> distributors;
+    private final Set<Integer> nonUpdatedDistributors;
 
     // Maps the distributor node index to a map of content node index to the
     // content node's stats.
@@ -39,6 +40,7 @@ public class ClusterStatsAggregator {
 
     ClusterStatsAggregator(Set<Integer> distributors, Set<Integer> storageNodes) {
         this.distributors = distributors;
+        nonUpdatedDistributors = new HashSet<>(distributors);
         aggregatedStats = new ContentClusterStats(storageNodes);
     }
 
@@ -46,6 +48,24 @@ public class ClusterStatsAggregator {
         return aggregatedStats;
     }
 
+    boolean hasUpdatesFromAllDistributors() {
+        return nonUpdatedDistributors.isEmpty();
+    }
+
+    boolean mayHaveBucketsPendingInGlobalSpace() {
+        if (!hasUpdatesFromAllDistributors()) {
+            return true;
+        }
+        AggregatedStatsMergePendingChecker checker = new AggregatedStatsMergePendingChecker(aggregatedStats);
+        for (Iterator<ContentNodeStats> itr = aggregatedStats.iterator(); itr.hasNext(); ) {
+            ContentNodeStats stats = itr.next();
+            if (checker.hasMergesPending(FixedBucketSpaces.globalSpace(), stats.getNodeIndex())) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     /**
      * Update the aggregator with the newest available stats from a distributor.
      */
@@ -53,6 +73,7 @@ public class ClusterStatsAggregator {
         if (!distributors.contains(distributorIndex)) {
             return;
         }
+        nonUpdatedDistributors.remove(distributorIndex);
         addStatsFromDistributor(distributorIndex, clusterStats);
     }
 
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsBucketsPendingState.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsBucketsPendingState.java
new file mode 100644
index 00000000000..e556841c835
--- /dev/null
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsBucketsPendingState.java
@@ -0,0 +1,34 @@
+// 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;
+
+/**
+ * Class tracking whether we have changes in buckets pending state in the 'global' bucket space.
+ *
+ * The state is considered changed if the previous and current cluster stats differs in whether
+ * they may have buckets pending in the 'global' bucket space. This signals that the ClusterStateBundle should be recomputed.
+ */
+public class ClusterStatsBucketsPendingState {
+
+    private ClusterStatsAggregator aggregator;
+    private boolean prevMayHaveBucketsPending;
+
+    public ClusterStatsBucketsPendingState(ClusterStatsAggregator aggregator) {
+        this.aggregator = aggregator;
+        this.prevMayHaveBucketsPending = false;
+    }
+
+    public void updateAggregator(ClusterStatsAggregator newAggregator) {
+        prevMayHaveBucketsPending = aggregator.mayHaveBucketsPendingInGlobalSpace();
+        aggregator = newAggregator;
+    }
+
+    public boolean stateHasChanged() {
+        if (!aggregator.hasUpdatesFromAllDistributors()) {
+            return false;
+        }
+        if (prevMayHaveBucketsPending != aggregator.mayHaveBucketsPendingInGlobalSpace()) {
+            return true;
+        }
+        return false;
+    }
+}
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregatorTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregatorTest.java
index 35391fc16f7..ac6417d0077 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregatorTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregatorTest.java
@@ -7,6 +7,8 @@ import org.junit.Test;
 import java.util.*;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 
 /**
  * @author hakonhall
@@ -26,6 +28,12 @@ public class ClusterStatsAggregatorTest {
         public void verify(ContentClusterStatsBuilder expectedStats) {
             assertEquals(expectedStats.build(), aggregator.getAggregatedStats());
         }
+        public boolean hasUpdatesFromAllDistributors() {
+            return aggregator.hasUpdatesFromAllDistributors();
+        }
+        public boolean mayHaveBucketsPendingInGlobalSpace() {
+            return aggregator.mayHaveBucketsPendingInGlobalSpace();
+        }
     }
 
     private static Set<Integer> distributorNodes(Integer... indices) {
@@ -109,4 +117,41 @@ public class ClusterStatsAggregatorTest {
         f.verify(new ContentClusterStatsBuilder().add(3));
     }
 
+    @Test
+    public void aggregator_tracks_when_it_has_updates_from_all_distributors() {
+        Fixture f = new Fixture(distributorNodes(1, 2), contentNodes(3));
+        assertFalse(f.hasUpdatesFromAllDistributors());
+        f.update(1, new ContentClusterStatsBuilder().add(3, "default"));
+        assertFalse(f.hasUpdatesFromAllDistributors());
+        f.update(1, new ContentClusterStatsBuilder().add(3, "default", 10, 1));
+        assertFalse(f.hasUpdatesFromAllDistributors());
+        f.update(2, new ContentClusterStatsBuilder().add(3, "default"));
+        assertTrue(f.hasUpdatesFromAllDistributors());
+    }
+
+    @Test
+    public void cluster_without_updates_from_all_distributors_may_have_buckets_pending() {
+        Fixture f = new Fixture(distributorNodes(1), contentNodes(3, 4));
+        assertTrue(f.mayHaveBucketsPendingInGlobalSpace());
+    }
+
+    @Test
+    public void cluster_may_have_buckets_pending_in_global_space_if_one_node_has_buckets_pending() {
+        Fixture f = new Fixture(distributorNodes(1), contentNodes(3, 4));
+        f.update(1, new ContentClusterStatsBuilder()
+                .add(3, "global", 10, 0)
+                .add(4, "global", 11, 1));
+        assertTrue(f.mayHaveBucketsPendingInGlobalSpace());
+    }
+
+    @Test
+    public void cluster_does_not_have_buckets_pending_in_global_space_if_no_nodes_have_buckets_pending() {
+        Fixture f = new Fixture(distributorNodes(1), contentNodes(3, 4));
+        f.update(1, new ContentClusterStatsBuilder()
+                .add(3, "global", 10, 0)
+                .add(4, "global", 11, 0)
+                .add(4, "default", 12, 1));
+        assertFalse(f.mayHaveBucketsPendingInGlobalSpace());
+    }
+
 }
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsBucketsPendingStateTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsBucketsPendingStateTest.java
new file mode 100644
index 00000000000..d785225d64d
--- /dev/null
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsBucketsPendingStateTest.java
@@ -0,0 +1,68 @@
+// 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.google.common.collect.Sets;
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class ClusterStatsBucketsPendingStateTest {
+
+    private static class Fixture {
+        private ClusterStatsAggregator aggregator;
+        private ClusterStatsBucketsPendingState state;
+
+        public Fixture() {
+            aggregator = new ClusterStatsAggregator(Sets.newHashSet(1), Sets.newHashSet(2));
+            state = new ClusterStatsBucketsPendingState(aggregator);
+        }
+
+        public void setBucketsPendingStats() {
+            updateStats(1);
+        }
+
+        public void setInSyncStats() {
+            updateStats(0);
+        }
+
+        public void updateStats(long bucketsPending) {
+            aggregator.updateForDistributor(1, new ContentClusterStatsBuilder()
+                    .add(2, "global", 5, bucketsPending).build());
+        }
+
+        public void updateAggregator() {
+            aggregator = new ClusterStatsAggregator(Sets.newHashSet(1), Sets.newHashSet(2));
+            state.updateAggregator(aggregator);
+        }
+
+        public boolean stateHasChanged() {
+            return state.stateHasChanged();
+        }
+
+    }
+
+    @Test
+    public void state_has_not_changed_if_not_all_distributors_are_updated() {
+        Fixture f = new Fixture();
+        assertFalse(f.stateHasChanged());
+    }
+
+    @Test
+    public void state_has_changed_if_previous_buckets_pending_stats_are_different_from_current() {
+        Fixture f = new Fixture();
+
+        f.setInSyncStats();
+        assertFalse(f.stateHasChanged());
+        f.setBucketsPendingStats();
+        assertTrue(f.stateHasChanged());
+
+        f.updateAggregator(); // previous stats may now have buckets pending
+
+        f.setInSyncStats();
+        assertTrue(f.stateHasChanged());
+        f.setBucketsPendingStats();
+        assertFalse(f.stateHasChanged());
+    }
+
+}