HADOOP-19363. Fixing audit tests

Most of this PR is trying to improve debugging of the auditor
invocation plane on the assumption that those flags being passed
down from the factory were causing problems.
None of those changes did any good, though they did marginally
improve debugging.

The actual problem was ordering of component startup during FS
init: the Auditor must be live before the AWS client is initialized.
Moved back to the right place and improved documentation.

Also: added a test to verify that setting flags would disable
the span checks, which is what we now require.

Change-Id: I108116f0775b71b1cf1c9a2bd5c95727f24f37bb

Author: steveloughran

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

@@ -230,6 +230,7 @@
 import static org.apache.hadoop.fs.s3a.Listing.toLocatedFileStatusIterator;
 import static org.apache.hadoop.fs.s3a.S3AUtils.*;
 import static org.apache.hadoop.fs.s3a.Statistic.*;
+import static org.apache.hadoop.fs.s3a.audit.AuditIntegration.isRejectOutOfSpan;
 import static org.apache.hadoop.fs.s3a.audit.S3AAuditConstants.INITIALIZE_SPAN;
 import static org.apache.hadoop.fs.s3a.auth.CredentialProviderListFactory.createAWSCredentialProviderList;
 import static org.apache.hadoop.fs.s3a.auth.RolePolicies.STATEMENT_ALLOW_KMS_RW;
@@ -256,6 +257,7 @@
 import static org.apache.hadoop.fs.s3a.impl.NetworkBinding.logDnsLookup;
 import static org.apache.hadoop.fs.s3a.impl.S3ExpressStorage.STORE_CAPABILITY_S3_EXPRESS_STORAGE;
 import static org.apache.hadoop.fs.s3a.impl.S3ExpressStorage.isS3ExpressStore;
+import static org.apache.hadoop.fs.s3a.impl.streams.StreamFactoryRequirements.Requirements.ExpectUnauditedGetRequests;
 import static org.apache.hadoop.fs.s3a.s3guard.S3Guard.checkNoS3Guard;
 import static org.apache.hadoop.fs.statistics.IOStatisticsLogging.logIOStatisticsAtLevel;
 import static org.apache.hadoop.fs.statistics.StoreStatisticNames.OBJECT_CONTINUE_LIST_REQUEST;
@@ -301,6 +303,9 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
 
   private String username;
 
