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 */
019package org.apache.hadoop.hbase.util;
020
021import java.io.FileNotFoundException;
022import java.io.IOException;
023import java.lang.reflect.InvocationTargetException;
024import java.lang.reflect.Method;
025import java.net.URI;
026import java.net.URISyntaxException;
027import java.util.List;
028import java.util.Locale;
029import java.util.Map;
030import java.util.concurrent.ConcurrentHashMap;
031
032import org.apache.hadoop.HadoopIllegalArgumentException;
033import org.apache.hadoop.conf.Configuration;
034import org.apache.hadoop.fs.FSDataOutputStream;
035import org.apache.hadoop.fs.FileStatus;
036import org.apache.hadoop.fs.FileSystem;
037import org.apache.hadoop.fs.LocatedFileStatus;
038import org.apache.hadoop.fs.Path;
039import org.apache.hadoop.fs.PathFilter;
040import org.apache.hadoop.fs.RemoteIterator;
041import org.apache.hadoop.fs.permission.FsPermission;
042import org.apache.hadoop.hbase.HConstants;
043import org.apache.hadoop.hbase.TableName;
044import org.apache.hadoop.ipc.RemoteException;
045import org.apache.yetus.audience.InterfaceAudience;
046import org.slf4j.Logger;
047import org.slf4j.LoggerFactory;
048
049import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
050import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
051
052/**
053 * Utility methods for interacting with the underlying file system.
054 */
055@InterfaceAudience.Private
056public abstract class CommonFSUtils {
057  private static final Logger LOG = LoggerFactory.getLogger(CommonFSUtils.class);
058
059  /** Parameter name for HBase WAL directory */
060  public static final String HBASE_WAL_DIR = "hbase.wal.dir";
061
062  /** Parameter to disable stream capability enforcement checks */
063  public static final String UNSAFE_STREAM_CAPABILITY_ENFORCE = "hbase.unsafe.stream.capability.enforce";
064
065  /** Full access permissions (starting point for a umask) */
066  public static final String FULL_RWX_PERMISSIONS = "777";
067
068  protected CommonFSUtils() {
069    super();
070  }
071
072  /**
073   * Compare of path component. Does not consider schema; i.e. if schemas
074   * different but <code>path</code> starts with <code>rootPath</code>,
075   * then the function returns true
076   * @param rootPath value to check for
077   * @param path subject to check
078   * @return True if <code>path</code> starts with <code>rootPath</code>
079   */
080  public static boolean isStartingWithPath(final Path rootPath, final String path) {
081    String uriRootPath = rootPath.toUri().getPath();
082    String tailUriPath = (new Path(path)).toUri().getPath();
083    return tailUriPath.startsWith(uriRootPath);
084  }
085
086  /**
087   * Compare path component of the Path URI; e.g. if hdfs://a/b/c and /a/b/c, it will compare the
088   * '/a/b/c' part. Does not consider schema; i.e. if schemas different but path or subpath matches,
089   * the two will equate.
090   * @param pathToSearch Path we will be trying to match against.
091   * @param pathTail what to match
092   * @return True if <code>pathTail</code> is tail on the path of <code>pathToSearch</code>
093   */
094  public static boolean isMatchingTail(final Path pathToSearch, String pathTail) {
095    return isMatchingTail(pathToSearch, new Path(pathTail));
096  }
097
098  /**
099   * Compare path component of the Path URI; e.g. if hdfs://a/b/c and /a/b/c, it will compare the
100   * '/a/b/c' part. If you passed in 'hdfs://a/b/c and b/c, it would return true.  Does not consider
101   * schema; i.e. if schemas different but path or subpath matches, the two will equate.
102   * @param pathToSearch Path we will be trying to match agains against
103   * @param pathTail what to match
104   * @return True if <code>pathTail</code> is tail on the path of <code>pathToSearch</code>
105   */
106  public static boolean isMatchingTail(final Path pathToSearch, final Path pathTail) {
107    if (pathToSearch.depth() != pathTail.depth()) {
108      return false;
109    }
110    Path tailPath = pathTail;
111    String tailName;
112    Path toSearch = pathToSearch;
113    String toSearchName;
114    boolean result = false;
115    do {
116      tailName = tailPath.getName();
117      if (tailName == null || tailName.length() <= 0) {
118        result = true;
119        break;
120      }
121      toSearchName = toSearch.getName();
122      if (toSearchName == null || toSearchName.length() <= 0) {
123        break;
124      }
125      // Move up a parent on each path for next go around.  Path doesn't let us go off the end.
126      tailPath = tailPath.getParent();
127      toSearch = toSearch.getParent();
128    } while(tailName.equals(toSearchName));
129    return result;
130  }
131
132  /**
133   * Delete if exists.
134   * @param fs filesystem object
135   * @param dir directory to delete
136   * @return True if deleted <code>dir</code>
137   * @throws IOException e
138   */
139  public static boolean deleteDirectory(final FileSystem fs, final Path dir)
140  throws IOException {
141    return fs.exists(dir) && fs.delete(dir, true);
142  }
143
144  /**
145   * Return the number of bytes that large input files should be optimally
146   * be split into to minimize i/o time.
147   *
148   * use reflection to search for getDefaultBlockSize(Path f)
149   * if the method doesn't exist, fall back to using getDefaultBlockSize()
150   *
151   * @param fs filesystem object
152   * @return the default block size for the path's filesystem
153   * @throws IOException e
154   */
155  public static long getDefaultBlockSize(final FileSystem fs, final Path path) throws IOException {
156    Method m = null;
157    Class<? extends FileSystem> cls = fs.getClass();
158    try {
159      m = cls.getMethod("getDefaultBlockSize", new Class<?>[] { Path.class });
160    } catch (NoSuchMethodException e) {
161      LOG.info("FileSystem doesn't support getDefaultBlockSize");
162    } catch (SecurityException e) {
163      LOG.info("Doesn't have access to getDefaultBlockSize on FileSystems", e);
164      m = null; // could happen on setAccessible()
165    }
166    if (m == null) {
167      return fs.getDefaultBlockSize(path);
168    } else {
169      try {
170        Object ret = m.invoke(fs, path);
171        return ((Long)ret).longValue();
172      } catch (Exception e) {
173        throw new IOException(e);
174      }
175    }
176  }
177
178  /*
179   * Get the default replication.
180   *
181   * use reflection to search for getDefaultReplication(Path f)
182   * if the method doesn't exist, fall back to using getDefaultReplication()
183   *
184   * @param fs filesystem object
185   * @param f path of file
186   * @return default replication for the path's filesystem
187   * @throws IOException e
188   */
189  public static short getDefaultReplication(final FileSystem fs, final Path path)
190      throws IOException {
191    Method m = null;
192    Class<? extends FileSystem> cls = fs.getClass();
193    try {
194      m = cls.getMethod("getDefaultReplication", new Class<?>[] { Path.class });
195    } catch (NoSuchMethodException e) {
196      LOG.info("FileSystem doesn't support getDefaultReplication");
197    } catch (SecurityException e) {
198      LOG.info("Doesn't have access to getDefaultReplication on FileSystems", e);
199      m = null; // could happen on setAccessible()
200    }
201    if (m == null) {
202      return fs.getDefaultReplication(path);
203    } else {
204      try {
205        Object ret = m.invoke(fs, path);
206        return ((Number)ret).shortValue();
207      } catch (Exception e) {
208        throw new IOException(e);
209      }
210    }
211  }
212
213  /**
214   * Returns the default buffer size to use during writes.
215   *
216   * The size of the buffer should probably be a multiple of hardware
217   * page size (4096 on Intel x86), and it determines how much data is
218   * buffered during read and write operations.
219   *
220   * @param fs filesystem object
221   * @return default buffer size to use during writes
222   */
223  public static int getDefaultBufferSize(final FileSystem fs) {
224    return fs.getConf().getInt("io.file.buffer.size", 4096);
225  }
226
227  /**
228   * Create the specified file on the filesystem. By default, this will:
229   * <ol>
230   * <li>apply the umask in the configuration (if it is enabled)</li>
231   * <li>use the fs configured buffer size (or 4096 if not set)</li>
232   * <li>use the default replication</li>
233   * <li>use the default block size</li>
234   * <li>not track progress</li>
235   * </ol>
236   *
237   * @param fs {@link FileSystem} on which to write the file
238   * @param path {@link Path} to the file to write
239   * @param perm intial permissions
240   * @param overwrite Whether or not the created file should be overwritten.
241   * @return output stream to the created file
242   * @throws IOException if the file cannot be created
243   */
244  public static FSDataOutputStream create(FileSystem fs, Path path,
245      FsPermission perm, boolean overwrite) throws IOException {
246    if (LOG.isTraceEnabled()) {
247      LOG.trace("Creating file=" + path + " with permission=" + perm + ", overwrite=" + overwrite);
248    }
249    return fs.create(path, perm, overwrite, getDefaultBufferSize(fs),
250        getDefaultReplication(fs, path), getDefaultBlockSize(fs, path), null);
251  }
252
253  /**
254   * Get the file permissions specified in the configuration, if they are
255   * enabled.
256   *
257   * @param fs filesystem that the file will be created on.
258   * @param conf configuration to read for determining if permissions are
259   *          enabled and which to use
260   * @param permssionConfKey property key in the configuration to use when
261   *          finding the permission
262   * @return the permission to use when creating a new file on the fs. If
263   *         special permissions are not specified in the configuration, then
264   *         the default permissions on the the fs will be returned.
265   */
266  public static FsPermission getFilePermissions(final FileSystem fs,
267      final Configuration conf, final String permssionConfKey) {
268    boolean enablePermissions = conf.getBoolean(
269        HConstants.ENABLE_DATA_FILE_UMASK, false);
270
271    if (enablePermissions) {
272      try {
273        FsPermission perm = new FsPermission(FULL_RWX_PERMISSIONS);
274        // make sure that we have a mask, if not, go default.
275        String mask = conf.get(permssionConfKey);
276        if (mask == null) {
277          return FsPermission.getFileDefault();
278        }
279        // appy the umask
280        FsPermission umask = new FsPermission(mask);
281        return perm.applyUMask(umask);
282      } catch (IllegalArgumentException e) {
283        LOG.warn(
284            "Incorrect umask attempted to be created: "
285                + conf.get(permssionConfKey)
286                + ", using default file permissions.", e);
287        return FsPermission.getFileDefault();
288      }
289    }
290    return FsPermission.getFileDefault();
291  }
292
293  /**
294   * Verifies root directory path is a valid URI with a scheme
295   *
296   * @param root root directory path
297   * @return Passed <code>root</code> argument.
298   * @throws IOException if not a valid URI with a scheme
299   */
300  public static Path validateRootPath(Path root) throws IOException {
301    try {
302      URI rootURI = new URI(root.toString());
303      String scheme = rootURI.getScheme();
304      if (scheme == null) {
305        throw new IOException("Root directory does not have a scheme");
306      }
307      return root;
308    } catch (URISyntaxException e) {
309      IOException io = new IOException("Root directory path is not a valid " +
310        "URI -- check your " + HConstants.HBASE_DIR + " configuration");
311      io.initCause(e);
312      throw io;
313    }
314  }
315
316  /**
317   * Checks for the presence of the WAL log root path (using the provided conf object) in the given
318   * path. If it exists, this method removes it and returns the String representation of remaining
319   * relative path.
320   * @param path must not be null
321   * @param conf must not be null
322   * @return String representation of the remaining relative path
323   * @throws IOException from underlying filesystem
324   */
325  public static String removeWALRootPath(Path path, final Configuration conf) throws IOException {
326    Path root = getWALRootDir(conf);
327    String pathStr = path.toString();
328    // check that the path is absolute... it has the root path in it.
329    if (!pathStr.startsWith(root.toString())) {
330      return pathStr;
331    }
332    // if not, return as it is.
333    return pathStr.substring(root.toString().length() + 1);// remove the "/" too.
334  }
335
336  /**
337   * Return the 'path' component of a Path.  In Hadoop, Path is an URI.  This
338   * method returns the 'path' component of a Path's URI: e.g. If a Path is
339   * <code>hdfs://example.org:9000/hbase_trunk/TestTable/compaction.dir</code>,
340   * this method returns <code>/hbase_trunk/TestTable/compaction.dir</code>.
341   * This method is useful if you want to print out a Path without qualifying
342   * Filesystem instance.
343   * @param p Filesystem Path whose 'path' component we are to return.
344   * @return Path portion of the Filesystem
345   */
346  public static String getPath(Path p) {
347    return p.toUri().getPath();
348  }
349
350  /**
351   * @param c configuration
352   * @return {@link Path} to hbase root directory from
353   *     configuration as a qualified Path.
354   * @throws IOException e
355   */
356  public static Path getRootDir(final Configuration c) throws IOException {
357    Path p = new Path(c.get(HConstants.HBASE_DIR));
358    FileSystem fs = p.getFileSystem(c);
359    return p.makeQualified(fs.getUri(), fs.getWorkingDirectory());
360  }
361
362  public static void setRootDir(final Configuration c, final Path root) throws IOException {
363    c.set(HConstants.HBASE_DIR, root.toString());
364  }
365
366  public static void setFsDefault(final Configuration c, final Path root) throws IOException {
367    c.set("fs.defaultFS", root.toString());    // for hadoop 0.21+
368  }
369
370  public static FileSystem getRootDirFileSystem(final Configuration c) throws IOException {
371    Path p = getRootDir(c);
372    return p.getFileSystem(c);
373  }
374
375  /**
376   * @param c configuration
377   * @return {@link Path} to hbase log root directory: e.g. {@value HBASE_WAL_DIR} from
378   *     configuration as a qualified Path. Defaults to HBase root dir.
379   * @throws IOException e
380   */
381  public static Path getWALRootDir(final Configuration c) throws IOException {
382    Path p = new Path(c.get(HBASE_WAL_DIR, c.get(HConstants.HBASE_DIR)));
383    if (!isValidWALRootDir(p, c)) {
384      return getRootDir(c);
385    }
386    FileSystem fs = p.getFileSystem(c);
387    return p.makeQualified(fs.getUri(), fs.getWorkingDirectory());
388  }
389
390  @VisibleForTesting
391  public static void setWALRootDir(final Configuration c, final Path root) throws IOException {
392    c.set(HBASE_WAL_DIR, root.toString());
393  }
394
395  public static FileSystem getWALFileSystem(final Configuration c) throws IOException {
396    Path p = getWALRootDir(c);
397    FileSystem fs = p.getFileSystem(c);
398    // hadoop-core does fs caching, so need to propogate this if set
399    String enforceStreamCapability = c.get(UNSAFE_STREAM_CAPABILITY_ENFORCE);
400    if (enforceStreamCapability != null) {
401      fs.getConf().set(UNSAFE_STREAM_CAPABILITY_ENFORCE, enforceStreamCapability);
402    }
403    return fs;
404  }
405
406  private static boolean isValidWALRootDir(Path walDir, final Configuration c) throws IOException {
407    Path rootDir = getRootDir(c);
408    FileSystem fs = walDir.getFileSystem(c);
409    Path qualifiedWalDir = walDir.makeQualified(fs.getUri(), fs.getWorkingDirectory());
410    if (!qualifiedWalDir.equals(rootDir)) {
411      if (qualifiedWalDir.toString().startsWith(rootDir.toString() + "/")) {
412        throw new IllegalStateException("Illegal WAL directory specified. " +
413            "WAL directories are not permitted to be under the root directory if set.");
414      }
415    }
416    return true;
417  }
418
419  /**
420   * Returns the {@link org.apache.hadoop.fs.Path} object representing the table directory under
421   * path rootdir
422   *
423   * @param rootdir qualified path of HBase root directory
424   * @param tableName name of table
425   * @return {@link org.apache.hadoop.fs.Path} for table
426   */
427  public static Path getTableDir(Path rootdir, final TableName tableName) {
428    return new Path(getNamespaceDir(rootdir, tableName.getNamespaceAsString()),
429        tableName.getQualifierAsString());
430  }
431
432  /**
433   * Returns the {@link org.apache.hadoop.hbase.TableName} object representing
434   * the table directory under
435   * path rootdir
436   *
437   * @param tablePath path of table
438   * @return {@link org.apache.hadoop.fs.Path} for table
439   */
440  public static TableName getTableName(Path tablePath) {
441    return TableName.valueOf(tablePath.getParent().getName(), tablePath.getName());
442  }
443
444  /**
445   * Returns the {@link org.apache.hadoop.fs.Path} object representing
446   * the namespace directory under path rootdir
447   *
448   * @param rootdir qualified path of HBase root directory
449   * @param namespace namespace name
450   * @return {@link org.apache.hadoop.fs.Path} for table
451   */
452  public static Path getNamespaceDir(Path rootdir, final String namespace) {
453    return new Path(rootdir, new Path(HConstants.BASE_NAMESPACE_DIR,
454        new Path(namespace)));
455  }
456
457  // this mapping means that under a federated FileSystem implementation, we'll
458  // only log the first failure from any of the underlying FileSystems at WARN and all others
459  // will be at DEBUG.
460  private static final Map<FileSystem, Boolean> warningMap =
461      new ConcurrentHashMap<FileSystem, Boolean>();
462
463  /**
464   * Sets storage policy for given path.
465   * If the passed path is a directory, we'll set the storage policy for all files
466   * created in the future in said directory. Note that this change in storage
467   * policy takes place at the FileSystem level; it will persist beyond this RS's lifecycle.
468   * If we're running on a version of FileSystem that doesn't support the given storage policy
469   * (or storage policies at all), then we'll issue a log message and continue.
470   *
471   * See http://hadoop.apache.org/docs/r2.6.0/hadoop-project-dist/hadoop-hdfs/ArchivalStorage.html
472   *
473   * @param fs We only do anything it implements a setStoragePolicy method
474   * @param path the Path whose storage policy is to be set
475   * @param storagePolicy Policy to set on <code>path</code>; see hadoop 2.6+
476   *   org.apache.hadoop.hdfs.protocol.HdfsConstants for possible list e.g
477   *   'COLD', 'WARM', 'HOT', 'ONE_SSD', 'ALL_SSD', 'LAZY_PERSIST'.
478   */
479  public static void setStoragePolicy(final FileSystem fs, final Path path,
480      final String storagePolicy) {
481    try {
482      setStoragePolicy(fs, path, storagePolicy, false);
483    } catch (IOException e) {
484      // should never arrive here
485      LOG.warn("We have chosen not to throw exception but some unexpectedly thrown out", e);
486    }
487  }
488
489  static void setStoragePolicy(final FileSystem fs, final Path path, final String storagePolicy,
490      boolean throwException) throws IOException {
491    if (storagePolicy == null) {
492      if (LOG.isTraceEnabled()) {
493        LOG.trace("We were passed a null storagePolicy, exiting early.");
494      }
495      return;
496    }
497    String trimmedStoragePolicy = storagePolicy.trim();
498    if (trimmedStoragePolicy.isEmpty()) {
499      if (LOG.isTraceEnabled()) {
500        LOG.trace("We were passed an empty storagePolicy, exiting early.");
501      }
502      return;
503    } else {
504      trimmedStoragePolicy = trimmedStoragePolicy.toUpperCase(Locale.ROOT);
505    }
506    if (trimmedStoragePolicy.equals(HConstants.DEFER_TO_HDFS_STORAGE_POLICY)) {
507      if (LOG.isTraceEnabled()) {
508        LOG.trace("We were passed the defer-to-hdfs policy {}, exiting early.",
509          trimmedStoragePolicy);
510      }
511      return;
512    }
513    try {
514      invokeSetStoragePolicy(fs, path, trimmedStoragePolicy);
515    } catch (IOException e) {
516      if (LOG.isTraceEnabled()) {
517        LOG.trace("Failed to invoke set storage policy API on FS", e);
518      }
519      if (throwException) {
520        throw e;
521      }
522    }
523  }
524
525  /*
526   * All args have been checked and are good. Run the setStoragePolicy invocation.
527   */
528  private static void invokeSetStoragePolicy(final FileSystem fs, final Path path,
529      final String storagePolicy) throws IOException {
530    Method m = null;
531    Exception toThrow = null;
532    try {
533      m = fs.getClass().getDeclaredMethod("setStoragePolicy",
534        new Class<?>[] { Path.class, String.class });
535      m.setAccessible(true);
536    } catch (NoSuchMethodException e) {
537      toThrow = e;
538      final String msg = "FileSystem doesn't support setStoragePolicy; HDFS-6584, HDFS-9345 " +
539          "not available. This is normal and expected on earlier Hadoop versions.";
540      if (!warningMap.containsKey(fs)) {
541        warningMap.put(fs, true);
542        LOG.warn(msg, e);
543      } else if (LOG.isDebugEnabled()) {
544        LOG.debug(msg, e);
545      }
546      m = null;
547    } catch (SecurityException e) {
548      toThrow = e;
549      final String msg = "No access to setStoragePolicy on FileSystem from the SecurityManager; " +
550          "HDFS-6584, HDFS-9345 not available. This is unusual and probably warrants an email " +
551          "to the user@hbase mailing list. Please be sure to include a link to your configs, and " +
552          "logs that include this message and period of time before it. Logs around service " +
553          "start up will probably be useful as well.";
554      if (!warningMap.containsKey(fs)) {
555        warningMap.put(fs, true);
556        LOG.warn(msg, e);
557      } else if (LOG.isDebugEnabled()) {
558        LOG.debug(msg, e);
559      }
560      m = null; // could happen on setAccessible() or getDeclaredMethod()
561    }
562    if (m != null) {
563      try {
564        m.invoke(fs, path, storagePolicy);
565        if (LOG.isDebugEnabled()) {
566          LOG.debug("Set storagePolicy=" + storagePolicy + " for path=" + path);
567        }
568      } catch (Exception e) {
569        toThrow = e;
570        // This swallows FNFE, should we be throwing it? seems more likely to indicate dev
571        // misuse than a runtime problem with HDFS.
572        if (!warningMap.containsKey(fs)) {
573          warningMap.put(fs, true);
574          LOG.warn("Unable to set storagePolicy=" + storagePolicy + " for path=" + path + ". " +
575              "DEBUG log level might have more details.", e);
576        } else if (LOG.isDebugEnabled()) {
577          LOG.debug("Unable to set storagePolicy=" + storagePolicy + " for path=" + path, e);
578        }
579        // check for lack of HDFS-7228
580        if (e instanceof InvocationTargetException) {
581          final Throwable exception = e.getCause();
582          if (exception instanceof RemoteException &&
583              HadoopIllegalArgumentException.class.getName().equals(
584                ((RemoteException)exception).getClassName())) {
585            if (LOG.isDebugEnabled()) {
586              LOG.debug("Given storage policy, '" +storagePolicy +"', was rejected and probably " +
587                "isn't a valid policy for the version of Hadoop you're running. I.e. if you're " +
588                "trying to use SSD related policies then you're likely missing HDFS-7228. For " +
589                "more information see the 'ArchivalStorage' docs for your Hadoop release.");
590            }
591          // Hadoop 2.8+, 3.0-a1+ added FileSystem.setStoragePolicy with a default implementation
592          // that throws UnsupportedOperationException
593          } else if (exception instanceof UnsupportedOperationException) {
594            if (LOG.isDebugEnabled()) {
595              LOG.debug("The underlying FileSystem implementation doesn't support " +
596                  "setStoragePolicy. This is probably intentional on their part, since HDFS-9345 " +
597                  "appears to be present in your version of Hadoop. For more information check " +
598                  "the Hadoop documentation on 'ArchivalStorage', the Hadoop FileSystem " +
599                  "specification docs from HADOOP-11981, and/or related documentation from the " +
600                  "provider of the underlying FileSystem (its name should appear in the " +
601                  "stacktrace that accompanies this message). Note in particular that Hadoop's " +
602                  "local filesystem implementation doesn't support storage policies.", exception);
603            }
604          }
605        }
606      }
607    }
608    if (toThrow != null) {
609      throw new IOException(toThrow);
610    }
611  }
612
613  /**
614   * @param conf must not be null
615   * @return True if this filesystem whose scheme is 'hdfs'.
616   * @throws IOException from underlying FileSystem
617   */
618  public static boolean isHDFS(final Configuration conf) throws IOException {
619    FileSystem fs = FileSystem.get(conf);
620    String scheme = fs.getUri().getScheme();
621    return scheme.equalsIgnoreCase("hdfs");
622  }
623
624  /**
625   * Checks if the given path is the one with 'recovered.edits' dir.
626   * @param path must not be null
627   * @return True if we recovered edits
628   */
629  public static boolean isRecoveredEdits(Path path) {
630    return path.toString().contains(HConstants.RECOVERED_EDITS_DIR);
631  }
632
633  /**
634   * @param conf must not be null
635   * @return Returns the filesystem of the hbase rootdir.
636   * @throws IOException from underlying FileSystem
637   */
638  public static FileSystem getCurrentFileSystem(Configuration conf)
639  throws IOException {
640    return getRootDir(conf).getFileSystem(conf);
641  }
642
643  /**
644   * Calls fs.listStatus() and treats FileNotFoundException as non-fatal
645   * This accommodates differences between hadoop versions, where hadoop 1
646   * does not throw a FileNotFoundException, and return an empty FileStatus[]
647   * while Hadoop 2 will throw FileNotFoundException.
648   *
649   * Where possible, prefer FSUtils#listStatusWithStatusFilter(FileSystem,
650   * Path, FileStatusFilter) instead.
651   *
652   * @param fs file system
653   * @param dir directory
654   * @param filter path filter
655   * @return null if dir is empty or doesn't exist, otherwise FileStatus array
656   */
657  public static FileStatus [] listStatus(final FileSystem fs,
658      final Path dir, final PathFilter filter) throws IOException {
659    FileStatus [] status = null;
660    try {
661      status = filter == null ? fs.listStatus(dir) : fs.listStatus(dir, filter);
662    } catch (FileNotFoundException fnfe) {
663      // if directory doesn't exist, return null
664      if (LOG.isTraceEnabled()) {
665        LOG.trace(dir + " doesn't exist");
666      }
667    }
668    if (status == null || status.length < 1) {
669      return null;
670    }
671    return status;
672  }
673
674  /**
675   * Calls fs.listStatus() and treats FileNotFoundException as non-fatal
676   * This would accommodates differences between hadoop versions
677   *
678   * @param fs file system
679   * @param dir directory
680   * @return null if dir is empty or doesn't exist, otherwise FileStatus array
681   */
682  public static FileStatus[] listStatus(final FileSystem fs, final Path dir) throws IOException {
683    return listStatus(fs, dir, null);
684  }
685
686  /**
687   * Calls fs.listFiles() to get FileStatus and BlockLocations together for reducing rpc call
688   *
689   * @param fs file system
690   * @param dir directory
691   * @return LocatedFileStatus list
692   */
693  public static List<LocatedFileStatus> listLocatedStatus(final FileSystem fs,
694      final Path dir) throws IOException {
695    List<LocatedFileStatus> status = null;
696    try {
697      RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = fs
698          .listFiles(dir, false);
699      while (locatedFileStatusRemoteIterator.hasNext()) {
700        if (status == null) {
701          status = Lists.newArrayList();
702        }
703        status.add(locatedFileStatusRemoteIterator.next());
704      }
705    } catch (FileNotFoundException fnfe) {
706      // if directory doesn't exist, return null
707      if (LOG.isTraceEnabled()) {
708        LOG.trace(dir + " doesn't exist");
709      }
710    }
711    return status;
712  }
713
714  /**
715   * Calls fs.delete() and returns the value returned by the fs.delete()
716   *
717   * @param fs must not be null
718   * @param path must not be null
719   * @param recursive delete tree rooted at path
720   * @return the value returned by the fs.delete()
721   * @throws IOException from underlying FileSystem
722   */
723  public static boolean delete(final FileSystem fs, final Path path, final boolean recursive)
724      throws IOException {
725    return fs.delete(path, recursive);
726  }
727
728  /**
729   * Calls fs.exists(). Checks if the specified path exists
730   *
731   * @param fs must not be null
732   * @param path must not be null
733   * @return the value returned by fs.exists()
734   * @throws IOException from underlying FileSystem
735   */
736  public static boolean isExists(final FileSystem fs, final Path path) throws IOException {
737    return fs.exists(path);
738  }
739
740  /**
741   * Log the current state of the filesystem from a certain root directory
742   * @param fs filesystem to investigate
743   * @param root root file/directory to start logging from
744   * @param LOG log to output information
745   * @throws IOException if an unexpected exception occurs
746   */
747  public static void logFileSystemState(final FileSystem fs, final Path root, Logger LOG)
748      throws IOException {
749    LOG.debug("File system contents for path " + root);
750    logFSTree(LOG, fs, root, "|-");
751  }
752
753  /**
754   * Recursive helper to log the state of the FS
755   *
756   * @see #logFileSystemState(FileSystem, Path, Logger)
757   */
758  private static void logFSTree(Logger LOG, final FileSystem fs, final Path root, String prefix)
759      throws IOException {
760    FileStatus[] files = listStatus(fs, root, null);
761    if (files == null) {
762      return;
763    }
764
765    for (FileStatus file : files) {
766      if (file.isDirectory()) {
767        LOG.debug(prefix + file.getPath().getName() + "/");
768        logFSTree(LOG, fs, file.getPath(), prefix + "---");
769      } else {
770        LOG.debug(prefix + file.getPath().getName());
771      }
772    }
773  }
774
775  public static boolean renameAndSetModifyTime(final FileSystem fs, final Path src, final Path dest)
776      throws IOException {
777    // set the modify time for TimeToLive Cleaner
778    fs.setTimes(src, EnvironmentEdgeManager.currentTime(), -1);
779    return fs.rename(src, dest);
780  }
781
782  /**
783   * Do our short circuit read setup.
784   * Checks buffer size to use and whether to do checksumming in hbase or hdfs.
785   * @param conf must not be null
786   */
787  public static void setupShortCircuitRead(final Configuration conf) {
788    // Check that the user has not set the "dfs.client.read.shortcircuit.skip.checksum" property.
789    boolean shortCircuitSkipChecksum =
790      conf.getBoolean("dfs.client.read.shortcircuit.skip.checksum", false);
791    boolean useHBaseChecksum = conf.getBoolean(HConstants.HBASE_CHECKSUM_VERIFICATION, true);
792    if (shortCircuitSkipChecksum) {
793      LOG.warn("Configuration \"dfs.client.read.shortcircuit.skip.checksum\" should not " +
794        "be set to true." + (useHBaseChecksum ? " HBase checksum doesn't require " +
795        "it, see https://issues.apache.org/jira/browse/HBASE-6868." : ""));
796      assert !shortCircuitSkipChecksum; //this will fail if assertions are on
797    }
798    checkShortCircuitReadBufferSize(conf);
799  }
800
801  /**
802   * Check if short circuit read buffer size is set and if not, set it to hbase value.
803   * @param conf must not be null
804   */
805  public static void checkShortCircuitReadBufferSize(final Configuration conf) {
806    final int defaultSize = HConstants.DEFAULT_BLOCKSIZE * 2;
807    final int notSet = -1;
808    // DFSConfigKeys.DFS_CLIENT_READ_SHORTCIRCUIT_BUFFER_SIZE_KEY is only defined in h2
809    final String dfsKey = "dfs.client.read.shortcircuit.buffer.size";
810    int size = conf.getInt(dfsKey, notSet);
811    // If a size is set, return -- we will use it.
812    if (size != notSet) {
813      return;
814    }
815    // But short circuit buffer size is normally not set.  Put in place the hbase wanted size.
816    int hbaseSize = conf.getInt("hbase." + dfsKey, defaultSize);
817    conf.setIfUnset(dfsKey, Integer.toString(hbaseSize));
818  }
819
820  private static class DfsBuilderUtility {
821    static Class<?> dfsClass = null;
822    static Method createMethod;
823    static Method overwriteMethod;
824    static Method bufferSizeMethod;
825    static Method blockSizeMethod;
826    static Method recursiveMethod;
827    static Method replicateMethod;
828    static Method replicationMethod;
829    static Method buildMethod;
830    static boolean allMethodsPresent = false;
831
832    static {
833      String dfsName = "org.apache.hadoop.hdfs.DistributedFileSystem";
834      String builderName = dfsName + "$HdfsDataOutputStreamBuilder";
835      Class<?> builderClass = null;
836
837      try {
838        dfsClass = Class.forName(dfsName);
839      } catch (ClassNotFoundException e) {
840        LOG.debug("{} not available, will not use builder API for file creation.", dfsName);
841      }
842      try {
843        builderClass = Class.forName(builderName);
844      } catch (ClassNotFoundException e) {
845        LOG.debug("{} not available, will not use builder API for file creation.", builderName);
846      }
847
848      if (dfsClass != null && builderClass != null) {
849        try {
850          createMethod = dfsClass.getMethod("createFile", Path.class);
851          overwriteMethod = builderClass.getMethod("overwrite", boolean.class);
852          bufferSizeMethod = builderClass.getMethod("bufferSize", int.class);
853          blockSizeMethod = builderClass.getMethod("blockSize", long.class);
854          recursiveMethod = builderClass.getMethod("recursive");
855          replicateMethod = builderClass.getMethod("replicate");
856          replicationMethod = builderClass.getMethod("replication", short.class);
857          buildMethod = builderClass.getMethod("build");
858
859          allMethodsPresent = true;
860          LOG.debug("Using builder API via reflection for DFS file creation.");
861        } catch (NoSuchMethodException e) {
862          LOG.debug("Could not find method on builder; will use old DFS API for file creation {}",
863              e.getMessage());
864        }
865      }
866    }
867
868    /**
869     * Attempt to use builder API via reflection to create a file with the given parameters and
870     * replication enabled.
871     */
872    static FSDataOutputStream createHelper(FileSystem fs, Path path, boolean overwritable,
873        int bufferSize, short replication, long blockSize, boolean isRecursive) throws IOException {
874      if (allMethodsPresent && dfsClass.isInstance(fs)) {
875        try {
876          Object builder;
877
878          builder = createMethod.invoke(fs, path);
879          builder = overwriteMethod.invoke(builder, overwritable);
880          builder = bufferSizeMethod.invoke(builder, bufferSize);
881          builder = blockSizeMethod.invoke(builder, blockSize);
882          if (isRecursive) {
883            builder = recursiveMethod.invoke(builder);
884          }
885          builder = replicateMethod.invoke(builder);
886          builder = replicationMethod.invoke(builder, replication);
887          return (FSDataOutputStream) buildMethod.invoke(builder);
888        } catch (IllegalAccessException | InvocationTargetException e) {
889          // Should have caught this failure during initialization, so log full trace here
890          LOG.warn("Couldn't use reflection with builder API", e);
891        }
892      }
893
894      if (isRecursive) {
895        return fs.create(path, overwritable, bufferSize, replication, blockSize, null);
896      }
897      return fs.createNonRecursive(path, overwritable, bufferSize, replication, blockSize, null);
898    }
899
900    /**
901     * Attempt to use builder API via reflection to create a file with the given parameters and
902     * replication enabled.
903     */
904    static FSDataOutputStream createHelper(FileSystem fs, Path path, boolean overwritable)
905        throws IOException {
906      if (allMethodsPresent && dfsClass.isInstance(fs)) {
907        try {
908          Object builder;
909
910          builder = createMethod.invoke(fs, path);
911          builder = overwriteMethod.invoke(builder, overwritable);
912          builder = replicateMethod.invoke(builder);
913          return (FSDataOutputStream) buildMethod.invoke(builder);
914        } catch (IllegalAccessException | InvocationTargetException e) {
915          // Should have caught this failure during initialization, so log full trace here
916          LOG.warn("Couldn't use reflection with builder API", e);
917        }
918      }
919
920      return fs.create(path, overwritable);
921    }
922  }
923
924  /**
925   * Attempt to use builder API via reflection to create a file with the given parameters and
926   * replication enabled.
927   * <p>
928   * Will not attempt to enable replication when passed an HFileSystem.
929   */
930  public static FSDataOutputStream createForWal(FileSystem fs, Path path, boolean overwritable)
931      throws IOException {
932    return DfsBuilderUtility.createHelper(fs, path, overwritable);
933  }
934
935  /**
936   * Attempt to use builder API via reflection to create a file with the given parameters and
937   * replication enabled.
938   * <p>
939   * Will not attempt to enable replication when passed an HFileSystem.
940   */
941  public static FSDataOutputStream createForWal(FileSystem fs, Path path, boolean overwritable,
942      int bufferSize, short replication, long blockSize, boolean isRecursive) throws IOException {
943    return DfsBuilderUtility.createHelper(fs, path, overwritable, bufferSize, replication,
944        blockSize, isRecursive);
945  }
946
947  // Holder singleton idiom. JVM spec ensures this will be run at most once per Classloader, and
948  // not until we attempt to reference it.
949  private static class StreamCapabilities {
950    public static final boolean PRESENT;
951    public static final Class<?> CLASS;
952    public static final Method METHOD;
953    static {
954      boolean tmp = false;
955      Class<?> clazz = null;
956      Method method = null;
957      try {
958        clazz = Class.forName("org.apache.hadoop.fs.StreamCapabilities");
959        method = clazz.getMethod("hasCapability", String.class);
960        tmp = true;
961      } catch(ClassNotFoundException|NoSuchMethodException|SecurityException exception) {
962        LOG.warn("Your Hadoop installation does not include the StreamCapabilities class from " +
963                 "HDFS-11644, so we will skip checking if any FSDataOutputStreams actually " +
964                 "support hflush/hsync. If you are running on top of HDFS this probably just " +
965                 "means you have an older version and this can be ignored. If you are running on " +
966                 "top of an alternate FileSystem implementation you should manually verify that " +
967                 "hflush and hsync are implemented; otherwise you risk data loss and hard to " +
968                 "diagnose errors when our assumptions are violated.");
969        LOG.debug("The first request to check for StreamCapabilities came from this stacktrace.",
970            exception);
971      } finally {
972        PRESENT = tmp;
973        CLASS = clazz;
974        METHOD = method;
975      }
976    }
977  }
978
979  /**
980   * If our FileSystem version includes the StreamCapabilities class, check if
981   * the given stream has a particular capability.
982   * @param stream capabilities are per-stream instance, so check this one specifically. must not be
983   *        null
984   * @param capability what to look for, per Hadoop Common's FileSystem docs
985   * @return true if there are no StreamCapabilities. false if there are, but this stream doesn't
986   *         implement it. return result of asking the stream otherwise.
987   */
988  public static boolean hasCapability(FSDataOutputStream stream, String capability) {
989    // be consistent whether or not StreamCapabilities is present
990    if (stream == null) {
991      throw new NullPointerException("stream parameter must not be null.");
992    }
993    // If o.a.h.fs.StreamCapabilities doesn't exist, assume everyone does everything
994    // otherwise old versions of Hadoop will break.
995    boolean result = true;
996    if (StreamCapabilities.PRESENT) {
997      // if StreamCapabilities is present, but the stream doesn't implement it
998      // or we run into a problem invoking the method,
999      // we treat that as equivalent to not declaring anything
1000      result = false;
1001      if (StreamCapabilities.CLASS.isAssignableFrom(stream.getClass())) {
1002        try {
1003          result = ((Boolean)StreamCapabilities.METHOD.invoke(stream, capability)).booleanValue();
1004        } catch (IllegalAccessException|IllegalArgumentException|InvocationTargetException
1005            exception) {
1006          LOG.warn("Your Hadoop installation's StreamCapabilities implementation doesn't match " +
1007              "our understanding of how it's supposed to work. Please file a JIRA and include " +
1008              "the following stack trace. In the mean time we're interpreting this behavior " +
1009              "difference as a lack of capability support, which will probably cause a failure.",
1010              exception);
1011        }
1012      }
1013    }
1014    return result;
1015  }
1016
1017  /**
1018   * Helper exception for those cases where the place where we need to check a stream capability
1019   * is not where we have the needed context to explain the impact and mitigation for a lack.
1020   */
1021  public static class StreamLacksCapabilityException extends Exception {
1022    public StreamLacksCapabilityException(String message, Throwable cause) {
1023      super(message, cause);
1024    }
1025    public StreamLacksCapabilityException(String message) {
1026      super(message);
1027    }
1028  }
1029
1030}