001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018
019package org.apache.hadoop.hbase.rsgroup;
020
021import java.io.IOException;
022import java.util.ArrayList;
023import java.util.Collections;
024import java.util.HashMap;
025import java.util.HashSet;
026import java.util.LinkedList;
027import java.util.List;
028import java.util.Map;
029import java.util.Set;
030import java.util.TreeMap;
031
032import org.apache.hadoop.conf.Configuration;
033import org.apache.hadoop.hbase.ClusterMetrics;
034import org.apache.hadoop.hbase.HBaseIOException;
035import org.apache.hadoop.hbase.HConstants;
036import org.apache.hadoop.hbase.ServerName;
037import org.apache.hadoop.hbase.TableName;
038import org.apache.hadoop.hbase.client.RegionInfo;
039import org.apache.hadoop.hbase.constraint.ConstraintException;
040import org.apache.hadoop.hbase.master.LoadBalancer;
041import org.apache.hadoop.hbase.master.MasterServices;
042import org.apache.hadoop.hbase.master.RegionPlan;
043import org.apache.hadoop.hbase.master.balancer.StochasticLoadBalancer;
044import org.apache.hadoop.hbase.net.Address;
045import org.apache.hadoop.util.ReflectionUtils;
046import org.apache.yetus.audience.InterfaceAudience;
047import org.slf4j.Logger;
048import org.slf4j.LoggerFactory;
049
050import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
051import org.apache.hbase.thirdparty.com.google.common.collect.ArrayListMultimap;
052import org.apache.hbase.thirdparty.com.google.common.collect.LinkedListMultimap;
053import org.apache.hbase.thirdparty.com.google.common.collect.ListMultimap;
054import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
055import org.apache.hbase.thirdparty.com.google.common.collect.Maps;
056
057/**
058 * GroupBasedLoadBalancer, used when Region Server Grouping is configured (HBase-6721)
059 * It does region balance based on a table's group membership.
060 *
061 * Most assignment methods contain two exclusive code paths: Online - when the group
062 * table is online and Offline - when it is unavailable.
063 *
064 * During Offline, assignments are assigned based on cached information in zookeeper.
065 * If unavailable (ie bootstrap) then regions are assigned randomly.
066 *
067 * Once the GROUP table has been assigned, the balancer switches to Online and will then
068 * start providing appropriate assignments for user tables.
069 *
070 */
071@InterfaceAudience.Private
072public class RSGroupBasedLoadBalancer implements RSGroupableBalancer {
073  private static final Logger LOG = LoggerFactory.getLogger(RSGroupBasedLoadBalancer.class);
074
075  private Configuration config;
076  private ClusterMetrics clusterStatus;
077  private MasterServices masterServices;
078  private volatile RSGroupInfoManager rsGroupInfoManager;
079  private LoadBalancer internalBalancer;
080
081  /**
082   * Used by reflection in {@link org.apache.hadoop.hbase.master.balancer.LoadBalancerFactory}.
083   */
084  @InterfaceAudience.Private
085  public RSGroupBasedLoadBalancer() {}
086
087  @Override
088  public Configuration getConf() {
089    return config;
090  }
091
092  @Override
093  public void setConf(Configuration conf) {
094    this.config = conf;
095  }
096
097  @Override
098  public void setClusterMetrics(ClusterMetrics sm) {
099    this.clusterStatus = sm;
100  }
101
102  @Override
103  public void setMasterServices(MasterServices masterServices) {
104    this.masterServices = masterServices;
105  }
106
107  @Override
108  public List<RegionPlan> balanceCluster(TableName tableName, Map<ServerName, List<RegionInfo>>
109      clusterState) throws HBaseIOException {
110    return balanceCluster(clusterState);
111  }
112
113  @Override
114  public List<RegionPlan> balanceCluster(Map<ServerName, List<RegionInfo>> clusterState)
115      throws HBaseIOException {
116    if (!isOnline()) {
117      throw new ConstraintException(RSGroupInfoManager.RSGROUP_TABLE_NAME +
118          " is not online, unable to perform balance");
119    }
120
121    Map<ServerName,List<RegionInfo>> correctedState = correctAssignments(clusterState);
122    List<RegionPlan> regionPlans = new ArrayList<>();
123
124    List<RegionInfo> misplacedRegions = correctedState.get(LoadBalancer.BOGUS_SERVER_NAME);
125    for (RegionInfo regionInfo : misplacedRegions) {
126      ServerName serverName = findServerForRegion(clusterState, regionInfo);
127      regionPlans.add(new RegionPlan(regionInfo, serverName, null));
128    }
129    try {
130      List<RSGroupInfo> rsgi = rsGroupInfoManager.listRSGroups();
131      for (RSGroupInfo info: rsgi) {
132        Map<ServerName, List<RegionInfo>> groupClusterState = new HashMap<>();
133        Map<TableName, Map<ServerName, List<RegionInfo>>> groupClusterLoad = new HashMap<>();
134        for (Address sName : info.getServers()) {
135          for(ServerName curr: clusterState.keySet()) {
136            if(curr.getAddress().equals(sName)) {
137              groupClusterState.put(curr, correctedState.get(curr));
138            }
139          }
140        }
141        groupClusterLoad.put(HConstants.ENSEMBLE_TABLE_NAME, groupClusterState);
142        this.internalBalancer.setClusterLoad(groupClusterLoad);
143        List<RegionPlan> groupPlans = this.internalBalancer
144            .balanceCluster(groupClusterState);
145        if (groupPlans != null) {
146          regionPlans.addAll(groupPlans);
147        }
148      }
149    } catch (IOException exp) {
150      LOG.warn("Exception while balancing cluster.", exp);
151      regionPlans.clear();
152    }
153    return regionPlans;
154  }
155
156  @Override
157  public Map<ServerName, List<RegionInfo>> roundRobinAssignment(
158      List<RegionInfo> regions, List<ServerName> servers) throws HBaseIOException {
159    Map<ServerName, List<RegionInfo>> assignments = Maps.newHashMap();
160    ListMultimap<String,RegionInfo> regionMap = ArrayListMultimap.create();
161    ListMultimap<String,ServerName> serverMap = ArrayListMultimap.create();
162    generateGroupMaps(regions, servers, regionMap, serverMap);
163    for(String groupKey : regionMap.keySet()) {
164      if (regionMap.get(groupKey).size() > 0) {
165        Map<ServerName, List<RegionInfo>> result =
166            this.internalBalancer.roundRobinAssignment(
167                regionMap.get(groupKey),
168                serverMap.get(groupKey));
169        if(result != null) {
170          if(result.containsKey(LoadBalancer.BOGUS_SERVER_NAME) &&
171              assignments.containsKey(LoadBalancer.BOGUS_SERVER_NAME)){
172            assignments.get(LoadBalancer.BOGUS_SERVER_NAME).addAll(
173              result.get(LoadBalancer.BOGUS_SERVER_NAME));
174          } else {
175            assignments.putAll(result);
176          }
177        }
178      }
179    }
180    return assignments;
181  }
182
183  @Override
184  public Map<ServerName, List<RegionInfo>> retainAssignment(
185      Map<RegionInfo, ServerName> regions, List<ServerName> servers) throws HBaseIOException {
186    try {
187      Map<ServerName, List<RegionInfo>> assignments = new TreeMap<>();
188      ListMultimap<String, RegionInfo> groupToRegion = ArrayListMultimap.create();
189      Set<RegionInfo> misplacedRegions = getMisplacedRegions(regions);
190      for (RegionInfo region : regions.keySet()) {
191        if (!misplacedRegions.contains(region)) {
192          String groupName = rsGroupInfoManager.getRSGroupOfTable(region.getTable());
193          if (groupName == null) {
194            LOG.info("Group not found for table " + region.getTable() + ", using default");
195            groupName = RSGroupInfo.DEFAULT_GROUP;
196          }
197          groupToRegion.put(groupName, region);
198        }
199      }
200      // Now the "groupToRegion" map has only the regions which have correct
201      // assignments.
202      for (String key : groupToRegion.keySet()) {
203        Map<RegionInfo, ServerName> currentAssignmentMap = new TreeMap<RegionInfo, ServerName>();
204        List<RegionInfo> regionList = groupToRegion.get(key);
205        RSGroupInfo info = rsGroupInfoManager.getRSGroup(key);
206        List<ServerName> candidateList = filterOfflineServers(info, servers);
207        for (RegionInfo region : regionList) {
208          currentAssignmentMap.put(region, regions.get(region));
209        }
210        if(candidateList.size() > 0) {
211          assignments.putAll(this.internalBalancer.retainAssignment(
212              currentAssignmentMap, candidateList));
213        }
214      }
215
216      for (RegionInfo region : misplacedRegions) {
217        String groupName = rsGroupInfoManager.getRSGroupOfTable(region.getTable());
218        if (groupName == null) {
219          LOG.info("Group not found for table " + region.getTable() + ", using default");
220          groupName = RSGroupInfo.DEFAULT_GROUP;
221        }
222        RSGroupInfo info = rsGroupInfoManager.getRSGroup(groupName);
223        List<ServerName> candidateList = filterOfflineServers(info, servers);
224        ServerName server = this.internalBalancer.randomAssignment(region,
225            candidateList);
226        if (server != null) {
227          if (!assignments.containsKey(server)) {
228            assignments.put(server, new ArrayList<>());
229          }
230          assignments.get(server).add(region);
231        } else {
232          //if not server is available assign to bogus so it ends up in RIT
233          if(!assignments.containsKey(LoadBalancer.BOGUS_SERVER_NAME)) {
234            assignments.put(LoadBalancer.BOGUS_SERVER_NAME, new ArrayList<>());
235          }
236          assignments.get(LoadBalancer.BOGUS_SERVER_NAME).add(region);
237        }
238      }
239      return assignments;
240    } catch (IOException e) {
241      throw new HBaseIOException("Failed to do online retain assignment", e);
242    }
243  }
244
245  @Override
246  public ServerName randomAssignment(RegionInfo region,
247      List<ServerName> servers) throws HBaseIOException {
248    ListMultimap<String,RegionInfo> regionMap = LinkedListMultimap.create();
249    ListMultimap<String,ServerName> serverMap = LinkedListMultimap.create();
250    generateGroupMaps(Lists.newArrayList(region), servers, regionMap, serverMap);
251    List<ServerName> filteredServers = serverMap.get(regionMap.keySet().iterator().next());
252    return this.internalBalancer.randomAssignment(region, filteredServers);
253  }
254
255  private void generateGroupMaps(
256    List<RegionInfo> regions,
257    List<ServerName> servers,
258    ListMultimap<String, RegionInfo> regionMap,
259    ListMultimap<String, ServerName> serverMap) throws HBaseIOException {
260    try {
261      for (RegionInfo region : regions) {
262        String groupName = rsGroupInfoManager.getRSGroupOfTable(region.getTable());
263        if (groupName == null) {
264          LOG.info("Group not found for table " + region.getTable() + ", using default");
265          groupName = RSGroupInfo.DEFAULT_GROUP;
266        }
267        regionMap.put(groupName, region);
268      }
269      for (String groupKey : regionMap.keySet()) {
270        RSGroupInfo info = rsGroupInfoManager.getRSGroup(groupKey);
271        serverMap.putAll(groupKey, filterOfflineServers(info, servers));
272        if(serverMap.get(groupKey).size() < 1) {
273          serverMap.put(groupKey, LoadBalancer.BOGUS_SERVER_NAME);
274        }
275      }
276    } catch(IOException e) {
277      throw new HBaseIOException("Failed to generate group maps", e);
278    }
279  }
280
281  private List<ServerName> filterOfflineServers(RSGroupInfo RSGroupInfo,
282                                                List<ServerName> onlineServers) {
283    if (RSGroupInfo != null) {
284      return filterServers(RSGroupInfo.getServers(), onlineServers);
285    } else {
286      LOG.warn("RSGroup Information found to be null. Some regions might be unassigned.");
287      return Collections.EMPTY_LIST;
288    }
289  }
290
291  /**
292   * Filter servers based on the online servers.
293   *
294   * @param servers
295   *          the servers
296   * @param onlineServers
297   *          List of servers which are online.
298   * @return the list
299   */
300  private List<ServerName> filterServers(Set<Address> servers,
301                                         List<ServerName> onlineServers) {
302    /**
303     * servers is actually a TreeSet (see {@link org.apache.hadoop.hbase.rsgroup.RSGroupInfo}),
304     * having its contains()'s time complexity as O(logn), which is good enough.
305     * TODO: consider using HashSet to pursue O(1) for contains() throughout the calling chain
306     * if needed. */
307    ArrayList<ServerName> finalList = new ArrayList<>();
308    for (ServerName onlineServer : onlineServers) {
309      if (servers.contains(onlineServer.getAddress())) {
310        finalList.add(onlineServer);
311      }
312    }
313
314    return finalList;
315  }
316
317  @VisibleForTesting
318  public Set<RegionInfo> getMisplacedRegions(
319      Map<RegionInfo, ServerName> regions) throws IOException {
320    Set<RegionInfo> misplacedRegions = new HashSet<>();
321    for(Map.Entry<RegionInfo, ServerName> region : regions.entrySet()) {
322      RegionInfo regionInfo = region.getKey();
323      ServerName assignedServer = region.getValue();
324      String groupName = rsGroupInfoManager.getRSGroupOfTable(regionInfo.getTable());
325      if (groupName == null) {
326        LOG.info("Group not found for table " + regionInfo.getTable() + ", using default");
327        groupName = RSGroupInfo.DEFAULT_GROUP;
328      }
329      RSGroupInfo info = rsGroupInfoManager.getRSGroup(groupName);
330      if (assignedServer == null) {
331        LOG.debug("There is no assigned server for {}", region);
332        continue;
333      }
334      RSGroupInfo otherInfo = rsGroupInfoManager.getRSGroupOfServer(assignedServer.getAddress());
335      if (info == null && otherInfo == null) {
336        LOG.warn("Couldn't obtain rs group information for {} on {}", region, assignedServer);
337        continue;
338      }
339      if ((info == null || !info.containsServer(assignedServer.getAddress()))) {
340        LOG.debug("Found misplaced region: " + regionInfo.getRegionNameAsString() +
341            " on server: " + assignedServer +
342            " found in group: " +  otherInfo +
343            " outside of group: " + (info == null ? "UNKNOWN" : info.getName()));
344        misplacedRegions.add(regionInfo);
345      }
346    }
347    return misplacedRegions;
348  }
349
350  private ServerName findServerForRegion(
351      Map<ServerName, List<RegionInfo>> existingAssignments, RegionInfo region) {
352    for (Map.Entry<ServerName, List<RegionInfo>> entry : existingAssignments.entrySet()) {
353      if (entry.getValue().contains(region)) {
354        return entry.getKey();
355      }
356    }
357
358    throw new IllegalStateException("Could not find server for region "
359        + region.getShortNameToLog());
360  }
361
362  private Map<ServerName, List<RegionInfo>> correctAssignments(
363       Map<ServerName, List<RegionInfo>> existingAssignments)
364  throws HBaseIOException{
365    Map<ServerName, List<RegionInfo>> correctAssignments = new TreeMap<>();
366    correctAssignments.put(LoadBalancer.BOGUS_SERVER_NAME, new LinkedList<>());
367    for (Map.Entry<ServerName, List<RegionInfo>> assignments : existingAssignments.entrySet()){
368      ServerName sName = assignments.getKey();
369      correctAssignments.put(sName, new LinkedList<>());
370      List<RegionInfo> regions = assignments.getValue();
371      for (RegionInfo region : regions) {
372        RSGroupInfo targetRSGInfo = null;
373        try {
374          String groupName = rsGroupInfoManager.getRSGroupOfTable(region.getTable());
375          if (groupName == null) {
376            LOG.info("Group not found for table " + region.getTable() + ", using default");
377            groupName = RSGroupInfo.DEFAULT_GROUP;
378          }
379          targetRSGInfo = rsGroupInfoManager.getRSGroup(groupName);
380        } catch (IOException exp) {
381          LOG.debug("RSGroup information null for region of table " + region.getTable(),
382              exp);
383        }
384        if ((targetRSGInfo == null) || (!targetRSGInfo.containsServer(sName.getAddress()))) {
385          correctAssignments.get(LoadBalancer.BOGUS_SERVER_NAME).add(region);
386        } else {
387          correctAssignments.get(sName).add(region);
388        }
389      }
390    }
391    return correctAssignments;
392  }
393
394  @Override
395  public void initialize() throws HBaseIOException {
396    try {
397      if (rsGroupInfoManager == null) {
398        List<RSGroupAdminEndpoint> cps =
399          masterServices.getMasterCoprocessorHost().findCoprocessors(RSGroupAdminEndpoint.class);
400        if (cps.size() != 1) {
401          String msg = "Expected one implementation of GroupAdminEndpoint but found " + cps.size();
402          LOG.error(msg);
403          throw new HBaseIOException(msg);
404        }
405        rsGroupInfoManager = cps.get(0).getGroupInfoManager();
406        if(rsGroupInfoManager == null){
407          String msg = "RSGroupInfoManager hasn't been initialized";
408          LOG.error(msg);
409          throw new HBaseIOException(msg);
410        }
411        rsGroupInfoManager.start();
412      }
413    } catch (IOException e) {
414      throw new HBaseIOException("Failed to initialize GroupInfoManagerImpl", e);
415    }
416
417    // Create the balancer
418    Class<? extends LoadBalancer> balancerKlass = config.getClass(HBASE_RSGROUP_LOADBALANCER_CLASS,
419        StochasticLoadBalancer.class, LoadBalancer.class);
420    internalBalancer = ReflectionUtils.newInstance(balancerKlass, config);
421    internalBalancer.setMasterServices(masterServices);
422    internalBalancer.setClusterMetrics(clusterStatus);
423    internalBalancer.setConf(config);
424    internalBalancer.initialize();
425  }
426
427  public boolean isOnline() {
428    if (this.rsGroupInfoManager == null) {
429      return false;
430    }
431
432    return this.rsGroupInfoManager.isOnline();
433  }
434
435  @Override
436  public void setClusterLoad(Map<TableName, Map<ServerName, List<RegionInfo>>> clusterLoad) {
437  }
438
439  @Override
440  public void regionOnline(RegionInfo regionInfo, ServerName sn) {
441  }
442
443  @Override
444  public void regionOffline(RegionInfo regionInfo) {
445  }
446
447  @Override
448  public void onConfigurationChange(Configuration conf) {
449    //DO nothing for now
450  }
451
452  @Override
453  public void stop(String why) {
454  }
455
456  @Override
457  public boolean isStopped() {
458    return false;
459  }
460
461  @VisibleForTesting
462  public void setRsGroupInfoManager(RSGroupInfoManager rsGroupInfoManager) {
463    this.rsGroupInfoManager = rsGroupInfoManager;
464  }
465
466  @Override
467  public void postMasterStartupInitialize() {
468    this.internalBalancer.postMasterStartupInitialize();
469  }
470}