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 java.util.ArrayList;
21 import java.util.Arrays;
22 import java.util.Collection;
23 import java.util.Collections;
24 import java.util.Comparator;
25 import java.util.Deque;
26 import java.util.HashMap;
27 import java.util.HashSet;
28 import java.util.Iterator;
29 import java.util.List;
30 import java.util.Map;
31 import java.util.Map.Entry;
32 import java.util.NavigableMap;
33 import java.util.Random;
34 import java.util.Set;
35 import java.util.TreeMap;
36
37 import org.apache.commons.lang.NotImplementedException;
38 import org.apache.commons.logging.Log;
39 import org.apache.commons.logging.LogFactory;
40 import org.apache.hadoop.conf.Configuration;
41 import org.apache.hadoop.hbase.ClusterStatus;
42 import org.apache.hadoop.hbase.HBaseConfiguration;
43 import org.apache.hadoop.hbase.HBaseIOException;
44 import org.apache.hadoop.hbase.HDFSBlocksDistribution;
45 import org.apache.hadoop.hbase.HRegionInfo;
46 import org.apache.hadoop.hbase.RegionLoad;
47 import org.apache.hadoop.hbase.ServerName;
48 import org.apache.hadoop.hbase.client.RegionReplicaUtil;
49 import org.apache.hadoop.hbase.master.LoadBalancer;
50 import org.apache.hadoop.hbase.master.MasterServices;
51 import org.apache.hadoop.hbase.master.RackManager;
52 import org.apache.hadoop.hbase.master.RegionPlan;
53 import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer.Cluster.Action.Type;
54 import org.apache.hadoop.util.StringUtils;
55
56 import com.google.common.annotations.VisibleForTesting;
57 import com.google.common.base.Joiner;
58 import com.google.common.collect.ArrayListMultimap;
59 import com.google.common.collect.Lists;
60 import com.google.common.collect.Sets;
61
62
63
64
65
66
67
68
69 public abstract class BaseLoadBalancer implements LoadBalancer {
70 protected static final int MIN_SERVER_BALANCE = 2;
71 private volatile boolean stopped = false;
72
73 private static final List<HRegionInfo> EMPTY_REGION_LIST = new ArrayList<HRegionInfo>(0);
74
75 protected RegionLocationFinder regionFinder;
76 protected boolean useRegionFinder;
77
78 private static class DefaultRackManager extends RackManager {
79 @Override
80 public String getRack(ServerName server) {
81 return UNKNOWN_RACK;
82 }
83 }
84
85
86
87
88 protected BaseLoadBalancer() {
89 metricsBalancer = new MetricsBalancer();
90 createRegionFinder();
91 }
92
93
94
95
96
97 protected BaseLoadBalancer(MetricsBalancer metricsBalancer) {
98 this.metricsBalancer = (metricsBalancer != null) ? metricsBalancer : new MetricsBalancer();
99 createRegionFinder();
100 }
101
102 private void createRegionFinder() {
103 useRegionFinder = config.getBoolean("hbase.master.balancer.uselocality", true);
104 if (useRegionFinder) {
105 regionFinder = new RegionLocationFinder();
106 }
107 }
108
109
110
111
112
113
114
115
116
117
118
119 protected static class Cluster {
120 ServerName[] servers;
121 String[] hosts;
122 String[] racks;
123 boolean multiServersPerHost = false;
124
125 ArrayList<String> tables;
126 HRegionInfo[] regions;
127 Deque<RegionLoad>[] regionLoads;
128 private RegionLocationFinder regionFinder;
129
130 int[][] regionLocations;
131
132 int[] serverIndexToHostIndex;
133 int[] serverIndexToRackIndex;
134
135 int[][] regionsPerServer;
136 int[][] regionsPerHost;
137 int[][] regionsPerRack;
138 int[][] primariesOfRegionsPerServer;
139 int[][] primariesOfRegionsPerHost;
140 int[][] primariesOfRegionsPerRack;
141
142 int[][] serversPerHost;
143 int[][] serversPerRack;
144 int[] regionIndexToServerIndex;
145 int[] initialRegionIndexToServerIndex;
146 int[] regionIndexToTableIndex;
147 int[][] numRegionsPerServerPerTable;
148 int[] numMaxRegionsPerTable;
149 int[] regionIndexToPrimaryIndex;
150 boolean hasRegionReplicas = false;
151
152 Integer[] serverIndicesSortedByRegionCount;
153 Integer[] serverIndicesSortedByLocality;
154
155 Map<String, Integer> serversToIndex;
156 Map<String, Integer> hostsToIndex;
157 Map<String, Integer> racksToIndex;
158 Map<String, Integer> tablesToIndex;
159 Map<HRegionInfo, Integer> regionsToIndex;
160 float[] localityPerServer;
161
162 int numServers;
163 int numHosts;
164 int numRacks;
165 int numTables;
166 int numRegions;
167
168 int numMovedRegions = 0;
169 Map<ServerName, List<HRegionInfo>> clusterState;
170
171 protected final RackManager rackManager;
172
173 private float[][] rackLocalities;
174
175 private int[][] regionsToMostLocalEntities;
176
177 protected Cluster(
178 Map<ServerName, List<HRegionInfo>> clusterState,
179 Map<String, Deque<RegionLoad>> loads,
180 RegionLocationFinder regionFinder,
181 RackManager rackManager) {
182 this(null, clusterState, loads, regionFinder, rackManager);
183 }
184
185 @SuppressWarnings("unchecked")
186 protected Cluster(
187 Collection<HRegionInfo> unassignedRegions,
188 Map<ServerName, List<HRegionInfo>> clusterState,
189 Map<String, Deque<RegionLoad>> loads,
190 RegionLocationFinder regionFinder,
191 RackManager rackManager) {
192
193 if (unassignedRegions == null) {
194 unassignedRegions = EMPTY_REGION_LIST;
195 }
196
197 serversToIndex = new HashMap<String, Integer>();
198 hostsToIndex = new HashMap<String, Integer>();
199 racksToIndex = new HashMap<String, Integer>();
200 tablesToIndex = new HashMap<String, Integer>();
201
202
203 tables = new ArrayList<String>();
204 this.rackManager = rackManager != null ? rackManager : new DefaultRackManager();
205
206 numRegions = 0;
207
208 List<List<Integer>> serversPerHostList = new ArrayList<List<Integer>>();
209 List<List<Integer>> serversPerRackList = new ArrayList<List<Integer>>();
210 this.clusterState = clusterState;
211 this.regionFinder = regionFinder;
212
213
214
215 for (ServerName sn : clusterState.keySet()) {
216 if (serversToIndex.get(sn.getHostAndPort()) == null) {
217 serversToIndex.put(sn.getHostAndPort(), numServers++);
218 }
219 if (!hostsToIndex.containsKey(sn.getHostname())) {
220 hostsToIndex.put(sn.getHostname(), numHosts++);
221 serversPerHostList.add(new ArrayList<Integer>(1));
222 }
223
224 int serverIndex = serversToIndex.get(sn.getHostAndPort());
225 int hostIndex = hostsToIndex.get(sn.getHostname());
226 serversPerHostList.get(hostIndex).add(serverIndex);
227
228 String rack = this.rackManager.getRack(sn);
229 if (!racksToIndex.containsKey(rack)) {
230 racksToIndex.put(rack, numRacks++);
231 serversPerRackList.add(new ArrayList<Integer>());
232 }
233 int rackIndex = racksToIndex.get(rack);
234 serversPerRackList.get(rackIndex).add(serverIndex);
235 }
236
237
238 for (Entry<ServerName, List<HRegionInfo>> entry : clusterState.entrySet()) {
239 numRegions += entry.getValue().size();
240 }
241 numRegions += unassignedRegions.size();
242
243 regionsToIndex = new HashMap<HRegionInfo, Integer>(numRegions);
244 servers = new ServerName[numServers];
245 serversPerHost = new int[numHosts][];
246 serversPerRack = new int[numRacks][];
247 regions = new HRegionInfo[numRegions];
248 regionIndexToServerIndex = new int[numRegions];
249 initialRegionIndexToServerIndex = new int[numRegions];
250 regionIndexToTableIndex = new int[numRegions];
251 regionIndexToPrimaryIndex = new int[numRegions];
252 regionLoads = new Deque[numRegions];
253
254 regionLocations = new int[numRegions][];
255 serverIndicesSortedByRegionCount = new Integer[numServers];
256 serverIndicesSortedByLocality = new Integer[numServers];
257 localityPerServer = new float[numServers];
258
259 serverIndexToHostIndex = new int[numServers];
260 serverIndexToRackIndex = new int[numServers];
261 regionsPerServer = new int[numServers][];
262 regionsPerHost = new int[numHosts][];
263 regionsPerRack = new int[numRacks][];
264 primariesOfRegionsPerServer = new int[numServers][];
265 primariesOfRegionsPerHost = new int[numHosts][];
266 primariesOfRegionsPerRack = new int[numRacks][];
267
268 int tableIndex = 0, regionIndex = 0, regionPerServerIndex = 0;
269
270 for (Entry<ServerName, List<HRegionInfo>> entry : clusterState.entrySet()) {
271 int serverIndex = serversToIndex.get(entry.getKey().getHostAndPort());
272
273
274
275 if (servers[serverIndex] == null ||
276 servers[serverIndex].getStartcode() < entry.getKey().getStartcode()) {
277 servers[serverIndex] = entry.getKey();
278 }
279
280 if (regionsPerServer[serverIndex] != null) {
281
282
283 regionsPerServer[serverIndex] = new int[entry.getValue().size() + regionsPerServer[serverIndex].length];
284 } else {
285 regionsPerServer[serverIndex] = new int[entry.getValue().size()];
286 }
287 primariesOfRegionsPerServer[serverIndex] = new int[regionsPerServer[serverIndex].length];
288 serverIndicesSortedByRegionCount[serverIndex] = serverIndex;
289 serverIndicesSortedByLocality[serverIndex] = serverIndex;
290 }
291
292 hosts = new String[numHosts];
293 for (Entry<String, Integer> entry : hostsToIndex.entrySet()) {
294 hosts[entry.getValue()] = entry.getKey();
295 }
296 racks = new String[numRacks];
297 for (Entry<String, Integer> entry : racksToIndex.entrySet()) {
298 racks[entry.getValue()] = entry.getKey();
299 }
300
301 for (Entry<ServerName, List<HRegionInfo>> entry : clusterState.entrySet()) {
302 int serverIndex = serversToIndex.get(entry.getKey().getHostAndPort());
303 regionPerServerIndex = 0;
304
305 int hostIndex = hostsToIndex.get(entry.getKey().getHostname());
306 serverIndexToHostIndex[serverIndex] = hostIndex;
307
308 int rackIndex = racksToIndex.get(this.rackManager.getRack(entry.getKey()));
309 serverIndexToRackIndex[serverIndex] = rackIndex;
310
311 for (HRegionInfo region : entry.getValue()) {
312 registerRegion(region, regionIndex, serverIndex, loads, regionFinder);
313 regionsPerServer[serverIndex][regionPerServerIndex++] = regionIndex;
314 regionIndex++;
315 }
316 }
317
318 for (HRegionInfo region : unassignedRegions) {
319 registerRegion(region, regionIndex, -1, loads, regionFinder);
320 regionIndex++;
321 }
322
323 for (int i = 0; i < serversPerHostList.size(); i++) {
324 serversPerHost[i] = new int[serversPerHostList.get(i).size()];
325 for (int j = 0; j < serversPerHost[i].length; j++) {
326 serversPerHost[i][j] = serversPerHostList.get(i).get(j);
327 }
328 if (serversPerHost[i].length > 1) {
329 multiServersPerHost = true;
330 }
331 }
332
333 for (int i = 0; i < serversPerRackList.size(); i++) {
334 serversPerRack[i] = new int[serversPerRackList.get(i).size()];
335 for (int j = 0; j < serversPerRack[i].length; j++) {
336 serversPerRack[i][j] = serversPerRackList.get(i).get(j);
337 }
338 }
339
340 numTables = tables.size();
341 numRegionsPerServerPerTable = new int[numServers][numTables];
342
343 for (int i = 0; i < numServers; i++) {
344 for (int j = 0; j < numTables; j++) {
345 numRegionsPerServerPerTable[i][j] = 0;
346 }
347 }
348
349 for (int i=0; i < regionIndexToServerIndex.length; i++) {
350 if (regionIndexToServerIndex[i] >= 0) {
351 numRegionsPerServerPerTable[regionIndexToServerIndex[i]][regionIndexToTableIndex[i]]++;
352 }
353 }
354
355 numMaxRegionsPerTable = new int[numTables];
356 for (int serverIndex = 0 ; serverIndex < numRegionsPerServerPerTable.length; serverIndex++) {
357 for (tableIndex = 0 ; tableIndex < numRegionsPerServerPerTable[serverIndex].length; tableIndex++) {
358 if (numRegionsPerServerPerTable[serverIndex][tableIndex] > numMaxRegionsPerTable[tableIndex]) {
359 numMaxRegionsPerTable[tableIndex] = numRegionsPerServerPerTable[serverIndex][tableIndex];
360 }
361 }
362 }
363
364 for (int i = 0; i < regions.length; i ++) {
365 HRegionInfo info = regions[i];
366 if (RegionReplicaUtil.isDefaultReplica(info)) {
367 regionIndexToPrimaryIndex[i] = i;
368 } else {
369 hasRegionReplicas = true;
370 HRegionInfo primaryInfo = RegionReplicaUtil.getRegionInfoForDefaultReplica(info);
371 regionIndexToPrimaryIndex[i] =
372 regionsToIndex.containsKey(primaryInfo) ?
373 regionsToIndex.get(primaryInfo):
374 -1;
375 }
376 }
377
378 for (int i = 0; i < regionsPerServer.length; i++) {
379 primariesOfRegionsPerServer[i] = new int[regionsPerServer[i].length];
380 for (int j = 0; j < regionsPerServer[i].length; j++) {
381 int primaryIndex = regionIndexToPrimaryIndex[regionsPerServer[i][j]];
382 primariesOfRegionsPerServer[i][j] = primaryIndex;
383 }
384
385 Arrays.sort(primariesOfRegionsPerServer[i]);
386 }
387
388
389 if (multiServersPerHost) {
390 for (int i = 0 ; i < serversPerHost.length; i++) {
391 int numRegionsPerHost = 0;
392 for (int j = 0; j < serversPerHost[i].length; j++) {
393 numRegionsPerHost += regionsPerServer[serversPerHost[i][j]].length;
394 }
395 regionsPerHost[i] = new int[numRegionsPerHost];
396 primariesOfRegionsPerHost[i] = new int[numRegionsPerHost];
397 }
398 for (int i = 0 ; i < serversPerHost.length; i++) {
399 int numRegionPerHostIndex = 0;
400 for (int j = 0; j < serversPerHost[i].length; j++) {
401 for (int k = 0; k < regionsPerServer[serversPerHost[i][j]].length; k++) {
402 int region = regionsPerServer[serversPerHost[i][j]][k];
403 regionsPerHost[i][numRegionPerHostIndex] = region;
404 int primaryIndex = regionIndexToPrimaryIndex[region];
405 primariesOfRegionsPerHost[i][numRegionPerHostIndex] = primaryIndex;
406 numRegionPerHostIndex++;
407 }
408 }
409
410 Arrays.sort(primariesOfRegionsPerHost[i]);
411 }
412 }
413
414
415 if (numRacks > 1) {
416 for (int i = 0 ; i < serversPerRack.length; i++) {
417 int numRegionsPerRack = 0;
418 for (int j = 0; j < serversPerRack[i].length; j++) {
419 numRegionsPerRack += regionsPerServer[serversPerRack[i][j]].length;
420 }
421 regionsPerRack[i] = new int[numRegionsPerRack];
422 primariesOfRegionsPerRack[i] = new int[numRegionsPerRack];
423 }
424
425 for (int i = 0 ; i < serversPerRack.length; i++) {
426 int numRegionPerRackIndex = 0;
427 for (int j = 0; j < serversPerRack[i].length; j++) {
428 for (int k = 0; k < regionsPerServer[serversPerRack[i][j]].length; k++) {
429 int region = regionsPerServer[serversPerRack[i][j]][k];
430 regionsPerRack[i][numRegionPerRackIndex] = region;
431 int primaryIndex = regionIndexToPrimaryIndex[region];
432 primariesOfRegionsPerRack[i][numRegionPerRackIndex] = primaryIndex;
433 numRegionPerRackIndex++;
434 }
435 }
436
437 Arrays.sort(primariesOfRegionsPerRack[i]);
438 }
439 }
440 }
441
442
443 private void registerRegion(HRegionInfo region, int regionIndex,
444 int serverIndex, Map<String, Deque<RegionLoad>> loads,
445 RegionLocationFinder regionFinder) {
446 String tableName = region.getTable().getNameAsString();
447 if (!tablesToIndex.containsKey(tableName)) {
448 tables.add(tableName);
449 tablesToIndex.put(tableName, tablesToIndex.size());
450 }
451 int tableIndex = tablesToIndex.get(tableName);
452
453 regionsToIndex.put(region, regionIndex);
454 regions[regionIndex] = region;
455 regionIndexToServerIndex[regionIndex] = serverIndex;
456 initialRegionIndexToServerIndex[regionIndex] = serverIndex;
457 regionIndexToTableIndex[regionIndex] = tableIndex;
458
459
460 if (loads != null) {
461 Deque<RegionLoad> rl = loads.get(region.getRegionNameAsString());
462
463 if (rl == null) {
464
465 rl = loads.get(region.getEncodedName());
466 }
467 regionLoads[regionIndex] = rl;
468 }
469
470 if (regionFinder != null) {
471
472 List<ServerName> loc = regionFinder.getTopBlockLocations(region);
473 regionLocations[regionIndex] = new int[loc.size()];
474 for (int i = 0; i < loc.size(); i++) {
475 regionLocations[regionIndex][i] = loc.get(i) == null ? -1
476 : (serversToIndex.get(loc.get(i).getHostAndPort()) == null ? -1
477 : serversToIndex.get(loc.get(i).getHostAndPort()));
478 }
479 }
480 }
481
482
483
484
485 public boolean serverHasTooFewRegions(int server) {
486 int minLoad = this.numRegions / numServers;
487 int numRegions = getNumRegions(server);
488 return numRegions < minLoad;
489 }
490
491
492
493
494
495 public float[][] getOrComputeRackLocalities() {
496 if (rackLocalities == null || regionsToMostLocalEntities == null) {
497 computeCachedLocalities();
498 }
499 return rackLocalities;
500 }
501
502
503
504
505
506 public int[] getOrComputeRegionsToMostLocalEntities(LocalityType type) {
507 if (rackLocalities == null || regionsToMostLocalEntities == null) {
508 computeCachedLocalities();
509 }
510 return regionsToMostLocalEntities[type.ordinal()];
511 }
512
513
514
515
516
517 public float getOrComputeLocality(int region, int entity, LocalityType type) {
518 switch (type) {
519 case SERVER:
520 return getLocalityOfRegion(region, entity);
521 case RACK:
522 return getOrComputeRackLocalities()[region][entity];
523 default:
524 throw new IllegalArgumentException("Unsupported LocalityType: " + type);
525 }
526 }
527
528
529
530
531
532 public double getOrComputeWeightedLocality(int region, int server, LocalityType type) {
533 return getRegionSizeMB(region) * getOrComputeLocality(region, server, type);
534 }
535
536
537
538
539 public int getRegionSizeMB(int region) {
540 Deque<RegionLoad> load = regionLoads[region];
541
542 if (load == null) {
543 return 0;
544 }
545 return regionLoads[region].getLast().getStorefileSizeMB();
546 }
547
548
549
550
551
552
553 private void computeCachedLocalities() {
554 rackLocalities = new float[numRegions][numServers];
555 regionsToMostLocalEntities = new int[LocalityType.values().length][numRegions];
556
557
558 for (int region = 0; region < numRegions; region++) {
559 int serverWithBestLocality = 0;
560 float bestLocalityForRegion = 0;
561 for (int server = 0; server < numServers; server++) {
562
563 float locality = getLocalityOfRegion(region, server);
564 int rack = serverIndexToRackIndex[server];
565 int numServersInRack = serversPerRack[rack].length;
566 rackLocalities[region][rack] += locality / numServersInRack;
567
568 if (locality > bestLocalityForRegion) {
569 serverWithBestLocality = server;
570 bestLocalityForRegion = locality;
571 }
572 }
573 regionsToMostLocalEntities[LocalityType.SERVER.ordinal()][region] = serverWithBestLocality;
574
575
576 int rackWithBestLocality = 0;
577 float bestRackLocalityForRegion = 0.0f;
578 for (int rack = 0; rack < numRacks; rack++) {
579 float rackLocality = rackLocalities[region][rack];
580 if (rackLocality > bestRackLocalityForRegion) {
581 bestRackLocalityForRegion = rackLocality;
582 rackWithBestLocality = rack;
583 }
584 }
585 regionsToMostLocalEntities[LocalityType.RACK.ordinal()][region] = rackWithBestLocality;
586 }
587
588 }
589
590
591
592
593 public int getRackForRegion(int region) {
594 return serverIndexToRackIndex[regionIndexToServerIndex[region]];
595 }
596
597 enum LocalityType {
598 SERVER,
599 RACK
600 }
601
602
603 public static class Action {
604 public static enum Type {
605 ASSIGN_REGION,
606 MOVE_REGION,
607 SWAP_REGIONS,
608 NULL,
609 }
610
611 public Type type;
612 public Action (Type type) {this.type = type;}
613
614 public Action undoAction() { return this; }
615 @Override
616 public String toString() { return type + ":";}
617 }
618
619 public static class AssignRegionAction extends Action {
620 public int region;
621 public int server;
622 public AssignRegionAction(int region, int server) {
623 super(Type.ASSIGN_REGION);
624 this.region = region;
625 this.server = server;
626 }
627 @Override
628 public Action undoAction() {
629
630
631 throw new NotImplementedException();
632 }
633 @Override
634 public String toString() {
635 return type + ": " + region + ":" + server;
636 }
637 }
638
639 public static class MoveRegionAction extends Action {
640 public int region;
641 public int fromServer;
642 public int toServer;
643
644 public MoveRegionAction(int region, int fromServer, int toServer) {
645 super(Type.MOVE_REGION);
646 this.fromServer = fromServer;
647 this.region = region;
648 this.toServer = toServer;
649 }
650 @Override
651 public Action undoAction() {
652 return new MoveRegionAction (region, toServer, fromServer);
653 }
654 @Override
655 public String toString() {
656 return type + ": " + region + ":" + fromServer + " -> " + toServer;
657 }
658 }
659
660 public static class SwapRegionsAction extends Action {
661 public int fromServer;
662 public int fromRegion;
663 public int toServer;
664 public int toRegion;
665 public SwapRegionsAction(int fromServer, int fromRegion, int toServer, int toRegion) {
666 super(Type.SWAP_REGIONS);
667 this.fromServer = fromServer;
668 this.fromRegion = fromRegion;
669 this.toServer = toServer;
670 this.toRegion = toRegion;
671 }
672 @Override
673 public Action undoAction() {
674 return new SwapRegionsAction (fromServer, toRegion, toServer, fromRegion);
675 }
676 @Override
677 public String toString() {
678 return type + ": " + fromRegion + ":" + fromServer + " <-> " + toRegion + ":" + toServer;
679 }
680 }
681
682 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="NM_FIELD_NAMING_CONVENTION",
683 justification="Mistake. Too disruptive to change now")
684 public static final Action NullAction = new Action(Type.NULL);
685
686 public void doAction(Action action) {
687 switch (action.type) {
688 case NULL: break;
689 case ASSIGN_REGION:
690
691 assert action instanceof AssignRegionAction: action.getClass();
692 AssignRegionAction ar = (AssignRegionAction) action;
693 regionsPerServer[ar.server] = addRegion(regionsPerServer[ar.server], ar.region);
694 regionMoved(ar.region, -1, ar.server);
695 break;
696 case MOVE_REGION:
697 assert action instanceof MoveRegionAction: action.getClass();
698 MoveRegionAction mra = (MoveRegionAction) action;
699 regionsPerServer[mra.fromServer] = removeRegion(regionsPerServer[mra.fromServer], mra.region);
700 regionsPerServer[mra.toServer] = addRegion(regionsPerServer[mra.toServer], mra.region);
701 regionMoved(mra.region, mra.fromServer, mra.toServer);
702 break;
703 case SWAP_REGIONS:
704 assert action instanceof SwapRegionsAction: action.getClass();
705 SwapRegionsAction a = (SwapRegionsAction) action;
706 regionsPerServer[a.fromServer] = replaceRegion(regionsPerServer[a.fromServer], a.fromRegion, a.toRegion);
707 regionsPerServer[a.toServer] = replaceRegion(regionsPerServer[a.toServer], a.toRegion, a.fromRegion);
708 regionMoved(a.fromRegion, a.fromServer, a.toServer);
709 regionMoved(a.toRegion, a.toServer, a.fromServer);
710 break;
711 default:
712 throw new RuntimeException("Uknown action:" + action.type);
713 }
714 }
715
716
717
718
719
720
721
722
723 boolean wouldLowerAvailability(HRegionInfo regionInfo, ServerName serverName) {
724 if (!serversToIndex.containsKey(serverName.getHostAndPort())) {
725 return false;
726 }
727 int server = serversToIndex.get(serverName.getHostAndPort());
728 int region = regionsToIndex.get(regionInfo);
729
730 int primary = regionIndexToPrimaryIndex[region];
731
732
733
734
735 if (contains(primariesOfRegionsPerServer[server], primary)) {
736
737 for (int i = 0; i < primariesOfRegionsPerServer.length; i++) {
738 if (i != server && !contains(primariesOfRegionsPerServer[i], primary)) {
739 return true;
740 }
741 }
742 return false;
743 }
744
745
746 if (multiServersPerHost) {
747 int host = serverIndexToHostIndex[server];
748 if (contains(primariesOfRegionsPerHost[host], primary)) {
749
750 for (int i = 0; i < primariesOfRegionsPerHost.length; i++) {
751 if (i != host && !contains(primariesOfRegionsPerHost[i], primary)) {
752 return true;
753 }
754 }
755 return false;
756 }
757 }
758
759
760 if (numRacks > 1) {
761 int rack = serverIndexToRackIndex[server];
762 if (contains(primariesOfRegionsPerRack[rack], primary)) {
763
764 for (int i = 0; i < primariesOfRegionsPerRack.length; i++) {
765 if (i != rack && !contains(primariesOfRegionsPerRack[i], primary)) {
766 return true;
767 }
768 }
769 return false;
770 }
771 }
772 return false;
773 }
774
775 void doAssignRegion(HRegionInfo regionInfo, ServerName serverName) {
776 if (!serversToIndex.containsKey(serverName.getHostAndPort())) {
777 return;
778 }
779 int server = serversToIndex.get(serverName.getHostAndPort());
780 int region = regionsToIndex.get(regionInfo);
781 doAction(new AssignRegionAction(region, server));
782 }
783
784 void regionMoved(int region, int oldServer, int newServer) {
785 regionIndexToServerIndex[region] = newServer;
786 if (initialRegionIndexToServerIndex[region] == newServer) {
787 numMovedRegions--;
788 } else if (oldServer >= 0 && initialRegionIndexToServerIndex[region] == oldServer) {
789 numMovedRegions++;
790 }
791 int tableIndex = regionIndexToTableIndex[region];
792 if (oldServer >= 0) {
793 numRegionsPerServerPerTable[oldServer][tableIndex]--;
794 }
795 numRegionsPerServerPerTable[newServer][tableIndex]++;
796
797
798 if (numRegionsPerServerPerTable[newServer][tableIndex] > numMaxRegionsPerTable[tableIndex]) {
799 numMaxRegionsPerTable[tableIndex] = numRegionsPerServerPerTable[newServer][tableIndex];
800 } else if (oldServer >= 0 && (numRegionsPerServerPerTable[oldServer][tableIndex] + 1)
801 == numMaxRegionsPerTable[tableIndex]) {
802
803 numMaxRegionsPerTable[tableIndex] = 0;
804 for (int serverIndex = 0 ; serverIndex < numRegionsPerServerPerTable.length; serverIndex++) {
805 if (numRegionsPerServerPerTable[serverIndex][tableIndex] > numMaxRegionsPerTable[tableIndex]) {
806 numMaxRegionsPerTable[tableIndex] = numRegionsPerServerPerTable[serverIndex][tableIndex];
807 }
808 }
809 }
810
811
812 int primary = regionIndexToPrimaryIndex[region];
813 if (oldServer >= 0) {
814 primariesOfRegionsPerServer[oldServer] = removeRegion(
815 primariesOfRegionsPerServer[oldServer], primary);
816 }
817 primariesOfRegionsPerServer[newServer] = addRegionSorted(
818 primariesOfRegionsPerServer[newServer], primary);
819
820
821 if (multiServersPerHost) {
822 int oldHost = oldServer >= 0 ? serverIndexToHostIndex[oldServer] : -1;
823 int newHost = serverIndexToHostIndex[newServer];
824 if (newHost != oldHost) {
825 regionsPerHost[newHost] = addRegion(regionsPerHost[newHost], region);
826 primariesOfRegionsPerHost[newHost] = addRegionSorted(primariesOfRegionsPerHost[newHost], primary);
827 if (oldHost >= 0) {
828 regionsPerHost[oldHost] = removeRegion(regionsPerHost[oldHost], region);
829 primariesOfRegionsPerHost[oldHost] = removeRegion(
830 primariesOfRegionsPerHost[oldHost], primary);
831 }
832 }
833 }
834
835
836 if (numRacks > 1) {
837 int oldRack = oldServer >= 0 ? serverIndexToRackIndex[oldServer] : -1;
838 int newRack = serverIndexToRackIndex[newServer];
839 if (newRack != oldRack) {
840 regionsPerRack[newRack] = addRegion(regionsPerRack[newRack], region);
841 primariesOfRegionsPerRack[newRack] = addRegionSorted(primariesOfRegionsPerRack[newRack], primary);
842 if (oldRack >= 0) {
843 regionsPerRack[oldRack] = removeRegion(regionsPerRack[oldRack], region);
844 primariesOfRegionsPerRack[oldRack] = removeRegion(
845 primariesOfRegionsPerRack[oldRack], primary);
846 }
847 }
848 }
849 }
850
851 int[] removeRegion(int[] regions, int regionIndex) {
852
853 int[] newRegions = new int[regions.length - 1];
854 int i = 0;
855 for (i = 0; i < regions.length; i++) {
856 if (regions[i] == regionIndex) {
857 break;
858 }
859 newRegions[i] = regions[i];
860 }
861 System.arraycopy(regions, i+1, newRegions, i, newRegions.length - i);
862 return newRegions;
863 }
864
865 int[] addRegion(int[] regions, int regionIndex) {
866 int[] newRegions = new int[regions.length + 1];
867 System.arraycopy(regions, 0, newRegions, 0, regions.length);
868 newRegions[newRegions.length - 1] = regionIndex;
869 return newRegions;
870 }
871
872 int[] addRegionSorted(int[] regions, int regionIndex) {
873 int[] newRegions = new int[regions.length + 1];
874 int i = 0;
875 for (i = 0; i < regions.length; i++) {
876 if (regions[i] > regionIndex) {
877 break;
878 }
879 }
880 System.arraycopy(regions, 0, newRegions, 0, i);
881 System.arraycopy(regions, i, newRegions, i+1, regions.length - i);
882 newRegions[i] = regionIndex;
883
884 return newRegions;
885 }
886
887 int[] replaceRegion(int[] regions, int regionIndex, int newRegionIndex) {
888 int i = 0;
889 for (i = 0; i < regions.length; i++) {
890 if (regions[i] == regionIndex) {
891 regions[i] = newRegionIndex;
892 break;
893 }
894 }
895 return regions;
896 }
897
898 void sortServersByRegionCount() {
899 Arrays.sort(serverIndicesSortedByRegionCount, numRegionsComparator);
900 }
901
902 int getNumRegions(int server) {
903 return regionsPerServer[server].length;
904 }
905
906 boolean contains(int[] arr, int val) {
907 return Arrays.binarySearch(arr, val) >= 0;
908 }
909
910 private Comparator<Integer> numRegionsComparator = new Comparator<Integer>() {
911 @Override
912 public int compare(Integer integer, Integer integer2) {
913 return Integer.compare(getNumRegions(integer), getNumRegions(integer2));
914 }
915 };
916
917 void sortServersByLocality() {
918 Arrays.sort(serverIndicesSortedByLocality, localityComparator);
919 }
920
921 float getLocality(int server) {
922 return localityPerServer[server];
923 }
924
925 private Comparator<Integer> localityComparator = new Comparator<Integer>() {
926 @Override
927 public int compare(Integer integer, Integer integer2) {
928 return Float.compare(getLocality(integer), getLocality(integer2));
929 }
930 };
931
932 int getLowestLocalityRegionServer() {
933 if (regionFinder == null) {
934 return -1;
935 } else {
936 sortServersByLocality();
937
938 int i = 0;
939 int lowestLocalityServerIndex = serverIndicesSortedByLocality[i];
940 while (localityPerServer[lowestLocalityServerIndex] == 0
941 && (regionsPerServer[lowestLocalityServerIndex].length == 0)) {
942 i++;
943 lowestLocalityServerIndex = serverIndicesSortedByLocality[i];
944 }
945 if (LOG.isTraceEnabled()) {
946 LOG.trace("Lowest locality region server with non zero regions is "
947 + servers[lowestLocalityServerIndex].getHostname() + " with locality "
948 + localityPerServer[lowestLocalityServerIndex]);
949 }
950 return lowestLocalityServerIndex;
951 }
952 }
953
954 int getLowestLocalityRegionOnServer(int serverIndex) {
955 if (regionFinder != null) {
956 float lowestLocality = 1.0f;
957 int lowestLocalityRegionIndex = -1;
958 if (regionsPerServer[serverIndex].length == 0) {
959
960 return -1;
961 }
962 for (int j = 0; j < regionsPerServer[serverIndex].length; j++) {
963 int regionIndex = regionsPerServer[serverIndex][j];
964 HDFSBlocksDistribution distribution = regionFinder
965 .getBlockDistribution(regions[regionIndex]);
966 float locality = distribution.getBlockLocalityIndex(servers[serverIndex].getHostname());
967
968 if (distribution.getUniqueBlocksTotalWeight() == 0) {
969 continue;
970 }
971 if (locality < lowestLocality) {
972 lowestLocality = locality;
973 lowestLocalityRegionIndex = j;
974 }
975 }
976 if (lowestLocalityRegionIndex == -1) {
977 return -1;
978 }
979 if (LOG.isTraceEnabled()) {
980 LOG.trace("Lowest locality region is "
981 + regions[regionsPerServer[serverIndex][lowestLocalityRegionIndex]]
982 .getRegionNameAsString() + " with locality " + lowestLocality
983 + " and its region server contains " + regionsPerServer[serverIndex].length
984 + " regions");
985 }
986 return regionsPerServer[serverIndex][lowestLocalityRegionIndex];
987 } else {
988 return -1;
989 }
990 }
991
992 float getLocalityOfRegion(int region, int server) {
993 if (regionFinder != null) {
994 HDFSBlocksDistribution distribution = regionFinder.getBlockDistribution(regions[region]);
995 return distribution.getBlockLocalityIndex(servers[server].getHostname());
996 } else {
997 return 0f;
998 }
999 }
1000
1001
1002
1003
1004
1005 int getLeastLoadedTopServerForRegion(int region, int currentServer) {
1006 if (regionFinder != null) {
1007 List<ServerName> topLocalServers = regionFinder.getTopBlockLocations(regions[region],
1008 servers[currentServer].getHostname());
1009 int leastLoadedServerIndex = -1;
1010 int load = Integer.MAX_VALUE;
1011 for (ServerName sn : topLocalServers) {
1012 if (!serversToIndex.containsKey(sn.getHostAndPort())) {
1013 continue;
1014 }
1015 int index = serversToIndex.get(sn.getHostAndPort());
1016 if (regionsPerServer[index] == null) {
1017 continue;
1018 }
1019 int tempLoad = regionsPerServer[index].length;
1020 if (tempLoad <= load) {
1021 leastLoadedServerIndex = index;
1022 load = tempLoad;
1023 }
1024 }
1025 if (leastLoadedServerIndex != -1) {
1026 if (LOG.isTraceEnabled()) {
1027 LOG.trace(
1028 "Pick the least loaded server " + servers[leastLoadedServerIndex].getHostname()
1029 + " with better locality for region " + regions[region]);
1030 }
1031 }
1032 return leastLoadedServerIndex;
1033 } else {
1034 return -1;
1035 }
1036 }
1037
1038 void calculateRegionServerLocalities() {
1039 if (regionFinder == null) {
1040 LOG.warn("Region location finder found null, skipping locality calculations.");
1041 return;
1042 }
1043 for (int i = 0; i < regionsPerServer.length; i++) {
1044 HDFSBlocksDistribution distribution = new HDFSBlocksDistribution();
1045 if (regionsPerServer[i].length > 0) {
1046 for (int j = 0; j < regionsPerServer[i].length; j++) {
1047 int regionIndex = regionsPerServer[i][j];
1048 distribution.add(regionFinder.getBlockDistribution(regions[regionIndex]));
1049 }
1050 } else {
1051 LOG.debug("Server " + servers[i].getHostname() + " had 0 regions.");
1052 }
1053 localityPerServer[i] = distribution.getBlockLocalityIndex(servers[i].getHostname());
1054 }
1055 }
1056
1057 @VisibleForTesting
1058 protected void setNumRegions(int numRegions) {
1059 this.numRegions = numRegions;
1060 }
1061
1062 @VisibleForTesting
1063 protected void setNumMovedRegions(int numMovedRegions) {
1064 this.numMovedRegions = numMovedRegions;
1065 }
1066
1067 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="SBSC_USE_STRINGBUFFER_CONCATENATION",
1068 justification="Not important but should be fixed")
1069 @Override
1070 public String toString() {
1071 String desc = "Cluster{" +
1072 "servers=[";
1073 for(ServerName sn:servers) {
1074 desc += sn.getHostAndPort() + ", ";
1075 }
1076 desc +=
1077 ", serverIndicesSortedByRegionCount="+
1078 Arrays.toString(serverIndicesSortedByRegionCount) +
1079 ", regionsPerServer=[";
1080
1081 for (int[]r:regionsPerServer) {
1082 desc += Arrays.toString(r);
1083 }
1084 desc += "]" +
1085 ", numMaxRegionsPerTable=" +
1086 Arrays.toString(numMaxRegionsPerTable) +
1087 ", numRegions=" +
1088 numRegions +
1089 ", numServers=" +
1090 numServers +
1091 ", numTables=" +
1092 numTables +
1093 ", numMovedRegions=" +
1094 numMovedRegions +
1095 '}';
1096 return desc;
1097 }
1098 }
1099
1100
1101 protected float slop;
1102 protected Configuration config = HBaseConfiguration.create();
1103 protected RackManager rackManager;
1104 private static final Random RANDOM = new Random(System.currentTimeMillis());
1105 private static final Log LOG = LogFactory.getLog(BaseLoadBalancer.class);
1106
1107 public static final String TABLES_ON_MASTER =
1108 "hbase.balancer.tablesOnMaster";
1109
1110 protected final Set<String> tablesOnMaster = new HashSet<String>();
1111 protected MetricsBalancer metricsBalancer = null;
1112 protected ClusterStatus clusterStatus = null;
1113 protected ServerName masterServerName;
1114 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="IS2_INCONSISTENT_SYNC",
1115 justification="The services is just assigned once when master start")
1116 protected MasterServices services;
1117
1118 protected static String[] getTablesOnMaster(Configuration conf) {
1119 String valueString = conf.get(TABLES_ON_MASTER);
1120 if (valueString != null) {
1121 valueString = valueString.trim();
1122 }
1123 if (valueString == null || valueString.equalsIgnoreCase("none")) {
1124 return null;
1125 }
1126 return StringUtils.getStrings(valueString);
1127 }
1128
1129
1130
1131
1132 public static boolean tablesOnMaster(Configuration conf) {
1133 String[] tables = getTablesOnMaster(conf);
1134 return tables != null && tables.length > 0;
1135 }
1136
1137 public static boolean userTablesOnMaster(Configuration conf) {
1138 String[] tables = getTablesOnMaster(conf);
1139 if (tables == null || tables.length == 0) {
1140 return false;
1141 }
1142 for (String tn:tables) {
1143 if (!tn.startsWith("hbase:")) {
1144 return true;
1145 }
1146 }
1147 return false;
1148 }
1149
1150 @Override
1151 public void setConf(Configuration conf) {
1152 setSlop(conf);
1153 if (slop < 0) slop = 0;
1154 else if (slop > 1) slop = 1;
1155
1156 this.config = conf;
1157 String[] tables = getTablesOnMaster(conf);
1158 if (tables != null && tables.length > 0) {
1159 Collections.addAll(tablesOnMaster, tables);
1160 }
1161 this.rackManager = new RackManager(getConf());
1162 if (useRegionFinder) {
1163 regionFinder.setConf(conf);
1164 }
1165 }
1166
1167 protected void setSlop(Configuration conf) {
1168 this.slop = conf.getFloat("hbase.regions.slop", (float) 0.2);
1169 }
1170
1171
1172
1173
1174
1175 public boolean shouldBeOnMaster(HRegionInfo region) {
1176 return tablesOnMaster.contains(region.getTable().getNameAsString())
1177 && region.getReplicaId() == HRegionInfo.DEFAULT_REPLICA_ID;
1178 }
1179
1180
1181
1182
1183 protected List<RegionPlan> balanceMasterRegions(
1184 Map<ServerName, List<HRegionInfo>> clusterMap) {
1185 if (masterServerName == null
1186 || clusterMap == null || clusterMap.size() <= 1) return null;
1187 List<RegionPlan> plans = null;
1188 List<HRegionInfo> regions = clusterMap.get(masterServerName);
1189 if (regions != null) {
1190 Iterator<ServerName> keyIt = null;
1191 for (HRegionInfo region: regions) {
1192 if (shouldBeOnMaster(region)) continue;
1193
1194
1195 if (keyIt == null || !keyIt.hasNext()) {
1196 keyIt = clusterMap.keySet().iterator();
1197 }
1198 ServerName dest = keyIt.next();
1199 if (masterServerName.equals(dest)) {
1200 if (!keyIt.hasNext()) {
1201 keyIt = clusterMap.keySet().iterator();
1202 }
1203 dest = keyIt.next();
1204 }
1205
1206
1207 RegionPlan plan = new RegionPlan(region, masterServerName, dest);
1208 if (plans == null) {
1209 plans = new ArrayList<RegionPlan>();
1210 }
1211 plans.add(plan);
1212 }
1213 }
1214 for (Map.Entry<ServerName, List<HRegionInfo>> server: clusterMap.entrySet()) {
1215 if (masterServerName.equals(server.getKey())) continue;
1216 for (HRegionInfo region: server.getValue()) {
1217 if (!shouldBeOnMaster(region)) continue;
1218
1219
1220 RegionPlan plan = new RegionPlan(region, server.getKey(), masterServerName);
1221 if (plans == null) {
1222 plans = new ArrayList<RegionPlan>();
1223 }
1224 plans.add(plan);
1225 }
1226 }
1227 return plans;
1228 }
1229
1230
1231
1232
1233 protected Map<ServerName, List<HRegionInfo>> assignMasterRegions(
1234 Collection<HRegionInfo> regions, List<ServerName> servers) {
1235 if (servers == null || regions == null || regions.isEmpty()) {
1236 return null;
1237 }
1238 Map<ServerName, List<HRegionInfo>> assignments
1239 = new TreeMap<ServerName, List<HRegionInfo>>();
1240 if (masterServerName != null && servers.contains(masterServerName)) {
1241 assignments.put(masterServerName, new ArrayList<HRegionInfo>());
1242 for (HRegionInfo region: regions) {
1243 if (shouldBeOnMaster(region)) {
1244 assignments.get(masterServerName).add(region);
1245 }
1246 }
1247 }
1248 return assignments;
1249 }
1250
1251 @Override
1252 public Configuration getConf() {
1253 return this.config;
1254 }
1255
1256 @Override
1257 public synchronized void setClusterStatus(ClusterStatus st) {
1258 this.clusterStatus = st;
1259 if (useRegionFinder) {
1260 regionFinder.setClusterStatus(st);
1261 }
1262 }
1263
1264 @Override
1265 public void setMasterServices(MasterServices masterServices) {
1266 masterServerName = masterServices.getServerName();
1267 this.services = masterServices;
1268 if (useRegionFinder) {
1269 this.regionFinder.setServices(masterServices);
1270 }
1271 }
1272
1273 @Override
1274 public void postMasterStartupInitialize() {
1275 if (services != null && regionFinder != null) {
1276 try {
1277 Set<HRegionInfo> regions =
1278 services.getAssignmentManager().getRegionStates().getRegionAssignments().keySet();
1279 regionFinder.refreshAndWait(regions);
1280 } catch (Exception e) {
1281 LOG.warn("Refreshing region HDFS Block dist failed with exception, ignoring", e);
1282 }
1283 }
1284 }
1285
1286 public void setRackManager(RackManager rackManager) {
1287 this.rackManager = rackManager;
1288 }
1289
1290 protected boolean needsBalance(Cluster c) {
1291 ClusterLoadState cs = new ClusterLoadState(c.clusterState);
1292 if (cs.getNumServers() < MIN_SERVER_BALANCE) {
1293 if (LOG.isDebugEnabled()) {
1294 LOG.debug("Not running balancer because only " + cs.getNumServers()
1295 + " active regionserver(s)");
1296 }
1297 return false;
1298 }
1299 if(areSomeRegionReplicasColocated(c)) return true;
1300
1301
1302 float average = cs.getLoadAverage();
1303 int floor = (int) Math.floor(average * (1 - slop));
1304 int ceiling = (int) Math.ceil(average * (1 + slop));
1305 if (!(cs.getMaxLoad() > ceiling || cs.getMinLoad() < floor)) {
1306 NavigableMap<ServerAndLoad, List<HRegionInfo>> serversByLoad = cs.getServersByLoad();
1307 if (LOG.isTraceEnabled()) {
1308
1309 LOG.trace("Skipping load balancing because balanced cluster; " +
1310 "servers=" + cs.getNumServers() +
1311 " regions=" + cs.getNumRegions() + " average=" + average +
1312 " mostloaded=" + serversByLoad.lastKey().getLoad() +
1313 " leastloaded=" + serversByLoad.firstKey().getLoad());
1314 }
1315 return false;
1316 }
1317 return true;
1318 }
1319
1320
1321
1322
1323
1324
1325
1326
1327 protected boolean areSomeRegionReplicasColocated(Cluster c) {
1328 return false;
1329 }
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348 @Override
1349 public Map<ServerName, List<HRegionInfo>> roundRobinAssignment(List<HRegionInfo> regions,
1350 List<ServerName> servers) {
1351 metricsBalancer.incrMiscInvocations();
1352 Map<ServerName, List<HRegionInfo>> assignments = assignMasterRegions(regions, servers);
1353 if (assignments != null && !assignments.isEmpty()) {
1354 servers = new ArrayList<ServerName>(servers);
1355
1356 servers.remove(masterServerName);
1357 List<HRegionInfo> masterRegions = assignments.get(masterServerName);
1358 if (!masterRegions.isEmpty()) {
1359 regions = new ArrayList<HRegionInfo>(regions);
1360 for (HRegionInfo region: masterRegions) {
1361 regions.remove(region);
1362 }
1363 }
1364 }
1365 if (regions == null || regions.isEmpty()) {
1366 return assignments;
1367 }
1368
1369 int numServers = servers == null ? 0 : servers.size();
1370 if (numServers == 0) {
1371 LOG.warn("Wanted to do round robin assignment but no servers to assign to");
1372 return null;
1373 }
1374
1375
1376
1377
1378
1379
1380 if (numServers == 1) {
1381 ServerName server = servers.get(0);
1382 assignments.put(server, new ArrayList<HRegionInfo>(regions));
1383 return assignments;
1384 }
1385
1386 Cluster cluster = createCluster(servers, regions);
1387 List<HRegionInfo> unassignedRegions = new ArrayList<HRegionInfo>();
1388
1389 roundRobinAssignment(cluster, regions, unassignedRegions,
1390 servers, assignments);
1391
1392 List<HRegionInfo> lastFewRegions = new ArrayList<HRegionInfo>();
1393
1394 int serverIdx = RANDOM.nextInt(numServers);
1395 for (HRegionInfo region : unassignedRegions) {
1396 boolean assigned = false;
1397 for (int j = 0; j < numServers; j++) {
1398 ServerName serverName = servers.get((j + serverIdx) % numServers);
1399 if (!cluster.wouldLowerAvailability(region, serverName)) {
1400 List<HRegionInfo> serverRegions = assignments.get(serverName);
1401 if (serverRegions == null) {
1402 serverRegions = new ArrayList<HRegionInfo>();
1403 assignments.put(serverName, serverRegions);
1404 }
1405 serverRegions.add(region);
1406 cluster.doAssignRegion(region, serverName);
1407 serverIdx = (j + serverIdx + 1) % numServers;
1408 assigned = true;
1409 break;
1410 }
1411 }
1412 if (!assigned) {
1413 lastFewRegions.add(region);
1414 }
1415 }
1416
1417
1418 for (HRegionInfo region : lastFewRegions) {
1419 int i = RANDOM.nextInt(numServers);
1420 ServerName server = servers.get(i);
1421 List<HRegionInfo> serverRegions = assignments.get(server);
1422 if (serverRegions == null) {
1423 serverRegions = new ArrayList<HRegionInfo>();
1424 assignments.put(server, serverRegions);
1425 }
1426 serverRegions.add(region);
1427 cluster.doAssignRegion(region, server);
1428 }
1429 return assignments;
1430 }
1431
1432 protected Cluster createCluster(List<ServerName> servers, Collection<HRegionInfo> regions) {
1433
1434
1435
1436
1437 Map<ServerName, List<HRegionInfo>> clusterState = getRegionAssignmentsByServer(regions);
1438
1439 for (ServerName server : servers) {
1440 if (!clusterState.containsKey(server)) {
1441 clusterState.put(server, EMPTY_REGION_LIST);
1442 }
1443 }
1444 return new Cluster(regions, clusterState, null, this.regionFinder,
1445 rackManager);
1446 }
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465 @Override
1466 public Map<HRegionInfo, ServerName> immediateAssignment(List<HRegionInfo> regions,
1467 List<ServerName> servers) {
1468 metricsBalancer.incrMiscInvocations();
1469 if (servers == null || servers.isEmpty()) {
1470 LOG.warn("Wanted to do random assignment but no servers to assign to");
1471 return null;
1472 }
1473
1474 Map<HRegionInfo, ServerName> assignments = new TreeMap<HRegionInfo, ServerName>();
1475 for (HRegionInfo region : regions) {
1476 assignments.put(region, randomAssignment(region, servers));
1477 }
1478 return assignments;
1479 }
1480
1481
1482
1483
1484 @Override
1485 public ServerName randomAssignment(HRegionInfo regionInfo, List<ServerName> servers) {
1486 metricsBalancer.incrMiscInvocations();
1487 if (servers != null && servers.contains(masterServerName)) {
1488 if (shouldBeOnMaster(regionInfo)) {
1489 return masterServerName;
1490 }
1491 servers = new ArrayList<ServerName>(servers);
1492
1493 servers.remove(masterServerName);
1494 }
1495
1496 int numServers = servers == null ? 0 : servers.size();
1497 if (numServers == 0) {
1498 LOG.warn("Wanted to do retain assignment but no servers to assign to");
1499 return null;
1500 }
1501 if (numServers == 1) {
1502 return servers.get(0);
1503 }
1504
1505 List<HRegionInfo> regions = Lists.newArrayList(regionInfo);
1506 Cluster cluster = createCluster(servers, regions);
1507 return randomAssignment(cluster, regionInfo, servers);
1508 }
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527 @Override
1528 public Map<ServerName, List<HRegionInfo>> retainAssignment(Map<HRegionInfo, ServerName> regions,
1529 List<ServerName> servers) {
1530
1531 metricsBalancer.incrMiscInvocations();
1532 Map<ServerName, List<HRegionInfo>> assignments
1533 = assignMasterRegions(regions.keySet(), servers);
1534 if (assignments != null && !assignments.isEmpty()) {
1535 servers = new ArrayList<ServerName>(servers);
1536
1537 servers.remove(masterServerName);
1538 List<HRegionInfo> masterRegions = assignments.get(masterServerName);
1539 if (!masterRegions.isEmpty()) {
1540 regions = new HashMap<HRegionInfo, ServerName>(regions);
1541 for (HRegionInfo region: masterRegions) {
1542 regions.remove(region);
1543 }
1544 }
1545 }
1546 if (regions == null || regions.isEmpty()) {
1547 return assignments;
1548 }
1549
1550 int numServers = servers == null ? 0 : servers.size();
1551 if (numServers == 0) {
1552 LOG.warn("Wanted to do retain assignment but no servers to assign to");
1553 return null;
1554 }
1555 if (numServers == 1) {
1556 ServerName server = servers.get(0);
1557 assignments.put(server, new ArrayList<HRegionInfo>(regions.keySet()));
1558 return assignments;
1559 }
1560
1561
1562
1563
1564
1565
1566
1567 ArrayListMultimap<String, ServerName> serversByHostname = ArrayListMultimap.create();
1568 for (ServerName server : servers) {
1569 assignments.put(server, new ArrayList<HRegionInfo>());
1570 serversByHostname.put(server.getHostname(), server);
1571 }
1572
1573
1574
1575
1576 Set<String> oldHostsNoLongerPresent = Sets.newTreeSet();
1577
1578
1579 List<HRegionInfo> randomAssignRegions = Lists.newArrayList();
1580
1581 int numRandomAssignments = 0;
1582 int numRetainedAssigments = 0;
1583
1584 for (Map.Entry<HRegionInfo, ServerName> entry : regions.entrySet()) {
1585 HRegionInfo region = entry.getKey();
1586 ServerName oldServerName = entry.getValue();
1587 List<ServerName> localServers = new ArrayList<ServerName>();
1588 if (oldServerName != null) {
1589 localServers = serversByHostname.get(oldServerName.getHostname());
1590 }
1591 if (localServers.isEmpty()) {
1592
1593 randomAssignRegions.add(region);
1594 if (oldServerName != null) {
1595 oldHostsNoLongerPresent.add(oldServerName.getHostname());
1596 }
1597 } else if (localServers.size() == 1) {
1598
1599 ServerName target = localServers.get(0);
1600 assignments.get(target).add(region);
1601 numRetainedAssigments++;
1602 } else {
1603
1604 if (localServers.contains(oldServerName)) {
1605 assignments.get(oldServerName).add(region);
1606 numRetainedAssigments++;
1607 } else {
1608 ServerName target = null;
1609 for (ServerName tmp : localServers) {
1610 if (tmp.getPort() == oldServerName.getPort()) {
1611 target = tmp;
1612 assignments.get(tmp).add(region);
1613 numRetainedAssigments++;
1614 break;
1615 }
1616 }
1617 if (target == null) {
1618 randomAssignRegions.add(region);
1619 }
1620 }
1621 }
1622 }
1623
1624
1625 if (randomAssignRegions.size() > 0) {
1626 Cluster cluster = createCluster(servers, regions.keySet());
1627 for (Map.Entry<ServerName, List<HRegionInfo>> entry : assignments.entrySet()) {
1628 ServerName sn = entry.getKey();
1629 for (HRegionInfo region : entry.getValue()) {
1630 cluster.doAssignRegion(region, sn);
1631 }
1632 }
1633 for (HRegionInfo region : randomAssignRegions) {
1634 ServerName target = randomAssignment(cluster, region, servers);
1635 assignments.get(target).add(region);
1636 cluster.doAssignRegion(region, target);
1637 numRandomAssignments++;
1638 }
1639 }
1640
1641 String randomAssignMsg = "";
1642 if (numRandomAssignments > 0) {
1643 randomAssignMsg =
1644 numRandomAssignments + " regions were assigned "
1645 + "to random hosts, since the old hosts for these regions are no "
1646 + "longer present in the cluster. These hosts were:\n "
1647 + Joiner.on("\n ").join(oldHostsNoLongerPresent);
1648 }
1649
1650 LOG.info("Reassigned " + regions.size() + " regions. " + numRetainedAssigments
1651 + " retained the pre-restart assignment. " + randomAssignMsg);
1652 return assignments;
1653 }
1654
1655 @Override
1656 public void initialize() throws HBaseIOException{
1657 }
1658
1659 @Override
1660 public void regionOnline(HRegionInfo regionInfo, ServerName sn) {
1661 }
1662
1663 @Override
1664 public void regionOffline(HRegionInfo regionInfo) {
1665 }
1666
1667 @Override
1668 public boolean isStopped() {
1669 return stopped;
1670 }
1671
1672 @Override
1673 public void stop(String why) {
1674 LOG.info("Load Balancer stop requested: "+why);
1675 stopped = true;
1676 }
1677
1678
1679
1680
1681 public void updateBalancerStatus(boolean status) {
1682 metricsBalancer.balancerStatus(status);
1683 }
1684
1685
1686
1687
1688 private ServerName randomAssignment(Cluster cluster, HRegionInfo regionInfo,
1689 List<ServerName> servers) {
1690 int numServers = servers.size();
1691 ServerName sn = null;
1692 final int maxIterations = numServers * 4;
1693 int iterations = 0;
1694
1695 do {
1696 int i = RANDOM.nextInt(numServers);
1697 sn = servers.get(i);
1698 } while (cluster.wouldLowerAvailability(regionInfo, sn)
1699 && iterations++ < maxIterations);
1700 cluster.doAssignRegion(regionInfo, sn);
1701 return sn;
1702 }
1703
1704
1705
1706
1707 private void roundRobinAssignment(Cluster cluster, List<HRegionInfo> regions,
1708 List<HRegionInfo> unassignedRegions, List<ServerName> servers,
1709 Map<ServerName, List<HRegionInfo>> assignments) {
1710
1711 int numServers = servers.size();
1712 int numRegions = regions.size();
1713 int max = (int) Math.ceil((float) numRegions / numServers);
1714 int serverIdx = 0;
1715 if (numServers > 1) {
1716 serverIdx = RANDOM.nextInt(numServers);
1717 }
1718 int regionIdx = 0;
1719
1720 for (int j = 0; j < numServers; j++) {
1721 ServerName server = servers.get((j + serverIdx) % numServers);
1722 List<HRegionInfo> serverRegions = new ArrayList<HRegionInfo>(max);
1723 for (int i = regionIdx; i < numRegions; i += numServers) {
1724 HRegionInfo region = regions.get(i % numRegions);
1725 if (cluster.wouldLowerAvailability(region, server)) {
1726 unassignedRegions.add(region);
1727 } else {
1728 serverRegions.add(region);
1729 cluster.doAssignRegion(region, server);
1730 }
1731 }
1732 assignments.put(server, serverRegions);
1733 regionIdx++;
1734 }
1735 }
1736
1737 protected Map<ServerName, List<HRegionInfo>> getRegionAssignmentsByServer(
1738 Collection<HRegionInfo> regions) {
1739 if (this.services != null && this.services.getAssignmentManager() != null) {
1740 return this.services.getAssignmentManager().getSnapShotOfAssignment(regions);
1741 } else {
1742 return new HashMap<ServerName, List<HRegionInfo>>();
1743 }
1744 }
1745
1746 @Override
1747 public void onConfigurationChange(Configuration conf) {
1748 }
1749 }