Add commonly used decoders to runtime modules

modules/runtime-common/src/main/scala/com/snowplowanalytics/snowplow/runtime/HealthProbe.scala

@@ -11,6 +11,7 @@ import cats.effect.{Async, Resource, Sync}
 import cats.data.Kleisli
 import cats.implicits._
 import com.comcast.ip4s.{Ipv4Address, Port}
+import io.circe.Decoder
 import org.http4s.ember.server.EmberServerBuilder
 import org.http4s.{HttpApp, Response, Status}
 import org.typelevel.log4cats.Logger
@@ -37,6 +38,12 @@ object HealthProbe {
       }
       .void
 
+  object decoders {
+    def portDecoder: Decoder[Port] = Decoder.decodeInt.emap { port =>
+      Port.fromInt(port).toRight("Invalid port")
+    }
+  }
+
   private def httpApp[F[_]: Sync](isHealthy: F[Status]): HttpApp[F] =
     Kleisli { _ =>
       isHealthy.flatMap {

modules/runtime-common/src/main/scala/com/snowplowanalytics/snowplow/runtime/Metrics.scala

@@ -11,6 +11,9 @@ import cats.effect.{Async, Sync}
 import cats.effect.kernel.{Ref, Resource}
 import cats.implicits._
 import fs2.Stream
+import io.circe.Decoder
+import io.circe.config.syntax._
+import io.circe.generic.semiauto._
 import org.typelevel.log4cats.{Logger, SelfAwareStructuredLogger}
 import org.typelevel.log4cats.slf4j.Slf4jLogger
 
@@ -43,6 +46,8 @@ abstract class Metrics[F[_]: Async, S <: Metrics.State](
 
 object Metrics {
 
+  /** Public API */
+
   case class StatsdConfig(
     hostname: String,
     port: Int,
@@ -51,6 +56,11 @@ object Metrics {
     prefix: String
   )
 
+  object StatsdConfig {
+    implicit def stasdConfigDecoder: Decoder[Option[StatsdConfig]] =
+      deriveDecoder[StatsdUnresolvedConfig].map(resolveConfig(_))
+  }
+
   trait State {
     def toKVMetrics: List[KVMetric]
   }
@@ -70,6 +80,30 @@ object Metrics {
     def metricType: MetricType
   }
 
+  /** Private implementation */
+
+  /**
+   * The raw config received by combinging user-provided config with snowplow defaults
+   *
+   * If user did not configure statsd, then hostname is None and all other params are defined via
+   * our defaults.
+   */
+  private case class StatsdUnresolvedConfig(
+    hostname: Option[String],
+    port: Int,
+    tags: Map[String, String],
+    period: FiniteDuration,
+    prefix: String
+  )
+
+  private def resolveConfig(from: StatsdUnresolvedConfig): Option[StatsdConfig] =
+    from match {
+      case StatsdUnresolvedConfig(Some(hostname), port, tags, period, prefix) =>
+        Some(StatsdConfig(hostname, port, tags, period, prefix))
+      case StatsdUnresolvedConfig(None, _, _, _, _) =>
+        None
+    }
+
   private implicit def logger[F[_]: Sync]: SelfAwareStructuredLogger[F] = Slf4jLogger.getLogger[F]
 
   private trait Reporter[F[_]] {

modules/runtime-common/src/test/scala/com/snowplowanalytics/snowplow/runtime/MetricsSpec.scala

@@ -0,0 +1,87 @@
+/*
+ * Copyright (c) 2023-present Snowplow Analytics Ltd. All rights reserved.
+ *
+ * This program is licensed to you under the Snowplow Community License Version 1.0,
+ * and you may not use this file except in compliance with the Snowplow Community License Version 1.0.
+ * You may obtain a copy of the Snowplow Community License Version 1.0 at https://docs.snowplow.io/community-license-1.0
+ */
+package com.snowplowanalytics.snowplow.runtime
+
+import cats.implicits._
+import io.circe.DecodingFailure
+import io.circe.literal._
+import org.specs2.Specification
+
+import scala.concurrent.duration.DurationLong
+
+class MetricsSpec extends Specification {
+
+  def is = s2"""
+  The statsd config decoder should:
+    Decode a valid JSON config when hostname is set $e1
+    Decode a valid JSON config when hostname is missing $e2
+    Not decode JSON if other required field is missing $e3
+
+  """
+
+  def e1 = {
+    val json = json"""
+    {
+      "hostname": "statsd.localdomain",
+      "port": 5432,
+      "tags": {
+        "abc": "xyz"
+      },
+      "period": "42 seconds",
+      "prefix": "foo.bar"
+    }
+    """
+
+    json.as[Option[Metrics.StatsdConfig]] must beRight.like { case Some(c: Metrics.StatsdConfig) =>
+      List(
+        c.hostname must beEqualTo("statsd.localdomain"),
+        c.port must beEqualTo(5432),
+        c.tags must beEqualTo(Map("abc" -> "xyz")),
+        c.period must beEqualTo(42.seconds),
+        c.prefix must beEqualTo("foo.bar")
+      ).reduce(_ and _)
+    }
+  }
+
+  def e2 = {
+    val json = json"""
+    {
+      "port": 5432,
+      "tags": {
+        "abc": "xyz"
+      },
+      "period": "42 seconds",
+      "prefix": "foo.bar"
+    }
+    """
+
+    json.as[Option[Metrics.StatsdConfig]] must beRight.like { case c: Option[Metrics.StatsdConfig] =>
+      c must beNone
+    }
+  }
+
+  def e3 = {
+
+    // missing port
+    val json = json"""
+    {
+      "hostname": "statsd.localdomain",
+      "tags": {
+        "abc": "xyz"
+      },
+      "period": "42 seconds",
+      "prefix": "foo.bar"
+    }
+    """
+
+    json.as[Option[Metrics.StatsdConfig]] must beLeft.like { case e: DecodingFailure =>
+      e.show must beEqualTo("DecodingFailure at .port: Missing required field")
+    }
+  }
+
+}

project/Dependencies.scala

@@ -157,13 +157,16 @@ object Dependencies {
     cats,
     catsEffectKernel,
     circeConfig,
+    circeGeneric,
     emberServer,
     fs2,
     igluClient,
     log4cats,
     slf4jApi,
     tracker,
-    trackerEmit
+    trackerEmit,
+    specs2,
+    circeLiteral % Test
   )
 
   val loadersCommonDependencies = Seq(