HADOOP-17244 directory marker deletion and s3guard

making sure the multi-object-delete unwinding process only
removes files and empty dir markers.

Testing is incomplete here, as the delete changes split file and
dir deletes in half, and only exec the dir delete requests after
file deletes go through.

so: need to add a test which creates a partially writeable dir tree with
only dir entries in. Not yet done.

Change-Id: If7ab07edaa57660b3fc57a567ae50ed4a82e9041

Author: steveloughran

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

@@ -2081,6 +2081,7 @@ protected void deleteObject(String key)
           DELETE_CONSIDERED_IDEMPOTENT,
           ()-> {
             incrementStatistic(OBJECT_DELETE_REQUESTS);
+            incrementStatistic(OBJECT_DELETE_OBJECTS);
             s3.deleteObject(bucket, key);
             return null;
           });
@@ -2127,9 +2128,14 @@ private void blockRootDelete(String key) throws InvalidRequestException {
   }
 
   /**
-   * Perform a bulk object delete operation.
+   * Perform a bulk object delete operation against S3; leaves S3Guard
+   * alone.
    * Increments the {@code OBJECT_DELETE_REQUESTS} and write
-   * operation statistics.
+   * operation statistics
+   * <p></p>
+   * {@code OBJECT_DELETE_OBJECTS} is updated with the actual number
+   * of objects deleted in the request.
+   * <p></p>
    * Retry policy: retry untranslated; delete considered idempotent.
    * If the request is throttled, this is logged in the throttle statistics,
    * with the counter set to the number of keys, rather than the number

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

@@ -531,7 +531,8 @@ private void asyncDeleteAction(
       throws IOException {
     List<DeleteObjectsResult.DeletedObject> deletedObjects = new ArrayList<>();
     try (DurationInfo ignored =
-             new DurationInfo(LOG, false, "Delete page of keys")) {
+             new DurationInfo(LOG, false,
+                 "Delete page of %d keys", keyList.size())) {
       DeleteObjectsResult result = null;
       List<Path> undeletedObjects = new ArrayList<>();
       if (!keyList.isEmpty()) {
@@ -541,10 +542,10 @@ private void asyncDeleteAction(
             .map(e -> e.keyVersion)
             .collect(Collectors.toList());
         LOG.debug("Deleting of {} file objects", files.size());
-        result = Invoker.once("Remove S3 Keys",
+        result = Invoker.once("Remove S3 Files",
             status.getPath().toString(),
             () -> callbacks.removeKeys(
-                keyList,
+                files,
                 false,
                 undeletedObjects,
                 state,
@@ -578,7 +579,7 @@ private void asyncDeleteAction(
         // (HADOOP-17244)
         metadataStore.deletePaths(pathList, state);
       }
-      if (auditDeletedKeys && result != null) {
+      if (auditDeletedKeys) {
         // audit the deleted keys
         if (deletedObjects.size() != keyList.size()) {
           // size mismatch

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

@@ -136,9 +136,9 @@ public Pair<List<KeyPath>, List<KeyPath>> splitUndeletedKeys(
         deleteException.getDeletedObjects().size());
     // convert the collection of keys being deleted into paths
     final List<KeyPath> pathsBeingDeleted = keysToKeyPaths(keysToDelete);
-    // Take this is list of paths
+    // Take this ist of paths
     // extract all undeleted entries contained in the exception and
-    // then removes them from the original list.
+    // then remove them from the original list.
     List<KeyPath> undeleted = removeUndeletedPaths(deleteException,
         pathsBeingDeleted,
         getStoreContext()::keyToPath);
@@ -209,15 +209,16 @@ public static List<KeyPath> convertToKeyPaths(
     List<Path> deletedPaths = new ArrayList<>();
     List<KeyPath> undeleted = outcome.getLeft();
     retainedMarkers.clear();
-    List<Path> undeletedPaths = toPathList(
-        (List<KeyPath>) undeleted);
+    List<Path> undeletedPaths = toPathList((List<KeyPath>) undeleted);
     // sort shorter keys first,
     // so that if the left key is longer than the first it is considered
     // smaller, so appears in the list first.
     // thus when we look for a dir being empty, we know it holds
     deleted.sort((l, r) -> r.getKey().length() - l.getKey().length());
 
-    // now go through and delete
+    // now go through and delete from S3Guard all paths listed in
+    // the result which are either files or directories with
+    // no children.
     deleted.forEach(kp -> {
       Path path = kp.getPath();
       try{

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

@@ -220,7 +220,7 @@ private void completeActiveCopies(String reason) throws IOException {
    *   </li>
    * </ol>
    * This method must only be called from the primary thread.
-   * @param path path to the object
+   * @param path path to the object.
    * @param key key of the object.
    */
   private void queueToDelete(Path path, String key) {
@@ -411,6 +411,7 @@ protected void recursiveDirectoryRename() throws IOException {
           destStatus.getPath());
       // Although the dir marker policy doesn't always need to do this,
       // it's simplest just to be consistent here.
+      // note: updates the metastore as well a S3.
       callbacks.deleteObjectAtPath(destStatus.getPath(), dstKey, false, null);
     }
 
