CDPD-26441. HADOOP-17705. S3A to add Config to set AWS region (apache#3020)

The option `fs.s3a.endpoint.region` can be used
to explicitly set the AWS region of a bucket.

This is needed when using AWS Private Link, as
the region cannot be automatically determined.

Contributed by Mehakmeet Singh

*Important*
This has proven to be a fairly brittle change. Consult the CDPD-26441 JIRA for details.
Test environments with the AWS CLI or run in EC2 do not pick up regressions, so
it is unlikely that test runs will identify major issues which surfaced in production.

When backported it MUST be accompanied by:

CDPD-27264. HADOOP-17771. S3AFS creation fails "Unable to find a region via the region provider chain."

This patch depends on CDPD-21205, which breaks the HBoss build/test.
Read the instructions there when cherrypicking.

Change-Id: If9b1b1d8ea3556ed4dfb2d164a8a6fa477dfc36d

Author: mehakmeet

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

@@ -1067,4 +1067,10 @@ private Constants() {
   public static final String STORE_CAPABILITY_DIRECTORY_MARKER_ACTION_DELETE
       = "fs.s3a.capability.directory.marker.action.delete";
 
+  /**
+   * AWS S3 region for the bucket. When set bypasses the construction of
+   * region through endpoint url.
+   */
+  public static final String AWS_REGION = "fs.s3a.endpoint.region";
+
 }

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

@@ -42,6 +42,7 @@
 import org.apache.hadoop.conf.Configured;
 import org.apache.hadoop.fs.s3a.statistics.impl.AwsStatisticsCollector;
 
+import static org.apache.hadoop.fs.s3a.Constants.AWS_REGION;
 import static org.apache.hadoop.fs.s3a.Constants.EXPERIMENTAL_AWS_INTERNAL_THROTTLING;
 import static org.apache.hadoop.fs.s3a.Constants.EXPERIMENTAL_AWS_INTERNAL_THROTTLING_DEFAULT;
 
@@ -132,7 +133,7 @@ protected AmazonS3 buildAmazonS3Client(
     // endpoint set up is a PITA
     AwsClientBuilder.EndpointConfiguration epr
         = createEndpointConfiguration(parameters.getEndpoint(),
-        awsConf);
+        awsConf, getConf().getTrimmed(AWS_REGION));
     if (epr != null) {
       // an endpoint binding was constructed: use it.
       b.withEndpointConfiguration(epr);
@@ -197,12 +198,14 @@ protected static AmazonS3 configureAmazonS3Client(AmazonS3 s3,
    *
    * @param endpoint possibly null endpoint.
    * @param awsConf config to build the URI from.
+   * @param awsRegion AWS S3 Region if the corresponding config is set.
    * @return a configuration for the S3 client builder.
    */
   @VisibleForTesting
   public static AwsClientBuilder.EndpointConfiguration
       createEndpointConfiguration(
-          final String endpoint, final ClientConfiguration awsConf) {
+      final String endpoint, final ClientConfiguration awsConf,
+      String awsRegion) {
     LOG.debug("Creating endpoint configuration for {}", endpoint);
     if (endpoint == null || endpoint.isEmpty()) {
       // the default endpoint...we should be using null at this point.
@@ -212,17 +215,19 @@ protected static AmazonS3 configureAmazonS3Client(AmazonS3 s3,
 
     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;
+    String region = awsRegion;
+    if (StringUtils.isBlank(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);

hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/index.md

@@ -796,6 +796,14 @@ options are covered in [Testing](./testing.md).
   </description>
 </property>
 
+<property>
+  <name>fs.s3a.endpoint.region</name>
+  <description>AWS S3 region for a bucket, which bypasses the parsing of
+ fs.s3a.endpoint to know the region. Would be helpful in avoiding errors
+ while using privateLink URL and explicitly set the bucket region.
+  </description>
+</property>
+
 <property>
   <name>fs.s3a.path.style.access</name>
   <value>false</value>

hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/troubleshooting_s3a.md

@@ -247,6 +247,32 @@ As an example, the endpoint for S3 Frankfurt is `s3.eu-central-1.amazonaws.com`:
   <value>s3.eu-central-1.amazonaws.com</value>
 </property>
 ```
+### <a name="AuthorizationHeaderMalformed"></a> "Authorization Header is Malformed"(400) exception when PrivateLink URL is used in "fs.s3a.endpoint"
+
+When [PrivateLink](https://docs.aws.amazon.com/AmazonS3/latest/userguide/privatelink-interface-endpoints.html) URL
+is used instead of standard s3a endpoint, it returns "authorization
+header is malformed" exception. So, if we set fs.s3a.endpoint=bucket.vpce
+-<some_string>.s3.ca-central-1.vpce.amazonaws.com and make s3 calls we get:
+```
+com.amazonaws.services.s3.model.AmazonS3Exception: The authorization header is malformed; the region 'vpce' is wrong; expecting 'ca-central-1'
+(Service: Amazon S3; Status Code: 400; Error Code: AuthorizationHeaderMalformed; Request ID: req-id; S3 Extended Request ID: req-id-2), S3 Extended Request ID: req-id-2:AuthorizationHeaderMalformed: The authorization
+header is malformed; the region 'vpce' is wrong; expecting 'ca-central-1' (Service: Amazon S3; Status Code: 400; Error Code: AuthorizationHeaderMalformed; Request ID: req-id;
+```
+Cause:
+
+Since, endpoint parsing is done in a way that it assumes the AWS S3 region
+would be the 2nd component of the `fs.s3a.endpoint` URL delimited by ".", in
+case of PrivateLink URL, it can't figure out the region and throws an
+authorization exception. Thus, to add support to using PrivateLink URLs we use `fs.s3a.endpoint.region`
+to set the region and bypass this parsing of `fs.s3a.endpoint`, in the case shown above to make it work we'll set the AWS
+S3 region as `ca-central-1`.
+
+```xml
+<property>
+  <name>fs.s3a.endpoint.region</name>
+  <value>ca-central-1</value>
+</property>
+```
 
 ### `Class does not implement AWSCredentialsProvider`
 

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AEndpointRegion.java

@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.s3a;
+
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.client.builder.AwsClientBuilder;
+import com.amazonaws.util.AwsHostNameUtils;
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+
+import static org.apache.hadoop.fs.s3a.Constants.AWS_REGION;
+
+/**
+ * Test to check correctness of S3A endpoint regions in
+ * {@link DefaultS3ClientFactory}.
+ */
+public class ITestS3AEndpointRegion extends AbstractS3ATestBase {
+
+  private static final String AWS_REGION_TEST = "test-region";
+  private static final String AWS_ENDPOINT_TEST = "test-endpoint";
+  private static final String AWS_ENDPOINT_TEST_WITH_REGION =
+      "test-endpoint.some-region.amazonaws.com";
+
+  /**
+   * Test to verify that setting a region with the config would bypass the
+   * construction of region from endpoint.
+   */
+  @Test
+  public void testWithRegionConfig() {
+    getFileSystem().getConf().set(AWS_REGION, AWS_REGION_TEST);
+
+    //Creating an endpoint config with a custom endpoint.
+    AwsClientBuilder.EndpointConfiguration epr = createEpr(AWS_ENDPOINT_TEST,
+        getFileSystem().getConf().getTrimmed(AWS_REGION));
+    //Checking if setting region config bypasses the endpoint region.
+    Assertions.assertThat(epr.getSigningRegion())
+        .describedAs("There is a region mismatch")
+        .isEqualTo(getFileSystem().getConf().get(AWS_REGION));
+  }
+
+  /**
+   * Test to verify that not setting the region config, would lead to using
+   * endpoint to construct the region.
+   */
+  @Test
+  public void testWithoutRegionConfig() {
+    getFileSystem().getConf().unset(AWS_REGION);
+
+    //Creating an endpoint config with a custom endpoint containing a region.
+    AwsClientBuilder.EndpointConfiguration eprRandom =
+        createEpr(AWS_ENDPOINT_TEST_WITH_REGION,
+            getFileSystem().getConf().getTrimmed(AWS_REGION));
+    String regionFromEndpoint =
+        AwsHostNameUtils
+            .parseRegionFromAwsPartitionPattern(AWS_ENDPOINT_TEST_WITH_REGION);
+    //Checking if not setting region config leads to constructing the region
+    // from endpoint.
+    Assertions.assertThat(eprRandom.getSigningRegion())
+        .describedAs("There is a region mismatch")
+        .isNotEqualTo(getFileSystem().getConf().get(AWS_REGION))
+        .isEqualTo(regionFromEndpoint);
+  }
+
+  /**
+   * Method to create EndpointConfiguration using an endpoint.
+   *
+   * @param endpoint the endpoint to be used for EndpointConfiguration creation.
+   * @return an instance of EndpointConfiguration.
+   */
+  private AwsClientBuilder.EndpointConfiguration createEpr(String endpoint,
+      String awsRegion) {
+    return DefaultS3ClientFactory.createEndpointConfiguration(endpoint,
+        new ClientConfiguration(), awsRegion);
+  }
+}

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/TestNetworkBinding.java

@@ -85,7 +85,7 @@ public void expectEndpoint(final String src,
       final boolean expectNull,
       final String expectRegion) {
     AwsClientBuilder.EndpointConfiguration epr =
-        createEndpointConfiguration(src, new ClientConfiguration());
+        createEndpointConfiguration(src, new ClientConfiguration(), src);
     String eprStr = epr == null
         ? "(empty)"
         : ("(" + epr.getServiceEndpoint() + " " + epr.getSigningRegion());