WIP: opentelemetry: use the new context propagation mechanism

servicetalk-opentelemetry-asynccontext/src/main/java/io/servicetalk/concurrent/api/OtelAgentCapturedContextProvider.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright © 2025 Apple Inc. and the ServiceTalk project 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
+ *
+ *   http://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 io.servicetalk.concurrent.api;
+
+import io.servicetalk.context.api.ContextMap;
+
+import io.opentelemetry.context.Context;
+
+public final class OtelAgentCapturedContextProvider implements CapturedContextProvider {
+
+  public OtelAgentCapturedContextProvider() {
+  }
+
+  @Override
+  public CapturedContext captureContext(CapturedContext underlying) {
+    return new WithOtelCapturedContext(Context.current(), underlying);
+  }
+
+  static final class WithOtelCapturedContext implements CapturedContext {
+
+    private final Context agentContext;
+    private final CapturedContext stContext;
+
+    WithOtelCapturedContext(Context agentContext, CapturedContext stContext) {
+      this.agentContext = agentContext;
+      this.stContext = stContext;
+    }
+
+    @Override
+    public ContextMap captured() {
+      return stContext.captured();
+    }
+
+    @Override
+    public Scope attachContext() {
+      Scope stScope = stContext.attachContext();
+      io.opentelemetry.context.Scope agentScope = agentContext.makeCurrent();
+      return () -> {
+        agentScope.close();
+        stScope.close();
+      };
+    }
+  }
+}

servicetalk-opentelemetry-asynccontext/src/main/resources/META-INF/services/io.servicetalk.concurrent.api.CapturedContextProvider

@@ -0,0 +1 @@
+io.servicetalk.concurrent.api.OtelAgentCapturedContextProvider

servicetalk-opentelemetry-http/src/main/java/io/servicetalk/opentelemetry/http/AbstractOpenTelemetryFilter.java

@@ -16,27 +16,49 @@
 
 package io.servicetalk.opentelemetry.http;
 
+import io.servicetalk.concurrent.PublisherSource;
+import io.servicetalk.concurrent.SingleSource;
+import io.servicetalk.concurrent.api.Publisher;
+import io.servicetalk.concurrent.api.Single;
+import io.servicetalk.concurrent.api.SourceAdapters;
 import io.servicetalk.http.api.HttpExecutionStrategies;
 import io.servicetalk.http.api.HttpExecutionStrategy;
 import io.servicetalk.http.api.HttpExecutionStrategyInfluencer;
 
-import io.opentelemetry.api.OpenTelemetry;
-import io.opentelemetry.api.trace.Tracer;
-import io.opentelemetry.context.propagation.ContextPropagators;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import io.servicetalk.http.api.StreamingHttpResponse;
 
 abstract class AbstractOpenTelemetryFilter implements HttpExecutionStrategyInfluencer {
     static final OpenTelemetryOptions DEFAULT_OPTIONS = new OpenTelemetryOptions.Builder().build();
     static final String INSTRUMENTATION_SCOPE_NAME = "io.servicetalk";
-    final Tracer tracer;
-    final ContextPropagators propagators;
-
-    AbstractOpenTelemetryFilter(final OpenTelemetry openTelemetry) {
-        this.tracer = openTelemetry.getTracer(INSTRUMENTATION_SCOPE_NAME);
-        this.propagators = openTelemetry.getPropagators();
-    }
 
     @Override
     public final HttpExecutionStrategy requiredOffloads() {
         return HttpExecutionStrategies.offloadNone();
     }
+
+    static Single<StreamingHttpResponse> withContext(Single<StreamingHttpResponse> response, Context context) {
+        return new Single<StreamingHttpResponse>() {
+            @Override
+            protected void handleSubscribe(SingleSource.Subscriber<? super StreamingHttpResponse> subscriber) {
+                try (Scope ignored = context.makeCurrent()) {
+                    SourceAdapters.toSource(response.map(resp ->
+                                    resp.transformMessageBody(body -> transformBody(body, context))))
+                            .subscribe(subscriber);
+                }
+            }
+        };
+    }
+
+    private static <T> Publisher<T> transformBody(Publisher<T> body, Context context) {
+        return new Publisher<T>() {
+            @Override
+            protected void handleSubscribe(PublisherSource.Subscriber<? super T> subscriber) {
+                try (Scope ignored = context.makeCurrent()) {
+                    SourceAdapters.toSource(body).subscribe(subscriber);
+                }
+            }
+        };
+    }
 }

servicetalk-opentelemetry-http/src/main/java/io/servicetalk/opentelemetry/http/OpenTelemetryHttpRequestFilter.java

@@ -32,6 +32,7 @@
 
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.api.trace.Tracer;
 import io.opentelemetry.context.Context;
@@ -43,8 +44,6 @@
 import io.opentelemetry.instrumentation.api.instrumenter.http.HttpClientAttributesExtractor;
 import io.opentelemetry.instrumentation.api.instrumenter.http.HttpClientMetrics;
 import io.opentelemetry.instrumentation.api.instrumenter.http.HttpSpanNameExtractor;
