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 org.apache.hadoop.hbase.client.RegionInfo; 022import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting; 023import org.apache.hbase.thirdparty.com.google.common.base.Throwables; 024import org.apache.hbase.thirdparty.com.google.common.collect.Iterators; 025import org.apache.hbase.thirdparty.com.google.common.primitives.Ints; 026 027import edu.umd.cs.findbugs.annotations.CheckForNull; 028 029import java.io.ByteArrayInputStream; 030import java.io.DataInputStream; 031import java.io.EOFException; 032import java.io.FileNotFoundException; 033import java.io.IOException; 034import java.io.InputStream; 035import java.io.InterruptedIOException; 036import java.lang.reflect.InvocationTargetException; 037import java.lang.reflect.Method; 038import java.net.InetSocketAddress; 039import java.util.ArrayList; 040import java.util.Arrays; 041import java.util.Collections; 042import java.util.HashMap; 043import java.util.Iterator; 044import java.util.LinkedList; 045import java.util.List; 046import java.util.Locale; 047import java.util.Map; 048import java.util.Vector; 049import java.util.concurrent.ArrayBlockingQueue; 050import java.util.concurrent.ConcurrentHashMap; 051import java.util.concurrent.ExecutionException; 052import java.util.concurrent.ExecutorService; 053import java.util.concurrent.Future; 054import java.util.concurrent.FutureTask; 055import java.util.concurrent.ThreadPoolExecutor; 056import java.util.concurrent.TimeUnit; 057import java.util.regex.Pattern; 058 059import org.apache.hadoop.conf.Configuration; 060import org.apache.hadoop.fs.BlockLocation; 061import org.apache.hadoop.fs.FSDataInputStream; 062import org.apache.hadoop.fs.FSDataOutputStream; 063import org.apache.hadoop.fs.FileStatus; 064import org.apache.hadoop.fs.FileSystem; 065import org.apache.hadoop.fs.Path; 066import org.apache.hadoop.fs.PathFilter; 067import org.apache.hadoop.fs.permission.FsAction; 068import org.apache.hadoop.fs.permission.FsPermission; 069import org.apache.hadoop.hbase.ClusterId; 070import org.apache.hadoop.hbase.HColumnDescriptor; 071import org.apache.hadoop.hbase.HConstants; 072import org.apache.hadoop.hbase.HDFSBlocksDistribution; 073import org.apache.hadoop.hbase.HRegionInfo; 074import org.apache.hadoop.hbase.TableName; 075import org.apache.yetus.audience.InterfaceAudience; 076import org.slf4j.Logger; 077import org.slf4j.LoggerFactory; 078import org.apache.hadoop.hbase.exceptions.DeserializationException; 079import org.apache.hadoop.hbase.fs.HFileSystem; 080import org.apache.hadoop.hbase.io.HFileLink; 081import org.apache.hadoop.hbase.master.HMaster; 082import org.apache.hadoop.hbase.regionserver.HRegion; 083import org.apache.hadoop.hbase.regionserver.StoreFileInfo; 084import org.apache.hadoop.hbase.security.AccessDeniedException; 085import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; 086import org.apache.hadoop.hbase.shaded.protobuf.generated.FSProtos; 087import org.apache.hadoop.hbase.util.HBaseFsck.ErrorReporter; 088import org.apache.hadoop.hdfs.DFSClient; 089import org.apache.hadoop.hdfs.DFSHedgedReadMetrics; 090import org.apache.hadoop.hdfs.DistributedFileSystem; 091import org.apache.hadoop.hdfs.protocol.HdfsConstants; 092import org.apache.hadoop.io.IOUtils; 093import org.apache.hadoop.ipc.RemoteException; 094import org.apache.hadoop.security.UserGroupInformation; 095import org.apache.hadoop.util.Progressable; 096import org.apache.hadoop.util.ReflectionUtils; 097import org.apache.hadoop.util.StringUtils; 098 099/** 100 * Utility methods for interacting with the underlying file system. 101 */ 102@InterfaceAudience.Private 103public abstract class FSUtils extends CommonFSUtils { 104 private static final Logger LOG = LoggerFactory.getLogger(FSUtils.class); 105 106 private static final String THREAD_POOLSIZE = "hbase.client.localityCheck.threadPoolSize"; 107 private static final int DEFAULT_THREAD_POOLSIZE = 2; 108 109 /** Set to true on Windows platforms */ 110 @VisibleForTesting // currently only used in testing. TODO refactor into a test class 111 public static final boolean WINDOWS = System.getProperty("os.name").startsWith("Windows"); 112 113 protected FSUtils() { 114 super(); 115 } 116 117 /** 118 * @return True is <code>fs</code> is instance of DistributedFileSystem 119 * @throws IOException 120 */ 121 public static boolean isDistributedFileSystem(final FileSystem fs) throws IOException { 122 FileSystem fileSystem = fs; 123 // If passed an instance of HFileSystem, it fails instanceof DistributedFileSystem. 124 // Check its backing fs for dfs-ness. 125 if (fs instanceof HFileSystem) { 126 fileSystem = ((HFileSystem)fs).getBackingFs(); 127 } 128 return fileSystem instanceof DistributedFileSystem; 129 } 130 131 /** 132 * Compare path component of the Path URI; e.g. if hdfs://a/b/c and /a/b/c, it will compare the 133 * '/a/b/c' part. If you passed in 'hdfs://a/b/c and b/c, it would return true. Does not consider 134 * schema; i.e. if schemas different but path or subpath matches, the two will equate. 135 * @param pathToSearch Path we will be trying to match. 136 * @param pathTail 137 * @return True if <code>pathTail</code> is tail on the path of <code>pathToSearch</code> 138 */ 139 public static boolean isMatchingTail(final Path pathToSearch, final Path pathTail) { 140 if (pathToSearch.depth() != pathTail.depth()) return false; 141 Path tailPath = pathTail; 142 String tailName; 143 Path toSearch = pathToSearch; 144 String toSearchName; 145 boolean result = false; 146 do { 147 tailName = tailPath.getName(); 148 if (tailName == null || tailName.length() <= 0) { 149 result = true; 150 break; 151 } 152 toSearchName = toSearch.getName(); 153 if (toSearchName == null || toSearchName.length() <= 0) break; 154 // Move up a parent on each path for next go around. Path doesn't let us go off the end. 155 tailPath = tailPath.getParent(); 156 toSearch = toSearch.getParent(); 157 } while(tailName.equals(toSearchName)); 158 return result; 159 } 160 161 public static FSUtils getInstance(FileSystem fs, Configuration conf) { 162 String scheme = fs.getUri().getScheme(); 163 if (scheme == null) { 164 LOG.warn("Could not find scheme for uri " + 165 fs.getUri() + ", default to hdfs"); 166 scheme = "hdfs"; 167 } 168 Class<?> fsUtilsClass = conf.getClass("hbase.fsutil." + 169 scheme + ".impl", FSHDFSUtils.class); // Default to HDFS impl 170 FSUtils fsUtils = (FSUtils)ReflectionUtils.newInstance(fsUtilsClass, conf); 171 return fsUtils; 172 } 173 174 /** 175 * Delete the region directory if exists. 176 * @param conf 177 * @param hri 178 * @return True if deleted the region directory. 179 * @throws IOException 180 */ 181 public static boolean deleteRegionDir(final Configuration conf, final HRegionInfo hri) 182 throws IOException { 183 Path rootDir = getRootDir(conf); 184 FileSystem fs = rootDir.getFileSystem(conf); 185 return deleteDirectory(fs, 186 new Path(getTableDir(rootDir, hri.getTable()), hri.getEncodedName())); 187 } 188 189 /** 190 * Create the specified file on the filesystem. By default, this will: 191 * <ol> 192 * <li>overwrite the file if it exists</li> 193 * <li>apply the umask in the configuration (if it is enabled)</li> 194 * <li>use the fs configured buffer size (or 4096 if not set)</li> 195 * <li>use the configured column family replication or default replication if 196 * {@link HColumnDescriptor#DEFAULT_DFS_REPLICATION}</li> 197 * <li>use the default block size</li> 198 * <li>not track progress</li> 199 * </ol> 200 * @param conf configurations 201 * @param fs {@link FileSystem} on which to write the file 202 * @param path {@link Path} to the file to write 203 * @param perm permissions 204 * @param favoredNodes 205 * @return output stream to the created file 206 * @throws IOException if the file cannot be created 207 */ 208 public static FSDataOutputStream create(Configuration conf, FileSystem fs, Path path, 209 FsPermission perm, InetSocketAddress[] favoredNodes) throws IOException { 210 if (fs instanceof HFileSystem) { 211 FileSystem backingFs = ((HFileSystem)fs).getBackingFs(); 212 if (backingFs instanceof DistributedFileSystem) { 213 // Try to use the favoredNodes version via reflection to allow backwards- 214 // compatibility. 215 short replication = Short.parseShort(conf.get(HColumnDescriptor.DFS_REPLICATION, 216 String.valueOf(HColumnDescriptor.DEFAULT_DFS_REPLICATION))); 217 try { 218 return (FSDataOutputStream) (DistributedFileSystem.class.getDeclaredMethod("create", 219 Path.class, FsPermission.class, boolean.class, int.class, short.class, long.class, 220 Progressable.class, InetSocketAddress[].class).invoke(backingFs, path, perm, true, 221 getDefaultBufferSize(backingFs), 222 replication > 0 ? replication : getDefaultReplication(backingFs, path), 223 getDefaultBlockSize(backingFs, path), null, favoredNodes)); 224 } catch (InvocationTargetException ite) { 225 // Function was properly called, but threw it's own exception. 226 throw new IOException(ite.getCause()); 227 } catch (NoSuchMethodException e) { 228 LOG.debug("DFS Client does not support most favored nodes create; using default create"); 229 if (LOG.isTraceEnabled()) LOG.trace("Ignoring; use default create", e); 230 } catch (IllegalArgumentException e) { 231 LOG.debug("Ignoring (most likely Reflection related exception) " + e); 232 } catch (SecurityException e) { 233 LOG.debug("Ignoring (most likely Reflection related exception) " + e); 234 } catch (IllegalAccessException e) { 235 LOG.debug("Ignoring (most likely Reflection related exception) " + e); 236 } 237 } 238 } 239 return create(fs, path, perm, true); 240 } 241 242 /** 243 * Checks to see if the specified file system is available 244 * 245 * @param fs filesystem 246 * @throws IOException e 247 */ 248 public static void checkFileSystemAvailable(final FileSystem fs) 249 throws IOException { 250 if (!(fs instanceof DistributedFileSystem)) { 251 return; 252 } 253 IOException exception = null; 254 DistributedFileSystem dfs = (DistributedFileSystem) fs; 255 try { 256 if (dfs.exists(new Path("/"))) { 257 return; 258 } 259 } catch (IOException e) { 260 exception = e instanceof RemoteException ? 261 ((RemoteException)e).unwrapRemoteException() : e; 262 } 263 try { 264 fs.close(); 265 } catch (Exception e) { 266 LOG.error("file system close failed: ", e); 267 } 268 IOException io = new IOException("File system is not available"); 269 io.initCause(exception); 270 throw io; 271 } 272 273 /** 274 * We use reflection because {@link DistributedFileSystem#setSafeMode( 275 * HdfsConstants.SafeModeAction action, boolean isChecked)} is not in hadoop 1.1 276 * 277 * @param dfs 278 * @return whether we're in safe mode 279 * @throws IOException 280 */ 281 private static boolean isInSafeMode(DistributedFileSystem dfs) throws IOException { 282 boolean inSafeMode = false; 283 try { 284 Method m = DistributedFileSystem.class.getMethod("setSafeMode", new Class<?> []{ 285 org.apache.hadoop.hdfs.protocol.HdfsConstants.SafeModeAction.class, boolean.class}); 286 inSafeMode = (Boolean) m.invoke(dfs, 287 org.apache.hadoop.hdfs.protocol.HdfsConstants.SafeModeAction.SAFEMODE_GET, true); 288 } catch (Exception e) { 289 if (e instanceof IOException) throw (IOException) e; 290 291 // Check whether dfs is on safemode. 292 inSafeMode = dfs.setSafeMode( 293 org.apache.hadoop.hdfs.protocol.HdfsConstants.SafeModeAction.SAFEMODE_GET); 294 } 295 return inSafeMode; 296 } 297 298 /** 299 * Check whether dfs is in safemode. 300 * @param conf 301 * @throws IOException 302 */ 303 public static void checkDfsSafeMode(final Configuration conf) 304 throws IOException { 305 boolean isInSafeMode = false; 306 FileSystem fs = FileSystem.get(conf); 307 if (fs instanceof DistributedFileSystem) { 308 DistributedFileSystem dfs = (DistributedFileSystem)fs; 309 isInSafeMode = isInSafeMode(dfs); 310 } 311 if (isInSafeMode) { 312 throw new IOException("File system is in safemode, it can't be written now"); 313 } 314 } 315 316 /** 317 * Verifies current version of file system 318 * 319 * @param fs filesystem object 320 * @param rootdir root hbase directory 321 * @return null if no version file exists, version string otherwise. 322 * @throws IOException e 323 * @throws org.apache.hadoop.hbase.exceptions.DeserializationException 324 */ 325 public static String getVersion(FileSystem fs, Path rootdir) 326 throws IOException, DeserializationException { 327 Path versionFile = new Path(rootdir, HConstants.VERSION_FILE_NAME); 328 FileStatus[] status = null; 329 try { 330 // hadoop 2.0 throws FNFE if directory does not exist. 331 // hadoop 1.0 returns null if directory does not exist. 332 status = fs.listStatus(versionFile); 333 } catch (FileNotFoundException fnfe) { 334 return null; 335 } 336 if (status == null || status.length == 0) return null; 337 String version = null; 338 byte [] content = new byte [(int)status[0].getLen()]; 339 FSDataInputStream s = fs.open(versionFile); 340 try { 341 IOUtils.readFully(s, content, 0, content.length); 342 if (ProtobufUtil.isPBMagicPrefix(content)) { 343 version = parseVersionFrom(content); 344 } else { 345 // Presume it pre-pb format. 346 InputStream is = new ByteArrayInputStream(content); 347 DataInputStream dis = new DataInputStream(is); 348 try { 349 version = dis.readUTF(); 350 } finally { 351 dis.close(); 352 } 353 } 354 } catch (EOFException eof) { 355 LOG.warn("Version file was empty, odd, will try to set it."); 356 } finally { 357 s.close(); 358 } 359 return version; 360 } 361 362 /** 363 * Parse the content of the ${HBASE_ROOTDIR}/hbase.version file. 364 * @param bytes The byte content of the hbase.version file. 365 * @return The version found in the file as a String. 366 * @throws DeserializationException 367 */ 368 static String parseVersionFrom(final byte [] bytes) 369 throws DeserializationException { 370 ProtobufUtil.expectPBMagicPrefix(bytes); 371 int pblen = ProtobufUtil.lengthOfPBMagic(); 372 FSProtos.HBaseVersionFileContent.Builder builder = 373 FSProtos.HBaseVersionFileContent.newBuilder(); 374 try { 375 ProtobufUtil.mergeFrom(builder, bytes, pblen, bytes.length - pblen); 376 return builder.getVersion(); 377 } catch (IOException e) { 378 // Convert 379 throw new DeserializationException(e); 380 } 381 } 382 383 /** 384 * Create the content to write into the ${HBASE_ROOTDIR}/hbase.version file. 385 * @param version Version to persist 386 * @return Serialized protobuf with <code>version</code> content and a bit of pb magic for a prefix. 387 */ 388 static byte [] toVersionByteArray(final String version) { 389 FSProtos.HBaseVersionFileContent.Builder builder = 390 FSProtos.HBaseVersionFileContent.newBuilder(); 391 return ProtobufUtil.prependPBMagic(builder.setVersion(version).build().toByteArray()); 392 } 393 394 /** 395 * Verifies current version of file system 396 * 397 * @param fs file system 398 * @param rootdir root directory of HBase installation 399 * @param message if true, issues a message on System.out 400 * 401 * @throws IOException e 402 * @throws DeserializationException 403 */ 404 public static void checkVersion(FileSystem fs, Path rootdir, boolean message) 405 throws IOException, DeserializationException { 406 checkVersion(fs, rootdir, message, 0, HConstants.DEFAULT_VERSION_FILE_WRITE_ATTEMPTS); 407 } 408 409 /** 410 * Verifies current version of file system 411 * 412 * @param fs file system 413 * @param rootdir root directory of HBase installation 414 * @param message if true, issues a message on System.out 415 * @param wait wait interval 416 * @param retries number of times to retry 417 * 418 * @throws IOException e 419 * @throws DeserializationException 420 */ 421 public static void checkVersion(FileSystem fs, Path rootdir, 422 boolean message, int wait, int retries) 423 throws IOException, DeserializationException { 424 String version = getVersion(fs, rootdir); 425 if (version == null) { 426 if (!metaRegionExists(fs, rootdir)) { 427 // rootDir is empty (no version file and no root region) 428 // just create new version file (HBASE-1195) 429 setVersion(fs, rootdir, wait, retries); 430 return; 431 } 432 } else if (version.compareTo(HConstants.FILE_SYSTEM_VERSION) == 0) return; 433 434 // version is deprecated require migration 435 // Output on stdout so user sees it in terminal. 436 String msg = "HBase file layout needs to be upgraded." 437 + " You have version " + version 438 + " and I want version " + HConstants.FILE_SYSTEM_VERSION 439 + ". Consult http://hbase.apache.org/book.html for further information about upgrading HBase." 440 + " Is your hbase.rootdir valid? If so, you may need to run " 441 + "'hbase hbck -fixVersionFile'."; 442 if (message) { 443 System.out.println("WARNING! " + msg); 444 } 445 throw new FileSystemVersionException(msg); 446 } 447 448 /** 449 * Sets version of file system 450 * 451 * @param fs filesystem object 452 * @param rootdir hbase root 453 * @throws IOException e 454 */ 455 public static void setVersion(FileSystem fs, Path rootdir) 456 throws IOException { 457 setVersion(fs, rootdir, HConstants.FILE_SYSTEM_VERSION, 0, 458 HConstants.DEFAULT_VERSION_FILE_WRITE_ATTEMPTS); 459 } 460 461 /** 462 * Sets version of file system 463 * 464 * @param fs filesystem object 465 * @param rootdir hbase root 466 * @param wait time to wait for retry 467 * @param retries number of times to retry before failing 468 * @throws IOException e 469 */ 470 public static void setVersion(FileSystem fs, Path rootdir, int wait, int retries) 471 throws IOException { 472 setVersion(fs, rootdir, HConstants.FILE_SYSTEM_VERSION, wait, retries); 473 } 474 475 476 /** 477 * Sets version of file system 478 * 479 * @param fs filesystem object 480 * @param rootdir hbase root directory 481 * @param version version to set 482 * @param wait time to wait for retry 483 * @param retries number of times to retry before throwing an IOException 484 * @throws IOException e 485 */ 486 public static void setVersion(FileSystem fs, Path rootdir, String version, 487 int wait, int retries) throws IOException { 488 Path versionFile = new Path(rootdir, HConstants.VERSION_FILE_NAME); 489 Path tempVersionFile = new Path(rootdir, HConstants.HBASE_TEMP_DIRECTORY + Path.SEPARATOR + 490 HConstants.VERSION_FILE_NAME); 491 while (true) { 492 try { 493 // Write the version to a temporary file 494 FSDataOutputStream s = fs.create(tempVersionFile); 495 try { 496 s.write(toVersionByteArray(version)); 497 s.close(); 498 s = null; 499 // Move the temp version file to its normal location. Returns false 500 // if the rename failed. Throw an IOE in that case. 501 if (!fs.rename(tempVersionFile, versionFile)) { 502 throw new IOException("Unable to move temp version file to " + versionFile); 503 } 504 } finally { 505 // Cleaning up the temporary if the rename failed would be trying 506 // too hard. We'll unconditionally create it again the next time 507 // through anyway, files are overwritten by default by create(). 508 509 // Attempt to close the stream on the way out if it is still open. 510 try { 511 if (s != null) s.close(); 512 } catch (IOException ignore) { } 513 } 514 LOG.info("Created version file at " + rootdir.toString() + " with version=" + version); 515 return; 516 } catch (IOException e) { 517 if (retries > 0) { 518 LOG.debug("Unable to create version file at " + rootdir.toString() + ", retrying", e); 519 fs.delete(versionFile, false); 520 try { 521 if (wait > 0) { 522 Thread.sleep(wait); 523 } 524 } catch (InterruptedException ie) { 525 throw (InterruptedIOException)new InterruptedIOException().initCause(ie); 526 } 527 retries--; 528 } else { 529 throw e; 530 } 531 } 532 } 533 } 534 535 /** 536 * Checks that a cluster ID file exists in the HBase root directory 537 * @param fs the root directory FileSystem 538 * @param rootdir the HBase root directory in HDFS 539 * @param wait how long to wait between retries 540 * @return <code>true</code> if the file exists, otherwise <code>false</code> 541 * @throws IOException if checking the FileSystem fails 542 */ 543 public static boolean checkClusterIdExists(FileSystem fs, Path rootdir, 544 int wait) throws IOException { 545 while (true) { 546 try { 547 Path filePath = new Path(rootdir, HConstants.CLUSTER_ID_FILE_NAME); 548 return fs.exists(filePath); 549 } catch (IOException ioe) { 550 if (wait > 0) { 551 LOG.warn("Unable to check cluster ID file in " + rootdir.toString() + 552 ", retrying in "+wait+"msec: "+StringUtils.stringifyException(ioe)); 553 try { 554 Thread.sleep(wait); 555 } catch (InterruptedException e) { 556 throw (InterruptedIOException)new InterruptedIOException().initCause(e); 557 } 558 } else { 559 throw ioe; 560 } 561 } 562 } 563 } 564 565 /** 566 * Returns the value of the unique cluster ID stored for this HBase instance. 567 * @param fs the root directory FileSystem 568 * @param rootdir the path to the HBase root directory 569 * @return the unique cluster identifier 570 * @throws IOException if reading the cluster ID file fails 571 */ 572 public static ClusterId getClusterId(FileSystem fs, Path rootdir) 573 throws IOException { 574 Path idPath = new Path(rootdir, HConstants.CLUSTER_ID_FILE_NAME); 575 ClusterId clusterId = null; 576 FileStatus status = fs.exists(idPath)? fs.getFileStatus(idPath): null; 577 if (status != null) { 578 int len = Ints.checkedCast(status.getLen()); 579 byte [] content = new byte[len]; 580 FSDataInputStream in = fs.open(idPath); 581 try { 582 in.readFully(content); 583 } catch (EOFException eof) { 584 LOG.warn("Cluster ID file " + idPath.toString() + " was empty"); 585 } finally{ 586 in.close(); 587 } 588 try { 589 clusterId = ClusterId.parseFrom(content); 590 } catch (DeserializationException e) { 591 throw new IOException("content=" + Bytes.toString(content), e); 592 } 593 // If not pb'd, make it so. 594 if (!ProtobufUtil.isPBMagicPrefix(content)) { 595 String cid = null; 596 in = fs.open(idPath); 597 try { 598 cid = in.readUTF(); 599 clusterId = new ClusterId(cid); 600 } catch (EOFException eof) { 601 LOG.warn("Cluster ID file " + idPath.toString() + " was empty"); 602 } finally { 603 in.close(); 604 } 605 rewriteAsPb(fs, rootdir, idPath, clusterId); 606 } 607 return clusterId; 608 } else { 609 LOG.warn("Cluster ID file does not exist at " + idPath.toString()); 610 } 611 return clusterId; 612 } 613 614 /** 615 * @param cid 616 * @throws IOException 617 */ 618 private static void rewriteAsPb(final FileSystem fs, final Path rootdir, final Path p, 619 final ClusterId cid) 620 throws IOException { 621 // Rewrite the file as pb. Move aside the old one first, write new 622 // then delete the moved-aside file. 623 Path movedAsideName = new Path(p + "." + System.currentTimeMillis()); 624 if (!fs.rename(p, movedAsideName)) throw new IOException("Failed rename of " + p); 625 setClusterId(fs, rootdir, cid, 100); 626 if (!fs.delete(movedAsideName, false)) { 627 throw new IOException("Failed delete of " + movedAsideName); 628 } 629 LOG.debug("Rewrote the hbase.id file as pb"); 630 } 631 632 /** 633 * Writes a new unique identifier for this cluster to the "hbase.id" file 634 * in the HBase root directory 635 * @param fs the root directory FileSystem 636 * @param rootdir the path to the HBase root directory 637 * @param clusterId the unique identifier to store 638 * @param wait how long (in milliseconds) to wait between retries 639 * @throws IOException if writing to the FileSystem fails and no wait value 640 */ 641 public static void setClusterId(FileSystem fs, Path rootdir, ClusterId clusterId, 642 int wait) throws IOException { 643 while (true) { 644 try { 645 Path idFile = new Path(rootdir, HConstants.CLUSTER_ID_FILE_NAME); 646 Path tempIdFile = new Path(rootdir, HConstants.HBASE_TEMP_DIRECTORY + 647 Path.SEPARATOR + HConstants.CLUSTER_ID_FILE_NAME); 648 // Write the id file to a temporary location 649 FSDataOutputStream s = fs.create(tempIdFile); 650 try { 651 s.write(clusterId.toByteArray()); 652 s.close(); 653 s = null; 654 // Move the temporary file to its normal location. Throw an IOE if 655 // the rename failed 656 if (!fs.rename(tempIdFile, idFile)) { 657 throw new IOException("Unable to move temp version file to " + idFile); 658 } 659 } finally { 660 // Attempt to close the stream if still open on the way out 661 try { 662 if (s != null) s.close(); 663 } catch (IOException ignore) { } 664 } 665 if (LOG.isDebugEnabled()) { 666 LOG.debug("Created cluster ID file at " + idFile.toString() + " with ID: " + clusterId); 667 } 668 return; 669 } catch (IOException ioe) { 670 if (wait > 0) { 671 LOG.warn("Unable to create cluster ID file in " + rootdir.toString() + 672 ", retrying in " + wait + "msec: " + StringUtils.stringifyException(ioe)); 673 try { 674 Thread.sleep(wait); 675 } catch (InterruptedException e) { 676 throw (InterruptedIOException)new InterruptedIOException().initCause(e); 677 } 678 } else { 679 throw ioe; 680 } 681 } 682 } 683 } 684 685 /** 686 * If DFS, check safe mode and if so, wait until we clear it. 687 * @param conf configuration 688 * @param wait Sleep between retries 689 * @throws IOException e 690 */ 691 public static void waitOnSafeMode(final Configuration conf, 692 final long wait) 693 throws IOException { 694 FileSystem fs = FileSystem.get(conf); 695 if (!(fs instanceof DistributedFileSystem)) return; 696 DistributedFileSystem dfs = (DistributedFileSystem)fs; 697 // Make sure dfs is not in safe mode 698 while (isInSafeMode(dfs)) { 699 LOG.info("Waiting for dfs to exit safe mode..."); 700 try { 701 Thread.sleep(wait); 702 } catch (InterruptedException e) { 703 throw (InterruptedIOException)new InterruptedIOException().initCause(e); 704 } 705 } 706 } 707 708 /** 709 * Checks if meta region exists 710 * 711 * @param fs file system 712 * @param rootdir root directory of HBase installation 713 * @return true if exists 714 * @throws IOException e 715 */ 716 @SuppressWarnings("deprecation") 717 public static boolean metaRegionExists(FileSystem fs, Path rootdir) 718 throws IOException { 719 Path metaRegionDir = 720 HRegion.getRegionDir(rootdir, HRegionInfo.FIRST_META_REGIONINFO); 721 return fs.exists(metaRegionDir); 722 } 723 724 /** 725 * Compute HDFS blocks distribution of a given file, or a portion of the file 726 * @param fs file system 727 * @param status file status of the file 728 * @param start start position of the portion 729 * @param length length of the portion 730 * @return The HDFS blocks distribution 731 */ 732 static public HDFSBlocksDistribution computeHDFSBlocksDistribution( 733 final FileSystem fs, FileStatus status, long start, long length) 734 throws IOException { 735 HDFSBlocksDistribution blocksDistribution = new HDFSBlocksDistribution(); 736 BlockLocation [] blockLocations = 737 fs.getFileBlockLocations(status, start, length); 738 for(BlockLocation bl : blockLocations) { 739 String [] hosts = bl.getHosts(); 740 long len = bl.getLength(); 741 blocksDistribution.addHostsAndBlockWeight(hosts, len); 742 } 743 744 return blocksDistribution; 745 } 746 747 /** 748 * Update blocksDistribution with blockLocations 749 * @param blocksDistribution the hdfs blocks distribution 750 * @param blockLocations an array containing block location 751 */ 752 static public void addToHDFSBlocksDistribution( 753 HDFSBlocksDistribution blocksDistribution, BlockLocation[] blockLocations) 754 throws IOException { 755 for (BlockLocation bl : blockLocations) { 756 String[] hosts = bl.getHosts(); 757 long len = bl.getLength(); 758 blocksDistribution.addHostsAndBlockWeight(hosts, len); 759 } 760 } 761 762 // TODO move this method OUT of FSUtils. No dependencies to HMaster 763 /** 764 * Returns the total overall fragmentation percentage. Includes hbase:meta and 765 * -ROOT- as well. 766 * 767 * @param master The master defining the HBase root and file system. 768 * @return A map for each table and its percentage. 769 * @throws IOException When scanning the directory fails. 770 */ 771 public static int getTotalTableFragmentation(final HMaster master) 772 throws IOException { 773 Map<String, Integer> map = getTableFragmentation(master); 774 return map != null && map.size() > 0 ? map.get("-TOTAL-") : -1; 775 } 776 777 /** 778 * Runs through the HBase rootdir and checks how many stores for each table 779 * have more than one file in them. Checks -ROOT- and hbase:meta too. The total 780 * percentage across all tables is stored under the special key "-TOTAL-". 781 * 782 * @param master The master defining the HBase root and file system. 783 * @return A map for each table and its percentage. 784 * 785 * @throws IOException When scanning the directory fails. 786 */ 787 public static Map<String, Integer> getTableFragmentation( 788 final HMaster master) 789 throws IOException { 790 Path path = getRootDir(master.getConfiguration()); 791 // since HMaster.getFileSystem() is package private 792 FileSystem fs = path.getFileSystem(master.getConfiguration()); 793 return getTableFragmentation(fs, path); 794 } 795 796 /** 797 * Runs through the HBase rootdir and checks how many stores for each table 798 * have more than one file in them. Checks -ROOT- and hbase:meta too. The total 799 * percentage across all tables is stored under the special key "-TOTAL-". 800 * 801 * @param fs The file system to use. 802 * @param hbaseRootDir The root directory to scan. 803 * @return A map for each table and its percentage. 804 * @throws IOException When scanning the directory fails. 805 */ 806 public static Map<String, Integer> getTableFragmentation( 807 final FileSystem fs, final Path hbaseRootDir) 808 throws IOException { 809 Map<String, Integer> frags = new HashMap<>(); 810 int cfCountTotal = 0; 811 int cfFragTotal = 0; 812 PathFilter regionFilter = new RegionDirFilter(fs); 813 PathFilter familyFilter = new FamilyDirFilter(fs); 814 List<Path> tableDirs = getTableDirs(fs, hbaseRootDir); 815 for (Path d : tableDirs) { 816 int cfCount = 0; 817 int cfFrag = 0; 818 FileStatus[] regionDirs = fs.listStatus(d, regionFilter); 819 for (FileStatus regionDir : regionDirs) { 820 Path dd = regionDir.getPath(); 821 // else its a region name, now look in region for families 822 FileStatus[] familyDirs = fs.listStatus(dd, familyFilter); 823 for (FileStatus familyDir : familyDirs) { 824 cfCount++; 825 cfCountTotal++; 826 Path family = familyDir.getPath(); 827 // now in family make sure only one file 828 FileStatus[] familyStatus = fs.listStatus(family); 829 if (familyStatus.length > 1) { 830 cfFrag++; 831 cfFragTotal++; 832 } 833 } 834 } 835 // compute percentage per table and store in result list 836 frags.put(FSUtils.getTableName(d).getNameAsString(), 837 cfCount == 0? 0: Math.round((float) cfFrag / cfCount * 100)); 838 } 839 // set overall percentage for all tables 840 frags.put("-TOTAL-", 841 cfCountTotal == 0? 0: Math.round((float) cfFragTotal / cfCountTotal * 100)); 842 return frags; 843 } 844 845 /** 846 * A {@link PathFilter} that returns only regular files. 847 */ 848 static class FileFilter extends AbstractFileStatusFilter { 849 private final FileSystem fs; 850 851 public FileFilter(final FileSystem fs) { 852 this.fs = fs; 853 } 854 855 @Override 856 protected boolean accept(Path p, @CheckForNull Boolean isDir) { 857 try { 858 return isFile(fs, isDir, p); 859 } catch (IOException e) { 860 LOG.warn("unable to verify if path=" + p + " is a regular file", e); 861 return false; 862 } 863 } 864 } 865 866 /** 867 * Directory filter that doesn't include any of the directories in the specified blacklist 868 */ 869 public static class BlackListDirFilter extends AbstractFileStatusFilter { 870 private final FileSystem fs; 871 private List<String> blacklist; 872 873 /** 874 * Create a filter on the givem filesystem with the specified blacklist 875 * @param fs filesystem to filter 876 * @param directoryNameBlackList list of the names of the directories to filter. If 877 * <tt>null</tt>, all directories are returned 878 */ 879 @SuppressWarnings("unchecked") 880 public BlackListDirFilter(final FileSystem fs, final List<String> directoryNameBlackList) { 881 this.fs = fs; 882 blacklist = 883 (List<String>) (directoryNameBlackList == null ? Collections.emptyList() 884 : directoryNameBlackList); 885 } 886 887 @Override 888 protected boolean accept(Path p, @CheckForNull Boolean isDir) { 889 if (!isValidName(p.getName())) { 890 return false; 891 } 892 893 try { 894 return isDirectory(fs, isDir, p); 895 } catch (IOException e) { 896 LOG.warn("An error occurred while verifying if [" + p.toString() 897 + "] is a valid directory. Returning 'not valid' and continuing.", e); 898 return false; 899 } 900 } 901 902 protected boolean isValidName(final String name) { 903 return !blacklist.contains(name); 904 } 905 } 906 907 /** 908 * A {@link PathFilter} that only allows directories. 909 */ 910 public static class DirFilter extends BlackListDirFilter { 911 912 public DirFilter(FileSystem fs) { 913 super(fs, null); 914 } 915 } 916 917 /** 918 * A {@link PathFilter} that returns usertable directories. To get all directories use the 919 * {@link BlackListDirFilter} with a <tt>null</tt> blacklist 920 */ 921 public static class UserTableDirFilter extends BlackListDirFilter { 922 public UserTableDirFilter(FileSystem fs) { 923 super(fs, HConstants.HBASE_NON_TABLE_DIRS); 924 } 925 926 @Override 927 protected boolean isValidName(final String name) { 928 if (!super.isValidName(name)) 929 return false; 930 931 try { 932 TableName.isLegalTableQualifierName(Bytes.toBytes(name)); 933 } catch (IllegalArgumentException e) { 934 LOG.info("INVALID NAME " + name); 935 return false; 936 } 937 return true; 938 } 939 } 940 941 /** 942 * Recover file lease. Used when a file might be suspect 943 * to be had been left open by another process. 944 * @param fs FileSystem handle 945 * @param p Path of file to recover lease 946 * @param conf Configuration handle 947 * @throws IOException 948 */ 949 public abstract void recoverFileLease(final FileSystem fs, final Path p, 950 Configuration conf, CancelableProgressable reporter) throws IOException; 951 952 public static List<Path> getTableDirs(final FileSystem fs, final Path rootdir) 953 throws IOException { 954 List<Path> tableDirs = new LinkedList<>(); 955 956 for(FileStatus status : 957 fs.globStatus(new Path(rootdir, 958 new Path(HConstants.BASE_NAMESPACE_DIR, "*")))) { 959 tableDirs.addAll(FSUtils.getLocalTableDirs(fs, status.getPath())); 960 } 961 return tableDirs; 962 } 963 964 /** 965 * @param fs 966 * @param rootdir 967 * @return All the table directories under <code>rootdir</code>. Ignore non table hbase folders such as 968 * .logs, .oldlogs, .corrupt folders. 969 * @throws IOException 970 */ 971 public static List<Path> getLocalTableDirs(final FileSystem fs, final Path rootdir) 972 throws IOException { 973 // presumes any directory under hbase.rootdir is a table 974 FileStatus[] dirs = fs.listStatus(rootdir, new UserTableDirFilter(fs)); 975 List<Path> tabledirs = new ArrayList<>(dirs.length); 976 for (FileStatus dir: dirs) { 977 tabledirs.add(dir.getPath()); 978 } 979 return tabledirs; 980 } 981 982 /** 983 * Filter for all dirs that don't start with '.' 984 */ 985 public static class RegionDirFilter extends AbstractFileStatusFilter { 986 // This pattern will accept 0.90+ style hex region dirs and older numeric region dir names. 987 final public static Pattern regionDirPattern = Pattern.compile("^[0-9a-f]*$"); 988 final FileSystem fs; 989 990 public RegionDirFilter(FileSystem fs) { 991 this.fs = fs; 992 } 993 994 @Override 995 protected boolean accept(Path p, @CheckForNull Boolean isDir) { 996 if (!regionDirPattern.matcher(p.getName()).matches()) { 997 return false; 998 } 999 1000 try { 1001 return isDirectory(fs, isDir, p); 1002 } catch (IOException ioe) { 1003 // Maybe the file was moved or the fs was disconnected. 1004 LOG.warn("Skipping file " + p +" due to IOException", ioe); 1005 return false; 1006 } 1007 } 1008 } 1009 1010 /** 1011 * Given a particular table dir, return all the regiondirs inside it, excluding files such as 1012 * .tableinfo 1013 * @param fs A file system for the Path 1014 * @param tableDir Path to a specific table directory <hbase.rootdir>/<tabledir> 1015 * @return List of paths to valid region directories in table dir. 1016 * @throws IOException 1017 */ 1018 public static List<Path> getRegionDirs(final FileSystem fs, final Path tableDir) throws IOException { 1019 // assumes we are in a table dir. 1020 List<FileStatus> rds = listStatusWithStatusFilter(fs, tableDir, new RegionDirFilter(fs)); 1021 if (rds == null) { 1022 return new ArrayList<>(); 1023 } 1024 List<Path> regionDirs = new ArrayList<>(rds.size()); 1025 for (FileStatus rdfs: rds) { 1026 Path rdPath = rdfs.getPath(); 1027 regionDirs.add(rdPath); 1028 } 1029 return regionDirs; 1030 } 1031 1032 public static Path getRegionDir(Path tableDir, RegionInfo region) { 1033 return new Path(tableDir, ServerRegionReplicaUtil.getRegionInfoForFs(region).getEncodedName()); 1034 } 1035 1036 /** 1037 * Filter for all dirs that are legal column family names. This is generally used for colfam 1038 * dirs <hbase.rootdir>/<tabledir>/<regiondir>/<colfamdir>. 1039 */ 1040 public static class FamilyDirFilter extends AbstractFileStatusFilter { 1041 final FileSystem fs; 1042 1043 public FamilyDirFilter(FileSystem fs) { 1044 this.fs = fs; 1045 } 1046 1047 @Override 1048 protected boolean accept(Path p, @CheckForNull Boolean isDir) { 1049 try { 1050 // throws IAE if invalid 1051 HColumnDescriptor.isLegalFamilyName(Bytes.toBytes(p.getName())); 1052 } catch (IllegalArgumentException iae) { 1053 // path name is an invalid family name and thus is excluded. 1054 return false; 1055 } 1056 1057 try { 1058 return isDirectory(fs, isDir, p); 1059 } catch (IOException ioe) { 1060 // Maybe the file was moved or the fs was disconnected. 1061 LOG.warn("Skipping file " + p +" due to IOException", ioe); 1062 return false; 1063 } 1064 } 1065 } 1066 1067 /** 1068 * Given a particular region dir, return all the familydirs inside it 1069 * 1070 * @param fs A file system for the Path 1071 * @param regionDir Path to a specific region directory 1072 * @return List of paths to valid family directories in region dir. 1073 * @throws IOException 1074 */ 1075 public static List<Path> getFamilyDirs(final FileSystem fs, final Path regionDir) throws IOException { 1076 // assumes we are in a region dir. 1077 FileStatus[] fds = fs.listStatus(regionDir, new FamilyDirFilter(fs)); 1078 List<Path> familyDirs = new ArrayList<>(fds.length); 1079 for (FileStatus fdfs: fds) { 1080 Path fdPath = fdfs.getPath(); 1081 familyDirs.add(fdPath); 1082 } 1083 return familyDirs; 1084 } 1085 1086 public static List<Path> getReferenceFilePaths(final FileSystem fs, final Path familyDir) throws IOException { 1087 List<FileStatus> fds = listStatusWithStatusFilter(fs, familyDir, new ReferenceFileFilter(fs)); 1088 if (fds == null) { 1089 return new ArrayList<>(); 1090 } 1091 List<Path> referenceFiles = new ArrayList<>(fds.size()); 1092 for (FileStatus fdfs: fds) { 1093 Path fdPath = fdfs.getPath(); 1094 referenceFiles.add(fdPath); 1095 } 1096 return referenceFiles; 1097 } 1098 1099 /** 1100 * Filter for HFiles that excludes reference files. 1101 */ 1102 public static class HFileFilter extends AbstractFileStatusFilter { 1103 final FileSystem fs; 1104 1105 public HFileFilter(FileSystem fs) { 1106 this.fs = fs; 1107 } 1108 1109 @Override 1110 protected boolean accept(Path p, @CheckForNull Boolean isDir) { 1111 if (!StoreFileInfo.isHFile(p)) { 1112 return false; 1113 } 1114 1115 try { 1116 return isFile(fs, isDir, p); 1117 } catch (IOException ioe) { 1118 // Maybe the file was moved or the fs was disconnected. 1119 LOG.warn("Skipping file " + p +" due to IOException", ioe); 1120 return false; 1121 } 1122 } 1123 } 1124 1125 /** 1126 * Filter for HFileLinks (StoreFiles and HFiles not included). 1127 * the filter itself does not consider if a link is file or not. 1128 */ 1129 public static class HFileLinkFilter implements PathFilter { 1130 1131 @Override 1132 public boolean accept(Path p) { 1133 return HFileLink.isHFileLink(p); 1134 } 1135 } 1136 1137 public static class ReferenceFileFilter extends AbstractFileStatusFilter { 1138 1139 private final FileSystem fs; 1140 1141 public ReferenceFileFilter(FileSystem fs) { 1142 this.fs = fs; 1143 } 1144 1145 @Override 1146 protected boolean accept(Path p, @CheckForNull Boolean isDir) { 1147 if (!StoreFileInfo.isReference(p)) { 1148 return false; 1149 } 1150 1151 try { 1152 // only files can be references. 1153 return isFile(fs, isDir, p); 1154 } catch (IOException ioe) { 1155 // Maybe the file was moved or the fs was disconnected. 1156 LOG.warn("Skipping file " + p +" due to IOException", ioe); 1157 return false; 1158 } 1159 } 1160 } 1161 1162 /** 1163 * Runs through the HBase rootdir/tablename and creates a reverse lookup map for 1164 * table StoreFile names to the full Path. 1165 * <br> 1166 * Example...<br> 1167 * Key = 3944417774205889744 <br> 1168 * Value = hdfs://localhost:51169/user/userid/-ROOT-/70236052/info/3944417774205889744 1169 * 1170 * @param map map to add values. If null, this method will create and populate one to return 1171 * @param fs The file system to use. 1172 * @param hbaseRootDir The root directory to scan. 1173 * @param tableName name of the table to scan. 1174 * @return Map keyed by StoreFile name with a value of the full Path. 1175 * @throws IOException When scanning the directory fails. 1176 * @throws InterruptedException 1177 */ 1178 public static Map<String, Path> getTableStoreFilePathMap(Map<String, Path> map, 1179 final FileSystem fs, final Path hbaseRootDir, TableName tableName) 1180 throws IOException, InterruptedException { 1181 return getTableStoreFilePathMap(map, fs, hbaseRootDir, tableName, null, null, null); 1182 } 1183 1184 /** 1185 * Runs through the HBase rootdir/tablename and creates a reverse lookup map for 1186 * table StoreFile names to the full Path. Note that because this method can be called 1187 * on a 'live' HBase system that we will skip files that no longer exist by the time 1188 * we traverse them and similarly the user of the result needs to consider that some 1189 * entries in this map may not exist by the time this call completes. 1190 * <br> 1191 * Example...<br> 1192 * Key = 3944417774205889744 <br> 1193 * Value = hdfs://localhost:51169/user/userid/-ROOT-/70236052/info/3944417774205889744 1194 * 1195 * @param resultMap map to add values. If null, this method will create and populate one to return 1196 * @param fs The file system to use. 1197 * @param hbaseRootDir The root directory to scan. 1198 * @param tableName name of the table to scan. 1199 * @param sfFilter optional path filter to apply to store files 1200 * @param executor optional executor service to parallelize this operation 1201 * @param errors ErrorReporter instance or null 1202 * @return Map keyed by StoreFile name with a value of the full Path. 1203 * @throws IOException When scanning the directory fails. 1204 * @throws InterruptedException 1205 */ 1206 public static Map<String, Path> getTableStoreFilePathMap( 1207 Map<String, Path> resultMap, 1208 final FileSystem fs, final Path hbaseRootDir, TableName tableName, final PathFilter sfFilter, 1209 ExecutorService executor, final ErrorReporter errors) throws IOException, InterruptedException { 1210 1211 final Map<String, Path> finalResultMap = 1212 resultMap == null ? new ConcurrentHashMap<>(128, 0.75f, 32) : resultMap; 1213 1214 // only include the directory paths to tables 1215 Path tableDir = FSUtils.getTableDir(hbaseRootDir, tableName); 1216 // Inside a table, there are compaction.dir directories to skip. Otherwise, all else 1217 // should be regions. 1218 final FamilyDirFilter familyFilter = new FamilyDirFilter(fs); 1219 final Vector<Exception> exceptions = new Vector<>(); 1220 1221 try { 1222 List<FileStatus> regionDirs = FSUtils.listStatusWithStatusFilter(fs, tableDir, new RegionDirFilter(fs)); 1223 if (regionDirs == null) { 1224 return finalResultMap; 1225 } 1226 1227 final List<Future<?>> futures = new ArrayList<>(regionDirs.size()); 1228 1229 for (FileStatus regionDir : regionDirs) { 1230 if (null != errors) { 1231 errors.progress(); 1232 } 1233 final Path dd = regionDir.getPath(); 1234 1235 if (!exceptions.isEmpty()) { 1236 break; 1237 } 1238 1239 Runnable getRegionStoreFileMapCall = new Runnable() { 1240 @Override 1241 public void run() { 1242 try { 1243 HashMap<String,Path> regionStoreFileMap = new HashMap<>(); 1244 List<FileStatus> familyDirs = FSUtils.listStatusWithStatusFilter(fs, dd, familyFilter); 1245 if (familyDirs == null) { 1246 if (!fs.exists(dd)) { 1247 LOG.warn("Skipping region because it no longer exists: " + dd); 1248 } else { 1249 LOG.warn("Skipping region because it has no family dirs: " + dd); 1250 } 1251 return; 1252 } 1253 for (FileStatus familyDir : familyDirs) { 1254 if (null != errors) { 1255 errors.progress(); 1256 } 1257 Path family = familyDir.getPath(); 1258 if (family.getName().equals(HConstants.RECOVERED_EDITS_DIR)) { 1259 continue; 1260 } 1261 // now in family, iterate over the StoreFiles and 1262 // put in map 1263 FileStatus[] familyStatus = fs.listStatus(family); 1264 for (FileStatus sfStatus : familyStatus) { 1265 if (null != errors) { 1266 errors.progress(); 1267 } 1268 Path sf = sfStatus.getPath(); 1269 if (sfFilter == null || sfFilter.accept(sf)) { 1270 regionStoreFileMap.put( sf.getName(), sf); 1271 } 1272 } 1273 } 1274 finalResultMap.putAll(regionStoreFileMap); 1275 } catch (Exception e) { 1276 LOG.error("Could not get region store file map for region: " + dd, e); 1277 exceptions.add(e); 1278 } 1279 } 1280 }; 1281 1282 // If executor is available, submit async tasks to exec concurrently, otherwise 1283 // just do serial sync execution 1284 if (executor != null) { 1285 Future<?> future = executor.submit(getRegionStoreFileMapCall); 1286 futures.add(future); 1287 } else { 1288 FutureTask<?> future = new FutureTask<>(getRegionStoreFileMapCall, null); 1289 future.run(); 1290 futures.add(future); 1291 } 1292 } 1293 1294 // Ensure all pending tasks are complete (or that we run into an exception) 1295 for (Future<?> f : futures) { 1296 if (!exceptions.isEmpty()) { 1297 break; 1298 } 1299 try { 1300 f.get(); 1301 } catch (ExecutionException e) { 1302 LOG.error("Unexpected exec exception! Should've been caught already. (Bug?)", e); 1303 // Shouldn't happen, we already logged/caught any exceptions in the Runnable 1304 } 1305 } 1306 } catch (IOException e) { 1307 LOG.error("Cannot execute getTableStoreFilePathMap for " + tableName, e); 1308 exceptions.add(e); 1309 } finally { 1310 if (!exceptions.isEmpty()) { 1311 // Just throw the first exception as an indication something bad happened 1312 // Don't need to propagate all the exceptions, we already logged them all anyway 1313 Throwables.propagateIfInstanceOf(exceptions.firstElement(), IOException.class); 1314 throw Throwables.propagate(exceptions.firstElement()); 1315 } 1316 } 1317 1318 return finalResultMap; 1319 } 1320 1321 public static int getRegionReferenceFileCount(final FileSystem fs, final Path p) { 1322 int result = 0; 1323 try { 1324 for (Path familyDir:getFamilyDirs(fs, p)){ 1325 result += getReferenceFilePaths(fs, familyDir).size(); 1326 } 1327 } catch (IOException e) { 1328 LOG.warn("Error Counting reference files.", e); 1329 } 1330 return result; 1331 } 1332 1333 /** 1334 * Runs through the HBase rootdir and creates a reverse lookup map for 1335 * table StoreFile names to the full Path. 1336 * <br> 1337 * Example...<br> 1338 * Key = 3944417774205889744 <br> 1339 * Value = hdfs://localhost:51169/user/userid/-ROOT-/70236052/info/3944417774205889744 1340 * 1341 * @param fs The file system to use. 1342 * @param hbaseRootDir The root directory to scan. 1343 * @return Map keyed by StoreFile name with a value of the full Path. 1344 * @throws IOException When scanning the directory fails. 1345 * @throws InterruptedException 1346 */ 1347 public static Map<String, Path> getTableStoreFilePathMap( 1348 final FileSystem fs, final Path hbaseRootDir) 1349 throws IOException, InterruptedException { 1350 return getTableStoreFilePathMap(fs, hbaseRootDir, null, null, null); 1351 } 1352 1353 /** 1354 * Runs through the HBase rootdir and creates a reverse lookup map for 1355 * table StoreFile names to the full Path. 1356 * <br> 1357 * Example...<br> 1358 * Key = 3944417774205889744 <br> 1359 * Value = hdfs://localhost:51169/user/userid/-ROOT-/70236052/info/3944417774205889744 1360 * 1361 * @param fs The file system to use. 1362 * @param hbaseRootDir The root directory to scan. 1363 * @param sfFilter optional path filter to apply to store files 1364 * @param executor optional executor service to parallelize this operation 1365 * @param errors ErrorReporter instance or null 1366 * @return Map keyed by StoreFile name with a value of the full Path. 1367 * @throws IOException When scanning the directory fails. 1368 * @throws InterruptedException 1369 */ 1370 public static Map<String, Path> getTableStoreFilePathMap( 1371 final FileSystem fs, final Path hbaseRootDir, PathFilter sfFilter, 1372 ExecutorService executor, ErrorReporter errors) 1373 throws IOException, InterruptedException { 1374 ConcurrentHashMap<String, Path> map = new ConcurrentHashMap<>(1024, 0.75f, 32); 1375 1376 // if this method looks similar to 'getTableFragmentation' that is because 1377 // it was borrowed from it. 1378 1379 // only include the directory paths to tables 1380 for (Path tableDir : FSUtils.getTableDirs(fs, hbaseRootDir)) { 1381 getTableStoreFilePathMap(map, fs, hbaseRootDir, 1382 FSUtils.getTableName(tableDir), sfFilter, executor, errors); 1383 } 1384 return map; 1385 } 1386 1387 /** 1388 * Filters FileStatuses in an array and returns a list 1389 * 1390 * @param input An array of FileStatuses 1391 * @param filter A required filter to filter the array 1392 * @return A list of FileStatuses 1393 */ 1394 public static List<FileStatus> filterFileStatuses(FileStatus[] input, 1395 FileStatusFilter filter) { 1396 if (input == null) return null; 1397 return filterFileStatuses(Iterators.forArray(input), filter); 1398 } 1399 1400 /** 1401 * Filters FileStatuses in an iterator and returns a list 1402 * 1403 * @param input An iterator of FileStatuses 1404 * @param filter A required filter to filter the array 1405 * @return A list of FileStatuses 1406 */ 1407 public static List<FileStatus> filterFileStatuses(Iterator<FileStatus> input, 1408 FileStatusFilter filter) { 1409 if (input == null) return null; 1410 ArrayList<FileStatus> results = new ArrayList<>(); 1411 while (input.hasNext()) { 1412 FileStatus f = input.next(); 1413 if (filter.accept(f)) { 1414 results.add(f); 1415 } 1416 } 1417 return results; 1418 } 1419 1420 /** 1421 * Calls fs.listStatus() and treats FileNotFoundException as non-fatal 1422 * This accommodates differences between hadoop versions, where hadoop 1 1423 * does not throw a FileNotFoundException, and return an empty FileStatus[] 1424 * while Hadoop 2 will throw FileNotFoundException. 1425 * 1426 * @param fs file system 1427 * @param dir directory 1428 * @param filter file status filter 1429 * @return null if dir is empty or doesn't exist, otherwise FileStatus list 1430 */ 1431 public static List<FileStatus> listStatusWithStatusFilter(final FileSystem fs, 1432 final Path dir, final FileStatusFilter filter) throws IOException { 1433 FileStatus [] status = null; 1434 try { 1435 status = fs.listStatus(dir); 1436 } catch (FileNotFoundException fnfe) { 1437 // if directory doesn't exist, return null 1438 if (LOG.isTraceEnabled()) { 1439 LOG.trace(dir + " doesn't exist"); 1440 } 1441 } 1442 1443 if (status == null || status.length < 1) { 1444 return null; 1445 } 1446 1447 if (filter == null) { 1448 return Arrays.asList(status); 1449 } else { 1450 List<FileStatus> status2 = filterFileStatuses(status, filter); 1451 if (status2 == null || status2.isEmpty()) { 1452 return null; 1453 } else { 1454 return status2; 1455 } 1456 } 1457 } 1458 1459 /** 1460 * Throw an exception if an action is not permitted by a user on a file. 1461 * 1462 * @param ugi 1463 * the user 1464 * @param file 1465 * the file 1466 * @param action 1467 * the action 1468 */ 1469 public static void checkAccess(UserGroupInformation ugi, FileStatus file, 1470 FsAction action) throws AccessDeniedException { 1471 if (ugi.getShortUserName().equals(file.getOwner())) { 1472 if (file.getPermission().getUserAction().implies(action)) { 1473 return; 1474 } 1475 } else if (contains(ugi.getGroupNames(), file.getGroup())) { 1476 if (file.getPermission().getGroupAction().implies(action)) { 1477 return; 1478 } 1479 } else if (file.getPermission().getOtherAction().implies(action)) { 1480 return; 1481 } 1482 throw new AccessDeniedException("Permission denied:" + " action=" + action 1483 + " path=" + file.getPath() + " user=" + ugi.getShortUserName()); 1484 } 1485 1486 private static boolean contains(String[] groups, String user) { 1487 for (String group : groups) { 1488 if (group.equals(user)) { 1489 return true; 1490 } 1491 } 1492 return false; 1493 } 1494 1495 /** 1496 * This function is to scan the root path of the file system to get the 1497 * degree of locality for each region on each of the servers having at least 1498 * one block of that region. 1499 * This is used by the tool {@link org.apache.hadoop.hbase.master.RegionPlacementMaintainer} 1500 * 1501 * @param conf 1502 * the configuration to use 1503 * @return the mapping from region encoded name to a map of server names to 1504 * locality fraction 1505 * @throws IOException 1506 * in case of file system errors or interrupts 1507 */ 1508 public static Map<String, Map<String, Float>> getRegionDegreeLocalityMappingFromFS( 1509 final Configuration conf) throws IOException { 1510 return getRegionDegreeLocalityMappingFromFS( 1511 conf, null, 1512 conf.getInt(THREAD_POOLSIZE, DEFAULT_THREAD_POOLSIZE)); 1513 1514 } 1515 1516 /** 1517 * This function is to scan the root path of the file system to get the 1518 * degree of locality for each region on each of the servers having at least 1519 * one block of that region. 1520 * 1521 * @param conf 1522 * the configuration to use 1523 * @param desiredTable 1524 * the table you wish to scan locality for 1525 * @param threadPoolSize 1526 * the thread pool size to use 1527 * @return the mapping from region encoded name to a map of server names to 1528 * locality fraction 1529 * @throws IOException 1530 * in case of file system errors or interrupts 1531 */ 1532 public static Map<String, Map<String, Float>> getRegionDegreeLocalityMappingFromFS( 1533 final Configuration conf, final String desiredTable, int threadPoolSize) 1534 throws IOException { 1535 Map<String, Map<String, Float>> regionDegreeLocalityMapping = new ConcurrentHashMap<>(); 1536 getRegionLocalityMappingFromFS(conf, desiredTable, threadPoolSize, null, 1537 regionDegreeLocalityMapping); 1538 return regionDegreeLocalityMapping; 1539 } 1540 1541 /** 1542 * This function is to scan the root path of the file system to get either the 1543 * mapping between the region name and its best locality region server or the 1544 * degree of locality of each region on each of the servers having at least 1545 * one block of that region. The output map parameters are both optional. 1546 * 1547 * @param conf 1548 * the configuration to use 1549 * @param desiredTable 1550 * the table you wish to scan locality for 1551 * @param threadPoolSize 1552 * the thread pool size to use 1553 * @param regionToBestLocalityRSMapping 1554 * the map into which to put the best locality mapping or null 1555 * @param regionDegreeLocalityMapping 1556 * the map into which to put the locality degree mapping or null, 1557 * must be a thread-safe implementation 1558 * @throws IOException 1559 * in case of file system errors or interrupts 1560 */ 1561 private static void getRegionLocalityMappingFromFS( 1562 final Configuration conf, final String desiredTable, 1563 int threadPoolSize, 1564 Map<String, String> regionToBestLocalityRSMapping, 1565 Map<String, Map<String, Float>> regionDegreeLocalityMapping) 1566 throws IOException { 1567 FileSystem fs = FileSystem.get(conf); 1568 Path rootPath = FSUtils.getRootDir(conf); 1569 long startTime = EnvironmentEdgeManager.currentTime(); 1570 Path queryPath; 1571 // The table files are in ${hbase.rootdir}/data/<namespace>/<table>/* 1572 if (null == desiredTable) { 1573 queryPath = new Path(new Path(rootPath, HConstants.BASE_NAMESPACE_DIR).toString() + "/*/*/*/"); 1574 } else { 1575 queryPath = new Path(FSUtils.getTableDir(rootPath, TableName.valueOf(desiredTable)).toString() + "/*/"); 1576 } 1577 1578 // reject all paths that are not appropriate 1579 PathFilter pathFilter = new PathFilter() { 1580 @Override 1581 public boolean accept(Path path) { 1582 // this is the region name; it may get some noise data 1583 if (null == path) { 1584 return false; 1585 } 1586 1587 // no parent? 1588 Path parent = path.getParent(); 1589 if (null == parent) { 1590 return false; 1591 } 1592 1593 String regionName = path.getName(); 1594 if (null == regionName) { 1595 return false; 1596 } 1597 1598 if (!regionName.toLowerCase(Locale.ROOT).matches("[0-9a-f]+")) { 1599 return false; 1600 } 1601 return true; 1602 } 1603 }; 1604 1605 FileStatus[] statusList = fs.globStatus(queryPath, pathFilter); 1606 1607 if (null == statusList) { 1608 return; 1609 } else { 1610 LOG.debug("Query Path: " + queryPath + " ; # list of files: " + 1611 statusList.length); 1612 } 1613 1614 // lower the number of threads in case we have very few expected regions 1615 threadPoolSize = Math.min(threadPoolSize, statusList.length); 1616 1617 // run in multiple threads 1618 ThreadPoolExecutor tpe = new ThreadPoolExecutor(threadPoolSize, 1619 threadPoolSize, 60, TimeUnit.SECONDS, 1620 new ArrayBlockingQueue<>(statusList.length)); 1621 try { 1622 // ignore all file status items that are not of interest 1623 for (FileStatus regionStatus : statusList) { 1624 if (null == regionStatus) { 1625 continue; 1626 } 1627 1628 if (!regionStatus.isDirectory()) { 1629 continue; 1630 } 1631 1632 Path regionPath = regionStatus.getPath(); 1633 if (null == regionPath) { 1634 continue; 1635 } 1636 1637 tpe.execute(new FSRegionScanner(fs, regionPath, 1638 regionToBestLocalityRSMapping, regionDegreeLocalityMapping)); 1639 } 1640 } finally { 1641 tpe.shutdown(); 1642 int threadWakeFrequency = conf.getInt(HConstants.THREAD_WAKE_FREQUENCY, 1643 60 * 1000); 1644 try { 1645 // here we wait until TPE terminates, which is either naturally or by 1646 // exceptions in the execution of the threads 1647 while (!tpe.awaitTermination(threadWakeFrequency, 1648 TimeUnit.MILLISECONDS)) { 1649 // printing out rough estimate, so as to not introduce 1650 // AtomicInteger 1651 LOG.info("Locality checking is underway: { Scanned Regions : " 1652 + tpe.getCompletedTaskCount() + "/" 1653 + tpe.getTaskCount() + " }"); 1654 } 1655 } catch (InterruptedException e) { 1656 throw (InterruptedIOException)new InterruptedIOException().initCause(e); 1657 } 1658 } 1659 1660 long overhead = EnvironmentEdgeManager.currentTime() - startTime; 1661 String overheadMsg = "Scan DFS for locality info takes " + overhead + " ms"; 1662 1663 LOG.info(overheadMsg); 1664 } 1665 1666 /** 1667 * Do our short circuit read setup. 1668 * Checks buffer size to use and whether to do checksumming in hbase or hdfs. 1669 * @param conf 1670 */ 1671 public static void setupShortCircuitRead(final Configuration conf) { 1672 // Check that the user has not set the "dfs.client.read.shortcircuit.skip.checksum" property. 1673 boolean shortCircuitSkipChecksum = 1674 conf.getBoolean("dfs.client.read.shortcircuit.skip.checksum", false); 1675 boolean useHBaseChecksum = conf.getBoolean(HConstants.HBASE_CHECKSUM_VERIFICATION, true); 1676 if (shortCircuitSkipChecksum) { 1677 LOG.warn("Configuration \"dfs.client.read.shortcircuit.skip.checksum\" should not " + 1678 "be set to true." + (useHBaseChecksum ? " HBase checksum doesn't require " + 1679 "it, see https://issues.apache.org/jira/browse/HBASE-6868." : "")); 1680 assert !shortCircuitSkipChecksum; //this will fail if assertions are on 1681 } 1682 checkShortCircuitReadBufferSize(conf); 1683 } 1684 1685 /** 1686 * Check if short circuit read buffer size is set and if not, set it to hbase value. 1687 * @param conf 1688 */ 1689 public static void checkShortCircuitReadBufferSize(final Configuration conf) { 1690 final int defaultSize = HConstants.DEFAULT_BLOCKSIZE * 2; 1691 final int notSet = -1; 1692 // DFSConfigKeys.DFS_CLIENT_READ_SHORTCIRCUIT_BUFFER_SIZE_KEY is only defined in h2 1693 final String dfsKey = "dfs.client.read.shortcircuit.buffer.size"; 1694 int size = conf.getInt(dfsKey, notSet); 1695 // If a size is set, return -- we will use it. 1696 if (size != notSet) return; 1697 // But short circuit buffer size is normally not set. Put in place the hbase wanted size. 1698 int hbaseSize = conf.getInt("hbase." + dfsKey, defaultSize); 1699 conf.setIfUnset(dfsKey, Integer.toString(hbaseSize)); 1700 } 1701 1702 /** 1703 * @param c 1704 * @return The DFSClient DFSHedgedReadMetrics instance or null if can't be found or not on hdfs. 1705 * @throws IOException 1706 */ 1707 public static DFSHedgedReadMetrics getDFSHedgedReadMetrics(final Configuration c) 1708 throws IOException { 1709 if (!isHDFS(c)) return null; 1710 // getHedgedReadMetrics is package private. Get the DFSClient instance that is internal 1711 // to the DFS FS instance and make the method getHedgedReadMetrics accessible, then invoke it 1712 // to get the singleton instance of DFSHedgedReadMetrics shared by DFSClients. 1713 final String name = "getHedgedReadMetrics"; 1714 DFSClient dfsclient = ((DistributedFileSystem)FileSystem.get(c)).getClient(); 1715 Method m; 1716 try { 1717 m = dfsclient.getClass().getDeclaredMethod(name); 1718 } catch (NoSuchMethodException e) { 1719 LOG.warn("Failed find method " + name + " in dfsclient; no hedged read metrics: " + 1720 e.getMessage()); 1721 return null; 1722 } catch (SecurityException e) { 1723 LOG.warn("Failed find method " + name + " in dfsclient; no hedged read metrics: " + 1724 e.getMessage()); 1725 return null; 1726 } 1727 m.setAccessible(true); 1728 try { 1729 return (DFSHedgedReadMetrics)m.invoke(dfsclient); 1730 } catch (IllegalAccessException e) { 1731 LOG.warn("Failed invoking method " + name + " on dfsclient; no hedged read metrics: " + 1732 e.getMessage()); 1733 return null; 1734 } catch (IllegalArgumentException e) { 1735 LOG.warn("Failed invoking method " + name + " on dfsclient; no hedged read metrics: " + 1736 e.getMessage()); 1737 return null; 1738 } catch (InvocationTargetException e) { 1739 LOG.warn("Failed invoking method " + name + " on dfsclient; no hedged read metrics: " + 1740 e.getMessage()); 1741 return null; 1742 } 1743 } 1744 1745}