natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-2.0' into cassandra-2.1 

Conflicts:
	CHANGES.txt
	src/java/org/apache/cassandra/auth/Auth.java
	src/java/org/apache/cassandra/service/ClientState.java
This commit is contained in:
Aleksey Yeschenko 2015-01-09 04:08:05 +03:00
parent 5674a96d4c e750ab238e
commit 8c5a959c97
8 changed files with 138 additions and 89 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -50,6 +50,8 @@
* Log failed host when preparing incremental repair (CASSANDRA-8228)
* Force config client mode in CQLSSTableWriter (CASSANDRA-8281)
Merged from 2.0:
* Introduce background cache refreshing to permissions cache
(CASSANDRA-8194)
* Fix race condition in StreamTransferTask that could lead to
infinite loops and premature sstable deletion (CASSANDRA-7704)
* Add an extra version check to MigrationTask (CASSANDRA-8462)

8
conf/cassandra.yaml
View File

@ -79,6 +79,14 @@ authorizer: AllowAllAuthorizer
# Will be disabled automatically for AllowAllAuthorizer.
permissions_validity_in_ms: 2000
# Refresh interval for permissions cache (if enabled).
# After this interval, cache entries become eligible for refresh. Upon next
# access, an async reload is scheduled and the old value returned until it
# completes. If permissions_validity_in_ms is non-zero, then this must be
# also.
# Defaults to the same value as permissions_validity_in_ms.
# permissions_update_interval_in_ms: 1000
# The partitioner is responsible for distributing groups of rows (by
# partition key) across nodes in the cluster. You should leave this
# alone for new clusters. The partitioner can NOT be changed without

55
src/java/org/apache/cassandra/auth/Auth.java
View File

