Merge pull request #17 from newrelic-experimental/camel-enhancements

Camel enhancements

.gradle/buildOutputCleanup/cache.properties

@@ -0,0 +1,2 @@
+#Thu Aug 31 07:21:04 CDT 2023
+gradle.version=7.3

camel-core-3.18/src/main/java/com/nr/instrumentation/apache/camel/NRAsyncProcessorWrapper.java

@@ -38,6 +38,10 @@ public boolean process(Exchange exchange, AsyncCallback callback) {
 				names = new String[] {"Custom","AsyncProcessor",delegate.getClass().getSimpleName(),"process"};
 			}
 		} else {
+			Object timerTaskValue = exchange.getProperty(Exchange.TIMER_NAME);
+			if(timerTaskValue != null) {
+				names = new String[] {"Custom","AsyncProcessor",delegate.getClass().getSimpleName(),"process","TimerTask",timerTaskValue.toString()};
+			}
 			names = new String[] {"Custom","AsyncProcessor",delegate.getClass().getSimpleName(),"process"};
 
 		}

camel-core-3.18/src/main/java/com/nr/instrumentation/apache/camel/Pipeline_instrumentation.java

@@ -8,6 +8,7 @@
 
 import com.newrelic.api.agent.NewRelic;
 import com.newrelic.api.agent.Trace;
+import com.newrelic.api.agent.TransactionNamePriority;
 import com.newrelic.api.agent.TransportType;
 import com.newrelic.api.agent.weaver.Weave;
 import com.newrelic.api.agent.weaver.Weaver;