+  /**
+   * Store back end.
+   */
   private S3AStore store;
 
   /**
@@ -683,13 +688,22 @@ public void initialize(URI name, Configuration originalConf)
       signerManager = new SignerManager(bucket, this, conf, owner);
       signerManager.initCustomSigners();
 
+      // start auditing
+      // extra configuration will be passed down later.
+      initializeAuditService();
+
+      // create the requestFactory.
+      // requires the audit manager to be initialized.
+      requestFactory = createRequestFactory();
+
       // create an initial span for all other operations.
       span = createSpan(INITIALIZE_SPAN, bucket, null);
 
       // creates the AWS client, including overriding auth chain if
       // the FS came with a DT
       // this may do some patching of the configuration (e.g. setting
       // the encryption algorithms)
+      // requires the audit manager to be initialized.
       ClientManager clientManager = createClientManager(name, delegationTokensEnabled);
 
       inputPolicy = S3AInputPolicy.getPolicy(
@@ -770,14 +784,6 @@ public void initialize(URI name, Configuration originalConf)
 
       int rateLimitCapacity = intOption(conf, S3A_IO_RATE_LIMIT, DEFAULT_S3A_IO_RATE_LIMIT, 0);
 
-      // start auditing.
-      // extra configuration will be passed down later.
-      initializeAuditService();
-
-      // create the requestFactory.
-      // requires the audit manager to be initialized.
-      requestFactory = createRequestFactory();
-
       // now create and initialize the store
       store = createS3AStore(clientManager, rateLimitCapacity);
       // the s3 client is created through the store, rather than
@@ -792,8 +798,7 @@ public void initialize(URI name, Configuration originalConf)
       // If the input stream can issue get requests outside spans,
       // the auditor is forced to disable rejection of unaudited requests.
       final EnumSet<AuditorFlags> flags = EnumSet.noneOf(AuditorFlags.class);
-      if (factoryRequirements.requires(
-          StreamFactoryRequirements.Requirements.ExpectUnauditedGetRequests)) {
+      if (factoryRequirements.requires(ExpectUnauditedGetRequests)) {
         flags.add(AuditorFlags.PermitOutOfBandOperations);
       }
       getAuditManager().setAuditFlags(flags);

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

@@ -38,6 +38,7 @@
 import static java.util.Objects.requireNonNull;
 import static org.apache.hadoop.fs.s3a.audit.S3AAuditConstants.AUDIT_ENABLED;
 import static org.apache.hadoop.fs.s3a.audit.S3AAuditConstants.AUDIT_ENABLED_DEFAULT;
+import static org.apache.hadoop.fs.s3a.audit.S3AAuditConstants.REJECT_OUT_OF_SPAN_OPERATIONS;
 import static org.apache.hadoop.fs.s3a.audit.impl.S3AInternalAuditConstants.AUDIT_SPAN_EXECUTION_ATTRIBUTE;
 
 /**
@@ -68,7 +69,7 @@ public static AuditManagerS3A createAndStartAuditManager(
       auditManager = new ActiveAuditManagerS3A(
           requireNonNull(iostatistics));
     } else {
-      LOG.debug("auditing is disabled");
+      LOG.debug(" is disabled");
       auditManager = stubAuditManager();
     }
     auditManager.init(conf);
@@ -186,4 +187,14 @@ public static boolean containsAuditException(Exception exception) {
         || exception.getCause() instanceof AuditFailureException;
   }
 
+  /**
+   * Check if the configuration is set to reject operations that are
+   * performed outside of an audit span.
+   *
+   * @param conf the configuration to check
+   * @return true if operations outside of an audit span should be rejected, false otherwise
+   */
+  public static boolean isRejectOutOfSpan(final Configuration conf) {
+    return conf.getBoolean(REJECT_OUT_OF_SPAN_OPERATIONS, false);
+  }
 }

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

@@ -23,6 +23,11 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.s3a.audit.AuditIntegration;
 import org.apache.hadoop.fs.s3a.audit.AuditorFlags;
 import org.apache.hadoop.fs.s3a.audit.OperationAuditor;
 import org.apache.hadoop.fs.s3a.audit.OperationAuditorOptions;
@@ -41,6 +46,9 @@
 public abstract class AbstractOperationAuditor extends AbstractService
     implements OperationAuditor {
 
+  private static final Logger LOG =
+      LoggerFactory.getLogger(AbstractOperationAuditor.class);
+
   /**
    * Base of IDs is a UUID.
    */
@@ -79,6 +87,11 @@ public abstract class AbstractOperationAuditor extends AbstractService
    */
   private final String auditorID = auditorUUID.toString();
 
+  /**
+   * Audit flags which can be passed down to subclasses.
+   */
+  private EnumSet<AuditorFlags> auditorFlags;
+
   /**
    * Construct.
    * @param name name
@@ -99,6 +112,15 @@ public void init(final OperationAuditorOptions opts) {
     init(opts.getConfiguration());
   }
 
+  @Override
+  protected void serviceInit(final Configuration conf) throws Exception {
+    super.serviceInit(conf);
+    setRejectOutOfSpan(AuditIntegration.isRejectOutOfSpan(conf));
+    LOG.debug("{}: Out of span operations will be {}",
+        getName(),
+        isRejectOutOfSpan() ? "rejected" : "ignored");
+  }
+
   @Override
   public String getAuditorId() {
     return auditorID;
@@ -129,8 +151,51 @@ protected final String createSpanID() {
         auditorID, SPAN_ID_COUNTER.incrementAndGet());
   }
 
+  /**
+   * Should out of scope ops be rejected?
+   */
+  protected boolean isRejectOutOfSpan() {
+    return rejectOutOfSpan.get();
+  }
+
+  /**
+   * Enable/disable out of span rejection.
+   * @param rejectOutOfSpan new value.
+   */
+  protected void setRejectOutOfSpan(boolean rejectOutOfSpan) {
+    this.rejectOutOfSpan.set(rejectOutOfSpan);
+  }
+
+  /**
+   * Update Auditor flags.
+   * Calls {@link #auditorFlagsChanged(EnumSet)} after the update.
+   * @param flags audit flags.
+   */
   @Override
   public void setAuditFlags(final EnumSet<AuditorFlags> flags) {
-    /* no-op */
+    auditorFlags = flags;
+    auditorFlagsChanged(flags);
+  }
+
+  /**
+   * Get the current set of auditor flags.
+   *
+   * @return the current set of auditor flags.
+   */
+  public EnumSet<AuditorFlags> getAuditorFlags() {
+    return auditorFlags;
   }
+
+  /**
+   * Notification that the auditor flags have been updated.
+   * @param flags audit flags.
+   */
+  protected void auditorFlagsChanged(EnumSet<AuditorFlags> flags) {
+    // if out of band operations are allowed, configuration settings are overridden
+    if (flags.contains(AuditorFlags.PermitOutOfBandOperations)) {
+      LOG.debug("Out of span operations are required by the stream factory");
+      setRejectOutOfSpan(false);
+    }
+  }
+
 }

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