@@ -422,7 +423,7 @@ protected void recursiveDirectoryRename() throws IOException {
         false);
 
     final RemoteIterator<S3ALocatedFileStatus> iterator =
-        callbacks.listFilesAndEmptyDirectories(parentPath,
+        callbacks.listFilesAndDirectoryMarkers(parentPath,
             sourceStatus,
             true,
             true);

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

@@ -21,6 +21,7 @@
 import com.amazonaws.services.s3.model.DeleteObjectsRequest;
 import com.amazonaws.services.s3.model.MultiObjectDeleteException;
 import com.google.common.collect.Lists;
+import org.assertj.core.api.Assertions;
 import org.junit.Assume;
 
 import org.apache.hadoop.conf.Configuration;
@@ -144,9 +145,10 @@ public void testMultiObjectDeleteNoPermissions() throws Throwable {
     Pair<List<KeyPath>, List<KeyPath>> pair =
         new MultiObjectDeleteSupport(fs.createStoreContext(), null)
         .splitUndeletedKeys(ex, keys);
-    assertEquals(undeleted, pair.getLeft());
+    assertEquals(undeleted, toPathList(pair.getLeft()));
     List<KeyPath> right = pair.getRight();
-    assertEquals("Wrong size for " + join(right), 1, right.size());
+    Assertions.assertThat(right)
+        .hasSize(1);
     assertEquals(markerPath, right.get(0).getPath());
   }
 

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestPartialRenamesDeletes.java

@@ -42,7 +42,6 @@
 import org.slf4j.LoggerFactory;
 
 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.contract.ContractTestUtils;
@@ -58,6 +57,7 @@
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.*;
 import static org.apache.hadoop.fs.s3a.S3AUtils.applyLocatedFiles;
 import static org.apache.hadoop.fs.s3a.Statistic.FILES_DELETE_REJECTED;
+import static org.apache.hadoop.fs.s3a.Statistic.OBJECT_DELETE_OBJECTS;
 import static org.apache.hadoop.fs.s3a.Statistic.OBJECT_DELETE_REQUESTS;
 import static org.apache.hadoop.fs.s3a.auth.RoleModel.Effects;
 import static org.apache.hadoop.fs.s3a.auth.RoleModel.Statement;
@@ -333,7 +333,8 @@ protected Configuration createConfiguration() {
         MAX_THREADS,
         MAXIMUM_CONNECTIONS,
         S3GUARD_DDB_BACKGROUND_SLEEP_MSEC_KEY,
-        DIRECTORY_MARKER_POLICY);
+        DIRECTORY_MARKER_POLICY,
+        BULK_DELETE_PAGE_SIZE);
     conf.setInt(MAX_THREADS, EXECUTOR_THREAD_COUNT);
     conf.setInt(MAXIMUM_CONNECTIONS, EXECUTOR_THREAD_COUNT * 2);
     // turn off prune delays, so as to stop scale tests creating
@@ -342,6 +343,10 @@ protected Configuration createConfiguration() {
     // use the keep policy to ensure that surplus markers exist
     // to complicate failures
     conf.set(DIRECTORY_MARKER_POLICY, DIRECTORY_MARKER_POLICY_KEEP);
+    // set the delete page size to its maximum to ensure that all
+    // entries are included in the same large delete, even on
+    // scale runs. This is needed for assertions on the result.
+    conf.setInt(BULK_DELETE_PAGE_SIZE, 1_000);
     return conf;
   }
 
