HADOOP-18820. Audit handler migration.

v1 conf load uses getTrimmed() and logs the list of values;
v2 ExecutionInterceptor will, if they implement Configurable,
   get the auditor Configuration

Change-Id: Ic0d49cbceb4a0c9fed9013c9a605f9b240801d73

Author: steveloughran

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

@@ -33,6 +33,8 @@
 import software.amazon.awssdk.http.SdkHttpRequest;
 import software.amazon.awssdk.http.SdkHttpResponse;
 import software.amazon.awssdk.transfer.s3.progress.TransferListener;
+
+import org.apache.hadoop.conf.Configurable;
 import org.apache.hadoop.util.Preconditions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -404,19 +406,26 @@ public List<ExecutionInterceptor> createExecutionInterceptors()
     List<ExecutionInterceptor> executionInterceptors = new ArrayList<>();
     executionInterceptors.add(this);
 
-    final String handlers = getConfig().get(AUDIT_REQUEST_HANDLERS);
-    if (handlers != null) {
-      V2Migration.v1RequestHandlersUsed();
+    final String handlers = getConfig().getTrimmed(AUDIT_REQUEST_HANDLERS, "");
+    if (!handlers.isEmpty()) {
+      // warn and ignore v1 handlers.
+      V2Migration.v1RequestHandlersUsed(handlers);
     }
 
-    // TODO: should we remove this and use Global/Service interceptors, see:
-    //  https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/core/interceptor/ExecutionInterceptor.html
+    // V2 SDK supports global/service interceptors, but they need to be configured on the
+    // classpath and don't get the filesystem/job configuration passed down.
     final Class<?>[] interceptors = getConfig().getClasses(AUDIT_EXECUTION_INTERCEPTORS);
     if (interceptors != null) {
       for (Class<?> handler : interceptors) {
         try {
+          LOG.debug("Adding intercept of class {}", handler);
           Constructor<?> ctor = handler.getConstructor();
-          executionInterceptors.add((ExecutionInterceptor) ctor.newInstance());
+          final ExecutionInterceptor interceptor = (ExecutionInterceptor) ctor.newInstance();
+          if (interceptor instanceof Configurable) {
+            // pass in the configuration.
+            ((Configurable) interceptor).setConf(getConfig());
+          }
+          executionInterceptors.add(interceptor);
         } catch (ExceptionInInitializerError e) {
           throw FutureIO.unwrapInnerException(e);
         } catch (Exception e) {

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

@@ -23,6 +23,7 @@
 
 import org.apache.hadoop.fs.store.LogExactlyOnce;
 
+import static org.apache.hadoop.fs.s3a.audit.S3AAuditConstants.AUDIT_REQUEST_HANDLERS;
 import static org.apache.hadoop.fs.s3a.impl.InternalConstants.SDK_V2_UPGRADE_LOG_NAME;
 
 /**
@@ -50,11 +51,12 @@ private V2Migration() { }
 
   /**
    * Notes use of request handlers.
+   * @param handlers handlers declared
    */
-  public static void v1RequestHandlersUsed() {
+  public static void v1RequestHandlersUsed(final String handlers) {
     WARN_OF_REQUEST_HANDLERS.warn(
-        "The request handler interface has changed in AWS SDK V2, use exception interceptors "
-            + "now S3A is upgraded to SDK V2");
+        "Ignoring V1 SDK request handlers set in {}: {}",
+        AUDIT_REQUEST_HANDLERS, handlers);
   }
 
 }

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

@@ -172,6 +172,9 @@ all of which must implement the interface `software.amazon.awssdk.core.intercept
 
 The list of classes is set in the configuration option `fs.s3a.audit.execution.interceptors`.
 
+Any class in the list which implements `org.apache.hadoop.conf.Configurable` will have
+`Configurable.setConf()` called with the filesystem configuration passed down.
+
 Before the upgrade to the V2 SDK, a list of extra subclasses of the AWS SDK `com.amazonaws.handlers.RequestHandler2`
 class could be declared in the option `fs.s3a.audit.request.handlers`;
 these would be wired up into the V1 request processing pipeline.

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

@@ -34,6 +34,7 @@
 import static org.apache.hadoop.fs.s3a.audit.AuditTestSupport.enableLoggingAuditor;
 import static org.apache.hadoop.fs.s3a.audit.AuditTestSupport.resetAuditOptions;
 import static org.apache.hadoop.fs.s3a.audit.S3AAuditConstants.AUDIT_EXECUTION_INTERCEPTORS;
+import static org.apache.hadoop.fs.s3a.audit.S3AAuditConstants.AUDIT_REQUEST_HANDLERS;
 import static org.apache.hadoop.fs.s3a.audit.S3AAuditConstants.UNAUDITED_OPERATION;
 import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.assertThatStatisticCounter;
 import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.lookupCounterStatistic;
@@ -59,6 +60,7 @@ public Configuration createConfiguration() {
     enableLoggingAuditor(conf);
     conf.set(AUDIT_EXECUTION_INTERCEPTORS,
         SimpleAWSExecutionInterceptor.CLASS);
+    conf.set(AUDIT_REQUEST_HANDLERS, "not-valid-class");
     return conf;
   }
 
@@ -114,8 +116,8 @@ public void testInvokeOutOfSpanRejected() throws Throwable {
   }
 
   @Test
-  public void testRequestHandlerBinding() throws Throwable {
-    describe("Verify that extra request handlers can be added and that they"
+  public void testExecutionInterceptorBinding() throws Throwable {
+    describe("Verify that extra ExecutionInterceptor can be added and that they"
         + " will be invoked during request execution");
     final long baseCount = SimpleAWSExecutionInterceptor.getInvocationCount();
     final S3AFileSystem fs = getFileSystem();
@@ -131,5 +133,9 @@ public void testRequestHandlerBinding() throws Throwable {
         .isGreaterThan(exec0);
     assertThatStatisticCounter(iostats(), AUDIT_FAILURE.getSymbol())
         .isZero();
+    Assertions.assertThat(SimpleAWSExecutionInterceptor.getStaticConf())
+        .describedAs("configuratin of SimpleAWSExecutionInterceptor")
+        .isNotNull()
+        .isSameAs(fs.getConf());
   }
 }

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

@@ -24,24 +24,31 @@
 import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
 import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
 
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
+
 /**
  * Simple AWS interceptor to verify dynamic loading of extra
  * execution interceptors during auditing setup.
  * The invocation counter tracks the count of calls to
  * {@link #beforeExecution}.
  */
-public final class SimpleAWSExecutionInterceptor implements ExecutionInterceptor {
+public final class SimpleAWSExecutionInterceptor extends Configured
+    implements ExecutionInterceptor {
 
   public static final String CLASS
       = "org.apache.hadoop.fs.s3a.audit.SimpleAWSExecutionInterceptor";
 
+  private static Configuration staticConf;
+
   /** Count of invocations. */
   private static final AtomicLong INVOCATIONS = new AtomicLong(0);
 
   @Override
   public void beforeExecution(Context.BeforeExecution context,
       ExecutionAttributes executionAttributes) {
     INVOCATIONS.incrementAndGet();
+    staticConf = getConf();
   }
 
   /**
@@ -51,4 +58,14 @@ public void beforeExecution(Context.BeforeExecution context,
   public static long getInvocationCount() {
     return INVOCATIONS.get();
   }
+
+  /**
+   * get the static conf, which is set the config of the
+   * last executor invoked.
+   * @return the static configuration.
+   */
+
+  public static Configuration getStaticConf() {
+    return staticConf;
+  }
 }