@@ -17,12 +18,11 @@ public abstract class Pipeline_instrumentation {
 
 	private String id = Weaver.callOriginal();
 
-	@Trace(dispatcher = true)
+	@Trace
 	public boolean process(Exchange exchange, AsyncCallback callback) {
 		Map<String, Object> attributes = new HashMap<String, Object>();
 		Util.recordExchange(attributes, exchange);
 		NewRelic.getAgent().getTracedMethod().addCustomAttributes(attributes);
-		NewRelic.getAgent().getTransaction().acceptDistributedTraceHeaders(TransportType.Other, new CamelHeaders(exchange));
 
 		if(id != null) {
 			NewRelic.getAgent().getTracedMethod().setMetricName("Custom","Pipeline","process",id);
@@ -43,8 +43,12 @@ public void run() {
 			if(exchange != null) {
 				Map<String, Object> attributes = new HashMap<String, Object>();
 				Util.recordExchange(attributes, exchange);
+				NewRelic.getAgent().getTracedMethod().addCustomAttributes(attributes);
 				NewRelic.getAgent().getTransaction().acceptDistributedTraceHeaders(TransportType.Other, new CamelHeaders(exchange));
-
+				String fromRoute = exchange.getFromRouteId();
+				if(fromRoute == null) fromRoute = "UnknownFromRoute";
+				NewRelic.getAgent().getTracedMethod().setMetricName("Custom","PipelineTask","run",fromRoute);
+				NewRelic.getAgent().getTransaction().setTransactionName(TransactionNamePriority.FRAMEWORK_HIGH, false, "Pipeline", "Custom","PipelineTask","run",fromRoute);
 			}
 			Weaver.callOriginal();
 		}

camel-core-3.18/src/main/java/com/nr/instrumentation/apache/camel/Util.java

@@ -2,8 +2,10 @@
 
 import java.lang.reflect.Method;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import org.apache.camel.CamelContext;
 import org.apache.camel.Endpoint;
@@ -22,6 +24,7 @@ public class Util {
 
 	static {
 		ignores = new ArrayList<>();
+		ignores.add("org.apache.camel.impl.engine.CamelInternalProcessor$AsyncAfterTask");
 	}
 
 	@SuppressWarnings("deprecation")
@@ -84,7 +87,7 @@ public static void recordExchange(Map<String, Object> attributes, Exchange excha
 			recordValue(attributes, "CamelContextManagementName", context.getManagementName());
 			Endpoint endpoint = exchange.getFromEndpoint();
 			if(endpoint != null) {
-				recordValue(attributes, "From_EndPointURI", endpoint.getEndpointUri());
+				recordValue(attributes, "From_EndPointURI", endpoint.getEndpointBaseUri());
 			}
 			recordValue(attributes, "FromRouteId", exchange.getFromRouteId());
 			if(exchange instanceof ExtendedExchange) {
@@ -117,4 +120,14 @@ public static void recordValue(Map<String,Object> attributes, String key, Object
 		}
 	}
 
+	public static void reportExchange(Exchange exchange) {
+		HashMap<String, Object> attributes = new HashMap<>();
+		recordValue(attributes, "ExchangeId", exchange.getExchangeId());
+		Map<String, Object> properties = exchange.getAllProperties();
+		Set<String> keys = properties.keySet();
+		for(String key : keys) {
+			recordValue(attributes, key, properties.get(key));
+		}
+		NewRelic.getAgent().getInsights().recordCustomEvent("Pipeline_Exchange", attributes);
+	}
 }

camel-core-3.18/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java

@@ -0,0 +1,67 @@
+package org.apache.camel.processor.errorhandler;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.camel.ExtendedExchange;
+
+import com.newrelic.api.agent.NewRelic;
+import com.newrelic.api.agent.weaver.Weave;
+import com.newrelic.api.agent.weaver.Weaver;
+
+@Weave
+public abstract class RedeliveryErrorHandler {
+
+	@Weave
+	protected static class SimpleTask {
+
+		private ExtendedExchange exchange = Weaver.callOriginal();		
+
+		public void run() {
+			List<String> names = new ArrayList<>();
+			names.add("Custom");
+			names.add("Camel");
+			names.add("RedeliveryErrorHandler");
+			names.add("SimpleTask");
+			if(exchange != null) {
+				String routeId = exchange.getFromRouteId();
+				if(routeId != null) {
+					names.add(routeId);
+				}
+
+			}
+			String[] metricNames = new String[names.size()];
+			names.toArray(metricNames);
+			NewRelic.getAgent().getTracedMethod().setMetricName(metricNames);
+			Weaver.callOriginal();
+		}
+
+	}
+
+	@Weave
+	protected static class RedeliveryTask {
+
+        private ExtendedExchange exchange = Weaver.callOriginal();
+
+
+		public void run() {
+			List<String> names = new ArrayList<>();
+			names.add("Custom");
+			names.add("Camel");
+			names.add("RedeliveryErrorHandler");
+			names.add("RedeliveryTask");
+			if(exchange != null) {
+				String routeId = exchange.getFromRouteId();
+				if(routeId != null) {
+					names.add(routeId);
+				}
+
+			}
+			String[] metricNames = new String[names.size()];
+			names.toArray(metricNames);
+			NewRelic.getAgent().getTracedMethod().setMetricName(metricNames);
+			Weaver.callOriginal();
+		}
+
+	}
+}

camel-core-3.18/src/main/java/org/apache/camel/support/DefaultAsyncProducer.java

@@ -28,6 +28,7 @@ public void process(Exchange exchange) {
 		Weaver.callOriginal();
 	}
 
+	@Trace(dispatcher = true)
 	public CompletableFuture<Exchange> processAsync(Exchange exchange) {
 		HashMap<String, Object> attributes = new HashMap<>();
 		Util.recordExchange(attributes, exchange);

camel-core-3.18/src/main/java/org/apache/camel/support/DefaultConsumer.java

@@ -5,6 +5,7 @@
 import org.apache.camel.Processor;
 import org.apache.camel.Route;
 
+import com.newrelic.api.agent.NewRelic;
 import com.newrelic.api.agent.weaver.MatchType;
 import com.newrelic.api.agent.weaver.Weave;
 import com.newrelic.api.agent.weaver.Weaver;
@@ -18,6 +19,8 @@ public abstract class DefaultConsumer {
 
 	public abstract Route getRoute();
 
+	public abstract String getRouteId();
+
 	public synchronized AsyncProcessor getAsyncProcessor() {
 		AsyncProcessor asyncProcessor = Weaver.callOriginal();
 		if(asyncProcessor != null && !(asyncProcessor instanceof NRAsyncProcessorWrapper)) {
@@ -33,4 +36,23 @@ public Processor getProcessor() {
 		}
 		return processor;
 	}
+
+	@Weave
+	private static class DefaultConsumerCallback {
+
+		private final DefaultConsumer consumer = Weaver.callOriginal();
+
+		@SuppressWarnings("unused")
+		public void done(boolean doneSync) {
+			if(consumer != null) {
+				String id = consumer.getRouteId();
+				if(id != null && !id.isEmpty()) {
+					NewRelic.getAgent().getTracedMethod().setMetricName("Custom","DefaultConsumerCallback",id);
+				}
+			}
+
+			Weaver.callOriginal();
+		}
+
+	}
 }

camel-core-3.9/src/main/java/com/nr/instrumentation/apache/camel/NRAsyncProcessorWrapper.java

@@ -38,6 +38,10 @@ public boolean process(Exchange exchange, AsyncCallback callback) {
 				names = new String[] {"Custom","AsyncProcessor",delegate.getClass().getSimpleName(),"process"};
 			}
 		} else {
+			Object timerTaskValue = exchange.getProperty(Exchange.TIMER_NAME);
+			if(timerTaskValue != null) {
+				names = new String[] {"Custom","AsyncProcessor",delegate.getClass().getSimpleName(),"process","TimerTask",timerTaskValue.toString()};
+			}
 			names = new String[] {"Custom","AsyncProcessor",delegate.getClass().getSimpleName(),"process"};
 
 		}

camel-core-3.9/src/main/java/com/nr/instrumentation/apache/camel/Pipeline_instrumentation.java

@@ -8,6 +8,7 @@
 
 import com.newrelic.api.agent.NewRelic;
 import com.newrelic.api.agent.Trace;
+import com.newrelic.api.agent.TransactionNamePriority;
 import com.newrelic.api.agent.TransportType;
 import com.newrelic.api.agent.weaver.Weave;
 import com.newrelic.api.agent.weaver.Weaver;
@@ -17,12 +18,11 @@ public abstract class Pipeline_instrumentation {
 
 	private String id = Weaver.callOriginal();
 
-	@Trace(dispatcher = true)
+	@Trace
 	public boolean process(Exchange exchange, AsyncCallback callback) {
 		Map<String, Object> attributes = new HashMap<String, Object>();
 		Util.recordExchange(attributes, exchange);
 		NewRelic.getAgent().getTracedMethod().addCustomAttributes(attributes);
-		NewRelic.getAgent().getTransaction().acceptDistributedTraceHeaders(TransportType.Other, new CamelHeaders(exchange));
 
 		if(id != null) {
 			NewRelic.getAgent().getTracedMethod().setMetricName("Custom","Pipeline","process",id);
@@ -43,11 +43,15 @@ public void run() {
 			if(exchange != null) {
 				Map<String, Object> attributes = new HashMap<String, Object>();
 				Util.recordExchange(attributes, exchange);
+				NewRelic.getAgent().getTracedMethod().addCustomAttributes(attributes);
 				NewRelic.getAgent().getTransaction().acceptDistributedTraceHeaders(TransportType.Other, new CamelHeaders(exchange));
-
+				String fromRoute = exchange.getFromRouteId();
+				if(fromRoute == null) fromRoute = "UnknownFromRoute";
+				NewRelic.getAgent().getTracedMethod().setMetricName("Custom","PipelineTask","run",fromRoute);
+				NewRelic.getAgent().getTransaction().setTransactionName(TransactionNamePriority.FRAMEWORK_HIGH, false, "Pipeline", "Custom","PipelineTask","run",fromRoute);
 			}
 			Weaver.callOriginal();
 		}
 	}
 
-}
+}

camel-core-3.9/src/main/java/com/nr/instrumentation/apache/camel/Util.java

@@ -2,8 +2,10 @@
 
 import java.lang.reflect.Method;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import org.apache.camel.CamelContext;
 import org.apache.camel.Endpoint;
@@ -22,6 +24,7 @@ public class Util {
 
 	static {
 		ignores = new ArrayList<>();
+		ignores.add("org.apache.camel.impl.engine.CamelInternalProcessor$AsyncAfterTask");
 	}
 
 	@SuppressWarnings("deprecation")
@@ -84,7 +87,7 @@ public static void recordExchange(Map<String, Object> attributes, Exchange excha
 			recordValue(attributes, "CamelContextManagementName", context.getManagementName());
 			Endpoint endpoint = exchange.getFromEndpoint();
 			if(endpoint != null) {
-				recordValue(attributes, "From_EndPointURI", endpoint.getEndpointUri());
+				recordValue(attributes, "From_EndPointURI", endpoint.getEndpointBaseUri());
 			}
 			recordValue(attributes, "FromRouteId", exchange.getFromRouteId());
 			if(exchange instanceof ExtendedExchange) {
@@ -116,4 +119,14 @@ public static void recordValue(Map<String,Object> attributes, String key, Object
 		}
 	}
 
+	public static void reportExchange(Exchange exchange) {
+		HashMap<String, Object> attributes = new HashMap<>();
+		recordValue(attributes, "ExchangeId", exchange.getExchangeId());
+		Map<String, Object> properties = exchange.getAllProperties();
+		Set<String> keys = properties.keySet();
+		for(String key : keys) {
+			recordValue(attributes, key, properties.get(key));
+		}
+		NewRelic.getAgent().getInsights().recordCustomEvent("Pipeline_Exchange", attributes);
+	}
 }

camel-core-3.9/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java

@@ -0,0 +1,67 @@
+package org.apache.camel.processor.errorhandler;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.camel.ExtendedExchange;
+
+import com.newrelic.api.agent.NewRelic;
+import com.newrelic.api.agent.weaver.Weave;
+import com.newrelic.api.agent.weaver.Weaver;
+
+@Weave
+public abstract class RedeliveryErrorHandler {
+
+	@Weave
+	protected static class SimpleTask {
+
+		private ExtendedExchange exchange = Weaver.callOriginal();		
+
+		public void run() {
+			List<String> names = new ArrayList<>();
+			names.add("Custom");
+			names.add("Camel");
+			names.add("RedeliveryErrorHandler");
+			names.add("SimpleTask");
+			if(exchange != null) {
+				String routeId = exchange.getFromRouteId();
+				if(routeId != null) {
+					names.add(routeId);
+				}
+
+			}
+			String[] metricNames = new String[names.size()];
+			names.toArray(metricNames);
+			NewRelic.getAgent().getTracedMethod().setMetricName(metricNames);
+			Weaver.callOriginal();
+		}
+
+	}
+
+	@Weave
+	protected static class RedeliveryTask {
+
+        private ExtendedExchange exchange = Weaver.callOriginal();
+
+
+		public void run() {
+			List<String> names = new ArrayList<>();
+			names.add("Custom");
+			names.add("Camel");
+			names.add("RedeliveryErrorHandler");
+			names.add("RedeliveryTask");
+			if(exchange != null) {
+				String routeId = exchange.getFromRouteId();
+				if(routeId != null) {
+					names.add(routeId);
+				}
+
+			}
+			String[] metricNames = new String[names.size()];
+			names.toArray(metricNames);
+			NewRelic.getAgent().getTracedMethod().setMetricName(metricNames);
+			Weaver.callOriginal();
+		}
+
+	}
+}

camel-core-3.9/src/main/java/org/apache/camel/spi/ReactiveExecutor.java

@@ -39,4 +39,4 @@ public boolean executeFromQueue() {
     	return Weaver.callOriginal();
     }
 
-}
+}

camel-core-3.9/src/main/java/org/apache/camel/support/DefaultAsyncProducer.java

@@ -28,6 +28,7 @@ public void process(Exchange exchange) {
 		Weaver.callOriginal();
 	}
 
+	@Trace(dispatcher = true)
 	public CompletableFuture<Exchange> processAsync(Exchange exchange) {
 		HashMap<String, Object> attributes = new HashMap<>();
 		Util.recordExchange(attributes, exchange);
@@ -53,4 +54,4 @@ public CompletableFuture<Exchange> processAsync(Exchange exchange) {
 		return result.whenComplete(consumer);
 	}
 
-}
+}