HADOOP-13551. AWS metrics wire-up (#2778)

Moves to the builder API for AWS S3 client creation, and
offers a similar style of API to the S3A FileSystem and tests, hiding
the details of which options are client, which are in AWS Conf,
and doing the wiring up of S3A statistics interfaces to the AWS
SDK internals. S3A Statistics, including IOStatistics, should now
count throttling events handled in the AWS SDK itself.

This patch restores endpoint determination by probes to US-East-1
if the client isn't configured with fs.s3a.endpoint.

Explicitly setting the endpoint will save the cost of these probe
HTTP requests.

Contributed by Steve Loughran.

Author: steveloughran

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

@@ -160,14 +160,33 @@ private Constants() {
       DEFAULT_SSL_CHANNEL_MODE =
           DelegatingSSLSocketFactory.SSLChannelMode.Default_JSSE;
 
-  //use a custom endpoint?
+  /**
+   * Endpoint. For v4 signing and/or better performance,
+   * this should be the specific endpoint of the region
+   * in which the bucket is hosted.
+   */
   public static final String ENDPOINT = "fs.s3a.endpoint";
 
   /**
-   * Default value of s3 endpoint. If not set explicitly using
-   * {@code AmazonS3#setEndpoint()}, this is used.
+   * Default value of s3 endpoint: {@value}.
+   * It tells the AWS client to work it out by asking the central
+   * endpoint where the bucket lives; caching that
+   * value in the client for the life of the process.
+   * <p>
+   * Note: previously this constant was defined as
+   * {@link #CENTRAL_ENDPOINT}, however the actual
+   * S3A client code used "" as the default when
+   * {@link #ENDPOINT} was unset.
+   * As core-default.xml also set the endpoint to "",
+   * the empty string has long been the <i>real</i>
+   * default value.
+   */
+  public static final String DEFAULT_ENDPOINT = "";
+
+  /**
+   * The central endpoint :{@value}.
    */
-  public static final String DEFAULT_ENDPOINT = "s3.amazonaws.com";
+  public static final String CENTRAL_ENDPOINT = "s3.amazonaws.com";
 
   //Enable path style access? Overrides default virtual hosting
   public static final String PATH_STYLE_ACCESS = "fs.s3a.path.style.access";

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

@@ -22,9 +22,8 @@
 import java.net.URI;
 
 import com.amazonaws.ClientConfiguration;
-import com.amazonaws.auth.AWSCredentialsProvider;
 import com.amazonaws.client.builder.AwsClientBuilder;
-import com.amazonaws.metrics.RequestMetricCollector;
+import com.amazonaws.handlers.RequestHandler2;
 import com.amazonaws.services.s3.AmazonS3;
 import com.amazonaws.services.s3.AmazonS3Client;
 import com.amazonaws.services.s3.AmazonS3ClientBuilder;
@@ -41,27 +40,22 @@
 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;
 import static org.apache.hadoop.fs.s3a.Constants.EXPERIMENTAL_AWS_INTERNAL_THROTTLING_DEFAULT;
-import static org.apache.hadoop.fs.s3a.Constants.PATH_STYLE_ACCESS;
 
 /**
  * The default {@link S3ClientFactory} implementation.
  * This calls the AWS SDK to configure and create an
- * {@link AmazonS3Client} that communicates with the S3 service.
+ * {@code AmazonS3Client} that communicates with the S3 service.
  */
 @InterfaceAudience.Private
 @InterfaceStability.Unstable
 public class DefaultS3ClientFactory extends Configured
     implements S3ClientFactory {
 
   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.
@@ -70,22 +64,21 @@ public class DefaultS3ClientFactory extends Configured
       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>
+   * Create the client by preparing the AwsConf configuration
+   * and then invoking {@code buildAmazonS3Client()}.
    */
   @Override
-  public AmazonS3 createS3Client(URI name,
-      final String bucket,
-      final AWSCredentialsProvider credentials,
-      final String userAgentSuffix,
-      final StatisticsFromAwsSdk statisticsFromAwsSdk) throws IOException {
+  public AmazonS3 createS3Client(
+      final URI uri,
+      final S3ClientCreationParameters parameters) throws IOException {
     Configuration conf = getConf();
     final ClientConfiguration awsConf = S3AUtils
-        .createAwsConf(conf, bucket, Constants.AWS_SERVICE_IDENTIFIER_S3);
+        .createAwsConf(conf,
+            uri.getHost(),
+            Constants.AWS_SERVICE_IDENTIFIER_S3);
+    // add any headers
+    parameters.getHeaders().forEach((h, v) ->
+        awsConf.addHeader(h, v));
 
     // When EXPERIMENTAL_AWS_INTERNAL_THROTTLING is false
     // throttling is explicitly disabled on the S3 client so that
@@ -96,111 +89,62 @@ public AmazonS3 createS3Client(URI name,
         conf.getBoolean(EXPERIMENTAL_AWS_INTERNAL_THROTTLING,
             EXPERIMENTAL_AWS_INTERNAL_THROTTLING_DEFAULT));
 
-    if (!StringUtils.isEmpty(userAgentSuffix)) {
-      awsConf.setUserAgentSuffix(userAgentSuffix);
+    if (!StringUtils.isEmpty(parameters.getUserAgentSuffix())) {
+      awsConf.setUserAgentSuffix(parameters.getUserAgentSuffix());
     }
-    // optional metrics
-    RequestMetricCollector metrics = statisticsFromAwsSdk != null
-        ? new AwsStatisticsCollector(statisticsFromAwsSdk)
-        : null;
 
-    return newAmazonS3Client(
-        credentials,
+    return buildAmazonS3Client(
         awsConf,
-        metrics,
-        conf.getTrimmed(ENDPOINT, ""),
-        conf.getBoolean(PATH_STYLE_ACCESS, false));
-  }
-
-  /**
-   * Create an {@link AmazonS3} client.
-   * Override this to provide an extended version of the client
-   * @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
-   */
-  protected AmazonS3 newAmazonS3Client(
-      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);
-    }
+        parameters);
   }
 
   /**
-   * Use the (newer) Builder SDK to create a an AWS S3 client.
+   * Use the Builder API to create 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
+   * This has a more complex endpoint configuration mechanism
+   * which initially caused problems; the
+   * {@code withForceGlobalBucketAccessEnabled(true)}
+   * command is critical here.
    * @param awsConf  AWS configuration
-   * @param metrics metrics collector or null
-   * @param endpoint endpoint string; may be ""
-   * @param pathStyleAccess enable path style access?
+   * @param parameters parameters
    * @return new AmazonS3 client
    */
-  private AmazonS3 buildAmazonS3Client(
-      final AWSCredentialsProvider credentials,
+  protected AmazonS3 buildAmazonS3Client(
       final ClientConfiguration awsConf,
-      final RequestMetricCollector metrics,
-      final String endpoint,
-      final boolean pathStyleAccess) {
+      final S3ClientCreationParameters parameters) {
     AmazonS3ClientBuilder b = AmazonS3Client.builder();
-    b.withCredentials(credentials);
+    b.withCredentials(parameters.getCredentialSet());
     b.withClientConfiguration(awsConf);
-    b.withPathStyleAccessEnabled(pathStyleAccess);
-    if (metrics != null) {
-      b.withMetricsCollector(metrics);
+    b.withPathStyleAccessEnabled(parameters.isPathStyleAccess());
+
+    if (parameters.getMetrics() != null) {
+      b.withMetricsCollector(
+          new AwsStatisticsCollector(parameters.getMetrics()));
+    }
+    if (parameters.getRequestHandlers() != null) {
+      b.withRequestHandlers(
+          parameters.getRequestHandlers().toArray(new RequestHandler2[0]));
+    }
+    if (parameters.getMonitoringListener() != null) {
+      b.withMonitoringListener(parameters.getMonitoringListener());
     }
 
     // endpoint set up is a PITA
-    //  client.setEndpoint("") is no longer available
     AwsClientBuilder.EndpointConfiguration epr
-        = createEndpointConfiguration(endpoint, awsConf);
+        = createEndpointConfiguration(parameters.getEndpoint(),
+        awsConf);
     if (epr != null) {
       // an endpoint binding was constructed: use it.
       b.withEndpointConfiguration(epr);
+    } else {
+      // no idea what the endpoint is, so tell the SDK
+      // to work it out at the cost of an extra HEAD request
+      b.withForceGlobalBucketAccessEnabled(true);
     }
     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>
@@ -226,31 +170,6 @@ protected static AmazonS3 configureAmazonS3Client(AmazonS3 s3,
         throw new IllegalArgumentException(msg, e);
       }
     }
-    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 pathStyleAccess enable path style access?
-   * @return the S3 client
-   */
-  protected static AmazonS3 applyS3ClientOptions(AmazonS3 s3,
-      final boolean pathStyleAccess) {
     if (pathStyleAccess) {
       LOG.debug("Enabling path style access!");
       s3.setS3ClientOptions(S3ClientOptions.builder()

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

@@ -19,8 +19,6 @@
 package org.apache.hadoop.fs.s3a;
 
 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;
@@ -31,31 +29,25 @@
  * This client is for testing <i>only</i>; it is in the production
  * {@code hadoop-aws} module to enable integration tests to use this
  * just by editing the Hadoop configuration used to bring up the client.
+ *
+ * The factory uses the older constructor-based instantiation/configuration
+ * of the client, so does not wire up metrics, handlers etc.
  */
 @InterfaceAudience.Private
 @InterfaceStability.Unstable
 public class InconsistentS3ClientFactory extends DefaultS3ClientFactory {
 
-  /**
-   * Create the inconsistent client.
-   * 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,
-      final RequestMetricCollector metrics,
-      final String endpoint,
-      final boolean pathStyleAccess) {
+  protected AmazonS3 buildAmazonS3Client(
+      final ClientConfiguration awsConf,
+      final S3ClientCreationParameters parameters) {
     LOG.warn("** FAILURE INJECTION ENABLED.  Do not run in production! **");
     InconsistentAmazonS3Client s3
-        = new InconsistentAmazonS3Client(credentials, awsConf, getConf());
-    configureAmazonS3Client(s3, endpoint, pathStyleAccess);
+        = new InconsistentAmazonS3Client(
+            parameters.getCredentialSet(), awsConf, getConf());
+    configureAmazonS3Client(s3,
+        parameters.getEndpoint(),
+        parameters.isPathStyleAccess());
     return s3;
   }
 }