@ -20,9 +20,6 @@ package org.apache.cassandra.auth;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import org.apache.commons.lang3.StringUtils;
@ -34,9 +31,9 @@ import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.cql3.statements.CFStatement;
import org.apache.cassandra.cql3.statements.CreateTableStatement;
import org.apache.cassandra.cql3.statements.SelectStatement;
@ -47,9 +44,8 @@ import org.apache.cassandra.locator.SimpleStrategy;
import org.apache.cassandra.service.*;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.Pair;
public class Auth implements AuthMBean
public class Auth
{
private static final Logger logger = LoggerFactory.getLogger(Auth.class);
@ -61,8 +57,10 @@ public class Auth implements AuthMBean
public static final String USERS_CF = "users";
// User-level permissions cache.
public static volatile LoadingCache<Pair<AuthenticatedUser, IResource>, Set<Permission>> permissionsCache = initPermissionsCache(null);
private static final PermissionsCache permissionsCache = new PermissionsCache(DatabaseDescriptor.getPermissionsValidity(),
DatabaseDescriptor.getPermissionsUpdateInterval(),
DatabaseDescriptor.getPermissionsCacheMaxEntries(),
DatabaseDescriptor.getAuthorizer());
private static final String USERS_CF_SCHEMA = String.format("CREATE TABLE %s.%s ("
+ "name text,"
@ -75,44 +73,9 @@ public class Auth implements AuthMBean
private static SelectStatement selectUserStatement;
public int getPermissionsValidity()
public static Set<Permission> getPermissions(AuthenticatedUser user, IResource resource)
{
return DatabaseDescriptor.getPermissionsValidity();
}
public void setPermissionsValidity(int timeoutInMs)
{
DatabaseDescriptor.setPermissionsValidity(timeoutInMs);
permissionsCache = initPermissionsCache(permissionsCache);
}
public void invalidatePermissionsCache()
{
permissionsCache = initPermissionsCache(null);
}
private static LoadingCache<Pair<AuthenticatedUser, IResource>, Set<Permission>> initPermissionsCache(LoadingCache<Pair<AuthenticatedUser, IResource>, Set<Permission>> oldCache)
{
if (DatabaseDescriptor.getAuthorizer() instanceof AllowAllAuthorizer)
return null;
int validityPeriod = DatabaseDescriptor.getPermissionsValidity();
if (validityPeriod <= 0)
return null;
LoadingCache<Pair<AuthenticatedUser, IResource>, Set<Permission>> newCache =
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);
}
});
if (oldCache != null)
newCache.putAll(oldCache.asMap());
return newCache;
return permissionsCache.getPermissions(user, resource);
}
/**

27
src/java/org/apache/cassandra/auth/AuthMBean.java
View File

@ -1,27 +0,0 @@
/*
* 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.cassandra.auth;
public interface AuthMBean
{
public int getPermissionsValidity();
public void setPermissionsValidity(int timeoutInMs);
public void invalidatePermissionsCache();
}

108
src/java/org/apache/cassandra/auth/PermissionsCache.java Normal file
View File

@ -0,0 +1,108 @@
/*
* 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.cassandra.auth;
import java.util.Set;
import java.util.concurrent.*;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListenableFutureTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
import org.apache.cassandra.utils.Pair;
public class PermissionsCache
{
private static final Logger logger = LoggerFactory.getLogger(PermissionsCache.class);
private final ThreadPoolExecutor cacheRefreshExecutor = new DebuggableThreadPoolExecutor("PermissionsCacheRefresh",
Thread.NORM_PRIORITY);
private final IAuthorizer authorizer;
private final LoadingCache<Pair<AuthenticatedUser, IResource>, Set<Permission>> cache;
public PermissionsCache(int validityPeriod, int updateInterval, int maxEntries, IAuthorizer authorizer)
{
this.authorizer = authorizer;
this.cache = initCache(validityPeriod, updateInterval, maxEntries);
}
public Set<Permission> getPermissions(AuthenticatedUser user, IResource resource)
{
if (cache == null)
return authorizer.authorize(user, resource);
try
{
return cache.get(Pair.create(user, resource));
}
catch (ExecutionException e)
{
throw new RuntimeException(e);
}
}
private LoadingCache<Pair<AuthenticatedUser, IResource>, Set<Permission>> initCache(int validityPeriod,
int updateInterval,
int maxEntries)
{
if (authorizer instanceof AllowAllAuthorizer)
return null;
if (validityPeriod <= 0)
return null;
return CacheBuilder.newBuilder()
.refreshAfterWrite(updateInterval, TimeUnit.MILLISECONDS)
.expireAfterWrite(validityPeriod, TimeUnit.MILLISECONDS)
.maximumSize(maxEntries)
.build(new CacheLoader<Pair<AuthenticatedUser, IResource>, Set<Permission>>()
{
public Set<Permission> load(Pair<AuthenticatedUser, IResource> userResource)
{
return authorizer.authorize(userResource.left, userResource.right);
}
public ListenableFuture<Set<Permission>> reload(final Pair<AuthenticatedUser, IResource> userResource,
final Set<Permission> oldValue)
{
ListenableFutureTask<Set<Permission>> task = ListenableFutureTask.create(new Callable<Set<Permission>>()
{
public Set<Permission>call() throws Exception
{
try
{
return authorizer.authorize(userResource.left, userResource.right);
}
catch (Exception e)
{
logger.debug("Error performing async refresh of user permissions", e);
throw e;
}
}
});
cacheRefreshExecutor.execute(task);
return task;
}
});
}
}

2
src/java/org/apache/cassandra/config/Config.java
View File

@ -43,6 +43,8 @@ public class Config
public String authenticator;
public String authorizer;
public int permissions_validity_in_ms = 2000;
public int permissions_cache_max_entries = 1000;
public int permissions_update_interval_in_ms = -1;
/* Hashing strategy Random or OPHF */
public String partitioner;

10
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -700,11 +700,17 @@ public class DatabaseDescriptor
return conf.permissions_validity_in_ms;
}
public static void setPermissionsValidity(int timeout)
public static int getPermissionsCacheMaxEntries()
{
conf.permissions_validity_in_ms = timeout;
return conf.permissions_cache_max_entries;
}
public static int getPermissionsUpdateInterval()
{
return conf.permissions_update_interval_in_ms == -1
? conf.permissions_validity_in_ms
: conf.permissions_update_interval_in_ms;
}
public static int getThriftFramedTransportSize()
{

15
src/java/org/apache/cassandra/service/ClientState.java
View File

@ -19,7 +19,6 @@ package org.apache.cassandra.service;
import java.net.SocketAddress;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicLong;
import com.google.common.collect.Iterables;
@ -42,7 +41,6 @@ import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.thrift.ThriftValidation;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.SemanticVersion;
/**
@ -351,17 +349,6 @@ public class ClientState
private Set<Permission> authorize(IResource resource)
{
// AllowAllAuthorizer or manually disabled caching.
if (Auth.permissionsCache == null)
return DatabaseDescriptor.getAuthorizer().authorize(user, resource);
try
{
return Auth.permissionsCache.get(Pair.create(user, resource));
}
catch (ExecutionException e)
{
throw new RuntimeException(e);
}
return Auth.getPermissions(user, resource);
}
}