Make OpenTelemetryTracingBackend generic - working for any effect (#2494)

Author: adamw

build.sbt

@@ -152,7 +152,7 @@ val zio2InteropRsVersion = "2.0.2"
 
 val oxVersion = "0.5.7"
 val sttpModelVersion = "1.7.12"
-val sttpSharedVersion = "1.4.2"
+val sttpSharedVersion = "1.5.0"
 
 val logback = "ch.qos.logback" % "logback-classic" % "1.5.14"
 

docs/backends/wrappers/opentelemetry.md

@@ -3,7 +3,7 @@
 Currently, the following OpenTelemetry features are supported:
 
 - metrics using `OpenTelemetryMetricsBackend`, wrapping any other backend
-- tracing using `OpenTelemetryTracingSyncBackend`, wrapping a synchronous backend
+- tracing using `OpenTelemetryTracingBackend`, wrapping a synchronous backend
 - tracing using `OpenTelemetryTracingZioBackend`, wrapping any ZIO2 backend
 - tracing using [trace4cats](https://github.com/trace4cats/trace4cats), wrapping a cats-effect backend
 
@@ -51,7 +51,7 @@ OpenTelemetryMetricsBackend(
 )
 ```
 
-## Tracing (synchronous)
+## Tracing 
 
 To use, add the following dependency to your project:
 
@@ -71,11 +71,10 @@ Other aspects of the backend can be configured as well:
 * how request, response, error attributes are computed
 
 ```{note}
-The backend relies on context passed using the default mechanism in OpenTelemetry SDK for Java, that is using 
-`ThreadLocal`s. This means that the backend will **not** work properly when combined with any asynchronous
-execution mechanisms, such as `Future`s.
-
-Moreover, for Java 21+, note that `ThreadLocal`s are not inherited when spawning a new virtual thread.
+Relies on the built-in OpenTelemetry Java SDK `ContextStorage` mechanism of propagating the tracing context;
+by default, this is using `ThreadLocal`s, which works with synchronous/direct-style environments. `Future`s are 
+supported through instrumentation provided by the OpenTelemetry javaagent. For functional effect systems, usually 
+a dedicated integration library is required.
 ```
 
 Example usage:
@@ -88,9 +87,9 @@ import io.opentelemetry.api.OpenTelemetry
 val sttpBackend: SyncBackend = ???
 val openTelemetry: OpenTelemetry = ???
 
-OpenTelemetryTracingSyncBackend(
+OpenTelemetryTracingBackend(
   sttpBackend,
-  OpenTelemetryTracingSyncConfig(
+  OpenTelemetryTracingConfig(
     openTelemetry,
     spanName = request => request.uri.pathSegments.segments.headOption.map(_.v).getOrElse("root")
   )

observability/opentelemetry-backend/src/main/scala/sttp/client4/opentelemetry/OpenTelemetryTracingBackend.scala

@@ -0,0 +1,137 @@
+package sttp.client4.opentelemetry
+
+import io.opentelemetry.api.OpenTelemetry
+import io.opentelemetry.context.Context
+import io.opentelemetry.context.propagation.TextMapSetter
+import sttp.capabilities.Effect
+import sttp.client4.Backend
+import sttp.client4.GenericBackend
+import sttp.client4.GenericRequest
+import sttp.client4.Response
+import sttp.client4.ResponseException
+import sttp.client4.StreamBackend
+import sttp.client4.SyncBackend
+import sttp.client4.WebSocketBackend
+import sttp.client4.WebSocketStreamBackend
+import sttp.client4.wrappers.DelegateBackend
+import sttp.client4.wrappers.FollowRedirectsBackend
+import sttp.monad.syntax._
+
+import scala.collection.mutable
+
+/** A backend wrapper which traces requests using OpenTelemetry.
+  *
+  * Span names and attributes are calculated using the provided [[OpenTelemetryTracingConfig]].
+  *
+  * To use, wrap your backend using this one, e.g.:
+  *
+  * {{{
+  * val rawBackend: SyncBackend = ???
+  * val openTelemetry: OpenTelemetry = ???
+  *
+  * val tracingBackend = OpenTelemetryTracingBackend(rawBackend, openTelemetry)
+  * }}}
+  *
+  * Relies on the built-in OpenTelemetry Java SDK [[io.opentelemetry.context.ContextStorage]] mechanism of propagating
+  * the tracing context; by default, this is using [[ThreadLocal]]s, which works with synchronous/direct-style
+  * environments. [[scala.concurrent.Future]]s are supported through instrumentation provided by the OpenTelemetry
+  * javaagent. For functional effect systems, usually a dedicated integration library is required.
+  */
+class OpenTelemetryTracingBackend[F[_], P](delegate: GenericBackend[F, P], config: OpenTelemetryTracingConfig)
+    extends DelegateBackend[F, P](delegate) {
+
+  private val setter = new TextMapSetter[mutable.Map[String, String]] {
+    def set(carrier: mutable.Map[String, String], key: String, value: String): Unit = {
+      val _ = carrier.put(key, value)
+    }
+  }
+
+  override def send[T](request: GenericRequest[T, P with Effect[F]]): F[Response[T]] = {
+    monad
+      .eval {
+        config.tracer
+          .spanBuilder(config.spanName(request))
+          .setAllAttributes(config.requestAttributes(request))
+          .startSpan()
+      }
+      .flatMap { span =>
+        monad.ensure2(
+          {
+            val scope = span.makeCurrent()
+            try {
+              val carrier = mutable.Map.empty[String, String]
+              config.propagators.getTextMapPropagator().inject(Context.current(), carrier, setter)
+
+              val requestWithTraceContext = request.headers(carrier.toMap)
+
+              delegate
+                .send(requestWithTraceContext)
+                .map { response =>
+                  span.setAllAttributes(config.responseAttributes(request, response))
+                  response
+                }
+                .handleError { case e: Throwable =>
+                  ResponseException.find(e) match {
+                    case Some(re) =>
+                      span.setAllAttributes(
+                        config.responseAttributes(request, Response((), re.response.code, request.onlyMetadata))
+                      )
+                    case _ =>
+                      span.setAllAttributes(config.errorAttributes(e))
+                  }
+                  monad.error(e)
+                }
+            } finally {
+              scope.close()
+            }
+          }, {
+            monad.eval(span.end())
+          }
+        )
+      }
+  }
+}
+
+object OpenTelemetryTracingBackend {
+  def apply(delegate: SyncBackend, openTelemetry: OpenTelemetry): SyncBackend =
+    apply(delegate, OpenTelemetryTracingConfig(openTelemetry))
+
+  def apply[F[_]](delegate: Backend[F], openTelemetry: OpenTelemetry): Backend[F] =
+    apply(delegate, OpenTelemetryTracingConfig(openTelemetry))
+
+  def apply[F[_]](delegate: WebSocketBackend[F], openTelemetry: OpenTelemetry): WebSocketBackend[F] =
+    apply(delegate, OpenTelemetryTracingConfig(openTelemetry))
+
+  def apply[F[_], S](delegate: StreamBackend[F, S], openTelemetry: OpenTelemetry): StreamBackend[F, S] =
+    apply(delegate, OpenTelemetryTracingConfig(openTelemetry))
+
+  def apply[F[_], S](
+      delegate: WebSocketStreamBackend[F, S],
+      openTelemetry: OpenTelemetry
+  ): WebSocketStreamBackend[F, S] =
+    apply(delegate, OpenTelemetryTracingConfig(openTelemetry))
+
+  def apply(delegate: SyncBackend, config: OpenTelemetryTracingConfig): SyncBackend = {
+    // redirects should be handled before tracing
+    FollowRedirectsBackend(new OpenTelemetryTracingBackend(delegate, config) with SyncBackend)
+  }
+
+  def apply[F[_]](delegate: Backend[F], config: OpenTelemetryTracingConfig): Backend[F] = {
+    FollowRedirectsBackend(new OpenTelemetryTracingBackend(delegate, config) with Backend[F])
+  }
+
+  def apply[F[_]](delegate: WebSocketBackend[F], config: OpenTelemetryTracingConfig): WebSocketBackend[F] = {
+    FollowRedirectsBackend(new OpenTelemetryTracingBackend(delegate, config) with WebSocketBackend[F])
+  }
+
+  def apply[F[_], S](delegate: StreamBackend[F, S], config: OpenTelemetryTracingConfig): StreamBackend[F, S] = {
+    FollowRedirectsBackend(new OpenTelemetryTracingBackend(delegate, config) with StreamBackend[F, S])
+  }
+
+  def apply[F[_], S](
+      delegate: WebSocketStreamBackend[F, S],
+      config: OpenTelemetryTracingConfig
+  ): WebSocketStreamBackend[F, S] = {
+    FollowRedirectsBackend(new OpenTelemetryTracingBackend(delegate, config) with WebSocketStreamBackend[F, S])
+  }
+}

observability/opentelemetry-backend/src/main/scala/sttp/client4/opentelemetry/OpenTelemetryTracingSyncConfig.scala observability/opentelemetry-backend/src/main/scala/sttp/client4/opentelemetry/OpenTelemetryTracingConfig.scala

@@ -8,7 +8,7 @@ import sttp.client4._
 
 import java.time.Clock
 
-case class OpenTelemetryTracingSyncConfig(
+case class OpenTelemetryTracingConfig(
     tracer: Tracer,
     propagators: ContextPropagators,
     clock: Clock,
@@ -18,7 +18,7 @@ case class OpenTelemetryTracingSyncConfig(
     errorAttributes: Throwable => Attributes
 )
 
-object OpenTelemetryTracingSyncConfig {
+object OpenTelemetryTracingConfig {
   def apply(
       openTelemetry: OpenTelemetry,
       clock: Clock = Clock.systemUTC(),
@@ -27,7 +27,7 @@ object OpenTelemetryTracingSyncConfig {
       responseAttributes: (GenericRequest[_, _], Response[_]) => Attributes =
         OpenTelemetryDefaults.responseAttributes _,
       errorAttributes: Throwable => Attributes = OpenTelemetryDefaults.errorAttributes _
-  ): OpenTelemetryTracingSyncConfig = usingTracer(
+  ): OpenTelemetryTracingConfig = usingTracer(
     openTelemetry
       .tracerBuilder(OpenTelemetryDefaults.instrumentationScopeName)
       .setInstrumentationVersion(OpenTelemetryDefaults.instrumentationScopeVersion)
@@ -49,8 +49,8 @@ object OpenTelemetryTracingSyncConfig {
       responseAttributes: (GenericRequest[_, _], Response[_]) => Attributes =
         OpenTelemetryDefaults.responseAttributes _,
       errorAttributes: Throwable => Attributes = OpenTelemetryDefaults.errorAttributes _
-  ): OpenTelemetryTracingSyncConfig =
-    OpenTelemetryTracingSyncConfig(
+  ): OpenTelemetryTracingConfig =
+    OpenTelemetryTracingConfig(
       tracer,
       propagators,
       clock,

observability/opentelemetry-backend/src/main/scala/sttp/client4/opentelemetry/OpenTelemetryTracingSyncBackend.scala

@@ -13,15 +13,15 @@ import io.opentelemetry.semconv.HttpAttributes
 import io.opentelemetry.semconv.ErrorAttributes
 import scala.collection.JavaConverters._
 
-class OpenTelemetryTracingSyncBackendTest extends AnyFlatSpec with Matchers {
+class OpenTelemetryTracingBackendTest extends AnyFlatSpec with Matchers {
   it should "capture successful spans" in {
     // given
     val testExporter = InMemorySpanExporter.create()
     val tracerProvider = SdkTracerProvider.builder().addSpanProcessor(SimpleSpanProcessor.create(testExporter)).build();
     val otel = OpenTelemetrySdk.builder().setTracerProvider(tracerProvider).build()
 
     val stubBackend = SyncBackendStub.whenAnyRequest.thenRespondOk()
-    val wrappedBackend = OpenTelemetryTracingSyncBackend(stubBackend, OpenTelemetryTracingSyncConfig(otel))
+    val wrappedBackend = OpenTelemetryTracingBackend(stubBackend, OpenTelemetryTracingConfig(otel))
 
     // when
     basicRequest.get(uri"http://test.com/foo").send(wrappedBackend)
@@ -46,7 +46,7 @@ class OpenTelemetryTracingSyncBackendTest extends AnyFlatSpec with Matchers {
     val otel = OpenTelemetrySdk.builder().setTracerProvider(tracerProvider).build()
 
     val stubBackend = SyncBackendStub.whenAnyRequest.thenRespond(throw new RuntimeException("test"))
-    val wrappedBackend = OpenTelemetryTracingSyncBackend(stubBackend, OpenTelemetryTracingSyncConfig(otel))
+    val wrappedBackend = OpenTelemetryTracingBackend(stubBackend, OpenTelemetryTracingConfig(otel))
 
     // when
     intercept[RuntimeException] {