Skip to content

Commit caab326

Browse files
Yavor16Yavor16
authored
Adopt azure sdk (#1857)
Adopt Azure sdk for Objectstore
1 parent 032a70e commit caab326

14 files changed

Lines changed: 689 additions & 93 deletions

File tree

‎multiapps-controller-persistence/pom.xml‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -104,8 +104,12 @@
104104
<artifactId>aws-s3</artifactId>
105105
</dependency>
106106
<dependency>
107-
<groupId>org.apache.jclouds.provider</groupId>
108-
<artifactId>azureblob</artifactId>
107+
<groupId>com.azure</groupId>
108+
<artifactId>azure-storage-blob</artifactId>
109+
</dependency>
110+
<dependency>
111+
<groupId>com.azure</groupId>
112+
<artifactId>azure-core-http-jdk-httpclient</artifactId>
109113
</dependency>
110114
<dependency>
111115
<groupId>org.apache.jclouds</groupId>

‎multiapps-controller-persistence/src/main/java/module-info.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,9 @@
2828
requires transitive org.cloudfoundry.multiapps.controller.api;
2929

3030
requires aliyun.sdk.oss;
31+
requires com.azure.core;
32+
requires com.azure.storage.blob;
33+
requires com.azure.core.http.jdk.httpclient;
3134
requires com.fasterxml.jackson.annotation;
3235
requires com.fasterxml.jackson.databind;
3336
requires com.google.auth;

‎multiapps-controller-persistence/src/main/java/org/cloudfoundry/multiapps/controller/persistence/Messages.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,8 @@ public final class Messages {
5151

5252
// ERROR log messages:
5353
public static final String UPLOAD_STREAM_FAILED_TO_CLOSE = "Cannot close file upload stream";
54+
public static final String CANNOT_PARSE_CONTAINER_URI_OF_OBJECT_STORE = "Cannot parse container_uri of object store";
55+
public static final String MISSING_CONTAINER_URI_IN_THE_CREDENTIALS = "Missing container uri in the credentials";
5456

5557
// WARN log messages:
5658
public static final String COULD_NOT_CLOSE_RESULT_SET = "Could not close result set.";

‎multiapps-controller-persistence/src/main/java/org/cloudfoundry/multiapps/controller/persistence/services/AzureObjectStoreFileStorage.java‎

Lines changed: 228 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,228 @@
1+
package org.cloudfoundry.multiapps.controller.persistence.services;
2+
3+
import java.io.IOException;
4+
import java.io.InputStream;
5+
import java.net.MalformedURLException;
6+
import java.net.URL;
7+
import java.time.LocalDateTime;
8+
import java.util.List;
9+
import java.util.Map;
10+
import java.util.Set;
11+
import java.util.function.Predicate;
12+
import java.util.stream.Collectors;
13+
14+
import com.azure.core.http.HttpClient;
15+
import com.azure.core.http.jdk.httpclient.JdkHttpClientBuilder;
16+
import com.azure.core.http.policy.ExponentialBackoffOptions;
17+
import com.azure.core.http.policy.RetryOptions;
18+
import com.azure.storage.blob.BlobClient;
19+
import com.azure.storage.blob.BlobContainerClient;
20+
import com.azure.storage.blob.BlobServiceClient;
21+
import com.azure.storage.blob.BlobServiceClientBuilder;
22+
import com.azure.storage.blob.models.BlobItem;
23+
import com.azure.storage.blob.models.BlobListDetails;
24+
import com.azure.storage.blob.models.BlobRange;
25+
import com.azure.storage.blob.models.BlobStorageException;
26+
import com.azure.storage.blob.models.ListBlobsOptions;
27+
import com.azure.storage.blob.models.ParallelTransferOptions;
28+
import com.azure.storage.blob.options.BlobParallelUploadOptions;
29+
import org.cloudfoundry.multiapps.controller.persistence.Messages;
30+
import org.cloudfoundry.multiapps.controller.persistence.model.FileEntry;
31+
import org.cloudfoundry.multiapps.controller.persistence.util.ObjectStoreConstants;
32+
import org.cloudfoundry.multiapps.controller.persistence.util.ObjectStoreFilter;
33+
import org.cloudfoundry.multiapps.controller.persistence.util.ObjectStoreMapper;
34+
35+
public class AzureObjectStoreFileStorage extends ObjectStoreFileStorage {
36+
37+
private static final String SAS_TOKEN = "sas_token";
38+
private static final String CONTAINER_NAME = "container_name";
39+
private static final String CONTAINER_URI = "container_uri";
40+
private static final long MAX_SINGLE_UPLOAD_SIZE = 15L * 1024 * 1024; // 15MB
41+
private static final long BLOCK_SIZE = 8L * 1024 * 1024; // 8MB
42+
private static final int MAX_CONCURRENCY = 5;
43+
private final HttpClient httpClient;
44+
private final BlobContainerClient containerClient;
45+
46+
public AzureObjectStoreFileStorage(Map<String, Object> credentials) {
47+
this.httpClient = new JdkHttpClientBuilder().build();
48+
this.containerClient = createContainerClient(credentials);
49+
}
50+
51+
@Override
52+
public void addFile(FileEntry fileEntry, InputStream content) throws FileStorageException {
53+
BlobClient blobClient = containerClient.getBlobClient(fileEntry.getId());
54+
try {
55+
ParallelTransferOptions pto = new ParallelTransferOptions().setMaxSingleUploadSizeLong(MAX_SINGLE_UPLOAD_SIZE)
56+
.setMaxConcurrency(MAX_CONCURRENCY)
57+
.setBlockSizeLong(BLOCK_SIZE);
58+
59+
BlobParallelUploadOptions blobParallelUploadOptions = new BlobParallelUploadOptions(content);
60+
blobParallelUploadOptions.setParallelTransferOptions(pto);
61+
blobParallelUploadOptions.setMetadata(ObjectStoreMapper.createFileEntryMetadata(fileEntry));
62+
63+
blobClient.uploadWithResponse(blobParallelUploadOptions,
64+
ObjectStoreConstants.AZURE_OBJECT_STORE_TOTAL_TIMEOUT_CONFIG_IN_MINUTES, null);
65+
} catch (BlobStorageException e) {
66+
throw new FileStorageException(e);
67+
}
68+
}
69+
70+
@Override
71+
public List<FileEntry> getFileEntriesWithoutContent(List<FileEntry> fileEntries) throws FileStorageException {
72+
Set<String> existingFiles = getAllEntriesNames();
73+
return fileEntries.stream()
74+
.filter(fileEntry -> !existingFiles.contains(fileEntry.getId()))
75+
.toList();
76+
}
77+
78+
@Override
79+
protected boolean existsInObjectStore(FileEntry fileEntry) {
80+
return containerClient.getBlobClient(fileEntry.getId())
81+
.exists();
82+
}
83+
84+
@Override
85+
public void deleteFile(String id, String space) throws FileStorageException {
86+
BlobClient blobClient = containerClient.getBlobClient(id);
87+
try {
88+
blobClient.deleteIfExists();
89+
} catch (BlobStorageException e) {
90+
throw new FileStorageException(e);
91+
}
92+
}
93+
94+
@Override
95+
public void deleteFilesBySpaceIds(List<String> spaceIds) throws FileStorageException {
96+
removeBlobsByFilter(blob -> ObjectStoreFilter.filterBySpaceIds(blob.getMetadata(), spaceIds));
97+
}
98+
99+
@Override
100+
public void deleteFilesBySpaceAndNamespace(String space, String namespace) {
101+
removeBlobsByFilter(blob -> ObjectStoreFilter.filterBySpaceAndNamespace(blob.getMetadata(), space, namespace));
102+
}
103+
104+
@Override
105+
public int deleteFilesModifiedBefore(LocalDateTime modificationTime) throws FileStorageException {
106+
return removeBlobsByFilter(
107+
blob -> ObjectStoreFilter.filterByModificationTime(blob.getMetadata(), blob.getName(), modificationTime));
108+
}
109+
110+
@Override
111+
public <T> T processFileContent(String space, String id, FileContentProcessor<T> fileContentProcessor) throws FileStorageException {
112+
FileEntry fileEntry = ObjectStoreMapper.createFileEntry(space, id);
113+
try (InputStream inputStream = openBlobInputStream(fileEntry)) {
114+
return fileContentProcessor.process(inputStream);
115+
} catch (Exception e) {
116+
throw new FileStorageException(e);
117+
}
118+
}
119+
120+
private InputStream openBlobInputStream(FileEntry fileEntry) throws FileStorageException {
121+
BlobClient blobClient = containerClient.getBlobClient(fileEntry.getId());
122+
try {
123+
return blobClient.openInputStream();
124+
} catch (BlobStorageException e) {
125+
throw new FileStorageException(e);
126+
}
127+
}
128+
129+
@Override
130+
public InputStream openInputStream(String space, String id) throws FileStorageException {
131+
FileEntry fileEntry = ObjectStoreMapper.createFileEntry(space, id);
132+
return openBlobInputStream(fileEntry);
133+
}
134+
135+
@Override
136+
public void testConnection() {
137+
containerClient.getBlobClient("test");
138+
}
139+
140+
@Override
141+
public void deleteFilesByIds(List<String> fileIds) throws FileStorageException {
142+
removeBlobsByFilter(blob -> fileIds.contains(blob.getName()));
143+
}
144+
145+
@Override
146+
public <T> T processArchiveEntryContent(FileContentToProcess fileContentToProcess, FileContentProcessor<T> fileContentProcessor)
147+
throws FileStorageException {
148+
FileEntry fileEntry = ObjectStoreMapper.createFileEntry(fileContentToProcess.getSpaceGuid(), fileContentToProcess.getGuid());
149+
BlobClient blobClient = containerClient.getBlobClient(fileEntry.getId());
150+
long contentSize = fileContentToProcess.getEndOffset() - fileContentToProcess.getStartOffset();
151+
BlobRange blobRange = new BlobRange(fileContentToProcess.getStartOffset(), contentSize);
152+
153+
try {
154+
return fileContentProcessor.process(blobClient.openInputStream(blobRange, null));
155+
} catch (IOException e) {
156+
throw new FileStorageException(e);
157+
}
158+
}
159+
160+
protected BlobContainerClient createContainerClient(Map<String, Object> credentials) {
161+
BlobServiceClient serviceClient = new BlobServiceClientBuilder().endpoint(getContainerUriEndpoint(credentials))
162+
.retryOptions(createRetryOptions())
163+
.httpClient(httpClient)
164+
.sasToken((String) credentials.get(SAS_TOKEN))
165+
.buildClient();
166+
167+
return serviceClient.getBlobContainerClient((String) credentials.get(CONTAINER_NAME));
168+
}
169+
170+
public String getContainerUriEndpoint(Map<String, Object> credentials) {
171+
if (!credentials.containsKey(CONTAINER_URI)) {
172+
throw new IllegalStateException(Messages.MISSING_CONTAINER_URI_IN_THE_CREDENTIALS);
173+
}
174+
try {
175+
URL containerUri = new URL((String) credentials.get(CONTAINER_URI));
176+
return new URL(containerUri.getProtocol(), containerUri.getHost(), containerUri.getPort(), "").toString();
177+
} catch (MalformedURLException e) {
178+
throw new IllegalStateException(Messages.CANNOT_PARSE_CONTAINER_URI_OF_OBJECT_STORE, e);
179+
}
180+
}
181+
182+
private RetryOptions createRetryOptions() {
183+
ExponentialBackoffOptions exponentialBackoffOptions = new ExponentialBackoffOptions().setBaseDelay(
184+
ObjectStoreConstants.OBJECT_STORE_INITIAL_RETRY_DELAY_CONFIG_IN_MILLIS)
185+
.setMaxDelay(
186+
ObjectStoreConstants.OBJECT_STORE_MAX_RETRY_DELAY_CONFIG_IN_SECONDS)
187+
.setMaxRetries(
188+
ObjectStoreConstants.OBJECT_STORE_MAX_ATTEMPTS_CONFIG);
189+
190+
return new RetryOptions(exponentialBackoffOptions);
191+
}
192+
193+
private int removeBlobsByFilter(Predicate<? super BlobItem> filter) {
194+
Set<String> blobNames = getEntryNames(filter);
195+
int deletedBlobsResult = 0;
196+
197+
if (blobNames.isEmpty()) {
198+
return deletedBlobsResult;
199+
}
200+
for (String blobName : blobNames) {
201+
BlobClient blobClient = containerClient.getBlobClient(blobName);
202+
if (blobClient.deleteIfExists()) {
203+
deletedBlobsResult++;
204+
}
205+
}
206+
207+
return deletedBlobsResult;
208+
}
209+
210+
protected Set<String> getEntryNames(Predicate<? super BlobItem> filter) {
211+
BlobListDetails blobListDetails = new BlobListDetails().setRetrieveMetadata(true);
212+
ListBlobsOptions listBlobsOptions = new ListBlobsOptions().setDetails(blobListDetails);
213+
214+
return containerClient.listBlobs(listBlobsOptions, ObjectStoreConstants.AZURE_OBJECT_STORE_TOTAL_TIMEOUT_CONFIG_IN_MINUTES)
215+
.stream()
216+
.filter(filter)
217+
.map(BlobItem::getName)
218+
.collect(Collectors.toSet());
219+
}
220+
221+
public Set<String> getAllEntriesNames() {
222+
ListBlobsOptions listOptions = new ListBlobsOptions();
223+
return containerClient.listBlobs(listOptions, ObjectStoreConstants.AZURE_OBJECT_STORE_TOTAL_TIMEOUT_CONFIG_IN_MINUTES)
224+
.stream()
225+
.map(BlobItem::getName)
226+
.collect(Collectors.toSet());
227+
}
228+
}

‎multiapps-controller-persistence/src/main/java/org/cloudfoundry/multiapps/controller/persistence/services/GcpObjectStoreFileStorage.java‎

Lines changed: 8 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@
55
import java.io.InputStream;
66
import java.nio.channels.Channels;
77
import java.text.MessageFormat;
8-
import java.time.Duration;
98
import java.time.LocalDateTime;
109
import java.util.ArrayList;
1110
import java.util.Base64;
@@ -28,6 +27,7 @@
2827
import com.google.cloud.storage.StorageRetryStrategy;
2928
import org.cloudfoundry.multiapps.controller.persistence.Messages;
3029
import org.cloudfoundry.multiapps.controller.persistence.model.FileEntry;
30+
import org.cloudfoundry.multiapps.controller.persistence.util.ObjectStoreConstants;
3131
import org.cloudfoundry.multiapps.controller.persistence.util.ObjectStoreFilter;
3232
import org.cloudfoundry.multiapps.controller.persistence.util.ObjectStoreMapper;
3333
import org.springframework.http.MediaType;
@@ -51,13 +51,8 @@ public class GcpObjectStoreFileStorage implements FileStorage {
5151

5252
private final String bucketName;
5353
private final Storage storage;
54-
private static final String BUCKET = "bucket";
55-
private static final int OBJECT_STORE_MAX_ATTEMPTS_CONFIG = 6;
56-
private static final double OBJECT_STORE_RETRY_DELAY_MULTIPLIER_CONFIG = 2.0;
57-
private static final Duration OBJECT_STORE_TOTAL_TIMEOUT_CONFIG_IN_MINUTES = Duration.ofMinutes(10);
58-
private static final Duration OBJECT_STORE_MAX_RETRY_DELAY_CONFIG_IN_SECONDS = Duration.ofSeconds(10);
59-
private static final Duration OBJECT_STORE_INITIAL_RETRY_DELAY_CONFIG_IN_MILLIS = Duration.ofMillis(250);
6054
private static final String BASE_64_ENCODED_PRIVATE_KEY_DATA = "base64EncodedPrivateKeyData";
55+
private static final String BUCKET = "bucket";
6156

6257
public GcpObjectStoreFileStorage(Map<String, Object> credentials) {
6358
this.bucketName = (String) credentials.get(BUCKET);
@@ -70,11 +65,12 @@ protected Storage createObjectStoreStorage(Map<String, Object> credentials) {
7065
.setStorageRetryStrategy(StorageRetryStrategy.getUniformStorageRetryStrategy())
7166
.setRetrySettings(
7267
RetrySettings.newBuilder()
73-
.setMaxAttempts(OBJECT_STORE_MAX_ATTEMPTS_CONFIG)
74-
.setTotalTimeoutDuration(OBJECT_STORE_TOTAL_TIMEOUT_CONFIG_IN_MINUTES)
75-
.setMaxRetryDelayDuration(OBJECT_STORE_MAX_RETRY_DELAY_CONFIG_IN_SECONDS)
76-
.setInitialRetryDelayDuration(OBJECT_STORE_INITIAL_RETRY_DELAY_CONFIG_IN_MILLIS)
77-
.setRetryDelayMultiplier(OBJECT_STORE_RETRY_DELAY_MULTIPLIER_CONFIG)
68+
.setMaxAttempts(ObjectStoreConstants.OBJECT_STORE_MAX_ATTEMPTS_CONFIG)
69+
.setTotalTimeoutDuration(ObjectStoreConstants.OBJECT_STORE_TOTAL_TIMEOUT_CONFIG_IN_MINUTES)
70+
.setMaxRetryDelayDuration(ObjectStoreConstants.OBJECT_STORE_MAX_RETRY_DELAY_CONFIG_IN_SECONDS)
71+
.setInitialRetryDelayDuration(
72+
ObjectStoreConstants.OBJECT_STORE_INITIAL_RETRY_DELAY_CONFIG_IN_MILLIS)
73+
.setRetryDelayMultiplier(ObjectStoreConstants.OBJECT_STORE_RETRY_DELAY_MULTIPLIER_CONFIG)
7874
.build())
7975
.build()
8076
.getService();

‎multiapps-controller-persistence/src/main/java/org/cloudfoundry/multiapps/controller/persistence/util/ObjectStoreConstants.java‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
package org.cloudfoundry.multiapps.controller.persistence.util;
2+
3+
import java.time.Duration;
4+
5+
public class ObjectStoreConstants {
6+
7+
private ObjectStoreConstants() {
8+
}
9+
10+
public static final int OBJECT_STORE_MAX_ATTEMPTS_CONFIG = 6;
11+
public static final double OBJECT_STORE_RETRY_DELAY_MULTIPLIER_CONFIG = 2.0;
12+
public static final Duration OBJECT_STORE_TOTAL_TIMEOUT_CONFIG_IN_MINUTES = Duration.ofMinutes(10);
13+
public static final Duration OBJECT_STORE_MAX_RETRY_DELAY_CONFIG_IN_SECONDS = Duration.ofSeconds(10);
14+
public static final Duration OBJECT_STORE_INITIAL_RETRY_DELAY_CONFIG_IN_MILLIS = Duration.ofMillis(250);
15+
public static final Duration AZURE_OBJECT_STORE_TOTAL_TIMEOUT_CONFIG_IN_MINUTES = Duration.ofMinutes(30);
16+
}

0 commit comments

Comments
 (0)