Retry indefinitely for s3 indices blob read errors (#103300)

This PR makes the s3 readBlob to retry indefinitely on either opening or
reading errors when the operation purpose is Inidces. It perform retries
with no delay within the configured number of max retries. Beyond that,
it retries with increased delay each time with a capped maximum of 10
seconds.

Relates: ES-6453

Author: ywangd

docs/changelog/103300.yaml

@@ -0,0 +1,5 @@
+pr: 103300
+summary: Retry indefinitely for s3 indices blob read errors
+area: Snapshot/Restore
+type: enhancement
+issues: []

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3RetryingInputStream.java

@@ -14,9 +14,9 @@
 import com.amazonaws.services.s3.model.S3Object;
 import com.amazonaws.services.s3.model.S3ObjectInputStream;
 
+import org.apache.logging.log4j.Level;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.apache.logging.log4j.util.Supplier;
 import org.elasticsearch.Version;
 import org.elasticsearch.common.blobstore.OperationPurpose;
 import org.elasticsearch.core.IOUtils;
@@ -48,7 +48,7 @@ class S3RetryingInputStream extends InputStream {
     private final String blobKey;
     private final long start;
     private final long end;
-    private final List<IOException> failures;
+    private final List<Exception> failures;
 
     private S3ObjectInputStream currentStream;
     private long currentStreamFirstOffset;
@@ -77,29 +77,34 @@ class S3RetryingInputStream extends InputStream {
         this.failures = new ArrayList<>(MAX_SUPPRESSED_EXCEPTIONS);
         this.start = start;
         this.end = end;
-        openStream();
+        openStreamWithRetry();
     }
 
-    private void openStream() throws IOException {
-        try (AmazonS3Reference clientReference = blobStore.clientReference()) {
-            final GetObjectRequest getObjectRequest = new GetObjectRequest(blobStore.bucket(), blobKey);
-            getObjectRequest.setRequestMetricCollector(blobStore.getMetricCollector(Operation.GET_OBJECT, purpose));
-            if (currentOffset > 0 || start > 0 || end < Long.MAX_VALUE - 1) {
-                assert start + currentOffset <= end
-                    : "requesting beyond end, start = " + start + " offset=" + currentOffset + " end=" + end;
-                getObjectRequest.setRange(Math.addExact(start, currentOffset), end);
-            }
-            final S3Object s3Object = SocketAccess.doPrivileged(() -> clientReference.client().getObject(getObjectRequest));
-            this.currentStreamFirstOffset = Math.addExact(start, currentOffset);
-            this.currentStreamLastOffset = Math.addExact(currentStreamFirstOffset, getStreamLength(s3Object));
-            this.currentStream = s3Object.getObjectContent();
-        } catch (final AmazonClientException e) {
-            if (e instanceof AmazonS3Exception amazonS3Exception) {
-                if (404 == amazonS3Exception.getStatusCode()) {
-                    throw addSuppressedExceptions(new NoSuchFileException("Blob object [" + blobKey + "] not found: " + e.getMessage()));
+    private void openStreamWithRetry() throws IOException {
+        while (true) {
+            try (AmazonS3Reference clientReference = blobStore.clientReference()) {
+                final GetObjectRequest getObjectRequest = new GetObjectRequest(blobStore.bucket(), blobKey);
+                getObjectRequest.setRequestMetricCollector(blobStore.getMetricCollector(Operation.GET_OBJECT, purpose));
+                if (currentOffset > 0 || start > 0 || end < Long.MAX_VALUE - 1) {
+                    assert start + currentOffset <= end
+                        : "requesting beyond end, start = " + start + " offset=" + currentOffset + " end=" + end;
+                    getObjectRequest.setRange(Math.addExact(start, currentOffset), end);
+                }
+                final S3Object s3Object = SocketAccess.doPrivileged(() -> clientReference.client().getObject(getObjectRequest));
+                this.currentStreamFirstOffset = Math.addExact(start, currentOffset);
+                this.currentStreamLastOffset = Math.addExact(currentStreamFirstOffset, getStreamLength(s3Object));
+                this.currentStream = s3Object.getObjectContent();
+                return;
+            } catch (AmazonClientException e) {
+                if (e instanceof AmazonS3Exception amazonS3Exception && 404 == amazonS3Exception.getStatusCode()) {
+                    throw addSuppressedExceptions(
+                        new NoSuchFileException("Blob object [" + blobKey + "] not found: " + amazonS3Exception.getMessage())
+                    );
                 }
+
+                final long delayInMillis = maybeLogAndComputeRetryDelay("opening", e);
+                delayBeforeRetry(delayInMillis);
             }
-            throw addSuppressedExceptions(e);
         }
     }
 
@@ -166,45 +171,104 @@ private void ensureOpen() {
     }
 
     private void reopenStreamOrFail(IOException e) throws IOException {
-        if (purpose == OperationPurpose.REPOSITORY_ANALYSIS) {
-            logger.warn(() -> format("""
-                failed reading [%s/%s] at offset [%s]""", blobStore.bucket(), blobKey, start + currentOffset), e);
-            throw e;
+        final long delayInMillis = maybeLogAndComputeRetryDelay("reading", e);
+        maybeAbort(currentStream);
+        IOUtils.closeWhileHandlingException(currentStream);
+
+        delayBeforeRetry(delayInMillis);
+        openStreamWithRetry();
+    }
+
+    // The method throws if the operation should *not* be retried. Otherwise, it keeps a record for the attempt and associated failure
+    // and compute the delay before retry.
+    private <T extends Exception> long maybeLogAndComputeRetryDelay(String action, T e) throws T {
+        if (shouldRetry(attempt) == false) {
+            final var finalException = addSuppressedExceptions(e);
+            logForFailure(action, finalException);
+            throw finalException;
         }
 
-        final int maxAttempts = blobStore.getMaxRetries() + 1;
+        // Log at info level for the 1st retry and every ~5 minutes afterward
+        logForRetry((attempt == 1 || attempt % 30 == 0) ? Level.INFO : Level.DEBUG, action, e);
+        if (failures.size() < MAX_SUPPRESSED_EXCEPTIONS) {
+            failures.add(e);
+        }
+        final long delayInMillis = getRetryDelayInMillis();
+        attempt += 1; // increment after computing delay because attempt affects the result
+        return delayInMillis;
+    }
+
+    private void logForFailure(String action, Exception e) {
+        logger.warn(
+            () -> format(
+                "failed %s [%s/%s] at offset [%s] with purpose [%s]",
+                action,
+                blobStore.bucket(),
+                blobKey,
+                start + currentOffset,
+                purpose.getKey()
+            ),
+            e
+        );
+    }
 
+    private void logForRetry(Level level, String action, Exception e) {
         final long meaningfulProgressSize = Math.max(1L, blobStore.bufferSizeInBytes() / 100L);
         final long currentStreamProgress = Math.subtractExact(Math.addExact(start, currentOffset), currentStreamFirstOffset);
         if (currentStreamProgress >= meaningfulProgressSize) {
             failuresAfterMeaningfulProgress += 1;
         }
-        final Supplier<String> messageSupplier = () -> format(
-            """
-                failed reading [%s/%s] at offset [%s]; this was attempt [%s] to read this blob which yielded [%s] bytes; in total \
-                [%s] of the attempts to read this blob have made meaningful progress and do not count towards the maximum number of \
-                retries; the maximum number of read attempts which do not make meaningful progress is [%s]""",
-            blobStore.bucket(),
-            blobKey,
-            start + currentOffset,
-            attempt,
-            currentStreamProgress,
-            failuresAfterMeaningfulProgress,
-            maxAttempts
+        logger.log(
+            level,
+            () -> format(
+                """
+                    failed %s [%s/%s] at offset [%s] with purpose [%s]; \
+                    this was attempt [%s] to read this blob which yielded [%s] bytes; in total \
+                    [%s] of the attempts to read this blob have made meaningful progress and do not count towards the maximum number of \
+                    retries; the maximum number of read attempts which do not make meaningful progress is [%s]""",
+                action,
+                blobStore.bucket(),
+                blobKey,
+                start + currentOffset,
+                purpose.getKey(),
+                attempt,
+                currentStreamProgress,
+                failuresAfterMeaningfulProgress,
+                maxRetriesForNoMeaningfulProgress()
+            ),
+            e
         );
-        if (attempt >= maxAttempts + failuresAfterMeaningfulProgress) {
-            final var finalException = addSuppressedExceptions(e);
-            logger.warn(messageSupplier, finalException);
-            throw finalException;
+    }
+
+    private boolean shouldRetry(int attempt) {
+        if (purpose == OperationPurpose.REPOSITORY_ANALYSIS) {
+            return false;
         }
-        logger.debug(messageSupplier, e);
-        attempt += 1;
-        if (failures.size() < MAX_SUPPRESSED_EXCEPTIONS) {
-            failures.add(e);
+        if (purpose == OperationPurpose.INDICES) {
+            return true;
         }
-        maybeAbort(currentStream);
-        IOUtils.closeWhileHandlingException(currentStream);
-        openStream();
+        final int maxAttempts = blobStore.getMaxRetries() + 1;
+        return attempt < maxAttempts + failuresAfterMeaningfulProgress;
+    }
+
+    private int maxRetriesForNoMeaningfulProgress() {
+        return purpose == OperationPurpose.INDICES ? Integer.MAX_VALUE : (blobStore.getMaxRetries() + 1);
+    }
+
+    private void delayBeforeRetry(long delayInMillis) {
+        try {
+            assert shouldRetry(attempt - 1) : "should not have retried";
+            Thread.sleep(delayInMillis);
+        } catch (InterruptedException e) {
+            logger.info("s3 input stream delay interrupted", e);
+            Thread.currentThread().interrupt();
+        }
+    }
+
+    // protected access for testing
+    protected long getRetryDelayInMillis() {
+        // Initial delay is 10 ms and cap max delay at 10 * 1024 millis, i.e. it retries every ~10 seconds at a minimum
+        return 10L << (Math.min(attempt - 1, 10));
     }
 
     @Override
@@ -247,7 +311,7 @@ public void reset() {
     }
 
     private <T extends Exception> T addSuppressedExceptions(T e) {
-        for (IOException failure : failures) {
+        for (Exception failure : failures) {
             e.addSuppressed(failure);
         }
         return e;

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3Service.java

@@ -178,6 +178,11 @@ S3ClientSettings settings(RepositoryMetadata repositoryMetadata) {
 
     // proxy for testing
     AmazonS3 buildClient(final S3ClientSettings clientSettings) {
+        final AmazonS3ClientBuilder builder = buildClientBuilder(clientSettings);
+        return SocketAccess.doPrivileged(builder::build);
+    }
+
+    protected AmazonS3ClientBuilder buildClientBuilder(S3ClientSettings clientSettings) {
         final AmazonS3ClientBuilder builder = AmazonS3ClientBuilder.standard();
         builder.withCredentials(buildCredentials(LOGGER, clientSettings, webIdentityTokenCredentialsProvider));
         builder.withClientConfiguration(buildConfiguration(clientSettings));
@@ -206,7 +211,7 @@ AmazonS3 buildClient(final S3ClientSettings clientSettings) {
         if (clientSettings.disableChunkedEncoding) {
             builder.disableChunkedEncoding();
         }
-        return SocketAccess.doPrivileged(builder::build);
+        return builder;
     }
 
     // pkg private for tests