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

Added spring-cloud-task support

Browse Source 
fixes gh-1903
This commit is contained in:
Marcin Grzejszczak
2021-04-21 10:11:28 +02:00
parent 6e8f86ed35
commit 2fca60a112
22 changed files with 792 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -18,8 +18,6 @@ package org.springframework.cloud.sleuth.instrument.messaging;
import java.util.Iterator;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.function.Function;
import org.apache.commons.logging.Log;
@@ -28,6 +26,8 @@ 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.SpanAndScope;
import org.springframework.cloud.sleuth.ThreadLocalSpan;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.propagation.Propagator;
import org.springframework.cloud.stream.binder.BinderType;
@@ -107,7 +107,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
private final Propagator propagator;
private final ThreadLocalSpan threadLocalSpan = new ThreadLocalSpan();
private final ThreadLocalSpan threadLocalSpan;
private final Function<String, String> remoteServiceNameMapper;
@@ -115,6 +115,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
Propagator.Setter<MessageHeaderAccessor> setter, Propagator.Getter<MessageHeaderAccessor> getter,
Function<String, String> remoteServiceNameMapper, MessageSpanCustomizer messageSpanCustomizer) {
this.tracer = tracer;
this.threadLocalSpan = new ThreadLocalSpan(tracer);
this.propagator = propagator;
this.injector = setter;
this.extractor = getter;
@@ -163,8 +164,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
}
private void setSpanInScope(Span span) {
Tracer.SpanInScope spanInScope = this.tracer.withSpan(span);
this.threadLocalSpan.set(new SpanAndScope(span, spanInScope));
this.threadLocalSpan.set(span);
if (log.isDebugEnabled()) {
log.debug("Put span in scope " + span);
}
@@ -362,8 +362,8 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
if (spanAndScope == null) {
return;
}
Span span = spanAndScope.span;
Tracer.SpanInScope scope = spanAndScope.scope;
Span span = spanAndScope.getSpan();
Tracer.SpanInScope scope = spanAndScope.getScope();
if (span.isNoop()) {
if (log.isDebugEnabled()) {
log.debug("Span " + span + " is noop - will stope the scope");
@@ -424,57 +424,3 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
}
}
class SpanAndScope {
final Span span;
final Tracer.SpanInScope scope;
SpanAndScope(Span span, Tracer.SpanInScope scope) {
this.span = span;
this.scope = scope;
}
}
class ThreadLocalSpan {
private static final Log log = LogFactory.getLog(ThreadLocalSpan.class);
final ThreadLocal<SpanAndScope> threadLocalSpan = new ThreadLocal<>();
final LinkedBlockingDeque<SpanAndScope> spans = new LinkedBlockingDeque<>();
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();
if (log.isDebugEnabled()) {
log.debug("Took span [" + span + "] from thread local");
}
this.threadLocalSpan.set(span);
}
catch (NoSuchElementException ex) {
if (log.isTraceEnabled()) {
log.trace("Failed to remove a span from the queue", ex);
}
}
}
}

65
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/task/TraceApplicationRunner.java Normal file
View File

@@ -0,0 +1,65 @@
/*
* Copyright 2018-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.task;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.Tracer;
/**
* Trace representation of a {@link ApplicationRunner}.
*
* @author Marcin Grzejszczak
* @since 3.1.0
*/
public class TraceApplicationRunner implements ApplicationRunner {
private final BeanFactory beanFactory;
private final ApplicationRunner delegate;
private final String beanName;
private Tracer tracer;
public TraceApplicationRunner(BeanFactory beanFactory, ApplicationRunner delegate, String beanName) {
this.beanFactory = beanFactory;
this.delegate = delegate;
this.beanName = beanName;
}
@Override
public void run(ApplicationArguments args) throws Exception {
Span span = tracer().nextSpan().name(this.beanName);
try (Tracer.SpanInScope spanInScope = tracer().withSpan(span.start())) {
this.delegate.run(args);
}
finally {
span.end();
}
}
private Tracer tracer() {
if (this.tracer == null) {
this.tracer = this.beanFactory.getBean(Tracer.class);
}
return this.tracer;
}
}

64
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/task/TraceCommandLineRunner.java Normal file
View File

@@ -0,0 +1,64 @@
/*
* Copyright 2018-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.task;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.Tracer;
/**
* Trace representation of a {@link CommandLineRunner}.
*
* @author Marcin Grzejszczak
* @since 3.1.0
*/
public class TraceCommandLineRunner implements CommandLineRunner {
private final BeanFactory beanFactory;
private final CommandLineRunner delegate;
private final String beanName;
private Tracer tracer;
public TraceCommandLineRunner(BeanFactory beanFactory, CommandLineRunner delegate, String beanName) {
this.beanFactory = beanFactory;
this.delegate = delegate;
this.beanName = beanName;
}
@Override
public void run(String... args) throws Exception {
Span span = tracer().nextSpan().name(this.beanName);
try (Tracer.SpanInScope spanInScope = tracer().withSpan(span.start())) {
this.delegate.run(args);
}
finally {
span.end();
}
}
private Tracer tracer() {
if (this.tracer == null) {
this.tracer = this.beanFactory.getBean(Tracer.class);
}
return this.tracer;
}
}

89
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/task/TraceTaskExecutionListener.java Normal file
View File

@@ -0,0 +1,89 @@
/*
* Copyright 2018-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.task;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.SpanAndScope;
import org.springframework.cloud.sleuth.ThreadLocalSpan;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.task.listener.TaskExecutionListener;
import org.springframework.cloud.task.repository.TaskExecution;
import org.springframework.core.Ordered;
/**
* Sets the span upon starting and closes it upon ending a task.
*
* @author Marcin Grzejszczak
* @since 3.1.0
*/
public class TraceTaskExecutionListener implements TaskExecutionListener, Ordered {
private static final Log log = LogFactory.getLog(TraceTaskExecutionListener.class);
private final Tracer tracer;
private final ThreadLocalSpan threadLocalSpan;
private final String projectName;
public TraceTaskExecutionListener(Tracer tracer, String projectName) {
this.tracer = tracer;
this.threadLocalSpan = new ThreadLocalSpan(tracer);
this.projectName = projectName;
}
@Override
public void onTaskStartup(TaskExecution taskExecution) {
Span span = this.tracer.nextSpan().name(this.projectName).start();
this.threadLocalSpan.set(span);
if (log.isDebugEnabled()) {
log.debug("Put the span [" + span + "] to thread local");
}
}
@Override
public void onTaskEnd(TaskExecution taskExecution) {
SpanAndScope spanAndScope = this.threadLocalSpan.get();
Span span = spanAndScope.getSpan();
span.end();
spanAndScope.getScope().close();
if (log.isDebugEnabled()) {
log.debug("Removed the [" + span + "] from thread local");
}
}
@Override
public void onTaskFailed(TaskExecution taskExecution, Throwable throwable) {
SpanAndScope spanAndScope = this.threadLocalSpan.get();
Span span = spanAndScope.getSpan();
span.error(throwable);
span.end();
spanAndScope.getScope().close();
if (log.isDebugEnabled()) {
log.debug("Removed the [" + span + "] from thread local and added error");
}
}
@Override
public int getOrder() {
return Ordered.HIGHEST_PRECEDENCE;
}
}