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

SQS Messaging Autoconfig (#1218)

Browse Source 
added support for SQS messaging tracing
This commit is contained in:
Brian Devins-Suresh
2019-10-03 06:18:32 -04:00
committed by Marcin Grzejszczak
parent c7b818753f
commit dc82c5cafc
10 changed files with 468 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
docs/src/main/asciidoc/spring-cloud-sleuth.adoc
View File

@@ -1381,6 +1381,13 @@ To block this feature, set `spring.sleuth.messaging.jms.enabled` to `false`.
IMPORTANT: We don't support baggage propagation for JMS
==== Spring Cloud AWS Messaging SQS
We instrument `@SqsListener` which is provided by `org.springframework.cloud:spring-cloud-aws-messaging`
so that tracing headers get extracted from the message and a trace gets put into the context.
To block this feature, set `spring.sleuth.messaging.sqs.enabled` to `false`.
=== Zuul
We instrument the Zuul Ribbon integration by enriching the Ribbon requests with tracing information.

9
pom.xml
View File

@@ -158,6 +158,13 @@
<type>pom</type>
<scope>import</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-aws-dependencies</artifactId>
<version>${spring-cloud-aws.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-gateway-dependencies</artifactId>
@@ -250,8 +257,8 @@
</spring-cloud-openfeign.version>
<spring-security-boot-autoconfigure.version>2.1.7.RELEASE
</spring-security-boot-autoconfigure.version>
<spring-cloud-aws.version>2.2.0.BUILD-SNAPSHOT</spring-cloud-aws.version>
<disable.nohttp.checks>false</disable.nohttp.checks>
<okhttp.version>3.10.0</okhttp.version>
<mockwebserver.version>3.10.0</mockwebserver.version>
<guava.version>20.0</guava.version>

5
spring-cloud-sleuth-core/pom.xml
View File

@@ -82,6 +82,11 @@
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-commons</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-aws-messaging</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-gateway</artifactId>

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

@@ -17,12 +17,15 @@
package org.springframework.cloud.sleuth.instrument.messaging;
import java.lang.reflect.Field;
import java.util.Collections;
import java.util.List;
import brave.Span;
import brave.Tracer;
import brave.Tracing;
import brave.jms.JmsTracing;
import brave.kafka.clients.KafkaTracing;
import brave.propagation.Propagation;
import brave.spring.rabbit.SpringRabbitTracing;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
@@ -49,6 +52,8 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.aws.messaging.config.QueueMessageHandlerFactory;
import org.springframework.cloud.aws.messaging.listener.QueueMessageHandler;
import org.springframework.cloud.sleuth.autoconfig.TraceAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -62,6 +67,11 @@ import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter;
import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ReflectionUtils;
/**
@@ -73,7 +83,8 @@ import org.springframework.util.ReflectionUtils;
*/
@Configuration
@ConditionalOnBean(Tracing.class)
@AutoConfigureAfter({ TraceAutoConfiguration.class })
@AutoConfigureAfter({ TraceAutoConfiguration.class,
TraceSpringMessagingAutoConfiguration.class })
@OnMessagingEnabled
@EnableConfigurationProperties(SleuthMessagingProperties.class)
public class TraceMessagingAutoConfiguration {
@@ -188,6 +199,28 @@ public class TraceMessagingAutoConfiguration {
}
@Configuration
@ConditionalOnProperty(value = "spring.sleuth.messaging.sqs.enabled",
matchIfMissing = true)
@ConditionalOnClass(QueueMessageHandler.class)
protected static class SleuthSqsConfiguration {
@Bean
TracingMethodMessageHandlerAdapter tracingMethodMessageHandlerAdapter(
Tracing tracing,
Propagation.Getter<MessageHeaderAccessor, String> traceMessagePropagationGetter) {
return new TracingMethodMessageHandlerAdapter(tracing,
traceMessagePropagationGetter);
}
@Bean
QueueMessageHandlerFactory sqsQueueMessageHandlerFactory(
TracingMethodMessageHandlerAdapter tracingMethodMessageHandlerAdapter) {
return new SqsQueueMessageHandlerFactory(tracingMethodMessageHandlerAdapter);
}
}
}
class SleuthRabbitBeanPostProcessor implements BeanPostProcessor {
@@ -422,3 +455,50 @@ class SleuthKafkaHeaderMapperBeanPostProcessor implements BeanPostProcessor {
}
}
class SqsQueueMessageHandlerFactory extends QueueMessageHandlerFactory {
private TracingMethodMessageHandlerAdapter handlerAdapter;
SqsQueueMessageHandlerFactory(TracingMethodMessageHandlerAdapter handlerAdapter) {
this.handlerAdapter = handlerAdapter;
}
@Override
public QueueMessageHandler createQueueMessageHandler() {
if (CollectionUtils.isEmpty(getMessageConverters())) {
return new SqsQueueMessageHandler(handlerAdapter, Collections.emptyList());
}
return new SqsQueueMessageHandler(handlerAdapter, getMessageConverters());
}
}
class SqsQueueMessageHandler extends QueueMessageHandler {
// copied from QueueMessageHandler
private static final String LOGICAL_RESOURCE_ID = "LogicalResourceId";
private TracingMethodMessageHandlerAdapter handlerAdapter;
SqsQueueMessageHandler(TracingMethodMessageHandlerAdapter handlerAdapter,
List<MessageConverter> messageConverters) {
super(messageConverters);
this.handlerAdapter = handlerAdapter;
}
@Override
public void handleMessage(Message<?> message) throws MessagingException {
handlerAdapter.wrapMethodMessageHandler(message, super::handleMessage,
this::messageSpanTagger);
}
private void messageSpanTagger(Span span, Message<?> message) {
span.remoteServiceName("sqs");
if (message.getHeaders().get(LOGICAL_RESOURCE_ID) != null) {
span.tag("sqs.queue_url",
message.getHeaders().get(LOGICAL_RESOURCE_ID).toString());
}
}
}

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

@@ -22,7 +22,6 @@ import brave.propagation.Propagation;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.sleuth.autoconfig.TraceAutoConfiguration;
@@ -44,7 +43,8 @@ import org.springframework.messaging.support.MessageHeaderAccessor;
@Configuration
@ConditionalOnClass(GlobalChannelInterceptor.class)
@ConditionalOnBean(Tracing.class)
@AutoConfigureAfter({ TraceAutoConfiguration.class })
@AutoConfigureAfter({ TraceAutoConfiguration.class,
TraceSpringMessagingAutoConfiguration.class })
@OnMessagingEnabled
@ConditionalOnProperty(value = "spring.sleuth.integration.enabled", matchIfMissing = true)
@EnableConfigurationProperties(SleuthMessagingProperties.class)
@@ -67,16 +67,4 @@ public class TraceSpringIntegrationAutoConfiguration {
traceMessagePropagationGetter);
}
@Bean
@ConditionalOnMissingBean
Propagation.Setter<MessageHeaderAccessor, String> traceMessagePropagationSetter() {
return MessageHeaderPropagation.INSTANCE;
}
@Bean
@ConditionalOnMissingBean
Propagation.Getter<MessageHeaderAccessor, String> traceMessagePropagationGetter() {
return MessageHeaderPropagation.INSTANCE;
}
}

