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}