sdk-metrics: add SdkMeter

sdk/metrics/src/main/scala/org/typelevel/otel4s/sdk/metrics/SdkMeter.scala

@@ -0,0 +1,105 @@
+/*
+ * Copyright 2024 Typelevel
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.typelevel.otel4s.sdk.metrics
+
+import cats.effect.Clock
+import cats.effect.MonadCancelThrow
+import cats.effect.std.Console
+import org.typelevel.otel4s.metrics.BatchCallback
+import org.typelevel.otel4s.metrics.Counter
+import org.typelevel.otel4s.metrics.Histogram
+import org.typelevel.otel4s.metrics.MeasurementValue
+import org.typelevel.otel4s.metrics.Meter
+import org.typelevel.otel4s.metrics.ObservableCounter
+import org.typelevel.otel4s.metrics.ObservableGauge
+import org.typelevel.otel4s.metrics.ObservableUpDownCounter
+import org.typelevel.otel4s.metrics.UpDownCounter
+import org.typelevel.otel4s.sdk.context.AskContext
+import org.typelevel.otel4s.sdk.metrics.internal.MeterSharedState
+
+/** The meter is responsible for creating instruments.
+  *
+  * @see
+  *   [[https://opentelemetry.io/docs/specs/otel/metrics/api/#meter]]
+  */
+private class SdkMeter[F[_]: MonadCancelThrow: Clock: Console: AskContext](
+    sharedState: MeterSharedState[F]
+) extends Meter[F] {
+
+  def counter[A: MeasurementValue](
+      name: String
+  ): Counter.Builder[F, A] =
+    if (SdkMeter.isValidName(name))
+      SdkCounter.Builder(name, sharedState)
+    else
+      NoopInstrumentBuilder.counter(name)
+
+  def histogram[A: MeasurementValue](
+      name: String
+  ): Histogram.Builder[F, A] =
+    if (SdkMeter.isValidName(name))
+      SdkHistogram.Builder(name, sharedState)
+    else
+      NoopInstrumentBuilder.histogram(name)
+
+  def upDownCounter[A: MeasurementValue](
+      name: String
+  ): UpDownCounter.Builder[F, A] =
+    if (SdkMeter.isValidName(name))
+      SdkUpDownCounter.Builder(name, sharedState)
+    else
+      NoopInstrumentBuilder.upDownCounter(name)
+
+  def observableGauge[A: MeasurementValue](
+      name: String
+  ): ObservableGauge.Builder[F, A] =
+    if (SdkMeter.isValidName(name))
+      SdkObservableGauge.Builder(name, sharedState)
+    else
+      NoopInstrumentBuilder.observableGauge(name)
+
+  def observableCounter[A: MeasurementValue](
+      name: String
+  ): ObservableCounter.Builder[F, A] =
+    if (SdkMeter.isValidName(name))
+      SdkObservableCounter.Builder(name, sharedState)
+    else
+      NoopInstrumentBuilder.observableCounter(name)
+
+  def observableUpDownCounter[A: MeasurementValue](
+      name: String
+  ): ObservableUpDownCounter.Builder[F, A] =
+    if (SdkMeter.isValidName(name))
+      SdkObservableUpDownCounter.Builder(name, sharedState)
+    else
+      NoopInstrumentBuilder.observableUpDownCounter(name)
+
+  val batchCallback: BatchCallback[F] =
+    new SdkBatchCallback[F](sharedState)
+
+}
+
+object SdkMeter {
+
+  // see https://opentelemetry.io/docs/specs/otel/metrics/api/#instrument-name-syntax
+  private val InstrumentNamePattern =
+    "([A-Za-z]){1}([A-Za-z0-9\\_\\-\\./]){0,254}".r
+
+  private def isValidName(name: String): Boolean =
+    name != null && SdkMeter.InstrumentNamePattern.matches(name)
+
+}

sdk/metrics/src/main/scala/org/typelevel/otel4s/sdk/metrics/SdkObservableCounter.scala

