From 8a610e616f724dad14ff46033ba52787525c542b Mon Sep 17 00:00:00 2001 From: slfan1989 <55643692+slfan1989@users.noreply.github.com> Date: Sun, 26 Nov 2023 16:01:35 +0800 Subject: [PATCH] YARN-11614. [Federation] Add Federation PolicyManager Validation Rules. (#6271) Contributed by Shilun Fan. Reviewed-by: Inigo Goiri Signed-off-by: Shilun Fan --- .../yarn/server/router/RouterServerUtil.java | 8 ++++++ .../rmadmin/FederationRMAdminInterceptor.java | 26 ++++++++++++++++--- .../TestFederationRMAdminInterceptor.java | 20 +++++++++++--- 3 files changed, 47 insertions(+), 7 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/RouterServerUtil.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/RouterServerUtil.java index 23902bdc61..744ddc8705 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/RouterServerUtil.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/RouterServerUtil.java @@ -822,4 +822,12 @@ public static boolean isRouterWebProxyEnable(Configuration conf) { return conf.getBoolean(YarnConfiguration.ROUTER_WEBAPP_PROXY_ENABLE, YarnConfiguration.DEFAULT_ROUTER_WEBAPP_PROXY_ENABLE); } + + public static boolean checkPolicyManagerValid(String policyManager, + List supportWeightList) throws YarnException { + if (supportWeightList.contains(policyManager)) { + return true; + } + return false; + } } \ No newline at end of file diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/rmadmin/FederationRMAdminInterceptor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/rmadmin/FederationRMAdminInterceptor.java index 7ade0d4007..230308e361 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/rmadmin/FederationRMAdminInterceptor.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/rmadmin/FederationRMAdminInterceptor.java @@ -75,6 +75,9 @@ import org.apache.hadoop.yarn.server.api.protocolrecords.DeleteFederationApplicationRequest; import org.apache.hadoop.yarn.server.api.protocolrecords.DeleteFederationApplicationResponse; import org.apache.hadoop.yarn.server.federation.failover.FederationProxyProviderUtil; +import org.apache.hadoop.yarn.server.federation.policies.manager.PriorityBroadcastPolicyManager; +import org.apache.hadoop.yarn.server.federation.policies.manager.WeightedHomePolicyManager; +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.federation.store.records.SubClusterInfo; @@ -92,6 +95,7 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.List; +import java.util.Arrays; import java.util.ArrayList; import java.util.Map; import java.util.HashMap; @@ -107,6 +111,8 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; +import static org.apache.hadoop.yarn.server.router.RouterServerUtil.checkPolicyManagerValid; + public class FederationRMAdminInterceptor extends AbstractRMAdminRequestInterceptor { private static final Logger LOG = @@ -115,6 +121,10 @@ public class FederationRMAdminInterceptor extends AbstractRMAdminRequestIntercep private static final String COMMA = ","; private static final String COLON = ":"; + private static final List SUPPORT_WEIGHT_MANAGERS = + new ArrayList<>(Arrays.asList(WeightedLocalityPolicyManager.class.getName(), + PriorityBroadcastPolicyManager.class.getName(), WeightedHomePolicyManager.class.getName())); + private Map adminRMProxies; private FederationStateStoreFacade federationFacade; private final Clock clock = new MonotonicClock(); @@ -924,6 +934,13 @@ public SaveFederationQueuePolicyResponse saveFederationQueuePolicy( RouterServerUtil.logAndThrowException("Missing Queue information.", null); } + String policyManagerClassName = request.getPolicyManagerClassName(); + if (!checkPolicyManagerValid(policyManagerClassName, SUPPORT_WEIGHT_MANAGERS)) { + routerMetrics.incrSaveFederationQueuePolicyFailedRetrieved(); + RouterServerUtil.logAndThrowException(policyManagerClassName + + " does not support the use of queue weights.", null); + } + String amRmWeight = federationQueueWeight.getAmrmWeight(); FederationQueueWeight.checkSubClusterQueueWeightRatioValid(amRmWeight); @@ -935,9 +952,6 @@ public SaveFederationQueuePolicyResponse saveFederationQueuePolicy( try { long startTime = clock.getTime(); - // Step1, get parameters. - String policyManagerClassName = request.getPolicyManagerClassName(); - // Step2, parse amRMPolicyWeights. Map amRMPolicyWeights = getSubClusterWeightMap(amRmWeight); @@ -1346,6 +1360,12 @@ private void saveFederationQueuePolicy(FederationQueueWeight federationQueueWeig RouterServerUtil.logAndThrowException("Missing PolicyManagerClassName information.", null); } + if (!checkPolicyManagerValid(policyManagerClassName, SUPPORT_WEIGHT_MANAGERS)) { + routerMetrics.incrSaveFederationQueuePolicyFailedRetrieved(); + RouterServerUtil.logAndThrowException(policyManagerClassName + + "does not support the use of queue weights.", null); + } + String amRmWeight = federationQueueWeight.getAmrmWeight(); FederationQueueWeight.checkSubClusterQueueWeightRatioValid(amRmWeight); diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/rmadmin/TestFederationRMAdminInterceptor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/rmadmin/TestFederationRMAdminInterceptor.java index 62f1eee845..a1339e4190 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/rmadmin/TestFederationRMAdminInterceptor.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/rmadmin/TestFederationRMAdminInterceptor.java @@ -643,15 +643,27 @@ public void testSaveFederationQueuePolicyErrorRequest() throws Exception { LambdaTestUtils.intercept(YarnException.class, "Missing Queue information.", () -> interceptor.saveFederationQueuePolicy(request)); - // routerWeight / amrmWeight - // The sum of the routerWeight is not equal to 1. + // PolicyManager needs to support weight FederationQueueWeight federationQueueWeight2 = FederationQueueWeight.newInstance( "SC-1:0.7,SC-2:0.3", "SC-1:0.8,SC-2:0.3", "1.0"); SaveFederationQueuePolicyRequest request2 = - SaveFederationQueuePolicyRequest.newInstance("root.a", federationQueueWeight2, "-"); + SaveFederationQueuePolicyRequest.newInstance("root.a", federationQueueWeight2, + "TestPolicyManager"); + LambdaTestUtils.intercept(YarnException.class, + "TestPolicyManager does not support the use of queue weights.", + () -> interceptor.saveFederationQueuePolicy(request2)); + + // routerWeight / amrmWeight + // The sum of the routerWeight is not equal to 1. + String policyTypeName = WeightedLocalityPolicyManager.class.getCanonicalName(); + FederationQueueWeight federationQueueWeight3 = FederationQueueWeight.newInstance( + "SC-1:0.7,SC-2:0.3", "SC-1:0.8,SC-2:0.3", "1.0"); + SaveFederationQueuePolicyRequest request3 = + SaveFederationQueuePolicyRequest.newInstance("root.a", federationQueueWeight3, + policyTypeName); LambdaTestUtils.intercept(YarnException.class, "The sum of ratios for all subClusters must be equal to 1.", - () -> interceptor.saveFederationQueuePolicy(request2)); + () -> interceptor.saveFederationQueuePolicy(request3)); } @Test