fix formatting, enable http-pipelining test + fixes

Author: samwright

instrumentation/pekko/pekko-http-1.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/pekkohttp/v1_0/server/PekkoFlowWrapper.java

@@ -43,7 +43,7 @@ public static Context getContext(OutHandler outHandler) {
       // be started before the first one has returned a response, because of this the first request
       // in the queue is always the one that is currently being processed.
       PekkoTracingRequest request =
-          ((TracingLogic.ApplicationOutHandler) outHandler).getRequests().peek();
+          ((TracingLogic.ApplicationOutHandler) outHandler).getRequests().poll();
       if (request != null) {
         return request.context;
       }
@@ -110,7 +110,8 @@ public void onPush() {
                   request
                       .getAttribute(PekkoTracingRequest.ATTR_KEY)
                       .orElse(PekkoTracingRequest.EMPTY);
-              if (tracingRequest == PekkoTracingRequest.EMPTY) {
+              if (tracingRequest != PekkoTracingRequest.EMPTY) {
+                // Remove HttpRequest attribute before passing it to user code
                 request = (HttpRequest) request.removeAttribute(PekkoTracingRequest.ATTR_KEY);
               }
               // event if span wasn't started we need to push TracingRequest to match response
@@ -138,19 +139,16 @@ public void onUpstreamFailure(Throwable exception) {
             @Override
             public void onPush() {
               HttpResponse response = grab(responseIn);
-              requests.poll();
               push(responseOut, response);
             }
 
             @Override
             public void onUpstreamFailure(Throwable exception) {
-              requests.clear();
               fail(responseOut, exception);
             }
 
             @Override
             public void onUpstreamFinish() {
-              requests.clear();
               completeStage();
             }
           });

instrumentation/pekko/pekko-http-1.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/pekkohttp/v1_0/server/PekkoHttpServerTracer.java

@@ -101,7 +101,7 @@ public void onPush() {
               if (instrumenter().shouldStart(parentContext, request)) {
                 Context context = instrumenter().start(parentContext, request);
                 context = PekkoRouteHolder.init(context);
-                tracingRequest = new PekkoTracingRequest(context, request);
+                tracingRequest = new PekkoTracingRequest(context, parentContext, request);
                 request =
                     (HttpRequest)
                         request.addAttribute(PekkoTracingRequest.ATTR_KEY, tracingRequest);
@@ -144,24 +144,41 @@ public void onPush() {
                   response = (HttpResponse) response.addHeaders(headers);
                 }
 
+                // When GraphInterpreterInstrumentation sets `tracingRequest.context` as the current
+                // context, the akka Envelope + Actor instrumentation propagates it all the way
+                // back to here and follows the HttpResponse up the stack of stages.
+                // If http-pipelining is enabled, it will also propagate this context to the handling
+                // of the next request, leading to context-leaking errors.
+                // To prevent this, we reset the context to what it was before creating it.
+                tracingRequest.parentContext.makeCurrent();
+
                 instrumenter().end(tracingRequest.context, tracingRequest.request, response, null);
               }
               push(responseOut, response);
             }
 
             @Override
             public void onUpstreamFailure(Throwable exception) {
-              endOngoingSpans(exception);
+              // End the span for the request that failed
+              PekkoTracingRequest tracingRequest = requests.poll();
+              if (tracingRequest != null && tracingRequest != PekkoTracingRequest.EMPTY) {
+                // see comment above
+                tracingRequest.parentContext.makeCurrent();
+                instrumenter()
+                    .end(
+                        tracingRequest.context,
+                        tracingRequest.request,
+                        PekkoHttpServerSingletons.errorResponse(),
+                        exception);
+              }
+
               fail(responseOut, exception);
             }
 
+
             @Override
             public void onUpstreamFinish() {
-              endOngoingSpans(null);
-              completeStage();
-            }
-
-            private void endOngoingSpans(Throwable exception) {
+              // End any ongoing spans, though there should be none.
               PekkoTracingRequest tracingRequest;
               while ((tracingRequest = requests.poll()) != null) {
                 if (tracingRequest == PekkoTracingRequest.EMPTY) {
@@ -172,8 +189,9 @@ private void endOngoingSpans(Throwable exception) {
                         tracingRequest.context,
                         tracingRequest.request,
                         PekkoHttpServerSingletons.errorResponse(),
-                        exception);
+                        null);
               }
+              completeStage();
             }
           });
     }

instrumentation/pekko/pekko-http-1.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/pekkohttp/v1_0/server/PekkoTracingRequest.java

@@ -5,19 +5,31 @@
 
 package io.opentelemetry.javaagent.instrumentation.pekkohttp.v1_0.server;
 
+import static io.opentelemetry.context.ContextKey.named;
+
 import io.opentelemetry.context.Context;
+import io.opentelemetry.context.ContextKey;
+import io.opentelemetry.context.ImplicitContextKeyed;
 import org.apache.pekko.http.scaladsl.model.AttributeKey;
 import org.apache.pekko.http.scaladsl.model.HttpRequest;
 
-public class PekkoTracingRequest {
+public class PekkoTracingRequest implements ImplicitContextKeyed {
+  private static final ContextKey<PekkoTracingRequest> CONTEXT_KEY = named("opentelemetry-pekko-tracing-request");
   static final AttributeKey<PekkoTracingRequest> ATTR_KEY =
       new AttributeKey<>("_otel_ctx", PekkoTracingRequest.class);
-  static final PekkoTracingRequest EMPTY = new PekkoTracingRequest(null, null);
+  static final PekkoTracingRequest EMPTY = new PekkoTracingRequest(null, null, null);
   final Context context;
+  final Context parentContext;
   final HttpRequest request;
 
-  PekkoTracingRequest(Context context, HttpRequest request) {
+  PekkoTracingRequest(Context context, Context parentContext, HttpRequest request) {
     this.context = context;
+    this.parentContext = parentContext;
     this.request = request;
   }
+
+  @Override
+  public Context storeInContext(Context context) {
+    return context.with(CONTEXT_KEY, this);
+  }
 }

instrumentation/pekko/pekko-http-1.0/javaagent/src/test/resources/application.conf

@@ -8,4 +8,7 @@ pekko.http {
       connecting-timeout = 5s
     }
   }
+  server {
+    pipelining-limit = 4 # default is 1
+  }
 }

instrumentation/pekko/pekko-http-1.0/javaagent/src/test/scala/io/opentelemetry/javaagent/instrumentation/pekkohttp/v1_0/PekkoHttpServerInstrumentationTestAsync.scala

@@ -25,11 +25,4 @@ class PekkoHttpServerInstrumentationTestAsync
 
   override protected def stopServer(server: Object): Unit =
     PekkoHttpTestAsyncWebServer.stop()
-
-  override protected def configure(
-      options: HttpServerTestOptions
-  ): Unit = {
-    super.configure(options)
-    options.setTestHttpPipelining(false)
-  }
 }

instrumentation/pekko/pekko-http-1.0/javaagent/src/test/scala/io/opentelemetry/javaagent/instrumentation/pekkohttp/v1_0/PekkoHttpTestAsyncWebServer.scala

@@ -64,7 +64,7 @@ object PekkoHttpTestAsyncWebServer {
     if (null == binding) {
       import scala.concurrent.duration._
       binding = Await.result(
-        Http().bindAndHandleAsync(asyncHandler, "localhost", port),
+        Http().bindAndHandleAsync(asyncHandler, "localhost", port, parallelism = 2),
         10.seconds
       )
     }