HADOOP-16697. Tune/audit S3A authoritative mode.

Contains:

HADOOP-16474. S3Guard ProgressiveRenameTracker to mark destination
              dirirectory as authoritative on success.
HADOOP-16684. S3guard bucket info to list a bit more about
              authoritative paths.
HADOOP-16722. S3GuardTool to support FilterFileSystem.

This patch improves the marking of newly created/import directory
trees in S3Guard DynamoDB tables as authoritative.

Specific changes:

 * Renamed directories are marked as authoritative if the entire
   operation succeeded (HADOOP-16474).
 * When updating parent table entries as part of any table write,
   there's no overwriting of their authoritative flag.

s3guard import changes:

* new -verbose flag to print out what is going on.

* The "s3guard import" command lets you declare that a directory tree
is to be marked as authoritative

  hadoop s3guard import -authoritative -verbose s3a://bucket/path

When importing a listing and a file is found, the import tool queries
the metastore and only updates the entry if the file is different from
before, where different == new timestamp, etag, or length. S3Guard can get
timestamp differences due to clock skew in PUT operations.

As the recursive list performed by the import command doesn't retrieve the
versionID, the existing entry may in fact be more complete.
When updating an existing due to clock skew the existing version ID
is propagated to the new entry (note: the etags must match; this is needed
to deal with inconsistent listings).

There is a new s3guard command to audit a s3guard bucket/path's
authoritative state:

  hadoop s3guard authoritative -check-config s3a://bucket/path

This is primarily for testing/auditing.

The s3guard bucket-info command also provides some more details on the
authoritative state of a store (HADOOP-16684).

Change-Id: I58001341c04f6f3597fcb4fcb1581ccefeb77d91

Author: steveloughran

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractRenameTest.java

@@ -204,7 +204,10 @@ public void testRenameWithNonEmptySubDir() throws Throwable {
     assertPathExists("not created in src/sub dir",
         new Path(srcSubDir, "subfile.txt"));
 
-    fs.rename(srcDir, finalDir);
+    boolean rename = fs.rename(srcDir, finalDir);
+    assertTrue("rename(" + srcDir + ", " + finalDir + ") failed",
+        rename);
+
     // Accept both POSIX rename behavior and CLI rename behavior
     if (renameRemoveEmptyDest) {
       // POSIX rename behavior

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractFSContractTestBase.java

@@ -82,6 +82,15 @@ public static void nameTestThread() {
     Thread.currentThread().setName("JUnit");
   }
 
+  @Before
+  public void nameThread() {
+    Thread.currentThread().setName("JUnit-" + getMethodName());
+  }
+
+  protected String getMethodName() {
+    return methodName.getMethodName();
+  }
+
   /**
    * This must be implemented by all instantiated test cases.
    * -provide the FS contract
@@ -172,6 +181,7 @@ protected int getTestTimeoutMillis() {
    */
   @Before
   public void setup() throws Exception {
+    Thread.currentThread().setName("setup");
     LOG.debug("== Setup ==");
     contract = createContract(createConfiguration());
     contract.init();
@@ -200,6 +210,7 @@ public void setup() throws Exception {
    */
   @After
   public void teardown() throws Exception {
+    Thread.currentThread().setName("teardown");
     LOG.debug("== Teardown ==");
     deleteTestDirInTeardown();
     LOG.debug("== Teardown complete ==");

hadoop-tools/hadoop-aws/dev-support/findbugs-exclude.xml

@@ -68,5 +68,10 @@
     <Method name="openFileWithOptions"/>
     <Bug pattern="RV_RETURN_VALUE_IGNORED_BAD_PRACTICE"/>
   </Match>
+  <Match>
+    <Class name="org.apache.hadoop.fs.s3a.s3guard.S3GuardTool$BucketInfo"/>
+    <Method name="run"/>
+    <Bug pattern="SF_SWITCH_FALLTHROUGH"/>
+  </Match>
 
 </FindBugsFilter>

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

@@ -178,6 +178,14 @@ public String getVersionId() {
     return versionId;
   }
 
+  /**
+   * set the S3 object versionId, else null.
+   * @param versionId version ID or null.
+   */
+  public void setVersionId(final String versionId) {
+    this.versionId = versionId;
+  }
+
   /** 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.

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

@@ -1245,7 +1245,8 @@ public boolean rename(Path src, Path dst) throws IOException {
     } catch (AmazonClientException e) {
       throw translateException("rename(" + src +", " + dst + ")", src, e);
     } catch (RenameFailedException e) {
-      LOG.debug(e.getMessage());
+      LOG.info("{}", e.getMessage());
+      LOG.debug("rename failure", e);
       return e.getExitCode();
     } catch (FileNotFoundException e) {
       LOG.debug(e.toString());
@@ -2477,7 +2478,7 @@ public FileStatus[] innerListStatus(Path f) throws FileNotFoundException,
    * @param path path
    * @return true if the path is auth
    */
-  protected boolean allowAuthoritative(final Path path) {
+  public boolean allowAuthoritative(final Path path) {
     return S3Guard.allowAuthoritative(path, this,
         allowAuthoritativeMetadataStore, allowAuthoritativePaths);
   }
@@ -2720,7 +2721,7 @@ S3AFileStatus innerGetFileStatus(final Path f,
                 + " s3modtime={}; msModTime={} updating metastore",
                 path, s3ModTime, msModTime);
             return S3Guard.putAndReturn(metadataStore, s3AFileStatus,
-                instrumentation, ttlTimeProvider);
+                ttlTimeProvider);
           }
         }
       }
@@ -2755,13 +2756,12 @@ S3AFileStatus innerGetFileStatus(final Path f,
       }
       // entry was found, save in S3Guard
       return S3Guard.putAndReturn(metadataStore, s3FileStatus,
-          instrumentation, ttlTimeProvider);
+          ttlTimeProvider);
     } else {
       // there was no entry in S3Guard
       // retrieve the data and update the metadata store in the process.
       return S3Guard.putAndReturn(metadataStore,
           s3GetFileStatus(path, key, probes, tombstones),
-          instrumentation,
           ttlTimeProvider);
     }
   }