-import io.opentelemetry.instrumentation.api.instrumenter.net.NetClientAttributesExtractor;
-import io.opentelemetry.semconv.SemanticAttributes;
 
 import java.util.function.UnaryOperator;
 
@@ -61,7 +60,9 @@
  * therefore will not see the {@link Span} for the current request/response.
  */
 public final class OpenTelemetryHttpRequestFilter extends AbstractOpenTelemetryFilter
-    implements StreamingHttpClientFilterFactory, StreamingHttpConnectionFilterFactory {
+        implements StreamingHttpClientFilterFactory, StreamingHttpConnectionFilterFactory {
+
+    private static final AttributeKey<String> PEER_SERVICE = AttributeKey.stringKey("peer.service");
 
     private final Instrumenter<HttpRequestMetaData, HttpResponseMetaData> instrumenter;
 
@@ -118,7 +119,6 @@ public OpenTelemetryHttpRequestFilter() {
      */
     OpenTelemetryHttpRequestFilter(final OpenTelemetry openTelemetry, String componentName,
                                    final OpenTelemetryOptions opentelemetryOptions) {
-        super(openTelemetry);
         SpanNameExtractor<HttpRequestMetaData> clientSpanNameExtractor =
                 HttpSpanNameExtractor.create(ServiceTalkHttpAttributesGetter.CLIENT_INSTANCE);
         InstrumenterBuilder<HttpRequestMetaData, HttpResponseMetaData> clientInstrumenterBuilder =
@@ -128,20 +128,18 @@ public OpenTelemetryHttpRequestFilter() {
         clientInstrumenterBuilder
                 .addAttributesExtractor(
                         HttpClientAttributesExtractor
-                        .builder(ServiceTalkHttpAttributesGetter.CLIENT_INSTANCE,
-                                ServiceTalkNetAttributesGetter.CLIENT_INSTANCE)
-                        .setCapturedRequestHeaders(opentelemetryOptions.capturedRequestHeaders())
-                        .setCapturedResponseHeaders(opentelemetryOptions.capturedResponseHeaders())
-                        .build())
-                .addAttributesExtractor(NetClientAttributesExtractor.create(
-                        ServiceTalkNetAttributesGetter.CLIENT_INSTANCE));
+                                .builder(ServiceTalkHttpAttributesGetter.CLIENT_INSTANCE,
+                                        ServiceTalkNetAttributesGetter.CLIENT_INSTANCE)
+                                .setCapturedRequestHeaders(opentelemetryOptions.capturedRequestHeaders())
+                                .setCapturedResponseHeaders(opentelemetryOptions.capturedResponseHeaders())
+                                .build());
         if (opentelemetryOptions.enableMetrics()) {
             clientInstrumenterBuilder.addOperationMetrics(HttpClientMetrics.get());
         }
         componentName = componentName.trim();
         if (!componentName.isEmpty()) {
             clientInstrumenterBuilder.addAttributesExtractor(
-                    AttributesExtractor.constant(SemanticAttributes.PEER_SERVICE, componentName));
+                    AttributesExtractor.constant(PEER_SERVICE, componentName));
         }
         instrumenter =
                 clientInstrumenterBuilder.buildClientInstrumenter(RequestHeadersPropagatorSetter.INSTANCE);
@@ -171,18 +169,23 @@ public Single<StreamingHttpResponse> request(final StreamingHttpRequest request)
 
     private Single<StreamingHttpResponse> trackRequest(final StreamingHttpRequester delegate,
                                                        final StreamingHttpRequest request) {
-        final Context parentContext = Context.current();
-        final Context context = instrumenter.start(parentContext, request);
-
-        final Scope scope = context.makeCurrent();
-        final ScopeTracker tracker = new ScopeTracker(scope, context, request, instrumenter);
-        Single<StreamingHttpResponse> response;
-        try {
-            response = delegate.request(request);
-        } catch (Throwable t) {
-            tracker.onError(t);
-            return Single.failed(t);
+        // This should be (essentially) the first filter in the client filter chain, so here is where we initialize
+        // all the little bits and pieces that will end up being part of the request context. This includes setting
+        // up the retry counter context entry since retries will happen further down in the filter chain.
+        Context current = Context.current();
+        if (!instrumenter.shouldStart(current, request)) {
+            return delegate.request(request);
+        }
+        final Context context = instrumenter.start(current, request);
+        try (Scope unused = context.makeCurrent()) {
+            final ScopeTracker tracker = new ScopeTracker(context, request, instrumenter);
+            try {
+                Single<StreamingHttpResponse> response = delegate.request(request);
+                return withContext(tracker.track(response), context);
+            } catch (Throwable t) {
+                tracker.onError(t);
+                return Single.failed(t);
+            }
         }
-        return tracker.track(response);
     }
 }