HADOOP-17065. Add Network Counters to ABFS (#2056)

Contributed by Mehakmeet Singh.

(cherry picked from commit 3472c3e)

Author: mehakmeet

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

@@ -41,7 +41,7 @@
 /**
  * Instrumentation of Abfs counters.
  */
-public class AbfsInstrumentation implements AbfsCounters {
+public class AbfsCountersImpl implements AbfsCounters {
 
   /**
    * Single context for all the Abfs counters to separate them from other
@@ -78,10 +78,17 @@ public class AbfsInstrumentation implements AbfsCounters {
       DIRECTORIES_DELETED,
       FILES_CREATED,
       FILES_DELETED,
-      ERROR_IGNORED
+      ERROR_IGNORED,
+      CONNECTIONS_MADE,
+      SEND_REQUESTS,
+      GET_RESPONSES,
+      BYTES_SENT,
+      BYTES_RECEIVED,
+      READ_THROTTLES,
+      WRITE_THROTTLES
   };
 
-  public AbfsInstrumentation(URI uri) {
+  public AbfsCountersImpl(URI uri) {
     UUID fileSystemInstanceId = UUID.randomUUID();
     registry.tag(REGISTRY_ID,
         "A unique identifier for the instance",

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

@@ -22,7 +22,7 @@
 
 /**
  * Statistic which are collected in Abfs.
- * Available as metrics in {@link AbfsInstrumentation}.
+ * Available as metrics in {@link AbfsCountersImpl}.
  */
 public enum AbfsStatistic {
 
@@ -57,7 +57,23 @@ public enum AbfsStatistic {
   FILES_DELETED("files_deleted",
       "Total number of files deleted from the object store."),
   ERROR_IGNORED("error_ignored",
-      "Errors caught and ignored.");
+      "Errors caught and ignored."),
+
+  //Network statistics.
+  CONNECTIONS_MADE("connections_made",
+      "Total number of times a connection was made with the data store."),
+  SEND_REQUESTS("send_requests",
+      "Total number of times http requests were sent to the data store."),
+  GET_RESPONSES("get_responses",
+      "Total number of times a response was received."),
+  BYTES_SENT("bytes_sent",
+      "Total bytes uploaded."),
+  BYTES_RECEIVED("bytes_received",
+      "Total bytes received."),
+  READ_THROTTLES("read_throttles",
+      "Total number of times a read operation is throttled."),
+  WRITE_THROTTLES("write_throttles",
+      "Total number of times a write operation is throttled.");
 
   private String statName;
   private String statDescription;

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

@@ -97,7 +97,7 @@ public class AzureBlobFileSystem extends FileSystem {
   private boolean delegationTokenEnabled = false;
   private AbfsDelegationTokenManager delegationTokenManager;
   private AbfsAuthorizer authorizer;
-  private AbfsCounters instrumentation;
+  private AbfsCounters abfsCounters;
 
   @Override
   public void initialize(URI uri, Configuration configuration)
@@ -109,9 +109,10 @@ public void initialize(URI uri, Configuration configuration)
     LOG.debug("Initializing AzureBlobFileSystem for {}", uri);
 
     this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority());
-    this.abfsStore = new AzureBlobFileSystemStore(uri, this.isSecureScheme(), configuration);
+    abfsCounters = new AbfsCountersImpl(uri);
+    this.abfsStore = new AzureBlobFileSystemStore(uri, this.isSecureScheme(),
+        configuration, abfsCounters);
     final AbfsConfiguration abfsConfiguration = abfsStore.getAbfsConfiguration();
-    instrumentation = new AbfsInstrumentation(uri);
     this.setWorkingDirectory(this.getHomeDirectory());
 
     if (abfsConfiguration.getCreateRemoteFileSystemDuringInitialization()) {
@@ -147,8 +148,8 @@ public String toString() {
     sb.append("uri=").append(uri);
     sb.append(", user='").append(abfsStore.getUser()).append('\'');
     sb.append(", primaryUserGroup='").append(abfsStore.getPrimaryGroup()).append('\'');
-    if (instrumentation != null) {
-      sb.append(", Statistics: {").append(instrumentation.formString("{", "=",
+    if (abfsCounters != null) {
+      sb.append(", Statistics: {").append(abfsCounters.formString("{", "=",
           "}", true));
       sb.append("}");
     }
@@ -395,7 +396,7 @@ private void statIncrement(AbfsStatistic statistic) {
    * @param statistic the Statistic to be incremented.
    */
   private void incrementStatistic(AbfsStatistic statistic) {
-    instrumentation.incrementCounter(statistic, 1);
+    abfsCounters.incrementCounter(statistic, 1);
   }
 
   /**
@@ -1256,6 +1257,6 @@ private void performAbfsAuthCheck(FsAction action, Path... paths)
 
   @VisibleForTesting
   Map<String, Long> getInstrumentationMap() {
-    return instrumentation.toMap();
+    return abfsCounters.toMap();
   }
 }

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

@@ -73,6 +73,7 @@
 import org.apache.hadoop.fs.azurebfs.oauth2.IdentityTransformer;
 import org.apache.hadoop.fs.azurebfs.services.AbfsAclHelper;
 import org.apache.hadoop.fs.azurebfs.services.AbfsClient;
+import org.apache.hadoop.fs.azurebfs.services.AbfsCounters;
 import org.apache.hadoop.fs.azurebfs.services.AbfsHttpOperation;
 import org.apache.hadoop.fs.azurebfs.services.AbfsInputStream;
 import org.apache.hadoop.fs.azurebfs.services.AbfsInputStreamContext;
@@ -132,8 +133,9 @@ public class AzureBlobFileSystemStore {
   private final UserGroupInformation userGroupInformation;
   private final IdentityTransformer identityTransformer;
 
-  public AzureBlobFileSystemStore(URI uri, boolean isSecureScheme, Configuration configuration)
-          throws IOException {
+  public AzureBlobFileSystemStore(URI uri, boolean isSecureScheme,
+                                  Configuration configuration,
+                                  AbfsCounters abfsCounters) throws IOException {
     this.uri = uri;
     String[] authorityParts = authorityParts(uri);
     final String fileSystemName = authorityParts[0];
@@ -163,7 +165,7 @@ public AzureBlobFileSystemStore(URI uri, boolean isSecureScheme, Configuration c
     this.authType = abfsConfiguration.getAuthType(accountName);
     boolean usingOauth = (authType == AuthType.OAuth);
     boolean useHttps = (usingOauth || abfsConfiguration.isHttpsAlwaysUsed()) ? true : isSecureScheme;
-    initializeClient(uri, fileSystemName, accountName, useHttps);
+    initializeClient(uri, fileSystemName, accountName, useHttps, abfsCounters);
     this.identityTransformer = new IdentityTransformer(abfsConfiguration.getRawConfiguration());
   }
 
@@ -927,7 +929,8 @@ public boolean isAtomicRenameKey(String key) {
     return isKeyForDirectorySet(key, azureAtomicRenameDirSet);
   }
 
-  private void initializeClient(URI uri, String fileSystemName, String accountName, boolean isSecure) throws AzureBlobFileSystemException {
+  private void initializeClient(URI uri, String fileSystemName,
+      String accountName, boolean isSecure, AbfsCounters abfsCounters) throws AzureBlobFileSystemException {
     if (this.client != null) {
       return;
     }
@@ -960,7 +963,7 @@ private void initializeClient(URI uri, String fileSystemName, String accountName
 
     this.client = new AbfsClient(baseUrl, creds, abfsConfiguration,
         new ExponentialRetryPolicy(abfsConfiguration.getMaxIoRetries()),
-        tokenProvider);
+        tokenProvider, abfsCounters);
   }
 
   private String getOctalNotation(FsPermission fsPermission) {

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

@@ -59,12 +59,14 @@ public class AbfsClient {
   private final String userAgent;
 
   private final AccessTokenProvider tokenProvider;
+  private final AbfsCounters abfsCounters;
 
 
   public AbfsClient(final URL baseUrl, final SharedKeyCredentials sharedKeyCredentials,
                     final AbfsConfiguration abfsConfiguration,
                     final ExponentialRetryPolicy exponentialRetryPolicy,
-                    final AccessTokenProvider tokenProvider) {
+                    final AccessTokenProvider tokenProvider,
+                    final AbfsCounters abfsCounters) {
     this.baseUrl = baseUrl;
     this.sharedKeyCredentials = sharedKeyCredentials;
     String baseUrlString = baseUrl.toString();
@@ -85,6 +87,7 @@ public AbfsClient(final URL baseUrl, final SharedKeyCredentials sharedKeyCredent
 
     this.userAgent = initializeUserAgent(abfsConfiguration, sslProviderName);
     this.tokenProvider = tokenProvider;
+    this.abfsCounters = abfsCounters;
   }
 
   public String getFileSystem() {
@@ -608,4 +611,12 @@ String initializeUserAgent(final AbfsConfiguration abfsConfiguration,
   URL getBaseUrl() {
     return baseUrl;
   }
+
+  /**
+   * Getter for abfsCounters from AbfsClient.
+   * @return AbfsCounters instance.
+   */
+  protected AbfsCounters getAbfsCounters() {
+    return abfsCounters;
+  }
 }

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

@@ -114,16 +114,19 @@ public void addBytesTransferred(long count, boolean isFailedOperation) {
 
   /**
    * Suspends the current storage operation, as necessary, to reduce throughput.
+   * @return true if Thread sleeps(Throttling occurs) else false.
    */
-  public void suspendIfNecessary() {
+  public boolean suspendIfNecessary() {
     int duration = sleepDuration;
     if (duration > 0) {
       try {
         Thread.sleep(duration);
+        return true;
       } catch (InterruptedException ie) {
         Thread.currentThread().interrupt();
       }
     }
+    return false;
   }
 
   @VisibleForTesting
@@ -269,4 +272,4 @@ static class AbfsOperationMetrics {
       this.operationsSuccessful = new AtomicLong();
     }
   }
-}
+}

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

@@ -23,6 +23,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hadoop.fs.azurebfs.AbfsStatistic;
 import org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations;
 
 /**
@@ -103,17 +104,24 @@ static void updateMetrics(AbfsRestOperationType operationType,
    * uses this to suspend the request, if necessary, to minimize errors and
    * maximize throughput.
    */
-  static void sendingRequest(AbfsRestOperationType operationType) {
+  static void sendingRequest(AbfsRestOperationType operationType,
+      AbfsCounters abfsCounters) {
     if (!isAutoThrottlingEnabled) {
       return;
     }
 
     switch (operationType) {
       case ReadFile:
-        singleton.readThrottler.suspendIfNecessary();
+        if (singleton.readThrottler.suspendIfNecessary()
+            && abfsCounters != null) {
+          abfsCounters.incrementCounter(AbfsStatistic.READ_THROTTLES, 1);
+        }
         break;
       case Append:
-        singleton.writeThrottler.suspendIfNecessary();
+        if (singleton.writeThrottler.suspendIfNecessary()
+            && abfsCounters != null) {
+          abfsCounters.incrementCounter(AbfsStatistic.WRITE_THROTTLES, 1);
+        }
         break;
       default:
         break;

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

@@ -27,6 +27,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hadoop.fs.azurebfs.AbfsStatistic;
 import org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AzureBlobFileSystemException;
@@ -62,6 +63,7 @@ public class AbfsRestOperation {
   private int bufferLength;
 
   private AbfsHttpOperation result;
+  private AbfsCounters abfsCounters;
 
   public AbfsHttpOperation getResult() {
     return result;
@@ -87,6 +89,7 @@ public AbfsHttpOperation getResult() {
     this.requestHeaders = requestHeaders;
     this.hasRequestBody = (AbfsHttpConstants.HTTP_METHOD_PUT.equals(method)
             || AbfsHttpConstants.HTTP_METHOD_PATCH.equals(method));
+    this.abfsCounters = client.getAbfsCounters();
   }
 
   /**
@@ -114,6 +117,7 @@ public AbfsHttpOperation getResult() {
     this.buffer = buffer;
     this.bufferOffset = bufferOffset;
     this.bufferLength = bufferLength;
+    this.abfsCounters = client.getAbfsCounters();
   }
 
   /**
@@ -149,6 +153,7 @@ private boolean executeHttpOperation(final int retryCount) throws AzureBlobFileS
     try {
       // initialize the HTTP request and open the connection
       httpOperation = new AbfsHttpOperation(url, method, requestHeaders);
+      incrementCounter(AbfsStatistic.CONNECTIONS_MADE, 1);
 
       // sign the HTTP request
       if (client.getAccessToken() == null) {
@@ -161,14 +166,19 @@ private boolean executeHttpOperation(final int retryCount) throws AzureBlobFileS
                 client.getAccessToken());
       }
 
-      AbfsClientThrottlingIntercept.sendingRequest(operationType);
+      AbfsClientThrottlingIntercept.sendingRequest(operationType, abfsCounters);
 
       if (hasRequestBody) {
         // HttpUrlConnection requires
         httpOperation.sendRequest(buffer, bufferOffset, bufferLength);
+        incrementCounter(AbfsStatistic.SEND_REQUESTS, 1);
+        incrementCounter(AbfsStatistic.BYTES_SENT, bufferLength);
       }
 
       httpOperation.processResponse(buffer, bufferOffset, bufferLength);
+      incrementCounter(AbfsStatistic.GET_RESPONSES, 1);
+      incrementCounter(AbfsStatistic.BYTES_RECEIVED,
+          httpOperation.getBytesReceived());
     } catch (IOException ex) {
       if (ex instanceof UnknownHostException) {
         LOG.warn(String.format("Unknown host name: %s. Retrying to resolve the host name...", httpOperation.getUrl().getHost()));
@@ -208,4 +218,16 @@ private boolean executeHttpOperation(final int retryCount) throws AzureBlobFileS
 
     return true;
   }
+
+  /**
+   * Incrementing Abfs counters with a long value.
+   *
+   * @param statistic the Abfs statistic that needs to be incremented.
+   * @param value     the value to be incremented by.
+   */
+  private void incrementCounter(AbfsStatistic statistic, long value) {
+    if (abfsCounters != null) {
+      abfsCounters.incrementCounter(statistic, value);
+    }
+  }
 }

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/AbstractAbfsIntegrationTest.java

@@ -376,9 +376,10 @@ protected AbfsOutputStream createAbfsOutputStreamWithFlushEnabled(
    * @param metricMap map of (String, Long) with statistics name as key and
    *                  statistics value as map value.
    */
-  protected void assertAbfsStatistics(AbfsStatistic statistic,
+  protected long assertAbfsStatistics(AbfsStatistic statistic,
       long expectedValue, Map<String, Long> metricMap) {
     assertEquals("Mismatch in " + statistic.getStatName(), expectedValue,
         (long) metricMap.get(statistic.getStatName()));
+    return expectedValue;
   }
 }