Add propagation of ZIO log annotations to OTEL metric attributes

docs/opentelemetry.md

@@ -187,6 +187,7 @@ object TracingApp extends ZIOAppDefault {
 
 To send [Metric signals](https://opentelemetry.io/docs/concepts/signals/metrics/), you will need a `Meter` service in your environment. For this, use the `OpenTelemetry.meter` layer which in turn requires an instance of `OpenTelemetry` provided by Java SDK and a suitable `ContextStorage` implementation. The `Meter` API lets you create [Counter](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/api.md#counter), [UpDownCounter](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/api.md#updowncounter), [Gauge](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/api.md#gauge), [Histogram](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/api.md#histogram) and their [asynchronous](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/api.md#asynchronous-instrument-api) (aka observable) counterparts. 
 As a rule of thumb, observable instruments must be initialized on an application startup. They are scoped, so you should not be worried about shutting them down manually.
+By default the metric instruments does not take ZIO log annotations into account. To turn it on pass `logAnnotated = true` parameter to the `OpenTelemetry.metrics` layer initializer.
 
 ```scala
 //> using scala "2.13.13"
@@ -316,12 +317,13 @@ object MetricsApp extends ZIOAppDefault {
           _                    <- messageLengthCounter.add(message.length, Attributes(Attribute.string("message", message)))
         } yield message
 
-        // By wrapping our logic into a span, we make the `messageLengthCounter` data points correlated with a "root_span" automatically
-        logic @@ tracing.aspects.root("root_span")
+        // By wrapping our logic into a span, we make the `messageLengthCounter` data points correlated with a "root_span" automatically.
+        // Additionally we implicitly add one more attribute to the `messageLenghtCounter` as it is wrapped into a `ZIO.logAnnotate` call.
+        ZIO.logAnnotate("zio", "annotation")(logic) @@ tracing.aspects.root("root_span")
       }
       .provide(
         otelSdkLayer,
-        OpenTelemetry.metrics(instrumentationScopeName),
+        OpenTelemetry.metrics(instrumentationScopeName, logAnnotated = true),
         OpenTelemetry.tracing(instrumentationScopeName),
         OpenTelemetry.contextZIO,
         tickCounterLayer,

opentelemetry/src/main/scala/zio/telemetry/opentelemetry/OpenTelemetry.scala

@@ -87,7 +87,8 @@ object OpenTelemetry {
   def metrics(
     instrumentationScopeName: String,
     instrumentationVersion: Option[String] = None,
-    schemaUrl: Option[String] = None
+    schemaUrl: Option[String] = None,
+    logAnnotated: Boolean = false
   ): URLayer[api.OpenTelemetry with ContextStorage, Meter with Instrument.Builder] = {
     val meterLayer   = ZLayer(
       ZIO.serviceWith[api.OpenTelemetry] { openTelemetry =>
@@ -99,7 +100,7 @@ object OpenTelemetry {
         builder.build()
       }
     )
-    val builderLayer = meterLayer >>> Instrument.Builder.live
+    val builderLayer = meterLayer >>> Instrument.Builder.live(logAnnotated)
 
     builderLayer >+> (builderLayer >>> Meter.live)
   }

opentelemetry/src/main/scala/zio/telemetry/opentelemetry/metrics/Counter.scala

@@ -5,7 +5,7 @@ import io.opentelemetry.api.metrics.LongCounter
 import io.opentelemetry.context.Context
 import zio._
 import zio.telemetry.opentelemetry.context.ContextStorage
-import zio.telemetry.opentelemetry.metrics.internal.Instrument
+import zio.telemetry.opentelemetry.metrics.internal.{Instrument, logAnnotatedAttributes}
 
 /**
  * A Counter instrument that records values of type `A`
@@ -42,14 +42,21 @@ trait Counter[-A] extends Instrument[A] {
 
 object Counter {
 
-  private[metrics] def long(counter: LongCounter, ctxStorage: ContextStorage): Counter[Long] =
+  private[metrics] def long(
+    counter: LongCounter,
+    ctxStorage: ContextStorage,
+    logAnnotated: Boolean
+  ): Counter[Long] =
     new Counter[Long] {
 
       override def record0(value: Long, attributes: Attributes = Attributes.empty, context: Context): Unit =
         counter.add(value, attributes, context)
 
       override def add(value: Long, attributes: Attributes = Attributes.empty)(implicit trace: Trace): UIO[Unit] =
-        ctxStorage.get.map(record0(value, attributes, _))
+        for {
+          annotated <- logAnnotatedAttributes(attributes, logAnnotated)
+          ctx       <- ctxStorage.get
+        } yield record0(value, annotated, ctx)
 
       override def inc(attributes: Attributes = Attributes.empty)(implicit trace: Trace): UIO[Unit] =
         add(1L, attributes)

opentelemetry/src/main/scala/zio/telemetry/opentelemetry/metrics/Histogram.scala

@@ -5,7 +5,7 @@ import io.opentelemetry.api.metrics.DoubleHistogram
 import io.opentelemetry.context.Context
 import zio._
 import zio.telemetry.opentelemetry.context.ContextStorage
-import zio.telemetry.opentelemetry.metrics.internal.Instrument
+import zio.telemetry.opentelemetry.metrics.internal.{Instrument, logAnnotatedAttributes}
 
 /**
  * A Histogram instrument that records values of type `A`
@@ -32,14 +32,21 @@ trait Histogram[-A] extends Instrument[A] {
 
 object Histogram {
 
-  private[metrics] def double(histogram: DoubleHistogram, ctxStorage: ContextStorage): Histogram[Double] =
+  private[metrics] def double(
+    histogram: DoubleHistogram,
+    ctxStorage: ContextStorage,
+    logAnnotated: Boolean
+  ): Histogram[Double] =
     new Histogram[Double] {
 
       override def record0(value: Double, attributes: Attributes = Attributes.empty, context: Context): Unit =
         histogram.record(value, attributes, context)
 
       override def record(value: Double, attributes: Attributes = Attributes.empty)(implicit trace: Trace): UIO[Unit] =
-        ctxStorage.get.map(record0(value, attributes, _))
+        for {
+          annotated <- logAnnotatedAttributes(attributes, logAnnotated)
+          ctx       <- ctxStorage.get
+        } yield record0(value, annotated, ctx)
 
     }
 

opentelemetry/src/main/scala/zio/telemetry/opentelemetry/metrics/UpDownCounter.scala

@@ -5,7 +5,7 @@ import io.opentelemetry.api.metrics.LongUpDownCounter
 import io.opentelemetry.context.Context
 import zio._
 import zio.telemetry.opentelemetry.context.ContextStorage
-import zio.telemetry.opentelemetry.metrics.internal.Instrument
+import zio.telemetry.opentelemetry.metrics.internal.{Instrument, logAnnotatedAttributes}
 
 /**
  * A UpDownCounter instrument that records values of type `A`
@@ -54,14 +54,21 @@ trait UpDownCounter[-A] extends Instrument[A] {
 
 object UpDownCounter {
 
-  private[metrics] def long(counter: LongUpDownCounter, ctxStorage: ContextStorage): UpDownCounter[Long] =
+  private[metrics] def long(
+    counter: LongUpDownCounter,
+    ctxStorage: ContextStorage,
+    logAnnotated: Boolean
+  ): UpDownCounter[Long] =
     new UpDownCounter[Long] {
 
       override def record0(value: Long, attributes: Attributes = Attributes.empty, context: Context): Unit =
         counter.add(value, attributes, context)
 
       override def add(value: Long, attributes: Attributes = Attributes.empty)(implicit trace: Trace): UIO[Unit] =
-        ctxStorage.get.map(record0(value, attributes, _))
+        for {
+          annotated <- logAnnotatedAttributes(attributes, logAnnotated)
+          ctx       <- ctxStorage.get
+        } yield record0(value, annotated, ctx)
 
       override def inc(attributes: Attributes = Attributes.empty)(implicit trace: Trace): UIO[Unit] =
         add(1L, attributes)

opentelemetry/src/main/scala/zio/telemetry/opentelemetry/metrics/internal/Instrument.scala

@@ -52,7 +52,7 @@ object Instrument {
 
   private[opentelemetry] object Builder {
 
-    def live: URLayer[api.metrics.Meter with ContextStorage, Builder] =
+    def live(logAnnotated: Boolean = false): URLayer[api.metrics.Meter with ContextStorage, Builder] =
       ZLayer(
         for {
           meter      <- ZIO.service[api.metrics.Meter]
@@ -69,7 +69,7 @@ object Instrument {
             unit.foreach(builder.setUnit)
             description.foreach(builder.setDescription)
 
-            Counter.long(builder.build(), ctxStorage)
+            Counter.long(builder.build(), ctxStorage, logAnnotated)
           }
 
           override def upDownCounter(
@@ -82,7 +82,7 @@ object Instrument {
             unit.foreach(builder.setUnit)
             description.foreach(builder.setDescription)
 
-            UpDownCounter.long(builder.build(), ctxStorage)
+            UpDownCounter.long(builder.build(), ctxStorage, logAnnotated)
           }
 
           override def histogram(
@@ -95,7 +95,7 @@ object Instrument {
             unit.foreach(builder.setUnit)
             description.foreach(builder.setDescription)
 
-            Histogram.double(builder.build(), ctxStorage)
+            Histogram.double(builder.build(), ctxStorage, logAnnotated)
           }
 
           override def observableCounter(

opentelemetry/src/main/scala/zio/telemetry/opentelemetry/metrics/internal/package.scala

@@ -1,6 +1,7 @@
 package zio.telemetry.opentelemetry.metrics
 
 import io.opentelemetry.api
+import zio._
 import zio.metrics.MetricLabel
 import zio.telemetry.opentelemetry.common.{Attribute, Attributes}
 
@@ -9,4 +10,18 @@ package object internal {
   private[metrics] def attributes(tags: Set[MetricLabel]): api.common.Attributes =
     Attributes(tags.map(t => Attribute.string(t.key, t.value)).toSeq: _*)
 
+  private[metrics] def logAnnotatedAttributes(attributes: api.common.Attributes, logAnnotated: Boolean)(implicit
+    trace: Trace
+  ): UIO[api.common.Attributes] =
+    if (logAnnotated)
+      for {
+        annotations <- ZIO.logAnnotations
+        annotated    = Attributes(annotations.map { case (k, v) => Attribute.string(k, v) }.toSeq: _*)
+        builder      = api.common.Attributes.builder()
+        _            = builder.putAll(annotated)
+        _            = builder.putAll(attributes)
+      } yield builder.build()
+    else
+      ZIO.succeed(attributes)
+
 }

opentelemetry/src/test/scala/zio/telemetry/opentelemetry/metrics/MeterTest.scala

@@ -16,15 +16,15 @@ object MeterTest extends ZIOSpecDefault {
   val inMemoryMetricReaderLayer: ZLayer[Any, Nothing, InMemoryMetricReader] =
     ZLayer(ZIO.succeed(InMemoryMetricReader.create()))
 
-  val meterLayer: ZLayer[InMemoryMetricReader with ContextStorage, Nothing, Meter] = {
+  def meterLayer(logAnnotated: Boolean = false): ZLayer[InMemoryMetricReader with ContextStorage, Nothing, Meter] = {
     val jmeter  = ZLayer {
       for {
         metricReader  <- ZIO.service[InMemoryMetricReader]
         meterProvider <- ZIO.succeed(SdkMeterProvider.builder().registerMetricReader(metricReader).build())
         meter         <- ZIO.succeed(meterProvider.get("MeterTest"))
       } yield meter
     }
-    val builder = jmeter >>> Instrument.Builder.live
+    val builder = jmeter >>> Instrument.Builder.live(logAnnotated)
 
     builder >>> Meter.live
   }
@@ -44,7 +44,8 @@ object MeterTest extends ZIOSpecDefault {
     suite("zio opentelemetry")(
       suite("Meter")(
         normalSpec,
-        contextualSpec
+        contextualSpec,
+        logAnnotatedSpec
       )
     )
 
@@ -131,8 +132,26 @@ object MeterTest extends ZIOSpecDefault {
             } yield assertTrue(metricValue == 14L)
           }
         )
+      },
+      test("zio log annotations are not included when turned off") {
+        ZIO.serviceWithZIO[Meter] { meter =>
+          for {
+            reader          <- ZIO.service[InMemoryMetricReader]
+            counter         <- meter.counter("test_counter")
+            _               <- ZIO.logAnnotate("zio", "annotation") {
+                                 counter.inc()
+                               }
+            metric           = reader.collectAllMetrics().asScala.toList.head
+            metricPoint      = metric.getLongSumData.getPoints.asScala.toList.head
+            metricValue      = metricPoint.getValue
+            metricAttributes = metricPoint.getAttributes()
+          } yield assertTrue(
+            metricValue == 1L,
+            metricAttributes == Attributes.empty
+          )
+        }
       }
-    ).provide(inMemoryMetricReaderLayer, meterLayer, ContextStorage.fiberRef, observableRefLayer)
+    ).provide(inMemoryMetricReaderLayer, meterLayer(), ContextStorage.fiberRef, observableRefLayer)
 
   private val contextualSpec =
     suite("contextual")(
@@ -155,6 +174,46 @@ object MeterTest extends ZIOSpecDefault {
           )
         }
       }
-    ).provide(inMemoryMetricReaderLayer, meterLayer, ContextStorage.fiberRef, TracingTest.tracingMockLayer)
+    ).provide(inMemoryMetricReaderLayer, meterLayer(), ContextStorage.fiberRef, TracingTest.tracingMockLayer)
+
+  private val logAnnotatedSpec =
+    suite("log annotated")(
+      test("new attributes") {
+        ZIO.serviceWithZIO[Meter] { meter =>
+          for {
+            reader          <- ZIO.service[InMemoryMetricReader]
+            counter         <- meter.counter("test_counter")
+            _               <- ZIO.logAnnotate("zio", "annotation") {
+                                 counter.inc()
+                               }
+            metric           = reader.collectAllMetrics().asScala.toList.head
+            metricPoint      = metric.getLongSumData.getPoints.asScala.toList.head
+            metricValue      = metricPoint.getValue
+            metricAttributes = metricPoint.getAttributes()
+          } yield assertTrue(
+            metricValue == 1L,
+            metricAttributes == Attributes(Attribute.string("zio", "annotation"))
+          )
+        }
+      },
+      test("instrumented attributes override log annotated") {
+        ZIO.serviceWithZIO[Meter] { meter =>
+          for {
+            reader          <- ZIO.service[InMemoryMetricReader]
+            counter         <- meter.counter("test_counter")
+            _               <- ZIO.logAnnotate("zio", "annotation") {
+                                 counter.inc(Attributes(Attribute.string("zio", "annotation2")))
+                               }
+            metric           = reader.collectAllMetrics().asScala.toList.head
+            metricPoint      = metric.getLongSumData.getPoints.asScala.toList.head
+            metricValue      = metricPoint.getValue
+            metricAttributes = metricPoint.getAttributes()
+          } yield assertTrue(
+            metricValue == 1L,
+            metricAttributes == Attributes(Attribute.string("zio", "annotation2"))
+          )
+        }
+      }
+    ).provide(inMemoryMetricReaderLayer, meterLayer(logAnnotated = true), ContextStorage.fiberRef)
 
 }

scala-cli/opentelemetry/MetricsApp.scala

@@ -125,12 +125,13 @@ object MetricsApp extends ZIOAppDefault {
           _                    <- messageLengthCounter.add(message.length, Attributes(Attribute.string("message", message)))
         } yield message
 
-        // By wrapping our logic into a span, we make the `messageLengthCounter` data points correlated with a "root_span" automatically
-        logic @@ tracing.aspects.root("root_span")
+        // By wrapping our logic into a span, we make the `messageLengthCounter` data points correlated with a "root_span" automatically.
+        // Additionally we implicitly add one more attribute to the `messageLenghtCounter` as it is wrapped into a `ZIO.logAnnotate` call.
+        ZIO.logAnnotate("zio", "annotation")(logic) @@ tracing.aspects.root("root_span")
       }
       .provide(
         otelSdkLayer,
-        OpenTelemetry.metrics(instrumentationScopeName),
+        OpenTelemetry.metrics(instrumentationScopeName, logAnnotated = true),
         OpenTelemetry.tracing(instrumentationScopeName),
         OpenTelemetry.contextZIO,
         tickCounterLayer,