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

2 files changed, 58 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;
+    }
+}