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