Prometheus client metrics support

karapace.config.json

@@ -27,5 +27,12 @@
     "registry_authfile": null,
     "topic_name": "_schemas",
     "protobuf_runtime_directory": "runtime",
-    "session_timeout_ms": 10000
+    "session_timeout_ms": 10000,
+    "stats_service": "statsd",
+    "metrics_extended": true,
+    "statsd_host": "127.0.0.1",
+    "statsd_port": 8125,
+    "prometheus_host": "127.0.0.1",
+    "prometheus_port": 8005,
+
 }

karapace/base_stats.py

@@ -0,0 +1,57 @@
+"""
+karapace - basestats
+
+Supports base class for statsd and prometheus protocols:
+
+Copyright (c) 2023 Aiven Ltd
+See LICENSE for details
+"""
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from contextlib import contextmanager
+from karapace.config import Config
+from karapace.sentry import get_sentry_client
+from typing import Final, Iterator
+
+import time
+
+
+class StatsClient(ABC):
+    @abstractmethod
+    def __init__(
+        self,
+        config: Config,
+    ) -> None:
+        self.sentry_client: Final = get_sentry_client(sentry_config=config.get("sentry", None))
+
+    @contextmanager
+    def timing_manager(self, metric: str, tags: dict | None = None) -> Iterator[None]:
+        start_time = time.monotonic()
+        yield
+        self.timing(metric, time.monotonic() - start_time, tags)
+
+    @abstractmethod
+    def gauge(self, metric: str, value: float, tags: dict | None = None) -> None:
+        pass
+
+    @abstractmethod
+    def increase(self, metric: str, inc_value: int = 1, tags: dict | None = None) -> None:
+        pass
+
+    @abstractmethod
+    def timing(self, metric: str, value: float, tags: dict | None = None) -> None:
+        pass
+
+    def unexpected_exception(self, ex: Exception, where: str, tags: dict | None = None) -> None:
+        all_tags = {
+            "exception": ex.__class__.__name__,
+            "where": where,
+        }
+        all_tags.update(tags or {})
+        self.increase("exception", tags=all_tags)
+        scope_args = {**(tags or {}), "where": where}
+        self.sentry_client.unexpected_exception(error=ex, where=where, tags=scope_args)
+
+    def close(self) -> None:
+        self.sentry_client.close()

karapace/config.py

@@ -78,6 +78,12 @@ class Config(TypedDict):
     name_strategy_validation: bool
     master_election_strategy: str
     protobuf_runtime_directory: str
+    stats_service: str
+    metrics_extended: bool
+    statsd_host: str
+    statsd_port: int
+    prometheus_host: str | None
+    prometheus_port: int | None
 
     sentry: NotRequired[Mapping[str, object]]
     tags: NotRequired[Mapping[str, object]]
@@ -150,6 +156,12 @@ class ConfigDefaults(Config, total=False):
     "name_strategy_validation": True,
     "master_election_strategy": "lowest",
     "protobuf_runtime_directory": "runtime",
+    "stats_service": "statsd",
+    "metrics_extended": True,
+    "statsd_host": "127.0.0.1",
+    "statsd_port": 8125,
+    "prometheus_host": "127.0.0.1",
+    "prometheus_port": 8005,
 }
 SECRET_CONFIG_OPTIONS = [SASL_PLAIN_PASSWORD]
 

karapace/metrics.py

