HADOOP-16599. Allow a SignerInitializer to be specified along with a

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

@@ -352,26 +352,30 @@ private Constants() {
 
   /**
    * List of custom Signers. The signer class will be loaded, and the signer
-   * name will be associated with this signer class in the S3 SDK. e.g. Single
-   * CustomSigner {@literal ->} 'CustomSigner:org.apache...CustomSignerClass Multiple
-   * CustomSigners {@literal ->} 'CSigner1:CustomSignerClass1,CSigner2:CustomerSignerClass2
+   * name will be associated with this signer class in the S3 SDK.
+   * Examples
+   * CustomSigner {@literal ->} 'CustomSigner:org.apache...CustomSignerClass'
+   * CustomSigners {@literal ->} 'CSigner1:CSigner1Class,CSigner2:CSigner2Class'
+   * Initializer {@literal ->} 'CSigner1:CSigner1Class:CSigner1InitializerClass'
+   * With Existing {@literal ->} 'AWS4Signer,CSigner1,CSigner2:CSigner2Class'
    */
   public static final String CUSTOM_SIGNERS = "fs.s3a.custom.signers";
 
   /**
    * There's 3 parameters that can be used to specify a non-default signing
-   * algorithm. fs.s3a.signing-algorithm - This property has existed for the
-   * longest time. If specified, without either of the other 2 properties being
-   * specified, this signing algorithm will be used for S3 and DDB (S3Guard).
-   * The other 2 properties override this value for S3 or DDB.
+   * algorithm.<br>
+   * fs.s3a.signing-algorithm - This property has existed for the longest time.
+   * If specified, without either of the other 2 properties being specified,
+   * this signing algorithm will be used for S3 and DDB (S3Guard). <br>
+   * The other 2 properties override this value for S3 or DDB. <br>
    * fs.s3a.s3.signing-algorithm - Allows overriding the S3 Signing algorithm.
    * This does not affect DDB. Specifying this property without specifying
    * fs.s3a.signing-algorithm will only update the signing algorithm for S3
-   * requests, and the default will be used for DDB fs.s3a.ddb.signing-algorithm
-   * - Allows overriding the DDB Signing algorithm. This does not affect S3.
-   * Specifying this property without specifying fs.s3a.signing-algorithm will
-   * only update the signing algorithm for DDB requests, and the default will be
-   * used for S3
+   * requests, and the default will be used for DDB.<br>
+   * fs.s3a.ddb.signing-algorithm - Allows overriding the DDB Signing algorithm.
+   * This does not affect S3. Specifying this property without specifying
+   * fs.s3a.signing-algorithm will only update the signing algorithm for
+   * DDB requests, and the default will be used for S3.
    */
   public static final String SIGNING_ALGORITHM = "fs.s3a.signing-algorithm";
 

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

@@ -95,6 +95,8 @@
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.Globber;
+import org.apache.hadoop.fs.s3a.auth.SignerManager;
+import org.apache.hadoop.fs.s3a.auth.delegation.DelegationTokenProvider;
 import org.apache.hadoop.fs.s3a.impl.ChangeDetectionPolicy;
 import org.apache.hadoop.fs.s3a.impl.ContextAccessors;
 import org.apache.hadoop.fs.s3a.impl.CopyOutcome;
@@ -108,6 +110,7 @@
 import org.apache.hadoop.fs.s3a.s3guard.BulkOperationState;
 import org.apache.hadoop.fs.s3a.select.InternalSelectConstants;
 import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.security.token.TokenIdentifier;
 import org.apache.hadoop.util.DurationInfo;
 import org.apache.hadoop.util.LambdaUtils;
 import org.apache.hadoop.fs.FileAlreadyExistsException;
@@ -182,7 +185,7 @@
 @InterfaceAudience.Private
 @InterfaceStability.Evolving
 public class S3AFileSystem extends FileSystem implements StreamCapabilities,
-    AWSPolicyProvider {
+    AWSPolicyProvider, DelegationTokenProvider {
   /**
    * Default blocksize as used in blocksize and FS status queries.
    */
@@ -362,8 +365,8 @@ public void initialize(URI name, Configuration originalConf)
       }
       useListV1 = (listVersion == 1);
 
-      signerManager = new SignerManager();
-      signerManager.initCustomSigners(conf);
+      signerManager = new SignerManager(bucket, this, conf, owner);
+      signerManager.initCustomSigners();
 
       // creates the AWS client, including overriding auth chain if
       // the FS came with a DT
@@ -1335,6 +1338,11 @@ private long innerRename(Path source, Path dest)
     return renameOperation.execute();
   }
 
+  @Override public Token<? extends TokenIdentifier> getFsDelegationToken()
+      throws IOException {
+    return getDelegationToken(null);
+  }
+
   /**
    * The callbacks made by the rename and delete operations.
    * This separation allows the operation to be factored out and

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

@@ -0,0 +1,53 @@
+/*
+ * 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.auth;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.s3a.auth.delegation.DelegationTokenProvider;
+import org.apache.hadoop.fs.s3a.S3AFileSystem;
+import org.apache.hadoop.security.UserGroupInformation;
+
+/**
+ * Interface which can be implemented to allow initialization of any custom
+ * signers which may be used by the {@link S3AFileSystem}.
+ */
+public interface AwsSignerInitializer {
+
+  /**
+   * Register a store instance.
+   *
+   * @param bucketName the bucket name
+   * @param storeConf the store configuration
+   * @param dtProvider delegation token provider for the store
+   * @param storeUgi ugi under which the store is operating
+   */
+  void registerStore(String bucketName, Configuration storeConf,
+      DelegationTokenProvider dtProvider, UserGroupInformation storeUgi);
+
+  /**
+   * Unregister a store instance.
+   *
+   * @param bucketName the bucket name
+   * @param storeConf the store configuration
+   * @param dtProvider delegation token provider for the store
+   * @param storeUgi ugi under which the store is operating
+   */
+  void unregisterStore(String bucketName, Configuration storeConf,
+      DelegationTokenProvider dtProvider, UserGroupInformation storeUgi);
+}

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

@@ -15,16 +15,22 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.hadoop.fs.s3a;
+package org.apache.hadoop.fs.s3a.auth;
 
-import com.amazonaws.auth.Signer;
-import com.amazonaws.auth.SignerFactory;
 import java.io.Closeable;
 import java.io.IOException;
+import java.util.LinkedList;
+import java.util.List;
+
+import com.amazonaws.auth.Signer;
+import com.amazonaws.auth.SignerFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.s3a.auth.delegation.DelegationTokenProvider;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.util.ReflectionUtils;
 
 import static org.apache.hadoop.fs.s3a.Constants.CUSTOM_SIGNERS;
 
@@ -36,17 +42,28 @@ public class SignerManager implements Closeable {
   private static final Logger LOG = LoggerFactory
       .getLogger(SignerManager.class);
 
+  private final List<AwsSignerInitializer> initializers = new LinkedList<>();
 
-  public SignerManager() {
+  private final String bucketName;
+  private final DelegationTokenProvider delegationTokenProvider;
+  private final Configuration ownerConf;
+  private final UserGroupInformation ownerUgi;
+
+  public SignerManager(String bucketName,
+      DelegationTokenProvider delegationTokenProvider, Configuration ownerConf,
+      UserGroupInformation ownerUgi) {
+    this.bucketName = bucketName;
+    this.delegationTokenProvider = delegationTokenProvider;
+    this.ownerConf = ownerConf;
+    this.ownerUgi = ownerUgi;
   }
 
   /**
    * Initialize custom signers and register them with the AWS SDK.
    *
-   * @param conf Hadoop configuration
    */
-  public void initCustomSigners(Configuration conf) {
-    String[] customSigners = conf.getTrimmedStrings(CUSTOM_SIGNERS);
+  public void initCustomSigners() {
+    String[] customSigners = ownerConf.getTrimmedStrings(CUSTOM_SIGNERS);
     if (customSigners == null || customSigners.length == 0) {
       // No custom signers specified, nothing to do.
       LOG.debug("No custom signers specified");
@@ -55,15 +72,41 @@ public void initCustomSigners(Configuration conf) {
 
     for (String customSigner : customSigners) {
       String[] parts = customSigner.split(":");
-      if (parts.length != 2) {
-        String message =
-            "Invalid format (Expected name:SignerClass) for CustomSigner: ["
-                + customSigner
-                + "]";
+      if (!(parts.length == 1 || parts.length == 2 || parts.length == 3)) {
+        String message = "Invalid format (Expected name, name:SignerClass,"
+            + " name:SignerClass:SignerInitializerClass)"
+            + " for CustomSigner: [" + customSigner + "]";
         LOG.error(message);
         throw new IllegalArgumentException(message);
       }
-      maybeRegisterSigner(parts[0], parts[1], conf);
+      if (parts.length == 1) {
+        // Nothing to do. Trying to use a pre-defined Signer
+      } else {
+        // Register any custom Signer
+        maybeRegisterSigner(parts[0], parts[1], ownerConf);
+
+        // If an initializer is specified, take care of instantiating it and
+        // setting it up
+        if (parts.length == 3) {
+          Class<? extends AwsSignerInitializer> clazz = null;
+          try {
+            clazz = (Class<? extends AwsSignerInitializer>) ownerConf
+                .getClassByName(parts[2]);
+          } catch (ClassNotFoundException e) {
+            throw new RuntimeException(String.format(
+                "SignerInitializer class" + " [%s] not found for signer [%s]",
+                parts[2], parts[0]), e);
+          }
+          LOG.debug("Creating signer initializer: [{}] for signer: [{}]",
+              parts[2], parts[0]);
+          AwsSignerInitializer signerInitializer = ReflectionUtils
+              .newInstance(clazz, null);
+          initializers.add(signerInitializer);
+          signerInitializer
+              .registerStore(bucketName, ownerConf, delegationTokenProvider,
+                  ownerUgi);
+        }
+      }
     }
   }
 
@@ -93,7 +136,12 @@ private static void maybeRegisterSigner(String signerName,
     }
   }
 
-  @Override
-  public void close() throws IOException {
+  @Override public void close() throws IOException {
+    LOG.debug("Unregistering fs from {} initializers", initializers.size());
+    for (AwsSignerInitializer initializer : initializers) {
+      initializer
+          .unregisterStore(bucketName, ownerConf, delegationTokenProvider,
+              ownerUgi);
+    }
   }
-}
+}

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

@@ -0,0 +1,31 @@
+/*
+ * 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.auth.delegation;
+
+import java.io.IOException;
+
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.security.token.TokenIdentifier;
+
+/**
+ * Interface for S3A Delegation Token access.
+ */
+public interface DelegationTokenProvider {
+  Token<? extends TokenIdentifier> getFsDelegationToken() throws IOException;
+}

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

@@ -1879,3 +1879,61 @@ To disable checksum verification in `distcp`, use the `-skipcrccheck` option:
 hadoop distcp -update -skipcrccheck -numListstatusThreads 40 /user/alice/datasets s3a://alice-backup/datasets
 ```
 
+### <a name="customsigners"></a> Advanced - Custom Signers
+
+AWS uees request signing to authenticate requests. In general, there should
+be no need to override the signers, and the defaults work out of the box.
+If, however, this is required - this section talks about how to configure
+custom signers. There’s 2 broad config categories to be set - one for
+registering a custom signer and another to specify usage.
+
+#### Registering Custom Signers
+```xml
+<property>
+  <name>fs.s3a.custom.signers</name>
+  <value>comma separated list of signers</value>
+  <!-- Example
+  <value>AWS4SignerType,CS1:CS1ClassName,CS2:CS2ClassName:CS2InitClass</value>
+  -->
+</property>
+```
+Acceptable value for each custom signer
+
+`SignerName`- this is used in case one of the default signers is being used.
+(E.g `AWS4SignerType`, `QueryStringSignerType`, `AWSS3V4SignerType`).
+If no custom signers are being used - this value does not need to be set.
+
+`SignerName:SignerClassName` - register a new signer with the specified name,
+and the class for this signer.
+The Signer Class must implement `com.amazonaws.auth.Signer`.
+
+`SignerName:SignerClassName:SignerInitializerClassName` - similar time above
+except also allows for a custom SignerInitializer
+(`org.apache.hadoop.fs.s3a.AwsSignerInitializer`) class to be specified.
+
+#### Usage of the Signers
+Signers can be set at a per service level(S3, dynamodb, etc) or a common
+signer for all services.
+
+```xml
+<property>
+  <name>fs.s3a.s3.signing-algorithm</name>
+  <value>${S3SignerName}</value>
+  <description>Specify the signer for S3</description>
+</property>
+
+<property>
+  <name>fs.s3a.ddb.signing-algorithm</name>
+  <value>${DdbSignerName}</value>
+  <description>Specify the signer for DDB</description>
+</property>
+
+<property>
+  <name>fs.s3a.signing-algorithm</name>
+  <value>${SignerName}</value>
+</property>
+```
+
+For a specific service, the service specific signer is looked up first.
+If that is not specified, the common signer is looked up. If this is
+not specified as well, SDK settings are used.