Fix metrics response (#1176)

Author: muralibasani

container/grafana/dashboards/karapace.json

@@ -501,7 +501,7 @@
           },
           "disableTextWrap": false,
           "editorMode": "builder",
-          "expr": "karapace_schema_reader_subjects_total",
+          "expr": "karapace_schema_reader_subjects_total{exported_job=\"karapace-schema-registry\"}",
           "fullMetaSearch": false,
           "includeNullMetadata": true,
           "instant": false,
@@ -574,7 +574,7 @@
           },
           "disableTextWrap": false,
           "editorMode": "builder",
-          "expr": "karapace_schema_reader_schemas_total",
+          "expr": "karapace_schema_reader_schemas_total{exported_job=\"karapace-schema-registry\"}",
           "fullMetaSearch": false,
           "includeNullMetadata": true,
           "instant": false,

container/opentelemetry/collector-config.yaml

@@ -14,10 +14,8 @@ exporters:
     endpoint: jaeger:4317
     tls:
       insecure: true
-  otlphttp/prometheus:
-    endpoint: http://prometheus:9090/api/v1/otlp
-    tls:
-      insecure: true
+  prometheus:
+    endpoint: "0.0.0.0:8889"
 
 service:
   pipelines:
@@ -26,4 +24,4 @@ service:
       exporters: [otlp]
     metrics:
       receivers: [otlp]
-      exporters: [otlphttp/prometheus]
+      exporters: [prometheus]

container/prometheus/prometheus.yml

@@ -14,15 +14,18 @@ rule_files:
 scrape_configs:
   - job_name: karapace-registry
     metrics_path: /metrics
+    scheme: https
+    tls_config:
+      ca_file: /opt/prometheus/certs/rootCA.pem
     static_configs:
       - targets:
-        - karapace-registry:8081
+        - karapace-schema-registry:8081
 
   - job_name: karapace-rest
     metrics_path: /metrics
     static_configs:
       - targets:
-        - karapace-rest:8082
+        - karapace-rest-proxy:8082
 
   - job_name: statsd-exporter
     metrics_path: /metrics
@@ -33,6 +36,7 @@ scrape_configs:
         - statsd-exporter:9102
 
   - job_name: opentelemetry-collector
+    metrics_path: /metrics
     static_configs:
       - targets:
-        - opentelemetry-collector:8888
+        - opentelemetry-collector:8889

src/karapace/api/routers/metrics.py

@@ -7,6 +7,7 @@
 from fastapi import APIRouter, Depends, Response
 from karapace.api.container import SchemaRegistryContainer
 from karapace.core.instrumentation.prometheus import PrometheusInstrumentation
