1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21 package org.apache.hadoop.hbase.rsgroup;
22
23 import com.google.common.annotations.VisibleForTesting;
24 import com.google.common.collect.ArrayListMultimap;
25 import com.google.common.collect.LinkedListMultimap;
26 import com.google.common.collect.ListMultimap;
27 import com.google.common.collect.Lists;
28 import com.google.common.collect.Maps;
29
30 import java.io.IOException;
31 import java.util.ArrayList;
32 import java.util.Collections;
33 import java.util.HashMap;
34 import java.util.HashSet;
35 import java.util.LinkedList;
36 import java.util.List;
37 import java.util.Map;
38 import java.util.Set;
39 import java.util.TreeMap;
40
41 import org.apache.commons.logging.Log;
42 import org.apache.commons.logging.LogFactory;
43 import org.apache.hadoop.conf.Configuration;
44 import org.apache.hadoop.hbase.ClusterStatus;
45 import org.apache.hadoop.hbase.HBaseIOException;
46 import org.apache.hadoop.hbase.HRegionInfo;
47 import org.apache.hadoop.hbase.ServerName;
48 import org.apache.hadoop.hbase.TableName;
49 import org.apache.hadoop.hbase.classification.InterfaceAudience;
50 import org.apache.hadoop.hbase.constraint.ConstraintException;
51 import org.apache.hadoop.hbase.master.LoadBalancer;
52 import org.apache.hadoop.hbase.master.MasterServices;
53 import org.apache.hadoop.hbase.master.RegionPlan;
54 import org.apache.hadoop.hbase.master.balancer.StochasticLoadBalancer;
55 import org.apache.hadoop.hbase.net.Address;
56 import org.apache.hadoop.util.ReflectionUtils;
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72 @InterfaceAudience.Private
73 public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalancer {
74
75 public static final String HBASE_GROUP_LOADBALANCER_CLASS = "hbase.group.grouploadbalancer.class";
76
77 private static final Log LOG = LogFactory.getLog(RSGroupBasedLoadBalancer.class);
78
79 private Configuration config;
80 private ClusterStatus clusterStatus;
81 private MasterServices masterServices;
82 private RSGroupInfoManager infoManager;
83 private LoadBalancer internalBalancer;
84
85
86 @InterfaceAudience.Private
87 public RSGroupBasedLoadBalancer() {
88 }
89
90
91 @InterfaceAudience.Private
92 public RSGroupBasedLoadBalancer(RSGroupInfoManager RSGroupInfoManager) {
93 this.infoManager = RSGroupInfoManager;
94 }
95
96 @Override
97 public Configuration getConf() {
98 return config;
99 }
100
101 @Override
102 public void setConf(Configuration conf) {
103 this.config = conf;
104 if (internalBalancer != null) {
105 internalBalancer.setConf(conf);
106 }
107 }
108
109 @Override
110 public void setClusterStatus(ClusterStatus st) {
111 this.clusterStatus = st;
112 if (internalBalancer != null) {
113 internalBalancer.setClusterStatus(st);
114 }
115 }
116
117 @Override
118 public void setMasterServices(MasterServices masterServices) {
119 this.masterServices = masterServices;
120 }
121
122 @Override
123 public List<RegionPlan> balanceCluster(TableName tableName, Map<ServerName, List<HRegionInfo>>
124 clusterState) throws HBaseIOException {
125 return balanceCluster(clusterState);
126 }
127
128 @Override
129 public List<RegionPlan> balanceCluster(Map<ServerName, List<HRegionInfo>> clusterState)
130 throws HBaseIOException {
131 if (!isOnline()) {
132 throw new ConstraintException(RSGroupInfoManager.RSGROUP_TABLE_NAME +
133 " is not online, unable to perform balance");
134 }
135
136 Map<ServerName,List<HRegionInfo>> correctedState = correctAssignments(clusterState);
137 List<RegionPlan> regionPlans = new ArrayList<RegionPlan>();
138
139 List<HRegionInfo> misplacedRegions = correctedState.get(LoadBalancer.BOGUS_SERVER_NAME);
140 for (HRegionInfo regionInfo : misplacedRegions) {
141 regionPlans.add(new RegionPlan(regionInfo, null, null));
142 }
143 try {
144
145 HashSet<ServerName> processedServers = new HashSet<>();
146
147
148 for (RSGroupInfo rsgroup : infoManager.listRSGroups()) {
149 Map<ServerName, List<HRegionInfo>> groupClusterState = new HashMap<>();
150 for (ServerName server : clusterState.keySet()) {
151 if (!processedServers.contains(server)
152 && rsgroup.containsServer(server.getAddress())) {
153 List<HRegionInfo> regionsOnServer = correctedState.get(server);
154 groupClusterState.put(server, regionsOnServer);
155 processedServers.add(server);
156 }
157 }
158
159 List<RegionPlan> groupPlans = this.internalBalancer
160 .balanceCluster(groupClusterState);
161 if (groupPlans != null) {
162 regionPlans.addAll(groupPlans);
163 }
164 }
165 } catch (IOException exp) {
166 LOG.warn("Exception while balancing cluster.", exp);
167 regionPlans.clear();
168 }
169 return regionPlans;
170 }
171
172 @Override
173 public Map<ServerName, List<HRegionInfo>> roundRobinAssignment(
174 List<HRegionInfo> regions, List<ServerName> servers) throws HBaseIOException {
175 Map<ServerName, List<HRegionInfo>> assignments = Maps.newHashMap();
176 ListMultimap<String,HRegionInfo> regionMap = ArrayListMultimap.create();
177 ListMultimap<String,ServerName> serverMap = ArrayListMultimap.create();
178 generateGroupMaps(regions, servers, regionMap, serverMap);
179 for(String groupKey : regionMap.keySet()) {
180 if (regionMap.get(groupKey).size() > 0) {
181 Map<ServerName, List<HRegionInfo>> result =
182 this.internalBalancer.roundRobinAssignment(
183 regionMap.get(groupKey),
184 serverMap.get(groupKey));
185 if(result != null) {
186 if(result.containsKey(LoadBalancer.BOGUS_SERVER_NAME) &&
187 assignments.containsKey(LoadBalancer.BOGUS_SERVER_NAME)){
188 assignments.get(LoadBalancer.BOGUS_SERVER_NAME).addAll(
189 result.get(LoadBalancer.BOGUS_SERVER_NAME));
190 } else {
191 assignments.putAll(result);
192 }
193 }
194 }
195 }
196 return assignments;
197 }
198
199 @Override
200 public Map<ServerName, List<HRegionInfo>> retainAssignment(
201 Map<HRegionInfo, ServerName> regions, List<ServerName> servers) throws HBaseIOException {
202 try {
203 Map<ServerName, List<HRegionInfo>> assignments = new TreeMap<ServerName, List<HRegionInfo>>();
204 ListMultimap<String, HRegionInfo> groupToRegion = ArrayListMultimap.create();
205 Set<HRegionInfo> misplacedRegions = getMisplacedRegions(regions);
206 for (HRegionInfo region : regions.keySet()) {
207 if (!misplacedRegions.contains(region)) {
208 String groupName = infoManager.getRSGroupOfTable(region.getTable());
209 if (groupName == null) {
210 LOG.info("Group not found for table " + region.getTable() + ", using default");
211 groupName = RSGroupInfo.DEFAULT_GROUP;
212 }
213 groupToRegion.put(groupName, region);
214 }
215 }
216
217
218 for (String key : groupToRegion.keySet()) {
219 Map<HRegionInfo, ServerName> currentAssignmentMap = new TreeMap<HRegionInfo, ServerName>();
220 List<HRegionInfo> regionList = groupToRegion.get(key);
221 RSGroupInfo info = infoManager.getRSGroup(key);
222 List<ServerName> candidateList = filterOfflineServers(info, servers);
223 for (HRegionInfo region : regionList) {
224 currentAssignmentMap.put(region, regions.get(region));
225 }
226 if(candidateList.size() > 0) {
227 assignments.putAll(this.internalBalancer.retainAssignment(
228 currentAssignmentMap, candidateList));
229 }
230 }
231
232 for (HRegionInfo region : misplacedRegions) {
233 String groupName = infoManager.getRSGroupOfTable(region.getTable());
234 if (groupName == null) {
235 LOG.info("Group not found for table " + region.getTable() + ", using default");
236 groupName = RSGroupInfo.DEFAULT_GROUP;
237 }
238 RSGroupInfo info = infoManager.getRSGroup(groupName);
239 List<ServerName> candidateList = filterOfflineServers(info, servers);
240 ServerName server = this.internalBalancer.randomAssignment(region,
241 candidateList);
242 if (server != null) {
243 if (!assignments.containsKey(server)) {
244 assignments.put(server, new ArrayList<HRegionInfo>());
245 }
246 assignments.get(server).add(region);
247 } else {
248
249 if(!assignments.containsKey(LoadBalancer.BOGUS_SERVER_NAME)) {
250 assignments.put(LoadBalancer.BOGUS_SERVER_NAME, new ArrayList<HRegionInfo>());
251 }
252 assignments.get(LoadBalancer.BOGUS_SERVER_NAME).add(region);
253 }
254 }
255 return assignments;
256 } catch (IOException e) {
257 throw new HBaseIOException("Failed to do online retain assignment", e);
258 }
259 }
260
261 @Override
262 public Map<HRegionInfo, ServerName> immediateAssignment(List<HRegionInfo> regions,
263 List<ServerName> servers) throws HBaseIOException {
264 throw new UnsupportedOperationException("immediateAssignment is not supported");
265 }
266
267 @Override
268 public ServerName randomAssignment(HRegionInfo region,
269 List<ServerName> servers) throws HBaseIOException {
270 ListMultimap<String,HRegionInfo> regionMap = LinkedListMultimap.create();
271 ListMultimap<String,ServerName> serverMap = LinkedListMultimap.create();
272 generateGroupMaps(Lists.newArrayList(region), servers, regionMap, serverMap);
273 List<ServerName> filteredServers = serverMap.get(regionMap.keySet().iterator().next());
274 return this.internalBalancer.randomAssignment(region, filteredServers);
275 }
276
277 private void generateGroupMaps(
278 List<HRegionInfo> regions,
279 List<ServerName> servers,
280 ListMultimap<String, HRegionInfo> regionMap,
281 ListMultimap<String, ServerName> serverMap) throws HBaseIOException {
282 try {
283 for (HRegionInfo region : regions) {
284 String groupName = infoManager.getRSGroupOfTable(region.getTable());
285 if (groupName == null) {
286 LOG.info("Group not found for table " + region.getTable() + ", using default");
287 groupName = RSGroupInfo.DEFAULT_GROUP;
288 }
289 regionMap.put(groupName, region);
290 }
291 for (String groupKey : regionMap.keySet()) {
292 RSGroupInfo info = infoManager.getRSGroup(groupKey);
293 serverMap.putAll(groupKey, filterOfflineServers(info, servers));
294 if(serverMap.get(groupKey).size() < 1) {
295 serverMap.put(groupKey, LoadBalancer.BOGUS_SERVER_NAME);
296 }
297 }
298 } catch(IOException e) {
299 throw new HBaseIOException("Failed to generate group maps", e);
300 }
301 }
302
303 private List<ServerName> filterOfflineServers(RSGroupInfo RSGroupInfo,
304 List<ServerName> onlineServers) {
305 if (RSGroupInfo != null) {
306 return filterServers(RSGroupInfo.getServers(), onlineServers);
307 } else {
308 LOG.debug("Group Information found to be null. Some regions might be unassigned.");
309 return Collections.emptyList();
310 }
311 }
312
313
314
315
316
317
318
319
320
321
322 private List<ServerName> filterServers(Set<Address> servers,
323 List<ServerName> onlineServers) {
324
325
326
327
328
329
330 ArrayList<ServerName> finalList = new ArrayList<>();
331 for (ServerName onlineServer : onlineServers) {
332 if (servers.contains(onlineServer.getAddress())) {
333 finalList.add(onlineServer);
334 }
335 }
336
337 return finalList;
338 }
339
340 @VisibleForTesting
341 public Set<HRegionInfo> getMisplacedRegions(
342 Map<HRegionInfo, ServerName> regions) throws IOException {
343 Set<HRegionInfo> misplacedRegions = new HashSet<HRegionInfo>();
344 for(Map.Entry<HRegionInfo, ServerName> region : regions.entrySet()) {
345 HRegionInfo regionInfo = region.getKey();
346 ServerName assignedServer = region.getValue();
347 String groupName = infoManager.getRSGroupOfTable(regionInfo.getTable());
348 if (groupName == null) {
349 LOG.info("Group not found for table " + regionInfo.getTable() + ", using default");
350 groupName = RSGroupInfo.DEFAULT_GROUP;
351 }
352 RSGroupInfo info = infoManager.getRSGroup(groupName);
353 if (assignedServer == null) {
354 LOG.debug("There is no assigned server for " + region);
355 continue;
356 }
357 RSGroupInfo otherInfo = infoManager.getRSGroupOfServer(assignedServer.getAddress());
358 if (info == null && otherInfo == null) {
359 LOG.warn("Couldn't obtain rs group information for " + region + " on " + assignedServer);
360 continue;
361 }
362 if ((info == null || !info.containsServer(assignedServer.getAddress()))) {
363 LOG.debug("Found misplaced region: " + regionInfo.getRegionNameAsString() +
364 " on server: " + assignedServer +
365 " found in group: " + otherInfo +
366 " outside of group: " + (info == null ? "UNKNOWN" : info.getName()));
367 misplacedRegions.add(regionInfo);
368 }
369 }
370 return misplacedRegions;
371 }
372
373 private Map<ServerName, List<HRegionInfo>> correctAssignments(
374 Map<ServerName, List<HRegionInfo>> existingAssignments) {
375 Map<ServerName, List<HRegionInfo>> correctAssignments =
376 new TreeMap<ServerName, List<HRegionInfo>>();
377 correctAssignments.put(LoadBalancer.BOGUS_SERVER_NAME, new LinkedList<HRegionInfo>());
378 for (Map.Entry<ServerName, List<HRegionInfo>> assignments : existingAssignments.entrySet()){
379 ServerName sName = assignments.getKey();
380 correctAssignments.put(sName, new LinkedList<HRegionInfo>());
381 List<HRegionInfo> regions = assignments.getValue();
382 for (HRegionInfo region : regions) {
383 RSGroupInfo info = null;
384 try {
385 String groupName = infoManager.getRSGroupOfTable(region.getTable());
386 if (groupName == null) {
387 LOG.info("Group not found for table " + region.getTable() + ", using default");
388 groupName = RSGroupInfo.DEFAULT_GROUP;
389 }
390 info = infoManager.getRSGroup(groupName);
391 } catch (IOException exp) {
392 LOG.debug("Group information null for region of table " + region.getTable(),
393 exp);
394 }
395 if ((info == null) || (!info.containsServer(sName.getAddress()))) {
396 correctAssignments.get(LoadBalancer.BOGUS_SERVER_NAME).add(region);
397 } else {
398 correctAssignments.get(sName).add(region);
399 }
400 }
401 }
402 return correctAssignments;
403 }
404
405 @Override
406 public void initialize() throws HBaseIOException {
407 try {
408 if (infoManager == null) {
409 List<RSGroupAdminEndpoint> cps =
410 masterServices.getMasterCoprocessorHost().findCoprocessors(RSGroupAdminEndpoint.class);
411 if (cps.size() != 1) {
412 String msg = "Expected one implementation of GroupAdminEndpoint but found " + cps.size();
413 LOG.error(msg);
414 throw new HBaseIOException(msg);
415 }
416 infoManager = cps.get(0).getGroupInfoManager();
417 if(infoManager == null){
418 String msg = "RSGroupInfoManager hasn't been initialized";
419 LOG.error(msg);
420 throw new HBaseIOException(msg);
421 }
422 infoManager.start();
423 }
424 } catch (IOException e) {
425 throw new HBaseIOException("Failed to initialize GroupInfoManagerImpl", e);
426 }
427
428
429 Class<? extends LoadBalancer> balancerKlass = config.getClass(
430 HBASE_GROUP_LOADBALANCER_CLASS,
431 StochasticLoadBalancer.class, LoadBalancer.class);
432 internalBalancer = ReflectionUtils.newInstance(balancerKlass, config);
433 if (clusterStatus != null) {
434 internalBalancer.setClusterStatus(clusterStatus);
435 }
436 internalBalancer.setMasterServices(masterServices);
437 internalBalancer.setConf(config);
438 internalBalancer.initialize();
439 }
440
441 public boolean isOnline() {
442 return infoManager != null && infoManager.isOnline();
443 }
444
445 @Override
446 public void regionOnline(HRegionInfo regionInfo, ServerName sn) {
447 }
448
449 @Override
450 public void regionOffline(HRegionInfo regionInfo) {
451 }
452
453 @Override
454 public void onConfigurationChange(Configuration conf) {
455
456 }
457
458 @Override
459 public void stop(String why) {
460 }
461
462 @Override
463 public boolean isStopped() {
464 return false;
465 }
466
467 @Override
468 public void postMasterStartupInitialize() {
469 this.internalBalancer.postMasterStartupInitialize();
470 }
471
472 public void updateBalancerStatus(boolean status) {
473 internalBalancer.updateBalancerStatus(status);
474 }
475 }