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 WAL region directory based on the given table name and region name
421   * @param conf configuration to determine WALRootDir
422   * @param tableName Table that the region is under
423   * @param encodedRegionName Region name used for creating the final region directory
424   * @return the region directory used to store WALs under the WALRootDir
425   * @throws IOException if there is an exception determining the WALRootDir
426   */
427  public static Path getWALRegionDir(final Configuration conf,
428      final TableName tableName, final String encodedRegionName)
429      throws IOException {
430    return new Path(getWALTableDir(conf, tableName),
431        encodedRegionName);
432  }
433
434  /**
435   * Returns the Table directory under the WALRootDir for the specified table name
436   * @param conf configuration used to get the WALRootDir
437   * @param tableName Table to get the directory for
438   * @return a path to the WAL table directory for the specified table
439   * @throws IOException if there is an exception determining the WALRootDir
440   */
441  public static Path getWALTableDir(final Configuration conf, final TableName tableName)
442      throws IOException {
443    return new Path(new Path(getWALRootDir(conf), tableName.getNamespaceAsString()),
444        tableName.getQualifierAsString());
445  }
446
447  /**
448   * Returns the {@link org.apache.hadoop.fs.Path} object representing the table directory under
449   * path rootdir
450   *
451   * @param rootdir qualified path of HBase root directory
452   * @param tableName name of table
453   * @return {@link org.apache.hadoop.fs.Path} for table
454   */
455  public static Path getTableDir(Path rootdir, final TableName tableName) {
456    return new Path(getNamespaceDir(rootdir, tableName.getNamespaceAsString()),
457        tableName.getQualifierAsString());
458  }
459
460  /**
461   * Returns the {@link org.apache.hadoop.hbase.TableName} object representing
462   * the table directory under
463   * path rootdir
464   *
465   * @param tablePath path of table
466   * @return {@link org.apache.hadoop.fs.Path} for table
467   */
468  public static TableName getTableName(Path tablePath) {
469    return TableName.valueOf(tablePath.getParent().getName(), tablePath.getName());
470  }
471
472  /**
473   * Returns the {@link org.apache.hadoop.fs.Path} object representing
474   * the namespace directory under path rootdir
475   *
476   * @param rootdir qualified path of HBase root directory
477   * @param namespace namespace name
478   * @return {@link org.apache.hadoop.fs.Path} for table
479   */
480  public static Path getNamespaceDir(Path rootdir, final String namespace) {
481    return new Path(rootdir, new Path(HConstants.BASE_NAMESPACE_DIR,
482        new Path(namespace)));
483  }
484
485  // this mapping means that under a federated FileSystem implementation, we'll
486  // only log the first failure from any of the underlying FileSystems at WARN and all others
487  // will be at DEBUG.
488  private static final Map<FileSystem, Boolean> warningMap =
489      new ConcurrentHashMap<FileSystem, Boolean>();
490
491  /**
492   * Sets storage policy for given path.
493   * If the passed path is a directory, we'll set the storage policy for all files
494   * created in the future in said directory. Note that this change in storage
495   * policy takes place at the FileSystem level; it will persist beyond this RS's lifecycle.
496   * If we're running on a version of FileSystem that doesn't support the given storage policy
497   * (or storage policies at all), then we'll issue a log message and continue.
498   *
499   * See http://hadoop.apache.org/docs/r2.6.0/hadoop-project-dist/hadoop-hdfs/ArchivalStorage.html
500   *
501   * @param fs We only do anything it implements a setStoragePolicy method
502   * @param path the Path whose storage policy is to be set
503   * @param storagePolicy Policy to set on <code>path</code>; see hadoop 2.6+
504   *   org.apache.hadoop.hdfs.protocol.HdfsConstants for possible list e.g
505   *   'COLD', 'WARM', 'HOT', 'ONE_SSD', 'ALL_SSD', 'LAZY_PERSIST'.
506   */
507  public static void setStoragePolicy(final FileSystem fs, final Path path,
508      final String storagePolicy) {
509    try {
510      setStoragePolicy(fs, path, storagePolicy, false);
511    } catch (IOException e) {
512      // should never arrive here
513      LOG.warn("We have chosen not to throw exception but some unexpectedly thrown out", e);
514    }
515  }
516
517  static void setStoragePolicy(final FileSystem fs, final Path path, final String storagePolicy,
518      boolean throwException) throws IOException {
519    if (storagePolicy == null) {
520      if (LOG.isTraceEnabled()) {
521        LOG.trace("We were passed a null storagePolicy, exiting early.");
522      }
523      return;
524    }
525    String trimmedStoragePolicy = storagePolicy.trim();
526    if (trimmedStoragePolicy.isEmpty()) {
527      if (LOG.isTraceEnabled()) {
528        LOG.trace("We were passed an empty storagePolicy, exiting early.");
529      }
530      return;
531    } else {
532      trimmedStoragePolicy = trimmedStoragePolicy.toUpperCase(Locale.ROOT);
533    }
534    if (trimmedStoragePolicy.equals(HConstants.DEFER_TO_HDFS_STORAGE_POLICY)) {
535      if (LOG.isTraceEnabled()) {
536        LOG.trace("We were passed the defer-to-hdfs policy {}, exiting early.",
537          trimmedStoragePolicy);
538      }
539      return;
540    }
541    try {
542      invokeSetStoragePolicy(fs, path, trimmedStoragePolicy);
543    } catch (IOException e) {
544      if (LOG.isTraceEnabled()) {
545        LOG.trace("Failed to invoke set storage policy API on FS", e);
546      }
547      if (throwException) {
548        throw e;
549      }
550    }
551  }
552
553  /*
554   * All args have been checked and are good. Run the setStoragePolicy invocation.
555   */
556  private static void invokeSetStoragePolicy(final FileSystem fs, final Path path,
557      final String storagePolicy) throws IOException {
558    Method m = null;
559    Exception toThrow = null;
560    try {
561      m = fs.getClass().getDeclaredMethod("setStoragePolicy",
562        new Class<?>[] { Path.class, String.class });
563      m.setAccessible(true);
564    } catch (NoSuchMethodException e) {
565      toThrow = e;
566      final String msg = "FileSystem doesn't support setStoragePolicy; HDFS-6584, HDFS-9345 " +
567          "not available. This is normal and expected on earlier Hadoop versions.";
568      if (!warningMap.containsKey(fs)) {
569        warningMap.put(fs, true);
570        LOG.warn(msg, e);
571      } else if (LOG.isDebugEnabled()) {
572        LOG.debug(msg, e);
573      }
574      m = null;
575    } catch (SecurityException e) {
576      toThrow = e;
577      final String msg = "No access to setStoragePolicy on FileSystem from the SecurityManager; " +
578          "HDFS-6584, HDFS-9345 not available. This is unusual and probably warrants an email " +
579          "to the user@hbase mailing list. Please be sure to include a link to your configs, and " +
580          "logs that include this message and period of time before it. Logs around service " +
581          "start up will probably be useful as well.";
582      if (!warningMap.containsKey(fs)) {
583        warningMap.put(fs, true);
584        LOG.warn(msg, e);
585      } else if (LOG.isDebugEnabled()) {
586        LOG.debug(msg, e);
587      }
588      m = null; // could happen on setAccessible() or getDeclaredMethod()
589    }
590    if (m != null) {
591      try {
592        m.invoke(fs, path, storagePolicy);
593        if (LOG.isDebugEnabled()) {
594          LOG.debug("Set storagePolicy=" + storagePolicy + " for path=" + path);
595        }
596      } catch (Exception e) {
597        toThrow = e;
598        // This swallows FNFE, should we be throwing it? seems more likely to indicate dev
599        // misuse than a runtime problem with HDFS.
600        if (!warningMap.containsKey(fs)) {
601          warningMap.put(fs, true);
602          LOG.warn("Unable to set storagePolicy=" + storagePolicy + " for path=" + path + ". " +
603              "DEBUG log level might have more details.", e);
604        } else if (LOG.isDebugEnabled()) {
605          LOG.debug("Unable to set storagePolicy=" + storagePolicy + " for path=" + path, e);
606        }
607        // check for lack of HDFS-7228
608        if (e instanceof InvocationTargetException) {
609          final Throwable exception = e.getCause();
610          if (exception instanceof RemoteException &&
611              HadoopIllegalArgumentException.class.getName().equals(
612                ((RemoteException)exception).getClassName())) {
613            if (LOG.isDebugEnabled()) {
614              LOG.debug("Given storage policy, '" +storagePolicy +"', was rejected and probably " +
615                "isn't a valid policy for the version of Hadoop you're running. I.e. if you're " +
616                "trying to use SSD related policies then you're likely missing HDFS-7228. For " +
617                "more information see the 'ArchivalStorage' docs for your Hadoop release.");
618            }
619          // Hadoop 2.8+, 3.0-a1+ added FileSystem.setStoragePolicy with a default implementation
620          // that throws UnsupportedOperationException
621          } else if (exception instanceof UnsupportedOperationException) {
622            if (LOG.isDebugEnabled()) {
623              LOG.debug("The underlying FileSystem implementation doesn't support " +
624                  "setStoragePolicy. This is probably intentional on their part, since HDFS-9345 " +
625                  "appears to be present in your version of Hadoop. For more information check " +
626                  "the Hadoop documentation on 'ArchivalStorage', the Hadoop FileSystem " +
627                  "specification docs from HADOOP-11981, and/or related documentation from the " +
628                  "provider of the underlying FileSystem (its name should appear in the " +
629                  "stacktrace that accompanies this message). Note in particular that Hadoop's " +
630                  "local filesystem implementation doesn't support storage policies.", exception);
631            }
632          }
633        }
634      }
635    }
636    if (toThrow != null) {
637      throw new IOException(toThrow);
638    }
639  }
640
641  /**
642   * @param conf must not be null
643   * @return True if this filesystem whose scheme is 'hdfs'.
644   * @throws IOException from underlying FileSystem
645   */
646  public static boolean isHDFS(final Configuration conf) throws IOException {
647    FileSystem fs = FileSystem.get(conf);
648    String scheme = fs.getUri().getScheme();
649    return scheme.equalsIgnoreCase("hdfs");
650  }
651
652  /**
653   * Checks if the given path is the one with 'recovered.edits' dir.
654   * @param path must not be null
655   * @return True if we recovered edits
656   */
657  public static boolean isRecoveredEdits(Path path) {
658    return path.toString().contains(HConstants.RECOVERED_EDITS_DIR);
659  }
660
661  /**
662   * @param conf must not be null
663   * @return Returns the filesystem of the hbase rootdir.
664   * @throws IOException from underlying FileSystem
665   */
666  public static FileSystem getCurrentFileSystem(Configuration conf)
667  throws IOException {
668    return getRootDir(conf).getFileSystem(conf);
669  }
670
671  /**
672   * Calls fs.listStatus() and treats FileNotFoundException as non-fatal
673   * This accommodates differences between hadoop versions, where hadoop 1
674   * does not throw a FileNotFoundException, and return an empty FileStatus[]
675   * while Hadoop 2 will throw FileNotFoundException.
676   *
677   * Where possible, prefer FSUtils#listStatusWithStatusFilter(FileSystem,
678   * Path, FileStatusFilter) instead.
679   *
680   * @param fs file system
681   * @param dir directory
682   * @param filter path filter
683   * @return null if dir is empty or doesn't exist, otherwise FileStatus array
684   */
685  public static FileStatus [] listStatus(final FileSystem fs,
686      final Path dir, final PathFilter filter) throws IOException {
687    FileStatus [] status = null;
688    try {
689      status = filter == null ? fs.listStatus(dir) : fs.listStatus(dir, filter);
690    } catch (FileNotFoundException fnfe) {
691      // if directory doesn't exist, return null
692      if (LOG.isTraceEnabled()) {
693        LOG.trace(dir + " doesn't exist");
694      }
695    }
696    if (status == null || status.length < 1) {
697      return null;
698    }
699    return status;
700  }
701
702  /**
703   * Calls fs.listStatus() and treats FileNotFoundException as non-fatal
704   * This would accommodates differences between hadoop versions
705   *
706   * @param fs file system
707   * @param dir directory
708   * @return null if dir is empty or doesn't exist, otherwise FileStatus array
709   */
710  public static FileStatus[] listStatus(final FileSystem fs, final Path dir) throws IOException {
711    return listStatus(fs, dir, null);
712  }
713
714  /**
715   * Calls fs.listFiles() to get FileStatus and BlockLocations together for reducing rpc call
716   *
717   * @param fs file system
718   * @param dir directory
719   * @return LocatedFileStatus list
720   */
721  public static List<LocatedFileStatus> listLocatedStatus(final FileSystem fs,
722      final Path dir) throws IOException {
723    List<LocatedFileStatus> status = null;
724    try {
725      RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = fs
726          .listFiles(dir, false);
727      while (locatedFileStatusRemoteIterator.hasNext()) {
728        if (status == null) {
729          status = Lists.newArrayList();
730        }
731        status.add(locatedFileStatusRemoteIterator.next());
732      }
733    } catch (FileNotFoundException fnfe) {
734      // if directory doesn't exist, return null
735      if (LOG.isTraceEnabled()) {
736        LOG.trace(dir + " doesn't exist");
737      }
738    }
739    return status;
740  }
741
742  /**
743   * Calls fs.delete() and returns the value returned by the fs.delete()
744   *
745   * @param fs must not be null
746   * @param path must not be null
747   * @param recursive delete tree rooted at path
748   * @return the value returned by the fs.delete()
749   * @throws IOException from underlying FileSystem
750   */
751  public static boolean delete(final FileSystem fs, final Path path, final boolean recursive)
752      throws IOException {
753    return fs.delete(path, recursive);
754  }
755
756  /**
757   * Calls fs.exists(). Checks if the specified path exists
758   *
759   * @param fs must not be null
760   * @param path must not be null
761   * @return the value returned by fs.exists()
762   * @throws IOException from underlying FileSystem
763   */
764  public static boolean isExists(final FileSystem fs, final Path path) throws IOException {
765    return fs.exists(path);
766  }
767
768  /**
769   * Log the current state of the filesystem from a certain root directory
770   * @param fs filesystem to investigate
771   * @param root root file/directory to start logging from
772   * @param LOG log to output information
773   * @throws IOException if an unexpected exception occurs
774   */
775  public static void logFileSystemState(final FileSystem fs, final Path root, Logger LOG)
776      throws IOException {
777    LOG.debug("File system contents for path " + root);
778    logFSTree(LOG, fs, root, "|-");
779  }
780
781  /**
782   * Recursive helper to log the state of the FS
783   *
784   * @see #logFileSystemState(FileSystem, Path, Logger)
785   */
786  private static void logFSTree(Logger LOG, final FileSystem fs, final Path root, String prefix)
787      throws IOException {
788    FileStatus[] files = listStatus(fs, root, null);
789    if (files == null) {
790      return;
791    }
792
793    for (FileStatus file : files) {
794      if (file.isDirectory()) {
795        LOG.debug(prefix + file.getPath().getName() + "/");
796        logFSTree(LOG, fs, file.getPath(), prefix + "---");
797      } else {
798        LOG.debug(prefix + file.getPath().getName());
799      }
800    }
801  }
802
803  public static boolean renameAndSetModifyTime(final FileSystem fs, final Path src, final Path dest)
804      throws IOException {
805    // set the modify time for TimeToLive Cleaner
806    fs.setTimes(src, EnvironmentEdgeManager.currentTime(), -1);
807    return fs.rename(src, dest);
808  }
809
810  /**
811   * Do our short circuit read setup.
812   * Checks buffer size to use and whether to do checksumming in hbase or hdfs.
813   * @param conf must not be null
814   */
815  public static void setupShortCircuitRead(final Configuration conf) {
816    // Check that the user has not set the "dfs.client.read.shortcircuit.skip.checksum" property.
817    boolean shortCircuitSkipChecksum =
818      conf.getBoolean("dfs.client.read.shortcircuit.skip.checksum", false);
819    boolean useHBaseChecksum = conf.getBoolean(HConstants.HBASE_CHECKSUM_VERIFICATION, true);
820    if (shortCircuitSkipChecksum) {
821      LOG.warn("Configuration \"dfs.client.read.shortcircuit.skip.checksum\" should not " +
822        "be set to true." + (useHBaseChecksum ? " HBase checksum doesn't require " +
823        "it, see https://issues.apache.org/jira/browse/HBASE-6868." : ""));
824      assert !shortCircuitSkipChecksum; //this will fail if assertions are on
825    }
826    checkShortCircuitReadBufferSize(conf);
827  }
828
829  /**
830   * Check if short circuit read buffer size is set and if not, set it to hbase value.
831   * @param conf must not be null
832   */
833  public static void checkShortCircuitReadBufferSize(final Configuration conf) {
834    final int defaultSize = HConstants.DEFAULT_BLOCKSIZE * 2;
835    final int notSet = -1;
836    // DFSConfigKeys.DFS_CLIENT_READ_SHORTCIRCUIT_BUFFER_SIZE_KEY is only defined in h2
837    final String dfsKey = "dfs.client.read.shortcircuit.buffer.size";
838    int size = conf.getInt(dfsKey, notSet);
839    // If a size is set, return -- we will use it.
840    if (size != notSet) {
841      return;
842    }
843    // But short circuit buffer size is normally not set.  Put in place the hbase wanted size.
844    int hbaseSize = conf.getInt("hbase." + dfsKey, defaultSize);
845    conf.setIfUnset(dfsKey, Integer.toString(hbaseSize));
846  }
847
848  private static class DfsBuilderUtility {
849    static Class<?> dfsClass = null;
850    static Method createMethod;
851    static Method overwriteMethod;
852    static Method bufferSizeMethod;
853    static Method blockSizeMethod;
854    static Method recursiveMethod;
855    static Method replicateMethod;
856    static Method replicationMethod;
857    static Method buildMethod;
858    static boolean allMethodsPresent = false;
859
860    static {
861      String dfsName = "org.apache.hadoop.hdfs.DistributedFileSystem";
862      String builderName = dfsName + "$HdfsDataOutputStreamBuilder";
863      Class<?> builderClass = null;
864
865      try {
866        dfsClass = Class.forName(dfsName);
867      } catch (ClassNotFoundException e) {
868        LOG.debug("{} not available, will not use builder API for file creation.", dfsName);
869      }
870      try {
871        builderClass = Class.forName(builderName);
872      } catch (ClassNotFoundException e) {
873        LOG.debug("{} not available, will not use builder API for file creation.", builderName);
874      }
875
876      if (dfsClass != null && builderClass != null) {
877        try {
878          createMethod = dfsClass.getMethod("createFile", Path.class);
879          overwriteMethod = builderClass.getMethod("overwrite", boolean.class);
880          bufferSizeMethod = builderClass.getMethod("bufferSize", int.class);
881          blockSizeMethod = builderClass.getMethod("blockSize", long.class);
882          recursiveMethod = builderClass.getMethod("recursive");
883          replicateMethod = builderClass.getMethod("replicate");
884          replicationMethod = builderClass.getMethod("replication", short.class);
885          buildMethod = builderClass.getMethod("build");
886
887          allMethodsPresent = true;
888          LOG.debug("Using builder API via reflection for DFS file creation.");
889        } catch (NoSuchMethodException e) {
890          LOG.debug("Could not find method on builder; will use old DFS API for file creation {}",
891              e.getMessage());
892        }
893      }
894    }
895
896    /**
897     * Attempt to use builder API via reflection to create a file with the given parameters and
898     * replication enabled.
899     */
900    static FSDataOutputStream createHelper(FileSystem fs, Path path, boolean overwritable,
901        int bufferSize, short replication, long blockSize, boolean isRecursive) throws IOException {
902      if (allMethodsPresent && dfsClass.isInstance(fs)) {
903        try {
904          Object builder;
905
906          builder = createMethod.invoke(fs, path);
907          builder = overwriteMethod.invoke(builder, overwritable);
908          builder = bufferSizeMethod.invoke(builder, bufferSize);
909          builder = blockSizeMethod.invoke(builder, blockSize);
910          if (isRecursive) {
911            builder = recursiveMethod.invoke(builder);
912          }
913          builder = replicateMethod.invoke(builder);
914          builder = replicationMethod.invoke(builder, replication);
915          return (FSDataOutputStream) buildMethod.invoke(builder);
916        } catch (IllegalAccessException | InvocationTargetException e) {
917          // Should have caught this failure during initialization, so log full trace here
918          LOG.warn("Couldn't use reflection with builder API", e);
919        }
920      }
921
922      if (isRecursive) {
923        return fs.create(path, overwritable, bufferSize, replication, blockSize, null);
924      }
925      return fs.createNonRecursive(path, overwritable, bufferSize, replication, blockSize, null);
926    }
927
928    /**
929     * Attempt to use builder API via reflection to create a file with the given parameters and
930     * replication enabled.
931     */
932    static FSDataOutputStream createHelper(FileSystem fs, Path path, boolean overwritable)
933        throws IOException {
934      if (allMethodsPresent && dfsClass.isInstance(fs)) {
935        try {
936          Object builder;
937
938          builder = createMethod.invoke(fs, path);
939          builder = overwriteMethod.invoke(builder, overwritable);
940          builder = replicateMethod.invoke(builder);
941          return (FSDataOutputStream) buildMethod.invoke(builder);
942        } catch (IllegalAccessException | InvocationTargetException e) {
943          // Should have caught this failure during initialization, so log full trace here
944          LOG.warn("Couldn't use reflection with builder API", e);
945        }
946      }
947
948      return fs.create(path, overwritable);
949    }
950  }
951
952  /**
953   * Attempt to use builder API via reflection to create a file with the given parameters and
954   * replication enabled.
955   * <p>
956   * Will not attempt to enable replication when passed an HFileSystem.
957   */
958  public static FSDataOutputStream createForWal(FileSystem fs, Path path, boolean overwritable)
959      throws IOException {
960    return DfsBuilderUtility.createHelper(fs, path, overwritable);
961  }
962
963  /**
964   * Attempt to use builder API via reflection to create a file with the given parameters and
965   * replication enabled.
966   * <p>
967   * Will not attempt to enable replication when passed an HFileSystem.
968   */
969  public static FSDataOutputStream createForWal(FileSystem fs, Path path, boolean overwritable,
970      int bufferSize, short replication, long blockSize, boolean isRecursive) throws IOException {
971    return DfsBuilderUtility.createHelper(fs, path, overwritable, bufferSize, replication,
972        blockSize, isRecursive);
973  }
974
975  // Holder singleton idiom. JVM spec ensures this will be run at most once per Classloader, and
976  // not until we attempt to reference it.
977  private static class StreamCapabilities {
978    public static final boolean PRESENT;
979    public static final Class<?> CLASS;
980    public static final Method METHOD;
981    static {
982      boolean tmp = false;
983      Class<?> clazz = null;
984      Method method = null;
985      try {
986        clazz = Class.forName("org.apache.hadoop.fs.StreamCapabilities");
987        method = clazz.getMethod("hasCapability", String.class);
988        tmp = true;
989      } catch(ClassNotFoundException|NoSuchMethodException|SecurityException exception) {
990        LOG.warn("Your Hadoop installation does not include the StreamCapabilities class from " +
991                 "HDFS-11644, so we will skip checking if any FSDataOutputStreams actually " +
992                 "support hflush/hsync. If you are running on top of HDFS this probably just " +
993                 "means you have an older version and this can be ignored. If you are running on " +
994                 "top of an alternate FileSystem implementation you should manually verify that " +
995                 "hflush and hsync are implemented; otherwise you risk data loss and hard to " +
996                 "diagnose errors when our assumptions are violated.");
997        LOG.debug("The first request to check for StreamCapabilities came from this stacktrace.",
998            exception);
999      } finally {
1000        PRESENT = tmp;
1001        CLASS = clazz;
1002        METHOD = method;
1003      }
1004    }
1005  }
1006
1007  /**
1008   * If our FileSystem version includes the StreamCapabilities class, check if
1009   * the given stream has a particular capability.
1010   * @param stream capabilities are per-stream instance, so check this one specifically. must not be
1011   *        null
1012   * @param capability what to look for, per Hadoop Common's FileSystem docs
1013   * @return true if there are no StreamCapabilities. false if there are, but this stream doesn't
1014   *         implement it. return result of asking the stream otherwise.
1015   */
1016  public static boolean hasCapability(FSDataOutputStream stream, String capability) {
1017    // be consistent whether or not StreamCapabilities is present
1018    if (stream == null) {
1019      throw new NullPointerException("stream parameter must not be null.");
1020    }
1021    // If o.a.h.fs.StreamCapabilities doesn't exist, assume everyone does everything
1022    // otherwise old versions of Hadoop will break.
1023    boolean result = true;
1024    if (StreamCapabilities.PRESENT) {
1025      // if StreamCapabilities is present, but the stream doesn't implement it
1026      // or we run into a problem invoking the method,
1027      // we treat that as equivalent to not declaring anything
1028      result = false;
1029      if (StreamCapabilities.CLASS.isAssignableFrom(stream.getClass())) {
1030        try {
1031          result = ((Boolean)StreamCapabilities.METHOD.invoke(stream, capability)).booleanValue();
1032        } catch (IllegalAccessException|IllegalArgumentException|InvocationTargetException
1033            exception) {
1034          LOG.warn("Your Hadoop installation's StreamCapabilities implementation doesn't match " +
1035              "our understanding of how it's supposed to work. Please file a JIRA and include " +
1036              "the following stack trace. In the mean time we're interpreting this behavior " +
1037              "difference as a lack of capability support, which will probably cause a failure.",
1038              exception);
1039        }
1040      }
1041    }
1042    return result;
1043  }
1044
1045  /**
1046   * Helper exception for those cases where the place where we need to check a stream capability
1047   * is not where we have the needed context to explain the impact and mitigation for a lack.
1048   */
1049  public static class StreamLacksCapabilityException extends Exception {
1050    public StreamLacksCapabilityException(String message, Throwable cause) {
1051      super(message, cause);
1052    }
1053    public StreamLacksCapabilityException(String message) {
1054      super(message);
1055    }
1056  }
1057
1058}