HADOOP-14556. S3A to support Delegation Tokens.

Contributed by Steve Loughran and Daryn Sharp.

Author: steveloughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StorageStatistics.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.fs;
 
 import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
 
 import java.util.Iterator;
 
@@ -37,9 +38,13 @@ public abstract class StorageStatistics {
    *
    * When adding new common statistic name constants, please make them unique.
    * By convention, they are implicitly unique:
-   *  - the name of the constants are uppercase, words separated by underscores.
-   *  - the value of the constants are lowercase of the constant names.
+   * <ul>
+   *   <li>the name of the constants are uppercase, words separated by
+   *   underscores.</li>
+   *   <li>the value of the constants are lowercase of the constant names.</li>
+   * </ul>
    */
+  @InterfaceStability.Evolving
   public interface CommonStatisticNames {
     // The following names are for file system operation invocations
     String OP_APPEND = "op_append";
@@ -49,6 +54,7 @@ public interface CommonStatisticNames {
     String OP_DELETE = "op_delete";
     String OP_EXISTS = "op_exists";
     String OP_GET_CONTENT_SUMMARY = "op_get_content_summary";
+    String OP_GET_DELEGATION_TOKEN = "op_get_delegation_token";
     String OP_GET_FILE_CHECKSUM = "op_get_file_checksum";
     String OP_GET_FILE_STATUS = "op_get_file_status";
     String OP_GET_STATUS = "op_get_status";

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/Job.java

@@ -42,6 +42,8 @@
 import org.apache.hadoop.mapreduce.util.ConfigUtil;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.yarn.api.records.ReservationId;
+
+import com.google.common.annotations.VisibleForTesting;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -1524,7 +1526,10 @@ public static Map<String, Boolean> getArchiveSharedCacheUploadPolicies(
     return getSharedCacheUploadPolicies(conf, false);
   }
 
-  private synchronized void connect()
+  /** Only for mocking via unit tests. */
+  @Private
+  @VisibleForTesting
+  synchronized void connect()
           throws IOException, InterruptedException, ClassNotFoundException {
     if (cluster == null) {
       cluster = 
@@ -1544,7 +1549,8 @@ boolean isConnected() {
 
   /** Only for mocking via unit tests. */
   @Private
-  public JobSubmitter getJobSubmitter(FileSystem fs, 
+  @VisibleForTesting
+  JobSubmitter getJobSubmitter(FileSystem fs,
       ClientProtocol submitClient) throws IOException {
     return new JobSubmitter(fs, submitClient);
   }

hadoop-project/pom.xml

@@ -1732,6 +1732,11 @@
           <artifactId>snakeyaml</artifactId>
           <version>${snakeyaml.version}</version>
         </dependency>
+        <dependency>
+          <groupId>org.hamcrest</groupId>
+          <artifactId>hamcrest-library</artifactId>
+          <version>1.3</version>
+        </dependency>
         <dependency>
           <groupId>org.assertj</groupId>
           <artifactId>assertj-core</artifactId>

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

@@ -21,6 +21,7 @@
 import java.io.Closeable;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -39,6 +40,7 @@
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.fs.s3a.auth.NoAuthWithAWSException;
+import org.apache.hadoop.fs.s3a.auth.NoAwsCredentialsException;
 import org.apache.hadoop.io.IOUtils;
 
 /**
@@ -52,7 +54,8 @@
  *   an {@link AmazonClientException}, that is rethrown, rather than
  *   swallowed.</li>
  *   <li>Has some more diagnostics.</li>
- *   <li>On failure, the last AmazonClientException raised is rethrown.</li>
+ *   <li>On failure, the last "relevant" AmazonClientException raised is
+ *   rethrown; exceptions other than 'no credentials' have priority.</li>
  *   <li>Special handling of {@link AnonymousAWSCredentials}.</li>
  * </ol>
  */
@@ -78,6 +81,12 @@ public class AWSCredentialProviderList implements AWSCredentialsProvider,
 
   private final AtomicBoolean closed = new AtomicBoolean(false);
 
+  /**
+   * The name, which is empty by default.
+   * Uses in the code assume if non empty there's a trailing space.
+   */
+  private String name = "";
+
   /**
    * Empty instance. This is not ready to be used.
    */
@@ -93,6 +102,29 @@ public AWSCredentialProviderList(
     this.providers.addAll(providers);
   }
 
+  /**
+   * Create with an initial list of providers.
+   * @param name name for error messages, may be ""
+   * @param providerArgs provider list.
+   */
+  public AWSCredentialProviderList(final String name,
+      final AWSCredentialsProvider... providerArgs) {
+    setName(name);
+    Collections.addAll(providers, providerArgs);
+  }
+
+  /**
+   * Set the name; adds a ": " if needed.
+   * @param name name to add, or "" for no name.
+   */
+  public void setName(final String name) {
+    if (!name.isEmpty() && !name.endsWith(": ")) {
+      this.name = name + ": ";
+    } else {
+      this.name = name;
+    }
+  }
+
   /**
    * Add a new provider.
    * @param p provider
@@ -101,6 +133,14 @@ public void add(AWSCredentialsProvider p) {
     providers.add(p);
   }
 
+  /**
+   * Add all providers from another list to this one.
+   * @param other the other list.
+   */
+  public void addAll(AWSCredentialProviderList other) {
+    providers.addAll(other.providers);
+  }
+
   /**
    * Refresh all child entries.
    */
@@ -123,7 +163,7 @@ public void refresh() {
   public AWSCredentials getCredentials() {
     if (isClosed()) {
       LOG.warn(CREDENTIALS_REQUESTED_WHEN_CLOSED);
-      throw new NoAuthWithAWSException(
+      throw new NoAuthWithAWSException(name +
           CREDENTIALS_REQUESTED_WHEN_CLOSED);
     }
     checkNotEmpty();
@@ -135,13 +175,27 @@ public AWSCredentials getCredentials() {
     for (AWSCredentialsProvider provider : providers) {
       try {
         AWSCredentials credentials = provider.getCredentials();
+        Preconditions.checkNotNull(credentials,
+            "Null credentials returned by %s", provider);
         if ((credentials.getAWSAccessKeyId() != null &&
             credentials.getAWSSecretKey() != null)
             || (credentials instanceof AnonymousAWSCredentials)) {
           lastProvider = provider;
           LOG.debug("Using credentials from {}", provider);
           return credentials;
         }
+      } catch (NoAwsCredentialsException e) {
+        // don't bother with the stack trace here as it is usually a
+        // minor detail.
+
+        // only update the last exception if it isn't set.
+        // Why so? Stops delegation token issues being lost on the fallback
+        // values.
+        if (lastException == null) {
+          lastException = e;
+        }
+        LOG.debug("No credentials from {}: {}",
+            provider, e.toString());
       } catch (AmazonClientException e) {
         lastException = e;
         LOG.debug("No credentials provided by {}: {}",
@@ -151,12 +205,16 @@ public AWSCredentials getCredentials() {
 
     // no providers had any credentials. Rethrow the last exception
     // or create a new one.
-    String message = "No AWS Credentials provided by "
+    String message =  name +  "No AWS Credentials provided by "
         + listProviderNames();
     if (lastException != null) {
       message += ": " + lastException;
     }
-    throw new NoAuthWithAWSException(message, lastException);
+    if (lastException instanceof CredentialInitializationException) {
+      throw lastException;
+    } else {
+      throw new NoAuthWithAWSException(message, lastException);
+    }
   }
 
   /**
@@ -175,7 +233,7 @@ List<AWSCredentialsProvider> getProviders() {
    */
   public void checkNotEmpty() {
     if (providers.isEmpty()) {
-      throw new NoAuthWithAWSException(NO_AWS_CREDENTIAL_PROVIDERS);
+      throw new NoAuthWithAWSException(name + NO_AWS_CREDENTIAL_PROVIDERS);
     }
   }
 
@@ -198,8 +256,10 @@ public String listProviderNames() {
   @Override
   public String toString() {
     return "AWSCredentialProviderList[" +
+        name +
         "refcount= " + refCount.get() + ": [" +
-        StringUtils.join(providers, ", ") + ']';
+        StringUtils.join(providers, ", ") + ']'
+        + (lastProvider != null ? (" last provider: " + lastProvider) : "");
   }
 
   /**
@@ -265,4 +325,12 @@ public void close() {
       }
     }
   }
+
+  /**
+   * Get the size of this list.
+   * @return the number of providers in the list.
+   */
+  public int size() {
+    return providers.size();
+  }
 }

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

@@ -49,7 +49,7 @@ private Constants() {
   // s3 secret key
   public static final String SECRET_KEY = "fs.s3a.secret.key";
 
-  // aws credentials provider
+  // aws credentials providers
   public static final String AWS_CREDENTIALS_PROVIDER =
       "fs.s3a.aws.credentials.provider";
 
@@ -61,18 +61,20 @@ private Constants() {
   public static final String S3A_SECURITY_CREDENTIAL_PROVIDER_PATH =
       "fs.s3a.security.credential.provider.path";
 
-  // session token for when using TemporaryAWSCredentialsProvider
+  /**
+   * session token for when using TemporaryAWSCredentialsProvider: : {@value}.
+   */
   public static final String SESSION_TOKEN = "fs.s3a.session.token";
 
   /**
-   * AWS Role to request.
+   * ARN of AWS Role to request: {@value}.
    */
   public static final String ASSUMED_ROLE_ARN =
       "fs.s3a.assumed.role.arn";
 
   /**
    * Session name for the assumed role, must be valid characters according
-   * to the AWS APIs.
+   * to the AWS APIs: {@value}.
    * If not set, one is generated from the current Hadoop/Kerberos username.
    */
   public static final String ASSUMED_ROLE_SESSION_NAME =
@@ -84,34 +86,50 @@ private Constants() {
   public static final String ASSUMED_ROLE_SESSION_DURATION =
       "fs.s3a.assumed.role.session.duration";
 
-  /** Security Token Service Endpoint. If unset, uses the default endpoint. */
+  /**
+   * Security Token Service Endpoint: {@value}.
+   * If unset, uses the default endpoint.
+   */
   public static final String ASSUMED_ROLE_STS_ENDPOINT =
       "fs.s3a.assumed.role.sts.endpoint";
 
   /**
-   * Region for the STS endpoint; only relevant if the endpoint
-   * is set.
+   * Default endpoint for session tokens: {@value}.
+   * This is the central STS endpoint which, for v3 signing, can
+   * issue STS tokens for any region.
+   */
+  public static final String DEFAULT_ASSUMED_ROLE_STS_ENDPOINT = "";
+
+  /**
+   * Region for the STS endpoint; needed if the endpoint
+   * is set to anything other then the central one.: {@value}.
    */
   public static final String ASSUMED_ROLE_STS_ENDPOINT_REGION =
       "fs.s3a.assumed.role.sts.endpoint.region";
 
   /**
    * Default value for the STS endpoint region; needed for
-   * v4 signing.
+   * v4 signing: {@value}.
    */
-  public static final String ASSUMED_ROLE_STS_ENDPOINT_REGION_DEFAULT =
-      "us-west-1";
+  public static final String ASSUMED_ROLE_STS_ENDPOINT_REGION_DEFAULT = "";
 
   /**
-   * Default duration of an assumed role.
+   * Default duration of an assumed role: {@value}.
    */
-  public static final String ASSUMED_ROLE_SESSION_DURATION_DEFAULT = "30m";
+  public static final String ASSUMED_ROLE_SESSION_DURATION_DEFAULT = "1h";
 
-  /** list of providers to authenticate for the assumed role. */
+  /**
+   * List of providers to authenticate for the assumed role: {@value}.
+   */
   public static final String ASSUMED_ROLE_CREDENTIALS_PROVIDER =
       "fs.s3a.assumed.role.credentials.provider";
 
-  /** JSON policy containing the policy to apply to the role. */
+  /**
+   * JSON policy containing the policy to apply to the role: {@value}.
+   * This is not used for delegation tokens, which generate the policy
+   * automatically, and restrict it to the S3, KMS and S3Guard services
+   * needed.
+   */
   public static final String ASSUMED_ROLE_POLICY =
       "fs.s3a.assumed.role.policy";
 
@@ -318,7 +336,10 @@ private Constants() {
   /** Prefix for S3A bucket-specific properties: {@value}. */
   public static final String FS_S3A_BUCKET_PREFIX = "fs.s3a.bucket.";
 
-  public static final int S3A_DEFAULT_PORT = -1;
+  /**
+   * Default port for this is 443: HTTPS.
+   */
+  public static final int S3A_DEFAULT_PORT = 443;
 
   public static final String USER_AGENT_PREFIX = "fs.s3a.user.agent.prefix";
 

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

@@ -28,6 +28,9 @@
 import com.amazonaws.services.s3.S3ClientOptions;
 import org.slf4j.Logger;
 
+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;
 
@@ -39,6 +42,8 @@
  * This which 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 {
 
@@ -47,9 +52,13 @@ public class DefaultS3ClientFactory extends Configured
   @Override
   public AmazonS3 createS3Client(URI name,
       final String bucket,
-      final AWSCredentialsProvider credentials) throws IOException {
+      final AWSCredentialsProvider credentials,
+      final String userAgentSuffix) throws IOException {
     Configuration conf = getConf();
     final ClientConfiguration awsConf = S3AUtils.createAwsConf(getConf(), bucket);
+    if (!StringUtils.isEmpty(userAgentSuffix)) {
+      awsConf.setUserAgentSuffix(userAgentSuffix);
+    }
     return configureAmazonS3Client(
         newAmazonS3Client(credentials, awsConf), conf);
   }

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

@@ -476,7 +476,7 @@ public void onFailure(String text,
   };
 
   /**
-   * Log summary at info, full stack at debug.
+   * Log retries at debug.
    */
   public static final Retried LOG_EVENT = new Retried() {
     @Override