16357938490

Committed 17 Jul 2025 11:06PM UTC coverage: 65.244% (-0.1%) from 65.359%

Build # 16357938490

Build Type

push

github

Committed by

web-flow

Commit Message

TRUNK-6318: Add S3 Storage Service (#5110)

Coverage Stats

111 of 156 new or added lines in 5 files covered. (71.15%)

48 existing lines in 8 files now uncovered.

23568 of 36123 relevant lines covered (65.24%)

0.65 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

65.88

/api/src/main/java/org/openmrs/api/storage/S3StorageService.java

/**
 * This Source Code Form is subject to the terms of the Mozilla Public License,
 * v. 2.0. If a copy of the MPL was not distributed with this file, You can
 * obtain one at http://mozilla.org/MPL/2.0/. OpenMRS is also distributed under
 * the terms of the Healthcare Disclaimer located at http://openmrs.org/license.
 *
 * Copyright (C) OpenMRS Inc. OpenMRS is a registered trademark and the OpenMRS
 * graphic logo is a trademark of OpenMRS Inc.
 */

package org.openmrs.api.storage;

import java.io.IOException;
import java.io.InputStream;
import java.io.InterruptedIOException;
import java.io.UncheckedIOException;
import java.nio.file.FileAlreadyExistsException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.stream.Stream;

import org.apache.commons.lang.StringUtils;
import org.openmrs.api.StorageService;
import org.openmrs.api.stream.StreamDataService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Conditional;
import org.springframework.stereotype.Service;
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
import software.amazon.awssdk.core.ResponseInputStream;
import software.amazon.awssdk.core.async.AsyncRequestBody;
import software.amazon.awssdk.core.async.AsyncResponseTransformer;
import software.amazon.awssdk.core.async.BlockingInputStreamAsyncRequestBody;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.s3.S3AsyncClient;
import software.amazon.awssdk.services.s3.S3AsyncClientBuilder;
import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
import software.amazon.awssdk.services.s3.model.DeleteObjectResponse;
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.ListObjectsV2Request;
import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
import software.amazon.awssdk.services.s3.model.PutObjectResponse;
import software.amazon.awssdk.services.s3.model.S3Exception;
import software.amazon.awssdk.services.s3.model.S3Object;
import software.amazon.awssdk.services.s3.model.S3Response;

/**
 * Amazon S3-based implementation of {@link StorageService}.
 * <p>
 * It uses S3AsyncClient under the hood for all operations to handle large files in the most performant way with
 * multipart enabled by default.
 * 
 * <p>It can be configured with the following properties: <i>storage.s3.bucketName, storage.s3.accessKeyId, 
 * storage.s3.secretAccessKey, storage.s3.region, storage.s3.multipartEnabled</i>
 * <p>
 * If not configured, the default bucket name is <i>'openmrs'</i>.
 * 
 * <p>
 * Credentials and region can also be provided by standard S3AsyncClient means. See 
 * <a href="https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/regions/providers/DefaultAwsRegionProviderChain.html">DefaultAwsRegionProviderChain</a>
 * and 
 * <a href="https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/DefaultCredentialsProvider.html">DefaultCredentialsProvider</a>
 * 
 * @since 2.8.0
 */
@Service
@Qualifier("s3")
@Conditional(StorageServiceCondition.class)
public class S3StorageService extends BaseStorageService implements StorageService {

    private static final Logger log = LoggerFactory.getLogger(S3StorageService.class);

    protected S3AsyncClient s3AsyncClient;
        
    private final String bucketName;

    @Autowired
    public S3StorageService(
     StreamDataService streamDataService,
        @Value("${storage.s3.accessKeyId:}") String accessKeyId,
        @Value("${storage.s3.secretAccessKey:}") String secretAccessKey,
        @Value("${storage.s3.region:}") String region,
                 @Value("${storage.s3.bucketName:openmrs}") String bucketName,
            @Value("${storage.s3.multipartEnabled:true}") boolean multipartEnabled
    ) {
        super(streamDataService);
                this.bucketName = bucketName;

                S3AsyncClientBuilder s3AsyncClientBuilder = S3AsyncClient.builder().multipartEnabled(multipartEnabled);

                if (StringUtils.isNotBlank(accessKeyId) || StringUtils.isNotBlank(secretAccessKey)) {
                        log.info("Using storage.s3.accessKeyId and storage.s3.secretAccessKey for S3 client");
                        if (StringUtils.isBlank(accessKeyId) || StringUtils.isBlank(secretAccessKey)) {
                                throw new IllegalArgumentException("Both storage.s3.accessKeyId and storage.s3.secretAccessKey " +
                                                "must be provided");
                        }
                        StaticCredentialsProvider awsCredentials = StaticCredentialsProvider.create(
                                AwsBasicCredentials.create(accessKeyId, secretAccessKey));
                        s3AsyncClientBuilder = s3AsyncClientBuilder.credentialsProvider(awsCredentials);
        }

                if (StringUtils.isNotBlank(region)) {
                        log.info("Using storage.s3.region '{}' for S3 client", region);
                        s3AsyncClientBuilder = s3AsyncClientBuilder.region(Region.of(region));
                }
                
                this.s3AsyncClient = s3AsyncClientBuilder.build();
    }
        
    public S3StorageService(StreamDataService streamService, S3AsyncClient s3AsyncClient, 
                                                        String bucketName) {
        super(streamService);
        this.s3AsyncClient = s3AsyncClient;
        this.bucketName = bucketName;
    }

    public InputStream getData(String key) throws IOException {
                CompletableFuture<ResponseInputStream<GetObjectResponse>> object = s3AsyncClient.getObject(
                        GetObjectRequest.builder().bucket(bucketName).key(encodeKey(key)).build(), AsyncResponseTransformer.toBlockingInputStream());
                return waitForResponse(object);
    }

        private <T> T waitForResponse(CompletableFuture<T> object) throws IOException {
                T result;
                try {
                        result = object.get();
                } catch (InterruptedException e) {
                        throw new InterruptedIOException(e.getMessage());
                } catch (ExecutionException e) {
                        throw new IOException(e);
                }
                return result;
        }

        public ObjectMetadata getMetadata(String key) throws IOException {
                HeadObjectRequest request = HeadObjectRequest.builder().bucket(bucketName).key(encodeKey(key)).build();
                CompletableFuture<HeadObjectResponse> headRequest = s3AsyncClient.headObject(request);
                HeadObjectResponse awsMetadata = waitForResponse(headRequest);
                ObjectMetadata metadata = new ObjectMetadata();
                metadata.setMimeType(awsMetadata.contentType());
                metadata.setLength(awsMetadata.contentLength());
                return metadata;
    }

    public Stream<String> getKeys(String moduleIdOrGroup, String keyPrefix) throws IOException {
                String key = newKey(moduleIdOrGroup, keyPrefix, null);
                
        ListObjectsV2Request req = ListObjectsV2Request.builder().bucket(bucketName)
                        .prefix(encodeKey(key)).build();
        CompletableFuture<ListObjectsV2Response> listObjectsRequest = s3AsyncClient.listObjectsV2(req);
                ListObjectsV2Response listObjects = waitForResponse(listObjectsRequest);
                
                return listObjects.contents().stream().map(S3Object::key).map(foundKey -> {
                        foundKey = decodeKey(foundKey);
                        String dirContent = foundKey.substring(key.length());
                        int subdir = dirContent.indexOf("/");
                        if (subdir != -1) {
                                // Return only subdirectories without their content
                                String dirOnly = dirContent.substring(0, subdir + 1);
                                return key + dirOnly;
                        } else {
                                return foundKey;
                        }
                }).distinct(); // Remove duplicate subdirectories
        }

    public boolean purgeData(String key) throws IOException {
                if (exists(key)) {
                        DeleteObjectRequest request = DeleteObjectRequest.builder().bucket(bucketName).key(encodeKey(key)).build();
                        CompletableFuture<DeleteObjectResponse> deleteRequest = s3AsyncClient.deleteObject(request);
                        return waitForBooleanResponse(deleteRequest);
                } else {
                        return false;
                }
    }
        
        private boolean waitForBooleanResponse(CompletableFuture<? extends S3Response> request) throws IOException {
                try {
                        request.get();
                        return true;
                } catch (InterruptedException e) {
                        throw new InterruptedIOException(e.getMessage());
                } catch (ExecutionException e) {
                        if (e.getCause() instanceof S3Exception) {
                                S3Exception s3e = (S3Exception) e.getCause();
                                if (s3e.statusCode() == 404) {
                                        return false;
                                }
                        }
                        throw new IOException(e);
                }
        }

    public boolean exists(String key) throws UncheckedIOException {
        HeadObjectRequest request = HeadObjectRequest.builder().bucket(bucketName).key(encodeKey(key)).build();
                CompletableFuture<HeadObjectResponse> headRequest = s3AsyncClient.headObject(request);
                try {
                        return waitForBooleanResponse(headRequest);
                } catch (IOException e) {
                        throw new UncheckedIOException(e);
                }
        }
        
    public String saveData(InputStream inputStream, ObjectMetadata metadata,
                           String moduleIdOrGroup, String keySuffix) throws IOException {
                metadata = (metadata == null) ? new ObjectMetadata() : metadata;

                String key = newKey(moduleIdOrGroup, keySuffix, metadata.getFilename());
                
                if (exists(key)) {
                        throw new FileAlreadyExistsException("Key " + key + " already exists");
                }
                
                PutObjectRequest request = PutObjectRequest.builder().bucket(bucketName).key(encodeKey(key))
                        .contentType(metadata.getMimeType()).contentLength(metadata.getLength()).build();

                try {
                        BlockingInputStreamAsyncRequestBody requestBody = AsyncRequestBody.forBlockingInputStream(metadata.getLength());
                        CompletableFuture<PutObjectResponse> putRequest = s3AsyncClient.putObject(request, requestBody);
                        requestBody.writeInputStream(inputStream);

                        waitForResponse(putRequest);
                } catch (Exception e) {
                        throw new IOException(e);
                }
                        
        return key;
    }
}

1	/**
2	* This Source Code Form is subject to the terms of the Mozilla Public License,
3	* v. 2.0. If a copy of the MPL was not distributed with this file, You can
4	* obtain one at http://mozilla.org/MPL/2.0/. OpenMRS is also distributed under
5	* the terms of the Healthcare Disclaimer located at http://openmrs.org/license.
6	*
7	* Copyright (C) OpenMRS Inc. OpenMRS is a registered trademark and the OpenMRS
8	* graphic logo is a trademark of OpenMRS Inc.
9	*/
10
11	package org.openmrs.api.storage;
12
13	import java.io.IOException;
14	import java.io.InputStream;
15	import java.io.InterruptedIOException;
16	import java.io.UncheckedIOException;
17	import java.nio.file.FileAlreadyExistsException;
18	import java.util.concurrent.CompletableFuture;
19	import java.util.concurrent.ExecutionException;
20	import java.util.stream.Stream;
21
22	import org.apache.commons.lang.StringUtils;
23	import org.openmrs.api.StorageService;
24	import org.openmrs.api.stream.StreamDataService;
25	import org.slf4j.Logger;
26	import org.slf4j.LoggerFactory;
27	import org.springframework.beans.factory.annotation.Autowired;
28	import org.springframework.beans.factory.annotation.Qualifier;
29	import org.springframework.beans.factory.annotation.Value;
30	import org.springframework.context.annotation.Conditional;
31	import org.springframework.stereotype.Service;
32	import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
33	import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
34	import software.amazon.awssdk.core.ResponseInputStream;
35	import software.amazon.awssdk.core.async.AsyncRequestBody;
36	import software.amazon.awssdk.core.async.AsyncResponseTransformer;
37	import software.amazon.awssdk.core.async.BlockingInputStreamAsyncRequestBody;
38	import software.amazon.awssdk.regions.Region;
39	import software.amazon.awssdk.services.s3.S3AsyncClient;
40	import software.amazon.awssdk.services.s3.S3AsyncClientBuilder;
41	import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
42	import software.amazon.awssdk.services.s3.model.DeleteObjectResponse;
43	import software.amazon.awssdk.services.s3.model.GetObjectRequest;
44	import software.amazon.awssdk.services.s3.model.GetObjectResponse;
45	import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
46	import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
47	import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
48	import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
49	import software.amazon.awssdk.services.s3.model.PutObjectRequest;
50	import software.amazon.awssdk.services.s3.model.PutObjectResponse;
51	import software.amazon.awssdk.services.s3.model.S3Exception;
52	import software.amazon.awssdk.services.s3.model.S3Object;
53	import software.amazon.awssdk.services.s3.model.S3Response;
54
55	/**
56	* Amazon S3-based implementation of {@link StorageService}.
57	* <p>
58	* It uses S3AsyncClient under the hood for all operations to handle large files in the most performant way with
59	* multipart enabled by default.
60	*
61	* <p>It can be configured with the following properties: <i>storage.s3.bucketName, storage.s3.accessKeyId,
62	* storage.s3.secretAccessKey, storage.s3.region, storage.s3.multipartEnabled</i>
63	* <p>
64	* If not configured, the default bucket name is <i>'openmrs'</i>.
65	*
66	* <p>
67	* Credentials and region can also be provided by standard S3AsyncClient means. See
68	* <a href="https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/regions/providers/DefaultAwsRegionProviderChain.html">DefaultAwsRegionProviderChain</a>
69	* and
70	* <a href="https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/DefaultCredentialsProvider.html">DefaultCredentialsProvider</a>
71	*
72	* @since 2.8.0
73	*/
74	@Service
75	@Qualifier("s3")
76	@Conditional(StorageServiceCondition.class)
77	public class S3StorageService extends BaseStorageService implements StorageService {
78
79	private static final Logger log = LoggerFactory.getLogger(S3StorageService.class);	1✔
80
81	protected S3AsyncClient s3AsyncClient;
82
83	private final String bucketName;
84
85	@Autowired
86	public S3StorageService(
87	StreamDataService streamDataService,
88	@Value("${storage.s3.accessKeyId:}") String accessKeyId,
89	@Value("${storage.s3.secretAccessKey:}") String secretAccessKey,
90	@Value("${storage.s3.region:}") String region,
91	@Value("${storage.s3.bucketName:openmrs}") String bucketName,
92	@Value("${storage.s3.multipartEnabled:true}") boolean multipartEnabled
93	) {
NEW 94	super(streamDataService);	×
NEW 95	this.bucketName = bucketName;	×
96
NEW 97	S3AsyncClientBuilder s3AsyncClientBuilder = S3AsyncClient.builder().multipartEnabled(multipartEnabled);	×
98
NEW 99	if (StringUtils.isNotBlank(accessKeyId) \|\| StringUtils.isNotBlank(secretAccessKey)) {	×
NEW 100	log.info("Using storage.s3.accessKeyId and storage.s3.secretAccessKey for S3 client");	×
NEW 101	if (StringUtils.isBlank(accessKeyId) \|\| StringUtils.isBlank(secretAccessKey)) {	×
NEW 102	throw new IllegalArgumentException("Both storage.s3.accessKeyId and storage.s3.secretAccessKey " +	×
103	"must be provided");
104	}
NEW 105	StaticCredentialsProvider awsCredentials = StaticCredentialsProvider.create(	×
NEW 106	AwsBasicCredentials.create(accessKeyId, secretAccessKey));	×
NEW 107	s3AsyncClientBuilder = s3AsyncClientBuilder.credentialsProvider(awsCredentials);	×
108	}
109
NEW 110	if (StringUtils.isNotBlank(region)) {	×
NEW 111	log.info("Using storage.s3.region '{}' for S3 client", region);	×
NEW 112	s3AsyncClientBuilder = s3AsyncClientBuilder.region(Region.of(region));	×
113	}
114
NEW 115	this.s3AsyncClient = s3AsyncClientBuilder.build();	×
NEW 116	}	×
117
118	public S3StorageService(StreamDataService streamService, S3AsyncClient s3AsyncClient,
119	String bucketName) {
120	super(streamService);	1✔
121	this.s3AsyncClient = s3AsyncClient;	1✔
122	this.bucketName = bucketName;	1✔
123	}	1✔
124
125	public InputStream getData(String key) throws IOException {
126	CompletableFuture<ResponseInputStream<GetObjectResponse>> object = s3AsyncClient.getObject(	1✔
127	GetObjectRequest.builder().bucket(bucketName).key(encodeKey(key)).build(), AsyncResponseTransformer.toBlockingInputStream());	1✔
128	return waitForResponse(object);	1✔
129	}
130
131	private <T> T waitForResponse(CompletableFuture<T> object) throws IOException {
132	T result;
133	try {
134	result = object.get();	1✔
NEW 135	} catch (InterruptedException e) {	×
NEW 136	throw new InterruptedIOException(e.getMessage());	×
137	} catch (ExecutionException e) {	1✔
138	throw new IOException(e);	1✔
139	}	1✔
140	return result;	1✔
141	}
142
143	public ObjectMetadata getMetadata(String key) throws IOException {
NEW 144	HeadObjectRequest request = HeadObjectRequest.builder().bucket(bucketName).key(encodeKey(key)).build();	×
NEW 145	CompletableFuture<HeadObjectResponse> headRequest = s3AsyncClient.headObject(request);	×
NEW 146	HeadObjectResponse awsMetadata = waitForResponse(headRequest);	×
NEW 147	ObjectMetadata metadata = new ObjectMetadata();	×
NEW 148	metadata.setMimeType(awsMetadata.contentType());	×
NEW 149	metadata.setLength(awsMetadata.contentLength());	×
NEW 150	return metadata;	×
151	}
152
153	public Stream<String> getKeys(String moduleIdOrGroup, String keyPrefix) throws IOException {
154	String key = newKey(moduleIdOrGroup, keyPrefix, null);	1✔
155
156	ListObjectsV2Request req = ListObjectsV2Request.builder().bucket(bucketName)	1✔
157	.prefix(encodeKey(key)).build();	1✔
158	CompletableFuture<ListObjectsV2Response> listObjectsRequest = s3AsyncClient.listObjectsV2(req);	1✔
159	ListObjectsV2Response listObjects = waitForResponse(listObjectsRequest);	1✔
160
161	return listObjects.contents().stream().map(S3Object::key).map(foundKey -> {	1✔
162	foundKey = decodeKey(foundKey);	1✔
163	String dirContent = foundKey.substring(key.length());	1✔
164	int subdir = dirContent.indexOf("/");	1✔
165	if (subdir != -1) {	1✔
166	// Return only subdirectories without their content
167	String dirOnly = dirContent.substring(0, subdir + 1);	1✔
168	return key + dirOnly;	1✔
169	} else {
170	return foundKey;	1✔
171	}
172	}).distinct(); // Remove duplicate subdirectories	1✔
173	}
174
175	public boolean purgeData(String key) throws IOException {
176	if (exists(key)) {	1✔
177	DeleteObjectRequest request = DeleteObjectRequest.builder().bucket(bucketName).key(encodeKey(key)).build();	1✔
178	CompletableFuture<DeleteObjectResponse> deleteRequest = s3AsyncClient.deleteObject(request);	1✔
179	return waitForBooleanResponse(deleteRequest);	1✔
180	} else {
181	return false;	1✔
182	}
183	}
184
185	private boolean waitForBooleanResponse(CompletableFuture<? extends S3Response> request) throws IOException {
186	try {
187	request.get();	1✔
188	return true;	1✔
NEW 189	} catch (InterruptedException e) {	×
NEW 190	throw new InterruptedIOException(e.getMessage());	×
191	} catch (ExecutionException e) {	1✔
192	if (e.getCause() instanceof S3Exception) {	1✔
193	S3Exception s3e = (S3Exception) e.getCause();	1✔
194	if (s3e.statusCode() == 404) {	1✔
195	return false;	1✔
196	}
197	}
NEW 198	throw new IOException(e);	×
199	}
200	}
201
202	public boolean exists(String key) throws UncheckedIOException {
203	HeadObjectRequest request = HeadObjectRequest.builder().bucket(bucketName).key(encodeKey(key)).build();	1✔
204	CompletableFuture<HeadObjectResponse> headRequest = s3AsyncClient.headObject(request);	1✔
205	try {
206	return waitForBooleanResponse(headRequest);	1✔
NEW 207	} catch (IOException e) {	×
NEW 208	throw new UncheckedIOException(e);	×
209	}
210	}
211
212	public String saveData(InputStream inputStream, ObjectMetadata metadata,
213	String moduleIdOrGroup, String keySuffix) throws IOException {
214	metadata = (metadata == null) ? new ObjectMetadata() : metadata;	1✔
215
216	String key = newKey(moduleIdOrGroup, keySuffix, metadata.getFilename());	1✔
217
218	if (exists(key)) {	1✔
219	throw new FileAlreadyExistsException("Key " + key + " already exists");	1✔
220	}
221
222	PutObjectRequest request = PutObjectRequest.builder().bucket(bucketName).key(encodeKey(key))	1✔
223	.contentType(metadata.getMimeType()).contentLength(metadata.getLength()).build();	1✔
224
225	try {
226	BlockingInputStreamAsyncRequestBody requestBody = AsyncRequestBody.forBlockingInputStream(metadata.getLength());	1✔
227	CompletableFuture<PutObjectResponse> putRequest = s3AsyncClient.putObject(request, requestBody);	1✔
228	requestBody.writeInputStream(inputStream);	1✔
229
230	waitForResponse(putRequest);	1✔
231	} catch (Exception e) {	1✔
232	throw new IOException(e);	1✔
233	}	1✔
234
235	return key;	1✔
236	}
237	}

openmrs / openmrs-core / 16357938490

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous