diff --git a/oap-storage/oap-storage-cloud-aws-s3/pom.xml b/oap-storage/oap-storage-cloud-aws-s3/pom.xml index f998baa9f1..a7c0bb3257 100644 --- a/oap-storage/oap-storage-cloud-aws-s3/pom.xml +++ b/oap-storage/oap-storage-cloud-aws-s3/pom.xml @@ -23,39 +23,10 @@ software.amazon.awssdk s3 - - software.amazon.awssdk.crt - aws-crt - software.amazon.awssdk - s3-transfer-manager - - - software.amazon.awssdk - s3 - - - software.amazon.awssdk - sdk-core - - - software.amazon.awssdk - utils - - - software.amazon.awssdk - annotations - - - software.amazon.awssdk - regions - - - software.amazon.awssdk.crt - aws-crt - - + apache5-client + 2.50.2 diff --git a/oap-storage/oap-storage-cloud-aws-s3/src/main/java/oap/storage/cloud/awss3/FileSystemCloudApiS3.java b/oap-storage/oap-storage-cloud-aws-s3/src/main/java/oap/storage/cloud/awss3/FileSystemCloudApiS3.java index cbdab9766d..108cc67da1 100644 --- a/oap-storage/oap-storage-cloud-aws-s3/src/main/java/oap/storage/cloud/awss3/FileSystemCloudApiS3.java +++ b/oap-storage/oap-storage-cloud-aws-s3/src/main/java/oap/storage/cloud/awss3/FileSystemCloudApiS3.java @@ -1,9 +1,6 @@ package oap.storage.cloud.awss3; -import io.micrometer.core.instrument.util.NamedThreadFactory; import lombok.extern.slf4j.Slf4j; -import oap.concurrent.Executors; -import oap.concurrent.ThreadPoolExecutor; import oap.io.Closeables; import oap.storage.cloud.BlobData; import oap.storage.cloud.CloudException; @@ -15,32 +12,40 @@ import oap.storage.cloud.PageSet; import oap.util.Lists; import oap.util.Maps; -import oap.util.Throwables; import org.apache.commons.lang3.NotImplementedException; import org.joda.time.DateTime; import org.joda.time.DateTimeZone; import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; -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.core.ResponseInputStream; +import software.amazon.awssdk.core.exception.SdkException; +import software.amazon.awssdk.core.sync.RequestBody; import software.amazon.awssdk.endpoints.Endpoint; +import software.amazon.awssdk.http.apache5.Apache5HttpClient; import software.amazon.awssdk.regions.Region; -import software.amazon.awssdk.services.s3.S3AsyncClient; -import software.amazon.awssdk.services.s3.S3CrtAsyncClientBuilder; +import software.amazon.awssdk.services.s3.S3Client; +import software.amazon.awssdk.services.s3.S3ClientBuilder; import software.amazon.awssdk.services.s3.endpoints.S3EndpointParams; import software.amazon.awssdk.services.s3.endpoints.internal.DefaultS3EndpointProvider; +import software.amazon.awssdk.services.s3.model.AbortMultipartUploadRequest; import software.amazon.awssdk.services.s3.model.BucketAlreadyExistsException; +import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest; +import software.amazon.awssdk.services.s3.model.CompletedMultipartUpload; +import software.amazon.awssdk.services.s3.model.CompletedPart; import software.amazon.awssdk.services.s3.model.CopyObjectRequest; import software.amazon.awssdk.services.s3.model.CreateBucketRequest; +import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest; import software.amazon.awssdk.services.s3.model.Delete; import software.amazon.awssdk.services.s3.model.DeleteBucketRequest; import software.amazon.awssdk.services.s3.model.DeleteObjectRequest; import software.amazon.awssdk.services.s3.model.DeleteObjectsRequest; import software.amazon.awssdk.services.s3.model.GetObjectRequest; +import software.amazon.awssdk.services.s3.model.GetObjectResponse; import software.amazon.awssdk.services.s3.model.HeadBucketRequest; 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.NoSuchBucketException; import software.amazon.awssdk.services.s3.model.NoSuchKeyException; import software.amazon.awssdk.services.s3.model.ObjectIdentifier; @@ -48,42 +53,35 @@ import software.amazon.awssdk.services.s3.model.S3Object; import software.amazon.awssdk.services.s3.model.Tag; import software.amazon.awssdk.services.s3.model.Tagging; -import software.amazon.awssdk.transfer.s3.S3TransferManager; -import software.amazon.awssdk.transfer.s3.model.CompletedUpload; -import software.amazon.awssdk.transfer.s3.model.Copy; -import software.amazon.awssdk.transfer.s3.model.CopyRequest; -import software.amazon.awssdk.transfer.s3.model.DownloadFileRequest; -import software.amazon.awssdk.transfer.s3.model.FileDownload; -import software.amazon.awssdk.transfer.s3.model.Upload; -import software.amazon.awssdk.transfer.s3.model.UploadRequest; +import software.amazon.awssdk.services.s3.model.UploadPartRequest; +import software.amazon.awssdk.services.s3.model.UploadPartResponse; import java.io.File; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; -import java.io.PipedInputStream; -import java.io.PipedOutputStream; import java.net.URI; import java.net.URISyntaxException; import java.nio.ByteBuffer; +import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.StandardCopyOption; import java.time.Instant; import java.util.ArrayList; import java.util.List; import java.util.Map; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; import static java.nio.charset.StandardCharsets.UTF_8; @Slf4j public class FileSystemCloudApiS3 implements FileSystemCloudApi { - private final S3AsyncClient s3Client; - private volatile S3TransferManager s3TransferManager; + private static final int PART_SIZE = 5 * 1024 * 1024; + + private final S3Client s3Client; public FileSystemCloudApiS3( FileSystemConfiguration fileSystemConfiguration, String bucketName ) { - S3CrtAsyncClientBuilder builder = S3AsyncClient.crtBuilder(); + S3ClientBuilder builder = S3Client.builder() + .httpClientBuilder( Apache5HttpClient.builder() ); Object regionObj = fileSystemConfiguration.get( "s3", bucketName, "jclouds.region" ); if( regionObj == null ) { @@ -125,253 +123,215 @@ private static DateTime instantToDateTime( Instant obj ) { } @Override - public CompletableFuture blobExistsAsync( CloudURI path ) { + public boolean blobExists( CloudURI path ) { HeadObjectRequest headObjectRequest = HeadObjectRequest.builder().bucket( path.container ).key( path.path ).build(); - return s3Client - .headObject( headObjectRequest ) - .thenApply( _ -> true ) - .exceptionallyCompose( ex -> { - if( ex.getCause() instanceof NoSuchBucketException || ex.getCause() instanceof NoSuchKeyException ) { - return CompletableFuture.completedFuture( false ); - } - - return CompletableFuture.failedFuture( propagate( ex ) ); - } ); + try { + s3Client.headObject( headObjectRequest ); + return true; + } catch( NoSuchBucketException | NoSuchKeyException e ) { + return false; + } catch( SdkException e ) { + throw new CloudException( e ); + } } @Override - public CompletableFuture containerExistsAsync( CloudURI path ) { + public boolean containerExists( CloudURI path ) { HeadBucketRequest headBucketRequest = HeadBucketRequest.builder() .bucket( path.container ) .build(); - return s3Client.headBucket( headBucketRequest ) - .thenApply( _ -> true ) - .exceptionallyCompose( e -> { - if( e.getCause() instanceof NoSuchBucketException ) { - return CompletableFuture.completedFuture( false ); - } - - return CompletableFuture.failedFuture( propagate( e ) ); - } ); + try { + s3Client.headBucket( headBucketRequest ); + return true; + } catch( NoSuchBucketException e ) { + return false; + } catch( SdkException e ) { + throw new CloudException( e ); + } } @Override - public CompletableFuture deleteBlobAsync( CloudURI path ) { + public void deleteBlob( CloudURI path ) { DeleteObjectRequest deleteRequest = DeleteObjectRequest.builder().bucket( path.container ).key( path.path ).build(); - return s3Client.deleteObject( deleteRequest ) - .thenAccept( _ -> {} ) - .exceptionallyCompose( e -> CompletableFuture.failedFuture( propagate( e ) ) ); + try { + s3Client.deleteObject( deleteRequest ); + } catch( SdkException e ) { + throw new CloudException( e ); + } } @Override - public CompletableFuture deleteContainerAsync( CloudURI path ) { - return s3Client.listObjectsV2( ListObjectsV2Request.builder().bucket( path.container ).build() ) - .thenCompose( listResponse -> { - List listObjects = listResponse.contents(); - - ArrayList objectsToDelete = new ArrayList<>(); - for( S3Object s3Object : listObjects ) { - objectsToDelete.add( ObjectIdentifier.builder().key( s3Object.key() ).build() ); - } + public void deleteContainer( CloudURI path ) { + try { + ListObjectsV2Response listResponse = s3Client.listObjectsV2( ListObjectsV2Request.builder().bucket( path.container ).build() ); + ArrayList objectsToDelete = new ArrayList<>(); + for( S3Object s3Object : listResponse.contents() ) { + objectsToDelete.add( ObjectIdentifier.builder().key( s3Object.key() ).build() ); + } + + if( !objectsToDelete.isEmpty() ) { DeleteObjectsRequest deleteObjectsRequest = DeleteObjectsRequest.builder() .bucket( path.container ) .delete( Delete.builder().objects( objectsToDelete ).build() ) .build(); - return s3Client.deleteObjects( deleteObjectsRequest ).thenCompose( deleteObjectsResponse -> { - DeleteBucketRequest deleteBucketRequest = DeleteBucketRequest.builder().bucket( path.container ).build(); + s3Client.deleteObjects( deleteObjectsRequest ); + } - return s3Client.deleteBucket( deleteBucketRequest ).thenAccept( _ -> {} ); - } ); - } ) - .exceptionallyCompose( e -> CompletableFuture.failedFuture( propagate( e ) ) ); + s3Client.deleteBucket( DeleteBucketRequest.builder().bucket( path.container ).build() ); + } catch( SdkException e ) { + throw new CloudException( e ); + } } @Override - public CompletableFuture createContainerAsync( CloudURI path ) { + public boolean createContainer( CloudURI path ) { CreateBucketRequest createBucketRequest = CreateBucketRequest.builder().bucket( path.container ).build(); - return s3Client.createBucket( createBucketRequest ) - .thenApply( _ -> true ) - .exceptionallyCompose( e -> { - if( e.getCause() instanceof BucketAlreadyExistsException ) { - return CompletableFuture.completedFuture( false ); - } - return CompletableFuture.failedFuture( propagate( e ) ); - } ); + + try { + s3Client.createBucket( createBucketRequest ); + return true; + } catch( BucketAlreadyExistsException e ) { + return false; + } catch( SdkException e ) { + throw new CloudException( e ); + } } @Override - public CompletableFuture deleteContainerIfEmptyAsync( CloudURI path ) { + public boolean deleteContainerIfEmpty( CloudURI path ) { DeleteBucketRequest deleteBucketRequest = DeleteBucketRequest.builder().bucket( path.container ).build(); - return s3Client.deleteBucket( deleteBucketRequest ) - .thenApply( _ -> true ) - .exceptionallyCompose( e -> { - if( e.getMessage().contains( "The bucket you tried to delete is not empty" ) ) { - return CompletableFuture.completedFuture( false ); - } - - return CompletableFuture.failedFuture( propagate( e ) ); - } ); + try { + s3Client.deleteBucket( deleteBucketRequest ); + return true; + } catch( SdkException e ) { + if( e.getMessage() != null && e.getMessage().contains( "The bucket you tried to delete is not empty" ) ) { + return false; + } + throw new CloudException( e ); + } } @Override - public CompletableFuture getMetadataAsync( CloudURI path ) { + public FileSystem.StorageItem getMetadata( CloudURI path ) { HeadObjectRequest headObjectRequest = HeadObjectRequest.builder().bucket( path.container ).key( path.path ).build(); - return s3Client.headObject( headObjectRequest ) - .thenApply( headObjectResponse -> { - return new FileSystem.StorageItem() { - @Override - public String getName() { - return path.toString(); - } + try { + HeadObjectResponse headObjectResponse = s3Client.headObject( headObjectRequest ); - @Override - public URI getUri() { - try { - return s3Client.utilities().getUrl( builder -> builder.bucket( path.container ).key( path.path ).build() ).toURI(); - } catch( URISyntaxException e ) { - throw new CloudException( e ); - } - } + return new FileSystem.StorageItem() { + @Override + public String getName() { + return path.toString(); + } - @Override - public String getETag() { - return headObjectResponse.eTag(); + @Override + public URI getUri() { + try { + return s3Client.utilities().getUrl( b -> b.bucket( path.container ).key( path.path ).build() ).toURI(); + } catch( URISyntaxException e ) { + throw new CloudException( e ); } + } - @Override - public DateTime getLastModified() { - return instantToDateTime( headObjectResponse.lastModified() ); - } + @Override + public String getETag() { + return headObjectResponse.eTag(); + } - @Override - public Long getSize() { - return headObjectResponse.contentLength(); - } + @Override + public DateTime getLastModified() { + return instantToDateTime( headObjectResponse.lastModified() ); + } - @Override - public String getContentType() { - return headObjectResponse.contentType(); - } - }; - } ) - .exceptionallyCompose( e -> { - if( e.getCause() instanceof NoSuchKeyException ) { - return CompletableFuture.completedFuture( null ); + @Override + public Long getSize() { + return headObjectResponse.contentLength(); } - return CompletableFuture.failedFuture( propagate( e ) ); - } ); + @Override + public String getContentType() { + return headObjectResponse.contentType(); + } + }; + } catch( NoSuchKeyException e ) { + return null; + } catch( SdkException e ) { + throw new CloudException( e ); + } } @Override - public CompletableFuture downloadFileAsync( CloudURI source, Path destination ) throws CloudException { - S3TransferManager s3TransferManager = getS3TransferManager(); + public void downloadFile( CloudURI source, Path destination ) throws CloudException { GetObjectRequest getObjectRequest = GetObjectRequest.builder().bucket( source.container ).key( source.path ).build(); - DownloadFileRequest downloadFileRequest = DownloadFileRequest.builder().getObjectRequest( getObjectRequest ).destination( destination ).build(); - FileDownload fileDownload = s3TransferManager.downloadFile( downloadFileRequest ); - return fileDownload.completionFuture() - .thenAccept( future -> {} ) - .exceptionallyCompose( e -> CompletableFuture.failedFuture( propagate( e ) ) ); - } - private S3TransferManager getS3TransferManager() { - if( s3TransferManager == null ) { - synchronized( this ) { - if( s3TransferManager == null ) { - s3TransferManager = S3TransferManager.builder().s3Client( s3Client ).build(); - } + try { + oap.io.Files.ensureFile( destination ); + + try( ResponseInputStream in = s3Client.getObject( getObjectRequest ) ) { + Files.copy( in, destination, StandardCopyOption.REPLACE_EXISTING ); } + } catch( IOException e ) { + throw new CloudException( e ); + } catch( SdkException e ) { + throw new CloudException( e ); } - return s3TransferManager; } @Override - public CompletableFuture copyAsync( CloudURI source, CloudURI destination ) { - S3TransferManager s3TransferManager = getS3TransferManager(); - CopyObjectRequest build = CopyObjectRequest.builder() + public void copy( CloudURI source, CloudURI destination ) { + CopyObjectRequest copyObjectRequest = CopyObjectRequest.builder() .sourceBucket( source.container ) .destinationBucket( destination.container ) .sourceKey( source.path ) .destinationKey( destination.path ) .build(); - Copy copy = s3TransferManager.copy( CopyRequest.builder().copyObjectRequest( build ).build() ); - return copy.completionFuture() - .thenAccept( _ -> {} ) - .exceptionallyCompose( e -> CompletableFuture.failedFuture( propagate( e ) ) ); + + try { + s3Client.copyObject( copyObjectRequest ); + } catch( SdkException e ) { + throw new CloudException( e ); + } } @Override - public CompletableFuture getInputStreamAsync( CloudURI path ) { - return s3Client - .getObject( GetObjectRequest.builder().bucket( path.container ).key( path.path ).build(), AsyncResponseTransformer.toBlockingInputStream() ) - .exceptionallyCompose( e -> CompletableFuture.failedFuture( propagate( e ) ) ); + public InputStream getInputStream( CloudURI path ) { + try { + return s3Client.getObject( GetObjectRequest.builder().bucket( path.container ).key( path.path ).build() ); + } catch( SdkException e ) { + throw new CloudException( e ); + } } @Override public OutputStream getOutputStream( CloudURI cloudURI, Map tags ) { - ThreadPoolExecutor threadPoolExecutor = null; - CompletableFuture completedUploadCompletableFuture = null; + return new MultipartUploadOutputStream( cloudURI, tags, null ); + } + @Override + public void upload( CloudURI cloudURI, BlobData blobData ) { try { - S3TransferManager s3TransferManager = getS3TransferManager(); - BlockingInputStreamAsyncRequestBody body = AsyncRequestBody.forBlockingInputStream( null ); - - PutObjectRequest putObjectRequest = PutObjectRequest.builder() - .bucket( cloudURI.container ).key( cloudURI.path ) - .tagging( getTagging( tags ) ) - .build(); - UploadRequest uploadRequest = UploadRequest.builder() - .putObjectRequest( putObjectRequest ) - .requestBody( body ) - .build(); - - Upload upload = s3TransferManager.upload( uploadRequest ); - completedUploadCompletableFuture = upload.completionFuture(); - - PipedOutputStream pipedOutputStream = new PipedOutputStream(); - PipedInputStream pingedInputStream = new PipedInputStream( pipedOutputStream ); - - threadPoolExecutor = Executors.newFixedBlockingThreadPool( 1, new NamedThreadFactory( "fs-write-" + cloudURI ) ); - Future future; - future = threadPoolExecutor.submit( () -> { - body.writeInputStream( pingedInputStream ); - } ); - - return new CloudOutputStream( s3TransferManager, completedUploadCompletableFuture, pipedOutputStream, future, threadPoolExecutor ); - } catch( Exception e ) { - log.error( e.getMessage(), e ); - - if( completedUploadCompletableFuture != null ) { - completedUploadCompletableFuture.completeExceptionally( e ); + switch( blobData.content ) { + case InputStream is -> uploadStream( cloudURI, is, blobData.tags, blobData.contentType ); + case String str -> putObject( cloudURI, blobData, RequestBody.fromString( str, UTF_8 ) ); + case byte[] bytes -> putObject( cloudURI, blobData, RequestBody.fromBytes( bytes ) ); + case ByteBuffer byteBuffer -> putObject( cloudURI, blobData, RequestBody.fromByteBuffer( byteBuffer ) ); + case File file -> putObject( cloudURI, blobData, RequestBody.fromFile( file ) ); + case Path path -> putObject( cloudURI, blobData, RequestBody.fromFile( path ) ); + case null -> putObject( cloudURI, blobData, RequestBody.empty() ); // "folder" + default -> throw new CloudException( "Unknown content type " + blobData.content.getClass() ); } - Closeables.close( threadPoolExecutor ); - Closeables.close( s3TransferManager ); - throw Throwables.propagate( e ); + } catch( SdkException e ) { + throw new CloudException( e ); } } - @Override - public CompletableFuture uploadAsync( CloudURI cloudURI, BlobData blobData ) { - S3TransferManager s3TransferManager = getS3TransferManager(); - AsyncRequestBody body = switch( blobData.content ) { - case InputStream _ -> AsyncRequestBody.forBlockingInputStream( null ); - case String str -> AsyncRequestBody.fromString( str, UTF_8 ); - case byte[] bytes -> AsyncRequestBody.fromBytes( bytes ); - case ByteBuffer byteBuffer -> AsyncRequestBody.fromByteBuffer( byteBuffer ); - case File file -> AsyncRequestBody.fromFile( file ); - case Path path -> AsyncRequestBody.fromFile( path ); - case null -> AsyncRequestBody.empty(); // "folder" - default -> throw new CloudException( "Unknown content type " + blobData.content.getClass() ); - }; - - + private void putObject( CloudURI cloudURI, BlobData blobData, RequestBody requestBody ) { PutObjectRequest.Builder putObjectRequestBuilder = PutObjectRequest.builder() .bucket( cloudURI.container ) .key( cloudURI.path ) @@ -384,24 +344,19 @@ public CompletableFuture uploadAsync( CloudURI cloudURI, BlobData blobData putObjectRequestBuilder.contentLength( blobData.contentLength ); } - UploadRequest uploadRequest = UploadRequest.builder() - .requestBody( body ) - .putObjectRequest( putObjectRequestBuilder.build() ) - .build(); - - Upload upload = s3TransferManager.upload( uploadRequest ); + s3Client.putObject( putObjectRequestBuilder.build(), requestBody ); + } - if( blobData.content instanceof InputStream is ) { - ( ( BlockingInputStreamAsyncRequestBody ) body ).writeInputStream( is ); + private void uploadStream( CloudURI cloudURI, InputStream inputStream, Map tags, String contentType ) { + try( OutputStream out = new MultipartUploadOutputStream( cloudURI, tags, contentType ) ) { + inputStream.transferTo( out ); + } catch( IOException e ) { + throw new CloudException( e ); } - - return upload.completionFuture() - .thenAccept( _ -> {} ) - .exceptionallyCompose( e -> CompletableFuture.failedFuture( propagate( e ) ) ); } @Override - public CompletableFuture> listAsync( CloudURI path, ListOptions listOptions ) { + public PageSet list( CloudURI path, ListOptions listOptions ) { ListObjectsV2Request.Builder builder = ListObjectsV2Request.builder().bucket( path.container ); if( !path.path.isEmpty() ) { builder.prefix( path.path ); @@ -412,103 +367,179 @@ public CompletableFuture> li if( listOptions.maxKeys != null ) { builder.maxKeys( listOptions.maxKeys ); } - return s3Client.listObjectsV2( builder.build() ) - .thenApply( listObjectsV2Response -> { - log.trace( " response {}", listObjectsV2Response ); - return new PageSet<>( listObjectsV2Response.nextContinuationToken(), Lists.map( listObjectsV2Response.contents(), obj -> new FileSystem.StorageItem() { - @Override - public String getName() { - return obj.key(); - } - @Override - public URI getUri() { - return s3Client.utilities().parseUri( URI.create( new CloudURI( path.scheme, path.container, obj.key() ).toString() ) ).uri(); - } + try { + ListObjectsV2Response listObjectsV2Response = s3Client.listObjectsV2( builder.build() ); + log.trace( " response {}", listObjectsV2Response ); - @Override - public String getETag() { - return obj.eTag(); - } + return new PageSet<>( listObjectsV2Response.nextContinuationToken(), Lists.map( listObjectsV2Response.contents(), obj -> new FileSystem.StorageItem() { + @Override + public String getName() { + return obj.key(); + } - @Override - public DateTime getLastModified() { - return instantToDateTime( obj.lastModified() ); - } + @Override + public URI getUri() { + return s3Client.utilities().parseUri( URI.create( new CloudURI( path.scheme, path.container, obj.key() ).toString() ) ).uri(); + } - @Override - public Long getSize() { - return obj.size(); - } + @Override + public String getETag() { + return obj.eTag(); + } - @Override - public String getContentType() { - throw new NotImplementedException(); - } - } ) ); - } ) - .exceptionallyCompose( e -> CompletableFuture.failedFuture( propagate( e ) ) ); + @Override + public DateTime getLastModified() { + return instantToDateTime( obj.lastModified() ); + } + + @Override + public Long getSize() { + return obj.size(); + } + + @Override + public String getContentType() { + throw new NotImplementedException(); + } + } ) ); + } catch( SdkException e ) { + throw new CloudException( e ); + } } @Override public void close() { - Closeables.close( s3TransferManager ); Closeables.close( s3Client ); } - public static class CloudOutputStream extends OutputStream { - private final S3TransferManager s3TransferManager; - private final CompletableFuture completedUploadCompletableFuture; - private final PipedOutputStream pipedOutputStream; - private final Future future; - private final ThreadPoolExecutor threadPoolExecutor; - - public CloudOutputStream( S3TransferManager s3TransferManager, - CompletableFuture completedUploadCompletableFuture, - PipedOutputStream pipedOutputStream, Future future, ThreadPoolExecutor threadPoolExecutor ) { - this.s3TransferManager = s3TransferManager; - this.completedUploadCompletableFuture = completedUploadCompletableFuture; - this.pipedOutputStream = pipedOutputStream; - this.future = future; - this.threadPoolExecutor = threadPoolExecutor; + private final class MultipartUploadOutputStream extends OutputStream { + private final CloudURI cloudURI; + private final Map tags; + private final String contentType; + private final List completedParts = new ArrayList<>(); + private final byte[] buffer = new byte[ PART_SIZE ]; + + private int position = 0; + private int partNumber = 1; + private String uploadId; + private boolean closed = false; + + MultipartUploadOutputStream( CloudURI cloudURI, Map tags, String contentType ) { + this.cloudURI = cloudURI; + this.tags = tags; + this.contentType = contentType; } @Override - public void write( int b ) throws IOException { - pipedOutputStream.write( b ); + public void write( int b ) { + if( position == buffer.length ) { + flushPart(); + } + buffer[ position++ ] = ( byte ) b; } @Override - public void write( byte[] b ) throws IOException { - pipedOutputStream.write( b ); + public void write( byte[] b, int off, int len ) { + int written = 0; + while( written < len ) { + if( position == buffer.length ) { + flushPart(); + } + int toCopy = Math.min( len - written, buffer.length - position ); + System.arraycopy( b, off + written, buffer, position, toCopy ); + position += toCopy; + written += toCopy; + } } - @Override - public void write( byte[] b, int off, int len ) throws IOException { - pipedOutputStream.write( b, off, len ); + private void flushPart() { + try { + if( uploadId == null ) { + uploadId = createMultipartUpload(); + } + uploadCurrentBuffer(); + position = 0; + } catch( SdkException e ) { + abortQuietly(); + throw new CloudException( e ); + } } - @Override - public void flush() throws IOException { - pipedOutputStream.flush(); + private String createMultipartUpload() { + CreateMultipartUploadRequest.Builder builder = CreateMultipartUploadRequest.builder() + .bucket( cloudURI.container ) + .key( cloudURI.path ) + .tagging( getTagging( tags ) ); + if( contentType != null ) { + builder.contentType( contentType ); + } + return s3Client.createMultipartUpload( builder.build() ).uploadId(); } - @Override - public void close() { - Closeables.close( pipedOutputStream ); + private void uploadCurrentBuffer() { + UploadPartRequest uploadPartRequest = UploadPartRequest.builder() + .bucket( cloudURI.container ) + .key( cloudURI.path ) + .uploadId( uploadId ) + .partNumber( partNumber ) + .build(); + + UploadPartResponse response = s3Client.uploadPart( uploadPartRequest, RequestBody.fromByteBuffer( ByteBuffer.wrap( buffer, 0, position ) ) ); + + completedParts.add( CompletedPart.builder().partNumber( partNumber ).eTag( response.eTag() ).build() ); + partNumber++; + } + private void abortQuietly() { + if( uploadId == null ) { + return; + } try { - future.get(); - } catch( InterruptedException | ExecutionException e ) { + s3Client.abortMultipartUpload( AbortMultipartUploadRequest.builder() + .bucket( cloudURI.container ) + .key( cloudURI.path ) + .uploadId( uploadId ) + .build() ); + } catch( SdkException e ) { log.error( e.getMessage(), e ); } + } + + @Override + public void close() { + if( closed ) { + return; + } + closed = true; + try { - completedUploadCompletableFuture.get(); - } catch( InterruptedException | ExecutionException e ) { - log.error( e.getMessage(), e ); + if( uploadId == null ) { + PutObjectRequest.Builder putObjectRequestBuilder = PutObjectRequest.builder() + .bucket( cloudURI.container ) + .key( cloudURI.path ) + .tagging( getTagging( tags ) ); + if( contentType != null ) { + putObjectRequestBuilder.contentType( contentType ); + } + s3Client.putObject( putObjectRequestBuilder.build(), RequestBody.fromByteBuffer( ByteBuffer.wrap( buffer, 0, position ) ) ); + return; + } + + if( position > 0 ) { + uploadCurrentBuffer(); + } + + s3Client.completeMultipartUpload( CompleteMultipartUploadRequest.builder() + .bucket( cloudURI.container ) + .key( cloudURI.path ) + .uploadId( uploadId ) + .multipartUpload( CompletedMultipartUpload.builder().parts( completedParts ).build() ) + .build() ); + } catch( SdkException e ) { + abortQuietly(); + throw new CloudException( e ); } - Closeables.close( threadPoolExecutor ); - Closeables.close( s3TransferManager ); } } } diff --git a/oap-storage/oap-storage-cloud-ftp/src/main/java/oap/storage/cloud/ftp/AbstractFileSystemCloudApiFtp.java b/oap-storage/oap-storage-cloud-ftp/src/main/java/oap/storage/cloud/ftp/AbstractFileSystemCloudApiFtp.java index 0ce1da3f41..e0db254058 100644 --- a/oap-storage/oap-storage-cloud-ftp/src/main/java/oap/storage/cloud/ftp/AbstractFileSystemCloudApiFtp.java +++ b/oap-storage/oap-storage-cloud-ftp/src/main/java/oap/storage/cloud/ftp/AbstractFileSystemCloudApiFtp.java @@ -36,7 +36,6 @@ import java.util.Comparator; import java.util.List; import java.util.Map; -import java.util.concurrent.CompletableFuture; import java.util.stream.Stream; import static dev.khbd.interp4j.core.Interpolations.s; @@ -233,39 +232,39 @@ private URI buildUri( CloudURI path ) { } @Override - public CompletableFuture blobExistsAsync( CloudURI path ) { + public boolean blobExists( CloudURI path ) { FTPClient client = borrow(); boolean healthy = false; try { boolean exists = findFile( client, absolute( path.path ) ) != null; healthy = true; - return CompletableFuture.completedFuture( exists ); + return exists; } catch( IOException e ) { - return CompletableFuture.failedFuture( new CloudException( e ) ); + throw new CloudException( e ); } finally { release( client, healthy ); } } @Override - public CompletableFuture containerExistsAsync( CloudURI path ) { + public boolean containerExists( CloudURI path ) { try { FTPClient client = borrow(); release( client, true ); - return CompletableFuture.completedFuture( true ); + return true; } catch( CloudException e ) { - return CompletableFuture.completedFuture( false ); + return false; } } @Override - public CompletableFuture deleteBlobAsync( CloudURI path ) { + public void deleteBlob( CloudURI path ) { FTPClient client = borrow(); boolean healthy = false; try { if( !client.deleteFile( absolute( path.path ) ) ) { healthy = true; - return CompletableFuture.failedFuture( new CloudException( "cannot delete " + path ) ); + throw new CloudException( "cannot delete " + path ); } if( removeEmptyFolders ) { @@ -273,9 +272,8 @@ public CompletableFuture deleteBlobAsync( CloudURI path ) { } healthy = true; - return CompletableFuture.completedFuture( null ); } catch( IOException e ) { - return CompletableFuture.failedFuture( new CloudException( e ) ); + throw new CloudException( e ); } finally { release( client, healthy ); } @@ -296,59 +294,58 @@ private void removeEmptyParents( FTPClient client, String dirPath ) throws IOExc } @Override - public CompletableFuture deleteContainerAsync( CloudURI path ) throws CloudException { + public void deleteContainer( CloudURI path ) throws CloudException { throw new CloudException( "not supported" ); } @Override - public CompletableFuture createContainerAsync( CloudURI path ) { - return CompletableFuture.completedFuture( false ); + public boolean createContainer( CloudURI path ) { + return false; } @Override - public CompletableFuture deleteContainerIfEmptyAsync( CloudURI path ) { - return CompletableFuture.completedFuture( false ); + public boolean deleteContainerIfEmpty( CloudURI path ) { + return false; } @Override - public CompletableFuture getMetadataAsync( CloudURI path ) { + public FileSystem.StorageItem getMetadata( CloudURI path ) { FTPClient client = borrow(); boolean healthy = false; try { FTPFile file = findFile( client, absolute( path.path ) ); healthy = true; - if( file == null ) return CompletableFuture.completedFuture( null ); + if( file == null ) return null; - return CompletableFuture.completedFuture( toStorageItem( path, file ) ); + return toStorageItem( path, file ); } catch( IOException e ) { - return CompletableFuture.failedFuture( new CloudException( e ) ); + throw new CloudException( e ); } finally { release( client, healthy ); } } @Override - public CompletableFuture downloadFileAsync( CloudURI source, Path destination ) { + public void downloadFile( CloudURI source, Path destination ) { FTPClient client = borrow(); boolean healthy = false; try { oap.io.Files.ensureFile( destination ); try( OutputStream out = Files.newOutputStream( destination ) ) { if( !client.retrieveFile( absolute( source.path ), out ) ) { - return CompletableFuture.failedFuture( new CloudException( "cannot download " + source ) ); + throw new CloudException( "cannot download " + source ); } } healthy = true; - return CompletableFuture.completedFuture( null ); } catch( IOException e ) { - return CompletableFuture.failedFuture( new CloudException( e ) ); + throw new CloudException( e ); } finally { release( client, healthy ); } } @Override - public CompletableFuture copyAsync( CloudURI source, CloudURI destination ) { + public void copy( CloudURI source, CloudURI destination ) { Preconditions.checkArgument( source.scheme.equals( destination.scheme ) ); FTPClient sourceClient = borrow(); @@ -360,7 +357,7 @@ public CompletableFuture copyAsync( CloudURI source, CloudURI destination if( in == null ) { sourceHealthy = true; destinationHealthy = true; - return CompletableFuture.failedFuture( new CloudException( "cannot open source stream " + source ) ); + throw new CloudException( "cannot open source stream " + source ); } ensureRemoteDirectory( destinationClient, parentOf( absolute( destination.path ) ) ); @@ -373,12 +370,10 @@ public CompletableFuture copyAsync( CloudURI source, CloudURI destination destinationHealthy = stored; if( !stored || !completed ) { - return CompletableFuture.failedFuture( new CloudException( "cannot copy " + source + " to " + destination ) ); + throw new CloudException( "cannot copy " + source + " to " + destination ); } - - return CompletableFuture.completedFuture( null ); } catch( IOException e ) { - return CompletableFuture.failedFuture( new CloudException( e ) ); + throw new CloudException( e ); } finally { release( sourceClient, sourceHealthy ); release( destinationClient, destinationHealthy ); @@ -386,18 +381,18 @@ public CompletableFuture copyAsync( CloudURI source, CloudURI destination } @Override - public CompletableFuture getInputStreamAsync( CloudURI path ) { + public InputStream getInputStream( CloudURI path ) { FTPClient client = borrow(); try { InputStream in = client.retrieveFileStream( absolute( path.path ) ); if( in == null ) { release( client, true ); - return CompletableFuture.failedFuture( new CloudException( "cannot open " + path ) ); + throw new CloudException( "cannot open " + path ); } - return CompletableFuture.completedFuture( new FtpInputStream( this, client, in ) ); + return new FtpInputStream( this, client, in ); } catch( IOException e ) { release( client, false ); - return CompletableFuture.failedFuture( new CloudException( e ) ); + throw new CloudException( e ); } } @@ -420,7 +415,7 @@ public OutputStream getOutputStream( CloudURI path, Map tags ) { } @Override - public CompletableFuture uploadAsync( CloudURI destination, BlobData blobData ) { + public void upload( CloudURI destination, BlobData blobData ) { FTPClient client = borrow(); boolean healthy = false; try { @@ -447,20 +442,19 @@ public CompletableFuture uploadAsync( CloudURI destination, BlobData blobD }; if( !stored ) { - return CompletableFuture.failedFuture( new CloudException( "cannot upload to " + destination ) ); + throw new CloudException( "cannot upload to " + destination ); } healthy = true; - return CompletableFuture.completedFuture( null ); } catch( IOException e ) { - return CompletableFuture.failedFuture( new CloudException( e ) ); + throw new CloudException( e ); } finally { release( client, healthy ); } } @Override - public CompletableFuture> listAsync( CloudURI path, ListOptions listOptions ) { + public PageSet list( CloudURI path, ListOptions listOptions ) { FTPClient client = borrow(); boolean healthy = false; try { @@ -482,9 +476,9 @@ public CompletableFuture> listAsync( C String nextToken = listOptions.maxKeys != null ? String.valueOf( skip + result.size() ) : null; healthy = true; - return CompletableFuture.completedFuture( new PageSet<>( nextToken, result ) ); + return new PageSet<>( nextToken, result ); } catch( IOException e ) { - return CompletableFuture.failedFuture( new CloudException( e ) ); + throw new CloudException( e ); } finally { release( client, healthy ); } diff --git a/oap-storage/oap-storage-cloud-test/src/test/java/oap/storage/cloud/FileSystemFileTest.java b/oap-storage/oap-storage-cloud-test/src/test/java/oap/storage/cloud/FileSystemFileTest.java index f6a246b6cd..5c5cfadfe2 100644 --- a/oap-storage/oap-storage-cloud-test/src/test/java/oap/storage/cloud/FileSystemFileTest.java +++ b/oap-storage/oap-storage-cloud-test/src/test/java/oap/storage/cloud/FileSystemFileTest.java @@ -21,7 +21,6 @@ import java.nio.file.Path; import java.util.LinkedHashMap; import java.util.Map; -import java.util.concurrent.TimeUnit; import static java.nio.charset.StandardCharsets.UTF_8; import static oap.testng.Asserts.assertFile; @@ -134,8 +133,7 @@ public void testCopy() { Files.write( path, "test string", ContentWriter.ofString() ); try( FileSystem fileSystem = new FileSystem( getFileSystemConfiguration() ) ) { - assertThat( fileSystem.copyAsync( fileSystem.toLocalFilePath( path ), new CloudURI( "file://logs/my-file.txt.gz" ), Map.of() ) ) - .succeedsWithin( 30, TimeUnit.SECONDS ); + fileSystem.copy( fileSystem.toLocalFilePath( path ), new CloudURI( "file://logs/my-file.txt.gz" ), Map.of() ); InputStream inputStream = fileSystem.getInputStream( new CloudURI( "file://logs/my-file.txt.gz" ) ); diff --git a/oap-storage/oap-storage-cloud-test/src/test/java/oap/storage/cloud/FileSystemFtpTest.java b/oap-storage/oap-storage-cloud-test/src/test/java/oap/storage/cloud/FileSystemFtpTest.java index 9a379e13f1..14a2df9ac3 100644 --- a/oap-storage/oap-storage-cloud-test/src/test/java/oap/storage/cloud/FileSystemFtpTest.java +++ b/oap-storage/oap-storage-cloud-test/src/test/java/oap/storage/cloud/FileSystemFtpTest.java @@ -113,8 +113,7 @@ public void testCopy() { Files.write( path, "test string", ContentWriter.ofString() ); try( FileSystem fileSystem = new FileSystem( getFileSystemConfiguration() ) ) { - assertThat( fileSystem.copyAsync( fileSystem.toLocalFilePath( path ), new CloudURI( "ftp://logs/my-file.txt.gz" ), Map.of() ) ) - .succeedsWithin( 30, TimeUnit.SECONDS ); + fileSystem.copy( fileSystem.toLocalFilePath( path ), new CloudURI( "ftp://logs/my-file.txt.gz" ), Map.of() ); InputStream inputStream = fileSystem.getInputStream( new CloudURI( "ftp://logs/my-file.txt.gz" ) ); diff --git a/oap-storage/oap-storage-cloud-test/src/test/java/oap/storage/cloud/FileSystemS3Test.java b/oap-storage/oap-storage-cloud-test/src/test/java/oap/storage/cloud/FileSystemS3Test.java index 5a59e194a6..76593c7cb1 100644 --- a/oap-storage/oap-storage-cloud-test/src/test/java/oap/storage/cloud/FileSystemS3Test.java +++ b/oap-storage/oap-storage-cloud-test/src/test/java/oap/storage/cloud/FileSystemS3Test.java @@ -23,7 +23,6 @@ import java.net.URISyntaxException; import java.nio.file.Path; import java.util.Map; -import java.util.concurrent.TimeUnit; import static java.nio.charset.StandardCharsets.UTF_8; import static oap.io.content.ContentReader.ofString; @@ -96,6 +95,32 @@ public void testGetOutputStream() throws IOException { } } + @Test + public void testGetOutputStreamMultipart() throws IOException { + int partSize = 5 * 1024 * 1024; + byte[] chunk = new byte[ 1024 * 1024 ]; + for( int i = 0; i < chunk.length; i++ ) { + chunk[i] = ( byte ) ( 'a' + ( i % 26 ) ); + } + + int totalSize = partSize + chunk.length; + + try( FileSystem fileSystem = new FileSystem( getFileSystemConfiguration() ) ) { + try( OutputStream outputStream = fileSystem.getOutputStream( new CloudURI( "s3://" + TEST_BUCKET + "/logs/big-file.bin" ), Map.of( "test-tag", "tag-val" ) ) ) { + int written = 0; + while( written < totalSize ) { + outputStream.write( chunk ); + written += chunk.length; + } + } + + assertThat( s3mockFixture.readTags( TEST_BUCKET, "logs/big-file.bin" ) ).contains( entry( "test-tag", "tag-val" ) ); + + byte[] content = s3mockFixture.readFile( TEST_BUCKET, "logs/big-file.bin", ContentReader.ofBytes(), Encoding.PLAIN ); + assertThat( content.length ).isEqualTo( totalSize ); + } + } + @Test public void testGetMetadata() throws URISyntaxException { Path path = testDirectoryFixture.testPath( "my-file.txt" ); @@ -133,8 +158,7 @@ public void testCopy() { Files.write( path, "test string", ContentWriter.ofString() ); try( FileSystem fileSystem = new FileSystem( getFileSystemConfiguration() ) ) { - assertThat( fileSystem.copyAsync( fileSystem.toLocalFilePath( path ), new CloudURI( "s3://" + TEST_BUCKET + "/logs/my-file.txt.gz" ), Map.of( "tag1", "va1", "tag2", "val2" ) ) ) - .succeedsWithin( 30, TimeUnit.SECONDS ); + fileSystem.copy( fileSystem.toLocalFilePath( path ), new CloudURI( "s3://" + TEST_BUCKET + "/logs/my-file.txt.gz" ), Map.of( "tag1", "va1", "tag2", "val2" ) ); InputStream inputStream = fileSystem.getInputStream( new CloudURI( "s3://" + TEST_BUCKET + "/logs/my-file.txt.gz" ) ); diff --git a/oap-storage/oap-storage-cloud/README.md b/oap-storage/oap-storage-cloud/README.md index 8366cd5b64..7feab3c23c 100644 --- a/oap-storage/oap-storage-cloud/README.md +++ b/oap-storage/oap-storage-cloud/README.md @@ -126,7 +126,7 @@ CloudURI defaultUri = fs.getDefaultURL( "reports/today.json" ); ### Operations reference -Every synchronous method has an `...Async` counterpart returning `CompletableFuture`. +All methods are synchronous/blocking. | Method | Description | |---|---| @@ -159,7 +159,7 @@ s3=com.example.MyS3CloudApi The class must have a constructor `(FileSystemConfiguration, String container)`. -Every method has a required `...Async(CloudURI, ...)` variant and a default synchronous wrapper that calls `.join()`. +Every method is a required synchronous, blocking method. --- diff --git a/oap-storage/oap-storage-cloud/src/main/java/oap/storage/cloud/FileSystem.java b/oap-storage/oap-storage-cloud/src/main/java/oap/storage/cloud/FileSystem.java index 92458182ad..7edb285c14 100644 --- a/oap-storage/oap-storage-cloud/src/main/java/oap/storage/cloud/FileSystem.java +++ b/oap-storage/oap-storage-cloud/src/main/java/oap/storage/cloud/FileSystem.java @@ -28,7 +28,6 @@ import java.util.List; import java.util.Map; import java.util.Properties; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; @Slf4j @@ -92,12 +91,6 @@ private FileSystemCloudApi getCloudApi( CloudURI cloudURI ) throws CloudExceptio } } - public CompletableFuture getInputStreamAsync( CloudURI path ) { - log.debug( "getInputStream {}", path ); - - return getCloudApi( path ).getInputStreamAsync( path ); - } - public InputStream getInputStream( CloudURI path ) throws CloudException { log.debug( "getInputStream {}", path ); @@ -108,16 +101,6 @@ public OutputStream getOutputStream( CloudURI cloudURI, Map tags return getCloudApi( cloudURI ).getOutputStream( cloudURI, tags ); } - public CompletableFuture downloadFileAsync( String source, Path destination ) { - return downloadFileAsync( new CloudURI( source ), destination ); - } - - public CompletableFuture downloadFileAsync( CloudURI source, Path destination ) { - log.debug( "downloadFile {} to {}", source, destination ); - - return getCloudApi( source ).downloadFileAsync( source, destination ); - } - public void downloadFile( String source, Path destination ) throws CloudException { downloadFile( new CloudURI( source ), destination ); } @@ -128,35 +111,23 @@ public void downloadFile( CloudURI source, Path destination ) throws CloudExcept getCloudApi( source ).downloadFile( source, destination ); } - public CompletableFuture uploadAsync( CloudURI destination, BlobData blobData ) { - log.debug( "upload byte[] to {} (blobData {})", destination, blobData ); - - return getCloudApi( destination ).uploadAsync( destination, blobData ); - } - public void upload( CloudURI destination, BlobData blobData ) throws CloudException { log.debug( "upload byte[] to {} (blobData {})", destination, blobData ); getCloudApi( destination ).upload( destination, blobData ); } - public CompletableFuture copyAsync( CloudURI source, CloudURI destination, Map tags ) { + public void copy( CloudURI source, CloudURI destination, Map tags ) throws CloudException { log.debug( "copy {} to {} (tags {})", source, destination, tags ); - FileSystemCloudApi sourceCloudApi = getCloudApi( source ); FileSystemCloudApi destinationCloudApi = getCloudApi( destination ); - return sourceCloudApi.getInputStreamAsync( source ) - .thenCompose( inputStream -> - destinationCloudApi.uploadAsync( destination, BlobData.builder().content( inputStream ).tags( tags ).build() ) - .thenAccept( _ -> Closeables.close( inputStream ) ) ); - } - - public void copy( CloudURI source, CloudURI destination, Map tags ) throws CloudException { - log.debug( "copy {} to {} (tags {})", source, destination, tags ); + if( isLocalFile( source ) ) { + destinationCloudApi.upload( destination, BlobData.builder().content( toFile( source ).toPath() ).tags( tags ).build() ); + return; + } FileSystemCloudApi sourceCloudApi = getCloudApi( source ); - FileSystemCloudApi destinationCloudApi = getCloudApi( destination ); try( InputStream inputStream = sourceCloudApi.getInputStream( source ) ) { destinationCloudApi.upload( destination, BlobData.builder().content( inputStream ).tags( tags ).build() ); @@ -166,20 +137,10 @@ public void copy( CloudURI source, CloudURI destination, Map tag } } - public CompletableFuture> listAsync( CloudURI path, ListOptions listOptions ) { - return getCloudApi( path ).listAsync( path, listOptions ); - } - public PageSet list( CloudURI path, ListOptions listOptions ) throws CloudException { return getCloudApi( path ).list( path, listOptions ); } - public CompletableFuture getMetadataAsync( CloudURI path ) { - log.debug( "getMetadata {}", path ); - - return getCloudApi( path ).getMetadataAsync( path ); - } - @Nullable public StorageItem getMetadata( CloudURI path ) throws CloudException { log.debug( "getMetadata {}", path ); @@ -187,66 +148,36 @@ public StorageItem getMetadata( CloudURI path ) throws CloudException { return getCloudApi( path ).getMetadata( path ); } - public CompletableFuture deleteBlobAsync( CloudURI path ) { - return getCloudApi( path ).deleteBlobAsync( path ); - } - public void deleteBlob( CloudURI path ) throws CloudException { log.debug( "deleteBlob {}", path ); getCloudApi( path ).deleteBlob( path ); } - public CompletableFuture deleteContainerIfEmptyAsync( CloudURI path ) { - return getCloudApi( path ).deleteContainerIfEmptyAsync( path ); - } - public boolean deleteContainerIfEmpty( CloudURI path ) { log.debug( "deleteContainerIfEmpty {}", path ); return getCloudApi( path ).deleteContainerIfEmpty( path ); } - public CompletableFuture deleteContainerAsync( CloudURI path ) { - return getCloudApi( path ).deleteContainerAsync( path ); - } - public void deleteContainer( CloudURI path ) throws CloudException { log.debug( "deleteContainer {}", path ); getCloudApi( path ).deleteContainer( path ); } - public CompletableFuture blobExistsAsync( CloudURI path ) { - log.debug( "blobExists {}", path ); - - return getCloudApi( path ).blobExistsAsync( path ); - } - public boolean blobExists( CloudURI path ) throws CloudException { log.debug( "blobExists {}", path ); return getCloudApi( path ).blobExists( path ); } - public CompletableFuture containerExistsAsync( CloudURI path ) { - log.debug( "containerExists {}", path ); - - return getCloudApi( path ).containerExistsAsync( path ); - } - public boolean containerExists( CloudURI path ) { log.debug( "containerExists {}", path ); return getCloudApi( path ).containerExists( path ); } - public CompletableFuture createContainerAsync( CloudURI path ) { - log.debug( "createContainer {}", path ); - - return getCloudApi( path ).createContainerAsync( path ); - } - public boolean createContainer( CloudURI path ) throws CloudException { log.debug( "createContainer {}", path ); @@ -271,6 +202,10 @@ public CloudURI toLocalFilePath( Path path ) { : Paths.get( "/" ).relativize( path ).toString() ); } + public boolean isLocalFile( CloudURI cloudURI ) { + return "file".equals( cloudURI.scheme ); + } + public File toFile( CloudURI cloudURI ) { Preconditions.checkArgument( "file".equals( cloudURI.scheme ) ); diff --git a/oap-storage/oap-storage-cloud/src/main/java/oap/storage/cloud/FileSystemCloudApi.java b/oap-storage/oap-storage-cloud/src/main/java/oap/storage/cloud/FileSystemCloudApi.java index 8a9d7464e1..73baca68da 100644 --- a/oap-storage/oap-storage-cloud/src/main/java/oap/storage/cloud/FileSystemCloudApi.java +++ b/oap-storage/oap-storage-cloud/src/main/java/oap/storage/cloud/FileSystemCloudApi.java @@ -4,139 +4,31 @@ import java.io.OutputStream; import java.nio.file.Path; import java.util.Map; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CompletionException; -import java.util.concurrent.ExecutionException; public interface FileSystemCloudApi extends AutoCloseable { - CompletableFuture blobExistsAsync( CloudURI path ); + boolean blobExists( CloudURI path ) throws CloudException; - default boolean blobExists( CloudURI path ) throws CloudException { - try { - return blobExistsAsync( path ).join(); - } catch( Exception e ) { - throw propagate( e ); - } - } + boolean containerExists( CloudURI path ) throws CloudException; - CompletableFuture containerExistsAsync( CloudURI path ); + void deleteBlob( CloudURI path ) throws CloudException; - default boolean containerExists( CloudURI path ) throws CloudException { - try { - return containerExistsAsync( path ).join(); - } catch( Exception e ) { - throw propagate( e ); - } - } + void deleteContainer( CloudURI path ) throws CloudException; - CompletableFuture deleteBlobAsync( CloudURI path ); + boolean createContainer( CloudURI path ) throws CloudException; - default void deleteBlob( CloudURI path ) throws CloudException { - try { - deleteBlobAsync( path ).join(); - } catch( Exception e ) { - throw propagate( e ); - } - } + boolean deleteContainerIfEmpty( CloudURI path ) throws CloudException; - CompletableFuture deleteContainerAsync( CloudURI path ) throws CloudException; + FileSystem.StorageItem getMetadata( CloudURI path ) throws CloudException; - default void deleteContainer( CloudURI path ) throws CloudException { - try { - deleteContainerAsync( path ).join(); - } catch( Exception e ) { - throw propagate( e ); - } - } + void downloadFile( CloudURI source, Path destination ) throws CloudException; - CompletableFuture createContainerAsync( CloudURI path ); + void copy( CloudURI source, CloudURI destination ) throws CloudException; - default boolean createContainer( CloudURI path ) throws CloudException { - try { - return createContainerAsync( path ).join(); - } catch( Exception e ) { - throw propagate( e ); - } - } - - CompletableFuture deleteContainerIfEmptyAsync( CloudURI path ); - - default boolean deleteContainerIfEmpty( CloudURI path ) throws CloudException { - try { - return deleteContainerIfEmptyAsync( path ).join(); - } catch( Exception e ) { - throw propagate( e ); - } - } - - CompletableFuture getMetadataAsync( CloudURI path ); - - default FileSystem.StorageItem getMetadata( CloudURI path ) throws CloudException { - try { - return getMetadataAsync( path ).join(); - } catch( Exception e ) { - throw propagate( e ); - } - } - - CompletableFuture downloadFileAsync( CloudURI source, Path destination ); - - default void downloadFile( CloudURI source, Path destination ) throws CloudException { - try { - downloadFileAsync( source, destination ).join(); - } catch( Exception e ) { - throw propagate( e ); - } - } - - CompletableFuture copyAsync( CloudURI source, CloudURI destination ); - - default void copy( CloudURI source, CloudURI destination ) throws CloudException { - try { - copyAsync( source, destination ).join(); - } catch( Exception e ) { - throw propagate( e ); - } - } - - CompletableFuture getInputStreamAsync( CloudURI path ); - - default InputStream getInputStream( CloudURI path ) throws CloudException { - try { - return getInputStreamAsync( path ).join(); - } catch( Exception e ) { - throw propagate( e ); - } - } + InputStream getInputStream( CloudURI path ) throws CloudException; OutputStream getOutputStream( CloudURI cloudURI, Map tags ); - CompletableFuture uploadAsync( CloudURI destination, BlobData blobData ); - - default void upload( CloudURI destination, BlobData blobData ) throws CloudException { - try { - uploadAsync( destination, blobData ).join(); - } catch( Exception e ) { - throw propagate( e ); - } - } - - CompletableFuture> listAsync( CloudURI path, ListOptions listOptions ); - - default PageSet list( CloudURI path, ListOptions listOptions ) throws CloudException { - try { - return listAsync( path, listOptions ).join(); - } catch( Exception e ) { - throw propagate( e ); - } - } + void upload( CloudURI destination, BlobData blobData ) throws CloudException; - default CloudException propagate( Throwable e ) throws CloudException { - return switch( e ) { - case CloudException ce -> ce; - case ExecutionException ee -> new CloudException( ee.getCause() ); - case CompletionException ce -> new CloudException( ce.getCause() ); - case null, default -> new CloudException( e ); - }; - } + PageSet list( CloudURI path, ListOptions listOptions ) throws CloudException; } diff --git a/oap-storage/oap-storage-cloud/src/main/java/oap/storage/cloud/FileSystemCloudApiLocalFs.java b/oap-storage/oap-storage-cloud/src/main/java/oap/storage/cloud/FileSystemCloudApiLocalFs.java index c5e072ab25..9dc256d6f8 100644 --- a/oap-storage/oap-storage-cloud/src/main/java/oap/storage/cloud/FileSystemCloudApiLocalFs.java +++ b/oap-storage/oap-storage-cloud/src/main/java/oap/storage/cloud/FileSystemCloudApiLocalFs.java @@ -22,7 +22,6 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; -import java.util.concurrent.CompletableFuture; import java.util.stream.Stream; @Slf4j @@ -44,8 +43,8 @@ public FileSystemCloudApiLocalFs( FileSystemConfiguration fileSystemConfiguratio } @Override - public CompletableFuture blobExistsAsync( CloudURI path ) throws CloudException { - return CompletableFuture.completedFuture( getPath( path ).toFile().exists() ); + public boolean blobExists( CloudURI path ) throws CloudException { + return getPath( path ).toFile().exists(); } protected Path getPath( CloudURI path ) { @@ -56,12 +55,12 @@ protected Path getPath( CloudURI path ) { } @Override - public CompletableFuture containerExistsAsync( CloudURI path ) throws CloudException { - return CompletableFuture.completedFuture( Files.isDirectory( getPath( path ) ) ); + public boolean containerExists( CloudURI path ) throws CloudException { + return Files.isDirectory( getPath( path ) ); } @Override - public CompletableFuture deleteBlobAsync( CloudURI path ) { + public void deleteBlob( CloudURI path ) { try { Path fsPath = getPath( path ); Files.delete( fsPath ); @@ -74,92 +73,83 @@ public CompletableFuture deleteBlobAsync( CloudURI path ) { parent = parent.getParent(); } } - - return CompletableFuture.completedFuture( null ); } catch( IOException e ) { - return CompletableFuture.failedFuture( new CloudException( e ) ); + throw new CloudException( e ); } } @Override - public CompletableFuture deleteContainerAsync( CloudURI path ) { - + public void deleteContainer( CloudURI path ) { Path fsPath = getPath( path ); if( !Files.isDirectory( fsPath ) ) { - return CompletableFuture.failedFuture( new CloudException( "Not a directory" ) ); + throw new CloudException( "Not a directory" ); } try { oap.io.Files.delete( fsPath ); } catch( Exception e ) { - return CompletableFuture.failedFuture( new CloudException( e ) ); + throw new CloudException( e ); } - - return CompletableFuture.completedFuture( null ); } @Override - public CompletableFuture createContainerAsync( CloudURI path ) { - return CompletableFuture.completedFuture( false ); + public boolean createContainer( CloudURI path ) { + return false; } @Override - public CompletableFuture deleteContainerIfEmptyAsync( CloudURI path ) { - return CompletableFuture.completedFuture( false ); + public boolean deleteContainerIfEmpty( CloudURI path ) { + return false; } @Override - public CompletableFuture getMetadataAsync( CloudURI path ) { + public FileSystem.StorageItem getMetadata( CloudURI path ) { try { Path fsPath = getPath( path ); - return CompletableFuture.completedFuture( new FileSystem.StorageItemImpl( + return new FileSystem.StorageItemImpl( fsPath.toString(), "", fsPath.toUri(), new DateTime( Files.getLastModifiedTime( fsPath ).toMillis(), DateTimeZone.UTC ), Files.size( fsPath ), - Files.isDirectory( fsPath ) ? "application/x-directory" : "" ) ); + Files.isDirectory( fsPath ) ? "application/x-directory" : "" ); } catch( NoSuchFileException e ) { - return CompletableFuture.completedFuture( null ); + return null; } catch( IOException e ) { - return CompletableFuture.failedFuture( new CloudException( e ) ); + throw new CloudException( e ); } } @Override - public CompletableFuture downloadFileAsync( CloudURI source, Path destination ) { + public void downloadFile( CloudURI source, Path destination ) { try { Files.copy( getPath( source ), destination ); - - return CompletableFuture.completedFuture( null ); } catch( IOException e ) { - return CompletableFuture.failedFuture( new CloudException( e ) ); + throw new CloudException( e ); } } @Override - public CompletableFuture copyAsync( CloudURI source, CloudURI destination ) { + public void copy( CloudURI source, CloudURI destination ) { Preconditions.checkArgument( source.scheme.equals( destination.scheme ) ); try { Files.copy( getPath( source ), getPath( destination ) ); - - return CompletableFuture.completedFuture( null ); } catch( IOException e ) { - return CompletableFuture.failedFuture( new CloudException( e ) ); + throw new CloudException( e ); } } @Override - public CompletableFuture getInputStreamAsync( CloudURI path ) { + public InputStream getInputStream( CloudURI path ) { try { Path fsPath = getPath( path ); - log.debug( "getInputStreamAsync '{}' -> '{}'", path, fsPath ); + log.debug( "getInputStream '{}' -> '{}'", path, fsPath ); - return CompletableFuture.completedFuture( Files.newInputStream( fsPath ) ); + return Files.newInputStream( fsPath ); } catch( IOException e ) { - return CompletableFuture.failedFuture( new CloudException( e ) ); + throw new CloudException( e ); } } @@ -178,11 +168,11 @@ public OutputStream getOutputStream( CloudURI path, Map tags ) t } @Override - public CompletableFuture uploadAsync( CloudURI destination, BlobData blobData ) throws CloudException { + public void upload( CloudURI destination, BlobData blobData ) throws CloudException { try { Path destinationFs = getPath( destination ); - log.debug( "uploadAsync '{}' -> '{}'", destination, destinationFs ); + log.debug( "upload '{}' -> '{}'", destination, destinationFs ); switch( blobData.content ) { case InputStream inputStream -> IoStreams.write( destinationFs, IoStreams.Encoding.PLAIN, inputStream ); @@ -201,14 +191,13 @@ public CompletableFuture uploadAsync( CloudURI destination, BlobData blobD } case null, default -> throw new CloudException( "Unknown content type " + blobData.content.getClass() ); } - return CompletableFuture.completedFuture( null ); } catch( IOException e ) { - return CompletableFuture.failedFuture( new CloudException( e ) ); + throw new CloudException( e ); } } @Override - public CompletableFuture> listAsync( CloudURI path, ListOptions listOptions ) { + public PageSet list( CloudURI path, ListOptions listOptions ) { try { Path filePath = getPath( path ); @@ -239,15 +228,14 @@ public CompletableFuture> listAsync( C Files.size( file ), "" ) ); } catch( IOException e ) { - return CompletableFuture.failedFuture( new CloudException( e ) ); + throw new CloudException( e ); } } - return CompletableFuture.completedFuture( new PageSet<>( - listOptions.maxKeys != null ? listOptions.maxKeys.toString() : null, list ) ); + return new PageSet<>( listOptions.maxKeys != null ? listOptions.maxKeys.toString() : null, list ); } catch( IOException e ) { - return CompletableFuture.failedFuture( new CloudException( e ) ); + throw new CloudException( e ); } } diff --git a/pom.xml b/pom.xml index de5b1cee57..600728de23 100644 --- a/pom.xml +++ b/pom.xml @@ -67,7 +67,7 @@ - 25.10.2 + 25.10.3 25.0.1 25.0.0