You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by bo...@apache.org on 2018/08/15 16:59:49 UTC
hadoop git commit: YARN-7708. [GPG] Load based policy generator.
Contributed by Young Chen.
Repository: hadoop
Updated Branches:
refs/heads/YARN-7402 b8e718082 -> 91dd58b76
YARN-7708. [GPG] Load based policy generator. Contributed by Young Chen.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/91dd58b7
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/91dd58b7
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/91dd58b7
Branch: refs/heads/YARN-7402
Commit: 91dd58b76a14c694145290e511999b847837e945
Parents: b8e7180
Author: Botong Huang <bo...@apache.org>
Authored: Wed Aug 15 09:45:50 2018 -0700
Committer: Botong Huang <bo...@apache.org>
Committed: Wed Aug 15 09:46:13 2018 -0700
----------------------------------------------------------------------
.../dev-support/findbugs-exclude.xml | 4 +
.../policygenerator/GlobalPolicy.java | 2 +-
.../policygenerator/LoadBasedGlobalPolicy.java | 255 +++++++++++++++++++
.../TestLoadBasedGlobalPolicy.java | 211 +++++++++++++++
4 files changed, 471 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/91dd58b7/hadoop-yarn-project/hadoop-yarn/dev-support/findbugs-exclude.xml
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/dev-support/findbugs-exclude.xml b/hadoop-yarn-project/hadoop-yarn/dev-support/findbugs-exclude.xml
index 9fcafad..6e062c7 100644
--- a/hadoop-yarn-project/hadoop-yarn/dev-support/findbugs-exclude.xml
+++ b/hadoop-yarn-project/hadoop-yarn/dev-support/findbugs-exclude.xml
@@ -209,6 +209,10 @@
<Class name="org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.RecoveryComparator" />
<Bug pattern="SE_COMPARATOR_SHOULD_BE_SERIALIZABLE" />
</Match>
+ <Match>
+ <Class name="org.apache.hadoop.yarn.server.globalpolicygenerator.policygenerator.LoadBasedGlobalPolicy$SortByDescendingLoad" />
+ <Bug pattern="SE_COMPARATOR_SHOULD_BE_SERIALIZABLE" />
+ </Match>
<!-- Ignore some irrelevant class name warning -->
<Match>
<Class name="org.apache.hadoop.yarn.api.records.SerializedException" />
http://git-wip-us.apache.org/repos/asf/hadoop/blob/91dd58b7/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-globalpolicygenerator/src/main/java/org/apache/hadoop/yarn/server/globalpolicygenerator/policygenerator/GlobalPolicy.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-globalpolicygenerator/src/main/java/org/apache/hadoop/yarn/server/globalpolicygenerator/policygenerator/GlobalPolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-globalpolicygenerator/src/main/java/org/apache/hadoop/yarn/server/globalpolicygenerator/policygenerator/GlobalPolicy.java
index 38d762d..fcd22c9 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-globalpolicygenerator/src/main/java/org/apache/hadoop/yarn/server/globalpolicygenerator/policygenerator/GlobalPolicy.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-globalpolicygenerator/src/main/java/org/apache/hadoop/yarn/server/globalpolicygenerator/policygenerator/GlobalPolicy.java
@@ -51,7 +51,7 @@ public abstract class GlobalPolicy implements Configurable {
* duplicate calls to the same * endpoints as the GlobalPolicy is invoked
* once per queue.
*/
- protected Map<Class, String> registerPaths() {
+ protected Map<Class<?>, String> registerPaths() {
// Default register nothing
return Collections.emptyMap();
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/91dd58b7/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-globalpolicygenerator/src/main/java/org/apache/hadoop/yarn/server/globalpolicygenerator/policygenerator/LoadBasedGlobalPolicy.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-globalpolicygenerator/src/main/java/org/apache/hadoop/yarn/server/globalpolicygenerator/policygenerator/LoadBasedGlobalPolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-globalpolicygenerator/src/main/java/org/apache/hadoop/yarn/server/globalpolicygenerator/policygenerator/LoadBasedGlobalPolicy.java
new file mode 100644
index 0000000..03bd48c
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-globalpolicygenerator/src/main/java/org/apache/hadoop/yarn/server/globalpolicygenerator/policygenerator/LoadBasedGlobalPolicy.java
@@ -0,0 +1,255 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.yarn.server.globalpolicygenerator.policygenerator;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
+import org.apache.hadoop.yarn.server.federation.policies.manager.FederationPolicyManager;
+import org.apache.hadoop.yarn.server.federation.policies.manager.WeightedLocalityPolicyManager;
+import org.apache.hadoop.yarn.server.federation.store.records.SubClusterId;
+import org.apache.hadoop.yarn.server.federation.store.records.SubClusterIdInfo;
+import org.apache.hadoop.yarn.server.globalpolicygenerator.GPGUtils;
+import org.apache.hadoop.yarn.server.resourcemanager.webapp.RMWSConsts;
+import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ClusterMetricsInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Load based policy that generates weighted policies by scaling
+ * the cluster load (based on pending) to a weight from 0.0 to 1.0.
+ */
+public class LoadBasedGlobalPolicy extends GlobalPolicy {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(LoadBasedGlobalPolicy.class);
+
+ private static final String FEDERATION_GPG_LOAD_BASED_PREFIX =
+ YarnConfiguration.FEDERATION_GPG_PREFIX + "policy.generator.load-based.";
+
+ public static final String FEDERATION_GPG_LOAD_BASED_MIN_PENDING =
+ FEDERATION_GPG_LOAD_BASED_PREFIX + "pending.minimum";
+ public static final int DEFAULT_FEDERATION_GPG_LOAD_BASED_MIN_PENDING = 100;
+
+ public static final String FEDERATION_GPG_LOAD_BASED_MAX_PENDING =
+ FEDERATION_GPG_LOAD_BASED_PREFIX + "pending.maximum";
+ public static final int DEFAULT_FEDERATION_GPG_LOAD_BASED_MAX_PENDING = 1000;
+
+ public static final String FEDERATION_GPG_LOAD_BASED_MIN_WEIGHT =
+ FEDERATION_GPG_LOAD_BASED_PREFIX + "weight.minimum";
+ public static final float DEFAULT_FEDERATION_GPG_LOAD_BASED_MIN_WEIGHT = 0.0f;
+
+ public static final String FEDERATION_GPG_LOAD_BASED_MAX_EDIT =
+ FEDERATION_GPG_LOAD_BASED_PREFIX + "edit.maximum";
+ public static final int DEFAULT_FEDERATION_GPG_LOAD_BASED_MAX_EDIT = 3;
+
+ public static final String FEDERATION_GPG_LOAD_BASED_SCALING =
+ FEDERATION_GPG_LOAD_BASED_PREFIX + "scaling";
+ public static final String DEFAULT_FEDERATION_GPG_LOAD_BASED_SCALING =
+ Scaling.LINEAR.name();
+
+ public enum Scaling {
+ LINEAR,
+ QUADRATIC,
+ LOG,
+ NONE
+ }
+
+ // Minimum pending count before the policy starts scaling down the weights
+ private int minPending;
+ // Maximum pending count before policy stops scaling down the weights
+ //(they'll be set to min weight)
+ private int maxPending;
+ // Minimum weight that a sub cluster will be assigned
+ private float minWeight;
+ // Maximum number of weights that can be scaled down simultaneously
+ private int maxEdit;
+ // Scaling type
+ private Scaling scaling = Scaling.NONE;
+
+ @Override
+ public void setConf(Configuration conf) {
+ super.setConf(conf);
+ minPending = conf.getInt(FEDERATION_GPG_LOAD_BASED_MIN_PENDING,
+ DEFAULT_FEDERATION_GPG_LOAD_BASED_MIN_PENDING);
+ maxPending = conf.getInt(FEDERATION_GPG_LOAD_BASED_MAX_PENDING,
+ DEFAULT_FEDERATION_GPG_LOAD_BASED_MAX_PENDING);
+ minWeight = conf.getFloat(FEDERATION_GPG_LOAD_BASED_MIN_WEIGHT,
+ DEFAULT_FEDERATION_GPG_LOAD_BASED_MIN_WEIGHT);
+ maxEdit = conf.getInt(FEDERATION_GPG_LOAD_BASED_MAX_EDIT,
+ DEFAULT_FEDERATION_GPG_LOAD_BASED_MAX_EDIT);
+ try {
+ scaling = Scaling.valueOf(conf.get(FEDERATION_GPG_LOAD_BASED_SCALING,
+ DEFAULT_FEDERATION_GPG_LOAD_BASED_SCALING));
+ } catch (IllegalArgumentException e) {
+ LOG.warn("Invalid scaling mode provided", e);
+ }
+ // Check that all configuration values are valid
+ if (!(minPending <= maxPending)) {
+ throw new YarnRuntimeException("minPending=" + minPending
+ + " must be less than or equal to maxPending=" + maxPending);
+ }
+ if (!(minWeight >= 0 && minWeight < 1)) {
+ throw new YarnRuntimeException(
+ "minWeight=" + minWeight + " must be within range [0,1)");
+ }
+ }
+
+ @Override
+ protected Map<Class<?>, String> registerPaths() {
+ // Register for the endpoints we want to receive information on
+ Map<Class<?>, String> map = new HashMap<>();
+ map.put(ClusterMetricsInfo.class, RMWSConsts.METRICS);
+ return map;
+ }
+
+ @Override
+ protected FederationPolicyManager updatePolicy(String queueName,
+ Map<SubClusterId, Map<Class, Object>> clusterInfo,
+ FederationPolicyManager currentManager) {
+ Map<SubClusterId, ClusterMetricsInfo> clusterMetrics = new HashMap<>();
+ for (Map.Entry<SubClusterId, Map<Class, Object>> e : clusterInfo
+ .entrySet()) {
+ clusterMetrics.put(e.getKey(),
+ (ClusterMetricsInfo) e.getValue().get(ClusterMetricsInfo.class));
+ }
+ if (currentManager == null) {
+ LOG.info("Creating load based weighted policy queue {}", queueName);
+ Map<SubClusterIdInfo, Float> weights = getTargetWeights(clusterMetrics);
+ WeightedLocalityPolicyManager manager =
+ new WeightedLocalityPolicyManager();
+ manager.setQueue(queueName);
+ manager.getWeightedPolicyInfo().setAMRMPolicyWeights(weights);
+ manager.getWeightedPolicyInfo().setRouterPolicyWeights(weights);
+ currentManager = manager;
+ } else if (currentManager instanceof WeightedLocalityPolicyManager) {
+ Map<SubClusterIdInfo, Float> weights = getTargetWeights(clusterMetrics);
+ LOG.info("Updating policy for queue {} based on cluster load to: {}",
+ queueName, weights);
+ WeightedLocalityPolicyManager manager =
+ (WeightedLocalityPolicyManager) currentManager;
+ manager.getWeightedPolicyInfo().setAMRMPolicyWeights(weights);
+ manager.getWeightedPolicyInfo().setRouterPolicyWeights(weights);
+ } else {
+ LOG.warn("Policy for queue {} is of type {}, expected {}", queueName,
+ currentManager.getClass(), WeightedLocalityPolicyManager.class);
+ }
+ return currentManager;
+ }
+
+ @VisibleForTesting
+ protected Map<SubClusterIdInfo, Float> getTargetWeights(
+ Map<SubClusterId, ClusterMetricsInfo> clusterMetrics) {
+ Map<SubClusterIdInfo, Float> weights =
+ GPGUtils.createUniformWeights(clusterMetrics.keySet());
+ List<SubClusterId> scs = new ArrayList<>(clusterMetrics.keySet());
+ // Sort the sub clusters into descending order based on pending load
+ scs.sort(new SortByDescendingLoad(clusterMetrics));
+ // Keep the top N loaded sub clusters
+ scs = scs.subList(0, Math.min(maxEdit, scs.size()));
+ for (SubClusterId sc : scs) {
+ LOG.info("Updating weight for sub cluster {}", sc.toString());
+ int pending = clusterMetrics.get(sc).getAppsPending();
+ if (pending <= minPending) {
+ LOG.info("Load ({}) is lower than minimum ({}), skipping", pending,
+ minPending);
+ } else if (pending < maxPending) {
+ float weight = 1.0f;
+ // The different scaling strategies should all map values from the
+ // range min_pending+1 to max_pending to the range min_weight to 1.0f
+ // so we pre process and simplify the domain to some value [1, MAX-MIN)
+ int val = pending - minPending;
+ int maxVal = maxPending - minPending;
+ switch (scaling) {
+ case NONE:
+ break;
+ case LINEAR:
+ weight = (float) (maxVal - val) / (float) (maxVal);
+ break;
+ case QUADRATIC:
+ double maxValQuad = Math.pow(maxVal, 2);
+ double valQuad = Math.pow(val, 2);
+ weight = (float) (maxValQuad - valQuad) / (float) (maxValQuad);
+ break;
+ case LOG:
+ double maxValLog = Math.log(maxVal);
+ double valLog = Math.log(val);
+ weight = (float) (maxValLog - valLog) / (float) (maxValLog);
+ break;
+ }
+ // Scale the weights to respect the config minimum
+ weight = weight * (1.0f - minWeight);
+ weight += minWeight;
+ weights.put(new SubClusterIdInfo(sc), weight);
+ LOG.info("Load ({}) is within maximum ({}), setting weights via {} "
+ + "scale to {}", pending, maxPending, scaling, weight);
+ } else {
+ weights.put(new SubClusterIdInfo(sc), minWeight);
+ LOG.info(
+ "Load ({}) exceeded maximum ({}), setting weight to minimum: {}",
+ pending, maxPending, minWeight);
+ }
+ }
+ validateWeights(weights);
+ return weights;
+ }
+
+ /**
+ * Helper to avoid all zero weights. If weights are all zero, they're reset
+ * to one
+ * @param weights weights to validate
+ */
+ private void validateWeights(Map<SubClusterIdInfo, Float> weights) {
+ for(Float w : weights.values()) {
+ // If we find a nonzero weight, we're validated
+ if(w > 0.0f) {
+ return;
+ }
+ }
+ LOG.warn("All " + weights.size()
+ + " generated weights were 0.0f. Resetting to 1.0f");
+ // All weights were zero. Reset all back to 1.0
+ for(SubClusterIdInfo id : weights.keySet()) {
+ weights.put(id, 1.0f);
+ }
+ }
+
+ private static final class SortByDescendingLoad
+ implements Comparator<SubClusterId> {
+
+ private Map<SubClusterId, ClusterMetricsInfo> clusterMetrics;
+
+ private SortByDescendingLoad(
+ Map<SubClusterId, ClusterMetricsInfo> clusterMetrics) {
+ this.clusterMetrics = clusterMetrics;
+ }
+
+ public int compare(SubClusterId a, SubClusterId b) {
+ // Sort by pending load
+ return clusterMetrics.get(b).getAppsPending() - clusterMetrics.get(a)
+ .getAppsPending();
+ }
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/91dd58b7/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-globalpolicygenerator/src/test/java/org/apache/hadoop/yarn/server/globalpolicygenerator/policygenerator/TestLoadBasedGlobalPolicy.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-globalpolicygenerator/src/test/java/org/apache/hadoop/yarn/server/globalpolicygenerator/policygenerator/TestLoadBasedGlobalPolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-globalpolicygenerator/src/test/java/org/apache/hadoop/yarn/server/globalpolicygenerator/policygenerator/TestLoadBasedGlobalPolicy.java
new file mode 100644
index 0000000..34c3e57
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-globalpolicygenerator/src/test/java/org/apache/hadoop/yarn/server/globalpolicygenerator/policygenerator/TestLoadBasedGlobalPolicy.java
@@ -0,0 +1,211 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.server.globalpolicygenerator.policygenerator;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.server.federation.store.records.SubClusterId;
+import org.apache.hadoop.yarn.server.federation.store.records.SubClusterIdInfo;
+import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ClusterMetricsInfo;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Unit test for the Load Based Global Policy.
+ */
+public class TestLoadBasedGlobalPolicy {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(TestLoadBasedGlobalPolicy.class);
+
+ private static final int NUM_SC = 3;
+ private static final float DELTA = 0.00001f;
+
+ private static final int MIN_PENDING = 100;
+ private static final int MAX_PENDING = 500;
+
+ private List<SubClusterId> subClusterIds;
+ private Map<SubClusterId, ClusterMetricsInfo> clusterMetricsInfos;
+ private Map<SubClusterIdInfo, Float> weights;
+
+ private Configuration conf;
+ private LoadBasedGlobalPolicy policyGenerator;
+
+ public TestLoadBasedGlobalPolicy() {
+ conf = new Configuration();
+ policyGenerator = new LoadBasedGlobalPolicy();
+ }
+
+ @Before
+ public void setUp() {
+
+ conf.setInt(LoadBasedGlobalPolicy.FEDERATION_GPG_LOAD_BASED_MAX_EDIT, 2);
+ conf.setInt(LoadBasedGlobalPolicy.FEDERATION_GPG_LOAD_BASED_MIN_PENDING,
+ MIN_PENDING);
+ conf.setInt(LoadBasedGlobalPolicy.FEDERATION_GPG_LOAD_BASED_MAX_PENDING,
+ MAX_PENDING);
+ conf.setFloat(LoadBasedGlobalPolicy.FEDERATION_GPG_LOAD_BASED_MIN_WEIGHT,
+ 0.0f);
+ conf.set(LoadBasedGlobalPolicy.FEDERATION_GPG_LOAD_BASED_SCALING,
+ LoadBasedGlobalPolicy.Scaling.LINEAR.name());
+ policyGenerator.setConf(conf);
+
+ subClusterIds = new ArrayList<>();
+ clusterMetricsInfos = new HashMap<>();
+ // Set up sub clusters
+ for (int i = 0; i < NUM_SC; ++i) {
+ // Sub cluster Id
+ SubClusterId id = SubClusterId.newInstance("sc" + i);
+ subClusterIds.add(id);
+
+ // Cluster metrics info
+ ClusterMetricsInfo metricsInfo = new ClusterMetricsInfo();
+ metricsInfo.setAppsPending(50);
+ clusterMetricsInfos.put(id, metricsInfo);
+ }
+ }
+
+ @Test
+ public void testSimpleTargetWeights() {
+ weights = policyGenerator.getTargetWeights(clusterMetricsInfos);
+ assertEquals(weights.size(), 3);
+ assertEquals(1.0, getWeight(0), DELTA);
+ assertEquals(1.0, getWeight(1), DELTA);
+ assertEquals(1.0, getWeight(2), DELTA);
+ }
+
+ @Test
+ public void testLoadTargetWeights() {
+ getMetric(0).setAppsPending(100);
+ weights = policyGenerator.getTargetWeights(clusterMetricsInfos);
+ assertEquals(weights.size(), 3);
+ assertEquals(1.0, getWeight(0), DELTA);
+ assertEquals(1.0, getWeight(1), DELTA);
+ assertEquals(1.0, getWeight(2), DELTA);
+ getMetric(0).setAppsPending(500);
+ weights = policyGenerator.getTargetWeights(clusterMetricsInfos);
+ assertEquals(weights.size(), 3);
+ assertEquals(0.0, getWeight(0), DELTA);
+ assertEquals(1.0, getWeight(1), DELTA);
+ assertEquals(1.0, getWeight(2), DELTA);
+ }
+
+ @Test
+ public void testMaxEdit() {
+ // The policy should be able to edit 2 weights
+ getMetric(0).setAppsPending(MAX_PENDING + 200);
+ getMetric(1).setAppsPending(MAX_PENDING + 100);
+ weights = policyGenerator.getTargetWeights(clusterMetricsInfos);
+ assertEquals(weights.size(), 3);
+ assertEquals(0.0, getWeight(0), DELTA);
+ assertEquals(0.0, getWeight(1), DELTA);
+ assertEquals(1.0, getWeight(2), DELTA);
+ // After updating the config, it should only edit the most loaded
+ conf.setInt(LoadBasedGlobalPolicy.FEDERATION_GPG_LOAD_BASED_MAX_EDIT, 1);
+ policyGenerator.setConf(conf);
+ weights = policyGenerator.getTargetWeights(clusterMetricsInfos);
+ assertEquals(weights.size(), 3);
+ assertEquals(0.0, getWeight(0), DELTA);
+ assertEquals(1.0, getWeight(1), DELTA);
+ assertEquals(1.0, getWeight(2), DELTA);
+ }
+
+ @Test
+ public void testMinWeight() {
+ // If a minimum weight is set, the generator should not go below it
+ conf.setFloat(LoadBasedGlobalPolicy.FEDERATION_GPG_LOAD_BASED_MIN_WEIGHT,
+ 0.5f);
+ policyGenerator.setConf(conf);
+ getMetric(0).setAppsPending(Integer.MAX_VALUE);
+ weights = policyGenerator.getTargetWeights(clusterMetricsInfos);
+ assertEquals(weights.size(), 3);
+ assertEquals(0.5, getWeight(0), DELTA);
+ assertEquals(1.0, getWeight(1), DELTA);
+ assertEquals(1.0, getWeight(2), DELTA);
+ }
+
+ @Test
+ public void testScaling() {
+ LOG.info("Testing that the generator weights are monotonically"
+ + " decreasing regardless of scaling method");
+ for (LoadBasedGlobalPolicy.Scaling scaling :
+ new LoadBasedGlobalPolicy.Scaling[] {
+ LoadBasedGlobalPolicy.Scaling.LINEAR,
+ LoadBasedGlobalPolicy.Scaling.QUADRATIC,
+ LoadBasedGlobalPolicy.Scaling.LOG }) {
+ LOG.info("Testing {} scaling...", scaling);
+ conf.set(LoadBasedGlobalPolicy.DEFAULT_FEDERATION_GPG_LOAD_BASED_SCALING,
+ scaling.name());
+ policyGenerator.setConf(conf);
+ // Test a continuous range for scaling
+ float prevWeight = 1.01f;
+ for (int load = 0; load < MAX_PENDING * 2; ++load) {
+ getMetric(0).setAppsPending(load);
+ weights = policyGenerator.getTargetWeights(clusterMetricsInfos);
+ if (load < MIN_PENDING) {
+ // Below the minimum load, it should stay 1.0f
+ assertEquals(1.0f, getWeight(0), DELTA);
+ } else if (load < MAX_PENDING) {
+ // In the specified range, the weight should consistently decrease
+ float weight = getWeight(0);
+ assertTrue(weight < prevWeight);
+ prevWeight = weight;
+ } else {
+ // Above the maximum load, it should stay 0.0f
+ assertEquals(0.0f, getWeight(0), DELTA);
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testNonZero() {
+ // If all generated weights are zero, they should be set back to one
+ conf.setFloat(LoadBasedGlobalPolicy.FEDERATION_GPG_LOAD_BASED_MIN_WEIGHT,
+ 0.0f);
+ conf.setInt(LoadBasedGlobalPolicy.FEDERATION_GPG_LOAD_BASED_MAX_EDIT, 3);
+ policyGenerator.setConf(conf);
+ getMetric(0).setAppsPending(Integer.MAX_VALUE);
+ getMetric(1).setAppsPending(Integer.MAX_VALUE);
+ getMetric(2).setAppsPending(Integer.MAX_VALUE);
+ weights = policyGenerator.getTargetWeights(clusterMetricsInfos);
+ assertEquals(weights.size(), 3);
+ assertEquals(1.0, getWeight(0), DELTA);
+ assertEquals(1.0, getWeight(1), DELTA);
+ assertEquals(1.0, getWeight(2), DELTA);
+ }
+
+ private float getWeight(int sc) {
+ return weights.get(new SubClusterIdInfo(subClusterIds.get(sc)));
+ }
+
+ private ClusterMetricsInfo getMetric(int sc) {
+ return clusterMetricsInfos.get(subClusterIds.get(sc));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org