HADOOP-17271. S3A connector to support IOStatistics. (#2580)

S3A connector to support the IOStatistics API of HADOOP-16830,

This is a major rework of the S3A Statistics collection to

* Embrace the IOStatistics APIs
* Move from direct references of S3AInstrumention statistics
  collectors to interface/implementation classes in new packages.
* Ubiquitous support of IOStatistics, including:
  S3AFileSystem, input and output streams, RemoteIterator instances
  provided in list calls.
* Adoption of new statistic names from hadoop-common

Regarding statistic collection, as well as all existing
statistics, the connector now records min/max/mean durations
of HTTP GET and HEAD requests, and those of LIST operations.

Contributed by Steve Loughran.

Author: steveloughran

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

@@ -74,4 +74,14 @@
     <Bug pattern="SF_SWITCH_FALLTHROUGH"/>
   </Match>
 
+  <!--
+  Some of the S3A Instrumentation classes increment volatile references from
+  within synchronized contexts; they use volatile to keep the cost
+  of these updates and reading them down.
+   -->
+  <Match>
+    <Class name="org.apache.hadoop.fs.s3a.S3AInstrumentation$InputStreamStatisticsImpl"/>
+    <Bug pattern="VO_VOLATILE_INCREMENT"/>
+  </Match>
+
 </FindBugsFilter>

hadoop-tools/hadoop-aws/pom.xml

@@ -201,6 +201,8 @@
                     <exclude>**/ITestMarkerToolRootOperations.java</exclude>
                     <!-- operations across the metastore -->
                     <exclude>**/ITestS3GuardDDBRootOperations.java</exclude>
+                    <!-- leave this until the end for better statistics -->
+                    <exclude>**/ITestAggregateIOStatistics.java</exclude>
                   </excludes>
                 </configuration>
               </execution>
@@ -250,6 +252,8 @@
                     <!-- operations across the metastore -->
                     <include>**/ITestS3AContractRootDir.java</include>
                     <include>**/ITestS3GuardDDBRootOperations.java</include>
+                    <!-- leave this until the end for better statistics -->
+                    <include>**/ITestAggregateIOStatistics.java</include>
                   </includes>
                 </configuration>
               </execution>

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

@@ -549,6 +549,13 @@ private Constants() {
   @InterfaceStability.Unstable
   public static final String INPUT_FADV_RANDOM = "random";
 
+  /**
+   * Gauge name for the input policy : {@value}.
+   * This references an enum currently exclusive to the S3A stream.
+   */
+  public static final String STREAM_READ_GAUGE_INPUT_POLICY =
+      "stream_read_gauge_input_policy";
+
   @InterfaceAudience.Private
   @InterfaceStability.Unstable
   public static final String S3_CLIENT_FACTORY_IMPL =

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

@@ -23,16 +23,26 @@
 
 import com.amazonaws.ClientConfiguration;
 import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.client.builder.AwsClientBuilder;
+import com.amazonaws.metrics.RequestMetricCollector;
 import com.amazonaws.services.s3.AmazonS3;
 import com.amazonaws.services.s3.AmazonS3Client;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
 import com.amazonaws.services.s3.S3ClientOptions;
+import com.amazonaws.services.s3.internal.ServiceUtils;
+import com.amazonaws.util.AwsHostNameUtils;
+import com.amazonaws.util.RuntimeHttpUtils;
+import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
 import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.s3a.statistics.StatisticsFromAwsSdk;
+import org.apache.hadoop.fs.s3a.statistics.impl.AwsStatisticsCollector;
 
 import static org.apache.hadoop.fs.s3a.Constants.EXPERIMENTAL_AWS_INTERNAL_THROTTLING;
 import static org.apache.hadoop.fs.s3a.Constants.ENDPOINT;
@@ -41,21 +51,38 @@
 
 /**
  * The default {@link S3ClientFactory} implementation.
- * This which calls the AWS SDK to configure and create an
+ * This calls the AWS SDK to configure and create an
  * {@link AmazonS3Client} that communicates with the S3 service.
  */
 @InterfaceAudience.Private
 @InterfaceStability.Unstable
 public class DefaultS3ClientFactory extends Configured
     implements S3ClientFactory {
 
-  protected static final Logger LOG = S3AFileSystem.LOG;
+  private static final String S3_SERVICE_NAME = "s3";
+  private static final String S3_SIGNER = "S3SignerType";
+  private static final String S3_V4_SIGNER = "AWSS3V4SignerType";
 
+  /**
+   * Subclasses refer to this.
+   */
+  protected static final Logger LOG =
+      LoggerFactory.getLogger(DefaultS3ClientFactory.class);
+
+  /**
+   * Create the client.
+   * <p>
+   * If the AWS stats are not null then a {@link AwsStatisticsCollector}.
+   * is created to bind to the two.
+   * <i>Important: until this binding works properly across regions,
+   * this should be null.</i>
+   */
   @Override
   public AmazonS3 createS3Client(URI name,
       final String bucket,
       final AWSCredentialsProvider credentials,
-      final String userAgentSuffix) throws IOException {
+      final String userAgentSuffix,
+      final StatisticsFromAwsSdk statisticsFromAwsSdk) throws IOException {
     Configuration conf = getConf();
     final ClientConfiguration awsConf = S3AUtils
         .createAwsConf(conf, bucket, Constants.AWS_SERVICE_IDENTIFIER_S3);
@@ -72,36 +99,124 @@ public AmazonS3 createS3Client(URI name,
     if (!StringUtils.isEmpty(userAgentSuffix)) {
       awsConf.setUserAgentSuffix(userAgentSuffix);
     }
-    return configureAmazonS3Client(
-        newAmazonS3Client(credentials, awsConf), conf);
+    // optional metrics
+    RequestMetricCollector metrics = statisticsFromAwsSdk != null
+        ? new AwsStatisticsCollector(statisticsFromAwsSdk)
+        : null;
+
+    return newAmazonS3Client(
+        credentials,
+        awsConf,
+        metrics,
+        conf.getTrimmed(ENDPOINT, ""),
+        conf.getBoolean(PATH_STYLE_ACCESS, false));
   }
 
   /**
-   * Wrapper around constructor for {@link AmazonS3} client.
+   * Create an {@link AmazonS3} client.
    * Override this to provide an extended version of the client
    * @param credentials credentials to use
    * @param awsConf  AWS configuration
-   * @return  new AmazonS3 client
+   * @param metrics metrics collector or null
+   * @param endpoint endpoint string; may be ""
+   * @param pathStyleAccess enable path style access?
+   * @return new AmazonS3 client
    */
   protected AmazonS3 newAmazonS3Client(
-      AWSCredentialsProvider credentials, ClientConfiguration awsConf) {
-    return new AmazonS3Client(credentials, awsConf);
+      final AWSCredentialsProvider credentials,
+      final ClientConfiguration awsConf,
+      final RequestMetricCollector metrics,
+      final String endpoint,
+      final boolean pathStyleAccess) {
+    if (metrics != null) {
+      LOG.debug("Building S3 client using the SDK builder API");
+      return buildAmazonS3Client(credentials, awsConf, metrics, endpoint,
+          pathStyleAccess);
+    } else {
+      LOG.debug("Building S3 client using the SDK builder API");
+      return classicAmazonS3Client(credentials, awsConf, endpoint,
+          pathStyleAccess);
+    }
   }
 
   /**
-   * Configure S3 client from the Hadoop configuration.
-   *
+   * Use the (newer) Builder SDK to create a an AWS S3 client.
+   * <p>
+   * This has a more complex endpoint configuration in a
+   * way which does not yet work in this code in a way
+   * which doesn't trigger regressions. So it is only used
+   * when SDK metrics are supplied.
+   * @param credentials credentials to use
+   * @param awsConf  AWS configuration
+   * @param metrics metrics collector or null
+   * @param endpoint endpoint string; may be ""
+   * @param pathStyleAccess enable path style access?
+   * @return new AmazonS3 client
+   */
+  private AmazonS3 buildAmazonS3Client(
+      final AWSCredentialsProvider credentials,
+      final ClientConfiguration awsConf,
+      final RequestMetricCollector metrics,
+      final String endpoint,
+      final boolean pathStyleAccess) {
+    AmazonS3ClientBuilder b = AmazonS3Client.builder();
+    b.withCredentials(credentials);
+    b.withClientConfiguration(awsConf);
+    b.withPathStyleAccessEnabled(pathStyleAccess);
+    if (metrics != null) {
+      b.withMetricsCollector(metrics);
+    }
+
+    // endpoint set up is a PITA
+    //  client.setEndpoint("") is no longer available
+    AwsClientBuilder.EndpointConfiguration epr
+        = createEndpointConfiguration(endpoint, awsConf);
+    if (epr != null) {
+      // an endpoint binding was constructed: use it.
+      b.withEndpointConfiguration(epr);
+    }
+    final AmazonS3 client = b.build();
+    return client;
+  }
+
+  /**
+   * Wrapper around constructor for {@link AmazonS3} client.
+   * Override this to provide an extended version of the client.
+   * <p>
+   * This uses a deprecated constructor -it is currently
+   * the only one which works for us.
+   * @param credentials credentials to use
+   * @param awsConf  AWS configuration
+   * @param endpoint endpoint string; may be ""
+   * @param pathStyleAccess enable path style access?
+   * @return new AmazonS3 client
+   */
+  @SuppressWarnings("deprecation")
+  private AmazonS3 classicAmazonS3Client(
+      AWSCredentialsProvider credentials,
+      ClientConfiguration awsConf,
+      final String endpoint,
+      final boolean pathStyleAccess) {
+    final AmazonS3 client = new AmazonS3Client(credentials, awsConf);
+    return configureAmazonS3Client(client, endpoint, pathStyleAccess);
+  }
+
+  /**
+   * Configure classic S3 client.
+   * <p>
    * This includes: endpoint, Path Access and possibly other
    * options.
    *
-   * @param conf Hadoop configuration
+   * @param s3 S3 Client.
+   * @param endPoint s3 endpoint, may be empty
+   * @param pathStyleAccess enable path style access?
    * @return S3 client
    * @throws IllegalArgumentException if misconfigured
    */
-  private static AmazonS3 configureAmazonS3Client(AmazonS3 s3,
-      Configuration conf)
+  protected static AmazonS3 configureAmazonS3Client(AmazonS3 s3,
+      final String endPoint,
+      final boolean pathStyleAccess)
       throws IllegalArgumentException {
-    String endPoint = conf.getTrimmed(ENDPOINT, "");
     if (!endPoint.isEmpty()) {
       try {
         s3.setEndpoint(endPoint);
@@ -111,31 +226,31 @@ private static AmazonS3 configureAmazonS3Client(AmazonS3 s3,
         throw new IllegalArgumentException(msg, e);
       }
     }
-    return applyS3ClientOptions(s3, conf);
+    return applyS3ClientOptions(s3, pathStyleAccess);
   }
 
   /**
    * Perform any tuning of the {@code S3ClientOptions} settings based on
    * the Hadoop configuration.
    * This is different from the general AWS configuration creation as
    * it is unique to S3 connections.
-   *
+   * <p>
    * The {@link Constants#PATH_STYLE_ACCESS} option enables path-style access
    * to S3 buckets if configured.  By default, the
    * behavior is to use virtual hosted-style access with URIs of the form
    * {@code http://bucketname.s3.amazonaws.com}
+   * <p>
    * Enabling path-style access and a
    * region-specific endpoint switches the behavior to use URIs of the form
    * {@code http://s3-eu-west-1.amazonaws.com/bucketname}.
    * It is common to use this when connecting to private S3 servers, as it
    * avoids the need to play with DNS entries.
    * @param s3 S3 client
-   * @param conf Hadoop configuration
+   * @param pathStyleAccess enable path style access?
    * @return the S3 client
    */
-  private static AmazonS3 applyS3ClientOptions(AmazonS3 s3,
-      Configuration conf) {
-    final boolean pathStyleAccess = conf.getBoolean(PATH_STYLE_ACCESS, false);
+  protected static AmazonS3 applyS3ClientOptions(AmazonS3 s3,
+      final boolean pathStyleAccess) {
     if (pathStyleAccess) {
       LOG.debug("Enabling path style access!");
       s3.setS3ClientOptions(S3ClientOptions.builder()
@@ -144,4 +259,54 @@ private static AmazonS3 applyS3ClientOptions(AmazonS3 s3,
     }
     return s3;
   }
+
+  /**
+   * Given an endpoint string, return an endpoint config, or null, if none
+   * is needed.
+   * <p>
+   * This is a pretty painful piece of code. It is trying to replicate
+   * what AwsClient.setEndpoint() does, because you can't
+   * call that setter on an AwsClient constructed via
+   * the builder, and you can't pass a metrics collector
+   * down except through the builder.
+   * <p>
+   * Note also that AWS signing is a mystery which nobody fully
+   * understands, especially given all problems surface in a
+   * "400 bad request" response, which, like all security systems,
+   * provides minimal diagnostics out of fear of leaking
+   * secrets.
+   *
+   * @param endpoint possibly null endpoint.
+   * @param awsConf config to build the URI from.
+   * @return a configuration for the S3 client builder.
+   */
+  @VisibleForTesting
+  public static AwsClientBuilder.EndpointConfiguration
+      createEndpointConfiguration(
+          final String endpoint, final ClientConfiguration awsConf) {
+    LOG.debug("Creating endpoint configuration for {}", endpoint);
+    if (endpoint == null || endpoint.isEmpty()) {
+      // the default endpoint...we should be using null at this point.
+      LOG.debug("Using default endpoint -no need to generate a configuration");
+      return null;
+    }
+
+    final URI epr = RuntimeHttpUtils.toUri(endpoint, awsConf);
+    LOG.debug("Endpoint URI = {}", epr);
+
+    String region;
+    if (!ServiceUtils.isS3USStandardEndpoint(endpoint)) {
+      LOG.debug("Endpoint {} is not the default; parsing", epr);
+      region = AwsHostNameUtils.parseRegion(
+          epr.getHost(),
+          S3_SERVICE_NAME);
+    } else {
+      // US-east, set region == null.
+      LOG.debug("Endpoint {} is the standard one; declare region as null", epr);
+      region = null;
+    }
+    LOG.debug("Region for endpoint {}, URI {} is determined as {}",
+        endpoint, epr, region);
+    return new AwsClientBuilder.EndpointConfiguration(endpoint, region);
+  }
 }

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

@@ -20,6 +20,7 @@
 
 import com.amazonaws.ClientConfiguration;
 import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.metrics.RequestMetricCollector;
 import com.amazonaws.services.s3.AmazonS3;
 
 import org.apache.hadoop.classification.InterfaceAudience;
@@ -40,12 +41,21 @@ public class InconsistentS3ClientFactory extends DefaultS3ClientFactory {
    * Logs a warning that this is being done.
    * @param credentials credentials to use
    * @param awsConf  AWS configuration
+   * @param metrics metric collector
+   * @param endpoint AWS endpoint
+   * @param pathStyleAccess should path style access be supported?
    * @return an inconsistent client.
    */
   @Override
   protected AmazonS3 newAmazonS3Client(AWSCredentialsProvider credentials,
-      ClientConfiguration awsConf) {
+      ClientConfiguration awsConf,
+      final RequestMetricCollector metrics,
+      final String endpoint,
+      final boolean pathStyleAccess) {
     LOG.warn("** FAILURE INJECTION ENABLED.  Do not run in production! **");
-    return new InconsistentAmazonS3Client(credentials, awsConf, getConf());
+    InconsistentAmazonS3Client s3
+        = new InconsistentAmazonS3Client(credentials, awsConf, getConf());
+    configureAmazonS3Client(s3, endpoint, pathStyleAccess);
+    return s3;
   }
 }