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.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
55
56
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 @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
119 private int maxSteps = 1000000;
120 private boolean runMaxSteps = false;
121 private int stepsPerRegion = 800;
122 private long maxRunningTime = 30 * 1000 * 1;
123 private int numRegionLoadsToRemember = 15;
124
125 private CandidateGenerator[] candidateGenerators;
126 private CostFromRegionLoadFunction[] regionLoadFunctions;
127 private CostFunction[] costFunctions;
128
129
130 private Double curOverallCost = 0d;
131 private Double[] tempFunctionCosts;
132 private Double[] curFunctionCosts;
133
134
135
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
145
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
227 try {
228
229 int tablesCount = isByTable ? services.getTableDescriptors().getAll().size() : 1;
230 int functionsCount = getCostFunctionNames().length;
231
232 updateMetricsSize(tablesCount * (functionsCount + 1));
233 } catch (Exception e) {
234 LOG.error("failed to get the size of all tables", e);
235 }
236 }
237
238
239
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
279
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
298
299
300
301 RegionLocationFinder finder = null;
302 if (this.localityCost != null && this.localityCost.getMultiplier() > 0) {
303 finder = this.regionFinder;
304 }
305
306
307
308
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
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
364 if (newCost < currentCost) {
365 currentCost = newCost;
366
367
368 curOverallCost = currentCost;
369 for (int i = 0; i < this.curFunctionCosts.length; i++) {
370 curFunctionCosts[i] = tempFunctionCosts[i];
371 }
372 } else {
373
374
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
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
413
414 private void updateStochasticCosts(TableName tableName, Double overall, Double[] subCosts) {
415 if (tableName == null) return;
416
417
418 if (metricsBalancer instanceof MetricsStochasticBalancer) {
419 MetricsStochasticBalancer balancer = (MetricsStochasticBalancer) metricsBalancer;
420
421 balancer.updateStochasticCost(tableName.getNameAsString(),
422 "Overall", "Overall cost", overall);
423
424
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
430 balancer.updateStochasticCost(tableName.getNameAsString(), costFunctionName,
431 "The percent of " + costFunctionName, costPercent);
432 }
433 }
434 }
435
436
437
438
439
440
441
442
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
469
470 private synchronized void updateRegionLoad() {
471
472
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
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
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
528
529
530
531
532
533
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
561 abstract static class CandidateGenerator {
562 abstract Cluster.Action generate(Cluster cluster);
563
564
565
566
567
568
569
570
571
572
573
574
575
576 protected int pickRandomRegion(Cluster cluster, int server, double chanceOfNoSwap) {
577
578 if (cluster.regionsPerServer[server].length == 0 || RANDOM.nextFloat() < chanceOfNoSwap) {
579
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
634 int thisRegionCount = cluster.getNumRegions(thisServer);
635 int otherRegionCount = cluster.getNumRegions(otherServer);
636
637
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
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
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
741 thisRegion = pickLowestLocalityRegionOnServer(cluster, thisServer);
742 }
743
744 if (thisRegion == -1) {
745 return Cluster.NullAction;
746 }
747
748
749 int otherServer = cluster.getLeastLoadedTopServerForRegion(thisRegion, thisServer);
750
751 if (otherServer == -1) {
752 return Cluster.NullAction;
753 }
754
755
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
776
777
778 static class RegionReplicaCandidateGenerator extends CandidateGenerator {
779
780 RandomCandidateGenerator randomGenerator = new RandomCandidateGenerator();
781
782
783
784
785
786
787
788
789
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
798
799
800 for (int j = 0; j <= primariesOfRegionsPerGroup.length; j++) {
801 int primary = j < primariesOfRegionsPerGroup.length
802 ? primariesOfRegionsPerGroup[j] : -1;
803 if (primary != currentPrimary) {
804 int numReplicas = j - currentPrimaryIndex;
805 if (numReplicas > 1) {
806
807 double currentRandom = RANDOM.nextDouble();
808
809
810 if (currentRandom > currentLargestRandom) {
811 selectedPrimaryIndex = currentPrimary;
812 currentLargestRandom = currentRandom;
813 }
814 }
815 currentPrimary = primary;
816 currentPrimaryIndex = j;
817 }
818 }
819
820
821
822 for (int j = 0; j < regionsPerGroup.length; j++) {
823 int regionIndex = regionsPerGroup[j];
824 if (selectedPrimaryIndex == regionIndexToPrimaryIndex[regionIndex]) {
825
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
847 if (regionIndex == -1) {
848
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
860
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
876 if (regionIndex == -1) {
877
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
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
913
914
915 void init(Cluster cluster) {
916 this.cluster = cluster;
917 }
918
919
920
921
922
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
951
952
953
954
955
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
965
966 double max = ((count - 1) * mean) + (total - mean);
967
968
969 double min;
970 if (count > total) {
971 min = ((count - total) * mean) + ((1 - mean) * total);
972 } else {
973
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
1001
1002
1003
1004
1005
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
1019
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
1035
1036 this.setMultiplier(conf.getFloat(MOVE_COST_KEY, DEFAULT_MOVE_COST));
1037
1038 maxMovesPercent = conf.getFloat(MAX_MOVES_PERCENT_KEY, DEFAULT_MAX_MOVE_PERCENT);
1039 }
1040
1041 @Override
1042 double cost() {
1043
1044 int maxMoves = Math.max((int) (cluster.numRegions * maxMovesPercent),
1045 DEFAULT_MAX_MOVES);
1046
1047 double moveCost = cluster.numMovedRegions;
1048
1049
1050
1051 if (moveCost > maxMoves) {
1052 return 1000000;
1053 }
1054
1055 return scale(0, Math.min(cluster.numRegions, maxMoves), moveCost);
1056 }
1057 }
1058
1059
1060
1061
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
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
1092
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
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
1132
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
1161
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
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
1196
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
1221
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
1252 long cost = 0;
1253
1254
1255 for(int regionIndex:cluster.regionsPerServer[i]) {
1256 Collection<RegionLoad> regionLoadList = cluster.regionLoads[regionIndex];
1257
1258
1259 if (regionLoadList != null) {
1260 cost += getRegionLoadCost(regionLoadList);
1261 }
1262 }
1263
1264
1265 stats[i] = cost;
1266 }
1267
1268
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
1293
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
1316
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
1337
1338
1339
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;
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
1360 maxCost = cluster.numHosts > 1 ? getMaxCost(cluster) : 0;
1361 costsPerGroup = new long[cluster.numHosts];
1362 primariesOfRegionsPerGroup = cluster.multiServersPerHost
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;
1373 }
1374
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
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
1400
1401
1402
1403
1404
1405
1406 protected long costPerGroup(int[] primariesOfRegions) {
1407 long cost = 0;
1408 int currentPrimary = -1;
1409 int currentPrimaryIndex = -1;
1410
1411
1412 for (int j = 0 ; j <= primariesOfRegions.length; j++) {
1413 int primary = j < primariesOfRegions.length ? primariesOfRegions[j] : -1;
1414 if (primary != currentPrimary) {
1415 int numReplicas = j - currentPrimaryIndex;
1416
1417 if (numReplicas > 1) {
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;
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
1449
1450
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;
1469 }
1470
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;
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
1494
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
1514
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
1535
1536 public static String composeAttributeName(String tableName, String costFunctionName) {
1537 return tableName + TABLE_FUNCTION_SEP + costFunctionName;
1538 }
1539 }