Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-24760 Add a config hbase.rsgroup.fallback.enable for RSGroup fallback feature (#2325) 

Signed-off-by: Guanghao Zhang <zghao@apache.org>
This commit is contained in:
XinSun 2020-08-31 10:53:05 +08:00 committed by GitHub
parent caeaf79f27
commit ad177059cf
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 81 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

95
hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupBasedLoadBalancer.java
View File

@ -20,12 +20,11 @@ package org.apache.hadoop.hbase.rsgroup;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.TreeMap;
@ -50,7 +49,6 @@ import org.slf4j.LoggerFactory;
import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
import org.apache.hbase.thirdparty.com.google.common.collect.ArrayListMultimap;
import org.apache.hbase.thirdparty.com.google.common.collect.LinkedListMultimap;
import org.apache.hbase.thirdparty.com.google.common.collect.ListMultimap;
import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
import org.apache.hbase.thirdparty.com.google.common.collect.Maps;
@ -80,15 +78,15 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer {
private LoadBalancer internalBalancer;
/**
* Define the config key of fallback groups
* Enabled only if this property is set
* Set this key to {@code true} to allow region fallback.
* Fallback to the default rsgroup first, then fallback to any group if no online servers in
* default rsgroup.
* Please keep balancer switch on at the same time, which is relied on to correct misplaced
* regions
*/
public static final String FALLBACK_GROUPS_KEY = "hbase.rsgroup.fallback.groups";
public static final String FALLBACK_GROUP_ENABLE_KEY = "hbase.rsgroup.fallback.enable";
private boolean fallbackEnabled = false;
private Set<String> fallbackGroups;
/**
* Used by reflection in {@link org.apache.hadoop.hbase.master.balancer.LoadBalancerFactory}.
@ -176,22 +174,14 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer {
public Map<ServerName, List<RegionInfo>> roundRobinAssignment(List<RegionInfo> regions,
List<ServerName> servers) throws HBaseIOException {
Map<ServerName, List<RegionInfo>> assignments = Maps.newHashMap();
ListMultimap<String, RegionInfo> regionMap = ArrayListMultimap.create();
ListMultimap<String, ServerName> serverMap = ArrayListMultimap.create();
generateGroupMaps(regions, servers, regionMap, serverMap);
for (String groupKey : regionMap.keySet()) {
if (regionMap.get(groupKey).size() > 0) {
List<Pair<List<RegionInfo>, List<ServerName>>> pairs =
generateGroupAssignments(regions, servers);
for (Pair<List<RegionInfo>, List<ServerName>> pair : pairs) {
Map<ServerName, List<RegionInfo>> result = this.internalBalancer
.roundRobinAssignment(regionMap.get(groupKey), serverMap.get(groupKey));
.roundRobinAssignment(pair.getFirst(), pair.getSecond());
if (result != null) {
if (result.containsKey(LoadBalancer.BOGUS_SERVER_NAME) &&
assignments.containsKey(LoadBalancer.BOGUS_SERVER_NAME)) {
assignments.get(LoadBalancer.BOGUS_SERVER_NAME)
.addAll(result.get(LoadBalancer.BOGUS_SERVER_NAME));
} else {
assignments.putAll(result);
}
}
result.forEach((server, regionInfos) ->
assignments.computeIfAbsent(server, s -> Lists.newArrayList()).addAll(regionInfos));
}
}
return assignments;
@ -243,37 +233,50 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer {
@Override
public ServerName randomAssignment(RegionInfo region,
List<ServerName> servers) throws HBaseIOException {
ListMultimap<String,RegionInfo> regionMap = LinkedListMultimap.create();
ListMultimap<String,ServerName> serverMap = LinkedListMultimap.create();
generateGroupMaps(Lists.newArrayList(region), servers, regionMap, serverMap);
List<ServerName> filteredServers = serverMap.get(regionMap.keySet().iterator().next());
List<Pair<List<RegionInfo>, List<ServerName>>> pairs =
generateGroupAssignments(Lists.newArrayList(region), servers);
List<ServerName> filteredServers = pairs.iterator().next().getSecond();
return this.internalBalancer.randomAssignment(region, filteredServers);
}
private void generateGroupMaps(List<RegionInfo> regions, List<ServerName> servers,
ListMultimap<String, RegionInfo> regionMap, ListMultimap<String, ServerName> serverMap)
throws HBaseIOException {
private List<Pair<List<RegionInfo>, List<ServerName>>> generateGroupAssignments(
List<RegionInfo> regions, List<ServerName> servers) throws HBaseIOException {
try {
ListMultimap<String, RegionInfo> regionMap = ArrayListMultimap.create();
ListMultimap<String, ServerName> serverMap = ArrayListMultimap.create();
RSGroupInfo defaultInfo = rsGroupInfoManager.getRSGroup(RSGroupInfo.DEFAULT_GROUP);
for (RegionInfo region : regions) {
String groupName = rsGroupInfoManager.getRSGroupOfTable(region.getTable());
if (groupName == null) {
LOG.debug("Group not found for table " + region.getTable() + ", using default");
groupName = RSGroupInfo.DEFAULT_GROUP;
}
String groupName =
Optional.ofNullable(rsGroupInfoManager.getRSGroupOfTable(region.getTable()))
.orElse(defaultInfo.getName());
regionMap.put(groupName, region);
}
for (String groupKey : regionMap.keySet()) {
RSGroupInfo info = rsGroupInfoManager.getRSGroup(groupKey);
serverMap.putAll(groupKey, filterOfflineServers(info, servers));
if (fallbackEnabled && serverMap.get(groupKey).isEmpty()) {
serverMap.putAll(groupKey, getFallBackCandidates(servers));
}
List<Pair<List<RegionInfo>, List<ServerName>>> result = Lists.newArrayList();
List<RegionInfo> fallbackRegions = Lists.newArrayList();
for (String groupKey : regionMap.keySet()) {
if (serverMap.get(groupKey).isEmpty()) {
serverMap.put(groupKey, LoadBalancer.BOGUS_SERVER_NAME);
fallbackRegions.addAll(regionMap.get(groupKey));
} else {
result.add(Pair.newPair(regionMap.get(groupKey), serverMap.get(groupKey)));
}
}
if (!fallbackRegions.isEmpty()) {
List<ServerName> candidates = null;
if (fallbackEnabled) {
candidates = getFallBackCandidates(servers);
}
candidates = (candidates == null || candidates.isEmpty()) ?
Lists.newArrayList(BOGUS_SERVER_NAME) : candidates;
result.add(Pair.newPair(fallbackRegions, candidates));
}
return result;
} catch(IOException e) {
throw new HBaseIOException("Failed to generate group maps", e);
throw new HBaseIOException("Failed to generate group assignments", e);
}
}
@ -389,11 +392,7 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer {
internalBalancer.setConf(config);
internalBalancer.initialize();
// init fallback groups
Collection<String> groups = config.getTrimmedStringCollection(FALLBACK_GROUPS_KEY);
if (groups != null && !groups.isEmpty()) {
this.fallbackEnabled = true;
this.fallbackGroups = new HashSet<>(groups);
}
this.fallbackEnabled = config.getBoolean(FALLBACK_GROUP_ENABLE_KEY, false);
}
public boolean isOnline() {
@ -476,15 +475,13 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer {
}
private List<ServerName> getFallBackCandidates(List<ServerName> servers) {
List<ServerName> serverNames = new ArrayList<>();
for (String fallbackGroup : fallbackGroups) {
List<ServerName> serverNames = null;
try {
RSGroupInfo info = rsGroupInfoManager.getRSGroup(fallbackGroup);
serverNames.addAll(filterOfflineServers(info, servers));
RSGroupInfo info = rsGroupInfoManager.getRSGroup(RSGroupInfo.DEFAULT_GROUP);
serverNames = filterOfflineServers(info, servers);
} catch (IOException e) {
LOG.error("Get group info for {} failed", fallbackGroup, e);
LOG.error("Failed to get default rsgroup info to fallback", e);
}
}
return serverNames;
return serverNames == null || serverNames.isEmpty() ? servers : serverNames;
}
}

53
hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroupsFallback.java
View File

@ -24,6 +24,7 @@ import java.util.Collections;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
@ -31,6 +32,8 @@ import org.apache.hadoop.hbase.master.assignment.AssignmentTestingUtil;
import org.apache.hadoop.hbase.net.Address;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.JVMClusterUtil;
import org.apache.hadoop.hbase.util.Threads;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
@ -54,8 +57,8 @@ public class TestRSGroupsFallback extends TestRSGroupsBase {
@BeforeClass
public static void setUp() throws Exception {
Configuration configuration = TEST_UTIL.getConfiguration();
configuration.set(RSGroupBasedLoadBalancer.FALLBACK_GROUPS_KEY, FALLBACK_GROUP);
Configuration conf = TEST_UTIL.getConfiguration();
conf.setBoolean(RSGroupBasedLoadBalancer.FALLBACK_GROUP_ENABLE_KEY, true);
setUpTestBeforeClass();
master.balanceSwitch(true);
}
@ -76,7 +79,7 @@ public class TestRSGroupsFallback extends TestRSGroupsBase {
}
@Test
public void testGroupFallback() throws Exception {
public void testFallback() throws Exception {
// add fallback group
addGroup(FALLBACK_GROUP, 1);
// add test group
@ -88,34 +91,31 @@ public class TestRSGroupsFallback extends TestRSGroupsBase {
admin.createTable(desc);
rsGroupAdmin.moveTables(Collections.singleton(tableName), groupName);
TEST_UTIL.waitUntilAllRegionsAssigned(tableName);
// server of test group crash
for (Address server : rsGroupAdmin.getRSGroupInfo(groupName).getServers()) {
AssignmentTestingUtil.crashRs(TEST_UTIL, getServerName(server), true);
}
TEST_UTIL.waitUntilNoRegionsInTransition(10000);
TEST_UTIL.waitUntilAllRegionsAssigned(tableName);
// server of test group crash, regions move to default group
crashRsInGroup(groupName);
assertRegionsInGroup(tableName, RSGroupInfo.DEFAULT_GROUP);
// regions move to fallback group
assertRegionsInGroup(FALLBACK_GROUP);
// server of default group crash, regions move to any other group
crashRsInGroup(RSGroupInfo.DEFAULT_GROUP);
assertRegionsInGroup(tableName, FALLBACK_GROUP);
// move a new server from default group
Address address = rsGroupAdmin.getRSGroupInfo(RSGroupInfo.DEFAULT_GROUP).getServers()
.iterator().next();
rsGroupAdmin.moveServers(Collections.singleton(address), groupName);
// correct misplaced regions
// add a new server to default group, regions move to default group
TEST_UTIL.getMiniHBaseCluster().startRegionServerAndWait(60000);
master.balance();
assertRegionsInGroup(tableName, RSGroupInfo.DEFAULT_GROUP);
TEST_UTIL.waitUntilNoRegionsInTransition(10000);
TEST_UTIL.waitUntilAllRegionsAssigned(tableName);
// regions move back
assertRegionsInGroup(groupName);
// add a new server to test group, regions move back
JVMClusterUtil.RegionServerThread t =
TEST_UTIL.getMiniHBaseCluster().startRegionServerAndWait(60000);
rsGroupAdmin.moveServers(
Collections.singleton(t.getRegionServer().getServerName().getAddress()), groupName);
master.balance();
assertRegionsInGroup(tableName, groupName);
TEST_UTIL.deleteTable(tableName);
}
private void assertRegionsInGroup(String group) throws IOException {
private void assertRegionsInGroup(TableName tableName, String group) throws IOException {
RSGroupInfo fallbackGroup = rsGroupAdmin.getRSGroupInfo(group);
master.getAssignmentManager().getRegionStates().getRegionsOfTable(tableName).forEach(region -> {
Address regionOnServer = master.getAssignmentManager().getRegionStates()
@ -124,4 +124,11 @@ public class TestRSGroupsFallback extends TestRSGroupsBase {
});
}
private void crashRsInGroup(String groupName) throws Exception {
for (Address server : rsGroupAdmin.getRSGroupInfo(groupName).getServers()) {
AssignmentTestingUtil.crashRs(TEST_UTIL, getServerName(server), true);
}
Threads.sleep(1000);
TEST_UTIL.waitUntilNoRegionsInTransition(60000);
}
}