HADOOP-19729. [ABFS][Perf] Network Profiling for Tailing Requests and Killing Bad Connections Proactively (#8043)

Contributed by Anuj Modi

Author: anujmodi2021

hadoop-tools/hadoop-azure/pom.xml

@@ -199,6 +199,13 @@
       <scope>compile</scope>
     </dependency>
 
+    <dependency>
+      <groupId>org.hdrhistogram</groupId>
+      <artifactId>HdrHistogram</artifactId>
+      <version>2.1.12</version>
+      <scope>compile</scope>
+    </dependency>
+
     <!--com.fasterxml.jackson is used by WASB, not ABFS-->
     <!--transitive dependency from Azure SDK-->
     <dependency>

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

@@ -601,6 +601,43 @@ public class AbfsConfiguration{
       DefaultValue = DEFAULT_FS_AZURE_ENABLE_CREATE_BLOB_IDEMPOTENCY)
   private boolean enableCreateIdempotency;
 
+  @BooleanConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_ENABLE_TAIL_LATENCY_TRACKER,
+      DefaultValue = DEFAULT_FS_AZURE_ENABLE_TAIL_LATENCY_TRACKER)
+  private boolean isTailLatencyTrackerEnabled;
+
+  @BooleanConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_ENABLE_TAIL_LATENCY_REQUEST_TIMEOUT,
+      DefaultValue = DEFAULT_FS_AZURE_ENABLE_TAIL_LATENCY_REQUEST_TIMEOUT)
+  private boolean isTailLatencyRequestTimeoutEnabled;
+
+  @IntegerConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_TAIL_LATENCY_PERCENTILE,
+      DefaultValue = DEFAULT_FS_AZURE_TAIL_LATENCY_PERCENTILE)
+  private int tailLatencyPercentile;
+
+  @IntegerConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_TAIL_LATENCY_MIN_DEVIATION,
+      DefaultValue = DEFAULT_FS_AZURE_TAIL_LATENCY_MIN_DEVIATION)
+  private int tailLatencyMinDeviation;
+
+  @IntegerConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_TAIL_LATENCY_MIN_SAMPLE_SIZE,
+      DefaultValue = DEFAULT_FS_AZURE_TAIL_LATENCY_MIN_SAMPLE_SIZE)
+  private int tailLatencyMinSampleSize;
+
+  @IntegerConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_TAIL_LATENCY_ANALYSIS_WINDOW_MILLIS,
+      DefaultValue = DEFAULT_FS_AZURE_TAIL_LATENCY_ANALYSIS_WINDOW_MILLIS)
+  private int tailLatencyAnalysisWindowInMillis;
+
+  @IntegerConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_TAIL_LATENCY_ANALYSIS_WINDOW_GRANULARITY,
+      DefaultValue = DEFAULT_FS_AZURE_TAIL_LATENCY_ANALYSIS_WINDOW_GRANULARITY,
+      MinValue = MIN_FS_AZURE_TAIL_LATENCY_ANALYSIS_WINDOW_GRANULARITY)
+  private int tailLatencyAnalysisWindowGranularity;
+
+  @IntegerConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_TAIL_LATENCY_PERCENTILE_COMPUTATION_INTERVAL_MILLIS,
+      DefaultValue = DEFAULT_FS_AZURE_TAIL_LATENCY_PERCENTILE_COMPUTATION_INTERVAL_MILLIS)
+  private int tailLatencyPercentileComputationIntervalInMillis;
+
+  @IntegerConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_TAIL_LATENCY_MAX_RETRY_COUNT,
+      DefaultValue = DEFAULT_FS_AZURE_TAIL_LATENCY_MAX_RETRY_COUNT)
+  private int tailLatencyMaxRetryCount;
+
   private String clientProvidedEncryptionKey;
   private String clientProvidedEncryptionKeySHA;
 
