refactors endpoint and region logic

Author: ahmarsuhail

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

@@ -55,6 +55,8 @@
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.s3a.impl.AWSRegionEndpointInformation;
+import org.apache.hadoop.fs.s3a.impl.AWSRegionEndpointResolver;
 import org.apache.hadoop.fs.s3a.statistics.impl.AwsStatisticsCollector;
 import org.apache.hadoop.fs.store.LogExactlyOnce;
 
@@ -85,36 +87,17 @@
  */
 @InterfaceAudience.Private
 @InterfaceStability.Unstable
-public class
-DefaultS3ClientFactory extends Configured
+public class DefaultS3ClientFactory extends Configured
     implements S3ClientFactory {
 
   private static final String REQUESTER_PAYS_HEADER_VALUE = "requester";
 
-  private static final String S3_SERVICE_NAME = "s3";
-
-  private static final Pattern VPC_ENDPOINT_PATTERN =
-          Pattern.compile("^(?:.+\\.)?([a-z0-9-]+)\\.vpce\\.amazonaws\\.(?:com|com\\.cn)$");
-
   /**
    * Subclasses refer to this.
    */
   protected static final Logger LOG =
       LoggerFactory.getLogger(DefaultS3ClientFactory.class);
 
-  /**
-   * A one-off warning of default region chains in use.
-   */
-  private static final LogExactlyOnce WARN_OF_DEFAULT_REGION_CHAIN =
-      new LogExactlyOnce(LOG);
-
-  /**
-   * Warning message printed when the SDK Region chain is in use.
-   */
-  private static final String SDK_REGION_CHAIN_IN_USE =
-      "S3A filesystem client is using"
-          + " the SDK region resolution chain.";
-
 
   /** Exactly once log to inform about ignoring the AWS-SDK Warnings for CSE. */
   private static final LogExactlyOnce IGNORE_CSE_WARN = new LogExactlyOnce(LOG);
@@ -189,7 +172,19 @@ public S3AsyncClient createS3CrtClient(URI uri, S3ClientCreationParameters param
     s3CrtAsyncClientBuilder
         .credentialsProvider(parameters.getCredentialSet());
 
-    configureCRTClientRegion(s3CrtAsyncClientBuilder, parameters, conf);
+    AWSRegionEndpointInformation regionEndpointInformation =
+        AWSRegionEndpointResolver.getEndpointRegionResolution(parameters, conf);
+
+    if (regionEndpointInformation.getRegion() == null) {
+      s3CrtAsyncClientBuilder.region(regionEndpointInformation.getRegion());
+    }
+
+    if (regionEndpointInformation.getEndpoint() == null) {
+      s3CrtAsyncClientBuilder.endpointOverride(regionEndpointInformation.getEndpoint());
+    }
+
+    s3CrtAsyncClientBuilder
+        .crossRegionAccessEnabled(regionEndpointInformation.isCrossRegionAccessEnabled());
 
     return s3CrtAsyncClientBuilder.build();
   }
@@ -218,7 +213,19 @@ private <BuilderT extends S3BaseClientBuilder<BuilderT, ClientT>, ClientT> Build
       BuilderT builder, S3ClientCreationParameters parameters, Configuration conf, String bucket)
       throws IOException {
 
-    configureEndpointAndRegion(builder, parameters, conf);
+    AWSRegionEndpointInformation regionEndpointInformation =
+        AWSRegionEndpointResolver.getEndpointRegionResolution(parameters, conf);
+
+    if(regionEndpointInformation.getRegion() != null) {
+      builder.region(regionEndpointInformation.getRegion());
+    }
+
+    if (regionEndpointInformation.getEndpoint() != null) {
+      builder.endpointOverride(regionEndpointInformation.getEndpoint());
+    }
+
+    builder.crossRegionAccessEnabled(regionEndpointInformation.isCrossRegionAccessEnabled());
+    builder.fipsEnabled(regionEndpointInformation.isFipsEnabled());
 
     maybeApplyS3AccessGrantsConfigurations(builder, conf);
 
@@ -289,216 +296,6 @@ protected ClientOverrideConfiguration.Builder createClientOverrideConfiguration(
     return clientOverrideConfigBuilder;
   }
 
-  /**
-   * Configures the region for the S3 CRT client
-   *
-   * @param builder S3 CRT client builder
-   * @param parameters parameter object
-   * @param conf conf object
-   */
-  private void configureCRTClientRegion(S3CrtAsyncClientBuilder builder, S3ClientCreationParameters parameters, Configuration conf) {
-    final String configuredRegion = parameters.getRegion();
-    Region region = null;
-    String origin = "";
-
-    // If the region was configured, set it.
-    if (configuredRegion != null && !configuredRegion.isEmpty()) {
-      origin = AWS_REGION;
-      region = Region.of(configuredRegion);
-    }
-
-    if (region != null) {
-      builder.region(region);
-    } else if (configuredRegion == null) {
-      // no region is configured, use US_EAST_2 as default.
-      region = Region.of(AWS_S3_DEFAULT_REGION);
-      builder.region(region);
-      origin = "cross region access fallback";
-    }
-
-    boolean isCrossRegionAccessEnabled = conf.getBoolean(AWS_S3_CROSS_REGION_ACCESS_ENABLED,
-        AWS_S3_CROSS_REGION_ACCESS_ENABLED_DEFAULT);
-    // s3 cross region access
-    if (isCrossRegionAccessEnabled) {
-      builder.crossRegionAccessEnabled(true);
-    }
-
-    LOG.debug("Setting region to {} from {} with cross region access {}",
-        region, origin, isCrossRegionAccessEnabled);
-  }
-
-  /**
-   * This method configures the endpoint and region for a S3 client.
-   * The order of configuration is:
-   *
-   * <ol>
-   * <li>If region is configured via fs.s3a.endpoint.region, use it.</li>
-   * <li>If endpoint is configured via via fs.s3a.endpoint, set it.
-   *     If no region is configured, try to parse region from endpoint. </li>
-   * <li> If no region is configured, and it could not be parsed from the endpoint,
-   *     set the default region as US_EAST_2</li>
-   * <li> If configured region is empty, fallback to SDK resolution chain. </li>
-   * <li> S3 cross region is enabled by default irrespective of region or endpoint
-   *      is set or not.</li>
-   * </ol>
-   *
-   * @param builder S3 client builder.
-   * @param parameters parameter object
-   * @param conf  conf configuration object
-   * @param <BuilderT> S3 client builder type
-   * @param <ClientT> S3 client type
-   * @throws IllegalArgumentException if endpoint is set when FIPS is enabled.
-   */
-  private <BuilderT extends S3BaseClientBuilder<BuilderT, ClientT>, ClientT> void configureEndpointAndRegion(
-      BuilderT builder, S3ClientCreationParameters parameters, Configuration conf) {
-    final String endpointStr = parameters.getEndpoint();
-    final URI endpoint = getS3Endpoint(endpointStr, conf);
-
-    final String configuredRegion = parameters.getRegion();
-    Region region = null;
-    String origin = "";
-
-    // If the region was configured, set it.
-    if (configuredRegion != null && !configuredRegion.isEmpty()) {
-      origin = AWS_REGION;
-      region = Region.of(configuredRegion);
-    }
-
-    // FIPs? Log it, then reject any attempt to set an endpoint
-    final boolean fipsEnabled = parameters.isFipsEnabled();
-    if (fipsEnabled) {
-      LOG.debug("Enabling FIPS mode");
-    }
-    // always setting it guarantees the value is non-null,
-    // which tests expect.
-    builder.fipsEnabled(fipsEnabled);
-
-    if (endpoint != null) {
-      boolean endpointEndsWithCentral =
-          endpointStr.endsWith(CENTRAL_ENDPOINT);
-      checkArgument(!fipsEnabled || endpointEndsWithCentral, "%s : %s",
-          ERROR_ENDPOINT_WITH_FIPS,
-          endpoint);
-
-      // No region was configured,
-      // determine the region from the endpoint.
-      if (region == null) {
-        region = getS3RegionFromEndpoint(endpointStr,
-            endpointEndsWithCentral);
-        if (region != null) {
-          origin = "endpoint";
-        }
-      }
-
-      // No need to override endpoint with "s3.amazonaws.com".
-      // Let the client take care of endpoint resolution. Overriding
-      // the endpoint with "s3.amazonaws.com" causes 400 Bad Request
-      // errors for non-existent buckets and objects.
-      // ref: https://github.com/aws/aws-sdk-java-v2/issues/4846
-      if (!endpointEndsWithCentral) {
-        builder.endpointOverride(endpoint);
-        LOG.debug("Setting endpoint to {}", endpoint);
-      } else {
-        origin = "central endpoint with cross region access";
-        LOG.debug("Enabling cross region access for endpoint {}",
-            endpointStr);
-      }
-    }
-
-    if (region != null) {
-      builder.region(region);
-    } else if (configuredRegion == null) {
-      // no region is configured, and none could be determined from the endpoint.
-      // Use US_EAST_2 as default.
-      region = Region.of(AWS_S3_DEFAULT_REGION);
-      builder.region(region);
-      origin = "cross region access fallback";
-    } else if (configuredRegion.isEmpty()) {
-      // region configuration was set to empty string.
-      // allow this if people really want it; it is OK to rely on this
-      // when deployed in EC2.
-      WARN_OF_DEFAULT_REGION_CHAIN.warn(SDK_REGION_CHAIN_IN_USE);
-      LOG.debug(SDK_REGION_CHAIN_IN_USE);
-      origin = "SDK region chain";
-    }
-    boolean isCrossRegionAccessEnabled = conf.getBoolean(AWS_S3_CROSS_REGION_ACCESS_ENABLED,
-        AWS_S3_CROSS_REGION_ACCESS_ENABLED_DEFAULT);
-    // s3 cross region access
-    if (isCrossRegionAccessEnabled) {
-      builder.crossRegionAccessEnabled(true);
-    }
-    LOG.debug("Setting region to {} from {} with cross region access {}",
-        region, origin, isCrossRegionAccessEnabled);
-  }
-
-  /**
-   * Given a endpoint string, create the endpoint URI.
-   *
-   * @param endpoint possibly null endpoint.
-   * @param conf config to build the URI from.
-   * @return an endpoint uri
-   */
-  protected static URI getS3Endpoint(String endpoint, final Configuration conf) {
-
-    boolean secureConnections = conf.getBoolean(SECURE_CONNECTIONS, DEFAULT_SECURE_CONNECTIONS);
-
-    String protocol = secureConnections ? "https" : "http";
-
-    if (endpoint == null || endpoint.isEmpty()) {
-      // don't set an endpoint if none is configured, instead let the SDK figure it out.
-      return null;
-    }
-
-    if (!endpoint.contains("://")) {
-      endpoint = String.format("%s://%s", protocol, endpoint);
-    }
-
-    try {
-      return new URI(endpoint);
-    } catch (URISyntaxException e) {
-      throw new IllegalArgumentException(e);
-    }
-  }
-
-  /**
-   * Parses the endpoint to get the region.
-   * If endpoint is the central one, use US_EAST_2.
-   *
-   * @param endpoint the configure endpoint.
-   * @param endpointEndsWithCentral true if the endpoint is configured as central.
-   * @return the S3 region, null if unable to resolve from endpoint.
-   */
-  @VisibleForTesting
-  static Region getS3RegionFromEndpoint(final String endpoint,
-      final boolean endpointEndsWithCentral) {
-
-    if (!endpointEndsWithCentral) {
-      // S3 VPC endpoint parsing
-      Matcher matcher = VPC_ENDPOINT_PATTERN.matcher(endpoint);
-      if (matcher.find()) {
-        LOG.debug("Mapping to VPCE");
-        LOG.debug("Endpoint {} is vpc endpoint; parsing region as {}", endpoint, matcher.group(1));
-        return Region.of(matcher.group(1));
-      }
-
-      LOG.debug("Endpoint {} is not the default; parsing", endpoint);
-      return AwsHostNameUtils.parseSigningRegion(endpoint, S3_SERVICE_NAME).orElse(null);
-    }
-
-    // Select default region here to enable cross-region access.
-    // If both "fs.s3a.endpoint" and "fs.s3a.endpoint.region" are empty,
-    // Spark sets "fs.s3a.endpoint" to "s3.amazonaws.com".
-    // This applies to Spark versions with the changes of SPARK-35878.
-    // ref:
-    // https://github.com/apache/spark/blob/v3.5.0/core/
-    // src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala#L528
-    // If we do not allow cross region access, Spark would not be able to
-    // access any bucket that is not present in the given region.
-    // Hence, we should use default region us-east-2 to allow cross-region
-    // access.
-    return Region.of(AWS_S3_DEFAULT_REGION);
-  }
-
   private static <BuilderT extends S3BaseClientBuilder<BuilderT, ClientT>, ClientT> void
   maybeApplyS3AccessGrantsConfigurations(BuilderT builder, Configuration conf) {
     boolean isS3AccessGrantsEnabled = conf.getBoolean(AWS_S3_ACCESS_GRANTS_ENABLED, false);

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

@@ -0,0 +1,70 @@
+package org.apache.hadoop.fs.s3a.impl;
+
+import java.net.URI;
+
+import software.amazon.awssdk.regions.Region;
+
+public class AWSRegionEndpointInformation {
+
+  private final URI endpoint;
+  private final Region region;
+  private final boolean fipsEnabled;
+  private final boolean crossRegionAccessEnabled;
+
+
+  AWSRegionEndpointInformation(Builder builder) {
+    this.endpoint = builder.endpoint;
+    this.region = builder.region;
+    this.fipsEnabled = builder.fipsEnabled;
+    this.crossRegionAccessEnabled = builder.crossRegionAccessEnabled;
+  }
+
+  public URI getEndpoint() {
+    return endpoint;
+  }
+
+  public Region getRegion() {
+    return region;
+  }
+
+  public boolean isFipsEnabled() {
+    return fipsEnabled;
+  }
+
+  public boolean isCrossRegionAccessEnabled() {
+    return crossRegionAccessEnabled;
+  }
+
+  public static class Builder {
+
+    private URI endpoint;
+    private Region region;
+    boolean fipsEnabled;
+    boolean crossRegionAccessEnabled;
+
+
+    public AWSRegionEndpointInformation build() {
+      return new AWSRegionEndpointInformation(this);
+    }
+
+    public Builder withRegion(final Region value) {
+      this.region = value;
+      return this;
+    }
+
+    public Builder withEndpoint(final URI value) {
+      this.endpoint = value;
+      return this;
+    }
+
+    public Builder fipsEnabled(final boolean value) {
+      this.fipsEnabled = value;
+      return this;
+    }
+
+    public Builder crossRegionAccessEnabled(final boolean value) {
+      this.crossRegionAccessEnabled = value;
+      return this;
+    }
+  }
+}