diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceMessagingAutoConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceMessagingAutoConfiguration.java index 762c5d74e..701712e31 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceMessagingAutoConfiguration.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceMessagingAutoConfiguration.java @@ -25,7 +25,6 @@ import brave.Tracer; import brave.Tracing; import brave.jms.JmsTracing; import brave.kafka.clients.KafkaTracing; -import brave.propagation.CurrentTraceContext; import brave.spring.rabbit.SpringRabbitTracing; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; @@ -57,6 +56,8 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Role; import org.springframework.jms.annotation.JmsListenerConfigurer; +import org.springframework.jms.config.JmsListenerEndpointRegistry; +import org.springframework.jms.config.TracingJmsListenerEndpointRegistry; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.AbstractMessageListenerContainer; import org.springframework.kafka.listener.MessageListener; @@ -129,6 +130,7 @@ public class TraceMessagingAutoConfiguration { @Configuration @ConditionalOnProperty(value = "spring.sleuth.messaging.jms.enabled", matchIfMissing = true) @ConditionalOnClass(JmsListenerConfigurer.class) + @ConditionalOnBean(JmsListenerEndpointRegistry.class) @Role(BeanDefinition.ROLE_INFRASTRUCTURE) protected static class SleuthJmsConfiguration { @@ -149,12 +151,22 @@ public class TraceMessagingAutoConfiguration { return new TracingConnectionFactoryBeanPostProcessor(beanFactory); } + @Bean + JmsListenerConfigurer configureTracing(BeanFactory beanFactory, + JmsListenerEndpointRegistry defaultRegistry) { + return registrar -> { + TracingJmsBeanPostProcessor processor = tracingJmsBeanPostProcessor( + beanFactory); + JmsListenerEndpointRegistry registry = registrar.getEndpointRegistry(); + registrar.setEndpointRegistry((JmsListenerEndpointRegistry) processor + .wrap(registry == null ? defaultRegistry : registry)); + }; + } + // Setup the tracing endpoint registry. @Bean - JmsListenerConfigurer configureTracing(JmsTracing jmsTracing, - CurrentTraceContext current) { - return registrar -> registrar.setEndpointRegistry( - new TracingJmsListenerEndpointRegistry(jmsTracing, current)); + TracingJmsBeanPostProcessor tracingJmsBeanPostProcessor(BeanFactory beanFactory) { + return new TracingJmsBeanPostProcessor(beanFactory); } } @@ -325,3 +337,32 @@ class MessageListenerMethodInterceptor } } + +class TracingJmsBeanPostProcessor implements BeanPostProcessor { + + private final BeanFactory beanFactory; + + TracingJmsBeanPostProcessor(BeanFactory beanFactory) { + this.beanFactory = beanFactory; + } + + @Override + public Object postProcessAfterInitialization(Object bean, String beanName) + throws BeansException { + return wrap(bean); + } + + Object wrap(Object bean) { + if (typeMatches(bean)) { + return new TracingJmsListenerEndpointRegistry( + (JmsListenerEndpointRegistry) bean, this.beanFactory); + } + return bean; + } + + private boolean typeMatches(Object bean) { + return bean instanceof JmsListenerEndpointRegistry + && !(bean instanceof TracingJmsListenerEndpointRegistry); + } + +} diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingConnectionFactoryBeanPostProcessor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingConnectionFactoryBeanPostProcessor.java index 718a3209e..b827710ba 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingConnectionFactoryBeanPostProcessor.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingConnectionFactoryBeanPostProcessor.java @@ -16,36 +16,23 @@ package org.springframework.cloud.sleuth.instrument.messaging; -import java.lang.reflect.Field; - import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.JMSContext; import javax.jms.JMSException; import javax.jms.Message; -import javax.jms.MessageConsumer; import javax.jms.MessageListener; -import javax.jms.Session; import javax.jms.XAConnection; import javax.jms.XAConnectionFactory; import javax.jms.XAJMSContext; -import brave.Span; import brave.jms.JmsTracing; -import brave.propagation.CurrentTraceContext; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.config.BeanPostProcessor; -import org.springframework.jms.config.JmsListenerContainerFactory; -import org.springframework.jms.config.JmsListenerEndpoint; -import org.springframework.jms.config.JmsListenerEndpointRegistry; -import org.springframework.jms.config.MethodJmsListenerEndpoint; -import org.springframework.jms.config.SimpleJmsListenerEndpoint; import org.springframework.jms.connection.CachingConnectionFactory; -import org.springframework.jms.listener.adapter.MessagingMessageListenerAdapter; import org.springframework.jms.listener.endpoint.JmsMessageEndpointManager; -import org.springframework.lang.Nullable; /** * {@link BeanPostProcessor} wrapping around JMS {@link ConnectionFactory}. @@ -244,160 +231,3 @@ class LazyMessageListener implements MessageListener { } } - -/** - * This ensures listeners end up continuing the trace from - * {@link MessageConsumer#receive()}. - */ -class TracingJmsListenerEndpointRegistry extends JmsListenerEndpointRegistry { - - final JmsTracing jmsTracing; - - final CurrentTraceContext current; - - // Not all state can be copied without using reflection - final Field messageHandlerMethodFactoryField; - - final Field embeddedValueResolverField; - - TracingJmsListenerEndpointRegistry(JmsTracing jmsTracing, - CurrentTraceContext current) { - this.jmsTracing = jmsTracing; - this.current = current; - this.messageHandlerMethodFactoryField = tryField("messageHandlerMethodFactory"); - this.embeddedValueResolverField = tryField("embeddedValueResolver"); - } - - @Nullable - static Field tryField(String name) { - try { - Field field = MethodJmsListenerEndpoint.class.getDeclaredField(name); - field.setAccessible(true); - return field; - } - catch (NoSuchFieldException e) { - return null; - } - } - - @Nullable - static T get(Object object, Field field) throws IllegalAccessException { - return (T) field.get(object); - } - - @Override - public void registerListenerContainer(JmsListenerEndpoint endpoint, - JmsListenerContainerFactory factory, boolean startImmediately) { - if (endpoint instanceof MethodJmsListenerEndpoint) { - endpoint = trace((MethodJmsListenerEndpoint) endpoint); - } - else if (endpoint instanceof SimpleJmsListenerEndpoint) { - endpoint = trace((SimpleJmsListenerEndpoint) endpoint); - } - super.registerListenerContainer(endpoint, factory, startImmediately); - } - - /** - * This wraps the {@link SimpleJmsListenerEndpoint#getMessageListener()} delegate in a - * new span. - * @param source jms endpoint - * @return wrapped endpoint - */ - SimpleJmsListenerEndpoint trace(SimpleJmsListenerEndpoint source) { - MessageListener delegate = source.getMessageListener(); - if (delegate == null) { - return source; - } - source.setMessageListener(this.jmsTracing.messageListener(delegate, false)); - return source; - } - - /** - * It would be better to trace by wrapping, but - * {@link MethodJmsListenerEndpoint#createMessageListenerInstance()}, is protected so - * we can't call it from outside code. In other words, a forwarding pattern can't be - * used. Instead, we copy state from the input. - *

- * NOTE: As {@linkplain MethodJmsListenerEndpoint} is neither final, nor effectively - * final. For this reason we can't ensure copying will get all state. For example, a - * subtype could hold state we aren't aware of, or change behavior. We can consider - * checking that input is not a subtype, and most conservatively leaving unknown - * subtypes untraced. - * @param source jms endpoint - * @return wrapped endpoint - */ - MethodJmsListenerEndpoint trace(MethodJmsListenerEndpoint source) { - // Skip out rather than incompletely copying the source - if (this.messageHandlerMethodFactoryField == null - || this.embeddedValueResolverField == null) { - return source; - } - - // We want the stock implementation, except we want to wrap the message listener - // in a new span - MethodJmsListenerEndpoint dest = new MethodJmsListenerEndpoint() { - @Override - protected MessagingMessageListenerAdapter createMessageListenerInstance() { - return new TracingMessagingMessageListenerAdapter( - TracingJmsListenerEndpointRegistry.this.jmsTracing, - TracingJmsListenerEndpointRegistry.this.current); - } - }; - - // set state from AbstractJmsListenerEndpoint - dest.setId(source.getId()); - dest.setDestination(source.getDestination()); - dest.setSubscription(source.getSubscription()); - dest.setSelector(source.getSelector()); - dest.setConcurrency(source.getConcurrency()); - - // set state from MethodJmsListenerEndpoint - dest.setBean(source.getBean()); - dest.setMethod(source.getMethod()); - dest.setMostSpecificMethod(source.getMostSpecificMethod()); - - try { - dest.setMessageHandlerMethodFactory( - get(source, this.messageHandlerMethodFactoryField)); - dest.setEmbeddedValueResolver(get(source, this.embeddedValueResolverField)); - } - catch (IllegalAccessException e) { - return source; // skip out rather than incompletely copying the source - } - return dest; - } - -} - -/** - * This wraps the message listener in a child span. - */ -final class TracingMessagingMessageListenerAdapter - extends MessagingMessageListenerAdapter { - - final JmsTracing jmsTracing; - - final CurrentTraceContext current; - - TracingMessagingMessageListenerAdapter(JmsTracing jmsTracing, - CurrentTraceContext current) { - this.jmsTracing = jmsTracing; - this.current = current; - } - - @Override - public void onMessage(Message message, Session session) throws JMSException { - Span span = this.jmsTracing.nextSpan(message).name("on-message").start(); - try (CurrentTraceContext.Scope ws = this.current.newScope(span.context())) { - super.onMessage(message, session); - } - catch (JMSException | RuntimeException | Error e) { - span.error(e); - throw e; - } - finally { - span.finish(); - } - } - -} diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/jms/config/TracingJmsListenerEndpointRegistry.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/jms/config/TracingJmsListenerEndpointRegistry.java new file mode 100644 index 000000000..a3cba2282 --- /dev/null +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/jms/config/TracingJmsListenerEndpointRegistry.java @@ -0,0 +1,297 @@ +/* + * Copyright 2013-2019 the original author or authors. + * + * Licensed 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.springframework.jms.config; + +import java.lang.reflect.Field; +import java.util.Collection; +import java.util.Set; + +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.MessageListener; +import javax.jms.Session; + +import brave.Span; +import brave.jms.JmsTracing; +import brave.propagation.CurrentTraceContext; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.ApplicationContext; +import org.springframework.context.event.ContextRefreshedEvent; +import org.springframework.jms.listener.MessageListenerContainer; +import org.springframework.jms.listener.adapter.MessagingMessageListenerAdapter; +import org.springframework.lang.Nullable; + +/** + * This ensures listeners end up continuing the trace from + * {@link javax.jms.MessageConsumer#receive()}. + * + * Internal class for Sleuth, do not use. Its API can change at anytime. + * + * Placed under this package cause we need to use the package scoped API. + * + * @author Marcin Grzejszczak + * @since 2.1.1 + */ +public final class TracingJmsListenerEndpointRegistry + extends JmsListenerEndpointRegistry { + + private final BeanFactory beanFactory; + + private final JmsListenerEndpointRegistry delegate; + + private JmsTracing jmsTracing; + + private CurrentTraceContext currentTraceContext; + + // Not all state can be copied without using reflection + final Field messageHandlerMethodFactoryField; + + final Field embeddedValueResolverField; + + public TracingJmsListenerEndpointRegistry(JmsListenerEndpointRegistry registry, + BeanFactory beanFactory) { + this.delegate = registry; + this.beanFactory = beanFactory; + this.messageHandlerMethodFactoryField = tryField("messageHandlerMethodFactory"); + this.embeddedValueResolverField = tryField("embeddedValueResolver"); + } + + private JmsTracing jmsTracing() { + if (this.jmsTracing == null) { + this.jmsTracing = this.beanFactory.getBean(JmsTracing.class); + } + return this.jmsTracing; + } + + private CurrentTraceContext currentTraceContext() { + if (this.currentTraceContext == null) { + this.currentTraceContext = this.beanFactory + .getBean(CurrentTraceContext.class); + } + return this.currentTraceContext; + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) { + this.delegate.setApplicationContext(applicationContext); + } + + @Override + public void onApplicationEvent(ContextRefreshedEvent event) { + this.delegate.onApplicationEvent(event); + } + + @Override + public MessageListenerContainer getListenerContainer(String id) { + return this.delegate.getListenerContainer(id); + } + + @Override + public Set getListenerContainerIds() { + return this.delegate.getListenerContainerIds(); + } + + @Override + public Collection getListenerContainers() { + return this.delegate.getListenerContainers(); + } + + @Override + public void registerListenerContainer(JmsListenerEndpoint endpoint, + JmsListenerContainerFactory factory) { + this.delegate.registerListenerContainer(wrapEndpoint(endpoint), factory); + } + + @Override + protected MessageListenerContainer createListenerContainer( + JmsListenerEndpoint endpoint, JmsListenerContainerFactory factory) { + return this.delegate.createListenerContainer(wrapEndpoint(endpoint), factory); + } + + @Override + public int getPhase() { + return this.delegate.getPhase(); + } + + @Override + public void start() { + this.delegate.start(); + } + + @Override + public void stop() { + this.delegate.stop(); + } + + @Override + public void stop(Runnable callback) { + this.delegate.stop(callback); + } + + @Override + public boolean isRunning() { + return this.delegate.isRunning(); + } + + @Override + public void destroy() { + this.delegate.destroy(); + } + + @Override + public boolean isAutoStartup() { + return this.delegate.isAutoStartup(); + } + + @Nullable + static Field tryField(String name) { + try { + Field field = MethodJmsListenerEndpoint.class.getDeclaredField(name); + field.setAccessible(true); + return field; + } + catch (NoSuchFieldException e) { + return null; + } + } + + @Nullable + static T get(Object object, Field field) throws IllegalAccessException { + return (T) field.get(object); + } + + @Override + public void registerListenerContainer(JmsListenerEndpoint endpoint, + JmsListenerContainerFactory factory, boolean startImmediately) { + this.delegate.registerListenerContainer(wrapEndpoint(endpoint), factory, + startImmediately); + } + + private JmsListenerEndpoint wrapEndpoint(JmsListenerEndpoint endpoint) { + if (endpoint instanceof MethodJmsListenerEndpoint) { + return trace((MethodJmsListenerEndpoint) endpoint); + } + else if (endpoint instanceof SimpleJmsListenerEndpoint) { + return trace((SimpleJmsListenerEndpoint) endpoint); + } + return endpoint; + } + + /** + * This wraps the {@link SimpleJmsListenerEndpoint#getMessageListener()} delegate in a + * new span. + * @param source jms endpoint + * @return wrapped endpoint + */ + SimpleJmsListenerEndpoint trace(SimpleJmsListenerEndpoint source) { + MessageListener delegate = source.getMessageListener(); + if (delegate == null) { + return source; + } + source.setMessageListener(jmsTracing().messageListener(delegate, false)); + return source; + } + + /** + * It would be better to trace by wrapping, but + * {@link MethodJmsListenerEndpoint#createMessageListenerInstance()}, is protected so + * we can't call it from outside code. In other words, a forwarding pattern can't be + * used. Instead, we copy state from the input. + *

+ * NOTE: As {@linkplain MethodJmsListenerEndpoint} is neither final, nor effectively + * final. For this reason we can't ensure copying will get all state. For example, a + * subtype could hold state we aren't aware of, or change behavior. We can consider + * checking that input is not a subtype, and most conservatively leaving unknown + * subtypes untraced. + * @param source jms endpoint + * @return wrapped endpoint + */ + MethodJmsListenerEndpoint trace(MethodJmsListenerEndpoint source) { + // Skip out rather than incompletely copying the source + if (this.messageHandlerMethodFactoryField == null + || this.embeddedValueResolverField == null) { + return source; + } + // We want the stock implementation, except we want to wrap the message listener + // in a new span + MethodJmsListenerEndpoint dest = new MethodJmsListenerEndpoint() { + @Override + protected MessagingMessageListenerAdapter createMessageListenerInstance() { + return new TracingMessagingMessageListenerAdapter(jmsTracing(), + currentTraceContext()); + } + }; + // set state from AbstractJmsListenerEndpoint + dest.setId(source.getId()); + dest.setDestination(source.getDestination()); + dest.setSubscription(source.getSubscription()); + dest.setSelector(source.getSelector()); + dest.setConcurrency(source.getConcurrency()); + // set state from MethodJmsListenerEndpoint + dest.setBean(source.getBean()); + dest.setMethod(source.getMethod()); + dest.setMostSpecificMethod(source.getMostSpecificMethod()); + try { + dest.setMessageHandlerMethodFactory( + get(source, this.messageHandlerMethodFactoryField)); + dest.setEmbeddedValueResolver(get(source, this.embeddedValueResolverField)); + } + catch (IllegalAccessException e) { + return source; // skip out rather than incompletely copying the source + } + return dest; + } + +} + +final class TracingMethodJmsListenerEndpoint extends MethodJmsListenerEndpoint { + +} + +/** + * This wraps the message listener in a child span. + */ +final class TracingMessagingMessageListenerAdapter + extends MessagingMessageListenerAdapter { + + final JmsTracing jmsTracing; + + final CurrentTraceContext current; + + TracingMessagingMessageListenerAdapter(JmsTracing jmsTracing, + CurrentTraceContext current) { + this.jmsTracing = jmsTracing; + this.current = current; + } + + @Override + public void onMessage(Message message, Session session) throws JMSException { + Span span = this.jmsTracing.nextSpan(message).name("on-message").start(); + try (CurrentTraceContext.Scope ws = this.current.newScope(span.context())) { + super.onMessage(message, session); + } + catch (JMSException | RuntimeException | Error e) { + span.error(e); + throw e; + } + finally { + span.finish(); + } + } + +} diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/JmsTracingConfigurationTest.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/JmsTracingConfigurationTest.java index fa2f33001..c00e60821 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/JmsTracingConfigurationTest.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/JmsTracingConfigurationTest.java @@ -51,6 +51,7 @@ import org.springframework.boot.test.context.assertj.AssertableApplicationContex import org.springframework.boot.test.context.runner.ApplicationContextRunner; import org.springframework.cloud.gateway.config.GatewayAutoConfiguration; import org.springframework.cloud.gateway.config.GatewayClassPathWarningAutoConfiguration; +import org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.jca.support.ResourceAdapterFactoryBean; @@ -72,11 +73,10 @@ import static org.assertj.core.api.Assertions.assertThat; public class JmsTracingConfigurationTest { final ApplicationContextRunner contextRunner = new ApplicationContextRunner() - .withConfiguration( - AutoConfigurations.of(AnnotationJmsListenerConfiguration.class, - XAConfiguration.class, SimpleJmsListenerConfiguration.class, - JcaJmsListenerConfiguration.class, - JmsTestTracingConfiguration.class)); + .withConfiguration(AutoConfigurations.of(JmsTestTracingConfiguration.class, + AnnotationJmsListenerConfiguration.class, XAConfiguration.class, + SimpleJmsListenerConfiguration.class, + JcaJmsListenerConfiguration.class)); static void clearSpans(AssertableApplicationContext ctx) throws JMSException { ctx.getBean(JmsTestTracingConfiguration.class).clearSpan(); @@ -137,8 +137,8 @@ public class JmsTracingConfigurationTest { assertThat(trace).allSatisfy(s -> assertThat(s.traceId()) .isEqualTo(trace.get(0).traceId())); - assertThat(trace).extracting(Span::name).contains("send", "receive", - "on-message"); + assertThat(trace).isNotNull().extracting(Span::name).contains("send", + "receive", "on-message"); }); } @@ -155,7 +155,7 @@ public class JmsTracingConfigurationTest { assertThat(trace).allSatisfy(s -> assertThat(s.traceId()) .isEqualTo(trace.get(0).traceId())); - assertThat(trace).extracting(Span::name) + assertThat(trace).isNotNull().extracting(Span::name) .containsExactlyInAnyOrder("send", "receive", "on-message"); }); } @@ -173,7 +173,7 @@ public class JmsTracingConfigurationTest { assertThat(trace).allSatisfy(s -> assertThat(s.traceId()) .isEqualTo(trace.get(0).traceId())); - assertThat(trace).extracting(Span::name) + assertThat(trace).isNotNull().extracting(Span::name) .containsExactlyInAnyOrder("send", "receive", "on-message"); }); } @@ -208,8 +208,8 @@ public class JmsTracingConfigurationTest { MessageListener simpleMessageListener(CurrentTraceContext current) { return message -> { // Didn't restart the trace - assertThat(current.get()).extracting(TraceContext::parentIdAsLong) - .isNotEqualTo(0L); + assertThat(current.get()).isNotNull() + .extracting(TraceContext::parentIdAsLong).isNotEqualTo(0L); }; } @@ -224,8 +224,8 @@ public class JmsTracingConfigurationTest { @JmsListener(destination = "myQueue") public void onMessage() { - assertThat(this.current.get()).extracting(TraceContext::parentIdAsLong) - .isNotEqualTo(0L); + assertThat(this.current.get()).isNotNull() + .extracting(TraceContext::parentIdAsLong).isNotEqualTo(0L); } } @@ -250,8 +250,8 @@ public class JmsTracingConfigurationTest { MessageListener simpleMessageListener(CurrentTraceContext current) { return message -> { // Didn't restart the trace - assertThat(current.get()).extracting(TraceContext::parentIdAsLong) - .isNotEqualTo(0L); + assertThat(current.get()).isNotNull() + .extracting(TraceContext::parentIdAsLong).isNotEqualTo(0L); }; } @@ -277,7 +277,8 @@ public class JmsTracingConfigurationTest { @Configuration @EnableAutoConfiguration(exclude = { GatewayAutoConfiguration.class, - GatewayClassPathWarningAutoConfiguration.class }) + GatewayClassPathWarningAutoConfiguration.class, + EurekaClientAutoConfiguration.class }) class JmsTestTracingConfiguration { static final String CONTEXT_LEAK = "context.leak";