SpringCloud/spring-cloud-sleuth
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Wrapping an existing JmsEndpointRegistry; fixes gh-1200 (#1211)

Browse Source 
Wrapping an existing JmsEndpointRegistry

fixes #1200
This commit is contained in:
Marcin Grzejszczak
2019-02-12 16:19:09 +01:00
committed by GitHub
parent 1e2d2f133a
commit 8648a34e98
4 changed files with 360 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceMessagingAutoConfiguration.java
View File

@@ -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<T extends MessageListener>
}
}
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);
}
}

170
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingConnectionFactoryBeanPostProcessor.java
View File

@@ -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> 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.
* <p>
* 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();
}
}
}

297
spring-cloud-sleuth-core/src/main/java/org/springframework/jms/config/TracingJmsListenerEndpointRegistry.java Normal file
View File

@@ -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<String> getListenerContainerIds() {
return this.delegate.getListenerContainerIds();
}
@Override
public Collection<MessageListenerContainer> 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> 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.
* <p>
* 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();
}
}
}

33
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/JmsTracingConfigurationTest.java
View File

@@ -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";