HADOOP-19181. S3A: IAMCredentialsProvider throttling results in auth failures (#8118)

Hard-code a 15 minute  advance time for credential renewal, so
even if a large number of processes simultaneously trying
to renew their tokens through IAM requests may trigger
throttling, the asynchronous refresh thread has minutes of
backoff and retry before the existing credentials become invalid.

+ Add a test case to explicitly instantiate the class and ask for credentials
  doesn't care about whether credentials are returned (EC2 runs) or
  if NoAwsCredentialsException is returned -any other exception is raised as
  a failure.
+ Make test suite subclass of HadoopTestBase to avoid instantiating s3a fs.

Contains HADOOP-19748: S3A: ITestAssumeRole tests failing now STS returns detailed
 error messages

Contributed by Steve Loughran

Author: steveloughran

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

@@ -20,6 +20,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.time.Duration;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -62,6 +63,12 @@ public class IAMInstanceCredentialsProvider
   private static final Logger LOG =
       LoggerFactory.getLogger(IAMInstanceCredentialsProvider.class);
 
+  /**
+   * How far in advance of credential expiry must IAM credentials be refreshed.
+   * See HADOOP-19181. S3A: IAMCredentialsProvider throttling results in AWS auth failures
+   */
+  public static final Duration TIME_BEFORE_EXPIRY = Duration.ofMinutes(5);
+
   /**
    * The credentials provider.
    * Initially a container credentials provider, but if that fails
@@ -130,8 +137,12 @@ private synchronized AwsCredentials getCredentials() {
         // close it to shut down any thread
         iamCredentialsProvider.close();
         isContainerCredentialsProvider = false;
+
+        // create an async credentials provider with a safe credential
+        // expiry time.
         iamCredentialsProvider = InstanceProfileCredentialsProvider.builder()
                 .asyncCredentialUpdateEnabled(true)
+                .staleTime(TIME_BEFORE_EXPIRY)
                 .build();
         return iamCredentialsProvider.resolveCredentials();
       } else {

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

@@ -55,11 +55,13 @@
 import org.apache.hadoop.fs.s3a.auth.CredentialProviderListFactory;
 import org.apache.hadoop.fs.s3a.auth.IAMInstanceCredentialsProvider;
 import org.apache.hadoop.fs.s3a.auth.NoAuthWithAWSException;
+import org.apache.hadoop.fs.s3a.auth.NoAwsCredentialsException;
 import org.apache.hadoop.fs.s3a.auth.ProfileAWSCredentialsProvider;
 import org.apache.hadoop.fs.s3a.auth.delegation.CountInvocationsProvider;
 import org.apache.hadoop.fs.s3a.impl.InstantiationIOException;
 import org.apache.hadoop.fs.s3a.test.PublicDatasetTestUtils;
 import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.test.HadoopTestBase;
 import org.apache.hadoop.util.Sets;
 
 import static org.apache.hadoop.fs.s3a.Constants.ASSUMED_ROLE_CREDENTIALS_PROVIDER;
@@ -79,7 +81,7 @@
 /**
  * Unit tests for {@link Constants#AWS_CREDENTIALS_PROVIDER} logic.
  */
-public class TestS3AAWSCredentialsProvider extends AbstractS3ATestBase {
+public class TestS3AAWSCredentialsProvider extends HadoopTestBase {
 
   /**
    * URI of the test file: this must be anonymously accessible.
@@ -921,4 +923,19 @@ public static AwsCredentialsProvider create() throws ClassNotFoundException {
     }
   }
 
+  @Test
+  public void testIAMInstanceCredentialsProvider() throws Throwable {
+
+    final Configuration conf =
+        createProviderConfiguration(IAMInstanceCredentialsProvider.class);
+    Path testFile = getExternalData(conf);
+    try (AWSCredentialProviderList list = createAWSCredentialProviderList(testFile.toUri(), conf)) {
+      try {
+        list.resolveCredentials();
+      } catch (NoAwsCredentialsException e) {
+        ///  this is OK.
+      }
+
+    }
+  }
 }

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

@@ -102,6 +102,8 @@ public class ITestAssumeRole extends AbstractS3ATestBase {
   private static final Statement STATEMENT_ALL_BUCKET_READ_ACCESS
       = statement(true, S3_ALL_BUCKETS, S3_BUCKET_READ_OPERATIONS);
 
+  public static final String MALFORMED_POLICY_DOCUMENT = "MalformedPolicyDocument";
+
   /**
    * test URI, built in setup.
    */
@@ -244,13 +246,13 @@ public void testAssumeRoleNoARN() throws Exception {
 
   @Test
   public void testAssumeRoleFSBadPolicy() throws Exception {
-    describe("Attemnpt to create the FS with malformed JSON");
+    describe("Attempt to create the FS with malformed JSON");
     Configuration conf = createAssumedRoleConfig();
     // add some malformed JSON
     conf.set(ASSUMED_ROLE_POLICY,  "}");
     expectFileSystemCreateFailure(conf,
         AWSBadRequestException.class,
-        "JSON");
+        MALFORMED_POLICY_DOCUMENT);
   }
 
   @Test
@@ -261,7 +263,7 @@ public void testAssumeRoleFSBadPolicy2() throws Exception {
     conf.set(ASSUMED_ROLE_POLICY, "{'json':'but not what AWS wants}");
     expectFileSystemCreateFailure(conf,
         AWSBadRequestException.class,
-        "Syntax errors in policy");
+        MALFORMED_POLICY_DOCUMENT);
   }
 
   @Test