| /** |
| * 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.nio.ByteBuffer; |
| import java.util.ArrayList; |
| import java.util.List; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext; |
| import org.apache.hadoop.yarn.api.records.Priority; |
| import org.apache.hadoop.yarn.api.records.Resource; |
| import org.apache.hadoop.yarn.api.records.ResourceRequest; |
| import org.apache.hadoop.yarn.conf.YarnConfiguration; |
| import org.apache.hadoop.yarn.exceptions.YarnException; |
| import org.apache.hadoop.yarn.server.federation.policies.FederationPolicyInitializationContext; |
| import org.apache.hadoop.yarn.server.federation.policies.dao.WeightedPolicyInfo; |
| import org.apache.hadoop.yarn.server.federation.policies.exceptions.FederationPolicyException; |
| import org.apache.hadoop.yarn.server.federation.resolver.SubClusterResolver; |
| 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.SubClusterPolicyConfiguration; |
| import org.apache.hadoop.yarn.server.federation.utils.FederationPoliciesTestUtil; |
| import org.apache.hadoop.yarn.util.resource.Resources; |
| import org.junit.Assert; |
| import org.junit.Before; |
| import org.junit.Test; |
| |
| /** |
| * Test class to validate the correctness of LocalityRouterPolicy. |
| */ |
| public class TestLocalityRouterPolicy extends TestWeightedRandomRouterPolicy { |
| |
| /* |
| * The MachineList for the default Resolver has the following nodes: |
| * |
| * node1<=>subcluster1 |
| * |
| * node2<=>subcluster2 |
| * |
| * noDE3<=>subcluster3 |
| * |
| * node4<=>subcluster3 |
| * |
| * subcluster0-rack0-host0<=>subcluster0 |
| * |
| * Subcluster1-RACK1-HOST1<=>subcluster1 |
| * |
| * SUBCLUSTER1-RACK1-HOST2<=>subcluster1 |
| * |
| * SubCluster2-RACK3-HOST3<=>subcluster2 |
| */ |
| |
| @Before |
| public void setUp() throws Exception { |
| setPolicy(new LocalityRouterPolicy()); |
| setPolicyInfo(new WeightedPolicyInfo()); |
| |
| configureWeights(4); |
| |
| // initialize policy with context |
| initializePolicy(new YarnConfiguration()); |
| } |
| |
| private void initializePolicy(Configuration conf) throws YarnException { |
| setFederationPolicyContext(new FederationPolicyInitializationContext()); |
| SubClusterResolver resolver = FederationPoliciesTestUtil.initResolver(); |
| getFederationPolicyContext().setFederationSubclusterResolver(resolver); |
| ByteBuffer buf = getPolicyInfo().toByteBuffer(); |
| getFederationPolicyContext().setSubClusterPolicyConfiguration( |
| SubClusterPolicyConfiguration |
| .newInstance("queue1", getPolicy().getClass().getCanonicalName(), |
| buf)); |
| getFederationPolicyContext().setHomeSubcluster(getHomeSubCluster()); |
| setupContext(); |
| } |
| |
| /** |
| * This test validates the correctness in case of the request has 1 node and |
| * the node belongs to an active subcluster. |
| */ |
| @Test |
| public void testNodeInActiveSubCluster() throws YarnException { |
| List<ResourceRequest> requests = new ArrayList<ResourceRequest>(); |
| requests.add(ResourceRequest |
| .newInstance(Priority.UNDEFINED, "node1", Resource.newInstance(10, 1), |
| 1)); |
| requests.add(ResourceRequest |
| .newInstance(Priority.UNDEFINED, "rack1", Resource.newInstance(10, 1), |
| 1)); |
| requests.add(ResourceRequest |
| .newInstance(Priority.UNDEFINED, ResourceRequest.ANY, |
| Resource.newInstance(10, 1), 1)); |
| ApplicationSubmissionContext asc = ApplicationSubmissionContext |
| .newInstance(null, null, null, null, null, false, false, 0, |
| Resources.none(), null, false, null, null); |
| asc.setAMContainerResourceRequests(requests); |
| |
| SubClusterId chosen = |
| ((FederationRouterPolicy) getPolicy()).getHomeSubcluster(asc, null); |
| // If node1 is active, we should choose the sub cluster with node1 |
| if (getActiveSubclusters().containsKey( |
| getFederationPolicyContext().getFederationSubclusterResolver() |
| .getSubClusterForNode("node1").getId())) { |
| Assert.assertEquals( |
| getFederationPolicyContext().getFederationSubclusterResolver() |
| .getSubClusterForNode("node1"), chosen); |
| } |
| // Regardless, we should choose an active SubCluster |
| Assert.assertTrue(getActiveSubclusters().containsKey(chosen)); |
| } |
| |
| /** |
| * This test validates the correctness in case of the request has multiple |
| * ResourceRequests. The tests without ResourceRequests are done in |
| * TestWeightedRandomRouterPolicy. |
| */ |
| @Test |
| public void testMultipleResourceRequests() throws YarnException { |
| List<ResourceRequest> requests = new ArrayList<ResourceRequest>(); |
| requests.add(ResourceRequest |
| .newInstance(Priority.UNDEFINED, "node1", Resource.newInstance(10, 1), |
| 1)); |
| requests.add(ResourceRequest |
| .newInstance(Priority.UNDEFINED, "node2", Resource.newInstance(10, 1), |
| 1)); |
| ApplicationSubmissionContext asc = ApplicationSubmissionContext |
| .newInstance(null, null, null, null, null, false, false, 0, |
| Resources.none(), null, false, null, null); |
| asc.setAMContainerResourceRequests(requests); |
| try { |
| ((FederationRouterPolicy) getPolicy()).getHomeSubcluster(asc, null); |
| Assert.fail(); |
| } catch (FederationPolicyException e) { |
| Assert.assertTrue( |
| e.getMessage().startsWith("Invalid number of resource requests: ")); |
| } |
| } |
| |
| /** |
| * This test validates the correctness in case of the request has 1 node and |
| * the node does not exist in the Resolver MachineList file. |
| */ |
| @Test |
| public void testNodeNotExists() throws YarnException { |
| List<ResourceRequest> requests = new ArrayList<ResourceRequest>(); |
| boolean relaxLocality = true; |
| requests.add(ResourceRequest |
| .newInstance(Priority.UNDEFINED, "node5", Resource.newInstance(10, 1), |
| 1, relaxLocality)); |
| requests.add(ResourceRequest |
| .newInstance(Priority.UNDEFINED, "rack1", Resource.newInstance(10, 1), |
| 1)); |
| requests.add(ResourceRequest |
| .newInstance(Priority.UNDEFINED, ResourceRequest.ANY, |
| Resource.newInstance(10, 1), 1)); |
| ApplicationSubmissionContext asc = ApplicationSubmissionContext |
| .newInstance(null, null, null, null, null, false, false, 0, |
| Resources.none(), null, false, null, null); |
| asc.setAMContainerResourceRequests(requests); |
| |
| try { |
| ((FederationRouterPolicy) getPolicy()).getHomeSubcluster(asc, null); |
| } catch (FederationPolicyException e) { |
| Assert.fail(); |
| } |
| } |
| |
| /** |
| * This test validates the correctness in case of the request has 1 node and |
| * the node is in a blacklist subclusters. |
| */ |
| @Test |
| public void testNodeInABlacklistSubCluster() throws YarnException { |
| // Blacklist SubCluster3 |
| String subClusterToBlacklist = "subcluster3"; |
| // Remember the current value of subcluster3 |
| Float value = |
| getPolicyInfo().getRouterPolicyWeights().get(subClusterToBlacklist); |
| getPolicyInfo().getRouterPolicyWeights() |
| .put(new SubClusterIdInfo(subClusterToBlacklist), 0.0f); |
| initializePolicy(new YarnConfiguration()); |
| |
| FederationPoliciesTestUtil |
| .initializePolicyContext(getFederationPolicyContext(), getPolicy(), |
| getPolicyInfo(), getActiveSubclusters(), new Configuration()); |
| |
| List<ResourceRequest> requests = new ArrayList<ResourceRequest>(); |
| boolean relaxLocality = true; |
| requests.add(ResourceRequest |
| .newInstance(Priority.UNDEFINED, "node4", Resource.newInstance(10, 1), |
| 1, relaxLocality)); |
| requests.add(ResourceRequest |
| .newInstance(Priority.UNDEFINED, "rack1", Resource.newInstance(10, 1), |
| 1)); |
| requests.add(ResourceRequest |
| .newInstance(Priority.UNDEFINED, ResourceRequest.ANY, |
| Resource.newInstance(10, 1), 1)); |
| ApplicationSubmissionContext asc = ApplicationSubmissionContext |
| .newInstance(null, null, null, null, null, false, false, 0, |
| Resources.none(), null, false, null, null); |
| asc.setAMContainerResourceRequests(requests); |
| |
| try { |
| SubClusterId targetId = |
| ((FederationRouterPolicy) getPolicy()).getHomeSubcluster(asc, null); |
| // The selected subcluster HAS no to be the same as the one blacklisted. |
| Assert.assertNotEquals(targetId.getId(), subClusterToBlacklist); |
| } catch (FederationPolicyException e) { |
| Assert.fail(); |
| } |
| |
| // Set again the previous value for the other tests |
| getPolicyInfo().getRouterPolicyWeights() |
| .put(new SubClusterIdInfo(subClusterToBlacklist), value); |
| } |
| |
| /** |
| * This test validates the correctness in case of the request has 1 node and |
| * the node is not in the policy weights. |
| */ |
| @Test |
| public void testNodeNotInPolicy() throws YarnException { |
| // Blacklist SubCluster3 |
| String subClusterToBlacklist = "subcluster3"; |
| // Remember the current value of subcluster3 |
| Float value = |
| getPolicyInfo().getRouterPolicyWeights().get(subClusterToBlacklist); |
| getPolicyInfo().getRouterPolicyWeights().remove(subClusterToBlacklist); |
| initializePolicy(new YarnConfiguration()); |
| |
| FederationPoliciesTestUtil |
| .initializePolicyContext(getFederationPolicyContext(), getPolicy(), |
| getPolicyInfo(), getActiveSubclusters(), new Configuration()); |
| |
| List<ResourceRequest> requests = new ArrayList<ResourceRequest>(); |
| boolean relaxLocality = true; |
| requests.add(ResourceRequest |
| .newInstance(Priority.UNDEFINED, "node4", Resource.newInstance(10, 1), |
| 1, relaxLocality)); |
| requests.add(ResourceRequest |
| .newInstance(Priority.UNDEFINED, "rack1", Resource.newInstance(10, 1), |
| 1)); |
| requests.add(ResourceRequest |
| .newInstance(Priority.UNDEFINED, ResourceRequest.ANY, |
| Resource.newInstance(10, 1), 1)); |
| ApplicationSubmissionContext asc = ApplicationSubmissionContext |
| .newInstance(null, null, null, null, null, false, false, 0, |
| Resources.none(), null, false, null, null); |
| asc.setAMContainerResourceRequests(requests); |
| |
| try { |
| SubClusterId targetId = |
| ((FederationRouterPolicy) getPolicy()).getHomeSubcluster(asc, null); |
| // The selected subcluster HAS no to be the same as the one blacklisted. |
| Assert.assertNotEquals(targetId.getId(), subClusterToBlacklist); |
| } catch (FederationPolicyException e) { |
| Assert.fail(); |
| } |
| |
| // Set again the previous value for the other tests |
| getPolicyInfo().getRouterPolicyWeights() |
| .put(new SubClusterIdInfo(subClusterToBlacklist), value); |
| } |
| } |
| |
