You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jb...@apache.org on 2012/08/29 20:18:20 UTC
[1/3] git commit: add request tracing patch by David Alves;
reviewed by jbellis for CASSANDRA-1123
Updated Branches:
refs/heads/trunk 501d7134f -> 5c94432b2
add request tracing
patch by David Alves; reviewed by jbellis for CASSANDRA-1123
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/5c94432b
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/5c94432b
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/5c94432b
Branch: refs/heads/trunk
Commit: 5c94432b239547ef3ee8326e9543200f67a2ed2f
Parents: 501d713
Author: Jonathan Ellis <jb...@apache.org>
Authored: Wed Aug 29 11:55:37 2012 -0500
Committer: Jonathan Ellis <jb...@apache.org>
Committed: Wed Aug 29 13:14:35 2012 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
NEWS.txt | 2 +
conf/log4j-server.properties | 10 +-
interface/cassandra.thrift | 4 +
.../org/apache/cassandra/thrift/Cassandra.java | 598 +++++++++++-
.../apache/cassandra/cache/AutoSavingCache.java | 3 +-
src/java/org/apache/cassandra/cli/Cli.g | 32 +-
src/java/org/apache/cassandra/cli/CliClient.java | 29 +-
.../concurrent/DebuggableThreadPoolExecutor.java | 109 ++-
.../org/apache/cassandra/concurrent/Stage.java | 2 +
.../apache/cassandra/concurrent/StageManager.java | 60 +-
.../org/apache/cassandra/config/CFMetaData.java | 48 +-
.../cassandra/config/DatabaseDescriptor.java | 12 +-
.../org/apache/cassandra/config/KSMetaData.java | 14 +-
src/java/org/apache/cassandra/config/Schema.java | 30 +-
.../org/apache/cassandra/db/ColumnFamilyStore.java | 34 +-
src/java/org/apache/cassandra/db/DefsTable.java | 6 +-
src/java/org/apache/cassandra/db/Directories.java | 2 +-
.../apache/cassandra/db/HintedHandOffManager.java | 12 +-
src/java/org/apache/cassandra/db/RowMutation.java | 2 +-
src/java/org/apache/cassandra/db/SystemTable.java | 28 +-
src/java/org/apache/cassandra/db/Table.java | 34 +-
src/java/org/apache/cassandra/net/MessageOut.java | 15 +-
.../org/apache/cassandra/net/MessagingService.java | 81 +-
.../apache/cassandra/service/CassandraDaemon.java | 4 +-
.../org/apache/cassandra/service/ClientState.java | 38 +-
.../apache/cassandra/service/MigrationManager.java | 2 +-
.../org/apache/cassandra/service/StorageProxy.java | 2 +-
.../apache/cassandra/service/StorageService.java | 25 +-
.../cassandra/service/StorageServiceMBean.java | 14 +
.../apache/cassandra/thrift/CassandraServer.java | 802 ++++++++++-----
.../apache/cassandra/thrift/ThriftValidation.java | 2 +-
src/java/org/apache/cassandra/tools/NodeCmd.java | 14 +-
src/java/org/apache/cassandra/tools/NodeProbe.java | 5 +
.../org/apache/cassandra/tracing/TraceState.java | 64 ++
src/java/org/apache/cassandra/tracing/Tracing.java | 256 +++++
.../apache/cassandra/tracing/TracingAppender.java | 65 ++
.../org/apache/cassandra/utils/ByteBufferUtil.java | 12 +
.../apache/cassandra/config/CFMetaDataTest.java | 5 +-
.../unit/org/apache/cassandra/config/DefsTest.java | 4 +-
.../service/StorageServiceServerTest.java | 3 +-
.../src/org/apache/cassandra/stress/Session.java | 2 +
42 files changed, 2090 insertions(+), 397 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 9e97872..271f5b7 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
1.2-dev
+ * debug tracing (CASSANDRA-1123)
* parallelize row cache loading (CASSANDRA-4282)
* Make compaction, flush JBOD-aware (CASSANDRA-4292)
* run local range scans on the read stage (CASSANDRA-3687)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/NEWS.txt
----------------------------------------------------------------------
diff --git a/NEWS.txt b/NEWS.txt
index 7c64af2..56bb1dc 100644
--- a/NEWS.txt
+++ b/NEWS.txt
@@ -53,6 +53,8 @@ Features
Also specifying initial_token will override any num_tokens setting.
- disk_failure_policy allows blacklisting failed disks in JBOD
configuration instead of erroring out indefinitely
+ - event tracing can be configured per-connection ("trace_next_query")
+ or globally/probabilistically ("nodetool settraceprobability")
1.1.3
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/conf/log4j-server.properties
----------------------------------------------------------------------
diff --git a/conf/log4j-server.properties b/conf/log4j-server.properties
index 086306e..6c9d15a 100644
--- a/conf/log4j-server.properties
+++ b/conf/log4j-server.properties
@@ -18,12 +18,13 @@
# (%l is slower.)
# output messages into a rolling log file as well as stdout
-log4j.rootLogger=INFO,stdout,R
+log4j.rootLogger=DEBUG,stdout,R,tracing
# stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%5p %d{HH:mm:ss,SSS} %m%n
+log4j.appender.stdout.Threshold=INFO
# rolling log file
log4j.appender.R=org.apache.log4j.RollingFileAppender
@@ -33,6 +34,13 @@ log4j.appender.R.layout=org.apache.log4j.PatternLayout
log4j.appender.R.layout.ConversionPattern=%5p [%t] %d{ISO8601} %F (line %L) %m%n
# Edit the next line to point to your logs directory
log4j.appender.R.File=/var/log/cassandra/system.log
+log4j.appender.R.Threshold=INFO
+
+log4j.appender.tracing=org.apache.cassandra.tracing.TracingAppender
+log4j.appender.tracing.layout=org.apache.log4j.PatternLayout
+log4j.appender.tracing.layout.ConversionPattern=%5p [%t] %d{ISO8601} %F (line %L) %m%n
+log4j.appender.tracing.Threshold=DEBUG
+
# Application logging options
#log4j.logger.org.apache.cassandra=DEBUG
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/interface/cassandra.thrift
----------------------------------------------------------------------
diff --git a/interface/cassandra.thrift b/interface/cassandra.thrift
index 1f735e6..7fdef49 100644
--- a/interface/cassandra.thrift
+++ b/interface/cassandra.thrift
@@ -721,6 +721,10 @@ service Cassandra {
4:required i32 keys_per_split)
throws (1:InvalidRequestException ire),
+ /** Enables tracing for the next query in this connection and returns the UUID for that trace session
+ The next query will be traced idependently of trace probability and the returned UUID can be used to query the trace keyspace */
+ binary trace_next_query(),
+
/** adds a column family. returns the new schema id. */
string system_add_column_family(1:required CfDef cf_def)
throws (1:InvalidRequestException ire, 2:SchemaDisagreementException sde),
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java
----------------------------------------------------------------------
diff --git a/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java b/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java
index ec4982e..fc179c9 100644
--- a/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java
+++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java
@@ -274,6 +274,12 @@ public class Cassandra {
public List<String> describe_splits(String cfName, String start_token, String end_token, int keys_per_split) throws InvalidRequestException, org.apache.thrift.TException;
/**
+ * Enables tracing for the next query in this connection and returns the UUID for that trace session
+ * The next query will be traced idependently of trace probability and the returned UUID can be used to query the trace keyspace
+ */
+ public ByteBuffer trace_next_query() throws org.apache.thrift.TException;
+
+ /**
* adds a column family. returns the new schema id.
*
* @param cf_def
@@ -402,6 +408,8 @@ public class Cassandra {
public void describe_splits(String cfName, String start_token, String end_token, int keys_per_split, org.apache.thrift.async.AsyncMethodCallback<AsyncClient.describe_splits_call> resultHandler) throws org.apache.thrift.TException;
+ public void trace_next_query(org.apache.thrift.async.AsyncMethodCallback<AsyncClient.trace_next_query_call> resultHandler) throws org.apache.thrift.TException;
+
public void system_add_column_family(CfDef cf_def, org.apache.thrift.async.AsyncMethodCallback<AsyncClient.system_add_column_family_call> resultHandler) throws org.apache.thrift.TException;
public void system_drop_column_family(String column_family, org.apache.thrift.async.AsyncMethodCallback<AsyncClient.system_drop_column_family_call> resultHandler) throws org.apache.thrift.TException;
@@ -1208,6 +1216,28 @@ public class Cassandra {
throw new org.apache.thrift.TApplicationException(org.apache.thrift.TApplicationException.MISSING_RESULT, "describe_splits failed: unknown result");
}
+ public ByteBuffer trace_next_query() throws org.apache.thrift.TException
+ {
+ send_trace_next_query();
+ return recv_trace_next_query();
+ }
+
+ public void send_trace_next_query() throws org.apache.thrift.TException
+ {
+ trace_next_query_args args = new trace_next_query_args();
+ sendBase("trace_next_query", args);
+ }
+
+ public ByteBuffer recv_trace_next_query() throws org.apache.thrift.TException
+ {
+ trace_next_query_result result = new trace_next_query_result();
+ receiveBase(result, "trace_next_query");
+ if (result.isSetSuccess()) {
+ return result.success;
+ }
+ throw new org.apache.thrift.TApplicationException(org.apache.thrift.TApplicationException.MISSING_RESULT, "trace_next_query failed: unknown result");
+ }
+
public String system_add_column_family(CfDef cf_def) throws InvalidRequestException, SchemaDisagreementException, org.apache.thrift.TException
{
send_system_add_column_family(cf_def);
@@ -2447,6 +2477,35 @@ public class Cassandra {
}
}
+ public void trace_next_query(org.apache.thrift.async.AsyncMethodCallback<trace_next_query_call> resultHandler) throws org.apache.thrift.TException {
+ checkReady();
+ trace_next_query_call method_call = new trace_next_query_call(resultHandler, this, ___protocolFactory, ___transport);
+ this.___currentMethod = method_call;
+ ___manager.call(method_call);
+ }
+
+ public static class trace_next_query_call extends org.apache.thrift.async.TAsyncMethodCall {
+ public trace_next_query_call(org.apache.thrift.async.AsyncMethodCallback<trace_next_query_call> resultHandler, org.apache.thrift.async.TAsyncClient client, org.apache.thrift.protocol.TProtocolFactory protocolFactory, org.apache.thrift.transport.TNonblockingTransport transport) throws org.apache.thrift.TException {
+ super(client, protocolFactory, transport, resultHandler, false);
+ }
+
+ public void write_args(org.apache.thrift.protocol.TProtocol prot) throws org.apache.thrift.TException {
+ prot.writeMessageBegin(new org.apache.thrift.protocol.TMessage("trace_next_query", org.apache.thrift.protocol.TMessageType.CALL, 0));
+ trace_next_query_args args = new trace_next_query_args();
+ args.write(prot);
+ prot.writeMessageEnd();
+ }
+
+ public ByteBuffer getResult() throws org.apache.thrift.TException {
+ if (getState() != org.apache.thrift.async.TAsyncMethodCall.State.RESPONSE_READ) {
+ throw new IllegalStateException("Method call not finished!");
+ }
+ org.apache.thrift.transport.TMemoryInputTransport memoryTransport = new org.apache.thrift.transport.TMemoryInputTransport(getFrameBuffer().array());
+ org.apache.thrift.protocol.TProtocol prot = client.getProtocolFactory().getProtocol(memoryTransport);
+ return (new Client(prot)).recv_trace_next_query();
+ }
+ }
+
public void system_add_column_family(CfDef cf_def, org.apache.thrift.async.AsyncMethodCallback<system_add_column_family_call> resultHandler) throws org.apache.thrift.TException {
checkReady();
system_add_column_family_call method_call = new system_add_column_family_call(cf_def, resultHandler, this, ___protocolFactory, ___transport);
@@ -2815,6 +2874,7 @@ public class Cassandra {
processMap.put("describe_snitch", new describe_snitch());
processMap.put("describe_keyspace", new describe_keyspace());
processMap.put("describe_splits", new describe_splits());
+ processMap.put("trace_next_query", new trace_next_query());
processMap.put("system_add_column_family", new system_add_column_family());
processMap.put("system_drop_column_family", new system_drop_column_family());
processMap.put("system_add_keyspace", new system_add_keyspace());
@@ -3395,6 +3455,22 @@ public class Cassandra {
}
}
+ private static class trace_next_query<I extends Iface> extends org.apache.thrift.ProcessFunction<I, trace_next_query_args> {
+ public trace_next_query() {
+ super("trace_next_query");
+ }
+
+ protected trace_next_query_args getEmptyArgsInstance() {
+ return new trace_next_query_args();
+ }
+
+ protected trace_next_query_result getResult(I iface, trace_next_query_args args) throws org.apache.thrift.TException {
+ trace_next_query_result result = new trace_next_query_result();
+ result.success = iface.trace_next_query();
+ return result;
+ }
+ }
+
private static class system_add_column_family<I extends Iface> extends org.apache.thrift.ProcessFunction<I, system_add_column_family_args> {
public system_add_column_family() {
super("system_add_column_family");
@@ -27268,8 +27344,6 @@ public class Cassandra {
private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
try {
- // it doesn't seem like you should have to do this, but java serialization is wacky, and doesn't call the default constructor.
- __isset_bit_vector = new BitSet(1);
read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
} catch (org.apache.thrift.TException te) {
throw new java.io.IOException(te);
@@ -27712,6 +27786,524 @@ public class Cassandra {
}
+ public static class trace_next_query_args implements org.apache.thrift.TBase<trace_next_query_args, trace_next_query_args._Fields>, java.io.Serializable, Cloneable {
+ private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("trace_next_query_args");
+
+
+
+ /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
+ public enum _Fields implements org.apache.thrift.TFieldIdEnum {
+;
+
+ private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
+
+ static {
+ for (_Fields field : EnumSet.allOf(_Fields.class)) {
+ byName.put(field.getFieldName(), field);
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, or null if its not found.
+ */
+ public static _Fields findByThriftId(int fieldId) {
+ switch(fieldId) {
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, throwing an exception
+ * if it is not found.
+ */
+ public static _Fields findByThriftIdOrThrow(int fieldId) {
+ _Fields fields = findByThriftId(fieldId);
+ if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+ return fields;
+ }
+
+ /**
+ * Find the _Fields constant that matches name, or null if its not found.
+ */
+ public static _Fields findByName(String name) {
+ return byName.get(name);
+ }
+
+ private final short _thriftId;
+ private final String _fieldName;
+
+ _Fields(short thriftId, String fieldName) {
+ _thriftId = thriftId;
+ _fieldName = fieldName;
+ }
+
+ public short getThriftFieldId() {
+ return _thriftId;
+ }
+
+ public String getFieldName() {
+ return _fieldName;
+ }
+ }
+ public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
+ static {
+ Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
+ metaDataMap = Collections.unmodifiableMap(tmpMap);
+ org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(trace_next_query_args.class, metaDataMap);
+ }
+
+ public trace_next_query_args() {
+ }
+
+ /**
+ * Performs a deep copy on <i>other</i>.
+ */
+ public trace_next_query_args(trace_next_query_args other) {
+ }
+
+ public trace_next_query_args deepCopy() {
+ return new trace_next_query_args(this);
+ }
+
+ @Override
+ public void clear() {
+ }
+
+ public void setFieldValue(_Fields field, Object value) {
+ switch (field) {
+ }
+ }
+
+ public Object getFieldValue(_Fields field) {
+ switch (field) {
+ }
+ throw new IllegalStateException();
+ }
+
+ /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
+ public boolean isSet(_Fields field) {
+ if (field == null) {
+ throw new IllegalArgumentException();
+ }
+
+ switch (field) {
+ }
+ throw new IllegalStateException();
+ }
+
+ @Override
+ public boolean equals(Object that) {
+ if (that == null)
+ return false;
+ if (that instanceof trace_next_query_args)
+ return this.equals((trace_next_query_args)that);
+ return false;
+ }
+
+ public boolean equals(trace_next_query_args that) {
+ if (that == null)
+ return false;
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ HashCodeBuilder builder = new HashCodeBuilder();
+
+ return builder.toHashCode();
+ }
+
+ public int compareTo(trace_next_query_args other) {
+ if (!getClass().equals(other.getClass())) {
+ return getClass().getName().compareTo(other.getClass().getName());
+ }
+
+ int lastComparison = 0;
+ trace_next_query_args typedOther = (trace_next_query_args)other;
+
+ return 0;
+ }
+
+ public _Fields fieldForId(int fieldId) {
+ return _Fields.findByThriftId(fieldId);
+ }
+
+ public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
+ org.apache.thrift.protocol.TField field;
+ iprot.readStructBegin();
+ while (true)
+ {
+ field = iprot.readFieldBegin();
+ if (field.type == org.apache.thrift.protocol.TType.STOP) {
+ break;
+ }
+ switch (field.id) {
+ default:
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, field.type);
+ }
+ iprot.readFieldEnd();
+ }
+ iprot.readStructEnd();
+
+ // check for required fields of primitive type, which can't be checked in the validate method
+ validate();
+ }
+
+ public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
+ validate();
+
+ oprot.writeStructBegin(STRUCT_DESC);
+ oprot.writeFieldStop();
+ oprot.writeStructEnd();
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("trace_next_query_args(");
+ boolean first = true;
+
+ sb.append(")");
+ return sb.toString();
+ }
+
+ public void validate() throws org.apache.thrift.TException {
+ // check for required fields
+ }
+
+ private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+ try {
+ write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
+ try {
+ read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ }
+
+ public static class trace_next_query_result implements org.apache.thrift.TBase<trace_next_query_result, trace_next_query_result._Fields>, java.io.Serializable, Cloneable {
+ private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("trace_next_query_result");
+
+ private static final org.apache.thrift.protocol.TField SUCCESS_FIELD_DESC = new org.apache.thrift.protocol.TField("success", org.apache.thrift.protocol.TType.STRING, (short)0);
+
+ public ByteBuffer success; // required
+
+ /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
+ public enum _Fields implements org.apache.thrift.TFieldIdEnum {
+ SUCCESS((short)0, "success");
+
+ private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
+
+ static {
+ for (_Fields field : EnumSet.allOf(_Fields.class)) {
+ byName.put(field.getFieldName(), field);
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, or null if its not found.
+ */
+ public static _Fields findByThriftId(int fieldId) {
+ switch(fieldId) {
+ case 0: // SUCCESS
+ return SUCCESS;
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, throwing an exception
+ * if it is not found.
+ */
+ public static _Fields findByThriftIdOrThrow(int fieldId) {
+ _Fields fields = findByThriftId(fieldId);
+ if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+ return fields;
+ }
+
+ /**
+ * Find the _Fields constant that matches name, or null if its not found.
+ */
+ public static _Fields findByName(String name) {
+ return byName.get(name);
+ }
+
+ private final short _thriftId;
+ private final String _fieldName;
+
+ _Fields(short thriftId, String fieldName) {
+ _thriftId = thriftId;
+ _fieldName = fieldName;
+ }
+
+ public short getThriftFieldId() {
+ return _thriftId;
+ }
+
+ public String getFieldName() {
+ return _fieldName;
+ }
+ }
+
+ // isset id assignments
+
+ public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
+ static {
+ Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
+ tmpMap.put(_Fields.SUCCESS, new org.apache.thrift.meta_data.FieldMetaData("success", org.apache.thrift.TFieldRequirementType.DEFAULT,
+ new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING , true)));
+ metaDataMap = Collections.unmodifiableMap(tmpMap);
+ org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(trace_next_query_result.class, metaDataMap);
+ }
+
+ public trace_next_query_result() {
+ }
+
+ public trace_next_query_result(
+ ByteBuffer success)
+ {
+ this();
+ this.success = success;
+ }
+
+ /**
+ * Performs a deep copy on <i>other</i>.
+ */
+ public trace_next_query_result(trace_next_query_result other) {
+ if (other.isSetSuccess()) {
+ this.success = org.apache.thrift.TBaseHelper.copyBinary(other.success);
+;
+ }
+ }
+
+ public trace_next_query_result deepCopy() {
+ return new trace_next_query_result(this);
+ }
+
+ @Override
+ public void clear() {
+ this.success = null;
+ }
+
+ public byte[] getSuccess() {
+ setSuccess(org.apache.thrift.TBaseHelper.rightSize(success));
+ return success == null ? null : success.array();
+ }
+
+ public ByteBuffer bufferForSuccess() {
+ return success;
+ }
+
+ public trace_next_query_result setSuccess(byte[] success) {
+ setSuccess(success == null ? (ByteBuffer)null : ByteBuffer.wrap(success));
+ return this;
+ }
+
+ public trace_next_query_result setSuccess(ByteBuffer success) {
+ this.success = success;
+ return this;
+ }
+
+ public void unsetSuccess() {
+ this.success = null;
+ }
+
+ /** Returns true if field success is set (has been assigned a value) and false otherwise */
+ public boolean isSetSuccess() {
+ return this.success != null;
+ }
+
+ public void setSuccessIsSet(boolean value) {
+ if (!value) {
+ this.success = null;
+ }
+ }
+
+ public void setFieldValue(_Fields field, Object value) {
+ switch (field) {
+ case SUCCESS:
+ if (value == null) {
+ unsetSuccess();
+ } else {
+ setSuccess((ByteBuffer)value);
+ }
+ break;
+
+ }
+ }
+
+ public Object getFieldValue(_Fields field) {
+ switch (field) {
+ case SUCCESS:
+ return getSuccess();
+
+ }
+ throw new IllegalStateException();
+ }
+
+ /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
+ public boolean isSet(_Fields field) {
+ if (field == null) {
+ throw new IllegalArgumentException();
+ }
+
+ switch (field) {
+ case SUCCESS:
+ return isSetSuccess();
+ }
+ throw new IllegalStateException();
+ }
+
+ @Override
+ public boolean equals(Object that) {
+ if (that == null)
+ return false;
+ if (that instanceof trace_next_query_result)
+ return this.equals((trace_next_query_result)that);
+ return false;
+ }
+
+ public boolean equals(trace_next_query_result that) {
+ if (that == null)
+ return false;
+
+ boolean this_present_success = true && this.isSetSuccess();
+ boolean that_present_success = true && that.isSetSuccess();
+ if (this_present_success || that_present_success) {
+ if (!(this_present_success && that_present_success))
+ return false;
+ if (!this.success.equals(that.success))
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ HashCodeBuilder builder = new HashCodeBuilder();
+
+ boolean present_success = true && (isSetSuccess());
+ builder.append(present_success);
+ if (present_success)
+ builder.append(success);
+
+ return builder.toHashCode();
+ }
+
+ public int compareTo(trace_next_query_result other) {
+ if (!getClass().equals(other.getClass())) {
+ return getClass().getName().compareTo(other.getClass().getName());
+ }
+
+ int lastComparison = 0;
+ trace_next_query_result typedOther = (trace_next_query_result)other;
+
+ lastComparison = Boolean.valueOf(isSetSuccess()).compareTo(typedOther.isSetSuccess());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (isSetSuccess()) {
+ lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.success, typedOther.success);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ return 0;
+ }
+
+ public _Fields fieldForId(int fieldId) {
+ return _Fields.findByThriftId(fieldId);
+ }
+
+ public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
+ org.apache.thrift.protocol.TField field;
+ iprot.readStructBegin();
+ while (true)
+ {
+ field = iprot.readFieldBegin();
+ if (field.type == org.apache.thrift.protocol.TType.STOP) {
+ break;
+ }
+ switch (field.id) {
+ case 0: // SUCCESS
+ if (field.type == org.apache.thrift.protocol.TType.STRING) {
+ this.success = iprot.readBinary();
+ } else {
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, field.type);
+ }
+ break;
+ default:
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, field.type);
+ }
+ iprot.readFieldEnd();
+ }
+ iprot.readStructEnd();
+
+ // check for required fields of primitive type, which can't be checked in the validate method
+ validate();
+ }
+
+ public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
+ oprot.writeStructBegin(STRUCT_DESC);
+
+ if (this.isSetSuccess()) {
+ oprot.writeFieldBegin(SUCCESS_FIELD_DESC);
+ oprot.writeBinary(this.success);
+ oprot.writeFieldEnd();
+ }
+ oprot.writeFieldStop();
+ oprot.writeStructEnd();
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("trace_next_query_result(");
+ boolean first = true;
+
+ sb.append("success:");
+ if (this.success == null) {
+ sb.append("null");
+ } else {
+ org.apache.thrift.TBaseHelper.toString(this.success, sb);
+ }
+ first = false;
+ sb.append(")");
+ return sb.toString();
+ }
+
+ public void validate() throws org.apache.thrift.TException {
+ // check for required fields
+ }
+
+ private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+ try {
+ write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
+ try {
+ read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ }
+
public static class system_add_column_family_args implements org.apache.thrift.TBase<system_add_column_family_args, system_add_column_family_args._Fields>, java.io.Serializable, Cloneable {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("system_add_column_family_args");
@@ -34878,8 +35470,6 @@ public class Cassandra {
private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
try {
- // it doesn't seem like you should have to do this, but java serialization is wacky, and doesn't call the default constructor.
- __isset_bit_vector = new BitSet(1);
read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
} catch (org.apache.thrift.TException te) {
throw new java.io.IOException(te);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/cache/AutoSavingCache.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cache/AutoSavingCache.java b/src/java/org/apache/cassandra/cache/AutoSavingCache.java
index 2e86672..2587619 100644
--- a/src/java/org/apache/cassandra/cache/AutoSavingCache.java
+++ b/src/java/org/apache/cassandra/cache/AutoSavingCache.java
@@ -31,6 +31,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.Table;
import org.apache.cassandra.db.compaction.CompactionInfo;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.compaction.OperationType;
@@ -208,7 +209,7 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
else
type = OperationType.UNKNOWN;
- info = new CompactionInfo(new CFMetaData("system", cacheType.toString(), null, null, null),
+ info = new CompactionInfo(new CFMetaData(Table.SYSTEM_KS, cacheType.toString(), null, null, null),
type,
0,
keys.size(),
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/cli/Cli.g
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cli/Cli.g b/src/java/org/apache/cassandra/cli/Cli.g
index 32bab47..6dee785 100644
--- a/src/java/org/apache/cassandra/cli/Cli.g
+++ b/src/java/org/apache/cassandra/cli/Cli.g
@@ -38,6 +38,8 @@ tokens {
NODE_DESCRIBE;
NODE_DESCRIBE_CLUSTER;
NODE_USE_TABLE;
+ NODE_TRACE_NEXT_QUERY;
+ NODE_SET_TRACE_PROBABILITY;
NODE_EXIT;
NODE_HELP;
NODE_NO_OP;
@@ -145,6 +147,8 @@ statement
| delColumnFamily
| delKeyspace
| useKeyspace
+ | traceNextQuery
+ | setTraceProbability
| delStatement
| getStatement
| helpStatement
@@ -173,6 +177,10 @@ helpStatement
-> ^(NODE_HELP NODE_CONNECT)
| HELP USE
-> ^(NODE_HELP NODE_USE_TABLE)
+ | HELP TRACE NEXT QUERY
+ -> ^(NODE_HELP NODE_TRACE_NEXT_QUERY)
+ | HELP SET TRACE PROBABILITY
+ -> ^(NODE_HELP NODE_SET_TRACE_PROBABILITY)
| HELP DESCRIBE
-> ^(NODE_HELP NODE_DESCRIBE)
| HELP DESCRIBE 'CLUSTER'
@@ -372,7 +380,16 @@ useKeyspace
: USE keyspace ( username )? ( password )?
-> ^(NODE_USE_TABLE keyspace ( username )? ( password )?)
;
+
+traceNextQuery
+ : TRACE NEXT QUERY
+ -> ^(NODE_TRACE_NEXT_QUERY)
+ ;
+setTraceProbability
+ : SET TRACE PROBABILITY tracingProbability
+ -> ^(NODE_SET_TRACE_PROBABILITY tracingProbability)
+ ;
keyValuePairExpr
: entityName ( (AND | WITH) keyValuePair )*
@@ -533,7 +550,6 @@ ip_address
-> ^(NODE_ID_LIST IP_ADDRESS)
;
-
port
: IntegerPositiveLiteral
;
@@ -543,6 +559,14 @@ incrementValue
| IntegerNegativeLiteral
;
+traceSessionId
+ : Identifier
+ ;
+
+tracingProbability
+ : DoubleLiteral
+ ;
+
//
// Lexer Section
//
@@ -552,13 +576,17 @@ incrementValue
//
// CLI is case-insensitive with respect to these keywords.
// However, they MUST be listed in upper case here.
-//
+//
CONFIG: 'CONFIG';
CONNECT: 'CONNECT';
COUNT: 'COUNT';
DEL: 'DEL';
DESCRIBE: 'DESCRIBE';
USE: 'USE';
+TRACE: 'TRACE';
+PROBABILITY: 'PROBABILITY';
+NEXT: 'NEXT';
+QUERY: 'QUERY';
GET: 'GET';
HELP: 'HELP';
EXIT: 'EXIT';
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/cli/CliClient.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cli/CliClient.java b/src/java/org/apache/cassandra/cli/CliClient.java
index 176f70a..bedcc59 100644
--- a/src/java/org/apache/cassandra/cli/CliClient.java
+++ b/src/java/org/apache/cassandra/cli/CliClient.java
@@ -21,17 +21,16 @@ import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.PrintStream;
-import java.nio.ByteBuffer;
-import java.nio.charset.CharacterCodingException;
import java.net.InetAddress;
import java.net.UnknownHostException;
+import java.nio.ByteBuffer;
+import java.nio.charset.CharacterCodingException;
import java.util.*;
+import com.google.common.base.Charsets;
import com.google.common.base.Predicate;
import com.google.common.collect.Collections2;
import org.apache.commons.lang.StringUtils;
-import com.google.common.base.Charsets;
-import com.google.common.base.Joiner;
import org.antlr.runtime.tree.Tree;
import org.apache.cassandra.auth.IAuthenticator;
@@ -42,7 +41,6 @@ import org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.locator.SimpleSnitch;
-import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.tools.NodeProbe;
import org.apache.cassandra.utils.ByteBufferUtil;
@@ -50,15 +48,11 @@ import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.UUIDGen;
import org.apache.thrift.TBaseHelper;
import org.apache.thrift.TException;
-import org.codehaus.jackson.JsonEncoding;
-import org.codehaus.jackson.JsonFactory;
-import org.codehaus.jackson.JsonGenerator;
-import org.codehaus.jackson.JsonParser;
-import org.codehaus.jackson.JsonToken;
-import org.yaml.snakeyaml.constructor.Constructor;
+import org.codehaus.jackson.*;
import org.yaml.snakeyaml.Loader;
import org.yaml.snakeyaml.TypeDescription;
import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.Constructor;
// Cli Client Side Library
public class CliClient
@@ -265,6 +259,9 @@ public class CliClient
case CliParser.NODE_USE_TABLE:
executeUseKeySpace(tree);
break;
+ case CliParser.NODE_TRACE_NEXT_QUERY:
+ executeTraceNextQuery();
+ break;
case CliParser.NODE_CONNECT:
executeConnect(tree);
break;
@@ -1983,6 +1980,16 @@ public class CliClient
}
}
+ private void executeTraceNextQuery() throws TException, CharacterCodingException
+ {
+ if (!CliMain.isConnected())
+ return;
+
+ UUID sessionId = TimeUUIDType.instance.compose(thriftClient.trace_next_query());
+
+ sessionState.out.println("Will trace next query. Session ID: " + sessionId.toString());
+ }
+
private void describeKeySpace(String keySpaceName, KsDef metadata) throws TException
{
NodeProbe probe = sessionState.getNodeProbe();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java b/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java
index ded7833..0d5c428 100644
--- a/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java
+++ b/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java
@@ -22,6 +22,11 @@ import java.util.concurrent.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.tracing.TraceState;
+import org.apache.cassandra.tracing.Tracing;
+
+import static org.apache.cassandra.tracing.Tracing.isTracing;
+
/**
* This class encorporates some Executor best practices for Cassandra. Most of the executors in the system
* should use or extend this. There are two main improvements over a vanilla TPE:
@@ -126,13 +131,63 @@ public class DebuggableThreadPoolExecutor extends ThreadPoolExecutor
protected void onFinalAccept(Runnable task) {}
protected void onFinalRejection(Runnable task) {}
+ // execute does not call newTaskFor
+ @Override
+ public void execute(Runnable command)
+ {
+ super.execute(isTracing() && !(command instanceof TraceSessionWrapper)
+ ? new TraceSessionWrapper<Object>(command, null)
+ : command);
+ }
+
+ @Override
+ protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T result)
+ {
+ if (isTracing() && !(runnable instanceof TraceSessionWrapper))
+ {
+ return new TraceSessionWrapper<T>(runnable, result);
+ }
+ return super.newTaskFor(runnable, result);
+ }
+
+ @Override
+ protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable)
+ {
+ if (isTracing() && !(callable instanceof TraceSessionWrapper))
+ {
+ return new TraceSessionWrapper<T>(callable);
+ }
+ return super.newTaskFor(callable);
+ }
+
@Override
protected void afterExecute(Runnable r, Throwable t)
{
- super.afterExecute(r,t);
+ super.afterExecute(r, t);
+
+ if (r instanceof TraceSessionWrapper)
+ {
+ logger.debug("completed executing {}", r);
+ // we have to reset trace state as its presence is what denotes the current thread is tracing
+ // and if left this thread might start tracing unrelated tasks
+ ((TraceSessionWrapper)r).reset();
+ }
+
logExceptionsAfterExecute(r, t);
}
+ @Override
+ protected void beforeExecute(Thread t, Runnable r)
+ {
+ if (r instanceof TraceSessionWrapper)
+ {
+ logger.debug("executing {}", r);
+ ((TraceSessionWrapper) r).setupContext();
+ }
+
+ super.beforeExecute(t, r);
+ }
+
/**
* Send @param t and any exception wrapped by @param r to the default uncaught exception handler,
* or log them if none such is set up
@@ -193,4 +248,56 @@ public class DebuggableThreadPoolExecutor extends ThreadPoolExecutor
return null;
}
+
+ /**
+ * Used to wrap a Runnable or Callable passed to submit or execute so we can clone the TraceSessionContext and move
+ * it into the worker thread.
+ *
+ * @param <T>
+ */
+ private static class TraceSessionWrapper<T> extends FutureTask<T> implements Callable<T>
+ {
+ private final TraceState state;
+ private Callable<T> callable;
+
+ // Using initializer because the ctor's provided by the FutureTask<> are all we need
+ {
+ state = Tracing.instance().get();
+ }
+
+ public TraceSessionWrapper(Runnable runnable, T result)
+ {
+ super(runnable, result);
+ }
+
+ public TraceSessionWrapper(Callable<T> callable)
+ {
+ super(callable);
+ this.callable = callable;
+ }
+
+ private void setupContext()
+ {
+ Tracing.instance().set(state);
+ }
+
+ private void reset()
+ {
+ Tracing.instance().set(null);
+ }
+
+ public T call() throws Exception
+ {
+ return callable.call();
+ }
+
+ @Override
+ public String toString()
+ {
+ return "TraceSessionWrapper{" +
+ "state=" + state +
+ ", callable=" + callable +
+ '}';
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/concurrent/Stage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/concurrent/Stage.java b/src/java/org/apache/cassandra/concurrent/Stage.java
index 9cf3e33..062bbe6 100644
--- a/src/java/org/apache/cassandra/concurrent/Stage.java
+++ b/src/java/org/apache/cassandra/concurrent/Stage.java
@@ -27,6 +27,7 @@ public enum Stage
ANTI_ENTROPY,
MIGRATION,
MISC,
+ TRACING,
INTERNAL_RESPONSE,
READ_REPAIR,
REPLICATE_ON_WRITE;
@@ -39,6 +40,7 @@ public enum Stage
case GOSSIP:
case MIGRATION:
case MISC:
+ case TRACING:
case STREAM:
case INTERNAL_RESPONSE:
return "internal";
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/concurrent/StageManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/concurrent/StageManager.java b/src/java/org/apache/cassandra/concurrent/StageManager.java
index e24d906..8359346 100644
--- a/src/java/org/apache/cassandra/concurrent/StageManager.java
+++ b/src/java/org/apache/cassandra/concurrent/StageManager.java
@@ -18,9 +18,12 @@
package org.apache.cassandra.concurrent;
import java.util.EnumMap;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.net.MessagingService;
import static org.apache.cassandra.config.DatabaseDescriptor.*;
@@ -32,6 +35,8 @@ import static org.apache.cassandra.config.DatabaseDescriptor.*;
*/
public class StageManager
{
+ private static final Logger logger = LoggerFactory.getLogger(StageManager.class);
+
private static final EnumMap<Stage, ThreadPoolExecutor> stages = new EnumMap<Stage, ThreadPoolExecutor>(Stage.class);
public static final long KEEPALIVE = 60; // seconds to keep "extra" threads alive for when idle
@@ -52,6 +57,25 @@ public class StageManager
stages.put(Stage.MIGRATION, new JMXEnabledThreadPoolExecutor(Stage.MIGRATION));
stages.put(Stage.MISC, new JMXEnabledThreadPoolExecutor(Stage.MISC));
stages.put(Stage.READ_REPAIR, multiThreadedStage(Stage.READ_REPAIR, Runtime.getRuntime().availableProcessors()));
+ stages.put(Stage.TRACING, tracingExecutor());
+ }
+
+ private static ThreadPoolExecutor tracingExecutor()
+ {
+ RejectedExecutionHandler reh = new RejectedExecutionHandler()
+ {
+ public void rejectedExecution(Runnable r, ThreadPoolExecutor executor)
+ {
+ MessagingService.instance().incrementDroppedMessages(MessagingService.Verb._TRACE);
+ }
+ };
+ return new ExecuteOnlyExecutor(1,
+ 1,
+ KEEPALIVE,
+ TimeUnit.SECONDS,
+ new ArrayBlockingQueue<Runnable>(1000),
+ new NamedThreadFactory(Stage.TRACING.getJmxName()),
+ reh);
}
private static ThreadPoolExecutor multiThreadedStage(Stage stage, int numThreads)
@@ -103,4 +127,34 @@ public class StageManager
StageManager.stages.get(stage).shutdownNow();
}
}
+
+ /**
+ * A TPE that disallows submit so that we don't need to worry about unwrapping exceptions on the
+ * tracing stage. See CASSANDRA-1123 for background.
+ */
+ private static class ExecuteOnlyExecutor extends ThreadPoolExecutor
+ {
+ public ExecuteOnlyExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)
+ {
+ super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
+ }
+
+ @Override
+ public Future<?> submit(Runnable task)
+ {
+ return super.submit(task);
+ }
+
+ @Override
+ public <T> Future<T> submit(Runnable task, T result)
+ {
+ return super.submit(task, result);
+ }
+
+ @Override
+ public <T> Future<T> submit(Callable<T> task)
+ {
+ return super.submit(task);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/config/CFMetaData.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java
index b124463..bb9a5a6 100644
--- a/src/java/org/apache/cassandra/config/CFMetaData.java
+++ b/src/java/org/apache/cassandra/config/CFMetaData.java
@@ -45,6 +45,7 @@ import org.apache.cassandra.io.compress.CompressionParameters;
import org.apache.cassandra.io.compress.SnappyCompressor;
import org.apache.cassandra.thrift.IndexType;
import org.apache.cassandra.thrift.InvalidRequestException;
+import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
@@ -75,13 +76,13 @@ public final class CFMetaData
public final static String DEFAULT_COMPRESSOR = SnappyCompressor.isAvailable() ? SnappyCompressor.class.getCanonicalName() : null;
@Deprecated
- public static final CFMetaData OldStatusCf = newSystemMetadata(SystemTable.OLD_STATUS_CF, 0, "unused", BytesType.instance, null);
+ public static final CFMetaData OldStatusCf = newSystemMetadata(Table.SYSTEM_KS, SystemTable.OLD_STATUS_CF, 0, "unused", BytesType.instance, null);
@Deprecated
- public static final CFMetaData OldHintsCf = newSystemMetadata(SystemTable.OLD_HINTS_CF, 1, "unused", BytesType.instance, BytesType.instance);
+ public static final CFMetaData OldHintsCf = newSystemMetadata(Table.SYSTEM_KS, SystemTable.OLD_HINTS_CF, 1, "unused", BytesType.instance, BytesType.instance);
@Deprecated
- public static final CFMetaData MigrationsCf = newSystemMetadata(DefsTable.OLD_MIGRATIONS_CF, 2, "unused", TimeUUIDType.instance, null);
+ public static final CFMetaData MigrationsCf = newSystemMetadata(Table.SYSTEM_KS, DefsTable.OLD_MIGRATIONS_CF, 2, "unused", TimeUUIDType.instance, null);
@Deprecated
- public static final CFMetaData SchemaCf = newSystemMetadata(DefsTable.OLD_SCHEMA_CF, 3, "unused", UTF8Type.instance, null);
+ public static final CFMetaData SchemaCf = newSystemMetadata(Table.SYSTEM_KS, DefsTable.OLD_SCHEMA_CF, 3, "unused", UTF8Type.instance, null);
public static final CFMetaData IndexCf = compile(5, "CREATE TABLE \"" + SystemTable.INDEX_CF + "\" ("
+ "table_name text,"
@@ -169,6 +170,26 @@ public final class CFMetaData
+ "cql_version text"
+ ") WITH COMMENT='information about the local node'");
+ public static final CFMetaData TraceSessionsCf = compile(14, "CREATE TABLE " + Tracing.SESSIONS_CF + " ("
+ + " session_id uuid PRIMARY KEY,"
+ + " coordinator inet,"
+ + " request text,"
+ + " started_at timestamp,"
+ + " finished_at timestamp,"
+ + " parameters map<text, text>"
+ + ") WITH COMMENT='traced sessions'", Tracing.TRACE_KS);
+
+ public static final CFMetaData TraceEventsCf = compile(15, "CREATE TABLE " + Tracing.EVENTS_CF + " ("
+ + " session_id uuid,"
+ + " event_id timeuuid,"
+ + " source inet,"
+ + " thread text,"
+ + " activity text,"
+ + " happened_at timestamp,"
+ + " source_elapsed int,"
+ + " PRIMARY KEY (session_id, event_id)"
+ + ");", Tracing.TRACE_KS);
+
public enum Caching
{
ALL, KEYS_ONLY, ROWS_ONLY, NONE;
@@ -258,12 +279,12 @@ public final class CFMetaData
this.init();
}
- private static CFMetaData compile(int id, String cql)
+ private static CFMetaData compile(int id, String cql, String keyspace)
{
try
{
CreateColumnFamilyStatement statement = (CreateColumnFamilyStatement) QueryProcessor.parseStatement(cql).prepare().statement;
- CFMetaData cfmd = newSystemMetadata(statement.columnFamily(), id, "", statement.comparator, null);
+ CFMetaData cfmd = newSystemMetadata(keyspace, statement.columnFamily(), id, "", statement.comparator, null);
statement.applyPropertiesTo(cfmd);
return cfmd;
}
@@ -277,6 +298,11 @@ public final class CFMetaData
}
}
+ private static CFMetaData compile(int id, String cql)
+ {
+ return compile(id, cql, Table.SYSTEM_KS);
+ }
+
private AbstractType<?> enforceSubccDefault(ColumnFamilyType cftype, AbstractType<?> subcc)
{
return (subcc == null) && (cftype == ColumnFamilyType.Super) ? BytesType.instance : subcc;
@@ -324,10 +350,10 @@ public final class CFMetaData
updateCfDef(); // init cqlCfDef
}
- private static CFMetaData newSystemMetadata(String cfName, int oldCfId, String comment, AbstractType<?> comparator, AbstractType<?> subcc)
+ private static CFMetaData newSystemMetadata(String keyspace, String cfName, int oldCfId, String comment, AbstractType<?> comparator, AbstractType<?> subcc)
{
ColumnFamilyType type = subcc == null ? ColumnFamilyType.Standard : ColumnFamilyType.Super;
- CFMetaData newCFMD = new CFMetaData(Table.SYSTEM_TABLE, cfName, type, comparator, subcc);
+ CFMetaData newCFMD = new CFMetaData(keyspace, cfName, type, comparator, subcc);
// adding old -> new style ID mapping to support backward compatibility
Schema.instance.addOldCfIdMapping(oldCfId, newCFMD.cfId);
@@ -1119,7 +1145,7 @@ public final class CFMetaData
*/
public RowMutation toSchemaUpdate(CFMetaData newState, long modificationTimestamp)
{
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(ksName));
+ RowMutation rm = new RowMutation(Table.SYSTEM_KS, SystemTable.getSchemaKSKey(ksName));
newState.toSchemaNoColumns(rm, modificationTimestamp);
@@ -1152,7 +1178,7 @@ public final class CFMetaData
*/
public RowMutation dropFromSchema(long timestamp)
{
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(ksName));
+ RowMutation rm = new RowMutation(Table.SYSTEM_KS, SystemTable.getSchemaKSKey(ksName));
ColumnFamily cf = rm.addOrGet(SystemTable.SCHEMA_COLUMNFAMILIES_CF);
int ldt = (int) (System.currentTimeMillis() / 1000);
@@ -1328,7 +1354,7 @@ public final class CFMetaData
*/
public RowMutation toSchema(long timestamp) throws ConfigurationException
{
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(ksName));
+ RowMutation rm = new RowMutation(Table.SYSTEM_KS, SystemTable.getSchemaKSKey(ksName));
toSchema(rm, timestamp);
return rm;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 0cb1d1b..2e22e95 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -458,11 +458,13 @@ public class DatabaseDescriptor
rowCacheProvider = FBUtilities.newCacheProvider(conf.row_cache_provider);
// Hardcoded system tables
- KSMetaData systemMeta = KSMetaData.systemKeyspace();
- for (CFMetaData cfm : systemMeta.cfMetaData().values())
- Schema.instance.load(cfm);
-
- Schema.instance.addSystemTable(systemMeta);
+ List<KSMetaData> systemKeyspaces = Arrays.asList(KSMetaData.systemKeyspace(), KSMetaData.traceKeyspace());
+ for (KSMetaData ksmd : systemKeyspaces)
+ {
+ for (CFMetaData cfm : ksmd.cfMetaData().values())
+ Schema.instance.load(cfm);
+ Schema.instance.setTableDefinition(ksmd);
+ }
/* Load the seeds for node contact points */
if (conf.seed_provider == null)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/config/KSMetaData.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/KSMetaData.java b/src/java/org/apache/cassandra/config/KSMetaData.java
index 9ce5a26..ac528f3 100644
--- a/src/java/org/apache/cassandra/config/KSMetaData.java
+++ b/src/java/org/apache/cassandra/config/KSMetaData.java
@@ -19,6 +19,7 @@ package org.apache.cassandra.config;
import java.util.*;
+import com.google.common.collect.ImmutableMap;
import org.apache.commons.lang.ObjectUtils;
import org.apache.commons.lang.StringUtils;
@@ -30,6 +31,7 @@ import org.apache.cassandra.locator.*;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.CfDef;
import org.apache.cassandra.thrift.KsDef;
+import org.apache.cassandra.tracing.Tracing;
import static org.apache.cassandra.utils.FBUtilities.*;
@@ -82,7 +84,13 @@ public final class KSMetaData
CFMetaData.OldHintsCf,
CFMetaData.MigrationsCf,
CFMetaData.SchemaCf);
- return new KSMetaData(Table.SYSTEM_TABLE, LocalStrategy.class, Collections.<String, String>emptyMap(), true, cfDefs);
+ return new KSMetaData(Table.SYSTEM_KS, LocalStrategy.class, Collections.<String, String>emptyMap(), true, cfDefs);
+ }
+
+ public static KSMetaData traceKeyspace()
+ {
+ List<CFMetaData> cfDefs = Arrays.asList(CFMetaData.TraceSessionsCf, CFMetaData.TraceEventsCf);
+ return new KSMetaData(Tracing.TRACE_KS, SimpleStrategy.class, ImmutableMap.of("replication_factor", "1"), true, cfDefs);
}
public static KSMetaData testMetadata(String name, Class<? extends AbstractReplicationStrategy> strategyClass, Map<String, String> strategyOptions, CFMetaData... cfDefs)
@@ -206,7 +214,7 @@ public final class KSMetaData
public RowMutation dropFromSchema(long timestamp)
{
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(name));
+ RowMutation rm = new RowMutation(Table.SYSTEM_KS, SystemTable.getSchemaKSKey(name));
rm.delete(new QueryPath(SystemTable.SCHEMA_KEYSPACES_CF), timestamp);
rm.delete(new QueryPath(SystemTable.SCHEMA_COLUMNFAMILIES_CF), timestamp);
rm.delete(new QueryPath(SystemTable.SCHEMA_COLUMNS_CF), timestamp);
@@ -216,7 +224,7 @@ public final class KSMetaData
public RowMutation toSchema(long timestamp)
{
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(name));
+ RowMutation rm = new RowMutation(Table.SYSTEM_KS, SystemTable.getSchemaKSKey(name));
ColumnFamily cf = rm.addOrGet(SystemTable.SCHEMA_KEYSPACES_CF);
cf.addColumn(Column.create(durableWrites, timestamp, "durable_writes"));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/config/Schema.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/Schema.java b/src/java/org/apache/cassandra/config/Schema.java
index d2fa4e6..d4ace31 100644
--- a/src/java/org/apache/cassandra/config/Schema.java
+++ b/src/java/org/apache/cassandra/config/Schema.java
@@ -22,21 +22,17 @@ import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.*;
-import com.google.common.collect.BiMap;
-import com.google.common.collect.HashBiMap;
-import org.apache.cassandra.service.MigrationManager;
+import com.google.common.collect.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.cassandra.db.ColumnFamilyType;
-import org.apache.cassandra.db.Row;
-import org.apache.cassandra.db.SystemTable;
+import org.apache.cassandra.db.*;
import org.apache.cassandra.db.Table;
-import org.apache.cassandra.db.UnknownColumnFamilyException;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.io.sstable.Descriptor;
+import org.apache.cassandra.service.MigrationManager;
+import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.Pair;
-
import org.cliffc.high_scale_lib.NonBlockingHashMap;
public class Schema
@@ -281,9 +277,8 @@ public class Schema
*/
public List<String> getNonSystemTables()
{
- List<String> tablesList = new ArrayList<String>(tables.keySet());
- tablesList.remove(Table.SYSTEM_TABLE);
- return Collections.unmodifiableList(tablesList);
+ ImmutableSet<String> system = ImmutableSet.of(Table.SYSTEM_KS, Tracing.TRACE_KS);
+ return ImmutableList.copyOf(Sets.difference(tables.keySet(), system));
}
/**
@@ -336,17 +331,8 @@ public class Schema
*/
public void setTableDefinition(KSMetaData ksm)
{
- if (ksm != null)
- tables.put(ksm.name, ksm);
- }
-
- /**
- * Add a new system table
- * @param systemTable The metadata describing new system table
- */
- public void addSystemTable(KSMetaData systemTable)
- {
- tables.put(systemTable.name, systemTable);
+ assert ksm != null;
+ tables.put(ksm.name, ksm);
}
/* ColumnFamily query/control methods */
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index f31276f..8ef686b 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -1149,9 +1149,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
assert columnFamily.equals(filter.getColumnFamilyName()) : filter.getColumnFamilyName();
+ ColumnFamily result = null;
+
long start = System.nanoTime();
try
{
+
if (!isRowCacheEnabled())
{
ColumnFamily cf = getTopLevelColumns(filter, gcBefore, false);
@@ -1161,23 +1164,36 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
// TODO this is necessary because when we collate supercolumns together, we don't check
// their subcolumns for relevance, so we need to do a second prune post facto here.
- return cf.isSuper() ? removeDeleted(cf, gcBefore) : removeDeletedCF(cf, gcBefore);
+ result = cf.isSuper() ? removeDeleted(cf, gcBefore) : removeDeletedCF(cf, gcBefore);
+
}
+ else
+ {
- UUID cfId = Schema.instance.getId(table.name, this.columnFamily);
- if (cfId == null)
- return null; // secondary index
+ UUID cfId = Schema.instance.getId(table.name, columnFamily);
+ if (cfId == null)
+ {
+ logger.debug("no id found for {}.{}", table.name, columnFamily);
+ return null;
+ }
- ColumnFamily cached = getThroughCache(cfId, filter);
- if (cached == null)
- return null;
+ ColumnFamily cached = getThroughCache(cfId, filter);
+ if (cached == null)
+ {
+ logger.debug("cached row is empty");
+ return null;
+ }
- return filterColumnFamily(cached, filter, gcBefore);
+ result = filterColumnFamily(cached, filter, gcBefore);
+ }
}
finally
{
readStats.addNano(System.nanoTime() - start);
}
+
+ logger.debug("Read {} columns", result.getColumnCount());
+ return result;
}
/**
@@ -1612,7 +1628,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public static List<ColumnFamilyStore> allUserDefined()
{
List<ColumnFamilyStore> cfses = new ArrayList<ColumnFamilyStore>();
- for (Table table : Sets.difference(ImmutableSet.copyOf(Table.all()), ImmutableSet.of(Table.open(Table.SYSTEM_TABLE))))
+ for (Table table : Sets.difference(ImmutableSet.copyOf(Table.all()), ImmutableSet.of(Table.open(Table.SYSTEM_KS))))
cfses.addAll(table.getColumnFamilyStores());
return cfses;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/db/DefsTable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/DefsTable.java b/src/java/org/apache/cassandra/db/DefsTable.java
index f156f39..e6c51e9 100644
--- a/src/java/org/apache/cassandra/db/DefsTable.java
+++ b/src/java/org/apache/cassandra/db/DefsTable.java
@@ -188,7 +188,7 @@ public class DefsTable
public static synchronized Collection<KSMetaData> loadFromStorage(UUID version) throws IOException
{
DecoratedKey vkey = StorageService.getPartitioner().decorateKey(toUTF8Bytes(version));
- Table defs = Table.open(Table.SYSTEM_TABLE);
+ Table defs = Table.open(Table.SYSTEM_KS);
ColumnFamilyStore cfStore = defs.getColumnFamilyStore(OLD_SCHEMA_CF);
ColumnFamily cf = cfStore.getColumnFamily(QueryFilter.getIdentityFilter(vkey, new QueryPath(OLD_SCHEMA_CF)));
IColumn avroschema = cf.getColumn(DEFINITION_SCHEMA_COLUMN_NAME);
@@ -216,8 +216,8 @@ public class DefsTable
dumpToStorage(keyspaces);
logger.info("Truncating deprecated system column families (migrations, schema)...");
- dropColumnFamily(Table.SYSTEM_TABLE, OLD_MIGRATIONS_CF);
- dropColumnFamily(Table.SYSTEM_TABLE, OLD_SCHEMA_CF);
+ dropColumnFamily(Table.SYSTEM_KS, OLD_MIGRATIONS_CF);
+ dropColumnFamily(Table.SYSTEM_KS, OLD_SCHEMA_CF);
}
return keyspaces;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/db/Directories.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/Directories.java b/src/java/org/apache/cassandra/db/Directories.java
index ee53fd2..895f931 100644
--- a/src/java/org/apache/cassandra/db/Directories.java
+++ b/src/java/org/apache/cassandra/db/Directories.java
@@ -473,7 +473,7 @@ public class Directories
boolean hasSystemKeyspace = false;
for (DataDirectory dir : dataFileLocations)
{
- File systemDir = new File(dir.location, Table.SYSTEM_TABLE);
+ File systemDir = new File(dir.location, Table.SYSTEM_KS);
hasSystemKeyspace |= (systemDir.exists() && systemDir.isDirectory());
File statusCFDir = new File(systemDir, SystemTable.SCHEMA_KEYSPACES_CF);
if (statusCFDir.exists())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/db/HintedHandOffManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/HintedHandOffManager.java b/src/java/org/apache/cassandra/db/HintedHandOffManager.java
index e271c78..bf6f594 100644
--- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java
+++ b/src/java/org/apache/cassandra/db/HintedHandOffManager.java
@@ -134,7 +134,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
private static void deleteHint(ByteBuffer tokenBytes, ByteBuffer columnName, long timestamp) throws IOException
{
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, tokenBytes);
+ RowMutation rm = new RowMutation(Table.SYSTEM_KS, tokenBytes);
rm.delete(new QueryPath(SystemTable.HINTS_CF, null, columnName), timestamp);
rm.applyUnsafe(); // don't bother with commitlog since we're going to flush as soon as we're done with delivery
}
@@ -160,7 +160,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
return;
UUID hostId = StorageService.instance.getTokenMetadata().getHostId(endpoint);
ByteBuffer hostIdBytes = ByteBuffer.wrap(UUIDGen.decompose(hostId));
- final RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, hostIdBytes);
+ final RowMutation rm = new RowMutation(Table.SYSTEM_KS, hostIdBytes);
rm.delete(new QueryPath(SystemTable.HINTS_CF), System.currentTimeMillis());
// execute asynchronously to avoid blocking caller (which may be processing gossip)
@@ -185,7 +185,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
private Future<?> compact() throws ExecutionException, InterruptedException
{
- final ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(SystemTable.HINTS_CF);
+ final ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_KS).getColumnFamilyStore(SystemTable.HINTS_CF);
hintStore.forceBlockingFlush();
ArrayList<Descriptor> descriptors = new ArrayList<Descriptor>();
for (SSTable sstable : hintStore.getSSTables())
@@ -272,7 +272,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
}
});
- ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(SystemTable.HINTS_CF);
+ ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_KS).getColumnFamilyStore(SystemTable.HINTS_CF);
if (hintStore.isEmpty())
return; // nothing to do, don't confuse users by logging a no-op handoff
@@ -396,7 +396,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
if (logger.isDebugEnabled())
logger.debug("Started scheduleAllDeliveries");
- ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(SystemTable.HINTS_CF);
+ ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_KS).getColumnFamilyStore(SystemTable.HINTS_CF);
IPartitioner p = StorageService.getPartitioner();
RowPosition minPos = p.getMinimumToken().minKeyBound();
Range<RowPosition> range = new Range<RowPosition>(minPos, minPos, p);
@@ -493,7 +493,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
List<Row> rows;
try
{
- rows = StorageProxy.getRangeSlice(new RangeSliceCommand("system", parent, predicate, range, null, LARGE_NUMBER), ConsistencyLevel.ONE);
+ rows = StorageProxy.getRangeSlice(new RangeSliceCommand(Table.SYSTEM_KS, parent, predicate, range, null, LARGE_NUMBER), ConsistencyLevel.ONE);
}
catch (Exception e)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/db/RowMutation.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/RowMutation.java b/src/java/org/apache/cassandra/db/RowMutation.java
index 06e8c37..4fcee8a 100644
--- a/src/java/org/apache/cassandra/db/RowMutation.java
+++ b/src/java/org/apache/cassandra/db/RowMutation.java
@@ -104,7 +104,7 @@ public class RowMutation implements IMutation
*/
public static RowMutation hintFor(RowMutation mutation, UUID targetId) throws IOException
{
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, UUIDType.instance.decompose(targetId));
+ RowMutation rm = new RowMutation(Table.SYSTEM_KS, UUIDType.instance.decompose(targetId));
UUID hintId = UUIDGen.makeType1UUIDFromHost(FBUtilities.getBroadcastAddress());
// determine the TTL for the RowMutation
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/db/SystemTable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/SystemTable.java b/src/java/org/apache/cassandra/db/SystemTable.java
index 6910905..23d9269 100644
--- a/src/java/org/apache/cassandra/db/SystemTable.java
+++ b/src/java/org/apache/cassandra/db/SystemTable.java
@@ -114,7 +114,7 @@ public class SystemTable
/** if system data becomes incompatible across versions of cassandra, that logic (and associated purging) is managed here */
private static void upgradeSystemData() throws IOException, ExecutionException, InterruptedException
{
- Table table = Table.open(Table.SYSTEM_TABLE);
+ Table table = Table.open(Table.SYSTEM_KS);
ColumnFamilyStore oldStatusCfs = table.getColumnFamilyStore(OLD_STATUS_CF);
if (oldStatusCfs.getSSTables().size() > 0)
{
@@ -243,7 +243,7 @@ public class SystemTable
{
try
{
- Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(cfname).forceBlockingFlush();
+ Table.open(Table.SYSTEM_KS).getColumnFamilyStore(cfname).forceBlockingFlush();
}
catch (ExecutionException e)
{
@@ -282,7 +282,7 @@ public class SystemTable
Table table;
try
{
- table = Table.open(Table.SYSTEM_TABLE);
+ table = Table.open(Table.SYSTEM_KS);
}
catch (AssertionError err)
{
@@ -389,7 +389,7 @@ public class SystemTable
public static boolean isIndexBuilt(String table, String indexName)
{
- ColumnFamilyStore cfs = Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(INDEX_CF);
+ ColumnFamilyStore cfs = Table.open(Table.SYSTEM_KS).getColumnFamilyStore(INDEX_CF);
QueryFilter filter = QueryFilter.getNamesFilter(decorate(ByteBufferUtil.bytes(table)),
new QueryPath(INDEX_CF),
ByteBufferUtil.bytes(indexName));
@@ -398,9 +398,9 @@ public class SystemTable
public static void setIndexBuilt(String table, String indexName)
{
- ColumnFamily cf = ColumnFamily.create(Table.SYSTEM_TABLE, INDEX_CF);
+ ColumnFamily cf = ColumnFamily.create(Table.SYSTEM_KS, INDEX_CF);
cf.addColumn(new Column(ByteBufferUtil.bytes(indexName), ByteBufferUtil.EMPTY_BYTE_BUFFER, FBUtilities.timestampMicros()));
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, ByteBufferUtil.bytes(table));
+ RowMutation rm = new RowMutation(Table.SYSTEM_KS, ByteBufferUtil.bytes(table));
rm.add(cf);
rm.apply();
forceBlockingFlush(INDEX_CF);
@@ -408,7 +408,7 @@ public class SystemTable
public static void setIndexRemoved(String table, String indexName)
{
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, ByteBufferUtil.bytes(table));
+ RowMutation rm = new RowMutation(Table.SYSTEM_KS, ByteBufferUtil.bytes(table));
rm.delete(new QueryPath(INDEX_CF, null, ByteBufferUtil.bytes(indexName)), FBUtilities.timestampMicros());
rm.apply();
forceBlockingFlush(INDEX_CF);
@@ -447,7 +447,7 @@ public class SystemTable
public static NodeId getCurrentLocalNodeId()
{
ByteBuffer id = null;
- Table table = Table.open(Table.SYSTEM_TABLE);
+ Table table = Table.open(Table.SYSTEM_KS);
QueryFilter filter = QueryFilter.getIdentityFilter(decorate(CURRENT_LOCAL_NODE_ID_KEY),
new QueryPath(NODE_ID_CF));
ColumnFamily cf = table.getColumnFamilyStore(NODE_ID_CF).getColumnFamily(filter);
@@ -480,7 +480,7 @@ public class SystemTable
*/
public static void writeCurrentLocalNodeId(NodeId oldNodeId, NodeId newNodeId, long now)
{
- ColumnFamily cf = ColumnFamily.create(Table.SYSTEM_TABLE, NODE_ID_CF);
+ ColumnFamily cf = ColumnFamily.create(Table.SYSTEM_KS, NODE_ID_CF);
cf.addColumn(new Column(newNodeId.bytes(), ByteBufferUtil.EMPTY_BYTE_BUFFER, now));
ColumnFamily cf2 = cf.cloneMe();
if (oldNodeId != null)
@@ -489,8 +489,8 @@ public class SystemTable
// tests use single digit long values for now, so use actual time.
cf2.addColumn(new DeletedColumn(oldNodeId.bytes(), (int)(System.currentTimeMillis() / 1000), now));
}
- RowMutation rmCurrent = new RowMutation(Table.SYSTEM_TABLE, CURRENT_LOCAL_NODE_ID_KEY);
- RowMutation rmAll = new RowMutation(Table.SYSTEM_TABLE, ALL_LOCAL_NODE_ID_KEY);
+ RowMutation rmCurrent = new RowMutation(Table.SYSTEM_KS, CURRENT_LOCAL_NODE_ID_KEY);
+ RowMutation rmAll = new RowMutation(Table.SYSTEM_KS, ALL_LOCAL_NODE_ID_KEY);
rmCurrent.add(cf2);
rmAll.add(cf);
rmCurrent.apply();
@@ -501,7 +501,7 @@ public class SystemTable
{
List<NodeId.NodeIdRecord> l = new ArrayList<NodeId.NodeIdRecord>();
- Table table = Table.open(Table.SYSTEM_TABLE);
+ Table table = Table.open(Table.SYSTEM_KS);
QueryFilter filter = QueryFilter.getIdentityFilter(decorate(ALL_LOCAL_NODE_ID_KEY),
new QueryPath(NODE_ID_CF));
ColumnFamily cf = table.getColumnFamilyStore(NODE_ID_CF).getColumnFamily(filter);
@@ -525,7 +525,7 @@ public class SystemTable
*/
public static ColumnFamilyStore schemaCFS(String cfName)
{
- return Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(cfName);
+ return Table.open(Table.SYSTEM_KS).getColumnFamilyStore(cfName);
}
public static List<Row> serializedSchema()
@@ -574,7 +574,7 @@ public class SystemTable
if (mutation == null)
{
- mutationMap.put(schemaRow.key, new RowMutation(Table.SYSTEM_TABLE, schemaRow));
+ mutationMap.put(schemaRow.key, new RowMutation(Table.SYSTEM_KS, schemaRow));
continue;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/db/Table.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java
index e41b11f..5b4520d 100644
--- a/src/java/org/apache/cassandra/db/Table.java
+++ b/src/java/org/apache/cassandra/db/Table.java
@@ -33,10 +33,13 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.*;
import org.apache.cassandra.db.commitlog.CommitLog;
-import org.apache.cassandra.db.filter.*;
+import org.apache.cassandra.db.filter.ColumnSlice;
+import org.apache.cassandra.db.filter.QueryFilter;
+import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.ByteBufferUtil;
/**
@@ -44,14 +47,14 @@ import org.apache.cassandra.utils.ByteBufferUtil;
*/
public class Table
{
- public static final String SYSTEM_TABLE = "system";
+ public static final String SYSTEM_KS = "system";
private static final Logger logger = LoggerFactory.getLogger(Table.class);
/**
* accesses to CFS.memtable should acquire this for thread safety.
* CFS.maybeSwitchMemtable should aquire the writeLock; see that method for the full explanation.
- *
+ * <p/>
* (Enabling fairness in the RRWL is observed to decrease throughput, so we leave it off.)
*/
public static final ReentrantReadWriteLock switchLock = new ReentrantReadWriteLock();
@@ -167,9 +170,8 @@ public class Table
* Take a snapshot of the specific column family, or the entire set of column families
* if columnFamily is null with a given timestamp
*
- * @param snapshotName the tag associated with the name of the snapshot. This value may not be null
+ * @param snapshotName the tag associated with the name of the snapshot. This value may not be null
* @param columnFamilyName the column family to snapshot or all on null
- *
* @throws IOException if the column family doesn't exist
*/
public void snapshot(String snapshotName, String columnFamilyName) throws IOException
@@ -224,7 +226,7 @@ public class Table
* Clear all the snapshots for a given table.
*
* @param snapshotName the user supplied snapshot name. It empty or null,
- * all the snapshots will be cleaned
+ * all the snapshots will be cleaned
*/
public void clearSnapshot(String snapshotName)
{
@@ -311,7 +313,9 @@ public class Table
cfs.invalidate();
}
- /** adds a cf to internal structures, ends up creating disk files). */
+ /**
+ * adds a cf to internal structures, ends up creating disk files).
+ */
public void initCf(UUID cfId, String cfName, boolean loadSSTables)
{
if (columnFamilyStores.containsKey(cfId))
@@ -345,15 +349,15 @@ public class Table
/**
* This method appends a row to the global CommitLog, then updates memtables and indexes.
*
- * @param mutation the row to write. Must not be modified after calling apply, since commitlog append
- * may happen concurrently, depending on the CL Executor type.
+ * @param mutation the row to write. Must not be modified after calling apply, since commitlog append
+ * may happen concurrently, depending on the CL Executor type.
* @param writeCommitLog false to disable commitlog append entirely
- * @param updateIndexes false to disable index updates (used by CollationController "defragmenting")
+ * @param updateIndexes false to disable index updates (used by CollationController "defragmenting")
*/
public void apply(RowMutation mutation, boolean writeCommitLog, boolean updateIndexes)
{
- if (logger.isDebugEnabled())
- logger.debug("applying mutation of row {}", ByteBufferUtil.bytesToHex(mutation.key()));
+ if (!mutation.getTable().equals(Tracing.TRACE_KS))
+ logger.debug("applying mutation");
// write the mutation to the commitlog and memtables
switchLock.readLock().lock();
@@ -520,9 +524,9 @@ public class Table
}
/**
- * @param key row to index
- * @param cfs ColumnFamily to index row in
- * @param indexedColumns columns to index, in comparator order
+ * @param key row to index
+ * @param cfs ColumnFamily to index row in
+ * @param idxNames columns to index, in comparator order
*/
public static void indexRow(DecoratedKey key, ColumnFamilyStore cfs, Set<String> idxNames)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5c94432b/src/java/org/apache/cassandra/net/MessageOut.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/MessageOut.java b/src/java/org/apache/cassandra/net/MessageOut.java
index 78546c8..14ef377 100644
--- a/src/java/org/apache/cassandra/net/MessageOut.java
+++ b/src/java/org/apache/cassandra/net/MessageOut.java
@@ -24,13 +24,19 @@ import java.net.InetAddress;
import java.util.Collections;
import java.util.Map;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.UUIDGen;
+
+import static org.apache.cassandra.tracing.Tracing.TRACE_HEADER;
+import static org.apache.cassandra.tracing.Tracing.isTracing;
public class MessageOut<T>
{
@@ -48,14 +54,19 @@ public class MessageOut<T>
public MessageOut(MessagingService.Verb verb, T payload, IVersionedSerializer<T> serializer)
{
- this(verb, payload, serializer, Collections.<String, byte[]>emptyMap());
+ this(verb,
+ payload,
+ serializer,
+ isTracing() ? ImmutableMap.of(TRACE_HEADER, UUIDGen.decompose(Tracing.instance().getSessionId()))
+ : Collections.<String, byte[]>emptyMap());
}
- public MessageOut(MessagingService.Verb verb, T payload, IVersionedSerializer<T> serializer, Map<String, byte[]> parameters)
+ private MessageOut(MessagingService.Verb verb, T payload, IVersionedSerializer<T> serializer, Map<String, byte[]> parameters)
{
this(FBUtilities.getBroadcastAddress(), verb, payload, serializer, parameters);
}
+ @VisibleForTesting
public MessageOut(InetAddress from, MessagingService.Verb verb, T payload, IVersionedSerializer<T> serializer, Map<String, byte[]> parameters)
{
this.from = from;