IoStatistics for AAL

Author: vaibhav5140

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

@@ -132,6 +132,9 @@ public final class StreamStatisticNames {
   public static final String STREAM_READ_OPERATIONS =
       "stream_read_operations";
 
+  /** Analytics GET requests made by stream. */
+  public static final String STREAM_READ_ANALYTICS_GET_REQUESTS = "stream_read_analytics_get_requests";
+
   /**
    * Count of readVectored() operations in an input stream.
    * Value: {@value}.

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

@@ -872,6 +872,7 @@ private InputStreamStatistics(
               StreamStatisticNames.STREAM_READ_CLOSE_OPERATIONS,
               StreamStatisticNames.STREAM_READ_OPENED,
               StreamStatisticNames.STREAM_READ_BYTES,
+              StreamStatisticNames.STREAM_READ_ANALYTICS_GET_REQUESTS,
               StreamStatisticNames.STREAM_READ_EXCEPTIONS,
               StreamStatisticNames.STREAM_READ_FULLY_OPERATIONS,
               StreamStatisticNames.STREAM_READ_OPERATIONS,
@@ -1128,6 +1129,11 @@ public void readVectoredBytesDiscarded(int discarded) {
       bytesDiscardedInVectoredIO.addAndGet(discarded);
     }
 
+    @Override
+    public void incrementAnalyticsGetRequests() {
+      increment(StreamStatisticNames.STREAM_READ_ANALYTICS_GET_REQUESTS);
+    }
+
     @Override
     public void executorAcquired(Duration timeInQueue) {
       // update the duration fields in the IOStatistics.

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

@@ -459,6 +459,11 @@ public enum Statistic {
       "Gauge of active memory in use",
       TYPE_GAUGE),
 
+  ANALYTICS_GET_REQUESTS(
+      StreamStatisticNames.STREAM_READ_ANALYTICS_GET_REQUESTS,
+      "GET requests made by analytics streams",
+      TYPE_COUNTER),
+
   /* Stream Write statistics */
 
   STREAM_WRITE_EXCEPTIONS(

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

@@ -0,0 +1,25 @@
+package org.apache.hadoop.fs.s3a.impl.streams;
+
+import org.apache.hadoop.fs.s3a.statistics.S3AInputStreamStatistics;
+import software.amazon.s3.analyticsaccelerator.util.RequestCallback;
+
+/**
+ * Implementation of AAL's RequestCallback interface that tracks analytics operations.
+ */
+public class AnalyticsRequestCallback implements RequestCallback {
+    private final S3AInputStreamStatistics statistics;
+
+    /**
+     * Create a new callback instance.
+     * @param statistics the statistics to update
+     */
+    public AnalyticsRequestCallback(S3AInputStreamStatistics statistics) {
+        this.statistics = statistics;
+    }
+
+    @Override
+    public void onGetRequest() {
+        statistics.incrementAnalyticsGetRequests();
+    }
+}
+

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

@@ -28,6 +28,7 @@
 import software.amazon.s3.analyticsaccelerator.util.InputPolicy;
 import software.amazon.s3.analyticsaccelerator.util.OpenStreamInformation;
 import software.amazon.s3.analyticsaccelerator.util.S3URI;
+import software.amazon.s3.analyticsaccelerator.util.RequestCallback;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -63,13 +64,21 @@ public AnalyticsStream(final ObjectReadParameters parameters,
   @Override
   public int read() throws IOException {
     throwIfClosed();
+    long pos = getPos();
+    getS3AStreamStatistics().readOperationStarted(pos, 1);
+
     int bytesRead;
     try {
       bytesRead = inputStream.read();
     } catch (IOException ioe) {
       onReadFailure(ioe);
       throw ioe;
     }
+
+    if (bytesRead != -1) {
+      getS3AStreamStatistics().bytesRead(1);
+    }
+
     return bytesRead;
   }
 
@@ -105,26 +114,42 @@ public synchronized long getPos() {
    */
   public int readTail(byte[] buf, int off, int len) throws IOException {
     throwIfClosed();
+    long pos = getPos();
+    getS3AStreamStatistics().readOperationStarted(pos, len);
+
     int bytesRead;
     try {
       bytesRead = inputStream.readTail(buf, off, len);
     } catch (IOException ioe) {
       onReadFailure(ioe);
       throw ioe;
     }
+
+    if (bytesRead > 0) {
+      getS3AStreamStatistics().bytesRead(bytesRead);
+    }
+
     return bytesRead;
   }
 
   @Override
   public int read(byte[] buf, int off, int len) throws IOException {
     throwIfClosed();
+    long pos = getPos();
+    getS3AStreamStatistics().readOperationStarted(pos, len);
+
     int bytesRead;
     try {
       bytesRead = inputStream.read(buf, off, len);
     } catch (IOException ioe) {
       onReadFailure(ioe);
       throw ioe;
     }
+
+    if (bytesRead > 0) {
+      getS3AStreamStatistics().bytesRead(bytesRead);
+    }
+
     return bytesRead;
   }
 
@@ -194,10 +219,13 @@ private void onReadFailure(IOException ioe) throws IOException {
   }
 
   private OpenStreamInformation buildOpenStreamInformation(ObjectReadParameters parameters) {
+
+    final RequestCallback requestCallback = new AnalyticsRequestCallback(getS3AStreamStatistics());
+
     OpenStreamInformation.OpenStreamInformationBuilder openStreamInformationBuilder =
         OpenStreamInformation.builder()
             .inputPolicy(mapS3AInputPolicyToAAL(parameters.getContext()
-            .getInputPolicy()));
+            .getInputPolicy())).requestCallback(requestCallback);
 
     if (parameters.getObjectAttributes().getETag() != null) {
       openStreamInformationBuilder.objectMetadata(ObjectMetadata.builder()

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

@@ -190,6 +190,10 @@ void readVectoredOperationStarted(int numIncomingRanges,
   long getVersionMismatches();
 
   long getInputPolicy();
+  /**
+   * Increment the counter for GET requests made by Analytics Accelerator Library.
+   */
+  void incrementAnalyticsGetRequests();
 
   /**
    * Get the value of a counter.

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

@@ -159,6 +159,9 @@ public void seekForwards(final long skipped,
         final long bytesReadInSeek) {
 
     }
+    @Override
+    public void incrementAnalyticsGetRequests() {
+    }
 
     @Override
     public long streamOpened() {

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

@@ -23,8 +23,8 @@
 import java.io.IOException;
 import java.io.InputStream;
 
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 import org.assertj.core.api.Assertions;
 
 import org.apache.hadoop.conf.Configuration;
@@ -35,6 +35,7 @@
 import org.apache.hadoop.fs.s3a.impl.streams.InputStreamType;
 import org.apache.hadoop.fs.s3a.impl.streams.ObjectInputStream;
 import org.apache.hadoop.fs.statistics.IOStatistics;
+import org.apache.hadoop.fs.statistics.StreamStatisticNames;
 
 import software.amazon.s3.analyticsaccelerator.S3SeekableInputStreamConfiguration;
 import software.amazon.s3.analyticsaccelerator.common.ConnectorConfiguration;
@@ -47,6 +48,8 @@
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
 import static org.apache.hadoop.fs.s3a.test.PublicDatasetTestUtils.getExternalData;
 import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.verifyStatisticCounterValue;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_BYTES;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_OPERATIONS;
 import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_ANALYTICS_OPENED;
 import static org.apache.hadoop.fs.statistics.StreamStatisticNames.ANALYTICS_STREAM_FACTORY_CLOSED;
 import static org.apache.hadoop.test.LambdaTestUtils.intercept;
@@ -67,7 +70,7 @@ public class ITestS3AAnalyticsAcceleratorStreamReading extends AbstractS3ATestBa
 
   private Path externalTestFile;
 
-  @Before
+  @BeforeEach
   public void setUp() throws Exception {
     super.setup();
     skipIfClientSideEncryption();
@@ -96,14 +99,20 @@ public void testConnectorFrameWorkIntegration() throws Throwable {
             .build().get()) {
       ioStats = inputStream.getIOStatistics();
       inputStream.seek(5);
-      inputStream.read(buffer, 0, 500);
+      int bytesRead = inputStream.read(buffer, 0, 500);
 
       final InputStream wrappedStream = inputStream.getWrappedStream();
       ObjectInputStream objectInputStream = (ObjectInputStream) wrappedStream;
 
       Assertions.assertThat(objectInputStream.streamType()).isEqualTo(InputStreamType.Analytics);
       Assertions.assertThat(objectInputStream.getInputPolicy())
           .isEqualTo(S3AInputPolicy.Sequential);
+
+      verifyStatisticCounterValue(ioStats, STREAM_READ_BYTES, bytesRead);
+      verifyStatisticCounterValue(ioStats, STREAM_READ_OPERATIONS, 1);
+
+      long streamBytesRead = objectInputStream.getS3AStreamStatistics().getBytesRead();
+      Assertions.assertThat(streamBytesRead).as("Stream statistics should track bytes read").isEqualTo(bytesRead);
     }
 
     verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_OPENED, 1);
@@ -129,11 +138,17 @@ public void testMalformedParquetFooter() throws IOException {
 
     byte[] buffer = new byte[500];
     IOStatistics ioStats;
+    int bytesRead;
 
     try (FSDataInputStream inputStream = getFileSystem().open(dest)) {
       ioStats = inputStream.getIOStatistics();
       inputStream.seek(5);
-      inputStream.read(buffer, 0, 500);
+      bytesRead = inputStream.read(buffer, 0, 500);
+
+      ObjectInputStream objectInputStream = (ObjectInputStream) inputStream.getWrappedStream();
+      long streamBytesRead = objectInputStream.getS3AStreamStatistics().getBytesRead();
+      Assertions.assertThat(streamBytesRead).as("Stream statistics should track bytes read").isEqualTo(bytesRead);
+
     }
 
     verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_OPENED, 1);
@@ -145,9 +160,9 @@ public void testMalformedParquetFooter() throws IOException {
    * can contain multiple row groups, this allows for further parallelisation, as each row group
    * can be processed independently.
    */
-  @Test
-  public void testMultiRowGroupParquet() throws Throwable {
-    describe("A parquet file is read successfully");
+@Test
+public void testMultiRowGroupParquet() throws Throwable {
+  describe("A parquet file is read successfully");
 
     Path dest = path("multi_row_group.parquet");
 
@@ -166,12 +181,16 @@ public void testMultiRowGroupParquet() throws Throwable {
     }
 
     verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_OPENED, 1);
-
+    verifyStatisticCounterValue(ioStats, StreamStatisticNames.STREAM_READ_ANALYTICS_GET_REQUESTS, 1);
     try (FSDataInputStream inputStream = getFileSystem().openFile(dest)
         .must(FS_OPTION_OPENFILE_READ_POLICY, FS_OPTION_OPENFILE_READ_POLICY_PARQUET)
         .build().get()) {
       ioStats = inputStream.getIOStatistics();
       inputStream.readFully(buffer, 0, (int) fileStatus.getLen());
+
+      verifyStatisticCounterValue(ioStats, STREAM_READ_BYTES, (int) fileStatus.getLen());
+      verifyStatisticCounterValue(ioStats, STREAM_READ_OPERATIONS, 1);
+
     }
 
     verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_OPENED, 1);

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

@@ -27,7 +27,6 @@
 import org.apache.hadoop.fs.contract.s3a.S3AContract;
 import org.apache.hadoop.fs.statistics.StreamStatisticNames;
 
-import static org.apache.hadoop.fs.s3a.S3ATestUtils.skipIfAnalyticsAcceleratorEnabled;
 import static org.apache.hadoop.fs.statistics.StreamStatisticNames.*;
 
 /**
@@ -78,13 +77,4 @@ public List<String> outputStreamStatisticKeys() {
         STREAM_WRITE_BLOCK_UPLOADS,
         STREAM_WRITE_EXCEPTIONS);
   }
-
-  @Override
-  public void testInputStreamStatisticRead() throws Throwable {
-    // Analytics accelerator currently does not support IOStatistics, this will be added as
-    // part of https://issues.apache.org/jira/browse/HADOOP-19364
-    skipIfAnalyticsAcceleratorEnabled(getContract().getConf(),
-        "Analytics Accelerator currently does not support stream statistics");
-    super.testInputStreamStatisticRead();
-  }
 }