Wire buckets pending state into state version tracker.

4 files changed, 65 insertions, 2 deletions

diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStateView.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStateView.java
index 6ce719a97de..62cd158c759 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStateView.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStateView.java
@@ -112,6 +112,10 @@ public class ClusterStateView {
                 StorageNodeStatsBridge.generate(hostInfo.getDistributor()));
     }
 
+    public ClusterStatsAggregator getStatsAggregator() {
+        return statsAggregator;
+    }
+
     public String toString() {
         return clusterState.toString();
     }
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
index e556841c835..dfa64d72521 100644
--- 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
@@ -17,8 +17,12 @@ public class ClusterStatsBucketsPendingState {
         this.prevMayHaveBucketsPending = false;
     }
 
-    public void updateAggregator(ClusterStatsAggregator newAggregator) {
+    public void syncBucketsPendingFlag() {
         prevMayHaveBucketsPending = aggregator.mayHaveBucketsPendingInGlobalSpace();
+    }
+
+    public void updateAggregator(ClusterStatsAggregator newAggregator) {
+        syncBucketsPendingFlag();
         aggregator = newAggregator;
     }
 
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 78e587a7446..c2be8047a13 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
@@ -33,12 +33,14 @@ public class StateVersionTracker {
     private ClusterStateBundle currentClusterState = latestCandidateState;
 
     private ClusterStateView clusterStateView;
+    private ClusterStatsBucketsPendingState bucketsPendingState;
 
     private final LinkedList<ClusterStateHistoryEntry> clusterStateHistory = new LinkedList<>();
     private int maxHistoryEntryCount = 50;
 
     StateVersionTracker() {
         clusterStateView = ClusterStateView.create(currentUnversionedState.getBaselineClusterState());
+        bucketsPendingState = new ClusterStatsBucketsPendingState(clusterStateView.getStatsAggregator());
     }
 
     void setVersionRetrievedFromZooKeeper(final int version) {
@@ -84,6 +86,7 @@ public class StateVersionTracker {
     public void updateLatestCandidateStateBundle(final ClusterStateBundle candidateBundle) {
         assert(latestCandidateState.getBaselineClusterState().getVersion() == 0);
         latestCandidateState = candidateBundle;
+        bucketsPendingState.syncBucketsPendingFlag();
     }
 
     /**
@@ -122,6 +125,7 @@ public class StateVersionTracker {
                 newStateBundle.getBaselineClusterState().getDistributionBitCount());
         // TODO should this take place in updateLatestCandidateStateBundle instead? I.e. does it require a consolidated state?
         clusterStateView = ClusterStateView.create(currentClusterState.getBaselineClusterState());
+        bucketsPendingState.updateAggregator(clusterStateView.getStatsAggregator());
     }
 
     private void recordCurrentStateInHistoryAtTime(final long currentTimeMs) {
@@ -138,7 +142,7 @@ public class StateVersionTracker {
     }
 
     boolean bucketSpaceMergeCompletionStateHasChanged() {
-        return false; // TODO wire changes in merge info
+        return bucketsPendingState.stateHasChanged();
     }
 
     /*
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/StateVersionTrackerTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/StateVersionTrackerTest.java
index 0f4d1fcdefc..0859ee0e409 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/StateVersionTrackerTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/StateVersionTrackerTest.java
@@ -6,6 +6,7 @@ import com.yahoo.vdslib.state.Node;
 import com.yahoo.vdslib.state.NodeState;
 import com.yahoo.vdslib.state.NodeType;
 import com.yahoo.vdslib.state.State;
+import com.yahoo.vespa.clustercontroller.core.hostinfo.HostInfo;
 import org.junit.Test;
 
 import java.text.ParseException;
@@ -17,6 +18,8 @@ import static org.hamcrest.core.Is.is;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 public class StateVersionTrackerTest {
 
@@ -267,4 +270,52 @@ public class StateVersionTrackerTest {
         assertTrue(versionTracker.candidateChangedEnoughFromCurrentToWarrantPublish());
     }
 
+    @Test
+    public void buckets_pending_state_is_tracked_between_cluster_states() {
+        final StateVersionTracker tracker = createWithMockedMetrics();
+        final NodeInfo distributorNode = mock(DistributorNodeInfo.class);
+        when(distributorNode.isDistributor()).thenReturn(true);
+        assertFalse(tracker.bucketSpaceMergeCompletionStateHasChanged());
+
+        tracker.updateLatestCandidateStateBundle(ClusterStateBundle
+                .ofBaselineOnly(stateWithoutAnnotations("distributor:1 storage:1")));
+        tracker.promoteCandidateToVersionedState(1234);
+        assertFalse(tracker.bucketSpaceMergeCompletionStateHasChanged());
+
+        // Give 'global' bucket space no buckets pending, which is the same as previous stats
+        tracker.handleUpdatedHostInfo(distributorNode, createHostInfo(0));
+        assertFalse(tracker.bucketSpaceMergeCompletionStateHasChanged());
+
+        // Give 'global' bucket space buckets pending, which is different from previous stats
+        tracker.handleUpdatedHostInfo(distributorNode, createHostInfo(1));
+        assertTrue(tracker.bucketSpaceMergeCompletionStateHasChanged());
+
+        tracker.updateLatestCandidateStateBundle(ClusterStateBundle
+                .ofBaselineOnly(stateWithoutAnnotations("distributor:1 storage:1")));
+        assertFalse(tracker.bucketSpaceMergeCompletionStateHasChanged());
+    }
+
+    private HostInfo createHostInfo(long bucketsPending) {
+        return HostInfo.createHostInfo(
+                "{\n" +
+                "\"cluster-state-version\": 2,\n" +
+                "\"distributor\": {\n" +
+                "  \"storage-nodes\": [\n" +
+                "    {\n" +
+                "      \"node-index\": 0,\n" +
+                "      \"bucket-spaces\": [\n" +
+                "        {\n" +
+                "          \"name\": \"global\"\n," +
+                "          \"buckets\": {\n" +
+                "            \"total\": 5,\n" +
+                "            \"pending\": " + bucketsPending + "\n" +
+                "          }\n" +
+                "        }\n" +
+                "      ]\n" +
+                "    }\n" +
+                "  ]\n" +
+                "}\n" +
+                "}");
+    }
+
 }