@@ -1970,4 +2007,80 @@ public int getBlobRenameDirConsumptionParallelism() {
   public int getBlobDeleteDirConsumptionParallelism() {
     return blobDeleteDirConsumptionParallelism;
   }
+
+  /**
+   * Checks if the tail latency tracker is enabled.
+   * @return true if enabled, false otherwise.
+   */
+  public boolean isTailLatencyTrackerEnabled() {
+    return isTailLatencyTrackerEnabled;
+  }
+
+  /**
+   * Checks if the tail latency request timeout feature is enabled.
+   * @return true if enabled, false otherwise.
+   */
+  public boolean isTailLatencyRequestTimeoutEnabled() {
+    return isTailLatencyTrackerEnabled && isTailLatencyRequestTimeoutEnabled
+        && getPreferredHttpOperationType().equals(HttpOperationType.APACHE_HTTP_CLIENT);
+  }
+
+  /**
+   * Gets the tail latency percentile to be tracked.
+   * @return the tail latency percentile.
+   */
+  public int getTailLatencyPercentile() {
+    return tailLatencyPercentile;
+  }
+
+  /**
+   * Gets the minimum deviation for tail latency tracking.
+   * @return the minimum deviation.
+   */
+  public int getTailLatencyMinDeviation() {
+    return tailLatencyMinDeviation;
+  }
+
+  /**
+   * Gets the minimum sample size for tail latency reporting.
+   * @return the minimum sample size.
+   */
+  public int getTailLatencyMinSampleSize() {
+    return tailLatencyMinSampleSize;
+  }
+
+  /**
+   * Gets the tail latency analysis window in milliseconds.
+   * @return the analysis window in milliseconds.
+   */
+  public int getTailLatencyAnalysisWindowInMillis() {
+    return tailLatencyAnalysisWindowInMillis;
+  }
+
+  /**
+   * Gets the interval for tail latency percentile computation in milliseconds.
+   * @return the computation interval in milliseconds.
+   */
+  public int getTailLatencyComputationIntervalInMillis() {
+    return tailLatencyPercentileComputationIntervalInMillis;
+  }
+
+  /**
+   * Gets the granularity of the tail latency analysis window.
+   * @return the analysis window granularity.
+   */
+  public int getTailLatencyAnalysisWindowGranularity() {
+    if (tailLatencyAnalysisWindowGranularity <= 0) {
+        return MIN_FS_AZURE_TAIL_LATENCY_ANALYSIS_WINDOW_GRANULARITY;
+    }
+    return tailLatencyAnalysisWindowGranularity;
+  }
+
+  /**
+   * Gets the maximum retry count for tail latency requests.
+   * @return the maximum retry count.
+   */
+  public int getTailLatencyMaxRetryCount() {
+    return tailLatencyMaxRetryCount;
+  }
 }

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

@@ -113,6 +113,7 @@
 import org.apache.hadoop.fs.azurebfs.services.ListingSupport;
 import org.apache.hadoop.fs.azurebfs.services.SharedKeyCredentials;
 import org.apache.hadoop.fs.azurebfs.services.StaticRetryPolicy;
+import org.apache.hadoop.fs.azurebfs.services.TailLatencyRequestTimeoutRetryPolicy;
 import org.apache.hadoop.fs.azurebfs.services.VersionedFileStatus;
 import org.apache.hadoop.fs.azurebfs.utils.Base64;
 import org.apache.hadoop.fs.azurebfs.utils.CRC64;
@@ -1817,6 +1818,8 @@ private AbfsClientContext populateAbfsClientContext() {
             new ExponentialRetryPolicy(abfsConfiguration))
         .withStaticRetryPolicy(
             new StaticRetryPolicy(abfsConfiguration))
+        .withTailLatencyRequestTimeoutRetryPolicy(
+            new TailLatencyRequestTimeoutRetryPolicy(abfsConfiguration))
         .withAbfsCounters(abfsCounters)
         .withAbfsPerfTracker(abfsPerfTracker)
         .build();

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