@@ -482,8 +487,11 @@ public void testRenameDirFailsInDelete() throws Throwable {
 
     // create a set of files
     // this is done in parallel as it is 10x faster on a long-haul test run.
-    List<Path> createdFiles = createFiles(fs, readOnlyDir, dirDepth, fileCount,
-        dirCount);
+    List<Path> dirs = new ArrayList<>(dirCount);
+    List<Path> createdFiles = createDirsAndFiles(fs, readOnlyDir, dirDepth,
+        fileCount, dirCount,
+        new ArrayList<>(fileCount),
+        dirs);
     // are they all there?
     int expectedFileCount = createdFiles.size();
     assertFileCount("files ready to rename", roleFS,
@@ -500,26 +508,36 @@ public void testRenameDirFailsInDelete() throws Throwable {
           MultiObjectDeleteException.class, deniedException);
       final List<Path> undeleted
           = extractUndeletedPaths(mde, fs::keyToQualifiedPath);
+
+      List<Path> expectedUndeletedFiles = new ArrayList<>(createdFiles);
+      if (getFileSystem().getDirectoryMarkerPolicy()
+          .keepDirectoryMarkers(readOnlyDir)) {
+        // directory markers are being retained,
+        // so will also be in the list of undeleted files
+        expectedUndeletedFiles.addAll(dirs);
+      }
       Assertions.assertThat(undeleted)
           .as("files which could not be deleted")
-          .hasSize(expectedFileCount)
-          .containsAll(createdFiles)
-          .containsExactlyInAnyOrderElementsOf(createdFiles);
+          .containsExactlyInAnyOrderElementsOf(expectedUndeletedFiles);
     }
     LOG.info("Result of renaming read-only files is as expected",
         deniedException);
     assertFileCount("files in the source directory", roleFS,
         readOnlyDir, expectedFileCount);
     // now lets look at the destination.
-    // even with S3Guard on, we expect the destination to match that of our
+    // even with S3Guard on, we expect the destination to match that of
     // the remote state.
     // the test will exist
     describe("Verify destination directory exists");
-    FileStatus st = roleFS.getFileStatus(writableDir);
-    assertTrue("Not a directory: " + st,
-        st.isDirectory());
+    assertIsDirectory(writableDir);
     assertFileCount("files in the dest directory", roleFS,
         writableDir, expectedFileCount);
+    // all directories in the source tree must still exist,
+    // which for S3Guard means no tombstone markers were added
+    LOG.info("Verifying all directories still exist");
+    for (Path dir : dirs) {
+      assertIsDirectory(dir);
+    }
   }
 
   @Test
