Merge pull request #350 from lichess-org/improve-kamon

Make kamon calls blocking

app/src/main/scala/App.scala

@@ -13,9 +13,9 @@ object App extends IOApp.Simple:
 
   def app: Resource[IO, Unit] =
     for
-      config <- AppConfig.load.toResource
+      config <- AppConfig.load().toResource
       _      <- Logger[IO].info(s"Starting lila-fishnet with config: $config").toResource
-      _      <- KamonInitiator.apply.init(config.kamon).toResource
+      _      <- KamonInitiator.apply().init(config.kamon).toResource
       res    <- AppResources.instance(config.redis)
       _      <- FishnetApp(res, config).run()
     yield ()
@@ -34,4 +34,4 @@ class FishnetApp(res: AppResources, config: AppConfig)(using Logger[IO]):
   private def createExecutor: Resource[IO, Executor] =
     val lilaClient = LilaClient(res.redisPubsub)
     val repository = StateRepository.instance(config.repository.path)
-    Executor.instance(lilaClient, repository, Monitor(), config.executor)
+    Monitor().toResource.flatMap(Executor.instance(lilaClient, repository, _, config.executor))

app/src/main/scala/AppConfig.scala

@@ -8,7 +8,7 @@ import com.comcast.ip4s.*
 
 object AppConfig:
 
-  def load: IO[AppConfig] = appConfig.load[IO]
+  def load(): IO[AppConfig] = appConfig.load[IO]
 
   def appConfig = (
     RedisConfig.config,

app/src/main/scala/KamonInitiator.scala

@@ -8,6 +8,6 @@ trait KamonInitiator:
   def init(config: KamonConfig): IO[Unit]
 
 object KamonInitiator:
-  def apply: KamonInitiator = new:
+  def apply(): KamonInitiator = new:
     def init(config: KamonConfig): IO[Unit] =
-      IO(Kamon.init()).whenA(config.enabled)
+      IO.blocking(Kamon.init()).whenA(config.enabled)

app/src/main/scala/Monitor.scala

@@ -1,6 +1,7 @@
 package lila.fishnet
 
 import cats.effect.IO
+import cats.syntax.all.*
 import kamon.Kamon
 import kamon.metric.Timer
 
@@ -20,17 +21,25 @@ object Monitor:
   val lvl8AcquiredTimeRequest = Kamon.timer("move.acquired.lvl8").withoutTags()
   val lvl1FullTimeRequest     = Kamon.timer("move.full.lvl1").withoutTags()
 
-  def apply(): Monitor = new:
-    def success(work: Work.Task): IO[Unit] =
-      IO.realTimeInstant.map: now =>
-        if work.request.level == 8 then
-          work.acquiredAt.foreach(at => record(lvl8AcquiredTimeRequest, at, now))
-        if work.request.level == 1 then record(lvl1FullTimeRequest, work.createdAt, now)
-
-    def updateSize(state: AppState): IO[Unit] =
-      IO(dbSize.update(state.size.toDouble)) *>
-        IO(dbQueued.update(state.count(_.nonAcquired).toDouble)) *>
-        IO(dbAcquired.update(state.count(_.isAcquired).toDouble)).void
-
-  private def record(timer: Timer, start: Instant, end: Instant): Unit =
-    val _ = timer.record(start.until(end, ChronoUnit.MILLIS), TimeUnit.MILLISECONDS)
+  def apply(): IO[Monitor] =
+    (
+      IO.blocking(Kamon.gauge("db.size").withoutTags()),
+      IO.blocking(Kamon.gauge("db.queued").withoutTags()),
+      IO.blocking(Kamon.gauge("db.acquired").withoutTags()),
+      IO.blocking(Kamon.timer("move.acquired.lvl8").withoutTags()),
+      IO.blocking(Kamon.timer("move.full.lvl1").withoutTags())
+    ).parMapN: (dbSize, dbQueued, dbAcquired, lvl8AcquiredTimeRequest, lvl1FullTimeRequest) =>
+      new Monitor:
+        def success(work: Work.Task): IO[Unit] =
+          IO.realTimeInstant.map: now =>
+            if work.request.level == 8 then
+              work.acquiredAt.foreach(at => record(lvl8AcquiredTimeRequest, at, now))
+            else if work.request.level == 1 then record(lvl1FullTimeRequest, work.createdAt, now)
+
+        def updateSize(state: AppState): IO[Unit] =
+          IO(dbSize.update(state.size.toDouble)) *>
+            IO(dbQueued.update(state.count(_.nonAcquired).toDouble)) *>
+            IO(dbAcquired.update(state.count(_.isAcquired).toDouble)).void
+
+        private def record(timer: Timer, start: Instant, end: Instant): Unit =
+          val _ = timer.record(start.until(end, ChronoUnit.MILLIS), TimeUnit.MILLISECONDS)