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

Merge branch '2.2.x'

Browse Source
This commit is contained in:
Marcin Grzejszczak
2020-12-01 22:05:23 +01:00
parent c6a84424ac f50fb221a7
commit a5f58c50b9
29 changed files with 307 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/brave/BraveAutoConfiguration.java
View File

@@ -43,6 +43,7 @@ import org.springframework.cloud.sleuth.autoconfig.TraceConfiguration;
import org.springframework.cloud.sleuth.autoconfig.brave.instrument.web.BraveHttpConfiguration;
import org.springframework.cloud.sleuth.brave.LocalServiceName;
import org.springframework.cloud.sleuth.internal.DefaultSpanNamer;
import org.springframework.cloud.sleuth.internal.SleuthContextListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
@@ -153,4 +154,9 @@ public class BraveAutoConfiguration {
return CurrentSpanCustomizer.create(tracing);
}
@Bean
SleuthContextListener sleuthContextListener() {
return new SleuthContextListener();
}
}

1
spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/messaging/BraveMessagingAutoConfiguration.java
View File

@@ -36,6 +36,7 @@ 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.instrument.messaging.SleuthMessagingProperties;
import org.springframework.cloud.sleuth.brave.instrument.messaging.ConditionalOnMessagingEnabled;
import org.springframework.cloud.sleuth.brave.instrument.messaging.ConsumerSampler;
import org.springframework.cloud.sleuth.brave.instrument.messaging.KafkaFactoryBeanPostProcessor;

7
spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/async/TraceAsyncAutoConfiguration.java
View File

@@ -23,8 +23,6 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.autoconfig.brave.BraveAutoConfiguration;
import org.springframework.cloud.sleuth.autoconfig.otel.OtelAutoConfiguration;
import org.springframework.cloud.sleuth.instrument.async.SleuthContextListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
@@ -41,9 +39,4 @@ import org.springframework.context.annotation.Configuration;
@AutoConfigureAfter({ BraveAutoConfiguration.class, OtelAutoConfiguration.class })
public class TraceAsyncAutoConfiguration {
@Bean
SleuthContextListener traceContextClosedListener() {
return new SleuthContextListener();
}
}

2
spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/brave/instrument/messaging/SleuthMessagingProperties.java → spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/messaging/SleuthMessagingProperties.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.sleuth.autoconfig.brave.instrument.messaging;
package org.springframework.cloud.sleuth.autoconfig.instrument.messaging;
import org.springframework.boot.context.properties.ConfigurationProperties;

23
spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/messaging/TraceSpringIntegrationAutoConfiguration.java
View File

