Abfs and s3a diagnostics to included detailed analysis of the local buffer dir

Reviews state of buffer dir, parent dirs, identifies permission issues,
then tries to create a temporary file through the LocalDirAllocator, just
as the filesystems will

Author: steveloughran

src/main/java/help.java

@@ -57,10 +57,10 @@ public static void main(String[] args) {
     printCommand("committerinfo", "Print committer information");
     printCommand("deleteobject", "Delete an S3 object");
     printCommand("gcscreds", "credential diagnostics for GCS. Warning: logs secrets");
-    printCommand("iampolicy", "generate IAM policy");
+    printCommand("iampolicy", "generate AWS IAM policy");
     printCommand("listobjects", "list S3 objects and their translated statuses");
     printCommand("regions\t", "emulate region lookup of AWS SDK");
-    printCommand("sessionkeys", "optional extra: generate session keys");
+    printCommand("sessionkeys", "generate AWS session keys from a long-lived login");
     println("%nThese are only available on some builds and require a compatible hadoop release");
 
     println("");

src/main/java/org/apache/hadoop/fs/store/diag/ADLDiagnosticsInfo.java

@@ -88,9 +88,9 @@ public List<URI> listEndpointsToProbe(final Configuration conf)
 
   @Override
   protected void validateConfig(final Printout printout,
-      final Configuration conf)
+      final Configuration conf, final boolean writeOperations)
       throws IOException {
-    super.validateConfig(printout, conf);
+    super.validateConfig(printout, conf, writeOperations);
     warnOnInvalidDomain(printout, ".azuredatalakestore.net",
         "https://docs.hortonworks.com/HDPDocuments/HDP2/HDP-2.6.5/bk_cloud-data-access/content/adls-uri.html");
   }

src/main/java/org/apache/hadoop/fs/store/diag/AbfsDiagnosticsInfo.java

@@ -31,9 +31,11 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations;
 import org.apache.hadoop.fs.store.StoreDurationInfo;
 
 import static org.apache.hadoop.fs.store.diag.CapabilityKeys.*;
