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 2021/04/15 13:49:38 UTC
[camel] branch master updated: CAMEL-16509 - Feature/fix zipkin
spans alternate (#5373)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/master by this push:
new 98628d9 CAMEL-16509 - Feature/fix zipkin spans alternate (#5373)
98628d9 is described below
commit 98628d9724cb29ea25583c756e61052ceff45cf6
Author: Samrat Dhillon <sa...@gmail.com>
AuthorDate: Thu Apr 15 09:49:12 2021 -0400
CAMEL-16509 - Feature/fix zipkin spans alternate (#5373)
* fixing incorrect span information generated by camel-zipkin in parallel processing of multicast or recipientlist. The original ZipkinState object ends up getting shared by different exchange object copies resulting in issues during push/pop operations. Added interface to allow safe copy of properties which can provide deep object copies for safe copy operations
* using thread safe datastructure for keeping span information
* fixing unused imports
* Try to perform safe copy of properties only if explicitly specified on the exchange
* cleaner way to implement copy safe properties
* refactoring and updating javadoc
Co-authored-by: Samrat Dhillon <sa...@innovapost.com>
---
.../java/org/apache/camel/zipkin/ZipkinState.java | 53 ++++++++-----------
.../java/org/apache/camel/zipkin/ZipkinTracer.java | 19 ++++---
.../org/apache/camel/zipkin/ZipkinStateTest.java | 30 ++++++++++-
.../java/org/apache/camel/ExtendedExchange.java | 18 +++++++
.../java/org/apache/camel/SafeCopyProperty.java | 40 +++++++++++++++
.../org/apache/camel/impl/DefaultExchangeTest.java | 38 ++++++++++++++
.../org/apache/camel/support/AbstractExchange.java | 60 +++++++++++++++++++++-
7 files changed, 216 insertions(+), 42 deletions(-)
diff --git a/components/camel-zipkin/src/main/java/org/apache/camel/zipkin/ZipkinState.java b/components/camel-zipkin/src/main/java/org/apache/camel/zipkin/ZipkinState.java
index 6cfa2f4..848ef02 100644
--- a/components/camel-zipkin/src/main/java/org/apache/camel/zipkin/ZipkinState.java
+++ b/components/camel-zipkin/src/main/java/org/apache/camel/zipkin/ZipkinState.java
@@ -16,13 +16,12 @@
*/
package org.apache.camel.zipkin;
-import java.util.ArrayDeque;
import java.util.Deque;
-import java.util.Iterator;
+import java.util.concurrent.ConcurrentLinkedDeque;
import brave.Span;
-import brave.propagation.TraceContextOrSamplingFlags;
import org.apache.camel.Exchange;
+import org.apache.camel.SafeCopyProperty;
/**
* The state of the zipkin trace which we store on the {@link Exchange}
@@ -30,18 +29,27 @@ import org.apache.camel.Exchange;
* This is needed to keep track of of correlating when an existing span is calling downstream service(s) and therefore
* must be able to correlate those service calls with the parent span.
*/
-public final class ZipkinState {
+public final class ZipkinState implements SafeCopyProperty {
public static final String KEY = "CamelZipkinState";
- private final Deque<Span> clientSpans = new ArrayDeque<>();
- private final Deque<Span> serverSpans = new ArrayDeque<>();
+ private final Deque<Span> clientSpans = new ConcurrentLinkedDeque<>();
+ private final Deque<Span> serverSpans = new ConcurrentLinkedDeque<>();
- public synchronized void pushClientSpan(Span span) {
+ public ZipkinState() {
+
+ }
+
+ private ZipkinState(ZipkinState state) {
+ this.clientSpans.addAll(state.clientSpans);
+ this.serverSpans.addAll(state.serverSpans);
+ }
+
+ public void pushClientSpan(Span span) {
clientSpans.push(span);
}
- public synchronized Span popClientSpan() {
+ public Span popClientSpan() {
if (!clientSpans.isEmpty()) {
return clientSpans.pop();
} else {
@@ -49,11 +57,11 @@ public final class ZipkinState {
}
}
- public synchronized void pushServerSpan(Span span) {
+ public void pushServerSpan(Span span) {
serverSpans.push(span);
}
- public synchronized Span popServerSpan() {
+ public Span popServerSpan() {
if (!serverSpans.isEmpty()) {
return serverSpans.pop();
} else {
@@ -61,7 +69,7 @@ public final class ZipkinState {
}
}
- private Span peekServerSpan() {
+ public Span peekServerSpan() {
if (!serverSpans.isEmpty()) {
return serverSpans.peek();
} else {
@@ -69,26 +77,9 @@ public final class ZipkinState {
}
}
- public synchronized Span findMatchingServerSpan(Exchange exchange) {
- String spanId = (String) exchange.getIn().getHeader(ZipkinConstants.SPAN_ID);
- Span lastSpan = peekServerSpan();
- if (spanId == null) {
- return lastSpan;
- }
- TraceContextOrSamplingFlags traceContext
- = ZipkinTracer.EXTRACTOR.extract(new CamelRequest(exchange.getIn(), Span.Kind.SERVER));
- if (traceContext.context().spanId() == lastSpan.context().spanId()) {
- return lastSpan;
- }
-
- Iterator<Span> spanItr = serverSpans.iterator();
- while (spanItr.hasNext()) {
- Span span = spanItr.next();
- if (span.context().spanId() == traceContext.context().spanId()) {
- return span;
- }
- }
- return lastSpan;
+ @Override
+ public ZipkinState safeCopy() {
+ return new ZipkinState(this);
}
}
diff --git a/components/camel-zipkin/src/main/java/org/apache/camel/zipkin/ZipkinTracer.java b/components/camel-zipkin/src/main/java/org/apache/camel/zipkin/ZipkinTracer.java
index 7883981..6fb0b2e 100644
--- a/components/camel-zipkin/src/main/java/org/apache/camel/zipkin/ZipkinTracer.java
+++ b/components/camel-zipkin/src/main/java/org/apache/camel/zipkin/ZipkinTracer.java
@@ -40,6 +40,7 @@ import org.apache.camel.CamelContextAware;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Expression;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.NamedNode;
import org.apache.camel.Route;
import org.apache.camel.RuntimeCamelException;
@@ -586,14 +587,15 @@ public class ZipkinTracer extends ServiceSupport implements RoutePolicyFactory,
private void clientRequest(Tracing brave, String serviceName, ExchangeSendingEvent event) {
// reuse existing span if we do multiple requests from the same
- ZipkinState state = event.getExchange().getProperty(ZipkinState.KEY, ZipkinState.class);
+ ExtendedExchange exchange = event.getExchange().adapt(ExtendedExchange.class);
+ ZipkinState state = exchange.getSafeCopyProperty(ZipkinState.KEY, ZipkinState.class);
if (state == null) {
state = new ZipkinState();
- event.getExchange().setProperty(ZipkinState.KEY, state);
+ exchange.setSafeCopyProperty(ZipkinState.KEY, state);
}
// if we started from a server span then lets reuse that when we call a
// downstream service
- Span last = state.findMatchingServerSpan(event.getExchange());
+ Span last = state.peekServerSpan();
Span span;
if (last != null) {
span = brave.tracer().newChild(last.context());
@@ -641,7 +643,8 @@ public class ZipkinTracer extends ServiceSupport implements RoutePolicyFactory,
private void clientResponse(Tracing brave, String serviceName, ExchangeSentEvent event) {
Span span = null;
- ZipkinState state = event.getExchange().getProperty(ZipkinState.KEY, ZipkinState.class);
+ ExtendedExchange exchange = event.getExchange().adapt(ExtendedExchange.class);
+ ZipkinState state = exchange.getSafeCopyProperty(ZipkinState.KEY, ZipkinState.class);
if (state != null) {
// only process if it was a zipkin client event
span = state.popClientSpan();
@@ -674,10 +677,11 @@ public class ZipkinTracer extends ServiceSupport implements RoutePolicyFactory,
private Span serverRequest(Tracing brave, String serviceName, Exchange exchange) {
// reuse existing span if we do multiple requests from the same
- ZipkinState state = exchange.getProperty(ZipkinState.KEY, ZipkinState.class);
+ ExtendedExchange extendedExchange = exchange.adapt(ExtendedExchange.class);
+ ZipkinState state = extendedExchange.getSafeCopyProperty(ZipkinState.KEY, ZipkinState.class);
if (state == null) {
state = new ZipkinState();
- exchange.setProperty(ZipkinState.KEY, state);
+ extendedExchange.setSafeCopyProperty(ZipkinState.KEY, state);
}
Span span = null;
Span.Kind spanKind = getConsumerComponentSpanKind(exchange.getFromEndpoint());
@@ -722,7 +726,8 @@ public class ZipkinTracer extends ServiceSupport implements RoutePolicyFactory,
private void serverResponse(Tracing brave, String serviceName, Exchange exchange) {
Span span = null;
- ZipkinState state = exchange.getProperty(ZipkinState.KEY, ZipkinState.class);
+ ExtendedExchange extendedExchange = exchange.adapt(ExtendedExchange.class);
+ ZipkinState state = extendedExchange.getSafeCopyProperty(ZipkinState.KEY, ZipkinState.class);
if (state != null) {
// only process if it was a zipkin server event
span = state.popServerSpan();
diff --git a/components/camel-zipkin/src/test/java/org/apache/camel/zipkin/ZipkinStateTest.java b/components/camel-zipkin/src/test/java/org/apache/camel/zipkin/ZipkinStateTest.java
index 00917b7..5ebfd69 100644
--- a/components/camel-zipkin/src/test/java/org/apache/camel/zipkin/ZipkinStateTest.java
+++ b/components/camel-zipkin/src/test/java/org/apache/camel/zipkin/ZipkinStateTest.java
@@ -53,16 +53,42 @@ public class ZipkinStateTest {
exchange2.getIn().setHeader(ZipkinConstants.PARENT_SPAN_ID, context.spanIdString());
exchange2.getIn().setHeader(ZipkinConstants.SPAN_ID, span2.context().spanIdString());
- Span retrived = state.findMatchingServerSpan(exchange2);
+ Span retrived = state.peekServerSpan();
assertThat(retrived.context().spanId()).isEqualTo(span2.context().spanId());
assertThat(retrived.context().parentId()).isEqualTo(span2.context().parentId());
assertThat(retrived.context().traceId()).isEqualTo(span2.context().traceId());
- retrived = state.findMatchingServerSpan(exchange1);
+ state.popServerSpan();
+
+ retrived = state.peekServerSpan();
assertThat(retrived.context().spanId()).isEqualTo(span1.context().spanId());
assertThat(retrived.context().parentId()).isEqualTo(span1.context().parentId());
assertThat(retrived.context().traceId()).isEqualTo(span1.context().traceId());
}
+ @Test
+ public void testZipkinStateSafeCopy() {
+ TraceContext context = TraceContext.newBuilder().traceId(1L).spanId(2L).parentId(3L).build();
+ TraceContextOrSamplingFlags sampling = TraceContextOrSamplingFlags.newBuilder(context).build();
+ Tracing tracing = Tracing.newBuilder().build();
+
+ Span span1 = tracing.tracer().nextSpan(sampling);
+ state.pushServerSpan(span1);
+
+ Span span2 = tracing.tracer().nextSpan(sampling);
+ state.pushServerSpan(span2);
+
+ Span span3 = tracing.tracer().nextSpan(sampling);
+ ZipkinState state2 = state.safeCopy();
+ state2.pushServerSpan(span3);
+
+ //original object intact
+ assertThat(state.peekServerSpan().context()).isNotEqualTo(span3.context()).usingRecursiveComparison();
+
+ //new object has the new span
+ assertThat(state2.peekServerSpan().context()).isEqualTo(span3.context()).usingRecursiveComparison();
+
+ }
+
}
diff --git a/core/camel-api/src/main/java/org/apache/camel/ExtendedExchange.java b/core/camel-api/src/main/java/org/apache/camel/ExtendedExchange.java
index 9ecd8e4..1cadcfb 100644
--- a/core/camel-api/src/main/java/org/apache/camel/ExtendedExchange.java
+++ b/core/camel-api/src/main/java/org/apache/camel/ExtendedExchange.java
@@ -204,4 +204,22 @@ public interface ExtendedExchange extends Exchange {
*/
void setDefaultConsumerCallback(AsyncCallback callback);
+ /**
+ * Method to set SafeCopyProperties on the exchange. {@link SafeCopyProperty}
+ *
+ * @param key
+ *
+ * @param SafeCopyProperty
+ */
+ void setSafeCopyProperty(String key, SafeCopyProperty value);
+
+ /**
+ * Method to get SafeCopyProperties from the exchange
+ *
+ * @param key
+ * @param type the type of the property
+ * @return
+ */
+
+ <T> T getSafeCopyProperty(String key, Class<T> type);
}
diff --git a/core/camel-api/src/main/java/org/apache/camel/SafeCopyProperty.java b/core/camel-api/src/main/java/org/apache/camel/SafeCopyProperty.java
new file mode 100644
index 0000000..c87b2b6
--- /dev/null
+++ b/core/camel-api/src/main/java/org/apache/camel/SafeCopyProperty.java
@@ -0,0 +1,40 @@
+/*
+ * 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;
+
+/**
+ *
+ * An interface that allows safe copy of property value object when creating copy of Exchange objects. Classes
+ * implementing this interface can be set as key value pair on exchange object via
+ * {@link ExtendedExchange#setSafeCopyProperty(String, SafeCopyProperty)}.
+ *
+ * When exchange object is copied it will invoke {@link SafeCopyProperty#safeCopy()} method on properties set using
+ * {@link ExtendedExchange#setSafeCopyProperty(String, SafeCopyProperty)}. This allows the property value object to
+ * return a copy object to be set on the target exchange object instead of the original value object. This protects the
+ * properties from unintended mutation when using parallelProcessing in Multicast or RecipientList EIP
+ */
+public interface SafeCopyProperty {
+
+ /**
+ * Implementations should implement this method to return a deep copy of the object which can be mutated
+ * independently of the original object.
+ *
+ * @return copy object usually of the same class type which implements this interface
+ */
+ SafeCopyProperty safeCopy();
+
+}
diff --git a/core/camel-core/src/test/java/org/apache/camel/impl/DefaultExchangeTest.java b/core/camel-core/src/test/java/org/apache/camel/impl/DefaultExchangeTest.java
index 8e77d35..7e91c42 100644
--- a/core/camel-core/src/test/java/org/apache/camel/impl/DefaultExchangeTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/impl/DefaultExchangeTest.java
@@ -27,11 +27,14 @@ import org.apache.camel.ExtendedExchange;
import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.SafeCopyProperty;
import org.apache.camel.TypeConversionException;
import org.apache.camel.support.DefaultExchange;
import org.apache.camel.support.DefaultMessage;
+import org.apache.camel.support.ExchangeHelper;
import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -42,6 +45,9 @@ import static org.junit.jupiter.api.Assertions.fail;
public class DefaultExchangeTest extends ExchangeTestSupport {
+ private static final String SAFE_PROPERTY = "SAFE_PROPERTY";
+ private static final String UNSAFE_PROPERTY = "UNSAFE_PROPERTY";
+
@Test
public void testBody() throws Exception {
assertNotNull(exchange.getIn().getBody());
@@ -294,6 +300,38 @@ public class DefaultExchangeTest extends ExchangeTestSupport {
assertEquals(sourceIn.getClass(), destIn.getClass(), "Dest message should be of the same type as source message");
}
+ @Test
+ public void testExchangeSafeCopy() {
+ DefaultExchange exchange = new DefaultExchange(context);
+ SafeProperty property = new SafeProperty();
+ UnsafeProperty unsafeProperty = new UnsafeProperty();
+ exchange.setSafeCopyProperty(SAFE_PROPERTY, property);
+ exchange.setProperty(UNSAFE_PROPERTY, unsafeProperty);
+
+ Exchange copy = ExchangeHelper.createCorrelatedCopy(exchange, false);
+
+ assertThat(copy.getProperty(SAFE_PROPERTY)).isNotSameAs(property);
+ assertThat(copy.getProperty(UNSAFE_PROPERTY)).isSameAs(unsafeProperty);
+
+ }
+
+ private static final class SafeProperty implements SafeCopyProperty {
+
+ private SafeProperty() {
+
+ }
+
+ @Override
+ public SafeProperty safeCopy() {
+ return new SafeProperty();
+ }
+
+ }
+
+ private static class UnsafeProperty {
+
+ }
+
public static class MyMessage extends DefaultMessage {
public MyMessage(CamelContext camelContext) {
super(camelContext);
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/AbstractExchange.java b/core/camel-support/src/main/java/org/apache/camel/support/AbstractExchange.java
index 42d578d..e3d2975 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/AbstractExchange.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/AbstractExchange.java
@@ -36,6 +36,7 @@ import org.apache.camel.ExtendedCamelContext;
import org.apache.camel.ExtendedExchange;
import org.apache.camel.Message;
import org.apache.camel.MessageHistory;
+import org.apache.camel.SafeCopyProperty;
import org.apache.camel.spi.HeadersMapFactory;
import org.apache.camel.spi.Synchronization;
import org.apache.camel.spi.UnitOfWork;
@@ -84,6 +85,7 @@ class AbstractExchange implements ExtendedExchange {
boolean redeliveryExhausted;
Boolean errorHandlerHandled;
AsyncCallback defaultConsumerCallback; // optimize (do not reset)
+ Map<String, SafeCopyProperty> safeCopyProperties;
public AbstractExchange(CamelContext context) {
this.context = context;
@@ -152,7 +154,11 @@ class AbstractExchange implements ExtendedExchange {
// copy properties after body as body may trigger lazy init
if (hasProperties()) {
- safeCopyProperties(getProperties(), exchange.getProperties());
+ copyProperties(getProperties(), exchange.getProperties());
+ }
+
+ if (hasSafeCopyProperties()) {
+ safeCopyProperties(this.safeCopyProperties, exchange.getSafeCopyProperties());
}
// copy over internal properties
System.arraycopy(internalProperties, 0, exchange.internalProperties, 0, internalProperties.length);
@@ -188,10 +194,17 @@ class AbstractExchange implements ExtendedExchange {
}
@SuppressWarnings("unchecked")
- private void safeCopyProperties(Map<String, Object> source, Map<String, Object> target) {
+ private void copyProperties(Map<String, Object> source, Map<String, Object> target) {
target.putAll(source);
}
+ private void safeCopyProperties(
+ Map<String, SafeCopyProperty> source, Map<String, SafeCopyProperty> target) {
+ source.entrySet().stream().forEach(entry -> {
+ target.put(entry.getKey(), entry.getValue().safeCopy());
+ });
+ }
+
@Override
public CamelContext getContext() {
return context;
@@ -250,6 +263,7 @@ class AbstractExchange implements ExtendedExchange {
internalProperties[key.ordinal()] = value;
}
+ @Override
public Object removeProperty(ExchangePropertyKey key) {
Object old = internalProperties[key.ordinal()];
internalProperties[key.ordinal()] = null;
@@ -428,6 +442,13 @@ class AbstractExchange implements ExtendedExchange {
return properties;
}
+ Map<String, SafeCopyProperty> getSafeCopyProperties() {
+ if (safeCopyProperties == null) {
+ this.safeCopyProperties = new ConcurrentHashMap<>(2);
+ }
+ return safeCopyProperties;
+ }
+
@Override
public Map<String, Object> getAllProperties() {
// include also internal properties (creates a new map)
@@ -443,6 +464,10 @@ class AbstractExchange implements ExtendedExchange {
return properties != null && !properties.isEmpty();
}
+ private boolean hasSafeCopyProperties() {
+ return safeCopyProperties != null && !safeCopyProperties.isEmpty();
+ }
+
@Override
public Message getIn() {
if (in == null) {
@@ -809,6 +834,7 @@ class AbstractExchange implements ExtendedExchange {
this.redeliveryExhausted = redeliveryExhausted;
}
+ @Override
public Boolean getErrorHandlerHandled() {
return errorHandlerHandled;
}
@@ -886,4 +912,34 @@ class AbstractExchange implements ExtendedExchange {
}
}
+ @Override
+ public void setSafeCopyProperty(String key, SafeCopyProperty value) {
+ if (value != null) {
+ // avoid the NullPointException
+ if (safeCopyProperties == null) {
+ this.safeCopyProperties = new ConcurrentHashMap<>(2);
+ }
+ safeCopyProperties.put(key, value);
+ } else if (safeCopyProperties != null) {
+ // if the value is null, we just remove the key from the map
+ safeCopyProperties.remove(key);
+ }
+
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <T> T getSafeCopyProperty(String key, Class<T> type) {
+ if (!hasSafeCopyProperties()) {
+ return null;
+ }
+ Object value = getSafeCopyProperties().get(key);
+
+ if (type.isInstance(value)) {
+ return (T) value;
+ }
+
+ return ExchangeHelper.convertToType(this, type, value);
+ }
+
}