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 2019/07/29 05:36:28 UTC
[camel] 04/10: CAMEL-13505: A new tracer - work in progress.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch CAMEL-13505
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 3db436a8113d787b6d6cf093ef7a98914330ba08
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Sun Jul 28 12:58:08 2019 +0200
CAMEL-13505: A new tracer - work in progress.
---
.../camel/processor/CamelInternalProcessor.java | 54 +++++++-
.../camel/processor/channel/DefaultChannel.java | 2 +-
.../camel/processor/interceptor/DefaultTracer.java | 144 ++++++++++-----------
.../org/apache/camel/model/RouteDefinition.java | 5 +
.../apache/camel/model/ToDynamicDefinition.java | 5 +
.../camel/component/seda/TracingWithDelayTest.java | 6 +-
6 files changed, 138 insertions(+), 78 deletions(-)
diff --git a/core/camel-base/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java b/core/camel-base/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java
index 95c58e8..444db23 100644
--- a/core/camel-base/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java
+++ b/core/camel-base/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java
@@ -20,6 +20,7 @@ import java.util.ArrayList;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
+import java.util.Objects;
import java.util.concurrent.RejectedExecutionException;
import org.apache.camel.AsyncCallback;
@@ -39,20 +40,20 @@ import org.apache.camel.processor.interceptor.DefaultBacklogTracerEventMessage;
import org.apache.camel.processor.interceptor.DefaultTracer;
import org.apache.camel.spi.CamelInternalProcessorAdvice;
import org.apache.camel.spi.Debugger;
-import org.apache.camel.spi.ExchangeFormatter;
import org.apache.camel.spi.InflightRepository;
import org.apache.camel.spi.ManagementInterceptStrategy.InstrumentationProcessor;
import org.apache.camel.spi.MessageHistoryFactory;
import org.apache.camel.spi.RouteContext;
import org.apache.camel.spi.RoutePolicy;
import org.apache.camel.spi.StreamCachingStrategy;
+import org.apache.camel.spi.Synchronization;
import org.apache.camel.spi.Transformer;
import org.apache.camel.spi.UnitOfWork;
import org.apache.camel.support.CamelContextHelper;
import org.apache.camel.support.MessageHelper;
import org.apache.camel.support.OrderedComparator;
+import org.apache.camel.support.SynchronizationAdapter;
import org.apache.camel.support.UnitOfWorkHelper;
-import org.apache.camel.support.processor.DefaultExchangeFormatter;
import org.apache.camel.support.processor.DelegateAsyncProcessor;
import org.apache.camel.util.StopWatch;
import org.slf4j.Logger;
@@ -777,14 +778,26 @@ public class CamelInternalProcessor extends DelegateAsyncProcessor {
private final DefaultTracer tracer;
private final NamedNode processorDefinition;
+ private final NamedNode routeDefinition;
+ private final boolean first;
+ private final Synchronization tracingAfterRoute;
- public TracingAdvice(DefaultTracer tracer, NamedNode processorDefinition) {
+ public TracingAdvice(DefaultTracer tracer, NamedNode processorDefinition, NamedNode routeDefinition, boolean first) {
this.tracer = tracer;
this.processorDefinition = processorDefinition;
+ this.routeDefinition = routeDefinition;
+ this.first = first;
+ this.tracingAfterRoute = new TracingAfterRoute(tracer, routeDefinition.getId());
}
@Override
public Object before(Exchange exchange) throws Exception {
+ // first time add before route and after route tracing
+ if (first) {
+ tracer.traceBeforeRoute(routeDefinition, exchange);
+ exchange.addOnCompletion(tracingAfterRoute);
+ }
+
tracer.trace(processorDefinition, exchange);
return null;
}
@@ -793,6 +806,41 @@ public class CamelInternalProcessor extends DelegateAsyncProcessor {
public void after(Exchange exchange, Object data) throws Exception {
// noop
}
+
+ private static final class TracingAfterRoute extends SynchronizationAdapter {
+
+ private final DefaultTracer tracer;
+ private final String routeId;
+
+ private TracingAfterRoute(DefaultTracer tracer, String routeId) {
+ this.tracer = tracer;
+ this.routeId = routeId;
+ }
+
+ @Override
+ public void onAfterRoute(Route route, Exchange exchange) {
+ if (routeId.equals(route.getId())) {
+ tracer.traceAfterRoute(route, exchange);
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ TracingAfterRoute that = (TracingAfterRoute) o;
+ return routeId.equals(that.routeId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(routeId);
+ }
+ }
}
/**
diff --git a/core/camel-base/src/main/java/org/apache/camel/processor/channel/DefaultChannel.java b/core/camel-base/src/main/java/org/apache/camel/processor/channel/DefaultChannel.java
index 5a12ac2..5bca38f 100644
--- a/core/camel-base/src/main/java/org/apache/camel/processor/channel/DefaultChannel.java
+++ b/core/camel-base/src/main/java/org/apache/camel/processor/channel/DefaultChannel.java
@@ -216,7 +216,7 @@ public class DefaultChannel extends CamelInternalProcessor implements Channel {
// add logger tracer
DefaultTracer tracer = getOrCreateDefaultTracer();
- addAdvice(new TracingAdvice(tracer, targetOutputDef));
+ addAdvice(new TracingAdvice(tracer, targetOutputDef, route, first));
}
// sort interceptors according to ordered
diff --git a/core/camel-base/src/main/java/org/apache/camel/processor/interceptor/DefaultTracer.java b/core/camel-base/src/main/java/org/apache/camel/processor/interceptor/DefaultTracer.java
index 177b7da..f4b5622 100644
--- a/core/camel-base/src/main/java/org/apache/camel/processor/interceptor/DefaultTracer.java
+++ b/core/camel-base/src/main/java/org/apache/camel/processor/interceptor/DefaultTracer.java
@@ -16,22 +16,19 @@
*/
package org.apache.camel.processor.interceptor;
-import java.util.List;
import java.util.Objects;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
-import org.apache.camel.MessageHistory;
import org.apache.camel.NamedNode;
+import org.apache.camel.Route;
import org.apache.camel.StaticService;
import org.apache.camel.spi.ExchangeFormatter;
import org.apache.camel.support.CamelContextHelper;
-import org.apache.camel.support.MessageHelper;
import org.apache.camel.support.PatternHelper;
import org.apache.camel.support.builder.ExpressionBuilder;
import org.apache.camel.support.processor.DefaultExchangeFormatter;
import org.apache.camel.support.service.ServiceSupport;
-import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.StringHelper;
import org.apache.camel.util.URISupport;
import org.slf4j.Logger;
@@ -39,12 +36,10 @@ import org.slf4j.LoggerFactory;
public class DefaultTracer extends ServiceSupport implements StaticService {
- // TODO: Output the result when the exchange is done <<<<
// TODO: Register this in JMX so it can be managed and turned on|off there
// TODO: Have operations to turn on|off routes on route mbean etc
// TODO: SPI to implement custom tracer
// TODO: Custom exchange formatter
- // TODO: Custom trace formatter
private static final String TRACING_OUTPUT = "%-6.6s [%-18.18s] [%-18.18s] [%-38.38s]";
@@ -52,12 +47,12 @@ public class DefaultTracer extends ServiceSupport implements StaticService {
private static final Logger LOG = LoggerFactory.getLogger("org.apache.camel.Tracing");
private final CamelContext camelContext;
private boolean enabled = true;
- private boolean history = false;
private ExchangeFormatter exchangeFormatter;
// a pattern to filter tracing nodes
private String tracePattern;
private transient String[] patterns;
+ private boolean traceBeforeAfterRoute = true;
public DefaultTracer(CamelContext camelContext) {
this.camelContext = camelContext;
@@ -85,63 +80,69 @@ public class DefaultTracer extends ServiceSupport implements StaticService {
@SuppressWarnings("unchecked")
public void trace(NamedNode node, Exchange exchange) {
if (shouldTrace(node)) {
- if (history) {
- String out = MessageHelper.dumpTracing(exchange, exchangeFormatter);
- LOG.info(out);
- } else {
- String routeId = ExpressionBuilder.routeIdExpression().evaluate(exchange, String.class);
- String id = node.getId();
-
- // are we going to a new route, then we would like to log an entry with the route input too
- boolean newRoute;
- String prevLabel = "";
- List<MessageHistory> list = exchange.getProperty(Exchange.MESSAGE_HISTORY, List.class);
- if (list == null || list.size() <= 1) {
- newRoute = true;
- if (exchange.getFromEndpoint() != null) {
- // we need to avoid leak the sensible information here
- // the sanitizeUri takes a very long time for very long string and the format cuts this to
- // 38 characters, anyway. Cut this to 60 characters. This will give enough space for removing
- // characters in the sanitizeUri method and will be reasonably fast
- prevLabel = "from[" + URISupport.sanitizeUri(StringHelper.limitLength(exchange.getFromEndpoint().getEndpointUri(), 60) + "]");
- }
- } else {
- // whats the last route id from message history
- MessageHistory prev = list.get(list.size() - 2);
- newRoute = !ObjectHelper.equal(routeId, prev.getRouteId());
- if (newRoute) {
- // we need to avoid leak the sensible information here
- // the sanitizeUri takes a very long time for very long string and the format cuts this to
- // 38 characters, anyway. Cut this to 60 characters. This will give enough space for removing
- // characters in the sanitizeUri method and will be reasonably fast
- prevLabel = "from[" + URISupport.sanitizeUri(StringHelper.limitLength(prev.getNode().getLabel(), 60) + "]");
- }
- }
- if (newRoute) {
- StringBuilder sb = new StringBuilder();
- sb.append(String.format(TRACING_OUTPUT, " ", routeId, routeId, prevLabel));
- sb.append(" ");
- String data = exchangeFormatter.format(exchange);
- sb.append(data);
- String out = sb.toString();
- LOG.info(out);
- }
+ String routeId = ExpressionBuilder.routeIdExpression().evaluate(exchange, String.class);
+ String id = node.getId();
+
+ // we need to avoid leak the sensible information here
+ // the sanitizeUri takes a very long time for very long string and the format cuts this to
+ // 38 characters, anyway. Cut this to 60 characters. This will give enough space for removing
+ // characters in the sanitizeUri method and will be reasonably fast
+ String label = URISupport.sanitizeUri(StringHelper.limitLength(node.getLabel(), 60));
+
+ StringBuilder sb = new StringBuilder();
+ sb.append(String.format(TRACING_OUTPUT, " ", routeId, id, label));
+ sb.append(" ");
+ String data = exchangeFormatter.format(exchange);
+ sb.append(data);
+ String out = sb.toString();
+ dumpTrace(out);
+ }
+ }
- // we need to avoid leak the sensible information here
- // the sanitizeUri takes a very long time for very long string and the format cuts this to
- // 38 characters, anyway. Cut this to 60 characters. This will give enough space for removing
- // characters in the sanitizeUri method and will be reasonably fast
- String label = URISupport.sanitizeUri(StringHelper.limitLength(node.getLabel(), 60));
-
- StringBuilder sb = new StringBuilder();
- sb.append(String.format(TRACING_OUTPUT, " >>>> ", routeId, id, label));
- sb.append(" ");
- String data = exchangeFormatter.format(exchange);
- sb.append(data);
- String out = sb.toString();
- LOG.info(out);
- }
+ public void traceBeforeRoute(NamedNode route, Exchange exchange) {
+ if (!traceBeforeAfterRoute) {
+ return;
}
+
+ // we need to avoid leak the sensible information here
+ // the sanitizeUri takes a very long time for very long string and the format cuts this to
+ // 38 characters, anyway. Cut this to 60 characters. This will give enough space for removing
+ // characters in the sanitizeUri method and will be reasonably fast
+ String uri = route.getLabel();
+ String label = "from[" + URISupport.sanitizeUri(StringHelper.limitLength(uri, 60) + "]");
+
+ StringBuilder sb = new StringBuilder();
+ sb.append(String.format(TRACING_OUTPUT, " ---> ", route.getId(), route.getId(), label));
+ sb.append(" ");
+ String data = exchangeFormatter.format(exchange);
+ sb.append(data);
+ String out = sb.toString();
+ LOG.info(out);
+ }
+
+ public void traceAfterRoute(Route route, Exchange exchange) {
+ if (!traceBeforeAfterRoute) {
+ return;
+ }
+
+ // we need to avoid leak the sensible information here
+ // the sanitizeUri takes a very long time for very long string and the format cuts this to
+ // 38 characters, anyway. Cut this to 60 characters. This will give enough space for removing
+ // characters in the sanitizeUri method and will be reasonably fast
+ String uri = route.getConsumer().getEndpoint().getEndpointUri();
+ String label = "from[" + URISupport.sanitizeUri(StringHelper.limitLength(uri, 60) + "]");
+
+ StringBuilder sb = new StringBuilder();
+ sb.append(String.format(TRACING_OUTPUT, " <--- ", route.getId(), route.getId(), label));
+ sb.append(" ");
+ String data = exchangeFormatter.format(exchange);
+ sb.append(data);
+ String out = sb.toString();
+ dumpTrace(out);
+ }
+
+ public void dumpTrace(String out) {
+ LOG.info(out);
}
/**
@@ -194,14 +195,6 @@ public class DefaultTracer extends ServiceSupport implements StaticService {
this.enabled = enabled;
}
- public boolean isHistory() {
- return history;
- }
-
- public void setHistory(boolean history) {
- this.history = history;
- }
-
public String getTracePattern() {
return tracePattern;
}
@@ -216,13 +209,20 @@ public class DefaultTracer extends ServiceSupport implements StaticService {
}
}
+ public boolean isTraceBeforeAfterRoute() {
+ return traceBeforeAfterRoute;
+ }
+
+ public void setTraceBeforeAfterRoute(boolean traceBeforeAfterRoute) {
+ this.traceBeforeAfterRoute = traceBeforeAfterRoute;
+ }
+
@Override
protected void doStart() throws Exception {
if (exchangeFormatter == null) {
DefaultExchangeFormatter formatter = new DefaultExchangeFormatter();
formatter.setShowExchangeId(true);
- // use multi line if using history
- formatter.setMultiline(isHistory());
+ formatter.setMultiline(false);
formatter.setShowHeaders(true);
formatter.setStyle(DefaultExchangeFormatter.OutputStyle.Default);
exchangeFormatter = formatter;
diff --git a/core/camel-core/src/main/java/org/apache/camel/model/RouteDefinition.java b/core/camel-core/src/main/java/org/apache/camel/model/RouteDefinition.java
index 9c4927d..3fe6a50 100644
--- a/core/camel-core/src/main/java/org/apache/camel/model/RouteDefinition.java
+++ b/core/camel-core/src/main/java/org/apache/camel/model/RouteDefinition.java
@@ -143,6 +143,11 @@ public class RouteDefinition extends ProcessorDefinition<RouteDefinition> implem
return "route";
}
+ @Override
+ public String getLabel() {
+ return input.getLabel();
+ }
+
// Fluent API
// -----------------------------------------------------------------------
diff --git a/core/camel-core/src/main/java/org/apache/camel/model/ToDynamicDefinition.java b/core/camel-core/src/main/java/org/apache/camel/model/ToDynamicDefinition.java
index ad9f46e..509c35a 100644
--- a/core/camel-core/src/main/java/org/apache/camel/model/ToDynamicDefinition.java
+++ b/core/camel-core/src/main/java/org/apache/camel/model/ToDynamicDefinition.java
@@ -70,6 +70,11 @@ public class ToDynamicDefinition extends NoOutputDefinition<ToDynamicDefinition>
return "DynamicTo[" + getLabel() + "]";
}
+ @Override
+ public String getLabel() {
+ return uri;
+ }
+
// Fluent API
// -------------------------------------------------------------------------
diff --git a/core/camel-core/src/test/java/org/apache/camel/component/seda/TracingWithDelayTest.java b/core/camel-core/src/test/java/org/apache/camel/component/seda/TracingWithDelayTest.java
index 83d28f3..33a8bc6 100644
--- a/core/camel-core/src/test/java/org/apache/camel/component/seda/TracingWithDelayTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/component/seda/TracingWithDelayTest.java
@@ -50,9 +50,11 @@ public class TracingWithDelayTest extends ContextTestSupport {
LOG.info("This is the processor being invoked between mock:a and mock:b");
}
})
- .to("mock:b").to("direct:c");
+ .to("mock:b").toD("direct:c")
+ .to("mock:result")
+ .transform(simple("${body}${body}"));
- from("direct:c").transform(constant("Bye World")).to("mock:result");
+ from("direct:c").transform(constant("Bye World")).to("mock:c");
}
};
}