Aws sdk 2.x netty.nio

src/hpc-postman/postman/collections/HPC Server API/.resources/definition.yaml

@@ -3,6 +3,8 @@ name: HPC Server API
 variables:
   lastSyncUploadPath: ""
   presignedDownloadUrl: ""
+  lastS3DownloadFileId: ""
+  lastS3DownloadTaskId: ""
 auth:
   - id: 236c22c3-dd78-4b41-8176-a23ea4721e7b
     type: basic

src/hpc-postman/postman/collections/HPC Server API/Data Object Registration/Sync Uplod/Download to AWS S3.request.yaml

@@ -3,8 +3,11 @@ name: Download to AWS S3
 method: POST
 url: "{{baseUrl}}{{basePath}}/v2/dataObject/{{path}}/download"
 scripts:
+  - type: afterResponse
+    code: "pm.collectionVariables.set('lastS3DownloadFileId', pm.variables.get('fileId')); const res = pm.response.json(); if (res.taskId) { pm.collectionVariables.set('lastS3DownloadTaskId', res.taskId); }"
+    language: text/javascript
   - type: beforeRequest
-    code: "pm.variables.set('path', pm.collectionVariables.get('lastSyncUploadPath')); "
+    code: "pm.variables.set('path', pm.collectionVariables.get('lastSyncUploadPath')); const now = new Date(); const ts = (now.getMonth()+1).toString().padStart(2,'0') + '-' + now.getDate().toString().padStart(2,'0') + '-' + now.getFullYear() + '-' + now.getHours().toString().padStart(2,'0') + '-' + now.getMinutes().toString().padStart(2,'0') + '-' + now.getSeconds().toString().padStart(2,'0'); pm.variables.set('fileId', pm.globals.get('aws-folder') + '/file-download-' + ts);"
     language: text/javascript
 order: 3863696254383541
 headers:
