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}