Skip to content

Commit 78fba94

Browse files
author
Luke Sikina
committed
[ALS-7200]
- Switch to async upload - Allows for 5G+ uploads to S3
1 parent c729973 commit 78fba94

File tree

7 files changed

+46
-37
lines changed

7 files changed

+46
-37
lines changed

‎uploader/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@
5959
</dependency>
6060
<dependency>
6161
<groupId>software.amazon.awssdk</groupId>
62-
<artifactId>apache-client</artifactId>
62+
<artifactId>netty-nio-client</artifactId>
6363
<version>${aws.version}</version>
6464
</dependency>
6565
<dependency>

‎uploader/src/main/java/edu/harvard/dbmi/avillach/dataupload/aws/AWSClientBuilder.java

+9-6
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,9 @@
88
import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
99
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
1010
import software.amazon.awssdk.http.SdkHttpClient;
11+
import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
12+
import software.amazon.awssdk.services.s3.S3AsyncClient;
13+
import software.amazon.awssdk.services.s3.S3AsyncClientBuilder;
1114
import software.amazon.awssdk.services.s3.S3Client;
1215
import software.amazon.awssdk.services.s3.S3ClientBuilder;
1316
import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
@@ -25,23 +28,23 @@ public class AWSClientBuilder {
2528

2629
private final Map<String, SiteAWSInfo> sites;
2730
private final StsClientProvider stsClientProvider;
28-
private final S3ClientBuilder s3ClientBuilder;
29-
private final SdkHttpClient sdkHttpClient;
31+
private final S3AsyncClientBuilder s3ClientBuilder;
32+
private final SdkAsyncHttpClient sdkHttpClient;
3033

3134
@Autowired
3235
public AWSClientBuilder(
3336
Map<String, SiteAWSInfo> sites,
3437
StsClientProvider stsClientProvider,
35-
S3ClientBuilder s3ClientBuilder,
36-
@Autowired(required = false) SdkHttpClient sdkHttpClient
38+
S3AsyncClientBuilder s3ClientBuilder,
39+
@Autowired(required = false) SdkAsyncHttpClient sdkHttpClient
3740
) {
3841
this.sites = sites;
3942
this.stsClientProvider = stsClientProvider;
4043
this.s3ClientBuilder = s3ClientBuilder;
4144
this.sdkHttpClient = sdkHttpClient;
4245
}
4346

44-
public Optional<S3Client> buildClientForSite(String siteName) {
47+
public Optional<S3AsyncClient> buildClientForSite(String siteName) {
4548
log.info("Building client for site {}", siteName);
4649
if (!sites.containsKey(siteName)) {
4750
log.warn("Could not find site {}", siteName);
@@ -78,7 +81,7 @@ public Optional<S3Client> buildClientForSite(String siteName) {
7881
return Optional.of(buildFromProvider(provider));
7982
}
8083

81-
private S3Client buildFromProvider(StaticCredentialsProvider provider) {
84+
private S3AsyncClient buildFromProvider(StaticCredentialsProvider provider) {
8285
if (sdkHttpClient == null) {
8386
return s3ClientBuilder.credentialsProvider(provider).build();
8487
}

‎uploader/src/main/java/edu/harvard/dbmi/avillach/dataupload/aws/S3StateVerifier.java

+4-2
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import org.springframework.beans.factory.annotation.Autowired;
88
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
99
import org.springframework.stereotype.Component;
10+
import software.amazon.awssdk.core.async.AsyncRequestBody;
1011
import software.amazon.awssdk.core.sync.RequestBody;
1112
import software.amazon.awssdk.services.s3.model.*;
1213

@@ -17,6 +18,7 @@
1718
import java.time.temporal.ChronoUnit;
1819
import java.util.Map;
1920
import java.util.Optional;
21+
import java.util.concurrent.CompletableFuture;
2022

2123
@ConditionalOnProperty(name = "production", havingValue = "true")
2224
@Component
@@ -59,7 +61,6 @@ private Optional<String> deleteFileFromBucket(String s, SiteAWSInfo info) {
5961
DeleteObjectRequest request = DeleteObjectRequest.builder().bucket(info.bucket()).key(s).build();
6062
return clientBuilder.buildClientForSite(info.siteName())
6163
.map(c -> c.deleteObject(request))
62-
.map(DeleteObjectResponse::deleteMarker)
6364
.map((ignored) -> s);
6465
}
6566

@@ -74,7 +75,8 @@ private String waitABit(String s) {
7475

7576
private Optional<String> uploadFileFromPath(Path p, SiteAWSInfo info) {
7677
LOG.info("Verifying upload capabilities");
77-
RequestBody body = RequestBody.fromFile(p.toFile());
78+
AsyncRequestBody body = AsyncRequestBody.fromFile(p.toFile());
79+
7880
PutObjectRequest request = PutObjectRequest.builder()
7981
.bucket(info.bucket())
8082
.serverSideEncryption(ServerSideEncryption.AWS_KMS)

‎uploader/src/main/java/edu/harvard/dbmi/avillach/dataupload/hpds/HttpClientConfig.java

+6-6
Original file line numberDiff line numberDiff line change
@@ -14,9 +14,9 @@
1414
import org.springframework.context.annotation.Bean;
1515
import org.springframework.context.annotation.Configuration;
1616
import org.springframework.util.StringUtils;
17-
import software.amazon.awssdk.http.SdkHttpClient;
18-
import software.amazon.awssdk.http.apache.ApacheHttpClient;
19-
import software.amazon.awssdk.http.apache.ProxyConfiguration;
17+
import software.amazon.awssdk.http.nio.netty.ProxyConfiguration;
18+
import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
19+
import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
2020

2121
@Configuration
2222
public class HttpClientConfig {
@@ -45,17 +45,17 @@ public HttpClient getHttpClient() {
4545

4646

4747
@Bean
48-
public SdkHttpClient getSdkClient() {
48+
public SdkAsyncHttpClient getSdkClient() {
4949
if (!StringUtils.hasLength(proxyUser)) {
5050
return null;
5151
}
5252
LOG.info("Found proxy user {}, will configure sdk proxy", proxyUser);
53-
ProxyConfiguration proxy = ProxyConfiguration.builder()
53+
var proxy = ProxyConfiguration.builder()
5454
.useSystemPropertyValues(true)
5555
.username(proxyUser)
5656
.password(proxyPassword)
5757
.build();
58-
return ApacheHttpClient.builder()
58+
return NettyNioAsyncHttpClient.builder()
5959
.proxyConfiguration(proxy)
6060
.build();
6161
}

‎uploader/src/main/java/edu/harvard/dbmi/avillach/dataupload/upload/DataUploadService.java

+2-1
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
1515
import org.springframework.stereotype.Service;
1616
import software.amazon.awssdk.awscore.exception.AwsServiceException;
17+
import software.amazon.awssdk.core.async.AsyncRequestBody;
1718
import software.amazon.awssdk.core.exception.SdkClientException;
1819
import software.amazon.awssdk.core.sync.RequestBody;
1920
import software.amazon.awssdk.services.s3.S3ClientBuilder;
@@ -119,7 +120,7 @@ private static void deleteFile(Path data) {
119120

120121
private boolean uploadFileFromPath(Path p, SiteAWSInfo site, String dir) {
121122
try {
122-
RequestBody body = RequestBody.fromFile(p.toFile());
123+
AsyncRequestBody body = AsyncRequestBody.fromFile(p.toFile());
123124
PutObjectRequest request = PutObjectRequest.builder()
124125
.bucket(site.bucket())
125126
.serverSideEncryption(ServerSideEncryption.AWS_KMS)

‎uploader/src/test/java/edu/harvard/dbmi/avillach/dataupload/aws/AWSClientBuilderTest.java

+15-15
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,8 @@
1313
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
1414
import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
1515
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
16-
import software.amazon.awssdk.services.s3.S3Client;
17-
import software.amazon.awssdk.services.s3.S3ClientBuilder;
16+
import software.amazon.awssdk.services.s3.S3AsyncClient;
17+
import software.amazon.awssdk.services.s3.S3AsyncClientBuilder;
1818
import software.amazon.awssdk.services.sts.StsClient;
1919
import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
2020
import software.amazon.awssdk.services.sts.model.AssumeRoleResponse;
@@ -38,7 +38,7 @@ class AWSClientBuilderTest {
3838
StsClientProvider stsClientProvider;
3939

4040
@MockBean
41-
S3ClientBuilder s3ClientBuilder;
41+
S3AsyncClientBuilder S3AsyncClientBuilder;
4242

4343
@Autowired
4444
AWSClientBuilder subject;
@@ -48,8 +48,8 @@ void shouldNotBuildClientIfSiteDNE() {
4848
Mockito.when(sites.get("Narnia"))
4949
.thenReturn(null);
5050

51-
Optional<S3Client> actual = subject.buildClientForSite("Narnia");
52-
Optional<S3Client> expected = Optional.empty();
51+
Optional<S3AsyncClient> actual = subject.buildClientForSite("Narnia");
52+
Optional<S3AsyncClient> expected = Optional.empty();
5353

5454
Assertions.assertEquals(expected, actual);
5555
}
@@ -72,8 +72,8 @@ void shouldNotBuildClientIfRoleRequestFails() {
7272
Mockito.when(stsClientProvider.createClient())
7373
.thenReturn(Optional.of(stsClient));
7474

75-
Optional<S3Client> actual = subject.buildClientForSite("bch");
76-
Optional<S3Client> expected = Optional.empty();
75+
Optional<S3AsyncClient> actual = subject.buildClientForSite("bch");
76+
Optional<S3AsyncClient> expected = Optional.empty();
7777

7878
Assertions.assertEquals(expected, actual);
7979
}
@@ -111,14 +111,14 @@ void shouldBuildClient() {
111111

112112
StaticCredentialsProvider provider = StaticCredentialsProvider.create(sessionCredentials);
113113
ArgumentMatcher<AwsCredentialsProvider> credsMatcher = (AwsCredentialsProvider p) -> p.toString().equals(provider.toString());
114-
S3Client s3Client = Mockito.mock(S3Client.class);
115-
Mockito.when(s3ClientBuilder.credentialsProvider(Mockito.argThat(credsMatcher)))
116-
.thenReturn(s3ClientBuilder);
117-
Mockito.when(s3ClientBuilder.build())
118-
.thenReturn(s3Client);
119-
120-
Optional<S3Client> actual = subject.buildClientForSite("bch");
121-
Optional<S3Client> expected = Optional.of(s3Client);
114+
S3AsyncClient S3AsyncClient = Mockito.mock(S3AsyncClient.class);
115+
Mockito.when(S3AsyncClientBuilder.credentialsProvider(Mockito.argThat(credsMatcher)))
116+
.thenReturn(S3AsyncClientBuilder);
117+
Mockito.when(S3AsyncClientBuilder.build())
118+
.thenReturn(S3AsyncClient);
119+
120+
Optional<S3AsyncClient> actual = subject.buildClientForSite("bch");
121+
Optional<S3AsyncClient> expected = Optional.of(S3AsyncClient);
122122

123123
Assertions.assertEquals(expected, actual);
124124
}

‎uploader/src/test/java/edu/harvard/dbmi/avillach/dataupload/upload/DataUploadServiceTest.java

+9-6
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,9 @@
1616
import org.springframework.boot.test.context.SpringBootTest;
1717
import org.springframework.test.util.ReflectionTestUtils;
1818
import software.amazon.awssdk.awscore.exception.AwsServiceException;
19+
import software.amazon.awssdk.core.async.AsyncRequestBody;
1920
import software.amazon.awssdk.core.sync.RequestBody;
21+
import software.amazon.awssdk.services.s3.S3AsyncClient;
2022
import software.amazon.awssdk.services.s3.S3Client;
2123
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
2224
import software.amazon.awssdk.services.s3.model.PutObjectResponse;
@@ -26,6 +28,7 @@
2628
import java.nio.file.Path;
2729
import java.util.Map;
2830
import java.util.Optional;
31+
import java.util.concurrent.CompletableFuture;
2932
import java.util.concurrent.Semaphore;
3033

3134
@SpringBootTest
@@ -44,7 +47,7 @@ class DataUploadServiceTest {
4447
private Path sharingRoot;
4548

4649
@Mock
47-
S3Client s3Client;
50+
S3AsyncClient s3Client;
4851

4952
@Mock
5053
AWSClientBuilder s3;
@@ -102,14 +105,14 @@ void shouldNotUploadDataIfAWSUpset(@TempDir Path tempDir) throws IOException, In
102105
Mockito.when(sharingRoot.toString()).thenReturn(tempDir.toString());
103106
Mockito.when(hpds.writePhenotypicData(q)).thenReturn(true);
104107
Mockito.when(s3.buildClientForSite("bch")).thenReturn(Optional.of(s3Client));
105-
Mockito.when(s3Client.putObject(Mockito.any(PutObjectRequest.class), Mockito.any(RequestBody.class)))
108+
Mockito.when(s3Client.putObject(Mockito.any(PutObjectRequest.class), Mockito.any(AsyncRequestBody.class)))
106109
.thenThrow(AwsServiceException.builder().build());
107110

108111
subject.uploadData(q, DataType.Phenotypic, "bch");
109112

110113
Mockito.verify(statusService, Mockito.times(1)).setPhenotypicStatus(q, UploadStatus.Querying);
111114
Mockito.verify(statusService, Mockito.times(1)).setPhenotypicStatus(q, UploadStatus.Uploading);
112-
Mockito.verify(s3Client, Mockito.times(1)).putObject(Mockito.any(PutObjectRequest.class), Mockito.any(RequestBody.class));
115+
Mockito.verify(s3Client, Mockito.times(1)).putObject(Mockito.any(PutObjectRequest.class), Mockito.any(AsyncRequestBody.class));
113116
Mockito.verify(statusService, Mockito.times(1)).setPhenotypicStatus(q, UploadStatus.Error);
114117
Mockito.verify(uploadLock, Mockito.times(1)).acquire();
115118
Mockito.verify(uploadLock, Mockito.times(1)).release();
@@ -129,14 +132,14 @@ void shouldUploadData(@TempDir Path tempDir) throws IOException, InterruptedExce
129132
Mockito.when(sharingRoot.toString()).thenReturn(tempDir.toString());
130133
Mockito.when(hpds.writePhenotypicData(q)).thenReturn(true);
131134
Mockito.when(s3.buildClientForSite("bch")).thenReturn(Optional.of(s3Client));
132-
Mockito.when(s3Client.putObject(Mockito.any(PutObjectRequest.class), Mockito.any(RequestBody.class)))
133-
.thenReturn(Mockito.mock(PutObjectResponse.class));
135+
Mockito.when(s3Client.putObject(Mockito.any(PutObjectRequest.class), Mockito.any(AsyncRequestBody.class)))
136+
.thenReturn(CompletableFuture.completedFuture(Mockito.mock(PutObjectResponse.class)));
134137

135138
subject.uploadData(q, DataType.Phenotypic, "bch");
136139

137140
Mockito.verify(statusService, Mockito.times(1)).setPhenotypicStatus(q, UploadStatus.Querying);
138141
Mockito.verify(statusService, Mockito.times(1)).setPhenotypicStatus(q, UploadStatus.Uploading);
139-
Mockito.verify(s3Client, Mockito.times(1)).putObject(Mockito.any(PutObjectRequest.class), Mockito.any(RequestBody.class));
142+
Mockito.verify(s3Client, Mockito.times(1)).putObject(Mockito.any(PutObjectRequest.class), Mockito.any(AsyncRequestBody.class));
140143
Mockito.verify(statusService, Mockito.times(1)).setPhenotypicStatus(q, UploadStatus.Uploaded);
141144
Assertions.assertFalse(Files.exists(fileToUpload));
142145
Mockito.verify(uploadLock, Mockito.times(1)).acquire();

0 commit comments

Comments
 (0)