Draft integration with IORuntimeMetrics

build.sbt

@@ -24,7 +24,7 @@ ThisBuild / crossScalaVersions := Seq(Scala213, "3.2.2")
 ThisBuild / scalaVersion := Scala213 // the default Scala
 
 val CatsVersion = "2.9.0"
-val CatsEffectVersion = "3.4.8"
+val CatsEffectVersion = "3.5-4848d2e-20230329T044947Z-SNAPSHOT"
 val CatsMtlVersion = "1.3.0"
 val FS2Version = "3.6.1"
 val MUnitVersion = "1.0.0-M7"
@@ -86,7 +86,7 @@ lazy val `core-metrics` = crossProject(JVMPlatform, JSPlatform, NativePlatform)
   .settings(
     name := "otel4s-core-metrics",
     libraryDependencies ++= Seq(
-      "org.typelevel" %%% "cats-effect-kernel" % CatsEffectVersion,
+      "org.typelevel" %%% "cats-effect" % CatsEffectVersion,
       "org.typelevel" %%% "cats-effect-testkit" % CatsEffectVersion % Test
     )
   )
@@ -232,6 +232,7 @@ lazy val examples = project
     name := "otel4s-examples",
     libraryDependencies ++= Seq(
       "io.opentelemetry" % "opentelemetry-exporter-otlp" % OpenTelemetryVersion,
+      "io.opentelemetry" % "opentelemetry-exporter-logging" % OpenTelemetryVersion,
       "io.opentelemetry" % "opentelemetry-sdk" % OpenTelemetryVersion,
       "io.opentelemetry" % "opentelemetry-sdk-extension-autoconfigure" % s"${OpenTelemetryVersion}-alpha",
       "io.opentelemetry" % "opentelemetry-extension-trace-propagators" % s"${OpenTelemetryVersion}" % Runtime

core/metrics/src/main/scala/org/typelevel/otel4s/metrics/preset/IOMetrics.scala

@@ -0,0 +1,84 @@
+package org.typelevel.otel4s.metrics
+package preset
+
+import cats.effect.{Resource, Sync}
+import cats.effect.unsafe._
+import cats.syntax.apply._
+import cats.syntax.functor._
+
+object IOMetrics {
+
+  def fromRuntimeMetrics[F[_]: Sync: Meter](
+      metrics: IORuntimeMetrics
+  ): Resource[F, Unit] = {
+    val meter = implicitly[Meter[F]]
+
+    val starvation = metrics.cpuStarvation.map { cpuStarvation =>
+      val starvationCount = meter
+        .observableUpDownCounter("cpu-starvation-count")
+        .createWithCallback(cb =>
+          Sync[F].defer(cb.record(cpuStarvation.starvationCount()))
+        )
+
+      val maxDrift = meter
+        .observableUpDownCounter("max-clock-drift-ms")
+        .createWithCallback(cb =>
+          Sync[F].defer(cb.record(cpuStarvation.maxClockDriftMs()))
+        )
+
+      val currentDrift = meter
+        .observableUpDownCounter("current-clock-drift-ms")
+        .createWithCallback(cb =>
+          Sync[F].defer(cb.record(cpuStarvation.currentClockDriftMs()))
+        )
+
+      (starvationCount, maxDrift, currentDrift).tupled.void
+    }
+
+    val compute = metrics.compute.map { compute =>
+      val wtc = meter
+        .observableUpDownCounter("worker-thread-count")
+        .createWithCallback(cb =>
+          Sync[F].defer(cb.record(compute.workerThreadCount().toLong))
+        )
+
+      val atc = meter
+        .observableUpDownCounter("active-thread-count")
+        .createWithCallback(cb =>
+          Sync[F].defer(cb.record(compute.activeThreadCount().toLong))
+        )
+
+      val stc = meter
+        .observableUpDownCounter("searching-thread-count")
+        .createWithCallback(cb =>
+          Sync[F].defer(cb.record(compute.searchingThreadCount().toLong))
+        )
+
+      val bwtc = meter
+        .observableUpDownCounter("blocker-worker-thread-count")
+        .createWithCallback(cb =>
+          Sync[F].defer(cb.record(compute.blockedWorkerThreadCount().toLong))
+        )
+
+      val lqfc = meter
+        .observableUpDownCounter("local-queue-fiber-count")
+        .createWithCallback(cb =>
+          Sync[F].defer(cb.record(compute.localQueueFiberCount()))
+        )
+
+      val sfc = meter
+        .observableUpDownCounter("suspended-fiber-count")
+        .createWithCallback(cb =>
+          Sync[F].defer(cb.record(compute.suspendedFiberCount()))
+        )
+
+      (wtc, atc, stc, bwtc, lqfc, sfc).tupled.void
+    }
+
+    for {
+      _ <- starvation.getOrElse(Resource.unit[F])
+      _ <- compute.getOrElse(Resource.unit[F])
+    } yield ()
+  }
+
+}

examples/src/main/scala/RuntimeMetricsExample.scala

@@ -0,0 +1,59 @@
+import cats.effect.std.Random
+import cats.effect.{IO, IOApp, Resource}
+import cats.syntax.foldable._
+import io.opentelemetry.sdk.OpenTelemetrySdk
+import io.opentelemetry.sdk.metrics.SdkMeterProvider
+import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader
+import org.typelevel.otel4s.java.OtelJava
+import org.typelevel.otel4s.metrics.preset.IOMetrics
+
+import scala.concurrent.duration._
+import scala.jdk.CollectionConverters._
+
+object RuntimeMetricsExample extends IOApp.Simple {
+  def run: IO[Unit] = {
+    val jMetricReader = InMemoryMetricReader.create()
+    val jSdk = {
+      val meterProvider = SdkMeterProvider
+        .builder()
+        .registerMetricReader(jMetricReader)
+        .build()
+
+      val sdk = OpenTelemetrySdk
+        .builder()
+        .setMeterProvider(meterProvider)
+        .build()
+
+      sdk
+    }
+
+    def printMetrics: IO[Unit] =
+      for {
+        _ <- IO.println("New cycle: ")
+        metrics <- IO.delay(jMetricReader.collectAllMetrics().asScala.toList)
+        _ <- metrics.traverse_(v => IO.println(v.getName + " = " + v.getData))
+      } yield ()
+
+    Resource
+      .eval(OtelJava.forAsync[IO](jSdk))
+      .evalMap(_.meterProvider.get("cats-effect-runtime-metrics"))
+      .use { implicit meter =>
+        IOMetrics.fromRuntimeMetrics[IO](runtime.metrics).surround {
+          printMetrics.delayBy(500.millis).foreverM.background.surround {
+            compute
+          }
+        }
+      }
+  }
+
+  private def compute: IO[Unit] =
+    Random.scalaUtilRandom[IO].flatMap { random =>
+      val io = random.betweenLong(10, 3000).flatMap { delay =>
+        if (delay % 2 == 0) IO.blocking(Thread.sleep(delay))
+        else IO.delay(Thread.sleep(delay))
+      }
+
+      IO.parReplicateAN(30)(100, io).void
+    }
+
+}