Akka http: propagate context to callbacks
MODIFY
instrumentation/akka-http-10.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/akkahttp/client/AkkaHttpClientInstrumentationModule.java
instrumentation/akka-http-10.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/akkahttp/client/AkkaHttpClientInstrumentationModule.java
package io.opentelemetry.javaagent.instrumentation.akkahttp.client;
-import static java.util.Collections.singletonList;
+import static java.util.Arrays.asList;
import com.google.auto.service.AutoService;
import io.opentelemetry.javaagent.extension.instrumentation.InstrumentationModule;
@@ -20,6 +20,6 @@ public class AkkaHttpClientInstrumentationModule extends InstrumentationModule {
@Override
public List<TypeInstrumentation> typeInstrumentations() {
- return singletonList(new HttpExtClientInstrumentation());
+ return asList(new HttpExtClientInstrumentation(), new PoolMasterActorInstrumentation());
}
}

ADD
instrumentation/akka-http-10.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/akkahttp/client/FutureWrapper.java
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.akkahttp.client;
+
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import scala.concurrent.ExecutionContext;
+import scala.concurrent.Future;
+import scala.concurrent.impl.Promise;
+import scala.runtime.AbstractFunction1;
+import scala.util.Try;
+
+public class FutureWrapper {
+ public static <T> Future<T> wrap(
+ Future<T> future, ExecutionContext executionContext, Context context) {
+ Promise.DefaultPromise<T> promise = new Promise.DefaultPromise<>();
+ future.onComplete(
+ new AbstractFunction1<Try<T>, Object>() {
+
+ @Override
+ public Object apply(Try<T> result) {
+ try (Scope ignored = context.makeCurrent()) {
+ return promise.complete(result);
+ }
+ }
+ },
+ executionContext);
+
+ return promise;
+ }
+}

MODIFY
instrumentation/akka-http-10.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/akkahttp/client/HttpExtClientInstrumentation.java
instrumentation/akka-http-10.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/akkahttp/client/HttpExtClientInstrumentation.java
@@ -70,7 +70,7 @@ public class HttpExtClientInstrumentation implements TypeInstrumentation {
public static void methodExit(
@Advice.Argument(0) HttpRequest request,
@Advice.This HttpExt thiz,
- @Advice.Return Future<HttpResponse> responseFuture,
+ @Advice.Return(readOnly = false) Future<HttpResponse> responseFuture,
@Advice.Thrown Throwable throwable,
@Advice.Local("otelContext") Context context,
@Advice.Local("otelScope") Scope scope) {
@@ -84,6 +84,10 @@ public class HttpExtClientInstrumentation implements TypeInstrumentation {
} else {
tracer().endExceptionally(context, throwable);
}
+ if (responseFuture != null) {
+ responseFuture =
+ FutureWrapper.wrap(responseFuture, thiz.system().dispatcher(), currentContext());
+ }
}
}
}

MODIFY
instrumentation/akka-http-10.0/javaagent/src/test/groovy/AkkaHttpClientInstrumentationTest.groovy
instrumentation/akka-http-10.0/javaagent/src/test/groovy/AkkaHttpClientInstrumentationTest.groovy
@@ -48,17 +48,11 @@ class AkkaHttpClientInstrumentationTest extends HttpClientTest<HttpRequest> impl
@Override
void sendRequestWithCallback(HttpRequest request, String method, URI uri, Map<String, String> headers, RequestResult requestResult) {
Http.get(system).singleRequest(request, materializer).whenComplete {response, throwable ->
+ if (throwable == null) {
response.discardEntityBytes(materializer)
- requestResult.complete({ response.status().intValue() }, throwable)
}
+ requestResult.complete({ response.status().intValue() }, throwable)
}
-
- // TODO(anuraaga): Context leak seems to prevent us from running asynchronous tests in a row.
- // Disable for now.
- // https://github.com/open-telemetry/opentelemetry-java-instrumentation/issues/2639
- @Override
- boolean testCallback() {
- false
}
@Override
@@ -66,12 +60,6 @@ class AkkaHttpClientInstrumentationTest extends HttpClientTest<HttpRequest> impl
false
}
- @Override
- boolean testRemoteConnection() {
- // Not sure how to properly set timeouts...
- return false
- }
-
@Override
SingleConnection createSingleConnection(String host, int port) {
// singleConnection test would require instrumentation to support requests made through pools

MODIFY
instrumentation/akka-http-10.0/javaagent/src/test/resources/application.conf
instrumentation/akka-http-10.0/javaagent/src/test/resources/application.conf
@@ -2,5 +2,10 @@ akka.http {
host-connection-pool {
// Limit maximum http backoff for tests
max-connection-backoff = 100ms
+ max-open-requests = 1024
+ max-retries = 0
+ client {
+ connecting-timeout = 5s
+ }
}
}

Akka http: propagate context to callbacks (#3263)

Co-authored-by: Trask Stalnaker <trask.stalnaker@gmail.com>