@@ -0,0 +1,103 @@
+"""
+karapace - metrics
+Supports collection of system metrics
+list of supported metrics:
+connections-active - The number of active HTTP(S) connections to server.
+                     Data collected inside aiohttp request handler.
+
+Copyright (c) 2023 Aiven Ltd
+See LICENSE for details
+"""
+from __future__ import annotations
+
+from karapace.base_stats import StatsClient
+from karapace.config import Config
+from karapace.prometheus import PrometheusClient
+from karapace.statsd import StatsdClient
+
+import threading
+
+
+class MetricsException(Exception):
+    pass
+
+
+class Singleton(type):
+    _instance: Singleton | None = None
+
+    def __call__(cls, *args: str, **kwargs: int) -> Singleton:
+        if cls._instance is None:
+            instance = super().__call__(*args, **kwargs)
+            cls._instance = instance
+        return cls._instance
+
+
+class Metrics(metaclass=Singleton):
+    stats_client: StatsClient
+
+    def __init__(
+        self,
+    ) -> None:
+        self.is_ready = False
+        self.lock = threading.Lock()
+        self.request_size_total = 0
+        self.request_count = 0
+
+    def setup(self, config: Config) -> None:
+        with self.lock:
+            if self.is_ready:
+                return
+
+            if not config.get("metrics_extended"):
+                return
+            stats_service = config.get("stats_service")
+            if stats_service == "statsd":
+                self.stats_client = StatsdClient(config=config)
+            elif stats_service == "prometheus":
+                self.stats_client = PrometheusClient(config=config)
+            else:
+                raise MetricsException('Config variable "stats_service" is not defined')
+            self.is_ready = True
+
+    def request(self, size: int) -> None:
+        if not self.is_ready or self.stats_client is None:
+            return
+        if not isinstance(self.stats_client, StatsClient):
+            raise RuntimeError("no StatsClient available")
+        self.stats_client.increase("request-size-total", size)
+        self.stats_client.increase("request-count", 1)
+
+    def response(self, size: int) -> None:
+        if not self.is_ready or self.stats_client is None:
+            return
+        if not isinstance(self.stats_client, StatsClient):
+            raise RuntimeError("no StatsClient available")
+        self.stats_client.increase("response-size-total", size)
+        self.stats_client.increase("response-count", 1)
+
+    def are_we_master(self, is_master: bool) -> None:
+        if not self.is_ready or self.stats_client is None:
+            return
+        if not isinstance(self.stats_client, StatsClient):
+            raise RuntimeError("no StatsClient available")
+        self.stats_client.gauge("master-slave-role", int(is_master))
+
+    def latency(self, latency_ms: float) -> None:
+        if not self.is_ready or self.stats_client is None:
+            return
+        if not isinstance(self.stats_client, StatsClient):
+            raise RuntimeError("no StatsClient available")
+        self.stats_client.timing("latency_ms", latency_ms)
+
+    def error(self) -> None:
+        if not self.is_ready or self.stats_client is None:
+            return
+        if not isinstance(self.stats_client, StatsClient):
+            raise RuntimeError("no StatsClient available")
+        self.stats_client.increase("error-total", 1)
+
+    def cleanup(self) -> None:
+        if self.stats_client:
+            self.stats_client.close()
+        if not self.is_ready:
+            return

karapace/prometheus.py

@@ -0,0 +1,124 @@
+"""
+karapace - prometheus
+
+Supports telegraf's statsd protocol extension for 'key=value' tags:
+
+  https://github.com/influxdata/telegraf/tree/master/plugins/inputs/statsd
+
+Copyright (c) 2023 Aiven Ltd
+See LICENSE for details
+"""
+from __future__ import annotations
+
+from karapace.base_stats import StatsClient
+from karapace.config import Config
+from prometheus_client import Counter, Gauge, REGISTRY, Summary
+from prometheus_client.exposition import make_wsgi_app
+from socketserver import ThreadingMixIn
+from typing import Final
+from wsgiref.simple_server import make_server, WSGIRequestHandler, WSGIServer
+
+import logging
+import socket
+import threading
+
+LOG = logging.getLogger(__name__)
+HOST: Final = "127.0.0.1"
+PORT: Final = 8005
+
+
+class PrometheusException(Exception):
+    pass
+
+
+class ThreadingWSGIServer(ThreadingMixIn, WSGIServer):
+    """Thread per request HTTP server."""
+
+    # Make worker threads "fire and forget". Beginning with Python 3.7 this
+    # prevents a memory leak because ``ThreadingMixIn`` starts to gather all
+    # non-daemon threads in a list in order to join on them at server close.
+    daemon_threads = True
+
+
+class _SilentHandler(WSGIRequestHandler):
+    """WSGI handler that does not log requests."""
+
+    # pylint: disable=W0622
+    def log_message(self, format, *args):
+        """Log nothing."""
+
+
+def get_family(address, port):
+    infos = socket.getaddrinfo(address, port)
+    family, _, _, _, sockaddr = next(iter(infos))
+    return family, sockaddr[0]
+
+
+class PrometheusClient(StatsClient):
+    server_is_active: bool = False
+
+    def __init__(self, config: Config) -> None:
+        super().__init__(config)
+        self.lock = threading.Lock()
+        self.httpd = None
+        self.thread = None
+        with self.lock:
+            _host = config.get("prometheus_host", None)
+            _port = config.get("prometheus_port", None)
+            if _host is None:
+                raise PrometheusException("prometheus_host host is undefined")
+            if _port is None:
+                raise PrometheusException("prometheus_host port is undefined")
+            if not PrometheusClient.server_is_active:
+                # We wrapped httpd server creation from prometheus client to allow stop this server"""
+                self.start_server(_host, _port)
+
+                PrometheusClient.server_is_active = True
+            else:
+                raise PrometheusException("Double instance of Prometheus interface")
+        self._gauge: dict[str, Gauge] = dict()
+        self._summary: dict[str, Summary] = dict()
+        self._counter: dict[str, Counter] = dict()
+
+    def gauge(self, metric: str, value: float, tags: dict | None = None) -> None:
+        m = self._gauge.get(metric)
+        if m is None:
+            m = Gauge(metric, metric)
+            self._gauge[metric] = m
+        m.set(value)
+
+    def increase(self, metric: str, inc_value: int = 1, tags: dict | None = None) -> None:
+        m = self._counter.get(metric)
+        if m is None:
+            m = Counter(metric, metric)
+            self._counter[metric] = m
+        m.inc(inc_value)
+
+    def timing(self, metric: str, value: float, tags: dict | None = None) -> None:
+        m = self._summary.get(metric)
+        if m is None:
+            m = Summary(metric, metric)
+            self._summary[metric] = m
+        m.observe(value)
+
+    def start_server(self, addr: str, port: int) -> None:
+        class TmpServer(ThreadingWSGIServer):
+            pass
+
+        TmpServer.address_family, addr = get_family(addr, port)
+        app = make_wsgi_app(REGISTRY)
+        self.httpd = make_server(addr, port, app, TmpServer, handler_class=_SilentHandler)
+        self.thread = threading.Thread(target=self.httpd.serve_forever)
+        self.thread.daemon = True
+        self.thread.start()
+
+    def stop_server(self) -> None:
+        self.httpd.shutdown()
+        self.httpd.server_close()
+        self.thread.join()
+
+    def close(self):
+        with self.lock:
+            if self.server_is_active:
+                self.stop_server()
+                PrometheusClient.server_is_active = False

