diff --git a/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/web/client/WebClientTests.java b/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/web/client/WebClientTests.java index 0e9cf4255..eef976520 100644 --- a/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/web/client/WebClientTests.java +++ b/spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/web/client/WebClientTests.java @@ -1,203 +1,205 @@ -/* - * Copyright 2013-2021 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.autoconfig.brave.instrument.web.client; - -import java.util.HashMap; -import java.util.Map; -import java.util.concurrent.Future; -import java.util.stream.Collectors; - -import brave.Span; -import brave.Tracer; -import brave.baggage.BaggagePropagation; -import brave.handler.SpanHandler; -import brave.propagation.B3Propagation; -import brave.sampler.Sampler; -import brave.test.TestSpanHandler; -import org.apache.http.HttpResponse; -import org.apache.http.client.methods.HttpGet; -import org.apache.http.concurrent.FutureCallback; -import org.apache.http.impl.client.BasicResponseHandler; -import org.apache.http.impl.client.HttpClientBuilder; -import org.apache.http.impl.nio.client.CloseableHttpAsyncClient; -import org.apache.http.impl.nio.client.HttpAsyncClientBuilder; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.autoconfigure.data.r2dbc.R2dbcDataAutoConfiguration; -import org.springframework.boot.autoconfigure.r2dbc.R2dbcAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.boot.web.server.LocalServerPort; -import org.springframework.cloud.gateway.config.GatewayAutoConfiguration; -import org.springframework.cloud.gateway.config.GatewayClassPathWarningAutoConfiguration; -import org.springframework.cloud.sleuth.DisableSecurity; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.http.HttpHeaders; -import org.springframework.test.annotation.DirtiesContext; -import org.springframework.web.bind.annotation.RequestHeader; -import org.springframework.web.bind.annotation.RequestMapping; -import org.springframework.web.bind.annotation.RestController; - -import static brave.Span.Kind.CLIENT; -import static brave.propagation.B3Propagation.Format.SINGLE_NO_PARENT; -import static org.assertj.core.api.BDDAssertions.then; - -@SpringBootTest(classes = WebClientTests.TestConfiguration.class, - webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, - properties = { "spring.sleuth.web.servlet.enabled=false", "spring.application.name=fooservice", - "spring.sleuth.web.client.skip-pattern=/skip.*" }) -@DirtiesContext -public class WebClientTests { - - @Autowired - HttpClientBuilder httpClientBuilder; // #845 - - @Autowired - HttpAsyncClientBuilder httpAsyncClientBuilder; // #845 - - @Autowired - TestSpanHandler spans; - - @Autowired - Tracer tracer; - - @LocalServerPort - int port; - - @Autowired - FooController fooController; - - @AfterEach - @BeforeEach - public void close() { - this.spans.clear(); - this.fooController.clear(); - } - - @Test - @SuppressWarnings("unchecked") - public void shouldAttachTraceIdWhenCallingAnotherServiceForHttpClient() throws Exception { - then(this.spans).isEmpty(); - Span span = this.tracer.nextSpan().name("foo").start(); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - String response = this.httpClientBuilder.build().execute(new HttpGet("http://localhost:" + this.port), - new BasicResponseHandler()); - - then(response).isNotEmpty(); - } - - then(this.tracer.currentSpan()).isNull(); - then(this.spans).isNotEmpty().extracting("traceId", String.class).containsOnly(span.context().traceIdString()); - then(this.spans.spans().stream().map(s -> s.kind().name()).collect(Collectors.toList())).contains("CLIENT"); - } - - @Test - @SuppressWarnings("unchecked") - public void shouldAttachTraceIdWhenCallingAnotherServiceForAsyncHttpClient() throws Exception { - Span span = this.tracer.nextSpan().name("foo").start(); - - CloseableHttpAsyncClient client = this.httpAsyncClientBuilder.build(); - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - client.start(); - Future future = client.execute(new HttpGet("http://localhost:" + this.port), - new FutureCallback() { - @Override - public void completed(HttpResponse result) { - - } - - @Override - public void failed(Exception ex) { - - } - - @Override - public void cancelled() { - - } - }); - then(future.get()).isNotNull(); - } - finally { - client.close(); - } - - then(this.tracer.currentSpan()).isNull(); - then(this.spans).isNotEmpty().extracting("traceId", String.class).containsOnly(span.context().traceIdString()); - then(this.spans.spans().stream().map(s -> s.kind().name()).collect(Collectors.toList())).contains("CLIENT"); - } - - @Configuration(proxyBeanMethods = false) - @EnableAutoConfiguration(exclude = { GatewayClassPathWarningAutoConfiguration.class, GatewayAutoConfiguration.class, - R2dbcAutoConfiguration.class, R2dbcDataAutoConfiguration.class }) - @DisableSecurity - public static class TestConfiguration { - - @Bean - BaggagePropagation.FactoryBuilder baggagePropagationFactoryBuilder() { - // Use b3 single format as it is less verbose - return BaggagePropagation.newFactoryBuilder( - B3Propagation.newFactoryBuilder().injectFormat(CLIENT, SINGLE_NO_PARENT).build()); - } - - @Bean - FooController fooController() { - return new FooController(); - } - - @Bean - Sampler testSampler() { - return Sampler.ALWAYS_SAMPLE; - } - - @Bean - SpanHandler testSpanHandler() { - return new TestSpanHandler(); - } - - } - - @RestController - public static class FooController { - - Span span; - - @RequestMapping("/") - public Map home(@RequestHeader HttpHeaders headers) { - Map map = new HashMap<>(); - for (String key : headers.keySet()) { - map.put(key, headers.getFirst(key)); - } - return map; - } - - public Span getSpan() { - return this.span; - } - - public void clear() { - this.span = null; - } - - } - -} +/* + * Copyright 2013-2021 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.autoconfig.brave.instrument.web.client; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.Future; +import java.util.stream.Collectors; + +import brave.Span; +import brave.Tracer; +import brave.baggage.BaggagePropagation; +import brave.handler.SpanHandler; +import brave.propagation.B3Propagation; +import brave.sampler.Sampler; +import brave.test.TestSpanHandler; +import org.apache.http.HttpResponse; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.concurrent.FutureCallback; +import org.apache.http.impl.client.BasicResponseHandler; +import org.apache.http.impl.client.HttpClientBuilder; +import org.apache.http.impl.nio.client.CloseableHttpAsyncClient; +import org.apache.http.impl.nio.client.HttpAsyncClientBuilder; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.autoconfigure.data.mongo.MongoDataAutoConfiguration; +import org.springframework.boot.autoconfigure.data.r2dbc.R2dbcDataAutoConfiguration; +import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration; +import org.springframework.boot.autoconfigure.r2dbc.R2dbcAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.web.server.LocalServerPort; +import org.springframework.cloud.gateway.config.GatewayAutoConfiguration; +import org.springframework.cloud.gateway.config.GatewayClassPathWarningAutoConfiguration; +import org.springframework.cloud.sleuth.DisableSecurity; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.http.HttpHeaders; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.web.bind.annotation.RequestHeader; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +import static brave.Span.Kind.CLIENT; +import static brave.propagation.B3Propagation.Format.SINGLE_NO_PARENT; +import static org.assertj.core.api.BDDAssertions.then; + +@SpringBootTest(classes = WebClientTests.TestConfiguration.class, + webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, + properties = {"spring.sleuth.web.servlet.enabled=false", "spring.application.name=fooservice", + "spring.sleuth.web.client.skip-pattern=/skip.*"}) +@DirtiesContext +public class WebClientTests { + + @Autowired + HttpClientBuilder httpClientBuilder; // #845 + + @Autowired + HttpAsyncClientBuilder httpAsyncClientBuilder; // #845 + + @Autowired + TestSpanHandler spans; + + @Autowired + Tracer tracer; + + @LocalServerPort + int port; + + @Autowired + FooController fooController; + + @AfterEach + @BeforeEach + public void close() { + this.spans.clear(); + this.fooController.clear(); + } + + @Test + @SuppressWarnings("unchecked") + public void shouldAttachTraceIdWhenCallingAnotherServiceForHttpClient() throws Exception { + then(this.spans).isEmpty(); + Span span = this.tracer.nextSpan().name("foo").start(); + + try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { + String response = this.httpClientBuilder.build().execute(new HttpGet("http://localhost:" + this.port), + new BasicResponseHandler()); + + then(response).isNotEmpty(); + } + + then(this.tracer.currentSpan()).isNull(); + then(this.spans).isNotEmpty().extracting("traceId", String.class).containsOnly(span.context().traceIdString()); + then(this.spans.spans().stream().map(s -> s.kind().name()).collect(Collectors.toList())).contains("CLIENT"); + } + + @Test + @SuppressWarnings("unchecked") + public void shouldAttachTraceIdWhenCallingAnotherServiceForAsyncHttpClient() throws Exception { + Span span = this.tracer.nextSpan().name("foo").start(); + + CloseableHttpAsyncClient client = this.httpAsyncClientBuilder.build(); + try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { + client.start(); + Future future = client.execute(new HttpGet("http://localhost:" + this.port), + new FutureCallback() { + @Override + public void completed(HttpResponse result) { + + } + + @Override + public void failed(Exception ex) { + + } + + @Override + public void cancelled() { + + } + }); + then(future.get()).isNotNull(); + } + finally { + client.close(); + } + + then(this.tracer.currentSpan()).isNull(); + then(this.spans).isNotEmpty().extracting("traceId", String.class).containsOnly(span.context().traceIdString()); + then(this.spans.spans().stream().map(s -> s.kind().name()).collect(Collectors.toList())).contains("CLIENT"); + } + + @Configuration(proxyBeanMethods = false) + @EnableAutoConfiguration(exclude = {GatewayClassPathWarningAutoConfiguration.class, GatewayAutoConfiguration.class, + R2dbcAutoConfiguration.class, R2dbcDataAutoConfiguration.class, MongoAutoConfiguration.class, MongoDataAutoConfiguration.class}) + @DisableSecurity + public static class TestConfiguration { + + @Bean + BaggagePropagation.FactoryBuilder baggagePropagationFactoryBuilder() { + // Use b3 single format as it is less verbose + return BaggagePropagation.newFactoryBuilder( + B3Propagation.newFactoryBuilder().injectFormat(CLIENT, SINGLE_NO_PARENT).build()); + } + + @Bean + FooController fooController() { + return new FooController(); + } + + @Bean + Sampler testSampler() { + return Sampler.ALWAYS_SAMPLE; + } + + @Bean + SpanHandler testSpanHandler() { + return new TestSpanHandler(); + } + + } + + @RestController + public static class FooController { + + Span span; + + @RequestMapping("/") + public Map home(@RequestHeader HttpHeaders headers) { + Map map = new HashMap<>(); + for (String key : headers.keySet()) { + map.put(key, headers.getFirst(key)); + } + return map; + } + + public Span getSpan() { + return this.span; + } + + public void clear() { + this.span = null; + } + + } + +} diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java index 786e7d0d0..25ed6e3b9 100644 --- a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java +++ b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java @@ -16,31 +16,24 @@ package org.springframework.cloud.sleuth.instrument.messaging; -import java.util.Iterator; -import java.util.Map; +import java.util.NoSuchElementException; +import java.util.Set; +import java.util.concurrent.LinkedBlockingDeque; import java.util.function.Function; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - import org.springframework.aop.support.AopUtils; import org.springframework.beans.BeansException; import org.springframework.cloud.sleuth.Span; -import org.springframework.cloud.sleuth.ThreadLocalSpan; import org.springframework.cloud.sleuth.Tracer; -import org.springframework.cloud.sleuth.WithThreadLocalSpan; import org.springframework.cloud.sleuth.propagation.Propagator; -import org.springframework.cloud.stream.binder.BinderType; -import org.springframework.cloud.stream.binder.BinderTypeRegistry; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; -import org.springframework.integration.channel.DirectChannel; +import org.springframework.core.log.LogAccessor; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; -import org.springframework.messaging.support.ChannelInterceptorAdapter; import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.ExecutorChannelInterceptor; import org.springframework.messaging.support.GenericMessage; @@ -55,18 +48,17 @@ import org.springframework.util.StringUtils; * a handler later calls {@link MessageHandler#handleMessage(Message)}. * * @author Marcin Grzejszczak + * @author Artem Bilan * @since 3.0.0 */ -public final class TracingChannelInterceptor extends ChannelInterceptorAdapter - implements ExecutorChannelInterceptor, ApplicationContextAware, WithThreadLocalSpan { +public final class TracingChannelInterceptor implements ExecutorChannelInterceptor, ApplicationContextAware { /** * Name of the class in Spring Cloud Stream that is a direct channel. */ - public static final String STREAM_DIRECT_CHANNEL = "org.springframework." - + "cloud.stream.messaging.DirectWithAttributesChannel"; + public static final String STREAM_DIRECT_CHANNEL = "org.springframework.cloud.stream.messaging.DirectWithAttributesChannel"; - private static final Log log = LogFactory.getLog(TracingChannelInterceptor.class); + private static final LogAccessor log = new LogAccessor(TracingChannelInterceptor.class); /** * Using the literal "broker" until we come up with a better solution. @@ -88,45 +80,47 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter */ private static final String REMOTE_SERVICE_NAME = "broker"; - final Tracer tracer; + private static final boolean hasDirectChannelClass = ClassUtils + .isPresent("org.springframework.integration.channel.DirectChannel", null); - final Propagator.Setter injector; - - final Propagator.Getter extractor; - - final MessageSpanCustomizer messageSpanCustomizer; - - private final boolean hasDirectChannelClass; - - private final boolean hasBinderTypeRegistry; + private static final boolean hasBinderTypeRegistry = ClassUtils + .isPresent("org.springframework.cloud.stream.binder.BinderTypeRegistry", null); // special case of a Stream - private final Class directWithAttributesChannelClass; + private static final Class directWithAttributesChannelClass = ClassUtils.isPresent(STREAM_DIRECT_CHANNEL, null) + ? ClassUtils.resolveClassName(STREAM_DIRECT_CHANNEL, null) : null; - private ApplicationContext applicationContext; + private final ThreadLocalSpan threadLocalSpan = new ThreadLocalSpan(); + + private final Tracer tracer; + + private final Propagator.Setter injector; + + private final Propagator.Getter extractor; + + private final MessageSpanCustomizer messageSpanCustomizer; private final Propagator propagator; - private final ThreadLocalSpan threadLocalSpan; - private final Function remoteServiceNameMapper; + private ApplicationContext applicationContext; + public TracingChannelInterceptor(Tracer tracer, Propagator propagator, Propagator.Setter setter, Propagator.Getter getter, Function remoteServiceNameMapper, MessageSpanCustomizer messageSpanCustomizer) { + this.tracer = tracer; - this.threadLocalSpan = new ThreadLocalSpan(tracer); this.propagator = propagator; this.injector = setter; this.extractor = getter; this.remoteServiceNameMapper = remoteServiceNameMapper; this.messageSpanCustomizer = messageSpanCustomizer; - this.hasDirectChannelClass = ClassUtils.isPresent("org.springframework.integration.channel.DirectChannel", - null); - this.hasBinderTypeRegistry = ClassUtils.isPresent("org.springframework.cloud.stream.binder.BinderTypeRegistry", - null); - this.directWithAttributesChannelClass = ClassUtils.isPresent(STREAM_DIRECT_CHANNEL, null) - ? ClassUtils.resolveClassName(STREAM_DIRECT_CHANNEL, null) : null; + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = applicationContext; } /** @@ -134,28 +128,19 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter */ @Override public Message preSend(Message message, MessageChannel channel) { - if (emptyMessage(message)) { - return message; - } Message retrievedMessage = getMessage(message); - if (log.isDebugEnabled()) { - log.debug("Received a message in pre-send " + retrievedMessage); - } + log.debug(() -> "Received a message in pre-send " + retrievedMessage); MessageHeaderAccessor headers = mutableHeaderAccessor(retrievedMessage); Span.Builder spanBuilder = this.propagator.extract(headers, this.extractor); MessageHeaderPropagatorSetter.removeAnyTraceHeaders(headers, this.propagator.fields()); spanBuilder = spanBuilder.kind(Span.Kind.PRODUCER); spanBuilder = this.messageSpanCustomizer.customizeSend(spanBuilder, message, channel) - .remoteServiceName(toRemoteServiceName(headers)); + .remoteServiceName(toRemoteServiceName(headers, remoteServiceNameMapper, applicationContext)); Span span = spanBuilder.start(); - if (log.isDebugEnabled()) { - log.debug("Extracted result from headers " + span); - } + log.debug(() -> "Extracted result from headers " + span); setSpanInScope(span); this.propagator.inject(span.context(), headers, this.injector); - if (log.isDebugEnabled()) { - log.debug("Created a new span in pre send " + span); - } + log.debug(() -> "Created a new span in pre send " + span); Message outputMessage = outputMessage(message, retrievedMessage, headers); if (isDirectChannel(channel)) { beforeHandle(outputMessage, channel, null); @@ -163,19 +148,28 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter return outputMessage; } - private String toRemoteServiceName(MessageHeaderAccessor headers) { + private void setSpanInScope(Span span) { + Tracer.SpanInScope spanInScope = this.tracer.withSpan(span); + this.threadLocalSpan.set(new SpanAndScope(span, spanInScope)); + log.debug(() -> "Put span in scope " + span); + } + + private static String toRemoteServiceName(MessageHeaderAccessor headers, + Function remoteServiceNameMapper, ApplicationContext applicationContext) { + for (String key : headers.getMessageHeaders().keySet()) { - String remoteServiceName = this.remoteServiceNameMapper.apply(key); + String remoteServiceName = remoteServiceNameMapper.apply(key); if (StringUtils.hasText(remoteServiceName)) { return remoteServiceName; } } - if (this.hasBinderTypeRegistry && this.applicationContext != null) { - BinderTypeRegistry typeRegistry = this.applicationContext.getBean(BinderTypeRegistry.class); - Iterator> iterator = typeRegistry.getAll().entrySet().iterator(); - if (iterator.hasNext()) { - String binderName = iterator.next().getKey(); - String remoteServiceName = this.remoteServiceNameMapper.apply(binderName); + + if (hasBinderTypeRegistry && applicationContext != null) { + org.springframework.cloud.stream.binder.BinderTypeRegistry typeRegistry = applicationContext + .getBean(org.springframework.cloud.stream.binder.BinderTypeRegistry.class); + Set binderNames = typeRegistry.getAll().keySet(); + for (String binderName : binderNames) { + String remoteServiceName = remoteServiceNameMapper.apply(binderName); if (StringUtils.hasText(remoteServiceName)) { return remoteServiceName; } @@ -194,43 +188,29 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter return new ErrorMessage(errorMessage.getPayload(), isWebSockets(headers) ? headers.getMessageHeaders() : new MessageHeaders(headers.getMessageHeaders()), errorMessage.getOriginalMessage()); } - headers.copyHeaders(new MessageHeaders(additionalHeaders.getMessageHeaders())); + headers.copyHeaders(additionalHeaders.getMessageHeaders()); return new GenericMessage<>(retrievedMessage.getPayload(), isWebSockets(headers) ? headers.getMessageHeaders() : new MessageHeaders(headers.getMessageHeaders())); } - private boolean isWebSockets(MessageHeaderAccessor headerAccessor) { + private static boolean isWebSockets(MessageHeaderAccessor headerAccessor) { return headerAccessor.getMessageHeaders().containsKey("stompCommand") || headerAccessor.getMessageHeaders().containsKey("simpMessageType"); } - private boolean isDirectChannel(MessageChannel channel) { + private static boolean isDirectChannel(MessageChannel channel) { Class targetClass = AopUtils.getTargetClass(channel); - boolean directChannel = this.hasDirectChannelClass && DirectChannel.class.isAssignableFrom(targetClass); - if (!directChannel) { - return false; - } - if (this.directWithAttributesChannelClass == null) { - return true; - } - return !isStreamSpecialDirectChannel(targetClass); - } - - private boolean isStreamSpecialDirectChannel(Class targetClass) { - return this.directWithAttributesChannelClass.isAssignableFrom(targetClass); + return (directWithAttributesChannelClass == null + || !directWithAttributesChannelClass.isAssignableFrom(targetClass)) && hasDirectChannelClass + && org.springframework.integration.channel.DirectChannel.class.isAssignableFrom(targetClass); } @Override public void afterSendCompletion(Message message, MessageChannel channel, boolean sent, Exception ex) { - if (emptyMessage(message)) { - return; - } if (isDirectChannel(channel)) { afterMessageHandled(message, channel, null, ex); } - if (log.isDebugEnabled()) { - log.debug("Will finish the current span after completion " + this.tracer.currentSpan()); - } + log.debug(() -> "Will finish the current span after completion " + this.tracer.currentSpan()); finishSpan(ex); } @@ -240,26 +220,15 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter */ @Override public Message postReceive(Message message, MessageChannel channel) { - if (emptyMessage(message)) { - return message; - } MessageHeaderAccessor headers = mutableHeaderAccessor(message); - if (log.isDebugEnabled()) { - log.debug("Received a message in post-receive " + message); - } + log.debug(() -> "Received a message in post-receive " + message); Span result = this.propagator.extract(headers, this.extractor).start(); - if (log.isDebugEnabled()) { - log.debug("Extracted result from headers " + result); - } + log.debug(() -> "Extracted result from headers " + result); Span span = consumerSpanReceive(message, channel, headers, result); setSpanInScope(span); - if (log.isDebugEnabled()) { - log.debug("Created a new span that will be injected in the headers " + span); - } + log.debug(() -> "Created a new span that will be injected in the headers " + span); this.propagator.inject(span.context(), headers, this.injector); - if (log.isDebugEnabled()) { - log.debug("Created a new span in post receive " + span); - } + log.debug(() -> "Created a new span in post receive " + span); headers.setImmutable(); if (message instanceof ErrorMessage) { ErrorMessage errorMessage = (ErrorMessage) message; @@ -275,18 +244,13 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter MessageHeaderPropagatorSetter.removeAnyTraceHeaders(headers, this.propagator.fields()); builder = builder.kind(Span.Kind.CONSUMER); builder = this.messageSpanCustomizer.customizeReceive(builder, message, channel); - builder = builder.remoteServiceName(toRemoteServiceName(headers)); + builder = builder.remoteServiceName(toRemoteServiceName(headers, remoteServiceNameMapper, applicationContext)); return builder.start(); } @Override public void afterReceiveCompletion(Message message, MessageChannel channel, Exception ex) { - if (emptyMessage(message)) { - return; - } - if (log.isDebugEnabled()) { - log.debug("Will finish the current span after receive completion " + this.tracer.currentSpan()); - } + log.debug(() -> "Will finish the current span after receive completion " + this.tracer.currentSpan()); finishSpan(ex); } @@ -296,16 +260,11 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter */ @Override public Message beforeHandle(Message message, MessageChannel channel, MessageHandler handler) { - if (emptyMessage(message)) { - return message; - } MessageHeaderAccessor headers = mutableHeaderAccessor(message); - if (log.isDebugEnabled()) { - log.debug("Received a message in before handle " + message); - } + log.debug(() -> "Received a message in before handle " + message); Span consumerSpan = consumerSpan(message, channel, headers); // create and scope a span for the message processor - Span handle = SleuthMessagingSpan.MESSAGING_SPAN.wrap(this.tracer.nextSpan(consumerSpan)); + Span handle = this.tracer.nextSpan(consumerSpan); handle = this.messageSpanCustomizer.customizeHandle(handle, message, channel).start(); if (log.isDebugEnabled()) { log.debug("Created consumer span " + handle); @@ -341,21 +300,42 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter @Override public void afterMessageHandled(Message message, MessageChannel channel, MessageHandler handler, Exception ex) { - if (emptyMessage(message)) { - return; - } - if (log.isDebugEnabled()) { - log.debug("Will finish the current span after message handled " + this.tracer.currentSpan()); - } + log.debug(() -> "Will finish the current span after message handled " + this.tracer.currentSpan()); finishSpan(ex); } - @Override - public ThreadLocalSpan getThreadLocalSpan() { - return this.threadLocalSpan; + void finishSpan(Exception error) { + SpanAndScope spanAndScope = getSpanFromThreadLocal(); + if (spanAndScope == null) { + return; + } + Span span = spanAndScope.span; + Tracer.SpanInScope scope = spanAndScope.scope; + if (span.isNoop()) { + log.debug(() -> "Span " + span + " is noop - will stop the scope"); + scope.close(); + return; + } + if (error != null) { // an error occurred, adding error to span + String message = error.getMessage(); + if (message == null) { + message = error.getClass().getSimpleName(); + } + span.tag("error", message); + } + log.debug(() -> "Will finish the and its corresponding scope " + span); + span.end(); + scope.close(); } - private MessageHeaderAccessor mutableHeaderAccessor(Message message) { + private SpanAndScope getSpanFromThreadLocal() { + SpanAndScope span = this.threadLocalSpan.get(); + log.debug(() -> "Took span [" + span + "] from thread local"); + this.threadLocalSpan.remove(); + return span; + } + + private static MessageHeaderAccessor mutableHeaderAccessor(Message message) { MessageHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, MessageHeaderAccessor.class); if (accessor != null && accessor.isMutable()) { return accessor; @@ -365,7 +345,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter return headers; } - private Message getMessage(Message message) { + private static Message getMessage(Message message) { Object payload = message.getPayload(); if (payload instanceof MessagingException) { MessagingException e = (MessagingException) payload; @@ -375,13 +355,57 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter return message; } - private boolean emptyMessage(Message message) { - return message == null; + private static class SpanAndScope { + + final Span span; + + final Tracer.SpanInScope scope; + + SpanAndScope(Span span, Tracer.SpanInScope scope) { + this.span = span; + this.scope = scope; + } + } - @Override - public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { - this.applicationContext = applicationContext; + private static class ThreadLocalSpan { + + private static final LogAccessor log = new LogAccessor(ThreadLocalSpan.class); + + private final ThreadLocal threadLocalSpan = new ThreadLocal<>(); + + private final LinkedBlockingDeque spans = new LinkedBlockingDeque<>(); + + ThreadLocalSpan() { + } + + void set(SpanAndScope spanAndScope) { + SpanAndScope scope = this.threadLocalSpan.get(); + if (scope != null) { + this.spans.addFirst(scope); + } + this.threadLocalSpan.set(spanAndScope); + } + + SpanAndScope get() { + return this.threadLocalSpan.get(); + } + + void remove() { + this.threadLocalSpan.remove(); + if (this.spans.isEmpty()) { + return; + } + try { + SpanAndScope span = this.spans.removeFirst(); + log.debug(() -> "Took span [" + span + "] from thread local"); + this.threadLocalSpan.set(span); + } + catch (NoSuchElementException ex) { + log.trace(ex, () -> "Failed to remove a span from the queue"); + } + } + } } diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaProducerTest.java b/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaProducerTest.java index 79b25ed2b..1692beac8 100644 --- a/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaProducerTest.java +++ b/tests/brave/spring-cloud-sleuth-instrumentation-kafka-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/kafka/KafkaProducerTest.java @@ -1,62 +1,66 @@ -/* - * Copyright 2013-2021 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.brave.instrument.kafka; - -import java.util.Optional; -import java.util.concurrent.TimeUnit; - -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.header.Header; -import org.assertj.core.api.BDDAssertions; -import org.junit.jupiter.api.Test; - -import org.springframework.cloud.sleuth.brave.BraveTestTracing; -import org.springframework.cloud.sleuth.test.TestTracingAware; - -public class KafkaProducerTest extends org.springframework.cloud.sleuth.instrument.kafka.KafkaProducerTest { - - BraveTestTracing testTracing; - - @Override - public TestTracingAware tracerTest() { - if (this.testTracing == null) { - this.testTracing = new BraveTestTracing(); - } - return this.testTracing; - } - - @Test - public void should_inject_native_headers() throws InterruptedException { - ProducerRecord producerRecord = new ProducerRecord<>(testTopic, "test", "test"); - startKafkaConsumer(); - - this.kafkaProducer.send(producerRecord); - ConsumerRecord consumerRecord = consumerRecords.poll(15, TimeUnit.SECONDS); - - BDDAssertions.then(consumerRecord).isNotNull(); - BDDAssertions.then(getHeaderValueOrNull(consumerRecord, "X-B3-TraceId")).isNotNull(); - BDDAssertions.then(getHeaderValueOrNull(consumerRecord, "X-B3-SpanId")).isNotNull(); - BDDAssertions.then(getHeaderValueOrNull(consumerRecord, "X-B3-Sampled")).isNotNull(); - } - - private static String getHeaderValueOrNull(ConsumerRecord consumerRecord, String header) { - return Optional.ofNullable(consumerRecord).map(ConsumerRecord::headers) - .map(headers -> headers.lastHeader(header)).map(Header::value).map(String::new).orElse(null); - } - -} +/* + * Copyright 2013-2021 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.brave.instrument.kafka; + +import java.util.Optional; +import java.util.concurrent.TimeUnit; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.header.Header; +import org.assertj.core.api.BDDAssertions; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.Test; + +import org.springframework.cloud.sleuth.brave.BraveTestTracing; +import org.springframework.cloud.sleuth.test.TestTracingAware; + +public class KafkaProducerTest extends org.springframework.cloud.sleuth.instrument.kafka.KafkaProducerTest { + + BraveTestTracing testTracing; + + @Override + public TestTracingAware tracerTest() { + if (this.testTracing == null) { + this.testTracing = new BraveTestTracing(); + } + return this.testTracing; + } + + @Test + public void should_inject_native_headers() throws InterruptedException { + ProducerRecord producerRecord = new ProducerRecord<>(testTopic, "test", "test"); + startKafkaConsumer(); + + this.kafkaProducer.send(producerRecord); + + Awaitility.await().atMost(1, TimeUnit.MINUTES).pollInterval(1, TimeUnit.SECONDS).untilAsserted(() -> { + ConsumerRecord consumerRecord = consumerRecords.poll(15, TimeUnit.SECONDS); + + BDDAssertions.then(consumerRecord).isNotNull(); + BDDAssertions.then(getHeaderValueOrNull(consumerRecord, "X-B3-TraceId")).isNotNull(); + BDDAssertions.then(getHeaderValueOrNull(consumerRecord, "X-B3-SpanId")).isNotNull(); + BDDAssertions.then(getHeaderValueOrNull(consumerRecord, "X-B3-Sampled")).isNotNull(); + }); + } + + private static String getHeaderValueOrNull(ConsumerRecord consumerRecord, String header) { + return Optional.ofNullable(consumerRecord).map(ConsumerRecord::headers) + .map(headers -> headers.lastHeader(header)).map(Header::value).map(String::new).orElse(null); + } + +} diff --git a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/r2dbc/R2dbcIntegrationTests.java b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/r2dbc/R2dbcIntegrationTests.java index c44d5a086..7844ae8ba 100644 --- a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/r2dbc/R2dbcIntegrationTests.java +++ b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/r2dbc/R2dbcIntegrationTests.java @@ -1,100 +1,102 @@ -/* - * Copyright 2013-2021 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.r2dbc; - -import java.time.Duration; -import java.util.List; -import java.util.Set; -import java.util.stream.Collectors; - -import io.r2dbc.spi.ConnectionFactory; -import org.junit.jupiter.api.Test; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.CommandLineRunner; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.cloud.sleuth.exporter.FinishedSpan; -import org.springframework.cloud.sleuth.test.TestSpanHandler; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.ComponentScan; -import org.springframework.context.annotation.Configuration; -import org.springframework.core.io.ClassPathResource; -import org.springframework.dao.DataAccessException; -import org.springframework.r2dbc.connection.init.ConnectionFactoryInitializer; -import org.springframework.r2dbc.connection.init.ResourceDatabasePopulator; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.TestPropertySource; - -import static org.assertj.core.api.BDDAssertions.then; - -@ContextConfiguration(classes = R2dbcIntegrationTests.TestConfig.class) -@TestPropertySource(properties = "spring.application.name=MyApplication") -public abstract class R2dbcIntegrationTests { - - @Autowired - TestSpanHandler spans; - - @Test - public void should_pass_tracing_information_when_using_r2dbc() { - Set traceIds = this.spans.reportedSpans().stream().map(FinishedSpan::getTraceId) - .collect(Collectors.toSet()); - then(traceIds).as("There's one traceid").hasSize(1); - Set spanIds = this.spans.reportedSpans().stream().map(FinishedSpan::getSpanId) - .collect(Collectors.toSet()); - - // 2 transactions - 9 database interactions - then(spanIds).as("There are 11 spans").hasSize(11); - List spanNames = this.spans.reportedSpans().stream().map(FinishedSpan::getName) - .collect(Collectors.toList()); - List remoteServiceNames = this.spans.reportedSpans().stream().map(FinishedSpan::getRemoteServiceName) - .collect(Collectors.toList()); - then(spanNames.stream().filter("tx"::equalsIgnoreCase).collect(Collectors.toList())).hasSize(2); - then(remoteServiceNames.stream().filter("h2"::equalsIgnoreCase).collect(Collectors.toList())).hasSize(9); - } - - @Configuration(proxyBeanMethods = false) - @EnableAutoConfiguration - @ComponentScan - public static class TestConfig { - - private static final Logger log = LoggerFactory.getLogger(TestConfig.class); - - @Bean - public CommandLineRunner demo(ReactiveNewTransactionService reactiveNewTransactionService) { - return (args) -> { - try { - reactiveNewTransactionService.newTransaction().block(Duration.ofSeconds(50)); - } - catch (DataAccessException e) { - log.info("Expected to throw an exception so that we see if rollback works", e); - } - }; - } - - @Bean - ConnectionFactoryInitializer initializer(ConnectionFactory connectionFactory) { - ConnectionFactoryInitializer initializer = new ConnectionFactoryInitializer(); - initializer.setConnectionFactory(connectionFactory); - initializer.setDatabasePopulator(new ResourceDatabasePopulator(new ClassPathResource("schema.sql"))); - return initializer; - } - - } - -} +/* + * Copyright 2013-2021 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.r2dbc; + +import java.time.Duration; +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; + +import io.r2dbc.spi.ConnectionFactory; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.CommandLineRunner; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.autoconfigure.jdbc.AutoConfigureTestDatabase; +import org.springframework.cloud.sleuth.exporter.FinishedSpan; +import org.springframework.cloud.sleuth.test.TestSpanHandler; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.io.ClassPathResource; +import org.springframework.dao.DataAccessException; +import org.springframework.r2dbc.connection.init.ConnectionFactoryInitializer; +import org.springframework.r2dbc.connection.init.ResourceDatabasePopulator; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.TestPropertySource; + +import static org.assertj.core.api.BDDAssertions.then; + +@ContextConfiguration(classes = R2dbcIntegrationTests.TestConfig.class) +@TestPropertySource(properties = { "spring.application.name=MyApplication", "jdbc:h2:mem:test;DB_CLOSE_DELAY=-1" }) +@AutoConfigureTestDatabase(replace = AutoConfigureTestDatabase.Replace.ANY) +public abstract class R2dbcIntegrationTests { + + @Autowired + TestSpanHandler spans; + + @Test + public void should_pass_tracing_information_when_using_r2dbc() { + Set traceIds = this.spans.reportedSpans().stream().map(FinishedSpan::getTraceId) + .collect(Collectors.toSet()); + then(traceIds).as("There's one traceid").hasSize(1); + Set spanIds = this.spans.reportedSpans().stream().map(FinishedSpan::getSpanId) + .collect(Collectors.toSet()); + + // 2 transactions - 9 database interactions + then(spanIds).as("There are 11 spans").hasSize(11); + List spanNames = this.spans.reportedSpans().stream().map(FinishedSpan::getName) + .collect(Collectors.toList()); + List remoteServiceNames = this.spans.reportedSpans().stream().map(FinishedSpan::getRemoteServiceName) + .collect(Collectors.toList()); + then(spanNames.stream().filter("tx"::equalsIgnoreCase).collect(Collectors.toList())).hasSize(2); + then(remoteServiceNames.stream().filter("h2"::equalsIgnoreCase).collect(Collectors.toList())).hasSize(9); + } + + @Configuration(proxyBeanMethods = false) + @EnableAutoConfiguration + @ComponentScan + public static class TestConfig { + + private static final Logger log = LoggerFactory.getLogger(TestConfig.class); + + @Bean + public CommandLineRunner demo(ReactiveNewTransactionService reactiveNewTransactionService) { + return (args) -> { + try { + reactiveNewTransactionService.newTransaction().block(Duration.ofSeconds(50)); + } + catch (DataAccessException e) { + log.info("Expected to throw an exception so that we see if rollback works", e); + } + }; + } + + @Bean + ConnectionFactoryInitializer initializer(ConnectionFactory connectionFactory) { + ConnectionFactoryInitializer initializer = new ConnectionFactoryInitializer(); + initializer.setConnectionFactory(connectionFactory); + initializer.setDatabasePopulator(new ResourceDatabasePopulator(new ClassPathResource("schema.sql"))); + return initializer; + } + + } + +} diff --git a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/integration/sampled/WebClientTests.java b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/integration/sampled/WebClientTests.java index 7af97c32b..618c83d0a 100644 --- a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/integration/sampled/WebClientTests.java +++ b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/integration/sampled/WebClientTests.java @@ -1,597 +1,597 @@ -/* - * Copyright 2013-2021 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.web.client.integration.sampled; - -import java.time.Duration; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Optional; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReference; -import java.util.stream.Collectors; -import java.util.stream.Stream; - -import javax.servlet.http.HttpServletRequest; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.awaitility.Awaitility; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Disabled; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.MethodSource; -import org.reactivestreams.Subscription; -import reactor.core.publisher.BaseSubscriber; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.autoconfigure.jmx.JmxAutoConfiguration; -import org.springframework.boot.autoconfigure.web.ServerProperties; -import org.springframework.boot.autoconfigure.web.servlet.error.BasicErrorController; -import org.springframework.boot.web.client.RestTemplateBuilder; -import org.springframework.boot.web.client.RestTemplateCustomizer; -import org.springframework.boot.web.server.LocalServerPort; -import org.springframework.boot.web.servlet.error.ErrorAttributes; -import org.springframework.cloud.client.DefaultServiceInstance; -import org.springframework.cloud.client.loadbalancer.LoadBalanced; -import org.springframework.cloud.loadbalancer.annotation.LoadBalancerClient; -import org.springframework.cloud.loadbalancer.core.ServiceInstanceListSupplier; -import org.springframework.cloud.loadbalancer.support.ServiceInstanceListSuppliers; -import org.springframework.cloud.openfeign.EnableFeignClients; -import org.springframework.cloud.openfeign.FeignClient; -import org.springframework.cloud.sleuth.Span; -import org.springframework.cloud.sleuth.Tracer; -import org.springframework.cloud.sleuth.exporter.FinishedSpan; -import org.springframework.cloud.sleuth.test.TestSpanHandler; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.http.HttpHeaders; -import org.springframework.http.ResponseEntity; -import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.TestPropertySource; -import org.springframework.web.bind.annotation.PathVariable; -import org.springframework.web.bind.annotation.RequestHeader; -import org.springframework.web.bind.annotation.RequestMapping; -import org.springframework.web.bind.annotation.RequestMethod; -import org.springframework.web.bind.annotation.RestController; -import org.springframework.web.client.HttpClientErrorException; -import org.springframework.web.client.RestTemplate; -import org.springframework.web.reactive.function.client.UnknownHttpStatusCodeException; -import org.springframework.web.reactive.function.client.WebClient; - -import static org.assertj.core.api.Assertions.fail; -import static org.assertj.core.api.BDDAssertions.then; - -@ContextConfiguration(classes = WebClientTests.TestConfiguration.class) -@TestPropertySource(properties = { "spring.sleuth.web.servlet.enabled=false", "spring.application.name=fooservice", - "spring.sleuth.web.client.skip-pattern=/skip.*" }) -@DirtiesContext -public abstract class WebClientTests { - - private static final Log log = LogFactory.getLog(WebClientTests.class); - - @Autowired - TestFeignInterface testFeignInterface; - - @Autowired - @LoadBalanced - RestTemplate template; - - @Autowired - WebClient webClient; - - @Autowired - WebClient.Builder webClientBuilder; - - @Autowired - TestSpanHandler spans; - - @Autowired - Tracer tracer; - - @Autowired - TestErrorController testErrorController; - - @Autowired - RestTemplateBuilder restTemplateBuilder; - - @LocalServerPort - int port; - - @Autowired - FooController fooController; - - @Autowired - MyRestTemplateCustomizer customizer; - - @AfterEach - @BeforeEach - public void close() { - this.spans.clear(); - this.testErrorController.clear(); - this.fooController.clear(); - } - - @BeforeEach - public void setup() { - log.info("Starting test"); - } - - @ParameterizedTest - @MethodSource("parametersForShouldCreateANewSpanWithClientSideTagsWhenNoPreviousTracingWasPresent") - @SuppressWarnings("unchecked") - public void shouldCreateANewSpanWithClientSideTagsWhenNoPreviousTracingWasPresent(ResponseEntityProvider provider) { - ResponseEntity response = provider.get(this); - - Awaitility.await().atMost(2, TimeUnit.SECONDS).untilAsserted(() -> { - then(getHeader(response, "b3")).isNull(); - then(this.spans).isNotEmpty(); - Optional noTraceSpan = this.spans.reportedSpans().stream() - .filter(span -> span.getName().contains("GET") && !span.getTags().isEmpty() - && span.getTags().containsKey(pathKey())) - .findFirst(); - then(noTraceSpan.isPresent()).isTrue(); - then(noTraceSpan.get().getTags()).containsEntry(pathKey(), "/notrace").containsEntry("http.method", "GET"); - // TODO: matches cause there is an issue with Feign not providing the full URL - // at the interceptor level - then(noTraceSpan.get().getTags().get(pathKey())).matches(".*/notrace"); - }); - thenThereIsNoCurrentSpan(); - } - - protected String pathKey() { - return "http.path"; - } - - private void thenThereIsNoCurrentSpan() { - log.info("Current span [" + this.tracer.currentSpan() + "]"); - then(this.tracer.currentSpan()).isNull(); - } - - static Stream parametersForShouldCreateANewSpanWithClientSideTagsWhenNoPreviousTracingWasPresent() { - return Stream.of((ResponseEntityProvider) (tests) -> tests.testFeignInterface.getNoTrace(), - (ResponseEntityProvider) (tests) -> tests.testFeignInterface.getNoTrace(), - (ResponseEntityProvider) (tests) -> tests.testFeignInterface.getNoTrace(), - (ResponseEntityProvider) (tests) -> tests.testFeignInterface.getNoTrace(), - (ResponseEntityProvider) (tests) -> tests.testFeignInterface.getNoTrace(), - (ResponseEntityProvider) (tests) -> tests.testFeignInterface.getNoTrace(), - (ResponseEntityProvider) (tests) -> tests.testFeignInterface.getNoTrace(), - (ResponseEntityProvider) (tests) -> tests.testFeignInterface.getNoTrace(), - (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/notrace", - String.class), - (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/notrace", - String.class), - (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/notrace", - String.class), - (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/notrace", - String.class), - (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/notrace", - String.class), - (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/notrace", - String.class), - (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/notrace", - String.class), - (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/notrace", - String.class)); - } - - @ParameterizedTest - @MethodSource("parametersForShouldAttachTraceIdWhenCallingAnotherService") - @SuppressWarnings("unchecked") - public void shouldAttachTraceIdWhenCallingAnotherService(ResponseEntityProvider provider) { - Span span = this.tracer.nextSpan().name("foo").start(); - - try (Tracer.SpanInScope ws = this.tracer.withSpan(span)) { - ResponseEntity response = provider.get(this); - - // https://github.com/spring-cloud/spring-cloud-sleuth/issues/327 - // we don't want to respond with any tracing data - then(getHeader(response, "b3")).isNull(); - } - finally { - span.end(); - } - - thenThereIsNoCurrentSpan(); - then(this.spans).isNotEmpty(); - } - - @Test - @SuppressWarnings("unchecked") - public void shouldAttachTraceIdWhenCallingAnotherServiceViaWebClient() { - Span span = this.tracer.nextSpan().name("foo").start(); - - try (Tracer.SpanInScope ws = this.tracer.withSpan(span)) { - this.webClient.get().uri("http://localhost:" + this.port + "/traceid").retrieve().bodyToMono(String.class) - .block(Duration.ofSeconds(5)); - } - finally { - span.end(); - } - thenThereIsNoCurrentSpan(); - then(this.spans.reportedSpans().stream().filter(r -> r.getKind() != null).map(r -> r.getKind().name()) - .collect(Collectors.toList())).isNotEmpty().contains("CLIENT"); - } - - @Test - @SuppressWarnings("unchecked") - public void shouldWorkWhenCustomStatusCodeIsReturned() { - Span span = this.tracer.nextSpan().name("foo").start(); - - try (Tracer.SpanInScope ws = this.tracer.withSpan(span)) { - this.webClient.get().uri("http://localhost:" + this.port + "/issue1462").retrieve().bodyToMono(String.class) - .block(Duration.ofSeconds(5)); - } - catch (UnknownHttpStatusCodeException ex) { - - } - finally { - span.end(); - } - - thenThereIsNoCurrentSpan(); - then(this.spans.reportedSpans().stream().filter(r -> r.getKind() != null).map(r -> r.getKind().name()) - .collect(Collectors.toList())).isNotEmpty().contains("CLIENT"); - } - - @Test - @SuppressWarnings("unchecked") - public void shouldUseUriTemplateInSpanName() { - Span span = this.tracer.nextSpan().name("foo").start(); - - try (Tracer.SpanInScope ws = this.tracer.withSpan(span)) { - this.webClientBuilder.baseUrl("http://localhost:" + this.port).build().get() - .uri("/prefix/{variable}/suffix", "value").retrieve().bodyToMono(String.class) - .block(Duration.ofSeconds(5)); - } - finally { - span.end(); - } - - thenThereIsNoCurrentSpan(); - then(this.spans.reportedSpans().stream().filter(r -> r.getKind() == Span.Kind.CLIENT).map(r -> r.getName()) - .collect(Collectors.toList())).isNotEmpty().contains(templatedName()); - } - - protected String templatedName() { - return "GET /prefix/{variable}/suffix"; - } - - /** - * Cancel before {@link Subscription#request(long)} means a network request was never - * sent - */ - @Test - @Disabled("flakey") - public void shouldNotTagOnCancel() { - this.webClient.get().uri("http://localhost:" + this.port + "/doNotSkip").retrieve().bodyToMono(String.class) - .subscribe(new BaseSubscriber() { - @Override - protected void hookOnSubscribe(Subscription subscription) { - cancel(); - } - }); - - then(this.spans).isEmpty(); - } - - @Test - public void shouldRespectSkipPattern() { - this.webClient.get().uri("http://localhost:" + this.port + "/skip").retrieve().bodyToMono(String.class) - .block(Duration.ofSeconds(5)); - then(this.spans).isEmpty(); - - this.webClient.get().uri("http://localhost:" + this.port + "/doNotSkip").retrieve().bodyToMono(String.class) - .block(Duration.ofSeconds(5)); - then(this.spans).isNotEmpty(); - } - - static Stream parametersForShouldAttachTraceIdWhenCallingAnotherService() { - return Stream.of((ResponseEntityProvider) (tests) -> tests.testFeignInterface.headers(), - (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/traceid", - String.class)); - } - - @ParameterizedTest - @MethodSource("parametersForShouldAttachTraceIdWhenUsingFeignClientWithoutResponseBody") - public void shouldAttachTraceIdWhenUsingFeignClientWithoutResponseBody(ResponseEntityProvider provider) { - Span span = this.tracer.nextSpan().name("foo").start(); - - try (Tracer.SpanInScope ws = this.tracer.withSpan(span)) { - provider.get(this); - } - finally { - span.end(); - } - - thenThereIsNoCurrentSpan(); - then(this.spans).isNotEmpty(); - } - - static Stream parametersForShouldAttachTraceIdWhenUsingFeignClientWithoutResponseBody() { - return Stream.of((ResponseEntityProvider) (tests) -> tests.testFeignInterface.noResponseBody(), - (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/noresponse", - String.class)); - } - - @Test - public void shouldCloseSpanWhenErrorControllerGetsCalled() { - try { - this.template.getForEntity("http://fooservice/nonExistent", String.class); - fail("An exception should be thrown"); - } - catch (HttpClientErrorException e) { - } - - thenThereIsNoCurrentSpan(); - Optional storedSpan = this.spans.reportedSpans().stream() - .filter(span -> "404".equals(span.getTags().get("http.status_code"))).findFirst(); - then(storedSpan.isPresent()).isTrue(); - this.spans.reportedSpans().stream().forEach(span -> { - int initialSize = span.getEvents().size(); - int distinctSize = span.getEvents().stream().map(Map.Entry::getValue).distinct() - .collect(Collectors.toList()).size(); - log.info("logs " + span.getEvents()); - then(initialSize).as("there are no duplicate log entries").isEqualTo(distinctSize); - }); - - then(this.spans.reportedSpans().stream().filter(r -> r.getKind() != null).map(r -> r.getKind().name()) - .collect(Collectors.toList())).isNotEmpty().contains("CLIENT"); - } - - @Test - public void shouldNotExecuteErrorControllerWhenUrlIsFound() { - this.template.getForEntity("http://fooservice/notrace", String.class); - - thenThereIsNoCurrentSpan(); - then(this.testErrorController.getSpan()).isNull(); - } - - @Test - public void should_wrap_rest_template_builders() { - Span span = this.tracer.nextSpan().name("foo").start(); - - try (Tracer.SpanInScope ws = this.tracer.withSpan(span)) { - RestTemplate template = this.restTemplateBuilder.build(); - - template.getForObject("http://localhost:" + this.port + "/traceid", String.class); - } - finally { - span.end(); - } - thenThereIsNoCurrentSpan(); - then(this.customizer.isExecuted()).isTrue(); - then(this.spans.reportedSpans().stream().filter(s -> s.getKind() != null).map(s -> s.getKind().name()) - .collect(Collectors.toList())).contains("CLIENT"); - } - - @Test - public void should_add_headers_eagerly() { - Span span = this.tracer.nextSpan().name("foo").start(); - - AtomicReference traceId = new AtomicReference<>(); - try (Tracer.SpanInScope ws = this.tracer.withSpan(span)) { - this.webClientBuilder.filter((request, exchange) -> { - traceId.set(request.headers().getFirst("b3")); - - return exchange.exchange(request); - }).build().get().uri("http://localhost:" + this.port + "/traceid").retrieve().bodyToMono(String.class) - .block(Duration.ofSeconds(5)); - } - finally { - span.end(); - } - then(traceId).doesNotHaveValue(null); - } - - private String getHeader(ResponseEntity response, String name) { - List headers = response.getHeaders().get(name); - return headers == null || headers.isEmpty() ? null : headers.get(0); - } - - @FeignClient("fooservice") - public interface TestFeignInterface { - - @RequestMapping(method = RequestMethod.GET, value = "/traceid") - ResponseEntity getTraceId(); - - @RequestMapping(method = RequestMethod.GET, value = "/notrace") - ResponseEntity getNoTrace(); - - @RequestMapping(method = RequestMethod.GET, value = "/") - ResponseEntity> headers(); - - @RequestMapping(method = RequestMethod.GET, value = "/noresponse") - ResponseEntity noResponseBody(); - - } - - @FunctionalInterface - interface ResponseEntityProvider { - - @SuppressWarnings("rawtypes") - ResponseEntity get(WebClientTests webClientTests); - - } - - @Configuration(proxyBeanMethods = false) - @EnableAutoConfiguration(exclude = JmxAutoConfiguration.class) - @EnableFeignClients - @LoadBalancerClient(value = "fooservice", configuration = SimpleLoadBalancerClientConfiguration.class) - public static class TestConfiguration { - - @Bean - FooController fooController() { - return new FooController(); - } - - @Bean - WebClientController webClientController() { - return new WebClientController(); - } - - @LoadBalanced - @Bean - public RestTemplate restTemplate() { - return new RestTemplate(); - } - - @Bean - TestErrorController testErrorController(ErrorAttributes errorAttributes, Tracer tracer) { - return new TestErrorController(errorAttributes, tracer); - } - - @Bean - WebClient webClient() { - return WebClient.builder().build(); - } - - @Bean - WebClient.Builder webClientBuilder() { - return WebClient.builder(); - } - - @Bean - RestTemplateCustomizer myRestTemplateCustomizer() { - return new MyRestTemplateCustomizer(); - } - - } - - static class MyRestTemplateCustomizer implements RestTemplateCustomizer { - - boolean executed; - - @Override - public void customize(RestTemplate restTemplate) { - this.executed = true; - } - - public boolean isExecuted() { - return this.executed; - } - - } - - public static class TestErrorController extends BasicErrorController { - - private final Tracer tracer; - - Span span; - - public TestErrorController(ErrorAttributes errorAttributes, Tracer tracer) { - super(errorAttributes, new ServerProperties().getError()); - this.tracer = tracer; - } - - @Override - public ResponseEntity> error(HttpServletRequest request) { - this.span = this.tracer.currentSpan(); - return super.error(request); - } - - public Span getSpan() { - return this.span; - } - - public void clear() { - this.span = null; - } - - } - - @RestController - public static class FooController { - - Span span; - - @RequestMapping(value = "/notrace", method = RequestMethod.GET) - public String notrace(@RequestHeader(name = "b3", required = false) String b3Single) { - then(b3Single).isNotNull(); - return "OK"; - } - - @RequestMapping(value = "/traceid", method = RequestMethod.GET) - public String traceId(@RequestHeader("b3") String b3Single) { - then(b3Single).isNotEmpty(); - return b3Single; - } - - @RequestMapping("/") - public Map home(@RequestHeader HttpHeaders headers) { - Map map = new HashMap<>(); - for (String key : headers.keySet()) { - map.put(key, headers.getFirst(key)); - } - return map; - } - - @RequestMapping("/noresponse") - public void noResponse(@RequestHeader("b3") String b3Single) { - then(b3Single).isNotEmpty(); - } - - public Span getSpan() { - return this.span; - } - - public void clear() { - this.span = null; - } - - } - - @RestController - public static class WebClientController { - - @RequestMapping(value = "/issue1462", method = RequestMethod.GET) - public ResponseEntity issue1462() { - return ResponseEntity.status(499).body("issue1462"); - } - - @RequestMapping(value = { "/skip", "/doNotSkip" }, method = RequestMethod.GET) - String skip() { - return "ok"; - } - - @RequestMapping(value = "/prefix/{variable}/suffix", method = RequestMethod.GET) - String pathVariable(@PathVariable("variable") String variable) { - return "variable = " + variable; - } - - } - - @Configuration(proxyBeanMethods = false) - public static class SimpleLoadBalancerClientConfiguration { - - @Value("${local.server.port}") - private int port = 0; - - @Bean - public ServiceInstanceListSupplier serviceInstanceListSupplier() { - return ServiceInstanceListSuppliers.from("fooservice", - new DefaultServiceInstance("fooservice" + "-1", "fooservice", "localhost", port, false)); - } - - } - -} +/* + * Copyright 2013-2021 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.web.client.integration.sampled; + +import java.time.Duration; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import javax.servlet.http.HttpServletRequest; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; +import org.reactivestreams.Subscription; +import reactor.core.publisher.BaseSubscriber; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.autoconfigure.jmx.JmxAutoConfiguration; +import org.springframework.boot.autoconfigure.web.ServerProperties; +import org.springframework.boot.autoconfigure.web.servlet.error.BasicErrorController; +import org.springframework.boot.web.client.RestTemplateBuilder; +import org.springframework.boot.web.client.RestTemplateCustomizer; +import org.springframework.boot.web.server.LocalServerPort; +import org.springframework.boot.web.servlet.error.ErrorAttributes; +import org.springframework.cloud.client.DefaultServiceInstance; +import org.springframework.cloud.client.loadbalancer.LoadBalanced; +import org.springframework.cloud.loadbalancer.annotation.LoadBalancerClient; +import org.springframework.cloud.loadbalancer.core.ServiceInstanceListSupplier; +import org.springframework.cloud.loadbalancer.support.ServiceInstanceListSuppliers; +import org.springframework.cloud.openfeign.EnableFeignClients; +import org.springframework.cloud.openfeign.FeignClient; +import org.springframework.cloud.sleuth.Span; +import org.springframework.cloud.sleuth.Tracer; +import org.springframework.cloud.sleuth.exporter.FinishedSpan; +import org.springframework.cloud.sleuth.test.TestSpanHandler; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.http.HttpHeaders; +import org.springframework.http.ResponseEntity; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.TestPropertySource; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.RequestHeader; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestMethod; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.client.HttpClientErrorException; +import org.springframework.web.client.RestTemplate; +import org.springframework.web.reactive.function.client.UnknownHttpStatusCodeException; +import org.springframework.web.reactive.function.client.WebClient; + +import static org.assertj.core.api.Assertions.fail; +import static org.assertj.core.api.BDDAssertions.then; + +@ContextConfiguration(classes = WebClientTests.TestConfiguration.class) +@TestPropertySource(properties = { "spring.sleuth.web.servlet.enabled=false", "spring.application.name=fooservice", + "spring.sleuth.web.client.skip-pattern=/skip.*" }) +@DirtiesContext +public abstract class WebClientTests { + + private static final Log log = LogFactory.getLog(WebClientTests.class); + + @Autowired + TestFeignInterface testFeignInterface; + + @Autowired + @LoadBalanced + RestTemplate template; + + @Autowired + WebClient webClient; + + @Autowired + WebClient.Builder webClientBuilder; + + @Autowired + TestSpanHandler spans; + + @Autowired + Tracer tracer; + + @Autowired + TestErrorController testErrorController; + + @Autowired + RestTemplateBuilder restTemplateBuilder; + + @LocalServerPort + int port; + + @Autowired + FooController fooController; + + @Autowired + MyRestTemplateCustomizer customizer; + + @AfterEach + @BeforeEach + public void close() { + this.spans.clear(); + this.testErrorController.clear(); + this.fooController.clear(); + } + + @BeforeEach + public void setup() { + log.info("Starting test"); + } + + @ParameterizedTest + @MethodSource("parametersForShouldCreateANewSpanWithClientSideTagsWhenNoPreviousTracingWasPresent") + @SuppressWarnings("unchecked") + public void shouldCreateANewSpanWithClientSideTagsWhenNoPreviousTracingWasPresent(ResponseEntityProvider provider) { + ResponseEntity response = provider.get(this); + + Awaitility.await().atMost(2, TimeUnit.SECONDS).untilAsserted(() -> { + then(getHeader(response, "b3")).isNull(); + then(this.spans).isNotEmpty(); + Optional noTraceSpan = this.spans.reportedSpans().stream() + .filter(span -> span.getName().contains("GET") && !span.getTags().isEmpty() + && span.getTags().containsKey(pathKey())) + .findFirst(); + then(noTraceSpan.isPresent()).isTrue(); + then(noTraceSpan.get().getTags()).containsEntry(pathKey(), "/notrace").containsEntry("http.method", "GET"); + // TODO: matches cause there is an issue with Feign not providing the full URL + // at the interceptor level + then(noTraceSpan.get().getTags().get(pathKey())).matches(".*/notrace"); + }); + thenThereIsNoCurrentSpan(); + } + + protected String pathKey() { + return "http.path"; + } + + private void thenThereIsNoCurrentSpan() { + log.info("Current span [" + this.tracer.currentSpan() + "]"); + then(this.tracer.currentSpan()).isNull(); + } + + static Stream parametersForShouldCreateANewSpanWithClientSideTagsWhenNoPreviousTracingWasPresent() { + return Stream.of((ResponseEntityProvider) (tests) -> tests.testFeignInterface.getNoTrace(), + (ResponseEntityProvider) (tests) -> tests.testFeignInterface.getNoTrace(), + (ResponseEntityProvider) (tests) -> tests.testFeignInterface.getNoTrace(), + (ResponseEntityProvider) (tests) -> tests.testFeignInterface.getNoTrace(), + (ResponseEntityProvider) (tests) -> tests.testFeignInterface.getNoTrace(), + (ResponseEntityProvider) (tests) -> tests.testFeignInterface.getNoTrace(), + (ResponseEntityProvider) (tests) -> tests.testFeignInterface.getNoTrace(), + (ResponseEntityProvider) (tests) -> tests.testFeignInterface.getNoTrace(), + (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/notrace", + String.class), + (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/notrace", + String.class), + (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/notrace", + String.class), + (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/notrace", + String.class), + (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/notrace", + String.class), + (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/notrace", + String.class), + (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/notrace", + String.class), + (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/notrace", + String.class)); + } + + @ParameterizedTest + @MethodSource("parametersForShouldAttachTraceIdWhenCallingAnotherService") + @SuppressWarnings("unchecked") + public void shouldAttachTraceIdWhenCallingAnotherService(ResponseEntityProvider provider) { + Span span = this.tracer.nextSpan().name("foo").start(); + + try (Tracer.SpanInScope ws = this.tracer.withSpan(span)) { + ResponseEntity response = provider.get(this); + + // https://github.com/spring-cloud/spring-cloud-sleuth/issues/327 + // we don't want to respond with any tracing data + then(getHeader(response, "b3")).isNull(); + } + finally { + span.end(); + } + + thenThereIsNoCurrentSpan(); + then(this.spans).isNotEmpty(); + } + + @Test + @SuppressWarnings("unchecked") + public void shouldAttachTraceIdWhenCallingAnotherServiceViaWebClient() { + Span span = this.tracer.nextSpan().name("foo").start(); + + try (Tracer.SpanInScope ws = this.tracer.withSpan(span)) { + this.webClient.get().uri("http://localhost:" + this.port + "/traceid").retrieve().bodyToMono(String.class) + .block(Duration.ofSeconds(5)); + } + finally { + span.end(); + } + thenThereIsNoCurrentSpan(); + then(this.spans.reportedSpans().stream().filter(r -> r.getKind() != null).map(r -> r.getKind().name()) + .collect(Collectors.toList())).isNotEmpty().contains("CLIENT"); + } + + @Test + @SuppressWarnings("unchecked") + public void shouldWorkWhenCustomStatusCodeIsReturned() { + Span span = this.tracer.nextSpan().name("foo").start(); + + try (Tracer.SpanInScope ws = this.tracer.withSpan(span)) { + this.webClient.get().uri("http://localhost:" + this.port + "/issue1462").retrieve().bodyToMono(String.class) + .block(Duration.ofSeconds(5)); + } + catch (UnknownHttpStatusCodeException ex) { + + } + finally { + span.end(); + } + + thenThereIsNoCurrentSpan(); + then(this.spans.reportedSpans().stream().filter(r -> r.getKind() != null).map(r -> r.getKind().name()) + .collect(Collectors.toList())).isNotEmpty().contains("CLIENT"); + } + + @Test + @SuppressWarnings("unchecked") + public void shouldUseUriTemplateInSpanName() { + Span span = this.tracer.nextSpan().name("foo").start(); + + try (Tracer.SpanInScope ws = this.tracer.withSpan(span)) { + this.webClientBuilder.baseUrl("http://localhost:" + this.port).build().get() + .uri("/prefix/{variable}/suffix", "value").retrieve().bodyToMono(String.class) + .block(Duration.ofSeconds(5)); + } + finally { + span.end(); + } + + thenThereIsNoCurrentSpan(); + then(this.spans.reportedSpans().stream().filter(r -> r.getKind() == Span.Kind.CLIENT).map(r -> r.getName()) + .collect(Collectors.toList())).isNotEmpty().contains(templatedName()); + } + + protected String templatedName() { + return "GET /prefix/{variable}/suffix"; + } + + /** + * Cancel before {@link Subscription#request(long)} means a network request was never + * sent + */ + @Test + @Disabled("flakey") + public void shouldNotTagOnCancel() { + this.webClient.get().uri("http://localhost:" + this.port + "/doNotSkip").retrieve().bodyToMono(String.class) + .subscribe(new BaseSubscriber() { + @Override + protected void hookOnSubscribe(Subscription subscription) { + cancel(); + } + }); + + then(this.spans).isEmpty(); + } + + @Test + public void shouldRespectSkipPattern() { + this.webClient.get().uri("http://localhost:" + this.port + "/skip").retrieve().bodyToMono(String.class) + .block(Duration.ofSeconds(5)); + then(this.spans).isEmpty(); + + this.webClient.get().uri("http://localhost:" + this.port + "/doNotSkip").retrieve().bodyToMono(String.class) + .block(Duration.ofSeconds(5)); + then(this.spans).isNotEmpty(); + } + + static Stream parametersForShouldAttachTraceIdWhenCallingAnotherService() { + return Stream.of((ResponseEntityProvider) (tests) -> tests.testFeignInterface.headers(), + (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/traceid", + String.class)); + } + + @ParameterizedTest + @MethodSource("parametersForShouldAttachTraceIdWhenUsingFeignClientWithoutResponseBody") + public void shouldAttachTraceIdWhenUsingFeignClientWithoutResponseBody(ResponseEntityProvider provider) { + Span span = this.tracer.nextSpan().name("foo").start(); + + try (Tracer.SpanInScope ws = this.tracer.withSpan(span)) { + provider.get(this); + } + finally { + span.end(); + } + + thenThereIsNoCurrentSpan(); + then(this.spans).isNotEmpty(); + } + + static Stream parametersForShouldAttachTraceIdWhenUsingFeignClientWithoutResponseBody() { + return Stream.of((ResponseEntityProvider) (tests) -> tests.testFeignInterface.noResponseBody(), + (ResponseEntityProvider) (tests) -> tests.template.getForEntity("http://fooservice/noresponse", + String.class)); + } + + @Test + public void shouldCloseSpanWhenErrorControllerGetsCalled() { + try { + this.template.getForEntity("http://fooservice/nonExistent", String.class); + fail("An exception should be thrown"); + } + catch (HttpClientErrorException e) { + } + + thenThereIsNoCurrentSpan(); + Optional storedSpan = this.spans.reportedSpans().stream() + .filter(span -> "404".equals(span.getTags().get("http.status_code"))).findFirst(); + then(storedSpan.isPresent()).isTrue(); + this.spans.reportedSpans().stream().forEach(span -> { + int initialSize = span.getEvents().size(); + int distinctSize = span.getEvents().stream().map(Map.Entry::getValue).distinct() + .collect(Collectors.toList()).size(); + log.info("logs " + span.getEvents()); + then(initialSize).as("there are no duplicate log entries").isEqualTo(distinctSize); + }); + + then(this.spans.reportedSpans().stream().filter(r -> r.getKind() != null).map(r -> r.getKind().name()) + .collect(Collectors.toList())).isNotEmpty().contains("CLIENT"); + } + + @Test + public void shouldNotExecuteErrorControllerWhenUrlIsFound() { + this.template.getForEntity("http://fooservice/notrace", String.class); + + thenThereIsNoCurrentSpan(); + then(this.testErrorController.getSpan()).isNull(); + } + + @Test + public void should_wrap_rest_template_builders() { + Span span = this.tracer.nextSpan().name("foo").start(); + + try (Tracer.SpanInScope ws = this.tracer.withSpan(span)) { + RestTemplate template = this.restTemplateBuilder.build(); + + template.getForObject("http://localhost:" + this.port + "/traceid", String.class); + } + finally { + span.end(); + } + thenThereIsNoCurrentSpan(); + then(this.customizer.isExecuted()).isTrue(); + then(this.spans.reportedSpans().stream().filter(s -> s.getKind() != null).map(s -> s.getKind().name()) + .collect(Collectors.toList())).contains("CLIENT"); + } + + @Test + public void should_add_headers_eagerly() { + Span span = this.tracer.nextSpan().name("foo").start(); + + AtomicReference traceId = new AtomicReference<>(); + try (Tracer.SpanInScope ws = this.tracer.withSpan(span)) { + this.webClientBuilder.filter((request, exchange) -> { + traceId.set(request.headers().getFirst("b3")); + + return exchange.exchange(request); + }).build().get().uri("http://localhost:" + this.port + "/traceid").retrieve().bodyToMono(String.class) + .block(Duration.ofSeconds(5)); + } + finally { + span.end(); + } + then(traceId).doesNotHaveValue(null); + } + + private String getHeader(ResponseEntity response, String name) { + List headers = response.getHeaders().get(name); + return headers == null || headers.isEmpty() ? null : headers.get(0); + } + + @FeignClient("fooservice") + public interface TestFeignInterface { + + @RequestMapping(method = RequestMethod.GET, value = "/traceid") + ResponseEntity getTraceId(); + + @RequestMapping(method = RequestMethod.GET, value = "/notrace") + ResponseEntity getNoTrace(); + + @RequestMapping(method = RequestMethod.GET, value = "/") + ResponseEntity> headers(); + + @RequestMapping(method = RequestMethod.GET, value = "/noresponse") + ResponseEntity noResponseBody(); + + } + + @FunctionalInterface + interface ResponseEntityProvider { + + @SuppressWarnings("rawtypes") + ResponseEntity get(WebClientTests webClientTests); + + } + + @Configuration(proxyBeanMethods = false) + @EnableAutoConfiguration(exclude = JmxAutoConfiguration.class) + @EnableFeignClients + @LoadBalancerClient(value = "fooservice", configuration = SimpleLoadBalancerClientConfiguration.class) + public static class TestConfiguration { + + @Bean + FooController fooController() { + return new FooController(); + } + + @Bean + WebClientController webClientController() { + return new WebClientController(); + } + + @LoadBalanced + @Bean + public RestTemplate restTemplate() { + return new RestTemplate(); + } + + @Bean + TestErrorController testErrorController(ErrorAttributes errorAttributes, Tracer tracer) { + return new TestErrorController(errorAttributes, tracer); + } + + @Bean + WebClient webClient() { + return WebClient.builder().build(); + } + + @Bean + WebClient.Builder webClientBuilder() { + return WebClient.builder(); + } + + @Bean + RestTemplateCustomizer myRestTemplateCustomizer() { + return new MyRestTemplateCustomizer(); + } + + } + + static class MyRestTemplateCustomizer implements RestTemplateCustomizer { + + boolean executed; + + @Override + public void customize(RestTemplate restTemplate) { + this.executed = true; + } + + public boolean isExecuted() { + return this.executed; + } + + } + + public static class TestErrorController extends BasicErrorController { + + private final Tracer tracer; + + Span span; + + public TestErrorController(ErrorAttributes errorAttributes, Tracer tracer) { + super(errorAttributes, new ServerProperties().getError()); + this.tracer = tracer; + } + + @Override + public ResponseEntity> error(HttpServletRequest request) { + this.span = this.tracer.currentSpan(); + return super.error(request); + } + + public Span getSpan() { + return this.span; + } + + public void clear() { + this.span = null; + } + + } + + @RestController + public static class FooController { + + Span span; + + @RequestMapping(value = "/notrace", method = RequestMethod.GET) + public String notrace(@RequestHeader(name = "b3", required = false) String b3Single) { + then(b3Single).isNotNull(); + return "OK"; + } + + @RequestMapping(value = "/traceid", method = RequestMethod.GET) + public String traceId(@RequestHeader("b3") String b3Single) { + then(b3Single).isNotEmpty(); + return b3Single; + } + + @RequestMapping("/") + public Map home(@RequestHeader HttpHeaders headers) { + Map map = new HashMap<>(); + for (String key : headers.keySet()) { + map.put(key, headers.getFirst(key)); + } + return map; + } + + @RequestMapping("/noresponse") + public void noResponse(@RequestHeader("b3") String b3Single) { + then(b3Single).isNotEmpty(); + } + + public Span getSpan() { + return this.span; + } + + public void clear() { + this.span = null; + } + + } + + @RestController + public static class WebClientController { + + @RequestMapping(value = "/issue1462", method = RequestMethod.GET) + public ResponseEntity issue1462() { + return ResponseEntity.status(499).body("issue1462"); + } + + @RequestMapping(value = { "/skip", "/doNotSkip" }, method = RequestMethod.GET) + String skip() { + return "ok"; + } + + @RequestMapping(value = "/prefix/{variable}/suffix", method = RequestMethod.GET) + String pathVariable(@PathVariable("variable") String variable) { + return "variable = " + variable; + } + + } + + @Configuration(proxyBeanMethods = false) + public static class SimpleLoadBalancerClientConfiguration { + + @Value("${local.server.port}") + private int port = 0; + + @Bean + public ServiceInstanceListSupplier serviceInstanceListSupplier() { + return ServiceInstanceListSuppliers.from("fooservice", + new DefaultServiceInstance("fooservice" + "-1", "fooservice", "localhost", port, false)); + } + + } + +}