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 */ 018package org.apache.hadoop.hbase.rsgroup; 019 020import java.io.ByteArrayInputStream; 021import java.io.IOException; 022import java.util.ArrayList; 023import java.util.Collection; 024import java.util.Collections; 025import java.util.HashMap; 026import java.util.HashSet; 027import java.util.LinkedList; 028import java.util.List; 029import java.util.Map; 030import java.util.Optional; 031import java.util.OptionalLong; 032import java.util.Set; 033import java.util.SortedSet; 034import java.util.TreeSet; 035import java.util.concurrent.Future; 036import java.util.function.Function; 037import java.util.regex.Pattern; 038import java.util.regex.PatternSyntaxException; 039import java.util.stream.Collectors; 040import org.apache.commons.lang3.StringUtils; 041import org.apache.hadoop.conf.Configuration; 042import org.apache.hadoop.hbase.Coprocessor; 043import org.apache.hadoop.hbase.DoNotRetryIOException; 044import org.apache.hadoop.hbase.HConstants; 045import org.apache.hadoop.hbase.NamespaceDescriptor; 046import org.apache.hadoop.hbase.ServerName; 047import org.apache.hadoop.hbase.TableDescriptors; 048import org.apache.hadoop.hbase.TableName; 049import org.apache.hadoop.hbase.client.AsyncClusterConnection; 050import org.apache.hadoop.hbase.client.AsyncTable; 051import org.apache.hadoop.hbase.client.BalanceRequest; 052import org.apache.hadoop.hbase.client.BalanceResponse; 053import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; 054import org.apache.hadoop.hbase.client.CoprocessorDescriptorBuilder; 055import org.apache.hadoop.hbase.client.Delete; 056import org.apache.hadoop.hbase.client.Get; 057import org.apache.hadoop.hbase.client.Mutation; 058import org.apache.hadoop.hbase.client.Put; 059import org.apache.hadoop.hbase.client.RegionInfo; 060import org.apache.hadoop.hbase.client.Result; 061import org.apache.hadoop.hbase.client.ResultScanner; 062import org.apache.hadoop.hbase.client.TableDescriptor; 063import org.apache.hadoop.hbase.client.TableDescriptorBuilder; 064import org.apache.hadoop.hbase.client.TableState; 065import org.apache.hadoop.hbase.constraint.ConstraintException; 066import org.apache.hadoop.hbase.coprocessor.MultiRowMutationEndpoint; 067import org.apache.hadoop.hbase.exceptions.DeserializationException; 068import org.apache.hadoop.hbase.master.LoadBalancer; 069import org.apache.hadoop.hbase.master.MasterServices; 070import org.apache.hadoop.hbase.master.RegionPlan; 071import org.apache.hadoop.hbase.master.RegionState; 072import org.apache.hadoop.hbase.master.ServerListener; 073import org.apache.hadoop.hbase.master.ServerManager; 074import org.apache.hadoop.hbase.master.TableStateManager; 075import org.apache.hadoop.hbase.master.assignment.RegionStateNode; 076import org.apache.hadoop.hbase.master.procedure.CreateTableProcedure; 077import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv; 078import org.apache.hadoop.hbase.master.procedure.MasterProcedureUtil; 079import org.apache.hadoop.hbase.master.procedure.ProcedureSyncWait; 080import org.apache.hadoop.hbase.net.Address; 081import org.apache.hadoop.hbase.procedure2.Procedure; 082import org.apache.hadoop.hbase.procedure2.ProcedureExecutor; 083import org.apache.hadoop.hbase.protobuf.ProtobufMagic; 084import org.apache.hadoop.hbase.regionserver.DisabledRegionSplitPolicy; 085import org.apache.hadoop.hbase.util.Bytes; 086import org.apache.hadoop.hbase.util.FutureUtils; 087import org.apache.hadoop.hbase.util.Pair; 088import org.apache.hadoop.hbase.util.Threads; 089import org.apache.hadoop.hbase.zookeeper.ZKUtil; 090import org.apache.hadoop.hbase.zookeeper.ZKWatcher; 091import org.apache.hadoop.hbase.zookeeper.ZNodePaths; 092import org.apache.hadoop.util.Shell; 093import org.apache.yetus.audience.InterfaceAudience; 094import org.apache.zookeeper.KeeperException; 095import org.slf4j.Logger; 096import org.slf4j.LoggerFactory; 097 098import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableMap; 099import org.apache.hbase.thirdparty.com.google.common.collect.Lists; 100import org.apache.hbase.thirdparty.com.google.common.collect.Maps; 101 102import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; 103import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.MutationProto; 104import org.apache.hadoop.hbase.shaded.protobuf.generated.MultiRowMutationProtos.MultiRowMutationService; 105import org.apache.hadoop.hbase.shaded.protobuf.generated.MultiRowMutationProtos.MutateRowsRequest; 106import org.apache.hadoop.hbase.shaded.protobuf.generated.MultiRowMutationProtos.MutateRowsResponse; 107import org.apache.hadoop.hbase.shaded.protobuf.generated.RSGroupProtos; 108 109/** 110 * This is an implementation of {@link RSGroupInfoManager} which makes use of an HBase table as the 111 * persistence store for the group information. It also makes use of zookeeper to store group 112 * information needed for bootstrapping during offline mode. 113 * <h2>Admin-managed and auto-managed groups</h2> An <b>admin-managed</b> group is a non-default 114 * group that is not regex-governed (see {@code hbase.rsgroup.regex.<group>}). Its servers are set 115 * by an administrator and are stored in hbase:rsgroup and ZooKeeper. A group whose regex is invalid 116 * is admin-managed until the regex is fixed. In that time the servers that the group holds in 117 * memory count as admin-managed when the membership is computed 118 * ({@code resolveRegexBasedRSGroupMembership}). After a master restart the group holds only what 119 * was stored at the last flush. 120 * <p/> 121 * An <b>auto-managed</b> group is 'default' or a regex-governed group. Its servers are computed 122 * from the live servers and, while the regex is valid, are not stored in hbase:rsgroup and 123 * ZooKeeper. 124 * <p/> 125 * Auto-managed groups exist to make RSGroup work with auto-scaling when HBase runs in a cloud 126 * environment such as AWS or GCP. An admin-managed group does not give the elasticity that 127 * automatic upscaling and downscaling need: each new RegionServer must be moved into the group by 128 * hand. It fits a fixed number of RegionServers, for example to host internal system tables like 129 * hbase:meta and hbase:namespace. End user facing tables must be placed on an elastic group, so 130 * RegionServers join and leave it automatically. That is the need for auto-managed groups. 131 * <h2>Concurrency</h2> RSGroup state is kept locally in Maps. There is a rsgroup name to cached 132 * RSGroupInfo Map at {@link RSGroupInfoHolder#groupName2Group}. These Maps are persisted to the 133 * hbase:rsgroup table (and cached in zk) on each modification. 134 * <p/> 135 * Mutations on state are synchronized but reads can continue without having to wait on an instance 136 * monitor, mutations do wholesale replace of the Maps on update -- Copy-On-Write; the local Maps of 137 * state are read-only, just-in-case (see flushConfig). 138 * <p/> 139 * Reads must not block else there is a danger we'll deadlock. 140 * <p/> 141 * Clients of this class, the {@link RSGroupAdminEndpoint} for example, want to query and then act 142 * on the results of the query modifying cache in zookeeper without another thread making 143 * intermediate modifications. These clients synchronize on the 'this' instance so no other has 144 * access concurrently. Reads must be able to continue concurrently. 145 */ 146@InterfaceAudience.Private 147final class RSGroupInfoManagerImpl implements RSGroupInfoManager { 148 private static final Logger LOG = LoggerFactory.getLogger(RSGroupInfoManagerImpl.class); 149 150 // Assigned before user tables 151 static final TableName RSGROUP_TABLE_NAME = 152 TableName.valueOf(NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR, "rsgroup"); 153 154 static final String KEEP_ONE_SERVER_IN_DEFAULT_ERROR_MESSAGE = 155 "should keep at least " + "one server in 'default' RSGroup."; 156 157 /** Define the config key of retries threshold when movements failed */ 158 static final String FAILED_MOVE_MAX_RETRY = "hbase.rsgroup.move.max.retry"; 159 160 /** Define the default number of retries */ 161 static final int DEFAULT_MAX_RETRY_VALUE = 50; 162 163 /** Config key prefix; {@code hbase.rsgroup.regex.<groupname>=<regex>} */ 164 static final String RS_GROUP_REGEX_PREFIX = "hbase.rsgroup.regex."; 165 private static final Pattern GROUP_NAME_PATTERN = Pattern.compile("[a-zA-Z0-9_]+"); 166 167 private static final String RS_GROUP_ZNODE = "rsgroup"; 168 169 static final byte[] META_FAMILY_BYTES = Bytes.toBytes("m"); 170 171 static final byte[] META_QUALIFIER_BYTES = Bytes.toBytes("i"); 172 173 static final String MIGRATE_THREAD_NAME = "Migrate-RSGroup-Tables"; 174 175 private static final byte[] ROW_KEY = { 0 }; 176 177 /** Table descriptor for <code>hbase:rsgroup</code> catalog table */ 178 private static final TableDescriptor RSGROUP_TABLE_DESC; 179 static { 180 TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(RSGROUP_TABLE_NAME) 181 .setColumnFamily(ColumnFamilyDescriptorBuilder.of(META_FAMILY_BYTES)) 182 .setRegionSplitPolicyClassName(DisabledRegionSplitPolicy.class.getName()); 183 try { 184 builder.setCoprocessor( 185 CoprocessorDescriptorBuilder.newBuilder(MultiRowMutationEndpoint.class.getName()) 186 .setPriority(Coprocessor.PRIORITY_SYSTEM).build()); 187 } catch (IOException ex) { 188 throw new Error(ex); 189 } 190 RSGROUP_TABLE_DESC = builder.build(); 191 } 192 193 // There two Maps are immutable and wholesale replaced on each modification 194 // so are safe to access concurrently. See class comment. 195 private static final class RSGroupInfoHolder { 196 final ImmutableMap<String, RSGroupInfo> groupName2Group; 197 final ImmutableMap<TableName, RSGroupInfo> tableName2Group; 198 199 RSGroupInfoHolder() { 200 this(Collections.emptyMap()); 201 } 202 203 RSGroupInfoHolder(Map<String, RSGroupInfo> rsGroupMap) { 204 ImmutableMap.Builder<String, RSGroupInfo> group2Name2GroupBuilder = ImmutableMap.builder(); 205 ImmutableMap.Builder<TableName, RSGroupInfo> tableName2GroupBuilder = ImmutableMap.builder(); 206 rsGroupMap.forEach((groupName, rsGroupInfo) -> { 207 group2Name2GroupBuilder.put(groupName, rsGroupInfo); 208 if (!groupName.equals(RSGroupInfo.DEFAULT_GROUP)) { 209 rsGroupInfo.getTables() 210 .forEach(tableName -> tableName2GroupBuilder.put(tableName, rsGroupInfo)); 211 } 212 }); 213 this.groupName2Group = group2Name2GroupBuilder.build(); 214 this.tableName2Group = tableName2GroupBuilder.build(); 215 } 216 } 217 218 private volatile RSGroupInfoHolder holder = new RSGroupInfoHolder(); 219 220 private final MasterServices masterServices; 221 private final AsyncClusterConnection conn; 222 private final ZKWatcher watcher; 223 private final RSGroupStartupWorker rsGroupStartupWorker; 224 private final ServerEventsListener serverEventsListener = new ServerEventsListener(); 225 // contains list of groups that were last flushed to persistent store 226 private Set<String> prevRSGroups = new HashSet<>(); 227 228 // Package visibility for testing 229 static class RSGroupMappingScript { 230 static final String RS_GROUP_MAPPING_SCRIPT = "hbase.rsgroup.table.mapping.script"; 231 static final String RS_GROUP_MAPPING_SCRIPT_TIMEOUT = 232 "hbase.rsgroup.table.mapping.script.timeout"; 233 234 private final String script; 235 private final long scriptTimeout; 236 237 RSGroupMappingScript(Configuration conf) { 238 script = conf.get(RS_GROUP_MAPPING_SCRIPT); 239 scriptTimeout = conf.getLong(RS_GROUP_MAPPING_SCRIPT_TIMEOUT, 5000); // 5 seconds 240 } 241 242 String getRSGroup(String namespace, String tablename) { 243 if (script == null || script.isEmpty()) { 244 return null; 245 } 246 Shell.ShellCommandExecutor rsgroupMappingScript = 247 new Shell.ShellCommandExecutor(new String[] { script, "", "" }, null, null, scriptTimeout); 248 249 String[] exec = rsgroupMappingScript.getExecString(); 250 exec[1] = namespace; 251 exec[2] = tablename; 252 try { 253 rsgroupMappingScript.execute(); 254 } catch (IOException e) { 255 // This exception may happen, like process doesn't have permission to run this script. 256 LOG.error("{}, placing {} back to default rsgroup", e.getMessage(), 257 TableName.valueOf(namespace, tablename)); 258 return RSGroupInfo.DEFAULT_GROUP; 259 } 260 return rsgroupMappingScript.getOutput().trim(); 261 } 262 } 263 264 private RSGroupMappingScript script; 265 266 private RSGroupInfoManagerImpl(MasterServices masterServices) { 267 this.masterServices = masterServices; 268 this.watcher = masterServices.getZooKeeper(); 269 this.conn = masterServices.getAsyncClusterConnection(); 270 this.rsGroupStartupWorker = new RSGroupStartupWorker(); 271 this.script = new RSGroupMappingScript(masterServices.getConfiguration()); 272 } 273 274 /** 275 * Called synchronously from {@code ServerListener#serverAdded}/{@code #serverRemoved}. Server 276 * membership of 'default' and of every regex-governed group is never persisted, so it is 277 * recomputed from the live servers and applied in memory only -- no {@code hbase:rsgroup}/ZK I/O. 278 */ 279 private synchronized void handleServerEvent() { 280 Map<String, RSGroupInfo> currentGroups = holder.groupName2Group; 281 Map<String, SortedSet<Address>> autoManagedServers = 282 computeAutoManagedRSGroupServers(currentGroups.values()); 283 Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(currentGroups); 284 applyAutoManagedRSGroupServers(newGroupMap, autoManagedServers); 285 dropRegexGovernedServersFromOtherGroups(newGroupMap, autoManagedServers); 286 for (Map.Entry<String, RSGroupInfo> entry : newGroupMap.entrySet()) { 287 if (!entry.getValue().getServers().equals(currentGroups.get(entry.getKey()).getServers())) { 288 resetRSGroupMap(newGroupMap); 289 LOG.info("Updated auto-managed RSGroup servers, {} servers", 290 autoManagedServers.values().stream().mapToInt(SortedSet::size).sum()); 291 return; 292 } 293 } 294 } 295 296 private synchronized void init() throws IOException { 297 refresh(false); 298 masterServices.getServerManager().registerListener(serverEventsListener); 299 } 300 301 static RSGroupInfoManager getInstance(MasterServices masterServices) throws IOException { 302 RSGroupInfoManagerImpl instance = new RSGroupInfoManagerImpl(masterServices); 303 instance.init(); 304 return instance; 305 } 306 307 public void start() { 308 // create system table of rsgroup 309 rsGroupStartupWorker.start(); 310 } 311 312 @Override 313 public synchronized void addRSGroup(RSGroupInfo rsGroupInfo) throws IOException { 314 checkGroupName(rsGroupInfo.getName()); 315 Map<String, RSGroupInfo> rsGroupMap = holder.groupName2Group; 316 if ( 317 rsGroupMap.get(rsGroupInfo.getName()) != null 318 || rsGroupInfo.getName().equals(RSGroupInfo.DEFAULT_GROUP) 319 ) { 320 throw new ConstraintException("Group already exists: " + rsGroupInfo.getName()); 321 } 322 Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(rsGroupMap); 323 newGroupMap.put(rsGroupInfo.getName(), rsGroupInfo); 324 flushConfig(newGroupMap); 325 LOG.info("Add group {} done.", rsGroupInfo.getName()); 326 } 327 328 private RSGroupInfo getRSGroupInfo(final String groupName) throws ConstraintException { 329 RSGroupInfo rsGroupInfo = holder.groupName2Group.get(groupName); 330 if (rsGroupInfo == null) { 331 throw new ConstraintException("RSGroup " + groupName + " does not exist"); 332 } 333 return rsGroupInfo; 334 } 335 336 /** Returns Set of online Servers named for their hostname and port (not ServerName). */ 337 private Set<Address> getOnlineServers() { 338 return masterServices.getServerManager().getOnlineServers().keySet().stream() 339 .map(ServerName::getAddress).collect(Collectors.toSet()); 340 } 341 342 /** 343 * Returns a copy of {@code src}, safe to mutate without affecting the live {@code 344 * holder.groupName2Group} entry -- callers must not mutate the object returned by 345 * {@link #getRSGroupInfo} or {@link #getRSGroupOfServer} in place, since a subsequent 346 * {@link #flushConfig(Map)} can still reject the change and the live state must stay untouched 347 * until the change is actually persisted. 348 */ 349 private static RSGroupInfo copyOf(RSGroupInfo src) { 350 RSGroupInfo copy = new RSGroupInfo(src); 351 // The copy constructor does not copy configuration, so do it by hand. 352 src.getConfiguration().forEach(copy::setConfiguration); 353 return copy; 354 } 355 356 public synchronized Set<Address> moveServers(Set<Address> servers, String srcGroup, 357 String dstGroup) throws IOException { 358 RSGroupInfo src = copyOf(getRSGroupInfo(srcGroup)); 359 RSGroupInfo dst = copyOf(getRSGroupInfo(dstGroup)); 360 Set<Address> movedServers = new HashSet<>(); 361 // If destination is 'default' rsgroup, only add servers that are online. If not online, drop 362 // it. If not 'default' group, add server to 'dst' rsgroup EVEN IF IT IS NOT online (could be a 363 // rsgroup of dead servers that are to come back later). 364 Set<Address> onlineServers = 365 dst.getName().equals(RSGroupInfo.DEFAULT_GROUP) ? getOnlineServers() : null; 366 for (Address el : servers) { 367 src.removeServer(el); 368 if (onlineServers != null) { 369 if (!onlineServers.contains(el)) { 370 if (LOG.isDebugEnabled()) { 371 LOG.debug("Dropping " + el + " during move-to-default RSGroup because not online"); 372 } 373 continue; 374 } 375 } 376 dst.addServer(el); 377 movedServers.add(el); 378 } 379 checkMoveAgreesWithRegex(movedServers, dst.getName()); 380 Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(holder.groupName2Group); 381 newGroupMap.put(src.getName(), src); 382 newGroupMap.put(dst.getName(), dst); 383 flushConfig(newGroupMap); 384 return movedServers; 385 } 386 387 /** 388 * Regex-based membership always wins: the next server event or refresh puts a server back where 389 * its hostname says, so an explicit move that disagrees would only make regions flap. Such a move 390 * is rejected. A server whose hostname matches no regex (or an ambiguous or dangling one, which 391 * fall back to ordinary placement) may go anywhere except into a regex-governed group. 392 */ 393 private void checkMoveAgreesWithRegex(Set<Address> movedServers, String dstGroup) 394 throws ConstraintException { 395 Map<String, Pattern> regexGroupMap = getRegexGroupMap(masterServices.getConfiguration()); 396 if (regexGroupMap.isEmpty() || movedServers.isEmpty()) { 397 return; 398 } 399 Map<Address, String> expectedGroups = 400 resolveServerAddrToRSGroupName(movedServers, regexGroupMap, holder.groupName2Group.keySet()); 401 for (Address server : movedServers) { 402 String expectedGroup = expectedGroups.get(server); 403 boolean isWrongMove = expectedGroup != null 404 ? !expectedGroup.equals(dstGroup) 405 : regexGroupMap.containsKey(dstGroup); 406 if (isWrongMove) { 407 String reason = expectedGroup != null 408 ? "its membership is decided by " + RS_GROUP_REGEX_PREFIX + expectedGroup 409 + ", under which its hostname belongs in '" + expectedGroup + "'" 410 : "its hostname does not match " + RS_GROUP_REGEX_PREFIX + dstGroup 411 + ", which decides the membership of that RSGroup"; 412 throw new ConstraintException("Server " + server + " cannot be moved to RSGroup '" 413 + dstGroup + "': " + reason + ". Change that config instead."); 414 } 415 } 416 } 417 418 @Override 419 public RSGroupInfo getRSGroupOfServer(Address serverHostPort) { 420 for (RSGroupInfo info : holder.groupName2Group.values()) { 421 if (info.containsServer(serverHostPort)) { 422 return info; 423 } 424 } 425 return null; 426 } 427 428 @Override 429 public RSGroupInfo getRSGroup(String groupName) { 430 return holder.groupName2Group.get(groupName); 431 } 432 433 @Override 434 public synchronized void removeRSGroup(String groupName) throws IOException { 435 RSGroupInfo rsGroupInfo = getRSGroupInfo(groupName); 436 int serverCount = rsGroupInfo.getServers().size(); 437 if (serverCount > 0) { 438 throw new ConstraintException("RSGroup " + groupName + " has " + serverCount 439 + " servers; you must remove these servers from the RSGroup before" 440 + " the RSGroup can be removed."); 441 } 442 for (TableDescriptor td : masterServices.getTableDescriptors().getAll().values()) { 443 if (td.getRegionServerGroup().map(groupName::equals).orElse(false)) { 444 throw new ConstraintException("RSGroup " + groupName + " is already referenced by " 445 + td.getTableName() + "; you must remove all the tables from the RSGroup before " 446 + "the RSGroup can be removed."); 447 } 448 } 449 for (NamespaceDescriptor ns : masterServices.getClusterSchema().getNamespaces()) { 450 String nsGroup = ns.getConfigurationValue(RSGroupInfo.NAMESPACE_DESC_PROP_GROUP); 451 if (nsGroup != null && nsGroup.equals(groupName)) { 452 throw new ConstraintException( 453 "RSGroup " + groupName + " is referenced by namespace: " + ns.getName()); 454 } 455 } 456 Map<String, RSGroupInfo> rsGroupMap = holder.groupName2Group; 457 if (!rsGroupMap.containsKey(groupName) || groupName.equals(RSGroupInfo.DEFAULT_GROUP)) { 458 throw new ConstraintException( 459 "Group " + groupName + " does not exist or is a reserved " + "group"); 460 } 461 Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(rsGroupMap); 462 newGroupMap.remove(groupName); 463 flushConfig(newGroupMap); 464 LOG.info("Remove group {} done", groupName); 465 } 466 467 @Override 468 public List<RSGroupInfo> listRSGroups() { 469 return Lists.newArrayList(holder.groupName2Group.values()); 470 } 471 472 @Override 473 public boolean isOnline() { 474 return rsGroupStartupWorker.isOnline(); 475 } 476 477 @Override 478 public synchronized void removeServers(Set<Address> servers) throws IOException { 479 if (servers == null || servers.isEmpty()) { 480 throw new ConstraintException("The set of servers to remove cannot be null or empty."); 481 } 482 483 // check the set of servers 484 checkForDeadOrOnlineServers(servers); 485 486 Map<String, RSGroupInfo> rsGroupInfos = new HashMap<String, RSGroupInfo>(); 487 for (Address el : servers) { 488 RSGroupInfo rsGroupInfo = getRSGroupOfServer(el); 489 if (rsGroupInfo != null) { 490 RSGroupInfo newRsGroupInfo = rsGroupInfos.get(rsGroupInfo.getName()); 491 if (newRsGroupInfo == null) { 492 newRsGroupInfo = copyOf(rsGroupInfo); 493 } 494 newRsGroupInfo.removeServer(el); 495 rsGroupInfos.put(newRsGroupInfo.getName(), newRsGroupInfo); 496 } else { 497 LOG.warn("Server " + el + " does not belong to any rsgroup."); 498 } 499 } 500 501 if (rsGroupInfos.size() > 0) { 502 Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(holder.groupName2Group); 503 newGroupMap.putAll(rsGroupInfos); 504 flushConfig(newGroupMap); 505 } 506 LOG.info("Remove decommissioned servers {} from RSGroup done", servers); 507 } 508 509 private List<RSGroupInfo> retrieveGroupListFromGroupTable() throws IOException { 510 List<RSGroupInfo> rsGroupInfoList = Lists.newArrayList(); 511 AsyncTable<?> table = conn.getTable(RSGROUP_TABLE_NAME); 512 try (ResultScanner scanner = table.getScanner(META_FAMILY_BYTES, META_QUALIFIER_BYTES)) { 513 for (Result result;;) { 514 result = scanner.next(); 515 if (result == null) { 516 break; 517 } 518 RSGroupProtos.RSGroupInfo proto = RSGroupProtos.RSGroupInfo 519 .parseFrom(result.getValue(META_FAMILY_BYTES, META_QUALIFIER_BYTES)); 520 rsGroupInfoList.add(ProtobufUtil.toGroupInfo(proto)); 521 } 522 } 523 return rsGroupInfoList; 524 } 525 526 private List<RSGroupInfo> retrieveGroupListFromZookeeper() throws IOException { 527 String groupBasePath = ZNodePaths.joinZNode(watcher.getZNodePaths().baseZNode, RS_GROUP_ZNODE); 528 List<RSGroupInfo> RSGroupInfoList = Lists.newArrayList(); 529 // Overwrite any info stored by table, this takes precedence 530 try { 531 if (ZKUtil.checkExists(watcher, groupBasePath) != -1) { 532 List<String> children = ZKUtil.listChildrenAndWatchForNewChildren(watcher, groupBasePath); 533 if (children == null) { 534 return RSGroupInfoList; 535 } 536 for (String znode : children) { 537 byte[] data = ZKUtil.getData(watcher, ZNodePaths.joinZNode(groupBasePath, znode)); 538 if (data != null && data.length > 0) { 539 ProtobufUtil.expectPBMagicPrefix(data); 540 ByteArrayInputStream bis = 541 new ByteArrayInputStream(data, ProtobufUtil.lengthOfPBMagic(), data.length); 542 RSGroupInfoList.add(ProtobufUtil.toGroupInfo(RSGroupProtos.RSGroupInfo.parseFrom(bis))); 543 } 544 } 545 LOG.debug("Read ZK GroupInfo count:" + RSGroupInfoList.size()); 546 } 547 } catch (KeeperException | DeserializationException | InterruptedException e) { 548 throw new IOException("Failed to read rsGroupZNode", e); 549 } 550 return RSGroupInfoList; 551 } 552 553 private void migrate(Collection<RSGroupInfo> groupList) { 554 TableDescriptors tds = masterServices.getTableDescriptors(); 555 ProcedureExecutor<MasterProcedureEnv> procExec = masterServices.getMasterProcedureExecutor(); 556 for (RSGroupInfo groupInfo : groupList) { 557 if (groupInfo.getName().equals(RSGroupInfo.DEFAULT_GROUP)) { 558 continue; 559 } 560 SortedSet<TableName> failedTables = new TreeSet<>(); 561 List<MigrateRSGroupProcedure> procs = new ArrayList<>(); 562 for (TableName tableName : groupInfo.getTables()) { 563 LOG.debug("Migrating {} in group {}", tableName, groupInfo.getName()); 564 TableDescriptor oldTd; 565 try { 566 oldTd = tds.get(tableName); 567 } catch (IOException e) { 568 LOG.warn("Failed to migrate {} in group {}", tableName, groupInfo.getName(), e); 569 failedTables.add(tableName); 570 continue; 571 } 572 if (oldTd == null) { 573 continue; 574 } 575 if (oldTd.getRegionServerGroup().isPresent()) { 576 // either we have already migrated it or that user has set the rs group using the new 577 // code which will set the group directly on table descriptor, skip. 578 LOG.debug("Skip migrating {} since it is already in group {}", tableName, 579 oldTd.getRegionServerGroup().get()); 580 continue; 581 } 582 // This is a bit tricky. Since we know that the region server group config in 583 // TableDescriptor will only be used at master side, it is fine to just update the table 584 // descriptor on file system and also the cache, without reopening all the regions. This 585 // will be much faster than the normal modifyTable. And when upgrading, we will update 586 // master first and then region server, so after all the region servers has been reopened, 587 // the new TableDescriptor will be loaded. 588 MigrateRSGroupProcedure proc = 589 new MigrateRSGroupProcedure(procExec.getEnvironment(), tableName); 590 procExec.submitProcedure(proc); 591 procs.add(proc); 592 } 593 for (MigrateRSGroupProcedure proc : procs) { 594 try { 595 ProcedureSyncWait.waitForProcedureToComplete(procExec, proc, 60000); 596 } catch (IOException e) { 597 LOG.warn("Failed to migrate rs group {} for table {}", groupInfo.getName(), 598 proc.getTableName()); 599 failedTables.add(proc.getTableName()); 600 } 601 } 602 LOG.debug("Done migrating {}, failed tables {}", groupInfo.getName(), failedTables); 603 synchronized (RSGroupInfoManagerImpl.this) { 604 Map<String, RSGroupInfo> rsGroupMap = holder.groupName2Group; 605 RSGroupInfo currentInfo = rsGroupMap.get(groupInfo.getName()); 606 if (currentInfo != null) { 607 RSGroupInfo newInfo = 608 new RSGroupInfo(currentInfo.getName(), currentInfo.getServers(), failedTables); 609 Map<String, RSGroupInfo> newGroupMap = new HashMap<>(rsGroupMap); 610 newGroupMap.put(groupInfo.getName(), newInfo); 611 try { 612 flushConfig(newGroupMap); 613 } catch (IOException e) { 614 LOG.warn("Failed to persist rs group {}", newInfo.getName(), e); 615 } 616 } 617 } 618 } 619 } 620 621 // Migrate the table rs group info from RSGroupInfo into the table descriptor 622 // Notice that we do not want to block the initialize so this will be done in background, and 623 // during the migrating, the rs group info maybe incomplete and cause region to be misplaced. 624 private void migrate() { 625 Thread migrateThread = new Thread(MIGRATE_THREAD_NAME) { 626 627 @Override 628 public void run() { 629 LOG.info("Start migrating table rs group config"); 630 while (!masterServices.isStopped()) { 631 Collection<RSGroupInfo> groups = holder.groupName2Group.values(); 632 boolean hasTables = groups.stream().anyMatch(r -> !r.getTables().isEmpty()); 633 if (!hasTables) { 634 break; 635 } 636 migrate(groups); 637 } 638 LOG.info("Done migrating table rs group info"); 639 } 640 }; 641 migrateThread.setDaemon(true); 642 migrateThread.start(); 643 } 644 645 /** 646 * Read rsgroup info from the source of truth, the hbase:rsgroup table. Update zk cache. Called on 647 * startup of the manager. 648 */ 649 private synchronized void refresh(boolean forceOnline) throws IOException { 650 LOG.debug("Refreshing RSGroup info from source of truth: forceOnline={}, isOnline={}", 651 forceOnline, isOnline()); 652 List<RSGroupInfo> groupList = new ArrayList<>(); 653 654 // Overwrite anything read from zk, group table is source of truth 655 // if online read from GROUP table 656 if (forceOnline || isOnline()) { 657 LOG.debug("Refreshing in Online mode."); 658 groupList.addAll(retrieveGroupListFromGroupTable()); 659 } else { 660 LOG.debug("Refreshing in Offline mode."); 661 groupList.addAll(retrieveGroupListFromZookeeper()); 662 } 663 664 // This is added to the last of the list so it overwrites the 'default' rsgroup loaded 665 // from region group table or zk 666 groupList.add(new RSGroupInfo(RSGroupInfo.DEFAULT_GROUP)); 667 668 // populate the data 669 HashMap<String, RSGroupInfo> newGroupMap = Maps.newHashMap(); 670 for (RSGroupInfo group : groupList) { 671 newGroupMap.put(group.getName(), group); 672 } 673 // Server membership for 'default' and for every regex-governed group is always recomputed 674 // on the fly -- never persisted nor trusted from storage. 675 Map<String, SortedSet<Address>> autoManagedServers = 676 computeAutoManagedRSGroupServers(groupList); 677 for (Map.Entry<String, SortedSet<Address>> entry : autoManagedServers.entrySet()) { 678 RSGroupInfo stored = newGroupMap.get(entry.getKey()); 679 if (!RSGroupInfo.DEFAULT_GROUP.equals(entry.getKey()) && !stored.getServers().isEmpty()) { 680 LOG.info( 681 "Ignoring servers {} stored for regex-governed RSGroup '{}'; its membership is " 682 + "computed on the fly and the stored copy is cleared on the next flush", 683 stored.getServers(), entry.getKey()); 684 } 685 } 686 applyAutoManagedRSGroupServers(newGroupMap, autoManagedServers); 687 dropRegexGovernedServersFromOtherGroups(newGroupMap, autoManagedServers); 688 resetRSGroupMap(newGroupMap); 689 updateCacheOfRSGroups(newGroupMap.keySet()); 690 LOG.info("Refresh completed successfully"); 691 } 692 693 private void flushConfigTable(Map<String, RSGroupInfo> groupMap) throws IOException { 694 List<Mutation> mutations = Lists.newArrayList(); 695 696 // populate deletes 697 for (String groupName : prevRSGroups) { 698 if (!groupMap.containsKey(groupName)) { 699 Delete d = new Delete(Bytes.toBytes(groupName)); 700 mutations.add(d); 701 } 702 } 703 704 // populate puts 705 for (RSGroupInfo gi : groupMap.values()) { 706 if (!gi.getName().equals(RSGroupInfo.DEFAULT_GROUP)) { 707 RSGroupProtos.RSGroupInfo proto = ProtobufUtil.toProtoGroupInfo(gi); 708 Put p = new Put(Bytes.toBytes(gi.getName())); 709 p.addColumn(META_FAMILY_BYTES, META_QUALIFIER_BYTES, proto.toByteArray()); 710 mutations.add(p); 711 } 712 } 713 714 if (mutations.size() > 0) { 715 multiMutate(mutations); 716 } 717 } 718 719 private synchronized void flushConfig() throws IOException { 720 flushConfig(holder.groupName2Group); 721 } 722 723 private synchronized void flushConfig(Map<String, RSGroupInfo> newGroupMap) throws IOException { 724 // For offline mode persistence is still unavailable 725 // We're refreshing in-memory state but only for servers in default group 726 if (!isOnline()) { 727 if (newGroupMap == holder.groupName2Group) { 728 // When newGroupMap is this.rsGroupMap itself, 729 // do not need to check default group and other groups as followed 730 return; 731 } 732 733 LOG.debug("Offline mode, cannot persist to {}", RSGROUP_TABLE_NAME); 734 735 Map<String, RSGroupInfo> oldGroupMap = Maps.newHashMap(holder.groupName2Group); 736 RSGroupInfo oldDefaultGroup = oldGroupMap.remove(RSGroupInfo.DEFAULT_GROUP); 737 RSGroupInfo newDefaultGroup = newGroupMap.remove(RSGroupInfo.DEFAULT_GROUP); 738 // Servers of regex-governed groups are computed on the fly and never persisted, so their 739 // changes are allowed here too. 740 if ( 741 !withoutRegexGovernedServers(oldGroupMap).equals(withoutRegexGovernedServers(newGroupMap)) 742 /* compare both tables and servers in other groups */ || !oldDefaultGroup.getTables() 743 .equals(newDefaultGroup.getTables()) 744 /* compare tables in default group */ 745 ) { 746 throw new IOException("Only servers in the default group and in regex-governed groups can " 747 + "be updated during offline mode"); 748 } 749 750 // Restore newGroupMap by putting its default group back 751 newGroupMap.put(RSGroupInfo.DEFAULT_GROUP, newDefaultGroup); 752 753 // Refresh rsGroupMap 754 // according to the inputted newGroupMap (an updated copy of rsGroupMap) 755 this.holder = new RSGroupInfoHolder(newGroupMap); 756 757 LOG.debug("New RSGroup map: {}", newGroupMap); 758 759 // Do not need to update tableMap 760 // because only the update on servers in default group is allowed above, 761 // or IOException will be thrown 762 return; 763 } 764 765 /* For online mode, persist to hbase:rsgroup and Zookeeper */ 766 LOG.debug("Online mode, persisting to {} and ZK", RSGROUP_TABLE_NAME); 767 // Servers of regex-governed groups are computed on the fly and never persisted 768 Map<String, RSGroupInfo> persistedGroupMap = withoutRegexGovernedServers(newGroupMap); 769 flushConfigTable(persistedGroupMap); 770 771 // Make changes visible after having been persisted to the source of truth 772 resetRSGroupMap(newGroupMap); 773 saveRSGroupMapToZK(persistedGroupMap); 774 updateCacheOfRSGroups(newGroupMap.keySet()); 775 LOG.info("Flush config done, new RSGroup map: {}", newGroupMap); 776 } 777 778 /** 779 * Returns a copy of {@code groupMap} where every regex-governed group has an empty server set. 780 * Their membership is recomputed from the live servers (see {@link #refresh}), so persisting it 781 * would only add I/O and leave stale data behind. 782 */ 783 private Map<String, RSGroupInfo> withoutRegexGovernedServers(Map<String, RSGroupInfo> groupMap) { 784 Set<String> regexGovernedGroups = getRegexGroupMap(masterServices.getConfiguration()).keySet(); 785 Map<String, RSGroupInfo> result = Maps.newHashMap(groupMap); 786 for (String groupName : regexGovernedGroups) { 787 RSGroupInfo info = groupMap.get(groupName); 788 if (info == null || info.getServers().isEmpty()) { 789 continue; 790 } 791 RSGroupInfo stripped = new RSGroupInfo(groupName); 792 stripped.addAllTables(info.getTables()); 793 info.getConfiguration().forEach(stripped::setConfiguration); 794 result.put(groupName, stripped); 795 } 796 return result; 797 } 798 799 private void saveRSGroupMapToZK(Map<String, RSGroupInfo> newGroupMap) throws IOException { 800 LOG.debug("Saving RSGroup info to ZK"); 801 try { 802 String groupBasePath = 803 ZNodePaths.joinZNode(watcher.getZNodePaths().baseZNode, RS_GROUP_ZNODE); 804 ZKUtil.createAndFailSilent(watcher, groupBasePath, ProtobufMagic.PB_MAGIC); 805 806 List<ZKUtil.ZKUtilOp> zkOps = new ArrayList<>(newGroupMap.size()); 807 for (String groupName : prevRSGroups) { 808 if (!newGroupMap.containsKey(groupName)) { 809 String znode = ZNodePaths.joinZNode(groupBasePath, groupName); 810 zkOps.add(ZKUtil.ZKUtilOp.deleteNodeFailSilent(znode)); 811 } 812 } 813 814 for (RSGroupInfo gi : newGroupMap.values()) { 815 if (!gi.getName().equals(RSGroupInfo.DEFAULT_GROUP)) { 816 String znode = ZNodePaths.joinZNode(groupBasePath, gi.getName()); 817 RSGroupProtos.RSGroupInfo proto = ProtobufUtil.toProtoGroupInfo(gi); 818 LOG.debug("Updating znode: " + znode); 819 ZKUtil.createAndFailSilent(watcher, znode); 820 zkOps.add(ZKUtil.ZKUtilOp.deleteNodeFailSilent(znode)); 821 zkOps.add(ZKUtil.ZKUtilOp.createAndFailSilent(znode, 822 ProtobufUtil.prependPBMagic(proto.toByteArray()))); 823 } 824 } 825 LOG.debug("Writing ZK GroupInfo count: " + zkOps.size()); 826 827 ZKUtil.multiOrSequential(watcher, zkOps, false); 828 } catch (KeeperException e) { 829 LOG.error("Failed to write to rsGroupZNode", e); 830 masterServices.abort("Failed to write to rsGroupZNode", e); 831 throw new IOException("Failed to write to rsGroupZNode", e); 832 } 833 } 834 835 /** 836 * Make changes visible. Caller must be synchronized on 'this'. 837 */ 838 private void resetRSGroupMap(Map<String, RSGroupInfo> newRSGroupMap) { 839 this.holder = new RSGroupInfoHolder(newRSGroupMap); 840 } 841 842 /** 843 * Update cache of rsgroups. Caller must be synchronized on 'this'. 844 * @param currentGroups Current list of Groups. 845 */ 846 private void updateCacheOfRSGroups(final Set<String> currentGroups) { 847 this.prevRSGroups.clear(); 848 this.prevRSGroups.addAll(currentGroups); 849 } 850 851 /** 852 * Parses {@code hbase.rsgroup.regex.<groupname>=<regex>} entries out of {@code conf}. Entries 853 * with an invalid group name, targeting the reserved {@link RSGroupInfo#DEFAULT_GROUP}, or with 854 * an unparsable regex are logged and skipped. 855 */ 856 static Map<String, Pattern> getRegexGroupMap(Configuration conf) { 857 Map<String, String> rsGroupNameToRegexMap = conf.getPropsWithPrefix(RS_GROUP_REGEX_PREFIX); 858 Map<String, Pattern> rsGroupNameToPatternMap = new HashMap<>(); 859 for (Map.Entry<String, String> e : rsGroupNameToRegexMap.entrySet()) { 860 if (e.getValue() == null) { 861 // getPropsWithPrefix reads names and values separately, so an entry removed by a concurrent 862 // configuration reload can show up here with a null value. Treat it as unset. 863 continue; 864 } 865 if (!GROUP_NAME_PATTERN.matcher(e.getKey()).matches()) { 866 LOG.warn("Ignoring {}{} -- '{}' is not a valid RSGroup name (it must match {})", 867 RS_GROUP_REGEX_PREFIX, e.getKey(), e.getKey(), GROUP_NAME_PATTERN.pattern()); 868 continue; 869 } 870 if (RSGroupInfo.DEFAULT_GROUP.equals(e.getKey())) { 871 LOG.warn("Ignoring {}{} -- regex-based membership cannot target the reserved '{}' group", 872 RS_GROUP_REGEX_PREFIX, e.getKey(), RSGroupInfo.DEFAULT_GROUP); 873 continue; 874 } 875 try { 876 rsGroupNameToPatternMap.put(e.getKey(), Pattern.compile(e.getValue())); 877 } catch (PatternSyntaxException ex) { 878 LOG.warn("Invalid regex '{}' for RSGroup '{}' ({}{}); ignoring this entry", e.getValue(), 879 e.getKey(), RS_GROUP_REGEX_PREFIX, e.getKey()); 880 } 881 } 882 LOG.debug("Resolved regex-based RSGroup membership config: {}", rsGroupNameToPatternMap); 883 return rsGroupNameToPatternMap; 884 } 885 886 private static List<String> getMatchingRSGroupNames(String hostname, 887 Map<String, Pattern> rsGroupNameToPatternMap) { 888 List<String> rsGroupNames = new ArrayList<>(); 889 for (Map.Entry<String, Pattern> e : rsGroupNameToPatternMap.entrySet()) { 890 if (e.getValue().matcher(hostname).matches()) { 891 rsGroupNames.add(e.getKey()); 892 } 893 } 894 return rsGroupNames; 895 } 896 897 /** 898 * Resolves every server in {@code onlineServers} whose hostname unambiguously matches exactly one 899 * entry of {@code rsGroupNameToPatternMap} naming a group in {@code existingGroupNames}. A server 900 * matching multiple regexes, or matching a regex for a group that does not (yet) exist, is 901 * omitted (warn-only) rather than resolved. 902 */ 903 static Map<Address, String> resolveServerAddrToRSGroupName(Set<Address> onlineServers, 904 Map<String, Pattern> rsGroupNameToPatternMap, Set<String> existingGroupNames) { 905 Map<Address, String> serverAddrToRSGroupName = new HashMap<>(); 906 if (rsGroupNameToPatternMap.isEmpty()) { 907 return serverAddrToRSGroupName; 908 } 909 Set<String> nonExistingRSGroupNames = new HashSet<>(); 910 for (Address server : onlineServers) { 911 List<String> matchingRSGroupNames = 912 getMatchingRSGroupNames(server.getHostName(), rsGroupNameToPatternMap); 913 if (matchingRSGroupNames.isEmpty()) { 914 continue; 915 } 916 if (matchingRSGroupNames.size() > 1) { 917 LOG.warn("Server {} hostname matches regexes for multiple RSGroups {}; treating it as " 918 + "unmatched by regex (falls back to default) until the overlapping " 919 + "hbase.rsgroup.regex.* entries are fixed", server, matchingRSGroupNames); 920 continue; 921 } 922 String rsGroupName = matchingRSGroupNames.get(0); 923 if (existingGroupNames.contains(rsGroupName)) { 924 serverAddrToRSGroupName.put(server, rsGroupName); 925 } else if (nonExistingRSGroupNames.add(rsGroupName)) { 926 LOG.warn( 927 "Config {}{} matches server {} but RSGroup '{}' does not exist -- create it with " 928 + "addRSGroup first; treating this server as unmatched by regex until then", 929 RS_GROUP_REGEX_PREFIX, rsGroupName, server, rsGroupName); 930 } 931 } 932 LOG.debug("Resolved server address to RSGroup name map: {}", serverAddrToRSGroupName); 933 return serverAddrToRSGroupName; 934 } 935 936 /** Returns {@code true} if every online server is either admin-managed or regex-matched. */ 937 static boolean wouldEmptyDefaultGroup(Set<Address> onlineServers, 938 Set<Address> adminManagedServers, Set<Address> regexMatchedServers) { 939 for (Address server : onlineServers) { 940 if (!adminManagedServers.contains(server) && !regexMatchedServers.contains(server)) { 941 return false; 942 } 943 } 944 return !onlineServers.isEmpty(); 945 } 946 947 static final class RegexBasedRSGroupMembershipResolution { 948 final Set<Address> adminManagedServers; 949 final Map<Address, String> regexMatchedServers; 950 951 RegexBasedRSGroupMembershipResolution(Set<Address> adminManagedServers, 952 Map<Address, String> regexMatchedServers) { 953 this.adminManagedServers = adminManagedServers; 954 this.regexMatchedServers = regexMatchedServers; 955 } 956 } 957 958 static RegexBasedRSGroupMembershipResolution resolveRegexBasedRSGroupMembership( 959 Map<String, Pattern> rsGroupNameToPatternMap, Set<Address> onlineServers, 960 Collection<RSGroupInfo> existingGroups) { 961 Set<String> existingGroupNames = 962 existingGroups.stream().map(RSGroupInfo::getName).collect(Collectors.toSet()); 963 Set<Address> adminManagedServers = new HashSet<>(); 964 for (RSGroupInfo existingGroup : existingGroups) { 965 if ( 966 !RSGroupInfo.DEFAULT_GROUP.equals(existingGroup.getName()) 967 && !rsGroupNameToPatternMap.containsKey(existingGroup.getName()) 968 ) { 969 adminManagedServers.addAll(existingGroup.getServers()); 970 } 971 } 972 Map<Address, String> regexMatchedServers = 973 resolveServerAddrToRSGroupName(onlineServers, rsGroupNameToPatternMap, existingGroupNames); 974 // A server stored in an admin-managed group that a regex now claims belongs to the regex group. 975 adminManagedServers.removeAll(regexMatchedServers.keySet()); 976 if (wouldEmptyDefaultGroup(onlineServers, adminManagedServers, regexMatchedServers.keySet())) { 977 // Regex placement is still applied: every server lands in its correct group. Operators are 978 // expected to configure regexes carefully, so this is a warning, not an error. 979 LOG.warn( 980 "Regex-based RSGroup membership leaves RSGroup '{}' with no online servers -- every " 981 + "online server is either admin-managed or matches a {}* entry. Tables in RSGroup '{}' " 982 + "(including system tables, unless they are in another RSGroup) have no server of " 983 + "their own to be assigned to; set {} to let them fall back to servers of other " 984 + "RSGroups", 985 RSGroupInfo.DEFAULT_GROUP, RS_GROUP_REGEX_PREFIX, RSGroupInfo.DEFAULT_GROUP, 986 RSGroupBasedLoadBalancer.FALLBACK_GROUP_ENABLE_KEY); 987 } 988 LOG.debug( 989 "Regex-based RSGroup membership resolution: onlineServers={}, " 990 + "adminManagedServers={}, regexMatchedServers={}", 991 onlineServers, adminManagedServers, regexMatchedServers); 992 return new RegexBasedRSGroupMembershipResolution(adminManagedServers, regexMatchedServers); 993 } 994 995 private Map<String, SortedSet<Address>> 996 computeAutoManagedRSGroupServers(Collection<RSGroupInfo> existingGroups) { 997 Set<Address> onlineServers = getOnlineServers(); 998 Map<String, Pattern> rsGroupNameToPatternMap = 999 getRegexGroupMap(masterServices.getConfiguration()); 1000 RegexBasedRSGroupMembershipResolution resolution = 1001 resolveRegexBasedRSGroupMembership(rsGroupNameToPatternMap, onlineServers, existingGroups); 1002 1003 Set<String> existingGroupNames = 1004 existingGroups.stream().map(RSGroupInfo::getName).collect(Collectors.toSet()); 1005 Map<String, SortedSet<Address>> result = new HashMap<>(); 1006 for (String rsGroupName : rsGroupNameToPatternMap.keySet()) { 1007 if (existingGroupNames.contains(rsGroupName)) { 1008 result.put(rsGroupName, new TreeSet<>()); 1009 } 1010 } 1011 result.put(RSGroupInfo.DEFAULT_GROUP, new TreeSet<>()); 1012 1013 for (Address server : onlineServers) { 1014 String regexGroupName = resolution.regexMatchedServers.get(server); 1015 if (regexGroupName != null) { 1016 result.get(regexGroupName).add(server); 1017 continue; 1018 } 1019 if (resolution.adminManagedServers.contains(server)) { 1020 continue; 1021 } 1022 result.get(RSGroupInfo.DEFAULT_GROUP).add(server); 1023 } 1024 for (Map.Entry<String, SortedSet<Address>> entry : result.entrySet()) { 1025 if (entry.getValue().isEmpty() && !RSGroupInfo.DEFAULT_GROUP.equals(entry.getKey())) { 1026 LOG.warn( 1027 "Regex-governed RSGroup '{}' has no online servers; any table assigned to it has " 1028 + "nowhere to go (check {}{} and the online servers' hostnames)", 1029 entry.getKey(), RS_GROUP_REGEX_PREFIX, entry.getKey()); 1030 } 1031 } 1032 LOG.debug("Computed auto-managed RSGroup server membership: {}", result); 1033 return result; 1034 } 1035 1036 private static void applyAutoManagedRSGroupServers(Map<String, RSGroupInfo> groupMap, 1037 Map<String, SortedSet<Address>> newRSGroupToServers) { 1038 for (Map.Entry<String, SortedSet<Address>> entry : newRSGroupToServers.entrySet()) { 1039 RSGroupInfo oldInfo = groupMap.get(entry.getKey()); 1040 if (oldInfo == null) { 1041 continue; 1042 } 1043 RSGroupInfo newInfo = new RSGroupInfo(entry.getKey(), entry.getValue()); 1044 newInfo.addAllTables(oldInfo.getTables()); 1045 oldInfo.getConfiguration().forEach(newInfo::setConfiguration); 1046 groupMap.put(entry.getKey(), newInfo); 1047 } 1048 } 1049 1050 /** 1051 * Stored membership of admin-managed groups is not recomputed on refresh, so a server that was 1052 * placed in one of them before a regex claimed it would end up in two groups. Unlike the runtime 1053 * paths, refresh cannot reject this: it runs in the startup worker, which retries forever on 1054 * failure and would keep the manager offline. Instead, resolve it the way 1055 * {@link #computeAutoManagedRSGroupServers} already does -- the regex wins -- in memory only; 1056 * storage is corrected by the next flush. 1057 */ 1058 private static void dropRegexGovernedServersFromOtherGroups(Map<String, RSGroupInfo> groupMap, 1059 Map<String, SortedSet<Address>> autoManagedServers) { 1060 Map<Address, String> regexGovernedServers = new HashMap<>(); 1061 for (Map.Entry<String, SortedSet<Address>> entry : autoManagedServers.entrySet()) { 1062 if (!RSGroupInfo.DEFAULT_GROUP.equals(entry.getKey())) { 1063 entry.getValue().forEach(server -> regexGovernedServers.put(server, entry.getKey())); 1064 } 1065 } 1066 if (regexGovernedServers.isEmpty()) { 1067 return; 1068 } 1069 for (Map.Entry<String, RSGroupInfo> entry : groupMap.entrySet()) { 1070 RSGroupInfo info = entry.getValue(); 1071 if (autoManagedServers.containsKey(info.getName())) { 1072 continue; 1073 } 1074 Set<Address> overlap = new TreeSet<>(); 1075 for (Address server : info.getServers()) { 1076 if (regexGovernedServers.containsKey(server)) { 1077 overlap.add(server); 1078 } 1079 } 1080 if (overlap.isEmpty()) { 1081 continue; 1082 } 1083 LOG.warn( 1084 "Servers {} are stored in RSGroup '{}' but match a {}* entry; placing them in the " 1085 + "regex-governed RSGroup instead and dropping the stored copy", 1086 overlap, info.getName(), RS_GROUP_REGEX_PREFIX); 1087 SortedSet<Address> kept = new TreeSet<>(info.getServers()); 1088 kept.removeAll(overlap); 1089 RSGroupInfo newInfo = new RSGroupInfo(info.getName(), kept); 1090 newInfo.addAllTables(info.getTables()); 1091 info.getConfiguration().forEach(newInfo::setConfiguration); 1092 entry.setValue(newInfo); 1093 } 1094 } 1095 1096 private class ServerEventsListener implements ServerListener { 1097 private final Logger LOG = LoggerFactory.getLogger(ServerEventsListener.class); 1098 1099 @Override 1100 public void serverAdded(ServerName serverName) { 1101 LOG.debug("Server added: {}", serverName); 1102 RSGroupInfoManagerImpl.this.handleServerEvent(); 1103 } 1104 1105 @Override 1106 public void serverRemoved(ServerName serverName) { 1107 LOG.debug("Server removed: {}", serverName); 1108 RSGroupInfoManagerImpl.this.handleServerEvent(); 1109 } 1110 } 1111 1112 private class RSGroupStartupWorker extends Thread { 1113 private final Logger LOG = LoggerFactory.getLogger(RSGroupStartupWorker.class); 1114 private volatile boolean online = false; 1115 1116 RSGroupStartupWorker() { 1117 super(RSGroupStartupWorker.class.getName() + "-" + masterServices.getServerName()); 1118 setDaemon(true); 1119 } 1120 1121 @Override 1122 public void run() { 1123 if (waitForGroupTableOnline()) { 1124 LOG.info("GroupBasedLoadBalancer is now online"); 1125 } else { 1126 LOG.warn("Quit without making region group table online"); 1127 } 1128 } 1129 1130 private boolean waitForGroupTableOnline() { 1131 while (isMasterRunning(masterServices)) { 1132 try { 1133 TableStateManager tsm = masterServices.getTableStateManager(); 1134 if (!tsm.isTablePresent(RSGROUP_TABLE_NAME)) { 1135 createRSGroupTable(); 1136 } 1137 // try reading from the table 1138 FutureUtils.get(conn.getTable(RSGROUP_TABLE_NAME).get(new Get(ROW_KEY))); 1139 LOG.info("RSGroup table={} is online, refreshing cached information", RSGROUP_TABLE_NAME); 1140 RSGroupInfoManagerImpl.this.refresh(true); 1141 online = true; 1142 // flush any inconsistencies between ZK and HTable 1143 RSGroupInfoManagerImpl.this.flushConfig(); 1144 // migrate after we are online. 1145 migrate(); 1146 return true; 1147 } catch (Exception e) { 1148 LOG.warn("Failed to perform check", e); 1149 // 100ms is short so let's just ignore the interrupt 1150 Threads.sleepWithoutInterrupt(100); 1151 } 1152 } 1153 return false; 1154 } 1155 1156 private void createRSGroupTable() throws IOException { 1157 OptionalLong optProcId = masterServices.getProcedures().stream() 1158 .filter(p -> p instanceof CreateTableProcedure).map(p -> (CreateTableProcedure) p) 1159 .filter(p -> p.getTableName().equals(RSGROUP_TABLE_NAME)).mapToLong(Procedure::getProcId) 1160 .findFirst(); 1161 long procId; 1162 if (optProcId.isPresent()) { 1163 procId = optProcId.getAsLong(); 1164 } else { 1165 LOG.debug("Creating group table {}", RSGROUP_TABLE_NAME); 1166 procId = masterServices.createSystemTable(RSGROUP_TABLE_DESC); 1167 } 1168 // wait for region to be online 1169 int tries = 600; 1170 while ( 1171 !(masterServices.getMasterProcedureExecutor().isFinished(procId)) 1172 && masterServices.getMasterProcedureExecutor().isRunning() && tries > 0 1173 ) { 1174 try { 1175 Thread.sleep(100); 1176 } catch (InterruptedException e) { 1177 throw new IOException("Wait interrupted ", e); 1178 } 1179 tries--; 1180 } 1181 if (tries <= 0) { 1182 throw new IOException("Failed to create group table in a given time."); 1183 } else { 1184 Procedure<?> result = masterServices.getMasterProcedureExecutor().getResult(procId); 1185 if (result != null && result.isFailed()) { 1186 throw new IOException( 1187 "Failed to create group table. " + MasterProcedureUtil.unwrapRemoteIOException(result)); 1188 } 1189 } 1190 } 1191 1192 public boolean isOnline() { 1193 return online; 1194 } 1195 } 1196 1197 private static boolean isMasterRunning(MasterServices masterServices) { 1198 return !masterServices.isAborted() && !masterServices.isStopped(); 1199 } 1200 1201 private void multiMutate(List<Mutation> mutations) throws IOException { 1202 MutateRowsRequest.Builder builder = MutateRowsRequest.newBuilder(); 1203 for (Mutation mutation : mutations) { 1204 if (mutation instanceof Put) { 1205 builder 1206 .addMutationRequest(ProtobufUtil.toMutation(MutationProto.MutationType.PUT, mutation)); 1207 } else if (mutation instanceof Delete) { 1208 builder 1209 .addMutationRequest(ProtobufUtil.toMutation(MutationProto.MutationType.DELETE, mutation)); 1210 } else { 1211 throw new DoNotRetryIOException( 1212 "multiMutate doesn't support " + mutation.getClass().getName()); 1213 } 1214 } 1215 MutateRowsRequest request = builder.build(); 1216 AsyncTable<?> table = conn.getTable(RSGROUP_TABLE_NAME); 1217 LOG.debug("Multimutating {} with {} mutations", RSGROUP_TABLE_NAME, mutations.size()); 1218 FutureUtils.get(table.<MultiRowMutationService, MutateRowsResponse> coprocessorService( 1219 MultiRowMutationService::newStub, 1220 (stub, controller, done) -> stub.mutateRows(controller, request, done), ROW_KEY)); 1221 LOG.info("Multimutating {} with {} mutations done", RSGROUP_TABLE_NAME, mutations.size()); 1222 } 1223 1224 private void checkGroupName(String groupName) throws ConstraintException { 1225 if (!GROUP_NAME_PATTERN.matcher(groupName).matches()) { 1226 throw new ConstraintException( 1227 "RSGroup name '" + groupName + "' must match " + GROUP_NAME_PATTERN.pattern()); 1228 } 1229 } 1230 1231 @Override 1232 public RSGroupInfo getRSGroupForTable(TableName tableName) throws IOException { 1233 return holder.tableName2Group.get(tableName); 1234 } 1235 1236 /** 1237 * Check if the set of servers are belong to dead servers list or online servers list. 1238 * @param servers servers to remove 1239 */ 1240 private void checkForDeadOrOnlineServers(Set<Address> servers) throws IOException { 1241 // This ugliness is because we only have Address, not ServerName. 1242 Set<Address> onlineServers = new HashSet<>(); 1243 List<ServerName> drainingServers = masterServices.getServerManager().getDrainingServersList(); 1244 for (ServerName server : masterServices.getServerManager().getOnlineServers().keySet()) { 1245 // Only online but not decommissioned servers are really online 1246 if (!drainingServers.contains(server)) { 1247 onlineServers.add(server.getAddress()); 1248 } 1249 } 1250 1251 Set<Address> deadServers = new HashSet<>(); 1252 for (ServerName server : masterServices.getServerManager().getDeadServers().copyServerNames()) { 1253 deadServers.add(server.getAddress()); 1254 } 1255 1256 for (Address address : servers) { 1257 if (onlineServers.contains(address)) { 1258 throw new DoNotRetryIOException( 1259 "Server " + address + " is an online server, not allowed to remove."); 1260 } 1261 if (deadServers.contains(address)) { 1262 throw new DoNotRetryIOException("Server " + address + " is on the dead servers list," 1263 + " Maybe it will come back again, not allowed to remove."); 1264 } 1265 } 1266 } 1267 1268 private void checkOnlineServersOnly(Set<Address> servers) throws IOException { 1269 // This uglyness is because we only have Address, not ServerName. 1270 // Online servers are keyed by ServerName. 1271 Set<Address> onlineServers = new HashSet<>(); 1272 for (ServerName server : masterServices.getServerManager().getOnlineServers().keySet()) { 1273 onlineServers.add(server.getAddress()); 1274 } 1275 for (Address address : servers) { 1276 if (!onlineServers.contains(address)) { 1277 throw new DoNotRetryIOException( 1278 "Server " + address + " is not an online server in 'default' RSGroup."); 1279 } 1280 } 1281 } 1282 1283 /** Returns List of Regions associated with this <code>server</code>. */ 1284 private List<RegionInfo> getRegions(final Address server) { 1285 LinkedList<RegionInfo> regions = new LinkedList<>(); 1286 for (Map.Entry<RegionInfo, ServerName> el : masterServices.getAssignmentManager() 1287 .getRegionStates().getRegionAssignments().entrySet()) { 1288 if (el.getValue() == null) { 1289 continue; 1290 } 1291 1292 if (el.getValue().getAddress().equals(server)) { 1293 addRegion(regions, el.getKey()); 1294 } 1295 } 1296 for (RegionStateNode state : masterServices.getAssignmentManager().getRegionsInTransition()) { 1297 if ( 1298 state.getRegionLocation() != null && state.getRegionLocation().getAddress().equals(server) 1299 ) { 1300 addRegion(regions, state.getRegionInfo()); 1301 } 1302 } 1303 return regions; 1304 } 1305 1306 private void addRegion(final LinkedList<RegionInfo> regions, RegionInfo hri) { 1307 // If meta, move it last otherwise other unassigns fail because meta is not 1308 // online for them to update state in. This is dodgy. Needs to be made more 1309 // robust. See TODO below. 1310 if (hri.isMetaRegion()) { 1311 regions.addLast(hri); 1312 } else { 1313 regions.addFirst(hri); 1314 } 1315 } 1316 1317 /** 1318 * Move every region from servers which are currently located on these servers, but should not be 1319 * located there. 1320 * @param movedServers the servers that are moved to new group 1321 * @param srcGrpServers all servers in the source group, excluding the movedServers 1322 * @param targetGroupName the target group 1323 * @param sourceGroupName the source group 1324 * @throws IOException if moving the server and tables fail 1325 */ 1326 private void moveServerRegionsFromGroup(Set<Address> movedServers, Set<Address> srcGrpServers, 1327 String targetGroupName, String sourceGroupName) throws IOException { 1328 moveRegionsBetweenGroups(movedServers, srcGrpServers, targetGroupName, sourceGroupName, 1329 rs -> getRegions(rs), info -> { 1330 try { 1331 String groupName = RSGroupUtil.getRSGroupInfo(masterServices, this, info.getTable()) 1332 .map(RSGroupInfo::getName).orElse(RSGroupInfo.DEFAULT_GROUP); 1333 return groupName.equals(targetGroupName); 1334 } catch (IOException e) { 1335 LOG.warn("Failed to test group for region {} and target group {}", info, targetGroupName); 1336 return false; 1337 } 1338 }); 1339 } 1340 1341 private <T> void moveRegionsBetweenGroups(Set<T> regionsOwners, Set<Address> newRegionsOwners, 1342 String targetGroupName, String sourceGroupName, Function<T, List<RegionInfo>> getRegionsInfo, 1343 Function<RegionInfo, Boolean> validation) throws IOException { 1344 // Get server names corresponding to given Addresses 1345 List<ServerName> movedServerNames = new ArrayList<>(regionsOwners.size()); 1346 List<ServerName> srcGrpServerNames = new ArrayList<>(newRegionsOwners.size()); 1347 for (ServerName serverName : masterServices.getServerManager().getOnlineServers().keySet()) { 1348 // In case region move failed in previous attempt, regionsOwners and newRegionsOwners 1349 // can have the same servers. So for all servers below both conditions to be checked 1350 if (newRegionsOwners.contains(serverName.getAddress())) { 1351 srcGrpServerNames.add(serverName); 1352 } 1353 if (regionsOwners.contains(serverName.getAddress())) { 1354 movedServerNames.add(serverName); 1355 } 1356 } 1357 List<Pair<RegionInfo, Future<byte[]>>> assignmentFutures = new ArrayList<>(); 1358 int retry = 0; 1359 Set<String> failedRegions = new HashSet<>(); 1360 IOException toThrow = null; 1361 do { 1362 assignmentFutures.clear(); 1363 failedRegions.clear(); 1364 for (ServerName owner : movedServerNames) { 1365 // Get regions that are associated with this server and filter regions by group tables. 1366 for (RegionInfo region : getRegionsInfo.apply((T) owner.getAddress())) { 1367 if (!validation.apply(region)) { 1368 LOG.info("Moving region {}, which does not belong to RSGroup {}", 1369 region.getShortNameToLog(), targetGroupName); 1370 // Move region back to source RSGroup servers 1371 ServerName dest = 1372 masterServices.getLoadBalancer().randomAssignment(region, srcGrpServerNames); 1373 if (dest == null) { 1374 failedRegions.add(region.getRegionNameAsString()); 1375 continue; 1376 } 1377 RegionPlan rp = new RegionPlan(region, owner, dest); 1378 try { 1379 Future<byte[]> future = masterServices.getAssignmentManager().moveAsync(rp); 1380 assignmentFutures.add(Pair.newPair(region, future)); 1381 } catch (IOException ioe) { 1382 failedRegions.add(region.getRegionNameAsString()); 1383 LOG.debug("Move region {} failed, will retry, current retry time is {}", 1384 region.getShortNameToLog(), retry, ioe); 1385 toThrow = ioe; 1386 } 1387 } 1388 } 1389 } 1390 waitForRegionMovement(assignmentFutures, failedRegions, sourceGroupName, retry); 1391 if (failedRegions.isEmpty()) { 1392 LOG.info("All regions from {} are moved back to {}", movedServerNames, sourceGroupName); 1393 return; 1394 } else { 1395 try { 1396 wait(1000); 1397 } catch (InterruptedException e) { 1398 LOG.warn("Sleep interrupted", e); 1399 Thread.currentThread().interrupt(); 1400 } 1401 retry++; 1402 } 1403 } while ( 1404 !failedRegions.isEmpty() && retry <= masterServices.getConfiguration() 1405 .getInt(FAILED_MOVE_MAX_RETRY, DEFAULT_MAX_RETRY_VALUE) 1406 ); 1407 1408 // has up to max retry time or there are no more regions to move 1409 if (!failedRegions.isEmpty()) { 1410 // print failed moved regions, for later process conveniently 1411 String msg = String.format("move regions for group %s failed, failed regions: %s", 1412 sourceGroupName, failedRegions); 1413 LOG.error(msg); 1414 throw new DoNotRetryIOException( 1415 msg + ", just record the last failed region's cause, more details in server log", toThrow); 1416 } 1417 } 1418 1419 /** 1420 * Wait for all the region move to complete. Keep waiting for other region movement completion 1421 * even if some region movement fails. 1422 */ 1423 private void waitForRegionMovement(List<Pair<RegionInfo, Future<byte[]>>> regionMoveFutures, 1424 Set<String> failedRegions, String sourceGroupName, int retryCount) { 1425 LOG.info("Moving {} region(s) to group {}, current retry={}", regionMoveFutures.size(), 1426 sourceGroupName, retryCount); 1427 for (Pair<RegionInfo, Future<byte[]>> pair : regionMoveFutures) { 1428 try { 1429 pair.getSecond().get(); 1430 if ( 1431 masterServices.getAssignmentManager().getRegionStates().getRegionState(pair.getFirst()) 1432 .isFailedOpen() 1433 ) { 1434 failedRegions.add(pair.getFirst().getRegionNameAsString()); 1435 } 1436 } catch (InterruptedException e) { 1437 // Dont return form there lets wait for other regions to complete movement. 1438 failedRegions.add(pair.getFirst().getRegionNameAsString()); 1439 LOG.warn("Sleep interrupted", e); 1440 } catch (Exception e) { 1441 failedRegions.add(pair.getFirst().getRegionNameAsString()); 1442 LOG.error("Move region {} to group {} failed, will retry on next attempt", 1443 pair.getFirst().getShortNameToLog(), sourceGroupName, e); 1444 } 1445 } 1446 } 1447 1448 private boolean isTableInGroup(TableName tableName, String groupName, 1449 Set<TableName> tablesInGroupCache) throws IOException { 1450 if (tablesInGroupCache.contains(tableName)) { 1451 return true; 1452 } 1453 if ( 1454 RSGroupUtil.getRSGroupInfo(masterServices, this, tableName).map(RSGroupInfo::getName) 1455 .orElse(RSGroupInfo.DEFAULT_GROUP).equals(groupName) 1456 ) { 1457 tablesInGroupCache.add(tableName); 1458 return true; 1459 } 1460 return false; 1461 } 1462 1463 private Map<String, RegionState> rsGroupGetRegionsInTransition(String groupName) 1464 throws IOException { 1465 Map<String, RegionState> rit = Maps.newTreeMap(); 1466 Set<TableName> tablesInGroupCache = new HashSet<>(); 1467 for (RegionStateNode regionNode : masterServices.getAssignmentManager() 1468 .getRegionsInTransition()) { 1469 TableName tn = regionNode.getTable(); 1470 if (isTableInGroup(tn, groupName, tablesInGroupCache)) { 1471 rit.put(regionNode.getRegionInfo().getEncodedName(), regionNode.toRegionState()); 1472 } 1473 } 1474 return rit; 1475 } 1476 1477 /** 1478 * This is an EXPENSIVE clone. Cloning though is the safest thing to do. Can't let out original 1479 * since it can change and at least the load balancer wants to iterate this exported list. Load 1480 * balancer should iterate over this list because cloned list will ignore disabled table and split 1481 * parent region cases. This method is invoked by {@link #balanceRSGroup} 1482 * @return A clone of current assignments for this group. 1483 */ 1484 Map<TableName, Map<ServerName, List<RegionInfo>>> getRSGroupAssignmentsByTable( 1485 TableStateManager tableStateManager, String groupName) throws IOException { 1486 Map<TableName, Map<ServerName, List<RegionInfo>>> result = Maps.newHashMap(); 1487 Set<TableName> tablesInGroupCache = new HashSet<>(); 1488 for (Map.Entry<RegionInfo, ServerName> entry : masterServices.getAssignmentManager() 1489 .getRegionStates().getRegionAssignments().entrySet()) { 1490 RegionInfo region = entry.getKey(); 1491 TableName tn = region.getTable(); 1492 ServerName server = entry.getValue(); 1493 if (isTableInGroup(tn, groupName, tablesInGroupCache)) { 1494 if ( 1495 tableStateManager.isTableState(tn, TableState.State.DISABLED, TableState.State.DISABLING) 1496 ) { 1497 continue; 1498 } 1499 if (region.isSplitParent()) { 1500 continue; 1501 } 1502 result.computeIfAbsent(tn, k -> new HashMap<>()) 1503 .computeIfAbsent(server, k -> new ArrayList<>()).add(region); 1504 } 1505 } 1506 RSGroupInfo rsGroupInfo = getRSGroupInfo(groupName); 1507 for (ServerName serverName : masterServices.getServerManager().getOnlineServers().keySet()) { 1508 if (rsGroupInfo.containsServer(serverName.getAddress())) { 1509 for (Map<ServerName, List<RegionInfo>> map : result.values()) { 1510 map.computeIfAbsent(serverName, k -> Collections.emptyList()); 1511 } 1512 } 1513 } 1514 return result; 1515 } 1516 1517 @Override 1518 public BalanceResponse balanceRSGroup(String groupName, BalanceRequest request) 1519 throws IOException { 1520 ServerManager serverManager = masterServices.getServerManager(); 1521 LoadBalancer balancer = masterServices.getLoadBalancer(); 1522 getRSGroupInfo(groupName); 1523 1524 BalanceResponse.Builder responseBuilder = BalanceResponse.newBuilder(); 1525 1526 synchronized (balancer) { 1527 // If balance not true, don't run balancer. 1528 if (!masterServices.isBalancerOn() && !request.isDryRun()) { 1529 return responseBuilder.build(); 1530 } 1531 1532 // Only allow one balance run at at time. 1533 Map<String, RegionState> groupRIT = rsGroupGetRegionsInTransition(groupName); 1534 if (groupRIT.size() > 0 && !request.isIgnoreRegionsInTransition()) { 1535 LOG.debug("Not running balancer because {} region(s) in transition: {}", groupRIT.size(), 1536 StringUtils.abbreviate( 1537 masterServices.getAssignmentManager().getRegionsInTransition().toString(), 256)); 1538 return responseBuilder.build(); 1539 } 1540 1541 if (serverManager.areDeadServersInProgress()) { 1542 LOG.debug("Not running balancer because processing dead regionserver(s): {}", 1543 serverManager.getDeadServers()); 1544 return responseBuilder.build(); 1545 } 1546 1547 // We balance per group instead of per table 1548 Map<TableName, Map<ServerName, List<RegionInfo>>> assignmentsByTable = 1549 getRSGroupAssignmentsByTable(masterServices.getTableStateManager(), groupName); 1550 List<RegionPlan> plans = balancer.balanceCluster(assignmentsByTable); 1551 boolean balancerRan = !plans.isEmpty(); 1552 1553 responseBuilder.setBalancerRan(balancerRan).setMovesCalculated(plans.size()); 1554 1555 if (balancerRan && !request.isDryRun()) { 1556 LOG.info("RSGroup balance {} starting with plan count: {}", groupName, plans.size()); 1557 List<RegionPlan> executed = masterServices.executeRegionPlansWithThrottling(plans); 1558 responseBuilder.setMovesExecuted(executed.size()); 1559 LOG.info("RSGroup balance " + groupName + " completed"); 1560 } 1561 1562 return responseBuilder.build(); 1563 } 1564 } 1565 1566 private void moveTablesAndWait(Set<TableName> tables, String targetGroup) throws IOException { 1567 LOG.debug("Moving {} tables to target group {}", tables.size(), targetGroup); 1568 List<Long> procIds = new ArrayList<Long>(); 1569 for (TableName tableName : tables) { 1570 TableDescriptor oldTd = masterServices.getTableDescriptors().get(tableName); 1571 if (oldTd == null) { 1572 continue; 1573 } 1574 TableDescriptor newTd = 1575 TableDescriptorBuilder.newBuilder(oldTd).setRegionServerGroup(targetGroup).build(); 1576 procIds.add( 1577 masterServices.modifyTable(tableName, newTd, HConstants.NO_NONCE, HConstants.NO_NONCE)); 1578 } 1579 for (long procId : procIds) { 1580 Procedure<?> proc = masterServices.getMasterProcedureExecutor().getProcedure(procId); 1581 if (proc == null) { 1582 continue; 1583 } 1584 ProcedureSyncWait.waitForProcedureToCompleteIOE(masterServices.getMasterProcedureExecutor(), 1585 proc, Long.MAX_VALUE); 1586 } 1587 LOG.info("Move tables done: moved {} tables to {}", tables.size(), targetGroup); 1588 if (LOG.isDebugEnabled()) { 1589 LOG.debug("Tables moved to {}: {}", targetGroup, tables); 1590 } 1591 } 1592 1593 @Override 1594 public void setRSGroup(Set<TableName> tables, String groupName) throws IOException { 1595 getRSGroupInfo(groupName); 1596 moveTablesAndWait(tables, groupName); 1597 } 1598 1599 public void moveServers(Set<Address> servers, String targetGroupName) throws IOException { 1600 if (servers == null) { 1601 throw new ConstraintException("The list of servers to move cannot be null."); 1602 } 1603 if (servers.isEmpty()) { 1604 // For some reason this difference between null servers and isEmpty is important distinction. 1605 // TODO. Why? Stuff breaks if I equate them. 1606 return; 1607 } 1608 if (StringUtils.isEmpty(targetGroupName)) { 1609 throw new ConstraintException("RSGroup cannot be null."); 1610 } 1611 1612 // Hold a lock on the manager instance while moving servers to prevent 1613 // another writer changing our state while we are working. 1614 synchronized (this) { 1615 // Presume first server's source group. Later ensure all servers are from this group. 1616 Address firstServer = servers.iterator().next(); 1617 RSGroupInfo srcGrp = getRSGroupOfServer(firstServer); 1618 if (srcGrp == null) { 1619 // Be careful. This exception message is tested for in TestRSGroupAdmin2... 1620 throw new ConstraintException( 1621 "Server " + firstServer + " is either offline or it does not exist."); 1622 } 1623 1624 // Only move online servers (when moving from 'default') or servers from other 1625 // groups. This prevents bogus servers from entering groups 1626 if (RSGroupInfo.DEFAULT_GROUP.equals(srcGrp.getName())) { 1627 if (srcGrp.getServers().size() <= servers.size()) { 1628 throw new ConstraintException(KEEP_ONE_SERVER_IN_DEFAULT_ERROR_MESSAGE); 1629 } 1630 checkOnlineServersOnly(servers); 1631 } 1632 // Ensure all servers are of same rsgroup. 1633 for (Address server : servers) { 1634 String tmpGroup = getRSGroupOfServer(server).getName(); 1635 if (!tmpGroup.equals(srcGrp.getName())) { 1636 throw new ConstraintException("Move server request should only come from one source " 1637 + "RSGroup. Expecting only " + srcGrp.getName() + " but contains " + tmpGroup); 1638 } 1639 } 1640 if (srcGrp.getServers().size() <= servers.size()) { 1641 // check if there are still tables reference this group 1642 for (TableDescriptor td : masterServices.getTableDescriptors().getAll().values()) { 1643 Optional<String> optGroupName = td.getRegionServerGroup(); 1644 if (optGroupName.isPresent() && optGroupName.get().equals(srcGrp.getName())) { 1645 throw new ConstraintException( 1646 "Cannot leave a RSGroup " + srcGrp.getName() + " that contains tables('" 1647 + td.getTableName() + "' at least) without servers to host them."); 1648 } 1649 } 1650 } 1651 1652 // MovedServers may be < passed in 'servers'. 1653 Set<Address> movedServers = moveServers(servers, srcGrp.getName(), targetGroupName); 1654 // moveServers (above) no longer mutates srcGrp in place -- re-fetch to see the post-move 1655 // remaining server set. 1656 RSGroupInfo refreshedSrcGrp = getRSGroupInfo(srcGrp.getName()); 1657 moveServerRegionsFromGroup(movedServers, refreshedSrcGrp.getServers(), targetGroupName, 1658 srcGrp.getName()); 1659 LOG.info("Move servers done: moved {} servers from {} to {}", movedServers.size(), 1660 srcGrp.getName(), targetGroupName); 1661 if (LOG.isDebugEnabled()) { 1662 LOG.debug("Servers moved from {} to {}: {}", srcGrp.getName(), targetGroupName, 1663 movedServers); 1664 } 1665 } 1666 } 1667 1668 @Override 1669 public String determineRSGroupInfoForTable(TableName tableName) { 1670 return script.getRSGroup(tableName.getNamespaceAsString(), tableName.getQualifierAsString()); 1671 } 1672 1673 @Override 1674 public synchronized void renameRSGroup(String oldName, String newName) throws IOException { 1675 if (oldName.equals(RSGroupInfo.DEFAULT_GROUP)) { 1676 throw new ConstraintException(RSGroupInfo.DEFAULT_GROUP + " can't be rename"); 1677 } 1678 checkGroupName(newName); 1679 // getRSGroupInfo validates old RSGroup existence. 1680 RSGroupInfo oldRSG = getRSGroupInfo(oldName); 1681 Map<String, RSGroupInfo> rsGroupMap = holder.groupName2Group; 1682 if (rsGroupMap.containsKey(newName)) { 1683 throw new ConstraintException("Group already exists: " + newName); 1684 } 1685 1686 Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(rsGroupMap); 1687 newGroupMap.remove(oldRSG.getName()); 1688 RSGroupInfo newRSG = new RSGroupInfo(newName, oldRSG.getServers()); 1689 newGroupMap.put(newName, newRSG); 1690 flushConfig(newGroupMap); 1691 Set<TableName> updateTables = masterServices.getTableDescriptors().getAll().values().stream() 1692 .filter(t -> oldName.equals(t.getRegionServerGroup().orElse(null))) 1693 .map(TableDescriptor::getTableName).collect(Collectors.toSet()); 1694 setRSGroup(updateTables, newName); 1695 LOG.info("Rename RSGroup done: {} => {}", oldName, newName); 1696 } 1697 1698 @Override 1699 public synchronized void updateRSGroupConfig(String groupName, Map<String, String> configuration) 1700 throws IOException { 1701 if (RSGroupInfo.DEFAULT_GROUP.equals(groupName)) { 1702 // We do not persist anything of default group, therefore, it is not supported to update 1703 // default group's configuration which lost once master down. 1704 throw new ConstraintException( 1705 "configuration of " + RSGroupInfo.DEFAULT_GROUP + " can't be stored persistently"); 1706 } 1707 RSGroupInfo rsGroupInfo = copyOf(getRSGroupInfo(groupName)); 1708 // Iterate over a snapshot of the keys, as removeConfiguration modifies the backing map. 1709 new ArrayList<>(rsGroupInfo.getConfiguration().keySet()) 1710 .forEach(rsGroupInfo::removeConfiguration); 1711 configuration.forEach((k, v) -> rsGroupInfo.setConfiguration(k, v)); 1712 Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(holder.groupName2Group); 1713 newGroupMap.put(groupName, rsGroupInfo); 1714 flushConfig(newGroupMap); 1715 } 1716}