Merge pull request apache#74 from ABFSDriver/aclApi

Acl API's to go to DFS endpoint and throttling metrics collection based on operation type update

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

@@ -1300,7 +1300,10 @@ public AbfsRestOperation setAcl(final String path, final String aclSpecString, f
     abfsUriQueryBuilder.addQuery(HttpQueryParams.QUERY_PARAM_ACTION, AbfsHttpConstants.SET_ACCESS_CONTROL);
     appendSASTokenToQuery(path, SASTokenProvider.SET_ACL_OPERATION, abfsUriQueryBuilder);
 
-    final URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
+    URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
+    if (url.toString().contains(WASB_DNS_PREFIX)) {
+      url = changePrefixFromBlobtoDfs(url);
+    }
     final AbfsRestOperation op = new AbfsRestOperation(
         AbfsRestOperationType.SetAcl,
         this,
@@ -1325,7 +1328,10 @@ public AbfsRestOperation getAclStatus(final String path, final boolean useUPN,
     abfsUriQueryBuilder.addQuery(HttpQueryParams.QUERY_PARAM_UPN, String.valueOf(useUPN));
     appendSASTokenToQuery(path, SASTokenProvider.GET_ACL_OPERATION, abfsUriQueryBuilder);
 
-    final URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
+    URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
+    if (url.toString().contains(WASB_DNS_PREFIX)) {
+      url = changePrefixFromBlobtoDfs(url);
+    }
     final AbfsRestOperation op = new AbfsRestOperation(
         AbfsRestOperationType.GetAcl,
         this,
@@ -1352,7 +1358,12 @@ public AbfsRestOperation checkAccess(String path, String rwx, TracingContext tra
     abfsUriQueryBuilder.addQuery(QUERY_PARAM_ACTION, CHECK_ACCESS);
     abfsUriQueryBuilder.addQuery(QUERY_FS_ACTION, rwx);
     appendSASTokenToQuery(path, SASTokenProvider.CHECK_ACCESS_OPERATION, abfsUriQueryBuilder);
-    final URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
+
+    URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
+    if (url.toString().contains(WASB_DNS_PREFIX)) {
+      url = changePrefixFromBlobtoDfs(url);
+    }
+
     AbfsRestOperation op = new AbfsRestOperation(
         AbfsRestOperationType.CheckAccess, this,
         AbfsHttpConstants.HTTP_METHOD_HEAD, url, createDefaultHeaders());

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

@@ -153,6 +153,7 @@ public void updateMetrics(AbfsRestOperationType operationType,
 
     switch (operationType) {
       case Append:
+      case PutBlock:
         contentLength = abfsHttpOperation.getBytesSent();
         if (contentLength == 0) {
           /*
@@ -170,6 +171,7 @@ public void updateMetrics(AbfsRestOperationType operationType,
         }
         break;
       case ReadFile:
+      case GetBlob:
         String range = abfsHttpOperation.getConnection().getRequestProperty(HttpHeaderConfigurations.RANGE);
         contentLength = getContentLengthIfKnown(range);
         if (contentLength > 0) {
@@ -192,12 +194,14 @@ public void sendingRequest(AbfsRestOperationType operationType,
       AbfsCounters abfsCounters) {
     switch (operationType) {
       case ReadFile:
+      case GetBlob:
         if (readThrottler.suspendIfNecessary()
             && abfsCounters != null) {
           abfsCounters.incrementCounter(AbfsStatistic.READ_THROTTLES, 1);
         }
         break;
       case Append:
+      case PutBlock:
         if (writeThrottler.suspendIfNecessary()
             && abfsCounters != null) {
           abfsCounters.incrementCounter(AbfsStatistic.WRITE_THROTTLES, 1);

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

@@ -104,13 +104,8 @@ public void testCreateStatistics() throws IOException {
     assertAbfsStatistics(AbfsStatistic.CALL_CREATE_NON_RECURSIVE, 1, metricMap);
     assertAbfsStatistics(AbfsStatistic.FILES_CREATED, 1, metricMap);
     // Child calls mkdirs for parent in case of blob.
-    if (getPrefixMode(fs) == PrefixMode.BLOB) {
-      assertAbfsStatistics(AbfsStatistic.DIRECTORIES_CREATED, 2, metricMap);
-      assertAbfsStatistics(AbfsStatistic.CALL_MKDIRS, 2, metricMap);
-    } else {
-      assertAbfsStatistics(AbfsStatistic.DIRECTORIES_CREATED, 1, metricMap);
-      assertAbfsStatistics(AbfsStatistic.CALL_MKDIRS, 1, metricMap);
-    }
+    assertAbfsStatistics(AbfsStatistic.DIRECTORIES_CREATED, 1, metricMap);
+    assertAbfsStatistics(AbfsStatistic.CALL_MKDIRS, 1, metricMap);
     assertAbfsStatistics(AbfsStatistic.CALL_GET_FILE_STATUS, 3, metricMap);
 
     //re-initialising Abfs to reset statistic values.
@@ -138,15 +133,9 @@ public void testCreateStatistics() throws IOException {
         metricMap);
     assertAbfsStatistics(AbfsStatistic.FILES_CREATED, NUMBER_OF_OPS, metricMap);
     // Child calls mkdirs for parent in case of blob.
-    if (getPrefixMode(fs) == PrefixMode.BLOB) {
-      assertAbfsStatistics(AbfsStatistic.DIRECTORIES_CREATED, 2 * NUMBER_OF_OPS,
+    assertAbfsStatistics(AbfsStatistic.DIRECTORIES_CREATED, NUMBER_OF_OPS,
               metricMap);
-      assertAbfsStatistics(AbfsStatistic.CALL_MKDIRS, 2 * NUMBER_OF_OPS, metricMap);
-    } else {
-      assertAbfsStatistics(AbfsStatistic.DIRECTORIES_CREATED, NUMBER_OF_OPS,
-              metricMap);
-      assertAbfsStatistics(AbfsStatistic.CALL_MKDIRS, NUMBER_OF_OPS, metricMap);
-    }
+    assertAbfsStatistics(AbfsStatistic.CALL_MKDIRS, NUMBER_OF_OPS, metricMap);
     assertAbfsStatistics(AbfsStatistic.CALL_GET_FILE_STATUS,
         1 + 2 * NUMBER_OF_OPS, metricMap);
   }

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

@@ -20,6 +20,7 @@
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.lang.reflect.Field;
+import java.nio.file.AccessDeniedException;
 import java.util.List;
 
 import org.apache.hadoop.fs.azurebfs.services.PrefixMode;
@@ -42,6 +43,7 @@
 
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_CREATE_REMOTE_FILESYSTEM_DURING_INITIALIZATION;
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ACCOUNT_AUTH_TYPE_PROPERTY_NAME;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ACCOUNT_IS_HNS_ENABLED;
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ACCOUNT_OAUTH_CLIENT_ENDPOINT;
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ACCOUNT_TOKEN_PROVIDER_TYPE_PROPERTY_NAME;
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ENABLE_CHECK_ACCESS;
@@ -109,6 +111,30 @@ private void setTestUserFs() throws Exception {
     this.testUserFs = FileSystem.newInstance(getRawConfiguration());
   }
 
+  private void setTestUserFsNonHNS() throws Exception {
+    AzureBlobFileSystemStore abfsStore = getAbfsStore(getFileSystem());
+    String accountName = this.getAccountName();
+    if (abfsStore.getPrefixMode() == PrefixMode.BLOB) {
+      if (abfsStore.getAbfsConfiguration().shouldEnableBlobEndPoint()) {
+        accountName = getAccountName().replace(ABFS_DNS_PREFIX, WASB_DNS_PREFIX);
+      }
+    }
+    checkIfConfigIsSet(FS_AZURE_ACCOUNT_OAUTH_CLIENT_ENDPOINT
+            + "." + accountName);
+    Configuration conf = Mockito.spy(getRawConfiguration());
+    setTestFsConf1(FS_AZURE_BLOB_FS_CLIENT_ID,
+            FS_AZURE_BLOB_FS_CHECKACCESS_TEST_CLIENT_ID, conf);
+    setTestFsConf1(FS_AZURE_BLOB_FS_CLIENT_SECRET,
+            FS_AZURE_BLOB_FS_CHECKACCESS_TEST_CLIENT_SECRET, conf);
+    conf.set(FS_AZURE_ACCOUNT_AUTH_TYPE_PROPERTY_NAME, AuthType.OAuth.name());
+    conf.set(FS_AZURE_ACCOUNT_TOKEN_PROVIDER_TYPE_PROPERTY_NAME + "."
+            + accountName, ClientCredsTokenProvider.class.getName());
+    conf.setBoolean(AZURE_CREATE_REMOTE_FILESYSTEM_DURING_INITIALIZATION,
+            false);
+    FileSystem testUserFsNonHns;
+    testUserFsNonHns = FileSystem.newInstance(conf);
+  }
+
   private void setTestFsConf(final String fsConfKey,
       final String testFsConfKey) {
     final String confKeyWithAccountName = fsConfKey + "." + getAccountName();
@@ -117,6 +143,14 @@ private void setTestFsConf(final String fsConfKey,
     getRawConfiguration().set(confKeyWithAccountName, confValue);
   }
 
+  private void setTestFsConf1(final String fsConfKey,
+                             final String testFsConfKey, Configuration conf) {
+    final String confKeyWithAccountName = fsConfKey + "." + getAccountName();
+    final String confValue = getConfiguration()
+            .getString(testFsConfKey, "");
+    conf.set(confKeyWithAccountName, confValue);
+  }
+
   @Test(expected = IllegalArgumentException.class)
   public void testCheckAccessWithNullPath() throws IOException {
     superUserFs.access(null, FsAction.READ);
@@ -184,14 +218,14 @@ public void testCheckAccessForAccountWithoutNS() throws Exception {
     checkIfConfigIsSet(FS_AZURE_BLOB_FS_CHECKACCESS_TEST_CLIENT_SECRET);
     checkIfConfigIsSet(FS_AZURE_BLOB_FS_CHECKACCESS_TEST_USER_GUID);
 
-    setTestUserFs();
+    Configuration configuration = Mockito.spy(getFileSystem().getConf());
+    FileSystem testUserFSWithoutNS;
+    testUserFSWithoutNS = FileSystem.newInstance(configuration);
 
     //  When the driver does not know if the account is HNS enabled or not it
     //  makes a server call and fails
-    intercept(AccessControlException.class,
-        "\"This request is not authorized to perform this operation using "
-            + "this permission.\", 403",
-        () -> testUserFs.access(new Path("/"), FsAction.READ));
+    intercept(Exception.class, "\"This request is not authorized to perform this operation using "
+            + "this permission.\", 403", this::setTestUserFsNonHNS);
 
     //  When the driver has already determined if the account is HNS enabled
     //  or not, and as the account is non HNS the AzureBlobFileSystem#access
@@ -204,8 +238,8 @@ public void testCheckAccessForAccountWithoutNS() throws Exception {
     Field abfsStoreField = AzureBlobFileSystem.class.getDeclaredField(
         "abfsStore");
     abfsStoreField.setAccessible(true);
-    abfsStoreField.set(testUserFs, mockAbfsStore);
-    testUserFs.access(new Path("/"), FsAction.READ);
+    abfsStoreField.set(testUserFSWithoutNS, mockAbfsStore);
+    testUserFSWithoutNS.access(new Path("/"), FsAction.READ);
 
     superUserFs.access(new Path("/"), FsAction.READ);
   }