cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject cassandra git commit: Add support for custom tracing implementations
Date Wed, 24 Feb 2016 18:29:23 GMT
Repository: cassandra
Updated Branches:
  refs/heads/trunk c94a9f236 -> bf25e668f


Add support for custom tracing implementations

Patch by Mick Semb Wever; reviewed by tjake for CASSANDRA-10392


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/bf25e668
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/bf25e668
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/bf25e668

Branch: refs/heads/trunk
Commit: bf25e668f416b1c279986d1b23fee2a0192d8022
Parents: c94a9f2
Author: Mick Semb Wever <mck@apache.org>
Authored: Wed Feb 24 10:26:44 2016 -0500
Committer: T Jake Luciani <jake@apache.org>
Committed: Wed Feb 24 13:28:18 2016 -0500

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 .../org/apache/cassandra/net/MessageOut.java    |   7 +-
 .../cassandra/net/OutboundTcpConnection.java    |   2 +-
 .../apache/cassandra/service/QueryState.java    |  12 +-
 .../cassandra/tracing/ExpiredTraceState.java    |  24 ++-
 .../apache/cassandra/tracing/TraceState.java    |  61 +------
 .../cassandra/tracing/TraceStateImpl.java       |  74 ++++++++
 .../org/apache/cassandra/tracing/Tracing.java   | 105 ++++++-----
 .../apache/cassandra/tracing/TracingImpl.java   |  89 ++++++++++
 .../transport/messages/ExecuteMessage.java      |   2 +-
 .../apache/cassandra/tracing/TracingTest.java   | 173 +++++++++++++++++++
 11 files changed, 437 insertions(+), 113 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/bf25e668/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 50a298e..bc21818 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 3.4
+ * Allow custom tracing implementations (CASSANDRA-10392)
  * Extract LoaderOptions to be able to be used from outside (CASSANDRA-10637)
  * fix OnDiskIndexTest to properly treat empty ranges (CASSANDRA-11205)
  * fix TrackerTest to handle new notifications (CASSANDRA-11178)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bf25e668/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 a524e7a..bc5c41b 100644
--- a/src/java/org/apache/cassandra/net/MessageOut.java
+++ b/src/java/org/apache/cassandra/net/MessageOut.java
@@ -33,10 +33,6 @@ import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.io.util.DataOutputPlus;
 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.TRACE_TYPE;
 import static org.apache.cassandra.tracing.Tracing.isTracing;
 
 public class MessageOut<T>
@@ -61,8 +57,7 @@ public class MessageOut<T>
              payload,
              serializer,
              isTracing()
-                 ? ImmutableMap.of(TRACE_HEADER, UUIDGen.decompose(Tracing.instance.getSessionId()),
-                                   TRACE_TYPE, new byte[] { Tracing.TraceType.serialize(Tracing.instance.getTraceType())
})
+                 ? Tracing.instance.getTraceHeaders()
                  : Collections.<String, byte[]>emptyMap());
     }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bf25e668/src/java/org/apache/cassandra/net/OutboundTcpConnection.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/OutboundTcpConnection.java b/src/java/org/apache/cassandra/net/OutboundTcpConnection.java