@@ -18,7 +21,7 @@ body:
       "s3DownloadDestination": {
         "destinationLocation": {
           "fileContainerId": "{{aws-bucket}}",
-          "fileId": "{{aws-folder}}/file-download"
+          "fileId": "{{fileId}}"
         },
         "account": {
           "accessKey": "{{vault:aws-key}}",

src/hpc-postman/postman/collections/HPC Server API/Data Object Registration/Sync Uplod/Get S3 Download Task Status.request.yaml

@@ -0,0 +1,8 @@
+$kind: http-request
+name: Get S3 Download Task Status
+method: GET
+url: '{{baseUrl}}{{basePath}}/dataObject/download/{{lastS3DownloadTaskId}}'
+order: 3863696254383542
+headers:
+  - key: Accept
+    value: 'application/json'

src/hpc-server/hpc-integration-impl/pom.xml

@@ -129,6 +129,10 @@
   			<groupId>software.amazon.awssdk.crt</groupId>
   			<artifactId>aws-crt</artifactId>
 		</dependency>
+		<dependency>
+  			<groupId>software.amazon.awssdk</groupId>
+  			<artifactId>netty-nio-client</artifactId>
+		</dependency>
 		<dependency>
     		<groupId>com.google.apis</groupId>
     		<artifactId>google-api-services-drive</artifactId>

src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/impl/HpcDataTransferProxyImpl.java → src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/v1/impl/HpcDataTransferProxyImpl.java

@@ -1,4 +1,4 @@
-package gov.nih.nci.hpc.integration.s3.impl;
+package gov.nih.nci.hpc.integration.s3.v1.impl;
 
 import static gov.nih.nci.hpc.integration.HpcDataTransferProxy.getArchiveDestinationLocation;
 import static gov.nih.nci.hpc.util.HpcUtil.toIntExact;

src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/impl/HpcS3Connection.java → src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/v1/impl/HpcS3Connection.java

@@ -8,7 +8,7 @@
  * Distributed under the OSI-approved BSD 3-Clause License. See
  * http://ncip.github.com/HPC/LICENSE.txt for details.
  */
-package gov.nih.nci.hpc.integration.s3.impl;
+package gov.nih.nci.hpc.integration.s3.v1.impl;
 
 import java.util.Base64;
 import java.util.HashSet;

src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/impl/HpcS3ProgressListener.java → src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/v1/impl/HpcS3ProgressListener.java

@@ -6,7 +6,7 @@
  * <p>Distributed under the OSI-approved BSD 3-Clause License. See
  * http://ncip.github.com/HPC/LICENSE.txt for details.
  */
-package gov.nih.nci.hpc.integration.s3.impl;
+package gov.nih.nci.hpc.integration.s3.v1.impl;
 
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;

src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/v2/impl/HpcS3Connection.java

@@ -31,30 +31,30 @@
 import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
 import software.amazon.awssdk.core.exception.SdkException;
-import software.amazon.awssdk.crt.CrtRuntimeException;
-import software.amazon.awssdk.crt.Log;
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.s3.S3AsyncClient;
 import software.amazon.awssdk.services.s3.S3Configuration;
-import software.amazon.awssdk.services.s3.S3CrtAsyncClientBuilder;
 import software.amazon.awssdk.services.s3.presigner.S3Presigner;
 import software.amazon.awssdk.transfer.s3.S3TransferManager;
 
 /**
- * HPC S3 Connection.
+ * HPC S3 Connection base class. Holds all logic shared between the concrete S3
+ * connection implementations. The only behavior that differs between
+ * implementations is how the {@link S3AsyncClient} is built (e.g. AWS CRT vs
+ * Netty-NIO), which is delegated to the abstract build methods.
  *
  * @author <a href="mailto:eran.rosenberg@nih.gov">Eran Rosenberg</a>
  */
-public class HpcS3Connection {
+public abstract class HpcS3Connection {
 	// ---------------------------------------------------------------------//
 	// Constants
 	// ---------------------------------------------------------------------//
 
 	// 5GB in bytes
-	private static final long FIVE_GB = 5368709120L;
+	protected static final long FIVE_GB = 5368709120L;
 
 	// Google Storage S3 URL.
-	private static final String GOOGLE_STORAGE_URL = "https://storage.googleapis.com";
+	protected static final String GOOGLE_STORAGE_URL = "https://storage.googleapis.com";
 
 	// ---------------------------------------------------------------------//
 	// Instance members
@@ -66,25 +66,33 @@ public class HpcS3Connection {
 
 	// The multipart upload minimum part size.
 	@Value("${hpc.integration.s3.minimumUploadPartSize}")
-	private Long minimumUploadPartSize = null;
+	protected Long minimumUploadPartSize = null;
 
 	// The multipart upload threshold.
 	@Value("${hpc.integration.s3.multipartUploadThreshold}")
-	private Long multipartUploadThreshold = null;
+	protected Long multipartUploadThreshold = null;
 
-	// The CRT log file (Optional).
-	@Value("${hpc.integration.s3.crtLogFile:#{null}}")
-	private String crtLogFile = null;
+	// The connection timeout - the time to wait to establish a new connection, in
+	// milliseconds. Same behavior across all S3 client implementations, so the
+	// property is shared.
+	@Value("${hpc.integration.s3.connectionTimeout}")
+	protected Integer connectionTimeout = null;
 
-	// AWS CRT to trust all certs config.
-	@Value("${hpc.integration.s3.trustAllCerts:false}")
-	private Boolean trustAllCerts = null;
+	// The max number of error retries. Same behavior across all S3 client
+	// implementations, so the property is shared.
+	@Value("${hpc.integration.s3.maxErrorRetries}")
+	protected Integer maxErrorRetries = null;
+
+	// Disable SSL certificate checking (for development/testing only). Shares the
+	// property with the v1 client since the behavior is the same.
+	@Value("${hpc.integration.s3.disableCertChecking:false}")
+	protected Boolean disableCertChecking = null;
 
 	// The executor service to be used by AWSTransferManager
 	private ExecutorService executorService = null;
 
 	// The logger instance.
-	private final Logger logger = LoggerFactory.getLogger(getClass().getName());
+	protected final Logger logger = LoggerFactory.getLogger(getClass().getName());
 
 	// ---------------------------------------------------------------------//
 	// Constructors
@@ -99,7 +107,7 @@ public class HpcS3Connection {
 	 * @param awsTransferManagerThreadPoolSize The thread pool size to be used for
 	 *                                         AWS transfer manager
 	 */
-	private HpcS3Connection(String pathStyleAccessEnabledProviders, int awsTransferManagerThreadPoolSize) {
+	protected HpcS3Connection(String pathStyleAccessEnabledProviders, int awsTransferManagerThreadPoolSize) {
 		for (String s3Provider : pathStyleAccessEnabledProviders.split(",")) {
 			this.pathStyleAccessEnabledProviders.add(HpcIntegratedSystem.fromValue(s3Provider));
 		}
@@ -110,21 +118,35 @@ private HpcS3Connection(String pathStyleAccessEnabledProviders, int awsTransferM
 	}
 
 	// ---------------------------------------------------------------------//
-	// Methods
+	// Abstract Methods
 	// ---------------------------------------------------------------------//
 
 	/**
-	 * Authenticate a (system) data transfer account to S3 (AWS or 3rd Party
-	 * Provider)
+	 * Build a S3 async client for a 'S3 3rd Party Provider' account.
 	 *
-	 * @param dataTransferAccount A data transfer account to authenticate.
-	 * @param s3URLorRegion       The S3 URL if authenticating with a 3rd party S3
-	 *                            Provider (Cleversafe, Cloudian, etc), or Region if
-	 *                            authenticating w/ AWS.
-	 * @return An authenticated TransferManager object, or null if authentication
-	 *         failed.
-	 * @throws HpcException if authentication failed
+	 * @param credentialsProvider    The S3 credentials provider.
+	 * @param endpoint               The S3 provider endpoint.
+	 * @param pathStyleAccessEnabled true if the S3 3rd Party provider supports path
+	 *                               style access.
+	 * @param thresholdInBytes       The multipart upload threshold in bytes.
+	 * @return A S3 async client.
 	 */
+	protected abstract S3AsyncClient buildS3ProviderAsyncClient(StaticCredentialsProvider credentialsProvider,
+			URI endpoint, boolean pathStyleAccessEnabled, long thresholdInBytes);
+
+	/**
+	 * Build a S3 async client for an AWS S3 account.
+	 *
+	 * @param credentialsProvider The AWS credentials provider.
+	 * @param region              The AWS account region.
+	 * @return A S3 async client.
+	 */
+	protected abstract S3AsyncClient buildAwsAsyncClient(StaticCredentialsProvider credentialsProvider, String region);
+
+	// ---------------------------------------------------------------------//
+	// Public Methods
+	// ---------------------------------------------------------------------//
+
 	public Object authenticate(HpcIntegratedSystemAccount dataTransferAccount, String s3URLorRegion)
 			throws HpcException {
 		if (dataTransferAccount.getIntegratedSystem().equals(HpcIntegratedSystem.AWS)) {
@@ -139,13 +161,6 @@ public Object authenticate(HpcIntegratedSystemAccount dataTransferAccount, Strin
 		}
 	}
 
-	/**
-	 * Authenticate a (user) S3 account (AWS or 3rd Party Provider)
-	 *
-	 * @param s3Account AWS S3 account.
-	 * @return TransferManager
-	 * @throws HpcException if authentication failed
-	 */
 	public Object authenticate(HpcS3Account s3Account) throws HpcException {
 		if (!StringUtils.isEmpty(s3Account.getRegion())) {
 			return authenticateAWS(s3Account.getAccessKey(), s3Account.getSecretKey(), s3Account.getRegion());
@@ -160,13 +175,6 @@ public Object authenticate(HpcS3Account s3Account) throws HpcException {
 		}
 	}
 
-	/**
-	 * Get S3 Transfer Manager from an authenticated token.
-	 *
-	 * @param authenticatedToken An authenticated token.
-	 * @return A transfer manager object.
-	 * @throws HpcException on invalid authentication token.
-	 */
 	public S3TransferManager getTransferManager(Object authenticatedToken) throws HpcException {
 		if (!(authenticatedToken instanceof HpcS3)) {
 			throw new HpcException("Invalid S3 authentication token", HpcErrorType.INVALID_REQUEST_INPUT);
@@ -175,13 +183,6 @@ public S3TransferManager getTransferManager(Object authenticatedToken) throws Hp
 		return ((HpcS3) authenticatedToken).transferManager;
 	}
 
-	/**
-	 * Get S3 Client from an authenticated token.
-	 *
-	 * @param authenticatedToken An authenticated token.
-	 * @return A S3 client object.
-	 * @throws HpcException on invalid authentication token.
-	 */
 	public S3AsyncClient getClient(Object authenticatedToken) throws HpcException {
 		if (!(authenticatedToken instanceof HpcS3)) {
 			throw new HpcException("Invalid S3 authentication token", HpcErrorType.INVALID_REQUEST_INPUT);
@@ -190,13 +191,6 @@ public S3AsyncClient getClient(Object authenticatedToken) throws HpcException {
 		return ((HpcS3) authenticatedToken).client;
 	}
 
-	/**
-	 * Get S3 Presigner from an authenticated token.
-	 *
-	 * @param authenticatedToken An authenticated token.
-	 * @return A S3 presigner object.
-	 * @throws HpcException on invalid authentication token.
-	 */
 	public S3Presigner getPresigner(Object authenticatedToken) throws HpcException {
 		if (!(authenticatedToken instanceof HpcS3)) {
 			throw new HpcException("Invalid S3 authentication token", HpcErrorType.INVALID_REQUEST_INPUT);
@@ -205,13 +199,6 @@ public S3Presigner getPresigner(Object authenticatedToken) throws HpcException {
 		return ((HpcS3) authenticatedToken).presigner;
 	}
 
-	/**
-	 * Get S3 Provider from an authenticated token.
-	 *
-	 * @param authenticatedToken An authenticated token.
-	 * @return A transfer manager object.
-	 * @throws HpcException on invalid authentication token.
-	 */
 	public HpcIntegratedSystem getS3Provider(Object authenticatedToken) throws HpcException {
 		if (!(authenticatedToken instanceof HpcS3)) {
 			return null;
@@ -224,7 +211,7 @@ public HpcIntegratedSystem getS3Provider(Object authenticatedToken) throws HpcEx
 	// Helper Methods
 	// ---------------------------------------------------------------------//
 
-	private class HpcS3 {
+	protected class HpcS3 {
 		private S3TransferManager transferManager = null;
 		private S3AsyncClient client = null;
 		private S3Presigner presigner = null;
@@ -262,24 +249,12 @@ private Object authenticateS3Provider(String username, String password, String u
 		HpcS3 s3 = new HpcS3();
 		s3.provider = s3Provider;
 
-		try {
-			// If configured, start the AWS CRT logger.
-			if (!StringUtils.isEmpty(crtLogFile)) {
-				Log.initLoggingToFile(Log.LogLevel.Trace, crtLogFile);
-			}
-
-			// Instantiate a S3 async client.
-			S3CrtAsyncClientBuilder crtAsyncClientBuilder = S3AsyncClient.crtBuilder()
-					.credentialsProvider(s3ProviderCredentialsProvider).forcePathStyle(pathStyleAccessEnabled)
-					.endpointOverride(uri).minimumPartSizeInBytes(minimumUploadPartSize)
-					.thresholdInBytes(url.equalsIgnoreCase(GOOGLE_STORAGE_URL) ? FIVE_GB : multipartUploadThreshold).maxConcurrency(1);
-
-			if (trustAllCerts) {
-				crtAsyncClientBuilder.httpConfiguration(builder -> builder.trustAllCertificatesEnabled(true));
-				logger.warn("hpc.integration.s3.trustAllCerts property is set to true. CRT cert validation is off");
-			}
+		long thresholdInBytes = url.equalsIgnoreCase(GOOGLE_STORAGE_URL) ? FIVE_GB : multipartUploadThreshold;
 
-			s3.client = crtAsyncClientBuilder.build();
+		try {
+			// Instantiate a S3 async client (implementation specific).
+			s3.client = buildS3ProviderAsyncClient(s3ProviderCredentialsProvider, uri, pathStyleAccessEnabled,
+					thresholdInBytes);
 
 			// Instantiate the S3 transfer manager.
 			s3.transferManager = S3TransferManager.builder().s3Client(s3.client).executor(executorService).build();
@@ -292,7 +267,7 @@ private Object authenticateS3Provider(String username, String password, String u
 
 			return s3;
 
-		} catch (SdkException | CrtRuntimeException e) {
+		} catch (SdkException e) {
 			throw new HpcException(
 					"[S3] Failed to authenticate S3 Provider: " + s3Provider.value() + "] - " + e.getMessage(),
 					HpcErrorType.DATA_TRANSFER_ERROR, e);
@@ -317,15 +292,8 @@ private Object authenticateAWS(String accessKey, String secretKey, String region
 		s3.provider = HpcIntegratedSystem.AWS;
 
 		try {
-			// If configured, start the AWS CRT logger.
-			if (!StringUtils.isEmpty(crtLogFile)) {
-				Log.initLoggingToFile(Log.LogLevel.Trace, crtLogFile);
-			}
-
-			// Instantiate a S3 async client.
-			s3.client = S3AsyncClient.crtBuilder().credentialsProvider(awsCredentialsProvider).region(Region.of(region))
-					.minimumPartSizeInBytes(minimumUploadPartSize)
-					.thresholdInBytes(multipartUploadThreshold).maxConcurrency(1).build();
+			// Instantiate a S3 async client (implementation specific).
+			s3.client = buildAwsAsyncClient(awsCredentialsProvider, region);
 
 			// Instantiate the S3 transfer manager.
 			s3.transferManager = S3TransferManager.builder().s3Client(s3.client).executor(executorService).build();
@@ -336,7 +304,7 @@ private Object authenticateAWS(String accessKey, String secretKey, String region
 
 			return s3;
 
-		} catch (SdkException | CrtRuntimeException e) {
+		} catch (SdkException e) {
 			throw new HpcException("[S3] Failed to authenticate S3 in region " + region + "] - " + e.getMessage(),
 					HpcErrorType.DATA_TRANSFER_ERROR, e);
 		}