@@ -44,7 +44,7 @@ private object SdkObservableCounter {
 
   final case class Builder[
       F[_]: MonadCancelThrow: Clock: Console: AskContext,
-      A: MeasurementValue: Numeric
+      A: MeasurementValue
   ](
       name: String,
       sharedState: MeterSharedState[F],
@@ -111,7 +111,17 @@ private object SdkObservableCounter {
       }
 
     def createObserver: F[ObservableMeasurement[F, A]] =
-      sharedState.registerObservableMeasurement[A](makeDescriptor).widen
+      MeasurementValue[A] match {
+        case MeasurementValue.LongMeasurementValue(cast) =>
+          sharedState
+            .registerObservableMeasurement[Long](makeDescriptor)
+            .map(_.contramap(cast))
+
+        case MeasurementValue.DoubleMeasurementValue(cast) =>
+          sharedState
+            .registerObservableMeasurement[Double](makeDescriptor)
+            .map(_.contramap(cast))
+      }
 
     private def makeDescriptor: InstrumentDescriptor.Asynchronous =
       InstrumentDescriptor.asynchronous(

sdk/metrics/src/main/scala/org/typelevel/otel4s/sdk/metrics/SdkObservableGauge.scala

@@ -43,7 +43,7 @@ private object SdkObservableGauge {
 
   final case class Builder[
       F[_]: MonadCancelThrow: Clock: Console: AskContext,
-      A: MeasurementValue: Numeric
+      A: MeasurementValue
   ](
       name: String,
       sharedState: MeterSharedState[F],
@@ -110,7 +110,17 @@ private object SdkObservableGauge {
       }
 
     def createObserver: F[ObservableMeasurement[F, A]] =
-      sharedState.registerObservableMeasurement[A](makeDescriptor).widen
+      MeasurementValue[A] match {
+        case MeasurementValue.LongMeasurementValue(cast) =>
+          sharedState
+            .registerObservableMeasurement[Long](makeDescriptor)
+            .map(_.contramap(cast))
+
+        case MeasurementValue.DoubleMeasurementValue(cast) =>
+          sharedState
+            .registerObservableMeasurement[Double](makeDescriptor)
+            .map(_.contramap(cast))
+      }
 
     private def makeDescriptor: InstrumentDescriptor.Asynchronous =
       InstrumentDescriptor.asynchronous(

sdk/metrics/src/main/scala/org/typelevel/otel4s/sdk/metrics/SdkObservableUpDownCounter.scala

@@ -43,7 +43,7 @@ private object SdkObservableUpDownCounter {
 
   final case class Builder[
       F[_]: MonadCancelThrow: Clock: Console: AskContext,
-      A: MeasurementValue: Numeric
+      A: MeasurementValue
   ](
       name: String,
       sharedState: MeterSharedState[F],
@@ -112,7 +112,17 @@ private object SdkObservableUpDownCounter {
       }
 
     def createObserver: F[ObservableMeasurement[F, A]] =
-      sharedState.registerObservableMeasurement[A](makeDescriptor).widen
+      MeasurementValue[A] match {
+        case MeasurementValue.LongMeasurementValue(cast) =>
+          sharedState
+            .registerObservableMeasurement[Long](makeDescriptor)
+            .map(_.contramap(cast))
+
+        case MeasurementValue.DoubleMeasurementValue(cast) =>
+          sharedState
+            .registerObservableMeasurement[Double](makeDescriptor)
+            .map(_.contramap(cast))
+      }
 
     private def makeDescriptor: InstrumentDescriptor.Asynchronous =
       InstrumentDescriptor.asynchronous(

sdk/metrics/src/main/scala/org/typelevel/otel4s/sdk/metrics/internal/CallbackRegistration.scala

@@ -29,7 +29,7 @@ private[metrics] final class CallbackRegistration[F[_]: MonadCancelThrow](
 ) {
 
   private val hasStorages: Boolean =
-    measurements.exists(_.storages.nonEmpty)
+    measurements.exists(_.hasStorages)
 
   /** Set the active reader on each observable measurement so that measurements
     * are only recorded to relevant storages.

sdk/metrics/src/main/scala/org/typelevel/otel4s/sdk/metrics/internal/SdkObservableMeasurement.scala

@@ -33,24 +33,20 @@ import org.typelevel.otel4s.sdk.metrics.data.TimeWindow
 import org.typelevel.otel4s.sdk.metrics.internal.exporter.RegisteredReader
 import org.typelevel.otel4s.sdk.metrics.internal.storage.MetricStorage
 
-private[metrics] final class SdkObservableMeasurement[
-    F[_]: Monad: Console,
-    A: MeasurementValue
-] private (
-    stateRef: Ref[F, SdkObservableMeasurement.State[F]],
-    val scope: InstrumentationScope,
-    val descriptor: InstrumentDescriptor,
-    val storages: Vector[MetricStorage.Asynchronous[F, A]]
-) extends ObservableMeasurement[F, A] {
-  import SdkObservableMeasurement._
-
-  private val isValid: A => Boolean =
-    MeasurementValue[A] match {
-      case MeasurementValue.LongMeasurementValue(_) =>
-        Function.const(true)
-      case MeasurementValue.DoubleMeasurementValue(cast) =>
-        v => !cast(v).isNaN
-    }
+private[metrics] sealed trait SdkObservableMeasurement[F[_], A]
+    extends ObservableMeasurement[F, A] { self =>
+
+  /** The scope associated with this measurement.
+    */
+  def scope: InstrumentationScope
+
+  /** The descriptor associated with this measurement.
+    */
+  def descriptor: InstrumentDescriptor
+
+  /** Whether the measurement has active storages.
+    */
+  def hasStorages: Boolean
 
   /** Sets an active reader and resets the state upon resource finalization.
     *
@@ -63,30 +59,23 @@ private[metrics] final class SdkObservableMeasurement[
   def withActiveReader(
       reader: RegisteredReader[F],
       timeWindow: TimeWindow
-  ): Resource[F, Unit] =
-    Resource.make(stateRef.set(State.WithReader(reader, timeWindow))) { _ =>
-      stateRef.set(State.Empty())
-    }
+  ): Resource[F, Unit]
 
-  def record(value: A, attributes: Attributes): F[Unit] =
-    stateRef.get
-      .flatMap {
-        case State.Empty() =>
-          Console[F].errorln(
-            "SdkObservableMeasurement: " +
-              s"trying to record a measurement for an instrument [${descriptor.name}] while the active reader is unset. " +
-              "Dropping the measurement."
-          )
-
-        case State.WithReader(reader, timeWindow) =>
-          val measurement =
-            AsynchronousMeasurement(timeWindow, attributes, value)
-
-          storages
-            .filter(_.reader == reader)
-            .traverse_(storage => storage.record(measurement))
-      }
-      .whenA(isValid(value))
+  final def contramap[B](f: B => A): SdkObservableMeasurement[F, B] =
+    new SdkObservableMeasurement[F, B] {
+      def scope: InstrumentationScope = self.scope
+      def descriptor: InstrumentDescriptor = self.descriptor
+      def hasStorages: Boolean = self.hasStorages
+
+      def withActiveReader(
+          reader: RegisteredReader[F],
+          timeWindow: TimeWindow
+      ): Resource[F, Unit] =
+        self.withActiveReader(reader, timeWindow)
+
+      def record(value: B, attributes: Attributes): F[Unit] =
+        self.record(f(value), attributes)
+    }
 
 }
 
@@ -109,11 +98,55 @@ private[metrics] object SdkObservableMeasurement {
   ): F[SdkObservableMeasurement[F, A]] =
     for {
       state <- Ref.of[F, State[F]](State.Empty())
-    } yield new SdkObservableMeasurement[F, A](
-      state,
-      scope,
-      descriptor,
-      storages
-    )
+    } yield new Impl[F, A](state, scope, descriptor, storages)
+
+  private final class Impl[
+      F[_]: Monad: Console,
+      A: MeasurementValue
+  ](
+      stateRef: Ref[F, SdkObservableMeasurement.State[F]],
+      val scope: InstrumentationScope,
+      val descriptor: InstrumentDescriptor,
+      storages: Vector[MetricStorage.Asynchronous[F, A]]
+  ) extends SdkObservableMeasurement[F, A] { self =>
+
+    private val isValid: A => Boolean =
+      MeasurementValue[A] match {
+        case MeasurementValue.LongMeasurementValue(_) =>
+          Function.const(true)
+        case MeasurementValue.DoubleMeasurementValue(cast) =>
+          v => !cast(v).isNaN
+      }
+
+    def withActiveReader(
+        reader: RegisteredReader[F],
+        timeWindow: TimeWindow
+    ): Resource[F, Unit] =
+      Resource.make(stateRef.set(State.WithReader(reader, timeWindow))) { _ =>
+        stateRef.set(State.Empty())
+      }
+
+    def record(value: A, attributes: Attributes): F[Unit] =
+      stateRef.get
+        .flatMap {
+          case State.Empty() =>
+            Console[F].errorln(
+              "SdkObservableMeasurement: " +
+                s"trying to record a measurement for an instrument [${descriptor.name}] while the active reader is unset. " +
+                "Dropping the measurement."
+            )
+
+          case State.WithReader(reader, timeWindow) =>
+            val measurement =
+              AsynchronousMeasurement(timeWindow, attributes, value)
+
+            storages
+              .filter(_.reader == reader)
+              .traverse_(storage => storage.record(measurement))
+        }
+        .whenA(isValid(value))
+
+    def hasStorages: Boolean = storages.nonEmpty
+  }
 
 }

sdk/metrics/src/test/scala/org/typelevel/otel4s/sdk/metrics/SdkBatchCallbackSuite.scala

@@ -46,7 +46,7 @@ class SdkBatchCallbackSuite extends CatsEffectSuite with ScalaCheckEffectSuite {
       Gen.option(Gen.alphaNumStr),
       Gen.either(Gen.long, Gen.double)
     ) { (resource, scope, window, attrs, name, description, unit, value) =>
-      def test[A: MeasurementValue: Numeric](value: A): IO[Unit] = {
+      def test[A: MeasurementValue](value: A): IO[Unit] = {
         val expectedCounter = MetricData(
           resource,
           scope,