+import static org.apache.hadoop.fs.store.diag.OptionSets.HADOOP_TMP_DIR;
 
 /**
  * Abfs diagnostics.
@@ -47,6 +49,43 @@ public class AbfsDiagnosticsInfo extends StoreDiagnosticsInfo {
 
   public static final String FS_AZURE_READAHEADQUEUE_DEPTH = "fs.azure.readaheadqueue.depth";
 
+
+  /**
+   * Buffer directory path for uploading AbfsOutputStream data blocks.
+   * Value: {@value}
+   */
+  public static final String FS_AZURE_BLOCK_UPLOAD_BUFFER_DIR =
+      "fs.azure.buffer.dir";
+
+  /**
+   * What data block buffer to use.
+   * <br>
+   * Options include: "disk"(Default), "array", and "bytebuffer".
+   * <br>
+   * Default is {@link FileSystemConfigurations#DATA_BLOCKS_BUFFER_DEFAULT}.
+   * Value: {@value}
+   */
+  public static final String DATA_BLOCKS_BUFFER =
+      "fs.azure.data.blocks.buffer";
+
+  /**
+   * Maximum Number of blocks a single output stream can have
+   * active (uploading, or queued to the central FileSystem
+   * instance's pool of queued operations.
+   * This stops a single stream overloading the shared thread pool.
+   * {@value}
+   * <p>
+   * Default is {@link FileSystemConfigurations#BLOCK_UPLOAD_ACTIVE_BLOCKS_DEFAULT}
+   */
+  public static final String FS_AZURE_BLOCK_UPLOAD_ACTIVE_BLOCKS =
+      "fs.azure.block.upload.active.blocks";
+
+  /**
+   * Limit of queued block upload operations before writes
+   * block for an OutputStream. Value: {@value}
+   */
+  public static final int BLOCK_UPLOAD_ACTIVE_BLOCKS_DEFAULT = 20;
+
   private static final Object[][] options = {
 
       {"abfs.external.authorization.class", false, false},
@@ -81,6 +120,9 @@ public class AbfsDiagnosticsInfo extends StoreDiagnosticsInfo {
       {"fs.azure.createRemoteFileSystemDuringInitialization", false, false},
       {"fs.azure.custom.token.fetch.retry.count", false, false},
       {"fs.azure.delegation.token.provider.type", false, false},
+      {DATA_BLOCKS_BUFFER, false, false},
+      {FS_AZURE_BLOCK_UPLOAD_BUFFER_DIR, false, false},
+      {FS_AZURE_BLOCK_UPLOAD_ACTIVE_BLOCKS, false, false},
       {"fs.azure.disable.outputstream.flush", false, false},
       {"fs.azure.enable.abfslistiterator", false, false},
       {"fs.azure.enable.autothrottling", false, false},
@@ -287,7 +329,7 @@ protected void addAccountOption(
       add(list, key + "." + getFsURI().getHost(), secret, sensitive);
     }
   }
-  
+
   @Override
   public String[] getClassnames(final Configuration conf) {
     return classnames;
@@ -304,11 +346,24 @@ public String[] getOptionalPathCapabilites() {
 
   @Override
   protected void validateConfig(final Printout printout,
-      final Configuration conf)
+      final Configuration conf,
+      final boolean writeOperations)
       throws IOException {
-    super.validateConfig(printout, conf);
+    super.validateConfig(printout, conf, writeOperations);
     warnOnInvalidDomain(printout, ".dfs.core.windows.net",
         "https://docs.microsoft.com/en-us/azure/storage/data-lake-storage/introduction-abfs-uri");
+
+    // look at block buffering
+    printout.heading("Output Buffering");
+    final String buffering = conf.getTrimmed(DATA_BLOCKS_BUFFER, "disk");
+    int activeBlocks = conf.getInt(FS_AZURE_BLOCK_UPLOAD_ACTIVE_BLOCKS,
+        BLOCK_UPLOAD_ACTIVE_BLOCKS_DEFAULT);
+    printout.println("Written data is buffered to %s with up to %d blocks queued per stream",
+        buffering, activeBlocks);
+    if ("disk".equals(buffering)) {
+      validateBufferDir(printout, conf, FS_AZURE_BLOCK_UPLOAD_BUFFER_DIR, HADOOP_TMP_DIR,
+          writeOperations);
+    }
   }
 
   @Override

src/main/java/org/apache/hadoop/fs/store/diag/GCSDiagnosticsInfo.java

@@ -220,9 +220,9 @@ public List<URI> listOptionalEndpointsToProbe(final Configuration conf)
 
   @Override
   protected void validateConfig(final Printout printout,
-      final Configuration conf)
+      final Configuration conf, final boolean writeOperations)
       throws IOException {
-    super.validateConfig(printout, conf);
+    super.validateConfig(printout, conf, writeOperations);
 
     // now print everything fs.s3a.ext, assuming that
     // there are no secrets in it. Don't do that.

src/main/java/org/apache/hadoop/fs/store/diag/OptionSets.java

@@ -155,6 +155,8 @@ public class OptionSets {
       MOZILLA_PUBLIC_SUFFIX_LIST
   };
 
+  public static final String HADOOP_TMP_DIR = "hadoop.tmp.dir";
+
 
   /**
    * The enhanced {@code openFile()} options.

src/main/java/org/apache/hadoop/fs/store/diag/S3ADiagnosticsInfo.java

@@ -18,7 +18,6 @@
 
 package org.apache.hadoop.fs.store.diag;
 
-import java.io.File;
 import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
@@ -33,7 +32,6 @@
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.LocalDirAllocator;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.store.StoreExitException;
 
@@ -77,11 +75,11 @@ public class S3ADiagnosticsInfo extends StoreDiagnosticsInfo {
   public static final String ASSUMED_ROLE_STS_ENDPOINT
       = "fs.s3a.assumed.role.sts.endpoint";
 
-  public static final String HADOOP_TMP_DIR = "hadoop.tmp.dir";
-
   //use a custom endpoint?
   public static final String ENDPOINT = "fs.s3a.endpoint";
+
   public static final String DEFAULT_ENDPOINT = "";
+
   public static final String REGION = "fs.s3a.endpoint.region";
 
   //Enable path style access? Overrides default virtual hosting
@@ -504,13 +502,13 @@ public Configuration patchConfigurationToInitalization(
       Method m = aClass.getMethod("propagateBucketOptions",
           Configuration.class,
           String.class);
-      return (Configuration)m.invoke(null, conf, getFsURI().getHost());
+      return (Configuration) m.invoke(null, conf, getFsURI().getHost());
     } catch (ClassNotFoundException e) {
       LOG.error("S3AUtils not found: hadoop-aws is not on the classpath", e);
       // this will carry on elsewhere
     } catch (NoSuchMethodException
-        | IllegalAccessException
-        | InvocationTargetException e) {
+             | IllegalAccessException
+             | InvocationTargetException e) {
       LOG.info("S3AUtils.propagateBucketOptions() not found; assume old Hadoop version");
     }
     return conf;
@@ -583,22 +581,15 @@ public List<URI> listOptionalEndpointsToProbe(final Configuration conf)
 
   @Override
   protected void validateConfig(final Printout printout,
-      final Configuration conf) throws IOException {
-    String bufferOption = conf.get(BUFFER_DIR) != null
-        ? BUFFER_DIR : HADOOP_TMP_DIR;
+      final Configuration conf,
+      final boolean writeOperations) throws IOException {
     printout.heading("S3A Config validation");
 
-    printout.println("Buffer configuration option %s = %s",
-        bufferOption, conf.get(bufferOption));
+    printout.heading("Output Buffering");
+    validateBufferDir(printout, conf, BUFFER_DIR, OptionSets.HADOOP_TMP_DIR,
+        writeOperations);
 
-    final LocalDirAllocator directoryAllocator = new LocalDirAllocator(
-        bufferOption);
-
-    File temp = directoryAllocator.createTmpFileForWrite("temp", 1, conf);
-
-    printout.println("Temporary files created in %s",
-        temp.getParentFile());
-    temp.delete();
+    printout.heading("Encryption");
 
     String encryption =
         conf.get("fs.s3a.server-side-encryption-algorithm", "").trim();
@@ -649,7 +640,8 @@ protected void validateConfig(final Printout printout,
     if (endpoint.startsWith("https:") || endpoint.startsWith("http:")) {
       printout.warn("Value of %s looks like a URL: %s", ENDPOINT, endpoint);
       printout.println("It SHOULD normally be a hostname or IP address");
-      printout.println("Unless you have a private store with a non-standard port or are using AWS S3 PrivateLink");
+      printout.println("Unless you have a private store with a non-standard port or"
+              + " are using AWS S3 PrivateLink");
       if (!pathStyleAccess) {
         printout.warn("You should probably set %s to true", PATH_STYLE_ACCESS);
       }
@@ -701,7 +693,8 @@ protected void validateConfig(final Printout printout,
 
     if (!dtbinding.isEmpty()) {
       printout.println("Delegation token binding %s is active", dtbinding);
-      printout.println("This will take over authentication from the settings in %s", AWS_CREDENTIALS_PROVIDER);
+      printout.println("This will take over authentication from the settings in %s",
+          AWS_CREDENTIALS_PROVIDER);
     } else {
       // TODO: analyse default values.
     }
@@ -752,14 +745,13 @@ protected void performanceHints(
         DIRECTORY_MARKER_RETENTION);
     hint(printout,
         "org.apache.hadoop.fs.s3a.s3guard.DynamoDBMetadataStore"
-            .equals(conf.get("fs.s3a.metadatastore.impl","")),
+            .equals(conf.get("fs.s3a.metadatastore.impl", "")),
         "S3Guard is no longer needed -decommission it");
 
 
     reviewReadPolicy(printout, conf);
 
 
-
     // TODO look at output buffer options
   }
 

src/main/java/org/apache/hadoop/fs/store/diag/StoreDiag.java

@@ -194,6 +194,8 @@ public int run(String[] args, PrintStream stream) throws Exception {
 
     // and its FS URI
     storeInfo = bindToStore(path.toUri());
+    final boolean writeOperations = hasOption(WRITE);
+
     printHadoopVersionInfo();
     printOSVersion();
     if (hasOption(SYSPROPS)) {
@@ -217,7 +219,7 @@ public int run(String[] args, PrintStream stream) throws Exception {
     printStoreConfiguration();
     probeRequiredAndOptionalClasses(hasOption(OPTIONAL));
     probeRequiredAndOptionalResources();
-    storeInfo.validateConfig(this, getConf());
+    storeInfo.validateConfig(this, getConf(), writeOperations);
     printPerformanceHints();
     probeForFileSystemClass(storeInfo.getScheme());
     if (storeInfo.printTLSInfo()) {
@@ -226,7 +228,7 @@ public int run(String[] args, PrintStream stream) throws Exception {
     probeAllEndpoints();
 
     // and the filesystem operations
-    executeFileSystemOperations(path, hasOption(WRITE));
+    executeFileSystemOperations(path, writeOperations);
 
     // dump JVM status
     printJVMStats();