Fix AggregatedStatsMergePendingChecker to take state of aggregated cluster stats into account.

We may have merges pending if:
  - we don't yet have updates from all distributors
  - we don't have stats for the content node in question

Also move mayHaveMergesPendingInGlobalSpace() to AggregatedStatsMergePendingChecker.

9 files changed, 134 insertions, 85 deletions

diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/AggregatedClusterStats.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/AggregatedClusterStats.java
new file mode 100644
index 00000000000..7a00627f7ae
--- /dev/null
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/AggregatedClusterStats.java
@@ -0,0 +1,13 @@
+// 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;
+
+/**
+ * Interface that gives a view over aggregated cluster stats that will change over time.
+ */
+public interface AggregatedClusterStats {
+
+    public boolean hasUpdatesFromAllDistributors();
+
+    public ContentClusterStats getStats();
+
+}
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 c9c1bbdbf79..71a582a2a4b 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
@@ -1,6 +1,10 @@
 // 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.Iterator;
+
 /**
  * Class checking whether a particular bucket space on a content node might have buckets pending.
  *
@@ -8,19 +12,37 @@ package com.yahoo.vespa.clustercontroller.core;
  */
 public class AggregatedStatsMergePendingChecker implements MergePendingChecker {
 
-    private final ContentClusterStats clusterStats;
+    private final AggregatedClusterStats stats;
 
-    public AggregatedStatsMergePendingChecker(ContentClusterStats clusterStats) {
-        this.clusterStats = clusterStats;
+    public AggregatedStatsMergePendingChecker(AggregatedClusterStats stats) {
+        this.stats = stats;
     }
 
     @Override
     public boolean mayHaveMergesPending(String bucketSpace, int contentNodeIndex) {
-        ContentNodeStats nodeStats = clusterStats.getContentNode(contentNodeIndex);
+        if (!stats.hasUpdatesFromAllDistributors()) {
+            return true;
+        }
+        ContentNodeStats nodeStats = stats.getStats().getContentNode(contentNodeIndex);
         if (nodeStats != null) {
             ContentNodeStats.BucketSpaceStats bucketSpaceStats = nodeStats.getBucketSpace(bucketSpace);
             if (bucketSpaceStats != null && bucketSpaceStats.mayHaveBucketsPending()) {
                 return true;
+            } else {
+                return false;
+            }
+        }
+        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/ClusterStatsAggregator.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregator.java
index ed2ca32dab0..9db4ce4e44a 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
@@ -46,8 +46,20 @@ public class ClusterStatsAggregator {
         aggregatedStats = new ContentClusterStats(storageNodes);
     }
 
-    public ContentClusterStats getAggregatedStats() {
-        return aggregatedStats;
+    public AggregatedClusterStats getAggregatedStats() {
+        return new AggregatedClusterStats() {
+
+            @Override
+            public boolean hasUpdatesFromAllDistributors() {
+                return nonUpdatedDistributors.isEmpty();
+            }
+
+            @Override
+            public ContentClusterStats getStats() {
+                return aggregatedStats;
+            }
+
+        };
     }
 
     public ContentNodeStats getAggregatedStatsForDistributor(int distributorIndex) {
@@ -61,26 +73,8 @@ public class ClusterStatsAggregator {
         return result;
     }
 
-    boolean hasUpdatesFromAllDistributors() {
-        return nonUpdatedDistributors.isEmpty();
-    }
-
-    boolean mayHaveBucketsPendingInGlobalSpace() {
-        if (!hasUpdatesFromAllDistributors()) {
-            return true;
-        }
-        MergePendingChecker checker = createMergePendingChecker();
-        for (Iterator<ContentNodeStats> itr = aggregatedStats.iterator(); itr.hasNext(); ) {
-            ContentNodeStats stats = itr.next();
-            if (checker.mayHaveMergesPending(FixedBucketSpaces.globalSpace(), stats.getNodeIndex())) {
-                return true;
-            }
-        }
-        return false;
-    }
-
     MergePendingChecker createMergePendingChecker() {
-        return new AggregatedStatsMergePendingChecker(aggregatedStats);
+        return new AggregatedStatsMergePendingChecker(getAggregatedStats());
     }
 
     /**
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 ed32a5f4b45..abab66cbde3 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
@@ -9,28 +9,34 @@ package com.yahoo.vespa.clustercontroller.core;
  */
 public class ClusterStatsChangeTracker {
 
-    private ClusterStatsAggregator aggregator;
-    private boolean prevMayHaveBucketsPending;
+    private AggregatedClusterStats aggregatedStats;
+    private AggregatedStatsMergePendingChecker checker;
+    private boolean prevMayHaveMergesPending;
 
-    public ClusterStatsChangeTracker(ClusterStatsAggregator aggregator) {
-        this.aggregator = aggregator;
-        this.prevMayHaveBucketsPending = false;
+    public ClusterStatsChangeTracker(AggregatedClusterStats aggregatedStats) {
+        setAggregatedStats(aggregatedStats);
+        prevMayHaveMergesPending = false;
+    }
+
+    private void setAggregatedStats(AggregatedClusterStats aggregatedStats) {
+        this.aggregatedStats = aggregatedStats;
+        checker = new AggregatedStatsMergePendingChecker(this.aggregatedStats);
     }
 
     public void syncBucketsPendingFlag() {
-        prevMayHaveBucketsPending = aggregator.mayHaveBucketsPendingInGlobalSpace();
+        prevMayHaveMergesPending = checker.mayHaveMergesPendingInGlobalSpace();
     }
 
-    public void updateAggregator(ClusterStatsAggregator newAggregator) {
+    public void updateAggregatedStats(AggregatedClusterStats newAggregatedStats) {
         syncBucketsPendingFlag();
-        aggregator = newAggregator;
+        setAggregatedStats(newAggregatedStats);
     }
 
     public boolean statsHaveChanged() {
-        if (!aggregator.hasUpdatesFromAllDistributors()) {
+        if (!aggregatedStats.hasUpdatesFromAllDistributors()) {
             return false;
         }
-        if (prevMayHaveBucketsPending != aggregator.mayHaveBucketsPendingInGlobalSpace()) {
+        if (prevMayHaveMergesPending != checker.mayHaveMergesPendingInGlobalSpace()) {
             return true;
         }
         return false;
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 1a57975c4bc..fcbc8ac1e21 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
@@ -39,7 +39,7 @@ public class StateVersionTracker {
 
     StateVersionTracker() {
         clusterStateView = ClusterStateView.create(currentUnversionedState.getBaselineClusterState());
-        clusterStatsChangeTracker = new ClusterStatsChangeTracker(clusterStateView.getStatsAggregator());
+        clusterStatsChangeTracker = new ClusterStatsChangeTracker(clusterStateView.getStatsAggregator().getAggregatedStats());
     }
 
     void setVersionRetrievedFromZooKeeper(final int version) {
@@ -128,7 +128,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());
-        clusterStatsChangeTracker.updateAggregator(clusterStateView.getStatsAggregator());
+        clusterStatsChangeTracker.updateAggregatedStats(clusterStateView.getStatsAggregator().getAggregatedStats());
     }
 
     private void recordCurrentStateInHistoryAtTime(final long currentTimeMs) {
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/status/statuspage/VdsClusterHtmlRendrer.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/status/statuspage/VdsClusterHtmlRendrer.java
index eb377718614..80aed35cfa1 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/status/statuspage/VdsClusterHtmlRendrer.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/status/statuspage/VdsClusterHtmlRendrer.java
@@ -299,7 +299,7 @@ public class VdsClusterHtmlRendrer {
         private static ContentNodeStats.BucketSpaceStats getStatsForContentNode(ClusterStatsAggregator statsAggregator,
                                                                                 NodeInfo nodeInfo,
                                                                                 String bucketSpace) {
-            ContentNodeStats nodeStats = statsAggregator.getAggregatedStats().getContentNode(nodeInfo.getNodeIndex());
+            ContentNodeStats nodeStats = statsAggregator.getAggregatedStats().getStats().getContentNode(nodeInfo.getNodeIndex());
             if (nodeStats != null) {
                 return nodeStats.getBucketSpace(bucketSpace);
             }
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 e5df589f8fe..15786dc6130 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
@@ -5,61 +5,103 @@ import org.junit.Test;
 
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 public class AggregatedStatsMergePendingCheckerTest {
 
     private static class Fixture {
 
+        private final AggregatedClusterStats mockAggregatedStats = mock(AggregatedClusterStats.class);
         private final AggregatedStatsMergePendingChecker checker;
 
-        public Fixture(ContentClusterStatsBuilder builder) {
-            this.checker = new AggregatedStatsMergePendingChecker(builder.build());
+        public Fixture(ContentClusterStatsBuilder builder, boolean hasUpdatesFromAllDistributors) {
+            when(mockAggregatedStats.getStats()).thenReturn(builder.build());
+            when(mockAggregatedStats.hasUpdatesFromAllDistributors()).thenReturn(hasUpdatesFromAllDistributors);
+            this.checker = new AggregatedStatsMergePendingChecker(mockAggregatedStats);
         }
 
-        public static Fixture fromBucketStats(long bucketsPending) {
+        public static Fixture fromBucketsPending(long bucketsPending) {
             return new Fixture(new ContentClusterStatsBuilder()
-                    .add(1, "default", 5, bucketsPending));
+                    .add(1, "default", 5, bucketsPending), true);
         }
 
         public static Fixture fromInvalidBucketStats() {
             return new Fixture(new ContentClusterStatsBuilder()
-                    .add(1, "default"));
+                    .add(1, "default"), true);
+        }
+
+        public static Fixture fromIncompleteStats() {
+            return new Fixture(new ContentClusterStatsBuilder(), false);
         }
 
         public boolean mayHaveMergesPending(String bucketSpace, int contentNodeIndex) {
             return checker.mayHaveMergesPending(bucketSpace, contentNodeIndex);
         }
 
+        public boolean mayHaveMergesPendingInGlobalSpace() {
+            return checker.mayHaveMergesPendingInGlobalSpace();
+        }
+
     }
 
     @Test
-    public void unknown_content_node_has_no_merges_pending() {
-        Fixture f = Fixture.fromBucketStats(1);
-        assertFalse(f.mayHaveMergesPending("default", 2));
+    public void unknown_content_node_may_have_merges_pending() {
+        Fixture f = Fixture.fromBucketsPending(1);
+        assertTrue(f.mayHaveMergesPending("default", 2));
     }
 
     @Test
     public void unknown_bucket_space_has_no_merges_pending() {
-        Fixture f = Fixture.fromBucketStats(1);
+        Fixture f = Fixture.fromBucketsPending(1);
         assertFalse(f.mayHaveMergesPending("global", 1));
     }
 
     @Test
     public void valid_bucket_space_stats_can_have_no_merges_pending() {
-        Fixture f = Fixture.fromBucketStats(0);
+        Fixture f = Fixture.fromBucketsPending(0);
         assertFalse(f.mayHaveMergesPending("default", 1));
     }
 
     @Test
-    public void valid_bucket_space_stats_can_have_merges_pending() {
-        Fixture f = Fixture.fromBucketStats(1);
+    public void valid_bucket_space_stats_may_have_merges_pending() {
+        Fixture f = Fixture.fromBucketsPending(1);
         assertTrue(f.mayHaveMergesPending("default", 1));
     }
 
     @Test
-    public void invalid_bucket_space_stats_has_merges_pending() {
+    public void invalid_bucket_space_stats_may_have_merges_pending() {
         Fixture f = Fixture.fromInvalidBucketStats();
         assertTrue(f.mayHaveMergesPending("default", 1));
     }
 
+    @Test
+    public void cluster_without_updates_from_all_distributors_may_have_merges_pending() {
+        Fixture f = Fixture.fromIncompleteStats();
+        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/ClusterStatsAggregatorTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregatorTest.java
index 90c9700e001..7f313b31a6c 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
@@ -29,7 +29,7 @@ public class ClusterStatsAggregatorTest {
         }
 
         public void verify(ContentClusterStatsBuilder expectedStats) {
-            assertEquals(expectedStats.build(), aggregator.getAggregatedStats());
+            assertEquals(expectedStats.build(), aggregator.getAggregatedStats().getStats());
         }
 
         public void verify(int distributorIndex, ContentNodeStatsBuilder expectedStats) {
@@ -37,12 +37,9 @@ public class ClusterStatsAggregatorTest {
         }
 
         public boolean hasUpdatesFromAllDistributors() {
-            return aggregator.hasUpdatesFromAllDistributors();
+            return aggregator.getAggregatedStats().hasUpdatesFromAllDistributors();
         }
 
-        public boolean mayHaveBucketsPendingInGlobalSpace() {
-            return aggregator.mayHaveBucketsPendingInGlobalSpace();
-        }
     }
 
     private static class FourNodesFixture extends Fixture {
@@ -146,31 +143,6 @@ public class ClusterStatsAggregatorTest {
     }
 
     @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());
-    }
-
-    @Test
     public void aggregator_can_provide_aggregated_stats_per_distributor() {
         Fixture f = new FourNodesFixture();
 
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 8670c512f14..0439532e02e 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
@@ -15,7 +15,7 @@ public class ClusterStatsChangeTrackerTest {
 
         public Fixture() {
             aggregator = new ClusterStatsAggregator(Sets.newHashSet(1), Sets.newHashSet(2));
-            tracker = new ClusterStatsChangeTracker(aggregator);
+            tracker = new ClusterStatsChangeTracker(aggregator.getAggregatedStats());
         }
 
         public void setBucketsPendingStats() {
@@ -31,9 +31,9 @@ public class ClusterStatsChangeTrackerTest {
                     .add(2, "global", 5, bucketsPending).build());
         }
 
-        public void updateAggregator() {
+        public void updateAggregatedStats() {
             aggregator = new ClusterStatsAggregator(Sets.newHashSet(1), Sets.newHashSet(2));
-            tracker.updateAggregator(aggregator);
+            tracker.updateAggregatedStats(aggregator.getAggregatedStats());
         }
 
         public boolean statsHaveChanged() {
@@ -57,7 +57,7 @@ public class ClusterStatsChangeTrackerTest {
         f.setBucketsPendingStats();
         assertTrue(f.statsHaveChanged());
 
-        f.updateAggregator(); // previous stats may now have buckets pending
+        f.updateAggregatedStats(); // previous stats may now have buckets pending
 
         f.setInSyncStats();
         assertTrue(f.statsHaveChanged());