From e3ce58546fc10d0235419b25bf12b9e6d995df04 Mon Sep 17 00:00:00 2001 From: Neil Sleightholm Date: Sun, 8 Nov 2026 09:22:49 +0000 Subject: [PATCH] Parse metrics in a worker thread to avoid blocking the event loop text_string_to_metric_families() and building the monitor models are CPU-bound and scale with the number of monitors. Running them on the event loop can stall the host application (e.g. Home Assistant) for seconds at a time. Offload them with asyncio.to_thread. Co-Authored-By: Claude Opus 5.5 --- pythonkuma/uptimekuma.py | 104 +++++++++++++++++++++++++-------------- tests/test_metrics.py | 27 +++++++++- 2 files changed, 94 insertions(+), 37 deletions(-) diff --git a/pythonkuma/uptimekuma.py b/pythonkuma/uptimekuma.py index 7d2d8ba..455be58 100644 --- a/pythonkuma/uptimekuma.py +++ b/pythonkuma/uptimekuma.py @@ -1,5 +1,6 @@ """Uptime Kuma client.""" +import asyncio from http import HTTPStatus from typing import Any @@ -86,7 +87,6 @@ async def metrics(self) -> dict[str | int, UptimeKumaMonitor]: If there is a connection error, timeout, or other client error during the request. """ - monitors: dict[str | int, dict[str, Any]] = {} url = self._base_url / "metrics" try: @@ -106,38 +106,70 @@ async def metrics(self) -> dict[str | int, UptimeKumaMonitor]: except ClientError as e: raise UptimeKumaConnectionException from e - try: - metrics = list(text_string_to_metric_families(await request.text())) - except ValueError as e: - raise UptimeKumaParseException from e - - for metric in metrics: - if metric.name == "app_version" and metric.samples: - self.version = UptimeKumaVersion.from_dict(metric.samples[0].labels) - if not metric.name.startswith("monitor"): - continue - for sample in metric.samples: - key = ( - int(monitor_id) - if (monitor_id := sample.labels.get("monitor_id")) - else sample.labels["monitor_name"] - ) - - tags = [ - f"{k}:{v}" if v else k - for k, v in sample.labels.items() - if k not in COMMON_LABELS - ] - name = ( - f"{sample.name}_{window}" - if (window := sample.labels.get("window")) - else sample.name - ) - - monitors.setdefault( - key, {**sample.labels, "monitor_tags": tags} - ).update({name: sample.value}) - - return { - key: UptimeKumaMonitor.from_dict(value) for key, value in monitors.items() - } + text = await request.text() + + # Parsing the Prometheus exposition format is CPU-bound and scales with + # the number of monitors, so run it in a worker thread to avoid blocking + # the event loop. + version, result = await asyncio.to_thread(_parse_metrics, text) + if version is not None: + self.version = version + + return result + + +def _parse_metrics( + text: str, +) -> tuple[UptimeKumaVersion | None, dict[str | int, UptimeKumaMonitor]]: + """Parse Prometheus metrics text into monitors. + + Runs synchronously; call it from a worker thread. + + Returns + ------- + tuple[UptimeKumaVersion | None, dict[str | int, UptimeKumaMonitor]] + The app version (if reported) and the monitors keyed by id or name. + + Raises + ------ + UptimeKumaParseException + If the metrics text cannot be parsed. + """ + monitors: dict[str | int, dict[str, Any]] = {} + version: UptimeKumaVersion | None = None + + try: + metrics = list(text_string_to_metric_families(text)) + except ValueError as e: + raise UptimeKumaParseException from e + + for metric in metrics: + if metric.name == "app_version" and metric.samples: + version = UptimeKumaVersion.from_dict(metric.samples[0].labels) + if not metric.name.startswith("monitor"): + continue + for sample in metric.samples: + key = ( + int(monitor_id) + if (monitor_id := sample.labels.get("monitor_id")) + else sample.labels["monitor_name"] + ) + + tags = [ + f"{k}:{v}" if v else k + for k, v in sample.labels.items() + if k not in COMMON_LABELS + ] + name = ( + f"{sample.name}_{window}" + if (window := sample.labels.get("window")) + else sample.name + ) + + monitors.setdefault(key, {**sample.labels, "monitor_tags": tags}).update( + {name: sample.value} + ) + + return version, { + key: UptimeKumaMonitor.from_dict(value) for key, value in monitors.items() + } diff --git a/tests/test_metrics.py b/tests/test_metrics.py index 83212bc..f299fe3 100644 --- a/tests/test_metrics.py +++ b/tests/test_metrics.py @@ -1,10 +1,14 @@ """Tests for pythonkuma.""" +from collections.abc import Iterable from http import HTTPStatus +import threading from typing import Any -from unittest.mock import AsyncMock, Mock +from unittest.mock import AsyncMock, Mock, patch from aiohttp import ClientError, ClientResponseError, ConnectionTimeoutError +from prometheus_client.metrics_core import Metric +from prometheus_client.parser import text_string_to_metric_families import pytest from syrupy.assertion import SnapshotAssertion @@ -82,3 +86,24 @@ async def test_metrics_parse_exceptions(mock_session: AsyncMock) -> None: with pytest.raises(UptimeKumaParseException): await uptime_kuma.metrics() + + +async def test_metrics_parsed_off_event_loop(mock_session: AsyncMock) -> None: + """Test prometheus metrics are parsed in a worker thread, not the event loop.""" + loop_thread = threading.get_ident() + parse_threads: list[int] = [] + + def recording_parser(text: str) -> Iterable[Metric]: + parse_threads.append(threading.get_ident()) + return text_string_to_metric_families(text) + + uptime_kuma = UptimeKuma(mock_session, "http://uptime.example.com", "test-apikey") + + with patch( + "pythonkuma.uptimekuma.text_string_to_metric_families", recording_parser + ): + monitors = await uptime_kuma.metrics() + + assert monitors + assert parse_threads + assert loop_thread not in parse_threads