@@ -530,5 +530,60 @@ public static String containerProperty(String property, String fsName, String ac
   /**Flag to enable/disable create idempotency during create operation: {@value}*/
   public static final String FS_AZURE_ENABLE_CREATE_BLOB_IDEMPOTENCY = "fs.azure.enable.create.blob.idempotency";
 
+  /**
+   * Flag to enable/disable tail latency tracker for AbfsRestOperation.
+   * When enabled, Client observed E2E latency will be tracked by a histogram.
+   * Regular p50, p99 and configured percentile latencies will be reported.
+   */
+  public static final String FS_AZURE_ENABLE_TAIL_LATENCY_TRACKER = "fs.azure.enable.tail.latency.tracker";
+
+  /**
+   * Flag to enable/disable tail latency based timeout for AbfsRestOperation.
+   * When enabled, if an operation's latency exceeds the currently reported tail
+   * latency by the tracker, the ongoing socket connection will be closed and
+   * the operation will be retried, up to the configured max retry count: {@value}
+   */
+  public static final String FS_AZURE_ENABLE_TAIL_LATENCY_REQUEST_TIMEOUT = "fs.azure.enable.tail.latency.timeout";
+
+  /**
+   * The percentile value to be considered as tail latency value.
+   * Default is 99.0 (99th percentile): {@value}
+   */
+  public static final String FS_AZURE_TAIL_LATENCY_PERCENTILE = "fs.azure.tail.latency.percentile";
+
+  /**
+   * The minimum deviation (in percentage) between p50 and tail latency
+   * percentile to trigger tail latency based request timeout: {@value}
+   */
+  public static final String FS_AZURE_TAIL_LATENCY_MIN_DEVIATION = "fs.azure.tail.latency.min.deviation";
+
+  /**
+   * The minimum sample size required before the histogram starts reporting latency data: {@value}
+   */
+  public static final String FS_AZURE_TAIL_LATENCY_MIN_SAMPLE_SIZE = "fs.azure.tail.latency.min.sample.size";
+
+  /**
+   * The time window (in milliseconds) over which the tail latency analysis is performed.
+   * Until the whole window is filled, the histogram will not report any latency data: {@value}
+   */
+  public static final String FS_AZURE_TAIL_LATENCY_ANALYSIS_WINDOW_MILLIS = "fs.azure.tail.latency.analysis.window.millis";
+
+  /**
+   * The granularity (in milliseconds) at which the tail latency analysis window is divided.
+   * This is to make sliding window calculations efficient and robust: {@value}
+   */
+  public static final String FS_AZURE_TAIL_LATENCY_ANALYSIS_WINDOW_GRANULARITY = "fs.azure.tail.latency.analysis.window.granularity";
+
+  /**
+   * Interval (in milliseconds) at which the tail latency percentile is computed
+   * and updated by the background thread for each operation type: {@value}
+   */
+  public static final String FS_AZURE_TAIL_LATENCY_PERCENTILE_COMPUTATION_INTERVAL_MILLIS = "fs.azure.tail.latency.percentile.computation.interval.millis";
+
+  /**
+   * Maximum number of retries for an operation when tail latency based timeout occur: {@value}
+   */
+  public static final String FS_AZURE_TAIL_LATENCY_MAX_RETRY_COUNT = "fs.azure.tail.latency.max.retry.count";
+
   private ConfigurationKeys() {}
 }

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

