HADOOP-19559: S3A Analytics-Accelerator Add IoStatistics support

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

@@ -132,6 +132,12 @@ public final class StreamStatisticNames {
   public static final String STREAM_READ_OPERATIONS =
       "stream_read_operations";
 
+  /** GET requests made by the analytics stream. */
+  public static final String STREAM_READ_ANALYTICS_GET_REQUESTS = "stream_read_analytics_get_requests";
+
+  /** HEAD requests made by the analytics stream. */
+  public static final String STREAM_READ_ANALYTICS_HEAD_REQUESTS = "stream_read_analytics_head_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,8 @@ private InputStreamStatistics(
               StreamStatisticNames.STREAM_READ_CLOSE_OPERATIONS,
               StreamStatisticNames.STREAM_READ_OPENED,
               StreamStatisticNames.STREAM_READ_BYTES,
+              StreamStatisticNames.STREAM_READ_ANALYTICS_GET_REQUESTS,
+              StreamStatisticNames.STREAM_READ_ANALYTICS_HEAD_REQUESTS,
               StreamStatisticNames.STREAM_READ_EXCEPTIONS,
               StreamStatisticNames.STREAM_READ_FULLY_OPERATIONS,
               StreamStatisticNames.STREAM_READ_OPERATIONS,
@@ -1128,6 +1130,15 @@ public void readVectoredBytesDiscarded(int discarded) {
       bytesDiscardedInVectoredIO.addAndGet(discarded);
     }
 
+    @Override
+    public void incrementAnalyticsGetRequests() {
+      increment(StreamStatisticNames.STREAM_READ_ANALYTICS_GET_REQUESTS);
+    }
+    @Override
+    public void incrementAnalyticsHeadRequests() {
+      increment(StreamStatisticNames.STREAM_READ_ANALYTICS_HEAD_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,14 @@ 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),
+  ANALYTICS_HEAD_REQUESTS(
+          StreamStatisticNames.STREAM_READ_ANALYTICS_HEAD_REQUESTS,
+          "HEAD 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,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.s3a.impl.streams;
+
+import org.apache.hadoop.fs.s3a.statistics.S3AInputStreamStatistics;
+import org.apache.hadoop.fs.statistics.DurationTracker;
+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();
+        // Update ACTION_HTTP_GET_REQUEST statistic
+        DurationTracker tracker = statistics.initiateGetRequest();
+        tracker.close();
+    }
+
+    @Override
+    public void onHeadRequest() {
+        statistics.incrementAnalyticsHeadRequests();
+    }
+}
+

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;
@@ -48,13 +49,17 @@
   private S3SeekableInputStream inputStream;
   private long lastReadCurrentPos = 0;
   private volatile boolean closed;
+  private final long contentLength;
+  private final long lengthLimit;
 
   public static final Logger LOG = LoggerFactory.getLogger(AnalyticsStream.class);
 
   public AnalyticsStream(final ObjectReadParameters parameters,
       final S3SeekableInputStreamFactory s3SeekableInputStreamFactory) throws IOException {
     super(InputStreamType.Analytics, parameters);
     S3ObjectAttributes s3Attributes = parameters.getObjectAttributes();
+    this.contentLength = s3Attributes.getLen();
+    this.lengthLimit = s3Attributes.getLen();
     this.inputStream = s3SeekableInputStreamFactory.createStream(S3URI.of(s3Attributes.getBucket(),
         s3Attributes.getKey()), buildOpenStreamInformation(parameters));
     getS3AStreamStatistics().streamOpened(InputStreamType.Analytics);
@@ -63,13 +68,23 @@
   @Override
   public int read() throws IOException {
     throwIfClosed();
+    if (getPos() >= lengthLimit) {
+      return -1; // EOF reached due to length limit
+    }
+    getS3AStreamStatistics().readOperationStarted(getPos(), 1);
+
     int bytesRead;
     try {
       bytesRead = inputStream.read();
     } catch (IOException ioe) {
       onReadFailure(ioe);
       throw ioe;
     }
+
+    if (bytesRead != -1) {
+      incrementBytesRead(1);
+    }
+
     return bytesRead;
   }
 
@@ -105,26 +120,47 @@
    */
   public int readTail(byte[] buf, int off, int len) throws IOException {
     throwIfClosed();
+    getS3AStreamStatistics().readOperationStarted(getPos(), len);
+
     int bytesRead;
     try {
       bytesRead = inputStream.readTail(buf, off, len);
     } catch (IOException ioe) {
       onReadFailure(ioe);
       throw ioe;
     }
+
+    if (bytesRead > 0) {
+      incrementBytesRead(bytesRead);
+    }
+
     return bytesRead;
   }
 
   @Override
   public int read(byte[] buf, int off, int len) throws IOException {
     throwIfClosed();
+    long pos = getPos();
+    if (pos >= lengthLimit) {
+      return -1; // EOF reached due to length limit
+    }
+
+    // Limit read length to not exceed the length limit
+    int maxRead = (int) Math.min(len, lengthLimit - pos);
+    getS3AStreamStatistics().readOperationStarted(pos, maxRead);
+
     int bytesRead;
     try {
-      bytesRead = inputStream.read(buf, off, len);
+      bytesRead = inputStream.read(buf, off, maxRead);
     } catch (IOException ioe) {
       onReadFailure(ioe);
       throw ioe;
     }
+
+    if (bytesRead > 0) {
+      incrementBytesRead(bytesRead);
+    }
+
     return bytesRead;
   }
 
@@ -194,10 +230,13 @@
   }
 
   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()
@@ -235,4 +274,16 @@
       throw new IOException(getKey() + ": " + FSExceptionMessages.STREAM_IS_CLOSED);
     }
   }
+
+  /**
+   * Increment the bytes read counter if there is a stats instance
+   * and the number of bytes read is more than zero.
+   * @param bytesRead number of bytes read
+   */
+  private void incrementBytesRead(long bytesRead) {
+    getS3AStreamStatistics().bytesRead(bytesRead);
+    if (getContext().getStats() != null && bytesRead > 0) {
+      getContext().getStats().incrementBytesRead(bytesRead);
+    }
+  }
 }

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

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

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

@@ -159,7 +159,12 @@ public void seekForwards(final long skipped,
         final long bytesReadInSeek) {
 
     }
-
+    @Override
+    public void incrementAnalyticsGetRequests() {
+    }
+    @Override
+    public void incrementAnalyticsHeadRequests() {
+    }
     @Override
     public long streamOpened() {
       return 0;