Improve logging for S3RetryingInputStream (#104892)

* Improve logging for S3RetryingInputStream

This PR adds a logging message when opening or read succeeds after
retries. See also
#103300 (comment)

It also makes the logging messages for retries exponentially
less so that the log size is more scalable when there are many threads
retrying. See also
#103300 (comment)

Relates: #103300

Author: ywangd

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

@@ -77,7 +77,9 @@ class S3RetryingInputStream extends InputStream {
         this.failures = new ArrayList<>(MAX_SUPPRESSED_EXCEPTIONS);
         this.start = start;
         this.end = end;
+        final int initialAttempt = attempt;
         openStreamWithRetry();
+        maybeLogForSuccessAfterRetries(initialAttempt, "opened");
     }
 
     private void openStreamWithRetry() throws IOException {
@@ -130,14 +132,16 @@ private long getStreamLength(final S3Object object) {
     @Override
     public int read() throws IOException {
         ensureOpen();
+        final int initialAttempt = attempt;
         while (true) {
             try {
                 final int result = currentStream.read();
                 if (result == -1) {
                     eof = true;
-                    return -1;
+                } else {
+                    currentOffset += 1;
                 }
-                currentOffset += 1;
+                maybeLogForSuccessAfterRetries(initialAttempt, "read");
                 return result;
             } catch (IOException e) {
                 reopenStreamOrFail(e);
@@ -148,14 +152,16 @@ public int read() throws IOException {
     @Override
     public int read(byte[] b, int off, int len) throws IOException {
         ensureOpen();
+        final int initialAttempt = attempt;
         while (true) {
             try {
                 final int bytesRead = currentStream.read(b, off, len);
                 if (bytesRead == -1) {
                     eof = true;
-                    return -1;
+                } else {
+                    currentOffset += bytesRead;
                 }
-                currentOffset += bytesRead;
+                maybeLogForSuccessAfterRetries(initialAttempt, "read");
                 return bytesRead;
             } catch (IOException e) {
                 reopenStreamOrFail(e);
@@ -192,8 +198,8 @@ private <T extends Exception> long maybeLogAndComputeRetryDelay(String action, T
             throw finalException;
         }
 
-        // 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);
+        // Log at info level for the 1st retry and then exponentially less
+        logForRetry(Integer.bitCount(attempt) == 1 ? Level.INFO : Level.DEBUG, action, e);
         if (failures.size() < MAX_SUPPRESSED_EXCEPTIONS) {
             failures.add(e);
         }
@@ -239,6 +245,19 @@ private void logForRetry(Level level, String action, Exception e) {
         );
     }
 
+    private void maybeLogForSuccessAfterRetries(int initialAttempt, String action) {
+        if (attempt > initialAttempt) {
+            logger.info(
+                "successfully {} input stream for [{}/{}] with purpose [{}] after [{}] retries",
+                action,
+                blobStore.bucket(),
+                blobKey,
+                purpose,
+                attempt - initialAttempt
+            );
+        }
+    }
+
     private long currentStreamProgress() {
         return Math.subtractExact(Math.addExact(start, currentOffset), currentStreamFirstOffset);
     }