adds in a cache_hit stat

Author: ahmarsuhail

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

@@ -499,6 +499,12 @@ public final class StreamStatisticNames {
    */
   public static final String STREAM_READ_PARQUET_FOOTER_PARSING_FAILED = "stream_read_parquet_footer_parsing_failed";
 
+  /**
+   * A cache hit occurs when the request range can be satisfied by the data in the cache.
+   */
+  public static final String STREAM_READ_CACHE_HIT = "stream_read_cache_hit";
+
+
   private StreamStatisticNames() {
   }
 

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

@@ -83,6 +83,7 @@
 import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_UNBUFFERED;
 import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_PREFETCHED_BYTES;
 import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_PARQUET_FOOTER_PARSING_FAILED;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_CACHE_HIT;
 import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.iostatisticsStore;
 import static org.apache.hadoop.fs.s3a.Statistic.*;
 
@@ -895,8 +896,9 @@ private InputStreamStatistics(
               StreamStatisticNames.STREAM_EVICT_BLOCKS_FROM_FILE_CACHE,
               StoreStatisticNames.ACTION_HTTP_HEAD_REQUEST,
               StreamStatisticNames.STREAM_READ_ANALYTICS_OPENED,
-              STREAM_READ_PREFETCHED_BYTES,
-              STREAM_READ_PARQUET_FOOTER_PARSING_FAILED
+              StreamStatisticNames.STREAM_READ_CACHE_HIT,
+              StreamStatisticNames.STREAM_READ_PREFETCHED_BYTES,
+              StreamStatisticNames.STREAM_READ_PARQUET_FOOTER_PARSING_FAILED
                   )
           .withGauges(STREAM_READ_GAUGE_INPUT_POLICY,
               STREAM_READ_BLOCKS_IN_FILE_CACHE.getSymbol(),
@@ -1154,6 +1156,11 @@ public void footerParsingFailed() {
       increment(STREAM_READ_PARQUET_FOOTER_PARSING_FAILED);
     }
 
+    @Override
+    public void streamReadCacheHit() {
+      increment(STREAM_READ_CACHE_HIT);
+    }
+
 
     @Override
     public void executorAcquired(Duration timeInQueue) {

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

@@ -458,6 +458,18 @@ public enum Statistic {
       StreamStatisticNames.STREAM_READ_ACTIVE_MEMORY_IN_USE,
       "Gauge of active memory in use",
       TYPE_GAUGE),
+  STREAM_READ_PREFETCH_BYTES(
+          StreamStatisticNames.STREAM_READ_PREFETCHED_BYTES,
+          "Bytes prefetched by AAL stream",
+          TYPE_COUNTER),
+  STREAM_READ_PARQUET_FOOTER_PARSING_FAILED(
+          StreamStatisticNames.STREAM_READ_PARQUET_FOOTER_PARSING_FAILED,
+          "Count of Parquet footer parsing failures encountered by AAL",
+          TYPE_COUNTER),
+  STREAM_READ_CACHE_HIT(
+          StreamStatisticNames.STREAM_READ_CACHE_HIT,
+          "Count of cache hits in AAL stream",
+          TYPE_COUNTER),
 
 
   /* Stream Write statistics */

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/streams/AnalyticsRequestCallback.java

@@ -60,6 +60,10 @@ public void onReadVectored(int numIncomingRanges, int numCombinedRanges) {
     statistics.readVectoredOperationStarted(numIncomingRanges, numCombinedRanges);
   }
 
+  @Override
+  public void onCacheHit() {
+    statistics.streamReadCacheHit();
+  }
 
 }
 

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

@@ -140,6 +140,11 @@ void readVectoredOperationStarted(int numIncomingRanges,
    */
   void footerParsingFailed();
 
+  /**
+   * If the request data is already in the data cache.
+   */
+  void streamReadCacheHit();
+
   @Override
   void close();
 

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

@@ -232,6 +232,11 @@ public void footerParsingFailed() {
 
     }
 
+    @Override
+    public void streamReadCacheHit() {
+
+    }
+
     @Override
     public void close() {
 

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AAnalyticsAcceleratorStreamReading.java

@@ -139,6 +139,15 @@ public void testConnectorFrameWorkIntegration() throws Throwable {
 
   @Test
   public void testSequentialPrefetching() throws IOException {
+
+    Configuration conf = getConfiguration();
+
+    // AAL uses a caffeine cache, and expires any prefetched data for a key 1s after it was last accessed by default.
+    // While this works well when running on EC2, for local testing, it can take more than 1s to download large chunks
+    // of data. Set this value to higher for testing to prevent early cache evictions.
+    conf.setInt(ANALYTICS_ACCELERATOR_CONFIGURATION_PREFIX +
+            "."  + "physicalio.cache.timeout", 10000);
+
     S3AFileSystem fs =
             (S3AFileSystem) FileSystem.get(externalTestFile.toUri(), getConfiguration());
     byte[] buffer = new byte[10 * ONE_MB];
@@ -162,9 +171,14 @@ public void testSequentialPrefetching() throws IOException {
       inputStream.readFully(buffer, 0, 2 *  ONE_MB);
       inputStream.readFully(buffer, 0, ONE_MB);
       verifyStatisticCounterValue(ioStats, STREAM_READ_PREFETCHED_BYTES, 3 * ONE_MB);
+      // Two cache hits, as the previous two reads were already prefetched.
+      verifyStatisticCounterValue(ioStats, STREAM_READ_CACHE_HIT, 2);
 
       // Another sequential read, GP will now prefetch the next 8MB of data.
       inputStream.readFully(buffer, 0, ONE_MB);
+      // Cache hit is still 2, as the previous read required a new GET request as it was outside the previously fetched
+      // 4MB.
+      verifyStatisticCounterValue(ioStats, STREAM_READ_CACHE_HIT, 2);
       // A total of 10MB is prefetched - 3MB and then 7MB.
       verifyStatisticCounterValue(ioStats, STREAM_READ_PREFETCHED_BYTES, 10 * ONE_MB);
 
@@ -173,7 +187,14 @@ public void testSequentialPrefetching() throws IOException {
       // Though the next GP should prefetch 16MB, since the file is ~23MB, only the bytes till EoF are prefetched:
       // 6291456 remaining bytes.
       verifyStatisticCounterValue(ioStats, STREAM_READ_PREFETCHED_BYTES, 10 * ONE_MB + 6291456);
+      inputStream.readFully(buffer, 0, 3 * ONE_MB);
+      verifyStatisticCounterValue(ioStats, STREAM_READ_CACHE_HIT, 3);
     }
+
+    // verify all AAL stats are passed to the FS.
+    verifyStatisticCounterValue(fs.getIOStatistics(), STREAM_READ_CACHE_HIT, 3);
+    verifyStatisticCounterValue(fs.getIOStatistics(), STREAM_READ_PARQUET_FOOTER_PARSING_FAILED, 0);
+    verifyStatisticCounterValue(fs.getIOStatistics(), STREAM_READ_PREFETCHED_BYTES, 10 * ONE_MB + 6291456);
   }
 
   @Test
@@ -234,6 +255,9 @@ public void testMalformedParquetFooter() throws IOException {
 
     // The footer is only prefetched once, but parsing is attempted on each stream open.
     verifyStatisticCounterValue(ioStats, STREAM_READ_PARQUET_FOOTER_PARSING_FAILED, 1);
+
+    // stat is passed up to the FS.
+    verifyStatisticCounterValue(getFileSystem().getIOStatistics(), STREAM_READ_PARQUET_FOOTER_PARSING_FAILED, 2);
   }
 
   /**
@@ -332,6 +356,9 @@ public void testRandomSeekPatternGets() throws Throwable {
       verifyStatisticCounterValue(ioStats, ACTION_HTTP_GET_REQUEST, 1);
       verifyStatisticCounterValue(ioStats, ACTION_HTTP_HEAD_REQUEST, 0);
     }
+
+    // We did 3 reads, and all of them were served from the cache
+    verifyStatisticCounterValue(getFileSystem().getIOStatistics(), STREAM_READ_CACHE_HIT, 3);
   }
 
 
@@ -364,5 +391,12 @@ public void testSequentialStreamsNoDuplicateGets() throws Throwable {
       // The second stream will not prefetch any bytes, as they have already been prefetched by stream 1.
       verifyStatisticCounterValue(stats2, STREAM_READ_PREFETCHED_BYTES, 0);
     }
+
+    // verify value is passed up to the FS
+    verifyStatisticCounterValue(getFileSystem().getIOStatistics(), STREAM_READ_PREFETCHED_BYTES, 1048575);
+
+    // We did 3 reads, all of them were served from the small object cache. In this case, the whole object was
+    // downloaded as soon as the stream to it was opened.
+    verifyStatisticCounterValue(getFileSystem().getIOStatistics(), STREAM_READ_CACHE_HIT, 3);
   }
 }