View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  package org.apache.hadoop.hbase.master.balancer;
19  
20  import com.google.common.annotations.VisibleForTesting;
21  import java.util.ArrayDeque;
22  import java.util.Arrays;
23  import java.util.Collection;
24  import java.util.Deque;
25  import java.util.HashMap;
26  import java.util.LinkedList;
27  import java.util.List;
28  import java.util.Map;
29  import java.util.Map.Entry;
30  import java.util.Random;
31  
32  import org.apache.commons.logging.Log;
33  import org.apache.commons.logging.LogFactory;
34  import org.apache.hadoop.hbase.classification.InterfaceAudience;
35  import org.apache.hadoop.conf.Configuration;
36  import org.apache.hadoop.hbase.ClusterStatus;
37  import org.apache.hadoop.hbase.HBaseInterfaceAudience;
38  import org.apache.hadoop.hbase.HConstants;
39  import org.apache.hadoop.hbase.HRegionInfo;
40  import org.apache.hadoop.hbase.RegionLoad;
41  import org.apache.hadoop.hbase.ServerLoad;
42  import org.apache.hadoop.hbase.ServerName;
43  import org.apache.hadoop.hbase.TableName;
44  import org.apache.hadoop.hbase.master.MasterServices;
45  import org.apache.hadoop.hbase.master.RegionPlan;
46  import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer.Cluster.Action;
47  import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer.Cluster.Action.Type;
48  import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer.Cluster.AssignRegionAction;
49  import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer.Cluster.MoveRegionAction;
50  import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer.Cluster.SwapRegionsAction;
51  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
52  
53  /**
54   * <p>This is a best effort load balancer. Given a Cost function F(C) =&gt; x It will
55   * randomly try and mutate the cluster to Cprime. If F(Cprime) &lt; F(C) then the
56   * new cluster state becomes the plan. It includes costs functions to compute the cost of:</p>
57   * <ul>
58   * <li>Region Load</li>
59   * <li>Table Load</li>
60   * <li>Data Locality</li>
61   * <li>Memstore Sizes</li>
62   * <li>Storefile Sizes</li>
63   * </ul>
64   *
65   *
66   * <p>Every cost function returns a number between 0 and 1 inclusive; where 0 is the lowest cost
67   * best solution, and 1 is the highest possible cost and the worst solution.  The computed costs are
68   * scaled by their respective multipliers:</p>
69   *
70   * <ul>
71   *   <li>hbase.master.balancer.stochastic.regionLoadCost</li>
72   *   <li>hbase.master.balancer.stochastic.moveCost</li>
73   *   <li>hbase.master.balancer.stochastic.tableLoadCost</li>
74   *   <li>hbase.master.balancer.stochastic.localityCost</li>
75   *   <li>hbase.master.balancer.stochastic.memstoreSizeCost</li>
76   *   <li>hbase.master.balancer.stochastic.storefileSizeCost</li>
77   * </ul>
78   *
79   * <p>In addition to the above configurations, the balancer can be tuned by the following
80   * configuration values:</p>
81   * <ul>
82   *   <li>hbase.master.balancer.stochastic.maxMoveRegions which
83   *   controls what the max number of regions that can be moved in a single invocation of this
84   *   balancer.</li>
85   *   <li>hbase.master.balancer.stochastic.stepsPerRegion is the coefficient by which the number of
86   *   regions is multiplied to try and get the number of times the balancer will
87   *   mutate all servers.</li>
88   *   <li>hbase.master.balancer.stochastic.maxSteps which controls the maximum number of times that
89   *   the balancer will try and mutate all the servers. The balancer will use the minimum of this
90   *   value and the above computation.</li>
91   * </ul>
92   *
93   * <p>This balancer is best used with hbase.master.loadbalance.bytable set to false
94   * so that the balancer gets the full picture of all loads on the cluster.</p>
95   */
96  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.CONFIG)
97  @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="IS2_INCONSISTENT_SYNC",
98    justification="Complaint is about costFunctions not being synchronized; not end of the world")
99  public class StochasticLoadBalancer extends BaseLoadBalancer {
100 
101   protected static final String STEPS_PER_REGION_KEY =
102       "hbase.master.balancer.stochastic.stepsPerRegion";
103   protected static final String MAX_STEPS_KEY =
104       "hbase.master.balancer.stochastic.maxSteps";
105   protected static final String RUN_MAX_STEPS_KEY =
106       "hbase.master.balancer.stochastic.runMaxSteps";
107   protected static final String MAX_RUNNING_TIME_KEY =
108       "hbase.master.balancer.stochastic.maxRunningTime";
109   protected static final String KEEP_REGION_LOADS =
110       "hbase.master.balancer.stochastic.numRegionLoadsToRemember";
111   private static final String TABLE_FUNCTION_SEP = "_";
112 
113   private static final Random RANDOM = new Random(System.currentTimeMillis());
114   private static final Log LOG = LogFactory.getLog(StochasticLoadBalancer.class);
115 
116   Map<String, Deque<RegionLoad>> loads = new HashMap<String, Deque<RegionLoad>>();
117 
118   // values are defaults
119   private int maxSteps = 1000000;
120   private boolean runMaxSteps = false;
121   private int stepsPerRegion = 800;
122   private long maxRunningTime = 30 * 1000 * 1; // 30 seconds.
123   private int numRegionLoadsToRemember = 15;
124 
125   private CandidateGenerator[] candidateGenerators;
126   private CostFromRegionLoadFunction[] regionLoadFunctions;
127   private CostFunction[] costFunctions; // FindBugs: Wants this protected; IS2_INCONSISTENT_SYNC
128 
129   // to save and report costs to JMX
130   private Double curOverallCost = 0d;
131   private Double[] tempFunctionCosts;
132   private Double[] curFunctionCosts;
133 
134   // Keep locality based picker and cost function to alert them
135   // when new services are offered
136   private LocalityBasedCandidateGenerator localityCandidateGenerator;
137   private LocalityCostFunction localityCost;
138   private RegionReplicaHostCostFunction regionReplicaHostCostFunction;
139   private RegionReplicaRackCostFunction regionReplicaRackCostFunction;
140   private boolean isByTable = false;
141   private TableName tableName = null;
142 
143   /**
144    * The constructor that pass a MetricsStochasticBalancer to BaseLoadBalancer to replace its
145    * default MetricsBalancer
146    */
147   public StochasticLoadBalancer() {
148     super(new MetricsStochasticBalancer());
149   }
150 
151   @Override
152   public void onConfigurationChange(Configuration conf) {
153     setConf(conf);
154   }
155 
156   @Override
157   public synchronized void setConf(Configuration conf) {
158     super.setConf(conf);
159     LOG.info("loading config");
160 
161     maxSteps = conf.getInt(MAX_STEPS_KEY, maxSteps);
162 
163     stepsPerRegion = conf.getInt(STEPS_PER_REGION_KEY, stepsPerRegion);
164     maxRunningTime = conf.getLong(MAX_RUNNING_TIME_KEY, maxRunningTime);
165     runMaxSteps = conf.getBoolean(RUN_MAX_STEPS_KEY, runMaxSteps);
166 
167     numRegionLoadsToRemember = conf.getInt(KEEP_REGION_LOADS, numRegionLoadsToRemember);
168     isByTable = conf.getBoolean(HConstants.HBASE_MASTER_LOADBALANCE_BYTABLE, isByTable);
169 
170     if (localityCandidateGenerator == null) {
171       localityCandidateGenerator = new LocalityBasedCandidateGenerator(services);
172     }
173     localityCost = new LocalityCostFunction(conf, services);
174 
175     if (candidateGenerators == null) {
176       candidateGenerators = new CandidateGenerator[] {
177           new RandomCandidateGenerator(),
178           new LoadCandidateGenerator(),
179           localityCandidateGenerator,
180           new RegionReplicaRackCandidateGenerator(),
181       };
182     }
183 
184     regionLoadFunctions = new CostFromRegionLoadFunction[] {
185       new ReadRequestCostFunction(conf),
186       new WriteRequestCostFunction(conf),
187       new MemstoreSizeCostFunction(conf),
188       new StoreFileCostFunction(conf)
189     };
190 
191     regionReplicaHostCostFunction = new RegionReplicaHostCostFunction(conf);
192     regionReplicaRackCostFunction = new RegionReplicaRackCostFunction(conf);
193 
194     costFunctions = new CostFunction[]{
195       new RegionCountSkewCostFunction(conf),
196       new PrimaryRegionCountSkewCostFunction(conf),
197       new MoveCostFunction(conf),
198       localityCost,
199       new TableSkewCostFunction(conf),
200       regionReplicaHostCostFunction,
201       regionReplicaRackCostFunction,
202       regionLoadFunctions[0],
203       regionLoadFunctions[1],
204       regionLoadFunctions[2],
205       regionLoadFunctions[3],
206     };
207 
208     curFunctionCosts= new Double[costFunctions.length];
209     tempFunctionCosts= new Double[costFunctions.length];
210 
211   }
212 
213   @Override
214   protected void setSlop(Configuration conf) {
215     this.slop = conf.getFloat("hbase.regions.slop", 0.001F);
216   }
217 
218   @Override
219   public synchronized void setClusterStatus(ClusterStatus st) {
220     super.setClusterStatus(st);
221     updateRegionLoad();
222     for(CostFromRegionLoadFunction cost : regionLoadFunctions) {
223       cost.setClusterStatus(st);
224     }
225 
226     // update metrics size
227     try {
228       // by-table or ensemble mode
229       int tablesCount = isByTable ? services.getTableDescriptors().getAll().size() : 1;
230       int functionsCount = getCostFunctionNames().length;
231 
232       updateMetricsSize(tablesCount * (functionsCount + 1)); // +1 for overall
233     } catch (Exception e) {
234       LOG.error("failed to get the size of all tables", e);
235     }
236   }
237 
238   /**
239    * Update the number of metrics that are reported to JMX
240    */
241   public void updateMetricsSize(int size) {
242     if (metricsBalancer instanceof MetricsStochasticBalancer) {
243         ((MetricsStochasticBalancer) metricsBalancer).updateMetricsSize(size);
244     }
245   }
246 
247   @Override
248   public synchronized void setMasterServices(MasterServices masterServices) {
249     super.setMasterServices(masterServices);
250     this.localityCost.setServices(masterServices);
251     this.localityCandidateGenerator.setServices(masterServices);
252 
253   }
254 
255   @Override
256   protected synchronized boolean areSomeRegionReplicasColocated(Cluster c) {
257     regionReplicaHostCostFunction.init(c);
258     if (regionReplicaHostCostFunction.cost() > 0) return true;
259     regionReplicaRackCostFunction.init(c);
260     if (regionReplicaRackCostFunction.cost() > 0) return true;
261     return false;
262   }
263 
264   @Override
265   public synchronized List<RegionPlan> balanceCluster(TableName tableName, Map<ServerName,
266     List<HRegionInfo>> clusterState) {
267     this.tableName = tableName;
268     return balanceCluster(clusterState);
269   }
270 
271   @VisibleForTesting
272   Cluster.Action nextAction(Cluster cluster) {
273     return candidateGenerators[(RANDOM.nextInt(candidateGenerators.length))]
274             .generate(cluster);
275   }
276 
277   /**
278    * Given the cluster state this will try and approach an optimal balance. This
279    * should always approach the optimal state given enough steps.
280    */
281   @Override
282   public synchronized List<RegionPlan> balanceCluster(Map<ServerName,
283     List<HRegionInfo>> clusterState) {
284     List<RegionPlan> plans = balanceMasterRegions(clusterState);
285     if (plans != null || clusterState == null || clusterState.size() <= 1) {
286       return plans;
287     }
288 
289     if (masterServerName != null && clusterState.containsKey(masterServerName)) {
290       if (clusterState.size() <= 2) {
291         return null;
292       }
293       clusterState = new HashMap<ServerName, List<HRegionInfo>>(clusterState);
294       clusterState.remove(masterServerName);
295     }
296 
297     // On clusters with lots of HFileLinks or lots of reference files,
298     // instantiating the storefile infos can be quite expensive.
299     // Allow turning this feature off if the locality cost is not going to
300     // be used in any computations.
301     RegionLocationFinder finder = null;
302     if (this.localityCost != null && this.localityCost.getMultiplier() > 0) {
303       finder = this.regionFinder;
304     }
305 
306     //The clusterState that is given to this method contains the state
307     //of all the regions in the table(s) (that's true today)
308     // Keep track of servers to iterate through them.
309     Cluster cluster = new Cluster(clusterState, loads, finder, rackManager);
310 
311     if (!needsBalance(cluster)) {
312       return null;
313     }
314 
315     long startTime = EnvironmentEdgeManager.currentTime();
316 
317     initCosts(cluster);
318 
319     double currentCost = computeCost(cluster, Double.MAX_VALUE);
320     curOverallCost = currentCost;
321     for (int i = 0; i < this.curFunctionCosts.length; i++) {
322       curFunctionCosts[i] = tempFunctionCosts[i];
323     }
324 
325     double initCost = currentCost;
326     double newCost = currentCost;
327 
328     long computedMaxSteps = 0;
329     if (runMaxSteps) {
330       computedMaxSteps = Math.max(this.maxSteps,
331           ((long)cluster.numRegions * (long)this.stepsPerRegion * (long)cluster.numServers));
332     } else {
333       long calculatedMaxSteps =
334           (long) cluster.numRegions * (long) this.stepsPerRegion * (long) cluster.numServers;
335       computedMaxSteps = Math.min(this.maxSteps, calculatedMaxSteps);
336       if (calculatedMaxSteps > maxSteps) {
337         LOG.warn(String.format("calculatedMaxSteps:%d for loadbalancer's stochastic walk is larger "
338                 + "than maxSteps:%d. Hence load balancing may not work well. Setting parameter "
339                 + "\"hbase.master.balancer.stochastic.runMaxSteps\" to true to overcome this issue."
340                 + "(This config change does not require service restart)", calculatedMaxSteps,
341             maxSteps));
342 
343       }
344     }
345     LOG.info("start StochasticLoadBalancer.balancer, initCost=" + currentCost +
346         " computedMaxSteps: " + computedMaxSteps);
347 
348     // Perform a stochastic walk to see if we can get a good fit.
349     long step;
350 
351     for (step = 0; step < computedMaxSteps; step++) {
352       Cluster.Action action = nextAction(cluster);
353 
354       if (action.type == Type.NULL) {
355         continue;
356       }
357 
358       cluster.doAction(action);
359       updateCostsWithAction(cluster, action);
360 
361       newCost = computeCost(cluster, currentCost);
362 
363       // Should this be kept?
364       if (newCost < currentCost) {
365         currentCost = newCost;
366 
367         // save for JMX
368         curOverallCost = currentCost;
369         for (int i = 0; i < this.curFunctionCosts.length; i++) {
370           curFunctionCosts[i] = tempFunctionCosts[i];
371         }
372       } else {
373         // Put things back the way they were before.
374         // TODO: undo by remembering old values
375         Action undoAction = action.undoAction();
376         cluster.doAction(undoAction);
377         updateCostsWithAction(cluster, undoAction);
378       }
379 
380       if (EnvironmentEdgeManager.currentTime() - startTime >
381           maxRunningTime) {
382         break;
383       }
384     }
385     long endTime = EnvironmentEdgeManager.currentTime();
386 
387     metricsBalancer.balanceCluster(endTime - startTime);
388 
389     // update costs metrics
390     updateStochasticCosts(tableName, curOverallCost, curFunctionCosts);
391     if (initCost > currentCost) {
392       plans = createRegionPlans(cluster);
393       if (LOG.isDebugEnabled()) {
394         LOG.debug("Finished computing new load balance plan.  Computation took "
395             + (endTime - startTime) + "ms to try " + step
396             + " different iterations.  Found a solution that moves "
397             + plans.size() + " regions; Going from a computed cost of "
398             + initCost + " to a new cost of " + currentCost);
399       }
400 
401       return plans;
402     }
403     if (LOG.isDebugEnabled()) {
404       LOG.debug("Could not find a better load balance plan.  Tried "
405           + step + " different configurations in " + (endTime - startTime)
406           + "ms, and did not find anything with a computed cost less than " + initCost);
407     }
408     return null;
409   }
410 
411   /**
412    * update costs to JMX
413    */
414   private void updateStochasticCosts(TableName tableName, Double overall, Double[] subCosts) {
415     if (tableName == null) return;
416 
417     // check if the metricsBalancer is MetricsStochasticBalancer before casting
418     if (metricsBalancer instanceof MetricsStochasticBalancer) {
419       MetricsStochasticBalancer balancer = (MetricsStochasticBalancer) metricsBalancer;
420       // overall cost
421       balancer.updateStochasticCost(tableName.getNameAsString(),
422         "Overall", "Overall cost", overall);
423 
424       // each cost function
425       for (int i = 0; i < costFunctions.length; i++) {
426         CostFunction costFunction = costFunctions[i];
427         String costFunctionName = costFunction.getClass().getSimpleName();
428         Double costPercent = (overall == 0) ? 0 : (subCosts[i] / overall);
429         // TODO: cost function may need a specific description
430         balancer.updateStochasticCost(tableName.getNameAsString(), costFunctionName,
431           "The percent of " + costFunctionName, costPercent);
432       }
433     }
434   }
435 
436 
437   /**
438    * Create all of the RegionPlan's needed to move from the initial cluster state to the desired
439    * state.
440    *
441    * @param cluster The state of the cluster
442    * @return List of RegionPlan's that represent the moves needed to get to desired final state.
443    */
444   private List<RegionPlan> createRegionPlans(Cluster cluster) {
445     List<RegionPlan> plans = new LinkedList<RegionPlan>();
446     for (int regionIndex = 0;
447          regionIndex < cluster.regionIndexToServerIndex.length; regionIndex++) {
448       int initialServerIndex = cluster.initialRegionIndexToServerIndex[regionIndex];
449       int newServerIndex = cluster.regionIndexToServerIndex[regionIndex];
450 
451       if (initialServerIndex != newServerIndex) {
452         HRegionInfo region = cluster.regions[regionIndex];
453         ServerName initialServer = cluster.servers[initialServerIndex];
454         ServerName newServer = cluster.servers[newServerIndex];
455 
456         if (LOG.isTraceEnabled()) {
457           LOG.trace("Moving Region " + region.getEncodedName() + " from server "
458               + initialServer.getHostname() + " to " + newServer.getHostname());
459         }
460         RegionPlan rp = new RegionPlan(region, initialServer, newServer);
461         plans.add(rp);
462       }
463     }
464     return plans;
465   }
466 
467   /**
468    * Store the current region loads.
469    */
470   private synchronized void updateRegionLoad() {
471     // We create a new hashmap so that regions that are no longer there are removed.
472     // However we temporarily need the old loads so we can use them to keep the rolling average.
473     Map<String, Deque<RegionLoad>> oldLoads = loads;
474     loads = new HashMap<String, Deque<RegionLoad>>();
475 
476     for (ServerName sn : clusterStatus.getServers()) {
477       ServerLoad sl = clusterStatus.getLoad(sn);
478       if (sl == null) {
479         continue;
480       }
481       for (Entry<byte[], RegionLoad> entry : sl.getRegionsLoad().entrySet()) {
482         String regionNameAsString = HRegionInfo.getRegionNameAsString(entry.getKey());
483         Deque<RegionLoad> rLoads = oldLoads.get(regionNameAsString);
484         if (rLoads == null) {
485           // There was nothing there
486           rLoads = new ArrayDeque<RegionLoad>();
487         } else if (rLoads.size() >= numRegionLoadsToRemember) {
488           rLoads.remove();
489         }
490         rLoads.add(entry.getValue());
491         loads.put(regionNameAsString, rLoads);
492       }
493     }
494 
495     for(CostFromRegionLoadFunction cost : regionLoadFunctions) {
496       cost.setLoads(loads);
497     }
498   }
499 
500   protected void initCosts(Cluster cluster) {
501     for (CostFunction c:costFunctions) {
502       c.init(cluster);
503     }
504   }
505 
506   protected void updateCostsWithAction(Cluster cluster, Action action) {
507     for (CostFunction c : costFunctions) {
508       c.postAction(action);
509     }
510   }
511 
512   /**
513    * Get the names of the cost functions
514    */
515   public String[] getCostFunctionNames() {
516     if (costFunctions == null) return null;
517     String[] ret = new String[costFunctions.length];
518     for (int i = 0; i < costFunctions.length; i++) {
519       CostFunction c = costFunctions[i];
520       ret[i] = c.getClass().getSimpleName();
521     }
522 
523     return ret;
524   }
525 
526   /**
527    * This is the main cost function.  It will compute a cost associated with a proposed cluster
528    * state.  All different costs will be combined with their multipliers to produce a double cost.
529    *
530    * @param cluster The state of the cluster
531    * @param previousCost the previous cost. This is used as an early out.
532    * @return a double of a cost associated with the proposed cluster state.  This cost is an
533    *         aggregate of all individual cost functions.
534    */
535   protected double computeCost(Cluster cluster, double previousCost) {
536     double total = 0;
537 
538     for (int i = 0; i < costFunctions.length; i++) {
539       CostFunction c = costFunctions[i];
540       this.tempFunctionCosts[i] = 0.0;
541 
542       if (c.getMultiplier() <= 0) {
543         continue;
544       }
545 
546       Float multiplier = c.getMultiplier();
547       Double cost = c.cost();
548 
549       this.tempFunctionCosts[i] = multiplier*cost;
550       total += this.tempFunctionCosts[i];
551 
552       if (total > previousCost) {
553         break;
554       }
555     }
556 
557     return total;
558   }
559 
560   /** Generates a candidate action to be applied to the cluster for cost function search */
561   abstract static class CandidateGenerator {
562     abstract Cluster.Action generate(Cluster cluster);
563 
564     /**
565      * From a list of regions pick a random one. Null can be returned which
566      * {@link StochasticLoadBalancer#balanceCluster(Map)} recognize as signal to try a region move
567      * rather than swap.
568      *
569      * @param cluster        The state of the cluster
570      * @param server         index of the server
571      * @param chanceOfNoSwap Chance that this will decide to try a move rather
572      *                       than a swap.
573      * @return a random {@link HRegionInfo} or null if an asymmetrical move is
574      *         suggested.
575      */
576     protected int pickRandomRegion(Cluster cluster, int server, double chanceOfNoSwap) {
577       // Check to see if this is just a move.
578       if (cluster.regionsPerServer[server].length == 0 || RANDOM.nextFloat() < chanceOfNoSwap) {
579         // signal a move only.
580         return -1;
581       }
582       int rand = RANDOM.nextInt(cluster.regionsPerServer[server].length);
583       return cluster.regionsPerServer[server][rand];
584 
585     }
586     protected int pickRandomServer(Cluster cluster) {
587       if (cluster.numServers < 1) {
588         return -1;
589       }
590 
591       return RANDOM.nextInt(cluster.numServers);
592     }
593 
594     protected int pickRandomRack(Cluster cluster) {
595       if (cluster.numRacks < 1) {
596         return -1;
597       }
598 
599       return RANDOM.nextInt(cluster.numRacks);
600     }
601 
602     protected int pickOtherRandomServer(Cluster cluster, int serverIndex) {
603       if (cluster.numServers < 2) {
604         return -1;
605       }
606       while (true) {
607         int otherServerIndex = pickRandomServer(cluster);
608         if (otherServerIndex != serverIndex) {
609           return otherServerIndex;
610         }
611       }
612     }
613 
614     protected int pickOtherRandomRack(Cluster cluster, int rackIndex) {
615       if (cluster.numRacks < 2) {
616         return -1;
617       }
618       while (true) {
619         int otherRackIndex = pickRandomRack(cluster);
620         if (otherRackIndex != rackIndex) {
621           return otherRackIndex;
622         }
623       }
624     }
625 
626     protected Cluster.Action pickRandomRegions(Cluster cluster,
627                                                        int thisServer,
628                                                        int otherServer) {
629       if (thisServer < 0 || otherServer < 0) {
630         return Cluster.NullAction;
631       }
632 
633       // Decide who is most likely to need another region
634       int thisRegionCount = cluster.getNumRegions(thisServer);
635       int otherRegionCount = cluster.getNumRegions(otherServer);
636 
637       // Assign the chance based upon the above
638       double thisChance = (thisRegionCount > otherRegionCount) ? 0 : 0.5;
639       double otherChance = (thisRegionCount <= otherRegionCount) ? 0 : 0.5;
640 
641       int thisRegion = pickRandomRegion(cluster, thisServer, thisChance);
642       int otherRegion = pickRandomRegion(cluster, otherServer, otherChance);
643 
644       return getAction(thisServer, thisRegion, otherServer, otherRegion);
645     }
646 
647     protected Cluster.Action getAction(int fromServer, int fromRegion,
648         int toServer, int toRegion) {
649       if (fromServer < 0 || toServer < 0) {
650         return Cluster.NullAction;
651       }
652       if (fromRegion > 0 && toRegion > 0) {
653         return new Cluster.SwapRegionsAction(fromServer, fromRegion,
654           toServer, toRegion);
655       } else if (fromRegion > 0) {
656         return new Cluster.MoveRegionAction(fromRegion, fromServer, toServer);
657       } else if (toRegion > 0) {
658         return new Cluster.MoveRegionAction(toRegion, toServer, fromServer);
659       } else {
660         return Cluster.NullAction;
661       }
662     }
663   }
664 
665   static class RandomCandidateGenerator extends CandidateGenerator {
666 
667     @Override
668     Cluster.Action generate(Cluster cluster) {
669 
670       int thisServer = pickRandomServer(cluster);
671 
672       // Pick the other server
673       int otherServer = pickOtherRandomServer(cluster, thisServer);
674 
675       return pickRandomRegions(cluster, thisServer, otherServer);
676     }
677   }
678 
679   static class LoadCandidateGenerator extends CandidateGenerator {
680 
681     @Override
682     Cluster.Action generate(Cluster cluster) {
683       cluster.sortServersByRegionCount();
684       int thisServer = pickMostLoadedServer(cluster, -1);
685       int otherServer = pickLeastLoadedServer(cluster, thisServer);
686 
687       return pickRandomRegions(cluster, thisServer, otherServer);
688     }
689 
690     private int pickLeastLoadedServer(final Cluster cluster, int thisServer) {
691       Integer[] servers = cluster.serverIndicesSortedByRegionCount;
692 
693       int index = 0;
694       while (servers[index] == null || servers[index] == thisServer) {
695         index++;
696         if (index == servers.length) {
697           return -1;
698         }
699       }
700       return servers[index];
701     }
702 
703     private int pickMostLoadedServer(final Cluster cluster, int thisServer) {
704       Integer[] servers = cluster.serverIndicesSortedByRegionCount;
705 
706       int index = servers.length - 1;
707       while (servers[index] == null || servers[index] == thisServer) {
708         index--;
709         if (index < 0) {
710           return -1;
711         }
712       }
713       return servers[index];
714     }
715   }
716 
717   static class LocalityBasedCandidateGenerator extends CandidateGenerator {
718 
719     private MasterServices masterServices;
720 
721     LocalityBasedCandidateGenerator(MasterServices masterServices) {
722       this.masterServices = masterServices;
723     }
724 
725     @Override
726     Cluster.Action generate(Cluster cluster) {
727       if (this.masterServices == null) {
728         int thisServer = pickRandomServer(cluster);
729         // Pick the other server
730         int otherServer = pickOtherRandomServer(cluster, thisServer);
731         return pickRandomRegions(cluster, thisServer, otherServer);
732       }
733 
734       int thisServer = pickRandomServer(cluster);
735       int thisRegion;
736       if (thisServer == -1) {
737         LOG.warn("Could not pick lowest locality region server");
738         return Cluster.NullAction;
739       } else {
740       // Pick lowest locality region on this server
741         thisRegion = pickLowestLocalityRegionOnServer(cluster, thisServer);
742       }
743 
744       if (thisRegion == -1) {
745         return Cluster.NullAction;
746       }
747 
748       // Pick the least loaded server with good locality for the region
749       int otherServer = cluster.getLeastLoadedTopServerForRegion(thisRegion, thisServer);
750 
751       if (otherServer == -1) {
752         return Cluster.NullAction;
753       }
754 
755       // Let the candidate region be moved to its highest locality server.
756       int otherRegion = -1;
757 
758       return getAction(thisServer, thisRegion, otherServer, otherRegion);
759     }
760 
761     private int pickLowestLocalityServer(Cluster cluster) {
762       return cluster.getLowestLocalityRegionServer();
763     }
764 
765     private int pickLowestLocalityRegionOnServer(Cluster cluster, int server) {
766       return cluster.getLowestLocalityRegionOnServer(server);
767     }
768 
769     void setServices(MasterServices services) {
770       this.masterServices = services;
771     }
772   }
773 
774   /**
775    * Generates candidates which moves the replicas out of the region server for
776    * co-hosted region replicas
777    */
778   static class RegionReplicaCandidateGenerator extends CandidateGenerator {
779 
780     RandomCandidateGenerator randomGenerator = new RandomCandidateGenerator();
781 
782     /**
783      * Randomly select one regionIndex out of all region replicas co-hosted in the same group
784      * (a group is a server, host or rack)
785      * @param primariesOfRegionsPerGroup either Cluster.primariesOfRegionsPerServer,
786      * primariesOfRegionsPerHost or primariesOfRegionsPerRack
787      * @param regionsPerGroup either Cluster.regionsPerServer, regionsPerHost or regionsPerRack
788      * @param regionIndexToPrimaryIndex Cluster.regionsIndexToPrimaryIndex
789      * @return a regionIndex for the selected primary or -1 if there is no co-locating
790      */
791     int selectCoHostedRegionPerGroup(int[] primariesOfRegionsPerGroup, int[] regionsPerGroup
792         , int[] regionIndexToPrimaryIndex) {
793       int currentPrimary = -1;
794       int currentPrimaryIndex = -1;
795       int selectedPrimaryIndex = -1;
796       double currentLargestRandom = -1;
797       // primariesOfRegionsPerGroup is a sorted array. Since it contains the primary region
798       // ids for the regions hosted in server, a consecutive repetition means that replicas
799       // are co-hosted
800       for (int j = 0; j <= primariesOfRegionsPerGroup.length; j++) {
801         int primary = j < primariesOfRegionsPerGroup.length
802             ? primariesOfRegionsPerGroup[j] : -1;
803         if (primary != currentPrimary) { // check for whether we see a new primary
804           int numReplicas = j - currentPrimaryIndex;
805           if (numReplicas > 1) { // means consecutive primaries, indicating co-location
806             // decide to select this primary region id or not
807             double currentRandom = RANDOM.nextDouble();
808             // we don't know how many region replicas are co-hosted, we will randomly select one
809             // using reservoir sampling (http://gregable.com/2007/10/reservoir-sampling.html)
810             if (currentRandom > currentLargestRandom) {
811               selectedPrimaryIndex = currentPrimary;
812               currentLargestRandom = currentRandom;
813             }
814           }
815           currentPrimary = primary;
816           currentPrimaryIndex = j;
817         }
818       }
819 
820       // we have found the primary id for the region to move. Now find the actual regionIndex
821       // with the given primary, prefer to move the secondary region.
822       for (int j = 0; j < regionsPerGroup.length; j++) {
823         int regionIndex = regionsPerGroup[j];
824         if (selectedPrimaryIndex == regionIndexToPrimaryIndex[regionIndex]) {
825           // always move the secondary, not the primary
826           if (selectedPrimaryIndex != regionIndex) {
827             return regionIndex;
828           }
829         }
830       }
831       return -1;
832     }
833 
834     @Override
835     Cluster.Action generate(Cluster cluster) {
836       int serverIndex = pickRandomServer(cluster);
837       if (cluster.numServers <= 1 || serverIndex == -1) {
838         return Cluster.NullAction;
839       }
840 
841       int regionIndex = selectCoHostedRegionPerGroup(
842         cluster.primariesOfRegionsPerServer[serverIndex],
843         cluster.regionsPerServer[serverIndex],
844         cluster.regionIndexToPrimaryIndex);
845 
846       // if there are no pairs of region replicas co-hosted, default to random generator
847       if (regionIndex == -1) {
848         // default to randompicker
849         return randomGenerator.generate(cluster);
850       }
851 
852       int toServerIndex = pickOtherRandomServer(cluster, serverIndex);
853       int toRegionIndex = pickRandomRegion(cluster, toServerIndex, 0.9f);
854       return getAction(serverIndex, regionIndex, toServerIndex, toRegionIndex);
855     }
856   }
857 
858   /**
859    * Generates candidates which moves the replicas out of the rack for
860    * co-hosted region replicas in the same rack
861    */
862   static class RegionReplicaRackCandidateGenerator extends RegionReplicaCandidateGenerator {
863     @Override
864     Cluster.Action generate(Cluster cluster) {
865       int rackIndex = pickRandomRack(cluster);
866       if (cluster.numRacks <= 1 || rackIndex == -1) {
867         return super.generate(cluster);
868       }
869 
870       int regionIndex = selectCoHostedRegionPerGroup(
871         cluster.primariesOfRegionsPerRack[rackIndex],
872         cluster.regionsPerRack[rackIndex],
873         cluster.regionIndexToPrimaryIndex);
874 
875       // if there are no pairs of region replicas co-hosted, default to random generator
876       if (regionIndex == -1) {
877         // default to randompicker
878         return randomGenerator.generate(cluster);
879       }
880 
881       int serverIndex = cluster.regionIndexToServerIndex[regionIndex];
882       int toRackIndex = pickOtherRandomRack(cluster, rackIndex);
883 
884       int rand = RANDOM.nextInt(cluster.serversPerRack[toRackIndex].length);
885       int toServerIndex = cluster.serversPerRack[toRackIndex][rand];
886       int toRegionIndex = pickRandomRegion(cluster, toServerIndex, 0.9f);
887       return getAction(serverIndex, regionIndex, toServerIndex, toRegionIndex);
888     }
889   }
890 
891   /**
892    * Base class of StochasticLoadBalancer's Cost Functions.
893    */
894   abstract static class CostFunction {
895 
896     private float multiplier = 0;
897 
898     protected Cluster cluster;
899 
900     CostFunction(Configuration c) {
901 
902     }
903 
904     float getMultiplier() {
905       return multiplier;
906     }
907 
908     void setMultiplier(float m) {
909       this.multiplier = m;
910     }
911 
912     /** Called once per LB invocation to give the cost function
913      * to initialize it's state, and perform any costly calculation.
914      */
915     void init(Cluster cluster) {
916       this.cluster = cluster;
917     }
918 
919     /** Called once per cluster Action to give the cost function
920      * an opportunity to update it's state. postAction() is always
921      * called at least once before cost() is called with the cluster
922      * that this action is performed on. */
923     void postAction(Action action) {
924       switch (action.type) {
925       case NULL: break;
926       case ASSIGN_REGION:
927         AssignRegionAction ar = (AssignRegionAction) action;
928         regionMoved(ar.region, -1, ar.server);
929         break;
930       case MOVE_REGION:
931         MoveRegionAction mra = (MoveRegionAction) action;
932         regionMoved(mra.region, mra.fromServer, mra.toServer);
933         break;
934       case SWAP_REGIONS:
935         SwapRegionsAction a = (SwapRegionsAction) action;
936         regionMoved(a.fromRegion, a.fromServer, a.toServer);
937         regionMoved(a.toRegion, a.toServer, a.fromServer);
938         break;
939       default:
940         throw new RuntimeException("Uknown action:" + action.type);
941       }
942     }
943 
944     protected void regionMoved(int region, int oldServer, int newServer) {
945     }
946 
947     abstract double cost();
948 
949     /**
950      * Function to compute a scaled cost using {@link DescriptiveStatistics}. It
951      * assumes that this is a zero sum set of costs.  It assumes that the worst case
952      * possible is all of the elements in one region server and the rest having 0.
953      *
954      * @param stats the costs
955      * @return a scaled set of costs.
956      */
957     protected double costFromArray(double[] stats) {
958       double totalCost = 0;
959       double total = getSum(stats);
960 
961       double count = stats.length;
962       double mean = total/count;
963 
964       // Compute max as if all region servers had 0 and one had the sum of all costs.  This must be
965       // a zero sum cost for this to make sense.
966       double max = ((count - 1) * mean) + (total - mean);
967 
968       // It's possible that there aren't enough regions to go around
969       double min;
970       if (count > total) {
971         min = ((count - total) * mean) + ((1 - mean) * total);
972       } else {
973         // Some will have 1 more than everything else.
974         int numHigh = (int) (total - (Math.floor(mean) * count));
975         int numLow = (int) (count - numHigh);
976 
977         min = (numHigh * (Math.ceil(mean) - mean)) + (numLow * (mean - Math.floor(mean)));
978 
979       }
980       min = Math.max(0, min);
981       for (int i=0; i<stats.length; i++) {
982         double n = stats[i];
983         double diff = Math.abs(mean - n);
984         totalCost += diff;
985       }
986 
987       double scaled =  scale(min, max, totalCost);
988       return scaled;
989     }
990 
991     private double getSum(double[] stats) {
992       double total = 0;
993       for(double s:stats) {
994         total += s;
995       }
996       return total;
997     }
998 
999     /**
1000      * Scale the value between 0 and 1.
1001      *
1002      * @param min   Min value
1003      * @param max   The Max value
1004      * @param value The value to be scaled.
1005      * @return The scaled value.
1006      */
1007     protected double scale(double min, double max, double value) {
1008       if (max <= min || value <= min) {
1009         return 0;
1010       }
1011       if ((max - min) == 0) return 0;
1012 
1013       return Math.max(0d, Math.min(1d, (value - min) / (max - min)));
1014     }
1015   }
1016 
1017   /**
1018    * Given the starting state of the regions and a potential ending state
1019    * compute cost based upon the number of regions that have moved.
1020    */
1021   static class MoveCostFunction extends CostFunction {
1022     private static final String MOVE_COST_KEY = "hbase.master.balancer.stochastic.moveCost";
1023     private static final String MAX_MOVES_PERCENT_KEY =
1024         "hbase.master.balancer.stochastic.maxMovePercent";
1025     private static final float DEFAULT_MOVE_COST = 7;
1026     private static final int DEFAULT_MAX_MOVES = 600;
1027     private static final float DEFAULT_MAX_MOVE_PERCENT = 0.25f;
1028 
1029     private final float maxMovesPercent;
1030 
1031     MoveCostFunction(Configuration conf) {
1032       super(conf);
1033 
1034       // Move cost multiplier should be the same cost or higher than the rest of the costs to ensure
1035       // that large benefits are need to overcome the cost of a move.
1036       this.setMultiplier(conf.getFloat(MOVE_COST_KEY, DEFAULT_MOVE_COST));
1037       // What percent of the number of regions a single run of the balancer can move.
1038       maxMovesPercent = conf.getFloat(MAX_MOVES_PERCENT_KEY, DEFAULT_MAX_MOVE_PERCENT);
1039     }
1040 
1041     @Override
1042     double cost() {
1043       // Try and size the max number of Moves, but always be prepared to move some.
1044       int maxMoves = Math.max((int) (cluster.numRegions * maxMovesPercent),
1045           DEFAULT_MAX_MOVES);
1046 
1047       double moveCost = cluster.numMovedRegions;
1048 
1049       // Don't let this single balance move more than the max moves.
1050       // This allows better scaling to accurately represent the actual cost of a move.
1051       if (moveCost > maxMoves) {
1052         return 1000000;   // return a number much greater than any of the other cost
1053       }
1054 
1055       return scale(0, Math.min(cluster.numRegions, maxMoves), moveCost);
1056     }
1057   }
1058 
1059   /**
1060    * Compute the cost of a potential cluster state from skew in number of
1061    * regions on a cluster.
1062    */
1063   static class RegionCountSkewCostFunction extends CostFunction {
1064     private static final String REGION_COUNT_SKEW_COST_KEY =
1065         "hbase.master.balancer.stochastic.regionCountCost";
1066     private static final float DEFAULT_REGION_COUNT_SKEW_COST = 500;
1067 
1068     private double[] stats = null;
1069 
1070     RegionCountSkewCostFunction(Configuration conf) {
1071       super(conf);
1072       // Load multiplier should be the greatest as it is the most general way to balance data.
1073       this.setMultiplier(conf.getFloat(REGION_COUNT_SKEW_COST_KEY, DEFAULT_REGION_COUNT_SKEW_COST));
1074     }
1075 
1076     @Override
1077     double cost() {
1078       if (stats == null || stats.length != cluster.numServers) {
1079         stats = new double[cluster.numServers];
1080       }
1081 
1082       for (int i =0; i < cluster.numServers; i++) {
1083         stats[i] = cluster.regionsPerServer[i].length;
1084       }
1085 
1086       return costFromArray(stats);
1087     }
1088   }
1089 
1090   /**
1091    * Compute the cost of a potential cluster state from skew in number of
1092    * primary regions on a cluster.
1093    */
1094   static class PrimaryRegionCountSkewCostFunction extends CostFunction {
1095     private static final String PRIMARY_REGION_COUNT_SKEW_COST_KEY =
1096         "hbase.master.balancer.stochastic.primaryRegionCountCost";
1097     private static final float DEFAULT_PRIMARY_REGION_COUNT_SKEW_COST = 500;
1098 
1099     private double[] stats = null;
1100 
1101     PrimaryRegionCountSkewCostFunction(Configuration conf) {
1102       super(conf);
1103       // Load multiplier should be the greatest as primary regions serve majority of reads/writes.
1104       this.setMultiplier(conf.getFloat(PRIMARY_REGION_COUNT_SKEW_COST_KEY,
1105         DEFAULT_PRIMARY_REGION_COUNT_SKEW_COST));
1106     }
1107 
1108     @Override
1109     double cost() {
1110       if (!cluster.hasRegionReplicas) {
1111         return 0;
1112       }
1113       if (stats == null || stats.length != cluster.numServers) {
1114         stats = new double[cluster.numServers];
1115       }
1116 
1117       for (int i =0; i < cluster.numServers; i++) {
1118         stats[i] = 0;
1119         for (int regionIdx : cluster.regionsPerServer[i]) {
1120           if (regionIdx == cluster.regionIndexToPrimaryIndex[regionIdx]) {
1121             stats[i] ++;
1122           }
1123         }
1124       }
1125 
1126       return costFromArray(stats);
1127     }
1128   }
1129 
1130   /**
1131    * Compute the cost of a potential cluster configuration based upon how evenly
1132    * distributed tables are.
1133    */
1134   static class TableSkewCostFunction extends CostFunction {
1135 
1136     private static final String TABLE_SKEW_COST_KEY =
1137         "hbase.master.balancer.stochastic.tableSkewCost";
1138     private static final float DEFAULT_TABLE_SKEW_COST = 35;
1139 
1140     TableSkewCostFunction(Configuration conf) {
1141       super(conf);
1142       this.setMultiplier(conf.getFloat(TABLE_SKEW_COST_KEY, DEFAULT_TABLE_SKEW_COST));
1143     }
1144 
1145     @Override
1146     double cost() {
1147       double max = cluster.numRegions;
1148       double min = ((double) cluster.numRegions) / cluster.numServers;
1149       double value = 0;
1150 
1151       for (int i = 0; i < cluster.numMaxRegionsPerTable.length; i++) {
1152         value += cluster.numMaxRegionsPerTable[i];
1153       }
1154 
1155       return scale(min, max, value);
1156     }
1157   }
1158 
1159   /**
1160    * Compute a cost of a potential cluster configuration based upon where
1161    * {@link org.apache.hadoop.hbase.regionserver.StoreFile}s are located.
1162    */
1163   static class LocalityCostFunction extends CostFunction {
1164 
1165     private static final String LOCALITY_COST_KEY = "hbase.master.balancer.stochastic.localityCost";
1166     private static final float DEFAULT_LOCALITY_COST = 25;
1167 
1168     private MasterServices services;
1169 
1170     LocalityCostFunction(Configuration conf, MasterServices srv) {
1171       super(conf);
1172       this.setMultiplier(conf.getFloat(LOCALITY_COST_KEY, DEFAULT_LOCALITY_COST));
1173       this.services = srv;
1174     }
1175 
1176     void setServices(MasterServices srvc) {
1177       this.services = srvc;
1178     }
1179 
1180     @Override
1181     double cost() {
1182       double max = 0;
1183       double cost = 0;
1184 
1185       // If there's no master so there's no way anything else works.
1186       if (this.services == null) {
1187         return cost;
1188       }
1189 
1190       for (int i = 0; i < cluster.regionLocations.length; i++) {
1191         max += 1;
1192         int serverIndex = cluster.regionIndexToServerIndex[i];
1193         int[] regionLocations = cluster.regionLocations[i];
1194 
1195         // If we can't find where the data is getTopBlock returns null.
1196         // so count that as being the best possible.
1197         if (regionLocations == null) {
1198           continue;
1199         }
1200 
1201         int index = -1;
1202         for (int j = 0; j < regionLocations.length; j++) {
1203           if (regionLocations[j] >= 0 && regionLocations[j] == serverIndex) {
1204             index = j;
1205             break;
1206           }
1207         }
1208 
1209         if (index < 0) {
1210           cost += 1;
1211         } else {
1212           cost += (1 - cluster.getLocalityOfRegion(i, serverIndex));
1213         }
1214       }
1215       return scale(0, max, cost);
1216     }
1217   }
1218 
1219   /**
1220    * Base class the allows writing costs functions from rolling average of some
1221    * number from RegionLoad.
1222    */
1223   abstract static class CostFromRegionLoadFunction extends CostFunction {
1224 
1225     private ClusterStatus clusterStatus = null;
1226     private Map<String, Deque<RegionLoad>> loads = null;
1227     private double[] stats = null;
1228     CostFromRegionLoadFunction(Configuration conf) {
1229       super(conf);
1230     }
1231 
1232     void setClusterStatus(ClusterStatus status) {
1233       this.clusterStatus = status;
1234     }
1235 
1236     void setLoads(Map<String, Deque<RegionLoad>> l) {
1237       this.loads = l;
1238     }
1239 
1240     @Override
1241     double cost() {
1242       if (clusterStatus == null || loads == null) {
1243         return 0;
1244       }
1245 
1246       if (stats == null || stats.length != cluster.numServers) {
1247         stats = new double[cluster.numServers];
1248       }
1249 
1250       for (int i =0; i < stats.length; i++) {
1251         //Cost this server has from RegionLoad
1252         long cost = 0;
1253 
1254         // for every region on this server get the rl
1255         for(int regionIndex:cluster.regionsPerServer[i]) {
1256           Collection<RegionLoad> regionLoadList =  cluster.regionLoads[regionIndex];
1257 
1258           // Now if we found a region load get the type of cost that was requested.
1259           if (regionLoadList != null) {
1260             cost += getRegionLoadCost(regionLoadList);
1261           }
1262         }
1263 
1264         // Add the total cost to the stats.
1265         stats[i] = cost;
1266       }
1267 
1268       // Now return the scaled cost from data held in the stats object.
1269       return costFromArray(stats);
1270     }
1271 
1272     protected double getRegionLoadCost(Collection<RegionLoad> regionLoadList) {
1273       double cost = 0;
1274 
1275       for (RegionLoad rl : regionLoadList) {
1276         double toAdd = getCostFromRl(rl);
1277 
1278         if (cost == 0) {
1279           cost = toAdd;
1280         } else {
1281           cost = (.5 * cost) + (.5 * toAdd);
1282         }
1283       }
1284 
1285       return cost;
1286     }
1287 
1288     protected abstract double getCostFromRl(RegionLoad rl);
1289   }
1290 
1291   /**
1292    * Compute the cost of total number of read requests  The more unbalanced the higher the
1293    * computed cost will be.  This uses a rolling average of regionload.
1294    */
1295 
1296   static class ReadRequestCostFunction extends CostFromRegionLoadFunction {
1297 
1298     private static final String READ_REQUEST_COST_KEY =
1299         "hbase.master.balancer.stochastic.readRequestCost";
1300     private static final float DEFAULT_READ_REQUEST_COST = 5;
1301 
1302     ReadRequestCostFunction(Configuration conf) {
1303       super(conf);
1304       this.setMultiplier(conf.getFloat(READ_REQUEST_COST_KEY, DEFAULT_READ_REQUEST_COST));
1305     }
1306 
1307 
1308     @Override
1309     protected double getCostFromRl(RegionLoad rl) {
1310       return rl.getReadRequestsCount();
1311     }
1312   }
1313 
1314   /**
1315    * Compute the cost of total number of write requests.  The more unbalanced the higher the
1316    * computed cost will be.  This uses a rolling average of regionload.
1317    */
1318   static class WriteRequestCostFunction extends CostFromRegionLoadFunction {
1319 
1320     private static final String WRITE_REQUEST_COST_KEY =
1321         "hbase.master.balancer.stochastic.writeRequestCost";
1322     private static final float DEFAULT_WRITE_REQUEST_COST = 5;
1323 
1324     WriteRequestCostFunction(Configuration conf) {
1325       super(conf);
1326       this.setMultiplier(conf.getFloat(WRITE_REQUEST_COST_KEY, DEFAULT_WRITE_REQUEST_COST));
1327     }
1328 
1329     @Override
1330     protected double getCostFromRl(RegionLoad rl) {
1331       return rl.getWriteRequestsCount();
1332     }
1333   }
1334 
1335   /**
1336    * A cost function for region replicas. We give a very high cost to hosting
1337    * replicas of the same region in the same host. We do not prevent the case
1338    * though, since if numReplicas > numRegionServers, we still want to keep the
1339    * replica open.
1340    */
1341   static class RegionReplicaHostCostFunction extends CostFunction {
1342     private static final String REGION_REPLICA_HOST_COST_KEY =
1343         "hbase.master.balancer.stochastic.regionReplicaHostCostKey";
1344     private static final float DEFAULT_REGION_REPLICA_HOST_COST_KEY = 100000;
1345 
1346     long maxCost = 0;
1347     long[] costsPerGroup; // group is either server, host or rack
1348     int[][] primariesOfRegionsPerGroup;
1349 
1350     public RegionReplicaHostCostFunction(Configuration conf) {
1351       super(conf);
1352       this.setMultiplier(conf.getFloat(REGION_REPLICA_HOST_COST_KEY,
1353         DEFAULT_REGION_REPLICA_HOST_COST_KEY));
1354     }
1355 
1356     @Override
1357     void init(Cluster cluster) {
1358       super.init(cluster);
1359       // max cost is the case where every region replica is hosted together regardless of host
1360       maxCost = cluster.numHosts > 1 ? getMaxCost(cluster) : 0;
1361       costsPerGroup = new long[cluster.numHosts];
1362       primariesOfRegionsPerGroup = cluster.multiServersPerHost // either server based or host based
1363           ? cluster.primariesOfRegionsPerHost
1364           : cluster.primariesOfRegionsPerServer;
1365       for (int i = 0 ; i < primariesOfRegionsPerGroup.length; i++) {
1366         costsPerGroup[i] = costPerGroup(primariesOfRegionsPerGroup[i]);
1367       }
1368     }
1369 
1370     long getMaxCost(Cluster cluster) {
1371       if (!cluster.hasRegionReplicas) {
1372         return 0; // short circuit
1373       }
1374       // max cost is the case where every region replica is hosted together regardless of host
1375       int[] primariesOfRegions = new int[cluster.numRegions];
1376       System.arraycopy(cluster.regionIndexToPrimaryIndex, 0, primariesOfRegions, 0,
1377           cluster.regions.length);
1378 
1379       Arrays.sort(primariesOfRegions);
1380 
1381       // compute numReplicas from the sorted array
1382       return costPerGroup(primariesOfRegions);
1383     }
1384 
1385     @Override
1386     double cost() {
1387       if (maxCost <= 0) {
1388         return 0;
1389       }
1390 
1391       long totalCost = 0;
1392       for (int i = 0 ; i < costsPerGroup.length; i++) {
1393         totalCost += costsPerGroup[i];
1394       }
1395       return scale(0, maxCost, totalCost);
1396     }
1397 
1398     /**
1399      * For each primary region, it computes the total number of replicas in the array (numReplicas)
1400      * and returns a sum of numReplicas-1 squared. For example, if the server hosts
1401      * regions a, b, c, d, e, f where a and b are same replicas, and c,d,e are same replicas, it
1402      * returns (2-1) * (2-1) + (3-1) * (3-1) + (1-1) * (1-1).
1403      * @param primariesOfRegions a sorted array of primary regions ids for the regions hosted
1404      * @return a sum of numReplicas-1 squared for each primary region in the group.
1405      */
1406     protected long costPerGroup(int[] primariesOfRegions) {
1407       long cost = 0;
1408       int currentPrimary = -1;
1409       int currentPrimaryIndex = -1;
1410       // primariesOfRegions is a sorted array of primary ids of regions. Replicas of regions
1411       // sharing the same primary will have consecutive numbers in the array.
1412       for (int j = 0 ; j <= primariesOfRegions.length; j++) {
1413         int primary = j < primariesOfRegions.length ? primariesOfRegions[j] : -1;
1414         if (primary != currentPrimary) { // we see a new primary
1415           int numReplicas = j - currentPrimaryIndex;
1416           // square the cost
1417           if (numReplicas > 1) { // means consecutive primaries, indicating co-location
1418             cost += (numReplicas - 1) * (numReplicas - 1);
1419           }
1420           currentPrimary = primary;
1421           currentPrimaryIndex = j;
1422         }
1423       }
1424 
1425       return cost;
1426     }
1427 
1428     @Override
1429     protected void regionMoved(int region, int oldServer, int newServer) {
1430       if (maxCost <= 0) {
1431         return; // no need to compute
1432       }
1433       if (cluster.multiServersPerHost) {
1434         int oldHost = cluster.serverIndexToHostIndex[oldServer];
1435         int newHost = cluster.serverIndexToHostIndex[newServer];
1436         if (newHost != oldHost) {
1437           costsPerGroup[oldHost] = costPerGroup(cluster.primariesOfRegionsPerHost[oldHost]);
1438           costsPerGroup[newHost] = costPerGroup(cluster.primariesOfRegionsPerHost[newHost]);
1439         }
1440       } else {
1441         costsPerGroup[oldServer] = costPerGroup(cluster.primariesOfRegionsPerServer[oldServer]);
1442         costsPerGroup[newServer] = costPerGroup(cluster.primariesOfRegionsPerServer[newServer]);
1443       }
1444     }
1445   }
1446 
1447   /**
1448    * A cost function for region replicas for the rack distribution. We give a relatively high
1449    * cost to hosting replicas of the same region in the same rack. We do not prevent the case
1450    * though.
1451    */
1452   static class RegionReplicaRackCostFunction extends RegionReplicaHostCostFunction {
1453     private static final String REGION_REPLICA_RACK_COST_KEY =
1454         "hbase.master.balancer.stochastic.regionReplicaRackCostKey";
1455     private static final float DEFAULT_REGION_REPLICA_RACK_COST_KEY = 10000;
1456 
1457     public RegionReplicaRackCostFunction(Configuration conf) {
1458       super(conf);
1459       this.setMultiplier(conf.getFloat(REGION_REPLICA_RACK_COST_KEY,
1460         DEFAULT_REGION_REPLICA_RACK_COST_KEY));
1461     }
1462 
1463     @Override
1464     void init(Cluster cluster) {
1465       this.cluster = cluster;
1466       if (cluster.numRacks <= 1) {
1467         maxCost = 0;
1468         return; // disabled for 1 rack
1469       }
1470       // max cost is the case where every region replica is hosted together regardless of rack
1471       maxCost = getMaxCost(cluster);
1472       costsPerGroup = new long[cluster.numRacks];
1473       for (int i = 0 ; i < cluster.primariesOfRegionsPerRack.length; i++) {
1474         costsPerGroup[i] = costPerGroup(cluster.primariesOfRegionsPerRack[i]);
1475       }
1476     }
1477 
1478     @Override
1479     protected void regionMoved(int region, int oldServer, int newServer) {
1480       if (maxCost <= 0) {
1481         return; // no need to compute
1482       }
1483       int oldRack = cluster.serverIndexToRackIndex[oldServer];
1484       int newRack = cluster.serverIndexToRackIndex[newServer];
1485       if (newRack != oldRack) {
1486         costsPerGroup[oldRack] = costPerGroup(cluster.primariesOfRegionsPerRack[oldRack]);
1487         costsPerGroup[newRack] = costPerGroup(cluster.primariesOfRegionsPerRack[newRack]);
1488       }
1489     }
1490   }
1491 
1492   /**
1493    * Compute the cost of total memstore size.  The more unbalanced the higher the
1494    * computed cost will be.  This uses a rolling average of regionload.
1495    */
1496   static class MemstoreSizeCostFunction extends CostFromRegionLoadFunction {
1497 
1498     private static final String MEMSTORE_SIZE_COST_KEY =
1499         "hbase.master.balancer.stochastic.memstoreSizeCost";
1500     private static final float DEFAULT_MEMSTORE_SIZE_COST = 5;
1501 
1502     MemstoreSizeCostFunction(Configuration conf) {
1503       super(conf);
1504       this.setMultiplier(conf.getFloat(MEMSTORE_SIZE_COST_KEY, DEFAULT_MEMSTORE_SIZE_COST));
1505     }
1506 
1507     @Override
1508     protected double getCostFromRl(RegionLoad rl) {
1509       return rl.getMemStoreSizeMB();
1510     }
1511   }
1512   /**
1513    * Compute the cost of total open storefiles size.  The more unbalanced the higher the
1514    * computed cost will be.  This uses a rolling average of regionload.
1515    */
1516   static class StoreFileCostFunction extends CostFromRegionLoadFunction {
1517 
1518     private static final String STOREFILE_SIZE_COST_KEY =
1519         "hbase.master.balancer.stochastic.storefileSizeCost";
1520     private static final float DEFAULT_STOREFILE_SIZE_COST = 5;
1521 
1522     StoreFileCostFunction(Configuration conf) {
1523       super(conf);
1524       this.setMultiplier(conf.getFloat(STOREFILE_SIZE_COST_KEY, DEFAULT_STOREFILE_SIZE_COST));
1525     }
1526 
1527     @Override
1528     protected double getCostFromRl(RegionLoad rl) {
1529       return rl.getStorefileSizeMB();
1530     }
1531   }
1532 
1533   /**
1534    * A helper function to compose the attribute name from tablename and costfunction name
1535    */
1536   public static String composeAttributeName(String tableName, String costFunctionName) {
1537     return tableName + TABLE_FUNCTION_SEP + costFunctionName;
1538   }
1539 }