HADOOP-14642. wasb: add support for caching Authorization and SASKeys. Contributed by Sivaguru Sankaridurg.

hadoop-common-project/hadoop-common/src/main/resources/core-default.xml

@@ -1343,7 +1343,14 @@
     configuration
   </description>
 </property>
-
+<property>
+  <name>fs.azure.authorization.caching.enable</name>
+  <value>true</value>
+  <description>
+    Config flag to enable caching of authorization results and saskeys in WASB.
+    This flag is relevant only when fs.azure.authorization is enabled.
+  </description>
+</property>
 
 <property>
   <name>io.seqfile.compress.blocksize</name>

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/conf/TestCommonConfigurationFields.java

@@ -115,6 +115,7 @@ public class TestCommonConfigurationFields extends TestConfigurationFieldsBase {
     xmlPropsToSkipCompare.add("fs.azure.local.sas.key.mode");
     xmlPropsToSkipCompare.add("fs.azure.secure.mode");
     xmlPropsToSkipCompare.add("fs.azure.authorization");
+    xmlPropsToSkipCompare.add("fs.azure.authorization.caching.enable");
 
     // Deprecated properties.  These should eventually be removed from the
     // class.

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

@@ -0,0 +1,232 @@
+/**
+ * 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.azure;
+import com.google.common.cache.Cache;
+import org.apache.hadoop.conf.Configuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.cache.CacheBuilder;
+
+/**
+ * Class that provides caching for Authorize and getSasUri calls
+ * @param <K> -  The cache key type
+ * @param <V> - The cached value type
+ */
+public class CachingAuthorizer<K, V> {
+
+  public static final Logger LOG = LoggerFactory
+      .getLogger(CachingAuthorizer.class);
+
+  private Cache<K, V> cache;
+  private boolean isEnabled = false;
+  private long cacheEntryExpiryPeriodInMinutes;
+  private String label;
+
+  public static final String KEY_AUTH_SERVICE_CACHING_ENABLE =
+      "fs.azure.authorization.caching.enable";
+
+  public static final boolean KEY_AUTH_SERVICE_CACHING_ENABLE_DEFAULT = false;
+
+  public static final String KEY_AUTH_SERVICE_CACHING_MAX_ENTRIES =
+      "fs.azure.authorization.caching.maxentries";
+
+  public static final int KEY_AUTH_SERVICE_CACHING_MAX_ENTRIES_DEFAULT = 512;
+
+  public CachingAuthorizer(long ttlInMinutes, String label) {
+    cacheEntryExpiryPeriodInMinutes = ttlInMinutes;
+    this.label = label;
+    if (cacheEntryExpiryPeriodInMinutes <= 0) {
+      isEnabled = false;
+    }
+  }
+
+
+  public void init(Configuration conf) {
+
+    isEnabled = conf.getBoolean(KEY_AUTH_SERVICE_CACHING_ENABLE, KEY_AUTH_SERVICE_CACHING_ENABLE_DEFAULT);
+
+    if (isEnabled) {
+      LOG.debug("{} : Initializing CachingAuthorizer instance", label);
+      cache = CacheBuilder.newBuilder()
+          .maximumSize(
+              conf.getInt(
+                  KEY_AUTH_SERVICE_CACHING_MAX_ENTRIES,
+                  KEY_AUTH_SERVICE_CACHING_MAX_ENTRIES_DEFAULT
+              )
+          )
+          .expireAfterWrite(cacheEntryExpiryPeriodInMinutes, TimeUnit.MINUTES)
+          .build();
+    }
+  }
+
+  /**
+   * @param key - Cache key
+   * @return null on cache-miss. true/false on cache-hit
+   */
+  public V get(K key) {
+    if (!isEnabled) {
+      return null;
+    }
+
+    V result = cache.getIfPresent(key);
+    if (result == null) {
+      LOG.debug("{}: CACHE MISS: {}", label, key.toString());
+    }
+    else {
+      LOG.debug("{}: CACHE HIT: {}, {}", label, key.toString(), result.toString());
+    }
+    return result;
+  }
+
+  public void put(K key, V value) {
+    if (isEnabled) {
+      LOG.debug("{}: CACHE PUT: {}, {}", label, key.toString(), value.toString());
+      cache.put(key, value);
+    }
+  }
+
+  public void clear() {
+    if (isEnabled) {
+      cache.invalidateAll();
+    }
+  }
+}
+
+/**
+ * POJO representing the cache key for authorization calls
+ */
+class CachedAuthorizerEntry {
+
+  private String path;
+  private String accessType;
+  private String owner;
+
+  CachedAuthorizerEntry(String path, String accessType, String owner) {
+    this.path = path;
+    this.accessType = accessType;
+    this.owner = owner;
+  }
+
+  public String getPath() {
+    return path;
+  }
+
+  public String getAccessType() {
+    return accessType;
+  }
+
+  public String getOwner() {
+    return owner;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (o == this) {
+      return true;
+    }
+
+    if (o == null) {
+      return false;
+    }
+
+    if (!(o instanceof CachedAuthorizerEntry)) {
+      return false;
+    }
+
+    CachedAuthorizerEntry c = (CachedAuthorizerEntry) o;
+    return
+      this.getPath().equals(c.getPath())
+      && this.getAccessType().equals(c.getAccessType())
+      && this.getOwner().equals(c.getOwner());
+  }
+
+  @Override
+  public int hashCode() {
+    return this.toString().hashCode();
+  }
+
+  @Override
+  public String toString() {
+    return path + ":" + accessType + ":" + owner;
+  }
+
+}
+
+
+/**
+ * POJO representing the cache key for sas-key calls
+ */
+class CachedSASKeyEntry {
+
+  private String storageAccount;
+  private String container;
+  private String path;
+
+  CachedSASKeyEntry(String storageAccount, String container, String path) {
+    this.storageAccount = storageAccount;
+    this.container = container;
+    this.path = path;
+  }
+
+  public String getStorageAccount() {
+    return storageAccount;
+  }
+
+  public String getContainer() {
+    return container;
+  }
+
+  public String getPath() {
+    return path;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (o == this) {
+      return true;
+    }
+
+    if (o == null) {
+      return false;
+    }
+
+    if (!(o instanceof CachedSASKeyEntry)) {
+      return false;
+    }
+
+    CachedSASKeyEntry c = (CachedSASKeyEntry) o;
+    return
+      this.getStorageAccount().equals(c.getStorageAccount())
+      && this.getContainer().equals(c.getContainer())
+      && this.getPath().equals(c.getPath());
+  }
+
+  @Override
+  public int hashCode() {
+    return this.toString().hashCode();
+  }
+
+  @Override
+  public String toString() {
+    return storageAccount + ":" + container + ":" + path;
+  }
+}

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

@@ -58,11 +58,14 @@ public class LocalSASKeyGeneratorImpl extends SASKeyGeneratorImpl {
    * Map to cache CloudStorageAccount instances.
    */
   private Map<String, CloudStorageAccount> storageAccountMap;
-
+  private CachingAuthorizer<CachedSASKeyEntry, URI> cache;
   private static final int HOURS_IN_DAY = 24;
+
   public LocalSASKeyGeneratorImpl(Configuration conf) {
     super(conf);
     storageAccountMap = new HashMap<String, CloudStorageAccount>();
+    cache = new CachingAuthorizer<>(getSasKeyExpiryPeriod(), "SASKEY");
+    cache.init(conf);
   }
 
   /**
@@ -74,11 +77,19 @@ public class LocalSASKeyGeneratorImpl extends SASKeyGeneratorImpl {
 
     try {
 
+      CachedSASKeyEntry cacheKey = new CachedSASKeyEntry(accountName, container, "/");
+      URI cacheResult = cache.get(cacheKey);
+      if (cacheResult != null) {
+        return cacheResult;
+      }
+
       CloudStorageAccount account =
           getSASKeyBasedStorageAccountInstance(accountName);
       CloudBlobClient client = account.createCloudBlobClient();
-      return client.getCredentials().transformUri(
+      URI sasKey = client.getCredentials().transformUri(
           client.getContainerReference(container).getUri());
+      cache.put(cacheKey, sasKey);
+      return sasKey;
 
     } catch (StorageException stoEx) {
       throw new SASKeyGenerationException("Encountered StorageException while"
@@ -146,7 +157,16 @@ public class LocalSASKeyGeneratorImpl extends SASKeyGeneratorImpl {
 
     CloudBlobContainer sc = null;
     CloudBlobClient client = null;
+    CachedSASKeyEntry cacheKey = null;
+
     try {
+
+      cacheKey = new CachedSASKeyEntry(accountName, container, relativePath);
+      URI cacheResult = cache.get(cacheKey);
+      if (cacheResult != null) {
+        return cacheResult;
+      }
+
       CloudStorageAccount account =
           getSASKeyBasedStorageAccountInstance(accountName);
       client = account.createCloudBlobClient();
@@ -175,7 +195,9 @@ public class LocalSASKeyGeneratorImpl extends SASKeyGeneratorImpl {
     }
 
     try {
-      return client.getCredentials().transformUri(blob.getUri());
+      URI sasKey = client.getCredentials().transformUri(blob.getUri());
+      cache.put(cacheKey, sasKey);
+      return sasKey;
     } catch (StorageException stoEx) {
       throw new SASKeyGenerationException("Encountered StorageException while "
           + "generating SAS key for Blob: " + relativePath + " inside "

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

@@ -2113,9 +2113,6 @@ public class NativeAzureFileSystem extends FileSystem {
 
     // Capture the absolute path and the path to key.
     Path absolutePath = makeAbsolute(f);
-
-    performAuthCheck(absolutePath, WasbAuthorizationOperations.READ, "getFileStatus", absolutePath);
-
     String key = pathToKey(absolutePath);
     if (key.length() == 0) { // root always exists
       return newDirectory(null, absolutePath);

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

@@ -22,6 +22,7 @@ import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 import com.fasterxml.jackson.databind.ObjectReader;
 import org.apache.hadoop.conf.Configuration;
@@ -106,12 +107,21 @@ public class RemoteSASKeyGeneratorImpl extends SASKeyGeneratorImpl {
   private static final String
       SAS_KEY_GENERATOR_HTTP_CLIENT_RETRY_POLICY_SPEC_DEFAULT =
       "10,3,100,2";
+  /**
+   * Saskey caching period
+   */
+  private static final String SASKEY_CACHEENTRY_EXPIRY_PERIOD =
+      "fs.azure.saskey.cacheentry.expiry.period";
 
   private WasbRemoteCallHelper remoteCallHelper = null;
   private boolean isKerberosSupportEnabled;
   private boolean isSpnegoTokenCacheEnabled;
   private RetryPolicy retryPolicy;
   private String[] commaSeparatedUrls;
+  private CachingAuthorizer<CachedSASKeyEntry, URI> cache;
+
+  private static final int HOURS_IN_DAY = 24;
+  private static final int MINUTES_IN_HOUR = 60;
 
   public RemoteSASKeyGeneratorImpl(Configuration conf) {
     super(conf);
@@ -141,6 +151,18 @@ public class RemoteSASKeyGeneratorImpl extends SASKeyGeneratorImpl {
     } else {
       this.remoteCallHelper = new WasbRemoteCallHelper(retryPolicy);
     }
+
+    /* Expire the cache entry five minutes before the actual saskey expiry, so that we never encounter a case
+     * where a stale sas-key-entry is picked up from the cache; which is expired on use.
+     */
+    long sasKeyExpiryPeriodInMinutes = getSasKeyExpiryPeriod() * HOURS_IN_DAY * MINUTES_IN_HOUR; // sas-expiry is in days, convert into mins
+    long cacheEntryDurationInMinutes =
+        conf.getTimeDuration(SASKEY_CACHEENTRY_EXPIRY_PERIOD, sasKeyExpiryPeriodInMinutes, TimeUnit.MINUTES);
+    cacheEntryDurationInMinutes = (cacheEntryDurationInMinutes > (sasKeyExpiryPeriodInMinutes - 5))
+        ? (sasKeyExpiryPeriodInMinutes - 5)
+        : cacheEntryDurationInMinutes;
+    this.cache = new CachingAuthorizer<>(cacheEntryDurationInMinutes, "SASKEY");
+    this.cache.init(conf);
     LOG.debug("Initialization of RemoteSASKeyGenerator instance successful");
   }
 
@@ -149,6 +171,13 @@ public class RemoteSASKeyGeneratorImpl extends SASKeyGeneratorImpl {
       String container) throws SASKeyGenerationException {
     RemoteSASKeyGenerationResponse sasKeyResponse = null;
     try {
+      CachedSASKeyEntry cacheKey = new CachedSASKeyEntry(storageAccount, container, "/");
+      URI cacheResult = cache.get(cacheKey);
+      if (cacheResult != null) {
+        return cacheResult;
+      }
+
+      LOG.debug("Generating Container SAS Key: Storage Account {}, Container {}", storageAccount, container);
       URIBuilder uriBuilder = new URIBuilder();
       uriBuilder.setPath("/" + CONTAINER_SAS_OP);
       uriBuilder.addParameter(STORAGE_ACCOUNT_QUERY_PARAM_NAME, storageAccount);
@@ -160,7 +189,9 @@ public class RemoteSASKeyGeneratorImpl extends SASKeyGeneratorImpl {
               uriBuilder.getQueryParams());
 
       if (sasKeyResponse.getResponseCode() == REMOTE_CALL_SUCCESS_CODE) {
-        return new URI(sasKeyResponse.getSasKey());
+        URI sasKey = new URI(sasKeyResponse.getSasKey());
+        cache.put(cacheKey, sasKey);
+        return sasKey;
       } else {
         throw new SASKeyGenerationException(
             "Remote Service encountered error in SAS Key generation : "
@@ -178,6 +209,15 @@ public class RemoteSASKeyGeneratorImpl extends SASKeyGeneratorImpl {
       String container, String relativePath) throws SASKeyGenerationException {
 
     try {
+      CachedSASKeyEntry cacheKey = new CachedSASKeyEntry(storageAccount, container, relativePath);
+      URI cacheResult = cache.get(cacheKey);
+      if (cacheResult != null) {
+        return cacheResult;
+      }
+
+      LOG.debug("Generating RelativePath SAS Key for relativePath {} inside Container {} inside Storage Account {}",
+          relativePath, container, storageAccount);
+
       URIBuilder uriBuilder = new URIBuilder();
       uriBuilder.setPath("/" + BLOB_SAS_OP);
       uriBuilder.addParameter(STORAGE_ACCOUNT_QUERY_PARAM_NAME, storageAccount);
@@ -190,7 +230,9 @@ public class RemoteSASKeyGeneratorImpl extends SASKeyGeneratorImpl {
           makeRemoteRequest(commaSeparatedUrls, uriBuilder.getPath(),
               uriBuilder.getQueryParams());
       if (sasKeyResponse.getResponseCode() == REMOTE_CALL_SUCCESS_CODE) {
-        return new URI(sasKeyResponse.getSasKey());
+        URI sasKey = new URI(sasKeyResponse.getSasKey());
+        cache.put(cacheKey, sasKey);
+        return sasKey;
       } else {
         throw new SASKeyGenerationException(
             "Remote Service encountered error in SAS Key generation : "

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

@@ -33,6 +33,7 @@ import org.apache.http.client.methods.HttpGet;
 import org.apache.http.client.utils.URIBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import java.util.concurrent.TimeUnit;
 
 import java.io.IOException;
 
@@ -95,11 +96,18 @@ public class RemoteWasbAuthorizerImpl implements WasbAuthorizerInterface {
   private static final String AUTHORIZER_HTTP_CLIENT_RETRY_POLICY_SPEC_DEFAULT =
       "10,3,100,2";
 
+  /**
+   * Authorization caching period
+   */
+  private static final String AUTHORIZATION_CACHEENTRY_EXPIRY_PERIOD =
+      "fs.azure.authorization.cacheentry.expiry.period";
+
   private WasbRemoteCallHelper remoteCallHelper = null;
   private boolean isKerberosSupportEnabled;
   private boolean isSpnegoTokenCacheEnabled;
   private RetryPolicy retryPolicy;
   private String[] commaSeparatedUrls = null;
+  private CachingAuthorizer<CachedAuthorizerEntry, Boolean> cache;
 
   @VisibleForTesting public void updateWasbRemoteCallHelper(
       WasbRemoteCallHelper helper) {
@@ -108,7 +116,7 @@ public class RemoteWasbAuthorizerImpl implements WasbAuthorizerInterface {
 
   @Override
   public void init(Configuration conf)
-      throws WasbAuthorizationException, IOException {
+      throws IOException {
     LOG.debug("Initializing RemoteWasbAuthorizerImpl instance");
     this.isKerberosSupportEnabled =
         conf.getBoolean(Constants.AZURE_KERBEROS_SUPPORT_PROPERTY_NAME, false);
@@ -131,14 +139,38 @@ public class RemoteWasbAuthorizerImpl implements WasbAuthorizerInterface {
     } else {
       this.remoteCallHelper = new WasbRemoteCallHelper(retryPolicy);
     }
+
+    this.cache = new CachingAuthorizer<>(
+        conf.getTimeDuration(AUTHORIZATION_CACHEENTRY_EXPIRY_PERIOD, 5L, TimeUnit.MINUTES), "AUTHORIZATION"
+    );
+    this.cache.init(conf);
   }
 
   @Override
   public boolean authorize(String wasbAbsolutePath, String accessType, String resourceOwner)
-      throws WasbAuthorizationException, IOException {
+      throws IOException {
+
+    /* Make an exception for the internal -RenamePending files */
+    if (wasbAbsolutePath.endsWith(NativeAzureFileSystem.FolderRenamePending.SUFFIX)) {
+      return true;
+    }
+
+    CachedAuthorizerEntry cacheKey = new CachedAuthorizerEntry(wasbAbsolutePath, accessType, resourceOwner);
+    Boolean cacheresult = cache.get(cacheKey);
+    if (cacheresult != null) {
+      return cacheresult;
+    }
+
+    boolean authorizeresult = authorizeInternal(wasbAbsolutePath, accessType, resourceOwner);
+    cache.put(cacheKey, authorizeresult);
+
+    return authorizeresult;
+  }
+
+  private boolean authorizeInternal(String wasbAbsolutePath, String accessType, String resourceOwner)
+          throws IOException {
 
     try {
-        /* Make an exception for the internal -RenamePending files */
       final URIBuilder uriBuilder = new URIBuilder();
       uriBuilder.setPath("/" + CHECK_AUTHORIZATION_OP);
       uriBuilder

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

@@ -38,7 +38,7 @@ public abstract class SASKeyGeneratorImpl implements SASKeyGeneratorInterface {
   /**
    * Default value for the SAS key expiry period in days. {@value}
    */
-  public static final long DEFAUL_CONTAINER_SAS_KEY_PERIOD = 90;
+  public static final long DEFAULT_CONTAINER_SAS_KEY_PERIOD = 90;
 
   private long sasKeyExpiryPeriod;
 
@@ -47,7 +47,7 @@ public abstract class SASKeyGeneratorImpl implements SASKeyGeneratorInterface {
   public SASKeyGeneratorImpl(Configuration conf) {
     this.conf = conf;
     this.sasKeyExpiryPeriod = conf.getTimeDuration(
-        KEY_SAS_KEY_EXPIRY_PERIOD, DEFAUL_CONTAINER_SAS_KEY_PERIOD,
+        KEY_SAS_KEY_EXPIRY_PERIOD, DEFAULT_CONTAINER_SAS_KEY_PERIOD,
         TimeUnit.DAYS);
   }
 

hadoop-tools/hadoop-azure/src/site/markdown/index.md

@@ -425,6 +425,44 @@ value takes a comma seperated list of user names who are allowed to perform chow
 </property>
 ```
 
+Caching of both SAS keys and Authorization responses can be enabled using the following setting:
+The cache settings are applicable only when fs.azure.authorization is enabled.
+The cache is maintained at a filesystem object level.
+```
+    <property>
+      <name>fs.azure.authorization.caching.enable</name>
+      <value>true</value>
+    </property>
+```
+
+The maximum number of entries that that cache can hold can be customized using the following setting:
+```
+    <property>
+      <name>fs.azure.authorization.caching.maxentries</name>
+      <value>512</value>
+    </property>
+```
+
+ The validity of an authorization cache-entry can be controlled using the following setting:
+ Setting the value to zero disables authorization-caching.
+ If the key is not specified, a default expiry duration of 5m takes effect.
+ ```
+    <property>
+      <name>fs.azure.authorization.cacheentry.expiry.period</name>
+      <value>5m</value>
+    </property>
+```
+
+ The validity of a SASKey cache-entry can be controlled using the following setting.
+ Setting the value to zero disables SASKey-caching.
+ If the key is not specified, the default expiry duration specified in the sas-key request takes effect.
+ ```
+    <property>
+      <name>fs.azure.saskey.cacheentry.expiry.period</name>
+      <value>90d</value>
+    </property>
+```
+
 ## Testing the hadoop-azure Module
 
 The hadoop-azure module includes a full suite of unit tests.  Most of the tests

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

@@ -21,6 +21,7 @@ package org.apache.hadoop.fs.azure;
 import static org.junit.Assume.assumeNotNull;
 
 import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.conf.Configuration;
 import org.junit.After;
 import org.junit.Before;
 import org.slf4j.Logger;
@@ -60,6 +61,10 @@ public abstract class AbstractWasbTestBase {
     }
   }
 
+  public Configuration getConfiguration() {
+    return new Configuration();
+  }
+
   protected abstract AzureBlobStorageTestAccount createTestAccount()
       throws Exception;
 }

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

@@ -20,6 +20,7 @@ package org.apache.hadoop.fs.azure;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import java.util.regex.Pattern;
 
 import org.apache.hadoop.security.UserGroupInformation;
@@ -35,6 +36,7 @@ public class MockWasbAuthorizerImpl implements WasbAuthorizerInterface {
 
   private Map<AuthorizationComponent, Boolean> authRules;
   private boolean performOwnerMatch;
+  private CachingAuthorizer<CachedAuthorizerEntry, Boolean> cache;
 
  // The full qualified URL to the root directory
   private String qualifiedPrefixUrl;
@@ -42,6 +44,7 @@ public class MockWasbAuthorizerImpl implements WasbAuthorizerInterface {
   public MockWasbAuthorizerImpl(NativeAzureFileSystem fs) {
     qualifiedPrefixUrl = new Path("/").makeQualified(fs.getUri(), fs.getWorkingDirectory())
         .toString().replaceAll("/$", "");
+    cache = new CachingAuthorizer<>(TimeUnit.MINUTES.convert(5L, TimeUnit.MINUTES), "AUTHORIZATION");
   }
 
   @Override
@@ -54,7 +57,8 @@ public class MockWasbAuthorizerImpl implements WasbAuthorizerInterface {
   if currentUserShortName is set to a string that is not empty
   */
   public void init(Configuration conf, boolean matchOwner) {
-    authRules = new HashMap<AuthorizationComponent, Boolean>();
+    cache.init(conf);
+    authRules = new HashMap<>();
     this.performOwnerMatch = matchOwner;
   }
 
@@ -76,6 +80,21 @@ public class MockWasbAuthorizerImpl implements WasbAuthorizerInterface {
       return true;
     }
 
+    CachedAuthorizerEntry cacheKey = new CachedAuthorizerEntry(wasbAbsolutePath, accessType, owner);
+    Boolean cacheresult = cache.get(cacheKey);
+    if (cacheresult != null) {
+      return cacheresult;
+    }
+
+    boolean authorizeresult = authorizeInternal(wasbAbsolutePath, accessType, owner);
+    cache.put(cacheKey, authorizeresult);
+
+    return authorizeresult;
+  }
+
+  private boolean authorizeInternal(String wasbAbsolutePath, String accessType, String owner)
+      throws WasbAuthorizationException {
+
     String currentUserShortName = "";
     if (this.performOwnerMatch) {
       try {
@@ -120,6 +139,7 @@ public class MockWasbAuthorizerImpl implements WasbAuthorizerInterface {
 
   public void deleteAllAuthRules() {
     authRules.clear();
+    cache.clear();
   }
 }
 

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

@@ -0,0 +1,60 @@
+/**
+ * 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.azure;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.contract.ContractTestUtils;
+import org.junit.Test;
+
+import static org.apache.hadoop.fs.azure.CachingAuthorizer.KEY_AUTH_SERVICE_CACHING_ENABLE;
+
+/**
+ * Test class to hold all WASB authorization caching related tests.
+ */
+public class TestNativeAzureFSAuthorizationCaching
+    extends TestNativeAzureFileSystemAuthorizationWithOwner {
+
+  private static final int DUMMY_TTL_VALUE = 5000;
+
+  @Override
+  public Configuration getConfiguration() {
+    Configuration conf = super.getConfiguration();
+    conf.set(KEY_AUTH_SERVICE_CACHING_ENABLE, "true");
+    return conf;
+  }
+
+  @Override
+  protected AzureBlobStorageTestAccount createTestAccount() throws Exception {
+    Configuration conf = getConfiguration();
+    return AzureBlobStorageTestAccount.create(conf);
+  }
+
+  /**
+   * Test to verify cache behavior -- assert that PUT overwrites value if present
+   */
+  @Test
+  public void testCachePut() throws Throwable {
+    CachingAuthorizer<String, Integer> cache = new CachingAuthorizer<>(DUMMY_TTL_VALUE, "TEST");
+    cache.init(getConfiguration());
+    cache.put("TEST", 1);
+    cache.put("TEST", 3);
+    int result = cache.get("TEST");
+    ContractTestUtils.assertTrue("Cache returned unexpected result", result == 3);
+  }
+}

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

@@ -49,11 +49,17 @@ public class TestNativeAzureFileSystemAuthorization
   protected MockWasbAuthorizerImpl authorizer;
 
   @Override
-  protected AzureBlobStorageTestAccount createTestAccount() throws Exception {
-    Configuration conf = new Configuration();
+  public Configuration getConfiguration() {
+    Configuration conf = super.getConfiguration();
     conf.set(NativeAzureFileSystem.KEY_AZURE_AUTHORIZATION, "true");
     conf.set(RemoteWasbAuthorizerImpl.KEY_REMOTE_AUTH_SERVICE_URLS, "http://localhost/");
     conf.set(NativeAzureFileSystem.AZURE_CHOWN_USERLIST_PROPERTY_NAME, "user1 , user2");
+    return conf;
+  }
+
+  @Override
+  protected AzureBlobStorageTestAccount createTestAccount() throws Exception {
+    Configuration conf = getConfiguration();
     return AzureBlobStorageTestAccount.create(conf);
   }
 
@@ -66,7 +72,8 @@ public class TestNativeAzureFileSystemAuthorization
         useSecureMode && useAuthorization);
 
     authorizer = new MockWasbAuthorizerImpl(fs);
-    authorizer.init(null);
+    authorizer.init(fs.getConf());
+    fs.updateWasbAuthorizer(authorizer);
   }
 
 
@@ -109,7 +116,6 @@ public class TestNativeAzureFileSystemAuthorization
     Path testPath = new Path(parentDir, "test.dat");
 
     authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), true);
-    authorizer.addAuthRule(testPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -135,7 +141,6 @@ public class TestNativeAzureFileSystemAuthorization
     Path testPath = new Path(parentDir, "test.dat");
 
     authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), true);
-    authorizer.addAuthRule(testPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -163,18 +168,14 @@ public class TestNativeAzureFileSystemAuthorization
     setExpectedFailureMessage("create", testPath);
 
     authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), true);
-    authorizer.addAuthRule(testPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
     fs.updateWasbAuthorizer(authorizer);
 
-    boolean initialCreateSucceeded = false;
     try {
       fs.create(testPath);
       ContractTestUtils.assertPathExists(fs, "testPath was not created", testPath);
-      initialCreateSucceeded = true;
       fs.create(testPath, true);
     }
     finally {
-      ContractTestUtils.assertTrue(initialCreateSucceeded);
       fs.delete(testPath, false);
     }
   }
@@ -191,19 +192,15 @@ public class TestNativeAzureFileSystemAuthorization
     Path testPath = new Path(parentDir, "test.dat");
 
     authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), true);
-    authorizer.addAuthRule(testPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
     authorizer.addAuthRule(testPath.toString(), WasbAuthorizationOperations.WRITE.toString(), true);
     fs.updateWasbAuthorizer(authorizer);
 
-    boolean initialCreateSucceeded = false;
     try {
       fs.create(testPath);
       ContractTestUtils.assertPathExists(fs, "testPath was not created", testPath);
-      initialCreateSucceeded = true;
       fs.create(testPath, true);
     }
     finally {
-      ContractTestUtils.assertTrue(initialCreateSucceeded);
       fs.delete(testPath, false);
     }
   }
@@ -299,8 +296,6 @@ public class TestNativeAzureFileSystemAuthorization
 
     authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), true); /* to create parentDir */
     authorizer.addAuthRule(parentDir.toString(), WasbAuthorizationOperations.WRITE.toString(), true); /* for rename */
-    authorizer.addAuthRule(srcPath.toString(), WasbAuthorizationOperations.READ.toString(), true); /* for exists */
-    authorizer.addAuthRule(dstPath.toString(), WasbAuthorizationOperations.READ.toString(), true); /* for exists */
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -331,8 +326,6 @@ public class TestNativeAzureFileSystemAuthorization
 
     authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), true); /* to create parent dir */
     authorizer.addAuthRule(parentDir.toString(), WasbAuthorizationOperations.WRITE.toString(), false);
-    authorizer.addAuthRule(srcPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
-    authorizer.addAuthRule(dstPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -365,8 +358,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), true); /* to create parent dir */
     authorizer.addAuthRule(parentSrcDir.toString(), WasbAuthorizationOperations.WRITE.toString(), true);
     authorizer.addAuthRule(parentDstDir.toString(), WasbAuthorizationOperations.WRITE.toString(), false);
-    authorizer.addAuthRule(srcPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
-    authorizer.addAuthRule(dstPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -396,8 +387,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), true); /* to create parent dirs */
     authorizer.addAuthRule(parentSrcDir.toString(), WasbAuthorizationOperations.WRITE.toString(), true);
     authorizer.addAuthRule(parentDstDir.toString(), WasbAuthorizationOperations.WRITE.toString(), true);
-    authorizer.addAuthRule(srcPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
-    authorizer.addAuthRule(dstPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -505,7 +494,6 @@ public class TestNativeAzureFileSystemAuthorization
     Path testPath = new Path(parentDir, "test.dat");
 
     authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), true);
-    authorizer.addAuthRule(testPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
     fs.updateWasbAuthorizer(authorizer);
     try {
       fs.create(testPath);
@@ -530,7 +518,6 @@ public class TestNativeAzureFileSystemAuthorization
     setExpectedFailureMessage("delete", testPath);
 
     authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), true);
-    authorizer.addAuthRule(testPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
     fs.updateWasbAuthorizer(authorizer);
     try {
       fs.create(testPath);
@@ -548,7 +535,6 @@ public class TestNativeAzureFileSystemAuthorization
       /* Restore permissions to force a successful delete */
       authorizer.deleteAllAuthRules();
       authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), true);
-      authorizer.addAuthRule(testPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
       fs.updateWasbAuthorizer(authorizer);
 
       fs.delete(testPath, false);
@@ -570,7 +556,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), true); // for create and delete
     authorizer.addAuthRule("/testDeleteIntermediateFolder*",
         WasbAuthorizationOperations.WRITE.toString(), true); // for recursive delete
-    authorizer.addAuthRule("/*", WasbAuthorizationOperations.READ.toString(), true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -586,34 +571,13 @@ public class TestNativeAzureFileSystemAuthorization
   }
 
   /**
-   * Positive test for getFileStatus
+   * Positive test for getFileStatus. No permissions are required for getting filestatus.
    * @throws Throwable
    */
   @Test
   public void testGetFileStatusPositive() throws Throwable {
 
     Path testPath = new Path("/");
-
-    authorizer.addAuthRule("/", WasbAuthorizationOperations.READ.toString(), true);
-    fs.updateWasbAuthorizer(authorizer);
-
-    ContractTestUtils.assertIsDirectory(fs, testPath);
-  }
-
-  /**
-   * Negative test for getFileStatus
-   * @throws Throwable
-   */
-  @Test //(expected=WasbAuthorizationException.class)
-  public void testGetFileStatusNegative() throws Throwable {
-
-    Path testPath = new Path("/");
-
-    setExpectedFailureMessage("getFileStatus", testPath);
-
-    authorizer.addAuthRule("/", WasbAuthorizationOperations.READ.toString(), false);
-    fs.updateWasbAuthorizer(authorizer);
-
     ContractTestUtils.assertIsDirectory(fs, testPath);
   }
 
@@ -627,7 +591,6 @@ public class TestNativeAzureFileSystemAuthorization
     Path testPath = new Path("/testMkdirsAccessCheckPositive/1/2/3");
 
     authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), true);
-    authorizer.addAuthRule(testPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -652,7 +615,6 @@ public class TestNativeAzureFileSystemAuthorization
     setExpectedFailureMessage("mkdirs", testPath);
 
     authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), false);
-    authorizer.addAuthRule(testPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -686,13 +648,12 @@ public class TestNativeAzureFileSystemAuthorization
    */
   @Test
   public void testSetOwnerThrowsForUnauthorisedUsers() throws Throwable {
+
     expectedEx.expect(WasbAuthorizationException.class);
 
     Path testPath = new Path("/testSetOwnerNegative");
-    MockWasbAuthorizerImpl authorizer = new MockWasbAuthorizerImpl(fs);
-    authorizer.init(null);
+
     authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), true);
-    authorizer.addAuthRule(testPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
     fs.updateWasbAuthorizer(authorizer);
 
     String owner = null;
@@ -723,11 +684,10 @@ public class TestNativeAzureFileSystemAuthorization
    * */
   @Test
   public void testSetOwnerSucceedsForAuthorisedUsers() throws Throwable {
-    Path testPath = new Path("/testsetownerpositive");
-    MockWasbAuthorizerImpl authorizer = new MockWasbAuthorizerImpl(fs);
-    authorizer.init(null);
+
+    Path testPath = new Path("/testSetOwnerPositive");
+
     authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), true);
-    authorizer.addAuthRule(testPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
     fs.updateWasbAuthorizer(authorizer);
 
     String newOwner = "newowner";
@@ -765,14 +725,14 @@ public class TestNativeAzureFileSystemAuthorization
    * */
   @Test
   public void testSetOwnerSucceedsForAnyUserWhenWildCardIsSpecified() throws Throwable {
+
     Configuration conf = fs.getConf();
     conf.set(NativeAzureFileSystem.AZURE_CHOWN_USERLIST_PROPERTY_NAME, "*");
-    Path testPath = new Path("/testsetownerpositivewildcard");
+    fs.setConf(conf);
+    Path testPath = new Path("/testSetOwnerPositiveWildcard");
 
-    MockWasbAuthorizerImpl authorizer = new MockWasbAuthorizerImpl(fs);
-    authorizer.init(null);
+    authorizer.init(conf);
     authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), true);
-    authorizer.addAuthRule(testPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
     fs.updateWasbAuthorizer(authorizer);
 
     String newOwner = "newowner";
@@ -809,16 +769,16 @@ public class TestNativeAzureFileSystemAuthorization
    */
   @Test
   public void testSetOwnerFailsForIllegalSetup() throws Throwable {
+
     expectedEx.expect(IllegalArgumentException.class);
 
     Configuration conf = fs.getConf();
     conf.set(NativeAzureFileSystem.AZURE_CHOWN_USERLIST_PROPERTY_NAME, "user1, *");
+    fs.setConf(conf);
     Path testPath = new Path("/testSetOwnerFailsForIllegalSetup");
 
-    MockWasbAuthorizerImpl authorizer = new MockWasbAuthorizerImpl(fs);
-    authorizer.init(null);
+    authorizer.init(conf);
     authorizer.addAuthRule("/", WasbAuthorizationOperations.WRITE.toString(), true);
-    authorizer.addAuthRule(testPath.toString(), WasbAuthorizationOperations.READ.toString(), true);
     fs.updateWasbAuthorizer(authorizer);
 
     String owner = null;

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

@@ -36,7 +36,7 @@ public class TestNativeAzureFileSystemAuthorizationWithOwner
   @Before
   public void beforeMethod() {
     super.beforeMethod();
-    authorizer.init(null, true);
+    authorizer.init(fs.getConf(), true);
   }
 
   /**

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

@@ -344,7 +344,7 @@ public class TestWasbRemoteCallHelper
     Mockito.when(mockHttpClient.execute(argThat(new HttpGetForServiceLocal())))
         .thenReturn(mockHttpResponseServiceLocal);
 
-    //Need 3 times because performop()  does 3 fs operations.
+    //Need 2 times because performop()  does 2 fs operations.
     Mockito.when(mockHttpEntity.getContent())
         .thenReturn(new ByteArrayInputStream(validJsonResponse()
             .getBytes(StandardCharsets.UTF_8)))
@@ -356,8 +356,8 @@ public class TestWasbRemoteCallHelper
 
     performop(mockHttpClient);
 
-    Mockito.verify(mockHttpClient, times(3)).execute(Mockito.argThat(new HttpGetForServiceLocal()));
-    Mockito.verify(mockHttpClient, times(3)).execute(Mockito.argThat(new HttpGetForService2()));
+    Mockito.verify(mockHttpClient, times(2)).execute(Mockito.argThat(new HttpGetForServiceLocal()));
+    Mockito.verify(mockHttpClient, times(2)).execute(Mockito.argThat(new HttpGetForService2()));
   }
 
   @Test

hadoop-tools/hadoop-azure/src/test/resources/azure-test.xml

@@ -31,10 +31,9 @@
 
   <property>
     <name>fs.azure.secure.mode</name>
-    <value>false</value>
+    <value>true</value>
   </property>
 
-
   <!-- Save the above configuration properties in a separate file named -->
   <!-- azure-auth-keys.xml in the same directory as this file. -->
   <!-- DO NOT ADD azure-auth-keys.xml TO REVISION CONTROL.  The keys to your -->