1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.util;
20
21 import com.google.common.annotations.VisibleForTesting;
22 import com.google.common.base.Throwables;
23 import com.google.common.collect.Iterators;
24 import com.google.common.collect.Lists;
25 import com.google.common.primitives.Ints;
26
27 import edu.umd.cs.findbugs.annotations.CheckForNull;
28
29 import java.io.ByteArrayInputStream;
30 import java.io.DataInputStream;
31 import java.io.EOFException;
32 import java.io.FileNotFoundException;
33 import java.io.IOException;
34 import java.io.InputStream;
35 import java.io.InterruptedIOException;
36 import java.lang.reflect.InvocationTargetException;
37 import java.lang.reflect.Method;
38 import java.net.InetSocketAddress;
39 import java.net.URI;
40 import java.net.URISyntaxException;
41 import java.util.ArrayList;
42 import java.util.Arrays;
43 import java.util.Collections;
44 import java.util.HashMap;
45 import java.util.Iterator;
46 import java.util.LinkedList;
47 import java.util.List;
48 import java.util.Locale;
49 import java.util.Map;
50 import java.util.Vector;
51 import java.util.concurrent.ArrayBlockingQueue;
52 import java.util.concurrent.ConcurrentHashMap;
53 import java.util.concurrent.ExecutionException;
54 import java.util.concurrent.ExecutorService;
55 import java.util.concurrent.Future;
56 import java.util.concurrent.FutureTask;
57 import java.util.concurrent.ThreadPoolExecutor;
58 import java.util.concurrent.TimeUnit;
59 import java.util.regex.Pattern;
60
61 import org.apache.commons.logging.Log;
62 import org.apache.commons.logging.LogFactory;
63 import org.apache.hadoop.HadoopIllegalArgumentException;
64 import org.apache.hadoop.conf.Configuration;
65 import org.apache.hadoop.fs.BlockLocation;
66 import org.apache.hadoop.fs.FSDataInputStream;
67 import org.apache.hadoop.fs.FSDataOutputStream;
68 import org.apache.hadoop.fs.FileStatus;
69 import org.apache.hadoop.fs.FileSystem;
70 import org.apache.hadoop.fs.LocatedFileStatus;
71 import org.apache.hadoop.fs.Path;
72 import org.apache.hadoop.fs.PathFilter;
73 import org.apache.hadoop.fs.RemoteIterator;
74 import org.apache.hadoop.fs.permission.FsAction;
75 import org.apache.hadoop.fs.permission.FsPermission;
76 import org.apache.hadoop.hbase.ClusterId;
77 import org.apache.hadoop.hbase.HColumnDescriptor;
78 import org.apache.hadoop.hbase.HConstants;
79 import org.apache.hadoop.hbase.HDFSBlocksDistribution;
80 import org.apache.hadoop.hbase.HRegionInfo;
81 import org.apache.hadoop.hbase.RemoteExceptionHandler;
82 import org.apache.hadoop.hbase.TableName;
83 import org.apache.hadoop.hbase.classification.InterfaceAudience;
84 import org.apache.hadoop.hbase.exceptions.DeserializationException;
85 import org.apache.hadoop.hbase.fs.HFileSystem;
86 import org.apache.hadoop.hbase.io.HFileLink;
87 import org.apache.hadoop.hbase.master.HMaster;
88 import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
89 import org.apache.hadoop.hbase.security.AccessDeniedException;
90 import org.apache.hadoop.hbase.util.HBaseFsck.ErrorReporter;
91 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
92 import org.apache.hadoop.hbase.protobuf.generated.FSProtos;
93 import org.apache.hadoop.hbase.regionserver.HRegion;
94 import org.apache.hadoop.hdfs.DistributedFileSystem;
95 import org.apache.hadoop.io.IOUtils;
96 import org.apache.hadoop.io.SequenceFile;
97 import org.apache.hadoop.ipc.RemoteException;
98 import org.apache.hadoop.security.UserGroupInformation;
99 import org.apache.hadoop.util.Progressable;
100 import org.apache.hadoop.util.ReflectionUtils;
101 import org.apache.hadoop.util.StringUtils;
102
103
104 import static org.apache.hadoop.hbase.HConstants.HBASE_DIR;
105
106
107
108
109 @InterfaceAudience.Private
110 public abstract class FSUtils {
111 private static final Log LOG = LogFactory.getLog(FSUtils.class);
112
113
114 public static final String FULL_RWX_PERMISSIONS = "777";
115 private static final String THREAD_POOLSIZE = "hbase.client.localityCheck.threadPoolSize";
116 private static final int DEFAULT_THREAD_POOLSIZE = 2;
117
118
119 public static final boolean WINDOWS = System.getProperty("os.name").startsWith("Windows");
120
121 protected FSUtils() {
122 super();
123 }
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141 public static void setStoragePolicy(final FileSystem fs, final Configuration conf,
142 final Path path, final String policyKey, final String defaultPolicy) {
143 String storagePolicy = conf.get(policyKey, defaultPolicy).toUpperCase(Locale.ROOT);
144 if (storagePolicy.equals(defaultPolicy)) {
145 if (LOG.isTraceEnabled()) {
146 LOG.trace("default policy of " + defaultPolicy + " requested, exiting early.");
147 }
148 return;
149 }
150 if (fs instanceof DistributedFileSystem) {
151 DistributedFileSystem dfs = (DistributedFileSystem)fs;
152
153 Class<? extends DistributedFileSystem> dfsClass = dfs.getClass();
154 Method m = null;
155 try {
156 m = dfsClass.getDeclaredMethod("setStoragePolicy",
157 new Class<?>[] { Path.class, String.class });
158 m.setAccessible(true);
159 } catch (NoSuchMethodException e) {
160 LOG.info("FileSystem doesn't support"
161 + " setStoragePolicy; --HDFS-6584 not available");
162 } catch (SecurityException e) {
163 LOG.info("Doesn't have access to setStoragePolicy on "
164 + "FileSystems --HDFS-6584 not available", e);
165 m = null;
166 }
167 if (m != null) {
168 try {
169 m.invoke(dfs, path, storagePolicy);
170 LOG.info("set " + storagePolicy + " for " + path);
171 } catch (Exception e) {
172
173 boolean probablyBadPolicy = false;
174 if (e instanceof InvocationTargetException) {
175 final Throwable exception = e.getCause();
176 if (exception instanceof RemoteException &&
177 HadoopIllegalArgumentException.class.getName().equals(
178 ((RemoteException)exception).getClassName())) {
179 LOG.warn("Given storage policy, '" + storagePolicy + "', was rejected and probably " +
180 "isn't a valid policy for the version of Hadoop you're running. I.e. if you're " +
181 "trying to use SSD related policies then you're likely missing HDFS-7228. For " +
182 "more information see the 'ArchivalStorage' docs for your Hadoop release.");
183 LOG.debug("More information about the invalid storage policy.", exception);
184 probablyBadPolicy = true;
185 }
186 }
187 if (!probablyBadPolicy) {
188
189
190 LOG.warn("Unable to set " + storagePolicy + " for " + path, e);
191 }
192 }
193 }
194 } else {
195 LOG.info("FileSystem isn't an instance of DistributedFileSystem; presuming it doesn't " +
196 "support setStoragePolicy.");
197 }
198 }
199
200
201
202
203
204 public static boolean isDistributedFileSystem(final FileSystem fs) throws IOException {
205 FileSystem fileSystem = fs;
206
207
208 if (fs instanceof HFileSystem) {
209 fileSystem = ((HFileSystem)fs).getBackingFs();
210 }
211 return fileSystem instanceof DistributedFileSystem;
212 }
213
214
215
216
217
218
219
220
221
222 public static boolean isStartingWithPath(final Path rootPath, final String path) {
223 String uriRootPath = rootPath.toUri().getPath();
224 String tailUriPath = (new Path(path)).toUri().getPath();
225 return tailUriPath.startsWith(uriRootPath);
226 }
227
228
229
230
231
232
233
234
235
236 public static boolean isMatchingTail(final Path pathToSearch, String pathTail) {
237 return isMatchingTail(pathToSearch, new Path(pathTail));
238 }
239
240
241
242
243
244
245
246
247
248 public static boolean isMatchingTail(final Path pathToSearch, final Path pathTail) {
249 if (pathToSearch.depth() != pathTail.depth()) return false;
250 Path tailPath = pathTail;
251 String tailName;
252 Path toSearch = pathToSearch;
253 String toSearchName;
254 boolean result = false;
255 do {
256 tailName = tailPath.getName();
257 if (tailName == null || tailName.length() <= 0) {
258 result = true;
259 break;
260 }
261 toSearchName = toSearch.getName();
262 if (toSearchName == null || toSearchName.length() <= 0) break;
263
264 tailPath = tailPath.getParent();
265 toSearch = toSearch.getParent();
266 } while(tailName.equals(toSearchName));
267 return result;
268 }
269
270 public static FSUtils getInstance(FileSystem fs, Configuration conf) {
271 String scheme = fs.getUri().getScheme();
272 if (scheme == null) {
273 LOG.warn("Could not find scheme for uri " +
274 fs.getUri() + ", default to hdfs");
275 scheme = "hdfs";
276 }
277 Class<?> fsUtilsClass = conf.getClass("hbase.fsutil." +
278 scheme + ".impl", FSHDFSUtils.class);
279 FSUtils fsUtils = (FSUtils)ReflectionUtils.newInstance(fsUtilsClass, conf);
280 return fsUtils;
281 }
282
283
284
285
286
287
288
289
290 public static boolean deleteDirectory(final FileSystem fs, final Path dir)
291 throws IOException {
292 return fs.exists(dir) && fs.delete(dir, true);
293 }
294
295
296
297
298
299
300
301
302 public static boolean deleteRegionDir(final Configuration conf, final HRegionInfo hri)
303 throws IOException {
304 Path rootDir = getRootDir(conf);
305 FileSystem fs = rootDir.getFileSystem(conf);
306 return deleteDirectory(fs,
307 new Path(getTableDir(rootDir, hri.getTable()), hri.getEncodedName()));
308 }
309
310
311
312
313
314
315
316
317
318
319
320
321 public static long getDefaultBlockSize(final FileSystem fs, final Path path) throws IOException {
322 Method m = null;
323 Class<? extends FileSystem> cls = fs.getClass();
324 try {
325 m = cls.getMethod("getDefaultBlockSize", new Class<?>[] { Path.class });
326 } catch (NoSuchMethodException e) {
327 LOG.info("FileSystem doesn't support getDefaultBlockSize");
328 } catch (SecurityException e) {
329 LOG.info("Doesn't have access to getDefaultBlockSize on FileSystems", e);
330 m = null;
331 }
332 if (m == null) {
333 return fs.getDefaultBlockSize(path);
334 } else {
335 try {
336 Object ret = m.invoke(fs, path);
337 return ((Long)ret).longValue();
338 } catch (Exception e) {
339 throw new IOException(e);
340 }
341 }
342 }
343
344
345
346
347
348
349
350
351
352
353
354
355 public static short getDefaultReplication(final FileSystem fs, final Path path) throws IOException {
356 Method m = null;
357 Class<? extends FileSystem> cls = fs.getClass();
358 try {
359 m = cls.getMethod("getDefaultReplication", new Class<?>[] { Path.class });
360 } catch (NoSuchMethodException e) {
361 LOG.info("FileSystem doesn't support getDefaultReplication");
362 } catch (SecurityException e) {
363 LOG.info("Doesn't have access to getDefaultReplication on FileSystems", e);
364 m = null;
365 }
366 if (m == null) {
367 return fs.getDefaultReplication(path);
368 } else {
369 try {
370 Object ret = m.invoke(fs, path);
371 return ((Number)ret).shortValue();
372 } catch (Exception e) {
373 throw new IOException(e);
374 }
375 }
376 }
377
378
379
380
381
382
383
384
385
386
387
388 public static int getDefaultBufferSize(final FileSystem fs) {
389 return fs.getConf().getInt("io.file.buffer.size", 4096);
390 }
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411 public static FSDataOutputStream create(Configuration conf, FileSystem fs, Path path,
412 FsPermission perm, InetSocketAddress[] favoredNodes) throws IOException {
413 if (fs instanceof HFileSystem) {
414 FileSystem backingFs = ((HFileSystem)fs).getBackingFs();
415 if (backingFs instanceof DistributedFileSystem) {
416
417
418 short replication = Short.parseShort(conf.get(HColumnDescriptor.DFS_REPLICATION,
419 String.valueOf(HColumnDescriptor.DEFAULT_DFS_REPLICATION)));
420 try {
421 return (FSDataOutputStream) (DistributedFileSystem.class.getDeclaredMethod("create",
422 Path.class, FsPermission.class, boolean.class, int.class, short.class, long.class,
423 Progressable.class, InetSocketAddress[].class).invoke(backingFs, path, perm, true,
424 getDefaultBufferSize(backingFs),
425 replication > 0 ? replication : getDefaultReplication(backingFs, path),
426 getDefaultBlockSize(backingFs, path), null, favoredNodes));
427 } catch (InvocationTargetException ite) {
428
429 throw new IOException(ite.getCause());
430 } catch (NoSuchMethodException e) {
431 LOG.debug("DFS Client does not support most favored nodes create; using default create");
432 if (LOG.isTraceEnabled()) LOG.trace("Ignoring; use default create", e);
433 } catch (IllegalArgumentException e) {
434 LOG.debug("Ignoring (most likely Reflection related exception) " + e);
435 } catch (SecurityException e) {
436 LOG.debug("Ignoring (most likely Reflection related exception) " + e);
437 } catch (IllegalAccessException e) {
438 LOG.debug("Ignoring (most likely Reflection related exception) " + e);
439 }
440 }
441 }
442 return create(fs, path, perm, true);
443 }
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462 public static FSDataOutputStream create(FileSystem fs, Path path,
463 FsPermission perm, boolean overwrite) throws IOException {
464 if (LOG.isTraceEnabled()) {
465 LOG.trace("Creating file=" + path + " with permission=" + perm + ", overwrite=" + overwrite);
466 }
467 return fs.create(path, perm, overwrite, getDefaultBufferSize(fs),
468 getDefaultReplication(fs, path), getDefaultBlockSize(fs, path), null);
469 }
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484 public static FsPermission getFilePermissions(final FileSystem fs,
485 final Configuration conf, final String permssionConfKey) {
486 boolean enablePermissions = conf.getBoolean(
487 HConstants.ENABLE_DATA_FILE_UMASK, false);
488
489 if (enablePermissions) {
490 try {
491 FsPermission perm = new FsPermission(FULL_RWX_PERMISSIONS);
492
493 String mask = conf.get(permssionConfKey);
494 if (mask == null)
495 return FsPermission.getFileDefault();
496
497 FsPermission umask = new FsPermission(mask);
498 return perm.applyUMask(umask);
499 } catch (IllegalArgumentException e) {
500 LOG.warn(
501 "Incorrect umask attempted to be created: "
502 + conf.get(permssionConfKey)
503 + ", using default file permissions.", e);
504 return FsPermission.getFileDefault();
505 }
506 }
507 return FsPermission.getFileDefault();
508 }
509
510
511
512
513
514
515
516 public static void checkFileSystemAvailable(final FileSystem fs)
517 throws IOException {
518 if (!(fs instanceof DistributedFileSystem)) {
519 return;
520 }
521 IOException exception = null;
522 DistributedFileSystem dfs = (DistributedFileSystem) fs;
523 try {
524 if (dfs.exists(new Path("/"))) {
525 return;
526 }
527 } catch (IOException e) {
528 exception = RemoteExceptionHandler.checkIOException(e);
529 }
530 try {
531 fs.close();
532 } catch (Exception e) {
533 LOG.error("file system close failed: ", e);
534 }
535 IOException io = new IOException("File system is not available");
536 io.initCause(exception);
537 throw io;
538 }
539
540
541
542
543
544
545
546
547
548 private static boolean isInSafeMode(DistributedFileSystem dfs) throws IOException {
549 boolean inSafeMode = false;
550 try {
551 Method m = DistributedFileSystem.class.getMethod("setSafeMode", new Class<?> []{
552 org.apache.hadoop.hdfs.protocol.HdfsConstants.SafeModeAction.class, boolean.class});
553 inSafeMode = (Boolean) m.invoke(dfs,
554 org.apache.hadoop.hdfs.protocol.HdfsConstants.SafeModeAction.SAFEMODE_GET, true);
555 } catch (Exception e) {
556 if (e instanceof IOException) throw (IOException) e;
557
558
559 inSafeMode = dfs.setSafeMode(
560 org.apache.hadoop.hdfs.protocol.HdfsConstants.SafeModeAction.SAFEMODE_GET);
561 }
562 return inSafeMode;
563 }
564
565
566
567
568
569
570 public static void checkDfsSafeMode(final Configuration conf)
571 throws IOException {
572 boolean isInSafeMode = false;
573 FileSystem fs = FileSystem.get(conf);
574 if (fs instanceof DistributedFileSystem) {
575 DistributedFileSystem dfs = (DistributedFileSystem)fs;
576 isInSafeMode = isInSafeMode(dfs);
577 }
578 if (isInSafeMode) {
579 throw new IOException("File system is in safemode, it can't be written now");
580 }
581 }
582
583
584
585
586
587
588
589
590
591
592 public static String getVersion(FileSystem fs, Path rootdir)
593 throws IOException, DeserializationException {
594 Path versionFile = new Path(rootdir, HConstants.VERSION_FILE_NAME);
595 FileStatus[] status = null;
596 try {
597
598
599 status = fs.listStatus(versionFile);
600 } catch (FileNotFoundException fnfe) {
601 return null;
602 }
603 if (status == null || status.length == 0) return null;
604 String version = null;
605 byte [] content = new byte [(int)status[0].getLen()];
606 FSDataInputStream s = fs.open(versionFile);
607 try {
608 IOUtils.readFully(s, content, 0, content.length);
609 if (ProtobufUtil.isPBMagicPrefix(content)) {
610 version = parseVersionFrom(content);
611 } else {
612
613 InputStream is = new ByteArrayInputStream(content);
614 DataInputStream dis = new DataInputStream(is);
615 try {
616 version = dis.readUTF();
617 } finally {
618 dis.close();
619 }
620 }
621 } catch (EOFException eof) {
622 LOG.warn("Version file was empty, odd, will try to set it.");
623 } finally {
624 s.close();
625 }
626 return version;
627 }
628
629
630
631
632
633
634
635 static String parseVersionFrom(final byte [] bytes)
636 throws DeserializationException {
637 ProtobufUtil.expectPBMagicPrefix(bytes);
638 int pblen = ProtobufUtil.lengthOfPBMagic();
639 FSProtos.HBaseVersionFileContent.Builder builder =
640 FSProtos.HBaseVersionFileContent.newBuilder();
641 try {
642 ProtobufUtil.mergeFrom(builder, bytes, pblen, bytes.length - pblen);
643 return builder.getVersion();
644 } catch (IOException e) {
645
646 throw new DeserializationException(e);
647 }
648 }
649
650
651
652
653
654
655 static byte [] toVersionByteArray(final String version) {
656 FSProtos.HBaseVersionFileContent.Builder builder =
657 FSProtos.HBaseVersionFileContent.newBuilder();
658 return ProtobufUtil.prependPBMagic(builder.setVersion(version).build().toByteArray());
659 }
660
661
662
663
664
665
666
667
668
669
670
671 public static void checkVersion(FileSystem fs, Path rootdir, boolean message)
672 throws IOException, DeserializationException {
673 checkVersion(fs, rootdir, message, 0, HConstants.DEFAULT_VERSION_FILE_WRITE_ATTEMPTS);
674 }
675
676
677
678
679
680
681
682
683
684
685
686
687
688 public static void checkVersion(FileSystem fs, Path rootdir,
689 boolean message, int wait, int retries)
690 throws IOException, DeserializationException {
691 String version = getVersion(fs, rootdir);
692 if (version == null) {
693 if (!metaRegionExists(fs, rootdir)) {
694
695
696 setVersion(fs, rootdir, wait, retries);
697 return;
698 }
699 } else if (version.compareTo(HConstants.FILE_SYSTEM_VERSION) == 0) return;
700
701
702
703 String msg = "HBase file layout needs to be upgraded."
704 + " You have version " + version
705 + " and I want version " + HConstants.FILE_SYSTEM_VERSION
706 + ". Consult http://hbase.apache.org/book.html for further information about upgrading HBase."
707 + " Is your hbase.rootdir valid? If so, you may need to run "
708 + "'hbase hbck -fixVersionFile'.";
709 if (message) {
710 System.out.println("WARNING! " + msg);
711 }
712 throw new FileSystemVersionException(msg);
713 }
714
715
716
717
718
719
720
721
722 public static void setVersion(FileSystem fs, Path rootdir)
723 throws IOException {
724 setVersion(fs, rootdir, HConstants.FILE_SYSTEM_VERSION, 0,
725 HConstants.DEFAULT_VERSION_FILE_WRITE_ATTEMPTS);
726 }
727
728
729
730
731
732
733
734
735
736
737 public static void setVersion(FileSystem fs, Path rootdir, int wait, int retries)
738 throws IOException {
739 setVersion(fs, rootdir, HConstants.FILE_SYSTEM_VERSION, wait, retries);
740 }
741
742
743
744
745
746
747
748
749
750
751
752
753 public static void setVersion(FileSystem fs, Path rootdir, String version,
754 int wait, int retries) throws IOException {
755 Path versionFile = new Path(rootdir, HConstants.VERSION_FILE_NAME);
756 Path tempVersionFile = new Path(rootdir, HConstants.HBASE_TEMP_DIRECTORY + Path.SEPARATOR +
757 HConstants.VERSION_FILE_NAME);
758 while (true) {
759 try {
760
761 FSDataOutputStream s = fs.create(tempVersionFile);
762 try {
763 s.write(toVersionByteArray(version));
764 s.close();
765 s = null;
766
767
768 if (!fs.rename(tempVersionFile, versionFile)) {
769 throw new IOException("Unable to move temp version file to " + versionFile);
770 }
771 } finally {
772
773
774
775
776
777 try {
778 if (s != null) s.close();
779 } catch (IOException ignore) { }
780 }
781 LOG.info("Created version file at " + rootdir.toString() + " with version=" + version);
782 return;
783 } catch (IOException e) {
784 if (retries > 0) {
785 LOG.debug("Unable to create version file at " + rootdir.toString() + ", retrying", e);
786 fs.delete(versionFile, false);
787 try {
788 if (wait > 0) {
789 Thread.sleep(wait);
790 }
791 } catch (InterruptedException ie) {
792 throw (InterruptedIOException)new InterruptedIOException().initCause(ie);
793 }
794 retries--;
795 } else {
796 throw e;
797 }
798 }
799 }
800 }
801
802
803
804
805
806
807
808
809
810 public static boolean checkClusterIdExists(FileSystem fs, Path rootdir,
811 int wait) throws IOException {
812 while (true) {
813 try {
814 Path filePath = new Path(rootdir, HConstants.CLUSTER_ID_FILE_NAME);
815 return fs.exists(filePath);
816 } catch (IOException ioe) {
817 if (wait > 0) {
818 LOG.warn("Unable to check cluster ID file in " + rootdir.toString() +
819 ", retrying in "+wait+"msec: "+StringUtils.stringifyException(ioe));
820 try {
821 Thread.sleep(wait);
822 } catch (InterruptedException e) {
823 throw (InterruptedIOException)new InterruptedIOException().initCause(e);
824 }
825 } else {
826 throw ioe;
827 }
828 }
829 }
830 }
831
832
833
834
835
836
837
838
839 public static ClusterId getClusterId(FileSystem fs, Path rootdir)
840 throws IOException {
841 Path idPath = new Path(rootdir, HConstants.CLUSTER_ID_FILE_NAME);
842 ClusterId clusterId = null;
843 FileStatus status = fs.exists(idPath)? fs.getFileStatus(idPath): null;
844 if (status != null) {
845 int len = Ints.checkedCast(status.getLen());
846 byte [] content = new byte[len];
847 FSDataInputStream in = fs.open(idPath);
848 try {
849 in.readFully(content);
850 } catch (EOFException eof) {
851 LOG.warn("Cluster ID file " + idPath.toString() + " was empty");
852 } finally{
853 in.close();
854 }
855 try {
856 clusterId = ClusterId.parseFrom(content);
857 } catch (DeserializationException e) {
858 throw new IOException("content=" + Bytes.toString(content), e);
859 }
860
861 if (!ProtobufUtil.isPBMagicPrefix(content)) {
862 String cid = null;
863 in = fs.open(idPath);
864 try {
865 cid = in.readUTF();
866 clusterId = new ClusterId(cid);
867 } catch (EOFException eof) {
868 LOG.warn("Cluster ID file " + idPath.toString() + " was empty");
869 } finally {
870 in.close();
871 }
872 rewriteAsPb(fs, rootdir, idPath, clusterId);
873 }
874 return clusterId;
875 } else {
876 LOG.warn("Cluster ID file does not exist at " + idPath.toString());
877 }
878 return clusterId;
879 }
880
881
882
883
884
885 private static void rewriteAsPb(final FileSystem fs, final Path rootdir, final Path p,
886 final ClusterId cid)
887 throws IOException {
888
889
890 Path movedAsideName = new Path(p + "." + System.currentTimeMillis());
891 if (!fs.rename(p, movedAsideName)) throw new IOException("Failed rename of " + p);
892 setClusterId(fs, rootdir, cid, 100);
893 if (!fs.delete(movedAsideName, false)) {
894 throw new IOException("Failed delete of " + movedAsideName);
895 }
896 LOG.debug("Rewrote the hbase.id file as pb");
897 }
898
899
900
901
902
903
904
905
906
907
908 public static void setClusterId(FileSystem fs, Path rootdir, ClusterId clusterId,
909 int wait) throws IOException {
910 while (true) {
911 try {
912 Path idFile = new Path(rootdir, HConstants.CLUSTER_ID_FILE_NAME);
913 Path tempIdFile = new Path(rootdir, HConstants.HBASE_TEMP_DIRECTORY +
914 Path.SEPARATOR + HConstants.CLUSTER_ID_FILE_NAME);
915
916 FSDataOutputStream s = fs.create(tempIdFile);
917 try {
918 s.write(clusterId.toByteArray());
919 s.close();
920 s = null;
921
922
923 if (!fs.rename(tempIdFile, idFile)) {
924 throw new IOException("Unable to move temp version file to " + idFile);
925 }
926 } finally {
927
928 try {
929 if (s != null) s.close();
930 } catch (IOException ignore) { }
931 }
932 if (LOG.isDebugEnabled()) {
933 LOG.debug("Created cluster ID file at " + idFile.toString() + " with ID: " + clusterId);
934 }
935 return;
936 } catch (IOException ioe) {
937 if (wait > 0) {
938 LOG.warn("Unable to create cluster ID file in " + rootdir.toString() +
939 ", retrying in " + wait + "msec: " + StringUtils.stringifyException(ioe));
940 try {
941 Thread.sleep(wait);
942 } catch (InterruptedException e) {
943 throw (InterruptedIOException)new InterruptedIOException().initCause(e);
944 }
945 } else {
946 throw ioe;
947 }
948 }
949 }
950 }
951
952
953
954
955
956
957
958
959 public static Path validateRootPath(Path root) throws IOException {
960 try {
961 URI rootURI = new URI(root.toString());
962 String scheme = rootURI.getScheme();
963 if (scheme == null) {
964 throw new IOException("Root directory does not have a scheme");
965 }
966 return root;
967 } catch (URISyntaxException e) {
968 IOException io = new IOException("Root directory path is not a valid " +
969 "URI -- check your " + HBASE_DIR + " configuration");
970 io.initCause(e);
971 throw io;
972 }
973 }
974
975
976
977
978
979
980
981
982
983 public static String removeWALRootPath(Path path, final Configuration conf) throws IOException {
984 Path root = getWALRootDir(conf);
985 String pathStr = path.toString();
986
987 if (!pathStr.startsWith(root.toString())) return pathStr;
988
989 return pathStr.substring(root.toString().length() + 1);
990 }
991
992
993
994
995
996
997
998 public static void waitOnSafeMode(final Configuration conf,
999 final long wait)
1000 throws IOException {
1001 FileSystem fs = FileSystem.get(conf);
1002 if (!(fs instanceof DistributedFileSystem)) return;
1003 DistributedFileSystem dfs = (DistributedFileSystem)fs;
1004
1005 while (isInSafeMode(dfs)) {
1006 LOG.info("Waiting for dfs to exit safe mode...");
1007 try {
1008 Thread.sleep(wait);
1009 } catch (InterruptedException e) {
1010 throw (InterruptedIOException)new InterruptedIOException().initCause(e);
1011 }
1012 }
1013 }
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025 public static String getPath(Path p) {
1026 return p.toUri().getPath();
1027 }
1028
1029
1030
1031
1032
1033
1034
1035 public static Path getRootDir(final Configuration c) throws IOException {
1036 Path p = new Path(c.get(HBASE_DIR));
1037 FileSystem fs = p.getFileSystem(c);
1038 return p.makeQualified(fs);
1039 }
1040
1041 public static void setRootDir(final Configuration c, final Path root) throws IOException {
1042 c.set(HBASE_DIR, root.toString());
1043 }
1044
1045 public static void setFsDefault(final Configuration c, final Path root) throws IOException {
1046 c.set("fs.defaultFS", root.toString());
1047 }
1048
1049 public static FileSystem getRootDirFileSystem(final Configuration c) throws IOException {
1050 Path p = getRootDir(c);
1051 return p.getFileSystem(c);
1052 }
1053
1054
1055
1056
1057
1058
1059
1060 public static Path getWALRootDir(final Configuration c) throws IOException {
1061 Path p = new Path(c.get(HFileSystem.HBASE_WAL_DIR, c.get(HBASE_DIR)));
1062 if (!isValidWALRootDir(p, c)) {
1063 return FSUtils.getRootDir(c);
1064 }
1065 FileSystem fs = p.getFileSystem(c);
1066 return p.makeQualified(fs);
1067 }
1068
1069 @VisibleForTesting
1070 public static void setWALRootDir(final Configuration c, final Path root) throws IOException {
1071 c.set(HFileSystem.HBASE_WAL_DIR, root.toString());
1072 }
1073
1074 public static FileSystem getWALFileSystem(final Configuration c) throws IOException {
1075 Path p = getWALRootDir(c);
1076 return p.getFileSystem(c);
1077 }
1078
1079 private static boolean isValidWALRootDir(Path walDir, final Configuration c) throws IOException {
1080 Path rootDir = FSUtils.getRootDir(c);
1081 if (!walDir.equals(rootDir)) {
1082 if (walDir.toString().startsWith(rootDir.toString() + "/")) {
1083 throw new IllegalStateException("Illegal WAL directory specified. " +
1084 "WAL directories are not permitted to be under the root directory if set.");
1085 }
1086 }
1087 return true;
1088 }
1089
1090
1091
1092
1093
1094
1095
1096
1097 public static Path getWALRegionDir(final Configuration conf,
1098 final HRegionInfo regionInfo)
1099 throws IOException {
1100 return new Path(getWALTableDir(conf, regionInfo.getTable()),
1101 regionInfo.getEncodedName());
1102 }
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112 @SuppressWarnings("deprecation")
1113 public static boolean metaRegionExists(FileSystem fs, Path rootdir)
1114 throws IOException {
1115 Path metaRegionDir =
1116 HRegion.getRegionDir(rootdir, HRegionInfo.FIRST_META_REGIONINFO);
1117 return fs.exists(metaRegionDir);
1118 }
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128 static public HDFSBlocksDistribution computeHDFSBlocksDistribution(
1129 final FileSystem fs, FileStatus status, long start, long length)
1130 throws IOException {
1131 HDFSBlocksDistribution blocksDistribution = new HDFSBlocksDistribution();
1132 BlockLocation [] blockLocations =
1133 fs.getFileBlockLocations(status, start, length);
1134 for(BlockLocation bl : blockLocations) {
1135 String [] hosts = bl.getHosts();
1136 long len = bl.getLength();
1137 blocksDistribution.addHostsAndBlockWeight(hosts, len);
1138 }
1139
1140 return blocksDistribution;
1141 }
1142
1143
1144
1145
1146
1147
1148 static public void addToHDFSBlocksDistribution(
1149 HDFSBlocksDistribution blocksDistribution, BlockLocation[] blockLocations)
1150 throws IOException {
1151 for (BlockLocation bl : blockLocations) {
1152 String[] hosts = bl.getHosts();
1153 long len = bl.getLength();
1154 blocksDistribution.addHostsAndBlockWeight(hosts, len);
1155 }
1156 }
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167 public static int getTotalTableFragmentation(final HMaster master)
1168 throws IOException {
1169 Map<String, Integer> map = getTableFragmentation(master);
1170 return map != null && map.size() > 0 ? map.get("-TOTAL-") : -1;
1171 }
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183 public static Map<String, Integer> getTableFragmentation(
1184 final HMaster master)
1185 throws IOException {
1186 Path path = getRootDir(master.getConfiguration());
1187
1188 FileSystem fs = path.getFileSystem(master.getConfiguration());
1189 return getTableFragmentation(fs, path);
1190 }
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202 public static Map<String, Integer> getTableFragmentation(
1203 final FileSystem fs, final Path hbaseRootDir)
1204 throws IOException {
1205 Map<String, Integer> frags = new HashMap<String, Integer>();
1206 int cfCountTotal = 0;
1207 int cfFragTotal = 0;
1208 PathFilter regionFilter = new RegionDirFilter(fs);
1209 PathFilter familyFilter = new FamilyDirFilter(fs);
1210 List<Path> tableDirs = getTableDirs(fs, hbaseRootDir);
1211 for (Path d : tableDirs) {
1212 int cfCount = 0;
1213 int cfFrag = 0;
1214 FileStatus[] regionDirs = fs.listStatus(d, regionFilter);
1215 for (FileStatus regionDir : regionDirs) {
1216 Path dd = regionDir.getPath();
1217
1218 FileStatus[] familyDirs = fs.listStatus(dd, familyFilter);
1219 for (FileStatus familyDir : familyDirs) {
1220 cfCount++;
1221 cfCountTotal++;
1222 Path family = familyDir.getPath();
1223
1224 FileStatus[] familyStatus = fs.listStatus(family);
1225 if (familyStatus.length > 1) {
1226 cfFrag++;
1227 cfFragTotal++;
1228 }
1229 }
1230 }
1231
1232 frags.put(FSUtils.getTableName(d).getNameAsString(),
1233 cfCount == 0? 0: Math.round((float) cfFrag / cfCount * 100));
1234 }
1235
1236 frags.put("-TOTAL-",
1237 cfCountTotal == 0? 0: Math.round((float) cfFragTotal / cfCountTotal * 100));
1238 return frags;
1239 }
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249 public static Path getTableDir(Path rootdir, final TableName tableName) {
1250 return new Path(getNamespaceDir(rootdir, tableName.getNamespaceAsString()),
1251 tableName.getQualifierAsString());
1252 }
1253
1254
1255
1256
1257
1258
1259
1260
1261 public static Path getWALTableDir(final Configuration conf, final TableName tableName)
1262 throws IOException {
1263 return new Path(new Path(getWALRootDir(conf), tableName.getNamespaceAsString()),
1264 tableName.getQualifierAsString());
1265 }
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275 public static TableName getTableName(Path tablePath) {
1276 return TableName.valueOf(tablePath.getParent().getName(), tablePath.getName());
1277 }
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287 public static Path getNamespaceDir(Path rootdir, final String namespace) {
1288 return new Path(rootdir, new Path(HConstants.BASE_NAMESPACE_DIR,
1289 new Path(namespace)));
1290 }
1291
1292
1293
1294
1295 static class FileFilter extends AbstractFileStatusFilter {
1296 private final FileSystem fs;
1297
1298 public FileFilter(final FileSystem fs) {
1299 this.fs = fs;
1300 }
1301
1302 @Override
1303 protected boolean accept(Path p, @CheckForNull Boolean isDir) {
1304 try {
1305 return isFile(fs, isDir, p);
1306 } catch (IOException e) {
1307 LOG.warn("unable to verify if path=" + p + " is a regular file", e);
1308 return false;
1309 }
1310 }
1311 }
1312
1313
1314
1315
1316 public static class BlackListDirFilter extends AbstractFileStatusFilter {
1317 private final FileSystem fs;
1318 private List<String> blacklist;
1319
1320
1321
1322
1323
1324
1325
1326 @SuppressWarnings("unchecked")
1327 public BlackListDirFilter(final FileSystem fs, final List<String> directoryNameBlackList) {
1328 this.fs = fs;
1329 blacklist =
1330 (List<String>) (directoryNameBlackList == null ? Collections.emptyList()
1331 : directoryNameBlackList);
1332 }
1333
1334 @Override
1335 protected boolean accept(Path p, @CheckForNull Boolean isDir) {
1336 if (!isValidName(p.getName())) {
1337 return false;
1338 }
1339
1340 try {
1341 return isDirectory(fs, isDir, p);
1342 } catch (IOException e) {
1343 LOG.warn("An error occurred while verifying if [" + p.toString()
1344 + "] is a valid directory. Returning 'not valid' and continuing.", e);
1345 return false;
1346 }
1347 }
1348
1349 protected boolean isValidName(final String name) {
1350 return !blacklist.contains(name);
1351 }
1352 }
1353
1354
1355
1356
1357 public static class DirFilter extends BlackListDirFilter {
1358
1359 public DirFilter(FileSystem fs) {
1360 super(fs, null);
1361 }
1362 }
1363
1364
1365
1366
1367
1368 public static class UserTableDirFilter extends BlackListDirFilter {
1369 public UserTableDirFilter(FileSystem fs) {
1370 super(fs, HConstants.HBASE_NON_TABLE_DIRS);
1371 }
1372
1373 @Override
1374 protected boolean isValidName(final String name) {
1375 if (!super.isValidName(name))
1376 return false;
1377
1378 try {
1379 TableName.isLegalTableQualifierName(Bytes.toBytes(name));
1380 } catch (IllegalArgumentException e) {
1381 LOG.info("INVALID NAME " + name);
1382 return false;
1383 }
1384 return true;
1385 }
1386 }
1387
1388
1389
1390
1391
1392
1393
1394
1395 public static boolean isAppendSupported(final Configuration conf) {
1396 boolean append = conf.getBoolean("dfs.support.append", false);
1397 if (append) {
1398 try {
1399
1400
1401
1402 SequenceFile.Writer.class.getMethod("syncFs", new Class<?> []{});
1403 append = true;
1404 } catch (SecurityException e) {
1405 } catch (NoSuchMethodException e) {
1406 append = false;
1407 }
1408 }
1409 if (!append) {
1410
1411 try {
1412 FSDataOutputStream.class.getMethod("hflush", new Class<?> []{});
1413 append = true;
1414 } catch (NoSuchMethodException e) {
1415 append = false;
1416 }
1417 }
1418 return append;
1419 }
1420
1421
1422
1423
1424
1425
1426 public static boolean isHDFS(final Configuration conf) throws IOException {
1427 FileSystem fs = FileSystem.get(conf);
1428 String scheme = fs.getUri().getScheme();
1429 return scheme.equalsIgnoreCase("hdfs");
1430 }
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440 public abstract void recoverFileLease(final FileSystem fs, final Path p,
1441 Configuration conf, CancelableProgressable reporter) throws IOException;
1442
1443 public static List<Path> getTableDirs(final FileSystem fs, final Path rootdir)
1444 throws IOException {
1445 List<Path> tableDirs = new LinkedList<Path>();
1446
1447 for(FileStatus status :
1448 fs.globStatus(new Path(rootdir,
1449 new Path(HConstants.BASE_NAMESPACE_DIR, "*")))) {
1450 tableDirs.addAll(FSUtils.getLocalTableDirs(fs, status.getPath()));
1451 }
1452 return tableDirs;
1453 }
1454
1455
1456
1457
1458
1459
1460
1461
1462 public static List<Path> getLocalTableDirs(final FileSystem fs, final Path rootdir)
1463 throws IOException {
1464
1465 FileStatus[] dirs = fs.listStatus(rootdir, new UserTableDirFilter(fs));
1466 List<Path> tabledirs = new ArrayList<Path>(dirs.length);
1467 for (FileStatus dir: dirs) {
1468 tabledirs.add(dir.getPath());
1469 }
1470 return tabledirs;
1471 }
1472
1473
1474
1475
1476
1477
1478 public static boolean isRecoveredEdits(Path path) {
1479 return path.toString().contains(HConstants.RECOVERED_EDITS_DIR);
1480 }
1481
1482
1483
1484
1485 public static class RegionDirFilter extends AbstractFileStatusFilter {
1486
1487 final public static Pattern regionDirPattern = Pattern.compile("^[0-9a-f]*$");
1488 final FileSystem fs;
1489
1490 public RegionDirFilter(FileSystem fs) {
1491 this.fs = fs;
1492 }
1493
1494 @Override
1495 protected boolean accept(Path p, @CheckForNull Boolean isDir) {
1496 if (!regionDirPattern.matcher(p.getName()).matches()) {
1497 return false;
1498 }
1499
1500 try {
1501 return isDirectory(fs, isDir, p);
1502 } catch (IOException ioe) {
1503
1504 LOG.warn("Skipping file " + p +" due to IOException", ioe);
1505 return false;
1506 }
1507 }
1508 }
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518 public static List<Path> getRegionDirs(final FileSystem fs, final Path tableDir) throws IOException {
1519
1520 List<FileStatus> rds = listStatusWithStatusFilter(fs, tableDir, new RegionDirFilter(fs));
1521 if (rds == null) {
1522 return new ArrayList<Path>();
1523 }
1524 List<Path> regionDirs = new ArrayList<Path>(rds.size());
1525 for (FileStatus rdfs: rds) {
1526 Path rdPath = rdfs.getPath();
1527 regionDirs.add(rdPath);
1528 }
1529 return regionDirs;
1530 }
1531
1532
1533
1534
1535
1536 public static class FamilyDirFilter extends AbstractFileStatusFilter {
1537 final FileSystem fs;
1538
1539 public FamilyDirFilter(FileSystem fs) {
1540 this.fs = fs;
1541 }
1542
1543 @Override
1544 protected boolean accept(Path p, @CheckForNull Boolean isDir) {
1545 try {
1546
1547 HColumnDescriptor.isLegalFamilyName(Bytes.toBytes(p.getName()));
1548 } catch (IllegalArgumentException iae) {
1549
1550 return false;
1551 }
1552
1553 try {
1554 return isDirectory(fs, isDir, p);
1555 } catch (IOException ioe) {
1556
1557 LOG.warn("Skipping file " + p +" due to IOException", ioe);
1558 return false;
1559 }
1560 }
1561 }
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571 public static List<Path> getFamilyDirs(final FileSystem fs, final Path regionDir) throws IOException {
1572
1573 FileStatus[] fds = fs.listStatus(regionDir, new FamilyDirFilter(fs));
1574 List<Path> familyDirs = new ArrayList<Path>(fds.length);
1575 for (FileStatus fdfs: fds) {
1576 Path fdPath = fdfs.getPath();
1577 familyDirs.add(fdPath);
1578 }
1579 return familyDirs;
1580 }
1581
1582 public static List<Path> getReferenceFilePaths(final FileSystem fs, final Path familyDir) throws IOException {
1583 List<FileStatus> fds = listStatusWithStatusFilter(fs, familyDir, new ReferenceFileFilter(fs));
1584 if (fds == null) {
1585 return new ArrayList<Path>();
1586 }
1587 List<Path> referenceFiles = new ArrayList<Path>(fds.size());
1588 for (FileStatus fdfs: fds) {
1589 Path fdPath = fdfs.getPath();
1590 referenceFiles.add(fdPath);
1591 }
1592 return referenceFiles;
1593 }
1594
1595
1596
1597
1598 public static class HFileFilter extends AbstractFileStatusFilter {
1599 final FileSystem fs;
1600
1601 public HFileFilter(FileSystem fs) {
1602 this.fs = fs;
1603 }
1604
1605 @Override
1606 protected boolean accept(Path p, @CheckForNull Boolean isDir) {
1607 if (!StoreFileInfo.isHFile(p)) {
1608 return false;
1609 }
1610
1611 try {
1612 return isFile(fs, isDir, p);
1613 } catch (IOException ioe) {
1614
1615 LOG.warn("Skipping file " + p +" due to IOException", ioe);
1616 return false;
1617 }
1618 }
1619 }
1620
1621
1622
1623
1624
1625 public static class HFileLinkFilter implements PathFilter {
1626
1627 @Override
1628 public boolean accept(Path p) {
1629 return HFileLink.isHFileLink(p);
1630 }
1631 }
1632
1633 public static class ReferenceFileFilter extends AbstractFileStatusFilter {
1634
1635 private final FileSystem fs;
1636
1637 public ReferenceFileFilter(FileSystem fs) {
1638 this.fs = fs;
1639 }
1640
1641 @Override
1642 protected boolean accept(Path p, @CheckForNull Boolean isDir) {
1643 if (!StoreFileInfo.isReference(p)) {
1644 return false;
1645 }
1646
1647 try {
1648
1649 return isFile(fs, isDir, p);
1650 } catch (IOException ioe) {
1651
1652 LOG.warn("Skipping file " + p +" due to IOException", ioe);
1653 return false;
1654 }
1655 }
1656 }
1657
1658
1659
1660
1661
1662
1663
1664 public static FileSystem getCurrentFileSystem(Configuration conf)
1665 throws IOException {
1666 return getRootDir(conf).getFileSystem(conf);
1667 }
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686 public static Map<String, Path> getTableStoreFilePathMap(Map<String, Path> map,
1687 final FileSystem fs, final Path hbaseRootDir, TableName tableName)
1688 throws IOException, InterruptedException {
1689 return getTableStoreFilePathMap(map, fs, hbaseRootDir, tableName, null, null, null);
1690 }
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714 public static Map<String, Path> getTableStoreFilePathMap(
1715 Map<String, Path> resultMap,
1716 final FileSystem fs, final Path hbaseRootDir, TableName tableName, final PathFilter sfFilter,
1717 ExecutorService executor, final ErrorReporter errors) throws IOException, InterruptedException {
1718
1719 final Map<String, Path> finalResultMap =
1720 resultMap == null ? new ConcurrentHashMap<String, Path>(128, 0.75f, 32) : resultMap;
1721
1722
1723 Path tableDir = FSUtils.getTableDir(hbaseRootDir, tableName);
1724
1725
1726 final FamilyDirFilter familyFilter = new FamilyDirFilter(fs);
1727 final Vector<Exception> exceptions = new Vector<Exception>();
1728
1729 try {
1730 List<FileStatus> regionDirs = FSUtils.listStatusWithStatusFilter(fs, tableDir, new RegionDirFilter(fs));
1731 if (regionDirs == null) {
1732 return finalResultMap;
1733 }
1734
1735 final List<Future<?>> futures = new ArrayList<Future<?>>(regionDirs.size());
1736
1737 for (FileStatus regionDir : regionDirs) {
1738 if (null != errors) {
1739 errors.progress();
1740 }
1741 final Path dd = regionDir.getPath();
1742
1743 if (!exceptions.isEmpty()) {
1744 break;
1745 }
1746
1747 Runnable getRegionStoreFileMapCall = new Runnable() {
1748 @Override
1749 public void run() {
1750 try {
1751 HashMap<String,Path> regionStoreFileMap = new HashMap<String, Path>();
1752 List<FileStatus> familyDirs = FSUtils.listStatusWithStatusFilter(fs, dd, familyFilter);
1753 if (familyDirs == null) {
1754 if (!fs.exists(dd)) {
1755 LOG.warn("Skipping region because it no longer exists: " + dd);
1756 } else {
1757 LOG.warn("Skipping region because it has no family dirs: " + dd);
1758 }
1759 return;
1760 }
1761 for (FileStatus familyDir : familyDirs) {
1762 if (null != errors) {
1763 errors.progress();
1764 }
1765 Path family = familyDir.getPath();
1766 if (family.getName().equals(HConstants.RECOVERED_EDITS_DIR)) {
1767 continue;
1768 }
1769
1770
1771 FileStatus[] familyStatus = fs.listStatus(family);
1772 for (FileStatus sfStatus : familyStatus) {
1773 if (null != errors) {
1774 errors.progress();
1775 }
1776 Path sf = sfStatus.getPath();
1777 if (sfFilter == null || sfFilter.accept(sf)) {
1778 regionStoreFileMap.put( sf.getName(), sf);
1779 }
1780 }
1781 }
1782 finalResultMap.putAll(regionStoreFileMap);
1783 } catch (Exception e) {
1784 LOG.error("Could not get region store file map for region: " + dd, e);
1785 exceptions.add(e);
1786 }
1787 }
1788 };
1789
1790
1791
1792 if (executor != null) {
1793 Future<?> future = executor.submit(getRegionStoreFileMapCall);
1794 futures.add(future);
1795 } else {
1796 FutureTask<?> future = new FutureTask<Object>(getRegionStoreFileMapCall, null);
1797 future.run();
1798 futures.add(future);
1799 }
1800 }
1801
1802
1803 for (Future<?> f : futures) {
1804 if (!exceptions.isEmpty()) {
1805 break;
1806 }
1807 try {
1808 f.get();
1809 } catch (ExecutionException e) {
1810 LOG.error("Unexpected exec exception! Should've been caught already. (Bug?)", e);
1811
1812 }
1813 }
1814 } catch (IOException e) {
1815 LOG.error("Cannot execute getTableStoreFilePathMap for " + tableName, e);
1816 exceptions.add(e);
1817 } finally {
1818 if (!exceptions.isEmpty()) {
1819
1820
1821 Throwables.propagateIfInstanceOf(exceptions.firstElement(), IOException.class);
1822 throw Throwables.propagate(exceptions.firstElement());
1823 }
1824 }
1825
1826 return finalResultMap;
1827 }
1828
1829 public static int getRegionReferenceFileCount(final FileSystem fs, final Path p) {
1830 int result = 0;
1831 try {
1832 for (Path familyDir:getFamilyDirs(fs, p)){
1833 result += getReferenceFilePaths(fs, familyDir).size();
1834 }
1835 } catch (IOException e) {
1836 LOG.warn("Error Counting reference files.", e);
1837 }
1838 return result;
1839 }
1840
1841
1842
1843
1844
1845
1846
1847
1848
1849
1850
1851
1852
1853
1854
1855 public static Map<String, Path> getTableStoreFilePathMap(
1856 final FileSystem fs, final Path hbaseRootDir)
1857 throws IOException, InterruptedException {
1858 return getTableStoreFilePathMap(fs, hbaseRootDir, null, null, null);
1859 }
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
1873
1874
1875
1876
1877
1878 public static Map<String, Path> getTableStoreFilePathMap(
1879 final FileSystem fs, final Path hbaseRootDir, PathFilter sfFilter,
1880 ExecutorService executor, ErrorReporter errors)
1881 throws IOException, InterruptedException {
1882 ConcurrentHashMap<String, Path> map = new ConcurrentHashMap<String, Path>(1024, 0.75f, 32);
1883
1884
1885
1886
1887
1888 for (Path tableDir : FSUtils.getTableDirs(fs, hbaseRootDir)) {
1889 getTableStoreFilePathMap(map, fs, hbaseRootDir,
1890 FSUtils.getTableName(tableDir), sfFilter, executor, errors);
1891 }
1892 return map;
1893 }
1894
1895
1896
1897
1898
1899
1900
1901
1902 public static List<FileStatus> filterFileStatuses(FileStatus[] input,
1903 FileStatusFilter filter) {
1904 if (input == null) return null;
1905 return filterFileStatuses(Iterators.forArray(input), filter);
1906 }
1907
1908
1909
1910
1911
1912
1913
1914
1915 public static List<FileStatus> filterFileStatuses(Iterator<FileStatus> input,
1916 FileStatusFilter filter) {
1917 if (input == null) return null;
1918 ArrayList<FileStatus> results = new ArrayList<FileStatus>();
1919 while (input.hasNext()) {
1920 FileStatus f = input.next();
1921 if (filter.accept(f)) {
1922 results.add(f);
1923 }
1924 }
1925 return results;
1926 }
1927
1928
1929
1930
1931
1932
1933
1934
1935
1936
1937
1938
1939 public static List<FileStatus> listStatusWithStatusFilter(final FileSystem fs,
1940 final Path dir, final FileStatusFilter filter) throws IOException {
1941 FileStatus [] status = null;
1942 try {
1943 status = fs.listStatus(dir);
1944 } catch (FileNotFoundException fnfe) {
1945
1946 if (LOG.isTraceEnabled()) {
1947 LOG.trace(dir + " doesn't exist");
1948 }
1949 }
1950
1951 if (status == null || status.length < 1) {
1952 return null;
1953 }
1954
1955 if (filter == null) {
1956 return Arrays.asList(status);
1957 } else {
1958 List<FileStatus> status2 = filterFileStatuses(status, filter);
1959 if (status2 == null || status2.isEmpty()) {
1960 return null;
1961 } else {
1962 return status2;
1963 }
1964 }
1965 }
1966
1967
1968
1969
1970
1971
1972
1973
1974
1975
1976
1977
1978
1979
1980
1981 public static FileStatus [] listStatus(final FileSystem fs,
1982 final Path dir, final PathFilter filter) throws IOException {
1983 FileStatus [] status = null;
1984 try {
1985 status = filter == null ? fs.listStatus(dir) : fs.listStatus(dir, filter);
1986 } catch (FileNotFoundException fnfe) {
1987
1988 if (LOG.isTraceEnabled()) {
1989 LOG.trace(dir + " doesn't exist");
1990 }
1991 }
1992 if (status == null || status.length < 1) return null;
1993 return status;
1994 }
1995
1996
1997
1998
1999
2000
2001
2002
2003
2004 public static FileStatus[] listStatus(final FileSystem fs, final Path dir) throws IOException {
2005 return listStatus(fs, dir, null);
2006 }
2007
2008
2009
2010
2011
2012
2013
2014
2015 public static List<LocatedFileStatus> listLocatedStatus(final FileSystem fs,
2016 final Path dir) throws IOException {
2017 List<LocatedFileStatus> status = null;
2018 try {
2019 RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = fs
2020 .listFiles(dir, false);
2021 while (locatedFileStatusRemoteIterator.hasNext()) {
2022 if (status == null) {
2023 status = Lists.newArrayList();
2024 }
2025 status.add(locatedFileStatusRemoteIterator.next());
2026 }
2027 } catch (FileNotFoundException fnfe) {
2028
2029 if (LOG.isTraceEnabled()) {
2030 LOG.trace(dir + " doesn't exist");
2031 }
2032 }
2033 return status;
2034 }
2035
2036
2037
2038
2039
2040
2041
2042
2043
2044
2045 public static boolean delete(final FileSystem fs, final Path path, final boolean recursive)
2046 throws IOException {
2047 return fs.delete(path, recursive);
2048 }
2049
2050
2051
2052
2053
2054
2055
2056
2057
2058 public static boolean isExists(final FileSystem fs, final Path path) throws IOException {
2059 return fs.exists(path);
2060 }
2061
2062
2063
2064
2065
2066
2067
2068
2069
2070
2071
2072 public static void checkAccess(UserGroupInformation ugi, FileStatus file,
2073 FsAction action) throws AccessDeniedException {
2074 if (ugi.getShortUserName().equals(file.getOwner())) {
2075 if (file.getPermission().getUserAction().implies(action)) {
2076 return;
2077 }
2078 } else if (contains(ugi.getGroupNames(), file.getGroup())) {
2079 if (file.getPermission().getGroupAction().implies(action)) {
2080 return;
2081 }
2082 } else if (file.getPermission().getOtherAction().implies(action)) {
2083 return;
2084 }
2085 throw new AccessDeniedException("Permission denied:" + " action=" + action
2086 + " path=" + file.getPath() + " user=" + ugi.getShortUserName());
2087 }
2088
2089 private static boolean contains(String[] groups, String user) {
2090 for (String group : groups) {
2091 if (group.equals(user)) {
2092 return true;
2093 }
2094 }
2095 return false;
2096 }
2097
2098
2099
2100
2101
2102
2103
2104
2105 public static void logFileSystemState(final FileSystem fs, final Path root, Log LOG)
2106 throws IOException {
2107 LOG.debug("Current file system:");
2108 logFSTree(LOG, fs, root, "|-");
2109 }
2110
2111
2112
2113
2114
2115
2116 private static void logFSTree(Log LOG, final FileSystem fs, final Path root, String prefix)
2117 throws IOException {
2118 FileStatus[] files = FSUtils.listStatus(fs, root, null);
2119 if (files == null) return;
2120
2121 for (FileStatus file : files) {
2122 if (file.isDirectory()) {
2123 LOG.debug(prefix + file.getPath().getName() + "/");
2124 logFSTree(LOG, fs, file.getPath(), prefix + "---");
2125 } else {
2126 LOG.debug(prefix + file.getPath().getName());
2127 }
2128 }
2129 }
2130
2131 public static boolean renameAndSetModifyTime(final FileSystem fs, final Path src, final Path dest)
2132 throws IOException {
2133
2134 fs.setTimes(src, EnvironmentEdgeManager.currentTime(), -1);
2135 return fs.rename(src, dest);
2136 }
2137
2138
2139
2140
2141
2142
2143
2144
2145
2146
2147
2148
2149
2150
2151 public static Map<String, Map<String, Float>> getRegionDegreeLocalityMappingFromFS(
2152 final Configuration conf) throws IOException {
2153 return getRegionDegreeLocalityMappingFromFS(
2154 conf, null,
2155 conf.getInt(THREAD_POOLSIZE, DEFAULT_THREAD_POOLSIZE));
2156
2157 }
2158
2159
2160
2161
2162
2163
2164
2165
2166
2167
2168
2169
2170
2171
2172
2173
2174
2175 public static Map<String, Map<String, Float>> getRegionDegreeLocalityMappingFromFS(
2176 final Configuration conf, final String desiredTable, int threadPoolSize)
2177 throws IOException {
2178 Map<String, Map<String, Float>> regionDegreeLocalityMapping =
2179 new ConcurrentHashMap<String, Map<String, Float>>();
2180 getRegionLocalityMappingFromFS(conf, desiredTable, threadPoolSize, null,
2181 regionDegreeLocalityMapping);
2182 return regionDegreeLocalityMapping;
2183 }
2184
2185
2186
2187
2188
2189
2190
2191
2192
2193
2194
2195
2196
2197
2198
2199
2200
2201
2202
2203
2204
2205 private static void getRegionLocalityMappingFromFS(
2206 final Configuration conf, final String desiredTable,
2207 int threadPoolSize,
2208 Map<String, String> regionToBestLocalityRSMapping,
2209 Map<String, Map<String, Float>> regionDegreeLocalityMapping)
2210 throws IOException {
2211 FileSystem fs = FileSystem.get(conf);
2212 Path rootPath = FSUtils.getRootDir(conf);
2213 long startTime = EnvironmentEdgeManager.currentTime();
2214 Path queryPath;
2215
2216
2217
2218
2219
2220
2221
2222
2223
2224
2225
2226
2227
2228
2229
2230
2231
2232
2233
2234
2235
2236
2237
2238
2239
2240
2241
2242
2243
2244
2245
2246
2247
2248
2249
2250
2251
2252
2253
2254
2255
2256
2257
2258
2259
2260
2261
2262
2263
2264
2265
2266
2267
2268
2269
2270
2271
2272
2273
2274
2275
2276
2277
2278
2279
2280
2281
2282
2283
2284
2285
2286
2287
2288
2289
2290
2291
2292
2293
2294
2295
2296
2297
2298
2299
2300
2301
2302
2303
2304
2305
2306
2307
2308
2309
2310
2311
2312
2313
2314
2315 public static void setupShortCircuitRead(final Configuration conf) {
2316
2317 boolean shortCircuitSkipChecksum =
2318 conf.getBoolean("dfs.client.read.shortcircuit.skip.checksum", false);
2319 boolean useHBaseChecksum = conf.getBoolean(HConstants.HBASE_CHECKSUM_VERIFICATION, true);
2320 if (shortCircuitSkipChecksum) {
2321 LOG.warn("Configuration \"dfs.client.read.shortcircuit.skip.checksum\" should not " +
2322 "be set to true." + (useHBaseChecksum ? " HBase checksum doesn't require " +
2323 "it, see https://issues.apache.org/jira/browse/HBASE-6868." : ""));
2324 assert !shortCircuitSkipChecksum;
2325 }
2326 checkShortCircuitReadBufferSize(conf);
2327 }
2328
2329
2330
2331
2332
2333 public static void checkShortCircuitReadBufferSize(final Configuration conf) {
2334 final int defaultSize = HConstants.DEFAULT_BLOCKSIZE * 2;
2335 final int notSet = -1;
2336
2337 final String dfsKey = "dfs.client.read.shortcircuit.buffer.size";
2338 int size = conf.getInt(dfsKey, notSet);
2339
2340 if (size != notSet) return;
2341
2342 int hbaseSize = conf.getInt("hbase." + dfsKey, defaultSize);
2343 conf.setIfUnset(dfsKey, Integer.toString(hbaseSize));
2344 }
2345 }