HADOOP-19766: ABFS. Force Blob Instance for FNS Accounts

hadoop-tools/hadoop-azure/dev-support/testrun-scripts/runtests.sh

@@ -48,14 +48,6 @@ runHNSSharedKeyDFSTest()
   triggerRun "HNS-SharedKey-DFS" "$accountName"  "$runTest" $processCount "$cleanUpTestContainers"
 }
 
-runNonHNSSharedKeyDFSTest()
-{
-  accountName=$(xmlstarlet sel -t -v '//property[name = "fs.azure.nonHnsTestAccountName"]/value' -n $azureTestXmlPath)
-  PROPERTIES=("fs.azure.account.auth.type")
-  VALUES=("SharedKey")
-  triggerRun "NonHNS-SharedKey-DFS" "$accountName" "$runTest" $processCount "$cleanUpTestContainers"
-}
-
 runAppendBlobHNSOAuthDFSTest()
 {
   accountName=$(xmlstarlet sel -t -v '//property[name = "fs.azure.hnsTestAccountName"]/value' -n $azureTestXmlPath)
@@ -73,14 +65,6 @@ runNonHNSSharedKeyBlobTest()
   triggerRun "NonHNS-SharedKey-Blob" "${accountName}_blob" "$runTest" $processCount "$cleanUpTestContainers"
 }
 
-runNonHNSOAuthDFSTest()
-{
-  accountName=$(xmlstarlet sel -t -v '//property[name = "fs.azure.nonHnsTestAccountName"]/value' -n $azureTestXmlPath)
-  PROPERTIES=("fs.azure.account.auth.type")
-  VALUES=("OAuth")
-  triggerRun "NonHNS-OAuth-DFS" "$accountName" "$runTest" $processCount "$cleanUpTestContainers"
-}
-
 runNonHNSOAuthBlobTest()
 {
   accountName=$(xmlstarlet sel -t -v '//property[name = "fs.azure.nonHnsTestAccountName"]/value' -n $azureTestXmlPath)
@@ -107,14 +91,6 @@ runHNSOAuthDFSIngressBlobTest()
   triggerRun "HNS-Oauth-DFS-IngressBlob" "$accountName" "$runTest" $processCount "$cleanUpTestContainers"
 }
 
-runNonHNSOAuthDFSIngressBlobTest()
-{
-  accountName=$(xmlstarlet sel -t -v '//property[name = "fs.azure.nonHnsTestAccountName"]/value' -n $azureTestXmlPath)
-  PROPERTIES=("fs.azure.account.auth.type" "fs.azure.ingress.service.type")
-  VALUES=("OAuth" "blob")
-  triggerRun "NonHNS-OAuth-DFS-IngressBlob" "$accountName" "$runTest" $processCount "$cleanUpTestContainers"
-}
-
 runTest=false
 cleanUpTestContainers=false
 echo 'Ensure below are complete before running script:'
@@ -181,7 +157,7 @@ done
 
 echo ' '
 echo 'Set the active test combination to run the action:'
-select combo in HNS-OAuth-DFS HNS-SharedKey-DFS NonHNS-SharedKey-DFS AppendBlob-HNS-OAuth-DFS NonHNS-SharedKey-Blob NonHNS-OAuth-DFS NonHNS-OAuth-Blob AppendBlob-NonHNS-OAuth-Blob HNS-Oauth-DFS-IngressBlob NonHNS-Oauth-DFS-IngressBlob AllCombinationsTestRun Quit
+select combo in HNS-OAuth-DFS HNS-SharedKey-DFS AppendBlob-HNS-OAuth-DFS NonHNS-SharedKey-Blob NonHNS-OAuth-Blob AppendBlob-NonHNS-OAuth-Blob HNS-Oauth-DFS-IngressBlob AllCombinationsTestRun Quit
 do
    case $combo in
       HNS-OAuth-DFS)
@@ -192,10 +168,6 @@ do
          runHNSSharedKeyDFSTest
          break
          ;;
-      NonHNS-SharedKey-DFS)
-         runNonHNSSharedKeyDFSTest
-         break
-         ;;
        AppendBlob-HNS-OAuth-DFS)
          runAppendBlobHNSOAuthDFSTest
          break
@@ -204,10 +176,6 @@ do
          runNonHNSSharedKeyBlobTest
          break
          ;;
-        NonHNS-OAuth-DFS)
-         runNonHNSOAuthDFSTest
-         break
-         ;;
         NonHNS-OAuth-Blob)
          runNonHNSOAuthBlobTest
          break
@@ -220,10 +188,6 @@ do
          runHNSOAuthDFSIngressBlobTest
          break
          ;;
