gerrit.onap Code Review - ccsdk/oran.git/blob

   1 /*-
   2  * ========================LICENSE_START=================================
   3  * ONAP : ccsdk oran
   4  * ======================================================================
   5  * Copyright (C) 2022 Nordix Foundation. All rights reserved.
   6  * ======================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ========================LICENSE_END===================================
  19  */
  20
  21 package org.onap.ccsdk.oran.a1policymanagementservice.datastore;
  22
  23 import java.lang.invoke.MethodHandles;
  24 import java.net.URI;
  25 import java.util.concurrent.CompletableFuture;
  26
  27 import org.onap.ccsdk.oran.a1policymanagementservice.configuration.ApplicationConfig;
  28 import org.slf4j.Logger;
  29 import org.slf4j.LoggerFactory;
  30
  31 import reactor.core.publisher.Flux;
  32 import reactor.core.publisher.Mono;
  33
  34 import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
  35 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
  36 import software.amazon.awssdk.core.BytesWrapper;
  37 import software.amazon.awssdk.core.ResponseBytes;
  38 import software.amazon.awssdk.core.async.AsyncRequestBody;
  39 import software.amazon.awssdk.core.async.AsyncResponseTransformer;
  40 import software.amazon.awssdk.regions.Region;
  41 import software.amazon.awssdk.services.s3.S3AsyncClient;
  42 import software.amazon.awssdk.services.s3.S3AsyncClientBuilder;
  43 import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
  44 import software.amazon.awssdk.services.s3.model.CreateBucketResponse;
  45 import software.amazon.awssdk.services.s3.model.DeleteBucketRequest;
  46 import software.amazon.awssdk.services.s3.model.DeleteBucketResponse;
  47 import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
  48 import software.amazon.awssdk.services.s3.model.DeleteObjectResponse;
  49 import software.amazon.awssdk.services.s3.model.GetObjectRequest;
  50 import software.amazon.awssdk.services.s3.model.GetObjectResponse;
  51 import software.amazon.awssdk.services.s3.model.ListObjectsRequest;
  52 import software.amazon.awssdk.services.s3.model.ListObjectsResponse;
  53 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
  54 import software.amazon.awssdk.services.s3.model.PutObjectResponse;
  55 import software.amazon.awssdk.services.s3.model.S3Object;
  56
  57 class S3ObjectStore implements DataStore {
  58     private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
  59     private final ApplicationConfig applicationConfig;
  60
  61     private static S3AsyncClient s3AsynchClient;
  62     private final String location;
  63
  64     public S3ObjectStore(ApplicationConfig applicationConfig, String location) {
  65         this.applicationConfig = applicationConfig;
  66         this.location = location;
  67
  68         getS3AsynchClient(applicationConfig);
  69     }
  70
  71     private static synchronized S3AsyncClient getS3AsynchClient(ApplicationConfig applicationConfig) {
  72         if (applicationConfig.isS3Enabled() && s3AsynchClient == null) {
  73             s3AsynchClient = getS3AsyncClientBuilder(applicationConfig).build();
  74         }
  75         return s3AsynchClient;
  76     }
  77
  78     private static S3AsyncClientBuilder getS3AsyncClientBuilder(ApplicationConfig applicationConfig) {
  79         URI uri = URI.create(applicationConfig.getS3EndpointOverride());
  80         return S3AsyncClient.builder() //
  81                 .region(Region.US_EAST_1) //
  82                 .endpointOverride(uri) //
  83                 .credentialsProvider(StaticCredentialsProvider.create( //
  84                         AwsBasicCredentials.create(applicationConfig.getS3AccessKeyId(), //
  85                                 applicationConfig.getS3SecretAccessKey())));
  86     }
  87
  88     @Override
  89     public Flux<String> listObjects(String prefix) {
  90         return listObjectsInBucket(bucket(), location + "/" + prefix).map(S3Object::key) //
  91                 .map(this::externalName);
  92     }
  93
  94     @Override
  95     public Mono<Boolean> deleteObject(String name) {
  96         DeleteObjectRequest request = DeleteObjectRequest.builder() //
  97                 .bucket(bucket()) //
  98                 .key(key(name)) //
  99                 .build();
 100
 101         CompletableFuture<DeleteObjectResponse> future = s3AsynchClient.deleteObject(request);
 102
 103         return Mono.fromFuture(future).map(resp -> true);
 104     }
 105
 106     @Override
 107     public Mono<byte[]> readObject(String name) {
 108         return getDataFromS3Object(bucket(), name);
 109     }
 110
 111     @Override
 112     public Mono<byte[]> writeObject(String name, byte[] fileData) {
 113
 114         PutObjectRequest request = PutObjectRequest.builder() //
 115                 .bucket(bucket()) //
 116                 .key(key(name)) //
 117                 .build();
 118
 119         AsyncRequestBody body = AsyncRequestBody.fromBytes(fileData);
 120
 121         CompletableFuture<PutObjectResponse> future = s3AsynchClient.putObject(request, body);
 122
 123         return Mono.fromFuture(future) //
 124                 .map(putObjectResponse -> fileData) //
 125                 .doOnError(t -> logger.error("Failed to store object '{}' in S3 {}", key(name), t.getMessage()));
 126     }
 127
 128     @Override
 129     public Mono<String> createDataStore() {
 130         return createS3Bucket(bucket());
 131     }
 132
 133     private Mono<String> createS3Bucket(String s3Bucket) {
 134
 135         CreateBucketRequest request = CreateBucketRequest.builder() //
 136                 .bucket(s3Bucket) //
 137                 .build();
 138
 139         CompletableFuture<CreateBucketResponse> future = s3AsynchClient.createBucket(request);
 140
 141         return Mono.fromFuture(future) //
 142                 .map(f -> s3Bucket) //
 143                 .doOnError(t -> logger.debug("Could not create S3 bucket: {}", t.getMessage()))
 144                 .onErrorResume(t -> Mono.just(s3Bucket));
 145     }
 146
 147     @Override
 148     public Mono<String> deleteAllObjects() {
 149         return listObjects("") //
 150                 .flatMap(this::deleteObject) //
 151                 .collectList() //
 152                 .map(resp -> "OK").onErrorResume(t -> Mono.just("NOK"));
 153     }
 154
 155     public Mono<String> deleteBucket() {
 156         DeleteBucketRequest request = DeleteBucketRequest.builder() //
 157                 .bucket(bucket()) //
 158                 .build();
 159
 160         CompletableFuture<DeleteBucketResponse> future = s3AsynchClient.deleteBucket(request);
 161
 162         return Mono.fromFuture(future) //
 163                 .doOnError(t -> logger.warn("Could not delete bucket: {}, reason: {}", bucket(), t.getMessage()))
 164                 .map(resp -> bucket()) //
 165                 .doOnNext(resp -> logger.debug("Deleted bucket: {}", bucket())).onErrorResume(t -> Mono.just("NOK"));
 166     }
 167
 168     private String bucket() {
 169         return applicationConfig.getS3Bucket();
 170     }
 171
 172     private Flux<S3Object> listObjectsInBucket(String bucket, String prefix) {
 173
 174         return listObjectsRequest(bucket, prefix, null) //
 175                 .expand(response -> listObjectsRequest(bucket, prefix, response)) //
 176                 .map(ListObjectsResponse::contents) //
 177                 .doOnNext(f -> logger.debug("Found objects in {}: {}", bucket, f.size())) //
 178                 .doOnError(t -> logger.warn("Error fromlist objects: {}", t.getMessage())) //
 179                 .flatMap(Flux::fromIterable) //
 180                 .doOnNext(obj -> logger.debug("Found object: {}", obj.key()));
 181     }
 182
 183     private Mono<ListObjectsResponse> listObjectsRequest(String bucket, String prefix,
 184             ListObjectsResponse prevResponse) {
 185         ListObjectsRequest.Builder builder = ListObjectsRequest.builder() //
 186                 .bucket(bucket) //
 187                 .maxKeys(1000) //
 188                 .prefix(prefix);
 189
 190         if (prevResponse != null) {
 191             if (Boolean.TRUE.equals(prevResponse.isTruncated())) {
 192                 builder.marker(prevResponse.nextMarker());
 193             } else {
 194                 return Mono.empty();
 195             }
 196         }
 197
 198         ListObjectsRequest listObjectsRequest = builder.build();
 199         CompletableFuture<ListObjectsResponse> future = s3AsynchClient.listObjects(listObjectsRequest);
 200         return Mono.fromFuture(future);
 201     }
 202
 203     private Mono<byte[]> getDataFromS3Object(String bucket, String name) {
 204
 205         GetObjectRequest request = GetObjectRequest.builder() //
 206                 .bucket(bucket) //
 207                 .key(key(name)) //
 208                 .build();
 209
 210         CompletableFuture<ResponseBytes<GetObjectResponse>> future =
 211                 s3AsynchClient.getObject(request, AsyncResponseTransformer.toBytes());
 212
 213         return Mono.fromFuture(future) //
 214                 .map(BytesWrapper::asByteArray) //
 215                 .doOnError(t -> logger.error("Failed to get file from S3, key:{}, bucket: {}, {}", key(name), bucket,
 216                         t.getMessage())) //
 217                 .doOnEach(n -> logger.debug("Read file from S3: {} {}", bucket, key(name))) //
 218                 .onErrorResume(t -> Mono.empty());
 219     }
 220
 221     private String key(String name) {
 222         return location + "/" + name;
 223     }
 224
 225     private String externalName(String internalName) {
 226         return internalName.substring(key("").length());
 227     }
 228
 229 }