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.BufferedReader; 023import java.io.DataInputStream; 024import java.io.DataOutputStream; 025import java.io.File; 026import java.io.FileInputStream; 027import java.io.FileOutputStream; 028import java.io.FileReader; 029import java.io.IOException; 030import java.util.ArrayList; 031import java.util.Collections; 032import java.util.EnumSet; 033import java.util.Iterator; 034import java.util.List; 035import java.util.Locale; 036import java.util.concurrent.Callable; 037import java.util.concurrent.CancellationException; 038import java.util.concurrent.ExecutionException; 039import java.util.concurrent.ExecutorService; 040import java.util.concurrent.Executors; 041import java.util.concurrent.Future; 042import java.util.concurrent.TimeUnit; 043import java.util.concurrent.TimeoutException; 044import org.apache.hadoop.conf.Configuration; 045import org.apache.hadoop.hbase.ClusterMetrics.Option; 046import org.apache.hadoop.hbase.HBaseConfiguration; 047import org.apache.hadoop.hbase.HConstants; 048import org.apache.hadoop.hbase.ServerName; 049import org.apache.hadoop.hbase.TableName; 050import org.apache.hadoop.hbase.client.Admin; 051import org.apache.hadoop.hbase.client.Connection; 052import org.apache.hadoop.hbase.client.ConnectionFactory; 053import org.apache.hadoop.hbase.client.Get; 054import org.apache.hadoop.hbase.client.RegionInfo; 055import org.apache.hadoop.hbase.client.Result; 056import org.apache.hadoop.hbase.client.ResultScanner; 057import org.apache.hadoop.hbase.client.Scan; 058import org.apache.hadoop.hbase.client.Table; 059import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter; 060import org.apache.hadoop.hbase.zookeeper.MetaTableLocator; 061import org.apache.hadoop.hbase.zookeeper.ZKWatcher; 062import org.apache.yetus.audience.InterfaceAudience; 063import org.slf4j.Logger; 064import org.slf4j.LoggerFactory; 065 066import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting; 067import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLine; 068 069/** 070 * Tool for loading/unloading regions to/from given regionserver This tool can be run from Command 071 * line directly as a utility. Supports Ack/No Ack mode for loading/unloading operations.Ack mode 072 * acknowledges if regions are online after movement while noAck mode is best effort mode that 073 * improves performance but will still move on if region is stuck/not moved. Motivation behind noAck 074 * mode being RS shutdown where even if a Region is stuck, upon shutdown master will move it 075 * anyways. This can also be used by constructiong an Object using the builder and then calling 076 * {@link #load()} or {@link #unload()} methods for the desired operations. 077 */ 078@InterfaceAudience.Public 079public class RegionMover extends AbstractHBaseTool { 080 public static final String MOVE_RETRIES_MAX_KEY = "hbase.move.retries.max"; 081 public static final String MOVE_WAIT_MAX_KEY = "hbase.move.wait.max"; 082 public static final String SERVERSTART_WAIT_MAX_KEY = "hbase.serverstart.wait.max"; 083 public static final int DEFAULT_MOVE_RETRIES_MAX = 5; 084 public static final int DEFAULT_MOVE_WAIT_MAX = 60; 085 public static final int DEFAULT_SERVERSTART_WAIT_MAX = 180; 086 static final Logger LOG = LoggerFactory.getLogger(RegionMover.class); 087 private RegionMoverBuilder rmbuilder; 088 private boolean ack = true; 089 private int maxthreads = 1; 090 private int timeout; 091 private String loadUnload; 092 private String hostname; 093 private String filename; 094 private String excludeFile; 095 private int port; 096 097 private RegionMover(RegionMoverBuilder builder) { 098 this.hostname = builder.hostname; 099 this.filename = builder.filename; 100 this.excludeFile = builder.excludeFile; 101 this.maxthreads = builder.maxthreads; 102 this.ack = builder.ack; 103 this.port = builder.port; 104 this.timeout = builder.timeout; 105 setConf(builder.conf); 106 } 107 108 private RegionMover() { 109 } 110 111 /** 112 * Builder for Region mover. Use the {@link #build()} method to create RegionMover object. Has 113 * {@link #filename(String)}, {@link #excludeFile(String)}, {@link #maxthreads(int)}, 114 * {@link #ack(boolean)}, {@link #timeout(int)} methods to set the corresponding options 115 */ 116 public static class RegionMoverBuilder { 117 private boolean ack = true; 118 private int maxthreads = 1; 119 private int timeout = Integer.MAX_VALUE; 120 private String hostname; 121 private String filename; 122 private String excludeFile = null; 123 private String defaultDir = System.getProperty("java.io.tmpdir"); 124 125 @VisibleForTesting 126 final int port; 127 private final Configuration conf; 128 129 public RegionMoverBuilder(String hostname) { 130 this(hostname, createConf()); 131 } 132 133 /** 134 * Creates a new configuration and sets region mover specific overrides 135 */ 136 private static Configuration createConf() { 137 Configuration conf = HBaseConfiguration.create(); 138 conf.setInt("hbase.client.prefetch.limit", 1); 139 conf.setInt("hbase.client.pause", 500); 140 conf.setInt("hbase.client.retries.number", 100); 141 return conf; 142 } 143 144 /** 145 * @param hostname Hostname to unload regions from or load regions to. Can be either hostname 146 * or hostname:port. 147 * @param conf Configuration object 148 */ 149 public RegionMoverBuilder(String hostname, Configuration conf) { 150 String[] splitHostname = hostname.toLowerCase().split(":"); 151 this.hostname = splitHostname[0]; 152 if (splitHostname.length == 2) { 153 this.port = Integer.parseInt(splitHostname[1]); 154 } else { 155 this.port = conf.getInt(HConstants.REGIONSERVER_PORT, HConstants.DEFAULT_REGIONSERVER_PORT); 156 } 157 this.filename = defaultDir + File.separator + System.getProperty("user.name") + this.hostname 158 + ":" + Integer.toString(this.port); 159 this.conf = conf; 160 } 161 162 /** 163 * Path of file where regions will be written to during unloading/read from during loading 164 * @param filename 165 * @return RegionMoverBuilder object 166 */ 167 public RegionMoverBuilder filename(String filename) { 168 this.filename = filename; 169 return this; 170 } 171 172 /** 173 * Set the max number of threads that will be used to move regions 174 */ 175 public RegionMoverBuilder maxthreads(int threads) { 176 this.maxthreads = threads; 177 return this; 178 } 179 180 /** 181 * Path of file containing hostnames to be excluded during region movement. Exclude file should 182 * have 'host:port' per line. Port is mandatory here as we can have many RS running on a single 183 * host. 184 */ 185 public RegionMoverBuilder excludeFile(String excludefile) { 186 this.excludeFile = excludefile; 187 return this; 188 } 189 190 /** 191 * Set ack/noAck mode. 192 * <p> 193 * In ack mode regions are acknowledged before and after moving and the move is retried 194 * hbase.move.retries.max times, if unsuccessful we quit with exit code 1.No Ack mode is a best 195 * effort mode,each region movement is tried once.This can be used during graceful shutdown as 196 * even if we have a stuck region,upon shutdown it'll be reassigned anyway. 197 * <p> 198 * @param ack 199 * @return RegionMoverBuilder object 200 */ 201 public RegionMoverBuilder ack(boolean ack) { 202 this.ack = ack; 203 return this; 204 } 205 206 /** 207 * Set the timeout for Load/Unload operation in seconds.This is a global timeout,threadpool for 208 * movers also have a separate time which is hbase.move.wait.max * number of regions to 209 * load/unload 210 * @param timeout in seconds 211 * @return RegionMoverBuilder object 212 */ 213 public RegionMoverBuilder timeout(int timeout) { 214 this.timeout = timeout; 215 return this; 216 } 217 218 /** 219 * This method builds the appropriate RegionMover object which can then be used to load/unload 220 * using load and unload methods 221 * @return RegionMover object 222 */ 223 public RegionMover build() { 224 return new RegionMover(this); 225 } 226 } 227 228 /** 229 * Loads the specified {@link #hostname} with regions listed in the {@link #filename} RegionMover 230 * Object has to be created using {@link #RegionMover(RegionMoverBuilder)} 231 * @return true if loading succeeded, false otherwise 232 * @throws ExecutionException 233 * @throws InterruptedException if the loader thread was interrupted 234 * @throws TimeoutException 235 */ 236 public boolean load() throws ExecutionException, InterruptedException, TimeoutException { 237 ExecutorService loadPool = Executors.newFixedThreadPool(1); 238 Future<Boolean> loadTask = loadPool.submit(new Load(this)); 239 loadPool.shutdown(); 240 try { 241 if (!loadPool.awaitTermination((long) this.timeout, TimeUnit.SECONDS)) { 242 LOG.warn("Timed out before finishing the loading operation. Timeout:" + this.timeout 243 + "sec"); 244 loadPool.shutdownNow(); 245 } 246 } catch (InterruptedException e) { 247 loadPool.shutdownNow(); 248 Thread.currentThread().interrupt(); 249 } 250 try { 251 return loadTask.get(5, TimeUnit.SECONDS); 252 } catch (InterruptedException e) { 253 LOG.warn("Interrupted while loading Regions on " + this.hostname, e); 254 throw e; 255 } catch (ExecutionException e) { 256 LOG.error("Error while loading regions on RegionServer " + this.hostname, e); 257 throw e; 258 } 259 } 260 261 private class Load implements Callable<Boolean> { 262 263 private RegionMover rm; 264 265 public Load(RegionMover rm) { 266 this.rm = rm; 267 } 268 269 @Override 270 public Boolean call() throws IOException { 271 Connection conn = ConnectionFactory.createConnection(rm.conf); 272 try { 273 List<RegionInfo> regionsToMove = readRegionsFromFile(rm.filename); 274 if (regionsToMove.isEmpty()) { 275 LOG.info("No regions to load.Exiting"); 276 return true; 277 } 278 Admin admin = conn.getAdmin(); 279 try { 280 loadRegions(admin, rm.hostname, rm.port, regionsToMove, rm.ack); 281 } finally { 282 admin.close(); 283 } 284 } catch (Exception e) { 285 LOG.error("Error while loading regions to " + rm.hostname, e); 286 return false; 287 } finally { 288 conn.close(); 289 } 290 return true; 291 } 292 } 293 294 /** 295 * Unload regions from given {@link #hostname} using ack/noAck mode and {@link #maxthreads}.In 296 * noAck mode we do not make sure that region is successfully online on the target region 297 * server,hence it is best effort.We do not unload regions to hostnames given in 298 * {@link #excludeFile}. 299 * @return true if unloading succeeded, false otherwise 300 * @throws InterruptedException if the unloader thread was interrupted 301 * @throws ExecutionException 302 * @throws TimeoutException 303 */ 304 public boolean unload() throws InterruptedException, ExecutionException, TimeoutException { 305 deleteFile(this.filename); 306 ExecutorService unloadPool = Executors.newFixedThreadPool(1); 307 Future<Boolean> unloadTask = unloadPool.submit(new Unload(this)); 308 unloadPool.shutdown(); 309 try { 310 if (!unloadPool.awaitTermination((long) this.timeout, TimeUnit.SECONDS)) { 311 LOG.warn("Timed out before finishing the unloading operation. Timeout:" + this.timeout 312 + "sec"); 313 unloadPool.shutdownNow(); 314 } 315 } catch (InterruptedException e) { 316 unloadPool.shutdownNow(); 317 Thread.currentThread().interrupt(); 318 } 319 try { 320 return unloadTask.get(5, TimeUnit.SECONDS); 321 } catch (InterruptedException e) { 322 LOG.warn("Interrupted while unloading Regions from " + this.hostname, e); 323 throw e; 324 } catch (ExecutionException e) { 325 LOG.error("Error while unloading regions from RegionServer " + this.hostname, e); 326 throw e; 327 } 328 } 329 330 private class Unload implements Callable<Boolean> { 331 332 List<RegionInfo> movedRegions = Collections.synchronizedList(new ArrayList<RegionInfo>()); 333 private RegionMover rm; 334 335 public Unload(RegionMover rm) { 336 this.rm = rm; 337 } 338 339 @Override 340 public Boolean call() throws IOException { 341 Connection conn = ConnectionFactory.createConnection(rm.conf); 342 try { 343 Admin admin = conn.getAdmin(); 344 // Get Online RegionServers 345 ArrayList<String> regionServers = getServers(admin); 346 if (LOG.isDebugEnabled()) { 347 LOG.debug("Online region servers:" + regionServers.toString()); 348 } 349 // Remove the host Region server from target Region Servers list 350 String server = stripServer(regionServers, hostname, port); 351 // Remove RS present in the exclude file 352 stripExcludes(regionServers, rm.excludeFile); 353 stripMaster(regionServers, admin); 354 unloadRegions(admin, server, regionServers, rm.ack, movedRegions); 355 } catch (Exception e) { 356 LOG.error("Error while unloading regions ", e); 357 return false; 358 } finally { 359 try { 360 conn.close(); 361 } catch (IOException e) { 362 // ignore 363 } 364 if (movedRegions != null) { 365 writeFile(rm.filename, movedRegions); 366 } 367 } 368 return true; 369 } 370 } 371 372 private void loadRegions(Admin admin, String hostname, int port, 373 List<RegionInfo> regionsToMove, boolean ack) throws Exception { 374 String server = null; 375 List<RegionInfo> movedRegions = Collections.synchronizedList(new ArrayList<RegionInfo>()); 376 int maxWaitInSeconds = 377 admin.getConfiguration().getInt(SERVERSTART_WAIT_MAX_KEY, DEFAULT_SERVERSTART_WAIT_MAX); 378 long maxWait = EnvironmentEdgeManager.currentTime() + maxWaitInSeconds * 1000; 379 while ((EnvironmentEdgeManager.currentTime() < maxWait) && (server == null)) { 380 try { 381 ArrayList<String> regionServers = getServers(admin); 382 // Remove the host Region server from target Region Servers list 383 server = stripServer(regionServers, hostname, port); 384 if (server != null) { 385 break; 386 } 387 } catch (IOException e) { 388 LOG.warn("Could not get list of region servers", e); 389 } catch (Exception e) { 390 LOG.info("hostname=" + hostname + " is not up yet, waiting"); 391 } 392 try { 393 Thread.sleep(500); 394 } catch (InterruptedException e) { 395 LOG.error("Interrupted while waiting for " + hostname + " to be up.Quitting now", e); 396 throw e; 397 } 398 } 399 if (server == null) { 400 LOG.error("Host:" + hostname + " is not up.Giving up."); 401 throw new Exception("Host to load regions not online"); 402 } 403 LOG.info("Moving " + regionsToMove.size() + " regions to " + server + " using " 404 + this.maxthreads + " threads.Ack mode:" + this.ack); 405 ExecutorService moveRegionsPool = Executors.newFixedThreadPool(this.maxthreads); 406 List<Future<Boolean>> taskList = new ArrayList<>(); 407 int counter = 0; 408 while (counter < regionsToMove.size()) { 409 RegionInfo region = regionsToMove.get(counter); 410 String currentServer = getServerNameForRegion(admin, region); 411 if (currentServer == null) { 412 LOG.warn("Could not get server for Region:" + region.getEncodedName() + " moving on"); 413 counter++; 414 continue; 415 } else if (server.equals(currentServer)) { 416 LOG.info("Region " + region.getRegionNameAsString() + 417 " is already on target server=" + server); 418 counter++; 419 continue; 420 } 421 if (ack) { 422 Future<Boolean> task = 423 moveRegionsPool.submit(new MoveWithAck(admin, region, currentServer, server, 424 movedRegions)); 425 taskList.add(task); 426 } else { 427 Future<Boolean> task = 428 moveRegionsPool.submit(new MoveWithoutAck(admin, region, currentServer, server, 429 movedRegions)); 430 taskList.add(task); 431 } 432 counter++; 433 } 434 moveRegionsPool.shutdown(); 435 long timeoutInSeconds = 436 regionsToMove.size() 437 * admin.getConfiguration().getLong(MOVE_WAIT_MAX_KEY, DEFAULT_MOVE_WAIT_MAX); 438 try { 439 if (!moveRegionsPool.awaitTermination(timeoutInSeconds, TimeUnit.SECONDS)) { 440 moveRegionsPool.shutdownNow(); 441 } 442 } catch (InterruptedException e) { 443 moveRegionsPool.shutdownNow(); 444 Thread.currentThread().interrupt(); 445 } 446 for (Future<Boolean> future : taskList) { 447 try { 448 // if even after shutdownNow threads are stuck we wait for 5 secs max 449 if (!future.get(5, TimeUnit.SECONDS)) { 450 LOG.error("Was Not able to move region....Exiting Now"); 451 throw new Exception("Could not move region Exception"); 452 } 453 } catch (InterruptedException e) { 454 LOG.error("Interrupted while waiting for Thread to Complete " + e.getMessage(), e); 455 throw e; 456 } catch (ExecutionException e) { 457 LOG.error("Got Exception From Thread While moving region " + e.getMessage(), e); 458 throw e; 459 } catch (CancellationException e) { 460 LOG.error("Thread for moving region cancelled. Timeout for cancellation:" 461 + timeoutInSeconds + "secs", e); 462 throw e; 463 } 464 } 465 } 466 467 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="DLS_DEAD_LOCAL_STORE", 468 justification="FB is wrong; its size is read") 469 private void unloadRegions(Admin admin, String server, ArrayList<String> regionServers, 470 boolean ack, List<RegionInfo> movedRegions) throws Exception { 471 List<RegionInfo> regionsToMove = new ArrayList<>();// FindBugs: DLS_DEAD_LOCAL_STORE 472 regionsToMove = getRegions(this.conf, server); 473 if (regionsToMove.isEmpty()) { 474 LOG.info("No Regions to move....Quitting now"); 475 return; 476 } else if (regionServers.isEmpty()) { 477 LOG.warn("No Regions were moved - no servers available"); 478 throw new Exception("No online region servers"); 479 } 480 while (true) { 481 regionsToMove = getRegions(this.conf, server); 482 regionsToMove.removeAll(movedRegions); 483 if (regionsToMove.isEmpty()) { 484 break; 485 } 486 int counter = 0; 487 LOG.info("Moving " + regionsToMove.size() + " regions from " + this.hostname + " to " 488 + regionServers.size() + " servers using " + this.maxthreads + " threads .Ack Mode:" 489 + ack); 490 ExecutorService moveRegionsPool = Executors.newFixedThreadPool(this.maxthreads); 491 List<Future<Boolean>> taskList = new ArrayList<>(); 492 int serverIndex = 0; 493 while (counter < regionsToMove.size()) { 494 if (ack) { 495 Future<Boolean> task = 496 moveRegionsPool.submit(new MoveWithAck(admin, regionsToMove.get(counter), server, 497 regionServers.get(serverIndex), movedRegions)); 498 taskList.add(task); 499 } else { 500 Future<Boolean> task = 501 moveRegionsPool.submit(new MoveWithoutAck(admin, regionsToMove.get(counter), server, 502 regionServers.get(serverIndex), movedRegions)); 503 taskList.add(task); 504 } 505 counter++; 506 serverIndex = (serverIndex + 1) % regionServers.size(); 507 } 508 moveRegionsPool.shutdown(); 509 long timeoutInSeconds = 510 regionsToMove.size() 511 * admin.getConfiguration().getLong(MOVE_WAIT_MAX_KEY, DEFAULT_MOVE_WAIT_MAX); 512 try { 513 if (!moveRegionsPool.awaitTermination(timeoutInSeconds, TimeUnit.SECONDS)) { 514 moveRegionsPool.shutdownNow(); 515 } 516 } catch (InterruptedException e) { 517 moveRegionsPool.shutdownNow(); 518 Thread.currentThread().interrupt(); 519 } 520 for (Future<Boolean> future : taskList) { 521 try { 522 // if even after shutdownNow threads are stuck we wait for 5 secs max 523 if (!future.get(5, TimeUnit.SECONDS)) { 524 LOG.error("Was Not able to move region....Exiting Now"); 525 throw new Exception("Could not move region Exception"); 526 } 527 } catch (InterruptedException e) { 528 LOG.error("Interrupted while waiting for Thread to Complete " + e.getMessage(), e); 529 throw e; 530 } catch (ExecutionException e) { 531 LOG.error("Got Exception From Thread While moving region " + e.getMessage(), e); 532 throw e; 533 } catch (CancellationException e) { 534 LOG.error("Thread for moving region cancelled. Timeout for cancellation:" 535 + timeoutInSeconds + "secs", e); 536 throw e; 537 } 538 } 539 } 540 } 541 542 /** 543 * Move Regions and make sure that they are up on the target server.If a region movement fails we 544 * exit as failure 545 */ 546 private class MoveWithAck implements Callable<Boolean> { 547 private Admin admin; 548 private RegionInfo region; 549 private String targetServer; 550 private List<RegionInfo> movedRegions; 551 private String sourceServer; 552 553 public MoveWithAck(Admin admin, RegionInfo regionInfo, String sourceServer, 554 String targetServer, List<RegionInfo> movedRegions) { 555 this.admin = admin; 556 this.region = regionInfo; 557 this.targetServer = targetServer; 558 this.movedRegions = movedRegions; 559 this.sourceServer = sourceServer; 560 } 561 562 @Override 563 public Boolean call() throws IOException, InterruptedException { 564 boolean moved = false; 565 int count = 0; 566 int retries = admin.getConfiguration().getInt(MOVE_RETRIES_MAX_KEY, DEFAULT_MOVE_RETRIES_MAX); 567 int maxWaitInSeconds = 568 admin.getConfiguration().getInt(MOVE_WAIT_MAX_KEY, DEFAULT_MOVE_WAIT_MAX); 569 long startTime = EnvironmentEdgeManager.currentTime(); 570 boolean sameServer = true; 571 // Assert we can scan the region in its current location 572 isSuccessfulScan(admin, region); 573 LOG.info("Moving region:" + region.getEncodedName() + " from " + sourceServer + " to " 574 + targetServer); 575 while (count < retries && sameServer) { 576 if (count > 0) { 577 LOG.info("Retry " + Integer.toString(count) + " of maximum " + Integer.toString(retries)); 578 } 579 count = count + 1; 580 admin.move(region.getEncodedNameAsBytes(), Bytes.toBytes(targetServer)); 581 long maxWait = startTime + (maxWaitInSeconds * 1000); 582 while (EnvironmentEdgeManager.currentTime() < maxWait) { 583 sameServer = isSameServer(admin, region, sourceServer); 584 if (!sameServer) { 585 break; 586 } 587 Thread.sleep(100); 588 } 589 } 590 if (sameServer) { 591 LOG.error("Region: " + region.getRegionNameAsString() + " stuck on " + this.sourceServer 592 + ",newServer=" + this.targetServer); 593 } else { 594 isSuccessfulScan(admin, region); 595 LOG.info("Moved Region " 596 + region.getRegionNameAsString() 597 + " cost:" 598 + String.format("%.3f", 599 (float) (EnvironmentEdgeManager.currentTime() - startTime) / 1000)); 600 moved = true; 601 movedRegions.add(region); 602 } 603 return moved; 604 } 605 } 606 607 /** 608 * Move Regions without Acknowledging.Usefule in case of RS shutdown as we might want to shut the 609 * RS down anyways and not abort on a stuck region. Improves movement performance 610 */ 611 private static class MoveWithoutAck implements Callable<Boolean> { 612 private Admin admin; 613 private RegionInfo region; 614 private String targetServer; 615 private List<RegionInfo> movedRegions; 616 private String sourceServer; 617 618 public MoveWithoutAck(Admin admin, RegionInfo regionInfo, String sourceServer, 619 String targetServer, List<RegionInfo> movedRegions) { 620 this.admin = admin; 621 this.region = regionInfo; 622 this.targetServer = targetServer; 623 this.movedRegions = movedRegions; 624 this.sourceServer = sourceServer; 625 } 626 627 @Override 628 public Boolean call() { 629 try { 630 LOG.info("Moving region:" + region.getEncodedName() + " from " + sourceServer + " to " 631 + targetServer); 632 admin.move(region.getEncodedNameAsBytes(), Bytes.toBytes(targetServer)); 633 LOG.info("Moved " + region.getEncodedName() + " from " + sourceServer + " to " 634 + targetServer); 635 } catch (Exception e) { 636 LOG.error("Error Moving Region:" + region.getEncodedName(), e); 637 } finally { 638 // we add region to the moved regions list in No Ack Mode since this is best effort 639 movedRegions.add(region); 640 } 641 return true; 642 } 643 } 644 645 private List<RegionInfo> readRegionsFromFile(String filename) throws IOException { 646 List<RegionInfo> regions = new ArrayList<>(); 647 File f = new File(filename); 648 if (!f.exists()) { 649 return regions; 650 } 651 FileInputStream fis = null; 652 DataInputStream dis = null; 653 try { 654 fis = new FileInputStream(f); 655 dis = new DataInputStream(fis); 656 int numRegions = dis.readInt(); 657 int index = 0; 658 while (index < numRegions) { 659 regions.add(RegionInfo.parseFromOrNull(Bytes.readByteArray(dis))); 660 index++; 661 } 662 } catch (IOException e) { 663 LOG.error("Error while reading regions from file:" + filename, e); 664 throw e; 665 } finally { 666 if (dis != null) { 667 dis.close(); 668 } 669 if (fis != null) { 670 fis.close(); 671 } 672 } 673 return regions; 674 } 675 676 /** 677 * Get online regions of the passed server 678 * @param conf 679 * @param server 680 * @return List of Regions online on the server 681 * @throws IOException 682 */ 683 private List<RegionInfo> getRegions(Configuration conf, String server) throws IOException { 684 Connection conn = ConnectionFactory.createConnection(conf); 685 try { 686 return conn.getAdmin().getRegions(ServerName.valueOf(server)); 687 } finally { 688 conn.close(); 689 } 690 } 691 692 /** 693 * Write the number of regions moved in the first line followed by regions moved in subsequent 694 * lines 695 * @param filename 696 * @param movedRegions 697 * @throws IOException 698 */ 699 private void writeFile(String filename, List<RegionInfo> movedRegions) throws IOException { 700 FileOutputStream fos = null; 701 DataOutputStream dos = null; 702 try { 703 fos = new FileOutputStream(filename); 704 dos = new DataOutputStream(fos); 705 dos.writeInt(movedRegions.size()); 706 for (RegionInfo region : movedRegions) { 707 Bytes.writeByteArray(dos, RegionInfo.toByteArray(region)); 708 } 709 } catch (IOException e) { 710 LOG.error("ERROR: Was Not able to write regions moved to output file but moved " 711 + movedRegions.size() + " regions", e); 712 throw e; 713 } finally { 714 if (dos != null) { 715 dos.close(); 716 } 717 if (fos != null) { 718 fos.close(); 719 } 720 } 721 } 722 723 /** 724 * Excludes the servername whose hostname and port portion matches the list given in exclude file 725 * @param regionServers 726 * @param excludeFile 727 * @throws IOException 728 */ 729 private void stripExcludes(ArrayList<String> regionServers, String excludeFile) 730 throws IOException { 731 if (excludeFile != null) { 732 ArrayList<String> excludes = readExcludes(excludeFile); 733 Iterator<String> i = regionServers.iterator(); 734 while (i.hasNext()) { 735 String rs = i.next(); 736 String rsPort = 737 rs.split(ServerName.SERVERNAME_SEPARATOR)[0] + ":" 738 + rs.split(ServerName.SERVERNAME_SEPARATOR)[1]; 739 if (excludes.contains(rsPort)) { 740 i.remove(); 741 } 742 } 743 LOG.info("Valid Region server targets are:" + regionServers.toString()); 744 LOG.info("Excluded Servers are" + excludes.toString()); 745 } 746 } 747 748 /** 749 * Exclude master from list of RSs to move regions to 750 * @param regionServers 751 * @param admin 752 * @throws IOException 753 */ 754 private void stripMaster(ArrayList<String> regionServers, Admin admin) throws IOException { 755 ServerName master = admin.getClusterMetrics(EnumSet.of(Option.MASTER)).getMasterName(); 756 String masterHostname = master.getHostname(); 757 int masterPort = master.getPort(); 758 try { 759 stripServer(regionServers, masterHostname, masterPort); 760 } catch (Exception e) { 761 LOG.warn("Could not remove master from list of RS", e); 762 } 763 } 764 765 /** 766 * @return List of servers from the exclude file in format 'hostname:port'. 767 */ 768 private ArrayList<String> readExcludes(String excludeFile) throws IOException { 769 ArrayList<String> excludeServers = new ArrayList<>(); 770 if (excludeFile == null) { 771 return excludeServers; 772 } else { 773 File f = new File(excludeFile); 774 String line; 775 BufferedReader br = null; 776 try { 777 br = new BufferedReader(new FileReader(f)); 778 while ((line = br.readLine()) != null) { 779 line = line.trim(); 780 if (!line.equals("")) { 781 excludeServers.add(line); 782 } 783 } 784 } catch (IOException e) { 785 LOG.warn("Exception while reading excludes file,continuing anyways", e); 786 } finally { 787 if (br != null) { 788 br.close(); 789 } 790 } 791 return excludeServers; 792 } 793 } 794 795 /** 796 * Remove the servername whose hostname and port portion matches from the passed array of servers. 797 * Returns as side-effect the servername removed. 798 * @param regionServers 799 * @param hostname 800 * @param port 801 * @return server removed from list of Region Servers 802 * @throws Exception 803 */ 804 private String stripServer(ArrayList<String> regionServers, String hostname, int port) 805 throws Exception { 806 String server = null; 807 String portString = Integer.toString(port); 808 Iterator<String> i = regionServers.iterator(); 809 int noOfRs = regionServers.size(); 810 while (i.hasNext()) { 811 server = i.next(); 812 String[] splitServer = server.split(ServerName.SERVERNAME_SEPARATOR); 813 if (splitServer[0].equalsIgnoreCase(hostname) && splitServer[1].equals(portString)) { 814 i.remove(); 815 return server; 816 } 817 } 818 if (regionServers.size() >= noOfRs) { 819 throw new Exception("Server " + hostname + ":" + Integer.toString(port) 820 + " is not in list of online servers(Offline/Incorrect)"); 821 } 822 return server; 823 } 824 825 /** 826 * Get Arraylist of Servers in the cluster 827 * @param admin 828 * @return ArrayList of online region servers 829 * @throws IOException 830 */ 831 private ArrayList<String> getServers(Admin admin) throws IOException { 832 ArrayList<ServerName> serverInfo = new ArrayList<>( 833 admin.getClusterMetrics(EnumSet.of(Option.LIVE_SERVERS)).getLiveServerMetrics().keySet()); 834 ArrayList<String> regionServers = new ArrayList<>(serverInfo.size()); 835 for (ServerName server : serverInfo) { 836 regionServers.add(server.getServerName().toLowerCase()); 837 } 838 return regionServers; 839 } 840 841 private void deleteFile(String filename) { 842 File f = new File(filename); 843 if (f.exists()) { 844 f.delete(); 845 } 846 } 847 848 /** 849 * Tries to scan a row from passed region 850 * @param admin 851 * @param region 852 * @throws IOException 853 */ 854 private void isSuccessfulScan(Admin admin, RegionInfo region) throws IOException { 855 Scan scan = new Scan(region.getStartKey()); 856 scan.setBatch(1); 857 scan.setCaching(1); 858 scan.setFilter(new FirstKeyOnlyFilter()); 859 try { 860 Table table = admin.getConnection().getTable(region.getTable()); 861 try { 862 ResultScanner scanner = table.getScanner(scan); 863 try { 864 scanner.next(); 865 } finally { 866 scanner.close(); 867 } 868 } finally { 869 table.close(); 870 } 871 } catch (IOException e) { 872 LOG.error("Could not scan region:" + region.getEncodedName(), e); 873 throw e; 874 } 875 } 876 877 /** 878 * Returns true if passed region is still on serverName when we look at hbase:meta. 879 * @param admin 880 * @param region 881 * @param serverName 882 * @return true if region is hosted on serverName otherwise false 883 * @throws IOException 884 */ 885 private boolean isSameServer(Admin admin, RegionInfo region, String serverName) 886 throws IOException { 887 String serverForRegion = getServerNameForRegion(admin, region); 888 if (serverForRegion != null && serverForRegion.equals(serverName)) { 889 return true; 890 } 891 return false; 892 } 893 894 /** 895 * Get servername that is up in hbase:meta hosting the given region. this is hostname + port + 896 * startcode comma-delimited. Can return null 897 * @param admin 898 * @param region 899 * @return regionServer hosting the given region 900 * @throws IOException 901 */ 902 private String getServerNameForRegion(Admin admin, RegionInfo region) throws IOException { 903 String server = null; 904 if (!admin.isTableEnabled(region.getTable())) { 905 return null; 906 } 907 if (region.isMetaRegion()) { 908 ZKWatcher zkw = new ZKWatcher(admin.getConfiguration(), "region_mover", null); 909 MetaTableLocator locator = new MetaTableLocator(); 910 int maxWaitInSeconds = 911 admin.getConfiguration().getInt(MOVE_WAIT_MAX_KEY, DEFAULT_MOVE_WAIT_MAX); 912 try { 913 server = locator.waitMetaRegionLocation(zkw, maxWaitInSeconds * 1000).toString(); 914 } catch (InterruptedException e) { 915 LOG.error("Interrupted while waiting for location of Meta", e); 916 } finally { 917 if (zkw != null) { 918 zkw.close(); 919 } 920 } 921 } else { 922 Table table = admin.getConnection().getTable(TableName.META_TABLE_NAME); 923 try { 924 Get get = new Get(region.getRegionName()); 925 get.addColumn(HConstants.CATALOG_FAMILY, HConstants.SERVER_QUALIFIER); 926 get.addColumn(HConstants.CATALOG_FAMILY, HConstants.STARTCODE_QUALIFIER); 927 Result result = table.get(get); 928 if (result != null) { 929 byte[] servername = 930 result.getValue(HConstants.CATALOG_FAMILY, HConstants.SERVER_QUALIFIER); 931 byte[] startcode = 932 result.getValue(HConstants.CATALOG_FAMILY, HConstants.STARTCODE_QUALIFIER); 933 if (servername != null) { 934 server = Bytes.toString(servername).replaceFirst(":", ",").toLowerCase() + "," + 935 Bytes.toLong(startcode); 936 } 937 } 938 } catch (IOException e) { 939 LOG.error("Could not get Server Name for region:" + region.getEncodedName(), e); 940 throw e; 941 } finally { 942 table.close(); 943 } 944 } 945 return server; 946 } 947 948 @Override 949 protected void addOptions() { 950 this.addRequiredOptWithArg("r", "regionserverhost", "region server <hostname>|<hostname:port>"); 951 this.addRequiredOptWithArg("o", "operation", "Expected: load/unload"); 952 this.addOptWithArg("m", "maxthreads", 953 "Define the maximum number of threads to use to unload and reload the regions"); 954 this.addOptWithArg("x", "excludefile", 955 "File with <hostname:port> per line to exclude as unload targets; default excludes only " 956 + "target host; useful for rack decommisioning."); 957 this.addOptWithArg("f", "filename", 958 "File to save regions list into unloading, or read from loading; " 959 + "default /tmp/<usernamehostname:port>"); 960 this.addOptNoArg("n", "noack", 961 "Turn on No-Ack mode(default: false) which won't check if region is online on target " 962 + "RegionServer, hence best effort. This is more performant in unloading and loading " 963 + "but might lead to region being unavailable for some time till master reassigns it " 964 + "in case the move failed"); 965 this.addOptWithArg("t", "timeout", "timeout in seconds after which the tool will exit " 966 + "irrespective of whether it finished or not;default Integer.MAX_VALUE"); 967 } 968 969 @Override 970 protected void processOptions(CommandLine cmd) { 971 String hostname = cmd.getOptionValue("r"); 972 rmbuilder = new RegionMoverBuilder(hostname); 973 if (cmd.hasOption('m')) { 974 rmbuilder.maxthreads(Integer.parseInt(cmd.getOptionValue('m'))); 975 } 976 if (cmd.hasOption('n')) { 977 rmbuilder.ack(false); 978 } 979 if (cmd.hasOption('f')) { 980 rmbuilder.filename(cmd.getOptionValue('f')); 981 } 982 if (cmd.hasOption('x')) { 983 rmbuilder.excludeFile(cmd.getOptionValue('x')); 984 } 985 if (cmd.hasOption('t')) { 986 rmbuilder.timeout(Integer.parseInt(cmd.getOptionValue('t'))); 987 } 988 this.loadUnload = cmd.getOptionValue("o").toLowerCase(Locale.ROOT); 989 } 990 991 @Override 992 protected int doWork() throws Exception { 993 boolean success; 994 RegionMover rm = rmbuilder.build(); 995 if (loadUnload.equalsIgnoreCase("load")) { 996 success = rm.load(); 997 } else if (loadUnload.equalsIgnoreCase("unload")) { 998 success = rm.unload(); 999 } else { 1000 printUsage(); 1001 success = false; 1002 } 1003 return (success ? 0 : 1); 1004 } 1005 1006 public static void main(String[] args) { 1007 new RegionMover().doStaticMain(args); 1008 } 1009}