Merge pull request apache#13 from passaro/HADOOP-18073-auditors

Hadoop 18073 auditors

Author: passaro

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

@@ -27,7 +27,6 @@
 import com.amazonaws.SdkClientException;
 import com.amazonaws.auth.AWSCredentialsProvider;
 import com.amazonaws.client.builder.AwsClientBuilder;
-import com.amazonaws.handlers.RequestHandler2;
 import com.amazonaws.regions.RegionUtils;
 import com.amazonaws.services.s3.AmazonS3;
 import com.amazonaws.services.s3.AmazonS3Builder;
@@ -53,6 +52,7 @@
 
 import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
 import software.amazon.awssdk.core.client.config.SdkAdvancedClientOption;
+import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
 import software.amazon.awssdk.core.retry.RetryPolicy;
 import software.amazon.awssdk.http.apache.ApacheHttpClient;
 import software.amazon.awssdk.http.apache.ProxyConfiguration;
@@ -226,6 +226,12 @@ public S3Client createS3ClientV2(
           parameters.getUserAgentSuffix());
     }
 
+    if (parameters.getExecutionInterceptors() != null) {
+      for (ExecutionInterceptor interceptor : parameters.getExecutionInterceptors()) {
+        clientOverrideConfigBuilder.addExecutionInterceptor(interceptor);
+      }
+    }
+
     clientOverrideConfigBuilder.retryPolicy(retryPolicyBuilder.build());
     httpClientBuilder.proxyConfiguration(proxyConfigBuilder.build());
 
@@ -253,8 +259,8 @@ public S3Client createS3ClientV2(
     s3ClientBuilder.serviceConfiguration(s3Configuration);
 
     // TODO: Some configuration done in configureBasicParams is not done yet.
-    //  Request handlers will be added during auditor work. Need to verify how metrics collection
-    //  can be done, as SDK V2 only seems to have a metrics publisher.
+    //  Need to verify how metrics collection can be done, as SDK V2 only
+    //  seems to have a metrics publisher.
 
     return s3ClientBuilder.build();
   }
@@ -359,10 +365,6 @@ private void configureBasicParams(AmazonS3Builder builder,
       builder.withMetricsCollector(
           new AwsStatisticsCollector(parameters.getMetrics()));
     }
-    if (parameters.getRequestHandlers() != null) {
-      builder.withRequestHandlers(
-          parameters.getRequestHandlers().toArray(new RequestHandler2[0]));
-    }
     if (parameters.getMonitoringListener() != null) {
       builder.withMonitoringListener(parameters.getMonitoringListener());
     }

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

@@ -77,17 +77,12 @@
 import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
 import software.amazon.awssdk.awscore.exception.AwsServiceException;
 import software.amazon.awssdk.core.sync.RequestBody;
-import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.s3.model.CopyObjectRequest;
 import software.amazon.awssdk.services.s3.model.CopyObjectResponse;
 import software.amazon.awssdk.services.s3.model.DeleteObjectsRequest;
 import software.amazon.awssdk.services.s3.model.DeleteObjectsResponse;
-import software.amazon.awssdk.services.s3.model.GetObjectRequest;
-import software.amazon.awssdk.services.s3.model.GetObjectResponse;
 import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
 import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
-import software.amazon.awssdk.services.s3.model.ListObjectsRequest;
-import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
 import software.amazon.awssdk.services.s3.model.ObjectCannedACL;
 import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
@@ -103,6 +98,7 @@
 import software.amazon.awssdk.transfer.s3.FileUpload;
 import software.amazon.awssdk.transfer.s3.S3TransferManager;
 import software.amazon.awssdk.transfer.s3.UploadFileRequest;
