diff --git a/build.sbt b/build.sbt index cfef6bce..ebf56f9c 100644 --- a/build.sbt +++ b/build.sbt @@ -129,12 +129,26 @@ lazy val otelContext: Project = (project in file("otel-context")) .settings( name := "otel-context", libraryDependencies ++= Seq( - "io.opentelemetry" % "opentelemetry-api" % "1.66.0", + "io.opentelemetry" % "opentelemetry-api" % "1.67.0", scalaTest ) ) .dependsOn(core % "test->test;compile->compile") +lazy val otel: Project = (project in file("otel")) + .settings( + name := "ox-telemetry", + Test / fork := true, + Test / javaOptions += "-Dio.opentelemetry.context.enableStrictContext=true", + libraryDependencies ++= Seq( + "io.opentelemetry" % "opentelemetry-api" % "1.67.0", + "io.opentelemetry" % "opentelemetry-sdk" % "1.67.0", + scalaTest, + "io.opentelemetry" % "opentelemetry-sdk-testing" % "1.67.0" % Test + ) + ) + .dependsOn(core % "test->test;compile->compile", otelContext % "test->test;compile->compile") + lazy val documentation: Project = (project in file("generated-doc")) // important: it must not be doc/ .enablePlugins(MdocPlugin) .settings( @@ -157,5 +171,6 @@ lazy val documentation: Project = (project in file("generated-doc")) // importan flowReactiveStreams, flowJson, cron, - otelContext + otelContext, + otel ) diff --git a/core/src/main/scala/ox/flow/FlowOps.scala b/core/src/main/scala/ox/flow/FlowOps.scala index 23de8a7e..92cde0c8 100644 --- a/core/src/main/scala/ox/flow/FlowOps.scala +++ b/core/src/main/scala/ox/flow/FlowOps.scala @@ -293,8 +293,6 @@ class FlowOps[+T]: val base: Flow[U] = expand[U](u => Iterator.single(u) ++ extrapolator(u)) initial.fold(base)(e => Flow.fromValues(e).concat(base)) - // - /** Applies the given mapping function `f` to each element emitted by this flow. The returned flow then emits the results. * * @param f diff --git a/doc/index.md b/doc/index.md index 9dadca2c..442b1aa3 100644 --- a/doc/index.md +++ b/doc/index.md @@ -93,6 +93,7 @@ In addition to this documentation, ScalaDocs can be browsed at [https://javadoc. integrations/mdc-logback integrations/cron4s integrations/otel-context + integrations/otel-tracing integrations/tapir integrations/sttp-client diff --git a/doc/integrations/otel-context.md b/doc/integrations/otel-context.md index 748c26d2..08204381 100644 --- a/doc/integrations/otel-context.md +++ b/doc/integrations/otel-context.md @@ -13,7 +13,7 @@ into traces, or metrics without the appropriate context. To fix this problem, the context must be propagated whenever a new virtual thread is created. One way to achieve this is by using a custom thread factory, provided by this module - `PropagatingVirtualThreadFactory`. It can be set -for the whole app when using [`OxApp`](../utils/oxapp.md), or manually through `oxThreadFactory`: +for the whole app when using [`OxApp`](../utils/oxapp.md), or manually through `setOxThreadFactory`: ```scala mdoc:compile-only import ox.* diff --git a/doc/integrations/otel-tracing.md b/doc/integrations/otel-tracing.md new file mode 100644 index 00000000..fcfa0d34 --- /dev/null +++ b/doc/integrations/otel-tracing.md @@ -0,0 +1,104 @@ +# OpenTelemetry tracing + +Dependency: + +```scala +"com.softwaremill.ox" %% "ox-telemetry" % "@VERSION@" +``` + +The module depends on [otel-context](otel-context.md), so it doesn't need to be added separately. An OpenTelemetry SDK +and exporter have to be configured by the application, as usual. + +Ox provides helpers for creating [OpenTelemetry](https://opentelemetry.io) spans around code that uses Ox's concurrency +operators, including [Flows](../streaming/flows.md). An opened span becomes a child of the current span, and becomes the +current span itself, until it's closed. + +Spans are created using an `OxTracing` instance, wrapping an OpenTelemetry `Tracer`: + +```scala mdoc:compile-only +import io.opentelemetry.api.trace.Tracer +import ox.telemetry.tracing.OxTracing + +def tracing(tracer: Tracer): OxTracing = OxTracing.native(tracer) +``` + +## Propagating the context + +The current span is kept in the OpenTelemetry `Context`, which by default isn't propagated to new virtual threads, e.g. +forks created in [`supervised`](../structured-concurrency/fork-join.md) scopes. `OxTracing.native` relies on the +`PropagatingVirtualThreadFactory` from [otel-context](otel-context.md) to do that, and prints a warning to stderr if +it's not set up. The factory can be set for the whole app when using [`OxApp`](../utils/oxapp.md), or manually through +`setOxThreadFactory`, before the thread factory is first used (that is, before any scope is started, or `OxTracing.native` +is used): + +```scala mdoc:compile-only +import ox.* +import ox.otel.context.PropagatingVirtualThreadFactory + +object MyApp extends OxApp: + override def settings: OxApp.Settings = OxApp.Settings.Default.copy( + threadFactory = Some(PropagatingVirtualThreadFactory()) + ) + + def run(args: Vector[String])(using Ox): ExitCode = ExitCode.Success +``` + +If you want to take care of context propagation yourself, use `OxTracing.custom`. + +## Spans around code and flows + +`byNameSpan` evaluates a code block within a span. The span is ended once the block completes, and if the block throws, +the exception is recorded on the span, and its status is set to `ERROR` (this includes an interruption). `Flow.span` +does the same for a [`Flow`](../streaming/flows.md): the span is opened when the flow is run, and ended when the flow +completes, fails, or is interrupted. Downstream stages are run within the span as well, and running the flow multiple +times opens a new span each time. + +In the example below, the `process-numbers` span is created in a fork, and thanks to the context being propagated, it's +a child of the `request` span: + +```scala mdoc:compile-only +import ox.* +import ox.flow.Flow +import ox.telemetry.tracing.* + +def example(tracing: OxTracing): Unit = + val result = supervised { + fork(Flow.fromValues(1, 2, 3).span(tracing)("process-numbers").runToList()).join() + }.byNameSpan(tracing)("request") + println(result) +``` + +Both accept optional `spanKind`, `attributes` and `links`, as well as a `beforeClose` callback, called right before the +span is closed, e.g. to set attributes or add events. The callback receives the span and the exception thrown by the +spanned code, if any, and has to return the span. If it throws, the exception is added as suppressed to the one thrown +by the spanned code; the span is closed in either case: + +```scala mdoc:compile-only +import ox.flow.Flow +import ox.telemetry.tracing.* + +def withAttributes(tracing: OxTracing): List[Int] = + Flow + .fromValues(1, 2, 3) + .span(tracing)("process-numbers", beforeClose = (span, error) => span.setAttribute("failed", error.isDefined)) + .runToList() +``` + +## Manual spans + +`spanUnsafe` opens a span, which has to be closed by the caller. Closing the returned `AutoCloseableSpan` ends the span +and restores the previously current span. It's unsafe, as forgetting to close it leaves the span open and the current +span unrestored; use it with [`useCloseableInScope`](../utils/resources.md), so that closing is guaranteed. +`currentSpanUnsafe` returns the current span, e.g. to add attributes: + +```scala mdoc:compile-only +import ox.* +import ox.telemetry.tracing.* + +def manual(tracing: OxTracing): Unit = + resourceScope { + val opened = useCloseableInScope(spanUnsafe(tracing)("manual-span")) + opened.span.setAttribute("key", "value") + currentSpanUnsafe(tracing).addEvent("working"): Unit + } +``` diff --git a/otel/src/main/scala/ox/telemetry/ContextHolder.scala b/otel/src/main/scala/ox/telemetry/ContextHolder.scala new file mode 100644 index 00000000..c3ab3379 --- /dev/null +++ b/otel/src/main/scala/ox/telemetry/ContextHolder.scala @@ -0,0 +1,48 @@ +package ox.telemetry + +import io.opentelemetry.context.{Context, Scope} +import ox.otel.context.PropagatingVirtualThreadFactory + +/** Abstraction over how [[ox.telemetry.tracing.OxTracing]] interacts with OpenTelemetry [[Context]]. + */ +sealed trait ContextHolder: + /** Returns the current [[Context]]. */ + def get(): Context + + /** Makes the given `context` current. Closing the returned [[Scope]] restores the previously current context. */ + def set(context: Context): Scope + +private[telemetry] object ContextHolder: + + /** Delegates to OpenTelemetry Context. Propagating the context to new (virtual) threads is left to the user. + */ + private[telemetry] object Delegating extends ContextHolder: + + override def get(): Context = Context.current() + + override def set(context: Context): Scope = context.makeCurrent() + + /** As [[Delegating]], but warns when [[ox.oxThreadFactory]] is not a [[PropagatingVirtualThreadFactory]], as in that case the context + * won't be propagated to forks. + */ + private[telemetry] object CheckingThreadFactory extends ContextHolder: + @volatile private var checked = false + private val wrapped: ContextHolder = Delegating + + override def get(): Context = + checkThreadFactory() + wrapped.get() + + override def set(context: Context): Scope = + checkThreadFactory() + wrapped.set(context) + + private def checkThreadFactory() = + if !checked then + if !ox.oxThreadFactory.isInstanceOf[PropagatingVirtualThreadFactory] then + System.err.println( + s"Default OpenTelemetry integration should use ${classOf[PropagatingVirtualThreadFactory].getName}, if you want to use OpenTelemetry with your own oxThreadFactory, you need to take care of propagation of OpenTelemetry context to newly created VirtualThreads" + ) + checked = true + end CheckingThreadFactory +end ContextHolder diff --git a/otel/src/main/scala/ox/telemetry/tracing/AutoCloseableSpan.scala b/otel/src/main/scala/ox/telemetry/tracing/AutoCloseableSpan.scala new file mode 100644 index 00000000..023c7aa6 --- /dev/null +++ b/otel/src/main/scala/ox/telemetry/tracing/AutoCloseableSpan.scala @@ -0,0 +1,19 @@ +package ox.telemetry.tracing + +import io.opentelemetry.api.trace.Span +import io.opentelemetry.context.Scope + +/** An opened [[Span]], which is also the current one, together with the [[Scope]] that made it current. + * + * Calling [[close]] ends the span and closes the scope, restoring the previously current span. Prefer scoped usage, e.g. + * [[ox.useCloseableInScope]], so that the span is always closed. + */ +sealed trait AutoCloseableSpan extends AutoCloseable: + /** The opened span, e.g. to set attributes or add events. */ + def span: Span + +private object AutoCloseableSpan: + private[ox] final class Impl(val span: Span, openedScope: Scope) extends AutoCloseableSpan: + def close(): Unit = + span.end() + openedScope.close() diff --git a/otel/src/main/scala/ox/telemetry/tracing/OxTracing.scala b/otel/src/main/scala/ox/telemetry/tracing/OxTracing.scala new file mode 100644 index 00000000..c1966039 --- /dev/null +++ b/otel/src/main/scala/ox/telemetry/tracing/OxTracing.scala @@ -0,0 +1,105 @@ +package ox.telemetry.tracing + +import io.opentelemetry.api.common.Attributes +import io.opentelemetry.api.trace.{Span, SpanContext, SpanKind, StatusCode, Tracer} +import ox.telemetry.ContextHolder +import ox.telemetry.tracing.flow.OxFlowTracing + +sealed trait OxTracing extends OxFlowTracing: + + protected def contextHolder: ContextHolder + + protected def tracer: Tracer + + /** Returns the current span. This is not unsafe per se, but the caller is responsible for using the span properly, e.g. not ending a span + * that is managed elsewhere. + */ + def currentSpanUnsafe(): Span = Span.fromContext(contextHolder.get()) + + /** Opens a new [[Span]] as a child of the current span. The new span becomes the current one. + * + * Unsafe, because the caller is responsible for closing the returned [[AutoCloseableSpan]], which ends the span and restores the + * previously current one. Scoped usage, e.g. [[ox.useCloseableInScope]], guarantees that. + * + * @return + * An [[AutoCloseableSpan]], which closes the opened span. + */ + def spanUnsafe( + spanName: String, + spanKind: SpanKind = SpanKind.INTERNAL, + attributes: Attributes = Attributes.empty(), + links: Seq[SpanContext] = Seq.empty + ): AutoCloseableSpan = + val previousContext = contextHolder.get() + val builder = tracer.spanBuilder(spanName).setSpanKind(spanKind).setAllAttributes(attributes).setParent(previousContext) + val withLinks = links.foldLeft(builder)((b, l) => b.addLink(l)) + val started = withLinks.startSpan() + val scope = contextHolder.set(previousContext.`with`(started)) + + AutoCloseableSpan.Impl(started, scope) + end spanUnsafe + + /** Evaluates `t` within a new [[Span]]. The span is opened before `t` is evaluated, as a child of the current span, and becomes the + * current one. It is ended once `t` completes, successfully or not, and the previously current span is restored. + * + * If `t` throws, the exception is recorded on the span and the span's status is set to [[io.opentelemetry.api.trace.StatusCode.ERROR]], + * before `beforeClose` is called. The exception is then re-thrown. + * + * To span a [[ox.flow.Flow]], use [[OxFlowTracing.span]]. + * + * @param beforeClose + * Run right before the span is closed, e.g. to set attributes or add events. The second element of the argument is the exception + * thrown by `t`, if any. If `beforeClose` itself throws, the exception is added as suppressed to the one thrown by `t`, or thrown if + * `t` succeeded. The span is closed in either case. + */ + def byNameSpan[T](t: => T)( + spanName: String, + spanKind: SpanKind = SpanKind.INTERNAL, + attributes: Attributes = Attributes.empty(), + links: Seq[SpanContext] = Seq.empty, + beforeClose: ((Span, Option[Throwable])) => Span = _._1 + ): T = + var error: Throwable | Null = null + var finalizer: AutoCloseableSpan | Null = null + try + val finalize = spanUnsafe(spanName, spanKind, attributes, links) + finalizer = finalize + t + catch + case e: Throwable => + error = e + if finalizer ne null then + finalizer.nn.span.recordException(e) + finalizer.nn.span.setStatus(StatusCode.ERROR, e.getMessage): Unit + throw e + finally + try if finalizer ne null then beforeClose(finalizer.nn.span, Option(error)): Unit + catch + case e: Throwable => + if error ne null then error.nn.addSuppressed(e) + else throw e + finally if finalizer ne null then finalizer.nn.close() + end try + end try + + end byNameSpan +end OxTracing + +object OxTracing: + + /** Tracing that relies on the [[ox.otel.context.PropagatingVirtualThreadFactory]] to propagate the context to forks. A warning is printed + * to stderr on first use, if the thread factory is not set up. + * + * @see + * [[ox.OxApp.Settings.threadFactory]], [[ox.setOxThreadFactory]] + */ + def native(tracer: Tracer): OxTracing = Impl(tracer, ContextHolder.CheckingThreadFactory) + + /** Tracing which leaves the propagation of the [[io.opentelemetry.context.Context]] to new threads to the user. For an out-of-the-box + * integration, use [[native]] together with the [[ox.otel.context.PropagatingVirtualThreadFactory]]. + */ + def custom(tracer: Tracer): OxTracing = Impl(tracer, ContextHolder.Delegating) + + private final class Impl(val tracer: Tracer, val contextHolder: ContextHolder) extends OxTracing + +end OxTracing diff --git a/otel/src/main/scala/ox/telemetry/tracing/flow/OxFlowTracing.scala b/otel/src/main/scala/ox/telemetry/tracing/flow/OxFlowTracing.scala new file mode 100644 index 00000000..f4cffa2e --- /dev/null +++ b/otel/src/main/scala/ox/telemetry/tracing/flow/OxFlowTracing.scala @@ -0,0 +1,29 @@ +package ox.telemetry.tracing.flow + +import io.opentelemetry.api.common.Attributes +import io.opentelemetry.api.trace.{Span, SpanContext, SpanKind} +import ox.flow.Flow +import ox.telemetry.tracing.OxTracing + +private[tracing] trait OxFlowTracing: + self: OxTracing => + + /** Returns a flow, which when run, evaluates `flow` within a new span. The span is opened when the flow is run, as a child of the span + * current at that point. It is ended when the flow completes, fails or is interrupted. Exceptions are recorded on the span as in + * [[OxTracing.byNameSpan]]. + * + * The parameters are the same as for [[OxTracing.byNameSpan]]. + */ + def span[T](flow: Flow[T])( + spanName: String, + spanKind: SpanKind = SpanKind.INTERNAL, + attributes: Attributes = Attributes.empty(), + links: Seq[SpanContext] = Seq.empty, + beforeClose: ((Span, Option[Throwable])) => Span = _._1 + ): Flow[T] = Flow.usingEmit: emit => + self.byNameSpan( + flow.runToEmit( + emit + ) + )(spanName, spanKind, attributes, links, beforeClose) +end OxFlowTracing diff --git a/otel/src/main/scala/ox/telemetry/tracing/flow/flowTracing.scala b/otel/src/main/scala/ox/telemetry/tracing/flow/flowTracing.scala new file mode 100644 index 00000000..e5c3d11f --- /dev/null +++ b/otel/src/main/scala/ox/telemetry/tracing/flow/flowTracing.scala @@ -0,0 +1,19 @@ +package ox.telemetry.tracing.flow + +import io.opentelemetry.api.common.Attributes +import io.opentelemetry.api.trace.{Span, SpanContext, SpanKind} +import ox.flow.Flow +import ox.telemetry.tracing.flow.OxFlowTracing + +extension [T](flow: Flow[T]) + /** Opens a span around the run of this flow, using the given [[ox.telemetry.tracing.OxTracing]], see + * [[ox.telemetry.tracing.OxTracing.span]]. + */ + def span(oxTracing: OxFlowTracing)( + spanName: String, + spanKind: SpanKind = SpanKind.INTERNAL, + attributes: Attributes = Attributes.empty(), + links: Seq[SpanContext] = Seq.empty, + beforeClose: ((Span, Option[Throwable])) => Span = _._1 + ): Flow[T] = oxTracing.span(flow)(spanName, spanKind, attributes, links, beforeClose) +end extension diff --git a/otel/src/main/scala/ox/telemetry/tracing/tracing.scala b/otel/src/main/scala/ox/telemetry/tracing/tracing.scala new file mode 100644 index 00000000..3b202d68 --- /dev/null +++ b/otel/src/main/scala/ox/telemetry/tracing/tracing.scala @@ -0,0 +1,29 @@ +package ox.telemetry.tracing + +import io.opentelemetry.api.common.Attributes +import io.opentelemetry.api.trace.{Span, SpanContext, SpanKind} +import ox.telemetry.tracing.OxTracing + +extension [T](t: => T) + /** Evaluates the by-name argument within a new span, see [[OxTracing.byNameSpan]]. */ + def byNameSpan(oxTracing: OxTracing)( + spanName: String, + spanKind: SpanKind = SpanKind.INTERNAL, + attributes: Attributes = Attributes.empty(), + links: Seq[SpanContext] = Seq.empty, + beforeClose: ((Span, Option[Throwable])) => Span = _._1 + ): T = oxTracing.byNameSpan(t)(spanName, spanKind, attributes, links, beforeClose) +end extension + +export ox.telemetry.tracing.flow.span + +/** Returns the current span, see [[OxTracing.currentSpanUnsafe]]. */ +def currentSpanUnsafe(oxTracing: OxTracing) = oxTracing.currentSpanUnsafe() + +/** Opens a new span, which has to be closed by the caller, see [[OxTracing.spanUnsafe]]. */ +def spanUnsafe(oxTracing: OxTracing)( + spanName: String, + spanKind: SpanKind = SpanKind.INTERNAL, + attributes: Attributes = Attributes.empty(), + links: Seq[SpanContext] = Seq.empty +): AutoCloseableSpan = oxTracing.spanUnsafe(spanName, spanKind, attributes, links) diff --git a/otel/src/test/scala/OxTelemetryTest.scala b/otel/src/test/scala/OxTelemetryTest.scala new file mode 100644 index 00000000..579c5c87 --- /dev/null +++ b/otel/src/test/scala/OxTelemetryTest.scala @@ -0,0 +1,416 @@ +import io.opentelemetry.api.metrics.Meter +import io.opentelemetry.api.trace.StatusCode +import io.opentelemetry.context.Context +import io.opentelemetry.sdk.OpenTelemetrySdk +import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter +import io.opentelemetry.sdk.trace.SdkTracerProvider +import io.opentelemetry.sdk.trace.`export`.SimpleSpanProcessor +import org.scalatest.BeforeAndAfterAll +import org.scalatest.matchers.should.Matchers +import org.scalatest.time.SpanSugar.convertIntToGrainOfTime +import org.scalatest.wordspec.AnyWordSpec +import ox.* +import ox.channels.BufferCapacity +import ox.either.ok +import ox.flow.Flow +import ox.otel.context.PropagatingVirtualThreadFactory +import ox.telemetry.tracing.* + +import java.util.UUID +import java.util.concurrent.Semaphore +import scala.collection.JavaConverters.asScalaBufferConverter +import scala.util.boundary.Label + +class OxTelemetryTest extends AnyWordSpec with Matchers with BeforeAndAfterAll: + override def beforeAll(): Unit = + setOxThreadFactory(new PropagatingVirtualThreadFactory) + + val exporter = InMemorySpanExporter.create() + + val tracerProvider = + SdkTracerProvider + .builder() + .addSpanProcessor(SimpleSpanProcessor.create(exporter)) + .build() + + val openTelemetry = + OpenTelemetrySdk + .builder() + .setTracerProvider(tracerProvider) + .build() + + val tracer = openTelemetry.getTracer("test") + + val myTracing: OxTracing = OxTracing.native(tracer) + + enum Level: + case Level0 + case Level1 + case Level1_First + case Level1_First_Child + + import Level.* + + class Fixture: + val prefix = s"[${UUID.randomUUID()}]" + val flow = Flow.fromValues(1, 2, 3) + + def testSpans = exporter.getFinishedSpanItems.asScala.filter(_.getName.startsWith(prefix)) + + def spansOf(level: Level) = testSpans.filter(_.getName.endsWith(level.toString)) + + def assertContextIsRoot() = assert(Context.current() == Context.root) + + def assertNesting(): Unit = + val level0 = spansOf(Level0) + val level1First = spansOf(Level1_First) + + level0.foreach { s => + assert(!s.getParentSpanContext.isValid) + } + + (spansOf(Level1) ++ level1First).foreach { s => + assert(level0.exists(_.getSpanId == s.getParentSpanId)) + } + + spansOf(Level1_First_Child).foreach { s => + assert(level1First.exists(_.getSpanId == s.getParentSpanId)) + } + end assertNesting + + def assertNestingAndRoot(): Unit = + assertNesting() + assertContextIsRoot() + end Fixture + + "Flow.span" when { + "run normally" should { + "not open a span until the flow is run" in new Fixture: + assert(testSpans.isEmpty) + assertNestingAndRoot() + + "open a single span for a multi-element flow" in new Fixture: + flow.span(myTracing)(s"$prefix-$Level0").runDrain() + + assert(spansOf(Level0).size == 1) + assert(testSpans.size == 1) + assertNestingAndRoot() + + "open one span per span() call when spans are nested" in new Fixture: + supervised { + val spanned1 = flow.span(myTracing)(s"$prefix-$Level1") ++ Flow.fromValues(4, 2) + + spanned1.span(myTracing)(s"$prefix-$Level0").runDrain() + + assert(spansOf(Level0).size == 1) + assert(spansOf(Level1).size == 1) + assert(testSpans.size == 2) + assertNestingAndRoot() + } + + "open a single span when downstream uses mapPar" in new Fixture: + flow.span(myTracing)(s"$prefix-$Level0").mapPar(3)(_ + 1).runDrain() + + assert(testSpans.size == 1) + + val only = spansOf(Level0).head + assert(only.getStatus.getStatusCode == StatusCode.UNSET) + assert(only.getEvents.isEmpty) + assertNestingAndRoot() + } + + "used under parent spans" should { + "use the current span as parent" in new Fixture: + val parent = myTracing.spanUnsafe(s"$prefix-$Level0") + + flow.span(myTracing)(s"$prefix-$Level1").runToList(): Unit + + parent.close() + + assert(spansOf(Level0).size == 1) + assert(spansOf(Level1).size == 1) + assertNestingAndRoot() + + "open one child span per element when flatMapped under a parent span" in new Fixture: + val spanned = flow.span(myTracing)(s"$prefix-$Level1") + + flow.flatMap { _ => spanned }.span(myTracing)(s"$prefix-$Level0").runToList(): Unit + + assert(spansOf(Level0).size == 1) + assert(spansOf(Level1).size == 3) + assertNestingAndRoot() + + "parent spans of concurrently run flows under the enclosing span" in new Fixture: + val spanned = flow.span(myTracing)(s"$prefix-$Level1") + + val parent = spanUnsafe(myTracing)(s"$prefix-$Level0") + + supervised { + (1 to 10).foreach(_ => fork(spanned.runToList())) + } + + parent.close() + + assert(spansOf(Level0).size == 1) + assert(spansOf(Level1).size == 10) + assertNestingAndRoot() + + "nest spans across forks and inner supervised scopes" in new Fixture: + val spanned = flow.span(myTracing)(s"$prefix-$Level1") + + supervised { + (1 to 10).map { + case 1 => + forkUser( + flow + .span(myTracing)(s"$prefix-$Level1_First") + .flatMap { el => + val innerFlow = supervised { + flow.span(myTracing)(s"$prefix-$Level1_First_Child", beforeClose = _._1.addEvent("boomed")) + } + innerFlow + } + .runToList() + ) + + case _ => forkUser(spanned.runToList()) + } + }.byNameSpan(myTracing)(s"$prefix-$Level0") + + assert(spansOf(Level1).size == 9) + assert(spansOf(Level1_First).size == 1) + assert(spansOf(Level1_First_Child).size == 3) + + assertNestingAndRoot() + } + + "a stage fails" should { + "record exception and set ERROR status when the flow fails" in new Fixture: + try + flow + .tap { case 2 => throw new Exception("boom"); case _ => () } + .span(myTracing)(s"$prefix-$Level0", beforeClose = (s, thr) => s.addEvent(thr.toString)) + .runToList() + catch case _ => () + + assert(testSpans.size == 1) + + val failed = spansOf(Level0).head + assert(failed.getStatus.getStatusCode == StatusCode.ERROR) + assert(failed.getStatus.getDescription == "boom") + assert(failed.getEvents.asScala.exists(_.getName == "exception")) + assertNestingAndRoot() + + "end span when a downstream stage fails after span" in new Fixture: + val spanned = flow.span(myTracing)(s"$prefix-$Level0", beforeClose = (s, thr) => s.addEvent(thr.toString)) + + try spanned.map { case 2 => throw new Exception("boom"); case n => n }.runToList() + catch case _ => () + + assert(spansOf(Level0).size == 1) + assert(testSpans.size == 1) + assertNestingAndRoot() + + "end span when downstream failure is recovered" in new Fixture: + flow.span(myTracing)(s"$prefix-$Level0").map { case 2 => throw new Exception("boom"); case x => x }.recover(_ => 5).runToList() + + assert(spansOf(Level0).size == 1) + assert(testSpans.size == 1) + assertNestingAndRoot() + + "keep the flow span as a child of the recovery span when failure is recovered" in new Fixture: + val spanned = flow.span(myTracing)(s"$prefix-$Level1") + + spanned + .map { case 2 => throw new Exception("boom"); case x => currentSpanUnsafe(myTracing).addEvent(x.toString); x } + .recover { x => + currentSpanUnsafe(myTracing).addEvent("recovery"); 5 + } + .span(myTracing)(s"$prefix-$Level0") + .runToList(): Unit + + assert(spansOf(Level0).size == 1) + assert(spansOf(Level1).size == 1) + assertNestingAndRoot() + + "end spans of concurrent flows when one of them fails" in new Fixture: + val spanned = flow.span(myTracing)(s"$prefix-$Level1") + + val parent = spanUnsafe(myTracing)(s"$prefix-$Level0") + + try + supervised { + (1 to 10).foreach { + case 1 => forkUser(spanned.tap(_ => throw new Exception("fds")).runToList()) + case _ => forkUser(spanned.runToList()) + } + } + catch case _ => () + end try + parent.close() + + assert(spansOf(Level0).size == 1) + assertNestingAndRoot() + + "end inner spans with event when nested flow fails in supervisedError" in new Fixture: + val em = EitherMode[String] + + val level0 = spanUnsafe(myTracing)(s"$prefix-$Level0") + + val spanned = flow.span(myTracing)(s"$prefix-$Level1") + + try + supervised { + (1 to 10).map { + case 1 => + forkUser { + flow + .span(myTracing)(s"$prefix-$Level1_First") + .flatMap { el => + { + val inner = supervisedError(em): + either: + val innerFlow = + em.pure( + flow.span(myTracing)(s"$prefix-$Level1_First_Child", beforeClose = _._1.addEvent(s"boomed at $el")).map { + case 1 => em.pureError("ops").ok(); + case x => x + } + ) + innerFlow.ok() + inner + }.toOption.get + } + .runToList() + } + case _ => forkUser(spanned.runToList()) + } + } + catch case _ => () + end try + + level0.close() + assert(spansOf(Level1_First_Child).size == 1) // because flow.flatMap is evaluated sequentially and exception ends whole flow + assertNestingAndRoot() + + "end inner spans with event when nested flow fails and failure is replaced by empty flow" in new Fixture: + val em = EitherMode[String] + + val level0 = spanUnsafe(myTracing)(s"$prefix-$Level0") + + val spanned = flow.span(myTracing)(s"$prefix-$Level1") + + try + supervised { + (1 to 10).map { + case 1 => + forkUser { + flow + .span(myTracing)(s"$prefix-$Level1_First") + .flatMap { el => + { + val inner = supervisedError(em): + either: + val innerFlow = + em.pure( + flow.span(myTracing)(s"$prefix-$Level1_First_Child", beforeClose = _._1.addEvent(s"boomed at $el")).map { + case 1 => em.pureError("ops").ok(); + case x => x + } + ) + innerFlow.ok() + inner + }.toOption.getOrElse(Flow.empty) + } + .runToList() + } + case _ => forkUser(spanned.runToList()) + } + } + catch case _ => () + end try + + level0.close() + assert(spansOf(Level1_First_Child).size == 1) // because flow.flatMap is evaluated sequentially and exception ends whole flow + assertNestingAndRoot() + + "end inner spans with event when nested flow fails inside enclosing supervisedError" in new Fixture: + val em = EitherMode[String] + + val level0 = spanUnsafe(myTracing)(s"$prefix-$Level0") + + val spanned = flow.span(myTracing)(s"$prefix-$Level1") + + try + supervisedError(em) { + either: + em.pure: + (1 to 10).map { + case 1 => + forkUser { + flow + .span(myTracing)(s"$prefix-$Level1_First") + .flatMap { el => + { + supervisedError(em): + val innerFlow = + em.pure( + flow.span(myTracing)(s"$prefix-$Level1_First_Child", beforeClose = _._1.addEvent(s"boomed at $el")).map { + case 1 => em.pureError("ops").ok(); + case x => x + } + ) + innerFlow + }.toOption.get + } + .runToList() + } + case _ => forkUser(spanned.runToList()) + } + } + catch case _ => () + end try + + level0.close() + assert(spansOf(Level1_First_Child).size == 1) // because flow.flatMap is evaluated sequentially and exception ends whole flow + assertNestingAndRoot() + } + + "cancelled" should { + "end span when enclosing supervised scope fails before stream completion" in new Fixture: + val semaphore = Semaphore(0) + + try + supervised { + forkUser { + semaphore.acquire() + throw new Exception("oops") + } + forkUser { + flow.span(myTracing)(s"$prefix-$Level0").tap { case 1 => semaphore.release(); case _ => sleep(100.seconds) }.runToList() + } + } + catch case err => () + end try + + assert(spansOf(Level0).size == 1) + assert(testSpans.size == 1) + assertNestingAndRoot() + + "end span when the enclosing fork is cancelled" in new Fixture: + val semaphore = Semaphore(0) + + supervised { + val forkWithSpan = forkCancellable { + flow.span(myTracing)(s"$prefix-$Level0").tap { case 1 => semaphore.release(); case _ => sleep(100.seconds) }.runToList() + } + semaphore.acquire() + forkWithSpan.cancelNow() + } + + assert(spansOf(Level0).size == 1) + assert(testSpans.size == 1) + assertNestingAndRoot() + } + } +end OxTelemetryTest