You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by fa...@apache.org on 2016/11/18 20:11:51 UTC
[1/2] hadoop git commit: HADOOP-13651 S3Guard: S3AFileSystem
Integration with MetadataStore
Repository: hadoop
Updated Branches:
refs/heads/HADOOP-13345 56b715f8e -> f4c37a579
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/MetadataStoreTestBase.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/MetadataStoreTestBase.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/MetadataStoreTestBase.java
index 462f8f9..fb46357 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/MetadataStoreTestBase.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/MetadataStoreTestBase.java
@@ -28,8 +28,9 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.FileNotFoundException;
import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -40,6 +41,8 @@ import java.util.Set;
* Main test class for MetadataStore implementations.
* Implementations should each create a test by subclassing this and
* overriding {@link #createContract()}.
+ * If your implementation may return missing results for recently set paths,
+ * override {@link MetadataStoreTestBase#allowMissing()}.
*/
public abstract class MetadataStoreTestBase extends Assert {
@@ -61,6 +64,17 @@ public abstract class MetadataStoreTestBase extends Assert {
*/
public abstract AbstractMSContract createContract();
+ /**
+ * Tests assume that implementations will return recently set results. If
+ * your implementation does not always hold onto metadata (e.g. LRU or
+ * time-based expiry) you can override this to return false.
+ * @return true if the test should succeed when null results are returned
+ * from the MetadataStore under test.
+ */
+ public boolean allowMissing() {
+ return false;
+ }
+
/** The MetadataStore contract used to test against. */
private AbstractMSContract contract;
@@ -100,8 +114,10 @@ public abstract class MetadataStoreTestBase extends Assert {
assertDirectorySize("/da1/db1", 1);
/* Check contents of dir status. */
- PathMetadata dirMeta = ms.get(new Path("/da1"));
- verifyDirStatus(dirMeta);
+ PathMetadata dirMeta = ms.get(strToPath("/da1"));
+ if (!allowMissing() || dirMeta != null) {
+ verifyDirStatus(dirMeta);
+ }
/* This already exists, and should silently replace it. */
ms.put(new PathMetadata(makeDirStatus("/da1/db1")));
@@ -117,10 +133,12 @@ public abstract class MetadataStoreTestBase extends Assert {
assertDirectorySize("/da1", 1);
assertDirectorySize("/da1/db1", 2);
assertEmptyDirs("/da2", "/da3");
- PathMetadata meta = ms.get(new Path("/da1/db1/fc2"));
- assertNotNull("Get file after put new.", meta);
- assertEquals("Cached file size correct.", 200,
- meta.getFileStatus().getLen());
+ PathMetadata meta = ms.get(strToPath("/da1/db1/fc2"));
+ if (!allowMissing() || meta != null) {
+ assertNotNull("Get file after put new.", meta);
+ assertEquals("Cached file size correct.", 200,
+ meta.getFileStatus().getLen());
+ }
}
@Test
@@ -129,53 +147,74 @@ public abstract class MetadataStoreTestBase extends Assert {
final String dirPath = "/a1/b1/c1/d1";
ms.put(new PathMetadata(makeFileStatus(filePath, 100)));
ms.put(new PathMetadata(makeDirStatus(dirPath)));
- PathMetadata meta = ms.get(new Path(filePath));
- verifyBasicFileStatus(meta);
+ PathMetadata meta = ms.get(strToPath(filePath));
+ if (!allowMissing() || meta != null) {
+ verifyBasicFileStatus(meta);
+ }
- ms.put(new PathMetadata(basicFileStatus(new Path(filePath), 9999, false)));
- meta = ms.get(new Path(filePath));
- assertEquals("Updated size", 9999, meta.getFileStatus().getLen());
+ ms.put(new PathMetadata(basicFileStatus(strToPath(filePath), 9999, false)));
+ meta = ms.get(strToPath(filePath));
+ if (!allowMissing() || meta != null) {
+ assertEquals("Updated size", 9999, meta.getFileStatus().getLen());
+ }
}
@Test
public void testRootDirPutNew() throws Exception {
- Path rootPath = new Path("/");
+ Path rootPath = strToPath("/");
ms.put(new PathMetadata(makeFileStatus("/file1", 100)));
DirListingMetadata dir = ms.listChildren(rootPath);
- assertNotNull("Root dir cached", dir);
- assertFalse("Root not fully cached", dir.isAuthoritative());
- assertNotNull("have root dir file listing", dir.getListing());
- assertEquals("One file in root dir", 1, dir.getListing().size());
- assertEquals("file1 in root dir", "/file1", dir.getListing().iterator()
- .next().getFileStatus().getPath().toString());
+ if (!allowMissing() || dir != null) {
+ assertNotNull("Root dir cached", dir);
+ assertFalse("Root not fully cached", dir.isAuthoritative());
+ assertNotNull("have root dir file listing", dir.getListing());
+ assertEquals("One file in root dir", 1, dir.getListing().size());
+ assertEquals("file1 in root dir", strToPath("/file1"),
+ dir.getListing().iterator().next().getFileStatus().getPath());
+ }
}
@Test
public void testDelete() throws Exception {
setUpDeleteTest();
- ms.delete(new Path("/ADirectory1/db1/file2"));
+ ms.delete(strToPath("/ADirectory1/db1/file2"));
/* Ensure delete happened. */
assertDirectorySize("/ADirectory1/db1", 1);
- PathMetadata meta = ms.get(new Path("/ADirectory1/db1/file2"));
+ PathMetadata meta = ms.get(strToPath("/ADirectory1/db1/file2"));
assertNull("File deleted", meta);
}
@Test
public void testDeleteSubtree() throws Exception {
- setUpDeleteTest();
- createNewDirs("/ADirectory1/db1/dc1", "/ADirectory1/db1/dc1/dd1");
+ deleteSubtreeHelper("");
+ }
+
+ @Test
+ public void testDeleteSubtreeHostPath() throws Exception {
+ deleteSubtreeHelper("s3a://test-bucket-name");
+ }
+
+ private void deleteSubtreeHelper(String pathPrefix) throws Exception {
+
+ String p = pathPrefix;
+ setUpDeleteTest(p);
+ createNewDirs(p + "/ADirectory1/db1/dc1", p + "/ADirectory1/db1/dc1/dd1");
ms.put(new PathMetadata(
- makeFileStatus("/ADirectory1/db1/dc1/dd1/deepFile", 100)));
- ms.deleteSubtree(new Path("/ADirectory1/db1"));
+ makeFileStatus(p + "/ADirectory1/db1/dc1/dd1/deepFile", 100)));
+ if (!allowMissing()) {
+ assertCached(p + "/ADirectory1/db1");
+ }
+ ms.deleteSubtree(strToPath(p + "/ADirectory1/db1/"));
- assertEmptyDirectory("/ADirectory1");
- assertNotCached("/ADirectory1/file1");
- assertNotCached("/ADirectory1/file2");
- assertNotCached("/ADirectory1/db1/dc1/dd1/deepFile");
- assertEmptyDirectory("/ADirectory2");
+ assertEmptyDirectory(p + "/ADirectory1");
+ assertNotCached(p + "/ADirectory1/db1");
+ assertNotCached(p + "/ADirectory1/file1");
+ assertNotCached(p + "/ADirectory1/file2");
+ assertNotCached(p + "/ADirectory1/db1/dc1/dd1/deepFile");
+ assertEmptyDirectory(p + "/ADirectory2");
}
@@ -188,7 +227,7 @@ public abstract class MetadataStoreTestBase extends Assert {
public void testDeleteRecursiveRoot() throws Exception {
setUpDeleteTest();
- ms.deleteSubtree(new Path("/"));
+ ms.deleteSubtree(strToPath("/"));
assertNotCached("/ADirectory1");
assertNotCached("/ADirectory2");
assertNotCached("/ADirectory2/db1");
@@ -199,22 +238,30 @@ public abstract class MetadataStoreTestBase extends Assert {
@Test
public void testDeleteNonExisting() throws Exception {
// Path doesn't exist, but should silently succeed
- ms.delete(new Path("/bobs/your/uncle"));
+ ms.delete(strToPath("/bobs/your/uncle"));
// Ditto.
- ms.deleteSubtree(new Path("/internets"));
+ ms.deleteSubtree(strToPath("/internets"));
}
private void setUpDeleteTest() throws IOException {
- createNewDirs("/ADirectory1", "/ADirectory2", "/ADirectory1/db1");
- ms.put(new PathMetadata(makeFileStatus("/ADirectory1/db1/file1", 100)));
- ms.put(new PathMetadata(makeFileStatus("/ADirectory1/db1/file2", 100)));
+ setUpDeleteTest("");
+ }
- PathMetadata meta = ms.get(new Path("/ADirectory1/db1/file2"));
- assertNotNull("Found test file", meta);
+ private void setUpDeleteTest(String prefix) throws IOException {
+ createNewDirs(prefix + "/ADirectory1", prefix + "/ADirectory2",
+ prefix + "/ADirectory1/db1");
+ ms.put(new PathMetadata(makeFileStatus(prefix + "/ADirectory1/db1/file1",
+ 100)));
+ ms.put(new PathMetadata(makeFileStatus(prefix + "/ADirectory1/db1/file2",
+ 100)));
- assertDirectorySize("/ADirectory1/db1", 2);
+ PathMetadata meta = ms.get(strToPath(prefix + "/ADirectory1/db1/file2"));
+ if (!allowMissing() || meta != null) {
+ assertNotNull("Found test file", meta);
+ assertDirectorySize(prefix + "/ADirectory1/db1", 2);
+ }
}
@Test
@@ -223,15 +270,19 @@ public abstract class MetadataStoreTestBase extends Assert {
final String dirPath = "/a1/b1/c1/d1";
ms.put(new PathMetadata(makeFileStatus(filePath, 100)));
ms.put(new PathMetadata(makeDirStatus(dirPath)));
- PathMetadata meta = ms.get(new Path(filePath));
- assertNotNull("Get found file", meta);
- verifyBasicFileStatus(meta);
+ PathMetadata meta = ms.get(strToPath(filePath));
+ if (!allowMissing() || meta != null) {
+ assertNotNull("Get found file", meta);
+ verifyBasicFileStatus(meta);
+ }
- meta = ms.get(new Path(dirPath));
- assertNotNull("Get found file (dir)", meta);
- assertTrue("Found dir", meta.getFileStatus().isDirectory());
+ meta = ms.get(strToPath(dirPath));
+ if (!allowMissing() || meta != null) {
+ assertNotNull("Get found file (dir)", meta);
+ assertTrue("Found dir", meta.getFileStatus().isDirectory());
+ }
- meta = ms.get(new Path("bollocks"));
+ meta = ms.get(strToPath("/bollocks"));
assertNull("Don't get non-existent file", meta);
}
@@ -241,20 +292,21 @@ public abstract class MetadataStoreTestBase extends Assert {
setupListStatus();
DirListingMetadata dirMeta;
- try {
- dirMeta = ms.listChildren(new Path("/"));
- if (dirMeta != null) {
- /* Cache has no way of knowing it has all entries for root unless we
- * specifically tell it via put() with
- * DirListingMetadata.isAuthoritative = true */
- assertFalse("Root dir is not cached, or partially cached",
- dirMeta.isAuthoritative());
- assertListingsEqual(dirMeta.getListing(), "/a1", "/a2");
- }
- } catch (FileNotFoundException f) { }
+ dirMeta = ms.listChildren(strToPath("/"));
+ if (!allowMissing()) {
+ assertNotNull(dirMeta);
+ /* Cache has no way of knowing it has all entries for root unless we
+ * specifically tell it via put() with
+ * DirListingMetadata.isAuthoritative = true */
+ assertFalse("Root dir is not cached, or partially cached",
+ dirMeta.isAuthoritative());
+ assertListingsEqual(dirMeta.getListing(), "/a1", "/a2");
+ }
- dirMeta = ms.listChildren(new Path("/a1"));
- assertListingsEqual(dirMeta.getListing(), "/a1/b1", "/a1/b2");
+ dirMeta = ms.listChildren(strToPath("/a1"));
+ if (!allowMissing() || dirMeta != null) {
+ assertListingsEqual(dirMeta.getListing(), "/a1/b1", "/a1/b2");
+ }
// TODO
// 1. Add properties query to MetadataStore interface
@@ -263,9 +315,11 @@ public abstract class MetadataStoreTestBase extends Assert {
// 3. If #1 is true, assert that directory is still fully cached here.
// assertTrue("Created dir is fully cached", dirMeta.isAuthoritative());
- dirMeta = ms.listChildren(new Path("/a1/b1"));
- assertListingsEqual(dirMeta.getListing(), "/a1/b1/file1", "/a1/b1/file2",
- "/a1/b1/c1");
+ dirMeta = ms.listChildren(strToPath("/a1/b1"));
+ if (!allowMissing() || dirMeta != null) {
+ assertListingsEqual(dirMeta.getListing(), "/a1/b1/file1", "/a1/b1/file2",
+ "/a1/b1/c1");
+ }
}
@Test
@@ -282,7 +336,7 @@ public abstract class MetadataStoreTestBase extends Assert {
public void testInvalidListChildren() throws Exception {
setupListStatus();
assertNull("missing path returns null",
- ms.listChildren(new Path("/a1/b1x")));
+ ms.listChildren(strToPath("/a1/b1x")));
}
@Test
@@ -293,20 +347,25 @@ public abstract class MetadataStoreTestBase extends Assert {
putListStatusFiles("/a1/b1", false, "/a1/b1/file1", "/a1/b1/file2");
// Assert root listing as expected
- DirListingMetadata dirMeta = ms.listChildren(new Path("/"));
- assertNotNull("Listing root", dirMeta);
- Collection<PathMetadata> entries = dirMeta.getListing();
- assertListingsEqual(entries, "/a1", "/a2", "/a3");
+ Collection<PathMetadata> entries;
+ DirListingMetadata dirMeta = ms.listChildren(strToPath("/"));
+ if (!allowMissing() || dirMeta != null) {
+ assertNotNull("Listing root", dirMeta);
+ entries = dirMeta.getListing();
+ assertListingsEqual(entries, "/a1", "/a2", "/a3");
+ }
// Assert src listing as expected
- dirMeta = ms.listChildren(new Path("/a1/b1"));
- assertNotNull("Listing /a1/b1", dirMeta);
- entries = dirMeta.getListing();
- assertListingsEqual(entries, "/a1/b1/file1", "/a1/b1/file2");
+ dirMeta = ms.listChildren(strToPath("/a1/b1"));
+ if (!allowMissing() || dirMeta != null) {
+ assertNotNull("Listing /a1/b1", dirMeta);
+ entries = dirMeta.getListing();
+ assertListingsEqual(entries, "/a1/b1/file1", "/a1/b1/file2");
+ }
// Do the move(): rename(/a1/b1, /b1)
- Collection<Path> srcPaths = Arrays.asList(new Path("/a1/b1"),
- new Path("/a1/b1/file1"), new Path("/a1/b1/file2"));
+ Collection<Path> srcPaths = Arrays.asList(strToPath("/a1/b1"),
+ strToPath("/a1/b1/file1"), strToPath("/a1/b1/file2"));
ArrayList<PathMetadata> destMetas = new ArrayList<>();
destMetas.add(new PathMetadata(makeDirStatus("/b1")));
@@ -315,24 +374,30 @@ public abstract class MetadataStoreTestBase extends Assert {
ms.move(srcPaths, destMetas);
// Assert src is no longer there
- dirMeta = ms.listChildren(new Path("/a1"));
- assertNotNull("Listing /a1", dirMeta);
- entries = dirMeta.getListing();
- assertListingsEqual(entries, "/a1/b2");
+ dirMeta = ms.listChildren(strToPath("/a1"));
+ if (!allowMissing() || dirMeta != null) {
+ assertNotNull("Listing /a1", dirMeta);
+ entries = dirMeta.getListing();
+ assertListingsEqual(entries, "/a1/b2");
+ }
- PathMetadata meta = ms.get(new Path("/a1/b1/file1"));
+ PathMetadata meta = ms.get(strToPath("/a1/b1/file1"));
// TODO allow return of PathMetadata with isDeleted == true
assertNull("Src path deleted", meta);
// Assert dest looks right
- meta = ms.get(new Path("/b1/file1"));
- assertNotNull("dest file not null", meta);
- verifyBasicFileStatus(meta);
+ meta = ms.get(strToPath("/b1/file1"));
+ if (!allowMissing() || meta != null) {
+ assertNotNull("dest file not null", meta);
+ verifyBasicFileStatus(meta);
+ }
- dirMeta = ms.listChildren(new Path("/b1"));
- assertNotNull("dest listing not null", dirMeta);
- entries = dirMeta.getListing();
- assertListingsEqual(entries, "/b1/file1", "/b1/file2");
+ dirMeta = ms.listChildren(strToPath("/b1"));
+ if (!allowMissing() || dirMeta != null) {
+ assertNotNull("dest listing not null", dirMeta);
+ entries = dirMeta.getListing();
+ assertListingsEqual(entries, "/b1/file1", "/b1/file2");
+ }
}
@@ -345,7 +410,7 @@ public abstract class MetadataStoreTestBase extends Assert {
Path[] output = new Path[paths.length];
for (int i = 0; i < paths.length; i++) {
- output[i] = new Path(parent, paths[i]);
+ output[i] = new Path(strToPath(parent), paths[i]);
}
return output;
}
@@ -353,7 +418,7 @@ public abstract class MetadataStoreTestBase extends Assert {
/** Modifies paths input array and returns it. */
private String[] buildPathStrings(String parent, String... paths) {
for (int i = 0; i < paths.length; i++) {
- Path p = new Path(parent, paths[i]);
+ Path p = new Path(strToPath(parent), paths[i]);
paths[i] = p.toString();
}
return paths;
@@ -362,12 +427,14 @@ public abstract class MetadataStoreTestBase extends Assert {
private void commonTestPutListStatus(final String parent) throws IOException {
putListStatusFiles(parent, true, buildPathStrings(parent, "file1", "file2",
"file3"));
- DirListingMetadata dirMeta = ms.listChildren(new Path(parent));
- assertNotNull("list after putListStatus", dirMeta);
- Collection<PathMetadata> entries = dirMeta.getListing();
- assertNotNull("listStatus has entries", entries);
- assertListingsEqual(entries, buildPathStrings(parent, "file1", "file2",
- "file3"));
+ DirListingMetadata dirMeta = ms.listChildren(strToPath(parent));
+ if (!allowMissing() || dirMeta != null) {
+ assertNotNull("list after putListStatus", dirMeta);
+ Collection<PathMetadata> entries = dirMeta.getListing();
+ assertNotNull("listStatus has entries", entries);
+ assertListingsEqual(entries,
+ buildPathStrings(parent, "file1", "file2", "file3"));
+ }
}
private void setupListStatus() throws IOException {
@@ -386,7 +453,7 @@ public abstract class MetadataStoreTestBase extends Assert {
Set<Path> b = new HashSet<>();
for (String ps : pathStrs) {
- b.add(new Path(ps));
+ b.add(strToPath(ps));
}
assertTrue("Same set of files", a.equals(b));
}
@@ -398,7 +465,7 @@ public abstract class MetadataStoreTestBase extends Assert {
metas.add(new PathMetadata(makeFileStatus(filenames[i], 100)));
}
DirListingMetadata dirMeta =
- new DirListingMetadata(new Path(dirPath), metas, authoritative);
+ new DirListingMetadata(strToPath(dirPath), metas, authoritative);
ms.put(dirMeta);
}
@@ -411,10 +478,14 @@ public abstract class MetadataStoreTestBase extends Assert {
private void assertDirectorySize(String pathStr, int size)
throws IOException {
- DirListingMetadata dirMeta = ms.listChildren(new Path(pathStr));
- assertNotNull("Directory " + pathStr + " in cache", dirMeta);
- assertEquals("Number of entries in dir " + pathStr, size,
- nonDeleted(dirMeta.getListing()).size());
+ DirListingMetadata dirMeta = ms.listChildren(strToPath(pathStr));
+ if (!allowMissing()) {
+ assertNotNull("Directory " + pathStr + " in cache", dirMeta);
+ }
+ if (!allowMissing() || dirMeta != null) {
+ assertEquals("Number of entries in dir " + pathStr, size,
+ nonDeleted(dirMeta.getListing()).size());
+ }
}
/** @return only file statuses which are *not* marked deleted. */
@@ -426,8 +497,32 @@ public abstract class MetadataStoreTestBase extends Assert {
private void assertNotCached(String pathStr) throws IOException {
// TODO this should return an entry with deleted flag set
- PathMetadata meta = ms.get(new Path(pathStr));
- assertNull(pathStr + " not cached.", meta);
+ Path path = strToPath(pathStr);
+ PathMetadata meta = ms.get(path);
+ assertNull(pathStr + " should not be cached.", meta);
+ }
+
+ private void assertCached(String pathStr) throws IOException {
+ Path path = strToPath(pathStr);
+ PathMetadata meta = ms.get(path);
+ assertNotNull(pathStr + " should be cached.", meta);
+ }
+
+ // Convenience to add scheme if missing
+ private Path strToPath(String p) {
+ Path path = new Path(p);
+ URI uri = path.toUri();
+ if (uri.getScheme() == null) {
+ String fsScheme = contract.getFileSystem().getScheme();
+ try {
+ return new Path(new URI(fsScheme, uri.getHost(), uri.getPath(),
+ uri.getFragment()));
+ } catch (URISyntaxException e) {
+ throw new IllegalArgumentException("FileStatus path invalid with " +
+ "scheme " + fsScheme + " added", e);
+ }
+ }
+ return path;
}
private void assertEmptyDirectory(String pathStr) throws IOException {
@@ -446,7 +541,7 @@ public abstract class MetadataStoreTestBase extends Assert {
}
private FileStatus makeFileStatus(String pathStr, int size) {
- return basicFileStatus(new Path(pathStr), size, false);
+ return basicFileStatus(strToPath(pathStr), size, false);
}
private void verifyBasicFileStatus(PathMetadata meta) {
@@ -462,7 +557,7 @@ public abstract class MetadataStoreTestBase extends Assert {
}
private FileStatus makeDirStatus(String pathStr) {
- return basicFileStatus(new Path(pathStr), 0, true);
+ return basicFileStatus(strToPath(pathStr), 0, true);
}
private void verifyDirStatus(PathMetadata meta) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/TestLocalMetadataStore.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/TestLocalMetadataStore.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/TestLocalMetadataStore.java
index c8b3a85..12aa9c6 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/TestLocalMetadataStore.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/TestLocalMetadataStore.java
@@ -20,9 +20,12 @@ package org.apache.hadoop.fs.s3a.s3guard;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.RawLocalFileSystem;
+import org.apache.hadoop.fs.Path;
+import org.junit.Test;
import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
/**
* MetadataStore unit test for {@link LocalMetadataStore}.
@@ -36,13 +39,13 @@ public class TestLocalMetadataStore extends MetadataStoreTestBase {
private FileSystem fs;
public LocalMSContract() {
- fs = new RawLocalFileSystem();
- Configuration config = fs.getConf();
- if (config == null) {
- config = new Configuration();
- }
+ Configuration config = new Configuration();
config.set(LocalMetadataStore.CONF_MAX_RECORDS, MAX_ENTRIES_STR);
- fs.setConf(config);
+ try {
+ fs = FileSystem.getLocal(config);
+ } catch (IOException e) {
+ fail("Error creating LocalFileSystem");
+ }
}
@Override
@@ -62,4 +65,39 @@ public class TestLocalMetadataStore extends MetadataStoreTestBase {
return new LocalMSContract();
}
-}
\ No newline at end of file
+ @Test
+ public void testClearByAncestor() {
+ Map<Path, String> map = new HashMap<>();
+
+ // 1. Test paths without scheme/host
+ assertClearResult(map, "", "/", 0);
+ assertClearResult(map, "", "/dirA/dirB", 2);
+ assertClearResult(map, "", "/invalid", 5);
+
+
+ // 2. Test paths w/ scheme/host
+ String p = "s3a://fake-bucket-name";
+ assertClearResult(map, p, "/", 0);
+ assertClearResult(map, p, "/dirA/dirB", 2);
+ assertClearResult(map, p, "/invalid", 5);
+ }
+
+ private static void populateMap(Map<Path, String> map, String prefix) {
+ String dummyVal = "dummy";
+ map.put(new Path(prefix + "/dirA/dirB/"), dummyVal);
+ map.put(new Path(prefix + "/dirA/dirB/dirC"), dummyVal);
+ map.put(new Path(prefix + "/dirA/dirB/dirC/file1"), dummyVal);
+ map.put(new Path(prefix + "/dirA/dirB/dirC/file2"), dummyVal);
+ map.put(new Path(prefix + "/dirA/file1"), dummyVal);
+ }
+
+ private static void assertClearResult(Map <Path, String> map,
+ String prefixStr, String pathStr, int leftoverSize) {
+ populateMap(map, prefixStr);
+ LocalMetadataStore.clearHashByAncestor(new Path(prefixStr + pathStr), map);
+ assertEquals(String.format("Map should have %d entries", leftoverSize),
+ leftoverSize, map.size());
+ map.clear();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/TestNullMetadataStore.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/TestNullMetadataStore.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/TestNullMetadataStore.java
new file mode 100644
index 0000000..5b18686
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/TestNullMetadataStore.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.s3a.s3guard;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+
+import java.io.IOException;
+
+/**
+ * Run MetadataStore unit tests on the NullMetadataStore implementation.
+ */
+public class TestNullMetadataStore extends MetadataStoreTestBase {
+ private static class NullMSContract extends AbstractMSContract {
+ @Override
+ public FileSystem getFileSystem() {
+ Configuration config = new Configuration();
+ try {
+ return FileSystem.getLocal(config);
+ } catch (IOException e) {
+ fail("Error creating LocalFileSystem");
+ return null;
+ }
+ }
+
+ @Override
+ public MetadataStore getMetadataStore() throws IOException {
+ return new NullMetadataStore();
+ }
+ }
+
+ /** This MetadataStore always says "I don't know, ask the backing store". */
+ @Override
+ public boolean allowMissing() {
+ return true;
+ }
+
+ @Override
+ public AbstractMSContract createContract() {
+ return new NullMSContract();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3ADirectoryPerformance.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3ADirectoryPerformance.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3ADirectoryPerformance.java
index d71364f..6d77e32 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3ADirectoryPerformance.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3ADirectoryPerformance.java
@@ -113,14 +113,15 @@ public class ITestS3ADirectoryPerformance extends S3AScaleTestBase {
listContinueRequests,
listStatusCalls,
getFileStatusCalls);
- assertEquals(listRequests.toString(), 2, listRequests.diff());
+ if (!fs.isMetadataStoreConfigured()) {
+ assertEquals(listRequests.toString(), 2, listRequests.diff());
+ }
reset(metadataRequests,
listRequests,
listContinueRequests,
listStatusCalls,
getFileStatusCalls);
-
} finally {
describe("deletion");
// deletion at the end of the run
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/test/resources/core-site.xml
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/resources/core-site.xml b/hadoop-tools/hadoop-aws/src/test/resources/core-site.xml
index 7d2046b..d76c4c8 100644
--- a/hadoop-tools/hadoop-aws/src/test/resources/core-site.xml
+++ b/hadoop-tools/hadoop-aws/src/test/resources/core-site.xml
@@ -30,6 +30,28 @@
<final>true</final>
</property>
+ <!-- Uncomment these two properties to run integration tests with
+ local metadata store (S3Guard).
+
+ <property>
+ <name>fs.s3a.metadatastore.authoritative</name>
+ <value>true</value>
+ </property>
+ <property>
+ <name>fs.s3a.metadatastore.impl</name>
+ <value>org.apache.hadoop.fs.s3a.s3guard.LocalMetadataStore</value>
+ </property>
+
+ -->
+
+ <!-- Scale integration tests may time out on slower connections
+ you can reduce the operation count like so to mitigate this.
+ <property>
+ <name>scale.test.operation.count</name>
+ <value>500</value>
+ </property>
+ -->
+
<!-- Turn security off for tests by default -->
<property>
<name>hadoop.security.authentication</name>
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[2/2] hadoop git commit: HADOOP-13651 S3Guard: S3AFileSystem
Integration with MetadataStore
Posted by fa...@apache.org.
HADOOP-13651 S3Guard: S3AFileSystem Integration with MetadataStore
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/f4c37a57
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/f4c37a57
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/f4c37a57
Branch: refs/heads/HADOOP-13345
Commit: f4c37a5794d01b8cc72d747ab465787902436961
Parents: 56b715f
Author: Aaron Fabbri <fa...@apache.org>
Authored: Mon Oct 17 13:53:46 2016 -0700
Committer: Aaron Fabbri <fa...@apache.org>
Committed: Tue Nov 15 20:05:12 2016 -0800
----------------------------------------------------------------------
.../src/main/resources/core-default.xml | 24 ++
.../org/apache/hadoop/fs/s3a/Constants.java | 5 +
.../hadoop/fs/s3a/S3ABlockOutputStream.java | 17 +-
.../org/apache/hadoop/fs/s3a/S3AFileStatus.java | 10 +
.../org/apache/hadoop/fs/s3a/S3AFileSystem.java | 194 ++++++++---
.../apache/hadoop/fs/s3a/S3AOutputStream.java | 14 +-
.../java/org/apache/hadoop/fs/s3a/S3AUtils.java | 34 +-
.../org/apache/hadoop/fs/s3a/UploadInfo.java | 43 +++
.../fs/s3a/s3guard/DirListingMetadata.java | 102 +++++-
.../fs/s3a/s3guard/LocalMetadataStore.java | 176 ++++++++--
.../hadoop/fs/s3a/s3guard/MetadataStore.java | 3 +
.../fs/s3a/s3guard/NullMetadataStore.java | 79 +++++
.../hadoop/fs/s3a/s3guard/PathMetadata.java | 19 +-
.../apache/hadoop/fs/s3a/s3guard/S3Guard.java | 334 +++++++++++++++++++
.../site/markdown/tools/hadoop-aws/s3guard.md | 26 ++
.../hadoop/fs/s3a/ITestS3ACredentialsInURL.java | 1 +
.../fs/s3a/ITestS3AFileOperationCost.java | 16 +-
.../org/apache/hadoop/fs/s3a/S3ATestUtils.java | 21 +-
.../s3a/fileContext/ITestS3AFileContextURI.java | 12 +-
.../fs/s3a/s3guard/MetadataStoreTestBase.java | 301 +++++++++++------
.../fs/s3a/s3guard/TestLocalMetadataStore.java | 54 ++-
.../fs/s3a/s3guard/TestNullMetadataStore.java | 58 ++++
.../s3a/scale/ITestS3ADirectoryPerformance.java | 5 +-
.../hadoop-aws/src/test/resources/core-site.xml | 22 ++
24 files changed, 1363 insertions(+), 207 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
index 327acfa..4f2c962 100644
--- a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
+++ b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
@@ -1211,6 +1211,30 @@
</property>
<property>
+ <name>fs.s3a.metadatastore.authoritative</name>
+ <value>false</value>
+ <description>
+ When true, allow MetadataStore implementations to act as source of
+ truth for getting file status and directory listings. Even if this
+ is set to true, MetadataStore implementations may choose not to
+ return authoritative results. If the configured MetadataStore does
+ not support being authoritative, this setting will have no effect.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.metadatastore.impl</name>
+ <value>org.apache.hadoop.fs.s3a.s3guard.NullMetadataStore</value>
+ <description>
+ Fully-qualified name of the class that implements the MetadataStore
+ to be used by s3a. The default class, NullMetadataStore, has no
+ effect: s3a will continue to treat the backing S3 service as the one
+ and only source of truth for file and directory metadata.
+ </description>
+</property>
+
+
+<property>
<name>fs.s3a.impl</name>
<value>org.apache.hadoop.fs.s3a.S3AFileSystem</value>
<description>The implementation class of the S3A Filesystem</description>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
index 65df0bf..518bd33 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
@@ -230,6 +230,11 @@ public final class Constants {
public static final String USER_AGENT_PREFIX = "fs.s3a.user.agent.prefix";
+ /** Whether or not to allow MetadataStore to be source of truth. */
+ public static final String METADATASTORE_AUTHORITATIVE =
+ "fs.s3a.metadatastore.authoritative";
+ public static final boolean DEFAULT_METADATASTORE_AUTHORITATIVE = false;
+
/** read ahead buffer size to prevent connection re-establishments. */
public static final String READAHEAD_RANGE = "fs.s3a.readahead.range";
public static final long DEFAULT_READAHEAD_RANGE = 64 * 1024;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ABlockOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ABlockOutputStream.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ABlockOutputStream.java
index b66a23f..1eb4195 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ABlockOutputStream.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ABlockOutputStream.java
@@ -81,6 +81,9 @@ class S3ABlockOutputStream extends OutputStream {
/** Size of all blocks. */
private final int blockSize;
+ /** Total bytes for uploads submitted so far. */
+ private long bytesSubmitted;
+
/** Callback for progress. */
private final ProgressListener progressListener;
private final ListeningExecutorService executorService;
@@ -302,6 +305,7 @@ class S3ABlockOutputStream extends OutputStream {
}
try {
multiPartUpload.uploadBlockAsync(getActiveBlock());
+ bytesSubmitted += getActiveBlock().dataSize();
} finally {
// set the block to null, so the next write will create a new block.
clearActiveBlock();
@@ -330,13 +334,14 @@ class S3ABlockOutputStream extends OutputStream {
this,
blockCount,
hasBlock ? block : "(none)");
+ long bytes = 0;
try {
if (multiPartUpload == null) {
if (hasBlock) {
// no uploads of data have taken place, put the single block up.
// This must happen even if there is no data, so that 0 byte files
// are created.
- putObject();
+ bytes = putObject();
}
} else {
// there has already been at least one block scheduled for upload;
@@ -350,6 +355,7 @@ class S3ABlockOutputStream extends OutputStream {
multiPartUpload.waitForAllPartUploads();
// then complete the operation
multiPartUpload.complete(partETags);
+ bytes = bytesSubmitted;
}
LOG.debug("Upload complete for {}", writeOperationHelper);
} catch (IOException ioe) {
@@ -364,7 +370,7 @@ class S3ABlockOutputStream extends OutputStream {
clearActiveBlock();
}
// All end of write operations, including deleting fake parent directories
- writeOperationHelper.writeSuccessful();
+ writeOperationHelper.writeSuccessful(bytes);
}
/**
@@ -372,8 +378,11 @@ class S3ABlockOutputStream extends OutputStream {
* is empty a 0-byte PUT will be invoked, as it is needed to create an
* entry at the far end.
* @throws IOException any problem.
+ * @return number of bytes uploaded. If thread was interrupted while
+ * waiting for upload to complete, returns zero with interrupted flag set
+ * on this thread.
*/
- private void putObject() throws IOException {
+ private int putObject() throws IOException {
LOG.debug("Executing regular upload for {}", writeOperationHelper);
final S3ADataBlocks.DataBlock block = getActiveBlock();
@@ -401,9 +410,11 @@ class S3ABlockOutputStream extends OutputStream {
//wait for completion
try {
putObjectResult.get();
+ return size;
} catch (InterruptedException ie) {
LOG.warn("Interrupted object upload", ie);
Thread.currentThread().interrupt();
+ return 0;
} catch (ExecutionException ee) {
throw extractException("regular upload", key, ee);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileStatus.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileStatus.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileStatus.java
index b0f08e3..62c4b0d 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileStatus.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileStatus.java
@@ -68,6 +68,16 @@ public class S3AFileStatus extends FileStatus {
return isEmptyDirectory;
}
+ /**
+ * Should not be called by clients. Only used so {@link org.apache.hadoop
+ * .fs.s3a.s3guard.MetadataStore} can maintain this flag when caching
+ * FileStatuses on behalf of s3a.
+ * @param value true iff empty
+ */
+ public void setIsEmptyDirectory(boolean value) {
+ isEmptyDirectory = value;
+ }
+
/** Compare if this object is equal to another object.
* @param o the object to be compared.
* @return true if two file status has the same path name; false if not.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
index 925abbc..9b8692a 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
@@ -86,6 +86,10 @@ import org.apache.hadoop.fs.PathIsNotEmptyDirectoryException;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.fs.StorageStatistics;
import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.fs.s3a.s3guard.DirListingMetadata;
+import org.apache.hadoop.fs.s3a.s3guard.MetadataStore;
+import org.apache.hadoop.fs.s3a.s3guard.PathMetadata;
+import org.apache.hadoop.fs.s3a.s3guard.S3Guard;
import org.apache.hadoop.fs.s3native.S3xLoginHelper;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.Progressable;
@@ -142,6 +146,8 @@ public class S3AFileSystem extends FileSystem {
private long readAhead;
private S3AInputPolicy inputPolicy;
private final AtomicBoolean closed = new AtomicBoolean(false);
+ private MetadataStore metadataStore;
+ private boolean allowAuthoritative;
// The maximum number of entries that can be deleted in any call to s3
private static final int MAX_ENTRIES_TO_DELETE = 1000;
@@ -242,10 +248,13 @@ public class S3AFileSystem extends FileSystem {
} else {
LOG.debug("Using S3AOutputStream");
}
+
+ metadataStore = S3Guard.getMetadataStore(this);
+ allowAuthoritative = conf.getBoolean(METADATASTORE_AUTHORITATIVE,
+ DEFAULT_METADATASTORE_AUTHORITATIVE);
} catch (AmazonClientException e) {
throw translateException("initializing ", new Path(name), e);
}
-
}
/**
@@ -651,14 +660,17 @@ public class S3AFileSystem extends FileSystem {
/**
* The inner rename operation. See {@link #rename(Path, Path)} for
* the description of the operation.
- * @param src path to be renamed
- * @param dst new path after rename
+ * @param source path to be renamed
+ * @param dest new path after rename
* @return true if rename is successful
* @throws IOException on IO failure.
* @throws AmazonClientException on failures inside the AWS SDK
*/
- private boolean innerRename(Path src, Path dst) throws IOException,
+ private boolean innerRename(Path source, Path dest) throws IOException,
AmazonClientException {
+ Path src = qualify(source);
+ Path dst = qualify(dest);
+
LOG.debug("Rename path {} to {}", src, dst);
incrementStatistic(INVOCATION_RENAME);
@@ -715,9 +727,20 @@ public class S3AFileSystem extends FileSystem {
}
}
+ // If we have a MetadataStore, track deletions/creations.
+ List<Path> srcPaths = null;
+ List<PathMetadata> dstMetas = null;
+ if (!S3Guard.isNullMetadataStore(metadataStore)) {
+ srcPaths = new ArrayList<>();
+ dstMetas = new ArrayList<>();
+ }
+ // HADOOP-13761 s3guard: retries when source paths are not visible yet
+ // TODO s3guard: performance: mark destination dirs as authoritative
+
// Ok! Time to start
if (srcStatus.isFile()) {
LOG.debug("rename: renaming file {} to {}", src, dst);
+ long length = srcStatus.getLen();
if (dstStatus != null && dstStatus.isDirectory()) {
String newDstKey = dstKey;
if (!newDstKey.endsWith("/")) {
@@ -726,9 +749,14 @@ public class S3AFileSystem extends FileSystem {
String filename =
srcKey.substring(pathToKey(src.getParent()).length()+1);
newDstKey = newDstKey + filename;
- copyFile(srcKey, newDstKey, srcStatus.getLen());
+ copyFile(srcKey, newDstKey, length);
+ S3Guard.addMoveFile(metadataStore, srcPaths, dstMetas, src,
+ keyToQualifiedPath(newDstKey), length, getDefaultBlockSize(dst),
+ username);
} else {
copyFile(srcKey, dstKey, srcStatus.getLen());
+ S3Guard.addMoveFile(metadataStore, srcPaths, dstMetas, src, dst,
+ length, getDefaultBlockSize(dst), username);
}
innerDelete(srcStatus, false);
} else {
@@ -765,11 +793,24 @@ public class S3AFileSystem extends FileSystem {
while (true) {
for (S3ObjectSummary summary : objects.getObjectSummaries()) {
- keysToDelete.add(
- new DeleteObjectsRequest.KeyVersion(summary.getKey()));
+ long length = summary.getSize();
+ keysToDelete
+ .add(new DeleteObjectsRequest.KeyVersion(summary.getKey()));
String newDstKey =
dstKey + summary.getKey().substring(srcKey.length());
- copyFile(summary.getKey(), newDstKey, summary.getSize());
+ copyFile(summary.getKey(), newDstKey, length);
+
+ if (!S3Guard.isNullMetadataStore(metadataStore)) {
+ Path srcPath = keyToQualifiedPath(summary.getKey());
+ Path dstPath = keyToQualifiedPath(newDstKey);
+ if (objectRepresentsDirectory(summary.getKey(), length)) {
+ S3Guard.addMoveDir(metadataStore, srcPaths, dstMetas, srcPath,
+ dstPath, username);
+ } else {
+ S3Guard.addMoveFile(metadataStore, srcPaths, dstMetas, srcPath,
+ dstPath, length, getDefaultBlockSize(dstPath), username);
+ }
+ }
if (keysToDelete.size() == MAX_ENTRIES_TO_DELETE) {
removeKeys(keysToDelete, true, false);
@@ -785,8 +826,13 @@ public class S3AFileSystem extends FileSystem {
break;
}
}
+
+ // We moved all the children, now move the top-level dir.
+ S3Guard.addMoveDir(metadataStore, srcPaths, dstMetas, src, dst, username);
}
+ metadataStore.move(srcPaths, dstMetas);
+
if (src.getParent() != dst.getParent()) {
deleteUnnecessaryFakeDirectories(dst.getParent());
createFakeDirectoryIfNecessary(src.getParent());
@@ -805,6 +851,11 @@ public class S3AFileSystem extends FileSystem {
return getObjectMetadata(pathToKey(path));
}
+ @VisibleForTesting
+ public boolean isMetadataStoreConfigured() {
+ return !S3Guard.isNullMetadataStore(metadataStore);
+ }
+
/**
* Increment a statistic by 1.
* @param statistic The operation to increment
@@ -1013,7 +1064,7 @@ public class S3AFileSystem extends FileSystem {
* @param putObjectRequest the request
* @return the upload initiated
*/
- public Upload putObject(PutObjectRequest putObjectRequest) {
+ public UploadInfo putObject(PutObjectRequest putObjectRequest) {
long len;
if (putObjectRequest.getFile() != null) {
len = putObjectRequest.getFile().length();
@@ -1024,7 +1075,7 @@ public class S3AFileSystem extends FileSystem {
try {
Upload upload = transfers.upload(putObjectRequest);
incrementPutCompletedStatistics(true, len);
- return upload;
+ return new UploadInfo(upload, len);
} catch (AmazonClientException e) {
incrementPutCompletedStatistics(false, len);
throw e;
@@ -1224,7 +1275,9 @@ public class S3AFileSystem extends FileSystem {
if (status.isEmptyDirectory()) {
LOG.debug("Deleting fake empty directory {}", key);
+ // HADOOP-13761 s3guard: retries here
deleteObject(key);
+ metadataStore.delete(f);
instrumentation.directoryDeleted();
} else {
LOG.debug("Getting objects for directory prefix {} to delete", key);
@@ -1240,6 +1293,7 @@ public class S3AFileSystem extends FileSystem {
LOG.debug("Got object to delete {}", summary.getKey());
if (keys.size() == MAX_ENTRIES_TO_DELETE) {
+ // HADOOP-13761 s3guard: retries
removeKeys(keys, true, false);
}
}
@@ -1248,16 +1302,19 @@ public class S3AFileSystem extends FileSystem {
objects = continueListObjects(objects);
} else {
if (!keys.isEmpty()) {
+ // HADOOP-13761 s3guard: retries
removeKeys(keys, false, false);
}
break;
}
}
}
+ metadataStore.deleteSubtree(f);
} else {
LOG.debug("delete: Path is a file");
instrumentation.fileDeleted(1);
deleteObject(key);
+ metadataStore.delete(f);
}
Path parent = f.getParent();
@@ -1345,6 +1402,11 @@ public class S3AFileSystem extends FileSystem {
key = key + '/';
}
+ DirListingMetadata dirMeta = metadataStore.listChildren(path);
+ if (allowAuthoritative && dirMeta != null && dirMeta.isAuthoritative()) {
+ return S3Guard.dirMetaToStatuses(dirMeta);
+ }
+
ListObjectsRequest request = createListObjectsRequest(key, "/");
LOG.debug("listStatus: doing listObjects for directory {}", key);
@@ -1357,7 +1419,7 @@ public class S3AFileSystem extends FileSystem {
while (files.hasNext()) {
result.add(files.next());
}
- return result.toArray(new FileStatus[result.size()]);
+ return S3Guard.dirListingUnion(metadataStore, path, result, dirMeta);
} else {
LOG.debug("Adding: rd (not a dir): {}", path);
FileStatus[] stats = new FileStatus[1];
@@ -1437,7 +1499,7 @@ public class S3AFileSystem extends FileSystem {
* Make the given path and all non-existent parents into
* directories.
* See {@link #mkdirs(Path, FsPermission)}
- * @param f path to create
+ * @param p path to create
* @param permission to apply to f
* @return true if a directory was created
* @throws FileAlreadyExistsException there is a file at the path specified
@@ -1446,11 +1508,17 @@ public class S3AFileSystem extends FileSystem {
*/
// TODO: If we have created an empty file at /foo/bar and we then call
// mkdirs for /foo/bar/baz/roo what happens to the empty file /foo/bar/?
- private boolean innerMkdirs(Path f, FsPermission permission)
+ private boolean innerMkdirs(Path p, FsPermission permission)
throws IOException, FileAlreadyExistsException, AmazonClientException {
+ Path f = qualify(p);
LOG.debug("Making directory: {}", f);
incrementStatistic(INVOCATION_MKDIRS);
FileStatus fileStatus;
+ List<Path> metadataStoreDirs = null;
+ if (!S3Guard.isNullMetadataStore(metadataStore)) {
+ metadataStoreDirs = new ArrayList<>();
+ }
+
try {
fileStatus = getFileStatus(f);
@@ -1460,7 +1528,11 @@ public class S3AFileSystem extends FileSystem {
throw new FileAlreadyExistsException("Path is a file: " + f);
}
} catch (FileNotFoundException e) {
+ // Walk path to root, ensuring closest ancestor is a directory, not file
Path fPart = f.getParent();
+ if (metadataStoreDirs != null) {
+ metadataStoreDirs.add(f);
+ }
do {
try {
fileStatus = getFileStatus(fPart);
@@ -1474,12 +1546,18 @@ public class S3AFileSystem extends FileSystem {
}
} catch (FileNotFoundException fnfe) {
instrumentation.errorIgnored();
+ // We create all missing directories in MetadataStore; it does not
+ // infer directories exist by prefix like S3.
+ if (metadataStoreDirs != null) {
+ metadataStoreDirs.add(fPart);
+ }
}
fPart = fPart.getParent();
} while (fPart != null);
String key = pathToKey(f);
createFakeDirectory(key);
+ S3Guard.makeDirsOrdered(metadataStore, metadataStoreDirs, username);
return true;
}
}
@@ -1496,20 +1574,30 @@ public class S3AFileSystem extends FileSystem {
final Path path = qualify(f);
String key = pathToKey(path);
LOG.debug("Getting path status for {} ({})", path , key);
+
+ // Check MetadataStore, if any.
+ PathMetadata pm = metadataStore.get(path);
+ if (pm != null) {
+ // HADOOP-13760: handle deleted files, i.e. PathMetadata#isDeleted() here
+ return (S3AFileStatus)pm.getFileStatus();
+ }
+
if (!key.isEmpty()) {
try {
ObjectMetadata meta = getObjectMetadata(key);
if (objectRepresentsDirectory(key, meta.getContentLength())) {
LOG.debug("Found exact file: fake directory");
- return new S3AFileStatus(true, path, username);
+ return S3Guard.putAndReturn(metadataStore,
+ new S3AFileStatus(true, path, username));
} else {
LOG.debug("Found exact file: normal file");
- return new S3AFileStatus(meta.getContentLength(),
- dateToLong(meta.getLastModified()),
- path,
- getDefaultBlockSize(path),
- username);
+ return S3Guard.putAndReturn(metadataStore,
+ new S3AFileStatus(meta.getContentLength(),
+ dateToLong(meta.getLastModified()),
+ path,
+ getDefaultBlockSize(path),
+ username));
}
} catch (AmazonServiceException e) {
if (e.getStatusCode() != 404) {
@@ -1527,16 +1615,18 @@ public class S3AFileSystem extends FileSystem {
if (objectRepresentsDirectory(newKey, meta.getContentLength())) {
LOG.debug("Found file (with /): fake directory");
- return new S3AFileStatus(true, path, username);
+ return S3Guard.putAndReturn(metadataStore,
+ new S3AFileStatus(true, path, username));
} else {
LOG.warn("Found file (with /): real file? should not happen: {}",
key);
- return new S3AFileStatus(meta.getContentLength(),
- dateToLong(meta.getLastModified()),
- path,
- getDefaultBlockSize(path),
- username);
+ return S3Guard.putAndReturn(metadataStore,
+ new S3AFileStatus(meta.getContentLength(),
+ dateToLong(meta.getLastModified()),
+ path,
+ getDefaultBlockSize(path),
+ username));
}
} catch (AmazonServiceException e) {
if (e.getStatusCode() != 404) {
@@ -1573,10 +1663,12 @@ public class S3AFileSystem extends FileSystem {
}
}
- return new S3AFileStatus(false, path, username);
+ return S3Guard.putAndReturn(metadataStore,
+ new S3AFileStatus(false, path, username));
} else if (key.isEmpty()) {
LOG.debug("Found root directory");
- return new S3AFileStatus(true, path, username);
+ return S3Guard.putAndReturn(metadataStore,
+ new S3AFileStatus(true, path, username));
}
} catch (AmazonServiceException e) {
if (e.getStatusCode() != 404) {
@@ -1652,12 +1744,13 @@ public class S3AFileSystem extends FileSystem {
final ObjectMetadata om = newObjectMetadata(srcfile.length());
PutObjectRequest putObjectRequest = newPutObjectRequest(key, om, srcfile);
- Upload up = putObject(putObjectRequest);
+ UploadInfo info = putObject(putObjectRequest);
+ Upload upload = info.getUpload();
ProgressableProgressListener listener = new ProgressableProgressListener(
- this, key, up, null);
- up.addProgressListener(listener);
+ this, key, upload, null);
+ upload.addProgressListener(listener);
try {
- up.waitForUploadResult();
+ upload.waitForUploadResult();
} catch (InterruptedException e) {
throw new InterruptedIOException("Interrupted copying " + src
+ " to " + dst + ", cancelling");
@@ -1665,7 +1758,7 @@ public class S3AFileSystem extends FileSystem {
listener.uploadCompleted();
// This will delete unnecessary fake parent directories
- finishedWrite(key);
+ finishedWrite(key, info.getLength());
if (delSrc) {
local.delete(src, false);
@@ -1689,6 +1782,10 @@ public class S3AFileSystem extends FileSystem {
transfers.shutdownNow(true);
transfers = null;
}
+ if (metadataStore != null) {
+ metadataStore.close();
+ metadataStore = null;
+ }
}
}
@@ -1756,10 +1853,26 @@ public class S3AFileSystem extends FileSystem {
/**
* Perform post-write actions.
* @param key key written to
+ * @param length total length of file written
*/
- public void finishedWrite(String key) {
- LOG.debug("Finished write to {}", key);
- deleteUnnecessaryFakeDirectories(keyToPath(key).getParent());
+ public void finishedWrite(String key, long length) {
+ LOG.debug("Finished write to {}, len {}", key, length);
+ Path p = keyToQualifiedPath(key);
+ deleteUnnecessaryFakeDirectories(p.getParent());
+
+ // See note about failure semantics in s3guard.md doc.
+ try {
+ if (!S3Guard.isNullMetadataStore(metadataStore)) {
+ S3AFileStatus status = createUploadFileStatus(p,
+ S3AUtils.objectRepresentsDirectory(key, length), length,
+ getDefaultBlockSize(p), username);
+ metadataStore.put(new PathMetadata(status));
+ }
+ } catch (IOException e) {
+ LOG.error("s3guard: Error updating MetadataStore for write to {}:",
+ key, e);
+ instrumentation.errorIgnored();
+ }
}
/**
@@ -1813,9 +1926,9 @@ public class S3AFileSystem extends FileSystem {
PutObjectRequest putObjectRequest = newPutObjectRequest(objectName,
newObjectMetadata(0L),
im);
- Upload upload = putObject(putObjectRequest);
+ UploadInfo info = putObject(putObjectRequest);
try {
- upload.waitForUploadResult();
+ info.getUpload().waitForUploadResult();
} catch (InterruptedException e) {
throw new InterruptedIOException("Interrupted creating " + objectName);
}
@@ -2060,6 +2173,9 @@ public class S3AFileSystem extends FileSystem {
new Listing.AcceptFilesOnly(path)));
}
} catch (AmazonClientException e) {
+ // TODO s3guard:
+ // 1. retry on file not found exception
+ // 2. merge listing with MetadataStore's view of directory tree
throw translateException("listFiles", path, e);
}
}
@@ -2170,8 +2286,8 @@ public class S3AFileSystem extends FileSystem {
/**
* Callback on a successful write.
*/
- void writeSuccessful() {
- finishedWrite(key);
+ void writeSuccessful(long length) {
+ finishedWrite(key, length);
}
/**
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AOutputStream.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AOutputStream.java
index 6ebc9e4..e723b75 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AOutputStream.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AOutputStream.java
@@ -20,7 +20,6 @@ package org.apache.hadoop.fs.s3a;
import com.amazonaws.AmazonClientException;
import com.amazonaws.services.s3.model.ObjectMetadata;
-import com.amazonaws.services.s3.transfer.Upload;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
@@ -101,19 +100,20 @@ public class S3AOutputStream extends OutputStream {
try {
final ObjectMetadata om = fs.newObjectMetadata(backupFile.length());
- Upload upload = fs.putObject(
+ UploadInfo info = fs.putObject(
fs.newPutObjectRequest(
key,
om,
backupFile));
ProgressableProgressListener listener =
- new ProgressableProgressListener(fs, key, upload, progress);
- upload.addProgressListener(listener);
+ new ProgressableProgressListener(fs, key, info.getUpload(), progress);
+ info.getUpload().addProgressListener(listener);
- upload.waitForUploadResult();
+ info.getUpload().waitForUploadResult();
listener.uploadCompleted();
- // This will delete unnecessary fake parent directories
- fs.finishedWrite(key);
+ // This will delete unnecessary fake parent directories, update any
+ // MetadataStore
+ fs.finishedWrite(key, info.getLength());
} catch (InterruptedException e) {
throw (InterruptedIOException) new InterruptedIOException(e.toString())
.initCause(e);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java
index 49f8862..028e1d8 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java
@@ -248,12 +248,38 @@ public final class S3AUtils {
S3ObjectSummary summary,
long blockSize,
String owner) {
- if (objectRepresentsDirectory(summary.getKey(), summary.getSize())) {
+ long size = summary.getSize();
+ return createFileStatus(keyPath,
+ objectRepresentsDirectory(summary.getKey(), size),
+ size, summary.getLastModified(), blockSize, owner);
+ }
+
+ /**
+ * Create a file status for object we just uploaded. For files, we use
+ * current time as modification time, since s3a uses S3's service-based
+ * modification time, which will not be available until we do a
+ * getFileStatus() later on.
+ * @param keyPath path for created object
+ * @param isDir true iff directory
+ * @param size file length
+ * @param blockSize block size for file status
+ * @param owner Hadoop username
+ * @return a status entry
+ */
+ public static S3AFileStatus createUploadFileStatus(Path keyPath,
+ boolean isDir, long size, long blockSize, String owner) {
+ Date date = isDir ? null : new Date();
+ return createFileStatus(keyPath, isDir, size, date, blockSize, owner);
+ }
+
+ /* Date 'modified' is ignored when isDir is true. */
+ private static S3AFileStatus createFileStatus(Path keyPath, boolean isDir,
+ long size, Date modified, long blockSize, String owner) {
+ if (isDir) {
return new S3AFileStatus(true, keyPath, owner);
} else {
- return new S3AFileStatus(summary.getSize(),
- dateToLong(summary.getLastModified()), keyPath,
- blockSize, owner);
+ return new S3AFileStatus(size, dateToLong(modified), keyPath, blockSize,
+ owner);
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/UploadInfo.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/UploadInfo.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/UploadInfo.java
new file mode 100644
index 0000000..b6d31af0
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/UploadInfo.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.s3a;
+
+import com.amazonaws.services.s3.transfer.Upload;
+
+/**
+ * Simple struct that contains information about a S3 upload.
+ */
+public class UploadInfo {
+ private Upload upload;
+ private long length;
+
+ public UploadInfo(Upload upload, long length) {
+ this.upload = upload;
+ this.length = length;
+ }
+
+ public Upload getUpload() {
+ return upload;
+ }
+
+ public long getLength() {
+ return length;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DirListingMetadata.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DirListingMetadata.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DirListingMetadata.java
index 5bfb6c0..c25ad3a 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DirListingMetadata.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DirListingMetadata.java
@@ -23,6 +23,8 @@ import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
+import java.net.URI;
+import java.net.URISyntaxException;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
@@ -39,8 +41,7 @@ import com.google.common.base.Preconditions;
public class DirListingMetadata {
/**
- * Convenience parameter for passing into
- * {@link DirListingMetadata#DirListingMetadata(Path, Collection, boolean)}.
+ * Convenience parameter for passing into constructor.
*/
public static final Collection<PathMetadata> EMPTY_DIR =
Collections.emptyList();
@@ -55,7 +56,8 @@ public class DirListingMetadata {
/**
* Create a directory listing metadata container.
*
- * @param path Path of the directory.
+ * @param path Path of the directory. If this path has a host component, then
+ * all paths added later via put() must also have the same host.
* @param listing Entries in the directory.
* @param isAuthoritative true iff listing is the full contents of the
* directory, and the calling client reports that this may be cached as
@@ -63,11 +65,15 @@ public class DirListingMetadata {
*/
public DirListingMetadata(Path path, Collection<PathMetadata> listing,
boolean isAuthoritative) {
- Preconditions.checkNotNull(path, "path must be non-null");
+
+ checkPathAbsolute(path);
this.path = path;
+
if (listing != null) {
for (PathMetadata entry : listing) {
- listMap.put(entry.getFileStatus().getPath(), entry);
+ Path childPath = entry.getFileStatus().getPath();
+ checkChildPath(childPath);
+ listMap.put(childPath, entry);
}
}
this.isAuthoritative = isAuthoritative;
@@ -88,6 +94,15 @@ public class DirListingMetadata {
}
/**
+ * @return number of entries tracked. This is not the same as the number
+ * of entries in the actual directory unless {@link #isAuthoritative()} is
+ * true.
+ */
+ public int numEntries() {
+ return listMap.size();
+ }
+
+ /**
* @return true iff this directory listing is full and authoritative within
* the scope of the {@code MetadataStore} that returned it.
*/
@@ -138,8 +153,7 @@ public class DirListingMetadata {
public void put(FileStatus childFileStatus) {
Preconditions.checkNotNull(childFileStatus,
"childFileStatus must be non-null");
- Path childPath = childFileStatus.getPath();
- checkChildPath(childPath);
+ Path childPath = childStatusToPathKey(childFileStatus);
listMap.put(childPath, new PathMetadata(childFileStatus));
}
@@ -153,17 +167,81 @@ public class DirListingMetadata {
}
/**
- * Performs pre-condition checks on child path arguments.
- *
+ * Log contents to supplied StringBuilder in a pretty fashion.
+ * @param sb target StringBuilder
+ */
+ public void prettyPrint(StringBuilder sb) {
+ sb.append(String.format("DirMeta %-20s %-18s",
+ path.toString(),
+ isAuthoritative ? "Authoritative" : "Not Authoritative"));
+ for (Map.Entry<Path, PathMetadata> entry : listMap.entrySet()) {
+ sb.append("\n key: ").append(entry.getKey()).append(": ");
+ entry.getValue().prettyPrint(sb);
+ }
+ sb.append("\n");
+ }
+
+ public String prettyPrint() {
+ StringBuilder sb = new StringBuilder();
+ prettyPrint(sb);
+ return sb.toString();
+ }
+ /**
+ * Checks that child path is valid.
* @param childPath path to check.
*/
private void checkChildPath(Path childPath) {
- Preconditions.checkNotNull(childPath, "childPath must be non-null");
- Preconditions.checkArgument(childPath.isAbsolute(), "childPath must be " +
- "absolute");
+ checkPathAbsolute(childPath);
+
+ // If this dir's path has host (and thus scheme), so must its children
+ URI parentUri = path.toUri();
+ if (parentUri.getHost() != null) {
+ URI childUri = childPath.toUri();
+ Preconditions.checkNotNull(childUri.getHost(), "Expected non-null URI " +
+ "host");
+ Preconditions.checkArgument(
+ childUri.getHost().equals(parentUri.getHost()));
+ Preconditions.checkNotNull(childUri.getScheme());
+ }
Preconditions.checkArgument(!childPath.isRoot(),
"childPath cannot be the root path");
Preconditions.checkArgument(childPath.getParent().equals(path),
"childPath must be a child of path");
}
+
+ /**
+ * For Path's that are handed in directly, we assert they are in consistent
+ * format with checkPath(). For paths that are supplied embedded in
+ * FileStatus', we attempt to fill in missing scheme and host, when this
+ * DirListingMetadata is associated with one.
+ *
+ * @return Path suitable for consistent hashtable lookups
+ */
+ private Path childStatusToPathKey(FileStatus status) {
+ Path p = status.getPath();
+ Preconditions.checkNotNull(p, "Child status' path cannot be null");
+ Preconditions.checkArgument(!p.isRoot(),
+ "childPath cannot be the root path");
+ Preconditions.checkArgument(p.getParent().equals(path),
+ "childPath must be a child of path");
+ URI uri = p.toUri();
+ URI parentUri = path.toUri();
+ // If FileStatus' path is missing host, but should have one, add it.
+ if (uri.getHost() == null && parentUri.getHost() != null) {
+ try {
+ return new Path(new URI(parentUri.getScheme(), parentUri.getHost(),
+ uri.getPath(), uri.getFragment()));
+ } catch (URISyntaxException e) {
+ throw new IllegalArgumentException("FileStatus path invalid with " +
+ " added " + parentUri.getScheme() + "://" + parentUri.getHost() +
+ " added", e);
+ }
+ }
+ return p;
+ }
+
+ private void checkPathAbsolute(Path p) {
+ Preconditions.checkNotNull(p, "path must be non-null");
+ Preconditions.checkArgument(p.isAbsolute(), "path must be absolute");
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/LocalMetadataStore.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/LocalMetadataStore.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/LocalMetadataStore.java
index 9f69189..63c931d 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/LocalMetadataStore.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/LocalMetadataStore.java
@@ -24,10 +24,13 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.s3a.S3AFileStatus;
+import org.apache.hadoop.fs.s3a.S3AFileSystem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.net.URI;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
@@ -48,6 +51,7 @@ import java.util.Map;
public class LocalMetadataStore implements MetadataStore {
public static final Logger LOG = LoggerFactory.getLogger(MetadataStore.class);
+ // TODO HADOOP-13649: use time instead of capacity for eviction.
public static final int DEFAULT_MAX_RECORDS = 128;
public static final String CONF_MAX_RECORDS =
"fs.metadatastore.local.max_records";
@@ -59,11 +63,28 @@ public class LocalMetadataStore implements MetadataStore {
private LruHashMap<Path, DirListingMetadata> dirHash;
private FileSystem fs;
+ private boolean isS3A;
+ /* Null iff this FS does not have an associated URI host. */
+ private String uriHost;
@Override
public void initialize(FileSystem fileSystem) throws IOException {
Preconditions.checkNotNull(fileSystem);
fs = fileSystem;
+
+ // Hate to take a dependency on S3A, but if the MetadataStore has to
+ // maintain S3AFileStatus#isEmptyDirectory, best to be able to to that
+ // under our own lock.
+ if (fs instanceof S3AFileSystem) {
+ isS3A = true;
+ }
+
+ URI fsURI = fs.getUri();
+ uriHost = fsURI.getHost();
+ if (uriHost != null && uriHost.equals("")) {
+ uriHost = null;
+ }
+
Configuration conf = fs.getConf();
Preconditions.checkNotNull(conf);
int maxRecords = conf.getInt(CONF_MAX_RECORDS, DEFAULT_MAX_RECORDS);
@@ -85,18 +106,15 @@ public class LocalMetadataStore implements MetadataStore {
doDelete(path, true);
}
- private synchronized void doDelete(Path path, boolean recursive) {
+ private synchronized void doDelete(Path p, boolean recursive) {
+ Path path = standardize(p);
// We could implement positive hit for 'deleted' files. For now we
// do not track them.
// Delete entry from file cache, then from cached parent directory, if any
- // Remove target file/dir
- fileHash.remove(path);
-
- // Update this and parent dir listing, if any
- dirHashDeleteFile(path);
+ removeHashEntries(path);
if (recursive) {
// Remove all entries that have this dir as path prefix.
@@ -106,14 +124,23 @@ public class LocalMetadataStore implements MetadataStore {
}
@Override
- public synchronized PathMetadata get(Path path) throws IOException {
- return fileHash.mruGet(path);
+ public synchronized PathMetadata get(Path p) throws IOException {
+ Path path = standardize(p);
+ PathMetadata m = fileHash.mruGet(path);
+ LOG.debug("get({}) -> {}", path, m == null ? "null" : m.prettyPrint());
+ return m;
}
@Override
- public synchronized DirListingMetadata listChildren(Path path) throws
+ public synchronized DirListingMetadata listChildren(Path p) throws
IOException {
- return dirHash.mruGet(path);
+ Path path = standardize(p);
+ DirListingMetadata listing = dirHash.mruGet(path);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("listChildren({}) -> {}", path,
+ listing == null ? "null" : listing.prettyPrint());
+ }
+ return listing;
}
@Override
@@ -130,11 +157,13 @@ public class LocalMetadataStore implements MetadataStore {
// 1. Delete pathsToDelete
for (Path p : pathsToDelete) {
+ LOG.debug("move: deleting metadata {}", p);
delete(p);
}
// 2. Create new destination path metadata
for (PathMetadata meta : pathsToCreate) {
+ LOG.debug("move: adding metadata {}", meta);
put(meta);
}
@@ -155,11 +184,15 @@ public class LocalMetadataStore implements MetadataStore {
@Override
public void put(PathMetadata meta) throws IOException {
+ Preconditions.checkNotNull(meta);
FileStatus status = meta.getFileStatus();
- Path path = status.getPath();
+ Path path = standardize(status.getPath());
synchronized (this) {
/* Add entry for this file. */
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("put {} -> {}", path, meta.prettyPrint());
+ }
fileHash.put(path, meta);
/* Directory case:
@@ -173,7 +206,6 @@ public class LocalMetadataStore implements MetadataStore {
* saving round trips to underlying store for subsequent listStatus()
*/
- /* If directory, go ahead and cache the fact that it is empty. */
if (status.isDirectory()) {
DirListingMetadata dir = dirHash.mruGet(path);
if (dir == null) {
@@ -184,22 +216,32 @@ public class LocalMetadataStore implements MetadataStore {
/* Update cached parent dir. */
Path parentPath = path.getParent();
- DirListingMetadata parent = dirHash.mruGet(parentPath);
- if (parent == null) {
+ if (parentPath != null) {
+ DirListingMetadata parent = dirHash.mruGet(parentPath);
+ if (parent == null) {
/* Track this new file's listing in parent. Parent is not
* authoritative, since there may be other items in it we don't know
* about. */
- parent = new DirListingMetadata(parentPath,
- DirListingMetadata.EMPTY_DIR, false);
- dirHash.put(parentPath, parent);
+ parent = new DirListingMetadata(parentPath,
+ DirListingMetadata.EMPTY_DIR, false);
+ dirHash.put(parentPath, parent);
+ } else {
+ // S3A-specific logic to maintain S3AFileStatus#isEmptyDirectory()
+ if (isS3A) {
+ setS3AIsEmpty(parentPath, false);
+ }
+ }
+ parent.put(status);
}
- parent.put(status);
}
}
@Override
public synchronized void put(DirListingMetadata meta) throws IOException {
- dirHash.put(meta.getPath(), meta);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("put dirMeta {}", meta.prettyPrint());
+ }
+ dirHash.put(standardize(meta.getPath()), meta);
}
@Override
@@ -233,11 +275,19 @@ public class LocalMetadataStore implements MetadataStore {
}
/**
- * Update dirHash to reflect deletion of file 'f'. Call with lock held.
+ * Update fileHash and dirHash to reflect deletion of file 'f'. Call with
+ * lock held.
*/
- private void dirHashDeleteFile(Path path) {
+ private void removeHashEntries(Path path) {
+
+ // Remove target file/dir
+ LOG.debug("delete file entry for {}", path);
+ fileHash.remove(path);
+
+ // Update this and parent dir listing, if any
/* If this path is a dir, remove its listing */
+ LOG.debug("removing listing of {}", path);
dirHash.remove(path);
/* Remove this path from parent's dir listing */
@@ -245,8 +295,92 @@ public class LocalMetadataStore implements MetadataStore {
if (parent != null) {
DirListingMetadata dir = dirHash.get(parent);
if (dir != null) {
+ LOG.debug("removing parent's entry for {} ", path);
dir.remove(path);
+
+ // S3A-specific logic dealing with S3AFileStatus#isEmptyDirectory()
+ if (isS3A) {
+ if (dir.isAuthoritative() && dir.numEntries() == 0) {
+ setS3AIsEmpty(parent, true);
+ } else if (dir.numEntries() == 0) {
+ // We do not know of any remaining entries in parent directory.
+ // However, we do not have authoritative listing, so there may
+ // still be some entries in the dir. Since we cannot know the
+ // proper state of the parent S3AFileStatus#isEmptyDirectory, we
+ // will invalidate our entries for it.
+ // Better than deleting entries would be marking them as "missing
+ // metadata". Deleting them means we lose consistent listing and
+ // ability to retry for eventual consistency for the parent path.
+
+ // TODO implement missing metadata feature
+ invalidateFileStatus(parent);
+ }
+ // else parent directory still has entries in it, isEmptyDirectory
+ // does not change
+ }
+
}
}
}
+
+ /**
+ * Invalidate any stored FileStatus's for path. Call with lock held.
+ * @param path path to invalidate
+ */
+ private void invalidateFileStatus(Path path) {
+ // TODO implement missing metadata feature
+ fileHash.remove(path);
+ Path parent = path.getParent();
+ if (parent != null) {
+ DirListingMetadata parentListing = dirHash.get(parent);
+ if (parentListing != null) {
+ parentListing.remove(path);
+ }
+ }
+ }
+
+ private void setS3AIsEmpty(Path path, boolean isEmpty) {
+ // Update any file statuses in fileHash
+ PathMetadata meta = fileHash.get(path);
+ if (meta != null) {
+ S3AFileStatus s3aStatus = (S3AFileStatus)meta.getFileStatus();
+ LOG.debug("Setting S3AFileStatus is empty dir ({}) for key {}, {}",
+ isEmpty, path, s3aStatus.getPath());
+ s3aStatus.setIsEmptyDirectory(isEmpty);
+ }
+ // Update any file statuses in dirHash
+ Path parent = path.getParent();
+ if (parent != null) {
+ DirListingMetadata dirMeta = dirHash.get(parent);
+ if (dirMeta != null) {
+ PathMetadata entry = dirMeta.get(path);
+ if (entry != null) {
+ S3AFileStatus s3aStatus = (S3AFileStatus)entry.getFileStatus();
+ LOG.debug("Setting S3AFileStatus is empty dir ({}) for key " +
+ "{}, {}", isEmpty, path, s3aStatus.getPath());
+ s3aStatus.setIsEmptyDirectory(isEmpty);
+ }
+ }
+ }
+ }
+
+ /**
+ * Return a "standardized" version of a path so we always have a consistent
+ * hash value. Also asserts the path is absolute, and contains host
+ * component.
+ * @param p input Path
+ * @return standardized version of Path, suitable for hash key
+ */
+ private Path standardize(Path p) {
+ Preconditions.checkArgument(p.isAbsolute(), "Path must be absolute");
+ URI uri = p.toUri();
+ if (uriHost != null) {
+ Preconditions.checkArgument(!isEmpty(uri.getHost()));
+ }
+ return p;
+ }
+
+ private static boolean isEmpty(String s) {
+ return (s == null || s.isEmpty());
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/MetadataStore.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/MetadataStore.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/MetadataStore.java
index 68009e3..f261fc9 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/MetadataStore.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/MetadataStore.java
@@ -89,6 +89,9 @@ public interface MetadataStore extends Closeable {
* MetadataStore. Clients provide explicit enumeration of the affected
* paths (recursively), before and after the rename.
*
+ * This operation is not atomic, unless specific implementations claim
+ * otherwise.
+ *
* On the need to provide an enumeration of directory trees instead of just
* source and destination paths:
* Since a MetadataStore does not have to track all metadata for the
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/NullMetadataStore.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/NullMetadataStore.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/NullMetadataStore.java
new file mode 100644
index 0000000..b3c9648
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/NullMetadataStore.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.s3a.s3guard;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.Collection;
+
+/**
+ * A no-op implementation of MetadataStore. Clients that use this
+ * implementation should behave the same as they would without any
+ * MetadataStore.
+ */
+public class NullMetadataStore implements MetadataStore {
+
+ @Override
+ public void initialize(FileSystem fs) throws IOException {
+ return;
+ }
+
+ @Override
+ public void close() throws IOException {
+ return;
+ }
+
+ @Override
+ public void delete(Path path) throws IOException {
+ return;
+ }
+
+ @Override
+ public void deleteSubtree(Path path) throws IOException {
+ return;
+ }
+
+ @Override
+ public PathMetadata get(Path path) throws IOException {
+ return null;
+ }
+
+ @Override
+ public DirListingMetadata listChildren(Path path) throws IOException {
+ return null;
+ }
+
+ @Override
+ public void move(Collection<Path> pathsToDelete,
+ Collection<PathMetadata> pathsToCreate) throws IOException {
+ return;
+ }
+
+ @Override
+ public void put(PathMetadata meta) throws IOException {
+ return;
+ }
+
+ @Override
+ public void put(DirListingMetadata meta) throws IOException {
+ return;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/PathMetadata.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/PathMetadata.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/PathMetadata.java
index eb2e520..d97d453 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/PathMetadata.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/PathMetadata.java
@@ -29,7 +29,7 @@ import org.apache.hadoop.fs.FileStatus;
*/
@InterfaceAudience.Private
@InterfaceStability.Evolving
-class PathMetadata {
+public class PathMetadata {
private final FileStatus fileStatus;
@@ -72,4 +72,21 @@ class PathMetadata {
"fileStatus=" + fileStatus +
'}';
}
+
+ /**
+ * Log contents to supplied StringBuilder in a pretty fashion.
+ * @param sb target StringBuilder
+ */
+ public void prettyPrint(StringBuilder sb) {
+ sb.append(String.format("%-5s %-20s %-7d",
+ fileStatus.isDirectory() ? "dir" : "file",
+ fileStatus.getPath().toString(), fileStatus.getLen()));
+ sb.append(fileStatus);
+ }
+
+ public String prettyPrint() {
+ StringBuilder sb = new StringBuilder();
+ prettyPrint(sb);
+ return sb.toString();
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/S3Guard.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/S3Guard.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/S3Guard.java
new file mode 100644
index 0000000..c998072
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/S3Guard.java
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.s3a.s3guard;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.s3a.S3AFileStatus;
+import org.apache.hadoop.fs.s3a.S3AUtils;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * Logic for integrating MetadataStore with S3A.
+ */
+final public class S3Guard {
+ private static final Logger LOG = LoggerFactory.getLogger(S3Guard.class);
+
+ /* Constants. */
+ public static final String S3_METADATA_STORE_IMPL =
+ "fs.s3a.metadatastore.impl";
+
+ // Utility class. All static functions.
+ private S3Guard() { }
+
+ /* Utility functions. */
+
+ /**
+ * Create a new instance of the configured MetadataStore.
+ * The returned MetadataStore will have been initialized via
+ * {@link MetadataStore#initialize(FileSystem)} by this function before
+ * returning it. Callers must clean up by calling
+ * {@link MetadataStore#close()} when done using the MetadataStore.
+ *
+ * If this fails to instantiate the class specified in config (fs.s3a
+ * .metadatastore.impl), or there is an error initializing it, this will log
+ * an error and return an instance of {@link NullMetadataStore} instead.
+ *
+ * @param fs FileSystem whose Configuration specifies which
+ * implementation to use.
+ * @return Reference to new MetadataStore.
+ */
+ public static MetadataStore getMetadataStore(FileSystem fs) {
+
+ Preconditions.checkNotNull(fs);
+ Configuration conf = fs.getConf();
+ Preconditions.checkNotNull(conf);
+ MetadataStore msInstance;
+ try {
+ Class<? extends MetadataStore> msClass = getMetadataStoreClass(conf);
+ msInstance = ReflectionUtils.newInstance(msClass, conf);
+ LOG.info("Using {} for {} filesystem", msClass.getSimpleName(),
+ fs.getScheme());
+ } catch (RuntimeException e) {
+ LOG.error("Failed to instantiate {}, using NullMetadataStore:",
+ conf.get(S3_METADATA_STORE_IMPL), e);
+ msInstance = new NullMetadataStore();
+ }
+ try {
+ msInstance.initialize(fs);
+ } catch (IOException ioe) {
+ LOG.error("Exception initializing MetadataStore, falling back to " +
+ "NullMetadataStore: ", ioe);
+ msInstance = new NullMetadataStore();
+ // no init needed for NullMetadataStore
+ }
+ return msInstance;
+ }
+
+ /**
+ * @param conf Configuration
+ * @return true if NullMetadataStore is configured for s3a, or if the
+ * configuration is mising.
+ */
+ public static boolean isNullMetadataStoreConfigured(Configuration conf) {
+ Class<? extends MetadataStore> msClass = getMetadataStoreClass(conf);
+ return msClass.equals(NullMetadataStore.class);
+ }
+
+ private static Class<? extends MetadataStore> getMetadataStoreClass(
+ Configuration conf) {
+ if (conf == null) {
+ return NullMetadataStore.class;
+ }
+
+ return conf.getClass(S3_METADATA_STORE_IMPL, NullMetadataStore.class,
+ MetadataStore.class);
+ }
+
+
+ /**
+ * Helper function which puts given S3AFileStatus in the MetadataStore and
+ * returns the same S3AFileStatus.
+ * @param ms MetadataStore to put() into.
+ * @param status status to store
+ * @return The same status as passed in
+ * @throws IOException if metadata store update failed
+ */
+ public static S3AFileStatus putAndReturn(MetadataStore ms,
+ S3AFileStatus status) throws IOException {
+
+ ms.put(new PathMetadata(status));
+ return status;
+ }
+
+ public static FileStatus[] dirMetaToStatuses(DirListingMetadata dirMeta) {
+ Collection<PathMetadata> listing = dirMeta.getListing();
+ FileStatus[] statuses = new FileStatus[listing.size()];
+
+ // HADOOP-13760: filter out deleted entries here
+ int i = 0;
+ for (PathMetadata pm : listing) {
+ statuses[i++] = pm.getFileStatus();
+ }
+
+ return statuses;
+ }
+
+ /**
+ * Given directory listing metadata from both the backing store, and the
+ * MetadataStore, merge the two sources of truth to create a consistent
+ * view of the current directory contents, which can be returned to clients.
+ *
+ * Also update the MetadataStore to reflect the resulting directory listing.
+ *
+ * @param ms MetadataStore to use.
+ * @param path path to directory
+ * @param backingStatuses Directory listing from the backing store.
+ * @param dirMeta Directory listing from MetadataStore. May be null.
+ * @return Final result of directory listing.
+ * @throws IOException if metadata store update failed
+ */
+ public static FileStatus[] dirListingUnion(MetadataStore ms, Path path,
+ List<FileStatus> backingStatuses, DirListingMetadata dirMeta)
+ throws IOException {
+
+ // Fast-path for NullMetadataStore
+ if (ms instanceof NullMetadataStore) {
+ return backingStatuses.toArray(new FileStatus[backingStatuses.size()]);
+ }
+
+ assertQualified(path);
+
+ if (dirMeta == null) {
+ // The metadataStore had zero state for this directory
+ dirMeta = new DirListingMetadata(path, DirListingMetadata.EMPTY_DIR,
+ false);
+ }
+
+ // Since we treat the MetadataStore as a "fresher" or "consistent" view
+ // of metadata, we always use its metadata first.
+
+ // Since the authoritative case is already handled outside this function,
+ // we will basically start with the set of directory entries in the
+ // DirListingMetadata, and add any that only exist in the backingStatuses.
+
+ // HADOOP-13760: filter out deleted files via PathMetadata#isDeleted() here
+
+ for (FileStatus s : backingStatuses) {
+
+ // Minor race condition here. Multiple threads could add to this
+ // mutable DirListingMetadata. Since it is backed by a
+ // ConcurrentHashMap, the last put() wins. I think this is ok.
+ // More concerning is two threads racing on listStatus() and delete().
+ // Any FileSystem has similar race conditions, but we could persist
+ // a stale entry longer. We could expose an atomic
+ // DirListingMetadata#putIfNotPresent()
+ if (dirMeta.get(s.getPath()) == null) {
+ dirMeta.put(s);
+ }
+ }
+
+ // TODO optimize for when allowAuthoritative = false
+ dirMeta.setAuthoritative(true); // This is the full directory contents
+ ms.put(dirMeta);
+ return dirMetaToStatuses(dirMeta);
+ }
+
+ /**
+ * Although NullMetadataStore does nothing, callers may wish to avoid work
+ * (fast path) when the NullMetadataStore is in use.
+ * @param ms The MetadataStore to test
+ * @return true iff the MetadataStore is the null, or no-op, implementation.
+ */
+ public static boolean isNullMetadataStore(MetadataStore ms) {
+ return (ms instanceof NullMetadataStore);
+ }
+
+ /**
+ * Update MetadataStore to reflect creation of the given directories.
+ * @param ms MetadataStore to update.
+ * @param dirs null, or an ordered list of directories from leaf to root.
+ * E.g. if /a/ exists, and mkdirs(/a/b/c/d) is called, this
+ * list will contain [/a/b/c/d, /a/b/c, /a/b]. /a/b/c/d is
+ * an empty, dir, and the other dirs only contain their child
+ * dir.
+ * @param owner Hadoop user name
+ */
+ public static void makeDirsOrdered(MetadataStore ms, List<Path> dirs,
+ String owner) {
+ if (dirs == null) {
+ return;
+ }
+
+ /* We discussed atomicity of this implementation.
+ * The concern is that multiple clients could race to write different
+ * cached directories to the MetadataStore. Two solutions are proposed:
+ * 1. Move mkdirs() into MetadataStore interface and let implementations
+ * ensure they are atomic.
+ * 2. Specify that the semantics of MetadataStore#putListStatus() is
+ * always additive, That is, if MetadataStore has listStatus() state
+ * for /a/b that contains [/a/b/file0, /a/b/file1], and we then call
+ * putListStatus(/a/b -> [/a/b/file2, /a/b/file3], isAuthoritative=true),
+ * then we will end up with final state of
+ * [/a/b/file0, /a/b/file1, /a/b/file2, /a/b/file3], isAuthoritative =
+ * true
+ */
+ S3AFileStatus prevStatus = null;
+ // Iterate from leaf to root
+ for (int i = 0; i < dirs.size(); i++) {
+ boolean isLeaf = (prevStatus == null);
+ Path f = dirs.get(i);
+ assertQualified(f);
+ S3AFileStatus status = new S3AFileStatus(isLeaf, f, owner);
+ Collection<PathMetadata> children;
+ if (isLeaf) {
+ children = DirListingMetadata.EMPTY_DIR;
+ } else {
+ children = new ArrayList<>(1);
+ children.add(new PathMetadata(prevStatus));
+ }
+ DirListingMetadata dirMeta = new DirListingMetadata(f, children, true);
+ try {
+ ms.put(dirMeta);
+ ms.put(new PathMetadata(status));
+ } catch (IOException ioe) {
+ LOG.error("MetadataStore#put() failure:", ioe);
+ return;
+ }
+ prevStatus = status;
+ }
+ }
+
+ /**
+ * Helper function that records the move of directory path, adding
+ * resulting metadata to the supplied lists. Does not store in MetadataStore.
+ * @param ms MetadataStore, used to make this a no-op, when it is
+ * NullMetadataStore.
+ * @param srcPaths stores the source path here
+ * @param dstMetas stores destination metadata here
+ * @param srcPath source path to store
+ * @param dstPath destination path to store
+ * @param owner Hadoop user name
+ */
+ public static void addMoveDir(MetadataStore ms, Collection<Path> srcPaths,
+ Collection<PathMetadata> dstMetas, Path srcPath, Path dstPath,
+ String owner) {
+ if (isNullMetadataStore(ms)) {
+ return;
+ }
+ assertQualified(srcPath);
+ assertQualified(dstPath);
+ S3AFileStatus dstStatus = new S3AFileStatus(true, dstPath, owner);
+ addMoveStatus(srcPaths, dstMetas, srcPath, dstStatus);
+ }
+
+ /**
+ * Like {@link #addMoveDir(MetadataStore, Collection, Collection, Path,
+ * Path, String)} (), but for files.
+ * @param ms MetadataStore, used to make this a no-op, when it is
+ * NullMetadataStore.
+ * @param srcPaths stores the source path here
+ * @param dstMetas stores destination metadata here
+ * @param srcPath source path to store
+ * @param dstPath destination path to store
+ * @param size length of file moved
+ * @param blockSize blocksize to associate with destination file
+ * @param owner Hadoop user name
+ */
+ public static void addMoveFile(MetadataStore ms, Collection<Path> srcPaths,
+ Collection<PathMetadata> dstMetas, Path srcPath, Path dstPath,
+ long size, long blockSize, String owner) {
+ if (isNullMetadataStore(ms)) {
+ return;
+ }
+ assertQualified(srcPath);
+ assertQualified(dstPath);
+ S3AFileStatus dstStatus = S3AUtils.createUploadFileStatus(dstPath, false,
+ size, blockSize, owner);
+ addMoveStatus(srcPaths, dstMetas, srcPath, dstStatus);
+ }
+
+ private static void addMoveStatus(Collection<Path> srcPaths,
+ Collection<PathMetadata> dstMetas, Path srcPath, S3AFileStatus dstStatus)
+ {
+ srcPaths.add(srcPath);
+ dstMetas.add(new PathMetadata(dstStatus));
+ }
+
+ public static void assertQualified(Path p) {
+ URI uri = p.toUri();
+ // Paths must include bucket in case MetadataStore is shared between
+ // multiple S3AFileSystem instances
+ Preconditions.checkNotNull(uri.getHost());
+
+ // I believe this should never fail, since there is a host?
+ Preconditions.checkNotNull(uri.getScheme());
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/s3guard.md
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/s3guard.md b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/s3guard.md
new file mode 100644
index 0000000..4b7c71a
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/s3guard.md
@@ -0,0 +1,26 @@
+<!---
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License. See accompanying LICENSE file.
+-->
+
+# S3Guard: Consistency and Metadata Caching for S3A
+
+## Failure Semantics
+
+Operations which modify metadata will make changes to S3 first. If, and only
+if, those operations succeed, the equivalent changes will be made to the
+MetadataStore.
+
+These changes to S3 and MetadataStore are not fully-transactional: If the S3
+operations succeed, and the subsequent MetadataStore updates fail, the S3
+changes will *not* be rolled back. In this case, an error message will be
+logged.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ACredentialsInURL.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ACredentialsInURL.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ACredentialsInURL.java
index b3d7abf..ed57d5d 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ACredentialsInURL.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ACredentialsInURL.java
@@ -119,6 +119,7 @@ public class ITestS3ACredentialsInURL extends Assert {
Configuration conf = new Configuration();
String fsname = conf.getTrimmed(TEST_FS_S3A_NAME, "");
Assume.assumeNotNull(fsname);
+ Assume.assumeFalse(S3ATestUtils.isMetadataStoreAuthoritative(conf));
URI original = new URI(fsname);
URI testURI = createUriWithEmbeddedSecrets(original, "user", "//");
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileOperationCost.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileOperationCost.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileOperationCost.java
index 7fb54b1..3e2d2b0 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileOperationCost.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileOperationCost.java
@@ -21,6 +21,7 @@ package org.apache.hadoop.fs.s3a;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.contract.ContractTestUtils;
+import org.apache.hadoop.fs.s3a.s3guard.S3Guard;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -64,7 +65,9 @@ public class ITestS3AFileOperationCost extends AbstractS3ATestBase {
resetMetricDiffs();
S3AFileStatus status = fs.getFileStatus(simpleFile);
assertTrue("not a file: " + status, status.isFile());
- metadataRequests.assertDiffEquals(1);
+ if (S3Guard.isNullMetadataStoreConfigured(fs.getConf())) {
+ metadataRequests.assertDiffEquals(1);
+ }
listRequests.assertDiffEquals(0);
}
@@ -81,7 +84,10 @@ public class ITestS3AFileOperationCost extends AbstractS3ATestBase {
resetMetricDiffs();
S3AFileStatus status = fs.getFileStatus(dir);
assertTrue("not empty: " + status, status.isEmptyDirectory());
- metadataRequests.assertDiffEquals(2);
+
+ if (S3Guard.isNullMetadataStoreConfigured(fs.getConf())) {
+ metadataRequests.assertDiffEquals(2);
+ }
listRequests.assertDiffEquals(0);
}
@@ -134,8 +140,10 @@ public class ITestS3AFileOperationCost extends AbstractS3ATestBase {
+ "\n" + ContractTestUtils.ls(fs, dir)
+ "\n" + fsState);
}
- metadataRequests.assertDiffEquals(2);
- listRequests.assertDiffEquals(1);
+ if (S3Guard.isNullMetadataStoreConfigured(fs.getConf())) {
+ metadataRequests.assertDiffEquals(2);
+ listRequests.assertDiffEquals(1);
+ }
}
@Test
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java
index 462914c..7dd4c49 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java
@@ -310,13 +310,30 @@ public final class S3ATestUtils {
* @return a path
*/
public static Path createTestPath(Path defVal) {
- String testUniqueForkId = System.getProperty(
- S3ATestConstants.TEST_UNIQUE_FORK_ID);
+ String testUniqueForkId =
+ System.getProperty(S3ATestConstants.TEST_UNIQUE_FORK_ID);
return testUniqueForkId == null ? defVal :
new Path("/" + testUniqueForkId, "test");
}
/**
+ * Is there a MetadataStore configured for s3a with authoritative enabled?
+ * @param conf Configuration to test.
+ * @return true iff there is a MetadataStore configured, and it is
+ * configured allow authoritative results. This can result in reducing
+ * round trips to S3 service for cached results, which may affect FS/FC
+ * statistics.
+ */
+ public static boolean isMetadataStoreAuthoritative(Configuration conf) {
+ if (conf == null) {
+ return Constants.DEFAULT_METADATASTORE_AUTHORITATIVE;
+ }
+ return conf.getBoolean(
+ Constants.METADATASTORE_AUTHORITATIVE,
+ Constants.DEFAULT_METADATASTORE_AUTHORITATIVE);
+ }
+
+ /**
* Reset all metrics in a list.
* @param metrics metrics to reset
*/
http://git-wip-us.apache.org/repos/asf/hadoop/blob/f4c37a57/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/fileContext/ITestS3AFileContextURI.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/fileContext/ITestS3AFileContextURI.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/fileContext/ITestS3AFileContextURI.java
index fff1fcb..62725d3 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/fileContext/ITestS3AFileContextURI.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/fileContext/ITestS3AFileContextURI.java
@@ -17,6 +17,8 @@ import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileContextURIBase;
import org.apache.hadoop.fs.s3a.S3ATestUtils;
+import org.apache.hadoop.fs.s3a.s3guard.S3Guard;
+import org.junit.Assume;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
@@ -26,9 +28,11 @@ import org.junit.Test;
*/
public class ITestS3AFileContextURI extends FileContextURIBase {
+ private Configuration conf;
+
@Before
public void setUp() throws IOException, Exception {
- Configuration conf = new Configuration();
+ conf = new Configuration();
fc1 = S3ATestUtils.createTestFileContext(conf);
fc2 = S3ATestUtils.createTestFileContext(conf); //different object, same FS
super.setUp();
@@ -41,4 +45,10 @@ public class ITestS3AFileContextURI extends FileContextURIBase {
// (the statistics tested with this method are not relevant for an S3FS)
}
+ @Test
+ @Override
+ public void testModificationTime() throws IOException {
+ Assume.assumeTrue(S3Guard.isNullMetadataStoreConfigured(conf));
+ super.testModificationTime();
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org