+import software.amazon.awssdk.transfer.s3.progress.TransferListener;
 
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.classification.InterfaceAudience;
@@ -967,7 +963,7 @@ private void bindAWSClient(URI name, boolean dtEnabled) throws IOException {
         .withPathStyleAccess(conf.getBoolean(PATH_STYLE_ACCESS, false))
         .withUserAgentSuffix(uaSuffix)
         .withRequesterPays(conf.getBoolean(ALLOW_REQUESTER_PAYS, DEFAULT_ALLOW_REQUESTER_PAYS))
-        .withRequestHandlers(auditManager.createRequestHandlers());
+        .withExecutionInterceptors(auditManager.createExecutionInterceptors());
 
     s3 = ReflectionUtils.newInstance(s3ClientFactoryClass, conf)
         .createS3Client(getUri(),
@@ -4269,13 +4265,11 @@ private CopyObjectResponse copyFile(String srcKey, String dstKey, long size,
 
     // TODO: Transfer manager currently only provides transfer listeners for upload,
     //  add progress listener for copy when this is supported.
-//    ProgressListener progressListener = progressEvent -> {
-//      switch (progressEvent.getEventType()) {
-//      case TRANSFER_PART_COMPLETED_EVENT:
+// TODO: Is the above still valid? Try to enable when logger issue is resolved.
+//    TransferListener progressListener = new TransferListener() {
+//      @Override
+//      public void transferComplete(Context.TransferComplete context) {
 //        incrementWriteOperations();
-//        break;
-//      default:
-//        break;
 //      }
 //    };
 
@@ -4319,7 +4313,13 @@ private CopyObjectResponse copyFile(String srcKey, String dstKey, long size,
           incrementStatistic(OBJECT_COPY_REQUESTS);
 
           Copy copy = transferManagerV2.copy(
-              CopyRequest.builder().copyObjectRequest(copyObjectRequestBuilder.build()).build());
+              CopyRequest.builder()
+                  .copyObjectRequest(copyObjectRequestBuilder.build())
+// TODO: Enable when logger issue is resolved.
+//                  .overrideConfiguration(c -> c
+//                      .addListener(getAuditManager().createTransferListener())
+//                      .addListener(progressListener))
+                  .build());
 
           CompletedCopy completedCopy = copy.completionFuture().join();
 

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

@@ -29,10 +29,10 @@
 import com.amazonaws.retry.RetryUtils;
 import com.amazonaws.services.s3.model.AmazonS3Exception;
 import com.amazonaws.services.s3.model.MultiObjectDeleteException;
-import org.apache.hadoop.classification.VisibleForTesting;
-import org.apache.hadoop.util.Preconditions;
 
 import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
@@ -49,6 +49,7 @@
 import org.apache.hadoop.fs.s3native.S3xLoginHelper;
 import org.apache.hadoop.net.ConnectTimeoutException;
 import org.apache.hadoop.security.ProviderUtils;
+import org.apache.hadoop.util.Preconditions;
 import org.apache.hadoop.util.VersionInfo;
 
 import org.apache.hadoop.util.Lists;

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

@@ -26,15 +26,16 @@
 import java.util.Map;
 
 import com.amazonaws.auth.AWSCredentialsProvider;
-import com.amazonaws.handlers.RequestHandler2;
 import com.amazonaws.monitoring.MonitoringListener;
 import com.amazonaws.services.s3.AmazonS3;
-import software.amazon.awssdk.services.s3.S3Client;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.fs.s3a.statistics.StatisticsFromAwsSdk;
 
+import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
+import software.amazon.awssdk.services.s3.S3Client;
+
 import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_ENDPOINT;
 
 /**
@@ -122,9 +123,9 @@ final class S3ClientCreationParameters {
     private boolean requesterPays;
 
     /**
-     * Request handlers; used for auditing, X-Ray etc.
-     */
-    private List<RequestHandler2> requestHandlers;
+     * Execution interceptors; used for auditing, X-Ray etc.
+     * */
+    private List<ExecutionInterceptor> executionInterceptors;
 
     /**
      * Suffix to UA.
@@ -138,22 +139,22 @@ final class S3ClientCreationParameters {
     private URI pathUri;
 
     /**
-     * List of request handlers to include in the chain
-     * of request execution in the SDK.
-     * @return the handler list
+     * List of execution interceptors to include in the chain
+     * of interceptors in the SDK.
+     * @return the interceptors list
      */
-    public List<RequestHandler2> getRequestHandlers() {
-      return requestHandlers;
+    public List<ExecutionInterceptor> getExecutionInterceptors() {
+      return executionInterceptors;
     }
 
     /**
-     * List of request handlers.
-     * @param handlers handler list.
+     * List of execution interceptors.
+     * @param interceptors interceptors list.
      * @return this object
      */
-    public S3ClientCreationParameters withRequestHandlers(
-        @Nullable final List<RequestHandler2> handlers) {
-      requestHandlers = handlers;
+    public S3ClientCreationParameters withExecutionInterceptors(
+        @Nullable final List<ExecutionInterceptor> interceptors) {
+      executionInterceptors = interceptors;
       return this;
     }
 

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

@@ -22,9 +22,6 @@
 import java.util.List;
 import java.util.Optional;
 
-import com.amazonaws.services.s3.model.ListNextBatchOfObjectsRequest;
-import com.amazonaws.services.s3.model.ObjectListing;
-import com.amazonaws.services.s3.model.SSEAwsKeyManagementParams;
 import com.amazonaws.services.s3.model.SSECustomerKey;
 import com.amazonaws.services.s3.model.SelectObjectContentRequest;
 
@@ -81,13 +78,6 @@ public interface RequestFactory {
    */
   ObjectCannedACL getCannedACL();
 
-  /**
-   * Create the AWS SDK structure used to configure SSE,
-   * if the encryption secrets contain the information/settings for this.
-   * @return an optional set of KMS Key settings
-   */
-  Optional<SSEAwsKeyManagementParams> generateSSEAwsKeyParams();
-
   /**
    * Create the SSE-C structure for the AWS SDK, if the encryption secrets
    * contain the information/settings for this.
@@ -242,16 +232,6 @@ ListObjectsRequest.Builder newListObjectsV1RequestBuilder(String key,
       String delimiter,
       int maxKeys);
 
-  /**
-   * Create the next V1 page list request, following
-   * on from the previous response.
-   * @param prev previous response
-   * @return the request
-   */
-
-  ListNextBatchOfObjectsRequest newListNextBatchOfObjectsRequest(
-      ObjectListing prev);
-
   /**
    * Create a V2 list request builder.
    * This will be recycled for any subsequent requests.

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

@@ -18,13 +18,8 @@
 
 package org.apache.hadoop.fs.s3a.audit;
 
-import com.amazonaws.AmazonWebServiceRequest;
-import com.amazonaws.Request;
-import com.amazonaws.Response;
-import com.amazonaws.SdkBaseException;
-import com.amazonaws.handlers.HandlerAfterAttemptContext;
-import com.amazonaws.handlers.HandlerBeforeAttemptContext;
-import com.amazonaws.http.HttpResponse;
+import software.amazon.awssdk.core.SdkRequest;
+import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
 
 import org.apache.hadoop.fs.s3a.Retries;
 
@@ -37,10 +32,10 @@
  * detect this and raise an exception.
  *
  * Look at the documentation for
- * {@code com.amazonaws.handlers.IRequestHandler2} for details
+ * {@code ExecutionInterceptor} for details
  * on the callbacks.
  */
-public interface AWSAuditEventCallbacks {
+public interface AWSAuditEventCallbacks extends ExecutionInterceptor {
 
   /**
    * Return a span ID which must be unique for all spans within
@@ -66,95 +61,8 @@ public interface AWSAuditEventCallbacks {
    * It is not invoked on any AWS requests created in the SDK.
    * Avoid raising exceptions or talking to any remote service;
    * this callback is for annotation rather than validation.
-   * @param request request request.
-   * @param <T> type of request
-   * @return the request, possibly modified.
+   * @param builder the request builder.
    */
-  default <T extends AmazonWebServiceRequest> T requestCreated(T request) {
-    return request;
-  }
+  default void requestCreated(SdkRequest.Builder builder) {}
 
-  /**
-   * Preflight preparation of AWS request.
-   * @param request request
-   * @param <T> type of request
-   * @return an updated request.
-   * @throws AuditFailureException for generic audit failures
-   * @throws SdkBaseException for other reasons.
-   */
-  @Retries.OnceRaw
-  default <T extends AmazonWebServiceRequest> T beforeExecution(T request)
-      throws AuditFailureException, SdkBaseException {
-    return request;
-  }
-
-  /**
-   * Callback after S3 responded to a request.
-   * @param request request
-   * @param response response.
-   * @throws AuditFailureException for generic audit failures
-   * @throws SdkBaseException for other reasons.
-   */
-  default void afterResponse(Request<?> request,
-      Response<?> response)
-      throws AuditFailureException, SdkBaseException {
-  }
-
-  /**
-   * Callback after a request resulted in an error.
-   * @param request request
-   * @param response response.
-   * @param exception exception raised.
-   * @throws AuditFailureException for generic audit failures
-   * @throws SdkBaseException for other reasons.
-   */
-  default void afterError(Request<?> request,
-      Response<?> response,
-      Exception exception)
-      throws AuditFailureException, SdkBaseException {
-  }
-
-  /**
-   * Request before marshalling.
-   * @param request request
-   * @return possibly modified request.
-   */
-  default AmazonWebServiceRequest beforeMarshalling(
-      AmazonWebServiceRequest request) {
-    return request;
-  }
-
-  /**
-   * Request before marshalling.
-   * @param request request
-   */
-  default void beforeRequest(Request<?> request) {
-  }
-
-  /**
-   * Before any attempt is made.
-   * @param context full context, including the request.
-   */
-  default void beforeAttempt(HandlerBeforeAttemptContext context) {
-  }
-
-  /**
-   * After any attempt is made.
-   * @param context full context, including the request.
-   */
-  default void afterAttempt(
-      HandlerAfterAttemptContext context) {
-  }
-
-  /**
-   * Before unmarshalling the response.
-   * @param request request made.
-   * @param httpResponse response received
-   * @return updated response.
-   */
-  default HttpResponse beforeUnmarshalling(
-      final Request<?> request,
-      final HttpResponse httpResponse) {
-    return httpResponse;
-  }
 }