index 7b6e26e..8b1ecf3 100644
--- a/src/java/org/apache/cassandra/net/OutboundTcpConnection.java
+++ b/src/java/org/apache/cassandra/net/OutboundTcpConnection.java
@@ -284,7 +284,7 @@ public class OutboundTcpConnection extends Thread
                 {
                     byte[] traceTypeBytes = qm.message.parameters.get(Tracing.TRACE_TYPE);
                     Tracing.TraceType traceType = traceTypeBytes == null ? Tracing.TraceType.QUERY
: Tracing.TraceType.deserialize(traceTypeBytes[0]);
-                    TraceState.mutateWithTracing(ByteBuffer.wrap(sessionBytes), message,
-1, traceType.getTTL());
+                    Tracing.instance.trace(ByteBuffer.wrap(sessionBytes), message, traceType.getTTL());
                 }
                 else
                 {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bf25e668/src/java/org/apache/cassandra/service/QueryState.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/QueryState.java b/src/java/org/apache/cassandra/service/QueryState.java
index ddbc959..c70c692 100644
--- a/src/java/org/apache/cassandra/service/QueryState.java
+++ b/src/java/org/apache/cassandra/service/QueryState.java
@@ -18,6 +18,9 @@
 package org.apache.cassandra.service;
 
 import java.net.InetAddress;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ThreadLocalRandom;
 
@@ -76,14 +79,19 @@ public class QueryState
 
     public void createTracingSession()
     {
+        createTracingSession(Collections.EMPTY_MAP);
+    }
+
+    public void createTracingSession(Map<String,ByteBuffer> customPayload)
+    {
         UUID session = this.preparedTracingSession;
         if (session == null)
         {
-            Tracing.instance.newSession();
+            Tracing.instance.newSession(customPayload);
         }
         else
         {
-            Tracing.instance.newSession(session);
+            Tracing.instance.newSession(session, customPayload);
             this.preparedTracingSession = null;
         }
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bf25e668/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java b/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java
index 5cc3c21..fbe2c33 100644
--- a/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java
+++ b/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java
@@ -1,5 +1,5 @@
 /*
- * 
+ *
  * 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
@@ -7,33 +7,39 @@
  * 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.tracing;
 
-import java.util.UUID;
-
 import org.apache.cassandra.utils.FBUtilities;
 
-public class ExpiredTraceState extends TraceState
+class ExpiredTraceState extends TraceState
 {
-    public ExpiredTraceState(UUID sessionId, Tracing.TraceType traceType)
+    private final TraceState delegate;
+
+    ExpiredTraceState(TraceState delegate)
     {
-        super(FBUtilities.getBroadcastAddress(), sessionId, traceType);
+        super(FBUtilities.getBroadcastAddress(), delegate.sessionId, delegate.traceType);
+        this.delegate = delegate;
     }
 
     public int elapsed()
     {
         return -1;
     }
+
+    protected void traceImpl(String message)
+    {
+        delegate.traceImpl(message);
+    }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bf25e668/src/java/org/apache/cassandra/tracing/TraceState.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/tracing/TraceState.java b/src/java/org/apache/cassandra/tracing/TraceState.java
index e882e67..5365d09 100644
--- a/src/java/org/apache/cassandra/tracing/TraceState.java
+++ b/src/java/org/apache/cassandra/tracing/TraceState.java
@@ -19,7 +19,6 @@ package org.apache.cassandra.tracing;
 
 import java.net.InetAddress;
 import java.nio.ByteBuffer;
-import java.util.Collections;
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -29,14 +28,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 import com.google.common.base.Stopwatch;
 import org.slf4j.helpers.MessageFormatter;
 
-import org.apache.cassandra.concurrent.Stage;
-import org.apache.cassandra.concurrent.StageManager;
-import org.apache.cassandra.db.ConsistencyLevel;
-import org.apache.cassandra.db.Mutation;
-import org.apache.cassandra.exceptions.OverloadedException;
-import org.apache.cassandra.service.StorageProxy;
 import org.apache.cassandra.utils.ByteBufferUtil;
-import org.apache.cassandra.utils.WrappedRunnable;
 import org.apache.cassandra.utils.progress.ProgressEvent;
 import org.apache.cassandra.utils.progress.ProgressEventNotifier;
 import org.apache.cassandra.utils.progress.ProgressListener;
@@ -45,7 +37,7 @@ import org.apache.cassandra.utils.progress.ProgressListener;
  * ThreadLocal state for a tracing session. The presence of an instance of this class as
a ThreadLocal denotes that an
  * operation is being traced.
  */
-public class TraceState implements ProgressEventNotifier
+public abstract class TraceState implements ProgressEventNotifier
 {
     public final UUID sessionId;
     public final InetAddress coordinator;
@@ -71,7 +63,7 @@ public class TraceState implements ProgressEventNotifier
     // See CASSANDRA-7626 for more details.
     private final AtomicInteger references = new AtomicInteger(1);
 
-    public TraceState(InetAddress coordinator, UUID sessionId, Tracing.TraceType traceType)
+    protected TraceState(InetAddress coordinator, UUID sessionId, Tracing.TraceType traceType)
     {
         assert coordinator != null;
         assert sessionId != null;
@@ -83,7 +75,7 @@ public class TraceState implements ProgressEventNotifier
         this.ttl = traceType.getTTL();
         watch = Stopwatch.createStarted();
         this.status = Status.IDLE;
-}
+    }
 
     /**
      * Activate notification with provided {@code tag} name.
@@ -151,7 +143,7 @@ public class TraceState implements ProgressEventNotifier
         return status;
     }
 
-    private synchronized void notifyActivity()
+    protected synchronized void notifyActivity()
     {
         status = Status.ACTIVE;
         notifyAll();
@@ -177,10 +169,7 @@ public class TraceState implements ProgressEventNotifier
         if (notify)
             notifyActivity();
 
-        final String threadName = Thread.currentThread().getName();
-        final int elapsed = elapsed();
-
-        executeMutation(TraceKeyspace.makeEventMutation(sessionIdBytes, message, elapsed,
threadName, ttl));
+        traceImpl(message);
 
         for (ProgressListener listener : listeners)
         {
@@ -188,45 +177,7 @@ public class TraceState implements ProgressEventNotifier
         }
     }
 
-    static void executeMutation(final Mutation mutation)
-    {
-        StageManager.getStage(Stage.TRACING).execute(new WrappedRunnable()
-        {
-            protected void runMayThrow() throws Exception
-            {
-            mutateWithCatch(mutation);
-            }
-        });
-    }
-
-    /**
-     * Called from {@link org.apache.cassandra.net.OutboundTcpConnection} for non-local traces
(traces
-     * that are not initiated by local node == coordinator).
-     */
-    public static void mutateWithTracing(final ByteBuffer sessionId, final String message,
final int elapsed, final int ttl)
-    {
-        final String threadName = Thread.currentThread().getName();
-
-        StageManager.getStage(Stage.TRACING).execute(new WrappedRunnable()
-        {
-            public void runMayThrow()
-            {
-                mutateWithCatch(TraceKeyspace.makeEventMutation(sessionId, message, elapsed,
threadName, ttl));
-            }
-        });
-    }
-
-    static void mutateWithCatch(Mutation mutation)
-    {
-        try
-        {
-            StorageProxy.mutate(Collections.singletonList(mutation), ConsistencyLevel.ANY);
-        }
-        catch (OverloadedException e)
-        {
-            Tracing.logger.warn("Too many nodes are overloaded to save trace events");
-        }
-    }
+    protected abstract void traceImpl(String message);
 
     public boolean acquireReference()
     {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bf25e668/src/java/org/apache/cassandra/tracing/TraceStateImpl.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/tracing/TraceStateImpl.java b/src/java/org/apache/cassandra/tracing/TraceStateImpl.java
new file mode 100644
index 0000000..1bc210a
--- /dev/null
+++ b/src/java/org/apache/cassandra/tracing/TraceStateImpl.java
@@ -0,0 +1,74 @@
+/*
+ * 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.tracing;
+
+import java.net.InetAddress;
+import java.util.Collections;
+import java.util.UUID;
+
+import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.concurrent.StageManager;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.exceptions.OverloadedException;
+import org.apache.cassandra.service.StorageProxy;
+import org.apache.cassandra.utils.WrappedRunnable;
+
+/**
+ * ThreadLocal state for a tracing session. The presence of an instance of this class as
a ThreadLocal denotes that an
+ * operation is being traced.
+ */
+public class TraceStateImpl extends TraceState
+{
+    public TraceStateImpl(InetAddress coordinator, UUID sessionId, Tracing.TraceType traceType)
+    {
+        super(coordinator, sessionId, traceType);
+    }
+
+    protected void traceImpl(String message)
+    {
+        final String threadName = Thread.currentThread().getName();
+        final int elapsed = elapsed();
+
+        executeMutation(TraceKeyspace.makeEventMutation(sessionIdBytes, message, elapsed,
threadName, ttl));
+    }
+
+    static void executeMutation(final Mutation mutation)
+    {
+        StageManager.getStage(Stage.TRACING).execute(new WrappedRunnable()
+        {
+            protected void runMayThrow() throws Exception
+            {
+                mutateWithCatch(mutation);
+            }
+        });
+    }
+
+    static void mutateWithCatch(Mutation mutation)
+    {
+        try
+        {
+            StorageProxy.mutate(Collections.singletonList(mutation), ConsistencyLevel.ANY);
+        }
+        catch (OverloadedException e)
+        {
+            Tracing.logger.warn("Too many nodes are overloaded to save trace events");
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bf25e668/src/java/org/apache/cassandra/tracing/Tracing.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/tracing/Tracing.java b/src/java/org/apache/cassandra/tracing/Tracing.java
index bf9cee7..e69645f 100644
--- a/src/java/org/apache/cassandra/tracing/Tracing.java
+++ b/src/java/org/apache/cassandra/tracing/Tracing.java
@@ -21,11 +21,13 @@ package org.apache.cassandra.tracing;
 
 import java.net.InetAddress;
 import java.nio.ByteBuffer;
+import java.util.Collections;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
+import com.google.common.collect.ImmutableMap;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -35,14 +37,15 @@ import org.apache.cassandra.db.marshal.TimeUUIDType;
 import org.apache.cassandra.net.MessageIn;
 import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.JVMStabilityInspector;
 import org.apache.cassandra.utils.UUIDGen;
 
 
 /**
  * A trace session context. Able to track and store trace sessions. A session is usually
a user initiated query, and may
- * have multiple local and remote events before it is completed. All events and sessions
are stored at keyspace.
+ * have multiple local and remote events before it is completed.
  */
-public class Tracing implements ExecutorLocal<TraceState>
+public abstract class Tracing implements ExecutorLocal<TraceState>
 {
     public static final String TRACE_HEADER = "TraceSession";
     public static final String TRACE_TYPE = "TraceType";
@@ -77,15 +80,34 @@ public class Tracing implements ExecutorLocal<TraceState>
         }
     }
 
-    static final Logger logger = LoggerFactory.getLogger(Tracing.class);
+    protected static final Logger logger = LoggerFactory.getLogger(Tracing.class);
 
     private final InetAddress localAddress = FBUtilities.getLocalAddress();
 
     private final ThreadLocal<TraceState> state = new ThreadLocal<>();
 
-    private final ConcurrentMap<UUID, TraceState> sessions = new ConcurrentHashMap<>();
+    protected final ConcurrentMap<UUID, TraceState> sessions = new ConcurrentHashMap<>();
 
-    public static final Tracing instance = new Tracing();
+    public static final Tracing instance;
+
+    static {
+        Tracing tracing = null;
+        String customTracingClass = System.getProperty("cassandra.custom_tracing_class");
+        if (null != customTracingClass)
+        {
+            try
+            {
+                tracing = FBUtilities.construct(customTracingClass, "Tracing");
+                logger.info("Using {} as tracing queries (as requested with -Dcassandra.custom_tracing_class)",
customTracingClass);
+            }
+            catch (Exception e)
+            {
+                JVMStabilityInspector.inspectThrowable(e);
+                logger.error("Cannot use class {} for tracing ({}), ignoring by defaulting
on normal tracing", customTracingClass, e.getMessage());
+            }
+        }
+        instance = null != tracing ? tracing : new TracingImpl();
+    }
 
     public UUID getSessionId()
     {
@@ -110,30 +132,33 @@ public class Tracing implements ExecutorLocal<TraceState>
      */
     public static boolean isTracing()
     {
-        return instance.state.get() != null;
+        return instance.get() != null;
     }
 
-    public UUID newSession()
+    public UUID newSession(Map<String,ByteBuffer> customPayload)
     {
         return newSession(TraceType.QUERY);
     }
 
     public UUID newSession(TraceType traceType)
     {
-        return newSession(TimeUUIDType.instance.compose(ByteBuffer.wrap(UUIDGen.getTimeUUIDBytes())),
traceType);
+        return newSession(
+                TimeUUIDType.instance.compose(ByteBuffer.wrap(UUIDGen.getTimeUUIDBytes())),
+                traceType,
+                Collections.EMPTY_MAP);
     }
 
-    public UUID newSession(UUID sessionId)
+    public UUID newSession(UUID sessionId, Map<String,ByteBuffer> customPayload)
     {
-        return newSession(sessionId, TraceType.QUERY);
+        return newSession(sessionId, TraceType.QUERY, Collections.EMPTY_MAP);
     }
 
-    private UUID newSession(UUID sessionId, TraceType traceType)
+    protected UUID newSession(UUID sessionId, TraceType traceType, Map<String,ByteBuffer>
customPayload)
     {
-        assert state.get() == null;
+        assert get() == null;
 
-        TraceState ts = new TraceState(localAddress, sessionId, traceType);
-        state.set(ts);
+        TraceState ts = newTraceState(localAddress, sessionId, traceType);
+        set(ts);
         sessions.put(sessionId, ts);
 
         return sessionId;
@@ -145,30 +170,29 @@ public class Tracing implements ExecutorLocal<TraceState>
             sessions.remove(state.sessionId);
     }
 
+
     /**
      * Stop the session and record its complete.  Called by coodinator when request is complete.
      */
     public void stopSession()
     {
-        TraceState state = this.state.get();
+        TraceState state = get();
         if (state == null) // inline isTracing to avoid implicit two calls to state.get()
         {
             logger.trace("request complete");
         }
         else
         {
-            final int elapsed = state.elapsed();
-            final ByteBuffer sessionId = state.sessionIdBytes;
-            final int ttl = state.ttl;
-
-            TraceState.executeMutation(TraceKeyspace.makeStopSessionMutation(sessionId, elapsed,
ttl));
+            stopSessionImpl();
 
             state.stop();
             sessions.remove(state.sessionId);
-            this.state.set(null);
+            set(null);
         }
     }
 
+    protected abstract void stopSessionImpl();
+
     public TraceState get()
     {
         return state.get();
@@ -189,24 +213,11 @@ public class Tracing implements ExecutorLocal<TraceState>
         return begin(request, null, parameters);
     }
 
-    public TraceState begin(final String request, final InetAddress client, final Map<String,
String> parameters)
-    {
-        assert isTracing();
-
-        final TraceState state = this.state.get();
-        final long startedAt = System.currentTimeMillis();
-        final ByteBuffer sessionId = state.sessionIdBytes;
-        final String command = state.traceType.toString();
-        final int ttl = state.ttl;
-
-        TraceState.executeMutation(TraceKeyspace.makeStartSessionMutation(sessionId, client,
parameters, request, startedAt, command, ttl));
-
-        return state;
-    }
+    public abstract TraceState begin(String request, InetAddress client, Map<String, String>
parameters);
 
     /**
      * Determines the tracing context from a message.  Does NOT set the threadlocal state.
-     * 
+     *
      * @param message The internode message
      */
     public TraceState initializeFromMessage(final MessageIn<?> message)
@@ -218,7 +229,7 @@ public class Tracing implements ExecutorLocal<TraceState>
 
         assert sessionBytes.length == 16;
         UUID sessionId = UUIDGen.getUUID(ByteBuffer.wrap(sessionBytes));
-        TraceState ts = sessions.get(sessionId);
+        TraceState ts = get(sessionId);
         if (ts != null && ts.acquireReference())
             return ts;
 
@@ -230,16 +241,26 @@ public class Tracing implements ExecutorLocal<TraceState>
         if (message.verb == MessagingService.Verb.REQUEST_RESPONSE)
         {
             // received a message for a session we've already closed out.  see CASSANDRA-5668
-            return new ExpiredTraceState(sessionId, traceType);
+            return new ExpiredTraceState(newTraceState(message.from, sessionId, traceType));
         }
         else
         {
-            ts = new TraceState(message.from, sessionId, traceType);
+            ts = newTraceState(message.from, sessionId, traceType);
             sessions.put(sessionId, ts);
             return ts;
         }
     }
 
+    public Map<String, byte[]> getTraceHeaders()
+    {
+        assert isTracing();
+
+        return ImmutableMap.of(
+                TRACE_HEADER, UUIDGen.decompose(Tracing.instance.getSessionId()),
+                TRACE_TYPE, new byte[] { Tracing.TraceType.serialize(Tracing.instance.getTraceType())
});
+    }
+
+    protected abstract TraceState newTraceState(InetAddress coordinator, UUID sessionId,
Tracing.TraceType traceType);
 
     // repair just gets a varargs method since it's so heavyweight anyway
     public static void traceRepair(String format, Object... args)
@@ -287,4 +308,10 @@ public class Tracing implements ExecutorLocal<TraceState>
 
         state.trace(format, args);
     }
+
+    /**
+     * Called from {@link org.apache.cassandra.net.OutboundTcpConnection} for non-local traces
(traces
+     * that are not initiated by local node == coordinator).
+     */
+    public abstract void trace(ByteBuffer sessionId, String message, int ttl);
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bf25e668/src/java/org/apache/cassandra/tracing/TracingImpl.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/tracing/TracingImpl.java b/src/java/org/apache/cassandra/tracing/TracingImpl.java
new file mode 100644
index 0000000..52ac183
--- /dev/null
+++ b/src/java/org/apache/cassandra/tracing/TracingImpl.java
@@ -0,0 +1,89 @@
+/*
+ * 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.tracing;
+
+import java.net.InetAddress;
+import java.nio.ByteBuffer;
+import java.util.Map;
+import java.util.UUID;
+
+import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.concurrent.StageManager;
+import org.apache.cassandra.utils.WrappedRunnable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+
+/**
+ * A trace session context. Able to track and store trace sessions. A session is usually
a user initiated query, and may
+ * have multiple local and remote events before it is completed. All events and sessions
are stored at keyspace.
+ */
+class TracingImpl extends Tracing
+{
+    private static final Logger logger = LoggerFactory.getLogger(TracingImpl.class);
+
+    public void stopSessionImpl() {
+        TraceState state = get();
+        int elapsed = state.elapsed();
+        ByteBuffer sessionId = state.sessionIdBytes;
+        int ttl = state.ttl;
+        TraceStateImpl.executeMutation(TraceKeyspace.makeStopSessionMutation(sessionId, elapsed,
ttl));
+    }
+
+    public TraceState begin(final String request, final InetAddress client, final Map<String,
String> parameters)
+    {
+        assert isTracing();
+
+        final TraceState state = get();
+        final long startedAt = System.currentTimeMillis();
+        final ByteBuffer sessionId = state.sessionIdBytes;
+        final String command = state.traceType.toString();
+        final int ttl = state.ttl;
+
+        TraceStateImpl.executeMutation(TraceKeyspace.makeStartSessionMutation(sessionId,
client, parameters, request, startedAt, command, ttl));
+
+        return state;
+    }
+
+    @Override
+    protected TraceState newTraceState(InetAddress coordinator, UUID sessionId, TraceType
traceType)
+    {
+        return new TraceStateImpl(coordinator, sessionId, traceType);
+    }
+
+    /**
+     * Called from {@link org.apache.cassandra.net.OutboundTcpConnection} for non-local traces
(traces
+     * that are not initiated by local node == coordinator).
+     */
+    public void trace(final ByteBuffer sessionId, final String message, final int ttl)
+    {
+        final String threadName = Thread.currentThread().getName();
+
+        StageManager.getStage(Stage.TRACING).execute(new WrappedRunnable()
+        {
+            public void runMayThrow()
+            {
+                TraceStateImpl.mutateWithCatch(TraceKeyspace.makeEventMutation(sessionId,
message, -1, threadName, ttl));
+            }
+        });
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bf25e668/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java b/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java
index e9923b4..df6acc0 100644
--- a/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java
+++ b/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java
@@ -110,7 +110,7 @@ public class ExecuteMessage extends Message.Request
 
             if (state.traceNextQuery())
             {
-                state.createTracingSession();
+                state.createTracingSession(getCustomPayload());
 
                 ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
                 if (options.getPageSize() > 0)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bf25e668/test/unit/org/apache/cassandra/tracing/TracingTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/tracing/TracingTest.java b/test/unit/org/apache/cassandra/tracing/TracingTest.java
new file mode 100644
index 0000000..1b0e507
--- /dev/null
+++ b/test/unit/org/apache/cassandra/tracing/TracingTest.java
@@ -0,0 +1,173 @@
+/*
+ * 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.tracing;
+
+import java.net.InetAddress;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import org.junit.Test;
+
+import org.apache.cassandra.utils.progress.ProgressEvent;
+import org.apache.cassandra.utils.progress.ProgressListener;
+
+public final class TracingTest
+{
+
+    @Test
+    public void test()
+    {
+        List<String> traces = new ArrayList<>();
+        Tracing tracing = new TracingImpl(traces);
+        tracing.newSession(Tracing.TraceType.NONE);
+        TraceState state = tracing.begin("test-request", Collections.<String,String>emptyMap());
+        state.trace("test-1");
+        state.trace("test-2");
+        state.trace("test-3");
+        tracing.stopSession();
+
+        assert null == tracing.get();
+        assert 4 == traces.size();
+        assert "test-request".equals(traces.get(0));
+        assert "test-1".equals(traces.get(1));
+        assert "test-2".equals(traces.get(2));
+        assert "test-3".equals(traces.get(3));
+    }
+
+    @Test
+    public void test_get()
+    {
+        List<String> traces = new ArrayList<>();
+        Tracing tracing = new TracingImpl(traces);
+        tracing.newSession(Tracing.TraceType.NONE);
+        tracing.begin("test-request", Collections.<String,String>emptyMap());
+        tracing.get().trace("test-1");
+        tracing.get().trace("test-2");
+        tracing.get().trace("test-3");
+        tracing.stopSession();
+
+        assert null == tracing.get();
+        assert 4 == traces.size();
+        assert "test-request".equals(traces.get(0));
+        assert "test-1".equals(traces.get(1));
+        assert "test-2".equals(traces.get(2));
+        assert "test-3".equals(traces.get(3));
+    }
+
+    @Test
+    public void test_get_uuid()
+    {
+        List<String> traces = new ArrayList<>();
+        Tracing tracing = new TracingImpl(traces);
+        UUID uuid = tracing.newSession(Tracing.TraceType.NONE);
+        tracing.begin("test-request", Collections.<String,String>emptyMap());
+        tracing.get(uuid).trace("test-1");
+        tracing.get(uuid).trace("test-2");
+        tracing.get(uuid).trace("test-3");
+        tracing.stopSession();
+
+        assert null == tracing.get();
+        assert 4 == traces.size();
+        assert "test-request".equals(traces.get(0));
+        assert "test-1".equals(traces.get(1));
+        assert "test-2".equals(traces.get(2));
+        assert "test-3".equals(traces.get(3));
+    }
+
+    @Test
+    public void test_states()
+    {
+        List<String> traces = new ArrayList<>();
+        Tracing tracing = new TracingImpl(traces);
+        tracing.newSession(Tracing.TraceType.REPAIR);
+        tracing.begin("test-request", Collections.<String,String>emptyMap());
+        tracing.get().enableActivityNotification("test-tag");
+        assert TraceState.Status.IDLE == tracing.get().waitActivity(1);
+        tracing.get().trace("test-1");
+        assert TraceState.Status.ACTIVE == tracing.get().waitActivity(1);
+        tracing.get().stop();
+        assert TraceState.Status.STOPPED == tracing.get().waitActivity(1);
+        tracing.stopSession();
+        assert null == tracing.get();
+    }
+
+    @Test
+    public void test_progress_listener()
+    {
+        List<String> traces = new ArrayList<>();
+        Tracing tracing = new TracingImpl(traces);
+        tracing.newSession(Tracing.TraceType.REPAIR);
+        tracing.begin("test-request", Collections.<String,String>emptyMap());
+        tracing.get().enableActivityNotification("test-tag");
+
+        tracing.get().addProgressListener(
+                new ProgressListener()
+                {
+                    public void progress(String tag, ProgressEvent pe)
+                    {
+                        assert "test-tag".equals(tag);
+                        assert "test-trace".equals(pe.getMessage());
+                    }
+                });
+
+        tracing.get().trace("test-trace");
+        tracing.stopSession();
+        assert null == tracing.get();
+    }
+
+    private class TracingImpl extends Tracing
+    {
+        private final List<String> traces;
+
+        public TracingImpl(List<String> traces)
+        {
+            this.traces = traces;
+        }
+
+        public void stopSessionImpl()
+        {}
+
+        public TraceState begin(String request, InetAddress ia, Map<String, String>
map)
+        {
+            traces.add(request);
+            return get();
+        }
+
+        protected TraceState newTraceState(InetAddress ia, UUID uuid, Tracing.TraceType tt)
+        {
+            return new TraceState(ia, uuid, tt)
+            {
+                protected void traceImpl(String string)
+                {
+                    traces.add(string);
+                }
+
+            };
+        }
+
+        public void trace(ByteBuffer bb, String string, int i)
+        {
+            throw new UnsupportedOperationException("Not supported yet.");
+        }
+    }
+}


Mime
View raw message