46
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceSpringMessagingAutoConfiguration.java Normal file
View File

@@ -0,0 +1,46 @@
/*
* 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
*
* https://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.cloud.sleuth.instrument.messaging;
import brave.propagation.Propagation;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.support.MessageHeaderAccessor;
@Configuration
@ConditionalOnClass(MessageHeaderAccessor.class)
@OnMessagingEnabled
@EnableConfigurationProperties(SleuthMessagingProperties.class)
class TraceSpringMessagingAutoConfiguration {
@Bean
@ConditionalOnMissingBean
Propagation.Setter<MessageHeaderAccessor, String> traceMessagePropagationSetter() {
return MessageHeaderPropagation.INSTANCE;
}
@Bean
@ConditionalOnMissingBean
Propagation.Getter<MessageHeaderAccessor, String> traceMessagePropagationGetter() {
return MessageHeaderPropagation.INSTANCE;
}
}

109
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingMethodMessageHandlerAdapter.java Normal file
View File

@@ -0,0 +1,109 @@
/*
* 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
*
* https://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.cloud.sleuth.instrument.messaging;
import java.util.function.BiConsumer;
import brave.Span;
import brave.Tracer;
import brave.Tracing;
import brave.propagation.Propagation;
import brave.propagation.TraceContext;
import brave.propagation.TraceContextOrSamplingFlags;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.MessageHeaderAccessor;
import static brave.Span.Kind.CONSUMER;
/**
* Adds tracing extraction to an instance of
* {@link org.springframework.messaging.handler.invocation.AbstractMethodMessageHandler}
* in a reusable way. When sub-classing a provider specific class of that type you would
* wrap the <pre>super.handleMessage(...)</pre> call with a call to this. See
* {@link org.springframework.cloud.sleuth.instrument.messaging.SqsQueueMessageHandler}
* for an example.
*
* This implementation also allows for supplying a {@link java.util.function.BiConsumer}
* instance that can be used to add queue specific tags and modifications to the span.
*
* @author Brian Devins-Suresh
*/
class TracingMethodMessageHandlerAdapter {
private Tracing tracing;
private Tracer tracer;
private TraceContext.Extractor<MessageHeaderAccessor> extractor;
TracingMethodMessageHandlerAdapter(Tracing tracing,
Propagation.Getter<MessageHeaderAccessor, String> traceMessagePropagationGetter) {
this.tracing = tracing;
this.tracer = tracing.tracer();
this.extractor = tracing.propagation().extractor(traceMessagePropagationGetter);
}
void wrapMethodMessageHandler(Message<?> message, MessageHandler messageHandler,
BiConsumer<Span, Message<?>> messageSpanTagger) {
TraceContextOrSamplingFlags extracted = extractAndClearHeaders(message);
Span consumerSpan = tracer.nextSpan(extracted);
Span listenerSpan = tracer.newChild(consumerSpan.context());
if (!consumerSpan.isNoop()) {
consumerSpan.name("next-message").kind(CONSUMER);
if (messageSpanTagger != null) {
messageSpanTagger.accept(consumerSpan, message);
}
// incur timestamp overhead only once
long timestamp = tracing.clock(consumerSpan.context())
.currentTimeMicroseconds();
consumerSpan.start(timestamp);
long consumerFinish = timestamp + 1L; // save a clock reading
consumerSpan.finish(consumerFinish);
// not using scoped span as we want to start with a pre-configured time
listenerSpan.name("on-message").start(consumerFinish);
}
try (Tracer.SpanInScope ws = tracer.withSpanInScope(listenerSpan)) {
messageHandler.handleMessage(message);
}
catch (Throwable t) {
listenerSpan.error(t);
throw t;
}
finally {
listenerSpan.finish();
}
}
private TraceContextOrSamplingFlags extractAndClearHeaders(Message<?> message) {
MessageHeaderAccessor headers = MessageHeaderAccessor.getMutableAccessor(message);
TraceContextOrSamplingFlags extracted = extractor.extract(headers);
for (String propagationKey : tracing.propagation().keys()) {
headers.removeHeader(propagationKey);
}
return extracted;
}
}

