You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2017/03/22 12:30:02 UTC
camel git commit: Support for async response handling and
multicast/parallel. Included local implementation of SpanManager to add
support for transferring managed span stack between threads.
Repository: camel
Updated Branches:
refs/heads/master f2cc3de50 -> a8910d540
Support for async response handling and multicast/parallel. Included local implementation of SpanManager to add support for transferring managed span stack between threads.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/a8910d54
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/a8910d54
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/a8910d54
Branch: refs/heads/master
Commit: a8910d5400e33a4a832e878932c7d14d43021cff
Parents: f2cc3de
Author: Gary Brown <ga...@brownuk.com>
Authored: Mon Mar 20 17:14:11 2017 +0000
Committer: Claus Ibsen <da...@apache.org>
Committed: Wed Mar 22 13:29:54 2017 +0100
----------------------------------------------------------------------
.../camel/opentracing/OpenTracingTracer.java | 50 +++--
.../concurrent/CamelSpanManager.java | 207 +++++++++++++++++++
.../OpenTracingExecutorServiceManager.java | 200 ++++++++++++++++++
.../CamelOpenTracingTestSupport.java | 75 ++++---
.../opentracing/MulticastParallelRouteTest.java | 80 +++++++
5 files changed, 574 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/a8910d54/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/OpenTracingTracer.java
----------------------------------------------------------------------
diff --git a/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/OpenTracingTracer.java b/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/OpenTracingTracer.java
index e662d24..d2f3b28 100644
--- a/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/OpenTracingTracer.java
+++ b/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/OpenTracingTracer.java
@@ -27,7 +27,6 @@ import io.opentracing.Span;
import io.opentracing.Tracer;
import io.opentracing.Tracer.SpanBuilder;
import io.opentracing.contrib.global.GlobalTracer;
-import io.opentracing.contrib.spanmanager.DefaultSpanManager;
import io.opentracing.contrib.spanmanager.SpanManager;
import io.opentracing.propagation.Format;
import io.opentracing.tag.Tags;
@@ -41,6 +40,8 @@ import org.apache.camel.api.management.ManagedResource;
import org.apache.camel.management.event.ExchangeSendingEvent;
import org.apache.camel.management.event.ExchangeSentEvent;
import org.apache.camel.model.RouteDefinition;
+import org.apache.camel.opentracing.concurrent.CamelSpanManager;
+import org.apache.camel.opentracing.concurrent.OpenTracingExecutorServiceManager;
import org.apache.camel.opentracing.propagation.CamelHeadersExtractAdapter;
import org.apache.camel.opentracing.propagation.CamelHeadersInjectAdapter;
import org.apache.camel.spi.RoutePolicy;
@@ -65,10 +66,12 @@ public class OpenTracingTracer extends ServiceSupport implements RoutePolicyFact
private static final Logger LOG = LoggerFactory.getLogger(OpenTracingTracer.class);
+ private static final String MANAGED_SPAN_PROPERTY = "ManagedSpan";
+
private static Map<String, SpanDecorator> decorators = new HashMap<>();
private final OpenTracingEventNotifier eventNotifier = new OpenTracingEventNotifier();
- private final SpanManager spanManager = DefaultSpanManager.getInstance();
+ private final CamelSpanManager spanManager = CamelSpanManager.getInstance();
private Tracer tracer;
private CamelContext camelContext;
@@ -103,6 +106,10 @@ public class OpenTracingTracer extends ServiceSupport implements RoutePolicyFact
try {
// start this service eager so we init before Camel is starting up
camelContext.addService(this, true, true);
+
+ // Wrap the ExecutorServiceManager with a SpanManager aware version
+ camelContext.setExecutorServiceManager(
+ new OpenTracingExecutorServiceManager(camelContext.getExecutorServiceManager(), spanManager));
} catch (Exception e) {
throw ObjectHelper.wrapRuntimeCamelException(e);
}
@@ -187,18 +194,26 @@ public class OpenTracingTracer extends ServiceSupport implements RoutePolicyFact
sd.pre(span, ese.getExchange(), ese.getEndpoint());
tracer.inject(span.context(), Format.Builtin.TEXT_MAP,
new CamelHeadersInjectAdapter(ese.getExchange().getIn().getHeaders()));
- spanManager.activate(span);
+ ese.getExchange().setProperty(MANAGED_SPAN_PROPERTY, spanManager.activate(span));
+ spanManager.clear();
if (LOG.isTraceEnabled()) {
LOG.trace("OpenTracing: start client span=" + span);
}
} else if (event instanceof ExchangeSentEvent) {
- SpanManager.ManagedSpan managedSpan = spanManager.current();
+ ExchangeSentEvent ese = (ExchangeSentEvent) event;
+ SpanManager.ManagedSpan managedSpan = (SpanManager.ManagedSpan)
+ ese.getExchange().getProperty(MANAGED_SPAN_PROPERTY);
+ if (managedSpan != null) {
+ spanManager.activate(managedSpan);
+ ese.getExchange().setProperty(MANAGED_SPAN_PROPERTY, null);
+ } else {
+ managedSpan = spanManager.current();
+ }
if (LOG.isTraceEnabled()) {
LOG.trace("OpenTracing: start client span=" + managedSpan.getSpan());
}
- SpanDecorator sd = getSpanDecorator(((ExchangeSentEvent) event).getEndpoint());
- sd.post(managedSpan.getSpan(), ((ExchangeSentEvent) event).getExchange(),
- ((ExchangeSentEvent) event).getEndpoint());
+ SpanDecorator sd = getSpanDecorator(ese.getEndpoint());
+ sd.post(managedSpan.getSpan(), ese.getExchange(), ese.getEndpoint());
managedSpan.getSpan().finish();
managedSpan.deactivate();
}
@@ -223,6 +238,11 @@ public class OpenTracingTracer extends ServiceSupport implements RoutePolicyFact
@Override
public void onExchangeBegin(Route route, Exchange exchange) {
+ // Check if continuing exchange on same thread
+ if (exchange.getProperties().containsKey(MANAGED_SPAN_PROPERTY)) {
+ spanManager.activate((SpanManager.ManagedSpan)exchange.getProperty(MANAGED_SPAN_PROPERTY));
+ exchange.setProperty(MANAGED_SPAN_PROPERTY, null);
+ }
SpanDecorator sd = getSpanDecorator(route.getEndpoint());
Span span = tracer.buildSpan(sd.getOperationName(exchange, route.getEndpoint()))
.asChildOf(tracer.extract(Format.Builtin.TEXT_MAP,
@@ -239,13 +259,17 @@ public class OpenTracingTracer extends ServiceSupport implements RoutePolicyFact
@Override
public void onExchangeDone(Route route, Exchange exchange) {
SpanManager.ManagedSpan managedSpan = spanManager.current();
- if (LOG.isTraceEnabled()) {
- LOG.trace("OpenTracing: finish server span=" + managedSpan.getSpan());
+ if (managedSpan.getSpan() != null) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("OpenTracing: finish server span=" + managedSpan.getSpan());
+ }
+ SpanDecorator sd = getSpanDecorator(route.getEndpoint());
+ sd.post(managedSpan.getSpan(), exchange, route.getEndpoint());
+ managedSpan.getSpan().finish();
+ managedSpan.deactivate();
+ } else {
+ LOG.warn("OpenTracing: could not find managed span for exchange=" + exchange);
}
- SpanDecorator sd = getSpanDecorator(route.getEndpoint());
- sd.post(managedSpan.getSpan(), exchange, route.getEndpoint());
- managedSpan.getSpan().finish();
- managedSpan.deactivate();
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/a8910d54/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/concurrent/CamelSpanManager.java
----------------------------------------------------------------------
diff --git a/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/concurrent/CamelSpanManager.java b/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/concurrent/CamelSpanManager.java
new file mode 100644
index 0000000..c5c34af
--- /dev/null
+++ b/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/concurrent/CamelSpanManager.java
@@ -0,0 +1,207 @@
+/**
+ * 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.camel.opentracing.concurrent;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import io.opentracing.NoopSpan;
+import io.opentracing.Span;
+import io.opentracing.contrib.spanmanager.SpanManager;
+
+/**
+ * Camel {@link SpanManager} implementation using {@link ThreadLocal} storage
+ * maintaining a stack-like structure of linked managed spans.
+ * <p>
+ * The linked managed spans provide the following stack unwinding algorithm:
+ * <ol>
+ * <li>If the deactivated span is not the <em>managed</em> span, the <em>current managed</em> span is left alone.</li>
+ * <li>Otherwise, the first parent that is <em>not yet deactivated</em> is set as the new managed span.</li>
+ * <li>If no managed parents remain, the <em>managed span</em> is cleared.</li>
+ * <li>Consecutive <code>deactivate()</code> calls for already-deactivated spans will be ignored.</li>
+ * </ol>
+ * <p>
+ * NOTE: This implementation has been copied and extended from opentracing-contrib/java-spanmanager
+ * project, to provide the additional functionality for transferring the managed spans from one thread
+ * to another. This functionality will soon be provided by the core opentracing-java project, at which
+ * time this implementation will be removed.
+ */
+public final class CamelSpanManager implements SpanManager {
+
+ private static final Logger LOGGER = Logger.getLogger(CamelSpanManager.class.getName());
+ private static final CamelSpanManager INSTANCE = new CamelSpanManager();
+ private static final ManagedSpan NO_MANAGED_SPAN = new NoManagedSpan();
+
+ private final ThreadLocal<LinkedManagedSpan> managed = new ThreadLocal<LinkedManagedSpan>();
+
+ private CamelSpanManager() {
+ }
+
+ /**
+ * @return The singleton instance of the camel span manager.
+ */
+ public static CamelSpanManager getInstance() {
+ return INSTANCE;
+ }
+
+ /**
+ * Stack unwinding algorithm that refreshes the currently managed span.
+ * <p>
+ * See {@link CamelSpanManager class javadoc} for a full description.
+ *
+ * @return The current non-deactivated LinkedManagedSpan or <code>null</code> if none remained.
+ */
+ private LinkedManagedSpan refreshCurrent() {
+ LinkedManagedSpan managedSpan = managed.get();
+ LinkedManagedSpan current = managedSpan;
+ while (current != null && current.deactivated.get()) { // Unwind stack if necessary.
+ current = current.parent;
+ }
+ if (current != managedSpan) { // refresh current if necessary.
+ if (current == null) {
+ managed.remove();
+ } else {
+ managed.set(current);
+ }
+ }
+ return current;
+ }
+
+ @Override
+ public ManagedSpan activate(Span span) {
+ LinkedManagedSpan managedSpan = new LinkedManagedSpan(span, refreshCurrent());
+ managed.set(managedSpan);
+ return managedSpan;
+ }
+
+ /**
+ * This method associates the supplied managed span with the current
+ * execution context (thread).
+ *
+ * @param managedSpan The managed span
+ */
+ public void activate(ManagedSpan managedSpan) {
+ if (managedSpan instanceof LinkedManagedSpan) {
+ managed.set((LinkedManagedSpan)managedSpan);
+ }
+ }
+
+ @Override
+ public ManagedSpan current() {
+ LinkedManagedSpan current = refreshCurrent();
+ return current != null ? current : NO_MANAGED_SPAN;
+ }
+
+ @Override
+ public void clear() {
+ managed.remove();
+ }
+
+ @Override
+ @Deprecated
+ public Span currentSpan() {
+ ManagedSpan current = current();
+ return current.getSpan() != null ? current.getSpan() : NoopSpan.INSTANCE;
+ }
+
+ /**
+ * @see #activate(Span)
+ * @deprecated renamed to activate()
+ */
+ @Deprecated
+ public ManagedSpan manage(Span span) {
+ return activate(span);
+ }
+
+ @Override
+ public String toString() {
+ return getClass().getSimpleName();
+ }
+
+ private final class LinkedManagedSpan implements ManagedSpan {
+ private final LinkedManagedSpan parent;
+ private final Span span;
+ private final AtomicBoolean deactivated = new AtomicBoolean(false);
+
+ private LinkedManagedSpan(Span span, LinkedManagedSpan parent) {
+ this.parent = parent;
+ this.span = span;
+ }
+
+ @Override
+ public Span getSpan() {
+ return span;
+ }
+
+ public void deactivate() {
+ if (deactivated.compareAndSet(false, true)) {
+ LinkedManagedSpan current = refreshCurrent(); // Trigger stack-unwinding algorithm.
+ LOGGER.log(Level.FINER, "Released {0}, current span is {1}.", new Object[]{this, current});
+ } else {
+ LOGGER.log(Level.FINEST, "No action needed, {0} was already deactivated.", this);
+ }
+ }
+
+ @Override
+ public void close() {
+ deactivate();
+ }
+
+ /**
+ * @see #deactivate()
+ * @deprecated renamed to deactivate()
+ */
+ @Deprecated
+ public void release() {
+ deactivate();
+ }
+
+ @Override
+ public String toString() {
+ return getClass().getSimpleName() + '{' + span + '}';
+ }
+ }
+
+ /**
+ * Empty implementation signifying there is no managed span.
+ */
+ private static final class NoManagedSpan implements ManagedSpan {
+ private NoManagedSpan() {
+ }
+
+ @Override
+ public Span getSpan() {
+ return null;
+ }
+
+ @Override
+ public void deactivate() {
+ // no-op
+ }
+
+ @Override
+ public void close() {
+ // no-op
+ }
+
+ @Override
+ public String toString() {
+ return getClass().getSimpleName();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a8910d54/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/concurrent/OpenTracingExecutorServiceManager.java
----------------------------------------------------------------------
diff --git a/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/concurrent/OpenTracingExecutorServiceManager.java b/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/concurrent/OpenTracingExecutorServiceManager.java
new file mode 100644
index 0000000..2b19a88
--- /dev/null
+++ b/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/concurrent/OpenTracingExecutorServiceManager.java
@@ -0,0 +1,200 @@
+/**
+ * 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.camel.opentracing.concurrent;
+
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+
+import io.opentracing.contrib.spanmanager.SpanManager;
+import io.opentracing.contrib.spanmanager.concurrent.SpanPropagatingExecutorService;
+
+import org.apache.camel.spi.ExecutorServiceManager;
+import org.apache.camel.spi.ThreadPoolFactory;
+import org.apache.camel.spi.ThreadPoolProfile;
+
+public class OpenTracingExecutorServiceManager implements ExecutorServiceManager {
+
+ private final ExecutorServiceManager delegate;
+ private final SpanManager spanManager;
+
+ public OpenTracingExecutorServiceManager(ExecutorServiceManager delegate, SpanManager spanManager) {
+ this.delegate = delegate;
+ this.spanManager = spanManager;
+ }
+
+ @Override
+ public void shutdown() throws Exception {
+ delegate.shutdown();
+ }
+
+ @Override
+ public void start() throws Exception {
+ delegate.start();
+ }
+
+ @Override
+ public void stop() throws Exception {
+ delegate.stop();
+ }
+
+ @Override
+ public boolean awaitTermination(ExecutorService arg0, long arg1) throws InterruptedException {
+ return delegate.awaitTermination(arg0, arg1);
+ }
+
+ @Override
+ public ThreadPoolProfile getDefaultThreadPoolProfile() {
+ return delegate.getDefaultThreadPoolProfile();
+ }
+
+ @Override
+ public long getShutdownAwaitTermination() {
+ return delegate.getShutdownAwaitTermination();
+ }
+
+ @Override
+ public String getThreadNamePattern() {
+ return delegate.getThreadNamePattern();
+ }
+
+ @Override
+ public ThreadPoolFactory getThreadPoolFactory() {
+ return delegate.getThreadPoolFactory();
+ }
+
+ @Override
+ public ThreadPoolProfile getThreadPoolProfile(String arg0) {
+ return delegate.getThreadPoolProfile(arg0);
+ }
+
+ @Override
+ public ExecutorService newCachedThreadPool(Object arg0, String arg1) {
+ return new SpanPropagatingExecutorService(delegate.newCachedThreadPool(arg0, arg1), spanManager);
+ }
+
+ @Override
+ public ScheduledExecutorService newDefaultScheduledThreadPool(Object arg0, String arg1) {
+ return delegate.newDefaultScheduledThreadPool(arg0, arg1);
+ }
+
+ @Override
+ public ExecutorService newDefaultThreadPool(Object arg0, String arg1) {
+ return new SpanPropagatingExecutorService(delegate.newDefaultThreadPool(arg0, arg1), spanManager);
+ }
+
+ @Override
+ public ExecutorService newFixedThreadPool(Object arg0, String arg1, int arg2) {
+ return new SpanPropagatingExecutorService(delegate.newFixedThreadPool(arg0, arg1, arg2), spanManager);
+ }
+
+ @Override
+ public ScheduledExecutorService newScheduledThreadPool(Object arg0, String arg1, int arg2) {
+ return delegate.newScheduledThreadPool(arg0, arg1, arg2);
+ }
+
+ @Override
+ public ScheduledExecutorService newScheduledThreadPool(Object arg0, String arg1, ThreadPoolProfile arg2) {
+ return delegate.newScheduledThreadPool(arg0, arg1, arg2);
+ }
+
+ @Override
+ public ScheduledExecutorService newScheduledThreadPool(Object arg0, String arg1, String arg2) {
+ return delegate.newScheduledThreadPool(arg0, arg1, arg2);
+ }
+
+ @Override
+ public ExecutorService newSingleThreadExecutor(Object arg0, String arg1) {
+ return new SpanPropagatingExecutorService(delegate.newSingleThreadExecutor(arg0, arg1), spanManager);
+ }
+
+ @Override
+ public ScheduledExecutorService newSingleThreadScheduledExecutor(Object arg0, String arg1) {
+ return delegate.newSingleThreadScheduledExecutor(arg0, arg1);
+ }
+
+ @Override
+ public Thread newThread(String arg0, Runnable arg1) {
+ return delegate.newThread(arg0, arg1);
+ }
+
+ @Override
+ public ExecutorService newThreadPool(Object arg0, String arg1, ThreadPoolProfile arg2) {
+ return new SpanPropagatingExecutorService(delegate.newThreadPool(arg0, arg1, arg2), spanManager);
+ }
+
+ @Override
+ public ExecutorService newThreadPool(Object arg0, String arg1, String arg2) {
+ return new SpanPropagatingExecutorService(delegate.newThreadPool(arg0, arg1, arg2), spanManager);
+ }
+
+ @Override
+ public ExecutorService newThreadPool(Object arg0, String arg1, int arg2, int arg3) {
+ return new SpanPropagatingExecutorService(delegate.newThreadPool(arg0, arg1, arg2, arg3), spanManager);
+ }
+
+ @Override
+ public void registerThreadPoolProfile(ThreadPoolProfile arg0) {
+ delegate.registerThreadPoolProfile(arg0);
+ }
+
+ @Override
+ public String resolveThreadName(String arg0) {
+ return delegate.resolveThreadName(arg0);
+ }
+
+ @Override
+ public void setDefaultThreadPoolProfile(ThreadPoolProfile arg0) {
+ delegate.setDefaultThreadPoolProfile(arg0);
+ }
+
+ @Override
+ public void setShutdownAwaitTermination(long arg0) {
+ delegate.setShutdownAwaitTermination(arg0);
+ }
+
+ @Override
+ public void setThreadNamePattern(String arg0) throws IllegalArgumentException {
+ delegate.setThreadNamePattern(arg0);
+ }
+
+ @Override
+ public void setThreadPoolFactory(ThreadPoolFactory arg0) {
+ delegate.setThreadPoolFactory(arg0);
+ }
+
+ @Override
+ public void shutdown(ExecutorService arg0) {
+ delegate.shutdown(arg0);
+ }
+
+ @Override
+ public void shutdownGraceful(ExecutorService arg0) {
+ delegate.shutdownGraceful(arg0);
+ }
+
+ @Override
+ public void shutdownGraceful(ExecutorService arg0, long arg1) {
+ delegate.shutdownGraceful(arg0, arg1);
+ }
+
+ @Override
+ public List<Runnable> shutdownNow(ExecutorService arg0) {
+ return delegate.shutdownNow(arg0);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a8910d54/components/camel-opentracing/src/test/java/org/apache/camel/opentracing/CamelOpenTracingTestSupport.java
----------------------------------------------------------------------
diff --git a/components/camel-opentracing/src/test/java/org/apache/camel/opentracing/CamelOpenTracingTestSupport.java b/components/camel-opentracing/src/test/java/org/apache/camel/opentracing/CamelOpenTracingTestSupport.java
index 6c7ad98..a48cee0 100644
--- a/components/camel-opentracing/src/test/java/org/apache/camel/opentracing/CamelOpenTracingTestSupport.java
+++ b/components/camel-opentracing/src/test/java/org/apache/camel/opentracing/CamelOpenTracingTestSupport.java
@@ -18,11 +18,15 @@ package org.apache.camel.opentracing;
import java.net.URI;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
import io.opentracing.Span;
+import io.opentracing.mock.MockSpan;
import io.opentracing.mock.MockTracer;
import io.opentracing.mock.MockTracer.Propagator;
import io.opentracing.tag.Tags;
@@ -63,42 +67,63 @@ public class CamelOpenTracingTestSupport extends CamelTestSupport {
}
protected void verify() {
+ verify(false);
+ }
+
+ protected void verify(boolean async) {
assertEquals("Incorrect number of spans", testdata.length, tracer.finishedSpans().size());
+ verifySameTrace();
+
+ List<MockSpan> spans = tracer.finishedSpans();
+ if (async) {
+ final List<MockSpan> unsortedSpans = spans;
+ spans = Arrays.asList(testdata).stream()
+ .map(td -> findSpan(td, unsortedSpans)).distinct().collect(Collectors.toList());
+ assertEquals("Incorrect number of spans after sorting", testdata.length, spans.size());
+ }
+
for (int i = 0; i < testdata.length; i++) {
- if (i > 0) {
- assertEquals(testdata[i].getLabel(), tracer.finishedSpans().get(0).context().traceId(),
- tracer.finishedSpans().get(i).context().traceId());
- }
+ verifySpan(i, testdata, spans);
+ }
+ }
- String component = (String) tracer.finishedSpans().get(i).tags().get(Tags.COMPONENT.getKey());
- assertNotNull(component);
- assertEquals(testdata[i].getLabel(),
- SpanDecorator.CAMEL_COMPONENT + URI.create((String) testdata[i].getUri()).getScheme(),
- component);
- assertEquals(testdata[i].getLabel(), testdata[i].getUri(),
- tracer.finishedSpans().get(i).tags().get("camel.uri"));
-
- // If span associated with TestSEDASpanDecorator, check that pre/post tags have been defined
- if ("camel-seda".equals(component)) {
- assertTrue(tracer.finishedSpans().get(i).tags().containsKey("pre"));
- assertTrue(tracer.finishedSpans().get(i).tags().containsKey("post"));
- }
+ protected MockSpan findSpan(SpanTestData testdata, List<MockSpan> spans) {
+ return spans.stream().filter(s -> s.operationName().equals(testdata.getOperation())
+ && s.tags().get("camel.uri").equals(testdata.getUri())
+ && s.tags().get(Tags.SPAN_KIND.getKey()).equals(testdata.getKind())).findFirst().orElse(null);
+ }
- assertEquals(testdata[i].getLabel(), testdata[i].getOperation(), tracer.finishedSpans().get(i).operationName());
+ protected void verifySpan(int index, SpanTestData[] testdata, List<MockSpan> spans) {
+ String component = (String) spans.get(index).tags().get(Tags.COMPONENT.getKey());
+ assertNotNull(component);
+ assertEquals(testdata[index].getLabel(),
+ SpanDecorator.CAMEL_COMPONENT + URI.create((String) testdata[index].getUri()).getScheme(),
+ component);
+ assertEquals(testdata[index].getLabel(), testdata[index].getUri(), spans.get(index).tags().get("camel.uri"));
+
+ // If span associated with TestSEDASpanDecorator, check that pre/post tags have been defined
+ if ("camel-seda".equals(component)) {
+ assertTrue(spans.get(index).tags().containsKey("pre"));
+ assertTrue(spans.get(index).tags().containsKey("post"));
+ }
- assertEquals(testdata[i].getLabel(), testdata[i].getKind(),
- tracer.finishedSpans().get(i).tags().get(Tags.SPAN_KIND.getKey()));
+ assertEquals(testdata[index].getLabel(), testdata[index].getOperation(), spans.get(index).operationName());
- if (testdata[i].getParentId() != -1) {
- assertEquals(testdata[i].getLabel(),
- tracer.finishedSpans().get(testdata[i].getParentId()).context().spanId(),
- tracer.finishedSpans().get(i).parentId());
- }
+ assertEquals(testdata[index].getLabel(), testdata[index].getKind(),
+ spans.get(index).tags().get(Tags.SPAN_KIND.getKey()));
+ if (testdata[index].getParentId() != -1) {
+ assertEquals(testdata[index].getLabel(),
+ spans.get(testdata[index].getParentId()).context().spanId(),
+ spans.get(index).parentId());
}
}
+ protected void verifySameTrace() {
+ assertEquals(1, tracer.finishedSpans().stream().map(s -> s.context().traceId()).distinct().count());
+ }
+
protected void verifyTraceSpanNumbers(int numOfTraces, int numSpansPerTrace) {
Map<Long, List<Span>> traces = new HashMap<Long, List<Span>>();
http://git-wip-us.apache.org/repos/asf/camel/blob/a8910d54/components/camel-opentracing/src/test/java/org/apache/camel/opentracing/MulticastParallelRouteTest.java
----------------------------------------------------------------------
diff --git a/components/camel-opentracing/src/test/java/org/apache/camel/opentracing/MulticastParallelRouteTest.java b/components/camel-opentracing/src/test/java/org/apache/camel/opentracing/MulticastParallelRouteTest.java
new file mode 100644
index 0000000..856f931
--- /dev/null
+++ b/components/camel-opentracing/src/test/java/org/apache/camel/opentracing/MulticastParallelRouteTest.java
@@ -0,0 +1,80 @@
+/**
+ * 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.camel.opentracing;
+
+import io.opentracing.tag.Tags;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.Test;
+
+public class MulticastParallelRouteTest extends CamelOpenTracingTestSupport {
+
+ private static SpanTestData[] testdata = {
+ new SpanTestData().setLabel("seda:b server").setUri("seda://b").setOperation("b")
+ .setKind(Tags.SPAN_KIND_SERVER).setParentId(1),
+ new SpanTestData().setLabel("seda:b client").setUri("seda://b").setOperation("b")
+ .setKind(Tags.SPAN_KIND_CLIENT).setParentId(4),
+ new SpanTestData().setLabel("seda:c server").setUri("seda://c").setOperation("c")
+ .setKind(Tags.SPAN_KIND_SERVER).setParentId(3),
+ new SpanTestData().setLabel("seda:c client").setUri("seda://c").setOperation("c")
+ .setKind(Tags.SPAN_KIND_CLIENT).setParentId(4),
+ new SpanTestData().setLabel("seda:a server").setUri("seda://a").setOperation("a")
+ .setKind(Tags.SPAN_KIND_SERVER).setParentId(5),
+ new SpanTestData().setLabel("seda:a client").setUri("seda://a").setOperation("a")
+ .setKind(Tags.SPAN_KIND_CLIENT).setParentId(6),
+ new SpanTestData().setLabel("direct:start server").setUri("direct://start").setOperation("start")
+ .setKind(Tags.SPAN_KIND_SERVER).setParentId(7),
+ new SpanTestData().setLabel("direct:start client").setUri("direct://start").setOperation("start")
+ .setKind(Tags.SPAN_KIND_CLIENT)
+ };
+
+ public MulticastParallelRouteTest() {
+ super(testdata);
+ }
+
+ @Test
+ public void testRoute() throws Exception {
+ template.requestBody("direct:start", "Hello");
+
+ verify(true);
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("direct:start").to("seda:a").routeId("start");
+
+ from("seda:a").routeId("a")
+ .log("routing at ${routeId}")
+ .multicast().parallelProcessing()
+ .to("seda:b", "seda:c")
+ .end()
+ .log("End of routing");
+
+ from("seda:b").routeId("b")
+ .log("routing at ${routeId}")
+ .delay(simple("${random(1000,2000)}"));
+
+ from("seda:c").routeId("c")
+ .log("routing at ${routeId}")
+ .delay(simple("${random(0,100)}"));
+ }
+ };
+ }
+}