HADOOP-19311: [ABFS] Implement Backoff and Read Footer metrics using IOStatistics Class

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/enums/AbfsReadFooterMetricsEnum.java

@@ -26,12 +26,12 @@
  * Enum representing various ABFS read footer metrics.
  */
 public enum AbfsReadFooterMetricsEnum {
-    TOTAL_FILES("totalFiles", "Total files in a file system", FILE,  TYPE_COUNTER),
+    TOTAL_FILES("totalFiles", "Total files read", FILE,  TYPE_COUNTER),
     FILE_LENGTH("fileLength", "File length", FILE, TYPE_GAUGE),
     SIZE_READ_BY_FIRST_READ("sizeReadByFirstRead", "Size read by first read", FILE, TYPE_GAUGE),
     OFFSET_DIFF_BETWEEN_FIRST_AND_SECOND_READ("offsetDiffBetweenFirstAndSecondRead", "Offset difference between first and second read", FILE, TYPE_GAUGE),
     READ_LEN_REQUESTED("readLenRequested", "Read length requested", FILE, TYPE_GAUGE),
-    READ_COUNT("readCount", "Read count", FILE, TYPE_COUNTER),
+    READ_COUNT("readCount", "Number of total reads", FILE, TYPE_COUNTER),
     FIRST_OFFSET_DIFF("firstOffsetDiff", "First offset difference", FILE, TYPE_GAUGE),
     SECOND_OFFSET_DIFF("secondOffsetDiff", "Second offset difference", FILE, TYPE_GAUGE);
 

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsBackoffMetrics.java

@@ -29,6 +29,7 @@
 import org.apache.hadoop.fs.azurebfs.statistics.AbstractAbfsStatisticsSource;
 import org.apache.hadoop.fs.statistics.impl.IOStatisticsStore;
 
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.EMPTY_STRING;
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.HUNDRED;
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.THOUSAND;
 import static org.apache.hadoop.fs.azurebfs.constants.MetricsConstants.RETRY;
@@ -55,6 +56,7 @@
 import static org.apache.hadoop.fs.azurebfs.enums.RetryValue.TWENTY_FIVE_AND_ABOVE;
 import static org.apache.hadoop.fs.azurebfs.enums.StatisticTypeEnum.TYPE_COUNTER;
 import static org.apache.hadoop.fs.azurebfs.enums.StatisticTypeEnum.TYPE_GAUGE;
+import static org.apache.hadoop.fs.azurebfs.utils.StringUtils.formatWithPrecision;
 import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.iostatisticsStore;
 
 /**
@@ -202,51 +204,68 @@ public void setMetricValue(AbfsBackoffMetricsEnum metric, long value) {
   1.RCTSI :- Request count that succeeded in x retries
   2.MMA :- Min Max Average (This refers to the backoff or sleep time between 2 requests)
   3.s :- seconds
-  4.BWT :- Number of Bandwidth throttled requests
-  5.IT :- Number of IOPS throttled requests
-  6.OT :- Number of Other throttled requests
-  7.NFR :- Number of requests which failed due to network errors
-  8.%RT :- Percentage of requests that are throttled
-  9.TRNR :- Total number of requests which succeeded without retrying
-  10.TRF :- Total number of requests which failed
-  11.TR :- Total number of requests which were made
-  12.MRC :- Max retry count across all requests
    */
-  @Override
-  public String toString() {
-    if (getMetricValue(TOTAL_NUMBER_OF_REQUESTS) == 0) {
-      return "";
+  private void getRetryMetrics(StringBuilder metricBuilder) {
+    for (RetryValue retryCount : RETRY_LIST) {
+      long totalRequests = getMetricValue(TOTAL_REQUESTS, retryCount);
+      metricBuilder.append("$RCTSI$_").append(retryCount.getValue())
+              .append("R=").append(getMetricValue(NUMBER_OF_REQUESTS_SUCCEEDED, retryCount));
+
+      if (totalRequests > 0) {
+        metricBuilder.append("$MMA$_").append(retryCount.getValue())
+                .append("R=").append(formatWithPrecision((double) getMetricValue(MIN_BACK_OFF, retryCount) / THOUSAND)).append("s")
+                .append(formatWithPrecision((double) getMetricValue(MAX_BACK_OFF, retryCount) / THOUSAND)).append("s")
+                .append(formatWithPrecision((double) getMetricValue(TOTAL_BACK_OFF, retryCount) / totalRequests / THOUSAND)).append("s");
+      } else {
+        metricBuilder.append("$MMA$_").append(retryCount.getValue()).append("R=0s");
+      }
     }
-    StringBuilder metricString = new StringBuilder();
+  }
+
+  /*
+  Acronyms :-
+  1.BWT :- Number of Bandwidth throttled requests
+  2.IT :- Number of IOPS throttled requests
+  3.OT :- Number of Other throttled requests
+  4.NFR :- Number of requests which failed due to network errors
+  5.%RT :- Percentage of requests that are throttled
+  6.TRNR :- Total number of requests which succeeded without retrying
+  7.TRF :- Total number of requests which failed
+  8.TR :- Total number of requests which were made
+  9.MRC :- Max retry count across all requests
+   */
+  private void getMmaMetrics(StringBuilder metricBuilder) {
     long totalRequestsThrottled = getMetricValue(NUMBER_OF_NETWORK_FAILED_REQUESTS)
             + getMetricValue(NUMBER_OF_IOPS_THROTTLED_REQUESTS)
             + getMetricValue(NUMBER_OF_OTHER_THROTTLED_REQUESTS)
             + getMetricValue(NUMBER_OF_BANDWIDTH_THROTTLED_REQUESTS);
     double percentageOfRequestsThrottled = ((double) totalRequestsThrottled / getMetricValue(TOTAL_NUMBER_OF_REQUESTS)) * HUNDRED;
 
-    for (RetryValue retryCount : RETRY_LIST) {
-      long totalRequests = getMetricValue(TOTAL_REQUESTS, retryCount);
-      metricString.append("$RCTSI$_").append(retryCount.getValue()).append("R=").append(getMetricValue(NUMBER_OF_REQUESTS_SUCCEEDED, retryCount));
-      if (totalRequests > 0) {
-        metricString.append("$MMA$_").append(retryCount.getValue()).append("R=")
-                .append(String.format("%.3f", (double) getMetricValue(MIN_BACK_OFF, retryCount) / THOUSAND)).append("s")
-                .append(String.format("%.3f", (double) getMetricValue(MAX_BACK_OFF, retryCount) / THOUSAND)).append("s")
-                .append(String.format("%.3f", (double) getMetricValue(TOTAL_BACK_OFF, retryCount) / totalRequests / THOUSAND)).append("s");
-      } else {
-        metricString.append("$MMA$_").append(retryCount.getValue()).append("R=0s");
-      }
-    }
-    metricString.append("$BWT=").append(getMetricValue(NUMBER_OF_BANDWIDTH_THROTTLED_REQUESTS))
+    metricBuilder.append("$BWT=").append(getMetricValue(NUMBER_OF_BANDWIDTH_THROTTLED_REQUESTS))
             .append("$IT=").append(getMetricValue(NUMBER_OF_IOPS_THROTTLED_REQUESTS))
             .append("$OT=").append(getMetricValue(NUMBER_OF_OTHER_THROTTLED_REQUESTS))
-            .append("$RT=").append(String.format("%.3f", percentageOfRequestsThrottled))
+            .append("$RT=").append(formatWithPrecision(percentageOfRequestsThrottled))
             .append("$NFR=").append(getMetricValue(NUMBER_OF_NETWORK_FAILED_REQUESTS))
             .append("$TRNR=").append(getMetricValue(NUMBER_OF_REQUESTS_SUCCEEDED_WITHOUT_RETRYING))
             .append("$TRF=").append(getMetricValue(NUMBER_OF_REQUESTS_FAILED))
             .append("$TR=").append(getMetricValue(TOTAL_NUMBER_OF_REQUESTS))
             .append("$MRC=").append(getMetricValue(MAX_RETRY_COUNT));
+  }
 
-    return metricString.toString();
+  /**
+   * Retrieves the string representation of the metrics.
+   *
+   * @return the string representation of the metrics
+   */
+  @Override
+  public String toString() {
+    if (getMetricValue(TOTAL_NUMBER_OF_REQUESTS) == 0) {
+      return EMPTY_STRING;
+    }
+    StringBuilder metricBuilder = new StringBuilder();
+    getRetryMetrics(metricBuilder);
+    getMmaMetrics(metricBuilder);
+    return metricBuilder.toString();
   }
 
   @VisibleForTesting

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsReadFooterMetrics.java

@@ -46,6 +46,7 @@
 import static org.apache.hadoop.fs.azurebfs.enums.FileType.NON_PARQUET;
 import static org.apache.hadoop.fs.azurebfs.enums.StatisticTypeEnum.TYPE_COUNTER;
 import static org.apache.hadoop.fs.azurebfs.enums.StatisticTypeEnum.TYPE_GAUGE;
+import static org.apache.hadoop.fs.azurebfs.utils.StringUtils.formatWithPrecision;
 import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.iostatisticsStore;
 
 /**
@@ -58,7 +59,7 @@ public class AbfsReadFooterMetrics extends AbstractAbfsStatisticsSource {
     /**
      * Inner class to handle file type checks.
      */
-    private static final class CheckFileType {
+    private static final class FileTypeMetrics {
         private final AtomicBoolean collectMetrics;
         private final AtomicBoolean collectMetricsForNextRead;
         private final AtomicBoolean collectLenMetrics;
@@ -68,7 +69,7 @@ private static final class CheckFileType {
         private String sizeReadByFirstRead;
         private String offsetDiffBetweenFirstAndSecondRead;
 
-        private CheckFileType() {
+        private FileTypeMetrics() {
             collectMetrics = new AtomicBoolean(false);
             collectMetricsForNextRead = new AtomicBoolean(false);
             collectLenMetrics = new AtomicBoolean(false);
@@ -151,7 +152,7 @@ private FileType getFileType() {
         }
     }
 
-    private final Map<String, CheckFileType> checkFileMap = new HashMap<>();
+    private final Map<String, FileTypeMetrics> fileTypeMetricsMap = new HashMap<>();
 
     /**
      * Constructor to initialize the IOStatisticsStore with counters and gauges.
@@ -230,7 +231,7 @@ public Long getTotalReadCount() {
      * @param filePathIdentifier the file path identifier
      */
     public void updateMap(String filePathIdentifier) {
-        checkFileMap.computeIfAbsent(filePathIdentifier, key -> new CheckFileType());
+        fileTypeMetricsMap.computeIfAbsent(filePathIdentifier, key -> new FileTypeMetrics());
     }
 
     /**
@@ -242,68 +243,104 @@ public void updateMap(String filePathIdentifier) {
      * @param nextReadPos the position of the next read
      */
     public void checkMetricUpdate(final String filePathIdentifier, final int len, final long contentLength, final long nextReadPos) {
-        CheckFileType checkFileType = checkFileMap.computeIfAbsent(filePathIdentifier, key -> new CheckFileType());
-        if (checkFileType.getReadCount() == 0 || (checkFileType.getReadCount() >= 1 && checkFileType.getCollectMetrics())) {
-            updateMetrics(checkFileType, len, contentLength, nextReadPos);
+        FileTypeMetrics fileTypeMetrics = fileTypeMetricsMap.computeIfAbsent(filePathIdentifier, key -> new FileTypeMetrics());
+        if (fileTypeMetrics.getReadCount() == 0 || (fileTypeMetrics.getReadCount() >= 1 && fileTypeMetrics.getCollectMetrics())) {
+            updateMetrics(fileTypeMetrics, len, contentLength, nextReadPos);
         }
     }
 
     /**
      * Updates metrics for a specific file identified by filePathIdentifier.
      *
-     * @param checkFileType      File metadata to know file type.
+     * @param fileTypeMetrics    File metadata to know file type.
      * @param len                The length of the read operation.
      * @param contentLength      The total content length of the file.
      * @param nextReadPos        The position of the next read operation.
      */
-    private void updateMetrics(CheckFileType checkFileType, int len, long contentLength, long nextReadPos) {
+    private void updateMetrics(FileTypeMetrics fileTypeMetrics, int len, long contentLength, long nextReadPos) {
         synchronized (this) {
-            checkFileType.incrementReadCount();
+            fileTypeMetrics.incrementReadCount();
         }
 
-        long readCount = checkFileType.getReadCount();
+        long readCount = fileTypeMetrics.getReadCount();
 
         if (readCount == 1) {
-            handleFirstRead(checkFileType, nextReadPos, len, contentLength);
+            handleFirstRead(fileTypeMetrics, nextReadPos, len, contentLength);
         } else if (readCount == 2) {
-            handleSecondRead(checkFileType, nextReadPos, len, contentLength);
+            handleSecondRead(fileTypeMetrics, nextReadPos, len, contentLength);
         } else {
-            handleFurtherRead(checkFileType, len);
+            handleFurtherRead(fileTypeMetrics, len);
         }
     }
 
-    private void handleFirstRead(CheckFileType checkFileType, long nextReadPos, int len, long contentLength) {
+    /**
+     * Handles the first read operation by checking if the current read position is near the end of the file.
+     * If it is, updates the {@link FileTypeMetrics} object to enable metrics collection and records the first read's
+     * offset and size.
+     *
+     * @param fileTypeMetrics The {@link FileTypeMetrics} object to update with metrics and read details.
+     * @param nextReadPos The position where the next read will start.
+     * @param len The length of the current read operation.
+     * @param contentLength The total length of the file content.
+     */
+    private void handleFirstRead(FileTypeMetrics fileTypeMetrics, long nextReadPos, int len, long contentLength) {
         if (nextReadPos >= contentLength - (long) Integer.parseInt(FOOTER_LENGTH) * ONE_KB) {
-            checkFileType.setCollectMetrics(true);
-            checkFileType.setCollectMetricsForNextRead(true);
-            checkFileType.setOffsetOfFirstRead(nextReadPos);
-            checkFileType.setSizeReadByFirstRead(len + "_" + Math.abs(contentLength - nextReadPos));
+            fileTypeMetrics.setCollectMetrics(true);
+            fileTypeMetrics.setCollectMetricsForNextRead(true);
+            fileTypeMetrics.setOffsetOfFirstRead(nextReadPos);
+            fileTypeMetrics.setSizeReadByFirstRead(len + "_" + Math.abs(contentLength - nextReadPos));
         }
     }
 
-    private void handleSecondRead(CheckFileType checkFileType, long nextReadPos, int len, long contentLength) {
-        if (checkFileType.getCollectMetricsForNextRead()) {
-            long offsetDiff = Math.abs(nextReadPos - checkFileType.getOffsetOfFirstRead());
-            checkFileType.setOffsetDiffBetweenFirstAndSecondRead(len + "_" + offsetDiff);
-            checkFileType.setCollectLenMetrics(true);
-            checkFileType.updateFileType();
-            updateMetricsData(checkFileType, len, contentLength);
+    /**
+     * Handles the second read operation by checking if metrics collection is enabled for the next read.
+     * If it is, calculates the offset difference between the first and second reads, updates the {@link FileTypeMetrics}
+     * object with this information, and sets the file type. Then, updates the metrics data.
+     *
+     * @param fileTypeMetrics The {@link FileTypeMetrics} object to update with metrics and read details.
+     * @param nextReadPos The position where the next read will start.
+     * @param len The length of the current read operation.
+     * @param contentLength The total length of the file content.
+     */
+    private void handleSecondRead(FileTypeMetrics fileTypeMetrics, long nextReadPos, int len, long contentLength) {
+        if (fileTypeMetrics.getCollectMetricsForNextRead()) {
+            long offsetDiff = Math.abs(nextReadPos - fileTypeMetrics.getOffsetOfFirstRead());
+            fileTypeMetrics.setOffsetDiffBetweenFirstAndSecondRead(len + "_" + offsetDiff);
+            fileTypeMetrics.setCollectLenMetrics(true);
+            fileTypeMetrics.updateFileType();
+            updateMetricsData(fileTypeMetrics, len, contentLength);
         }
     }
 
-    private synchronized void handleFurtherRead(CheckFileType checkFileType, int len) {
-        if (checkFileType.getCollectLenMetrics() && checkFileType.getFileType() != null) {
-            FileType fileType = checkFileType.getFileType();
+    /**
+     * Handles further read operations beyond the second read. If metrics collection is enabled and the file type is set,
+     * updates the read length requested and increments the read count for the specific file type.
+     *
+     * @param fileTypeMetrics The {@link FileTypeMetrics} object containing metrics and read details.
+     * @param len The length of the current read operation.
+     */
+    private synchronized void handleFurtherRead(FileTypeMetrics fileTypeMetrics, int len) {
+        if (fileTypeMetrics.getCollectLenMetrics() && fileTypeMetrics.getFileType() != null) {
+            FileType fileType = fileTypeMetrics.getFileType();
             updateMetricValue(fileType, READ_LEN_REQUESTED, len);
             incrementMetricValue(fileType, READ_COUNT);
         }
     }
 
-    private synchronized void updateMetricsData(CheckFileType checkFileType, int len, long contentLength) {
-        long sizeReadByFirstRead = Long.parseLong(checkFileType.getSizeReadByFirstRead().split("_")[0]);
-        long firstOffsetDiff = Long.parseLong(checkFileType.getSizeReadByFirstRead().split("_")[1]);
-        long secondOffsetDiff = Long.parseLong(checkFileType.getOffsetDiffBetweenFirstAndSecondRead().split("_")[1]);
-        FileType fileType = checkFileType.getFileType();
+    /**
+     * Updates the metrics data for a specific file identified by the {@link FileTypeMetrics} object.
+     * This method calculates and updates various metrics such as read length requested, file length,
+     * size read by the first read, and offset differences between reads.
+     *
+     * @param fileTypeMetrics The {@link FileTypeMetrics} object containing metrics and read details.
+     * @param len The length of the current read operation.
+     * @param contentLength The total length of the file content.
+     */
+    private synchronized void updateMetricsData(FileTypeMetrics fileTypeMetrics, int len, long contentLength) {
+        long sizeReadByFirstRead = Long.parseLong(fileTypeMetrics.getSizeReadByFirstRead().split("_")[0]);
+        long firstOffsetDiff = Long.parseLong(fileTypeMetrics.getSizeReadByFirstRead().split("_")[1]);
+        long secondOffsetDiff = Long.parseLong(fileTypeMetrics.getOffsetDiffBetweenFirstAndSecondRead().split("_")[1]);
+        FileType fileType = fileTypeMetrics.getFileType();
 
         updateMetricValue(fileType, READ_LEN_REQUESTED, len + sizeReadByFirstRead);
         updateMetricValue(fileType, FILE_LENGTH, contentLength);
@@ -314,39 +351,27 @@ private synchronized void updateMetricsData(CheckFileType checkFileType, int len
         incrementMetricValue(fileType, TOTAL_FILES);
     }
 
-    /**
-     * Returns the read footer metrics for a given file type.
-     *
-     * @param fileType the type of the file
-     * @return the read footer metrics as a string
-     */
-    public String getReadFooterMetrics(FileType fileType) {
-        StringBuilder readFooterMetric = new StringBuilder();
-        appendMetrics(readFooterMetric, fileType);
-        return readFooterMetric.toString();
-    }
-
     private void appendMetrics(StringBuilder metricBuilder, FileType fileType) {
         long totalFiles = getMetricValue(fileType, TOTAL_FILES);
         long readCount = getMetricValue(fileType, READ_COUNT);
         if (totalFiles <= 0 || readCount <= 0) {
             return;
         }
 
-        String sizeReadByFirstRead = String.format("%.3f", getMetricValue(fileType, SIZE_READ_BY_FIRST_READ) / (double) totalFiles);
-        String offsetDiffBetweenFirstAndSecondRead = String.format("%.3f",
-                getMetricValue(fileType, OFFSET_DIFF_BETWEEN_FIRST_AND_SECOND_READ) / (double) totalFiles);
+        String sizeReadByFirstRead = formatWithPrecision(getMetricValue(fileType, SIZE_READ_BY_FIRST_READ) / (double) totalFiles);
+        String offsetDiffBetweenFirstAndSecondRead = formatWithPrecision(getMetricValue(fileType,
+                OFFSET_DIFF_BETWEEN_FIRST_AND_SECOND_READ) / (double) totalFiles);
 
         if (NON_PARQUET.equals(fileType)) {
-            sizeReadByFirstRead += "_" + String.format("%.3f", getMetricValue(fileType, FIRST_OFFSET_DIFF) / (double) totalFiles);
-            offsetDiffBetweenFirstAndSecondRead += "_" + String.format("%.3f", getMetricValue(fileType, SECOND_OFFSET_DIFF) / (double) totalFiles);
+            sizeReadByFirstRead += "_" + formatWithPrecision(getMetricValue(fileType, FIRST_OFFSET_DIFF) / (double) totalFiles);
+            offsetDiffBetweenFirstAndSecondRead += "_" + formatWithPrecision(getMetricValue(fileType, SECOND_OFFSET_DIFF) / (double) totalFiles);
         }
 
         metricBuilder.append("$").append(fileType)
                 .append(":$FR=").append(sizeReadByFirstRead)
                 .append("$SR=").append(offsetDiffBetweenFirstAndSecondRead)
-                .append("$FL=").append(String.format("%.3f", getMetricValue(fileType, FILE_LENGTH) / (double) totalFiles))
-                .append("$RL=").append(String.format("%.3f", getMetricValue(fileType, READ_LEN_REQUESTED) / (double) readCount));
+                .append("$FL=").append(formatWithPrecision(getMetricValue(fileType, FILE_LENGTH) / (double) totalFiles))
+                .append("$RL=").append(formatWithPrecision(getMetricValue(fileType, READ_LEN_REQUESTED) / (double) readCount));
     }
 
     /**