karapace/rapu.py

@@ -9,7 +9,7 @@
 from accept_types import get_best_match
 from http import HTTPStatus
 from karapace.config import Config, create_server_ssl_context
-from karapace.statsd import StatsClient
+from karapace.metrics import Metrics
 from karapace.utils import json_decode, json_encode
 from karapace.version import __version__
 from typing import Callable, Dict, NoReturn, Optional, overload, Union
@@ -134,6 +134,8 @@ def __init__(
         if content_type:
             self.headers["Content-Type"] = content_type
         super().__init__(f"HTTPResponse {status.value}")
+        if not is_success(status):
+            Metrics().error()
 
     def ok(self) -> bool:
         """True if resposne has a 2xx status_code"""
@@ -166,7 +168,8 @@ def __init__(
         self.app_request_metric = f"{app_name}_request"
         self.app = self._create_aiohttp_application(config=config)
         self.log = logging.getLogger(self.app_name)
-        self.stats = StatsClient(config=config)
+        Metrics().setup(config)
+        self.stats = Metrics().stats_client
         self.app.on_shutdown.append(self.close_by_app)
         self.not_ready_handler = not_ready_handler
 
@@ -183,7 +186,7 @@ async def close(self) -> None:
         set as hook because the awaitables have to run inside the event loop
         created by the aiohttp library.
         """
-        self.stats.close()
+        Metrics().cleanup()
 
     @staticmethod
     def cors_and_server_headers_for_request(*, request, origin="*"):  # pylint: disable=unused-argument
@@ -269,15 +272,21 @@ async def _handle_request(
             url=request.url,
             path_for_stats=path_for_stats,
         )
+
         try:
             if request.method == "OPTIONS":
                 origin = request.headers.get("Origin")
                 if not origin:
                     raise HTTPResponse(body="OPTIONS missing Origin", status=HTTPStatus.BAD_REQUEST)
                 headers = self.cors_and_server_headers_for_request(request=rapu_request, origin=origin)
+
                 raise HTTPResponse(body=b"", status=HTTPStatus.OK, headers=headers)
 
             body = await request.read()
+            if body:
+                Metrics().request(len(body))
+            else:
+                Metrics().request(0)
             if json_request:
                 if not body:
                     raise HTTPResponse(body="Missing request JSON body", status=HTTPStatus.BAD_REQUEST)
@@ -406,6 +415,8 @@ async def _handle_request(
             self.log.exception("Unexpected error handling user request: %s %s", request.method, request.url)
             resp = aiohttp.web.Response(text="Internal Server Error", status=HTTPStatus.INTERNAL_SERVER_ERROR.value)
         finally:
+            Metrics().response(resp.content_length)
+            Metrics().latency((time.monotonic() - start_time) * 1000)
             self.stats.timing(
                 self.app_request_metric,
                 time.monotonic() - start_time,