Test propagation of trace info over stream scopes

Author: iRevive

build.sbt

@@ -33,7 +33,7 @@ lazy val core = crossProject(JVMPlatform, JSPlatform)
     libraryDependencies ++= Seq(
       "org.typelevel" %%% "cats-core" % "2.8.0",
       "org.typelevel" %%% "cats-effect" % "3.3.13",
-      "co.fs2" %%% "fs2-core" % "3.2.11",
+      "org.scodec" %%% "scodec-bits" % "1.1.34",
       "org.scalameta" %%% "munit" % "0.7.29" % Test,
       "org.typelevel" %%% "munit-cats-effect-3" % "1.0.7" % Test,
       "org.typelevel" %%% "cats-effect-testkit" % "3.3.13" % Test
@@ -69,6 +69,7 @@ lazy val java = crossProject(JVMPlatform)
       "io.opentelemetry" % "opentelemetry-api" % "1.15.0",
       "io.opentelemetry" % "opentelemetry-sdk" % "1.15.0" % Test,
       "io.opentelemetry" % "opentelemetry-sdk-testing" % "1.15.0" % Test,
+      "co.fs2" %%% "fs2-core" % "3.2.11" % Test,
       "org.scalameta" %% "munit" % "0.7.29" % Test,
       "org.typelevel" %% "munit-cats-effect-3" % "1.0.7" % Test,
       "org.typelevel" %%% "cats-effect-testkit" % "3.3.13" % Test

java/src/test/scala/org/typelevel/otel4s/java/trace/TracerSuite.scala

@@ -19,6 +19,7 @@ package org.typelevel.otel4s.java.trace
 import cats.effect.IO
 import cats.effect.Resource
 import cats.effect.testkit.TestControl
+import fs2.Stream
 import io.opentelemetry.api.common.Attributes
 import io.opentelemetry.api.trace.StatusCode
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo
@@ -451,6 +452,51 @@ class TracerSuite extends CatsEffectSuite {
     }
   }
 
+  test("propagate trace info over stream scopes") {
+    def expected(now: FiniteDuration) =
+      SpanNode(
+        "span",
+        now,
+        now,
+        List(
+          SpanNode("span-3", now, now, Nil),
+          SpanNode("span-2", now, now, Nil)
+        )
+      )
+
+    def flow(tracer: Tracer[IO]): Stream[IO, Unit] =
+      for {
+        span <- Stream.resource(tracer.span("span"))
+        _ <- Stream.eval(
+          tracer.currentSpanContext.assertEquals(Some(span.context))
+        )
+        span2 <- Stream.resource(tracer.span("span-2"))
+        _ <- Stream.eval(
+          tracer.currentSpanContext.assertEquals(Some(span2.context))
+        )
+        span3 <- Stream.resource(
+          tracer.spanBuilder("span-3").withParent(span.context).createAuto
+        )
+        _ <- Stream.eval(
+          tracer.currentSpanContext.assertEquals(Some(span3.context))
+        )
+      } yield ()
+
+    TestControl.executeEmbed {
+      for {
+        now <- IO.monotonic.delayBy(1.second) // otherwise returns 0
+        sdk <- makeSdk()
+        tracer <- sdk.provider.tracer("tracer").get
+        _ <- tracer.currentSpanContext.assertEquals(None)
+        _ <- flow(tracer).compile.drain
+        _ <- tracer.currentSpanContext.assertEquals(None)
+        spans <- sdk.finishedSpans
+        tree <- IO.pure(SpanNode.fromSpans(spans))
+        // _ <- IO.println(tree.map(SpanNode.render).mkString("\n"))
+      } yield assertEquals(tree, List(expected(now)))
+    }
+  }
+
   private def assertIdsNotEqual(s1: Span[IO], s2: Span[IO]): Unit = {
     assertNotEquals(s1.context.traceIdHex, s2.context.traceIdHex)
     assertNotEquals(s1.context.spanIdHex, s2.context.spanIdHex)