HADOOP-16916: ABFS: Delegation SAS generator for integration with Ranger

hadoop-tools/hadoop-azure/dev-support/findbugs-exclude.xml

@@ -15,6 +15,23 @@
    limitations under the License.
 -->
 <FindBugsFilter>
+     <!-- This reference equality check is an intentional light weight
+          check to avoid re-validating the token when re-used. -->
+     <Match>
+       <Class name="org.apache.hadoop.fs.azurebfs.utils.CachedSASToken" />
+       <Method name="update" />
+       <Bug pattern="ES_COMPARING_PARAMETER_STRING_WITH_EQ" />
+     </Match>
+
+     <!-- This is intentional. The unsynchronized field access is safe
+          and only synchronized access is used when using the sasToken
+          for authorization. -->
+     <Match>
+       <Class name="org.apache.hadoop.fs.azurebfs.utils.CachedSASToken" />
+       <Field name="sasToken" />
+       <Bug pattern="IS2_INCONSISTENT_SYNC" />
+     </Match>
+
      <!-- It is okay to skip up to end of file. No need to check return value. -->
      <Match>
        <Class name="org.apache.hadoop.fs.azure.AzureNativeFileSystemStore" />
@@ -24,7 +41,7 @@
      </Match>
 
      <!-- Returning fully loaded array to iterate through is a convenience
-	  and helps performance. -->
+          and helps performance. -->
      <Match>
        <Class name="org.apache.hadoop.fs.azure.NativeAzureFileSystem$FolderRenamePending" />
        <Method name="getFiles" />
@@ -40,7 +57,7 @@
      </Match>
 
      <!-- Using a key set iterator is fine because this is not a performance-critical
-	  method. -->
+          method. -->
      <Match>
        <Class name="org.apache.hadoop.fs.azure.PageBlobOutputStream" />
        <Method name="logAllStackTraces" />

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

@@ -210,6 +210,11 @@ public class AbfsConfiguration{
           DefaultValue = DEFAULT_ABFS_LATENCY_TRACK)
   private boolean trackLatency;
 
+  @LongConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_SAS_TOKEN_RENEW_PERIOD_FOR_STREAMS,
+      MinValue = 0,
+      DefaultValue = DEFAULT_SAS_TOKEN_RENEW_PERIOD_FOR_STREAMS_IN_SECONDS)
+  private long sasTokenRenewPeriodForStreamsInSeconds;
+
   private Map<String, String> storageAccountKeys;
 
   public AbfsConfiguration(final Configuration rawConfig, String accountName)
@@ -451,6 +456,10 @@ public boolean isCheckAccessEnabled() {
     return this.isCheckAccessEnabled;
   }
 
+  public long getSasTokenRenewPeriodForStreamsInSeconds() {
+    return this.sasTokenRenewPeriodForStreamsInSeconds;
+  }
+
   public String getAzureBlockLocationHost() {
     return this.azureBlockLocationHost;
   }

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

@@ -642,15 +642,17 @@ public void setXAttr(final Path path, final String name, final byte[] value, fin
       throw new IllegalArgumentException("A valid name and value must be specified.");
     }
 
+    Path qualifiedPath = makeQualified(path);
+
     try {
-      Hashtable<String, String> properties = abfsStore.getPathStatus(path);
+      Hashtable<String, String> properties = abfsStore.getPathStatus(qualifiedPath);
       String xAttrName = ensureValidAttributeName(name);
       boolean xAttrExists = properties.containsKey(xAttrName);
       XAttrSetFlag.validate(name, xAttrExists, flag);
 
       String xAttrValue = abfsStore.decodeAttribute(value);
       properties.put(xAttrName, xAttrValue);
-      abfsStore.setPathProperties(path, properties);
+      abfsStore.setPathProperties(qualifiedPath, properties);
     } catch (AzureBlobFileSystemException ex) {
       checkException(path, ex);
     }
@@ -675,9 +677,11 @@ public byte[] getXAttr(final Path path, final String name)
       throw new IllegalArgumentException("A valid name must be specified.");
     }
 
