Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.stream.Collectors;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.ClusterMetrics;
import org.apache.hadoop.hbase.HBaseIOException;
Expand Down Expand Up @@ -129,6 +131,7 @@ public synchronized List<RegionPlan> balanceCluster(
try {
// For each rsgroup
for (RSGroupInfo rsgroup : rsGroupInfoManager.listRSGroups()) {
LOG.debug("Balancing RSGroup={}", rsgroup.getName());
Map<TableName, Map<ServerName, List<RegionInfo>>> loadOfTablesInGroup = new HashMap<>();
for (Map.Entry<TableName, Map<ServerName, List<RegionInfo>>> entry : correctedLoadOfAllTable
.entrySet()) {
Expand Down Expand Up @@ -234,6 +237,11 @@ private List<Pair<List<RegionInfo>, List<ServerName>>> generateGroupAssignments(
if (!fallbackRegions.isEmpty()) {
List<ServerName> candidates = null;
if (isFallbackEnabled()) {
if (LOG.isDebugEnabled()) {
LOG.debug("Falling back {} regions to servers outside their RSGroup. Regions: {}",
fallbackRegions.size(), fallbackRegions.stream()
.map(RegionInfo::getRegionNameAsString).collect(Collectors.toSet()));
}
candidates = getFallBackCandidates(servers);
}
candidates = (candidates == null || candidates.isEmpty()) ?
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -259,6 +259,9 @@ private synchronized void updateDefaultServers() {
// do not need to persist, as we do not persist default group.
resetRSGroupMap(newGroupMap);
LOG.info("Updated default servers, {} servers", newDefaultGroupInfo.getServers().size());
if (LOG.isDebugEnabled()) {
LOG.debug("New default servers list: {}", newDefaultGroupInfo.getServers());
}
}

private synchronized void init() throws IOException {
Expand Down Expand Up @@ -300,6 +303,7 @@ public synchronized void addRSGroup(RSGroupInfo rsGroupInfo) throws IOException
Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(rsGroupMap);
newGroupMap.put(rsGroupInfo.getName(), rsGroupInfo);
flushConfig(newGroupMap);
LOG.info("Add group {} done.", rsGroupInfo.getName());
}

private RSGroupInfo getRSGroupInfo(final String groupName) throws ConstraintException {
Expand Down Expand Up @@ -333,7 +337,7 @@ public synchronized Set<Address> moveServers(Set<Address> servers, String srcGro
if (onlineServers != null) {
if (!onlineServers.contains(el)) {
if (LOG.isDebugEnabled()) {
LOG.debug("Dropping " + el + " during move-to-default rsgroup because not online");
LOG.debug("Dropping " + el + " during move-to-default RSGroup because not online");
}
continue;
}
Expand Down Expand Up @@ -375,8 +379,8 @@ public synchronized void removeRSGroup(String groupName) throws IOException {
for (TableDescriptor td : masterServices.getTableDescriptors().getAll().values()) {
if (td.getRegionServerGroup().map(groupName::equals).orElse(false)) {
throw new ConstraintException("RSGroup " + groupName + " is already referenced by " +
td.getTableName() + "; you must remove all the tables from the rsgroup before " +
"the rsgroup can be removed.");
td.getTableName() + "; you must remove all the tables from the RSGroup before " +
"the RSGroup can be removed.");
}
}
for (NamespaceDescriptor ns : masterServices.getClusterSchema().getNamespaces()) {
Expand All @@ -394,6 +398,7 @@ public synchronized void removeRSGroup(String groupName) throws IOException {
Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(rsGroupMap);
newGroupMap.remove(groupName);
flushConfig(newGroupMap);
LOG.info("Remove group {} done", groupName);
}

@Override
Expand Down Expand Up @@ -641,6 +646,8 @@ private synchronized void flushConfig(Map<String, RSGroupInfo> newGroupMap) thro
return;
}

LOG.debug("Offline mode, cannot persist to {}", RSGROUP_TABLE_NAME);

Map<String, RSGroupInfo> oldGroupMap = Maps.newHashMap(holder.groupName2Group);
RSGroupInfo oldDefaultGroup = oldGroupMap.remove(RSGroupInfo.DEFAULT_GROUP);
RSGroupInfo newDefaultGroup = newGroupMap.remove(RSGroupInfo.DEFAULT_GROUP);
Expand All @@ -657,22 +664,27 @@ private synchronized void flushConfig(Map<String, RSGroupInfo> newGroupMap) thro
// according to the inputted newGroupMap (an updated copy of rsGroupMap)
this.holder = new RSGroupInfoHolder(newGroupMap);

LOG.debug("New RSGroup map: {}", newGroupMap);

// Do not need to update tableMap
// because only the update on servers in default group is allowed above,
// or IOException will be thrown
return;
}

/* For online mode, persist to hbase:rsgroup and Zookeeper */
LOG.debug("Online mode, persisting to {} and ZK", RSGROUP_TABLE_NAME);
flushConfigTable(newGroupMap);

// Make changes visible after having been persisted to the source of truth
resetRSGroupMap(newGroupMap);
saveRSGroupMapToZK(newGroupMap);
updateCacheOfRSGroups(newGroupMap.keySet());
LOG.info("Flush config done, new RSGroup map: {}", newGroupMap);
}

private void saveRSGroupMapToZK(Map<String, RSGroupInfo> newGroupMap) throws IOException {
LOG.debug("Saving RSGroup info to ZK");
try {
String groupBasePath =
ZNodePaths.joinZNode(watcher.getZNodePaths().baseZNode, RS_GROUP_ZNODE);
Expand Down Expand Up @@ -802,6 +814,7 @@ private void createRSGroupTable() throws IOException {
if (optProcId.isPresent()) {
procId = optProcId.getAsLong();
} else {
LOG.debug("Creating group table {}", RSGROUP_TABLE_NAME);
procId = masterServices.createSystemTable(RSGROUP_TABLE_DESC);
}
// wait for region to be online
Expand Down Expand Up @@ -851,9 +864,11 @@ private void multiMutate(List<Mutation> mutations) throws IOException {
}
MutateRowsRequest request = builder.build();
AsyncTable<?> table = conn.getTable(RSGROUP_TABLE_NAME);
LOG.debug("Multimutating {} with {} mutations", RSGROUP_TABLE_NAME, mutations.size());
FutureUtils.get(table.<MultiRowMutationService, MutateRowsResponse> coprocessorService(
MultiRowMutationService::newStub,
(stub, controller, done) -> stub.mutateRows(controller, request, done), ROW_KEY));
LOG.info("Multimutating {} with {} mutations done", RSGROUP_TABLE_NAME, mutations.size());
}

private void checkGroupName(String groupName) throws ConstraintException {
Expand All @@ -873,7 +888,7 @@ public RSGroupInfo getRSGroupForTable(TableName tableName) throws IOException {
* @param servers servers to remove
*/
private void checkForDeadOrOnlineServers(Set<Address> servers) throws IOException {
// This uglyness is because we only have Address, not ServerName.
// This ugliness is because we only have Address, not ServerName.
Set<Address> onlineServers = new HashSet<>();
List<ServerName> drainingServers = masterServices.getServerManager().getDrainingServersList();
for (ServerName server : masterServices.getServerManager().getOnlineServers().keySet()) {
Expand Down Expand Up @@ -1003,7 +1018,7 @@ private <T> void moveRegionsBetweenGroups(Set<T> regionsOwners, Set<Address> new
// Get regions that are associated with this server and filter regions by group tables.
for (RegionInfo region : getRegionsInfo.apply((T) owner.getAddress())) {
if (!validation.apply(region)) {
LOG.info("Moving region {}, which do not belong to RSGroup {}",
LOG.info("Moving region {}, which does not belong to RSGroup {}",
region.getShortNameToLog(), targetGroupName);
// Move region back to source RSGroup servers
ServerName dest =
Expand Down Expand Up @@ -1197,6 +1212,7 @@ public BalanceResponse balanceRSGroup(String groupName, BalanceRequest request)
}

private void moveTablesAndWait(Set<TableName> tables, String targetGroup) throws IOException {
LOG.debug("Moving {} tables to target group {}", tables.size(), targetGroup);
List<Long> procIds = new ArrayList<Long>();
for (TableName tableName : tables) {
TableDescriptor oldTd = masterServices.getTableDescriptors().get(tableName);
Expand All @@ -1216,6 +1232,10 @@ private void moveTablesAndWait(Set<TableName> tables, String targetGroup) throws
ProcedureSyncWait.waitForProcedureToCompleteIOE(masterServices.getMasterProcedureExecutor(),
proc, Long.MAX_VALUE);
}
LOG.info("Move tables done: moved {} tables to {}", tables.size(), targetGroup);
if (LOG.isDebugEnabled()) {
LOG.debug("Tables moved to {}: {}", targetGroup, tables);
}
}

@Override
Expand Down Expand Up @@ -1281,7 +1301,12 @@ public void moveServers(Set<Address> servers, String targetGroupName) throws IOE
Set<Address> movedServers = moveServers(servers, srcGrp.getName(),
targetGroupName);
moveServerRegionsFromGroup(movedServers, srcGrp.getServers(), targetGroupName, srcGrp.getName());
LOG.info("Move servers done: {} => {}", srcGrp.getName(), targetGroupName);
LOG.info("Move servers done: moved {} servers from {} to {}", movedServers.size(),
srcGrp.getName(), targetGroupName);
if (LOG.isDebugEnabled()) {
LOG.debug("Servers moved from {} to {}: {}", srcGrp.getName(), targetGroupName,
movedServers);
}
}
}

Expand Down Expand Up @@ -1315,6 +1340,7 @@ public synchronized void renameRSGroup(String oldName, String newName) throws IO
.map(TableDescriptor::getTableName)
.collect(Collectors.toSet());
setRSGroup(updateTables, newName);
LOG.info("Rename RSGroup done: {} => {}", oldName, newName);
}

@Override
Expand Down