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}