@@ -3177,12 +3177,12 @@ protected synchronized void stopAllServices() {
     HadoopExecutors.shutdown(unboundedThreadPool, LOG,
         THREAD_POOL_SHUTDOWN_DELAY_SECONDS, TimeUnit.SECONDS);
     unboundedThreadPool = null;
-    closeAutocloseables(LOG, credentials);
     cleanupWithLogger(LOG,
         metadataStore,
         instrumentation,
         delegationTokens.orElse(null),
         signerManager);
+    closeAutocloseables(LOG, credentials);
     delegationTokens = Optional.empty();
     signerManager = null;
     credentials = null;
@@ -3529,13 +3529,21 @@ void finishedWrite(String key, long length, String eTag, String versionId,
           activeState = stateToClose;
         }
         S3Guard.addAncestors(metadataStore, p, ttlTimeProvider, activeState);
+        final boolean isDir = objectRepresentsDirectory(key, length);
         S3AFileStatus status = createUploadFileStatus(p,
-            S3AUtils.objectRepresentsDirectory(key, length), length,
+            isDir, length,
             getDefaultBlockSize(p), username, eTag, versionId);
-        S3Guard.putAndReturn(metadataStore, status,
-            instrumentation,
-            ttlTimeProvider,
-            activeState);
+        if (!isDir) {
+          S3Guard.putAndReturn(metadataStore, status,
+              ttlTimeProvider,
+              activeState);
+        } else {
+          // this is a directory marker so put it as such.
+          status.setIsEmptyDirectory(Tristate.TRUE);
+          S3Guard.putAuthDirectoryMarker(metadataStore, status,
+              ttlTimeProvider,
+              activeState);
+        }
       }
     } catch (IOException e) {
       if (failOnMetadataWriteError) {

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

@@ -25,6 +25,7 @@
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.fs.FileSystem.Statistics;
+import org.apache.hadoop.fs.s3a.s3guard.MetastoreInstrumentation;
 import org.apache.hadoop.metrics2.AbstractMetric;
 import org.apache.hadoop.metrics2.MetricStringBuilder;
 import org.apache.hadoop.metrics2.MetricsCollector;
@@ -188,6 +189,7 @@ public class S3AInstrumentation implements Closeable, MetricsSource {
       S3GUARD_METADATASTORE_RECORD_WRITES,
       S3GUARD_METADATASTORE_RETRY,
       S3GUARD_METADATASTORE_THROTTLED,
+      S3GUARD_METADATASTORE_AUTHORITATIVE_DIRECTORIES_UPDATED,
       STORE_IO_THROTTLED,
       DELEGATION_TOKENS_ISSUED,
       FILES_DELETE_REJECTED
@@ -562,11 +564,11 @@ public InputStreamStatistics newInputStreamStatistics() {
   }
 
   /**
-   * Create a S3Guard instrumentation instance.
+   * Create a MetastoreInstrumentation instrumentation instance.
    * There's likely to be at most one instance of this per FS instance.
    * @return the S3Guard instrumentation point.
    */
-  public S3GuardInstrumentation getS3GuardInstrumentation() {
+  public MetastoreInstrumentation getS3GuardInstrumentation() {
     return s3GuardInstrumentation;
   }
 
@@ -1127,43 +1129,35 @@ public String toString() {
   /**
    * Instrumentation exported to S3Guard.
    */
-  public final class S3GuardInstrumentation {
+  private final class S3GuardInstrumentation
+      implements MetastoreInstrumentation {
 
-    /** Initialized event. */
+    @Override
     public void initialized() {
       incrementCounter(S3GUARD_METADATASTORE_INITIALIZATION, 1);
     }
 
+    @Override
     public void storeClosed() {
 
     }
 
-    /**
-     * Throttled request.
-     */
+    @Override
     public void throttled() {
       // counters are incremented by owner.
     }
 
-    /**
-     * S3Guard is retrying after a (retryable) failure.
-     */
+    @Override
     public void retrying() {
       // counters are incremented by owner.
     }
 
-    /**
-     * Records have been read.
-     * @param count the number of records read
-     */
+    @Override
     public void recordsDeleted(int count) {
       incrementCounter(S3GUARD_METADATASTORE_RECORD_DELETES, count);
     }
 
-    /**
-     * Records have been read.
-     * @param count the number of records read
-     */
+    @Override
     public void recordsRead(int count) {
       incrementCounter(S3GUARD_METADATASTORE_RECORD_READS, count);
     }
@@ -1172,10 +1166,25 @@ public void recordsRead(int count) {
      * records have been written (including deleted).
      * @param count number of records written.
      */
+    @Override
     public void recordsWritten(int count) {
       incrementCounter(S3GUARD_METADATASTORE_RECORD_WRITES, count);
     }
 
+    @Override
+    public void directoryMarkedAuthoritative() {
+      incrementCounter(S3GUARD_METADATASTORE_AUTHORITATIVE_DIRECTORIES_UPDATED,
+          1);
+    }
+
+    @Override
+    public void entryAdded(final long durationNanos) {
+      addValueToQuantiles(
+          S3GUARD_METADATASTORE_PUT_PATH_LATENCY,
+          durationNanos);
+      incrementCounter(S3GUARD_METADATASTORE_PUT_PATH_REQUEST, 1);
+    }
+
   }
 
   /**

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

@@ -225,6 +225,9 @@ public enum Statistic {
   S3GUARD_METADATASTORE_THROTTLE_RATE(
       "s3guard_metadatastore_throttle_rate",
       "S3Guard metadata store throttle rate"),
+  S3GUARD_METADATASTORE_AUTHORITATIVE_DIRECTORIES_UPDATED(
+      "s3guard_metadatastore_authoritative_directories_updated",
+      "S3Guard metadata store authoritative directories updated from S3"),
 
   STORE_IO_THROTTLED("store_io_throttled", "Requests throttled and retried"),