diff options
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()); + } + +} |