HADOOP-19559. S3A: Prerequisite features for AAL default ON. (#8067)

Backports:

* HADOOP-19394. S3A: Integrate with AAL's readVectored().  (#7720)

* HADOOP-19664. S3A: Analytics stream to use Java sync client. (#7909)

* HADOOP-19698. S3A: Add AAL dependency to LICENSE-binary.

* HADOOP-19587. S3A: Adds in support for SSE-C to AAL (#7906)

* HADOOP-19365. S3A: Adds in support for auditing for AAL. (#7723)

Author: ahmarsuhail

LICENSE-binary

@@ -361,6 +361,7 @@ org.xerial.snappy:snappy-java:1.1.10.4
 org.yaml:snakeyaml:2.0
 org.wildfly.openssl:wildfly-openssl:2.2.5.Final
 software.amazon.awssdk:bundle:2.29.52
+software.amazon.s3.analyticsaccelerator:analyticsaccelerator-s3:1.3.0
 
 --------------------------------------------------------------------------------
 This product bundles various third-party components under other open source

hadoop-project/pom.xml

@@ -204,7 +204,7 @@
     <aws-java-sdk.version>1.12.720</aws-java-sdk.version>
     <aws-java-sdk-v2.version>2.29.52</aws-java-sdk-v2.version>
     <amazon-s3-encryption-client-java.version>3.1.1</amazon-s3-encryption-client-java.version>
-    <amazon-s3-analyticsaccelerator-s3.version>1.2.1</amazon-s3-analyticsaccelerator-s3.version>
+    <amazon-s3-analyticsaccelerator-s3.version>1.3.0</amazon-s3-analyticsaccelerator-s3.version>
     <aws.eventstream.version>1.0.1</aws.eventstream.version>
     <hsqldb.version>2.7.1</hsqldb.version>
     <frontend-maven-plugin.version>1.11.2</frontend-maven-plugin.version>

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

@@ -1928,7 +1928,10 @@ private FSDataInputStream executeOpen(
         .withCallbacks(createInputStreamCallbacks(auditSpan))
         .withContext(readContext.build())
         .withObjectAttributes(createObjectAttributes(path, fileStatus))
-        .withStreamStatistics(inputStreamStats);
+        .withStreamStatistics(inputStreamStats)
+        .withEncryptionSecrets(getEncryptionSecrets())
+        .withAuditSpan(auditSpan);
+
     return new FSDataInputStream(getStore().readObject(parameters));
   }
 

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

@@ -21,6 +21,7 @@
 import java.util.List;
 
 import software.amazon.awssdk.core.SdkRequest;
+import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
 import software.amazon.awssdk.services.s3.model.AbortMultipartUploadRequest;
 import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
 import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
@@ -50,6 +51,8 @@
 import static org.apache.hadoop.fs.statistics.StoreStatisticNames.OBJECT_LIST_REQUEST;
 import static org.apache.hadoop.fs.statistics.StoreStatisticNames.OBJECT_PUT_REQUEST;
 import static org.apache.hadoop.fs.statistics.StoreStatisticNames.STORE_EXISTS_PROBE;
+import static software.amazon.s3.analyticsaccelerator.request.Constants.OPERATION_NAME;
+import static software.amazon.s3.analyticsaccelerator.request.Constants.SPAN_ID;
 
 /**
  * Extract information from a request.
@@ -193,6 +196,18 @@ private RequestInfo writing(final String verb,
         || request instanceof CreateSessionRequest;
   }
 
+  /**
+   * If spanId and operation name are set by dependencies such as AAL, then this returns true. Allows for auditing
+   * of requests which are made outside S3A's requestFactory.
+   *
+   * @param executionAttributes request execution attributes
+   * @return true if request is audited outside of current span
+   */
+  public static boolean isRequestAuditedOutsideOfCurrentSpan(ExecutionAttributes executionAttributes) {
+   return executionAttributes.getAttribute(SPAN_ID) != null
+            && executionAttributes.getAttribute(OPERATION_NAME) != null;
+  }
+
   /**
    * Predicate which returns true if the request is part of the
    * multipart upload API -and which therefore must be rejected

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

@@ -61,6 +61,7 @@
 import static org.apache.hadoop.fs.s3a.Constants.MULTIPART_UPLOADS_ENABLED;
 import static org.apache.hadoop.fs.s3a.audit.AWSRequestAnalyzer.isRequestMultipartIO;
 import static org.apache.hadoop.fs.s3a.audit.AWSRequestAnalyzer.isRequestNotAlwaysInSpan;
+import static org.apache.hadoop.fs.s3a.audit.AWSRequestAnalyzer.isRequestAuditedOutsideOfCurrentSpan;
 import static org.apache.hadoop.fs.s3a.audit.S3AAuditConstants.OUTSIDE_SPAN;
 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;
@@ -69,6 +70,8 @@
 import static org.apache.hadoop.fs.s3a.commit.CommitUtils.extractJobID;
 import static org.apache.hadoop.fs.s3a.impl.HeaderProcessing.HEADER_REFERRER;
 import static org.apache.hadoop.fs.s3a.statistics.impl.StatisticsFromAwsSdkImpl.mapErrorStatusCodeToStatisticName;
+import static software.amazon.s3.analyticsaccelerator.request.Constants.OPERATION_NAME;
+import static software.amazon.s3.analyticsaccelerator.request.Constants.SPAN_ID;
 
 /**
  * The LoggingAuditor logs operations at DEBUG (in SDK Request) and
@@ -85,7 +88,6 @@ public class LoggingAuditor
   private static final Logger LOG =
       LoggerFactory.getLogger(LoggingAuditor.class);
 
-
   /**
    * Some basic analysis for the logs.
    */
@@ -267,7 +269,14 @@ HttpReferrerAuditHeader getReferrer(AuditSpanS3A span) {
    */
   private class LoggingAuditSpan extends AbstractAuditSpanImpl {
 
-    private final HttpReferrerAuditHeader referrer;
+    private HttpReferrerAuditHeader referrer;
+
+    /**
+     * Builder for the referrer header. Requests that execute outside S3A, such as in AAL, will initially have SpanId
+     * of  the outside-span operation. For such requests, the spanId and operation name in this builder is overwritten
+     * in the modifyHttpRequest execution interceptor.
+     */
+    private final HttpReferrerAuditHeader.Builder headerBuilder;
 
     /**
      * Attach Range of data for GetObject Request.
@@ -300,7 +309,7 @@ private LoggingAuditSpan(
         final String path2) {
       super(spanId, operationName);
 
-      this.referrer = HttpReferrerAuditHeader.builder()
+      this.headerBuilder = HttpReferrerAuditHeader.builder()
           .withContextId(getAuditorId())
           .withSpanId(spanId)
           .withOperationName(operationName)
@@ -312,8 +321,9 @@ private LoggingAuditSpan(
               currentThreadID())
           .withAttribute(PARAM_TIMESTAMP, Long.toString(getTimestamp()))
           .withEvaluated(context.getEvaluatedEntries())
-          .withFilter(filters)
-          .build();
+          .withFilter(filters);
+
+      this.referrer = this.headerBuilder.build();
 
       this.description = referrer.buildHttpReferrer();
     }
@@ -384,6 +394,26 @@ public SdkHttpRequest modifyHttpRequest(Context.ModifyHttpRequest context,
       SdkHttpRequest httpRequest = context.httpRequest();
       SdkRequest sdkRequest = context.request();
 
+      // If spanId and operationName are set in execution attributes, then use these values,
+      // instead of the ones in the current span. This is useful when requests are happening in dependencies such as
+      // the analytics accelerator library (AAL), where they cannot be attached to the correct span. In which case, AAL
+      // will attach the current spanId and operationName via execution attributes during it's request creation. These
+      // can then used to update the values in the logger and referrer header. Without this overwriting, the operation
+      // name and corresponding span will be whichever is active on the thread the request is getting executed on.
+      boolean isRequestAuditedOutsideCurrentSpan = isRequestAuditedOutsideOfCurrentSpan(executionAttributes);
+
+      String spanId = isRequestAuditedOutsideCurrentSpan ?
+              executionAttributes.getAttribute(SPAN_ID) : getSpanId();
+
+      String operationName = isRequestAuditedOutsideCurrentSpan ?
+              executionAttributes.getAttribute(OPERATION_NAME) : getOperationName();
+
+      if (isRequestAuditedOutsideCurrentSpan) {
+        this.headerBuilder.withSpanId(spanId);
+        this.headerBuilder.withOperationName(operationName);
+        this.referrer = this.headerBuilder.build();
+      }
+
       // attach range for GetObject requests
       attachRangeFromRequest(httpRequest, executionAttributes);
 
@@ -400,11 +430,12 @@ public SdkHttpRequest modifyHttpRequest(Context.ModifyHttpRequest context,
             .appendHeader(HEADER_REFERRER, header)
             .build();
       }
+
       if (LOG.isDebugEnabled()) {
         LOG.debug("[{}] {} Executing {} with {}; {}",
             currentThreadID(),
-            getSpanId(),
-            getOperationName(),
+            spanId,
+            operationName,
             analyzer.analyze(context.request()),
             header);
       }
@@ -533,10 +564,12 @@ public void beforeExecution(Context.BeforeExecution context,
           + analyzer.analyze(context.request());
       final String unaudited = getSpanId() + " "
           + UNAUDITED_OPERATION + " " + error;
+      // If request is attached to a span in the modifyHttpRequest, as is the case for requests made by AAL, treat it
+      // as an audited request.
       if (isRequestNotAlwaysInSpan(context.request())) {
-        // can get by auditing during a copy, so don't overreact
+        // can get by auditing during a copy, so don't overreact.
         LOG.debug(unaudited);
-      } else {
+      } else if (!isRequestAuditedOutsideOfCurrentSpan(executionAttributes)) {
         final RuntimeException ex = new AuditFailureException(unaudited);
         LOG.debug(unaudited, ex);
         if (isRejectOutOfSpan()) {
@@ -547,5 +580,4 @@ public void beforeExecution(Context.BeforeExecution context,
       super.beforeExecution(context, executionAttributes);
     }
   }
-
 }

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

@@ -993,10 +993,9 @@ public InputStreamType streamType() {
   private class FactoryCallbacks implements StreamFactoryCallbacks {
 
     @Override
-    public S3AsyncClient getOrCreateAsyncClient(final boolean requireCRT) throws IOException {
-      // Needs support of the CRT before the requireCRT can be used
-      LOG.debug("Stream factory requested async client");
-      return clientManager().getOrCreateAsyncClient();
+    public S3Client getOrCreateSyncClient() throws IOException {
+      LOG.debug("Stream factory requested sync client");
+      return clientManager().getOrCreateS3Client();
     }
 
     @Override

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

@@ -21,10 +21,22 @@
 
 import java.io.EOFException;
 import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+import java.util.function.IntFunction;
+import java.util.Optional;
 
+import org.apache.hadoop.fs.s3a.S3AEncryptionMethods;
+import org.apache.hadoop.fs.s3a.auth.delegation.EncryptionSecretOperations;
 import software.amazon.s3.analyticsaccelerator.S3SeekableInputStreamFactory;
 import software.amazon.s3.analyticsaccelerator.S3SeekableInputStream;
+import software.amazon.s3.analyticsaccelerator.common.ObjectRange;
+import software.amazon.s3.analyticsaccelerator.request.EncryptionSecrets;
 import software.amazon.s3.analyticsaccelerator.request.ObjectMetadata;
+import software.amazon.s3.analyticsaccelerator.request.StreamAuditContext;
 import software.amazon.s3.analyticsaccelerator.util.InputPolicy;
 import software.amazon.s3.analyticsaccelerator.util.OpenStreamInformation;
 import software.amazon.s3.analyticsaccelerator.util.S3URI;
@@ -37,6 +49,11 @@
 import org.apache.hadoop.fs.s3a.Retries;
 import org.apache.hadoop.fs.s3a.S3AInputPolicy;
 import org.apache.hadoop.fs.s3a.S3ObjectAttributes;
+import org.apache.hadoop.fs.FileRange;
+import org.apache.hadoop.fs.VectoredReadUtils;
+
+import static org.apache.hadoop.fs.VectoredReadUtils.LOG_BYTE_BUFFER_RELEASED;
+
 
 /**
  * Analytics stream creates a stream using aws-analytics-accelerator-s3. This stream supports
@@ -128,6 +145,42 @@ public int read(byte[] buf, int off, int len) throws IOException {
     return bytesRead;
   }
 
+  /**
+   * Pass to {@link #readVectored(List, IntFunction, Consumer)}
+   * with the {@link VectoredReadUtils#LOG_BYTE_BUFFER_RELEASED} releaser.
+   * {@inheritDoc}
+   */
+  @Override
+  public void readVectored(List<? extends FileRange> ranges,
+                                        IntFunction<ByteBuffer> allocate) throws IOException {
+    readVectored(ranges, allocate, LOG_BYTE_BUFFER_RELEASED);
+  }
+
+  /**
+   * Pass to {@link #readVectored(List, IntFunction, Consumer)}
+   * with the {@link VectoredReadUtils#LOG_BYTE_BUFFER_RELEASED} releaser.
+   * {@inheritDoc}
+   */
+  @Override
+  public void readVectored(final List<? extends FileRange> ranges,
+                           final IntFunction<ByteBuffer> allocate,
+                           final Consumer<ByteBuffer> release) throws IOException {
+    LOG.debug("AAL: Starting vectored read on path {} for ranges {} ", getPathStr(), ranges);
+    throwIfClosed();
+
+    List<ObjectRange> objectRanges = new ArrayList<>();
+
+    for (FileRange range : ranges) {
+      CompletableFuture<ByteBuffer> result = new CompletableFuture<>();
+      ObjectRange objectRange = new ObjectRange(result, range.getOffset(), range.getLength());
+      objectRanges.add(objectRange);
+      range.setData(result);
+    }
+
+    // AAL does not do any range coalescing, so input and combined ranges are the same.
+    this.getS3AStreamStatistics().readVectoredOperationStarted(ranges.size(), ranges.size());
+    inputStream.readVectored(objectRanges, allocate, release);
+  }
 
   @Override
   public boolean seekToNewSource(long l) throws IOException {
@@ -205,6 +258,18 @@ private OpenStreamInformation buildOpenStreamInformation(ObjectReadParameters pa
           .etag(parameters.getObjectAttributes().getETag()).build());
     }
 
+
+    if (parameters.getEncryptionSecrets().getEncryptionMethod() == S3AEncryptionMethods.SSE_C) {
+      EncryptionSecretOperations.getSSECustomerKey(parameters.getEncryptionSecrets())
+              .ifPresent(base64customerKey -> openStreamInformationBuilder.encryptionSecrets(
+              EncryptionSecrets.builder().sseCustomerKey(Optional.of(base64customerKey)).build()));
+    }
+
+    openStreamInformationBuilder.streamAuditContext(StreamAuditContext.builder()
+                    .operationName(parameters.getAuditSpan().getOperationName())
+                    .spanId(parameters.getAuditSpan().getSpanId())
+                    .build());
+
     return openStreamInformationBuilder.build();
   }
 

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

@@ -23,9 +23,10 @@
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import software.amazon.s3.analyticsaccelerator.S3SdkObjectClient;
+
 import software.amazon.s3.analyticsaccelerator.S3SeekableInputStreamConfiguration;
 import software.amazon.s3.analyticsaccelerator.S3SeekableInputStreamFactory;
+import software.amazon.s3.analyticsaccelerator.S3SyncSdkObjectClient;
 import software.amazon.s3.analyticsaccelerator.common.ConnectorConfiguration;
 
 import org.apache.hadoop.conf.Configuration;
@@ -96,8 +97,7 @@ public StreamFactoryRequirements factoryRequirements() {
     vectorContext.setMinSeekForVectoredReads(0);
 
     return new StreamFactoryRequirements(0,
-            0, vectorContext,
-            StreamFactoryRequirements.Requirements.ExpectUnauditedGetRequests);
+            0, vectorContext);
   }
 
   @Override
@@ -118,7 +118,7 @@ private S3SeekableInputStreamFactory getOrCreateS3SeekableInputStreamFactory()
 
   private CallableRaisingIOE<S3SeekableInputStreamFactory> createS3SeekableInputStreamFactory() {
     return () -> new S3SeekableInputStreamFactory(
-            new S3SdkObjectClient(callbacks().getOrCreateAsyncClient(requireCrt)),
+            new S3SyncSdkObjectClient(callbacks().getOrCreateSyncClient()),
             seekableInputStreamConfiguration);
   }
 

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

@@ -20,7 +20,7 @@
 
 import java.io.IOException;
 
-import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.S3Client;
 
 import org.apache.hadoop.fs.s3a.Statistic;
 import org.apache.hadoop.fs.StreamCapabilities;
@@ -80,12 +80,11 @@ ObjectInputStream readObject(ObjectReadParameters parameters)
   interface StreamFactoryCallbacks {
 
     /**
-     * Get the Async S3Client, raising a failure to create as an IOException.
-     * @param requireCRT is the CRT required.
+     * Get the Sync S3Client, raising a failure to create as an IOException.
      * @return the Async S3 client
      * @throws IOException failure to create the client.
      */
-    S3AsyncClient getOrCreateAsyncClient(boolean requireCRT) throws IOException;
+    S3Client getOrCreateSyncClient() throws IOException;
 
     void incrementFactoryStatistic(Statistic statistic);
   }