Merge pull request #8 from SOFTNETWORK-APP/feature/prometheusHttpMetrics

feat: add HTTP metrics recording to Prometheus for request latency and s…

Author: fupelaqu

build.sbt

@@ -16,7 +16,7 @@ ThisBuild / organization := "app.softnetwork"
 
 name := "generic-persistence-api"
 
-ThisBuild / version := "0.8.5"
+ThisBuild / version := "0.8.6"
 
 lazy val moduleSettings = Seq(
   crossScalaVersions := Seq(scala212, scala213),

kv/src/main/resources/reference.conf

@@ -0,0 +1,6 @@
+# KeyValue extends both ProtobufDomainObject (proto) and KvState/State (chill).
+# Without this explicit binding, Akka finds multiple serializers and may pick
+# the wrong one, causing snapshot deserialization failures.
+akka.actor.serialization-bindings {
+  "app.softnetwork.kv.model.KeyValue" = proto
+}

project/Versions.scala

@@ -2,6 +2,10 @@ object Versions {
 
   val akka = "2.6.20" // TODO 2.6.20 -> 2.8.3
 
+  // Prometheus client_java 1.x — HTTP-route metrics recorded into PrometheusRegistry.defaultRegistry
+  // (the shared registry a downstream /metrics endpoint serves). Story 13.6 Phase B.
+  val prometheus = "1.7.0"
+
   val akkaHttp = "10.2.10" // TODO 10.2.10 -> 10.5.3
 
   val akkaHttpJson4s = "1.39.2" //1.37.0 -> 1.39.2

server/build.sbt

@@ -19,4 +19,17 @@ val tapir = Seq(
   "com.softwaremill.sttp.tapir" %% "tapir-swagger-ui-bundle" % Versions.tapir
 )
 
-libraryDependencies ++= akkaHttp ++ tapir
+// Story 13.6 Phase B — record HTTP-route rate+latency into PrometheusRegistry.defaultRegistry.
+val prometheus = Seq(
+  "io.prometheus" % "prometheus-metrics-core" % Versions.prometheus
+)
+
+// Route-level test for the HttpMetrics directive (akka-http-testkit + text exposition to assert
+// registry samples). Test-scope only.
+val httpMetricsTest = Seq(
+  "com.typesafe.akka" %% "akka-http-testkit" % Versions.akkaHttp % Test,
+  "io.prometheus" % "prometheus-metrics-exposition-textformats" % Versions.prometheus % Test,
+  "org.scalatest" %% "scalatest" % Versions.scalatest % Test
+)
+
+libraryDependencies ++= akkaHttp ++ tapir ++ prometheus ++ httpMetricsTest

server/src/main/scala/app/softnetwork/api/server/ApiRoutes.scala

@@ -54,26 +54,31 @@ trait ApiRoutes extends Directives with GrpcServices with DefaultComplete {
 
   final def mainRoutes: ActorSystem[_] => Route = system => {
     val routes = concat((HealthCheckService :: apiRoutes(system)).map(_.route): _*)
-    handleRejections(rejectionHandler) {
-      handleExceptions(exceptionHandler) {
-        logRequestResult("RestAll") {
-          pathPrefix(config.ServerSettings.RootPath) {
-            Try(
-              respondWithHeaders(RawHeader("Api-Version", applicationVersion)) {
-                routes
-              }
-            ) match {
-              case Success(s) => s
-              case Failure(f) =>
-                log.error(f.getMessage, f.getCause)
-                complete(
-                  HttpResponse(
-                    StatusCodes.InternalServerError,
-                    entity = f.getMessage
+    // Story 13.6 Phase B — record method / normalised-path / status + latency for every request into
+    // PrometheusRegistry.defaultRegistry. Wraps the WHOLE pipeline (outside handleRejections /
+    // handleExceptions) so the final response — rejection/exception ones included — is observed.
+    HttpMetrics.withMetrics {
+      handleRejections(rejectionHandler) {
+        handleExceptions(exceptionHandler) {
+          logRequestResult("RestAll") {
+            pathPrefix(config.ServerSettings.RootPath) {
+              Try(
+                respondWithHeaders(RawHeader("Api-Version", applicationVersion)) {
+                  routes
+                }
+              ) match {
+                case Success(s) => s
+                case Failure(f) =>
+                  log.error(f.getMessage, f.getCause)
+                  complete(
+                    HttpResponse(
+                      StatusCodes.InternalServerError,
+                      entity = f.getMessage
+                    )
                   )
-                )
-            }
-          } ~ grpcRoutes(system)
+              }
+            } ~ grpcRoutes(system)
+          }
         }
       }
     }

server/src/main/scala/app/softnetwork/api/server/HttpMetrics.scala

@@ -0,0 +1,70 @@
+package app.softnetwork.api.server
+
+import akka.http.scaladsl.server.Directives._
+import akka.http.scaladsl.server.Route
+import io.prometheus.metrics.core.metrics.{Counter, Histogram}
+
+/** Story 13.6 Phase B — HTTP request rate + latency, recorded into the global
+  * `PrometheusRegistry.defaultRegistry`. A downstream service's `/metrics` endpoint (served from
+  * the same default registry) exposes these series; the `service` label is added at scrape time by
+  * the ServiceMonitor relabeling (these are library-defined series with a fixed label set).
+  *
+  * `path` is normalised (id-like segments collapsed to `:id`) to bound cardinality, since the raw
+  * request path can embed UUIDs / numeric ids.
+  */
+object HttpMetrics {
+
+  private val requests: Counter = Counter
+    .builder()
+    .name("http_requests")
+    .help("HTTP requests, by method / normalised path / status")
+    .labelNames("method", "path", "status")
+    .register()
+
+  private val duration: Histogram = Histogram
+    .builder()
+    .name("http_request_duration_seconds")
+    .help("HTTP request duration in seconds, by method / normalised path")
+    .labelNames("method", "path")
+    .classicUpperBounds(0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0)
+    .register()
+
+  def record(method: String, path: String, status: Int, seconds: Double): Unit = {
+    val p = normalizePath(path)
+    requests.labelValues(method, p, status.toString).inc()
+    duration.labelValues(method, p).observe(seconds)
+  }
+
+  /** akka-http directive: times the request and records method / normalised-path / status + latency
+    * when the inner route completes. Wrap it OUTSIDE rejection/exception handling so `mapResponse`
+    * observes the FINAL response (rejection- and exception-derived responses included).
+    */
+  def withMetrics(inner: Route): Route =
+    extractRequest { req =>
+      val startNanos = System.nanoTime()
+      mapResponse { resp =>
+        record(
+          req.method.value,
+          req.uri.path.toString,
+          resp.status.intValue(),
+          (System.nanoTime() - startNanos) / 1e9d
+        )
+        resp
+      }(inner)
+    }
+
+  private val HexLike = "^[0-9a-fA-F-]+$".r
+  private val DigitsOnly = "^[0-9]+$".r
+
+  /** Collapse id-like segments (UUID/hex >= 8 chars, or all-digits) to `:id`. */
+  def normalizePath(path: String): String =
+    path
+      .split("/", -1)
+      .map { seg =>
+        if (seg.isEmpty) seg
+        else if (seg.length >= 8 && HexLike.pattern.matcher(seg).matches()) ":id"
+        else if (DigitsOnly.pattern.matcher(seg).matches()) ":id"
+        else seg
+      }
+      .mkString("/")
+}

server/src/test/scala/app/softnetwork/api/server/HttpMetricsSpec.scala

@@ -0,0 +1,83 @@
+package app.softnetwork.api.server
+
+import akka.http.scaladsl.model.StatusCodes
+import akka.http.scaladsl.server.{Directives, ExceptionHandler, RejectionHandler, Route}
+import akka.http.scaladsl.testkit.ScalatestRouteTest
+import io.prometheus.metrics.expositionformats.PrometheusTextFormatWriter
+import io.prometheus.metrics.model.registry.PrometheusRegistry
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+import java.io.ByteArrayOutputStream
+
+/** Story 13.6 Phase B — proves the HttpMetrics directive emits request/latency samples into the
+  * default registry for normal, rejection and exception responses, and that `normalizePath`
+  * collapses id-like segments.
+  */
+class HttpMetricsSpec extends AnyWordSpec with Matchers with ScalatestRouteTest with Directives {
+
+  // Mirror the ApiRoutes wrapping: metrics OUTSIDE rejection/exception handling so the final response
+  // (including the rejection-derived 404 and the exception-derived 500) is observed.
+  private val exceptionHandler = ExceptionHandler { case _: RuntimeException =>
+    complete(StatusCodes.InternalServerError -> "boom")
+  }
+
+  private val route: Route =
+    HttpMetrics.withMetrics {
+      handleRejections(RejectionHandler.default) {
+        handleExceptions(exceptionHandler) {
+          concat(
+            path("ping")(get(complete("pong"))),
+            path("licenses" / Segment)(id => get(complete(id))),
+            path("boom")(get(failWith(new RuntimeException("boom"))))
+          )
+        }
+      }
+    }
+
+  private def scrapeText(): String = {
+    val writer = PrometheusTextFormatWriter.builder().build()
+    val out = new ByteArrayOutputStream()
+    writer.write(out, PrometheusRegistry.defaultRegistry.scrape())
+    out.toString("UTF-8")
+  }
+
+  "HttpMetrics.normalizePath" should {
+    "collapse numeric and uuid/hex segments to :id" in {
+      HttpMetrics.normalizePath("/api/licenses/123") shouldBe "/api/licenses/:id"
+      HttpMetrics.normalizePath(
+        "/api/licenses/550e8400-e29b-41d4-a716-446655440000"
+      ) shouldBe "/api/licenses/:id"
+    }
+    "leave non-id segments untouched" in {
+      HttpMetrics.normalizePath("/api/healthcheck") shouldBe "/api/healthcheck"
+      HttpMetrics.normalizePath("/ping") shouldBe "/ping"
+    }
+  }
+
+  "The HttpMetrics directive" should {
+    "record a 200, normalising an id segment in the path label" in {
+      Get("/ping") ~> route ~> check { status shouldBe StatusCodes.OK }
+      Get("/licenses/550e8400-e29b-41d4-a716-446655440000") ~> route ~> check {
+        status shouldBe StatusCodes.OK
+      }
+      val text = scrapeText()
+      text should include("""http_requests_total{method="GET",path="/ping",status="200"}""")
+      text should include("""http_requests_total{method="GET",path="/licenses/:id",status="200"}""")
+      // histogram observed too
+      text should include("""http_request_duration_seconds_count{method="GET",path="/ping"}""")
+    }
+
+    "record a rejection-derived 404 response" in {
+      Get("/does-not-exist") ~> route ~> check { status shouldBe StatusCodes.NotFound }
+      scrapeText() should include(
+        """http_requests_total{method="GET",path="/does-not-exist",status="404"}"""
+      )
+    }
+
+    "record an exception-derived 500 response" in {
+      Get("/boom") ~> route ~> check { status shouldBe StatusCodes.InternalServerError }
+      scrapeText() should include("""http_requests_total{method="GET",path="/boom",status="500"}""")
+    }
+  }
+}