HADOOP-19645. [ABFS][ReadAheadV2] Improve Metrics for Read Calls to identify type of read done.

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

@@ -128,6 +128,7 @@ public final class AbfsHttpConstants {
   public static final String STAR = "*";
   public static final String COMMA = ",";
   public static final String COLON = ":";
+  public static final String HYPHEN = "-";
   public static final String EQUAL = "=";
   public static final String QUESTION_MARK = "?";
   public static final String AND_MARK = "&";
@@ -174,6 +175,8 @@ public final class AbfsHttpConstants {
   public static final char CHAR_STAR = '*';
   public static final char CHAR_PLUS = '+';
 
+  public static final int SPLIT_NO_LIMIT = -1;
+
   /**
    * Specifies the version of the REST protocol used for processing the request.
    * Versions should be added in enum list in ascending chronological order.

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

@@ -0,0 +1,66 @@
+/**
+ * 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.constants;
+
+/**
+ * Enumeration for different types of read operations triggered by AbfsInputStream.
+ */
+public enum ReadType {
+  /**
+   * Synchronous read from the storage service. No optimization is being applied.
+   */
+  DIRECT_READ("DR"),
+  /**
+   * Synchronous read from the storage service where optimization were considered but found disabled.
+   */
+  NORMAL_READ("NR"),
+  /**
+   * Asynchronous read from the storage service for filling up cache.
+   */
+  PREFETCH_READ("PR"),
+  /**
+   * Synchronous read from the storage service when nothing was found in cache.
+   */
+  MISSEDCACHE_READ("MR"),
+  /**
+   * Synchronous read from the storage service for reading the footer of a file.
+   * Only triggered when footer read optimization kicks in.
+   */
+  FOOTER_READ("FR"),
+  /**
+   * Synchronous read from the storage service for reading a small file fully.
+   * Only triggered when small file read optimization kicks in.
+   */
+  SMALLFILE_READ("SR"),
+  /**
+   * None of the above read types were applicable.
+   */
+  UNKNOWN_READ("UR");
+
+  private final String readType;
+
+  ReadType(String readType) {
+    this.readType = readType;
+  }
+
+  @Override
+  public String toString() {
+    return readType;
+  }
+}

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

@@ -27,6 +27,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.fs.PositionedReadable;
+import org.apache.hadoop.fs.azurebfs.constants.ReadType;
 import org.apache.hadoop.fs.impl.BackReference;
 import org.apache.hadoop.util.Preconditions;
 
@@ -165,6 +166,7 @@ public AbfsInputStream(
     this.tracingContext = new TracingContext(tracingContext);
     this.tracingContext.setOperation(FSOperationType.READ);
     this.tracingContext.setStreamID(inputStreamId);
+    this.tracingContext.setReadType(ReadType.UNKNOWN_READ);
     this.context = abfsInputStreamContext;
     readAheadBlockSize = abfsInputStreamContext.getReadAheadBlockSize();
     if (abfsReadFooterMetrics != null) {
@@ -227,7 +229,9 @@ public int read(long position, byte[] buffer, int offset, int length)
     if (streamStatistics != null) {
       streamStatistics.readOperationStarted();
     }
-    int bytesRead = readRemote(position, buffer, offset, length, tracingContext);
+    TracingContext tc = new TracingContext(tracingContext);
+    tc.setReadType(ReadType.DIRECT_READ);
+    int bytesRead = readRemote(position, buffer, offset, length, tc);
     if (statistics != null) {
       statistics.incrementBytesRead(bytesRead);
     }
@@ -345,6 +349,8 @@ private int readOneBlock(final byte[] b, final int off, final int len) throws IO
         buffer = new byte[bufferSize];
       }
 
+      // Reset Read Type back to normal and set again based on code flow.
+      tracingContext.setReadType(ReadType.NORMAL_READ);
       if (alwaysReadBufferSize) {
         bytesRead = readInternal(fCursor, buffer, 0, bufferSize, false);
       } else {
@@ -385,6 +391,7 @@ private int readFileCompletely(final byte[] b, final int off, final int len)
     // data need to be copied to user buffer from index bCursor, bCursor has
     // to be the current fCusor
     bCursor = (int) fCursor;
+    tracingContext.setReadType(ReadType.SMALLFILE_READ);
     return optimisedRead(b, off, len, 0, contentLength);
   }
 
@@ -405,6 +412,7 @@ private int readLastBlock(final byte[] b, final int off, final int len)
     bCursor = (int) (fCursor - lastBlockStart);
     // 0 if contentlength is < buffersize
     long actualLenToRead = min(footerReadSize, contentLength);
+    tracingContext.setReadType(ReadType.FOOTER_READ);
     return optimisedRead(b, off, len, lastBlockStart, actualLenToRead);
   }
 
@@ -428,6 +436,7 @@ private int optimisedRead(final byte[] b, final int off, final int len,
       }
     } catch (IOException e) {
       LOG.debug("Optimized read failed. Defaulting to readOneBlock {}", e);
+      tracingContext.setReadType(ReadType.NORMAL_READ);
       restorePointerState();
       return readOneBlock(b, off, len);
     } finally {
@@ -442,6 +451,7 @@ private int optimisedRead(final byte[] b, final int off, final int len,
     //  bCursor that means the user requested data has not been read.
     if (fCursor < contentLength && bCursor > limit) {
       restorePointerState();
+      tracingContext.setReadType(ReadType.NORMAL_READ);
       return readOneBlock(b, off, len);
     }
     return copyToUserBuffer(b, off, len);
@@ -520,6 +530,7 @@ private int readInternal(final long position, final byte[] b, final int offset,
       LOG.debug("read ahead enabled issuing readheads num = {}", numReadAheads);
       TracingContext readAheadTracingContext = new TracingContext(tracingContext);
       readAheadTracingContext.setPrimaryRequestID();
+      readAheadTracingContext.setReadType(ReadType.PREFETCH_READ);
       while (numReadAheads > 0 && nextOffset < contentLength) {
         LOG.debug("issuing read ahead requestedOffset = {} requested size {}",
             nextOffset, nextSize);
@@ -544,7 +555,9 @@ private int readInternal(final long position, final byte[] b, final int offset,
       }
 
       // got nothing from read-ahead, do our own read now
-      receivedBytes = readRemote(position, b, offset, length, new TracingContext(tracingContext));
+      TracingContext tc = new TracingContext(tracingContext);
+      tc.setReadType(ReadType.MISSEDCACHE_READ);
+      receivedBytes = readRemote(position, b, offset, length, tc);
       return receivedBytes;
     } else {
       LOG.debug("read ahead disabled, reading remote");
@@ -578,6 +591,7 @@ int readRemote(long position, byte[] b, int offset, int length, TracingContext t
         streamStatistics.remoteReadOperation();
       }
       LOG.trace("Trigger client.read for path={} position={} offset={} length={}", path, position, offset, length);
+      tracingContext.setPosition(String.valueOf(position));
       op = client.read(path, position, b, offset, length,
           tolerateOobAppends ? "*" : eTag, cachedSasToken.get(),
           contextEncryptionAdapter, tracingContext);

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

@@ -19,6 +19,7 @@
 package org.apache.hadoop.fs.azurebfs.utils;
 
 import org.apache.hadoop.fs.azurebfs.constants.FSOperationType;
+import org.apache.hadoop.fs.azurebfs.constants.ReadType;
 
 /**
  * Interface for testing identifiers tracked via TracingContext
@@ -32,4 +33,5 @@ public interface Listener {
   void setOperation(FSOperationType operation);
   void updateIngressHandler(String ingressHandler);
   void updatePosition(String position);
+  void updateReadType(ReadType readType);
 }

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

@@ -25,9 +25,12 @@
 
 import org.apache.hadoop.fs.azurebfs.constants.FSOperationType;
 import org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations;
+import org.apache.hadoop.fs.azurebfs.constants.ReadType;
 import org.apache.hadoop.fs.azurebfs.services.AbfsClient;
 import org.apache.hadoop.fs.azurebfs.services.AbfsHttpOperation;
 
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.CHAR_HYPHEN;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.COLON;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.EMPTY_STRING;
 import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.CONNECTION_TIMEOUT_ABBREVIATION;
 
@@ -67,6 +70,7 @@ public class TracingContext {
   private String position = EMPTY_STRING;
   private String metricResults = EMPTY_STRING;
   private String metricHeader = EMPTY_STRING;
+  private ReadType readType = ReadType.UNKNOWN_READ;
 
   /**
    * If {@link #primaryRequestId} is null, this field shall be set equal
@@ -77,8 +81,7 @@ public class TracingContext {
    * this field shall not be set.
    */
   private String primaryRequestIdForRetry;
-
-  private Integer operatedBlobCount = null;
+  private Integer operatedBlobCount = 1; // Only relevant for rename-delete over blob endpoint where it will be explicitly set.
 
   private static final Logger LOG = LoggerFactory.getLogger(AbfsClient.class);
   public static final int MAX_CLIENT_CORRELATION_ID_LENGTH = 72;
@@ -142,6 +145,7 @@ public TracingContext(TracingContext originalTracingContext) {
       this.listener = originalTracingContext.listener.getClone();
     }
     this.metricResults = originalTracingContext.metricResults;
+    this.readType = originalTracingContext.readType;
   }
   public static String validateClientCorrelationID(String clientCorrelationID) {
     if ((clientCorrelationID.length() > MAX_CLIENT_CORRELATION_ID_LENGTH)
@@ -181,8 +185,24 @@ public void setListener(Listener listener) {
   }
 
   /**
-   * Concatenate all identifiers separated by (:) into a string and set into
+   * Concatenate all components separated by (:) into a string and set into
    * X_MS_CLIENT_REQUEST_ID header of the http operation
+   * Following are the components in order of concatenation:
+   * <ul>
+   *   <li>version - not present for versions less than v1</li>
+   *   <li>clientCorrelationId</li>
+   *   <li>clientRequestId</li>
+   *   <li>fileSystemId</li>
+   *   <li>primaryRequestId</li>
+   *   <li>streamId</li>
+   *   <li>opType</li>
+   *   <li>retryHeader - this contains retryCount, failureReason and retryPolicy underscore separated</li>
+   *   <li>ingressHandler</li>
+   *   <li>position of read/write in the remote file</li>
+   *   <li>operatedBlobCount - number of blobs operated on by this request</li>
+   *   <li>httpOperationHeader - suffix for network library used</li>
+   *   <li>operationSpecificHeader - different operation types can publish info relevant to that operation</li>
+   * </ul>
    * @param httpOperation AbfsHttpOperation instance to set header into
    *                      connection
    * @param previousFailure Failure seen before this API trigger on same operation
@@ -193,32 +213,33 @@ public void setListener(Listener listener) {
   public void constructHeader(AbfsHttpOperation httpOperation, String previousFailure, String retryPolicyAbbreviation) {
     clientRequestId = UUID.randomUUID().toString();
     switch (format) {
-    case ALL_ID_FORMAT: // Optional IDs (e.g. streamId) may be empty
-      header =
-          clientCorrelationID + ":" + clientRequestId + ":" + fileSystemID + ":"
-              + getPrimaryRequestIdForHeader(retryCount > 0) + ":" + streamID
-              + ":" + opType + ":" + retryCount;
-      header = addFailureReasons(header, previousFailure, retryPolicyAbbreviation);
-      if (!(ingressHandler.equals(EMPTY_STRING))) {
-        header += ":" + ingressHandler;
-      }
-      if (!(position.equals(EMPTY_STRING))) {
-        header += ":" + position;
-      }
-      if (operatedBlobCount != null) {
-        header += (":" + operatedBlobCount);
-      }
-      header += (":" + httpOperation.getTracingContextSuffix());
-      metricHeader += !(metricResults.trim().isEmpty()) ? metricResults  : "";
+    case ALL_ID_FORMAT:
+      header = TracingHeaderVersion.V1.getVersion() + COLON
+          + clientCorrelationID + COLON
+          + clientRequestId + COLON
+          + fileSystemID + COLON
+          + getPrimaryRequestIdForHeader(retryCount > 0) + COLON
+          + streamID + COLON
+          + opType + COLON
+          + getRetryHeader(previousFailure, retryPolicyAbbreviation) + COLON
+          + ingressHandler + COLON
+          + position + COLON
+          + operatedBlobCount + COLON
+          + httpOperation.getTracingContextSuffix() + COLON
+          + getOperationSpecificHeader(opType);
+
+      metricHeader += !(metricResults.trim().isEmpty()) ? metricResults  : EMPTY_STRING;
       break;
     case TWO_ID_FORMAT:
-      header = clientCorrelationID + ":" + clientRequestId;
-      metricHeader += !(metricResults.trim().isEmpty()) ? metricResults  : "";
+      header = TracingHeaderVersion.V1.getVersion() + COLON
+          + clientCorrelationID + COLON + clientRequestId;
+      metricHeader += !(metricResults.trim().isEmpty()) ? metricResults  : EMPTY_STRING;
       break;
     default:
       //case SINGLE_ID_FORMAT
-      header = clientRequestId;
-      metricHeader += !(metricResults.trim().isEmpty()) ? metricResults  : "";
+      header = TracingHeaderVersion.V1.getVersion() + COLON
+          + clientRequestId;
+      metricHeader += !(metricResults.trim().isEmpty()) ? metricResults  : EMPTY_STRING;
     }
     if (listener != null) { //for testing
       listener.callTracingHeaderValidator(header, format);
@@ -234,7 +255,7 @@ public void constructHeader(AbfsHttpOperation httpOperation, String previousFail
     * of the x-ms-client-request-id header in case of retry of the same API-request.
     */
     if (primaryRequestId.isEmpty() && previousFailure == null) {
-      String[] clientRequestIdParts = clientRequestId.split("-");
+      String[] clientRequestIdParts = clientRequestId.split(String.valueOf(CHAR_HYPHEN));
       primaryRequestIdForRetry = clientRequestIdParts[
           clientRequestIdParts.length - 1];
     }
@@ -265,6 +286,34 @@ private String addFailureReasons(final String header,
     return String.format("%s_%s", header, previousFailure);
   }
 
+  private String getRetryHeader(final String previousFailure, String retryPolicyAbbreviation) {
+    String retryHeader = String.format("%d", retryCount);
+    if (previousFailure == null) {
+      return retryHeader;
+    }
+    if (CONNECTION_TIMEOUT_ABBREVIATION.equals(previousFailure) && retryPolicyAbbreviation != null) {
+      return String.format("%s_%s_%s", retryHeader, previousFailure, retryPolicyAbbreviation);
+    }
+    return String.format("%s_%s", retryHeader, previousFailure);
+  }
+
+  private String getOperationSpecificHeader(FSOperationType opType) {
+    // Similar header can be added for other operations in the future.
+    switch (opType) {
+      case READ:
+        return getReadSpecificHeader();
+      default:
+        return EMPTY_STRING; // no operation specific header
+    }
+  }
+
+  private String getReadSpecificHeader() {
+    // More information on read can be added to this header in the future.
+    // As underscore separated values.
+    String readHeader = String.format("%s", readType.toString());
+    return readHeader;
+  }
+
   public void setOperatedBlobCount(Integer count) {
     operatedBlobCount = count;
   }
@@ -322,4 +371,15 @@ public void setPosition(final String position) {
       listener.updatePosition(position);
     }
   }
+
+  public void setReadType(ReadType readType) {
+    this.readType = readType;
+    if (listener != null) {
+      listener.updateReadType(readType);
+    }
+  }
+
+  public ReadType getReadType() {
+    return readType;
+  }
 }