View Javadoc

1   /**
2    * Copyright The Apache Software Foundation
3    *
4    * Licensed to the Apache Software Foundation (ASF) under one
5    * or more contributor license agreements.  See the NOTICE file
6    * distributed with this work for additional information
7    * regarding copyright ownership.  The ASF licenses this file
8    * to you under the Apache License, Version 2.0 (the
9    * "License"); you may not use this file except in compliance
10   * with the License.  You may obtain a copy of the License at
11   *
12   *     http://www.apache.org/licenses/LICENSE-2.0
13   *
14   * Unless required by applicable law or agreed to in writing, software
15   * distributed under the License is distributed on an "AS IS" BASIS,
16   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17   * See the License for the specific language governing permissions and
18   * limitations under the License.
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   * GroupBasedLoadBalancer, used when Region Server Grouping is configured (HBase-6721)
60   * It does region balance based on a table's group membership.
61   *
62   * Most assignment methods contain two exclusive code paths: Online - when the group
63   * table is online and Offline - when it is unavailable.
64   *
65   * During Offline, assignments are assigned based on cached information in zookeeper.
66   * If unavailable (ie bootstrap) then regions are assigned randomly.
67   *
68   * Once the GROUP table has been assigned, the balancer switches to Online and will then
69   * start providing appropriate assignments for user tables.
70   *
71   */
72  @InterfaceAudience.Private
73  public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalancer {
74    /** Config for pluggable load balancers */
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    //used during reflection by LoadBalancerFactory
86    @InterfaceAudience.Private
87    public RSGroupBasedLoadBalancer() {
88    }
89  
90    //This constructor should only be used for unit testing
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       // Record which region servers have been processed,so as to skip them after processed
145       HashSet<ServerName> processedServers = new HashSet<>();
146 
147       // For each rsgroup
148       for (RSGroupInfo rsgroup : infoManager.listRSGroups()) {
149         Map<ServerName, List<HRegionInfo>> groupClusterState = new HashMap<>();
150         for (ServerName server : clusterState.keySet()) { // for each region server
151           if (!processedServers.contains(server) // server is not processed yet
152               && rsgroup.containsServer(server.getAddress())) { // server belongs to this rsgroup
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       // Now the "groupToRegion" map has only the regions which have correct
217       // assignments.
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           //if not server is available assign to bogus so it ends up in RIT
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    * Filter servers based on the online servers.
315    *
316    * @param servers
317    *          the servers
318    * @param onlineServers
319    *          List of servers which are online.
320    * @return the list
321    */
322   private List<ServerName> filterServers(Set<Address> servers,
323       List<ServerName> onlineServers) {
324     /**
325      * servers is actually a TreeSet (see {@link org.apache.hadoop.hbase.rsgroup.RSGroupInfo}),
326      * having its contains()'s time complexity as O(logn), which is good enough.
327      * TODO: consider using HashSet to pursue O(1) for contains() throughout the calling chain
328      * if needed.
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     // Create the balancer
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     //DO nothing for now
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 }