Skip to content

Commit 5b24367

Browse files
committed
fixup
1 parent 8f69645 commit 5b24367

7 files changed

Lines changed: 328 additions & 15 deletions

File tree

braintrust-sdk/instrumentation/anthropic_2_2_0/src/main/java/dev/braintrust/instrumentation/anthropic/v2_2_0/BraintrustAnthropic.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,14 +19,14 @@ public final class BraintrustAnthropic {
1919
/** Instrument Anthropic client with Braintrust traces. */
2020
public static AnthropicClient wrap(OpenTelemetry openTelemetry, AnthropicClient client) {
2121
instrument(openTelemetry, client);
22-
return client;
22+
return ContextCapturingProxy.wrap(client, AnthropicClient.class);
2323
}
2424

2525
/** Instrument an async Anthropic client with Braintrust traces. */
2626
public static AnthropicClientAsync wrap(
2727
OpenTelemetry openTelemetry, AnthropicClientAsync client) {
2828
instrument(openTelemetry, client);
29-
return client;
29+
return ContextCapturingProxy.wrap(client, AnthropicClientAsync.class);
3030
}
3131

3232
/**
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,160 @@
1+
package dev.braintrust.instrumentation.anthropic.v2_2_0;
2+
3+
import com.anthropic.core.Params;
4+
import io.opentelemetry.api.trace.Span;
5+
import io.opentelemetry.api.trace.SpanContext;
6+
import java.lang.reflect.InvocationHandler;
7+
import java.lang.reflect.InvocationTargetException;
8+
import java.lang.reflect.Method;
9+
import java.lang.reflect.Proxy;
10+
import java.util.Map;
11+
import java.util.concurrent.ConcurrentHashMap;
12+
import lombok.extern.slf4j.Slf4j;
13+
14+
/**
15+
* Captures the caller's OTel context at the service-call boundary of a wrapped anthropic-java
16+
* client.
17+
*
18+
* <p>anthropic-java dispatches async requests through CompletableFuture continuations on the common
19+
* pool, so by the time {@link TracingHttpClient#executeAsync} runs, the caller's thread-local
20+
* context (e.g. an active application span) is gone. The service-method invocation itself, however,
21+
* happens on the caller's thread — and the only data that travels from there into the HTTP layer is
22+
* the request. So this proxy rewrites any {@link Params} argument to carry the current span as an
23+
* internal {@code traceparent}-format header, which {@link TracingHttpClient} extracts (and strips
24+
* from the outgoing request) to parent the LLM span.
25+
*
26+
* <p>Purely reflective and API-shape based, so it works for every service and endpoint: methods
27+
* returning other {@code com.anthropic} interfaces (service accessors like {@code messages()},
28+
* {@code async()}) return proxied instances, so context capture follows the whole call graph.
29+
*/
30+
@Slf4j
31+
final class ContextCapturingProxy implements InvocationHandler {
32+
33+
/** Internal correlation header; never sent — TracingHttpClient removes it. */
34+
static final String CONTEXT_HEADER = "x-braintrust-otel-traceparent";
35+
36+
/** Per-params-class reflection handles: [toBuilder, putAdditionalHeader, build]. */
37+
private static final Map<Class<?>, Method[]> PARAMS_METHODS = new ConcurrentHashMap<>();
38+
39+
private static final Method[] UNSUPPORTED = new Method[0];
40+
41+
private final Object delegate;
42+
43+
private ContextCapturingProxy(Object delegate) {
44+
this.delegate = delegate;
45+
}
46+
47+
/** Wraps {@code delegate} in a context-capturing proxy of {@code iface}. Idempotent. */
48+
@SuppressWarnings("unchecked")
49+
static <T> T wrap(T delegate, Class<T> iface) {
50+
if (delegate == null || isContextCapturingProxy(delegate)) {
51+
return delegate;
52+
}
53+
return (T)
54+
Proxy.newProxyInstance(
55+
iface.getClassLoader(),
56+
new Class<?>[] {iface},
57+
new ContextCapturingProxy(delegate));
58+
}
59+
60+
private static boolean isContextCapturingProxy(Object o) {
61+
return Proxy.isProxyClass(o.getClass())
62+
&& Proxy.getInvocationHandler(o) instanceof ContextCapturingProxy;
63+
}
64+
65+
@Override
66+
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
67+
Object[] invokeArgs = injectContextHeader(args);
68+
Object result;
69+
try {
70+
result = method.invoke(delegate, invokeArgs);
71+
} catch (InvocationTargetException e) {
72+
throw e.getCause();
73+
}
74+
// Follow the service graph: accessors like messages(), async() return com.anthropic
75+
// interfaces whose methods must also capture context.
76+
Class<?> returnType = method.getReturnType();
77+
if (result != null
78+
&& returnType.isInterface()
79+
&& returnType.getName().startsWith("com.anthropic.")) {
80+
return Proxy.newProxyInstance(
81+
returnType.getClassLoader(),
82+
new Class<?>[] {returnType},
83+
new ContextCapturingProxy(result));
84+
}
85+
return result;
86+
}
87+
88+
/** Rewrites any {@link Params} argument to carry the current span as an internal header. */
89+
private Object[] injectContextHeader(Object[] args) {
90+
if (args == null) {
91+
return null;
92+
}
93+
String traceparent = currentTraceparent();
94+
if (traceparent == null) {
95+
return args;
96+
}
97+
Object[] result = args;
98+
for (int i = 0; i < args.length; i++) {
99+
if (args[i] instanceof Params params) {
100+
Object rewritten = withContextHeader(params, traceparent);
101+
if (rewritten != null) {
102+
if (result == args) {
103+
result = args.clone();
104+
}
105+
result[i] = rewritten;
106+
}
107+
}
108+
}
109+
return result;
110+
}
111+
112+
private static String currentTraceparent() {
113+
SpanContext spanContext = Span.current().getSpanContext();
114+
if (!spanContext.isValid()) {
115+
return null;
116+
}
117+
return "00-"
118+
+ spanContext.getTraceId()
119+
+ "-"
120+
+ spanContext.getSpanId()
121+
+ "-"
122+
+ spanContext.getTraceFlags().asHex();
123+
}
124+
125+
/**
126+
* {@code params.toBuilder().putAdditionalHeader(CONTEXT_HEADER, traceparent).build()}, done
127+
* reflectively so it works for every generated params type. Returns {@code null} (leaving the
128+
* original untouched) when the shape doesn't match.
129+
*/
130+
private static Object withContextHeader(Params params, String traceparent) {
131+
Method[] methods =
132+
PARAMS_METHODS.computeIfAbsent(
133+
params.getClass(), ContextCapturingProxy::resolveParamsMethods);
134+
if (methods == UNSUPPORTED) {
135+
return null;
136+
}
137+
try {
138+
Object builder = methods[0].invoke(params);
139+
methods[1].invoke(builder, CONTEXT_HEADER, traceparent);
140+
return methods[2].invoke(builder);
141+
} catch (Exception e) {
142+
log.debug("failed to inject context header into {}", params.getClass().getName(), e);
143+
return null;
144+
}
145+
}
146+
147+
private static Method[] resolveParamsMethods(Class<?> paramsClass) {
148+
try {
149+
Method toBuilder = paramsClass.getMethod("toBuilder");
150+
Class<?> builderClass = toBuilder.getReturnType();
151+
Method putHeader =
152+
builderClass.getMethod("putAdditionalHeader", String.class, String.class);
153+
Method build = builderClass.getMethod("build");
154+
return new Method[] {toBuilder, putHeader, build};
155+
} catch (NoSuchMethodException e) {
156+
log.debug("params type {} has no header builder shape", paramsClass.getName());
157+
return UNSUPPORTED;
158+
}
159+
}
160+
}

