addresses review comments, adds in duration tracker for prefetch ops

Author: ahmarsuhail

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/StreamStatisticNames.java

@@ -393,12 +393,6 @@ public final class StreamStatisticNames {
   public static final String STREAM_READ_PREFETCH_OPERATIONS
       = "stream_read_prefetch_operations";
 
-  /**
-   * Total number of failed prefetching operations.
-   */
-  public static final String STREAM_READ_FAILED_PREFETCH_OPERATIONS
-      = "stream_read_failed_prefetch_operations";
-
   /**
    * Total number of block in disk cache.
    */

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/common/CachingBlockManager.java

@@ -33,6 +33,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hadoop.fs.statistics.DurationTracker;
+
 import static java.util.Objects.requireNonNull;
 
 import static org.apache.hadoop.io.IOUtils.cleanupWithLogger;
@@ -307,6 +309,7 @@ private void readBlock(BufferData data, boolean isPrefetch, BufferData.State...
     }
 
     BlockOperations.Operation op = null;
+    DurationTracker tracker = null;
 
     synchronized (data) {
       try {
@@ -328,7 +331,7 @@ private void readBlock(BufferData data, boolean isPrefetch, BufferData.State...
         }
 
         if (isPrefetch) {
-          prefetchingStatistics.prefetchOperationStarted();
+          tracker = prefetchingStatistics.prefetchOperationStarted();
           op = this.ops.prefetch(data.getBlockNumber());
         } else {
           op = this.ops.getRead(data.getBlockNumber());
@@ -341,20 +344,26 @@ private void readBlock(BufferData data, boolean isPrefetch, BufferData.State...
         this.read(buffer, offset, size);
         buffer.flip();
         data.setReady(expectedState);
-
-        if (isPrefetch) {
-          prefetchingStatistics.prefetchOperationCompleted();
-        }
       } catch (Exception e) {
         String message = String.format("error during readBlock(%s)", data.getBlockNumber());
         LOG.error(message, e);
+
+        if(isPrefetch) {
+          tracker.failed();
+        }
+
         this.numReadErrors.incrementAndGet();
         data.setDone();
         throw e;
       } finally {
         if (op != null) {
           this.ops.end(op);
         }
+
+        if (isPrefetch) {
+          prefetchingStatistics.prefetchOperationCompleted();
+          tracker.close();
+        }
       }
     }
   }

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/common/PrefetchingStatistics.java

@@ -21,14 +21,16 @@
 
 import java.time.Duration;
 
+import org.apache.hadoop.fs.statistics.DurationTracker;
 import org.apache.hadoop.fs.statistics.IOStatisticsSource;
 
 public interface PrefetchingStatistics extends IOStatisticsSource {
 
   /**
    * A prefetch operation has started.
+   * @return duration tracker
    */
-  void prefetchOperationStarted();
+  DurationTracker prefetchOperationStarted();
 
   /**
    * A block has been saved to the file cache.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java

@@ -803,8 +803,6 @@ private final class InputStreamStatistics
     private final AtomicLong readOperations;
     private final AtomicLong readFullyOperations;
     private final AtomicLong seekOperations;
-    private final AtomicLong prefetchReadOperations;
-    private final AtomicLong failedPrefetchReadOperations;
 
     /** Bytes read by the application and any when draining streams . */
     private final AtomicLong totalBytesRead;
@@ -838,9 +836,7 @@ private InputStreamStatistics(
               StreamStatisticNames.STREAM_READ_SEEK_BYTES_SKIPPED,
               StreamStatisticNames.STREAM_READ_TOTAL_BYTES,
               StreamStatisticNames.STREAM_READ_UNBUFFERED,
-              StreamStatisticNames.STREAM_READ_VERSION_MISMATCHES,
-              StreamStatisticNames.STREAM_READ_PREFETCH_OPERATIONS,
-              StreamStatisticNames.STREAM_READ_FAILED_PREFETCH_OPERATIONS)
+              StreamStatisticNames.STREAM_READ_VERSION_MISMATCHES)
           .withGauges(STREAM_READ_GAUGE_INPUT_POLICY,
               STREAM_READ_BLOCKS_IN_FILE_CACHE.getSymbol(),
               STREAM_READ_ACTIVE_PREFETCH_OPERATIONS.getSymbol(),
@@ -851,6 +847,7 @@ private InputStreamStatistics(
               StoreStatisticNames.ACTION_FILE_OPENED,
               StreamStatisticNames.STREAM_READ_REMOTE_STREAM_ABORTED,
               StreamStatisticNames.STREAM_READ_REMOTE_STREAM_DRAINED,
+              StreamStatisticNames.STREAM_READ_PREFETCH_OPERATIONS,
               StreamStatisticNames.STREAM_READ_REMOTE_BLOCK_READ,
               StreamStatisticNames.STREAM_READ_BLOCK_ACQUIRE_AND_READ)
           .build();
@@ -889,10 +886,6 @@ private InputStreamStatistics(
           StreamStatisticNames.STREAM_READ_SEEK_OPERATIONS);
       totalBytesRead = st.getCounterReference(
           StreamStatisticNames.STREAM_READ_TOTAL_BYTES);
-      prefetchReadOperations =
-          st.getCounterReference(StreamStatisticNames.STREAM_READ_PREFETCH_OPERATIONS);
-      failedPrefetchReadOperations =
-          st.getCounterReference(StreamStatisticNames.STREAM_READ_FAILED_PREFETCH_OPERATIONS);
       setIOStatistics(st);
       // create initial snapshot of merged statistics
       mergedStats = snapshotIOStatistics(st);
@@ -1316,9 +1309,9 @@ public DurationTracker initiateInnerStreamClose(final boolean abort) {
     }
 
     @Override
-    public void prefetchOperationStarted() {
+    public DurationTracker prefetchOperationStarted() {
       incAllGauges(STREAM_READ_ACTIVE_PREFETCH_OPERATIONS, 1);
-      prefetchReadOperations.incrementAndGet();
+      return trackDuration(StreamStatisticNames.STREAM_READ_PREFETCH_OPERATIONS);
     }
 
     @Override

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/statistics/impl/EmptyS3AStatisticsContext.java

@@ -209,8 +209,8 @@ public void unbuffered() {
     }
 
     @Override
-    public void prefetchOperationStarted() {
-
+    public DurationTracker prefetchOperationStarted() {
+      return stubDurationTracker();
     }
 
     @Override

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/common/EmptyPrefetchingStatistics.java

@@ -21,10 +21,25 @@
 
 import java.time.Duration;
 
-public final class EmptyPrefetchingStatistics implements PrefetchingStatistics {
-  @Override
-  public void prefetchOperationStarted() {
+import org.apache.hadoop.fs.statistics.DurationTracker;
+
+import static org.apache.hadoop.fs.statistics.IOStatisticsSupport.stubDurationTracker;
+
+ public final class EmptyPrefetchingStatistics implements PrefetchingStatistics {
+
+   private static final EmptyPrefetchingStatistics emptyPrefetchingStatistics =
+       new EmptyPrefetchingStatistics();
 
+   private EmptyPrefetchingStatistics() {
+   }
+
+   public static EmptyPrefetchingStatistics getInstance() {
+     return emptyPrefetchingStatistics;
+   }
+
+  @Override
+  public DurationTracker prefetchOperationStarted() {
+    return stubDurationTracker();
   }
 
   @Override

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/common/TestBlockCache.java

@@ -40,7 +40,7 @@ public class TestBlockCache extends AbstractHadoopTestBase {
   public void testArgChecks() throws Exception {
     // Should not throw.
     BlockCache cache =
-        new SingleFilePerBlockCache(new EmptyPrefetchingStatistics());
+        new SingleFilePerBlockCache(EmptyPrefetchingStatistics.getInstance());
 
     ByteBuffer buffer = ByteBuffer.allocate(16);
 

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/common/TestBufferPool.java

@@ -21,6 +21,7 @@
 
 import org.junit.Test;
 
+import org.apache.hadoop.fs.s3a.statistics.S3AInputStreamStatistics;
 import org.apache.hadoop.fs.s3a.statistics.impl.EmptyS3AStatisticsContext;
 import org.apache.hadoop.test.AbstractHadoopTestBase;
 
@@ -37,30 +38,32 @@ public class TestBufferPool extends AbstractHadoopTestBase {
 
   @Test
   public void testArgChecks() throws Exception {
+    S3AInputStreamStatistics s3AInputStreamStatistics =
+        new EmptyS3AStatisticsContext().newInputStreamStatistics();
+
     // Should not throw.
-    BufferPool pool = new BufferPool(POOL_SIZE, BUFFER_SIZE,
-        EmptyS3AStatisticsContext.EMPTY_INPUT_STREAM_STATISTICS);
+    BufferPool pool = new BufferPool(POOL_SIZE, BUFFER_SIZE, s3AInputStreamStatistics);
 
     // Verify it throws correctly.
     ExceptionAsserts.assertThrows(
         IllegalArgumentException.class,
         "'size' must be a positive integer",
-        () -> new BufferPool(0, 10, EmptyS3AStatisticsContext.EMPTY_INPUT_STREAM_STATISTICS));
+        () -> new BufferPool(0, 10, s3AInputStreamStatistics));
 
     ExceptionAsserts.assertThrows(
         IllegalArgumentException.class,
         "'size' must be a positive integer",
-        () -> new BufferPool(-1, 10, EmptyS3AStatisticsContext.EMPTY_INPUT_STREAM_STATISTICS));
+        () -> new BufferPool(-1, 10, s3AInputStreamStatistics));
 
     ExceptionAsserts.assertThrows(
         IllegalArgumentException.class,
         "'bufferSize' must be a positive integer",
-        () -> new BufferPool(10, 0, EmptyS3AStatisticsContext.EMPTY_INPUT_STREAM_STATISTICS));
+        () -> new BufferPool(10, 0, s3AInputStreamStatistics));
 
     ExceptionAsserts.assertThrows(
         IllegalArgumentException.class,
         "'bufferSize' must be a positive integer",
-        () -> new BufferPool(1, -10, EmptyS3AStatisticsContext.EMPTY_INPUT_STREAM_STATISTICS));
+        () -> new BufferPool(1, -10, s3AInputStreamStatistics));
 
     ExceptionAsserts.assertThrows(
         NullPointerException.class,