test otel span is passed from pekko routes to actors

instrumentation/pekko/pekko-http-1.0/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/pekkohttp/v1_0/PekkoHttpTestSetup.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.pekkohttp.v1_0;
+
+import static io.opentelemetry.instrumentation.testing.junit.http.AbstractHttpServerUsingTest.TEST_CLIENT_IP;
+import static io.opentelemetry.instrumentation.testing.junit.http.AbstractHttpServerUsingTest.TEST_USER_AGENT;
+
+import io.opentelemetry.instrumentation.test.utils.PortUtils;
+import io.opentelemetry.testing.internal.armeria.client.ClientFactory;
+import io.opentelemetry.testing.internal.armeria.client.WebClient;
+import io.opentelemetry.testing.internal.armeria.client.logging.LoggingClient;
+import io.opentelemetry.testing.internal.armeria.common.HttpHeaderNames;
+import java.time.Duration;
+
+public final class PekkoHttpTestSetup {
+
+  private final int port;
+  private final WebClient client;
+
+  public PekkoHttpTestSetup() {
+    port = PortUtils.findOpenPort();
+    client =
+        WebClient.builder()
+            .responseTimeout(Duration.ofMinutes(1))
+            .writeTimeout(Duration.ofMinutes(1))
+            .factory(ClientFactory.builder().connectTimeout(Duration.ofMinutes(1)).build())
+            .setHeader(HttpHeaderNames.USER_AGENT, TEST_USER_AGENT)
+            .setHeader(HttpHeaderNames.X_FORWARDED_FOR, TEST_CLIENT_IP)
+            .decorator(LoggingClient.newDecorator())
+            .build();
+  }
+
+  public int getPort() {
+    return port;
+  }
+
+  public WebClient getClient() {
+    return client;
+  }
+}

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

@@ -0,0 +1,45 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.pekkohttp.v1_0
+
+import java.net.URI
+
+import io.opentelemetry.testing.internal.armeria.common.{
+  AggregatedHttpRequest,
+  HttpMethod
+}
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.{AfterAll, BeforeAll, Test, TestInstance}
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class PekkoHttpServerWithActorTest {
+
+  val setup = new PekkoHttpTestSetup()
+
+  @BeforeAll def setupOptions(): Unit = {
+    PekkoHttpTestWebServerWithActor.start(setup.getPort())
+  }
+
+  @AfterAll def cleanup(): Unit = {
+    PekkoHttpTestWebServerWithActor.stop()
+  }
+
+  @Test def testSpan(): Unit = {
+    val address = new URI(s"http://localhost:${setup.getPort()}/")
+    val request = AggregatedHttpRequest.of(
+      HttpMethod.GET,
+      address.resolve("/test").toString
+    )
+    val response = setup.getClient().execute(request).aggregate.join
+    assertThat(response.status.code).isEqualTo(200)
+    val responseText = response.contentUtf8
+    val splits = responseText.split("\n")
+    assertThat(splits.length).isEqualTo(2)
+    val routeSpan = splits(0).substring(6, splits(0).length)
+    val actorSpan = splits(1).substring(6, splits(1).length)
+    assertThat(routeSpan).isEqualTo(actorSpan)
+  }
+}

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

@@ -0,0 +1,67 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.pekkohttp.v1_0
+
+import io.opentelemetry.api.trace.Span
+import org.apache.pekko.actor.{Actor, ActorSystem, Props}
+import org.apache.pekko.http.scaladsl.Http
+import org.apache.pekko.http.scaladsl.Http.ServerBinding
+import org.apache.pekko.http.scaladsl.server.Directives._
+import org.apache.pekko.pattern.ask
+import org.apache.pekko.util.Timeout
+
+import scala.concurrent.{Await, ExecutionContext}
+import scala.concurrent.duration.DurationInt
+
+object PekkoHttpTestWebServerWithActor {
+  implicit val system: ActorSystem = ActorSystem("http-server-with-actor")
+  // needed for the future flatMap/onComplete in the end
+  implicit val executionContext: ExecutionContext = system.dispatcher
+
+  private case object TestMessage
+  private class SpanTestActor extends Actor {
+    def receive = { case TestMessage =>
+      sender() ! spanSummary(Span.current())
+    }
+  }
+
+  val spanTestActor = system.actorOf(Props[SpanTestActor]())
+
+  var route = get {
+    path("test") {
+      complete {
+        val otelSummary = spanSummary(Span.current())
+        spanTestActor.ask(TestMessage)(Timeout(5.seconds)).mapTo[String].map {
+          actorSummary =>
+            s"Route=$otelSummary\nActor=$actorSummary"
+        }
+      }
+    }
+  }
+
+  private var binding: ServerBinding = null
+
+  def start(port: Int): Unit = synchronized {
+    if (null == binding) {
+      binding =
+        Await.result(Http().bindAndHandle(route, "localhost", port), 10.seconds)
+    }
+  }
+
+  def stop(): Unit = synchronized {
+    if (null != binding) {
+      binding.unbind()
+      system.terminate()
+      binding = null
+    }
+  }
+
+  def spanSummary(span: Span): String = {
+    val spanId = span.getSpanContext().getSpanId()
+    val traceId = span.getSpanContext().getTraceId()
+    s"Span(traceId=$traceId, spanId=$spanId)"
+  }
+}