-        NonHNS-Oauth-DFS-IngressBlob)
-         runNonHNSOAuthDFSIngressBlobTest
-         break
-         ;;
       AllCombinationsTestRun)
         if [ $runTest == false ]
         then
@@ -232,14 +196,11 @@ do
         fi
         runHNSOAuthDFSTest
         runHNSSharedKeyDFSTest
-        runNonHNSSharedKeyDFSTest
         runAppendBlobHNSOAuthDFSTest
         runNonHNSSharedKeyBlobTest
-        runNonHNSOAuthDFSTest
         runNonHNSOAuthBlobTest
         runAppendBlobNonHNSOAuthBlobTest
         runHNSOAuthDFSIngressBlobTest
-        runNonHNSOAuthDFSIngressBlobTest
          break
          ;;
       Quit)

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

@@ -80,7 +80,6 @@
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.EMPTY_STRING;
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.*;
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.*;
-import static org.apache.hadoop.fs.azurebfs.services.AbfsErrors.INCORRECT_INGRESS_TYPE;
 
 /**
  * Configuration for Azure Blob FileSystem.
@@ -93,7 +92,7 @@ public class AbfsConfiguration{
   private final String accountName;
   private String fsName;
   // Service type identified from URL used to initialize FileSystem.
-  private final AbfsServiceType fsConfiguredServiceType;
+  private AbfsServiceType fsConfiguredServiceTypeFromUrl;
   private final boolean isSecure;
   private static final Logger LOG = LoggerFactory.getLogger(AbfsConfiguration.class);
   private Trilean isNamespaceEnabled = null;
@@ -663,18 +662,18 @@ public class AbfsConfiguration{
    * Constructor for AbfsConfiguration for specified service type.
    * @param rawConfig used to initialize the configuration.
    * @param accountName the name of the azure storage account.
-   * @param fsConfiguredServiceType service type configured for the file system.
+   * @param fsConfiguredServiceTypeFromUrl service type configured for the file system.
    * @throws IllegalAccessException if the field is not accessible.
    * @throws IOException if an I/O error occurs.
    */
   public AbfsConfiguration(final Configuration rawConfig,
       String accountName,
-      AbfsServiceType fsConfiguredServiceType)
+      AbfsServiceType fsConfiguredServiceTypeFromUrl)
       throws IllegalAccessException, IOException {
     this.rawConfig = ProviderUtils.excludeIncompatibleCredentialProviders(
         rawConfig, AzureBlobFileSystem.class);
     this.accountName = accountName;
-    this.fsConfiguredServiceType = fsConfiguredServiceType;
+    this.fsConfiguredServiceTypeFromUrl = fsConfiguredServiceTypeFromUrl;
     this.isSecure = getBoolean(FS_AZURE_SECURE_MODE, false);
 
     Field[] fields = this.getClass().getDeclaredFields();
@@ -701,16 +700,16 @@ public AbfsConfiguration(final Configuration rawConfig,
    * @param rawConfig used to initialize the configuration.
    * @param accountName the name of the azure storage account.
    * @param fsName the name of the file system (container name).
-   * @param fsConfiguredServiceType service type configured for the file system.
+   * @param fsConfiguredServiceTypeFromUrl service type configured for the file system.
    * @throws IllegalAccessException if the field is not accessible.
    * @throws IOException if an I/O error occurs.
    */
   public AbfsConfiguration(final Configuration rawConfig,
       String accountName,
       String fsName,
-      AbfsServiceType fsConfiguredServiceType)
+      AbfsServiceType fsConfiguredServiceTypeFromUrl)
       throws IllegalAccessException, IOException {
-    this(rawConfig, accountName, fsConfiguredServiceType);
+    this(rawConfig, accountName, fsConfiguredServiceTypeFromUrl);
     this.fsName = fsName;
   }
 
@@ -749,7 +748,16 @@ public Trilean getIsNamespaceEnabledAccount() {
    * @return the service type.
    */
   public AbfsServiceType getFsConfiguredServiceType() {
-    return getCaseInsensitiveEnum(FS_AZURE_FNS_ACCOUNT_SERVICE_TYPE, fsConfiguredServiceType);
+    return getCaseInsensitiveEnum(FS_AZURE_FNS_ACCOUNT_SERVICE_TYPE, fsConfiguredServiceTypeFromUrl);
+  }
+
+  /**
+   * Returns the service type identified from the URL used to initialize the FileSystem.
+   *
+   * @return the configured AbfsServiceType from the URL
+   */
+  public AbfsServiceType getFsConfiguredServiceTypeFromUrl() {
+    return fsConfiguredServiceTypeFromUrl;
   }
 
   /**
@@ -790,13 +798,9 @@ public void validateConfiguredServiceType(boolean isHNSEnabled)
     if (isHNSEnabled && getConfiguredServiceTypeForFNSAccounts() == AbfsServiceType.BLOB) {
       throw new InvalidConfigurationValueException(
           FS_AZURE_FNS_ACCOUNT_SERVICE_TYPE, "Service Type Cannot be BLOB for HNS Account");
-    } else if (isHNSEnabled && fsConfiguredServiceType == AbfsServiceType.BLOB) {
+    } else if (isHNSEnabled && fsConfiguredServiceTypeFromUrl == AbfsServiceType.BLOB) {
       throw new InvalidConfigurationValueException(FS_DEFAULT_NAME_KEY,
           "Blob Endpoint Url Cannot be used to initialize filesystem for HNS Account");
-    } else if (getFsConfiguredServiceType() == AbfsServiceType.BLOB
-        && getIngressServiceType() == AbfsServiceType.DFS) {
-      throw new InvalidConfigurationValueException(
-          FS_AZURE_INGRESS_SERVICE_TYPE, INCORRECT_INGRESS_TYPE);
     }
   }
 
@@ -1799,6 +1803,14 @@ void setReadAheadEnabled(final boolean enabledReadAhead) {
     this.enabledReadAhead = enabledReadAhead;
   }
 
+  /**
+   * Sets the configured service type.
+   * Used to update the service type identified from the URL.
+   */
+  void setFsConfiguredServiceType(AbfsServiceType serviceType) {
+    this.fsConfiguredServiceTypeFromUrl = serviceType;
+  }
+
   public int getReadAheadRange() {
     return this.readAheadRange;
   }

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

@@ -129,6 +129,7 @@
 import static org.apache.hadoop.fs.azurebfs.constants.FSOperationType.CREATE_FILESYSTEM;
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.BLOCK_UPLOAD_ACTIVE_BLOCKS_DEFAULT;
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.DATA_BLOCKS_BUFFER_DEFAULT;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemUriSchemes.ABFS_DFS_DOMAIN_NAME;
 import static org.apache.hadoop.fs.azurebfs.constants.InternalConstants.CAPABILITY_SAFE_READAHEAD;
 import static org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_CREATE_ON_ROOT;
 import static org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_INVALID_ABFS_STATE;
@@ -315,6 +316,18 @@ public void initialize(URI uri, Configuration configuration)
       throw new InvalidConfigurationValueException(FS_AZURE_ACCOUNT_IS_HNS_ENABLED, ex);
     }
 
+    /*
+     * For FNS accounts, restrict the endpoint and service type to Blob
+     * For FNS-DFS, also update the tracing context to add metric to show endpoint conversion.
+     */
+    if (!tryGetIsNamespaceEnabled(new TracingContext(initFSTracingContext))) {
+      LOG.debug("FNS account detected; restricting service type to Blob.");
+      abfsStore.restrictServiceTypeToBlob();
+      if (uri.toString().contains(ABFS_DFS_DOMAIN_NAME)) {
+        initFSTracingContext.setFNSEndpointConverted();
+      }
+    }
+
     // Create the file system if it does not exist.
     if (abfsConfiguration.getCreateRemoteFileSystemDuringInitialization()) {
       TracingContext createFSTracingContext = new TracingContext(initFSTracingContext);
@@ -1861,4 +1874,3 @@ public IOStatistics getIOStatistics() {
     return abfsCounters != null ? abfsCounters.getIOStatistics() : null;
   }
 }
-

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

@@ -798,9 +798,11 @@ private AbfsOutputStreamContext populateAbfsOutputStreamContext(
       ContextEncryptionAdapter contextEncryptionAdapter,
       TracingContext tracingContext) {
     int bufferSize = abfsConfiguration.getWriteBufferSize();
+
     if (isAppendBlob && bufferSize > FileSystemConfigurations.APPENDBLOB_MAX_WRITE_BUFFER_SIZE) {
       bufferSize = FileSystemConfigurations.APPENDBLOB_MAX_WRITE_BUFFER_SIZE;
     }
+
     return new AbfsOutputStreamContext(abfsConfiguration.getSasTokenRenewPeriodForStreamsInSeconds())
             .withWriteBufferSize(bufferSize)
             .enableExpectHeader(abfsConfiguration.isExpectHeaderEnabled())
@@ -824,7 +826,7 @@ private AbfsOutputStreamContext populateAbfsOutputStreamContext(
             .withWriteThreadPoolManager(writeThreadPoolSizeManager)
             .withTracingContext(tracingContext)
             .withAbfsBackRef(fsBackRef)
-            .withIngressServiceType(abfsConfiguration.getIngressServiceType())
+            .withIngressServiceType(clientHandler.getIngressServiceType())
             .withDFSToBlobFallbackEnabled(abfsConfiguration.isDfsToBlobFallbackEnabled())
             .withETag(eTag)
             .build();
@@ -1906,6 +1908,17 @@ private AbfsPerfInfo startTracking(String callerName, String calleeName) {
     return new AbfsPerfInfo(abfsPerfTracker, callerName, calleeName);
   }
 
+  /**
+   * Restricts all service types to BLOB when FNS account detected
+   * Updates the client to reflect the new default service type.
+   */
+  public void restrictServiceTypeToBlob() {
+    clientHandler.setDefaultServiceType(AbfsServiceType.BLOB);
+    clientHandler.setIngressServiceType(AbfsServiceType.BLOB);
+    getAbfsConfiguration().setFsConfiguredServiceType(AbfsServiceType.BLOB);
+    this.client = clientHandler.getClient();
+  }
+
   /**
    * Permissions class contain provided permission and umask in octalNotation.
    * If the object is created for namespace-disabled account, the permission and

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

@@ -42,6 +42,14 @@ public class MsiTokenProvider extends AccessTokenProvider {
 
   private static final Logger LOG = LoggerFactory.getLogger(AccessTokenProvider.class);
 
+  /**
+   * Constructs an MsiTokenProvider.
+   *
+   * @param authEndpoint the authentication endpoint for MSI
+   * @param tenantGuid   the tenant GUID
+   * @param clientId     the client ID for MSI
+   * @param authority    the authority URL
+   */
   public MsiTokenProvider(final String authEndpoint, final String tenantGuid,
       final String clientId, final String authority) {
     this.authEndpoint = authEndpoint;

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

@@ -1373,7 +1373,8 @@ String initializeUserAgent(final AbfsConfiguration abfsConfiguration,
     // Add a unique identifier in FNS-Blob user agent string
     // Current filesystem init restricts HNS-Blob combination
     // so namespace check not required.
-    if (abfsConfiguration.getFsConfiguredServiceType() == BLOB) {
+    // We need to rely on URL check to identify Blob service instead of user config
+    if (abfsConfiguration.getFsConfiguredServiceTypeFromUrl() == BLOB) {
       sb.append(SEMICOLON)
           .append(SINGLE_WHITE_SPACE)
           .append(FNS_BLOB_USER_AGENT_IDENTIFIER);

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

@@ -91,6 +91,33 @@ private void initServiceType(final AbfsConfiguration abfsConfiguration) {
     this.ingressServiceType = abfsConfiguration.getIngressServiceType();
   }
 
+  /**
+   * Sets the default service type.
+   *
+   * @param defaultServiceType the service type to set as default
+   */
+  public void setDefaultServiceType(AbfsServiceType defaultServiceType) {
+    this.defaultServiceType = defaultServiceType;
+  }
+
+  /**
+   * Sets the ingress service type.
+   *
+   * @param ingressServiceType the ingress service type
+   */
+  public void setIngressServiceType(AbfsServiceType ingressServiceType) {
+    this.ingressServiceType = ingressServiceType;
+  }
+
+  /**
+   * Gets the default ingress service type.
+   *
+   * @return the default ingress service type
+   */
+  public AbfsServiceType getIngressServiceType() {
+    return ingressServiceType;
+  }
+
   /**
    * Get the AbfsClient based on the default service type.
    * @return AbfsClient

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

@@ -320,7 +320,11 @@ private AzureIngressHandler createNewHandler(AbfsServiceType serviceType,
       boolean isSwitch,
       AzureBlockManager blockManager) throws IOException {
     this.client = clientHandler.getClient(serviceType);
-    if (isDFSToBlobFallbackEnabled && serviceTypeAtInit != AbfsServiceType.DFS) {
+
+    // Check ingress service type is also set to DFS along with enabling the config for fallback
+    // Separate ingress service type is only allowed for HNS accounts
+    if (isDFSToBlobFallbackEnabled && client.getIsNamespaceEnabled()
+        && serviceTypeAtInit != AbfsServiceType.DFS) {
       throw new InvalidConfigurationValueException(
           "The ingress service type must be configured as DFS");
     }

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

@@ -29,6 +29,10 @@
 
 @InterfaceAudience.Private
 @InterfaceStability.Unstable
+
+/**
+ * Interface for listing support in Azure Blob File System.
+ */
 public interface ListingSupport {
 
   /**

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

@@ -32,6 +32,7 @@ public interface Listener {
   Listener getClone();
   void setOperation(FSOperationType operation);
   void updateIngressHandler(String ingressHandler);
+  void updateFNSEndpointConverted();
   void updatePosition(String position);
   void updateReadType(ReadType readType);
 }