镜像站点 · 本页由第三方 GitHub 只读镜像提供,非 GitHub 官方站点,不接受任何登录或凭据输入。前往 github.com
Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -1,18 +1,34 @@
package datadog.trace.instrumentation.resilience4j;

import static datadog.trace.bootstrap.instrumentation.java.concurrent.ExcludeFilter.ExcludeType.COMPLETABLE_FUTURE_CALLBACK;
import static java.util.Collections.singleton;
import static java.util.Collections.singletonMap;

import com.google.auto.service.AutoService;
import datadog.trace.agent.tooling.ExcludeFilterProvider;
import datadog.trace.agent.tooling.Instrumenter;
import datadog.trace.agent.tooling.InstrumenterModule;
import datadog.trace.bootstrap.instrumentation.java.concurrent.ExcludeFilter;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map;

@AutoService(InstrumenterModule.class)
public class Resilience4jModule extends InstrumenterModule.Tracing {
public class Resilience4jModule extends InstrumenterModule.Tracing
implements ExcludeFilterProvider {

public Resilience4jModule() {
super("resilience4j");
}

@Override
public Map<ExcludeFilter.ExcludeType, ? extends Collection<String>> excludedClasses() {
return singletonMap(
COMPLETABLE_FUTURE_CALLBACK,
singleton(packageName + ".WrapperWithContext$SpanFinishingCallback"));
}

@Override
public List<Instrumenter> typeInstrumentations() {
return Arrays.asList(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
Expand Down Expand Up @@ -213,11 +214,7 @@ public SupplierOfCompletionStageWithContext(
@Override
public CompletionStage<?> get() {
try (ContextScope ignore = activateScope()) {
return delegate
.get()
.whenComplete((v, e) -> {
finishSpanIfNeeded();
});
return delegate.get().whenComplete(new SpanFinishingCallback(this));
}
}
}
Expand All @@ -237,17 +234,27 @@ public Future<?> get() {
try (ContextScope ignore = activateScope()) {
Future<?> future = delegate.get();
if (future instanceof CompletableFuture) {
((CompletableFuture<?>) future)
.whenComplete((v, e) -> {
finishSpanIfNeeded();
});
((CompletableFuture<?>) future).whenComplete(new SpanFinishingCallback(this));
return future;
}
return new FinishOnGetFuture<>(future, this);
}
}
}

private static final class SpanFinishingCallback implements BiConsumer<Object, Throwable> {
private final WrapperWithContext<?> owner;

SpanFinishingCallback(WrapperWithContext<?> owner) {
this.owner = owner;
}

@Override
public void accept(Object value, Throwable error) {
owner.finishSpanIfNeeded();
}
}

private static final class FinishOnGetFuture<V> implements Future<V> {
private final Future<V> delegate;
private final WrapperWithContext<?> context;
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,121 @@
import datadog.trace.agent.test.InstrumentationSpecification
import datadog.trace.agent.test.scopediag.ScopeDiagnostics
import datadog.trace.bootstrap.instrumentation.api.AgentSpan
import datadog.trace.instrumentation.resilience4j.Resilience4jSpan
import datadog.trace.instrumentation.resilience4j.Resilience4jSpanDecorator
import datadog.trace.instrumentation.resilience4j.WrapperWithContext.SupplierOfCompletionStageWithContext
import datadog.trace.instrumentation.resilience4j.WrapperWithContext.SupplierOfFutureWithContext

import java.util.concurrent.CompletableFuture
import java.util.concurrent.CompletionException
import java.util.function.BiConsumer
import java.util.function.Supplier

import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateSpan
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activeSpan
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.startSpan

class Resilience4jCallbackExclusionTest extends InstrumentationSpecification {
def "#kind finisher skips capture: completed=#completed failure=#failure nested=#nested"() {
setup:
def source = new CompletableFuture<String>()
def cause = new IllegalArgumentException("delegate failed")
if (completed) {
complete(source, failure ? cause : null)
}
AgentSpan resilience = null
AgentSpan delegateChild = null
Supplier delegate = {
resilience = activeSpan()
delegateChild = startSpan("test", "delegate.child")
delegateChild.finish()
source
} as Supplier
def wrapper = kind == "stage" ? new SupplierOfCompletionStageWithContext(delegate, Resilience4jSpanDecorator.DECORATE, null) :
new SupplierOfFutureWithContext(delegate, Resilience4jSpanDecorator.DECORATE, null)
def application = startSpan("test", "application")
def existing = nested ? Resilience4jSpan.start() : null
def scope = existing == null ? null : activateSpan(existing)
def callback = new ApplicationCallback()
int before = captures()

when:
def returned = wrapper.get()
int finisherCaptures = captures() - before
assert activeSpan() == existing
scope?.close()
scope = activateSpan(application)
before = captures()
def downstream = returned.whenComplete(callback)
int applicationCaptures = captures() - before
scope.close()
scope = null
complete(source, failure ? cause : null)

then:
finisherCaptures == 0
applicationCaptures == (completed ? 0 : 1)
returned.is(source) == (kind == "future")
resilience.operationName.toString() == "resilience4j"
delegateChild.parentId == resilience.spanId
callback.calls == 1
callback.parent.is(application)
callback.child.parentId == application.spanId
if (failure) {
try {
downstream.join()
assert false: "expected exceptional completion"
} catch (CompletionException expected) {
assert expected.cause.is(cause)
}
} else {
assert downstream.join() == "value"
}
!resilience.isError()
if (nested) {
assert resilience.is(existing)
assert TEST_WRITER.size() == 0
existing.finish()
}
TEST_WRITER.waitForTraces(1)
TEST_WRITER[0].size() == 2
activeSpan() == null

cleanup:
scope?.close()
complete(source, failure ? cause : null)
existing?.finish()
application.finish()
TEST_WRITER.waitForTraces(2)
assert TEST_WRITER.collectMany { it }.size() == 4

where:
[kind, completed, failure, nested] << [["stage", "future"], [false, true], [false, true], [false, true]].combinations()
}

private static int captures() {
ScopeDiagnostics.report().records().size()
}

private static void complete(CompletableFuture source, Throwable failure) {
if (failure == null) {
source.complete("value")
} else {
source.completeExceptionally(failure)
}
}

private static class ApplicationCallback implements BiConsumer<String, Throwable> {
int calls
AgentSpan parent
AgentSpan child

@Override
void accept(String value, Throwable failure) {
calls++
parent = activeSpan()
child = startSpan("test", "application.callback")
child.finish()
}
}
}
Loading