Fixed the build
This commit is contained in:
Marcin Grzejszczak
@@ -1,173 +1,175 @@
|
||||
/*
|
||||
* 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.tx;
|
||||
|
||||
import javax.annotation.PostConstruct;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
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.transaction.PlatformTransactionManager;
|
||||
import org.springframework.transaction.TransactionDefinition;
|
||||
import org.springframework.transaction.TransactionException;
|
||||
import org.springframework.transaction.TransactionStatus;
|
||||
|
||||
/**
|
||||
* A trace representation of a {@link PlatformTransactionManager}.
|
||||
*
|
||||
* @author Marcin Grzejszczak
|
||||
* @since 3.1.0
|
||||
*/
|
||||
public class TracePlatformTransactionManager implements PlatformTransactionManager {
|
||||
|
||||
private static final Log log = LogFactory.getLog(TracePlatformTransactionManager.class);
|
||||
|
||||
private final PlatformTransactionManager delegate;
|
||||
|
||||
private final BeanFactory beanFactory;
|
||||
|
||||
private Tracer tracer;
|
||||
|
||||
volatile ThreadLocalSpan threadLocalSpan;
|
||||
|
||||
public TracePlatformTransactionManager(PlatformTransactionManager delegate, BeanFactory beanFactory) {
|
||||
this.delegate = delegate;
|
||||
this.beanFactory = beanFactory;
|
||||
}
|
||||
|
||||
@PostConstruct
|
||||
void initialize() {
|
||||
if (this.threadLocalSpan == null) {
|
||||
this.threadLocalSpan = new ThreadLocalSpan(tracer());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public TransactionStatus getTransaction(TransactionDefinition definition) throws TransactionException {
|
||||
initialize();
|
||||
SpanAndScope spanAndScope = this.threadLocalSpan.get();
|
||||
Span currentSpan = spanAndScope != null ? spanAndScope.getSpan() : tracer().currentSpan();
|
||||
Span span = fallbackSpan();
|
||||
try {
|
||||
TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());
|
||||
TransactionStatus status = this.delegate.getTransaction(definition);
|
||||
span = taggedSpan(currentSpan, span, def, status);
|
||||
return status;
|
||||
}
|
||||
catch (Exception e) {
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug(
|
||||
"Exception occurred while trying to get a transaction, will mark the span with error and report it");
|
||||
}
|
||||
span.error(e);
|
||||
span.end();
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
Span fallbackSpan() {
|
||||
return tracer().nextSpan().name("tx").start();
|
||||
}
|
||||
|
||||
private Span taggedSpan(Span currentSpan, Span span, TransactionDefinition def, TransactionStatus status) {
|
||||
if (status.isNewTransaction() || currentSpan == null) {
|
||||
log.info("Creating new span cause a new transaction is started");
|
||||
TracePlatformTransactionManagerTags.tag(span, def, this.delegate.getClass());
|
||||
}
|
||||
else {
|
||||
span = currentSpan;
|
||||
}
|
||||
this.threadLocalSpan.set(span);
|
||||
return span;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void commit(TransactionStatus status) throws TransactionException {
|
||||
SpanAndScope spanAndScope = this.threadLocalSpan.get();
|
||||
if (spanAndScope == null) {
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("No span and scope found - this shouldn't happen, sth is wrong");
|
||||
}
|
||||
this.delegate.commit(status);
|
||||
return;
|
||||
}
|
||||
Exception ex = null;
|
||||
Span span = spanAndScope.getSpan();
|
||||
try {
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("Wrapping commit");
|
||||
}
|
||||
this.delegate.commit(status);
|
||||
}
|
||||
catch (Exception e) {
|
||||
ex = e;
|
||||
span.error(e);
|
||||
throw e;
|
||||
}
|
||||
finally {
|
||||
span.event("tx commit");
|
||||
span.end();
|
||||
if (ex == null) {
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("No exception was found - will clear thread local span");
|
||||
}
|
||||
this.threadLocalSpan.remove();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void rollback(TransactionStatus status) throws TransactionException {
|
||||
SpanAndScope spanAndScope = this.threadLocalSpan.get();
|
||||
if (spanAndScope == null) {
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("No span and scope found - this shouldn't happen, sth is wrong");
|
||||
}
|
||||
this.delegate.rollback(status);
|
||||
return;
|
||||
}
|
||||
Span span = spanAndScope.getSpan();
|
||||
try {
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("Wrapping rollback");
|
||||
}
|
||||
this.delegate.rollback(status);
|
||||
}
|
||||
catch (Exception e) {
|
||||
span.error(e);
|
||||
throw e;
|
||||
}
|
||||
finally {
|
||||
span.event("tx rollback");
|
||||
span.end();
|
||||
this.threadLocalSpan.remove();
|
||||
}
|
||||
}
|
||||
|
||||
private Tracer tracer() {
|
||||
if (this.tracer == null) {
|
||||
this.tracer = this.beanFactory.getBean(Tracer.class);
|
||||
}
|
||||
return this.tracer;
|
||||
}
|
||||
|
||||
}
|
||||
/*
|
||||
* 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.tx;
|
||||
|
||||
import javax.annotation.PostConstruct;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
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.transaction.PlatformTransactionManager;
|
||||
import org.springframework.transaction.TransactionDefinition;
|
||||
import org.springframework.transaction.TransactionException;
|
||||
import org.springframework.transaction.TransactionStatus;
|
||||
|
||||
/**
|
||||
* A trace representation of a {@link PlatformTransactionManager}.
|
||||
*
|
||||
* @author Marcin Grzejszczak
|
||||
* @since 3.1.0
|
||||
*/
|
||||
public class TracePlatformTransactionManager implements PlatformTransactionManager {
|
||||
|
||||
private static final Log log = LogFactory.getLog(TracePlatformTransactionManager.class);
|
||||
|
||||
private final PlatformTransactionManager delegate;
|
||||
|
||||
private final BeanFactory beanFactory;
|
||||
|
||||
private Tracer tracer;
|
||||
|
||||
volatile ThreadLocalSpan threadLocalSpan;
|
||||
|
||||
public TracePlatformTransactionManager(PlatformTransactionManager delegate, BeanFactory beanFactory) {
|
||||
this.delegate = delegate;
|
||||
this.beanFactory = beanFactory;
|
||||
}
|
||||
|
||||
@PostConstruct
|
||||
void initialize() {
|
||||
if (this.threadLocalSpan == null) {
|
||||
this.threadLocalSpan = new ThreadLocalSpan(tracer());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public TransactionStatus getTransaction(TransactionDefinition definition) throws TransactionException {
|
||||
initialize();
|
||||
SpanAndScope spanAndScope = this.threadLocalSpan.get();
|
||||
Span currentSpan = spanAndScope != null ? spanAndScope.getSpan() : tracer().currentSpan();
|
||||
Span span = fallbackSpan();
|
||||
try {
|
||||
TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());
|
||||
TransactionStatus status = this.delegate.getTransaction(definition);
|
||||
span = taggedSpan(currentSpan, span, def, status);
|
||||
return status;
|
||||
}
|
||||
catch (Exception e) {
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug(
|
||||
"Exception occurred while trying to get a transaction, will mark the span with error and report it");
|
||||
}
|
||||
span.error(e);
|
||||
span.end();
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
Span fallbackSpan() {
|
||||
return tracer().nextSpan().name("tx").start();
|
||||
}
|
||||
|
||||
private Span taggedSpan(Span currentSpan, Span span, TransactionDefinition def, TransactionStatus status) {
|
||||
if (status.isNewTransaction() || currentSpan == null) {
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("Creating new span cause a new transaction is started");
|
||||
}
|
||||
TracePlatformTransactionManagerTags.tag(span, def, this.delegate.getClass());
|
||||
}
|
||||
else {
|
||||
span = currentSpan;
|
||||
}
|
||||
this.threadLocalSpan.set(span);
|
||||
return span;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void commit(TransactionStatus status) throws TransactionException {
|
||||
SpanAndScope spanAndScope = this.threadLocalSpan.get();
|
||||
if (spanAndScope == null) {
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("No span and scope found - this shouldn't happen, sth is wrong");
|
||||
}
|
||||
this.delegate.commit(status);
|
||||
return;
|
||||
}
|
||||
Exception ex = null;
|
||||
Span span = spanAndScope.getSpan();
|
||||
try {
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("Wrapping commit");
|
||||
}
|
||||
this.delegate.commit(status);
|
||||
}
|
||||
catch (Exception e) {
|
||||
ex = e;
|
||||
span.error(e);
|
||||
throw e;
|
||||
}
|
||||
finally {
|
||||
span.event("tx.commit");
|
||||
span.end();
|
||||
if (ex == null) {
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("No exception was found - will clear thread local span");
|
||||
}
|
||||
this.threadLocalSpan.remove();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void rollback(TransactionStatus status) throws TransactionException {
|
||||
SpanAndScope spanAndScope = this.threadLocalSpan.get();
|
||||
if (spanAndScope == null) {
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("No span and scope found - this shouldn't happen, sth is wrong");
|
||||
}
|
||||
this.delegate.rollback(status);
|
||||
return;
|
||||
}
|
||||
Span span = spanAndScope.getSpan();
|
||||
try {
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("Wrapping rollback");
|
||||
}
|
||||
this.delegate.rollback(status);
|
||||
}
|
||||
catch (Exception e) {
|
||||
span.error(e);
|
||||
throw e;
|
||||
}
|
||||
finally {
|
||||
span.event("tx.rollback");
|
||||
span.end();
|
||||
this.threadLocalSpan.remove();
|
||||
}
|
||||
}
|
||||
|
||||
private Tracer tracer() {
|
||||
if (this.tracer == null) {
|
||||
this.tracer = this.beanFactory.getBean(Tracer.class);
|
||||
}
|
||||
return this.tracer;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,189 +1,187 @@
|
||||
/*
|
||||
* 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.tx;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
import org.springframework.cloud.sleuth.CurrentTraceContext;
|
||||
import org.springframework.cloud.sleuth.Span;
|
||||
import org.springframework.cloud.sleuth.SpanAndScope;
|
||||
import org.springframework.cloud.sleuth.TraceContext;
|
||||
import org.springframework.cloud.sleuth.Tracer;
|
||||
import org.springframework.transaction.ReactiveTransaction;
|
||||
import org.springframework.transaction.ReactiveTransactionManager;
|
||||
import org.springframework.transaction.TransactionDefinition;
|
||||
import org.springframework.transaction.TransactionException;
|
||||
|
||||
/**
|
||||
* A trace representation of a {@link ReactiveTransactionManager}.
|
||||
*
|
||||
* @author Marcin Grzejszczak
|
||||
* @since 3.1.0
|
||||
*/
|
||||
public class TraceReactiveTransactionManager implements ReactiveTransactionManager {
|
||||
|
||||
private static final Log log = LogFactory.getLog(TraceReactiveTransactionManager.class);
|
||||
|
||||
private final ReactiveTransactionManager delegate;
|
||||
|
||||
private final BeanFactory beanFactory;
|
||||
|
||||
private Tracer tracer;
|
||||
|
||||
private CurrentTraceContext currentTraceContext;
|
||||
|
||||
public TraceReactiveTransactionManager(ReactiveTransactionManager delegate, BeanFactory beanFactory) {
|
||||
this.delegate = delegate;
|
||||
this.beanFactory = beanFactory;
|
||||
}
|
||||
|
||||
private Tracer tracer() {
|
||||
if (this.tracer == null) {
|
||||
this.tracer = this.beanFactory.getBean(Tracer.class);
|
||||
}
|
||||
return this.tracer;
|
||||
}
|
||||
|
||||
private CurrentTraceContext currentTraceContext() {
|
||||
if (this.currentTraceContext == null) {
|
||||
this.currentTraceContext = this.beanFactory.getBean(CurrentTraceContext.class);
|
||||
}
|
||||
return this.currentTraceContext;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Mono<ReactiveTransaction> getReactiveTransaction(TransactionDefinition definition)
|
||||
throws TransactionException {
|
||||
return Mono.deferContextual(contextView -> {
|
||||
return this.delegate.getReactiveTransaction(definition).map(tx -> {
|
||||
Span span = span(contextView);
|
||||
if (tx.isNewTransaction() || span == null) {
|
||||
if (span == null) {
|
||||
span = tracer().nextSpan().name("tx").start();
|
||||
}
|
||||
else {
|
||||
span = tracer().nextSpan(span).name("tx").start();
|
||||
}
|
||||
TracePlatformTransactionManagerTags.tag(span, definition, this.delegate.getClass());
|
||||
}
|
||||
Tracer.SpanInScope withSpan = tracer().withSpan(span);
|
||||
SpanAndScope spanAndScope = new SpanAndScope(span, withSpan);
|
||||
return new TraceReactiveTransaction(tx, spanAndScope);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
private Span span(reactor.util.context.ContextView contextView) {
|
||||
Span span = contextView.getOrDefault(Span.class, null);
|
||||
if (span == null) {
|
||||
TraceContext traceContext = contextView.getOrDefault(TraceContext.class, null);
|
||||
if (traceContext == null) {
|
||||
Span currentSpan = tracer().currentSpan();
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("There's no Span or TraceContext in the reactor context. Current span is [" + currentSpan
|
||||
+ "]");
|
||||
}
|
||||
span = currentSpan;
|
||||
}
|
||||
else {
|
||||
span = spanFromContext(traceContext);
|
||||
}
|
||||
}
|
||||
return span;
|
||||
}
|
||||
|
||||
private Span spanFromContext(TraceContext traceContext) {
|
||||
try (CurrentTraceContext.Scope scope = currentTraceContext().maybeScope(traceContext)) {
|
||||
return tracer().currentSpan();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Mono<Void> commit(ReactiveTransaction transaction) throws TransactionException {
|
||||
if (!(transaction instanceof TraceReactiveTransaction)) {
|
||||
return this.delegate.commit(transaction);
|
||||
}
|
||||
TraceReactiveTransaction reactiveTransaction = (TraceReactiveTransaction) transaction;
|
||||
SpanAndScope spanAndScope = reactiveTransaction.spanAndScope;
|
||||
Span span = spanAndScope.getSpan();
|
||||
Tracer.SpanInScope scope = spanAndScope.getScope();
|
||||
return this.delegate.commit(reactiveTransaction.delegate)
|
||||
// TODO: Fix me when this is resolved in Reactor
|
||||
// .doOnSubscribe(__ -> scope.close())
|
||||
.doOnError(span::error).doOnSuccess(signalType -> {
|
||||
span.end();
|
||||
scope.close();
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Mono<Void> rollback(ReactiveTransaction transaction) throws TransactionException {
|
||||
if (!(transaction instanceof TraceReactiveTransaction)) {
|
||||
return this.delegate.rollback(transaction);
|
||||
}
|
||||
TraceReactiveTransaction reactiveTransaction = (TraceReactiveTransaction) transaction;
|
||||
SpanAndScope spanAndScope = reactiveTransaction.spanAndScope;
|
||||
Span span = spanAndScope.getSpan();
|
||||
Tracer.SpanInScope scope = spanAndScope.getScope();
|
||||
return this.delegate.rollback(reactiveTransaction.delegate)
|
||||
// TODO: Fix me when this is resolved in Reactor
|
||||
// .doOnSubscribe(__ -> scope.close())
|
||||
.doOnError(span::error).doFinally(signalType -> {
|
||||
span.end();
|
||||
if (scope != null) {
|
||||
scope.close();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
static class TraceReactiveTransaction implements ReactiveTransaction {
|
||||
|
||||
final ReactiveTransaction delegate;
|
||||
|
||||
final SpanAndScope spanAndScope;
|
||||
|
||||
TraceReactiveTransaction(ReactiveTransaction delegate, SpanAndScope spanAndScope) {
|
||||
this.delegate = delegate;
|
||||
this.spanAndScope = spanAndScope;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isNewTransaction() {
|
||||
return this.delegate.isNewTransaction();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setRollbackOnly() {
|
||||
this.delegate.setRollbackOnly();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isRollbackOnly() {
|
||||
return this.delegate.isRollbackOnly();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isCompleted() {
|
||||
return this.delegate.isCompleted();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
/*
|
||||
* 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.tx;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
import org.springframework.cloud.sleuth.CurrentTraceContext;
|
||||
import org.springframework.cloud.sleuth.Span;
|
||||
import org.springframework.cloud.sleuth.SpanAndScope;
|
||||
import org.springframework.cloud.sleuth.TraceContext;
|
||||
import org.springframework.cloud.sleuth.Tracer;
|
||||
import org.springframework.transaction.ReactiveTransaction;
|
||||
import org.springframework.transaction.ReactiveTransactionManager;
|
||||
import org.springframework.transaction.TransactionDefinition;
|
||||
import org.springframework.transaction.TransactionException;
|
||||
|
||||
/**
|
||||
* A trace representation of a {@link ReactiveTransactionManager}.
|
||||
*
|
||||
* @author Marcin Grzejszczak
|
||||
* @since 3.1.0
|
||||
*/
|
||||
public class TraceReactiveTransactionManager implements ReactiveTransactionManager {
|
||||
|
||||
private static final Log log = LogFactory.getLog(TraceReactiveTransactionManager.class);
|
||||
|
||||
private final ReactiveTransactionManager delegate;
|
||||
|
||||
private final BeanFactory beanFactory;
|
||||
|
||||
private Tracer tracer;
|
||||
|
||||
private CurrentTraceContext currentTraceContext;
|
||||
|
||||
public TraceReactiveTransactionManager(ReactiveTransactionManager delegate, BeanFactory beanFactory) {
|
||||
this.delegate = delegate;
|
||||
this.beanFactory = beanFactory;
|
||||
}
|
||||
|
||||
private Tracer tracer() {
|
||||
if (this.tracer == null) {
|
||||
this.tracer = this.beanFactory.getBean(Tracer.class);
|
||||
}
|
||||
return this.tracer;
|
||||
}
|
||||
|
||||
private CurrentTraceContext currentTraceContext() {
|
||||
if (this.currentTraceContext == null) {
|
||||
this.currentTraceContext = this.beanFactory.getBean(CurrentTraceContext.class);
|
||||
}
|
||||
return this.currentTraceContext;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Mono<ReactiveTransaction> getReactiveTransaction(TransactionDefinition definition)
|
||||
throws TransactionException {
|
||||
return Mono.deferContextual(contextView -> this.delegate.getReactiveTransaction(definition).map(tx -> {
|
||||
Span span = span(contextView);
|
||||
if (tx.isNewTransaction() || span == null) {
|
||||
if (span == null) {
|
||||
span = tracer().nextSpan().name("tx").start();
|
||||
}
|
||||
else {
|
||||
span = tracer().nextSpan(span).name("tx").start();
|
||||
}
|
||||
TracePlatformTransactionManagerTags.tag(span, definition, this.delegate.getClass());
|
||||
}
|
||||
Tracer.SpanInScope withSpan = tracer().withSpan(span);
|
||||
SpanAndScope spanAndScope = new SpanAndScope(span, withSpan);
|
||||
return new TraceReactiveTransaction(tx, spanAndScope);
|
||||
}));
|
||||
}
|
||||
|
||||
private Span span(reactor.util.context.ContextView contextView) {
|
||||
Span span = contextView.getOrDefault(Span.class, null);
|
||||
if (span == null) {
|
||||
TraceContext traceContext = contextView.getOrDefault(TraceContext.class, null);
|
||||
if (traceContext == null) {
|
||||
Span currentSpan = tracer().currentSpan();
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("There's no Span or TraceContext in the reactor context. Current span is [" + currentSpan
|
||||
+ "]");
|
||||
}
|
||||
span = currentSpan;
|
||||
}
|
||||
else {
|
||||
span = spanFromContext(traceContext);
|
||||
}
|
||||
}
|
||||
return span;
|
||||
}
|
||||
|
||||
private Span spanFromContext(TraceContext traceContext) {
|
||||
try (CurrentTraceContext.Scope scope = currentTraceContext().maybeScope(traceContext)) {
|
||||
return tracer().currentSpan();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Mono<Void> commit(ReactiveTransaction transaction) throws TransactionException {
|
||||
if (!(transaction instanceof TraceReactiveTransaction)) {
|
||||
return this.delegate.commit(transaction);
|
||||
}
|
||||
TraceReactiveTransaction reactiveTransaction = (TraceReactiveTransaction) transaction;
|
||||
SpanAndScope spanAndScope = reactiveTransaction.spanAndScope;
|
||||
Span span = spanAndScope.getSpan();
|
||||
Tracer.SpanInScope scope = spanAndScope.getScope();
|
||||
return this.delegate.commit(reactiveTransaction.delegate)
|
||||
// TODO: Fix me when this is resolved in Reactor
|
||||
// .doOnSubscribe(__ -> scope.close())
|
||||
.doOnError(span::error).doOnSuccess(signalType -> {
|
||||
span.end();
|
||||
scope.close();
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Mono<Void> rollback(ReactiveTransaction transaction) throws TransactionException {
|
||||
if (!(transaction instanceof TraceReactiveTransaction)) {
|
||||
return this.delegate.rollback(transaction);
|
||||
}
|
||||
TraceReactiveTransaction reactiveTransaction = (TraceReactiveTransaction) transaction;
|
||||
SpanAndScope spanAndScope = reactiveTransaction.spanAndScope;
|
||||
Span span = spanAndScope.getSpan();
|
||||
Tracer.SpanInScope scope = spanAndScope.getScope();
|
||||
return this.delegate.rollback(reactiveTransaction.delegate)
|
||||
// TODO: Fix me when this is resolved in Reactor
|
||||
// .doOnSubscribe(__ -> scope.close())
|
||||
.doOnError(span::error).doFinally(signalType -> {
|
||||
span.end();
|
||||
if (scope != null) {
|
||||
scope.close();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
static class TraceReactiveTransaction implements ReactiveTransaction {
|
||||
|
||||
final ReactiveTransaction delegate;
|
||||
|
||||
final SpanAndScope spanAndScope;
|
||||
|
||||
TraceReactiveTransaction(ReactiveTransaction delegate, SpanAndScope spanAndScope) {
|
||||
this.delegate = delegate;
|
||||
this.spanAndScope = spanAndScope;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isNewTransaction() {
|
||||
return this.delegate.isNewTransaction();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setRollbackOnly() {
|
||||
this.delegate.setRollbackOnly();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isRollbackOnly() {
|
||||
return this.delegate.isRollbackOnly();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isCompleted() {
|
||||
return this.delegate.isCompleted();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,105 +1,113 @@
|
||||
/*
|
||||
* 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.tracer;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import org.springframework.cloud.sleuth.Span;
|
||||
import org.springframework.cloud.sleuth.TraceContext;
|
||||
|
||||
/**
|
||||
* A noop implementation. Does nothing.
|
||||
*
|
||||
* @author Marcin Grzejszczak
|
||||
* @since 3.0.0
|
||||
*/
|
||||
class SimpleSpanBuilder implements Span.Builder {
|
||||
|
||||
List<String> events = new ArrayList<>();
|
||||
|
||||
Map<String, String> tags = new HashMap<>();
|
||||
|
||||
Throwable error;
|
||||
|
||||
Span.Kind spanKind;
|
||||
|
||||
String remoteServiceName;
|
||||
|
||||
String name;
|
||||
|
||||
@Override
|
||||
public Span.Builder setParent(TraceContext context) {
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Span.Builder setNoParent() {
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Span.Builder name(String name) {
|
||||
this.name = name;
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Span.Builder event(String value) {
|
||||
this.events.add(value);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Span.Builder tag(String key, String value) {
|
||||
this.tags.put(key, value);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Span.Builder error(Throwable throwable) {
|
||||
this.error = throwable;
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Span.Builder kind(Span.Kind spanKind) {
|
||||
this.spanKind = spanKind;
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Span.Builder remoteServiceName(String remoteServiceName) {
|
||||
this.remoteServiceName = remoteServiceName;
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Span start() {
|
||||
SimpleSpan span = new SimpleSpan();
|
||||
this.tags.forEach(span::tag);
|
||||
this.events.forEach(span::event);
|
||||
span.remoteServiceName(this.remoteServiceName);
|
||||
span.error(this.error);
|
||||
span.spanKind = this.spanKind;
|
||||
span.name(this.name);
|
||||
return span;
|
||||
}
|
||||
|
||||
}
|
||||
/*
|
||||
* 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.tracer;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import org.springframework.cloud.sleuth.Span;
|
||||
import org.springframework.cloud.sleuth.TraceContext;
|
||||
|
||||
/**
|
||||
* A noop implementation. Does nothing.
|
||||
*
|
||||
* @author Marcin Grzejszczak
|
||||
* @since 3.0.0
|
||||
*/
|
||||
class SimpleSpanBuilder implements Span.Builder {
|
||||
|
||||
List<String> events = new ArrayList<>();
|
||||
|
||||
Map<String, String> tags = new HashMap<>();
|
||||
|
||||
Throwable error;
|
||||
|
||||
Span.Kind spanKind;
|
||||
|
||||
String remoteServiceName;
|
||||
|
||||
String name;
|
||||
|
||||
SimpleTracer simpleTracer;
|
||||
|
||||
public SimpleSpanBuilder(SimpleTracer simpleTracer) {
|
||||
this.simpleTracer = simpleTracer;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Span.Builder setParent(TraceContext context) {
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Span.Builder setNoParent() {
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Span.Builder name(String name) {
|
||||
this.name = name;
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Span.Builder event(String value) {
|
||||
this.events.add(value);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Span.Builder tag(String key, String value) {
|
||||
this.tags.put(key, value);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Span.Builder error(Throwable throwable) {
|
||||
this.error = throwable;
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Span.Builder kind(Span.Kind spanKind) {
|
||||
this.spanKind = spanKind;
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Span.Builder remoteServiceName(String remoteServiceName) {
|
||||
this.remoteServiceName = remoteServiceName;
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Span start() {
|
||||
SimpleSpan span = new SimpleSpan();
|
||||
this.tags.forEach(span::tag);
|
||||
this.events.forEach(span::event);
|
||||
span.remoteServiceName(this.remoteServiceName);
|
||||
span.error(this.error);
|
||||
span.spanKind = this.spanKind;
|
||||
span.name(this.name);
|
||||
span.start();
|
||||
simpleTracer.spans.add(span);
|
||||
return span;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -92,7 +92,7 @@ public class SimpleTracer implements Tracer {
|
||||
|
||||
@Override
|
||||
public Span.Builder spanBuilder() {
|
||||
return new SimpleSpanBuilder();
|
||||
return new SimpleSpanBuilder(this);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
Reference in New Issue
Block a user