HADOOP-15183: more work on DDB tests cleaning up/pruning; the AncestorState failure on a file found is now optionally downgradeable to a warn.

Still unsure what to do there, as its a symptom of a problem: there's a file in the store which is a parent of another file. For prune, we can downgrade this
to a warn and then rely on FSCK to fix

Change-Id: I2f2cf9b85bbd3acd8a606e669c1b7e27a9db16f4

Author: steveloughran

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/StoreContext.java

@@ -316,4 +316,13 @@ public File createTempFile(String prefix, long size) throws IOException {
     return tempFileFactory.apply(prefix, size);
   }
 
+  /**
+   * Get the location of the bucket.
+   * @return the bucket location.
+   * @throws IOException failure.
+   */
+  public String getBucketLocation() throws IOException {
+    return getBucketLocation.apply();
+  }
+
 }

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/DynamoDBMetadataStore.java

@@ -818,7 +818,7 @@ private Collection<DDBPathMetadata> completeAncestry(
       ancestorsToAdd.add(entry);
       Path parent = path.getParent();
       while (!parent.isRoot()) {
-        if (ancestorState.findDirectory(parent)) {
+        if (ancestorState.findEntry(parent, true)) {
           break;
         }
         LOG.debug("auto-create ancestor path {} for child path {}",
@@ -1180,7 +1180,7 @@ List<DDBPathMetadata> fullPathsToPut(DDBPathMetadata meta,
     Path path = meta.getFileStatus().getPath().getParent();
     while (path != null && !path.isRoot()) {
       synchronized (ancestorState) {
-        if (ancestorState.findDirectory(path)) {
+        if (ancestorState.findEntry(path, true)) {
           break;
         }
       }
@@ -1336,6 +1336,7 @@ public void prune(long modTime) throws IOException {
   @Override
   @Retries.RetryTranslated
   public void prune(long modTime, String keyPrefix) throws IOException {
+    LOG.debug("Prune files under {} with oldes time {}", keyPrefix, modTime);
     int itemCount = 0;
     try {
       Collection<Path> deletionBatch =
@@ -2187,41 +2188,47 @@ public String toString() {
      * @param p path to check
      * @return true if the state has an entry
      */
-    public boolean contains(Path p) {
+    boolean contains(Path p) {
       return ancestry.containsKey(p);
     }
 
-    public DDBPathMetadata put(Path p, DDBPathMetadata md) {
+    DDBPathMetadata put(Path p, DDBPathMetadata md) {
       return ancestry.put(p, md);
     }
 
-    public DDBPathMetadata put(DDBPathMetadata md) {
+    DDBPathMetadata put(DDBPathMetadata md) {
       return ancestry.put(md.getFileStatus().getPath(), md);
     }
 
-    public DDBPathMetadata get(Path p) {
+    DDBPathMetadata get(Path p) {
       return ancestry.get(p);
     }
 
     /**
-     * Find a directory entry, raising an exception if there is a file
-     * at the path.
+     * Find an entry in the ancestor state, warning and optionally
+     * raising an exception if there is a file at the path.
      * @param path path to look up
+     * @param failOnFile fail if a file was found.
      * @return true iff a directory was found in the ancestor state.
      * @throws PathIOException if there was a file at the path.
      */
-    public boolean findDirectory(Path path) throws PathIOException {
+    boolean findEntry(
+        final Path path,
+        final boolean failOnFile) throws PathIOException {
       final DDBPathMetadata ancestor = get(path);
       if (ancestor != null) {
         // there's an entry in the ancestor state
         if (!ancestor.getFileStatus().isDirectory()) {
           // but: its a file, which means this update is now inconsistent.
-          throw new PathIOException(path.toString(), E_INCONSISTENT_UPDATE);
-        } else {
-          // the ancestor is a directory, so break out the ancestor creation
-          // loop.
-          return true;
+          final String message = E_INCONSISTENT_UPDATE + " entry is " + ancestor
+              .getFileStatus();
+          LOG.error(message);
+          if (failOnFile) {
+            // errors trigger failure
+            throw new PathIOException(path.toString(), message);
+          }
         }
+        return true;
       } else {
         return false;
       }

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/AbstractS3GuardToolTestBase.java

@@ -264,12 +264,13 @@ private void testPruneCommand(Configuration cmdConf, Path parent,
       String...args) throws Exception {
     Path keepParent = path("prune-cli-keep");
     StopWatch timer = new StopWatch();
+    final S3AFileSystem fs = getFileSystem();
     try {
       S3GuardTool.Prune cmd = new S3GuardTool.Prune(cmdConf);
       cmd.setMetadataStore(ms);
 
-      getFileSystem().mkdirs(parent);
-      getFileSystem().mkdirs(keepParent);
+      fs.mkdirs(parent);
+      fs.mkdirs(keepParent);
       createFile(new Path(parent, "stale"), true, true);
       createFile(new Path(keepParent, "stale-to-keep"), true, true);
 
@@ -291,8 +292,10 @@ private void testPruneCommand(Configuration cmdConf, Path parent,
       assertMetastoreListingCount(keepParent,
           "This child should have been kept (prefix restriction).", 1);
     } finally {
-      getFileSystem().delete(parent, true);
-      ms.prune(Long.MAX_VALUE);
+      fs.delete(parent, true);
+      fs.delete(keepParent, true);
+      ms.prune(Long.MAX_VALUE, fs.pathToKey(parent));
+      ms.prune(Long.MAX_VALUE, fs.pathToKey(keepParent));
     }
   }
 

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/ITestDynamoDBMetadataStore.java

@@ -263,26 +263,49 @@ public void tearDown() throws Exception {
   private void deleteAllMetadata() throws IOException {
     // The following is a way to be sure the table will be cleared and there
     // will be no leftovers after the test.
-    ThrottleTracker throttleTracker = new ThrottleTracker(ddbmsStatic);
-    final Path root = strToPath("/");
+    deleteMetadataUnderPath(ddbmsStatic, strToPath("/"), true);
+  }
+
+  /**
+   * Delete all metadata under a path.
+   * Attempt to use prune first as it scales slightly better.
+   * @param ms store
+   * @param path path to prune under
+   * @param suppressErrors should errors be suppressed?
+   * @throws IOException if there is a failure and suppressErrors == false
+   */
+  public static void deleteMetadataUnderPath(final DynamoDBMetadataStore ms,
+      final Path path, final boolean suppressErrors) throws IOException {
+    ThrottleTracker throttleTracker = new ThrottleTracker(ms);
     try (DurationInfo ignored = new DurationInfo(LOG, true, "prune")) {
-      ddbmsStatic.prune(System.currentTimeMillis(),
-          PathMetadataDynamoDBTranslation.pathToParentKey(root));
+      ms.prune(System.currentTimeMillis(),
+          PathMetadataDynamoDBTranslation.pathToParentKey(path));
       LOG.info("Throttle statistics: {}", throttleTracker);
+    } catch (IOException ioe) {
+      // prune failed. warn and then fall back to forget.
+      LOG.warn("Failed to prune {}", path, ioe);
+      if (!suppressErrors) {
+        throw ioe;
+      }
     }
     // and after the pruning, make sure all other metadata is gone
     int forgotten = 0;
     try (DurationInfo ignored = new DurationInfo(LOG, true, "forget")) {
-      PathMetadata meta = ddbmsStatic.get(root);
+      PathMetadata meta = ms.get(path);
       if (meta != null) {
-        for (DescendantsIterator desc = new DescendantsIterator(ddbmsStatic,
+        for (DescendantsIterator desc = new DescendantsIterator(ms,
             meta);
             desc.hasNext(); ) {
           forgotten++;
-          ddbmsStatic.forgetMetadata(desc.next().getPath());
+          ms.forgetMetadata(desc.next().getPath());
         }
         LOG.info("Forgot {} entries", forgotten);
       }
+    } catch (IOException ioe) {
+      LOG.warn("Failed to forget entries under {}", path, ioe);
+      if (!suppressErrors) {
+        throw ioe;
+      }
     }
     LOG.info("Throttle statistics: {}", throttleTracker);
   }
@@ -491,7 +514,9 @@ private void doBatchWriteForOneSet(int index) throws IOException {
           numNewMetas,
           getDynamoMetadataStore());
     }
-    deleteAllMetadata();
+    // The following is a way to be sure the table will be cleared and there
+    // will be no leftovers after the test.
+    deleteMetadataUnderPath(ddbmsStatic, strToPath("/"), false);
   }
 
   /**

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/ITestDynamoDBMetadataStoreScale.java

@@ -229,6 +229,8 @@ public void test_030_BatchedWrite() throws Exception {
     try {
       describe("Running %d iterations of batched put, size %d", iterations,
           BATCH_SIZE);
+      Path base = path(getMethodName());
+      final String pathKey = base.toUri().getPath();
 
       ThrottleTracker result = execute("prune",
           1,
@@ -237,7 +239,8 @@ public void test_030_BatchedWrite() throws Exception {
             ThrottleTracker tracker = new ThrottleTracker(ddbms);
             long pruneItems = 0;
             for (long i = 0; i < iterations; i++) {
-              Path longPath = pathOfDepth(BATCH_SIZE, String.valueOf(i));
+              Path longPath = pathOfDepth(BATCH_SIZE,
+                  pathKey, String.valueOf(i));
               S3AFileStatus status = basicFileStatus(longPath, 0, false,
                   12345);
               PathMetadata pm = new PathMetadata(status);
@@ -251,7 +254,7 @@ public void test_030_BatchedWrite() throws Exception {
 
               if (pruneItems == BATCH_SIZE) {
                 describe("pruning files");
-                ddbms.prune(Long.MAX_VALUE /* all files */);
+                ddbms.prune(Long.MAX_VALUE, pathKey);
                 pruneItems = 0;
               }
               if (tracker.probe()) {
@@ -365,23 +368,28 @@ public void test_080_fullPathsToPut() throws Throwable {
     Path base = new Path("s3a://example.org/test_080_fullPathsToPut");
     Path child = new Path(base, "child");
     List<PathMetadata> pms = new ArrayList<>();
-    try(BulkOperationState bulkUpdate
-            = ddbms.initiateBulkWrite(child)) {
-      ddbms.put(new PathMetadata(makeDirStatus(base)), bulkUpdate);
-      ddbms.put(new PathMetadata(makeDirStatus(child)), bulkUpdate);
-      ddbms.getInvoker().retry("set up directory tree",
-          base.toString(),
-          true,
-          () -> ddbms.put(pms, bulkUpdate));
-    }
-    try(BulkOperationState bulkUpdate
-            = ddbms.initiateBulkWrite(child)) {
-      DDBPathMetadata dirData = ddbms.get(child, true);
-      execute("put",
-          OPERATIONS_PER_THREAD,
-          expectThrottling(),
-          () -> ddbms.fullPathsToPut(dirData, bulkUpdate)
-      );
+    try {
+      try (BulkOperationState bulkUpdate
+              = ddbms.initiateBulkWrite(child)) {
+        ddbms.put(new PathMetadata(makeDirStatus(base)), bulkUpdate);
+        ddbms.put(new PathMetadata(makeDirStatus(child)), bulkUpdate);
+        ddbms.getInvoker().retry("set up directory tree",
+            base.toString(),
+            true,
+            () -> ddbms.put(pms, bulkUpdate));
+      }
+      try (BulkOperationState bulkUpdate
+              = ddbms.initiateBulkWrite(child)) {
+        DDBPathMetadata dirData = ddbms.get(child, true);
+        execute("put",
+            OPERATIONS_PER_THREAD,
+            expectThrottling(),
+            () -> ddbms.fullPathsToPut(dirData, bulkUpdate)
+        );
+      }
+    } finally {
+      ddbms.forgetMetadata(child);
+      ddbms.forgetMetadata(base);
     }
   }
 
@@ -533,22 +541,21 @@ public ThrottleTracker execute(String operation,
    * @param ms store
    * @param pm path to clean up
    */
-  private void cleanupMetadata(MetadataStore ms, PathMetadata pm) {
+  private void cleanupMetadata(DynamoDBMetadataStore ms, PathMetadata pm) {
     Path path = pm.getFileStatus().getPath();
     try {
-      ddbms.getInvoker().retry("clean up", path.toString(), true,
-          () -> ms.forgetMetadata(path));
+      ITestDynamoDBMetadataStore.deleteMetadataUnderPath(ms, path, true);
     } catch (IOException ioe) {
       // Ignore.
       LOG.info("Ignoring error while cleaning up {} in database", path, ioe);
     }
   }
 
-  private Path pathOfDepth(long n, @Nullable String fileSuffix) {
+  private Path pathOfDepth(long n,
+      String name, @Nullable String fileSuffix) {
     StringBuilder sb = new StringBuilder();
     for (long i = 0; i < n; i++) {
-      sb.append(i == 0 ? "/" + this.getClass().getSimpleName() : "lvl");
-      sb.append(i);
+      sb.append(i == 0 ? "/" + name : String.format("level-%03d", i));
       if (i == n - 1 && fileSuffix != null) {
         sb.append(fileSuffix);
       }

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/ITestS3GuardToolLocal.java

@@ -417,7 +417,8 @@ private void uploadCommandAssertCount(S3AFileSystem fs, String options[],
         String[] fields = line.split("\\s");
         if (fields.length == 4 && fields[0].equals(Uploads.TOTAL)) {
           int parsedUploads = Integer.valueOf(fields[1]);
-          LOG.debug("Matched CLI output: {} {} {} {}", fields);
+          LOG.debug("Matched CLI output: {} {} {} {}",
+              fields[0], fields[1], fields[2], fields[3]);
           assertEquals("Unexpected number of uploads", numUploads,
               parsedUploads);
           return;

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/TestDynamoDBMiscOperations.java

@@ -26,7 +26,10 @@
 import com.amazonaws.waiters.WaiterTimedOutException;
 import org.junit.Test;
 
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.PathIOException;
 import org.apache.hadoop.fs.s3a.AWSClientIOException;
+import org.apache.hadoop.fs.s3a.S3AFileStatus;
 import org.apache.hadoop.test.HadoopTestBase;
 import org.apache.hadoop.fs.Path;
 
@@ -109,4 +112,56 @@ public void testInnerListChildrenDirectoryNpe() throws Exception {
         ddbms.getDirListingMetadataFromDirMetaAndList(p, metas, dirPathMeta));
   }
 
+  @Test
+  public void testAncestorStateForDir() throws Throwable {
+    final DynamoDBMetadataStore.AncestorState ancestorState
+        = new DynamoDBMetadataStore.AncestorState(null);
+
+    // path 1 is a directory
+    final Path path1 = new Path("s3a://bucket/1");
+    final S3AFileStatus status1 = new S3AFileStatus(true,
+        path1, "hadoop");
+    final DDBPathMetadata md1 = new DDBPathMetadata(
+        status1);
+    ancestorState.put(md1);
+    assertTrue("Status not found in ancestors",
+        ancestorState.contains(path1));
+    final DDBPathMetadata result = ancestorState.get(path1);
+    assertEquals(status1, result.getFileStatus());
+    assertTrue("Lookup failed",
+        ancestorState.findEntry(path1, true));
+    final Path path2 = new Path("s3a://bucket/2");
+    assertFalse("Lookup didn't fail",
+        ancestorState.findEntry(path2, true));
+    assertFalse("Lookup didn't fail",
+        ancestorState.contains(path2));
+    assertNull("Lookup didn't fail",
+        ancestorState.get(path2));
+  }
+
+  @Test
+  public void testAncestorStateForFile() throws Throwable {
+    final DynamoDBMetadataStore.AncestorState ancestorState
+        = new DynamoDBMetadataStore.AncestorState(null);
+
+    // path 1 is a file
+    final Path path1 = new Path("s3a://bucket/1");
+    final S3AFileStatus status1 = new S3AFileStatus(
+        1024_1024_1024L,
+        0,
+        path1,
+        32_000_000,
+        "hadoop",
+        "e4",
+        "f5");
+    final DDBPathMetadata md1 = new DDBPathMetadata(
+        status1);
+    ancestorState.put(md1);
+    assertTrue("Lookup failed",
+        ancestorState.findEntry(path1, false));
+    intercept(PathIOException.class,
+        DynamoDBMetadataStore.E_INCONSISTENT_UPDATE,
+        () -> ancestorState.findEntry(path1, true));
+  }
+
 }