mirror of https://github.com/apache/cassandra
move CS threadlocals into single object. patch by stuhood, reviewed by gdusbabek. CASSANDRA-1237
git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@980225 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Dusbabek
parent
3f9f50725b
commit
da2d195919
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -56,6 +56,7 @@ import org.apache.cassandra.db.marshal.MarshalException;
|
|||
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 Cassandra {
|
|||
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 Cassandra {
|
|||
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 Cassandra {
|
|||
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 Cassandra {
|
|||
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 Cassandra {
|
|||
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 Cassandra {
|
|||
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 Cassandra {
|
|||
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 Cassandra {
|
|||
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 Cassandra {
|
|||
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 Cassandra {
|
|||
*/
|
||||
private void schedule()
|
||||
{
|
||||
requestScheduler.queue(Thread.currentThread(), requestSchedulerId.get());
|
||||
requestScheduler.queue(Thread.currentThread(), clientState.getSchedulingId());
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -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));
|
||||
}
|
||||
}
|
||||
|
|
@ -189,7 +189,7 @@ public class CassandraDaemon extends org.apache.cassandra.service.AbstractCassan
|
|||
protected void afterExecute(Runnable r, Throwable t)
|
||||
{
|
||||
super.afterExecute(r, t);
|
||||
cassandraServer.logout();
|
||||
cassandraServer.clientState.logout();
|
||||
}
|
||||
};
|
||||
serverEngine = new CustomTThreadPoolServer(new TProcessorFactory(processor),
|
||||
|
|
|
|||
|
|
@ -29,6 +29,8 @@ import org.apache.cassandra.db.migration.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.IPartitioner;
|
|||
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 Cassandra.Iface
|
|||
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 Cassandra.Iface
|
|||
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 Cassandra.Iface
|
|||
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 Cassandra.Iface
|
|||
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 Cassandra.Iface
|
|||
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 Cassandra.Iface
|
|||
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 Cassandra.Iface
|
|||
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 Cassandra.Iface
|
|||
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 Cassandra.Iface
|
|||
}
|
||||
}
|
||||
|
||||
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 Cassandra.Iface
|
|||
|
||||
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 Cassandra.Iface
|
|||
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 Cassandra.Iface
|
|||
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 Cassandra.Iface
|
|||
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 Cassandra.Iface
|
|||
|
||||
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 Cassandra.Iface
|
|||
|
||||
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 Cassandra.Iface
|
|||
*/
|
||||
private void schedule()
|
||||
{
|
||||
requestScheduler.queue(Thread.currentThread(), requestSchedulerId.get());
|
||||
requestScheduler.queue(Thread.currentThread(), clientState.getSchedulingId());
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -792,7 +756,8 @@ public class CassandraServer implements Cassandra.Iface
|
|||
|
||||
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 Cassandra.Iface
|
|||
|
||||
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 Cassandra.Iface
|
|||
|
||||
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 Cassandra.Iface
|
|||
|
||||
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 Cassandra.Iface
|
|||
|
||||
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 Cassandra.Iface
|
|||
|
||||
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 Cassandra.Iface
|
|||
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
|
||||
|
|
|
|||
Loading…
Reference in New Issue