Implement caching of authorization results

Patch by Aleksey Yeschenko; reviewed by Jonathan Ellis for
CASSANDRA-4295

CHANGES.txt

@@ -15,6 +15,7 @@
  * add UseCondCardMark XX jvm settings on jdk 1.7 (CASSANDRA-4366)
  * CQL3 refactor to allow conversion function (CASSANDRA-5226)
  * Fix drop of sstables in some circumstance (CASSANDRA-5232)
+ * Implement caching of authorization results (CASSANDRA-4295)
 
 
 1.2.1

conf/cassandra.yaml

@@ -60,6 +60,11 @@ authenticator: org.apache.cassandra.auth.AllowAllAuthenticator
 # authorization backend, implementing IAuthorizer; used to limit access/provide permissions
 authorizer: org.apache.cassandra.auth.AllowAllAuthorizer
 
+# Validity period for permissions cache (fetching permissions can be an
+# expensive operation depending on the authorizer). Defaults to 2000,
+# set to 0 to disable. Will be disabled automatically for AllowAllAuthorizer.
+permissions_validity_in_ms: 2000
+
 # The partitioner is responsible for distributing rows (by key) across
 # nodes in the cluster.  Any IPartitioner may be used, including your
 # own as long as it is on the classpath.  Out of the box, Cassandra

src/java/org/apache/cassandra/auth/AuthenticatedUser.java

@@ -17,6 +17,8 @@
  */
 package org.apache.cassandra.auth;
 
+import com.google.common.base.Objects;
+
 /**
  * Returned from IAuthenticator#authenticate(), represents an authenticated user everywhere internally.
  */
@@ -61,4 +63,24 @@ public class AuthenticatedUser
     {
         return String.format("#<User %s>", name);
     }
+
+    @Override
+    public boolean equals(Object o)
+    {
+        if (this == o)
+            return true;
+
+        if (!(o instanceof AuthenticatedUser))
+            return false;
+
+        AuthenticatedUser u = (AuthenticatedUser) o;
+
+        return Objects.equal(this.name, u.name);
+    }
+
+    @Override
+    public int hashCode()
+    {
+        return Objects.hashCode(name);
+    }
 }

src/java/org/apache/cassandra/config/Config.java

@@ -33,6 +33,7 @@ public class Config
     public String authenticator;
     public String authority; // for backwards compatibility - will log a warning.
     public String authorizer;
+    public int permissions_validity_in_ms = 2000;
 
     /* Hashing strategy Random or OPHF */
     public String partitioner;

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -604,6 +604,11 @@ public class DatabaseDescriptor
         return authorizer;
     }
 
+    public static int getPermissionsValidity()
+    {
+        return conf.permissions_validity_in_ms;
+    }
+
     public static int getThriftMaxMessageLength()
     {
         return conf.thrift_max_message_length_in_mb * 1024 * 1024;

src/java/org/apache/cassandra/cql3/statements/BatchStatement.java

@@ -64,23 +64,8 @@ public class BatchStatement extends ModificationStatement
     @Override
     public void checkAccess(ClientState state) throws InvalidRequestException, UnauthorizedException
     {
-        Map<String, Set<String>> cfamsSeen = new HashMap<String, Set<String>>();
         for (ModificationStatement statement : statements)
-        {
-            String ks = statement.keyspace();
-            String cf = statement.columnFamily();
-
-            if (!cfamsSeen.containsKey(ks))
-                cfamsSeen.put(ks, new HashSet<String>());
-
-            // Avoid unnecessary authorization.
-            Set<String> cfs = cfamsSeen.get(ks);
-            if (!(cfs.contains(cf)))
-            {
-                state.hasColumnFamilyAccess(ks, cf, Permission.MODIFY);
-                cfs.add(cf);
-            }
-        }
+            state.hasColumnFamilyAccess(statement.keyspace(), statement.columnFamily(), Permission.MODIFY);
     }
 
     public void validate(ClientState state) throws InvalidRequestException

src/java/org/apache/cassandra/service/ClientState.java

@@ -18,7 +18,12 @@
 package org.apache.cassandra.service;
 
 import java.util.*;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
 
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -31,6 +36,7 @@ import org.apache.cassandra.db.Table;
 import org.apache.cassandra.exceptions.AuthenticationException;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.exceptions.UnauthorizedException;
+import org.apache.cassandra.utils.Pair;
 import org.apache.cassandra.utils.SemanticVersion;
 
 /**
@@ -44,6 +50,9 @@ public class ClientState
     private static final Set<IResource> READABLE_SYSTEM_RESOURCES = new HashSet<IResource>(5);
     private static final Set<IResource> PROTECTED_AUTH_RESOURCES = new HashSet<IResource>();
 
+    // User-level permissions cache.
+    private static final LoadingCache<Pair<AuthenticatedUser, IResource>, Set<Permission>> permissionsCache = initPermissionsCache();
+
     static
     {
         // We want these system cfs to be always readable since many tools rely on them (nodetool, cqlsh, bulkloader, etc.)
@@ -239,8 +248,39 @@ public class ClientState
         return new SemanticVersion[]{ cql, cql3 };
     }
 
+    private static LoadingCache<Pair<AuthenticatedUser, IResource>, Set<Permission>> initPermissionsCache()
+    {
+        if (DatabaseDescriptor.getAuthorizer() instanceof AllowAllAuthorizer)
+            return null;
+
+        int validityPeriod = DatabaseDescriptor.getPermissionsValidity();
+        if (validityPeriod <= 0)
+            return null;
+
+        return CacheBuilder.newBuilder().expireAfterWrite(validityPeriod, TimeUnit.MILLISECONDS)
+                                        .build(new CacheLoader<Pair<AuthenticatedUser, IResource>, Set<Permission>>()
+                                        {
+                                            public Set<Permission> load(Pair<AuthenticatedUser, IResource> userResource)
+                                            {
+                                                return DatabaseDescriptor.getAuthorizer().authorize(userResource.left,
+                                                                                                    userResource.right);
+                                            }
+                                        });
+    }
+
     private Set<Permission> authorize(IResource resource)
     {
-        return DatabaseDescriptor.getAuthorizer().authorize(user, resource);
+        // AllowAllAuthorizer or manually disabled caching.
+        if (permissionsCache == null)
+            return DatabaseDescriptor.getAuthorizer().authorize(user, resource);
+
+        try
+        {
+            return permissionsCache.get(Pair.create(user, resource));
+        }
+        catch (ExecutionException e)
+        {
+            throw new RuntimeException(e);
+        }
     }
 }

src/java/org/apache/cassandra/thrift/CassandraServer.java

@@ -659,7 +659,6 @@ public class CassandraServer implements Cassandra.Iface
                                                boolean allowCounterMutations)
     throws RequestValidationException
     {
-        List<String> cfamsSeen = new ArrayList<String>();
         List<IMutation> rowMutations = new ArrayList<IMutation>();
         ThriftClientState cState = state();
         String keyspace = cState.getKeyspace();
@@ -678,12 +677,7 @@ public class CassandraServer implements Cassandra.Iface
             {
                 String cfName = columnFamilyMutations.getKey();
 
-                // Avoid unneeded authorizations
-                if (!(cfamsSeen.contains(cfName)))
-                {
-                    cState.hasColumnFamilyAccess(keyspace, cfName, Permission.MODIFY);
-                    cfamsSeen.add(cfName);
-                }
+                cState.hasColumnFamilyAccess(keyspace, cfName, Permission.MODIFY);
 
                 CFMetaData metadata = ThriftValidation.validateColumnFamily(keyspace, cfName);
                 ThriftValidation.validateKey(metadata, key);