@@ -21,7 +21,6 @@
 import javax.annotation.Nullable;
 import java.io.IOException;
 import java.util.Collection;
-import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
@@ -47,7 +46,6 @@
 import org.apache.hadoop.fs.s3a.audit.AuditFailureException;
 import org.apache.hadoop.fs.s3a.audit.AuditOperationRejectedException;
 import org.apache.hadoop.fs.s3a.audit.AuditSpanS3A;
-import org.apache.hadoop.fs.s3a.audit.AuditorFlags;
 import org.apache.hadoop.fs.store.LogExactlyOnce;
 import org.apache.hadoop.fs.store.audit.HttpReferrerAuditHeader;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -67,7 +65,6 @@
 import static org.apache.hadoop.fs.s3a.audit.S3AAuditConstants.REFERRER_HEADER_ENABLED;
 import static org.apache.hadoop.fs.s3a.audit.S3AAuditConstants.REFERRER_HEADER_ENABLED_DEFAULT;
 import static org.apache.hadoop.fs.s3a.audit.S3AAuditConstants.REFERRER_HEADER_FILTER;
-import static org.apache.hadoop.fs.s3a.audit.S3AAuditConstants.REJECT_OUT_OF_SPAN_OPERATIONS;
 import static org.apache.hadoop.fs.s3a.audit.S3AAuditConstants.UNAUDITED_OPERATION;
 import static org.apache.hadoop.fs.s3a.commit.CommitUtils.extractJobID;
 import static org.apache.hadoop.fs.s3a.impl.HeaderProcessing.HEADER_REFERRER;
@@ -99,11 +96,6 @@ public class LoggingAuditor
    */
   private AuditSpanS3A warningSpan;
 
-  /**
-   * Should out of scope ops be rejected?
-   */
-  private boolean rejectOutOfSpan;
-
   /**
    * Map of attributes which will be added to all operations.
    */
