Datadog

build.sbt

@@ -87,4 +87,8 @@ lazy val root = (project in file("."))
     libraryDependencies += "org.apache.logging.log4j" % "log4j-to-slf4j" % "2.24.3",
     libraryDependencies += "ch.qos.logback" % "logback-classic" % "1.5.16",
     libraryDependencies += "net.logstash.logback" % "logstash-logback-encoder" % "8.0",
+
+    // Metrics
+    libraryDependencies += "dev.zio" %% "zio-metrics-connectors" % "2.3.1",
+    libraryDependencies += "dev.zio" %% "zio-metrics-connectors-datadog" % "2.3.1",
   )

src/main/scala/services/base/DeclaredMetrics.scala

@@ -0,0 +1,14 @@
+package com.sneaksanddata.arcane.framework
+package services.base
+
+import zio.metrics.Metric
+import zio.metrics.Metric.Counter
+
+/**
+ * A object that contains the declared metrics names.
+ */
+object DeclaredMetrics:
+  /**
+   * Number of rows received from the source
+   */
+  val rowsIncoming: Counter[Long] = Metric.counter("rows_incoming")

src/main/scala/services/base/DimensionsProvider.scala

@@ -0,0 +1,15 @@
+package com.sneaksanddata.arcane.framework
+package services.base
+
+import scala.collection.immutable.SortedMap
+
+/**
+ * Provides the metrics dimensions.
+ */
+trait DimensionsProvider:
+  /**
+   * Provides the dimensions.
+   *
+   * @return The dimensions.
+   */
+  def getDimensions: SortedMap[String, String]

src/main/scala/services/base/MetricsService.scala

@@ -0,0 +1,17 @@
+package com.sneaksanddata.arcane.framework
+package services.base
+
+import scala.collection.immutable.SortedMap
+
+/**
+ * A trait that represents a metrics service.
+ */
+trait MetricsService:
+
+  /**
+   * Increments the counter.
+   *
+   * @param name The name of the counter.
+   * @param value The value to increment the counter by.
+   */
+  def incrementCounter(name: String, dimensions: SortedMap[String, String], value: Long): Unit

src/main/scala/services/metrics/ArcaneDimensionsProvider.scala

@@ -0,0 +1,69 @@
+package com.sneaksanddata.arcane.framework
+package services.metrics
+
+import models.app.StreamContext
+import services.base.DimensionsProvider
+import services.metrics.ArcaneDimensionsProvider.camelCaseToSnakeCase
+
+import zio.{ZIO, ZLayer}
+
+import scala.annotation.tailrec
+import scala.collection.immutable.SortedMap
+
+/**
+ * A dimensions provider that provides dimensions for DataDog metrics service.
+ *
+ * @param streamContext The stream context.
+ */
+class ArcaneDimensionsProvider(streamContext: StreamContext) extends DimensionsProvider:
+  /**
+   * Provides the metrics dimensions.
+   *
+   * @return The dimensions.
+   */
+  def getDimensions: SortedMap[String, String] = SortedMap(
+    "arcane.sneaksanddata.com/kind" -> streamContext.streamKind.camelCaseToSnakeCase,
+    "arcane.sneaksanddata.com/mode" -> getStreamMode(streamContext.IsBackfilling),
+    "arcane.sneaksanddata.com/stream_id" -> streamContext.streamId,
+  )
+
+  private def getStreamMode(isBackfilling: Boolean): String = if isBackfilling then "backfill" else "stream"
+
+/**
+ * The companion object for the ArcaneDimensionsProvider class.
+ */
+object ArcaneDimensionsProvider:
+  /**
+   * Creates a new instance of the ArcaneDimensionsProvider.
+   *
+   * @param streamContext The stream context.
+   * @return The ArcaneDimensionsProvider instance.
+   */
+  def apply(streamContext: StreamContext): ArcaneDimensionsProvider = new ArcaneDimensionsProvider(streamContext)
+
+  /**
+   * The ZLayer that creates the ArcaneDimensionsProvider.
+   */
+  val layer: ZLayer[StreamContext, Nothing, DimensionsProvider] =
+    ZLayer {
+      for
+        context <- ZIO.service[StreamContext]
+      yield ArcaneDimensionsProvider(context)
+    }
+
+  /**
+   * Converts a camel case string to a snake case string.
+   *
+   * @return The converted string.
+   */
+  extension (s: String) def camelCaseToSnakeCase: String = toSnakeCase(s, 0, new StringBuilder())
+
+  @tailrec
+  private def toSnakeCase(s: String, from: Int, acc: StringBuilder): String =
+    if from == s.length then
+      acc.toString()
+    else
+      if Character.isUpperCase(s.charAt(from)) && acc.nonEmpty && !Character.isUpperCase(acc.charAt(acc.length-1)) then
+        acc.append("_")
+      acc.append(java.lang.Character.toLowerCase(s.charAt(from)))
+      toSnakeCase(s, from + 1, acc)