diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/AbstractAMRMProxyPolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/AbstractAMRMProxyPolicy.java index e853744e106..07cd6db1909 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/AbstractAMRMProxyPolicy.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/AbstractAMRMProxyPolicy.java @@ -20,9 +20,12 @@ import java.util.Map; +import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse; +import org.apache.hadoop.yarn.exceptions.YarnException; import org.apache.hadoop.yarn.server.federation.policies.AbstractConfigurableFederationPolicy; import org.apache.hadoop.yarn.server.federation.policies.dao.WeightedPolicyInfo; import org.apache.hadoop.yarn.server.federation.policies.exceptions.FederationPolicyInitializationException; +import org.apache.hadoop.yarn.server.federation.store.records.SubClusterId; import org.apache.hadoop.yarn.server.federation.store.records.SubClusterIdInfo; /** @@ -44,4 +47,9 @@ public void validate(WeightedPolicyInfo newPolicyInfo) } } + @Override + public void notifyOfResponse(SubClusterId subClusterId, + AllocateResponse response) throws YarnException { + // By default, a stateless policy does not care about responses + } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/LocalAMRMProxyPolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/LocalAMRMProxyPolicy.java new file mode 100644 index 00000000000..ef16e9f3a8f --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/LocalAMRMProxyPolicy.java @@ -0,0 +1,68 @@ +/* + * 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.federation.policies.amrmproxy; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import org.apache.hadoop.yarn.api.records.ResourceRequest; +import org.apache.hadoop.yarn.exceptions.YarnException; +import org.apache.hadoop.yarn.server.federation.policies.FederationPolicyInitializationContext; +import org.apache.hadoop.yarn.server.federation.policies.FederationPolicyInitializationContextValidator; +import org.apache.hadoop.yarn.server.federation.policies.exceptions.FederationPolicyException; +import org.apache.hadoop.yarn.server.federation.policies.exceptions.FederationPolicyInitializationException; +import org.apache.hadoop.yarn.server.federation.store.records.SubClusterId; + +/** + * An implementation of the {@link FederationAMRMProxyPolicy} that simply + * sends the {@link ResourceRequest} to the home subcluster. + */ +public class LocalAMRMProxyPolicy extends AbstractAMRMProxyPolicy { + + /** Identifier of the local subcluster. */ + private SubClusterId localSubcluster; + + @Override + public void reinitialize( + FederationPolicyInitializationContext policyContext) + throws FederationPolicyInitializationException { + + FederationPolicyInitializationContextValidator + .validate(policyContext, this.getClass().getCanonicalName()); + setPolicyContext(policyContext); + + this.localSubcluster = policyContext.getHomeSubcluster(); + } + + @Override + public Map> splitResourceRequests( + List resourceRequests) throws YarnException { + + getActiveSubclusters(); + if (localSubcluster == null) { + throw new FederationPolicyException("No local subcluster available"); + } + + List resourceRequestsCopy = + new ArrayList<>(resourceRequests); + return Collections.singletonMap(localSubcluster, resourceRequestsCopy); + } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/manager/LocalPolicyManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/manager/LocalPolicyManager.java new file mode 100644 index 00000000000..f71f340e7c5 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/manager/LocalPolicyManager.java @@ -0,0 +1,60 @@ +/** + * 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.federation.policies.manager; + +import java.nio.ByteBuffer; +import java.util.Collections; + +import org.apache.hadoop.yarn.server.federation.policies.amrmproxy.LocalAMRMProxyPolicy; +import org.apache.hadoop.yarn.server.federation.policies.dao.WeightedPolicyInfo; +import org.apache.hadoop.yarn.server.federation.policies.exceptions.FederationPolicyInitializationException; +import org.apache.hadoop.yarn.server.federation.policies.router.LocalRouterPolicy; +import org.apache.hadoop.yarn.server.federation.store.records.SubClusterIdInfo; +import org.apache.hadoop.yarn.server.federation.store.records.SubClusterPolicyConfiguration; + +/** + * Policy manager which uses the {@link LocalRouterPolicy} for the Router and + * {@link LocalAMRMProxyPolicy} as the AMRMProxy policy to find an RM. + */ +public class LocalPolicyManager extends AbstractPolicyManager { + + /** Imaginary configuration to fulfill the super class. */ + private WeightedPolicyInfo weightedPolicyInfo; + + public LocalPolicyManager() { + + weightedPolicyInfo = new WeightedPolicyInfo(); + weightedPolicyInfo.setRouterPolicyWeights( + Collections.singletonMap(new SubClusterIdInfo(""), 1.0f)); + weightedPolicyInfo.setAMRMPolicyWeights( + Collections.singletonMap(new SubClusterIdInfo(""), 1.0f)); + + // Hard-codes two compatible policies for Router and AMRMProxy. + routerFederationPolicy = LocalRouterPolicy.class; + amrmProxyFederationPolicy = LocalAMRMProxyPolicy.class; + } + + @Override + public SubClusterPolicyConfiguration serializeConf() + throws FederationPolicyInitializationException { + + ByteBuffer buf = weightedPolicyInfo.toByteBuffer(); + return SubClusterPolicyConfiguration.newInstance( + getQueue(), this.getClass().getCanonicalName(), buf); + } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/router/LocalRouterPolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/router/LocalRouterPolicy.java new file mode 100644 index 00000000000..87ae58b8825 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/router/LocalRouterPolicy.java @@ -0,0 +1,85 @@ +/** + * 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.federation.policies.router; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext; +import org.apache.hadoop.yarn.exceptions.YarnException; +import org.apache.hadoop.yarn.server.federation.policies.FederationPolicyInitializationContext; +import org.apache.hadoop.yarn.server.federation.policies.exceptions.FederationPolicyException; +import org.apache.hadoop.yarn.server.federation.policies.exceptions.FederationPolicyInitializationException; +import org.apache.hadoop.yarn.server.federation.store.records.SubClusterId; +import org.apache.hadoop.yarn.server.federation.store.records.SubClusterInfo; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * This policy always sends to the local subcluster. If the local is not + * available it will rely on {@link WeightedRandomRouterPolicy}. + */ +public class LocalRouterPolicy extends WeightedRandomRouterPolicy { + + private static final Logger LOG = + LoggerFactory.getLogger(LocalRouterPolicy.class); + + /** Identifier of the local subcluster. */ + private SubClusterId localSubcluster; + + @Override + public void reinitialize( + FederationPolicyInitializationContext initializationContext) + throws FederationPolicyInitializationException { + super.reinitialize(initializationContext); + this.localSubcluster = initializationContext.getHomeSubcluster(); + if (this.localSubcluster == null) { + LOG.warn("Cannot get the local home subcluster, it will use random"); + } + } + + @Override + public SubClusterId getHomeSubcluster( + ApplicationSubmissionContext appSubmissionContext, + List blacklist) throws YarnException { + + // null checks and default-queue behavior + validate(appSubmissionContext); + + // If we don't know the local subcluster, use random + if (this.localSubcluster == null) { + return super.getHomeSubcluster(appSubmissionContext, blacklist); + } + + Map activeSubclusters = + getActiveSubclusters(); + if (!activeSubclusters.containsKey(this.localSubcluster)) { + throw new FederationPolicyException("The local subcluster (" + + this.localSubcluster + ") is not available"); + } + if (blacklist.contains(this.localSubcluster)) { + throw new FederationPolicyException("The local subcluster (" + + this.localSubcluster + ") is black lister"); + } + return localSubcluster; + } + +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/TestLocalAMRMProxyPolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/TestLocalAMRMProxyPolicy.java new file mode 100644 index 00000000000..daa40f9a1ae --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/TestLocalAMRMProxyPolicy.java @@ -0,0 +1,81 @@ +/* + * 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.federation.policies.amrmproxy; + +import static org.apache.hadoop.yarn.server.federation.utils.FederationPoliciesTestUtil.createResourceRequests; +import static org.apache.hadoop.yarn.server.federation.utils.FederationPoliciesTestUtil.initializePolicyContext; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.List; +import java.util.Map; + +import org.apache.hadoop.yarn.api.records.ResourceRequest; +import org.apache.hadoop.yarn.server.federation.policies.BaseFederationPoliciesTest; +import org.apache.hadoop.yarn.server.federation.policies.dao.WeightedPolicyInfo; +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; +import org.apache.hadoop.yarn.server.federation.store.records.SubClusterState; +import org.junit.Before; +import org.junit.Test; + +/** + * Simple test class for the {@link LocalAMRMProxyPolicy}. + */ +public class TestLocalAMRMProxyPolicy extends BaseFederationPoliciesTest { + + private static final int NUM_SUBCLUSTERS = 4; + + @Before + public void setUp() throws Exception { + setPolicy(new LocalAMRMProxyPolicy()); + // needed for base test to work + setPolicyInfo(mock(WeightedPolicyInfo.class)); + + for (int i = 0; i < NUM_SUBCLUSTERS; i++) { + SubClusterIdInfo sc = new SubClusterIdInfo("sc" + i); + SubClusterInfo sci = mock(SubClusterInfo.class); + when(sci.getState()).thenReturn(SubClusterState.SC_RUNNING); + when(sci.getSubClusterId()).thenReturn(sc.toId()); + getActiveSubclusters().put(sc.toId(), sci); + } + + initializePolicyContext( + getPolicy(), mock(WeightedPolicyInfo.class), getActiveSubclusters(), "sc2"); + } + + @Test + public void testSplitAllocateRequest() throws Exception { + // verify the request only goes to the local subcluster + String[] hosts = new String[] {"host0", "host1", "host2", "host3"}; + List resourceRequests = createResourceRequests( + hosts, 2 * 1024, 2, 1, 3, null, false); + + FederationAMRMProxyPolicy federationPolicy = + (FederationAMRMProxyPolicy)getPolicy(); + Map> response = + federationPolicy.splitResourceRequests(resourceRequests); + assertEquals(1, response.size()); + assertNotNull(response.get(SubClusterId.newInstance("sc2"))); + assertEquals(9, response.get(SubClusterId.newInstance("sc2")).size()); + } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/manager/TestLocalPolicyManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/manager/TestLocalPolicyManager.java new file mode 100644 index 00000000000..858cff7aacd --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/manager/TestLocalPolicyManager.java @@ -0,0 +1,39 @@ +/** + * 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.federation.policies.manager; + +import org.apache.hadoop.yarn.server.federation.policies.amrmproxy.LocalAMRMProxyPolicy; +import org.apache.hadoop.yarn.server.federation.policies.router.LocalRouterPolicy; +import org.junit.Before; + +/** + * Simple test of {@link LocalPolicyManager}. + */ +public class TestLocalPolicyManager extends BasePolicyManagerTest { + + @Before + public void setup() { + + wfp = new LocalPolicyManager(); + + //set expected params that the base test class will use for tests + expectedPolicyManager = LocalPolicyManager.class; + expectedAMRMProxyPolicy = LocalAMRMProxyPolicy.class; + expectedRouterPolicy = LocalRouterPolicy.class; + } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/utils/FederationPoliciesTestUtil.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/utils/FederationPoliciesTestUtil.java index 4954197ecfe..ce7512cceee 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/utils/FederationPoliciesTestUtil.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/utils/FederationPoliciesTestUtil.java @@ -141,11 +141,21 @@ public static void initializePolicyContext( public static void initializePolicyContext( ConfigurableFederationPolicy policy, - WeightedPolicyInfo policyInfo, Map activeSubclusters) throws YarnException { + WeightedPolicyInfo policyInfo, + Map activeSubclusters) + throws YarnException { + initializePolicyContext( + policy, policyInfo, activeSubclusters, "homesubcluster"); + } + + public static void initializePolicyContext( + ConfigurableFederationPolicy policy, + WeightedPolicyInfo policyInfo, + Map activeSubclusters, + String subclusterId) throws YarnException { FederationPolicyInitializationContext context = new FederationPolicyInitializationContext(null, initResolver(), - initFacade(), SubClusterId.newInstance("homesubcluster")); + initFacade(), SubClusterId.newInstance(subclusterId)); initializePolicyContext(context, policy, policyInfo, activeSubclusters); }