+from karapace.rapu import HTTPResponse
 
 metrics_router = APIRouter(
     prefix=PrometheusInstrumentation.METRICS_ENDPOINT_PATH,
@@ -20,4 +21,10 @@
 async def metrics(
     prometheus: PrometheusInstrumentation = Depends(Provide[SchemaRegistryContainer.karapace_container.prometheus]),
 ) -> Response:
-    return Response(content=await prometheus.serve_metrics(), media_type=prometheus.CONTENT_TYPE_LATEST)
+    try:
+        await prometheus.serve_metrics()
+    except HTTPResponse as ex:
+        # FastAPI needs to extract the body and return a Response
+        # generate_latest() returns bytes, so ex.body is already bytes
+        content = ex.body if isinstance(ex.body, bytes) else ex.body.encode("utf-8")
+        return Response(content=content, media_type=prometheus.CONTENT_TYPE_LATEST)

src/karapace/core/instrumentation/prometheus.py

@@ -10,9 +10,10 @@
 
 from aiohttp.web import middleware, Request, Response
 from collections.abc import Awaitable, Callable
-from karapace.rapu import RestApp
+from http import HTTPStatus
+from karapace.rapu import HTTPResponse, RestApp
 from prometheus_client import CollectorRegistry, Counter, Gauge, generate_latest, Histogram
-from typing import Final
+from typing import Final, NoReturn
 
 import logging
 import time
@@ -70,8 +71,14 @@ def setup_metrics(cls, *, app: RestApp) -> None:
         app.app[cls.karapace_http_requests_in_progress] = cls.karapace_http_requests_in_progress
 
     @classmethod
-    async def serve_metrics(cls) -> bytes:
-        return generate_latest(cls.registry)
+    async def serve_metrics(cls) -> NoReturn:
+        metrics_data = generate_latest(cls.registry)
+        # Raise HTTPResponse for RestApp compatibility
+        raise HTTPResponse(
+            body=metrics_data,
+            status=HTTPStatus.OK,
+            content_type=cls.CONTENT_TYPE_LATEST,
+        )
 
     @classmethod
     @middleware

src/karapace/core/schema_reader.py

@@ -367,6 +367,8 @@ def _is_ready(self) -> bool:
         if ready:
             self.max_messages_to_process = MAX_MESSAGES_TO_CONSUME_AFTER_STARTUP
             LOG.info("Ready in %s seconds", time.monotonic() - self.start_time)
+            # Initialize metrics with current database state when becoming ready
+            self._initialize_metrics()
         return ready
 
     def highest_offset(self) -> int:
@@ -499,6 +501,33 @@ def _update_is_ready_flag(self) -> None:
             with self._ready_lock:
                 self._ready = new_ready_flag
 
+    def _update_schema_gauges(self) -> tuple[int, int, int, int]:
+        """Update all schema-related gauge metrics with current database state.
+
+        Returns:
+            Tuple of (num_subjects, num_schemas, live_versions, soft_deleted_versions)
+        """
+        num_schemas = self.database.num_schemas()
+        num_subjects = self.database.num_subjects()
+        live_versions, soft_deleted_versions = self.database.num_schema_versions()
+
+        self.stats.set_schemas_num_total(value=num_schemas)
+        self.stats.set_subjects_num_total(value=num_subjects)
+        self.stats.set_schema_versions_num_total(live_versions=live_versions, soft_deleted_versions=soft_deleted_versions)
+
+        return num_subjects, num_schemas, live_versions, soft_deleted_versions
+
+    def _initialize_metrics(self) -> None:
+        """Initialize metrics with current database state."""
+        num_subjects, num_schemas, live_versions, soft_deleted_versions = self._update_schema_gauges()
+        LOG.info(
+            "Metrics initialized: subjects=%s, schemas=%s, live_versions=%s, soft_deleted_versions=%s",
+            num_subjects,
+            num_schemas,
+            live_versions,
+            soft_deleted_versions,
+        )
+
     def _report_schema_metrics(
         self,
         schema_records_processed_keymode_canonical: int,
@@ -510,20 +539,9 @@ def _report_schema_metrics(
             with_deprecated_key=schema_records_processed_keymode_deprecated_karapace,
         )
 
-        # Update following gauges only if there is a possibility of a change.
-        records_processed = bool(
-            schema_records_processed_keymode_canonical or schema_records_processed_keymode_deprecated_karapace
-        )
-        if records_processed:
-            num_schemas = self.database.num_schemas()
-            num_subjects = self.database.num_subjects()
-            self.stats.set_schemas_num_total(value=num_schemas)
-            self.stats.set_subjects_num_total(value=num_subjects)
-
-            live_versions, soft_deleted_versions = self.database.num_schema_versions()
-            self.stats.set_schema_versions_num_total(
-                live_versions=live_versions, soft_deleted_versions=soft_deleted_versions
-            )
+        # Always update gauges to reflect current database state
+        # This ensures metrics stay accurate even if no new records are processed
+        self._update_schema_gauges()
 
     def _handle_msg_config(self, key: dict, value: dict | None) -> None:
         subject = key.get("subject")

src/karapace/core/stats.py

@@ -37,6 +37,8 @@ def __init__(self, *, config: Config, meter: Meter) -> None:
         self.sentry_client: Final = get_sentry_client(sentry_config=(config.sentry or None))
         self._meter = meter
 
+        LOG.info("Initializing StatsClient with tags: %s", self._tags)
+
         # Supports labels for keymode
         self._schema_records_processed_counter: Final[Counter] = self._meter.get_meter().create_counter(
             name=METRIC_SCHEMA_TOPIC_RECORDS_PROCESSED_COUNT,
@@ -67,12 +69,20 @@ def schema_records_processed(self, *, with_canonical_key: int, with_deprecated_k
         )
 
     def set_schemas_num_total(self, *, value: int) -> None:
+        LOG.debug("Setting schemas gauge to %s with attributes %s", value, self._tags)
         self._total_schemas_gauge.set(amount=value, attributes=self._tags)
 
     def set_subjects_num_total(self, *, value: int) -> None:
+        LOG.debug("Setting subjects gauge to %s with attributes %s", value, self._tags)
         self._total_subjects_gauge.set(amount=value, attributes=self._tags)
 
     def set_schema_versions_num_total(self, *, live_versions: int, soft_deleted_versions: int) -> None:
+        LOG.debug(
+            "Setting schema versions gauge: live=%s, soft_deleted=%s with attributes %s",
+            live_versions,
+            soft_deleted_versions,
+            self._tags,
+        )
         self._schema_versions_gauge.set(amount=live_versions, attributes={"state": "live", **self._tags})
         self._schema_versions_gauge.set(amount=soft_deleted_versions, attributes={"state": "soft_deleted", **self._tags})
 

tests/unit/instrumentation/test_prometheus.py

@@ -6,6 +6,7 @@
 """
 
 import logging
+from http import HTTPStatus
 from unittest.mock import AsyncMock, MagicMock, call, patch
 
 import aiohttp.web
@@ -14,7 +15,7 @@
 from prometheus_client import CollectorRegistry, Counter, Gauge, Histogram
 
 from karapace.core.instrumentation.prometheus import PrometheusInstrumentation
-from karapace.rapu import RestApp
+from karapace.rapu import HTTPResponse, RestApp
 
 
 # Simple request object implementing mapping semantics for START_TIME used by tests
@@ -130,9 +131,19 @@ def test_setup_metrics(self, caplog: LogCaptureFixture, prometheus: PrometheusIn
 
     @patch("karapace.core.instrumentation.prometheus.generate_latest")
     async def test_serve_metrics(self, generate_latest: MagicMock, prometheus: PrometheusInstrumentation) -> None:
-        await prometheus.serve_metrics()
+        mock_metrics_data = b"# HELP test_metric Test metric\n# TYPE test_metric counter\ntest_metric 1\n"
+        generate_latest.return_value = mock_metrics_data
+
+        with pytest.raises(HTTPResponse) as exc_info:
+            await prometheus.serve_metrics()
+
         generate_latest.assert_called_once_with(prometheus.registry)
 
+        # Verify HTTPResponse has correct attributes
+        assert exc_info.value.status == HTTPStatus.OK
+        assert exc_info.value.headers.get("Content-Type") == prometheus.CONTENT_TYPE_LATEST
+        assert exc_info.value.body == mock_metrics_data
+
     @patch("karapace.core.instrumentation.prometheus.time")
     async def test_http_request_metrics_middleware(
         self,