braintrust-sdk/instrumentation/anthropic_2_2_0/src/main/java/dev/braintrust/instrumentation/anthropic/v2_2_0/TracingHttpClient.java

Lines changed: 57 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,9 @@
1212
import dev.braintrust.json.BraintrustJsonMapper;
1313
import io.opentelemetry.api.OpenTelemetry;
1414
import io.opentelemetry.api.trace.Span;
15+
import io.opentelemetry.api.trace.SpanContext;
16+
import io.opentelemetry.api.trace.TraceFlags;
17+
import io.opentelemetry.api.trace.TraceState;
1518
import io.opentelemetry.api.trace.Tracer;
1619
import io.opentelemetry.context.Context;
1720
import java.io.BufferedReader;
@@ -25,6 +28,7 @@
2528
import java.util.concurrent.atomic.AtomicBoolean;
2629
import java.util.concurrent.atomic.AtomicLong;
2730
import javax.annotation.Nonnull;
31+
import javax.annotation.Nullable;
2832
import lombok.extern.slf4j.Slf4j;
2933

3034
@Slf4j
@@ -33,10 +37,11 @@ public class TracingHttpClient implements HttpClient {
3337
private final HttpClient underlying;
3438

3539
/**
36-
* OTel context captured when the client was instrumented. Frameworks (e.g. Spring AI 2.x) may
37-
* dispatch async/streaming requests on executors or schedulers where the caller's thread-local
38-
* context is lost — which would orphan the LLM span. When the executing thread has no active
39-
* context, we fall back to the context active at wrap time.
40+
* OTel context captured when the client was instrumented. anthropic-java (and frameworks like
41+
* Spring AI 2.x) dispatch async/streaming requests on executors where the caller's thread-local
42+
* context is lost — which would orphan the LLM span. The primary fix is the context header
43+
* injected by {@link ContextCapturingProxy}; when neither that header nor an active context is
44+
* present, we fall back to the context active at wrap time.
4045
*/
4146
private final Context instrumentationContext;
4247

@@ -46,8 +51,8 @@ public TracingHttpClient(OpenTelemetry openTelemetry, HttpClient underlying) {
4651
this.instrumentationContext = Context.current();
4752
}
4853

49-
private Span startLlmSpan() {
50-
Context parent = Context.current();
54+
private Span startLlmSpan(@Nullable Context headerContext) {
55+
Context parent = headerContext != null ? headerContext : Context.current();
5156
if (parent == Context.root() && instrumentationContext != Context.root()) {
5257
parent = instrumentationContext;
5358
}
@@ -56,6 +61,46 @@ private Span startLlmSpan() {
5661
.startSpan();
5762
}
5863

64+
/**
65+
* Extracts the caller context injected by {@link ContextCapturingProxy} (if any) and strips the
66+
* internal header from the outgoing request.
67+
*/
68+
private static ExtractedRequest extractCallerContext(HttpRequest request) {
69+
var values = request.headers().values(ContextCapturingProxy.CONTEXT_HEADER);
70+
if (values.isEmpty()) {
71+
return new ExtractedRequest(request, null);
72+
}
73+
Context context = contextFromTraceparent(values.get(0));
74+
HttpRequest stripped =
75+
request.toBuilder()
76+
.replaceHeaders(ContextCapturingProxy.CONTEXT_HEADER, java.util.List.of())
77+
.build();
78+
return new ExtractedRequest(stripped, context);
79+
}
80+
81+
/** Parses a W3C {@code traceparent} value ({@code 00-<traceId>-<spanId>-<flags>}). */
82+
@Nullable
83+
private static Context contextFromTraceparent(String traceparent) {
84+
try {
85+
String[] parts = traceparent.split("-");
86+
SpanContext spanContext =
87+
SpanContext.create(
88+
parts[1],
89+
parts[2],
90+
TraceFlags.fromHex(parts[3], 0),
91+
TraceState.getDefault());
92+
if (!spanContext.isValid()) {
93+
return null;
94+
}
95+
return Context.root().with(Span.wrap(spanContext));
96+
} catch (Exception e) {
97+
log.debug("invalid context header value: {}", traceparent, e);
98+
return null;
99+
}
100+
}
101+
102+
private record ExtractedRequest(HttpRequest request, @Nullable Context callerContext) {}
103+
59104
@Override
60105
public void close() {
61106
underlying.close();
@@ -64,9 +109,10 @@ public void close() {
64109
@Override
65110
public @Nonnull HttpResponse execute(
66111
@Nonnull HttpRequest httpRequest, @Nonnull RequestOptions requestOptions) {
67-
var span = startLlmSpan();
112+
var extracted = extractCallerContext(httpRequest);
113+
var span = startLlmSpan(extracted.callerContext());
68114
try (var ignored = span.makeCurrent()) {
69-
var bufferedRequest = bufferRequestBody(httpRequest);
115+
var bufferedRequest = bufferRequestBody(extracted.request());
70116

71117
String inputJson =
72118
bufferedRequest.body() != null
@@ -93,9 +139,10 @@ public void close() {
93139
@Override
94140
public @Nonnull CompletableFuture<HttpResponse> executeAsync(
95141
@Nonnull HttpRequest httpRequest, @Nonnull RequestOptions requestOptions) {
96-
var span = startLlmSpan();
142+
var extracted = extractCallerContext(httpRequest);
143+
var span = startLlmSpan(extracted.callerContext());
97144
try {
98-
var bufferedRequest = bufferRequestBody(httpRequest);
145+
var bufferedRequest = bufferRequestBody(extracted.request());
99146
String inputJson =
100147
bufferedRequest.body() != null
101148
? readBodyAsString(bufferedRequest.body())

braintrust-sdk/instrumentation/anthropic_2_2_0/src/main/java/dev/braintrust/instrumentation/anthropic/v2_2_0/auto/AnthropicInstrumentationModule.java

Lines changed: 33 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
55

66
import com.anthropic.client.AnthropicClient;
7+
import com.anthropic.client.AnthropicClientAsync;
78
import com.google.auto.service.AutoService;
89
import dev.braintrust.instrumentation.InstrumentationModule;
910
import dev.braintrust.instrumentation.TypeInstrumentation;
@@ -32,14 +33,18 @@ public List<String> getHelperClassNames() {
3233
MANUAL_INSTRUMENTATION_PACKAGE + "TracingHttpClient$1",
3334
MANUAL_INSTRUMENTATION_PACKAGE + "TracingHttpClient$TeeingStreamHttpResponse",
3435
MANUAL_INSTRUMENTATION_PACKAGE + "TracingHttpClient$TeeInputStream",
36+
MANUAL_INSTRUMENTATION_PACKAGE + "TracingHttpClient$ExtractedRequest",
3537
MANUAL_INSTRUMENTATION_PACKAGE + "BraintrustAnthropic",
38+
MANUAL_INSTRUMENTATION_PACKAGE + "ContextCapturingProxy",
3639
"dev.braintrust.json.BraintrustJsonMapper",
3740
"dev.braintrust.instrumentation.InstrumentationSemConv");
3841
}
3942

4043
@Override
4144
public List<TypeInstrumentation> typeInstrumentations() {
42-
return List.of(new AnthropicOkHttpClientBuilderInstrumentation());
45+
return List.of(
46+
new AnthropicOkHttpClientBuilderInstrumentation(),
47+
new AnthropicOkHttpClientAsyncBuilderInstrumentation());
4348
}
4449

4550
public static class AnthropicOkHttpClientBuilderInstrumentation implements TypeInstrumentation {
@@ -67,4 +72,31 @@ public static void build(
6772
GlobalOpenTelemetry.get(), (AnthropicClient) returnedObject);
6873
}
6974
}
75+
76+
public static class AnthropicOkHttpClientAsyncBuilderInstrumentation
77+
implements TypeInstrumentation {
78+
@Override
79+
public ElementMatcher<TypeDescription> typeMatcher() {
80+
return named("com.anthropic.client.okhttp.AnthropicOkHttpClientAsync$Builder");
81+
}
82+
83+
@Override
84+
public void transform(TypeTransformer transformer) {
85+
transformer.applyAdviceToMethod(
86+
named("build").and(takesArguments(0)),
87+
AnthropicInstrumentationModule.class.getName()
88+
+ "$AnthropicOkHttpClientAsyncBuilderAdvice");
89+
}
90+
}
91+
92+
private static class AnthropicOkHttpClientAsyncBuilderAdvice {
93+
@Advice.OnMethodExit
94+
public static void build(
95+
@Advice.Return(readOnly = false, typing = Assigner.Typing.DYNAMIC)
96+
Object returnedObject) {
97+
returnedObject =
98+
BraintrustAnthropic.wrap(
99+
GlobalOpenTelemetry.get(), (AnthropicClientAsync) returnedObject);
100+
}
101+
}
70102
}

0 commit comments

Comments
 (0)