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}