+    Path qualifiedPath = makeQualified(path);
+
     byte[] value = null;
     try {
-      Hashtable<String, String> properties = abfsStore.getPathStatus(path);
+      Hashtable<String, String> properties = abfsStore.getPathStatus(qualifiedPath);
       String xAttrName = ensureValidAttributeName(name);
       if (properties.containsKey(xAttrName)) {
         String xAttrValue = properties.get(xAttrName);

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

@@ -142,5 +142,8 @@ public static String accountProperty(String property, String account) {
   /** Key for SAS token provider **/
   public static final String FS_AZURE_SAS_TOKEN_PROVIDER_TYPE = "fs.azure.sas.token.provider.type";
 
+  /** For performance, AbfsInputStream/AbfsOutputStream re-use SAS tokens until the expiry is within this number of seconds. **/
+  public static final String FS_AZURE_SAS_TOKEN_RENEW_PERIOD_FOR_STREAMS = "fs.azure.sas.token.renew.period.for.streams";
+
   private ConfigurationKeys() {}
 }

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

@@ -76,6 +76,7 @@ public final class FileSystemConfigurations {
   public static final boolean DEFAULT_USE_UPN = false;
   public static final boolean DEFAULT_ENABLE_CHECK_ACCESS = false;
   public static final boolean DEFAULT_ABFS_LATENCY_TRACK = false;
+  public static final long DEFAULT_SAS_TOKEN_RENEW_PERIOD_FOR_STREAMS_IN_SECONDS = 120;
 
   public static final String DEFAULT_FS_AZURE_USER_AGENT_PREFIX = EMPTY_STRING;
   public static final String DEFAULT_VALUE_UNKNOWN = "UNKNOWN";

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

@@ -32,22 +32,23 @@
 @InterfaceStability.Unstable
 public interface SASTokenProvider {
 
-  String CONCAT_SOURCE_OPERATION = "concat-source";
-  String CONCAT_TARGET_OPERATION = "concat-target";
-  String CREATEFILE_OPERATION = "create";
+  String CHECK_ACCESS_OPERATION = "check-access";
+  String CREATE_FILE_OPERATION = "create-file";
   String DELETE_OPERATION = "delete";
-  String EXECUTE_OPERATION = "execute";
-  String GETACL_OPERATION = "getaclstatus";
-  String GETFILESTATUS_OPERATION = "getfilestatus";
-  String LISTSTATUS_OPERATION = "liststatus";
-  String MKDIR_OPERATION = "mkdir";
+  String DELETE_RECURSIVE_OPERATION = "delete-recursive";
+  String GET_ACL_OPERATION = "get-acl";
+  String GET_STATUS_OPERATION = "get-status";
+  String GET_PROPERTIES_OPERATION = "get-properties";
+  String LIST_OPERATION = "list";
+  String CREATE_DIRECTORY_OPERATION = "create-directory";
   String READ_OPERATION = "read";
   String RENAME_SOURCE_OPERATION = "rename-source";
   String RENAME_DESTINATION_OPERATION = "rename-destination";
-  String SETACL_OPERATION = "setacl";
-  String SETOWNER_OPERATION = "setowner";
-  String SETPERMISSION_OPERATION = "setpermission";
-  String APPEND_OPERATION = "write";
+  String SET_ACL_OPERATION = "set-acl";
+  String SET_OWNER_OPERATION = "set-owner";
+  String SET_PERMISSION_OPERATION = "set-permission";
+  String SET_PROPERTIES_OPERATION = "set-properties";
+  String WRITE_OPERATION = "write";
 
   /**
    * Initialize authorizer for Azure Blob File System.

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

@@ -52,6 +52,7 @@ public final class AzureADAuthenticator {
 
   private static final Logger LOG = LoggerFactory.getLogger(AzureADAuthenticator.class);
   private static final String RESOURCE_NAME = "https://storage.azure.com/";
+  private static final String SCOPE = "https://storage.azure.com/.default";
   private static final int CONNECT_TIMEOUT = 30 * 1000;
   private static final int READ_TIMEOUT = 30 * 1000;
 
@@ -85,9 +86,14 @@ public static AzureADToken getTokenUsingClientCreds(String authEndpoint,
     Preconditions.checkNotNull(authEndpoint, "authEndpoint");
     Preconditions.checkNotNull(clientId, "clientId");
     Preconditions.checkNotNull(clientSecret, "clientSecret");
+    boolean isVersion2AuthenticationEndpoint = authEndpoint.contains("/oauth2/v2.0/");
 
     QueryParams qp = new QueryParams();
-    qp.add("resource", RESOURCE_NAME);
+    if (isVersion2AuthenticationEndpoint) {
+      qp.add("scope", SCOPE);
+    } else {
+      qp.add("resource", RESOURCE_NAME);
+    }
     qp.add("grant_type", "client_credentials");
     qp.add("client_id", clientId);
     qp.add("client_secret", clientSecret);