Source code

001/*
002 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
003 * agreements. See the NOTICE file distributed with this work for additional information regarding
004 * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
005 * "License"); you may not use this file except in compliance with the License. You may obtain a
006 * copy of the License at
007 * <p>
008 * http://www.apache.org/licenses/LICENSE-2.0
009 * <p>
010 * Unless required by applicable law or agreed to in writing, software distributed under the License
011 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
012 * or implied. See the License for the specific language governing permissions and limitations under
013 * the License.
014 */
015package org.apache.hadoop.hbase.master.balancer;
016
017import static junit.framework.TestCase.assertNotNull;
018import static junit.framework.TestCase.assertTrue;
019import static org.junit.Assert.assertNull;
020import java.io.IOException;
021import java.util.Arrays;
022import java.util.Collections;
023import java.util.LinkedList;
024import java.util.List;
025import java.util.Map;
026import java.util.Queue;
027import java.util.Random;
028import java.util.TreeMap;
029import java.util.concurrent.ThreadLocalRandom;
030
031import org.apache.hadoop.fs.FileSystem;
032import org.apache.hadoop.hbase.HBaseClassTestRule;
033import org.apache.hadoop.hbase.HBaseTestingUtility;
034import org.apache.hadoop.hbase.HConstants;
035import org.apache.hadoop.hbase.ServerName;
036import org.apache.hadoop.hbase.client.RegionInfo;
037import org.apache.hadoop.hbase.client.RegionReplicaUtil;
038import org.apache.hadoop.hbase.master.RackManager;
039import org.apache.hadoop.hbase.master.RegionPlan;
040import org.apache.hadoop.hbase.testclassification.MasterTests;
041import org.apache.hadoop.hbase.testclassification.MediumTests;
042import org.junit.BeforeClass;
043import org.junit.ClassRule;
044import org.junit.Test;
045import org.junit.experimental.categories.Category;
046import org.slf4j.Logger;
047import org.slf4j.LoggerFactory;
048
049@Category({ MasterTests.class, MediumTests.class })
050public class TestStochasticLoadBalancerHeterogeneousCost extends BalancerTestBase {
051  @ClassRule
052  public static final HBaseClassTestRule CLASS_RULE =
053      HBaseClassTestRule.forClass(TestStochasticLoadBalancerHeterogeneousCost.class);
054
055  private static final Logger LOG =
056      LoggerFactory.getLogger(TestStochasticLoadBalancerHeterogeneousCost.class);
057  private static final double ALLOWED_WINDOW = 1.20;
058  private static final HBaseTestingUtility HTU = new HBaseTestingUtility();
059  private static String RULES_FILE;
060
061  @BeforeClass
062  public static void beforeAllTests() throws IOException {
063    BalancerTestBase.conf = HTU.getConfiguration();
064    BalancerTestBase.conf.setFloat("hbase.master.balancer.stochastic.regionCountCost", 0);
065    BalancerTestBase.conf.setFloat("hbase.master.balancer.stochastic.primaryRegionCountCost", 0);
066    BalancerTestBase.conf.setFloat("hbase.master.balancer.stochastic.tableSkewCost", 0);
067    BalancerTestBase.conf.setBoolean("hbase.master.balancer.stochastic.runMaxSteps", true);
068    BalancerTestBase.conf.set(StochasticLoadBalancer.COST_FUNCTIONS_COST_FUNCTIONS_KEY,
069      HeterogeneousRegionCountCostFunction.class.getName());
070    // Need to ensure test dir has been created.
071    assertTrue(FileSystem.get(HTU.getConfiguration()).mkdirs(HTU.getDataTestDir()));
072    RULES_FILE = HTU.getDataTestDir(
073      TestStochasticLoadBalancerHeterogeneousCostRules.DEFAULT_RULES_FILE_NAME).toString();
074    BalancerTestBase.conf.set(
075      HeterogeneousRegionCountCostFunction.HBASE_MASTER_BALANCER_HETEROGENEOUS_RULES_FILE,
076      RULES_FILE);
077    BalancerTestBase.loadBalancer = new StochasticLoadBalancer();
078    BalancerTestBase.loadBalancer.setConf(BalancerTestBase.conf);
079  }
080
081  @Test
082  public void testDefault() throws IOException {
083    final List<String> rules = Collections.emptyList();
084
085    final int numNodes = 2;
086    final int numRegions = 300;
087    final int numRegionsPerServer = 250;
088
089    // Initial state: { rs1:50 , rs0:250 }
090    // Cluster can hold 300/400 regions (75%)
091    // Expected balanced Cluster: { rs0:150 , rs1:150 }
092    this.testHeterogeneousWithCluster(numNodes, numRegions, numRegionsPerServer, rules);
093  }
094
095  @Test
096  public void testOneGroup() throws IOException {
097    final List<String> rules = Collections.singletonList("rs.* 100");
098
099    final int numNodes = 4;
100    final int numRegions = 300;
101    final int numRegionsPerServer = 30;
102
103    // Initial state: { rs0:30 , rs1:30 , rs2:30 , rs3:210 }.
104    // The cluster can hold 300/400 regions (75%)
105    // Expected balanced Cluster: { rs0:75 , rs1:75 , rs2:75 , rs3:75 }
106    this.testHeterogeneousWithCluster(numNodes, numRegions, numRegionsPerServer, rules);
107  }
108
109  @Test
110  public void testTwoGroups() throws IOException {
111    final List<String> rules = Arrays.asList("rs[0-4] 200", "rs[5-9] 50");
112
113    final int numNodes = 10;
114    final int numRegions = 500;
115    final int numRegionsPerServer = 50;
116
117    // Initial state: { rs0:50 , rs1:50 , rs2:50 , rs3:50 , rs4:50 , rs5:50 , rs6:50 , rs7:50 ,
118    // rs8:50 , rs9:50 }
119    // the cluster can hold 500/1250 regions (40%)
120    // Expected balanced Cluster: { rs5:20 , rs6:20 , rs7:20 , rs8:20 , rs9:20 , rs0:80 , rs1:80 ,
121    // rs2:80 , rs3:80 , rs4:80 }
122    this.testHeterogeneousWithCluster(numNodes, numRegions, numRegionsPerServer, rules);
123  }
124
125  @Test
126  public void testFourGroups() throws IOException {
127    final List<String> rules = Arrays.asList("rs[1-3] 200", "rs[4-7] 250", "rs[8-9] 100");
128
129    final int numNodes = 10;
130    final int numRegions = 800;
131    final int numRegionsPerServer = 80;
132
133    // Initial state: { rs0:80 , rs1:80 , rs2:80 , rs3:80 , rs4:80 , rs5:80 , rs6:80 , rs7:80 ,
134    // rs8:80 , rs9:80 }
135    // Cluster can hold 800/2000 regions (40%)
136    // Expected balanced Cluster: { rs8:40 , rs9:40 , rs2:80 , rs3:80 , rs1:82 , rs0:94 , rs4:96 ,
137    // rs5:96 , rs6:96 , rs7:96 }
138    this.testHeterogeneousWithCluster(numNodes, numRegions, numRegionsPerServer, rules);
139  }
140
141  @Test
142  public void testOverloaded() throws IOException {
143    final List<String> rules = Collections.singletonList("rs[0-1] 50");
144
145    final int numNodes = 2;
146    final int numRegions = 120;
147    final int numRegionsPerServer = 60;
148
149    TestStochasticLoadBalancerHeterogeneousCostRules.createRulesFile(RULES_FILE);
150    final Map<ServerName, List<RegionInfo>> serverMap =
151        this.createServerMap(numNodes, numRegions, numRegionsPerServer, 1, 1);
152    final List<RegionPlan> plans =
153        BalancerTestBase.loadBalancer.balanceTable(HConstants.ENSEMBLE_TABLE_NAME, serverMap);
154    // As we disabled all the other cost functions, balancing only according to
155    // the heterogeneous cost function should return nothing.
156    assertNull(plans);
157  }
158
159  private void testHeterogeneousWithCluster(final int numNodes, final int numRegions,
160      final int numRegionsPerServer, final List<String> rules) throws IOException {
161
162    TestStochasticLoadBalancerHeterogeneousCostRules.createRulesFile(RULES_FILE, rules);
163    final Map<ServerName, List<RegionInfo>> serverMap =
164        this.createServerMap(numNodes, numRegions, numRegionsPerServer, 1, 1);
165    this.testWithCluster(serverMap, null, true, false);
166  }
167
168  protected void testWithCluster(final Map<ServerName, List<RegionInfo>> serverMap,
169      final RackManager rackManager, final boolean assertFullyBalanced,
170      final boolean assertFullyBalancedForReplicas) {
171    final List<ServerAndLoad> list = this.convertToList(serverMap);
172    LOG.info("Mock Cluster : " + this.printMock(list) + " " + this.printStats(list));
173
174    BalancerTestBase.loadBalancer.setRackManager(rackManager);
175
176    // Run the balancer.
177    final List<RegionPlan> plans =
178        BalancerTestBase.loadBalancer.balanceTable(HConstants.ENSEMBLE_TABLE_NAME, serverMap);
179    assertNotNull(plans);
180
181    // Check to see that this actually got to a stable place.
182    if (assertFullyBalanced || assertFullyBalancedForReplicas) {
183      // Apply the plan to the mock cluster.
184      final List<ServerAndLoad> balancedCluster = this.reconcile(list, plans, serverMap);
185
186      // Print out the cluster loads to make debugging easier.
187      LOG.info("Mock Balanced cluster : " + this.printMock(balancedCluster));
188
189      if (assertFullyBalanced) {
190        final List<RegionPlan> secondPlans =
191            BalancerTestBase.loadBalancer.balanceTable(HConstants.ENSEMBLE_TABLE_NAME, serverMap);
192        assertNull(secondPlans);
193
194        // create external cost function to retrieve limit
195        // for each RS
196        final HeterogeneousRegionCountCostFunction cf =
197            new HeterogeneousRegionCountCostFunction(conf);
198        assertNotNull(cf);
199        BaseLoadBalancer.Cluster cluster =
200            new BaseLoadBalancer.Cluster(serverMap, null, null, null);
201        cf.init(cluster);
202
203        // checking that we all hosts have a number of regions below their limit
204        for (final ServerAndLoad serverAndLoad : balancedCluster) {
205          final ServerName sn = serverAndLoad.getServerName();
206          final int numberRegions = serverAndLoad.getLoad();
207          final int limit = cf.findLimitForRS(sn);
208
209          double usage = (double) numberRegions / (double) limit;
210          LOG.debug(
211            sn.getHostname() + ":" + numberRegions + "/" + limit + "(" + (usage * 100) + "%)");
212
213          // as the balancer is stochastic, we cannot check exactly the result of the balancing,
214          // hence the allowedWindow parameter
215          assertTrue("Host " + sn.getHostname() + " should be below "
216              + cf.overallUsage * ALLOWED_WINDOW * 100 + "%; " + cf.overallUsage +
217              ", " + usage + ", " + numberRegions + ", " + limit,
218            usage <= cf.overallUsage * ALLOWED_WINDOW);
219        }
220      }
221
222      if (assertFullyBalancedForReplicas) {
223        this.assertRegionReplicaPlacement(serverMap, rackManager);
224      }
225    }
226  }
227
228  @Override
229  protected Map<ServerName, List<RegionInfo>> createServerMap(int numNodes, int numRegions,
230      int numRegionsPerServer, int replication, int numTables) {
231    // construct a cluster of numNodes, having a total of numRegions. Each RS will hold
232    // numRegionsPerServer many regions except for the last one, which will host all the
233    // remaining regions
234    int[] cluster = new int[numNodes];
235    for (int i = 0; i < numNodes; i++) {
236      cluster[i] = numRegionsPerServer;
237    }
238    cluster[cluster.length - 1] = numRegions - ((cluster.length - 1) * numRegionsPerServer);
239    Map<ServerName, List<RegionInfo>> clusterState = mockClusterServers(cluster, numTables);
240    if (replication > 0) {
241      // replicate the regions to the same servers
242      for (List<RegionInfo> regions : clusterState.values()) {
243        int length = regions.size();
244        for (int i = 0; i < length; i++) {
245          for (int r = 1; r < replication; r++) {
246            regions.add(RegionReplicaUtil.getRegionInfoForReplica(regions.get(i), r));
247          }
248        }
249      }
250    }
251
252    return clusterState;
253  }
254
255  @Override
256  protected TreeMap<ServerName, List<RegionInfo>> mockClusterServers(int[] mockCluster,
257      int numTables) {
258    int numServers = mockCluster.length;
259    TreeMap<ServerName, List<RegionInfo>> servers = new TreeMap<>();
260    for (int i = 0; i < numServers; i++) {
261      int numRegions = mockCluster[i];
262      ServerAndLoad sal = createServer("rs" + i);
263      List<RegionInfo> regions = randomRegions(numRegions, numTables);
264      servers.put(sal.getServerName(), regions);
265    }
266    return servers;
267  }
268
269  private Queue<ServerName> serverQueue = new LinkedList<>();
270
271  private ServerAndLoad createServer(final String host) {
272    if (!this.serverQueue.isEmpty()) {
273      ServerName sn = this.serverQueue.poll();
274      return new ServerAndLoad(sn, 0);
275    }
276    Random rand = ThreadLocalRandom.current();
277    int port = rand.nextInt(60000);
278    long startCode = rand.nextLong();
279    ServerName sn = ServerName.valueOf(host, port, startCode);
280    return new ServerAndLoad(sn, 0);
281  }
282}