@@ -618,7 +636,13 @@ public void testPartialDirDelete() throws Throwable {
     S3AFileSystem fs = getFileSystem();
     StoreContext storeContext = fs.createStoreContext();
 
-    List<Path> readOnlyFiles = createFiles(fs, readOnlyDir,
+    List<Path> dirs = new ArrayList<>(dirCount);
+    List<Path> readOnlyFiles = createDirsAndFiles(
+        fs, readOnlyDir, dirDepth,
+        fileCount, dirCount,
+        new ArrayList<>(fileCount),
+        dirs);
+        createFiles(fs, readOnlyDir,
         dirDepth, fileCount, dirCount);
     List<Path> deletableFiles = createFiles(fs,
         writableDir, dirDepth, fileCount, dirCount);
@@ -642,16 +666,19 @@ public void testPartialDirDelete() throws Throwable {
     // this set can be deleted by the role FS
     MetricDiff rejectionCount = new MetricDiff(roleFS, FILES_DELETE_REJECTED);
     MetricDiff deleteVerbCount = new MetricDiff(roleFS, OBJECT_DELETE_REQUESTS);
+    MetricDiff deleteObjectCount = new MetricDiff(roleFS, OBJECT_DELETE_OBJECTS);
 
     describe("Trying to delete read only directory");
     AccessDeniedException ex = expectDeleteForbidden(readOnlyDir);
     if (multiDelete) {
       // multi-delete status checks
       extractCause(MultiObjectDeleteException.class, ex);
+      deleteVerbCount.assertDiffEquals("Wrong delete request count", 1);
+      deleteObjectCount.assertDiffEquals("Wrong count of objects in delete request",
+          readOnlyFiles.size());
       rejectionCount.assertDiffEquals("Wrong rejection count",
           readOnlyFiles.size());
-      deleteVerbCount.assertDiffEquals("Wrong delete count", 1);
-      reset(rejectionCount, deleteVerbCount);
+      reset(rejectionCount, deleteVerbCount, deleteObjectCount);
     }
     // all the files are still there? (avoid in scale test due to cost)
     if (!scaleTest) {
@@ -669,6 +696,9 @@ public void testPartialDirDelete() throws Throwable {
           mde, keyPaths, storeContext::keyToPath);
       final List<Path> undeleted = toPathList(
           undeletedKeyPaths);
+      deleteObjectCount.assertDiffEquals(
+          "Wrong count of objects in delete request",
+          allFiles.size());
       Assertions.assertThat(undeleted)
           .as("files which could not be deleted")
           .containsExactlyInAnyOrderElementsOf(readOnlyFiles);
@@ -691,7 +721,7 @@ public void testPartialDirDelete() throws Throwable {
 
     Assertions.assertThat(readOnlyListing)
         .as("ReadOnly directory " + directoryList)
-        .containsAll(readOnlyFiles);
+        .containsExactlyInAnyOrderElementsOf(readOnlyFiles);
   }
 
   /**
@@ -785,7 +815,7 @@ private static CompletableFuture<Path> put(FileSystem fs,
   }
 
   /**
-   * Parallel-touch a set of files in the destination directory.
+   * Build a set of files in a directory tree.
    * @param fs filesystem
    * @param destDir destination
    * @param depth file depth
@@ -798,10 +828,31 @@ public static List<Path> createFiles(final FileSystem fs,
       final int depth,
       final int fileCount,
       final int dirCount) throws IOException {
-    List<CompletableFuture<Path>> futures = new ArrayList<>(fileCount);
-    List<Path> paths = new ArrayList<>(fileCount);
-    List<Path> dirs = new ArrayList<>(dirCount);
+    return createDirsAndFiles(fs, destDir, depth, fileCount, dirCount,
+        new ArrayList<Path>(fileCount),
+        new ArrayList<Path>(dirCount));
+  }
+
+  /**
+   * Build a set of files in a directory tree.
+   * @param fs filesystem
+   * @param destDir destination
+   * @param depth file depth
+   * @param fileCount number of files to create.
+   * @param dirCount number of dirs to create at each level
+   * @param paths [out] list of file paths created
+   * @param dirs [out] list of directory paths created.
+   * @return the list of files created.
+   */
+  public static List<Path> createDirsAndFiles(final FileSystem fs,
+      final Path destDir,
+      final int depth,
+      final int fileCount,
+      final int dirCount,
+      final List<Path> paths,
+      final List<Path> dirs) throws IOException {
     buildPaths(paths, dirs, destDir, depth, fileCount, dirCount);
+    List<CompletableFuture<Path>> futures = new ArrayList<>(paths.size() + dirs.size());
 
     // create directories. With dir marker retention, that adds more entries
     // to cause deletion issues
@@ -817,7 +868,7 @@ public static List<Path> createFiles(final FileSystem fs,
     }
 
     try (DurationInfo ignore =
-            new DurationInfo(LOG, "Creating %d files", fileCount)) {
+            new DurationInfo(LOG, "Creating %d files", paths.size())) {
       for (Path path : paths) {
         futures.add(put(fs, path, path.getName()));
       }

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/AbstractS3ACostTest.java

@@ -151,6 +151,7 @@ public void setup() throws Exception {
             INVOCATION_COPY_FROM_LOCAL_FILE,
             OBJECT_COPY_REQUESTS,
             OBJECT_DELETE_REQUESTS,
+            OBJECT_DELETE_OBJECTS,
             OBJECT_LIST_REQUESTS,
             OBJECT_METADATA_REQUESTS,
             OBJECT_PUT_BYTES,