[XEN-3146-migrate] only log progress events every N bytes

Author: NotAPenguin0

integration-tests/src/test/resources/solr.xml

@@ -13,6 +13,7 @@
 
             <bool name="s3.path.style.access.enabled">${S3_PATH_STYLE_ACCESS_ENABLED:false}</bool>
             <bool name="s3.client.checksumValidationEnabled">${S3_CLIENT_CHECKSUM_VALIDATION_ENABLED:true}</bool>
+            <int name="s3.client.progressLogByteInterval">${S3_CLIENT_PROGRESS_LOG_BYTE_INTERVAL:4194304}</int>
         </repository>
     </backup>
 </solr>

solr-backup/src/main/java/eu/xenit/solr/backup/s3/S3BackupRepositoryConfig.java

@@ -37,6 +37,7 @@ public class S3BackupRepositoryConfig {
     public static final String S3_CLIENT_CHECKSUM_VALIDATION_ENABLED = "s3.client.checksum.validation.enabled";
     public static final String S3_PROXY_HOST = "s3.proxy.host";
     public static final String S3_PROXY_PORT = "s3.proxy.port";
+    public static final String S3_CLIENT_PROGRESS_LOG_BYTE_INTERVAL = "s3.client.progressLogByteInterval";
 
     private final String bucketName;
 
@@ -56,6 +57,8 @@ public class S3BackupRepositoryConfig {
 
     private final Boolean checksumValidationEnabled;
 
+    private final Integer progressLogByteInterval;
+
 
     public S3BackupRepositoryConfig(NamedList<?> config) {
         region = getStringConfig(config, S3_REGION);
@@ -67,13 +70,14 @@ public S3BackupRepositoryConfig(NamedList<?> config) {
         secretKey = getStringConfig(config, S3_SECRET_KEY);
         pathStyleAccessEnabled = getBooleanConfig(config, S3_PATH_STYLE_ACCESS_ENABLED);
         checksumValidationEnabled = getBooleanConfig(config, S3_CLIENT_CHECKSUM_VALIDATION_ENABLED);
+        progressLogByteInterval = getIntConfig(config, S3_CLIENT_PROGRESS_LOG_BYTE_INTERVAL);
     }
 
     /**
      * @return a {@link S3StorageClient} from the provided config.
      */
     public S3StorageClient buildClient() throws URISyntaxException {
-        return new S3StorageClient(bucketName, region, proxyHost, proxyPort, endpoint, accessKey, secretKey, pathStyleAccessEnabled, checksumValidationEnabled);
+        return new S3StorageClient(this);
     }
 
     private static String getStringConfig(NamedList<?> config, String property) {

solr-backup/src/main/java/eu/xenit/solr/backup/s3/S3OutputStream.java

@@ -55,26 +55,22 @@ public class S3OutputStream extends OutputStream {
     static final int MIN_PART_SIZE = 5242880;
 
     private final S3Client s3Client;
-    private final String bucketName;
     private final String key;
-    // TODO: upgrade from 1.x sdk
-    // private final SyncProgressListener progressListener;
     private volatile boolean closed;
     private final ByteBuffer buffer;
     private MultipartUpload multiPartUpload;
+    private final S3BackupRepositoryConfig configuration;
 
-    public S3OutputStream(S3Client s3Client, String key, String bucketName) {
+    public S3OutputStream(S3Client s3Client, String key, S3BackupRepositoryConfig configuration) {
         this.s3Client = s3Client;
-        this.bucketName = bucketName;
+        this.configuration = configuration;
         this.key = key;
         this.closed = false;
         this.buffer = ByteBuffer.allocate(PART_SIZE);
-        // TODO: upgrade from 1.x sdk
-        //this.progressListener = new ConnectProgressListener();
         this.multiPartUpload = null;
 
         if (log.isDebugEnabled()) {
-            log.debug("Created S3OutputStream for bucketName '{}' key '{}'", bucketName, key);
+            log.debug("Created S3OutputStream for bucketName '{}' key '{}'", this.configuration.getBucketName(), key);
         }
     }
 
@@ -134,7 +130,7 @@ private void uploadPart(boolean isLastPart) throws IOException {
 
         if (multiPartUpload == null) {
             if (log.isDebugEnabled()) {
-                log.debug("New multi-part upload for bucketName '{}' key '{}'", bucketName, key);
+                log.debug("New multi-part upload for bucketName '{}' key '{}'", this.configuration.getBucketName(), key);
             }
             multiPartUpload = newMultipartUpload();
         }
@@ -144,7 +140,7 @@ private void uploadPart(boolean isLastPart) throws IOException {
             if (multiPartUpload != null) {
                 multiPartUpload.abort();
                 if (log.isDebugEnabled()) {
-                    log.debug("Multipart upload aborted for bucketName '{}' key '{}'.", bucketName, key);
+                    log.debug("Multipart upload aborted for bucketName '{}' key '{}'.", this.configuration.getBucketName(), key);
                 }
             }
             throw new S3Exception("Part upload failed: ", e);
@@ -187,7 +183,7 @@ public void close() throws IOException {
     private MultipartUpload newMultipartUpload() throws IOException {
         CreateMultipartUploadRequest initRequest =
                 CreateMultipartUploadRequest.builder()
-                        .bucket(bucketName)
+                        .bucket(this.configuration.getBucketName())
                         .key(key)
                         .build();
         try {
@@ -207,7 +203,7 @@ public MultipartUpload(String uploadId) {
             if (log.isDebugEnabled()) {
                 log.debug(
                         "Initiated multi-part upload for bucketName '{}' key '{}' with id '{}'",
-                        bucketName,
+                        configuration.getBucketName(),
                         key,
                         uploadId);
             }
@@ -224,15 +220,24 @@ void uploadPart(ByteArrayInputStream inputStream, int partSize) {
              * - Pass `contentLength` to request
              * - Wrap the input stream into a progress listening input stream.
              */
-            Consumer<Long> progressListener = (bytesTransferred) -> {
-                log.debug("Progress: {} bytes", bytesTransferred);
+            Consumer<Long> progressListener = new Consumer<>() {
+                private Long bytesSinceLastCheckpoint = 0L;
+
+                public void accept(Long bytesTransferred) {
+                    // Only log every interval of bytes, instead of each event
+                    bytesSinceLastCheckpoint += bytesTransferred;
+                    if (bytesSinceLastCheckpoint > configuration.getProgressLogByteInterval()) {
+                        log.debug("Progress: {} bytes", bytesTransferred);
+                        bytesSinceLastCheckpoint = 0L;
+                    }
+                }
             };
             InputStream trackedStream = new ProgressTrackingInputStream(inputStream, progressListener);
             RequestBody body = RequestBody.fromInputStream(trackedStream, partSize);
 
             UploadPartRequest request =
                     UploadPartRequest.builder()
-                            .bucket(bucketName)
+                            .bucket(configuration.getBucketName())
                             .key(key)
                             .uploadId(uploadId)
                             .partNumber(currentPartNumber)
@@ -259,7 +264,7 @@ void complete() {
             }
             CompleteMultipartUploadRequest completeRequest =
                     CompleteMultipartUploadRequest.builder()
-                            .bucket(bucketName)
+                            .bucket(configuration.getBucketName())
                             .key(key)
                             .uploadId(uploadId)
                             .multipartUpload(
@@ -277,7 +282,7 @@ public void abort() {
             try {
                 s3Client.abortMultipartUpload(AbortMultipartUploadRequest
                         .builder()
-                        .bucket(bucketName)
+                        .bucket(configuration.getBucketName())
                         .key(key)
                         .uploadId(uploadId)
                         .build());

solr-backup/src/main/java/eu/xenit/solr/backup/s3/S3StorageClient.java

@@ -93,36 +93,24 @@ class S3StorageClient {
     private static final Set<String> NOT_FOUND_CODES = Set.of("NoSuchKey", "404 Not Found");
 
     private final S3Client s3Client;
+    private final S3BackupRepositoryConfig configuration;
 
-    /**
-     * The S3 bucket where we read/write all data.
-     */
-    private final String bucketName;
-
-    S3StorageClient(
-            String bucketName, String region, String proxyHost, int proxyPort, String endpoint, String accessKey, String secretKey, Boolean pathStyleAccessEnabled, Boolean checksumValidationEnabled) throws URISyntaxException {
-        this(createInternalClient(region, proxyHost, proxyPort, endpoint, accessKey, secretKey, pathStyleAccessEnabled, checksumValidationEnabled), bucketName);
+    S3StorageClient(S3BackupRepositoryConfig config) throws URISyntaxException {
+        this(createInternalClient(config), config);
     }
 
     @VisibleForTesting
-    S3StorageClient(S3Client s3Client, String bucketName) {
+    S3StorageClient(S3Client s3Client, S3BackupRepositoryConfig configuration) {
         this.s3Client = s3Client;
-        this.bucketName = bucketName;
+        this.configuration = configuration;
     }
 
-    private static S3Client createInternalClient(
-            String region,
-            String proxyHost,
-            int proxyPort,
-            String endpoint,
-            String accessKey,
-            String secretKey, Boolean pathStyleAccessEnabled,
-            Boolean checksumValidationEnabled) throws URISyntaxException {
+    private static S3Client createInternalClient(S3BackupRepositoryConfig config) throws URISyntaxException {
 
         S3ClientBuilder clientBuilder = S3Client.builder();
 
         S3Configuration configuration = S3Configuration.builder()
-                .checksumValidationEnabled(checksumValidationEnabled)
+                .checksumValidationEnabled(config.getChecksumValidationEnabled())
                 .build();
         clientBuilder.serviceConfiguration(configuration);
 
@@ -131,9 +119,9 @@ private static S3Client createInternalClient(
          * not on a general client configuration object.
          */
         ApacheHttpClient.Builder httpClientBuilder = ApacheHttpClient.builder();
-        if (!StringUtils.isEmpty(proxyHost)) {
+        if (!StringUtils.isEmpty(config.getProxyHost())) {
             ProxyConfiguration.Builder proxyConfigBuilder = ProxyConfiguration.builder()
-                    .endpoint(URI.create(proxyHost + ":" + proxyPort));
+                    .endpoint(URI.create(config.getProxyHost() + ":" + config.getProxyPort()));
             httpClientBuilder.proxyConfiguration(proxyConfigBuilder.build());
         }
         clientBuilder.httpClientBuilder(httpClientBuilder);
@@ -145,9 +133,9 @@ private static S3Client createInternalClient(
          */
         clientBuilder.overrideConfiguration(ClientOverrideConfiguration.builder().build());
 
-        if (!(StringUtils.isEmpty(accessKey) || StringUtils.isEmpty(secretKey))) {
+        if (!(StringUtils.isEmpty(config.getAccessKey()) || StringUtils.isEmpty(config.getSecretKey()))) {
             clientBuilder.credentialsProvider(
-                    StaticCredentialsProvider.create(AwsBasicCredentials.create(accessKey, secretKey)));
+                    StaticCredentialsProvider.create(AwsBasicCredentials.create(config.getAccessKey(), config.getSecretKey())));
         } else {
             log.info("No accessKey or secretKey configured, using default credentials provider chain");
         }
@@ -156,17 +144,17 @@ private static S3Client createInternalClient(
          * SDK v2 Migration: `setEndpointConfiguration` from v1 is replaced by
          * `endpointOverride`. The region must still be set separately.
          */
-        if (!StringUtils.isEmpty(endpoint)) {
-            clientBuilder.endpointOverride(new URI(endpoint));
+        if (!StringUtils.isEmpty(config.getEndpoint())) {
+            clientBuilder.endpointOverride(new URI(config.getEndpoint()));
         }
-        clientBuilder.region(Region.of(region));
+        clientBuilder.region(Region.of(config.getRegion()));
 
         /*
          * SDK v2 Migration: The method `withPathStyleAccessEnabled(boolean)` from v1 is
          * replaced by `forcePathStyle(boolean)` in v2.
          */
-        if (pathStyleAccessEnabled != null) {
-            clientBuilder.forcePathStyle(pathStyleAccessEnabled);
+        if (config.getPathStyleAccessEnabled() != null) {
+            clientBuilder.forcePathStyle(config.getPathStyleAccessEnabled());
         }
 
         return clientBuilder.build();
@@ -192,7 +180,7 @@ void createDirectory(String path) throws S3Exception {
              * In v2, request parameters like bucket and key are set using builder methods.
              */
             PutObjectRequest putRequest = PutObjectRequest.builder()
-                    .bucket(bucketName)
+                    .bucket(this.configuration.getBucketName())
                     .key(path)
                     .contentType(S3_DIR_CONTENT_TYPE)
                     .metadata(Collections.singletonMap("Content-Type", S3_DIR_CONTENT_TYPE))
@@ -257,7 +245,7 @@ String[] listDir(String path) throws S3Exception {
 
         // The request MUST include the delimiter.
         ListObjectsV2Request listRequest = ListObjectsV2Request.builder()
-                .bucket(bucketName)
+                .bucket(this.configuration.getBucketName())
                 .prefix(prefix)
                 .delimiter(S3_FILE_PATH_DELIMITER)
                 .build();
@@ -312,7 +300,7 @@ HeadObjectResponse getObjectMetadata(String path) throws SdkException {
          * This is the standard v2 way to retrieve object metadata without fetching the object's content.
          */
         HeadObjectRequest request = HeadObjectRequest.builder()
-                .bucket(bucketName)
+                .bucket(this.configuration.getBucketName())
                 .key(path)
                 .build();
         return s3Client.headObject(request);
@@ -418,7 +406,10 @@ InputStream pullStream(String path) throws S3Exception {
         path = sanitizedFilePath(path);
 
         try {
-            ResponseInputStream<GetObjectResponse> requestedObject = s3Client.getObject(GetObjectRequest.builder().bucket(bucketName).key(path)
+            ResponseInputStream<GetObjectResponse> requestedObject = s3Client
+                    .getObject(GetObjectRequest.builder()
+                            .bucket(this.configuration.getBucketName())
+                            .key(path)
                     .build());
             // This InputStream instance needs to be closed by the caller
             return requestedObject;
@@ -441,7 +432,7 @@ OutputStream pushStream(String path) throws S3Exception {
         }
 
         try {
-            return new S3OutputStream(s3Client, path, bucketName);
+            return new S3OutputStream(s3Client, path, this.configuration);
         } catch (SdkException ase) {
             throw handleAmazonException(ase);
         }
@@ -519,7 +510,10 @@ Collection<String> deleteObjects(Collection<String> entries, int batchSize) thro
         if (deleteIndividually) {
             for (ObjectIdentifier k : keysToDelete) {
                 try {
-                    s3Client.deleteObject(DeleteObjectRequest.builder().bucket(bucketName).key(k.key())
+                    s3Client.deleteObject(DeleteObjectRequest
+                            .builder()
+                            .bucket(this.configuration.getBucketName())
+                            .key(k.key())
                             .build());
                     deletedPaths.add(k.key());
                 } catch (SdkException e) {
@@ -541,7 +535,7 @@ private DeleteObjectsRequest createBatchDeleteRequest(List<ObjectIdentifier> key
                 .build();
 
         return DeleteObjectsRequest.builder()
-                .bucket(bucketName)
+                .bucket(this.configuration.getBucketName())
                 .delete(deleteAction)
                 .build();
     }
@@ -554,7 +548,7 @@ private List<String> listAll(String path) throws S3Exception {
          * `ListObjectsV2Request`.
          */
         ListObjectsV2Request listRequest = ListObjectsV2Request.builder()
-                .bucket(bucketName)
+                .bucket(this.configuration.getBucketName())
                 .prefix(prefix)
                 .build();