Hadoop 18325: ABFS: Add correlated metric support for ABFS operations

hadoop-tools/hadoop-azure/src/config/checkstyle-suppressions.xml

@@ -44,6 +44,16 @@
 <suppressions>
     <suppress checks="ParameterNumber|MagicNumber"
               files="org[\\/]apache[\\/]hadoop[\\/]fs[\\/]azurebfs[\\/]AzureBlobFileSystemStore.java"/>
+    <suppress checks="MagicNumber"
+              files="org[\\/]apache[\\/]hadoop[\\/]fs[\\/]azurebfs[\\/]AbfsDriverMetrics.java"/>
+    <suppress checks="MagicNumber"
+              files="org[\\/]apache[\\/]hadoop[\\/]fs[\\/]azurebfs[\\/]AbfsBackoffMetrics.java"/>
+    <suppress checks="MagicNumber"
+              files="org[\\/]apache[\\/]hadoop[\\/]fs[\\/]azurebfs[\\/]services[\\/]AbfsRestOperation.java"/>
+    <suppress checks="MagicNumber"
+              files="org[\\/]apache[\\/]hadoop[\\/]fs[\\/]azurebfs[\\/]services[\\/]AbfsInputStream.java"/>
+    <suppress checks="MagicNumber"
+              files="org[\\/]apache[\\/]hadoop[\\/]fs[\\/]azurebfs[\\/]services[\\/]TestAbfsRestOperation.java"/>
     <suppress checks="ParameterNumber|MagicNumber"
               files="org[\\/]apache[\\/]hadoop[\\/]fs[\\/]azurebfs[\\/]utils[\\/]Base64.java"/>
     <suppress checks="ParameterNumber|VisibilityModifier"

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

@@ -0,0 +1,217 @@
+/**
+ * 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.azurebfs;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.Map;
+import java.util.ArrayList;
+import java.util.Arrays;
+
+public class AbfsBackoffMetrics {
+
+  private AtomicLong numberOfRequestsSucceeded;
+
+  private AtomicLong minBackoff;
+
+  private AtomicLong maxBackoff;
+
+  private AtomicLong totalRequests;
+
+  private AtomicLong totalBackoff;
+
+  private String retryCount;
+
+  private AtomicLong numberOfIOPSThrottledRequests;
+
+  private AtomicLong numberOfBandwidthThrottledRequests;
+
+  private AtomicLong numberOfOtherThrottledRequests;
+
+  private AtomicLong numberOfNetworkFailedRequests;
+
+  private AtomicLong maxRetryCount;
+
+  private AtomicLong totalNumberOfRequests;
+
+  private AtomicLong numberOfRequestsSucceededWithoutRetrying;
+
+  private AtomicLong numberOfRequestsFailed;
+
+  private final Map<String, AbfsBackoffMetrics> metricsMap
+      = new ConcurrentHashMap<>();
+
+  public AbfsBackoffMetrics() {
+    initializeMap();
+    this.numberOfIOPSThrottledRequests = new AtomicLong();
+    this.numberOfBandwidthThrottledRequests = new AtomicLong();
+    this.numberOfOtherThrottledRequests = new AtomicLong();
+    this.totalNumberOfRequests = new AtomicLong();
+    this.maxRetryCount = new AtomicLong();
+    this.numberOfRequestsSucceededWithoutRetrying = new AtomicLong();
+    this.numberOfRequestsFailed = new AtomicLong();
+    this.numberOfNetworkFailedRequests = new AtomicLong();
+  }
+
+  public AbfsBackoffMetrics(String retryCount) {
+    this.retryCount = retryCount;
+    this.numberOfRequestsSucceeded = new AtomicLong();
+    this.minBackoff = new AtomicLong(Long.MAX_VALUE);
+    this.maxBackoff = new AtomicLong();
+    this.totalRequests = new AtomicLong();
+    this.totalBackoff = new AtomicLong();
+  }
+
+  private void initializeMap() {
+    ArrayList<String> retryCountList = new ArrayList<String>(
+        Arrays.asList("1", "2", "3", "4", "5_15", "15_25", "25AndAbove"));
+    for (String s : retryCountList) {
+      metricsMap.put(s, new AbfsBackoffMetrics(s));
+    }
+  }
+
+  public AtomicLong getNumberOfRequestsSucceeded() {
+    return numberOfRequestsSucceeded;
+  }
+
+  public AtomicLong getMinBackoff() {
+    return minBackoff;
+  }
+
+  public AtomicLong getMaxBackoff() {
+    return maxBackoff;
+  }
+
+  public AtomicLong getTotalRequests() {
+    return totalRequests;
+  }
+
+  public AtomicLong getTotalBackoff() {
+    return totalBackoff;
+  }
+
+  public String getRetryCount() {
+    return retryCount;
+  }
+
+  public AtomicLong getNumberOfIOPSThrottledRequests() {
+    return numberOfIOPSThrottledRequests;
+  }
+
+  public AtomicLong getNumberOfBandwidthThrottledRequests() {
+    return numberOfBandwidthThrottledRequests;
+  }
+
+  public AtomicLong getNumberOfOtherThrottledRequests() {
+    return numberOfOtherThrottledRequests;
+  }
+
+  public AtomicLong getMaxRetryCount() {
+    return maxRetryCount;
+  }
+
+  public AtomicLong getTotalNumberOfRequests() {
+    return totalNumberOfRequests;
+  }
+
+  public Map<String, AbfsBackoffMetrics> getMetricsMap() {
+    return metricsMap;
+  }
+
+  public AtomicLong getNumberOfRequestsSucceededWithoutRetrying() {
+    return numberOfRequestsSucceededWithoutRetrying;
+  }
+
+  public AtomicLong getNumberOfRequestsFailed() {
+    return numberOfRequestsFailed;
+  }
+
+  public AtomicLong getNumberOfNetworkFailedRequests() {
+    return numberOfNetworkFailedRequests;
+  }
+
+  /*
+    Acronyms :-
+    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() {
+    StringBuilder metricString = new StringBuilder();
+    long totalRequestsThrottled = numberOfBandwidthThrottledRequests.get()
+        + numberOfIOPSThrottledRequests.get()
+        + numberOfOtherThrottledRequests.get();
+    double percentageOfRequestsThrottled =
+        ((double) totalRequestsThrottled / totalNumberOfRequests.get()) * 100;
+    for (Map.Entry<String, AbfsBackoffMetrics> entry : metricsMap.entrySet()) {
+      metricString.append("#RCTSI#_").append(entry.getKey())
+          .append("R_").append("=")
+          .append(entry.getValue().getNumberOfRequestsSucceeded()).append(" ");
+      long totalRequests = entry.getValue().getTotalRequests().get();
+      if (totalRequests > 0) {
+        metricString.append("#MMA#_").append(entry.getKey())
+            .append("R_").append("=")
+            .append(String.format("%.3f",
+                (double) entry.getValue().getMinBackoff().get() / 1000L))
+            .append("s ")
+            .append(String.format("%.3f",
+                (double) entry.getValue().getMaxBackoff().get() / 1000L))
+            .append("s ")
+            .append(String.format("%.3f",
+                ((double) entry.getValue().getTotalBackoff().get() / totalRequests)
+                    / 1000L))
+            .append("s ");
+      } else {
+        metricString.append("#MMA#_").append(entry.getKey())
+            .append("R_").append("=0s ");
+      }
+    }
+    metricString.append("#BWT=")
+        .append(numberOfBandwidthThrottledRequests)
+        .append(" #IT=")
+        .append(numberOfIOPSThrottledRequests)
+        .append(" #OT=")
+        .append(numberOfOtherThrottledRequests)
+        .append(" #%RT=")
+        .append(String.format("%.3f", percentageOfRequestsThrottled))
+        .append(" #NFR=")
+        .append(numberOfNetworkFailedRequests)
+        .append(" #TRNR=")
+        .append(numberOfRequestsSucceededWithoutRetrying)
+        .append(" #TRF=")
+        .append(numberOfRequestsFailed)
+        .append(" #TR=")
+        .append(totalNumberOfRequests)
+        .append(" #MRC=")
+        .append(maxRetryCount);
+
+    return metricString + " ";
+  }
+}
+

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

@@ -759,6 +759,13 @@ public TracingHeaderFormat getTracingHeaderFormat() {
     return getEnum(FS_AZURE_TRACINGHEADER_FORMAT, TracingHeaderFormat.ALL_ID_FORMAT);
   }
 
+  /**
+   * Enum config to allow user to pick format of x-ms-client-request-id header
+   * @return tracingContextFormat config if valid, else default ALL_ID_FORMAT
+   */
+  public TracingHeaderFormat getTracingMetricHeaderFormat() {
+    return getEnum(FS_AZURE_TRACINGMETRICHEADER_FORMAT, TracingHeaderFormat.EMPTY);
+  }
   public AuthType getAuthType(String accountName) {
     return getEnum(FS_AZURE_ACCOUNT_AUTH_TYPE_PROPERTY_NAME, AuthType.SharedKey);
   }

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