@@ -16,6 +16,8 @@
package org.springframework.cloud.sleuth.autoconfig.instrument.messaging;
import java.util.function.Function;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.AnyNestedCondition;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
@@ -39,6 +41,7 @@ import org.springframework.integration.channel.interceptor.GlobalChannelIntercep
import org.springframework.integration.config.GlobalChannelInterceptor;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
/**
* {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration
@@ -70,9 +73,25 @@ public class TraceSpringIntegrationAutoConfiguration {
@Bean
TracingChannelInterceptor traceChannelInterceptor(Tracer tracer, Propagator propagator,
Propagator.Setter<MessageHeaderAccessor> traceMessagePropagationSetter,
Propagator.Getter<MessageHeaderAccessor> traceMessagePropagationGetter) {
Propagator.Getter<MessageHeaderAccessor> traceMessagePropagationGetter,
SleuthMessagingProperties properties) {
return new TracingChannelInterceptor(tracer, propagator, traceMessagePropagationSetter,
traceMessagePropagationGetter);
traceMessagePropagationGetter, remoteServiceNameMapper(properties));
}
static Function<String, String> remoteServiceNameMapper(SleuthMessagingProperties properties) {
return s -> {
if (!StringUtils.hasText(s)) {
return null;
}
if (s.startsWith("amqp") || s.startsWith("rabbit")) {
return properties.getMessaging().getRabbit().getRemoteServiceName();
}
else if (s.startsWith("kafka")) {
return properties.getMessaging().getKafka().getRemoteServiceName();
}
return null;
};
}
}

2
spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/messaging/TraceSpringMessagingAutoConfiguration.java
View File

@@ -29,7 +29,7 @@ import org.springframework.messaging.support.MessageHeaderAccessor;
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(MessageHeaderAccessor.class)
@ConditionalOnProperty(value = "spring.sleuth.messaging.enabled", matchIfMissing = true)
@EnableConfigurationProperties(SleuthIntegrationMessagingProperties.class)
@EnableConfigurationProperties({ SleuthIntegrationMessagingProperties.class, SleuthMessagingProperties.class })
class TraceSpringMessagingAutoConfiguration {
@Bean

18
spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/messaging/TraceWebSocketAutoConfiguration.java
View File

@@ -61,6 +61,9 @@ class TraceWebSocketAutoConfiguration extends AbstractWebSocketMessageBrokerConf
@Autowired
Propagator.Getter<MessageHeaderAccessor> getter;
@Autowired
SleuthMessagingProperties sleuthMessagingProperties;
@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
// The user must register their own endpoints
@@ -68,20 +71,23 @@ class TraceWebSocketAutoConfiguration extends AbstractWebSocketMessageBrokerConf
@Override
public void configureMessageBroker(MessageBrokerRegistry registry) {
registry.configureBrokerChannel()
.setInterceptors(new TracingChannelInterceptor(this.tracer, this.propagator, this.setter, this.getter));
registry.configureBrokerChannel().setInterceptors(new TracingChannelInterceptor(this.tracer, this.propagator,
this.setter, this.getter,
TraceSpringIntegrationAutoConfiguration.remoteServiceNameMapper(this.sleuthMessagingProperties)));
}
@Override
public void configureClientOutboundChannel(ChannelRegistration registration) {
registration
.setInterceptors(new TracingChannelInterceptor(this.tracer, this.propagator, this.setter, this.getter));
registration.setInterceptors(new TracingChannelInterceptor(this.tracer, this.propagator, this.setter,
this.getter,
TraceSpringIntegrationAutoConfiguration.remoteServiceNameMapper(this.sleuthMessagingProperties)));
}
@Override
public void configureClientInboundChannel(ChannelRegistration registration) {
registration
.setInterceptors(new TracingChannelInterceptor(this.tracer, this.propagator, this.setter, this.getter));
registration.setInterceptors(new TracingChannelInterceptor(this.tracer, this.propagator, this.setter,
this.getter,
TraceSpringIntegrationAutoConfiguration.remoteServiceNameMapper(this.sleuthMessagingProperties)));
}
}

6
spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/otel/OtelAutoConfiguration.java
View File

@@ -45,6 +45,7 @@ import org.springframework.cloud.sleuth.autoconfig.SleuthBaggageProperties;
import org.springframework.cloud.sleuth.autoconfig.SleuthSpanFilterProperties;
import org.springframework.cloud.sleuth.autoconfig.SleuthTracerProperties;
import org.springframework.cloud.sleuth.autoconfig.TraceConfiguration;
import org.springframework.cloud.sleuth.internal.SleuthContextListener;
import org.springframework.cloud.sleuth.otel.bridge.OtelOpenTelemetry;
import org.springframework.cloud.sleuth.otel.bridge.SpanExporterCustomizer;
import org.springframework.context.annotation.Bean;
@@ -137,4 +138,9 @@ public class OtelAutoConfiguration {
return Sampler.traceIdRatioBased(otelProperties.getTraceIdRatioBased());
}
@Bean
SleuthContextListener sleuthContextListener() {
return new SleuthContextListener();
}
}

4
spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/instrument/web/TraceWebServletConfigurationTests.java
View File

@@ -40,7 +40,8 @@ import static org.assertj.core.api.Assertions.assertThat;
public class TraceWebServletConfigurationTests {
private final WebApplicationContextRunner contextRunner = new WebApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(TraceWebAutoConfiguration.class)).withUserConfiguration(TestConfig.class);
.withConfiguration(AutoConfigurations.of(TraceWebAutoConfiguration.class))
.withUserConfiguration(TestConfig.class);
@Test
public void shouldNotCreateTracedWebBeansWhenServletClassMissing() {
@@ -78,6 +79,7 @@ public class TraceWebServletConfigurationTests {
HttpServerHandler httpServerHandler() {
return BDDMockito.mock(HttpServerHandler.class);
}
}
}

68
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/LazyTracingClientHttpRequestInterceptorTests.java Normal file
View File

@@ -0,0 +1,68 @@
/*
* 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.web.client;
import java.io.IOException;
import brave.spring.web.TracingClientHttpRequestInterceptor;
import org.junit.jupiter.api.Test;
import org.mockito.BDDMockito;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.http.HttpRequest;
import org.springframework.http.client.ClientHttpRequestExecution;
import org.springframework.http.client.ClientHttpRequestInterceptor;
import static org.mockito.BDDMockito.then;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
class LazyTracingClientHttpRequestInterceptorTests {
@Test
void should_not_retrieve_bean_from_context_when_context_is_not_ready() throws IOException {
BeanFactory beanFactory = mock(BeanFactory.class);
LazyTracingClientHttpRequestInterceptor interceptor = new LazyTracingClientHttpRequestInterceptor(beanFactory);
interceptor.intercept(mock(HttpRequest.class), new byte[0], mock(ClientHttpRequestExecution.class));
then(beanFactory).should(never()).getBean(TracingClientHttpRequestInterceptor.class);
}
@Test
void should_retrieve_bean_from_context_when_context_is_ready() throws IOException {
BeanFactory beanFactory = mock(BeanFactory.class);
ClientHttpRequestInterceptor requestInterceptor = mock(ClientHttpRequestInterceptor.class);
LazyTracingClientHttpRequestInterceptor interceptor = new LazyTracingClientHttpRequestInterceptor(beanFactory) {
@Override
ClientHttpRequestInterceptor interceptor() {
return requestInterceptor;
}
@Override
boolean isContextUnusable() {
return false;
}
};
interceptor.intercept(mock(HttpRequest.class), new byte[0], mock(ClientHttpRequestExecution.class));
then(requestInterceptor).should().intercept(BDDMockito.any(HttpRequest.class), BDDMockito.any(byte[].class),
BDDMockito.any(ClientHttpRequestExecution.class));
}
}

1
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceAsyncTaskExecutor.java
View File

@@ -26,6 +26,7 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.cloud.sleuth.SpanNamer;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.internal.ContextUtil;
import org.springframework.cloud.sleuth.internal.DefaultSpanNamer;
import org.springframework.core.task.AsyncTaskExecutor;

1
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceExecutor.java
View File

@@ -25,6 +25,7 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.cloud.sleuth.SpanNamer;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.internal.ContextUtil;
import org.springframework.cloud.sleuth.internal.DefaultSpanNamer;
/**

54
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceScheduledThreadPoolExecutor.java
View File

@@ -40,6 +40,7 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.cloud.sleuth.SpanNamer;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.internal.ContextUtil;
import org.springframework.cloud.sleuth.internal.DefaultSpanNamer;
import org.springframework.util.ReflectionUtils;
@@ -144,61 +145,76 @@ class LazyTraceScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor {
makeAccessibleIfNotNull(this.newTaskForCallable);
}
private Runnable traceRunnableWhenContextReady(Runnable delegate) {
if (isContextUnusable()) {
return delegate;
}
return new TraceRunnable(tracing(), spanNamer(), delegate, this.beanName);
}
boolean isContextUnusable() {
return ContextUtil.isContextUnusable(this.beanFactory);
}
private <V> Callable<V> traceCallableWhenContextReady(Callable<V> delegate) {
if (isContextUnusable()) {
return delegate;
}
return new TraceCallable<>(tracing(), spanNamer(), delegate, this.beanName);
}
@Override
@SuppressWarnings("unchecked")
public <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
return (RunnableScheduledFuture<V>) ReflectionUtils.invokeMethod(this.decorateTaskRunnable, this.delegate,
new TraceRunnable(tracing(), spanNamer(), runnable, this.beanName), task);
traceRunnableWhenContextReady(runnable), task);
}
@Override
@SuppressWarnings("unchecked")
public <V> RunnableScheduledFuture<V> decorateTask(Callable<V> callable, RunnableScheduledFuture<V> task) {
return (RunnableScheduledFuture<V>) ReflectionUtils.invokeMethod(this.decorateTaskCallable, this.delegate,
new TraceCallable<>(tracing(), spanNamer(), callable, this.beanName), task);
traceCallableWhenContextReady(callable), task);
}
@Override
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
return this.delegate.schedule(new TraceRunnable(tracing(), spanNamer(), command, this.beanName), delay, unit);
return this.delegate.schedule(traceRunnableWhenContextReady(command), delay, unit);
}
@Override
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
return this.delegate.schedule(new TraceCallable<>(tracing(), spanNamer(), callable, this.beanName), delay,
unit);
return this.delegate.schedule(traceCallableWhenContextReady(callable), delay, unit);
}
@Override
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
return this.delegate.scheduleAtFixedRate(new TraceRunnable(tracing(), spanNamer(), command, this.beanName),
initialDelay, period, unit);
return this.delegate.scheduleAtFixedRate(traceRunnableWhenContextReady(command), initialDelay, period, unit);
}
@Override
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
return this.delegate.scheduleWithFixedDelay(new TraceRunnable(tracing(), spanNamer(), command, this.beanName),
initialDelay, delay, unit);
return this.delegate.scheduleWithFixedDelay(traceRunnableWhenContextReady(command), initialDelay, delay, unit);
}
@Override
public void execute(Runnable command) {
this.delegate.execute(new TraceRunnable(tracing(), spanNamer(), command, this.beanName));
this.delegate.execute(traceRunnableWhenContextReady(command));
}
@Override
public Future<?> submit(Runnable task) {
return this.delegate.submit(new TraceRunnable(tracing(), spanNamer(), task, this.beanName));
return this.delegate.submit(traceRunnableWhenContextReady(task));
}
@Override
public <T> Future<T> submit(Runnable task, T result) {
return this.delegate.submit(new TraceRunnable(tracing(), spanNamer(), task, this.beanName), result);
return this.delegate.submit(traceRunnableWhenContextReady(task), result);
}
@Override
public <T> Future<T> submit(Callable<T> task) {
return this.delegate.submit(new TraceCallable<>(tracing(), spanNamer(), task, this.beanName));
return this.delegate.submit(traceCallableWhenContextReady(task));
}
@Override
@@ -382,14 +398,12 @@ class LazyTraceScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor {
@Override
public void beforeExecute(Thread t, Runnable r) {
ReflectionUtils.invokeMethod(this.beforeExecute, this.delegate, t,
new TraceRunnable(tracing(), spanNamer(), r, this.beanName));
ReflectionUtils.invokeMethod(this.beforeExecute, this.delegate, t, traceRunnableWhenContextReady(r));
}
@Override
public void afterExecute(Runnable r, Throwable t) {
ReflectionUtils.invokeMethod(this.afterExecute, this.delegate,
new TraceRunnable(tracing(), spanNamer(), r, this.beanName), t);
ReflectionUtils.invokeMethod(this.afterExecute, this.delegate, traceRunnableWhenContextReady(r), t);
}
@Override
@@ -401,14 +415,14 @@ class LazyTraceScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor {
@SuppressWarnings("unchecked")
public <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return (RunnableFuture<T>) ReflectionUtils.invokeMethod(this.newTaskForRunnable, this.delegate,
new TraceRunnable(tracing(), spanNamer(), runnable, this.beanName), value);
traceRunnableWhenContextReady(runnable), value);
}
@Override
@SuppressWarnings("unchecked")
public <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return (RunnableFuture<T>) ReflectionUtils.invokeMethod(this.newTaskForCallable, this.delegate,
new TraceCallable<>(tracing(), spanNamer(), callable, this.beanName));
traceCallableWhenContextReady(callable));
}
@Override
@@ -420,7 +434,7 @@ class LazyTraceScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor {
List<Callable<T>> ts = new ArrayList<>();
for (Callable<T> task : tasks) {
if (!(task instanceof TraceCallable)) {
ts.add(new TraceCallable<>(tracing(), spanNamer(), task, this.beanName));
ts.add(traceCallableWhenContextReady(task));
}
}
return ts;

1
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceThreadPoolTaskExecutor.java
View File

@@ -29,6 +29,7 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.cloud.sleuth.SpanNamer;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.internal.ContextUtil;
import org.springframework.cloud.sleuth.internal.DefaultSpanNamer;
import org.springframework.core.task.TaskDecorator;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

60
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceThreadPoolTaskScheduler.java
View File

@@ -36,6 +36,7 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.cloud.sleuth.SpanNamer;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.internal.ContextUtil;
import org.springframework.cloud.sleuth.internal.DefaultSpanNamer;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.Trigger;
@@ -101,6 +102,20 @@ class LazyTraceThreadPoolTaskScheduler extends ThreadPoolTaskScheduler {
}
}
private Runnable traceRunnableWhenContextReady(Runnable delegate) {
if (ContextUtil.isContextUnusable(this.beanFactory)) {
return delegate;
}
return new TraceRunnable(tracing(), spanNamer(), delegate, this.beanName);
}
private <V> Callable<V> traceCallableWhenContextReady(Callable<V> delegate) {
if (ContextUtil.isContextUnusable(this.beanFactory)) {
return delegate;
}
return new TraceCallable<>(tracing(), spanNamer(), delegate, this.beanName);
}
@Override
public void setPoolSize(int poolSize) {
this.delegate.setPoolSize(poolSize);
@@ -177,38 +192,37 @@ class LazyTraceThreadPoolTaskScheduler extends ThreadPoolTaskScheduler {
@Override
public void execute(Runnable task) {
this.delegate.execute(new TraceRunnable(tracing(), spanNamer(), task, this.beanName));
this.delegate.execute(traceRunnableWhenContextReady(task));
}
@Override
public void execute(Runnable task, long startTimeout) {
this.delegate.execute(new TraceRunnable(tracing(), spanNamer(), task, this.beanName), startTimeout);
this.delegate.execute(traceRunnableWhenContextReady(task), startTimeout);
}
@Override
public Future<?> submit(Runnable task) {
return this.delegate.submit(new TraceRunnable(tracing(), spanNamer(), task, this.beanName));
return this.delegate.submit(traceRunnableWhenContextReady(task));
}
@Override
public <T> Future<T> submit(Callable<T> task) {
return this.delegate.submit(new TraceCallable<>(tracing(), spanNamer(), task, this.beanName));
return this.delegate.submit(traceCallableWhenContextReady(task));
}
@Override
public ListenableFuture<?> submitListenable(Runnable task) {
return this.delegate.submitListenable(new TraceRunnable(tracing(), spanNamer(), task, this.beanName));
return this.delegate.submitListenable(traceRunnableWhenContextReady(task));
}
@Override
public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
return this.delegate.submitListenable(new TraceCallable<>(tracing(), spanNamer(), task, this.beanName));
return this.delegate.submitListenable(traceCallableWhenContextReady(task));
}
@Override
public void cancelRemainingTask(Runnable task) {
ReflectionUtils.invokeMethod(this.cancelRemainingTask, this.delegate,
new TraceRunnable(tracing(), spanNamer(), task, this.beanName));
ReflectionUtils.invokeMethod(this.cancelRemainingTask, this.delegate, traceRunnableWhenContextReady(task));
}
@Override
@@ -219,36 +233,32 @@ class LazyTraceThreadPoolTaskScheduler extends ThreadPoolTaskScheduler {
@Override
@Nullable
public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
return this.delegate.schedule(new TraceRunnable(tracing(), spanNamer(), task, this.beanName), trigger);
return this.delegate.schedule(traceRunnableWhenContextReady(task), trigger);
}
@Override
public ScheduledFuture<?> schedule(Runnable task, Date startTime) {
return this.delegate.schedule(new TraceRunnable(tracing(), spanNamer(), task, this.beanName), startTime);
return this.delegate.schedule(traceRunnableWhenContextReady(task), startTime);
}
@Override
public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Date startTime, long period) {
return this.delegate.scheduleAtFixedRate(new TraceRunnable(tracing(), spanNamer(), task, this.beanName),
startTime, period);
return this.delegate.scheduleAtFixedRate(traceRunnableWhenContextReady(task), startTime, period);
}
@Override
public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, long period) {
return this.delegate.scheduleAtFixedRate(new TraceRunnable(tracing(), spanNamer(), task, this.beanName),
period);
return this.delegate.scheduleAtFixedRate(traceRunnableWhenContextReady(task), period);
}
@Override
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Date startTime, long delay) {
return this.delegate.scheduleWithFixedDelay(new TraceRunnable(tracing(), spanNamer(), task, this.beanName),
startTime, delay);
return this.delegate.scheduleWithFixedDelay(traceRunnableWhenContextReady(task), startTime, delay);
}
@Override
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, long delay) {
return this.delegate.scheduleWithFixedDelay(new TraceRunnable(tracing(), spanNamer(), task, this.beanName),
delay);
return this.delegate.scheduleWithFixedDelay(traceRunnableWhenContextReady(task), delay);
}
@Override
@@ -367,31 +377,27 @@ class LazyTraceThreadPoolTaskScheduler extends ThreadPoolTaskScheduler {
@Override
public ScheduledFuture<?> schedule(Runnable task, Instant startTime) {
return this.delegate.schedule(new TraceRunnable(tracing(), spanNamer(), task, this.beanName), startTime);
return this.delegate.schedule(traceRunnableWhenContextReady(task), startTime);
}
@Override
public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Instant startTime, Duration period) {
return this.delegate.scheduleAtFixedRate(new TraceRunnable(tracing(), spanNamer(), task, this.beanName),
startTime, period);
return this.delegate.scheduleAtFixedRate(traceRunnableWhenContextReady(task), startTime, period);
}
@Override
public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Duration period) {
return this.delegate.scheduleAtFixedRate(new TraceRunnable(tracing(), spanNamer(), task, this.beanName),
period);
return this.delegate.scheduleAtFixedRate(traceRunnableWhenContextReady(task), period);
}
@Override
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Instant startTime, Duration delay) {
return this.delegate.scheduleWithFixedDelay(new TraceRunnable(tracing(), spanNamer(), task, this.beanName),
startTime, delay);
return this.delegate.scheduleWithFixedDelay(traceRunnableWhenContextReady(task), startTime, delay);
}
@Override
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Duration delay) {
return this.delegate.scheduleWithFixedDelay(new TraceRunnable(tracing(), spanNamer(), task, this.beanName),
delay);
return this.delegate.scheduleWithFixedDelay(traceRunnableWhenContextReady(task), delay);
}
private Tracer tracing() {

1
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/TraceableExecutorService.java
View File

@@ -29,6 +29,7 @@ import java.util.concurrent.TimeoutException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.cloud.sleuth.SpanNamer;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.internal.ContextUtil;
/**
* A decorator class for {@link ExecutorService} to support tracing in Executors.

1
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/TraceableScheduledExecutorService.java
View File

@@ -23,6 +23,7 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.cloud.sleuth.internal.ContextUtil;
/**
* A decorator class for {@link ScheduledExecutorService} to support tracing in Executors.

52
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java
View File

@@ -16,16 +16,24 @@
package org.springframework.cloud.sleuth.instrument.messaging;
import java.util.Iterator;
import java.util.Map;
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.Tracer;
import org.springframework.cloud.sleuth.internal.SpanNameUtil;
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.AbstractMessageChannel;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.context.IntegrationObjectSupport;
@@ -40,6 +48,7 @@ import org.springframework.messaging.support.ExecutorChannelInterceptor;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;
/**
* This starts and propagates {@link Span.Kind#PRODUCER} span for each message sent (via
@@ -50,7 +59,8 @@ import org.springframework.util.ClassUtils;
* @author Marcin Grzejszczak
* @since 3.0.0
*/
public final class TracingChannelInterceptor extends ChannelInterceptorAdapter implements ExecutorChannelInterceptor {
public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
implements ExecutorChannelInterceptor, ApplicationContextAware {
/**
* Name of the class in Spring Cloud Stream that is a direct channel.
@@ -90,35 +100,37 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i
private final boolean hasDirectChannelClass;
private final boolean hasBinderTypeRegistry;
// special case of a Stream
private final Class<?> directWithAttributesChannelClass;
private ApplicationContext applicationContext;
private final Propagator propagator;
private final ThreadLocalSpan threadLocalSpan = new ThreadLocalSpan();
TracingChannelInterceptor(Tracer tracer, Propagator propagator) {
this(tracer, propagator, MessageHeaderPropagation.INSTANCE, MessageHeaderPropagation.INSTANCE);
}
private final Function<String, String> remoteServiceNameMapper;
public TracingChannelInterceptor(Tracer tracer, Propagator propagator,
Propagator.Setter<MessageHeaderAccessor> setter, Propagator.Getter<MessageHeaderAccessor> getter) {
Propagator.Setter<MessageHeaderAccessor> setter, Propagator.Getter<MessageHeaderAccessor> getter,
Function<String, String> remoteServiceNameMapper) {
this.tracer = tracer;
this.propagator = propagator;
this.injector = setter;
this.extractor = getter;
this.remoteServiceNameMapper = remoteServiceNameMapper;
this.integrationObjectSupportPresent = ClassUtils
.isPresent("org.springframework.integration.context.IntegrationObjectSupport", null);
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;
}
public static TracingChannelInterceptor create(Tracer tracer, Propagator propagator) {
return new TracingChannelInterceptor(tracer, propagator);
}
/**
* Use this to create a span for processing the given message. Note: the result has no
* name and is not started. This creates a child from identifiers extracted from the
@@ -182,11 +194,20 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i
private String toRemoteServiceName(MessageHeaderAccessor headers) {
for (String key : headers.getMessageHeaders().keySet()) {
if (key.startsWith("kafka_")) {
return "kafka";
String remoteServiceName = this.remoteServiceNameMapper.apply(key);
if (StringUtils.hasText(remoteServiceName)) {
return remoteServiceName;
}
else if (key.startsWith("amqp_")) {
return "rabbitmq";
}
if (this.hasBinderTypeRegistry && this.applicationContext != null) {
BinderTypeRegistry typeRegistry = this.applicationContext.getBean(BinderTypeRegistry.class);
Iterator<Map.Entry<String, BinderType>> iterator = typeRegistry.getAll().entrySet().iterator();
if (iterator.hasNext()) {
String binderName = iterator.next().getKey();
String remoteServiceName = this.remoteServiceNameMapper.apply(binderName);
if (StringUtils.hasText(remoteServiceName)) {
return remoteServiceName;
}
}
}
return REMOTE_SERVICE_NAME;
@@ -456,6 +477,11 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i
return message == null;
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
}
}
class SpanAndScope {

5
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java
View File

@@ -86,14 +86,11 @@ public abstract class ReactorSleuth {
}
if (!springContext.isActive()) {
boolean assertOn = false;
assert assertOn = true; // gives a message in unit test failures
if (log.isTraceEnabled() || assertOn) {
if (log.isTraceEnabled()) {
String message = "Spring Context [" + springContext
+ "] is not yet refreshed. This is unexpected. Reactor Context is [" + sub.currentContext()
+ "] and name is [" + name(sub) + "]";
log.trace(message);
assert false : message; // should never happen, but don't break.
}
return sub;
}

10
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/LazyTraceClientHttpRequestInterceptor.java
View File

@@ -20,6 +20,7 @@ import java.io.IOException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.cloud.sleuth.instrument.web.mvc.TracingClientHttpRequestInterceptor;
import org.springframework.cloud.sleuth.internal.ContextUtil;
import org.springframework.http.HttpRequest;
import org.springframework.http.client.ClientHttpRequestExecution;
import org.springframework.http.client.ClientHttpRequestInterceptor;
@@ -44,10 +45,17 @@ public class LazyTraceClientHttpRequestInterceptor implements ClientHttpRequestI
@Override
public ClientHttpResponse intercept(HttpRequest request, byte[] body, ClientHttpRequestExecution execution)
throws IOException {
if (isContextUnusable()) {
return execution.execute(request, body);
}
return interceptor().intercept(request, body, execution);
}
private TracingClientHttpRequestInterceptor interceptor() {
boolean isContextUnusable() {
return ContextUtil.isContextUnusable(this.beanFactory);
}
ClientHttpRequestInterceptor interceptor() {
if (this.interceptor == null) {
this.interceptor = this.beanFactory.getBean(TracingClientHttpRequestInterceptor.class);
}

15
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/ContextUtil.java → spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/internal/ContextUtil.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.sleuth.instrument.async;
package org.springframework.cloud.sleuth.internal;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -22,12 +22,12 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactory;
/**
* Utility class that verifies that context is in creation.
* Utility class that verifies that context is in creation. Do not use.
*
* @author Marcin Grzejszczak
* @since 2.1.0
*/
final class ContextUtil {
public final class ContextUtil {
private ContextUtil() {
throw new IllegalStateException("Can't instantiate a utility class");
@@ -35,8 +35,13 @@ final class ContextUtil {
private static final Log log = LogFactory.getLog(ContextUtil.class);
static boolean isContextUnusable(BeanFactory beanFactory) {
SleuthContextListener listener = SleuthContextListener.getBean(beanFactory);
/**
* @param beanFactory bean facotry
* @return {@code true} when context is not ready to be used
*/
public static boolean isContextUnusable(BeanFactory beanFactory) {
org.springframework.cloud.sleuth.internal.SleuthContextListener listener = org.springframework.cloud.sleuth.internal.SleuthContextListener
.getBean(beanFactory);
boolean contextUnusable = listener.isUnusable();
if (contextUnusable && log.isDebugEnabled()) {
log.debug("Context [" + Integer.toHexString(beanFactory.hashCode())

27
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/SleuthContextListener.java → spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/internal/SleuthContextListener.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.sleuth.instrument.async;
package org.springframework.cloud.sleuth.internal;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -33,10 +33,10 @@ import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.context.event.SmartApplicationListener;
/**
* Checks whether Spring Context is up and running.
* Internal tool used by Sleuth. Do not use.
*
* @author Marcin Grzejszczak
* @since 3.0.0
* @since 2.2.5
*/
public class SleuthContextListener implements SmartApplicationListener {
@@ -58,11 +58,18 @@ public class SleuthContextListener implements SmartApplicationListener {
this.closed = closed;
}
static SleuthContextListener getBean(BeanFactory beanFactory) {
if (beanFactory instanceof ConfigurableApplicationContext) {
beanFactory = ((ConfigurableApplicationContext) beanFactory).getBeanFactory();
/**
* Returns an instance of the {@link SleuthContextListener} that might have already
* been initialized.
* @param beanFactory bean factory
* @return instance of {@link SleuthContextListener}
*/
public static SleuthContextListener getBean(BeanFactory beanFactory) {
BeanFactory bf = beanFactory;
if (bf instanceof ConfigurableApplicationContext) {
bf = ((ConfigurableApplicationContext) bf).getBeanFactory();
}
return CACHE.getOrDefault(beanFactory, new SleuthContextListener());
return CACHE.getOrDefault(bf, new SleuthContextListener());
}
@Override
@@ -90,7 +97,11 @@ public class SleuthContextListener implements SmartApplicationListener {
}
}
boolean isUnusable() {
/**
* Verifies if context is unusable.
* @return true when Spring Context has NOT yet been started
*/
public boolean isUnusable() {
return !this.refreshed.get() || this.closed.get();
}

7
spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceScheduledThreadPoolExecutorTests.java
View File

@@ -87,7 +87,12 @@ public class LazyTraceScheduledThreadPoolExecutorTests {
doReturn(tracer).when(beanFactory).getBean(Tracer.class);
doReturn(parent).when(tracer).currentSpan();
doReturn(spanNamer).when(beanFactory).getBean(SpanNamer.class);
this.executor = spy(new LazyTraceScheduledThreadPoolExecutor(1, beanFactory, delegate, BEAN_NAME));
this.executor = spy(new LazyTraceScheduledThreadPoolExecutor(1, beanFactory, delegate, BEAN_NAME) {
@Override
boolean isContextUnusable() {
return false;
}
});
}
@Test

2
spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/async/SleuthContextListenerTest.java → spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/internal/SleuthContextListenerTest.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.sleuth.instrument.async;
package org.springframework.cloud.sleuth.internal;
import org.assertj.core.api.BDDAssertions;
import org.junit.jupiter.api.Test;

1
tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/LazyTraceThreadPoolTaskSchedulerTests.java
View File

@@ -35,6 +35,7 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.cloud.sleuth.SpanNamer;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.internal.DefaultSpanNamer;
import org.springframework.cloud.sleuth.internal.SleuthContextListenerAccessor;
import org.springframework.cloud.sleuth.test.TestTracingAwareSupplier;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

2
tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/SleuthContextListenerAccessor.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.sleuth.instrument.async;
package org.springframework.cloud.sleuth.internal;
import java.util.concurrent.atomic.AtomicBoolean;

1
tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/TraceableExecutorServiceTests.java
View File

@@ -46,6 +46,7 @@ import org.springframework.cloud.sleuth.SpanNamer;
import org.springframework.cloud.sleuth.TraceContext;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.internal.DefaultSpanNamer;
import org.springframework.cloud.sleuth.internal.SleuthContextListenerAccessor;
import org.springframework.cloud.sleuth.test.TestTracingAwareSupplier;
import static java.util.stream.Collectors.toList;

1
tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/async/TraceableScheduledExecutorServiceTest.java
View File

@@ -36,6 +36,7 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.cloud.sleuth.SpanNamer;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.internal.DefaultSpanNamer;
import org.springframework.cloud.sleuth.internal.SleuthContextListenerAccessor;
import org.springframework.cloud.sleuth.test.TestTracingAwareSupplier;
/**

28
tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java
View File

@@ -20,12 +20,15 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import org.assertj.core.api.BDDAssertions;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.autoconfig.instrument.messaging.SleuthMessagingProperties;
import org.springframework.cloud.sleuth.autoconfig.instrument.messaging.TraceSpringIntegrationAutoConfiguration;
import org.springframework.cloud.sleuth.exporter.FinishedSpan;
import org.springframework.cloud.sleuth.test.TestSpanHandler;
import org.springframework.cloud.sleuth.test.TestTracingAwareSupplier;
@@ -42,14 +45,37 @@ import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.ExecutorChannelInterceptor;
import org.springframework.messaging.support.ExecutorSubscribableChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.StringUtils;
import static org.assertj.core.api.Assertions.assertThat;
import static org.springframework.messaging.support.NativeMessageHeaderAccessor.NATIVE_HEADERS;
public abstract class TracingChannelInterceptorTest implements TestTracingAwareSupplier {
/**
* Align with
* {@link TraceSpringIntegrationAutoConfiguration#remoteServiceNameMapper(SleuthMessagingProperties)}.
* @param properties messaging properties
* @return function mapping remote service name
*/
static Function<String, String> remoteServiceNameMapper(SleuthMessagingProperties properties) {
return s -> {
if (!StringUtils.hasText(s)) {
return null;
}
if (s.startsWith("amqp") || s.startsWith("rabbit")) {
return properties.getMessaging().getRabbit().getRemoteServiceName();
}
else if (s.startsWith("kafka")) {
return properties.getMessaging().getKafka().getRemoteServiceName();
}
return null;
};
}
protected ChannelInterceptor interceptor = new TracingChannelInterceptor(tracerTest().tracing().tracer(),
tracerTest().tracing().propagator());
tracerTest().tracing().propagator(), MessageHeaderPropagation.INSTANCE, MessageHeaderPropagation.INSTANCE,
remoteServiceNameMapper(new SleuthMessagingProperties()));
protected TestSpanHandler spans = tracerTest().handler();