2
spring-cloud-sleuth-core/src/main/resources/META-INF/spring.factories
View File

@@ -24,10 +24,12 @@ org.springframework.cloud.sleuth.instrument.grpc.TraceGrpcAutoConfiguration,\
org.springframework.cloud.sleuth.instrument.messaging.SleuthKafkaStreamsConfiguration,\
org.springframework.cloud.sleuth.instrument.messaging.TraceMessagingAutoConfiguration,\
org.springframework.cloud.sleuth.instrument.messaging.TraceSpringIntegrationAutoConfiguration,\
org.springframework.cloud.sleuth.instrument.messaging.TraceSpringMessagingAutoConfiguration,\
org.springframework.cloud.sleuth.instrument.messaging.websocket.TraceWebSocketAutoConfiguration,\
org.springframework.cloud.sleuth.instrument.opentracing.OpentracingAutoConfiguration,\
org.springframework.cloud.sleuth.instrument.redis.TraceRedisAutoConfiguration,\
org.springframework.cloud.sleuth.instrument.quartz.TraceQuartzAutoConfiguration
# Environment Post Processor
org.springframework.boot.env.EnvironmentPostProcessor=\
org.springframework.cloud.sleuth.autoconfig.TraceEnvironmentPostProcessor

135
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/ITTracingMethodMessageHandlerAdapterTests.java Normal file
View File

