1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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.ArrayList;
23 import java.util.Arrays;
24 import java.util.Collection;
25 import java.util.Collections;
26 import java.util.Deque;
27 import java.util.HashMap;
28 import java.util.LinkedList;
29 import java.util.List;
30 import java.util.Map;
31 import java.util.Map.Entry;
32 import java.util.Random;
33
34 import org.apache.commons.logging.Log;
35 import org.apache.commons.logging.LogFactory;
36 import org.apache.hadoop.hbase.classification.InterfaceAudience;
37 import org.apache.hadoop.conf.Configuration;
38 import org.apache.hadoop.hbase.ClusterStatus;
39 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
40 import org.apache.hadoop.hbase.HConstants;
41 import org.apache.hadoop.hbase.HRegionInfo;
42 import org.apache.hadoop.hbase.RegionLoad;
43 import org.apache.hadoop.hbase.ServerLoad;
44 import org.apache.hadoop.hbase.ServerName;
45 import org.apache.hadoop.hbase.TableName;
46 import org.apache.hadoop.hbase.master.MasterServices;
47 import org.apache.hadoop.hbase.master.RegionPlan;
48 import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer.Cluster.Action;
49 import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer.Cluster.Action.Type;
50 import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer.Cluster.AssignRegionAction;
51 import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer.Cluster.LocalityType;
52 import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer.Cluster.MoveRegionAction;
53 import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer.Cluster.SwapRegionsAction;
54 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
55
56 import com.google.common.base.Optional;
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.CONFIG)
102 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="IS2_INCONSISTENT_SYNC",
103 justification="Complaint is about costFunctions not being synchronized; not end of the world")
104 public class StochasticLoadBalancer extends BaseLoadBalancer {
105
106 protected static final String STEPS_PER_REGION_KEY =
107 "hbase.master.balancer.stochastic.stepsPerRegion";
108 protected static final String MAX_STEPS_KEY =
109 "hbase.master.balancer.stochastic.maxSteps";
110 protected static final String RUN_MAX_STEPS_KEY =
111 "hbase.master.balancer.stochastic.runMaxSteps";
112 protected static final String MAX_RUNNING_TIME_KEY =
113 "hbase.master.balancer.stochastic.maxRunningTime";
114 protected static final String KEEP_REGION_LOADS =
115 "hbase.master.balancer.stochastic.numRegionLoadsToRemember";
116 private static final String TABLE_FUNCTION_SEP = "_";
117 protected static final String MIN_COST_NEED_BALANCE_KEY =
118 "hbase.master.balancer.stochastic.minCostNeedBalance";
119
120 private static final Random RANDOM = new Random(System.currentTimeMillis());
121 private static final Log LOG = LogFactory.getLog(StochasticLoadBalancer.class);
122
123 Map<String, Deque<RegionLoad>> loads = new HashMap<String, Deque<RegionLoad>>();
124
125
126 private int maxSteps = 1000000;
127 private boolean runMaxSteps = false;
128 private int stepsPerRegion = 800;
129 private long maxRunningTime = 30 * 1000 * 1;
130 private int numRegionLoadsToRemember = 15;
131 private float minCostNeedBalance = 0.05f;
132
133 private CandidateGenerator[] candidateGenerators;
134 private CostFromRegionLoadFunction[] regionLoadFunctions;
135 private CostFunction[] costFunctions;
136
137
138 private Double curOverallCost = 0d;
139 private Double[] tempFunctionCosts;
140 private Double[] curFunctionCosts;
141
142
143
144 private LocalityBasedCandidateGenerator localityCandidateGenerator;
145 private ServerLocalityCostFunction localityCost;
146 private RackLocalityCostFunction rackLocalityCost;
147 private RegionReplicaHostCostFunction regionReplicaHostCostFunction;
148 private RegionReplicaRackCostFunction regionReplicaRackCostFunction;
149 private boolean isByTable = false;
150 private TableName tableName = null;
151
152
153
154
155
156 public StochasticLoadBalancer() {
157 super(new MetricsStochasticBalancer());
158 }
159
160 @Override
161 public void onConfigurationChange(Configuration conf) {
162 setConf(conf);
163 }
164
165 @Override
166 public synchronized void setConf(Configuration conf) {
167 super.setConf(conf);
168 LOG.info("loading config");
169
170 maxSteps = conf.getInt(MAX_STEPS_KEY, maxSteps);
171
172 stepsPerRegion = conf.getInt(STEPS_PER_REGION_KEY, stepsPerRegion);
173 maxRunningTime = conf.getLong(MAX_RUNNING_TIME_KEY, maxRunningTime);
174 runMaxSteps = conf.getBoolean(RUN_MAX_STEPS_KEY, runMaxSteps);
175
176 numRegionLoadsToRemember = conf.getInt(KEEP_REGION_LOADS, numRegionLoadsToRemember);
177 isByTable = conf.getBoolean(HConstants.HBASE_MASTER_LOADBALANCE_BYTABLE, isByTable);
178
179 minCostNeedBalance = conf.getFloat(MIN_COST_NEED_BALANCE_KEY, minCostNeedBalance);
180
181 if (localityCandidateGenerator == null) {
182 localityCandidateGenerator = new LocalityBasedCandidateGenerator(services);
183 }
184 localityCost = new ServerLocalityCostFunction(conf, services);
185 rackLocalityCost = new RackLocalityCostFunction(conf, services);
186
187 if (candidateGenerators == null) {
188 candidateGenerators = new CandidateGenerator[] {
189 new RandomCandidateGenerator(),
190 new LoadCandidateGenerator(),
191 localityCandidateGenerator,
192 new RegionReplicaRackCandidateGenerator(),
193 };
194 }
195
196 regionLoadFunctions = new CostFromRegionLoadFunction[] {
197 new ReadRequestCostFunction(conf),
198 new WriteRequestCostFunction(conf),
199 new MemstoreSizeCostFunction(conf),
200 new StoreFileCostFunction(conf)
201 };
202
203 regionReplicaHostCostFunction = new RegionReplicaHostCostFunction(conf);
204 regionReplicaRackCostFunction = new RegionReplicaRackCostFunction(conf);
205
206 costFunctions = new CostFunction[]{
207 new RegionCountSkewCostFunction(conf),
208 new PrimaryRegionCountSkewCostFunction(conf),
209 new MoveCostFunction(conf),
210 localityCost,
211 rackLocalityCost,
212 new TableSkewCostFunction(conf),
213 regionReplicaHostCostFunction,
214 regionReplicaRackCostFunction,
215 regionLoadFunctions[0],
216 regionLoadFunctions[1],
217 regionLoadFunctions[2],
218 regionLoadFunctions[3],
219 };
220
221 curFunctionCosts= new Double[costFunctions.length];
222 tempFunctionCosts= new Double[costFunctions.length];
223
224 }
225
226 @Override
227 protected void setSlop(Configuration conf) {
228 this.slop = conf.getFloat("hbase.regions.slop", 0.001F);
229 }
230
231 @Override
232 public synchronized void setClusterStatus(ClusterStatus st) {
233 super.setClusterStatus(st);
234 updateRegionLoad();
235 for(CostFromRegionLoadFunction cost : regionLoadFunctions) {
236 cost.setClusterStatus(st);
237 }
238
239
240 try {
241
242 int tablesCount = isByTable ? services.getTableDescriptors().getAll().size() : 1;
243 int functionsCount = getCostFunctionNames().length;
244
245 updateMetricsSize(tablesCount * (functionsCount + 1));
246 } catch (Exception e) {
247 LOG.error("failed to get the size of all tables", e);
248 }
249 }
250
251
252
253
254 public void updateMetricsSize(int size) {
255 if (metricsBalancer instanceof MetricsStochasticBalancer) {
256 ((MetricsStochasticBalancer) metricsBalancer).updateMetricsSize(size);
257 }
258 }
259
260 @Override
261 public synchronized void setMasterServices(MasterServices masterServices) {
262 super.setMasterServices(masterServices);
263 this.localityCost.setServices(masterServices);
264 this.rackLocalityCost.setServices(masterServices);
265 this.localityCandidateGenerator.setServices(masterServices);
266 }
267
268 @Override
269 protected synchronized boolean areSomeRegionReplicasColocated(Cluster c) {
270 regionReplicaHostCostFunction.init(c);
271 if (regionReplicaHostCostFunction.cost() > 0) return true;
272 regionReplicaRackCostFunction.init(c);
273 if (regionReplicaRackCostFunction.cost() > 0) return true;
274 return false;
275 }
276
277 @Override
278 protected boolean needsBalance(Cluster cluster) {
279 ClusterLoadState cs = new ClusterLoadState(cluster.clusterState);
280 if (cs.getNumServers() < MIN_SERVER_BALANCE) {
281 if (LOG.isDebugEnabled()) {
282 LOG.debug("Not running balancer because only " + cs.getNumServers()
283 + " active regionserver(s)");
284 }
285 return false;
286 }
287 if (areSomeRegionReplicasColocated(cluster)) {
288 return true;
289 }
290
291 double total = 0.0;
292 float sumMultiplier = 0.0f;
293 for (CostFunction c : costFunctions) {
294 float multiplier = c.getMultiplier();
295 if (multiplier <= 0) {
296 continue;
297 }
298 if (!c.isNeeded()) {
299 LOG.debug(c.getClass().getName() + " indicated that its cost should not be considered");
300 continue;
301 }
302 sumMultiplier += multiplier;
303 total += c.cost() * multiplier;
304 }
305
306 if (total <= 0 || sumMultiplier <= 0
307 || (sumMultiplier > 0 && (total / sumMultiplier) < minCostNeedBalance)) {
308 LOG.info("Skipping load balancing because balanced cluster; " + "total cost is " + total
309 + ", sum multiplier is " + sumMultiplier + " min cost which need balance is "
310 + minCostNeedBalance);
311 return false;
312 }
313 return true;
314 }
315
316 @Override
317 public synchronized List<RegionPlan> balanceCluster(TableName tableName, Map<ServerName,
318 List<HRegionInfo>> clusterState) {
319 this.tableName = tableName;
320 return balanceCluster(clusterState);
321 }
322
323 @VisibleForTesting
324 Cluster.Action nextAction(Cluster cluster) {
325 return candidateGenerators[(RANDOM.nextInt(candidateGenerators.length))]
326 .generate(cluster);
327 }
328
329
330
331
332
333 @Override
334 public synchronized List<RegionPlan> balanceCluster(Map<ServerName,
335 List<HRegionInfo>> clusterState) {
336 List<RegionPlan> plans = balanceMasterRegions(clusterState);
337 if (plans != null || clusterState == null || clusterState.size() <= 1) {
338 return plans;
339 }
340
341 if (masterServerName != null && clusterState.containsKey(masterServerName)) {
342 if (clusterState.size() <= 2) {
343 return null;
344 }
345 clusterState = new HashMap<ServerName, List<HRegionInfo>>(clusterState);
346 clusterState.remove(masterServerName);
347 }
348
349
350
351
352
353 RegionLocationFinder finder = null;
354 if ((this.localityCost != null && this.localityCost.getMultiplier() > 0)
355 || (this.rackLocalityCost != null && this.rackLocalityCost.getMultiplier() > 0)) {
356 finder = this.regionFinder;
357 }
358
359
360
361
362 Cluster cluster = new Cluster(clusterState, loads, finder, rackManager);
363
364 long startTime = EnvironmentEdgeManager.currentTime();
365
366 initCosts(cluster);
367
368 if (!needsBalance(cluster)) {
369 return null;
370 }
371
372 double currentCost = computeCost(cluster, Double.MAX_VALUE);
373 curOverallCost = currentCost;
374 for (int i = 0; i < this.curFunctionCosts.length; i++) {
375 curFunctionCosts[i] = tempFunctionCosts[i];
376 }
377
378 double initCost = currentCost;
379 double newCost = currentCost;
380
381 long computedMaxSteps = 0;
382 if (runMaxSteps) {
383 computedMaxSteps = Math.max(this.maxSteps,
384 ((long)cluster.numRegions * (long)this.stepsPerRegion * (long)cluster.numServers));
385 } else {
386 long calculatedMaxSteps =
387 (long) cluster.numRegions * (long) this.stepsPerRegion * (long) cluster.numServers;
388 computedMaxSteps = Math.min(this.maxSteps, calculatedMaxSteps);
389 if (calculatedMaxSteps > maxSteps) {
390 LOG.warn(String.format("calculatedMaxSteps:%d for loadbalancer's stochastic walk is larger "
391 + "than maxSteps:%dß. Hence load balancing may not work well. Setting parameter "
392 + "\"hbase.master.balancer.stochastic.runMaxSteps\" to true to overcome this issue."
393 + "(This config change does not require service restart)", calculatedMaxSteps,
394 maxRunningTime));
395
396 }
397 }
398 LOG.info("start StochasticLoadBalancer.balancer, initCost=" + currentCost + ", functionCost="
399 + functionCost() + " computedMaxSteps: " + computedMaxSteps);
400
401
402 long step;
403
404 for (step = 0; step < computedMaxSteps; step++) {
405 Cluster.Action action = nextAction(cluster);
406
407 if (action.type == Type.NULL) {
408 continue;
409 }
410
411 cluster.doAction(action);
412 updateCostsWithAction(cluster, action);
413
414 newCost = computeCost(cluster, currentCost);
415
416
417 if (newCost < currentCost) {
418 currentCost = newCost;
419
420
421 curOverallCost = currentCost;
422 for (int i = 0; i < this.curFunctionCosts.length; i++) {
423 curFunctionCosts[i] = tempFunctionCosts[i];
424 }
425 } else {
426
427
428 Action undoAction = action.undoAction();
429 cluster.doAction(undoAction);
430 updateCostsWithAction(cluster, undoAction);
431 }
432
433 if (EnvironmentEdgeManager.currentTime() - startTime >
434 maxRunningTime) {
435 break;
436 }
437 }
438 long endTime = EnvironmentEdgeManager.currentTime();
439
440 metricsBalancer.balanceCluster(endTime - startTime);
441
442
443 updateStochasticCosts(tableName, curOverallCost, curFunctionCosts);
444 if (initCost > currentCost) {
445 plans = createRegionPlans(cluster);
446 if (LOG.isDebugEnabled()) {
447 LOG.debug("Finished computing new load balance plan. Computation took "
448 + (endTime - startTime) + "ms to try " + step
449 + " different iterations. Found a solution that moves "
450 + plans.size() + " regions; Going from a computed cost of "
451 + initCost + " to a new cost of " + currentCost);
452 }
453
454 return plans;
455 }
456 if (LOG.isDebugEnabled()) {
457 LOG.debug("Could not find a better load balance plan. Tried "
458 + step + " different configurations in " + (endTime - startTime)
459 + "ms, and did not find anything with a computed cost less than " + initCost);
460 }
461 return null;
462 }
463
464
465
466
467 private void updateStochasticCosts(TableName tableName, Double overall, Double[] subCosts) {
468 if (tableName == null) return;
469
470
471 if (metricsBalancer instanceof MetricsStochasticBalancer) {
472 MetricsStochasticBalancer balancer = (MetricsStochasticBalancer) metricsBalancer;
473
474 balancer.updateStochasticCost(tableName.getNameAsString(),
475 "Overall", "Overall cost", overall);
476
477
478 for (int i = 0; i < costFunctions.length; i++) {
479 CostFunction costFunction = costFunctions[i];
480 String costFunctionName = costFunction.getClass().getSimpleName();
481 Double costPercent = (overall == 0) ? 0 : (subCosts[i] / overall);
482
483 balancer.updateStochasticCost(tableName.getNameAsString(), costFunctionName,
484 "The percent of " + costFunctionName, costPercent);
485 }
486 }
487 }
488
489 private String functionCost() {
490 StringBuilder builder = new StringBuilder();
491 for (CostFunction c:costFunctions) {
492 builder.append(c.getClass().getSimpleName());
493 builder.append(" : (");
494 builder.append(c.getMultiplier());
495 builder.append(", ");
496 builder.append(c.cost());
497 builder.append("); ");
498 }
499 return builder.toString();
500 }
501
502
503
504
505
506
507
508
509 private List<RegionPlan> createRegionPlans(Cluster cluster) {
510 List<RegionPlan> plans = new LinkedList<RegionPlan>();
511 for (int regionIndex = 0;
512 regionIndex < cluster.regionIndexToServerIndex.length; regionIndex++) {
513 int initialServerIndex = cluster.initialRegionIndexToServerIndex[regionIndex];
514 int newServerIndex = cluster.regionIndexToServerIndex[regionIndex];
515
516 if (initialServerIndex != newServerIndex) {
517 HRegionInfo region = cluster.regions[regionIndex];
518 ServerName initialServer = cluster.servers[initialServerIndex];
519 ServerName newServer = cluster.servers[newServerIndex];
520
521 if (LOG.isTraceEnabled()) {
522 LOG.trace("Moving Region " + region.getEncodedName() + " from server "
523 + initialServer.getHostname() + " to " + newServer.getHostname());
524 }
525 RegionPlan rp = new RegionPlan(region, initialServer, newServer);
526 plans.add(rp);
527 }
528 }
529 return plans;
530 }
531
532
533
534
535 private synchronized void updateRegionLoad() {
536
537
538 Map<String, Deque<RegionLoad>> oldLoads = loads;
539 loads = new HashMap<String, Deque<RegionLoad>>();
540
541 for (ServerName sn : clusterStatus.getServers()) {
542 ServerLoad sl = clusterStatus.getLoad(sn);
543 if (sl == null) {
544 continue;
545 }
546 for (Entry<byte[], RegionLoad> entry : sl.getRegionsLoad().entrySet()) {
547 String regionNameAsString = HRegionInfo.getRegionNameAsString(entry.getKey());
548 Deque<RegionLoad> rLoads = oldLoads.get(regionNameAsString);
549 if (rLoads == null) {
550
551 rLoads = new ArrayDeque<RegionLoad>();
552 } else if (rLoads.size() >= numRegionLoadsToRemember) {
553 rLoads.remove();
554 }
555 rLoads.add(entry.getValue());
556 loads.put(regionNameAsString, rLoads);
557 }
558 }
559
560 for(CostFromRegionLoadFunction cost : regionLoadFunctions) {
561 cost.setLoads(loads);
562 }
563 }
564
565 protected void initCosts(Cluster cluster) {
566 for (CostFunction c:costFunctions) {
567 c.init(cluster);
568 }
569 }
570
571 protected void updateCostsWithAction(Cluster cluster, Action action) {
572 for (CostFunction c : costFunctions) {
573 c.postAction(action);
574 }
575 }
576
577
578
579
580 public String[] getCostFunctionNames() {
581 if (costFunctions == null) return null;
582 String[] ret = new String[costFunctions.length];
583 for (int i = 0; i < costFunctions.length; i++) {
584 CostFunction c = costFunctions[i];
585 ret[i] = c.getClass().getSimpleName();
586 }
587
588 return ret;
589 }
590
591
592
593
594
595
596
597
598
599
600 protected double computeCost(Cluster cluster, double previousCost) {
601 double total = 0;
602
603 for (int i = 0; i < costFunctions.length; i++) {
604 CostFunction c = costFunctions[i];
605 this.tempFunctionCosts[i] = 0.0;
606
607 if (c.getMultiplier() <= 0) {
608 continue;
609 }
610
611 Float multiplier = c.getMultiplier();
612 Double cost = c.cost();
613
614 this.tempFunctionCosts[i] = multiplier*cost;
615 total += this.tempFunctionCosts[i];
616
617 if (total > previousCost) {
618 break;
619 }
620 }
621
622 return total;
623 }
624
625
626 abstract static class CandidateGenerator {
627 abstract Cluster.Action generate(Cluster cluster);
628
629
630
631
632
633
634
635
636
637
638
639
640
641 protected int pickRandomRegion(Cluster cluster, int server, double chanceOfNoSwap) {
642
643 if (cluster.regionsPerServer[server].length == 0 || RANDOM.nextFloat() < chanceOfNoSwap) {
644
645 return -1;
646 }
647 int rand = RANDOM.nextInt(cluster.regionsPerServer[server].length);
648 return cluster.regionsPerServer[server][rand];
649
650 }
651 protected int pickRandomServer(Cluster cluster) {
652 if (cluster.numServers < 1) {
653 return -1;
654 }
655
656 return RANDOM.nextInt(cluster.numServers);
657 }
658
659 protected int pickRandomRack(Cluster cluster) {
660 if (cluster.numRacks < 1) {
661 return -1;
662 }
663
664 return RANDOM.nextInt(cluster.numRacks);
665 }
666
667 protected int pickOtherRandomServer(Cluster cluster, int serverIndex) {
668 if (cluster.numServers < 2) {
669 return -1;
670 }
671 while (true) {
672 int otherServerIndex = pickRandomServer(cluster);
673 if (otherServerIndex != serverIndex) {
674 return otherServerIndex;
675 }
676 }
677 }
678
679 protected int pickOtherRandomRack(Cluster cluster, int rackIndex) {
680 if (cluster.numRacks < 2) {
681 return -1;
682 }
683 while (true) {
684 int otherRackIndex = pickRandomRack(cluster);
685 if (otherRackIndex != rackIndex) {
686 return otherRackIndex;
687 }
688 }
689 }
690
691 protected Cluster.Action pickRandomRegions(Cluster cluster,
692 int thisServer,
693 int otherServer) {
694 if (thisServer < 0 || otherServer < 0) {
695 return Cluster.NullAction;
696 }
697
698
699 int thisRegionCount = cluster.getNumRegions(thisServer);
700 int otherRegionCount = cluster.getNumRegions(otherServer);
701
702
703 double thisChance = (thisRegionCount > otherRegionCount) ? 0 : 0.5;
704 double otherChance = (thisRegionCount <= otherRegionCount) ? 0 : 0.5;
705
706 int thisRegion = pickRandomRegion(cluster, thisServer, thisChance);
707 int otherRegion = pickRandomRegion(cluster, otherServer, otherChance);
708
709 return getAction(thisServer, thisRegion, otherServer, otherRegion);
710 }
711
712 protected Cluster.Action getAction(int fromServer, int fromRegion,
713 int toServer, int toRegion) {
714 if (fromServer < 0 || toServer < 0) {
715 return Cluster.NullAction;
716 }
717 if (fromRegion > 0 && toRegion > 0) {
718 return new Cluster.SwapRegionsAction(fromServer, fromRegion,
719 toServer, toRegion);
720 } else if (fromRegion > 0) {
721 return new Cluster.MoveRegionAction(fromRegion, fromServer, toServer);
722 } else if (toRegion > 0) {
723 return new Cluster.MoveRegionAction(toRegion, toServer, fromServer);
724 } else {
725 return Cluster.NullAction;
726 }
727 }
728
729
730
731
732 protected List<Integer> getRandomIterationOrder(int length) {
733 ArrayList<Integer> order = new ArrayList<>(length);
734 for (int i = 0; i < length; i++) {
735 order.add(i);
736 }
737 Collections.shuffle(order);
738 return order;
739 }
740 }
741
742 static class RandomCandidateGenerator extends CandidateGenerator {
743
744 @Override
745 Cluster.Action generate(Cluster cluster) {
746
747 int thisServer = pickRandomServer(cluster);
748
749
750 int otherServer = pickOtherRandomServer(cluster, thisServer);
751
752 return pickRandomRegions(cluster, thisServer, otherServer);
753 }
754 }
755
756 static class LoadCandidateGenerator extends CandidateGenerator {
757
758 @Override
759 Cluster.Action generate(Cluster cluster) {
760 cluster.sortServersByRegionCount();
761 int thisServer = pickMostLoadedServer(cluster, -1);
762 int otherServer = pickLeastLoadedServer(cluster, thisServer);
763
764 return pickRandomRegions(cluster, thisServer, otherServer);
765 }
766
767 private int pickLeastLoadedServer(final Cluster cluster, int thisServer) {
768 Integer[] servers = cluster.serverIndicesSortedByRegionCount;
769
770 int index = 0;
771 while (servers[index] == null || servers[index] == thisServer) {
772 index++;
773 if (index == servers.length) {
774 return -1;
775 }
776 }
777 return servers[index];
778 }
779
780 private int pickMostLoadedServer(final Cluster cluster, int thisServer) {
781 Integer[] servers = cluster.serverIndicesSortedByRegionCount;
782
783 int index = servers.length - 1;
784 while (servers[index] == null || servers[index] == thisServer) {
785 index--;
786 if (index < 0) {
787 return -1;
788 }
789 }
790 return servers[index];
791 }
792 }
793
794 static class LocalityBasedCandidateGenerator extends CandidateGenerator {
795
796 private MasterServices masterServices;
797
798 LocalityBasedCandidateGenerator(MasterServices masterServices) {
799 this.masterServices = masterServices;
800 }
801
802 @Override
803 Cluster.Action generate(Cluster cluster) {
804 if (this.masterServices == null) {
805 int thisServer = pickRandomServer(cluster);
806
807 int otherServer = pickOtherRandomServer(cluster, thisServer);
808 return pickRandomRegions(cluster, thisServer, otherServer);
809 }
810
811
812 for (int region : getRandomIterationOrder(cluster.numRegions)) {
813 int currentServer = cluster.regionIndexToServerIndex[region];
814 if (currentServer != cluster.getOrComputeRegionsToMostLocalEntities(LocalityType.SERVER)[region]) {
815 Optional<Action> potential = tryMoveOrSwap(
816 cluster,
817 currentServer,
818 region,
819 cluster.getOrComputeRegionsToMostLocalEntities(LocalityType.SERVER)[region]
820 );
821 if (potential.isPresent()) {
822 return potential.get();
823 }
824 }
825 }
826 return Cluster.NullAction;
827 }
828
829
830
831
832
833 private Optional<Action> tryMoveOrSwap(Cluster cluster,
834 int fromServer,
835 int fromRegion,
836 int toServer) {
837
838 if (cluster.serverHasTooFewRegions(toServer)) {
839 return Optional.of(getAction(fromServer, fromRegion, toServer, -1));
840 }
841
842
843 double fromRegionLocalityDelta =
844 getWeightedLocality(cluster, fromRegion, toServer) - getWeightedLocality(cluster, fromRegion, fromServer);
845 for (int toRegionIndex : getRandomIterationOrder(cluster.regionsPerServer[toServer].length)) {
846 int toRegion = cluster.regionsPerServer[toServer][toRegionIndex];
847 double toRegionLocalityDelta =
848 getWeightedLocality(cluster, toRegion, fromServer) - getWeightedLocality(cluster, toRegion, toServer);
849
850 if (fromRegionLocalityDelta + toRegionLocalityDelta >= 0) {
851 return Optional.of(getAction(fromServer, fromRegion, toServer, toRegion));
852 }
853 }
854
855 return Optional.absent();
856 }
857
858 private double getWeightedLocality(Cluster cluster, int region, int server) {
859 return cluster.getOrComputeWeightedLocality(region, server, LocalityType.SERVER);
860 }
861
862 void setServices(MasterServices services) {
863 this.masterServices = services;
864 }
865 }
866
867
868
869
870
871 static class RegionReplicaCandidateGenerator extends CandidateGenerator {
872
873 RandomCandidateGenerator randomGenerator = new RandomCandidateGenerator();
874
875
876
877
878
879
880
881
882
883
884 int selectCoHostedRegionPerGroup(int[] primariesOfRegionsPerGroup, int[] regionsPerGroup
885 , int[] regionIndexToPrimaryIndex) {
886 int currentPrimary = -1;
887 int currentPrimaryIndex = -1;
888 int selectedPrimaryIndex = -1;
889 double currentLargestRandom = -1;
890
891
892
893 for (int j = 0; j <= primariesOfRegionsPerGroup.length; j++) {
894 int primary = j < primariesOfRegionsPerGroup.length
895 ? primariesOfRegionsPerGroup[j] : -1;
896 if (primary != currentPrimary) {
897 int numReplicas = j - currentPrimaryIndex;
898 if (numReplicas > 1) {
899
900 double currentRandom = RANDOM.nextDouble();
901
902
903 if (currentRandom > currentLargestRandom) {
904 selectedPrimaryIndex = currentPrimary;
905 currentLargestRandom = currentRandom;
906 }
907 }
908 currentPrimary = primary;
909 currentPrimaryIndex = j;
910 }
911 }
912
913
914
915 for (int j = 0; j < regionsPerGroup.length; j++) {
916 int regionIndex = regionsPerGroup[j];
917 if (selectedPrimaryIndex == regionIndexToPrimaryIndex[regionIndex]) {
918
919 if (selectedPrimaryIndex != regionIndex) {
920 return regionIndex;
921 }
922 }
923 }
924 return -1;
925 }
926
927 @Override
928 Cluster.Action generate(Cluster cluster) {
929 int serverIndex = pickRandomServer(cluster);
930 if (cluster.numServers <= 1 || serverIndex == -1) {
931 return Cluster.NullAction;
932 }
933
934 int regionIndex = selectCoHostedRegionPerGroup(
935 cluster.primariesOfRegionsPerServer[serverIndex],
936 cluster.regionsPerServer[serverIndex],
937 cluster.regionIndexToPrimaryIndex);
938
939
940 if (regionIndex == -1) {
941
942 return randomGenerator.generate(cluster);
943 }
944
945 int toServerIndex = pickOtherRandomServer(cluster, serverIndex);
946 int toRegionIndex = pickRandomRegion(cluster, toServerIndex, 0.9f);
947 return getAction(serverIndex, regionIndex, toServerIndex, toRegionIndex);
948 }
949 }
950
951
952
953
954
955 static class RegionReplicaRackCandidateGenerator extends RegionReplicaCandidateGenerator {
956 @Override
957 Cluster.Action generate(Cluster cluster) {
958 int rackIndex = pickRandomRack(cluster);
959 if (cluster.numRacks <= 1 || rackIndex == -1) {
960 return super.generate(cluster);
961 }
962
963 int regionIndex = selectCoHostedRegionPerGroup(
964 cluster.primariesOfRegionsPerRack[rackIndex],
965 cluster.regionsPerRack[rackIndex],
966 cluster.regionIndexToPrimaryIndex);
967
968
969 if (regionIndex == -1) {
970
971 return randomGenerator.generate(cluster);
972 }
973
974 int serverIndex = cluster.regionIndexToServerIndex[regionIndex];
975 int toRackIndex = pickOtherRandomRack(cluster, rackIndex);
976
977 int rand = RANDOM.nextInt(cluster.serversPerRack[toRackIndex].length);
978 int toServerIndex = cluster.serversPerRack[toRackIndex][rand];
979 int toRegionIndex = pickRandomRegion(cluster, toServerIndex, 0.9f);
980 return getAction(serverIndex, regionIndex, toServerIndex, toRegionIndex);
981 }
982 }
983
984
985
986
987 abstract static class CostFunction {
988
989 private float multiplier = 0;
990
991 protected Cluster cluster;
992
993 CostFunction(Configuration c) {
994 }
995
996 boolean isNeeded() {
997 return true;
998 }
999 float getMultiplier() {
1000 return multiplier;
1001 }
1002
1003 void setMultiplier(float m) {
1004 this.multiplier = m;
1005 }
1006
1007
1008
1009
1010 void init(Cluster cluster) {
1011 this.cluster = cluster;
1012 }
1013
1014
1015
1016
1017
1018 void postAction(Action action) {
1019 switch (action.type) {
1020 case NULL: break;
1021 case ASSIGN_REGION:
1022 AssignRegionAction ar = (AssignRegionAction) action;
1023 regionMoved(ar.region, -1, ar.server);
1024 break;
1025 case MOVE_REGION:
1026 MoveRegionAction mra = (MoveRegionAction) action;
1027 regionMoved(mra.region, mra.fromServer, mra.toServer);
1028 break;
1029 case SWAP_REGIONS:
1030 SwapRegionsAction a = (SwapRegionsAction) action;
1031 regionMoved(a.fromRegion, a.fromServer, a.toServer);
1032 regionMoved(a.toRegion, a.toServer, a.fromServer);
1033 break;
1034 default:
1035 throw new RuntimeException("Uknown action:" + action.type);
1036 }
1037 }
1038
1039 protected void regionMoved(int region, int oldServer, int newServer) {
1040 }
1041
1042 abstract double cost();
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052 protected double costFromArray(double[] stats) {
1053 double totalCost = 0;
1054 double total = getSum(stats);
1055
1056 double count = stats.length;
1057 double mean = total/count;
1058
1059
1060
1061 double max = ((count - 1) * mean) + (total - mean);
1062
1063
1064 double min;
1065 if (count > total) {
1066 min = ((count - total) * mean) + ((1 - mean) * total);
1067 } else {
1068
1069 int numHigh = (int) (total - (Math.floor(mean) * count));
1070 int numLow = (int) (count - numHigh);
1071
1072 min = (numHigh * (Math.ceil(mean) - mean)) + (numLow * (mean - Math.floor(mean)));
1073
1074 }
1075 min = Math.max(0, min);
1076 for (int i=0; i<stats.length; i++) {
1077 double n = stats[i];
1078 double diff = Math.abs(mean - n);
1079 totalCost += diff;
1080 }
1081
1082 double scaled = scale(min, max, totalCost);
1083 return scaled;
1084 }
1085
1086 private double getSum(double[] stats) {
1087 double total = 0;
1088 for(double s:stats) {
1089 total += s;
1090 }
1091 return total;
1092 }
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102 protected double scale(double min, double max, double value) {
1103 if (max <= min || value <= min) {
1104 return 0;
1105 }
1106 if ((max - min) == 0) return 0;
1107
1108 return Math.max(0d, Math.min(1d, (value - min) / (max - min)));
1109 }
1110 }
1111
1112
1113
1114
1115
1116 static class MoveCostFunction extends CostFunction {
1117 private static final String MOVE_COST_KEY = "hbase.master.balancer.stochastic.moveCost";
1118 private static final String MAX_MOVES_PERCENT_KEY =
1119 "hbase.master.balancer.stochastic.maxMovePercent";
1120 private static final float DEFAULT_MOVE_COST = 7;
1121 private static final int DEFAULT_MAX_MOVES = 600;
1122 private static final float DEFAULT_MAX_MOVE_PERCENT = 0.25f;
1123
1124 private final float maxMovesPercent;
1125
1126 MoveCostFunction(Configuration conf) {
1127 super(conf);
1128
1129
1130
1131 this.setMultiplier(conf.getFloat(MOVE_COST_KEY, DEFAULT_MOVE_COST));
1132
1133 maxMovesPercent = conf.getFloat(MAX_MOVES_PERCENT_KEY, DEFAULT_MAX_MOVE_PERCENT);
1134 }
1135
1136 @Override
1137 double cost() {
1138
1139 int maxMoves = Math.max((int) (cluster.numRegions * maxMovesPercent),
1140 DEFAULT_MAX_MOVES);
1141
1142 double moveCost = cluster.numMovedRegions;
1143
1144
1145
1146 if (moveCost > maxMoves) {
1147 return 1000000;
1148 }
1149
1150 return scale(0, Math.min(cluster.numRegions, maxMoves), moveCost);
1151 }
1152 }
1153
1154
1155
1156
1157
1158 static class RegionCountSkewCostFunction extends CostFunction {
1159 private static final String REGION_COUNT_SKEW_COST_KEY =
1160 "hbase.master.balancer.stochastic.regionCountCost";
1161 private static final float DEFAULT_REGION_COUNT_SKEW_COST = 500;
1162
1163 private double[] stats = null;
1164
1165 RegionCountSkewCostFunction(Configuration conf) {
1166 super(conf);
1167
1168 this.setMultiplier(conf.getFloat(REGION_COUNT_SKEW_COST_KEY, DEFAULT_REGION_COUNT_SKEW_COST));
1169 }
1170
1171 @Override
1172 double cost() {
1173 if (stats == null || stats.length != cluster.numServers) {
1174 stats = new double[cluster.numServers];
1175 }
1176
1177 for (int i =0; i < cluster.numServers; i++) {
1178 stats[i] = cluster.regionsPerServer[i].length;
1179 }
1180
1181 return costFromArray(stats);
1182 }
1183 }
1184
1185
1186
1187
1188
1189 static class PrimaryRegionCountSkewCostFunction extends CostFunction {
1190 private static final String PRIMARY_REGION_COUNT_SKEW_COST_KEY =
1191 "hbase.master.balancer.stochastic.primaryRegionCountCost";
1192 private static final float DEFAULT_PRIMARY_REGION_COUNT_SKEW_COST = 500;
1193
1194 private double[] stats = null;
1195
1196 PrimaryRegionCountSkewCostFunction(Configuration conf) {
1197 super(conf);
1198
1199 this.setMultiplier(conf.getFloat(PRIMARY_REGION_COUNT_SKEW_COST_KEY,
1200 DEFAULT_PRIMARY_REGION_COUNT_SKEW_COST));
1201 }
1202
1203 @Override
1204 double cost() {
1205 if (!cluster.hasRegionReplicas) {
1206 return 0;
1207 }
1208 if (stats == null || stats.length != cluster.numServers) {
1209 stats = new double[cluster.numServers];
1210 }
1211
1212 for (int i =0; i < cluster.numServers; i++) {
1213 stats[i] = 0;
1214 for (int regionIdx : cluster.regionsPerServer[i]) {
1215 if (regionIdx == cluster.regionIndexToPrimaryIndex[regionIdx]) {
1216 stats[i] ++;
1217 }
1218 }
1219 }
1220
1221 return costFromArray(stats);
1222 }
1223 }
1224
1225
1226
1227
1228
1229 static class TableSkewCostFunction extends CostFunction {
1230
1231 private static final String TABLE_SKEW_COST_KEY =
1232 "hbase.master.balancer.stochastic.tableSkewCost";
1233 private static final float DEFAULT_TABLE_SKEW_COST = 35;
1234
1235 TableSkewCostFunction(Configuration conf) {
1236 super(conf);
1237 this.setMultiplier(conf.getFloat(TABLE_SKEW_COST_KEY, DEFAULT_TABLE_SKEW_COST));
1238 }
1239
1240 @Override
1241 double cost() {
1242 double max = cluster.numRegions;
1243 double min = ((double) cluster.numRegions) / cluster.numServers;
1244 double value = 0;
1245
1246 for (int i = 0; i < cluster.numMaxRegionsPerTable.length; i++) {
1247 value += cluster.numMaxRegionsPerTable[i];
1248 }
1249
1250 return scale(min, max, value);
1251 }
1252 }
1253
1254
1255
1256
1257
1258 static abstract class LocalityBasedCostFunction extends CostFunction {
1259
1260 private final LocalityType type;
1261
1262 private double bestLocality;
1263 private double locality;
1264
1265 private MasterServices services;
1266
1267 LocalityBasedCostFunction(Configuration conf,
1268 MasterServices srv,
1269 LocalityType type,
1270 String localityCostKey,
1271 float defaultLocalityCost) {
1272 super(conf);
1273 this.type = type;
1274 this.setMultiplier(conf.getFloat(localityCostKey, defaultLocalityCost));
1275 this.services = srv;
1276 this.locality = 0.0;
1277 this.bestLocality = 0.0;
1278 }
1279
1280
1281
1282
1283 abstract int regionIndexToEntityIndex(int region);
1284
1285 public void setServices(MasterServices srvc) {
1286 this.services = srvc;
1287 }
1288
1289 @Override
1290 void init(Cluster cluster) {
1291 super.init(cluster);
1292 locality = 0.0;
1293 bestLocality = 0.0;
1294
1295
1296 if (this.services == null) {
1297 return;
1298 }
1299
1300 for (int region = 0; region < cluster.numRegions; region++) {
1301 locality += getWeightedLocality(region, regionIndexToEntityIndex(region));
1302 bestLocality += getWeightedLocality(region, getMostLocalEntityForRegion(region));
1303 }
1304
1305
1306
1307
1308 locality = bestLocality == 0 ? 1.0 : locality / bestLocality;
1309 }
1310
1311 @Override
1312 protected void regionMoved(int region, int oldServer, int newServer) {
1313 int oldEntity = type == LocalityType.SERVER ? oldServer : cluster.serverIndexToRackIndex[oldServer];
1314 int newEntity = type == LocalityType.SERVER ? newServer : cluster.serverIndexToRackIndex[newServer];
1315 if (this.services == null) {
1316 return;
1317 }
1318 double localityDelta = getWeightedLocality(region, newEntity) - getWeightedLocality(region, oldEntity);
1319 double normalizedDelta = bestLocality == 0 ? 0.0 : localityDelta / bestLocality;
1320 locality += normalizedDelta;
1321 }
1322
1323 @Override
1324 double cost() {
1325 return 1 - locality;
1326 }
1327
1328 private int getMostLocalEntityForRegion(int region) {
1329 return cluster.getOrComputeRegionsToMostLocalEntities(type)[region];
1330 }
1331
1332 private double getWeightedLocality(int region, int entity) {
1333 return cluster.getOrComputeWeightedLocality(region, entity, type);
1334 }
1335
1336 }
1337
1338 static class ServerLocalityCostFunction extends LocalityBasedCostFunction {
1339
1340 private static final String LOCALITY_COST_KEY = "hbase.master.balancer.stochastic.localityCost";
1341 private static final float DEFAULT_LOCALITY_COST = 25;
1342
1343 ServerLocalityCostFunction(Configuration conf, MasterServices srv) {
1344 super(
1345 conf,
1346 srv,
1347 LocalityType.SERVER,
1348 LOCALITY_COST_KEY,
1349 DEFAULT_LOCALITY_COST
1350 );
1351 }
1352
1353 @Override
1354 int regionIndexToEntityIndex(int region) {
1355 return cluster.regionIndexToServerIndex[region];
1356 }
1357 }
1358
1359 static class RackLocalityCostFunction extends LocalityBasedCostFunction {
1360
1361 private static final String RACK_LOCALITY_COST_KEY = "hbase.master.balancer.stochastic.rackLocalityCost";
1362 private static final float DEFAULT_RACK_LOCALITY_COST = 15;
1363
1364 public RackLocalityCostFunction(Configuration conf, MasterServices services) {
1365 super(
1366 conf,
1367 services,
1368 LocalityType.RACK,
1369 RACK_LOCALITY_COST_KEY,
1370 DEFAULT_RACK_LOCALITY_COST
1371 );
1372 }
1373
1374 @Override
1375 int regionIndexToEntityIndex(int region) {
1376 return cluster.getRackForRegion(region);
1377 }
1378 }
1379
1380
1381
1382
1383
1384 abstract static class CostFromRegionLoadFunction extends CostFunction {
1385
1386 private ClusterStatus clusterStatus = null;
1387 private Map<String, Deque<RegionLoad>> loads = null;
1388 private double[] stats = null;
1389 CostFromRegionLoadFunction(Configuration conf) {
1390 super(conf);
1391 }
1392
1393 void setClusterStatus(ClusterStatus status) {
1394 this.clusterStatus = status;
1395 }
1396
1397 void setLoads(Map<String, Deque<RegionLoad>> l) {
1398 this.loads = l;
1399 }
1400
1401 @Override
1402 double cost() {
1403 if (clusterStatus == null || loads == null) {
1404 return 0;
1405 }
1406
1407 if (stats == null || stats.length != cluster.numServers) {
1408 stats = new double[cluster.numServers];
1409 }
1410
1411 for (int i =0; i < stats.length; i++) {
1412
1413 long cost = 0;
1414
1415
1416 for(int regionIndex:cluster.regionsPerServer[i]) {
1417 Collection<RegionLoad> regionLoadList = cluster.regionLoads[regionIndex];
1418
1419
1420 if (regionLoadList != null) {
1421 cost = (long) (cost + getRegionLoadCost(regionLoadList));
1422 }
1423 }
1424
1425
1426 stats[i] = cost;
1427 }
1428
1429
1430 return costFromArray(stats);
1431 }
1432
1433 protected double getRegionLoadCost(Collection<RegionLoad> regionLoadList) {
1434 double cost = 0;
1435
1436 for (RegionLoad rl : regionLoadList) {
1437 cost += getCostFromRl(rl);
1438 }
1439 return cost / regionLoadList.size();
1440 }
1441
1442
1443 protected abstract double getCostFromRl(RegionLoad rl);
1444 }
1445
1446
1447
1448
1449
1450
1451 abstract static class CostFromRegionLoadAsRateFunction extends CostFromRegionLoadFunction {
1452
1453 CostFromRegionLoadAsRateFunction(Configuration conf) {
1454 super(conf);
1455 }
1456
1457 @Override
1458 protected double getRegionLoadCost(Collection<RegionLoad> regionLoadList) {
1459 double cost = 0;
1460 double previous = 0;
1461 boolean isFirst = true;
1462 for (RegionLoad rl : regionLoadList) {
1463 double current = getCostFromRl(rl);
1464 if (isFirst) {
1465 isFirst = false;
1466 } else {
1467 cost += current - previous;
1468 }
1469 previous = current;
1470 }
1471 return Math.max(0, cost / (regionLoadList.size() - 1));
1472 }
1473 }
1474
1475
1476
1477
1478
1479
1480 static class ReadRequestCostFunction extends CostFromRegionLoadAsRateFunction {
1481
1482 private static final String READ_REQUEST_COST_KEY =
1483 "hbase.master.balancer.stochastic.readRequestCost";
1484 private static final float DEFAULT_READ_REQUEST_COST = 5;
1485
1486 ReadRequestCostFunction(Configuration conf) {
1487 super(conf);
1488 this.setMultiplier(conf.getFloat(READ_REQUEST_COST_KEY, DEFAULT_READ_REQUEST_COST));
1489 }
1490
1491
1492 @Override
1493 protected double getCostFromRl(RegionLoad rl) {
1494 return rl.getReadRequestsCount();
1495 }
1496 }
1497
1498
1499
1500
1501
1502 static class WriteRequestCostFunction extends CostFromRegionLoadAsRateFunction {
1503
1504 private static final String WRITE_REQUEST_COST_KEY =
1505 "hbase.master.balancer.stochastic.writeRequestCost";
1506 private static final float DEFAULT_WRITE_REQUEST_COST = 5;
1507
1508 WriteRequestCostFunction(Configuration conf) {
1509 super(conf);
1510 this.setMultiplier(conf.getFloat(WRITE_REQUEST_COST_KEY, DEFAULT_WRITE_REQUEST_COST));
1511 }
1512
1513 @Override
1514 protected double getCostFromRl(RegionLoad rl) {
1515 return rl.getWriteRequestsCount();
1516 }
1517 }
1518
1519
1520
1521
1522
1523
1524
1525 static class RegionReplicaHostCostFunction extends CostFunction {
1526 private static final String REGION_REPLICA_HOST_COST_KEY =
1527 "hbase.master.balancer.stochastic.regionReplicaHostCostKey";
1528 private static final float DEFAULT_REGION_REPLICA_HOST_COST_KEY = 100000;
1529
1530 long maxCost = 0;
1531 long[] costsPerGroup;
1532 int[][] primariesOfRegionsPerGroup;
1533
1534 public RegionReplicaHostCostFunction(Configuration conf) {
1535 super(conf);
1536 this.setMultiplier(conf.getFloat(REGION_REPLICA_HOST_COST_KEY,
1537 DEFAULT_REGION_REPLICA_HOST_COST_KEY));
1538 }
1539
1540 @Override
1541 void init(Cluster cluster) {
1542 super.init(cluster);
1543
1544 maxCost = cluster.numHosts > 1 ? getMaxCost(cluster) : 0;
1545 costsPerGroup = new long[cluster.numHosts];
1546 primariesOfRegionsPerGroup = cluster.multiServersPerHost
1547 ? cluster.primariesOfRegionsPerHost
1548 : cluster.primariesOfRegionsPerServer;
1549 for (int i = 0 ; i < primariesOfRegionsPerGroup.length; i++) {
1550 costsPerGroup[i] = costPerGroup(primariesOfRegionsPerGroup[i]);
1551 }
1552 }
1553
1554 long getMaxCost(Cluster cluster) {
1555 if (!cluster.hasRegionReplicas) {
1556 return 0;
1557 }
1558
1559 int[] primariesOfRegions = new int[cluster.numRegions];
1560 System.arraycopy(cluster.regionIndexToPrimaryIndex, 0, primariesOfRegions, 0,
1561 cluster.regions.length);
1562
1563 Arrays.sort(primariesOfRegions);
1564
1565
1566 return costPerGroup(primariesOfRegions);
1567 }
1568
1569 @Override
1570 boolean isNeeded() {
1571 return cluster.hasRegionReplicas;
1572 }
1573
1574 @Override
1575 double cost() {
1576 if (maxCost <= 0) {
1577 return 0;
1578 }
1579
1580 long totalCost = 0;
1581 for (int i = 0 ; i < costsPerGroup.length; i++) {
1582 totalCost += costsPerGroup[i];
1583 }
1584 return scale(0, maxCost, totalCost);
1585 }
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595 protected long costPerGroup(int[] primariesOfRegions) {
1596 long cost = 0;
1597 int currentPrimary = -1;
1598 int currentPrimaryIndex = -1;
1599
1600
1601 for (int j = 0 ; j <= primariesOfRegions.length; j++) {
1602 int primary = j < primariesOfRegions.length ? primariesOfRegions[j] : -1;
1603 if (primary != currentPrimary) {
1604 int numReplicas = j - currentPrimaryIndex;
1605
1606 if (numReplicas > 1) {
1607 cost += (numReplicas - 1) * (numReplicas - 1);
1608 }
1609 currentPrimary = primary;
1610 currentPrimaryIndex = j;
1611 }
1612 }
1613
1614 return cost;
1615 }
1616
1617 @Override
1618 protected void regionMoved(int region, int oldServer, int newServer) {
1619 if (maxCost <= 0) {
1620 return;
1621 }
1622 if (cluster.multiServersPerHost) {
1623 int oldHost = cluster.serverIndexToHostIndex[oldServer];
1624 int newHost = cluster.serverIndexToHostIndex[newServer];
1625 if (newHost != oldHost) {
1626 costsPerGroup[oldHost] = costPerGroup(cluster.primariesOfRegionsPerHost[oldHost]);
1627 costsPerGroup[newHost] = costPerGroup(cluster.primariesOfRegionsPerHost[newHost]);
1628 }
1629 } else {
1630 costsPerGroup[oldServer] = costPerGroup(cluster.primariesOfRegionsPerServer[oldServer]);
1631 costsPerGroup[newServer] = costPerGroup(cluster.primariesOfRegionsPerServer[newServer]);
1632 }
1633 }
1634 }
1635
1636
1637
1638
1639
1640
1641 static class RegionReplicaRackCostFunction extends RegionReplicaHostCostFunction {
1642 private static final String REGION_REPLICA_RACK_COST_KEY =
1643 "hbase.master.balancer.stochastic.regionReplicaRackCostKey";
1644 private static final float DEFAULT_REGION_REPLICA_RACK_COST_KEY = 10000;
1645
1646 public RegionReplicaRackCostFunction(Configuration conf) {
1647 super(conf);
1648 this.setMultiplier(conf.getFloat(REGION_REPLICA_RACK_COST_KEY,
1649 DEFAULT_REGION_REPLICA_RACK_COST_KEY));
1650 }
1651
1652 @Override
1653 void init(Cluster cluster) {
1654 this.cluster = cluster;
1655 if (cluster.numRacks <= 1) {
1656 maxCost = 0;
1657 return;
1658 }
1659
1660 maxCost = getMaxCost(cluster);
1661 costsPerGroup = new long[cluster.numRacks];
1662 for (int i = 0 ; i < cluster.primariesOfRegionsPerRack.length; i++) {
1663 costsPerGroup[i] = costPerGroup(cluster.primariesOfRegionsPerRack[i]);
1664 }
1665 }
1666
1667 @Override
1668 protected void regionMoved(int region, int oldServer, int newServer) {
1669 if (maxCost <= 0) {
1670 return;
1671 }
1672 int oldRack = cluster.serverIndexToRackIndex[oldServer];
1673 int newRack = cluster.serverIndexToRackIndex[newServer];
1674 if (newRack != oldRack) {
1675 costsPerGroup[oldRack] = costPerGroup(cluster.primariesOfRegionsPerRack[oldRack]);
1676 costsPerGroup[newRack] = costPerGroup(cluster.primariesOfRegionsPerRack[newRack]);
1677 }
1678 }
1679 }
1680
1681
1682
1683
1684
1685 static class MemstoreSizeCostFunction extends CostFromRegionLoadAsRateFunction {
1686
1687 private static final String MEMSTORE_SIZE_COST_KEY =
1688 "hbase.master.balancer.stochastic.memstoreSizeCost";
1689 private static final float DEFAULT_MEMSTORE_SIZE_COST = 5;
1690
1691 MemstoreSizeCostFunction(Configuration conf) {
1692 super(conf);
1693 this.setMultiplier(conf.getFloat(MEMSTORE_SIZE_COST_KEY, DEFAULT_MEMSTORE_SIZE_COST));
1694 }
1695
1696 @Override
1697 protected double getCostFromRl(RegionLoad rl) {
1698 return rl.getMemStoreSizeMB();
1699 }
1700 }
1701
1702
1703
1704
1705 static class StoreFileCostFunction extends CostFromRegionLoadFunction {
1706
1707 private static final String STOREFILE_SIZE_COST_KEY =
1708 "hbase.master.balancer.stochastic.storefileSizeCost";
1709 private static final float DEFAULT_STOREFILE_SIZE_COST = 5;
1710
1711 StoreFileCostFunction(Configuration conf) {
1712 super(conf);
1713 this.setMultiplier(conf.getFloat(STOREFILE_SIZE_COST_KEY, DEFAULT_STOREFILE_SIZE_COST));
1714 }
1715
1716 @Override
1717 protected double getCostFromRl(RegionLoad rl) {
1718 return rl.getStorefileSizeMB();
1719 }
1720 }
1721
1722
1723
1724
1725 public static String composeAttributeName(String tableName, String costFunctionName) {
1726 return tableName + TABLE_FUNCTION_SEP + costFunctionName;
1727 }
1728 }