You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by gd...@apache.org on 2010/07/28 23:02:05 UTC
svn commit: r980225 - in /cassandra/trunk/src/java/org/apache/cassandra:
auth/AuthenticatedUser.java avro/CassandraServer.java
service/ClientState.java thrift/CassandraDaemon.java
thrift/CassandraServer.java
Author: gdusbabek
Date: Wed Jul 28 21:02:05 2010
New Revision: 980225
URL: http://svn.apache.org/viewvc?rev=980225&view=rev
Log:
move CS threadlocals into single object. patch by stuhood, reviewed by gdusbabek. CASSANDRA-1237
Added:
cassandra/trunk/src/java/org/apache/cassandra/service/ClientState.java
Modified:
cassandra/trunk/src/java/org/apache/cassandra/auth/AuthenticatedUser.java
cassandra/trunk/src/java/org/apache/cassandra/avro/CassandraServer.java
cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraDaemon.java
cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java
Modified: cassandra/trunk/src/java/org/apache/cassandra/auth/AuthenticatedUser.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/auth/AuthenticatedUser.java?rev=980225&r1=980224&r2=980225&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/auth/AuthenticatedUser.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/auth/AuthenticatedUser.java Wed Jul 28 21:02:05 2010
@@ -71,6 +71,6 @@ public class AuthenticatedUser
@Override
public String toString()
{
- return "#<User %s groups=%s>".format(username, groups);
+ return String.format("#<User %s groups=%s>", username, groups);
}
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/avro/CassandraServer.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/avro/CassandraServer.java?rev=980225&r1=980224&r2=980225&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/avro/CassandraServer.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/avro/CassandraServer.java Wed Jul 28 21:02:05 2010
@@ -56,6 +56,7 @@ import org.apache.cassandra.db.marshal.M
import org.apache.cassandra.db.migration.AddKeyspace;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.scheduler.IRequestScheduler;
+import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.StorageService;
@@ -75,22 +76,8 @@ public class CassandraServer implements
private final static String D_CF_COMPTYPE = "BytesType";
private final static String D_CF_SUBCOMPTYPE = "";
- private ThreadLocal<AccessLevel> loginDone = new ThreadLocal<AccessLevel>()
- {
- @Override
- protected AccessLevel initialValue()
- {
- return AccessLevel.NONE;
- }
- };
-
- // Session keyspace.
- private ThreadLocal<String> curKeyspace = new ThreadLocal<String>();
-
- /*
- * An associated Id for scheduling the requests
- */
- private ThreadLocal<String> requestSchedulerId = new ThreadLocal<String>();
+ // thread local state containing session information
+ private final ClientState clientState = new ClientState();
/*
* RequestScheduler to perform the scheduling of incoming requests
@@ -108,7 +95,7 @@ public class CassandraServer implements
if (logger.isDebugEnabled())
logger.debug("get");
- AvroValidation.validateColumnPath(curKeyspace.get(), columnPath);
+ AvroValidation.validateColumnPath(clientState.getKeyspace(), columnPath);
// FIXME: This is repetitive.
byte[] column, super_column;
@@ -118,7 +105,7 @@ public class CassandraServer implements
QueryPath path = new QueryPath(columnPath.column_family.toString(), column == null ? null : super_column);
List<byte[]> nameAsList = Arrays.asList(column == null ? super_column : column);
AvroValidation.validateKey(key.array());
- ReadCommand command = new SliceByNamesReadCommand(curKeyspace.get(), key.array(), path, nameAsList);
+ ReadCommand command = new SliceByNamesReadCommand(clientState.getKeyspace(), key.array(), path, nameAsList);
Map<DecoratedKey<?>, ColumnFamily> cfamilies = readColumnFamily(Arrays.asList(command), consistencyLevel);
ColumnFamily cf = cfamilies.get(StorageService.getPartitioner().decorateKey(command.key));
@@ -289,7 +276,7 @@ public class CassandraServer implements
GenericArray<ByteBuffer> keys = new GenericData.Array<ByteBuffer>(1, bytesArray);
keys.add(key);
- return multigetSliceInternal(curKeyspace.get(), keys, columnParent, predicate, consistencyLevel).iterator().next().columns;
+ return multigetSliceInternal(clientState.getKeyspace(), keys, columnParent, predicate, consistencyLevel).iterator().next().columns;
}
private GenericArray<CoscsMapEntry> multigetSliceInternal(String keyspace, GenericArray<ByteBuffer> keys,
@@ -366,7 +353,7 @@ public class CassandraServer implements
if (logger.isDebugEnabled())
logger.debug("multiget_slice");
- return multigetSliceInternal(curKeyspace.get(), keys, columnParent, predicate, consistencyLevel);
+ return multigetSliceInternal(clientState.getKeyspace(), keys, columnParent, predicate, consistencyLevel);
}
@Override
@@ -377,10 +364,10 @@ public class CassandraServer implements
logger.debug("insert");
AvroValidation.validateKey(key.array());
- AvroValidation.validateColumnParent(curKeyspace.get(), parent);
- AvroValidation.validateColumn(curKeyspace.get(), parent, column);
+ AvroValidation.validateColumnParent(clientState.getKeyspace(), parent);
+ AvroValidation.validateColumn(clientState.getKeyspace(), parent, column);
- RowMutation rm = new RowMutation(curKeyspace.get(), key.array());
+ RowMutation rm = new RowMutation(clientState.getKeyspace(), key.array());
try
{
rm.add(new QueryPath(parent.column_family.toString(),
@@ -407,10 +394,10 @@ public class CassandraServer implements
logger.debug("remove");
AvroValidation.validateKey(key.array());
- AvroValidation.validateColumnPath(curKeyspace.get(), columnPath);
+ AvroValidation.validateColumnPath(clientState.getKeyspace(), columnPath);
IClock dbClock = AvroValidation.validateClock(clock);
- RowMutation rm = new RowMutation(curKeyspace.get(), key.array());
+ RowMutation rm = new RowMutation(clientState.getKeyspace(), key.array());
byte[] superName = columnPath.super_column == null ? null : columnPath.super_column.array();
rm.delete(new QueryPath(columnPath.column_family.toString(), superName), dbClock);
@@ -474,9 +461,9 @@ public class CassandraServer implements
String cfName = cfMutations.getKey().toString();
for (Mutation mutation : cfMutations.getValue())
- AvroValidation.validateMutation(curKeyspace.get(), cfName, mutation);
+ AvroValidation.validateMutation(clientState.getKeyspace(), cfName, mutation);
}
- rowMutations.add(getRowMutationFromMutations(curKeyspace.get(), pair.key.array(), cfToMutations));
+ rowMutations.add(getRowMutationFromMutations(clientState.getKeyspace(), pair.key.array(), cfToMutations));
}
if (consistencyLevel == ConsistencyLevel.ZERO)
@@ -611,16 +598,7 @@ public class CassandraServer implements
throw newInvalidRequestException("Keyspace does not exist");
}
- // If switching, invalidate previous access level; force a new login.
- if (this.curKeyspace.get() != null && !this.curKeyspace.get().equals(keyspaceStr))
- loginDone.set(AccessLevel.NONE);
-
- this.curKeyspace.set(keyspaceStr);
-
- if (DatabaseDescriptor.getRequestSchedulerId().equals(Config.RequestSchedulerId.keyspace)) {
- requestSchedulerId.set(curKeyspace.get());
- }
-
+ clientState.setKeyspace(keyspaceStr);
return null;
}
@@ -742,7 +720,7 @@ public class CassandraServer implements
*/
private void schedule()
{
- requestScheduler.queue(Thread.currentThread(), requestSchedulerId.get());
+ requestScheduler.queue(Thread.currentThread(), clientState.getSchedulingId());
}
/**
Added: cassandra/trunk/src/java/org/apache/cassandra/service/ClientState.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/ClientState.java?rev=980225&view=auto
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/ClientState.java (added)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/ClientState.java Wed Jul 28 21:02:05 2010
@@ -0,0 +1,146 @@
+/**
+ * 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.service;
+
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.auth.AuthenticatedUser;
+import org.apache.cassandra.avro.AccessLevel;
+import org.apache.cassandra.config.Config.RequestSchedulerId;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.KSMetaData;
+import org.apache.cassandra.thrift.AuthenticationException;
+import org.apache.cassandra.thrift.AuthorizationException;
+import org.apache.cassandra.thrift.InvalidRequestException;
+
+/**
+ * A container for per-client, thread-local state that Avro/Thrift threads must hold.
+ */
+public class ClientState
+{
+ private static Logger logger = LoggerFactory.getLogger(ClientState.class);
+
+ // true if the keyspace should be used as the scheduling id
+ private final boolean SCHEDULE_ON_KEYSPACE = DatabaseDescriptor.getRequestSchedulerId().equals(RequestSchedulerId.keyspace);
+
+ // Current user for the session
+ private final ThreadLocal<AuthenticatedUser> user = new ThreadLocal<AuthenticatedUser>()
+ {
+ @Override
+ public AuthenticatedUser initialValue()
+ {
+ return DatabaseDescriptor.getAuthenticator().defaultUser();
+ }
+ };
+
+ // Keyspace and keyspace AccessLevels associated with the session
+ private final ThreadLocal<String> keyspace = new ThreadLocal<String>();
+ private final ThreadLocal<AccessLevel> keyspaceAccess = new ThreadLocal<AccessLevel>();
+
+ /**
+ * Called when the keyspace or user have changed.
+ */
+ private void updateKeyspaceAccess()
+ {
+ if (user.get() == null)
+ // user is not logged in
+ keyspaceAccess.set(null);
+ else if (user.get().isSuper)
+ // super user
+ keyspaceAccess.set(AccessLevel.FULL);
+ else if (keyspace.get() != null)
+ {
+ // lookup the access level for the user in the current keyspace
+ KSMetaData ksm = DatabaseDescriptor.getTableDefinition(keyspace.get());
+ keyspaceAccess.set(user.get().levelFor(ksm.usersAccess, ksm.groupsAccess));
+ }
+ else
+ // user is logged in, but no keyspace is set
+ keyspaceAccess.set(null);
+ }
+
+ public String getKeyspace()
+ {
+ return keyspace.get();
+ }
+
+ public void setKeyspace(String ks)
+ {
+ keyspace.set(ks);
+ updateKeyspaceAccess();
+ }
+
+ public String getSchedulingId()
+ {
+ if (SCHEDULE_ON_KEYSPACE)
+ return keyspace.get();
+ return "default";
+ }
+
+ /**
+ * Attempts to login this client with the given credentials map.
+ * TODO: Kill thrift exceptions
+ */
+ public void login(Map<String,String> credentials) throws AuthenticationException, AuthorizationException
+ {
+ AuthenticatedUser user = DatabaseDescriptor.getAuthenticator().login(credentials);
+ if (logger.isDebugEnabled())
+ logger.debug("logged in: {}", user);
+ this.user.set(user);
+ updateKeyspaceAccess();
+ }
+
+ public void logout()
+ {
+ if (logger.isDebugEnabled())
+ logger.debug("logged out: {}", user.get());
+ user.remove();
+ keyspace.remove();
+ keyspaceAccess.remove();
+ }
+
+ /**
+ * Confirms that the client thread has the given AccessLevel in the 'base' context (where Keyspace
+ * management occurs).
+ */
+ public void hasBaseAccess(AccessLevel level) throws InvalidRequestException
+ {
+ if (user.get() == null)
+ throw new InvalidRequestException("You have not logged in");
+ // FIXME: only checking for the super user until 1271 lands
+ if (!user.get().isSuper)
+ throw new InvalidRequestException("Only a 'super' user may modify keyspaces");
+ }
+
+ /**
+ * Confirms that the client thread has the given AccessLevel in the context of the current Keyspace.
+ */
+ public void hasKeyspaceAccess(AccessLevel level) throws InvalidRequestException
+ {
+ if (user.get() == null)
+ throw new InvalidRequestException("You have not logged in");
+ if (keyspaceAccess.get() == null)
+ throw new InvalidRequestException("You have not set a keyspace for this session");
+ if (keyspaceAccess.get().ordinal() < level.ordinal())
+ throw new InvalidRequestException(String.format("Your user (%s) does not have permission to perform %s operations", user, level));
+ }
+}
Modified: cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraDaemon.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraDaemon.java?rev=980225&r1=980224&r2=980225&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraDaemon.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraDaemon.java Wed Jul 28 21:02:05 2010
@@ -189,7 +189,7 @@ public class CassandraDaemon extends org
protected void afterExecute(Runnable r, Throwable t)
{
super.afterExecute(r, t);
- cassandraServer.logout();
+ cassandraServer.clientState.logout();
}
};
serverEngine = new CustomTThreadPoolServer(new TProcessorFactory(processor),
Modified: cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java?rev=980225&r1=980224&r2=980225&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java Wed Jul 28 21:02:05 2010
@@ -29,6 +29,8 @@ import org.apache.cassandra.db.migration
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+// importing avro.AccessLevel hides thrift.AccessLevel
+import org.apache.cassandra.avro.AccessLevel;
import org.apache.cassandra.auth.AllowAllAuthenticator;
import org.apache.cassandra.auth.AuthenticatedUser;
import org.apache.cassandra.concurrent.StageManager;
@@ -54,6 +56,7 @@ import org.apache.cassandra.dht.IPartiti
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.scheduler.IRequestScheduler;
+import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.StorageService;
import org.apache.thrift.TException;
@@ -66,24 +69,8 @@ public class CassandraServer implements
private final static List<ColumnOrSuperColumn> EMPTY_COLUMNS = Collections.emptyList();
private final static List<Column> EMPTY_SUBCOLUMNS = Collections.emptyList();
- // will be set only by login()
- private ThreadLocal<AuthenticatedUser> loginDone = new ThreadLocal<AuthenticatedUser>() {
- @Override
- public AuthenticatedUser initialValue()
- {
- return DatabaseDescriptor.getAuthenticator().defaultUser();
- }
- };
-
- /*
- * Keyspace associated with session
- */
- private ThreadLocal<String> keySpace = new ThreadLocal<String>();
-
- /*
- * An associated Id for scheduling the requests
- */
- private ThreadLocal<String> requestSchedulerId = new ThreadLocal<String>();
+ // thread local state containing session information
+ public final ClientState clientState = new ClientState();
/*
* RequestScheduler to perform the scheduling of incoming requests
@@ -226,12 +213,12 @@ public class CassandraServer implements
return thrift_clock;
}
- private static Map<String,org.apache.cassandra.avro.AccessLevel> unthriftifyAccessMap(Map<String,AccessLevel> map)
+ private static Map<String,AccessLevel> unthriftifyAccessMap(Map<String,org.apache.cassandra.thrift.AccessLevel> map)
{
- Map<String,org.apache.cassandra.avro.AccessLevel> out = new HashMap<String,org.apache.cassandra.avro.AccessLevel>();
+ Map<String,AccessLevel> out = new HashMap<String,AccessLevel>();
if (map == null)
return out;
- for (Map.Entry<String,AccessLevel> entry : map.entrySet())
+ for (Map.Entry<String,org.apache.cassandra.thrift.AccessLevel> entry : map.entrySet())
out.put(entry.getKey(), Enum.valueOf(org.apache.cassandra.avro.AccessLevel.class, entry.getValue().name()));
return out;
}
@@ -277,8 +264,8 @@ public class CassandraServer implements
if (logger.isDebugEnabled())
logger.debug("get_slice");
- checkKeyspaceAndLoginAuthorized(AccessLevel.READONLY);
- return multigetSliceInternal(keySpace.get(), Arrays.asList(key), column_parent, predicate, consistency_level).get(key);
+ clientState.hasKeyspaceAccess(AccessLevel.READONLY);
+ return multigetSliceInternal(clientState.getKeyspace(), Arrays.asList(key), column_parent, predicate, consistency_level).get(key);
}
public Map<byte[], List<ColumnOrSuperColumn>> multiget_slice(List<byte[]> keys, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
@@ -287,9 +274,9 @@ public class CassandraServer implements
if (logger.isDebugEnabled())
logger.debug("multiget_slice");
- checkKeyspaceAndLoginAuthorized(AccessLevel.READONLY);
+ clientState.hasKeyspaceAccess(AccessLevel.READONLY);
- return multigetSliceInternal(keySpace.get(), keys, column_parent, predicate, consistency_level);
+ return multigetSliceInternal(clientState.getKeyspace(), keys, column_parent, predicate, consistency_level);
}
private Map<byte[], List<ColumnOrSuperColumn>> multigetSliceInternal(String keyspace, List<byte[]> keys, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
@@ -326,8 +313,8 @@ public class CassandraServer implements
if (logger.isDebugEnabled())
logger.debug("get");
- checkKeyspaceAndLoginAuthorized(AccessLevel.READONLY);
- String keyspace = keySpace.get();
+ clientState.hasKeyspaceAccess(AccessLevel.READONLY);
+ String keyspace = clientState.getKeyspace();
ThriftValidation.validateColumnPath(keyspace, column_path);
@@ -354,7 +341,7 @@ public class CassandraServer implements
if (logger.isDebugEnabled())
logger.debug("get_count");
- checkKeyspaceAndLoginAuthorized(AccessLevel.READONLY);
+ clientState.hasKeyspaceAccess(AccessLevel.READONLY);
return get_slice(key, column_parent, predicate, consistency_level).size();
}
@@ -365,7 +352,7 @@ public class CassandraServer implements
if (logger.isDebugEnabled())
logger.debug("multiget_count");
- checkKeyspaceAndLoginAuthorized(AccessLevel.READONLY);
+ clientState.hasKeyspaceAccess(AccessLevel.READONLY);
Map<byte[], Integer> counts = new HashMap<byte[], Integer>();
Map<byte[], List<ColumnOrSuperColumn>> columnFamiliesMap = multigetSliceInternal(table, keys, column_parent, predicate, consistency_level);
@@ -382,14 +369,14 @@ public class CassandraServer implements
if (logger.isDebugEnabled())
logger.debug("insert");
- checkKeyspaceAndLoginAuthorized(AccessLevel.READWRITE);
+ clientState.hasKeyspaceAccess(AccessLevel.READWRITE);
ThriftValidation.validateKey(key);
- ThriftValidation.validateColumnParent(keySpace.get(), column_parent);
- ThriftValidation.validateColumn(keySpace.get(), column_parent, column);
+ ThriftValidation.validateColumnParent(clientState.getKeyspace(), column_parent);
+ ThriftValidation.validateColumn(clientState.getKeyspace(), column_parent, column);
IClock cassandra_clock = ThriftValidation.validateClock(column.clock);
- RowMutation rm = new RowMutation(keySpace.get(), key);
+ RowMutation rm = new RowMutation(clientState.getKeyspace(), key);
try
{
rm.add(new QueryPath(column_parent.column_family, column_parent.super_column, column.name), column.value, cassandra_clock, column.ttl);
@@ -425,7 +412,7 @@ public class CassandraServer implements
}
}
- checkKeyspaceAndLoginAuthorized(needed);
+ clientState.hasKeyspaceAccess(needed);
List<RowMutation> rowMutations = new ArrayList<RowMutation>();
for (Map.Entry<byte[], Map<String, List<Mutation>>> mutationEntry: mutation_map.entrySet())
@@ -440,10 +427,10 @@ public class CassandraServer implements
for (Mutation mutation : columnFamilyMutations.getValue())
{
- ThriftValidation.validateMutation(keySpace.get(), cfName, mutation);
+ ThriftValidation.validateMutation(clientState.getKeyspace(), cfName, mutation);
}
}
- rowMutations.add(RowMutation.getRowMutationFromMutations(keySpace.get(), key, columnFamilyToMutations));
+ rowMutations.add(RowMutation.getRowMutationFromMutations(clientState.getKeyspace(), key, columnFamilyToMutations));
}
doInsert(consistency_level, rowMutations);
@@ -455,14 +442,14 @@ public class CassandraServer implements
if (logger.isDebugEnabled())
logger.debug("remove");
- checkKeyspaceAndLoginAuthorized(AccessLevel.FULL);
+ clientState.hasKeyspaceAccess(AccessLevel.FULL);
ThriftValidation.validateKey(key);
- ThriftValidation.validateColumnPathOrParent(keySpace.get(), column_path);
+ ThriftValidation.validateColumnPathOrParent(clientState.getKeyspace(), column_path);
IClock cassandra_clock = ThriftValidation.validateClock(clock);
- RowMutation rm = new RowMutation(keySpace.get(), key);
+ RowMutation rm = new RowMutation(clientState.getKeyspace(), key);
rm.delete(new QueryPath(column_path), cassandra_clock);
doInsert(consistency_level, Arrays.asList(rm));
@@ -530,8 +517,8 @@ public class CassandraServer implements
if (logger.isDebugEnabled())
logger.debug("range_slice");
- String keyspace = keySpace.get();
- checkKeyspaceAndLoginAuthorized(AccessLevel.READONLY);
+ String keyspace = clientState.getKeyspace();
+ clientState.hasKeyspaceAccess(AccessLevel.READONLY);
return getRangeSlicesInternal(keyspace, column_parent, range, predicate, consistency_level);
}
@@ -600,11 +587,11 @@ public class CassandraServer implements
if (logger.isDebugEnabled())
logger.debug("scan");
- checkKeyspaceAndLoginAuthorized(AccessLevel.READONLY);
+ clientState.hasKeyspaceAccess(AccessLevel.READONLY);
if (row_predicate.keys != null)
{
- Map<byte[], List<ColumnOrSuperColumn>> rowMap = multigetSliceInternal(keySpace.get(), row_predicate.keys, column_parent, column_predicate, consistency_level);
+ Map<byte[], List<ColumnOrSuperColumn>> rowMap = multigetSliceInternal(clientState.getKeyspace(), row_predicate.keys, column_parent, column_predicate, consistency_level);
List<KeySlice> rows = new ArrayList<KeySlice>(rowMap.size());
for (Map.Entry<byte[], List<ColumnOrSuperColumn>> entry : rowMap.entrySet())
{
@@ -615,12 +602,12 @@ public class CassandraServer implements
if (row_predicate.key_range != null)
{
- return getRangeSlicesInternal(keySpace.get(), column_parent, row_predicate.key_range, column_predicate, consistency_level);
+ return getRangeSlicesInternal(clientState.getKeyspace(), column_parent, row_predicate.key_range, column_predicate, consistency_level);
}
if (row_predicate.index_clause != null)
{
- return scanIndexInternal(keySpace.get(), column_parent, row_predicate.index_clause, column_predicate, consistency_level);
+ return scanIndexInternal(clientState.getKeyspace(), column_parent, row_predicate.index_clause, column_predicate, consistency_level);
}
throw new InvalidRequestException("row predicate must specify keys, key_range, or index_clause");
@@ -709,30 +696,7 @@ public class CassandraServer implements
public void login(AuthenticationRequest auth_request) throws AuthenticationException, AuthorizationException, TException
{
- AuthenticatedUser user = DatabaseDescriptor.getAuthenticator().login(auth_request.getCredentials());
-
- if (logger.isDebugEnabled())
- logger.debug("login confirmed; user is " + user);
-
- loginDone.set(user);
- }
-
- public void logout()
- {
- keySpace.remove();
- loginDone.remove();
-
- if (logger.isDebugEnabled())
- logger.debug("logout complete");
- }
-
- protected void checkKeyspaceAndLoginAuthorized(AccessLevel level) throws InvalidRequestException
- {
- if (loginDone.get() == null)
- throw new InvalidRequestException("You have not logged in");
-
- // FIXME: if no keyspace set, check against global authlist. otherwise, check
- // against keyspace authlist
+ clientState.login(auth_request.getCredentials());
}
/**
@@ -740,7 +704,7 @@ public class CassandraServer implements
*/
private void schedule()
{
- requestScheduler.queue(Thread.currentThread(), requestSchedulerId.get());
+ requestScheduler.queue(Thread.currentThread(), clientState.getSchedulingId());
}
/**
@@ -792,7 +756,8 @@ public class CassandraServer implements
public String system_add_column_family(CfDef cf_def) throws InvalidRequestException, TException
{
- checkKeyspaceAndLoginAuthorized(AccessLevel.FULL);
+ clientState.hasBaseAccess(AccessLevel.FULL);
+
try
{
applyMigrationOnStage(new AddColumnFamily(convertToCFMetaData(cf_def)));
@@ -814,11 +779,11 @@ public class CassandraServer implements
public String system_drop_column_family(String column_family) throws InvalidRequestException, TException
{
- checkKeyspaceAndLoginAuthorized(AccessLevel.FULL);
+ clientState.hasBaseAccess(AccessLevel.FULL);
try
{
- applyMigrationOnStage(new DropColumnFamily(keySpace.get(), column_family, true));
+ applyMigrationOnStage(new DropColumnFamily(clientState.getKeyspace(), column_family, true));
return DatabaseDescriptor.getDefsVersion().toString();
}
catch (ConfigurationException e)
@@ -837,11 +802,11 @@ public class CassandraServer implements
public String system_rename_column_family(String old_name, String new_name) throws InvalidRequestException, TException
{
- checkKeyspaceAndLoginAuthorized(AccessLevel.FULL);
+ clientState.hasBaseAccess(AccessLevel.FULL);
try
{
- applyMigrationOnStage(new RenameColumnFamily(keySpace.get(), old_name, new_name));
+ applyMigrationOnStage(new RenameColumnFamily(clientState.getKeyspace(), old_name, new_name));
return DatabaseDescriptor.getDefsVersion().toString();
}
catch (ConfigurationException e)
@@ -907,7 +872,7 @@ public class CassandraServer implements
public String system_drop_keyspace(String keyspace) throws InvalidRequestException, TException
{
- checkKeyspaceAndLoginAuthorized(AccessLevel.FULL);
+ clientState.hasBaseAccess(AccessLevel.FULL);
try
{
@@ -930,7 +895,7 @@ public class CassandraServer implements
public String system_rename_keyspace(String old_name, String new_name) throws InvalidRequestException, TException
{
- checkKeyspaceAndLoginAuthorized(AccessLevel.FULL);
+ clientState.hasBaseAccess(AccessLevel.FULL);
try
{
@@ -993,12 +958,12 @@ public class CassandraServer implements
public void truncate(String cfname) throws InvalidRequestException, UnavailableException, TException
{
- logger.debug("truncating {} in {}", cfname, keySpace.get());
- checkKeyspaceAndLoginAuthorized(AccessLevel.FULL);
+ logger.debug("truncating {} in {}", cfname, clientState.getKeyspace());
+ clientState.hasKeyspaceAccess(AccessLevel.READWRITE);
try
{
schedule();
- StorageProxy.truncateBlocking(keySpace.get(), cfname);
+ StorageProxy.truncateBlocking(clientState.getKeyspace(), cfname);
}
catch (TimeoutException e)
{
@@ -1021,8 +986,7 @@ public class CassandraServer implements
throw new InvalidRequestException("Keyspace does not exist");
}
- keySpace.set(keyspace);
- requestSchedulerId.set(keyspace);
+ clientState.setKeyspace(keyspace);
}
public Map<String, List<String>> check_schema_agreement() throws TException, InvalidRequestException