001/** 002 * 003 * Licensed to the Apache Software Foundation (ASF) under one 004 * or more contributor license agreements. See the NOTICE file 005 * distributed with this work for additional information 006 * regarding copyright ownership. The ASF licenses this file 007 * to you under the Apache License, Version 2.0 (the 008 * "License"); you may not use this file except in compliance 009 * with the License. You may obtain a copy of the License at 010 * 011 * http://www.apache.org/licenses/LICENSE-2.0 012 * 013 * Unless required by applicable law or agreed to in writing, software 014 * distributed under the License is distributed on an "AS IS" BASIS, 015 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 016 * See the License for the specific language governing permissions and 017 * limitations under the License. 018 */ 019 020package org.apache.hadoop.hbase.util; 021 022import java.io.BufferedInputStream; 023import java.io.BufferedOutputStream; 024import java.io.Closeable; 025import java.io.DataInputStream; 026import java.io.DataOutputStream; 027import java.io.File; 028import java.io.FileInputStream; 029import java.io.FileOutputStream; 030import java.io.IOException; 031import java.nio.file.Files; 032import java.nio.file.Paths; 033import java.util.ArrayList; 034import java.util.Collections; 035import java.util.EnumSet; 036import java.util.Iterator; 037import java.util.List; 038import java.util.Locale; 039import java.util.concurrent.Callable; 040import java.util.concurrent.CancellationException; 041import java.util.concurrent.ExecutionException; 042import java.util.concurrent.ExecutorService; 043import java.util.concurrent.Executors; 044import java.util.concurrent.Future; 045import java.util.concurrent.TimeUnit; 046import java.util.concurrent.TimeoutException; 047import java.util.function.Predicate; 048 049import org.apache.commons.io.IOUtils; 050import org.apache.hadoop.conf.Configuration; 051import org.apache.hadoop.hbase.ClusterMetrics.Option; 052import org.apache.hadoop.hbase.HBaseConfiguration; 053import org.apache.hadoop.hbase.HConstants; 054import org.apache.hadoop.hbase.MetaTableAccessor; 055import org.apache.hadoop.hbase.ServerName; 056import org.apache.hadoop.hbase.client.Admin; 057import org.apache.hadoop.hbase.client.Connection; 058import org.apache.hadoop.hbase.client.ConnectionFactory; 059import org.apache.hadoop.hbase.client.RegionInfo; 060import org.apache.hadoop.hbase.client.ResultScanner; 061import org.apache.hadoop.hbase.client.Scan; 062import org.apache.hadoop.hbase.client.Table; 063import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter; 064import org.apache.yetus.audience.InterfaceAudience; 065import org.slf4j.Logger; 066import org.slf4j.LoggerFactory; 067 068import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting; 069import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLine; 070 071/** 072 * Tool for loading/unloading regions to/from given regionserver This tool can be run from Command 073 * line directly as a utility. Supports Ack/No Ack mode for loading/unloading operations.Ack mode 074 * acknowledges if regions are online after movement while noAck mode is best effort mode that 075 * improves performance but will still move on if region is stuck/not moved. Motivation behind noAck 076 * mode being RS shutdown where even if a Region is stuck, upon shutdown master will move it 077 * anyways. This can also be used by constructiong an Object using the builder and then calling 078 * {@link #load()} or {@link #unload()} methods for the desired operations. 079 */ 080@InterfaceAudience.Public 081public class RegionMover extends AbstractHBaseTool implements Closeable { 082 public static final String MOVE_RETRIES_MAX_KEY = "hbase.move.retries.max"; 083 public static final String MOVE_WAIT_MAX_KEY = "hbase.move.wait.max"; 084 public static final String SERVERSTART_WAIT_MAX_KEY = "hbase.serverstart.wait.max"; 085 public static final int DEFAULT_MOVE_RETRIES_MAX = 5; 086 public static final int DEFAULT_MOVE_WAIT_MAX = 60; 087 public static final int DEFAULT_SERVERSTART_WAIT_MAX = 180; 088 static final Logger LOG = LoggerFactory.getLogger(RegionMover.class); 089 private RegionMoverBuilder rmbuilder; 090 private boolean ack = true; 091 private int maxthreads = 1; 092 private int timeout; 093 private String loadUnload; 094 private String hostname; 095 private String filename; 096 private String excludeFile; 097 private int port; 098 private Connection conn; 099 private Admin admin; 100 101 private RegionMover(RegionMoverBuilder builder) throws IOException { 102 this.hostname = builder.hostname; 103 this.filename = builder.filename; 104 this.excludeFile = builder.excludeFile; 105 this.maxthreads = builder.maxthreads; 106 this.ack = builder.ack; 107 this.port = builder.port; 108 this.timeout = builder.timeout; 109 setConf(builder.conf); 110 this.conn = ConnectionFactory.createConnection(conf); 111 this.admin = conn.getAdmin(); 112 } 113 114 private RegionMover() { 115 } 116 117 @Override 118 public void close() { 119 IOUtils.closeQuietly(this.admin); 120 IOUtils.closeQuietly(this.conn); 121 } 122 123 /** 124 * Builder for Region mover. Use the {@link #build()} method to create RegionMover object. Has 125 * {@link #filename(String)}, {@link #excludeFile(String)}, {@link #maxthreads(int)}, 126 * {@link #ack(boolean)}, {@link #timeout(int)} methods to set the corresponding options 127 */ 128 public static class RegionMoverBuilder { 129 private boolean ack = true; 130 private int maxthreads = 1; 131 private int timeout = Integer.MAX_VALUE; 132 private String hostname; 133 private String filename; 134 private String excludeFile = null; 135 private String defaultDir = System.getProperty("java.io.tmpdir"); 136 @VisibleForTesting 137 final int port; 138 private final Configuration conf; 139 140 public RegionMoverBuilder(String hostname) { 141 this(hostname, createConf()); 142 } 143 144 /** 145 * Creates a new configuration and sets region mover specific overrides 146 */ 147 private static Configuration createConf() { 148 Configuration conf = HBaseConfiguration.create(); 149 conf.setInt("hbase.client.prefetch.limit", 1); 150 conf.setInt("hbase.client.pause", 500); 151 conf.setInt("hbase.client.retries.number", 100); 152 return conf; 153 } 154 155 /** 156 * @param hostname Hostname to unload regions from or load regions to. Can be either hostname 157 * or hostname:port. 158 * @param conf Configuration object 159 */ 160 public RegionMoverBuilder(String hostname, Configuration conf) { 161 String[] splitHostname = hostname.toLowerCase().split(":"); 162 this.hostname = splitHostname[0]; 163 if (splitHostname.length == 2) { 164 this.port = Integer.parseInt(splitHostname[1]); 165 } else { 166 this.port = conf.getInt(HConstants.REGIONSERVER_PORT, HConstants.DEFAULT_REGIONSERVER_PORT); 167 } 168 this.filename = defaultDir + File.separator + System.getProperty("user.name") + this.hostname 169 + ":" + Integer.toString(this.port); 170 this.conf = conf; 171 } 172 173 /** 174 * Path of file where regions will be written to during unloading/read from during loading 175 * @param filename 176 * @return RegionMoverBuilder object 177 */ 178 public RegionMoverBuilder filename(String filename) { 179 this.filename = filename; 180 return this; 181 } 182 183 /** 184 * Set the max number of threads that will be used to move regions 185 */ 186 public RegionMoverBuilder maxthreads(int threads) { 187 this.maxthreads = threads; 188 return this; 189 } 190 191 /** 192 * Path of file containing hostnames to be excluded during region movement. Exclude file should 193 * have 'host:port' per line. Port is mandatory here as we can have many RS running on a single 194 * host. 195 */ 196 public RegionMoverBuilder excludeFile(String excludefile) { 197 this.excludeFile = excludefile; 198 return this; 199 } 200 201 /** 202 * Set ack/noAck mode. 203 * <p> 204 * In ack mode regions are acknowledged before and after moving and the move is retried 205 * hbase.move.retries.max times, if unsuccessful we quit with exit code 1.No Ack mode is a best 206 * effort mode,each region movement is tried once.This can be used during graceful shutdown as 207 * even if we have a stuck region,upon shutdown it'll be reassigned anyway. 208 * <p> 209 * @param ack 210 * @return RegionMoverBuilder object 211 */ 212 public RegionMoverBuilder ack(boolean ack) { 213 this.ack = ack; 214 return this; 215 } 216 217 /** 218 * Set the timeout for Load/Unload operation in seconds.This is a global timeout,threadpool for 219 * movers also have a separate time which is hbase.move.wait.max * number of regions to 220 * load/unload 221 * @param timeout in seconds 222 * @return RegionMoverBuilder object 223 */ 224 public RegionMoverBuilder timeout(int timeout) { 225 this.timeout = timeout; 226 return this; 227 } 228 229 /** 230 * This method builds the appropriate RegionMover object which can then be used to load/unload 231 * using load and unload methods 232 * @return RegionMover object 233 */ 234 public RegionMover build() throws IOException { 235 return new RegionMover(this); 236 } 237 } 238 239 /** 240 * Move Regions and make sure that they are up on the target server.If a region movement fails we 241 * exit as failure 242 */ 243 private class MoveWithAck implements Callable<Boolean> { 244 private RegionInfo region; 245 private ServerName targetServer; 246 private List<RegionInfo> movedRegions; 247 private ServerName sourceServer; 248 249 public MoveWithAck(RegionInfo regionInfo, ServerName sourceServer, 250 ServerName targetServer, List<RegionInfo> movedRegions) { 251 this.region = regionInfo; 252 this.targetServer = targetServer; 253 this.movedRegions = movedRegions; 254 this.sourceServer = sourceServer; 255 } 256 257 @Override 258 public Boolean call() throws IOException, InterruptedException { 259 boolean moved = false; 260 int count = 0; 261 int retries = admin.getConfiguration().getInt(MOVE_RETRIES_MAX_KEY, DEFAULT_MOVE_RETRIES_MAX); 262 int maxWaitInSeconds = 263 admin.getConfiguration().getInt(MOVE_WAIT_MAX_KEY, DEFAULT_MOVE_WAIT_MAX); 264 long startTime = EnvironmentEdgeManager.currentTime(); 265 boolean sameServer = true; 266 // Assert we can scan the region in its current location 267 isSuccessfulScan(region); 268 LOG.info("Moving region:" + region.getEncodedName() + " from " + sourceServer + " to " 269 + targetServer); 270 while (count < retries && sameServer) { 271 if (count > 0) { 272 LOG.info("Retry " + Integer.toString(count) + " of maximum " + Integer.toString(retries)); 273 } 274 count = count + 1; 275 admin.move(region.getEncodedNameAsBytes(), Bytes.toBytes(targetServer.getServerName())); 276 long maxWait = startTime + (maxWaitInSeconds * 1000); 277 while (EnvironmentEdgeManager.currentTime() < maxWait) { 278 sameServer = isSameServer(region, sourceServer); 279 if (!sameServer) { 280 break; 281 } 282 Thread.sleep(100); 283 } 284 } 285 if (sameServer) { 286 LOG.error("Region: " + region.getRegionNameAsString() + " stuck on " + this.sourceServer 287 + ",newServer=" + this.targetServer); 288 } else { 289 isSuccessfulScan(region); 290 LOG.info("Moved Region " 291 + region.getRegionNameAsString() 292 + " cost:" 293 + String.format("%.3f", 294 (float) (EnvironmentEdgeManager.currentTime() - startTime) / 1000)); 295 moved = true; 296 movedRegions.add(region); 297 } 298 return moved; 299 } 300 } 301 302 /** 303 * Move Regions without Acknowledging.Usefule in case of RS shutdown as we might want to shut the 304 * RS down anyways and not abort on a stuck region. Improves movement performance 305 */ 306 private class MoveWithoutAck implements Callable<Boolean> { 307 private RegionInfo region; 308 private ServerName targetServer; 309 private List<RegionInfo> movedRegions; 310 private ServerName sourceServer; 311 312 public MoveWithoutAck(RegionInfo regionInfo, ServerName sourceServer, 313 ServerName targetServer, List<RegionInfo> movedRegions) { 314 this.region = regionInfo; 315 this.targetServer = targetServer; 316 this.movedRegions = movedRegions; 317 this.sourceServer = sourceServer; 318 } 319 320 @Override 321 public Boolean call() { 322 try { 323 LOG.info("Moving region:" + region.getEncodedName() + " from " + sourceServer + " to " 324 + targetServer); 325 admin.move(region.getEncodedNameAsBytes(), Bytes.toBytes(targetServer.getServerName())); 326 LOG.info("Moved " + region.getEncodedName() + " from " + sourceServer + " to " 327 + targetServer); 328 } catch (Exception e) { 329 LOG.error("Error Moving Region:" + region.getEncodedName(), e); 330 } finally { 331 // we add region to the moved regions list in No Ack Mode since this is best effort 332 movedRegions.add(region); 333 } 334 return true; 335 } 336 } 337 338 /** 339 * Loads the specified {@link #hostname} with regions listed in the {@link #filename} RegionMover 340 * Object has to be created using {@link #RegionMover(RegionMoverBuilder)} 341 * @return true if loading succeeded, false otherwise 342 */ 343 public boolean load() throws ExecutionException, InterruptedException, TimeoutException { 344 ExecutorService loadPool = Executors.newFixedThreadPool(1); 345 Future<Boolean> loadTask = loadPool.submit(() -> { 346 try { 347 List<RegionInfo> regionsToMove = readRegionsFromFile(filename); 348 if (regionsToMove.isEmpty()) { 349 LOG.info("No regions to load.Exiting"); 350 return true; 351 } 352 loadRegions(regionsToMove); 353 } catch (Exception e) { 354 LOG.error("Error while loading regions to " + hostname, e); 355 return false; 356 } 357 return true; 358 }); 359 return waitTaskToFinish(loadPool, loadTask, "loading"); 360 } 361 362 private void loadRegions(List<RegionInfo> regionsToMove) 363 throws Exception { 364 ServerName server = getTargetServer(); 365 List<RegionInfo> movedRegions = Collections.synchronizedList(new ArrayList<>()); 366 LOG.info( 367 "Moving " + regionsToMove.size() + " regions to " + server + " using " + this.maxthreads 368 + " threads.Ack mode:" + this.ack); 369 370 ExecutorService moveRegionsPool = Executors.newFixedThreadPool(this.maxthreads); 371 List<Future<Boolean>> taskList = new ArrayList<>(); 372 int counter = 0; 373 while (counter < regionsToMove.size()) { 374 RegionInfo region = regionsToMove.get(counter); 375 ServerName currentServer = getServerNameForRegion(region); 376 if (currentServer == null) { 377 LOG.warn( 378 "Could not get server for Region:" + region.getRegionNameAsString() + " moving on"); 379 counter++; 380 continue; 381 } else if (server.equals(currentServer)) { 382 LOG.info( 383 "Region " + region.getRegionNameAsString() + " is already on target server=" + server); 384 counter++; 385 continue; 386 } 387 if (ack) { 388 Future<Boolean> task = 389 moveRegionsPool.submit(new MoveWithAck(region, currentServer, server, movedRegions)); 390 taskList.add(task); 391 } else { 392 Future<Boolean> task = 393 moveRegionsPool.submit(new MoveWithoutAck(region, currentServer, server, movedRegions)); 394 taskList.add(task); 395 } 396 counter++; 397 } 398 399 moveRegionsPool.shutdown(); 400 long timeoutInSeconds = regionsToMove.size() * admin.getConfiguration() 401 .getLong(MOVE_WAIT_MAX_KEY, DEFAULT_MOVE_WAIT_MAX); 402 waitMoveTasksToFinish(moveRegionsPool, taskList, timeoutInSeconds); 403 } 404 405 /** 406 * Unload regions from given {@link #hostname} using ack/noAck mode and {@link #maxthreads}.In 407 * noAck mode we do not make sure that region is successfully online on the target region 408 * server,hence it is best effort.We do not unload regions to hostnames given in 409 * {@link #excludeFile}. 410 * @return true if unloading succeeded, false otherwise 411 */ 412 public boolean unload() throws InterruptedException, ExecutionException, TimeoutException { 413 deleteFile(this.filename); 414 ExecutorService unloadPool = Executors.newFixedThreadPool(1); 415 Future<Boolean> unloadTask = unloadPool.submit(() -> { 416 List<RegionInfo> movedRegions = Collections.synchronizedList(new ArrayList<>()); 417 try { 418 // Get Online RegionServers 419 List<ServerName> regionServers = new ArrayList<>(); 420 regionServers.addAll( 421 admin.getClusterMetrics(EnumSet.of(Option.LIVE_SERVERS)).getLiveServerMetrics() 422 .keySet()); 423 // Remove the host Region server from target Region Servers list 424 ServerName server = stripServer(regionServers, hostname, port); 425 // Remove RS present in the exclude file 426 stripExcludes(regionServers); 427 stripMaster(regionServers); 428 if (regionServers.isEmpty()) { 429 LOG.warn("No Regions were moved - no servers available"); 430 return false; 431 } 432 unloadRegions(server, regionServers, movedRegions); 433 } catch (Exception e) { 434 LOG.error("Error while unloading regions ", e); 435 return false; 436 } finally { 437 if (movedRegions != null) { 438 writeFile(filename, movedRegions); 439 } 440 } 441 return true; 442 }); 443 return waitTaskToFinish(unloadPool, unloadTask, "unloading"); 444 } 445 446 private void unloadRegions(ServerName server, List<ServerName> regionServers, 447 List<RegionInfo> movedRegions) throws Exception { 448 while (true) { 449 List<RegionInfo> regionsToMove = admin.getRegions(server); 450 regionsToMove.removeAll(movedRegions); 451 if (regionsToMove.isEmpty()) { 452 LOG.info("No Regions to move....Quitting now"); 453 break; 454 } 455 int counter = 0; 456 LOG.info("Moving " + regionsToMove.size() + " regions from " + this.hostname + " to " 457 + regionServers.size() + " servers using " + this.maxthreads + " threads .Ack Mode:" 458 + ack); 459 ExecutorService moveRegionsPool = Executors.newFixedThreadPool(this.maxthreads); 460 List<Future<Boolean>> taskList = new ArrayList<>(); 461 int serverIndex = 0; 462 while (counter < regionsToMove.size()) { 463 if (ack) { 464 Future<Boolean> task = moveRegionsPool.submit( 465 new MoveWithAck(regionsToMove.get(counter), server, regionServers.get(serverIndex), 466 movedRegions)); 467 taskList.add(task); 468 } else { 469 Future<Boolean> task = moveRegionsPool.submit( 470 new MoveWithoutAck(regionsToMove.get(counter), server, regionServers.get(serverIndex), 471 movedRegions)); 472 taskList.add(task); 473 } 474 counter++; 475 serverIndex = (serverIndex + 1) % regionServers.size(); 476 } 477 moveRegionsPool.shutdown(); 478 long timeoutInSeconds = regionsToMove.size() * admin.getConfiguration() 479 .getLong(MOVE_WAIT_MAX_KEY, DEFAULT_MOVE_WAIT_MAX); 480 waitMoveTasksToFinish(moveRegionsPool, taskList, timeoutInSeconds); 481 } 482 } 483 484 private boolean waitTaskToFinish(ExecutorService pool, Future<Boolean> task, String operation) 485 throws TimeoutException, InterruptedException, ExecutionException { 486 pool.shutdown(); 487 try { 488 if (!pool.awaitTermination((long) this.timeout, TimeUnit.SECONDS)) { 489 LOG.warn( 490 "Timed out before finishing the " + operation + " operation. Timeout: " + this.timeout 491 + "sec"); 492 pool.shutdownNow(); 493 } 494 } catch (InterruptedException e) { 495 pool.shutdownNow(); 496 Thread.currentThread().interrupt(); 497 } 498 try { 499 return task.get(5, TimeUnit.SECONDS); 500 } catch (InterruptedException e) { 501 LOG.warn("Interrupted while " + operation + " Regions on " + this.hostname, e); 502 throw e; 503 } catch (ExecutionException e) { 504 LOG.error("Error while " + operation + " regions on RegionServer " + this.hostname, e); 505 throw e; 506 } 507 } 508 509 private void waitMoveTasksToFinish(ExecutorService moveRegionsPool, 510 List<Future<Boolean>> taskList, long timeoutInSeconds) throws Exception { 511 try { 512 if (!moveRegionsPool.awaitTermination(timeoutInSeconds, TimeUnit.SECONDS)) { 513 moveRegionsPool.shutdownNow(); 514 } 515 } catch (InterruptedException e) { 516 moveRegionsPool.shutdownNow(); 517 Thread.currentThread().interrupt(); 518 } 519 for (Future<Boolean> future : taskList) { 520 try { 521 // if even after shutdownNow threads are stuck we wait for 5 secs max 522 if (!future.get(5, TimeUnit.SECONDS)) { 523 LOG.error("Was Not able to move region....Exiting Now"); 524 throw new Exception("Could not move region Exception"); 525 } 526 } catch (InterruptedException e) { 527 LOG.error("Interrupted while waiting for Thread to Complete " + e.getMessage(), e); 528 throw e; 529 } catch (ExecutionException e) { 530 LOG.error("Got Exception From Thread While moving region " + e.getMessage(), e); 531 throw e; 532 } catch (CancellationException e) { 533 LOG.error("Thread for moving region cancelled. Timeout for cancellation:" + timeoutInSeconds 534 + "secs", e); 535 throw e; 536 } 537 } 538 } 539 540 private ServerName getTargetServer() throws Exception { 541 ServerName server = null; 542 int maxWaitInSeconds = 543 admin.getConfiguration().getInt(SERVERSTART_WAIT_MAX_KEY, DEFAULT_SERVERSTART_WAIT_MAX); 544 long maxWait = EnvironmentEdgeManager.currentTime() + maxWaitInSeconds * 1000; 545 while (EnvironmentEdgeManager.currentTime() < maxWait) { 546 try { 547 List<ServerName> regionServers = new ArrayList<>(); 548 regionServers.addAll( 549 admin.getClusterMetrics(EnumSet.of(Option.LIVE_SERVERS)).getLiveServerMetrics() 550 .keySet()); 551 // Remove the host Region server from target Region Servers list 552 server = stripServer(regionServers, hostname, port); 553 if (server != null) { 554 break; 555 } else { 556 LOG.warn("Server " + hostname + ":" + port + " is not up yet, waiting"); 557 } 558 } catch (IOException e) { 559 LOG.warn("Could not get list of region servers", e); 560 } 561 Thread.sleep(500); 562 } 563 if (server == null) { 564 LOG.error("Server " + hostname + ":" + port + " is not up. Giving up."); 565 throw new Exception("Server " + hostname + ":" + port + " to load regions not online"); 566 } 567 return server; 568 } 569 570 private List<RegionInfo> readRegionsFromFile(String filename) throws IOException { 571 List<RegionInfo> regions = new ArrayList<>(); 572 File f = new File(filename); 573 if (!f.exists()) { 574 return regions; 575 } 576 try (DataInputStream dis = new DataInputStream( 577 new BufferedInputStream(new FileInputStream(f)))) { 578 int numRegions = dis.readInt(); 579 int index = 0; 580 while (index < numRegions) { 581 regions.add(RegionInfo.parseFromOrNull(Bytes.readByteArray(dis))); 582 index++; 583 } 584 } catch (IOException e) { 585 LOG.error("Error while reading regions from file:" + filename, e); 586 throw e; 587 } 588 return regions; 589 } 590 591 /** 592 * Write the number of regions moved in the first line followed by regions moved in subsequent 593 * lines 594 */ 595 private void writeFile(String filename, List<RegionInfo> movedRegions) throws IOException { 596 try (DataOutputStream dos = new DataOutputStream( 597 new BufferedOutputStream(new FileOutputStream(filename)))) { 598 dos.writeInt(movedRegions.size()); 599 for (RegionInfo region : movedRegions) { 600 Bytes.writeByteArray(dos, RegionInfo.toByteArray(region)); 601 } 602 } catch (IOException e) { 603 LOG.error( 604 "ERROR: Was Not able to write regions moved to output file but moved " + movedRegions 605 .size() + " regions", e); 606 throw e; 607 } 608 } 609 610 private void deleteFile(String filename) { 611 File f = new File(filename); 612 if (f.exists()) { 613 f.delete(); 614 } 615 } 616 617 /** 618 * @return List of servers from the exclude file in format 'hostname:port'. 619 */ 620 private List<String> readExcludes(String excludeFile) throws IOException { 621 List<String> excludeServers = new ArrayList<>(); 622 if (excludeFile == null) { 623 return excludeServers; 624 } else { 625 try { 626 Files.readAllLines(Paths.get(excludeFile)).stream().map(String::trim) 627 .filter(((Predicate<String>) String::isEmpty).negate()).map(String::toLowerCase) 628 .forEach(excludeServers::add); 629 } catch (IOException e) { 630 LOG.warn("Exception while reading excludes file, continuing anyways", e); 631 } 632 return excludeServers; 633 } 634 } 635 636 /** 637 * Excludes the servername whose hostname and port portion matches the list given in exclude file 638 */ 639 private void stripExcludes(List<ServerName> regionServers) throws IOException { 640 if (excludeFile != null) { 641 List<String> excludes = readExcludes(excludeFile); 642 Iterator<ServerName> i = regionServers.iterator(); 643 while (i.hasNext()) { 644 String rs = i.next().getServerName(); 645 String rsPort = rs.split(ServerName.SERVERNAME_SEPARATOR)[0].toLowerCase() + ":" + rs 646 .split(ServerName.SERVERNAME_SEPARATOR)[1]; 647 if (excludes.contains(rsPort)) { 648 i.remove(); 649 } 650 } 651 LOG.info("Valid Region server targets are:" + regionServers.toString()); 652 LOG.info("Excluded Servers are" + excludes.toString()); 653 } 654 } 655 656 /** 657 * Exclude master from list of RSs to move regions to 658 */ 659 private void stripMaster(List<ServerName> regionServers) throws IOException { 660 ServerName master = admin.getClusterMetrics(EnumSet.of(Option.MASTER)).getMasterName(); 661 stripServer(regionServers, master.getHostname(), master.getPort()); 662 } 663 664 /** 665 * Remove the servername whose hostname and port portion matches from the passed array of servers. 666 * Returns as side-effect the servername removed. 667 * @return server removed from list of Region Servers 668 */ 669 private ServerName stripServer(List<ServerName> regionServers, String hostname, int port) { 670 ServerName server = null; 671 String portString = Integer.toString(port); 672 Iterator<ServerName> i = regionServers.iterator(); 673 while (i.hasNext()) { 674 server = i.next(); 675 String[] splitServer = server.getServerName().split(ServerName.SERVERNAME_SEPARATOR); 676 if (splitServer[0].equalsIgnoreCase(hostname) && splitServer[1].equals(portString)) { 677 i.remove(); 678 return server; 679 } 680 } 681 return server; 682 } 683 684 /** 685 * Tries to scan a row from passed region 686 */ 687 private void isSuccessfulScan(RegionInfo region) throws IOException { 688 Scan scan = new Scan().withStartRow(region.getStartKey()).setRaw(true).setOneRowLimit() 689 .setMaxResultSize(1L).setCaching(1).setFilter(new FirstKeyOnlyFilter()) 690 .setCacheBlocks(false); 691 try (Table table = conn.getTable(region.getTable()); 692 ResultScanner scanner = table.getScanner(scan)) { 693 scanner.next(); 694 } catch (IOException e) { 695 LOG.error("Could not scan region:" + region.getEncodedName(), e); 696 throw e; 697 } 698 } 699 700 /** 701 * Returns true if passed region is still on serverName when we look at hbase:meta. 702 * @return true if region is hosted on serverName otherwise false 703 */ 704 private boolean isSameServer(RegionInfo region, ServerName serverName) 705 throws IOException { 706 ServerName serverForRegion = getServerNameForRegion(region); 707 if (serverForRegion != null && serverForRegion.equals(serverName)) { 708 return true; 709 } 710 return false; 711 } 712 713 /** 714 * Get servername that is up in hbase:meta hosting the given region. this is hostname + port + 715 * startcode comma-delimited. Can return null 716 * @return regionServer hosting the given region 717 */ 718 private ServerName getServerNameForRegion(RegionInfo region) throws IOException { 719 if (!admin.isTableEnabled(region.getTable())) { 720 return null; 721 } 722 return MetaTableAccessor.getRegionLocation(conn, region).getServerName(); 723 } 724 725 @Override 726 protected void addOptions() { 727 this.addRequiredOptWithArg("r", "regionserverhost", "region server <hostname>|<hostname:port>"); 728 this.addRequiredOptWithArg("o", "operation", "Expected: load/unload"); 729 this.addOptWithArg("m", "maxthreads", 730 "Define the maximum number of threads to use to unload and reload the regions"); 731 this.addOptWithArg("x", "excludefile", 732 "File with <hostname:port> per line to exclude as unload targets; default excludes only " 733 + "target host; useful for rack decommisioning."); 734 this.addOptWithArg("f", "filename", 735 "File to save regions list into unloading, or read from loading; " 736 + "default /tmp/<usernamehostname:port>"); 737 this.addOptNoArg("n", "noack", 738 "Turn on No-Ack mode(default: false) which won't check if region is online on target " 739 + "RegionServer, hence best effort. This is more performant in unloading and loading " 740 + "but might lead to region being unavailable for some time till master reassigns it " 741 + "in case the move failed"); 742 this.addOptWithArg("t", "timeout", "timeout in seconds after which the tool will exit " 743 + "irrespective of whether it finished or not;default Integer.MAX_VALUE"); 744 } 745 746 @Override 747 protected void processOptions(CommandLine cmd) { 748 String hostname = cmd.getOptionValue("r"); 749 rmbuilder = new RegionMoverBuilder(hostname); 750 if (cmd.hasOption('m')) { 751 rmbuilder.maxthreads(Integer.parseInt(cmd.getOptionValue('m'))); 752 } 753 if (cmd.hasOption('n')) { 754 rmbuilder.ack(false); 755 } 756 if (cmd.hasOption('f')) { 757 rmbuilder.filename(cmd.getOptionValue('f')); 758 } 759 if (cmd.hasOption('x')) { 760 rmbuilder.excludeFile(cmd.getOptionValue('x')); 761 } 762 if (cmd.hasOption('t')) { 763 rmbuilder.timeout(Integer.parseInt(cmd.getOptionValue('t'))); 764 } 765 this.loadUnload = cmd.getOptionValue("o").toLowerCase(Locale.ROOT); 766 } 767 768 @Override 769 protected int doWork() throws Exception { 770 boolean success; 771 try (RegionMover rm = rmbuilder.build()) { 772 if (loadUnload.equalsIgnoreCase("load")) { 773 success = rm.load(); 774 } else if (loadUnload.equalsIgnoreCase("unload")) { 775 success = rm.unload(); 776 } else { 777 printUsage(); 778 success = false; 779 } 780 } 781 return (success ? 0 : 1); 782 } 783 784 public static void main(String[] args) { 785 new RegionMover().doStaticMain(args); 786 } 787}