diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 31fadc49..8a913136 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -30,7 +30,14 @@ jobs: os: [ubuntu-22.04] scala: [2.12, 2.13, 3] java: [temurin@8] - project: [rootJS, rootJVM, rootNative] + project: [coreJVM, coreJS, coreNative, otel4sJVM, otel4sJS, otel4sNative] + exclude: + - project: otel4sJVM + scala: 2.12 + - project: otel4sJS + scala: 2.12 + - project: otel4sNative + scala: 2.12 runs-on: ${{ matrix.os }} timeout-minutes: 60 steps: @@ -83,11 +90,11 @@ jobs: - name: Make target directories if: github.event_name != 'pull_request' && (startsWith(github.ref, 'refs/tags/v') || github.ref == 'refs/heads/main') - run: mkdir -p core/.native/target core/.js/target core/.jvm/target project/target + run: mkdir -p otel4s/.native/target core/.native/target otel4s/.js/target otel4s/.jvm/target core/.js/target core/.jvm/target project/target - name: Compress target directories if: github.event_name != 'pull_request' && (startsWith(github.ref, 'refs/tags/v') || github.ref == 'refs/heads/main') - run: tar cf targets.tar core/.native/target core/.js/target core/.jvm/target project/target + run: tar cf targets.tar otel4s/.native/target core/.native/target otel4s/.js/target otel4s/.jvm/target core/.js/target core/.jvm/target project/target - name: Upload target directories if: github.event_name != 'pull_request' && (startsWith(github.ref, 'refs/tags/v') || github.ref == 'refs/heads/main') @@ -127,92 +134,152 @@ jobs: if: matrix.java == 'temurin@8' && steps.setup-java-temurin-8.outputs.cache-hit == 'false' run: sbt +update - - name: Download target directories (2.12, rootJS) + - name: Download target directories (2.12, coreJVM) uses: actions/download-artifact@v6 with: - name: target-${{ matrix.os }}-${{ matrix.java }}-2.12-rootJS + name: target-${{ matrix.os }}-${{ matrix.java }}-2.12-coreJVM - - name: Inflate target directories (2.12, rootJS) + - name: Inflate target directories (2.12, coreJVM) run: | tar xf targets.tar rm targets.tar - - name: Download target directories (2.12, rootJVM) + - name: Download target directories (2.12, coreJS) uses: actions/download-artifact@v6 with: - name: target-${{ matrix.os }}-${{ matrix.java }}-2.12-rootJVM + name: target-${{ matrix.os }}-${{ matrix.java }}-2.12-coreJS - - name: Inflate target directories (2.12, rootJVM) + - name: Inflate target directories (2.12, coreJS) run: | tar xf targets.tar rm targets.tar - - name: Download target directories (2.12, rootNative) + - name: Download target directories (2.12, coreNative) uses: actions/download-artifact@v6 with: - name: target-${{ matrix.os }}-${{ matrix.java }}-2.12-rootNative + name: target-${{ matrix.os }}-${{ matrix.java }}-2.12-coreNative - - name: Inflate target directories (2.12, rootNative) + - name: Inflate target directories (2.12, coreNative) run: | tar xf targets.tar rm targets.tar - - name: Download target directories (2.13, rootJS) + - name: Download target directories (2.13, coreJVM) uses: actions/download-artifact@v6 with: - name: target-${{ matrix.os }}-${{ matrix.java }}-2.13-rootJS + name: target-${{ matrix.os }}-${{ matrix.java }}-2.13-coreJVM - - name: Inflate target directories (2.13, rootJS) + - name: Inflate target directories (2.13, coreJVM) run: | tar xf targets.tar rm targets.tar - - name: Download target directories (2.13, rootJVM) + - name: Download target directories (2.13, coreJS) uses: actions/download-artifact@v6 with: - name: target-${{ matrix.os }}-${{ matrix.java }}-2.13-rootJVM + name: target-${{ matrix.os }}-${{ matrix.java }}-2.13-coreJS - - name: Inflate target directories (2.13, rootJVM) + - name: Inflate target directories (2.13, coreJS) run: | tar xf targets.tar rm targets.tar - - name: Download target directories (2.13, rootNative) + - name: Download target directories (2.13, coreNative) uses: actions/download-artifact@v6 with: - name: target-${{ matrix.os }}-${{ matrix.java }}-2.13-rootNative + name: target-${{ matrix.os }}-${{ matrix.java }}-2.13-coreNative - - name: Inflate target directories (2.13, rootNative) + - name: Inflate target directories (2.13, coreNative) run: | tar xf targets.tar rm targets.tar - - name: Download target directories (3, rootJS) + - name: Download target directories (2.13, otel4sJVM) uses: actions/download-artifact@v6 with: - name: target-${{ matrix.os }}-${{ matrix.java }}-3-rootJS + name: target-${{ matrix.os }}-${{ matrix.java }}-2.13-otel4sJVM - - name: Inflate target directories (3, rootJS) + - name: Inflate target directories (2.13, otel4sJVM) run: | tar xf targets.tar rm targets.tar - - name: Download target directories (3, rootJVM) + - name: Download target directories (2.13, otel4sJS) uses: actions/download-artifact@v6 with: - name: target-${{ matrix.os }}-${{ matrix.java }}-3-rootJVM + name: target-${{ matrix.os }}-${{ matrix.java }}-2.13-otel4sJS - - name: Inflate target directories (3, rootJVM) + - name: Inflate target directories (2.13, otel4sJS) run: | tar xf targets.tar rm targets.tar - - name: Download target directories (3, rootNative) + - name: Download target directories (2.13, otel4sNative) uses: actions/download-artifact@v6 with: - name: target-${{ matrix.os }}-${{ matrix.java }}-3-rootNative + name: target-${{ matrix.os }}-${{ matrix.java }}-2.13-otel4sNative - - name: Inflate target directories (3, rootNative) + - name: Inflate target directories (2.13, otel4sNative) + run: | + tar xf targets.tar + rm targets.tar + + - name: Download target directories (3, coreJVM) + uses: actions/download-artifact@v6 + with: + name: target-${{ matrix.os }}-${{ matrix.java }}-3-coreJVM + + - name: Inflate target directories (3, coreJVM) + run: | + tar xf targets.tar + rm targets.tar + + - name: Download target directories (3, coreJS) + uses: actions/download-artifact@v6 + with: + name: target-${{ matrix.os }}-${{ matrix.java }}-3-coreJS + + - name: Inflate target directories (3, coreJS) + run: | + tar xf targets.tar + rm targets.tar + + - name: Download target directories (3, coreNative) + uses: actions/download-artifact@v6 + with: + name: target-${{ matrix.os }}-${{ matrix.java }}-3-coreNative + + - name: Inflate target directories (3, coreNative) + run: | + tar xf targets.tar + rm targets.tar + + - name: Download target directories (3, otel4sJVM) + uses: actions/download-artifact@v6 + with: + name: target-${{ matrix.os }}-${{ matrix.java }}-3-otel4sJVM + + - name: Inflate target directories (3, otel4sJVM) + run: | + tar xf targets.tar + rm targets.tar + + - name: Download target directories (3, otel4sJS) + uses: actions/download-artifact@v6 + with: + name: target-${{ matrix.os }}-${{ matrix.java }}-3-otel4sJS + + - name: Inflate target directories (3, otel4sJS) + run: | + tar xf targets.tar + rm targets.tar + + - name: Download target directories (3, otel4sNative) + uses: actions/download-artifact@v6 + with: + name: target-${{ matrix.os }}-${{ matrix.java }}-3-otel4sNative + + - name: Inflate target directories (3, otel4sNative) run: | tar xf targets.tar rm targets.tar diff --git a/build.sbt b/build.sbt index 7813cd5d..23a30cdb 100644 --- a/build.sbt +++ b/build.sbt @@ -11,7 +11,18 @@ ThisBuild / startYear := Some(2019) ThisBuild / licenses := Seq(License.MIT) ThisBuild / tlSiteApiUrl := Some(url("https://www.javadoc.io/doc/org.typelevel/keypool_2.12")) -lazy val root = tlCrossRootProject.aggregate(core) +lazy val root = tlCrossRootProject.aggregate(core, otel4s) + +ThisBuild / githubWorkflowBuildMatrixAdditions := { + val projects = core.componentProjects ++ otel4s.componentProjects + + Map("project" -> projects.map(_.id).toList) +} + +ThisBuild / githubWorkflowBuildMatrixExclusions ++= { + val projects = otel4s.componentProjects.map(_.id) + projects.map(project => MatrixExclude(Map("project" -> project, "scala" -> "2.12"))) +} lazy val core = crossProject(JVMPlatform, JSPlatform, NativePlatform) .crossType(CrossType.Pure) @@ -50,15 +61,41 @@ lazy val core = crossProject(JVMPlatform, JSPlatform, NativePlatform) ) ) +lazy val otel4s = crossProject(JVMPlatform, JSPlatform, NativePlatform) + .crossType(CrossType.Pure) + .in(file("otel4s")) + .dependsOn(core) + .enablePlugins(BuildInfoPlugin) + .settings(commonSettings) + .settings( + name := "keypool-otel4s", + startYear := Some(2024), + crossScalaVersions := Seq(Scala213, Scala3), + libraryDependencies ++= Seq( + "org.typelevel" %%% "otel4s-core-metrics" % otel4sV, + "org.typelevel" %%% "otel4s-semconv-metrics-experimental" % otel4sV % Test, + "org.typelevel" %%% "otel4s-sdk-metrics-testkit" % otel4sSdkV % Test + ), + buildInfoPackage := "org.typelevel.keypool.otel4s", + buildInfoOptions += sbtbuildinfo.BuildInfoOption.PackagePrivate, + buildInfoKeys := Seq[BuildInfoKey]( + "version" -> version.value + ), + mimaPreviousArtifacts ~= { _.filterNot(_.revision.startsWith("0.4")) } + ) + lazy val docs = project .in(file("site")) .settings(commonSettings) - .dependsOn(core.jvm) + .dependsOn(core.jvm, otel4s.jvm) .enablePlugins(TypelevelSitePlugin) val catsV = "2.13.0" val catsEffectV = "3.7.1" +val otel4sV = "1.1.0" +val otel4sSdkV = "0.19.2" + val munitV = "1.3.6" val munitCatsEffectV = "2.2.0" diff --git a/core/src/main/scala/org/typelevel/keypool/KeyPool.scala b/core/src/main/scala/org/typelevel/keypool/KeyPool.scala index 40bcd697..136fe548 100644 --- a/core/src/main/scala/org/typelevel/keypool/KeyPool.scala +++ b/core/src/main/scala/org/typelevel/keypool/KeyPool.scala @@ -67,7 +67,8 @@ object KeyPool { private[keypool] val kpMaxIdle: Int, private[keypool] val kpMaxTotal: Int, private[keypool] val kpMaxTotalSem: RequestSemaphore[F], - private[keypool] val kpVar: Ref[F, PoolMap[A, (B, F[Unit])]] + private[keypool] val kpVar: Ref[F, PoolMap[A, (B, F[Unit])]], + private[keypool] val kpMetrics: Metrics[F] ) extends KeyPool[F, A, B] { def take(k: A): Resource[F, Managed[F, B]] = @@ -114,7 +115,8 @@ object KeyPool { * Make a 'KeyPool' inactive and destroy all idle resources. */ private[keypool] def destroy[F[_]: MonadThrow, A, B]( - kpVar: Ref[F, PoolMap[A, (B, F[Unit])]] + kpVar: Ref[F, PoolMap[A, (B, F[Unit])]], + metrics: Metrics[F] ): F[Unit] = for { m <- kpVar.getAndSet(PoolMap.closed[A, (B, F[Unit])]) _ <- m match { @@ -123,7 +125,9 @@ object KeyPool { m2.toList.traverse_ { case (_, pl) => pl.toList .traverse_ { case (_, r) => - r._2.attempt.void + metrics.idleDec >> + metrics.resourceDestroyed(Metrics.DestructionReason.PoolClosed) >> + r._2.attempt.void } } } @@ -137,6 +141,7 @@ object KeyPool { idleTimeAllowedInPoolNanos: FiniteDuration, durationBetweenEvictionRuns: FiniteDuration, kpVar: Ref[F, PoolMap[A, (B, F[Unit])]], + metrics: Metrics[F], onReaperException: Throwable => F[Unit] )(implicit F: Temporal[F]): F[Unit] = { // We are going to do non-referentially transparent things as we may be waiting for our modification to go through @@ -199,11 +204,18 @@ object KeyPool { val (m_, toDestroy) = findStale(now, idleCount, m) ( m_, - toDestroy.traverse_(_._2._2).attempt.flatMap { - case Left(t) => onReaperException(t) - // .handleErrorWith(t => F.delay(t.printStackTrace())) // CHEATING? - case Right(()) => F.unit - } + toDestroy + .traverse_(r => + metrics.idleDec >> + metrics.resourceDestroyed(Metrics.DestructionReason.IdleTimeout) >> + r._2._2 + ) + .attempt + .flatMap { + case Left(t) => onReaperException(t) + // .handleErrorWith(t => F.delay(t.printStackTrace())) // CHEATING? + case Right(()) => F.unit + } ) } } @@ -239,6 +251,9 @@ object KeyPool { r: B, destroy: F[Unit] ): F[Unit] = { + def destroyFor(reason: Metrics.DestructionReason): F[Unit] = + kp.kpMetrics.resourceDestroyed(reason) >> destroy + def addToList[Z]( now: FiniteDuration, maxCount: Int, @@ -254,22 +269,29 @@ object KeyPool { else (l, Some(x)) } } + def go(now: FiniteDuration, pc: PoolMap[A, (B, F[Unit])]): (PoolMap[A, (B, F[Unit])], F[Unit]) = pc match { - case p @ PoolClosed() => (p, destroy) + case p @ PoolClosed() => (p, destroyFor(Metrics.DestructionReason.PoolClosed)) case p @ PoolOpen(idleCount, m) => - if (idleCount > kp.kpMaxIdle) (p, destroy) + if (kp.kpMaxIdle == 0 || idleCount >= kp.kpMaxIdle) + (p, destroyFor(Metrics.DestructionReason.MaxIdle)) else m.get(k) match { case None => val cnt_ = idleCount + 1 val m_ = PoolMap.open(cnt_, m + (k -> One((r, destroy), now))) - (m_, Applicative[F].pure(())) + (m_, kp.kpMetrics.idleInc) case Some(l) => val (l_, mx) = addToList(now, kp.kpMaxPerKey(k), (r, destroy), l) val cnt_ = idleCount + mx.fold(1)(_ => 0) val m_ = PoolMap.open(cnt_, m + (k -> l_)) - (m_, mx.fold(Applicative[F].unit)(_ => destroy)) + ( + m_, + mx.fold(kp.kpMetrics.idleInc)(_ => + destroyFor(Metrics.DestructionReason.MaxPerKey) + ) + ) } } @@ -295,21 +317,32 @@ object KeyPool { } } + def allocateNew: F[(B, F[Unit])] = + kp.kpMetrics.createDuration.surround(kp.kpRes(k).allocated) + for { + acquisition <- kp.kpMetrics.acquire _ <- kp.kpMaxTotalSem.permit optR <- Resource.eval(kp.kpVar.modify(go)) + _ <- Resource.eval(kp.kpMetrics.idleDec.whenA(optR.nonEmpty)) releasedState <- Resource.eval(Ref[F].of[Reusable](kp.kpDefaultReuseState)) resource <- Resource.makeFull[F, (B, F[Unit])] { poll => - optR.fold(poll(kp.kpRes(k).allocated))(r => Applicative[F].pure(r)) + optR.fold(poll(allocateNew))(r => Applicative[F].pure(r)) } { resource => for { reusable <- releasedState.get out <- reusable match { case Reusable.Reuse => put(kp, k, resource._1, resource._2).attempt.void - case Reusable.DontReuse => resource._2.attempt.void + case Reusable.DontReuse => + kp.kpMetrics.resourceDestroyed( + Metrics.DestructionReason.NotReusable + ) >> resource._2.attempt.void } } yield out } + _ <- Resource.eval(acquisition.complete) + _ <- kp.kpMetrics.inUseCount + _ <- kp.kpMetrics.useDuration } yield new Managed(resource._1, optR.isDefined, releasedState) } @@ -322,7 +355,8 @@ object KeyPool { val kpMaxIdle: Int, val kpMaxTotal: Int, val fairness: Fairness, - val onReaperException: Throwable => F[Unit] + val onReaperException: Throwable => F[Unit], + val metricsProvider: Metrics.Provider[F] ) { private def copy( kpRes: A => Resource[F, B] = this.kpRes, @@ -333,7 +367,8 @@ object KeyPool { kpMaxIdle: Int = this.kpMaxIdle, kpMaxTotal: Int = this.kpMaxTotal, fairness: Fairness = this.fairness, - onReaperException: Throwable => F[Unit] = this.onReaperException + onReaperException: Throwable => F[Unit] = this.onReaperException, + metricsProvider: Metrics.Provider[F] = this.metricsProvider ): Builder[F, A, B] = new Builder[F, A, B]( kpRes, kpDefaultReuseState, @@ -343,7 +378,8 @@ object KeyPool { kpMaxIdle, kpMaxTotal, fairness, - onReaperException + onReaperException, + metricsProvider ) def doOnCreate(f: B => F[Unit]): Builder[F, A, B] = @@ -378,19 +414,23 @@ object KeyPool { def withOnReaperException(f: Throwable => F[Unit]): Builder[F, A, B] = copy(onReaperException = f) + def withMetricsProvider(metricsProvider: Metrics.Provider[F]): Builder[F, A, B] = + copy(metricsProvider = metricsProvider) + def build: Resource[F, KeyPool[F, A, B]] = { def keepRunning[Z](fa: F[Z]): F[Z] = fa.onError { case e => onReaperException(e) }.attempt >> keepRunning(fa) for { + kpMetrics <- Resource.eval(metricsProvider.get) kpVar <- Resource.make( Ref[F].of[PoolMap[A, (B, F[Unit])]](PoolMap.open(0, Map.empty[A, PoolList[(B, F[Unit])]])) - )(kpVar => KeyPool.destroy(kpVar)) + )(kpVar => KeyPool.destroy(kpVar, kpMetrics)) kpMaxTotalSem <- Resource.eval(RequestSemaphore[F](fairness, kpMaxTotal)) _ <- (idleTimeAllowedInPool, durationBetweenEvictionRuns) match { case (fdI: FiniteDuration, fdE: FiniteDuration) if fdE >= 0.seconds => val idleNanos = 0.seconds.max(fdI) keepRunning( - KeyPool.reap(idleNanos, fdE, kpVar, onReaperException) + KeyPool.reap(idleNanos, fdE, kpVar, kpMetrics, onReaperException) ).background.void case _ => Applicative[Resource[F, *]].unit @@ -402,7 +442,8 @@ object KeyPool { kpMaxIdle, kpMaxTotal, kpMaxTotalSem, - kpVar + kpVar, + kpMetrics ) } @@ -420,7 +461,8 @@ object KeyPool { Defaults.maxIdle, Defaults.maxTotal, Defaults.fairness, - Defaults.onReaperException[F] + Defaults.onReaperException[F], + Defaults.metricsProvider ) def apply[F[_]: Temporal, A, B]( @@ -440,6 +482,7 @@ object KeyPool { def onReaperException[F[_]: Applicative] = { (t: Throwable) => Function.const(Applicative[F].unit)(t) } + def metricsProvider[F[_]: Applicative]: Metrics.Provider[F] = Metrics.Provider.noop } } } diff --git a/core/src/main/scala/org/typelevel/keypool/KeyPoolBuilder.scala b/core/src/main/scala/org/typelevel/keypool/KeyPoolBuilder.scala index d6bc4c3e..97128eb7 100644 --- a/core/src/main/scala/org/typelevel/keypool/KeyPoolBuilder.scala +++ b/core/src/main/scala/org/typelevel/keypool/KeyPoolBuilder.scala @@ -21,6 +21,7 @@ package org.typelevel.keypool +import internal.{Metrics, PoolList, PoolMap} import cats._ import cats.syntax.all._ import cats.effect.kernel._ @@ -87,16 +88,17 @@ final class KeyPoolBuilder[F[_]: Temporal, A, B] private ( def keepRunning[Z](fa: F[Z]): F[Z] = fa.onError { case e => onReaperException(e) }.attempt >> keepRunning(fa) for { + kpMetrics <- Resource.pure(Metrics.noop) kpVar <- Resource.make( Ref[F].of[PoolMap[A, (B, F[Unit])]](PoolMap.open(0, Map.empty[A, PoolList[(B, F[Unit])]])) - )(kpVar => KeyPool.destroy(kpVar)) + )(kpVar => KeyPool.destroy(kpVar, kpMetrics)) kpMaxTotalSem <- Resource.eval(RequestSemaphore[F](Fairness.Fifo, kpMaxTotal)) _ <- idleTimeAllowedInPool match { case fd: FiniteDuration => val nanos = 0.seconds.max(fd) val durationBetweenEvictionRuns = 5.seconds // the previous default keepRunning( - KeyPool.reap(nanos, durationBetweenEvictionRuns, kpVar, onReaperException) + KeyPool.reap(nanos, durationBetweenEvictionRuns, kpVar, kpMetrics, onReaperException) ).background.void case _ => Applicative[Resource[F, *]].unit @@ -108,7 +110,8 @@ final class KeyPoolBuilder[F[_]: Temporal, A, B] private ( kpMaxIdle, kpMaxTotal, kpMaxTotalSem, - kpVar + kpVar, + kpMetrics ) } diff --git a/core/src/main/scala/org/typelevel/keypool/Pool.scala b/core/src/main/scala/org/typelevel/keypool/Pool.scala index 1e1bffd3..9b883fed 100644 --- a/core/src/main/scala/org/typelevel/keypool/Pool.scala +++ b/core/src/main/scala/org/typelevel/keypool/Pool.scala @@ -24,6 +24,7 @@ package org.typelevel.keypool import cats._ import cats.effect.kernel._ import cats.syntax.all._ +import org.typelevel.keypool.internal.Metrics import scala.concurrent.duration._ /** @@ -78,7 +79,8 @@ object Pool { val kpMaxIdle: Int, val kpMaxTotal: Int, val fairness: Fairness, - val onReaperException: Throwable => F[Unit] + val onReaperException: Throwable => F[Unit], + val metricsProvider: Metrics.Provider[F] ) { private def copy( kpRes: Resource[F, B] = this.kpRes, @@ -88,7 +90,8 @@ object Pool { kpMaxIdle: Int = this.kpMaxIdle, kpMaxTotal: Int = this.kpMaxTotal, fairness: Fairness = this.fairness, - onReaperException: Throwable => F[Unit] = this.onReaperException + onReaperException: Throwable => F[Unit] = this.onReaperException, + metricsProvider: Metrics.Provider[F] = this.metricsProvider ): Builder[F, B] = new Builder[F, B]( kpRes, kpDefaultReuseState, @@ -97,7 +100,8 @@ object Pool { kpMaxIdle, kpMaxTotal, fairness, - onReaperException + onReaperException, + metricsProvider ) def doOnCreate(f: B => F[Unit]): Builder[F, B] = @@ -129,6 +133,9 @@ object Pool { def withOnReaperException(f: Throwable => F[Unit]): Builder[F, B] = copy(onReaperException = f) + def withMetricsProvider(metricsProvider: Metrics.Provider[F]): Builder[F, B] = + copy(metricsProvider = metricsProvider) + private def toKeyPoolBuilder: KeyPool.Builder[F, Unit, B] = new KeyPool.Builder( kpRes = _ => kpRes, @@ -139,7 +146,8 @@ object Pool { kpMaxIdle = kpMaxIdle, kpMaxTotal = kpMaxTotal, fairness = fairness, - onReaperException = onReaperException + onReaperException = onReaperException, + metricsProvider = metricsProvider ) def build: Resource[F, Pool[F, B]] = { @@ -163,7 +171,8 @@ object Pool { Defaults.maxIdle, Defaults.maxTotal, Defaults.fairness, - Defaults.onReaperException[F] + Defaults.onReaperException[F], + Defaults.metricsProvider ) def apply[F[_]: Temporal, B]( @@ -182,6 +191,7 @@ object Pool { def onReaperException[F[_]: Applicative] = { (t: Throwable) => Function.const(Applicative[F].unit)(t) } + def metricsProvider[F[_]: Applicative]: Metrics.Provider[F] = Metrics.Provider.noop } } } diff --git a/core/src/main/scala/org/typelevel/keypool/internal/Metrics.scala b/core/src/main/scala/org/typelevel/keypool/internal/Metrics.scala new file mode 100644 index 00000000..b73b9ee6 --- /dev/null +++ b/core/src/main/scala/org/typelevel/keypool/internal/Metrics.scala @@ -0,0 +1,101 @@ +/* + * Copyright (c) 2019 Typelevel + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to + * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of + * the Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER + * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN + * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ + +package org.typelevel.keypool.internal + +import cats.Applicative +import cats.effect.kernel.Resource + +/** Pool metrics. Implementations must not fail pool operations. */ +sealed trait Metrics[F[_]] { + + /** Tracks an acquisition until a resource is ready. */ + def acquire: Resource[F, Metrics.Acquisition[F]] + + /** Adds one idle resource. */ + def idleInc: F[Unit] + + /** Removes one idle resource. */ + def idleDec: F[Unit] + + /** Tracks a resource while it is in use. */ + def inUseCount: Resource[F, Unit] + + /** Records how long a resource is used. */ + def useDuration: Resource[F, Unit] + + /** Records how long it takes to create a new resource. */ + def createDuration: Resource[F, Unit] + + /** Records a resource removed permanently from the pool. */ + def resourceDestroyed(reason: Metrics.DestructionReason): F[Unit] + +} + +object Metrics { + private[keypool] trait Unsealed[F[_]] extends Metrics[F] + + sealed trait DestructionReason extends Product with Serializable + + object DestructionReason { + case object IdleTimeout extends DestructionReason + case object MaxIdle extends DestructionReason + case object MaxPerKey extends DestructionReason + case object NotReusable extends DestructionReason + case object PoolClosed extends DestructionReason + } + + trait Acquisition[F[_]] { + + /** Marks the resource as ready. */ + def complete: F[Unit] + + /** Finishes an acquisition that did not complete. */ + private[keypool] def finish(exitCase: Resource.ExitCase): F[Unit] + } + + trait Provider[F[_]] { + def get: F[Metrics[F]] + } + + object Provider { + def noop[F[_]: Applicative]: Provider[F] = + new Provider[F] { + def get: F[Metrics[F]] = Applicative[F].pure(Metrics.noop) + } + } + + def noop[F[_]: Applicative]: Metrics[F] = + new Metrics[F] { + def idleInc: F[Unit] = Applicative[F].unit + def idleDec: F[Unit] = Applicative[F].unit + def acquire: Resource[F, Acquisition[F]] = + Resource.pure(new Acquisition[F] { + def complete: F[Unit] = Applicative[F].unit + private[keypool] def finish(exitCase: Resource.ExitCase): F[Unit] = Applicative[F].unit + }) + def inUseCount: Resource[F, Unit] = Resource.unit + def useDuration: Resource[F, Unit] = Resource.unit + def createDuration: Resource[F, Unit] = Resource.unit + def resourceDestroyed(reason: DestructionReason): F[Unit] = Applicative[F].unit + } + +} diff --git a/otel4s/src/main/scala/org/typelevel/keypool/otel4s/Otel4sMetrics.scala b/otel4s/src/main/scala/org/typelevel/keypool/otel4s/Otel4sMetrics.scala new file mode 100644 index 00000000..539112cb --- /dev/null +++ b/otel4s/src/main/scala/org/typelevel/keypool/otel4s/Otel4sMetrics.scala @@ -0,0 +1,565 @@ +/* + * Copyright (c) 2024 Typelevel + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to + * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of + * the Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER + * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN + * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ + +package org.typelevel.keypool.otel4s + +import java.util.concurrent.TimeUnit + +import cats.effect.kernel.{Clock, Ref, Resource, Temporal} +import cats.syntax.flatMap.* +import cats.syntax.functor.* +import cats.syntax.traverse.* +import org.typelevel.keypool.internal.Metrics +import org.typelevel.otel4s.{Attribute, Attributes} +import org.typelevel.otel4s.metrics.{BucketBoundaries, Histogram, MeterProvider} + +import scala.concurrent.duration.FiniteDuration + +object Otel4sMetrics { + + /** Configuration for an emitted instrument. */ + sealed trait InstrumentConfig { + + /** Instrument name. */ + def name: String + + /** Instrument unit. */ + def unit: String + + /** Instrument description. */ + def description: String + + } + + object InstrumentConfig { + + /** Resource-destruction counter configuration. */ + sealed trait Counter extends InstrumentConfig { + + /** Attributes added to each measurement, based on why the resource was destroyed. */ + def attributes: Metrics.DestructionReason => Attributes + } + + /** Up-down counter configuration. */ + sealed trait UpDownCounter extends InstrumentConfig { + + /** Attributes added to each measurement. */ + def attributes: Attributes + } + + /** Duration histogram configuration. */ + sealed trait Histogram extends InstrumentConfig { + + /** Attributes added to each measurement, based on how the measured resource use completed. */ + def attributes: Resource.ExitCase => Attributes + + /** Unit used to record durations. */ + def timeUnit: TimeUnit + + /** Histogram boundaries in [[timeUnit]]. */ + def explicitBucketBoundaries: BucketBoundaries + + final def unit: String = + timeUnit match { + case TimeUnit.NANOSECONDS => "ns" + case TimeUnit.MICROSECONDS => "us" + case TimeUnit.MILLISECONDS => "ms" + case TimeUnit.SECONDS => "s" + case TimeUnit.MINUTES => "min" + case TimeUnit.HOURS => "h" + case TimeUnit.DAYS => "d" + } + } + + /** Creates an up-down counter configuration. */ + def upDownCounter( + name: String, + unit: String, + description: String, + attributes: Attributes + ): UpDownCounter = + UpDownCounterImpl(name, unit, description, attributes) + + /** Creates a resource-destruction counter configuration. */ + def counter( + name: String, + unit: String, + description: String, + attributes: Attributes + ): Counter = + counter(name, unit, description, _ => attributes) + + /** Creates a resource-destruction counter with reason-dependent attributes. */ + def counter( + name: String, + unit: String, + description: String, + attributes: Metrics.DestructionReason => Attributes + ): Counter = + CounterImpl(name, unit, description, attributes) + + /** Creates a duration histogram configuration. */ + def histogram( + name: String, + timeUnit: TimeUnit, + description: String, + attributes: Attributes, + explicitBucketBoundaries: BucketBoundaries + ): Histogram = + histogram(name, timeUnit, description, _ => attributes, explicitBucketBoundaries) + + /** Creates a duration histogram configuration with exit-case-dependent attributes. */ + def histogram( + name: String, + timeUnit: TimeUnit, + description: String, + attributes: Resource.ExitCase => Attributes, + explicitBucketBoundaries: BucketBoundaries + ): Histogram = + HistogramImpl(name, timeUnit, description, attributes, explicitBucketBoundaries) + + private final case class UpDownCounterImpl( + name: String, + unit: String, + description: String, + attributes: Attributes + ) extends UpDownCounter + + private final case class CounterImpl( + name: String, + unit: String, + description: String, + attributes: Metrics.DestructionReason => Attributes + ) extends Counter + + private final case class HistogramImpl( + name: String, + timeUnit: TimeUnit, + description: String, + attributes: Resource.ExitCase => Attributes, + explicitBucketBoundaries: BucketBoundaries + ) extends Histogram + + } + + /** Configuration for [[Otel4sMetrics]]. */ + sealed trait Config { + private[otel4s] def meterName: String + private[otel4s] def constAttributes: Attributes + private[otel4s] def idleInstrument: Option[InstrumentConfig.UpDownCounter] + private[otel4s] def inUseInstrument: Option[InstrumentConfig.UpDownCounter] + private[otel4s] def useDurationInstrument: Option[InstrumentConfig.Histogram] + private[otel4s] def pendingAcquireInstrument: Option[InstrumentConfig.UpDownCounter] + private[otel4s] def acquireDurationInstrument: Option[InstrumentConfig.Histogram] + private[otel4s] def createDurationInstrument: Option[InstrumentConfig.Histogram] + private[otel4s] def destroyedInstrument: Option[InstrumentConfig.Counter] + + /** Replaces the constant attributes attached to every measurement. */ + def withConstAttributes(attributes: Attributes): Config + + /** Appends constant attributes to every measurement. */ + def addConstAttributes(head: Attribute[?], tail: Attribute[?]*): Config + + /** Replaces the idle-resource instrument. */ + def withIdleInstrument(instrument: InstrumentConfig.UpDownCounter): Config + + /** Disables the idle-resource instrument. */ + def withoutIdle: Config + + /** Replaces the in-use-resource instrument. */ + def withInUseInstrument(instrument: InstrumentConfig.UpDownCounter): Config + + /** Disables the in-use-resource instrument. */ + def withoutInUse: Config + + /** Replaces the in-use-duration instrument. */ + def withUseDurationInstrument(instrument: InstrumentConfig.Histogram): Config + + /** Disables the in-use-duration instrument. */ + def withoutUseDuration: Config + + /** Replaces the pending-acquisition instrument. */ + def withPendingAcquireInstrument(instrument: InstrumentConfig.UpDownCounter): Config + + /** Disables the pending-acquisition instrument. */ + def withoutPendingAcquire: Config + + /** Replaces the acquire-duration instrument. */ + def withAcquireDurationInstrument(instrument: InstrumentConfig.Histogram): Config + + /** Disables the acquire-duration instrument. */ + def withoutAcquireDuration: Config + + /** Replaces the resource-creation-duration instrument. */ + def withCreateDurationInstrument(instrument: InstrumentConfig.Histogram): Config + + /** Disables the resource-creation-duration instrument. */ + def withoutCreateDuration: Config + + /** Replaces the resource-destruction counter. */ + def withDestroyedInstrument(instrument: InstrumentConfig.Counter): Config + + /** Disables the resource-destruction counter. */ + def withoutDestroyed: Config + } + + object Config { + + /** Default instrument configuration. */ + object Defaults { + val meterName: String = "org.typelevel.keypool" + + val histogramBucketBoundaries: BucketBoundaries = + BucketBoundaries(.005, .01, .025, .05, .075, .1, .25, .5, .75, 1, 2.5, 5, 7.5, 10) + + val idleInstrument: InstrumentConfig.UpDownCounter = + InstrumentConfig.upDownCounter( + name = "keypool.resource.count", + unit = "{resource}", + description = "The number of resources currently in the pool, by state.", + attributes = Attributes(Attribute("keypool.resource.state", "idle")) + ) + + val inUseInstrument: InstrumentConfig.UpDownCounter = + InstrumentConfig.upDownCounter( + name = "keypool.resource.count", + unit = "{resource}", + description = "The number of resources currently in the pool, by state.", + attributes = Attributes(Attribute("keypool.resource.state", "used")) + ) + + val useDurationInstrument: InstrumentConfig.Histogram = + InstrumentConfig.histogram( + name = "keypool.resource.use.duration", + timeUnit = TimeUnit.SECONDS, + description = "The duration between borrowing a resource and returning it to the pool.", + attributes = Attributes.empty, + explicitBucketBoundaries = histogramBucketBoundaries + ) + + val pendingAcquireInstrument: InstrumentConfig.UpDownCounter = + InstrumentConfig.upDownCounter( + name = "keypool.acquire.pending", + unit = "{request}", + description = "The number of requests currently waiting to acquire a resource.", + attributes = Attributes.empty + ) + + val acquireDurationInstrument: InstrumentConfig.Histogram = + InstrumentConfig.histogram( + name = "keypool.acquire.duration", + timeUnit = TimeUnit.SECONDS, + description = "The time it took to obtain a resource from the pool.", + attributes = Attributes.empty, + explicitBucketBoundaries = histogramBucketBoundaries + ) + + val createDurationInstrument: InstrumentConfig.Histogram = + InstrumentConfig.histogram( + name = "keypool.resource.create.duration", + timeUnit = TimeUnit.SECONDS, + description = "The time it took to create a new resource.", + attributes = Attributes.empty, + explicitBucketBoundaries = histogramBucketBoundaries + ) + + val destroyedInstrument: InstrumentConfig.Counter = + InstrumentConfig.counter( + name = "keypool.resource.destroyed", + unit = "{resource}", + description = "The number of resources removed permanently from the pool.", + attributes = reason => Attributes(Attribute("keypool.destroy.reason", reasonName(reason))) + ) + + private def reasonName(reason: Metrics.DestructionReason): String = + reason match { + case Metrics.DestructionReason.IdleTimeout => "idle_timeout" + case Metrics.DestructionReason.MaxIdle => "max_idle" + case Metrics.DestructionReason.MaxPerKey => "max_per_key" + case Metrics.DestructionReason.NotReusable => "not_reusable" + case Metrics.DestructionReason.PoolClosed => "pool_closed" + } + } + + /** Default metrics configuration. */ + val default: Config = + ConfigImpl( + meterName = Defaults.meterName, + constAttributes = Attributes.empty, + idleInstrument = Some(Defaults.idleInstrument), + inUseInstrument = Some(Defaults.inUseInstrument), + useDurationInstrument = Some(Defaults.useDurationInstrument), + pendingAcquireInstrument = Some(Defaults.pendingAcquireInstrument), + acquireDurationInstrument = Some(Defaults.acquireDurationInstrument), + createDurationInstrument = Some(Defaults.createDurationInstrument), + destroyedInstrument = Some(Defaults.destroyedInstrument) + ) + + private final case class ConfigImpl( + meterName: String, + constAttributes: Attributes, + idleInstrument: Option[InstrumentConfig.UpDownCounter], + inUseInstrument: Option[InstrumentConfig.UpDownCounter], + useDurationInstrument: Option[InstrumentConfig.Histogram], + pendingAcquireInstrument: Option[InstrumentConfig.UpDownCounter], + acquireDurationInstrument: Option[InstrumentConfig.Histogram], + createDurationInstrument: Option[InstrumentConfig.Histogram], + destroyedInstrument: Option[InstrumentConfig.Counter] + ) extends Config { + + def withConstAttributes(attributes: Attributes): Config = + copy(constAttributes = attributes) + + def addConstAttributes(head: Attribute[?], tail: Attribute[?]*): Config = + copy(constAttributes = constAttributes + head ++ tail) + + def withIdleInstrument(instrument: InstrumentConfig.UpDownCounter): Config = + copy(idleInstrument = Some(instrument)) + + def withoutIdle: Config = + copy(idleInstrument = None) + + def withInUseInstrument(instrument: InstrumentConfig.UpDownCounter): Config = + copy(inUseInstrument = Some(instrument)) + + def withoutInUse: Config = + copy(inUseInstrument = None) + + def withUseDurationInstrument(instrument: InstrumentConfig.Histogram): Config = + copy(useDurationInstrument = Some(instrument)) + + def withoutUseDuration: Config = + copy(useDurationInstrument = None) + + def withPendingAcquireInstrument(instrument: InstrumentConfig.UpDownCounter): Config = + copy(pendingAcquireInstrument = Some(instrument)) + + def withoutPendingAcquire: Config = + copy(pendingAcquireInstrument = None) + + def withAcquireDurationInstrument(instrument: InstrumentConfig.Histogram): Config = + copy(acquireDurationInstrument = Some(instrument)) + + def withoutAcquireDuration: Config = + copy(acquireDurationInstrument = None) + + def withCreateDurationInstrument(instrument: InstrumentConfig.Histogram): Config = + copy(createDurationInstrument = Some(instrument)) + + def withoutCreateDuration: Config = + copy(createDurationInstrument = None) + + def withDestroyedInstrument(instrument: InstrumentConfig.Counter): Config = + copy(destroyedInstrument = Some(instrument)) + + def withoutDestroyed: Config = + copy(destroyedInstrument = None) + } + + } + + /** + * Creates metrics provider using otel4s `MeterProvider`. + * + * @example + * {{{ + * val config = Otel4sMetrics.Config.default + * .withConstAttributes(Attributes(Attribute("pool.name", "db-pool"))) + * + * Otel4sMetrics.provider[IO](config) + * }}} + */ + def provider[F[_]: Temporal: MeterProvider]( + config: Config + ): Metrics.Provider[F] = + new Metrics.Provider[F] { + def get: F[Metrics[F]] = + for { + meter <- MeterProvider[F].meter(config.meterName).withVersion(BuildInfo.version).get + + idle <- config.idleInstrument.traverse { instrument => + meter + .upDownCounter[Long](instrument.name) + .withUnit(instrument.unit) + .withDescription(instrument.description) + .create + .tupleLeft(instrument) + } + + inUse <- config.inUseInstrument.traverse { instrument => + meter + .upDownCounter[Long](instrument.name) + .withUnit(instrument.unit) + .withDescription(instrument.description) + .create + .tupleLeft(instrument) + } + + useDurationHistogram <- config.useDurationInstrument.traverse { instrument => + meter + .histogram[Double](instrument.name) + .withUnit(instrument.unit) + .withDescription(instrument.description) + .withExplicitBucketBoundaries(instrument.explicitBucketBoundaries) + .create + .tupleLeft(instrument) + } + + pendingAcquire <- config.pendingAcquireInstrument.traverse { instrument => + meter + .upDownCounter[Long](instrument.name) + .withUnit(instrument.unit) + .withDescription(instrument.description) + .create + .tupleLeft(instrument) + } + + acquireDuration <- config.acquireDurationInstrument.traverse { instrument => + meter + .histogram[Double](instrument.name) + .withUnit(instrument.unit) + .withDescription(instrument.description) + .withExplicitBucketBoundaries(instrument.explicitBucketBoundaries) + .create + .tupleLeft(instrument) + } + + createDurationHistogram <- config.createDurationInstrument.traverse { instrument => + meter + .histogram[Double](instrument.name) + .withUnit(instrument.unit) + .withDescription(instrument.description) + .withExplicitBucketBoundaries(instrument.explicitBucketBoundaries) + .create + .tupleLeft(instrument) + } + + destroyed <- config.destroyedInstrument.traverse { instrument => + meter + .counter[Long](instrument.name) + .withUnit(instrument.unit) + .withDescription(instrument.description) + .create + .tupleLeft(instrument) + } + } yield new Metrics.Unsealed[F] { + + private def attributes(instrument: InstrumentConfig.UpDownCounter): Attributes = + config.constAttributes ++ instrument.attributes + + private def attributes( + instrument: InstrumentConfig.Counter, + reason: Metrics.DestructionReason + ): Attributes = + config.constAttributes ++ instrument.attributes(reason) + + private def attributes( + instrument: InstrumentConfig.Histogram + ): Resource.ExitCase => Attributes = + exitCase => config.constAttributes ++ instrument.attributes(exitCase) + + private def recordElapsed( + histogram: Option[(InstrumentConfig.Histogram, Histogram[F, Double])], + startedAt: FiniteDuration, + exitCase: Resource.ExitCase + ): F[Unit] = + histogram.fold(Temporal[F].unit) { case (instrument, histogram) => + Clock[F].monotonic.flatMap { finishedAt => + val duration = + (finishedAt - startedAt).toNanos.toDouble / instrument.timeUnit + .toNanos(1L) + .toDouble + histogram.record(duration, attributes(instrument)(exitCase)) + } + } + + /** + * Completes when the resource is ready, while the finalizer covers failed or canceled + * acquisitions. We cannot rely on the finalizer alone because it runs when the borrowed + * resource is returned, which would include use time and keep the pending count elevated. + * The guard ensures the metrics are recorded only once. + */ + val acquire: Resource[F, Metrics.Acquisition[F]] = + for { + startedAt <- Resource.eval(Clock[F].monotonic) + completed <- Resource.eval(Ref.of[F, Boolean](false)) + _ <- Resource.eval( + pendingAcquire.fold(Temporal[F].unit) { case (instrument, counter) => + counter.inc(attributes(instrument)) + } + ) + acquisition = new Metrics.Acquisition[F] { + def complete: F[Unit] = + finish(Resource.ExitCase.Succeeded) + + private[keypool] def finish(exitCase: Resource.ExitCase): F[Unit] = + Temporal[F].uncancelable { _ => + completed.flatModify { + case true => (true, Temporal[F].unit) + case false => + val decrementPending = + pendingAcquire.fold(Temporal[F].unit) { case (instrument, counter) => + counter.dec(attributes(instrument)) + } + ( + true, + decrementPending >> recordElapsed(acquireDuration, startedAt, exitCase) + ) + } + } + } + _ <- Resource.onFinalizeCase(acquisition.finish) + } yield acquisition + + val idleInc: F[Unit] = + idle.fold(Temporal[F].unit) { case (instrument, counter) => + counter.inc(attributes(instrument)) + } + + val idleDec: F[Unit] = + idle.fold(Temporal[F].unit) { case (instrument, counter) => + counter.dec(attributes(instrument)) + } + + val inUseCount: Resource[F, Unit] = + inUse.fold(Resource.unit[F]) { case (instrument, counter) => + Resource.make(counter.inc(attributes(instrument)))(_ => + counter.dec(attributes(instrument)) + ) + } + + val useDuration: Resource[F, Unit] = + useDurationHistogram.fold(Resource.unit[F]) { case (instrument, histogram) => + histogram.recordDuration(instrument.timeUnit, attributes(instrument)) + } + + val createDuration: Resource[F, Unit] = + createDurationHistogram.fold(Resource.unit[F]) { case (instrument, histogram) => + histogram.recordDuration(instrument.timeUnit, attributes(instrument)) + } + + def resourceDestroyed(reason: Metrics.DestructionReason): F[Unit] = + destroyed.fold(Temporal[F].unit) { case (instrument, counter) => + counter.inc(attributes(instrument, reason)) + } + } + } +} diff --git a/otel4s/src/test/scala/org/typelevel/keypool/DatabaseSemanticConventionsSpec.scala b/otel4s/src/test/scala/org/typelevel/keypool/DatabaseSemanticConventionsSpec.scala new file mode 100644 index 00000000..8528b664 --- /dev/null +++ b/otel4s/src/test/scala/org/typelevel/keypool/DatabaseSemanticConventionsSpec.scala @@ -0,0 +1,239 @@ +/* + * Copyright (c) 2024 Typelevel + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to + * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of + * the Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER + * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN + * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ + +package org.typelevel.keypool + +import java.util.concurrent.TimeUnit + +import cats.effect.* +import cats.effect.testkit.* +import munit.CatsEffectSuite +import org.typelevel.keypool.otel4s.Otel4sMetrics +import org.typelevel.otel4s.Attributes +import org.typelevel.otel4s.metrics.{BucketBoundaries, MeterProvider} +import org.typelevel.otel4s.sdk.metrics.data.{MetricData, PointData} +import org.typelevel.otel4s.sdk.testkit.AttributesExpectation +import org.typelevel.otel4s.sdk.testkit.metrics.{ + MetricExpectation, + MetricExpectations, + MetricsTestkit, + PointExpectation, + PointSetExpectation +} +import org.typelevel.otel4s.semconv.{MetricSpec, Requirement} +import org.typelevel.otel4s.semconv.experimental.attributes.DbExperimentalAttributes +import org.typelevel.otel4s.semconv.experimental.metrics.DbExperimentalMetrics + +import scala.concurrent.duration.* + +class DatabaseSemanticConventionsSpec extends CatsEffectSuite { + + test("Configure instruments to follow the database connection pool semantic conventions") { + TestControl.executeEmbed { + MetricsTestkit.inMemory[IO]().use { sdk => + implicit val meterProvider: MeterProvider[IO] = sdk.meterProvider + + val poolAttributes = + Attributes(DbExperimentalAttributes.DbClientConnectionPoolName("test")) + + val connectionCountSpec = DbExperimentalMetrics.ClientConnectionCount + val useTimeSpec = DbExperimentalMetrics.ClientConnectionUseTime + val waitTimeSpec = DbExperimentalMetrics.ClientConnectionWaitTime + val createTimeSpec = DbExperimentalMetrics.ClientConnectionCreateTime + val pendingRequestsSpec = DbExperimentalMetrics.ClientConnectionPendingRequests + + def connectionCount(state: String): Otel4sMetrics.InstrumentConfig.UpDownCounter = + Otel4sMetrics.InstrumentConfig.upDownCounter( + name = connectionCountSpec.name, + unit = connectionCountSpec.unit, + description = connectionCountSpec.description, + attributes = Attributes(DbExperimentalAttributes.DbClientConnectionState(state)) + ) + + val config = Otel4sMetrics.Config.default + .withConstAttributes(poolAttributes) + .withIdleInstrument(connectionCount("idle")) + .withInUseInstrument(connectionCount("used")) + .withUseDurationInstrument( + Otel4sMetrics.InstrumentConfig.histogram( + name = useTimeSpec.name, + timeUnit = TimeUnit.SECONDS, + description = useTimeSpec.description, + attributes = Attributes.empty, + explicitBucketBoundaries = HistogramBuckets + ) + ) + .withPendingAcquireInstrument( + Otel4sMetrics.InstrumentConfig.upDownCounter( + name = pendingRequestsSpec.name, + unit = pendingRequestsSpec.unit, + description = pendingRequestsSpec.description, + attributes = Attributes.empty + ) + ) + .withAcquireDurationInstrument( + Otel4sMetrics.InstrumentConfig.histogram( + name = waitTimeSpec.name, + timeUnit = TimeUnit.SECONDS, + description = waitTimeSpec.description, + attributes = Attributes.empty, + explicitBucketBoundaries = HistogramBuckets + ) + ) + .withCreateDurationInstrument( + Otel4sMetrics.InstrumentConfig.histogram( + name = createTimeSpec.name, + timeUnit = TimeUnit.SECONDS, + description = createTimeSpec.description, + attributes = Attributes.empty, + explicitBucketBoundaries = HistogramBuckets + ) + ) + .withoutDestroyed + + Pool + .Builder(Ref.of[IO, Int](1), nothing) + .withMetricsProvider(Otel4sMetrics.provider[IO](config)) + .withMaxIdle(1) + .build + .use { pool => + pool.take + .surround(sdk.collectMetrics.delayBy(1.second)) + .product(sdk.collectMetrics) + .map { case (inUse, afterUse) => + val usedAttributes = + poolAttributes + DbExperimentalAttributes.DbClientConnectionState("used") + val idleAttributes = + poolAttributes + DbExperimentalAttributes.DbClientConnectionState("idle") + + val waitTime = semanticHistogram(waitTimeSpec) + .exactlyPoints( + PointExpectation.histogram + .stats(PointData.Histogram.Stats(0.0, 0.0, 0.0, 1)) + .boundaries(HistogramBuckets) + .counts(1L, 0L, 0L, 0L, 0L) + .attributesExact(poolAttributes) + ) + val createTime = semanticHistogram(createTimeSpec) + .exactlyPoints( + PointExpectation.histogram + .stats(PointData.Histogram.Stats(0.0, 0.0, 0.0, 1)) + .boundaries(HistogramBuckets) + .counts(1L, 0L, 0L, 0L, 0L) + .attributesExact(poolAttributes) + ) + val pendingRequests = semanticSum(pendingRequestsSpec) + .exactlyPoints(PointExpectation.numeric(0L).attributesExact(poolAttributes)) + + assertNotEmitted(inUse, useTimeSpec.name) + assertMetrics( + inUse, + semanticSum(connectionCountSpec) + .exactlyPoints( + PointExpectation.numeric(1L).attributesExact(usedAttributes) + ), + pendingRequests, + waitTime, + createTime + ) + + assertMetrics( + afterUse, + semanticSum(connectionCountSpec) + .exactlyPoints( + PointExpectation.numeric(1L).attributesExact(idleAttributes), + PointExpectation.numeric(0L).attributesExact(usedAttributes) + ), + pendingRequests, + semanticHistogram(useTimeSpec) + .exactlyPoints( + PointExpectation.histogram + .stats(PointData.Histogram.Stats(1.0, 1.0, 1.0, 1)) + .boundaries(HistogramBuckets) + .counts(0L, 1L, 0L, 0L, 0L) + .attributesExact(poolAttributes) + ), + waitTime, + createTime + ) + } + } + } + } + } + + private def semanticSum(spec: MetricSpec): MetricExpectation.Numeric[Long] = { + val attributes = requiredAttributes(spec) + + MetricExpectation + .sum[Long](spec.name) + .description(spec.description) + .unit(spec.unit) + .pointsWhere("all points should have the attributes required by the semantic convention")( + _.forall(point => attributes.matches(point.attributes)) + ) + } + + private def semanticHistogram(spec: MetricSpec): MetricExpectation.Histogram = + MetricExpectation + .histogram(spec.name) + .description(spec.description) + .unit(spec.unit) + .points( + PointSetExpectation.forall( + PointExpectation.histogram.attributes(requiredAttributes(spec)) + ) + ) + + private def requiredAttributes(spec: MetricSpec): AttributesExpectation = { + val requiredKeys = spec.attributeSpecs.collect { + case attribute if attribute.requirement.level == Requirement.Level.Required => + attribute.key + } + + AttributesExpectation.where( + s"required attributes: ${requiredKeys.map(_.name).sorted.mkString(", ")}" + ) { attributes => + requiredKeys.forall(key => attributes.exists(_.key == key)) + } + } + + private def assertMetrics(metrics: List[MetricData], expected: MetricExpectation*): Unit = + MetricExpectations.checkAll(metrics, expected.toList) match { + case Right(_) => () + case Left(mismatches) => fail(MetricExpectations.format(mismatches)) + } + + private def assertNotEmitted(metrics: List[MetricData], names: String*): Unit = + names.foreach { name => + assert( + !MetricExpectations.exists(metrics, MetricExpectation.name(name)), + clues(name, metrics.map(_.name)) + ) + } + + private val HistogramBuckets: BucketBoundaries = + BucketBoundaries(Vector(0.01, 1.0, 100.0, 1000.0)) + + private def nothing(ref: Ref[IO, Int]): IO[Unit] = + ref.get.void + +} diff --git a/otel4s/src/test/scala/org/typelevel/keypool/PoolMetricsSpec.scala b/otel4s/src/test/scala/org/typelevel/keypool/PoolMetricsSpec.scala new file mode 100644 index 00000000..de0fbd1c --- /dev/null +++ b/otel4s/src/test/scala/org/typelevel/keypool/PoolMetricsSpec.scala @@ -0,0 +1,433 @@ +/* + * Copyright (c) 2024 Typelevel + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to + * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of + * the Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER + * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN + * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ + +package org.typelevel.keypool + +import cats.effect.* +import cats.effect.testkit.* +import munit.CatsEffectSuite +import org.typelevel.keypool.internal.Metrics +import org.typelevel.keypool.otel4s.Otel4sMetrics +import org.typelevel.otel4s.{Attribute, Attributes} +import org.typelevel.otel4s.metrics.{BucketBoundaries, MeterProvider} +import org.typelevel.otel4s.sdk.metrics.data.{MetricData, PointData, TimeWindow} +import org.typelevel.otel4s.sdk.testkit.metrics.{ + MetricExpectation, + MetricExpectations, + MetricsTestkit, + PointExpectation +} + +import scala.concurrent.duration.* +import scala.util.control.NoStackTrace + +class PoolMetricsSpec extends CatsEffectSuite { + test("Metrics should be empty for unused pool") { + createTestkit.use { testkit => + mkPool(testkit.meterProvider) + .surround(testkit.collectMetrics) + .map(metrics => assertEquals(metrics, Nil)) + } + } + + test("In use: increment on acquire and decrement on release") { + poolTest() { (sdk, pool) => + for { + inUse <- pool.take.surround(sdk.collectMetrics) + afterUse <- sdk.collectMetrics + } yield { + assertMetrics(inUse, resourceCount("used" -> 1L)) + assertMetrics(afterUse, resourceCount("idle" -> 1L, "used" -> 0L)) + } + } + } + + test("In use: increment on acquire and decrement on release (failure)") { + val exception = new RuntimeException("Something went wrong") with NoStackTrace + + poolTest() { (sdk, pool) => + for { + deferred <- IO.deferred[List[MetricData]] + _ <- pool.take + .surround(sdk.collectMetrics.flatMap(deferred.complete) >> IO.raiseError(exception)) + .attempt + inUse <- deferred.get + afterUse <- sdk.collectMetrics + } yield { + assertMetrics(inUse, resourceCount("used" -> 1L)) + assertMetrics(afterUse, resourceCount("idle" -> 1L, "used" -> 0L)) + } + } + } + + test("Idle: keep 0 when `maxIdle` is 0") { + poolTest(_.withMaxIdle(0)) { (sdk, pool) => + for { + inUse <- pool.take.surround(sdk.collectMetrics) + afterUse <- sdk.collectMetrics + } yield { + assertMetrics(inUse, resourceCount("used" -> 1L)) + assertMetrics( + afterUse, + resourceCount("used" -> 0L), + destroyedMetric("max_idle", 1L) + ) + } + } + } + + test("Idle: keep 1 when `maxIdle` is 1") { + poolTest(_.withMaxIdle(1)) { (sdk, pool) => + for { + inUse <- pool.take.surround(sdk.collectMetrics) + afterUse <- sdk.collectMetrics + } yield { + assertMetrics(inUse, resourceCount("used" -> 1L)) + assertMetrics(afterUse, resourceCount("idle" -> 1L, "used" -> 0L)) + } + } + } + + test("Idle: decrement on reaper cleanup") { + poolTest(_.withMaxIdle(1).withIdleTimeAllowedInPool(1.second)) { (sdk, pool) => + for { + inUse <- pool.take.surround(sdk.collectMetrics) + afterUse <- sdk.collectMetrics + afterSleep <- sdk.collectMetrics.delayBy(6.seconds) + } yield { + assertMetrics(inUse, resourceCount("used" -> 1L)) + assertMetrics(afterUse, resourceCount("idle" -> 1L, "used" -> 0L)) + assertMetrics( + afterSleep, + resourceCount("idle" -> 0L, "used" -> 0L), + destroyedMetric("idle_timeout", 1L) + ) + } + } + + } + + test("Resource count remains accurate when an idle resource is reused") { + poolTest(_.withMaxIdle(1)) { (sdk, pool) => + for { + _ <- pool.take.use_ // create and return one idle resource + duringReuse <- pool.take.surround(sdk.collectMetrics) + afterReuse <- sdk.collectMetrics + } yield { + assertMetrics(duringReuse, resourceCount("idle" -> 0L, "used" -> 1L)) + assertMetrics(afterReuse, resourceCount("idle" -> 1L, "used" -> 0L)) + } + } + } + + test("Resource count returns to zero when the pool closes") { + TestControl.executeEmbed { + createTestkit.use { sdk => + implicit val meterProvider: MeterProvider[IO] = sdk.meterProvider + for { + _ <- Pool + .Builder(Ref.of[IO, Int](1), nothing) + .withMetricsProvider(metricsProvider) + .withMaxIdle(1) + .build + .use(_.take.use_) + afterClose <- sdk.collectMetrics + } yield assertMetrics( + afterClose, + resourceCount("idle" -> 0L, "used" -> 0L), + destroyedMetric("pool_closed", 1L) + ) + } + } + } + + test("Pending count and acquire duration include time waiting for a permit") { + poolTest(_.withMaxTotal(1).withMaxIdle(1)) { (sdk, pool) => + for { + holderStarted <- IO.deferred[Unit] + releaseHolder <- IO.deferred[Unit] + holder <- pool.take.use(_ => holderStarted.complete(()) >> releaseHolder.get).start + _ <- holderStarted.get + waiter <- pool.take.use_.start + pending <- sdk.collectMetrics.delayBy(1.second) + _ <- releaseHolder.complete(()) + _ <- holder.joinWithNever + _ <- waiter.joinWithNever + completed <- sdk.collectMetrics + } yield { + assertMetrics(pending, currentMetric(PendingAcquire, 1L)) + assertMetrics( + completed, + currentMetric(PendingAcquire, 0L), + histogramMetric( + AcquireDuration, + TimeWindow(Duration.Zero, 1.second), + PointData.Histogram.Stats(1.0, 0.0, 1.0, 2), + List(1, 1, 0, 0, 0) + ) + ) + } + } + } + + test("Generate valid metrics") { + poolTest() { (sdk, pool) => + pool.take + .surround(sdk.collectMetrics.delayBy(1.second)) + .product(sdk.collectMetrics) + .map { case (inUse, afterUse) => + val acquireDuration = histogramMetric( + AcquireDuration, + TimeWindow(Duration.Zero, 1.second), + PointData.Histogram.Stats(0.0, 0.0, 0.0, 1), + List(1, 0, 0, 0, 0) + ) + val createDuration = histogramMetric( + CreateDuration, + TimeWindow(Duration.Zero, 1.second), + PointData.Histogram.Stats(0.0, 0.0, 0.0, 1), + List(1, 0, 0, 0, 0) + ) + + assertNotEmitted(inUse, UseDuration) + assertMetrics( + inUse, + resourceCount("used" -> 1L), + currentMetric(PendingAcquire, 0L), + acquireDuration, + createDuration + ) + + assertMetrics( + afterUse, + resourceCount("idle" -> 1L, "used" -> 0L), + currentMetric(PendingAcquire, 0L), + histogramMetric( + UseDuration, + TimeWindow(Duration.Zero, 1.second), + PointData.Histogram.Stats(1.0, 1.0, 1.0, 1), + List(0, 1, 0, 0, 0) + ), + acquireDuration, + createDuration + ) + } + } + } + + test("Histogram attributes can depend on the resource exit case") { + val exception = new RuntimeException("Something went wrong") with NoStackTrace + + TestControl.executeEmbed { + createTestkit.use { sdk => + implicit val meterProvider: MeterProvider[IO] = sdk.meterProvider + + val config = Otel4sMetrics.Config.default + .withConstAttributes(PoolAttributes) + .withoutIdle + .withoutInUse + .withUseDurationInstrument( + Otel4sMetrics.InstrumentConfig.histogram( + name = UseDuration, + timeUnit = java.util.concurrent.TimeUnit.SECONDS, + description = "For how long a resource is in use.", + attributes = + exitCase => Attributes(Attribute("pool.exit_case", exitCaseName(exitCase))), + explicitBucketBoundaries = HistogramBuckets + ) + ) + .withoutPendingAcquire + .withoutAcquireDuration + .withoutCreateDuration + .withoutDestroyed + + Pool + .Builder(Ref.of[IO, Int](1), nothing) + .withMetricsProvider(Otel4sMetrics.provider[IO](config)) + .withMaxIdle(0) + .build + .use { pool => + for { + _ <- pool.take.use(_ => IO.unit) + _ <- pool.take.use(_ => IO.raiseError[Unit](exception)).attempt + started <- IO.deferred[Unit] + fiber <- pool.take.use(_ => started.complete(()) >> IO.never[Unit]).start + _ <- started.get + _ <- fiber.cancel + metrics <- sdk.collectMetrics + } yield assertMetrics( + metrics, + MetricExpectation + .histogram(UseDuration) + .exactlyPoints( + PointExpectation.histogram.attributesExact(exitCaseAttributes("succeeded")), + PointExpectation.histogram.attributesExact(exitCaseAttributes("errored")), + PointExpectation.histogram.attributesExact(exitCaseAttributes("canceled")) + ) + ) + } + } + } + } + + private def poolTest( + customize: Pool.Builder[IO, Ref[IO, Int]] => Pool.Builder[IO, Ref[IO, Int]] = identity + )(scenario: (MetricsTestkit[IO], Pool[IO, Ref[IO, Int]]) => IO[Unit]): IO[Unit] = + TestControl.executeEmbed { + createTestkit.use { sdk => + implicit val meterProvider: MeterProvider[IO] = sdk.meterProvider + val builder = Pool + .Builder(Ref.of[IO, Int](1), nothing) + .withMetricsProvider(metricsProvider) + + customize(builder).build.use(pool => scenario(sdk, pool)) + } + } + + private def mkPool(meterProvider: MeterProvider[IO]) = { + implicit val implicitMeterProvider: MeterProvider[IO] = meterProvider + Pool + .Builder( + Ref.of[IO, Int](1), + nothing + ) + .withMetricsProvider(metricsProvider) + .withMaxTotal(10) + .build + } + + private def metricsProvider(implicit M: MeterProvider[IO]): Metrics.Provider[IO] = + Otel4sMetrics.provider[IO]( + Otel4sMetrics.Config.default + .withConstAttributes(PoolAttributes) + .withUseDurationInstrument( + Otel4sMetrics.InstrumentConfig.histogram( + name = UseDuration, + timeUnit = java.util.concurrent.TimeUnit.SECONDS, + description = "For how long a resource is in use.", + attributes = Attributes.empty, + explicitBucketBoundaries = HistogramBuckets + ) + ) + .withAcquireDurationInstrument( + Otel4sMetrics.InstrumentConfig.histogram( + name = AcquireDuration, + timeUnit = java.util.concurrent.TimeUnit.SECONDS, + description = "How long does it take to acquire a resource.", + attributes = Attributes.empty, + explicitBucketBoundaries = HistogramBuckets + ) + ) + .withCreateDurationInstrument( + Otel4sMetrics.InstrumentConfig.histogram( + name = CreateDuration, + timeUnit = java.util.concurrent.TimeUnit.SECONDS, + description = "How long does it take to create a resource.", + attributes = Attributes.empty, + explicitBucketBoundaries = HistogramBuckets + ) + ) + ) + + private def createTestkit: Resource[IO, MetricsTestkit[IO]] = + MetricsTestkit.inMemory[IO]() + + private def assertMetrics(metrics: List[MetricData], expected: MetricExpectation*): Unit = + MetricExpectations.checkAll(metrics, expected.toList) match { + case Right(_) => () + case Left(mismatches) => fail(MetricExpectations.format(mismatches)) + } + + private def assertNotEmitted(metrics: List[MetricData], names: String*): Unit = + names.foreach { name => + assert( + !MetricExpectations.exists(metrics, MetricExpectation.name(name)), + clues(name, metrics.map(_.name)) + ) + } + + private def currentMetric(name: String, value: Long): MetricExpectation = + MetricExpectation + .sum[Long](name) + .exactlyPoints(PointExpectation.numeric(value).attributesExact(PoolAttributes)) + + private def resourceCount(points: (String, Long)*): MetricExpectation = { + val expected = points.map { case (state, value) => + PointExpectation + .numeric(value) + .attributesExact(PoolAttributes + Attribute("keypool.resource.state", state)) + } + MetricExpectation.sum[Long](ResourceCount).exactlyPoints(expected.head, expected.tail*) + } + + private def destroyedMetric(reason: String, value: Long): MetricExpectation = + MetricExpectation + .sum[Long](Destroyed) + .exactlyPoints( + PointExpectation + .numeric(value) + .attributesExact(PoolAttributes + Attribute("keypool.destroy.reason", reason)) + ) + + private def histogramMetric( + name: String, + timeWindow: TimeWindow, + stats: PointData.Histogram.Stats, + counts: List[Long] + ): MetricExpectation = + MetricExpectation + .histogram(name) + .exactlyPoints( + PointExpectation.histogram + .stats(stats) + .boundaries(HistogramBuckets) + .counts(counts) + .attributesExact(PoolAttributes) + .where(s"time window should be $timeWindow")(_.timeWindow == timeWindow) + ) + + private val HistogramBuckets: BucketBoundaries = + BucketBoundaries(Vector(0.01, 1.0, 100.0, 1000.0)) + + private val PoolAttributes: Attributes = + Attributes(Attribute("pool.name", "test")) + + private def exitCaseAttributes(exitCase: String): Attributes = + PoolAttributes + Attribute("pool.exit_case", exitCase) + + private def exitCaseName(exitCase: Resource.ExitCase): String = + exitCase match { + case Resource.ExitCase.Succeeded => "succeeded" + case Resource.ExitCase.Errored(_) => "errored" + case Resource.ExitCase.Canceled => "canceled" + } + + private val ResourceCount = "keypool.resource.count" + private val UseDuration = "keypool.resource.use.duration" + private val PendingAcquire = "keypool.acquire.pending" + private val AcquireDuration = "keypool.acquire.duration" + private val CreateDuration = "keypool.resource.create.duration" + private val Destroyed = "keypool.resource.destroyed" + + private def nothing(ref: Ref[IO, Int]): IO[Unit] = + ref.get.void + +} diff --git a/project/plugins.sbt b/project/plugins.sbt index d455b0f4..e0b9bace 100644 --- a/project/plugins.sbt +++ b/project/plugins.sbt @@ -6,3 +6,4 @@ addSbtPlugin("org.portable-scala" % "sbt-scala-native-crossproject" % "1.3.2") addSbtPlugin("com.github.cb372" % "sbt-explicit-dependencies" % "0.3.1") addSbtPlugin("org.typelevel" % "sbt-typelevel" % sbtTypelevelVersion) addSbtPlugin("org.typelevel" % "sbt-typelevel-site" % sbtTypelevelVersion) +addSbtPlugin("com.eed3si9n" % "sbt-buildinfo" % "0.13.1")