@@ -241,6 +241,7 @@ public final class FileSystemConfigurations {
   public static final int RATE_LIMIT_DEFAULT = 1_000;
 
   public static final int ZERO = 0;
+  public static final double ZERO_D = 0.0;
   public static final int HUNDRED = 100;
   public static final double HUNDRED_D = 100.0;
   public static final long THOUSAND = 1000L;
@@ -392,5 +393,16 @@ public final class FileSystemConfigurations {
 
   public static final boolean DEFAULT_FS_AZURE_ENABLE_CREATE_BLOB_IDEMPOTENCY = true;
 
+  public static final boolean DEFAULT_FS_AZURE_ENABLE_TAIL_LATENCY_TRACKER = false;
+  public static final boolean DEFAULT_FS_AZURE_ENABLE_TAIL_LATENCY_REQUEST_TIMEOUT = false;
+  public static final int DEFAULT_FS_AZURE_TAIL_LATENCY_PERCENTILE = 99;
+  public static final int DEFAULT_FS_AZURE_TAIL_LATENCY_MIN_DEVIATION = 200;
+  public static final int DEFAULT_FS_AZURE_TAIL_LATENCY_MIN_SAMPLE_SIZE = 100;
+  public static final int DEFAULT_FS_AZURE_TAIL_LATENCY_ANALYSIS_WINDOW_MILLIS = 60_000;
+  public static final int DEFAULT_FS_AZURE_TAIL_LATENCY_ANALYSIS_WINDOW_GRANULARITY = 10;
+  public static final int MIN_FS_AZURE_TAIL_LATENCY_ANALYSIS_WINDOW_GRANULARITY = 1;
+  public static final int DEFAULT_FS_AZURE_TAIL_LATENCY_PERCENTILE_COMPUTATION_INTERVAL_MILLIS = 500;
+  public static final int DEFAULT_FS_AZURE_TAIL_LATENCY_MAX_RETRY_COUNT = 1;
+
   private FileSystemConfigurations() {}
 }

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/exceptions/TailLatencyRequestTimeoutException.java

@@ -0,0 +1,43 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.contracts.exceptions;
+
+import java.util.concurrent.TimeoutException;
+import static org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_TAIL_LATENCY_REQUEST_TIMEOUT;
+
+/**
+ * Thrown when a request takes more time than the current reported tail latency.
+ */
+public class TailLatencyRequestTimeoutException extends AzureBlobFileSystemException {
+
+  /**
+   * Constructs a TailLatencyRequestTimeoutException with TimeoutException as the cause.
+   * @param innerException the TimeoutException that caused this exception
+   */
+  public TailLatencyRequestTimeoutException(TimeoutException innerException) {
+    super(ERR_TAIL_LATENCY_REQUEST_TIMEOUT, innerException);
+  }
+
+  /**
+   * Constructs a TailLatencyRequestTimeoutException without a cause.
+   */
+  public TailLatencyRequestTimeoutException() {
+    super(ERR_TAIL_LATENCY_REQUEST_TIMEOUT);
+  }
+}

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

@@ -91,11 +91,19 @@ public class AbfsAHCHttpOperation extends AbfsHttpOperation {
    */
   private final AbfsApacheHttpClient abfsApacheHttpClient;
 
+  /**
+   * Timeout in milliseconds that defines maximum allowed time to execute operation.
+   * This timeout starts when execution starts and includes E2E processing time of request.
+   * This is based on tail latency observed in the system.
+   */
+  private final long tailLatencyTimeout;
+
   public AbfsAHCHttpOperation(final URL url,
       final String method,
       final List<AbfsHttpHeader> requestHeaders,
       final Duration connectionTimeout,
       final Duration readTimeout,
+      final long tailLatencyTimeout,
       final AbfsApacheHttpClient abfsApacheHttpClient,
       final AbfsClient abfsClient) throws IOException {
     super(LOG, url, method, requestHeaders, connectionTimeout, readTimeout,
@@ -104,6 +112,7 @@ public AbfsAHCHttpOperation(final URL url,
         || HTTP_METHOD_PATCH.equals(method)
         || HTTP_METHOD_POST.equals(method);
     this.abfsApacheHttpClient = abfsApacheHttpClient;
+    this.tailLatencyTimeout = tailLatencyTimeout;
     LOG.debug("Creating AbfsAHCHttpOperation for URL: {}, method: {}",
         url, method);
 
@@ -159,6 +168,10 @@ AbfsManagedHttpClientContext getHttpClientContext() {
     return new AbfsManagedHttpClientContext();
   }
 
+  long getTailLatencyTimeout() {
+    return tailLatencyTimeout;
+  }
+
   /**{@inheritDoc}*/
   @Override
   protected InputStream getErrorStream() throws IOException {
@@ -273,7 +286,7 @@ HttpResponse executeRequest() throws IOException {
     try {
       LOG.debug("Executing request: {}", httpRequestBase);
       HttpResponse response = abfsApacheHttpClient.execute(httpRequestBase,
-          abfsHttpClientContext, getConnectionTimeout(), getReadTimeout());
+          abfsHttpClientContext, getConnectionTimeout(), getReadTimeout(), getTailLatencyTimeout());
       setConnectionTimeMs(abfsHttpClientContext.getConnectTime());
       setSendRequestTimeMs(abfsHttpClientContext.getSendTime());
       setRecvResponseTimeMs(abfsHttpClientContext.getReadTime());