@@ -173,8 +165,6 @@ public LoggingAuditor() {
   @Override
   protected void serviceInit(final Configuration conf) throws Exception {
     super.serviceInit(conf);
-    rejectOutOfSpan = conf.getBoolean(
-        REJECT_OUT_OF_SPAN_OPERATIONS, false);
     // attach the job ID if there is one in the configuration used
     // to create this file.
     String jobID = extractJobID(conf);
@@ -189,6 +179,7 @@ protected void serviceInit(final Configuration conf) throws Exception {
         currentContext, createSpanID(), null, null);
     isMultipartUploadEnabled = conf.getBoolean(MULTIPART_UPLOADS_ENABLED,
               DEFAULT_MULTIPART_UPLOAD_ENABLED);
+    LOG.debug("Initialized {}", this);
   }
 
   @Override
@@ -197,7 +188,7 @@ public String toString() {
         "LoggingAuditor{");
     sb.append("ID='").append(getAuditorId()).append('\'');
     sb.append(", headerEnabled=").append(headerEnabled);
-    sb.append(", rejectOutOfSpan=").append(rejectOutOfSpan);
+    sb.append(", rejectOutOfSpan=").append(isRejectOutOfSpan());
     sb.append(", isMultipartUploadEnabled=").append(isMultipartUploadEnabled);
     sb.append('}');
     return sb.toString();
@@ -217,15 +208,6 @@ public AuditSpanS3A createSpan(final String operation,
     return span;
   }
 
-  @Override
-  public void setAuditFlags(final EnumSet<AuditorFlags> flags) {
-    // if out of band operations are allowed, configuration settings are overridden
-    if (flags.contains(AuditorFlags.PermitOutOfBandOperations)) {
-      LOG.debug("Out of span operations are required by the stream factory");
-      rejectOutOfSpan = false;
-    }
-  }
-
   /**
    * Get/Prepare the active context for a span.
    * @return the common audit context.
@@ -275,6 +257,7 @@ private void setLastHeader(final String lastHeader) {
   HttpReferrerAuditHeader getReferrer(AuditSpanS3A span) {
     return ((LoggingAuditSpan) span).getReferrer();
   }
+
   /**
    * Span which logs at debug and sets the HTTP referrer on
    * invocations.
@@ -556,7 +539,7 @@ public void beforeExecution(Context.BeforeExecution context,
       } else {
         final RuntimeException ex = new AuditFailureException(unaudited);
         LOG.debug(unaudited, ex);
-        if (rejectOutOfSpan) {
+        if (isRejectOutOfSpan()) {
           throw ex;
         }
       }

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

@@ -79,7 +79,7 @@ public NoopAuditManagerS3A() {
   @Override
   protected void serviceInit(final Configuration conf) throws Exception {
     super.serviceInit(conf);
-    NoopAuditor audit = new NoopAuditor(this);
+    NoopAuditor audit = new NoopAuditor("NoopAuditor", this);
     final OperationAuditorOptions options =
         OperationAuditorOptions.builder()
             .withConfiguration(conf)

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

@@ -47,23 +47,29 @@ public class NoopAuditor extends AbstractOperationAuditor {
    * Constructor.
    * This will be used when the auditor is created through
    * configuration and classloading.
+   * @param name  auditor name
    */
-  public NoopAuditor() {
-    this(null);
+  public NoopAuditor(String name) {
+    this(name, null);
   }
 
   /**
    * Constructor when explicitly created within
    * the {@link NoopAuditManagerS3A}.
+   * @param name  auditor name
    * @param activationCallbacks Activation callbacks.
    */
   public NoopAuditor(
-      NoopSpan.SpanActivationCallbacks activationCallbacks) {
-    super("NoopAuditor");
+      String name, NoopSpan.SpanActivationCallbacks activationCallbacks) {
+    super(name);
     this.unbondedSpan = createSpan("unbonded", null, null);
     this.activationCallbacks = activationCallbacks;
   }
 
+  public NoopAuditor() {
+    this("NoopAuditor");
+  }
+
   @Override
   public AuditSpanS3A createSpan(
       final String operation,
@@ -86,7 +92,7 @@ public AuditSpanS3A getUnbondedSpan() {
    */
   public static NoopAuditor createAndStartNoopAuditor(Configuration conf,
       NoopSpan.SpanActivationCallbacks activationCallbacks) {
-    NoopAuditor noop = new NoopAuditor(activationCallbacks);
+    NoopAuditor noop = new NoopAuditor("NoopAuditor", activationCallbacks);
     final OperationAuditorOptions options =
         OperationAuditorOptions.builder()
             .withConfiguration(conf)

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

@@ -20,6 +20,9 @@
 
 import java.io.IOException;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.s3a.S3AFileStatus;
@@ -33,10 +36,14 @@ public class AccessCheckingAuditor  extends NoopAuditor {
   public static final String CLASS =
       "org.apache.hadoop.fs.s3a.audit.AccessCheckingAuditor";
 
+  private static final Logger LOG =
+       LoggerFactory.getLogger(AccessCheckingAuditor.class);
+
   /** Flag to enable/disable access. */
   private boolean accessAllowed = true;
 
   public AccessCheckingAuditor() {
+    super("AccessCheckingAuditor");
   }
 
   public void setAccessAllowed(final boolean accessAllowed) {
@@ -48,6 +55,9 @@ public boolean checkAccess(final Path path,
       final S3AFileStatus status,
       final FsAction mode)
       throws IOException {
+
+    LOG.debug("Check access to {}; allowed={}",
+        path, accessAllowed);
     return accessAllowed;
   }
 }