@@ -21,7 +21,8 @@
 import java.net.URI;
 import java.util.Map;
 import java.util.UUID;
-
+import java.util.List;
+import java.util.ArrayList;
 import org.apache.hadoop.classification.VisibleForTesting;
 
 import org.apache.hadoop.fs.azurebfs.services.AbfsCounters;
@@ -33,9 +34,10 @@
 import org.apache.hadoop.metrics2.lib.MetricsRegistry;
 import org.apache.hadoop.metrics2.lib.MutableCounterLong;
 import org.apache.hadoop.metrics2.lib.MutableMetric;
-
+import org.apache.hadoop.fs.azurebfs.services.AbfsReadFooterMetrics;
 import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.*;
 import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.iostatisticsStore;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * Instrumentation of Abfs counters.
@@ -63,6 +65,10 @@ public class AbfsCountersImpl implements AbfsCounters {
 
   private final IOStatisticsStore ioStatisticsStore;
 
+  private AtomicReference<AbfsBackoffMetrics> abfsBackoffMetrics = null;
+
+  private List<AbfsReadFooterMetrics> readFooterMetricsList;
+
   private static final AbfsStatistic[] STATISTIC_LIST = {
       CALL_CREATE,
       CALL_OPEN,
@@ -121,6 +127,8 @@ public AbfsCountersImpl(URI uri) {
       ioStatisticsStoreBuilder.withDurationTracking(durationStats.getStatName());
     }
     ioStatisticsStore = ioStatisticsStoreBuilder.build();
+    abfsBackoffMetrics = new AtomicReference<>(new AbfsBackoffMetrics());
+    readFooterMetricsList = new ArrayList<>();
   }
 
   /**
@@ -188,6 +196,14 @@ private MetricsRegistry getRegistry() {
     return registry;
   }
 
+  public AbfsBackoffMetrics getAbfsBackoffMetrics() {
+    return abfsBackoffMetrics.get();
+  }
+
+  public List<AbfsReadFooterMetrics> getAbfsReadFooterMetrics() {
+    return readFooterMetricsList;
+  }
+
   /**
    * {@inheritDoc}
    *