@@ -0,0 +1,135 @@
/*
* 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
*
* https://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.cloud.sleuth.instrument.messaging;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import brave.Span;
import brave.Tracing;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.aws.messaging.listener.QueueMessageHandler;
import org.springframework.cloud.aws.messaging.listener.annotation.SqsListener;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.context.junit4.SpringRunner;
import static org.assertj.core.api.Assertions.assertThat;
import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.NONE;
@SpringBootTest(
classes = ITTracingMethodMessageHandlerAdapterTests.TestingConfiguration.class,
webEnvironment = NONE)
@RunWith(SpringRunner.class)
public class ITTracingMethodMessageHandlerAdapterTests {
private static final String TRACE_ID = "12345678123456781234567812345678";
private static final String SPAN_ID = "1234567812345678";
@Autowired
ApplicationContext applicationContext;
@Autowired
SqsQueueMessageHandlerFactory messageHandlerFactory;
@Autowired
TestingMessageHandler testingMessageHandler;
@Autowired
Tracing tracing;
private QueueMessageHandler messageHandler;
@Before
public void setup() {
messageHandler = messageHandlerFactory.createQueueMessageHandler();
messageHandler.setApplicationContext(applicationContext);
messageHandler.afterPropertiesSet();
}
@Test
public void aSpanGetsPutIntoScopeWithoutHeadersOnTheMessage() {
AtomicReference<Span> probedSpan = new AtomicReference<>();
testingMessageHandler.withTestProbe(((headers, s) -> {
probedSpan.set(tracing.tracer().currentSpan());
}));
messageHandler.handleMessage(new GenericMessage<>("message",
Collections.singletonMap("LogicalResourceId", "test")));
assertThat(probedSpan.get()).isNotNull();
}
@Test
public void theSpanThatIsInTheHeadersIsUsedForTheTraceScope() {
AtomicReference<Span> probedSpan = new AtomicReference<>();
testingMessageHandler.withTestProbe(((headers, s) -> {
probedSpan.set(tracing.tracer().currentSpan());
}));
Map<String, Object> headers = new HashMap<>();
headers.put("LogicalResourceId", "test");
headers.put("X-B3-TraceId", TRACE_ID);
headers.put("X-B3-SpanId", SPAN_ID);
headers.put("X-B3-Sampled", "1");
messageHandler.handleMessage(new GenericMessage<>("message", headers));
assertThat(probedSpan.get()).isNotNull();
assertThat(probedSpan.get().context().traceIdString()).isEqualTo(TRACE_ID);
assertThat(probedSpan.get().context().sampled()).isTrue();
}
@EnableAutoConfiguration
@Configuration
static class TestingConfiguration {
@Bean
TestingMessageHandler testingMessageHandler() {
return new TestingMessageHandler();
}
}
static class TestingMessageHandler {
private BiConsumer<MessageHeaders, String> testProbe;
void withTestProbe(BiConsumer<MessageHeaders, String> consumer) {
this.testProbe = consumer;
}
@SqsListener("test")
public void handle(MessageHeaders header, String payload) {
testProbe.accept(header, payload);
}
}
}

73
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/SqsQueueMessageHandlerTests.java Normal file
View File

@@ -0,0 +1,73 @@
/*
* 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
*
* https://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.cloud.sleuth.instrument.messaging;
import java.util.Collections;
import java.util.function.BiConsumer;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnit;
import org.mockito.junit.MockitoRule;
import org.mockito.quality.Strictness;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.GenericMessage;
import static org.assertj.core.api.Assertions.assertThat;
public class SqsQueueMessageHandlerTests {
@Rule
public MockitoRule rule = MockitoJUnit.rule().strictness(Strictness.STRICT_STUBS);
@Mock
TracingMethodMessageHandlerAdapter adapter;
SqsQueueMessageHandler subject;
@Before
public void setup() {
subject = new SqsQueueMessageHandler(adapter, Collections.emptyList());
}
@Test
public void sqsQueueMessageHandlerDelegatesToAdapter() {
ArgumentCaptor<Message> messageCapture = ArgumentCaptor.forClass(Message.class);
ArgumentCaptor<MessageHandler> handlerCapture = ArgumentCaptor
.forClass(MessageHandler.class);
ArgumentCaptor<BiConsumer> spanTaggerCapture = ArgumentCaptor
.forClass(BiConsumer.class);
Mockito.doNothing().when(adapter).wrapMethodMessageHandler(
messageCapture.capture(), handlerCapture.capture(),
spanTaggerCapture.capture());
subject.handleMessage(new GenericMessage<>("a"));
Mockito.verify(adapter, Mockito.times(1)).wrapMethodMessageHandler(Mockito.any(),
Mockito.any(), Mockito.any());
assertThat(messageCapture.getValue().getPayload().toString()).isEqualTo("a");
assertThat(handlerCapture.getValue()).isNotNull();
assertThat(spanTaggerCapture.getValue()).isNotNull();
}
}