diff --git a/CHANGELOG.md b/CHANGELOG.md index 5e8a1482c..2a67fb9e4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,8 @@ and this project adheres to - 🐛(frontend) reduce PostHog volume from web vitals and opt_in spam #2701 - ✨(backend) expose the attachment max size in the config endpoint #2577 - ✨(frontend) warn before uploading an attachment over the size limit #2577 +- ✨(backend) measure the calls to yhub and to the converters, the database + pool and the celery queue - ✨(backend) add a `LoadTest` configuration and its `loadtest` application, minting sessions for load tests - ✨(collaboration) add opt-in prometheus metrics to yhub, server and worker, diff --git a/documentation/env.md b/documentation/env.md index bc87a0d2d..85befe7d6 100644 --- a/documentation/env.md +++ b/documentation/env.md @@ -134,6 +134,7 @@ These are the environment variables you can set for the `impress-backend` contai | OIDC_USE_NONCE | Use nonce for OIDC | true | | POSTHOG_KEY | Posthog key for analytics | | | PROMETHEUS_API_KEY | Bearer token required on `/metrics`. Required when the metrics are enabled (the application refuses to start without it). Can be given as a file with PROMETHEUS_API_KEY_FILE | | +| PROMETHEUS_CELERY_QUEUE_METRICS_ENABLED | When the metrics are enabled, report the length of the default Celery queue, asked to the broker at every scrape | True | | PROMETHEUS_DB_METRICS_ENABLED | When the metrics are enabled, also count and time the SQL queries (swaps the postgresql engine for the instrumented one of django-prometheus) | True | | PROMETHEUS_METRICS_ENABLED | Instrument the application with django-prometheus and serve `/metrics`. OFF by default. See documentation/metrics.md | False | | PROMETHEUS_MULTIPROC_DIR | Directory the uvicorn workers share their metrics through. Local to the host or pod, owned by the user running the application | `/impress-prometheus-` | diff --git a/documentation/metrics.md b/documentation/metrics.md index 6ba516782..4fc2437d6 100644 --- a/documentation/metrics.md +++ b/documentation/metrics.md @@ -49,6 +49,34 @@ scrape_configs: With `DB_PSYCOPG_POOL_ENABLED`, `django_db_new_connections_total` counts the connections taken from the pool, not the connections opened to Postgres. +- Calls to the other services, in + `docs_outgoing_request_duration_seconds{service,operation,method,status}` and + `docs_outgoing_requests_inflight{service,operation,method}`. `service` is + `yhub`, `y-provider` or `docspec`; `operation` is the endpoint (`ydoc`, + `create-ydoc`, `reset-connections`, `convert`, ...), never the url; `status` + is the http status, `timeout` when the call was given up on, `error` when it + never got an answer. These calls are made inside requests (`duplicate`, + `formatted-content`, document creation from a file) and from the Celery + tasks: the in-flight gauge is what piles up when the collaboration server + slows down. +- The psycopg pool, when `DB_PSYCOPG_POOL_ENABLED` is on: + `docs_db_pool_size`, `docs_db_pool_available` and + `docs_db_pool_requests_waiting` (what the pool of each worker looked like at + the end of its last request, added up), and the exact counters of the pool: + `docs_db_pool_requests_total`, `docs_db_pool_requests_queued_total`, + `docs_db_pool_requests_wait_seconds_total`, + `docs_db_pool_requests_errors_total`, `docs_db_pool_connections_total`, + `docs_db_pool_connections_errors_total`. A rising + `rate(docs_db_pool_requests_queued_total)` with + `rate(docs_db_pool_requests_wait_seconds_total)` is the application waiting + for connections, before Postgres shows anything. +- `docs_celery_queue_length{queue}`: tasks waiting on the default Celery queue, + asked to the broker when the metrics are scraped (Redis/Valkey brokers only; + turn it off with `PROMETHEUS_CELERY_QUEUE_METRICS_ENABLED=False`). It is one + queue for the whole deployment, so every replica reports the same number: + read it with `max`, never `sum`. The Celery workers have no endpoint of + their own, which is why the backend reports it. + No label ever carries a path, a user or a document identifier. A request that matches no route is counted under ``. @@ -82,6 +110,11 @@ application refuses a directory owned by another user or a symbolic link. Set `PROMETHEUS_MULTIPROC_DIR` to put it elsewhere — it must be writable, and local to the host or the pod: it is **not** shared between replicas. +The gauges (in-flight calls, pool state) are kept per worker process and added +up over the living ones. uvicorn has no hook telling when a worker is gone, so +whichever worker answers a scrape first drops the gauge files of the processes +that no longer exist. + Known limit: uvicorn recycles its workers (`--limit-max-requests`) and has no hook to tell when one is gone. Every new worker writes two new 64 KiB files, and the files of the workers that are gone must stay — their counters are part diff --git a/src/backend/core/apps.py b/src/backend/core/apps.py index b8ce64883..611945c08 100644 --- a/src/backend/core/apps.py +++ b/src/backend/core/apps.py @@ -13,7 +13,10 @@ class CoreConfig(AppConfig): def ready(self): """ - Import signals when the app is ready. + Import signals when the app is ready, and wire the measurements driven + by Django's own signals (a no-op unless the metrics are enabled). """ # pylint: disable=import-outside-toplevel, unused-import - from . import signals # noqa: PLC0415 + from . import instrumentation, signals # noqa: PLC0415 + + instrumentation.connect_signals() diff --git a/src/backend/core/instrumentation.py b/src/backend/core/instrumentation.py new file mode 100644 index 000000000..45c3c548d --- /dev/null +++ b/src/backend/core/instrumentation.py @@ -0,0 +1,78 @@ +"""Hooks the application calls to be measured, cheap when nothing is measured. + +The Prometheus metrics are opt-in (PROMETHEUS_METRICS_ENABLED). This module is +what the rest of the code imports: it has no dependency on prometheus_client, and +only loads `core.metrics` — where the metrics are defined — once they are enabled. +With the metrics off, every hook here is a test of a setting and nothing else. +""" + +import time +from contextlib import contextmanager + +from django.conf import settings + +import requests + + +class ObservedRequest: + """What the caller tells about the outgoing request it is making.""" + + def __init__(self): + self.status = None + + +@contextmanager +def outgoing_request(service, operation, method="POST"): + """ + Time an HTTP call to another service. + + with outgoing_request("yhub", "ydoc", "GET") as observed: + response = requests.get(...) + observed.status = response.status_code + + The status label is the HTTP status the caller reported, `timeout` when the + call was given up on, `error` when it never got an answer. Neither the url + nor a document id is ever a label: `operation` names the endpoint. + """ + observed = ObservedRequest() + if not settings.PROMETHEUS_METRICS_ENABLED: + yield observed + return + + # pylint: disable-next=import-outside-toplevel + from core import metrics # noqa: PLC0415 + + labels = {"service": service, "operation": operation, "method": method.upper()} + metrics.OUTGOING_REQUESTS_INFLIGHT.labels(**labels).inc() + start = time.perf_counter() + status = "error" + try: + yield observed + status = str(observed.status) if observed.status is not None else "unknown" + except requests.Timeout: + status = "timeout" + raise + finally: + metrics.OUTGOING_REQUESTS_INFLIGHT.labels(**labels).dec() + metrics.OUTGOING_REQUEST_DURATION.labels(**labels, status=status).observe( + time.perf_counter() - start + ) + + +def connect_signals(): + """ + Wire the measurements that are driven by Django's signals. Called once the + applications are loaded, and a no-op unless the metrics are enabled. + """ + if not settings.PROMETHEUS_METRICS_ENABLED: + return + + # pylint: disable-next=import-outside-toplevel + from django.core.signals import request_finished # noqa: PLC0415 + + # pylint: disable-next=import-outside-toplevel + from core import metrics # noqa: PLC0415 + + request_finished.connect( + metrics.record_database_pool, dispatch_uid="core.metrics.database_pool" + ) diff --git a/src/backend/core/metrics.py b/src/backend/core/metrics.py index cdf5718d9..fb51fb321 100644 --- a/src/backend/core/metrics.py +++ b/src/backend/core/metrics.py @@ -2,27 +2,246 @@ Enabled by PROMETHEUS_METRICS_ENABLED and protected by `core.middleware.PrometheusAuthMiddleware` — see documentation/metrics.md. + +Only imported when the metrics are enabled: the rest of the code goes through +`core.instrumentation`, which does not depend on prometheus_client. """ +import logging +import os +import re import socket +from django.conf import settings +from django.db import connections from django.http import HttpResponse from django.views.decorators.http import require_safe -from prometheus_client import CONTENT_TYPE_LATEST, CollectorRegistry, generate_latest -from prometheus_client.core import Metric +from prometheus_client import ( + CONTENT_TYPE_LATEST, + CollectorRegistry, + Counter, + Gauge, + Histogram, + generate_latest, + multiprocess, +) +from prometheus_client.core import GaugeMetricFamily, Metric from prometheus_client.multiprocess import MultiProcessCollector +logger = logging.getLogger(__name__) + HOSTNAME_LABEL = "hostname" +# A call to another service answers in milliseconds when all is well and in tens +# of seconds when that service is drowning: both ends have to be readable. +LATENCY_BUCKETS = (0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60) -class HostnameLabelCollector: +OUTGOING_REQUEST_DURATION = Histogram( + "docs_outgoing_request_duration_seconds", + "Duration of the HTTP calls made to the other services (yhub, converters)", + ["service", "operation", "method", "status"], + buckets=LATENCY_BUCKETS, +) +OUTGOING_REQUESTS_INFLIGHT = Gauge( + "docs_outgoing_requests_inflight", + "HTTP calls to the other services that have not been answered yet", + ["service", "operation", "method"], + # one value per worker process: add up those that are still alive + multiprocess_mode="livesum", +) + +# The psycopg pool of each worker process (DB_PSYCOPG_POOL_ENABLED). The gauges +# are what the pool looked like at the end of the last request of each worker; +# the counters are exact, the pool counts them itself. +DATABASE_POOL_SIZE = Gauge( + "docs_db_pool_size", + "Connections the pools hold, idle or not", + ["alias"], + multiprocess_mode="livesum", +) +DATABASE_POOL_AVAILABLE = Gauge( + "docs_db_pool_available", + "Idle connections in the pools", + ["alias"], + multiprocess_mode="livesum", +) +DATABASE_POOL_REQUESTS_WAITING = Gauge( + "docs_db_pool_requests_waiting", + "Requests queued for a connection, all of them being taken", + ["alias"], + multiprocess_mode="livesum", +) +DATABASE_POOL_REQUESTS = Counter( + "docs_db_pool_requests_total", "Connections asked to the pools", ["alias"] +) +DATABASE_POOL_REQUESTS_QUEUED = Counter( + "docs_db_pool_requests_queued_total", + "Connections asked to the pools that had to wait for one", + ["alias"], +) +DATABASE_POOL_REQUESTS_WAIT_SECONDS = Counter( + "docs_db_pool_requests_wait_seconds_total", + "Time spent waiting for a connection of the pools", + ["alias"], +) +DATABASE_POOL_REQUESTS_ERRORS = Counter( + "docs_db_pool_requests_errors_total", + "Connections asked to the pools that were not given (timeout, queue full)", + ["alias"], +) +DATABASE_POOL_CONNECTIONS = Counter( + "docs_db_pool_connections_total", + "Connections the pools opened to the database", + ["alias"], +) +DATABASE_POOL_CONNECTIONS_ERRORS = Counter( + "docs_db_pool_connections_errors_total", + "Connections the pools failed to open", + ["alias"], +) + +LIVE_GAUGE_FILE = re.compile(r"^gauge_live[a-z]*_(?P\d+)\.db$") + + +def record_database_pool(**_kwargs): """ - The metrics of every worker process of this host, labelled with its name. + Record the state of the psycopg pools of this process. Connected to + `request_finished`, so that it costs a request nothing while it runs. + + `pop_stats` hands the counters over and resets them, which is what lets them + be added to ours whatever the number of worker processes. + """ + for alias in connections: + connection = connections[alias] + if not connection.settings_dict.get("OPTIONS", {}).get("pool"): + continue + try: + stats = connection.pool.pop_stats() + except Exception: # pylint: disable=broad-exception-caught + # a measurement never fails a request + logger.debug("could not read the database pool stats", exc_info=True) + continue + DATABASE_POOL_SIZE.labels(alias).set(stats.get("pool_size", 0)) + DATABASE_POOL_AVAILABLE.labels(alias).set(stats.get("pool_available", 0)) + DATABASE_POOL_REQUESTS_WAITING.labels(alias).set( + stats.get("requests_waiting", 0) + ) + DATABASE_POOL_REQUESTS.labels(alias).inc(stats.get("requests_num", 0)) + DATABASE_POOL_REQUESTS_QUEUED.labels(alias).inc(stats.get("requests_queued", 0)) + DATABASE_POOL_REQUESTS_WAIT_SECONDS.labels(alias).inc( + stats.get("requests_wait_ms", 0) / 1000 + ) + DATABASE_POOL_REQUESTS_ERRORS.labels(alias).inc(stats.get("requests_errors", 0)) + DATABASE_POOL_CONNECTIONS.labels(alias).inc(stats.get("connections_num", 0)) + DATABASE_POOL_CONNECTIONS_ERRORS.labels(alias).inc( + stats.get("connections_errors", 0) + ) + + +def forget_dead_processes(): + """ + Drop the gauges of the worker processes that are gone. + + A gauge that is summed over the living processes is kept in one file per + process, and prometheus_client only removes it when told the process died. + gunicorn has a hook for that, uvicorn has none: whoever answers a scrape + checks instead. The directory is local to this host, so are the pids. + Counters and histograms are left alone: what a dead worker counted stays part + of the totals. + """ + directory = os.environ.get("PROMETHEUS_MULTIPROC_DIR") + if not directory: + return + try: + names = os.listdir(directory) + except OSError: + return + pids = {int(m["pid"]) for m in map(LIVE_GAUGE_FILE.match, names) if m} + for pid in pids: + try: + os.kill(pid, 0) + except ProcessLookupError: + multiprocess.mark_process_dead(pid) + except OSError: + # it exists and is not ours to signal: alive is all that matters + continue + + +class CeleryQueueCollector: + """ + The number of tasks waiting on the default Celery queue, asked to the broker + when the metrics are scraped. + + The workers have no endpoint of their own, and the queue is what tells that + they are not keeping up: the delete, restore and access cascades all land on + it. One queue for the whole deployment: every replica reports the same + number, which is to be read with max(), never summed. + """ + + def collect(self): + """Yield the gauge, or nothing when the broker cannot be asked.""" + length = self.queue_length() + if length is None: + return + gauge = GaugeMetricFamily( + "docs_celery_queue_length", + "Tasks waiting on the default Celery queue, for the whole deployment " + "(use max, not sum)", + labels=["queue"], + ) + gauge.add_metric([self.queue_name()], length) + yield gauge + + @staticmethod + def queue_name(): + """The queue the tasks are sent to: none of them names another one.""" + # pylint: disable-next=import-outside-toplevel + from impress.celery_app import app # noqa: PLC0415 + + return app.conf.task_default_queue + + def queue_length(self): + """Ask the broker, quickly: a scrape must not hang on it.""" + # pylint: disable-next=import-outside-toplevel + from impress.celery_app import app # noqa: PLC0415 + + try: + with app.connection_for_read() as connection: + if connection.transport.driver_type != "redis": + return None + connection.ensure_connection(max_retries=1, timeout=2) + channel = connection.default_channel + queue = self.queue_name() + # kombu keeps one redis list per priority step + keys = [ + f"{queue}{channel.sep}{priority}" if priority else queue + for priority in channel.priority_steps + ] + return sum(channel.client.llen(key) for key in keys) + except Exception: # pylint: disable=broad-exception-caught + logger.debug("could not read the celery queue length", exc_info=True) + return None + + +class WorkerProcessesCollector: + """ + The metrics of every worker process of this host. uvicorn runs several workers and the scrape is answered by one of them: the multiprocess collector reads what they all wrote to PROMETHEUS_MULTIPROC_DIR and adds it up, instead of reporting the numbers of that worker alone. + """ + + def collect(self): + """Yield the aggregated metrics.""" + # a registry of its own: the collector registers itself in the one it is given + yield from MultiProcessCollector(CollectorRegistry()).collect() + + +class HostnameLabelled: + """ + The metrics of another collector, labelled with the name of this host. The label is for a scraper that goes through a load balancer (an ingress in front of several replicas): each scrape is then answered by a different @@ -31,13 +250,13 @@ class HostnameLabelCollector: each replica has its own series, sampled whenever it happens to answer. """ - def __init__(self): + def __init__(self, collector): + self.collector = collector self.hostname = socket.gethostname() def collect(self): - """Yield the aggregated metrics with the hostname added to every sample.""" - # a registry of its own: the collector registers itself in the one it is given - for metric in MultiProcessCollector(CollectorRegistry()).collect(): + """Yield the metrics with the hostname added to every sample.""" + for metric in self.collector.collect(): labelled = Metric(metric.name, metric.documentation, metric.type) for sample in metric.samples: labelled.add_sample( @@ -53,6 +272,9 @@ class HostnameLabelCollector: @require_safe def metrics_view(request): """Serve the metrics in the Prometheus text format.""" + forget_dead_processes() registry = CollectorRegistry() - registry.register(HostnameLabelCollector()) + registry.register(HostnameLabelled(WorkerProcessesCollector())) + if settings.PROMETHEUS_CELERY_QUEUE_METRICS_ENABLED: + registry.register(HostnameLabelled(CeleryQueueCollector())) return HttpResponse(generate_latest(registry), content_type=CONTENT_TYPE_LATEST) diff --git a/src/backend/core/services/converter_services.py b/src/backend/core/services/converter_services.py index 7a550d18c..26ca3bd57 100644 --- a/src/backend/core/services/converter_services.py +++ b/src/backend/core/services/converter_services.py @@ -7,6 +7,7 @@ from django.conf import settings import requests +from core.instrumentation import outgoing_request from core.services import mime_types from core.services.jwt_services import Audiences, JWTService @@ -64,16 +65,18 @@ class DocSpecConverter: def _request(self, url, data, content_type): """Make a request to the DocSpec API.""" - response = requests.post( - url, - headers={ - "Content-Type": content_type, - "Accept": mime_types.BLOCKNOTE, - }, - data=data, - timeout=settings.CONVERSION_API_TIMEOUT, - verify=settings.CONVERSION_API_SECURE, - ) + with outgoing_request("docspec", "convert") as observed: + response = requests.post( + url, + headers={ + "Content-Type": content_type, + "Accept": mime_types.BLOCKNOTE, + }, + data=data, + timeout=settings.CONVERSION_API_TIMEOUT, + verify=settings.CONVERSION_API_SECURE, + ) + observed.status = response.status_code if not response.ok: logger.error( "DocSpec API error: url=%s, status=%d, response=%s", @@ -114,17 +117,19 @@ class YdocConverter: def _request(self, url, data, content_type, accept): """Make a request to the Y-Provider API.""" - response = requests.post( - url, - data=data, - headers={ - "Authorization": self.auth_header, - "Content-Type": content_type, - "Accept": accept, - }, - timeout=settings.CONVERSION_API_TIMEOUT, - verify=settings.CONVERSION_API_SECURE, - ) + with outgoing_request("y-provider", "convert") as observed: + response = requests.post( + url, + data=data, + headers={ + "Authorization": self.auth_header, + "Content-Type": content_type, + "Accept": accept, + }, + timeout=settings.CONVERSION_API_TIMEOUT, + verify=settings.CONVERSION_API_SECURE, + ) + observed.status = response.status_code if not response.ok: logger.error( "Y-Provider API error: url=%s, status=%d, response=%s", diff --git a/src/backend/core/services/yhub_services.py b/src/backend/core/services/yhub_services.py index 327fa55f6..944cd59e0 100644 --- a/src/backend/core/services/yhub_services.py +++ b/src/backend/core/services/yhub_services.py @@ -35,6 +35,7 @@ from django.conf import settings import requests +from core.instrumentation import outgoing_request from core.services.jwt_services import Audiences, JWKSClient, JWTService logger = logging.getLogger(__name__) @@ -193,19 +194,23 @@ class YHubService: endpoints do not all answer with the same payload. An endpoint doing more than answering a document passes its own timeout. """ + # the endpoint is what the call is measured under: `{base}/{prefix}/{endpoint}/…` + endpoint = url.removeprefix(f"{self.base_url}/{self.api_prefix}/").split("/")[0] try: - response = requests.request( - method, - url, - data=data, - headers={ - "Authorization": self.auth_header, - # what makes yhub answer JSON rather than its lib0 encoding - "Accept": "application/json", - **(headers or {}), - }, - timeout=timeout or self.timeout, - ) + with outgoing_request("yhub", endpoint, method) as observed: + response = requests.request( + method, + url, + data=data, + headers={ + "Authorization": self.auth_header, + # what makes yhub answer JSON rather than its lib0 encoding + "Accept": "application/json", + **(headers or {}), + }, + timeout=timeout or self.timeout, + ) + observed.status = response.status_code except requests.RequestException as err: logger.exception("yhub service error: url=%s", url) raise ServiceUnavailableError( diff --git a/src/backend/core/tests/test_instrumentation.py b/src/backend/core/tests/test_instrumentation.py new file mode 100644 index 000000000..1ae40ead1 --- /dev/null +++ b/src/backend/core/tests/test_instrumentation.py @@ -0,0 +1,271 @@ +""" +Test the measurements the application takes of itself: outgoing calls, database +pool and Celery queue. +""" + +import os +import subprocess +import sys +from unittest import mock + +from django.core.signals import request_finished + +import pytest +import requests +from prometheus_client import REGISTRY, CollectorRegistry, generate_latest + +from core import factories, instrumentation, metrics +from core.services.yhub_services import APIError, YHubService + +pytestmark = pytest.mark.django_db + +DURATION = "docs_outgoing_request_duration_seconds_count" +INFLIGHT = "docs_outgoing_requests_inflight" + + +def sample(name, **labels): + """The value of a sample of the default registry, 0 when it does not exist.""" + return REGISTRY.get_sample_value(name, labels) or 0 + + +@pytest.fixture(name="metrics_enabled") +def metrics_enabled_fixture(settings): + """Turn the measurements on, as PROMETHEUS_METRICS_ENABLED does.""" + settings.PROMETHEUS_METRICS_ENABLED = True + return settings + + +def test_instrumentation_outgoing_request_disabled(settings): + """With the metrics off a call is not measured, and goes through untouched.""" + settings.PROMETHEUS_METRICS_ENABLED = False + labels = {"service": "test-off", "operation": "op", "method": "GET"} + + with instrumentation.outgoing_request("test-off", "op", "get") as observed: + observed.status = 200 + + assert sample(DURATION, **labels, status="200") == 0 + + +@pytest.mark.usefixtures("metrics_enabled") +def test_instrumentation_outgoing_request_counts_the_status(): + """A call is counted under the status its caller reported, and is in flight meanwhile.""" + labels = {"service": "test-status", "operation": "op", "method": "GET"} + + with instrumentation.outgoing_request("test-status", "op", "get") as observed: + assert sample(INFLIGHT, **labels) == 1 + observed.status = 503 + + assert sample(INFLIGHT, **labels) == 0 + assert sample(DURATION, **labels, status="503") == 1 + + +@pytest.mark.usefixtures("metrics_enabled") +@pytest.mark.parametrize( + "error,status", + [ + (requests.ReadTimeout("too slow"), "timeout"), + (requests.ConnectTimeout("too slow"), "timeout"), + (requests.ConnectionError("refused"), "error"), + (ValueError("anything else"), "error"), + ], +) +def test_instrumentation_outgoing_request_failures(error, status): + """A call given up on is told apart from one that failed, and both are re-raised.""" + labels = {"service": f"test-{status}", "operation": "op", "method": "POST"} + before = sample(DURATION, **labels, status=status) + + with pytest.raises(type(error)): + with instrumentation.outgoing_request(f"test-{status}", "op"): + raise error + + assert sample(DURATION, **labels, status=status) == before + 1 + assert sample(INFLIGHT, **labels) == 0 + + +@pytest.mark.usefixtures("metrics_enabled") +def test_instrumentation_yhub_client_is_measured_by_endpoint(): + """The yhub client is measured under the endpoint it calls, never under the url.""" + document = factories.DocumentFactory() + labels = {"service": "yhub", "operation": "ydoc", "method": "GET"} + ok_before = sample(DURATION, **labels, status="200") + missing_before = sample(DURATION, **labels, status="404") + + response = mock.Mock(ok=True, status_code=200, content=b"\\x00\\x00") + with mock.patch("core.services.yhub_services.requests.request") as mock_request: + mock_request.return_value = response + YHubService().request("get", YHubService().build_url("ydoc", document)) + mock_request.return_value = mock.Mock( + ok=False, status_code=404, text="", json=mock.Mock(return_value={}) + ) + with pytest.raises(APIError): + YHubService().request("get", YHubService().build_url("ydoc", document)) + + assert sample(DURATION, **labels, status="200") == ok_before + 1 + assert sample(DURATION, **labels, status="404") == missing_before + 1 + # the document never makes it to a label + assert str(document.id) not in generate_latest(REGISTRY).decode() + + +class FakePool: + """A psycopg pool, as far as its statistics go.""" + + def __init__(self, stats=None, error=None): + self.stats = stats + self.error = error + + def pop_stats(self): + """Hand the counters over, as psycopg does.""" + if self.error: + raise self.error + return self.stats + + +class FakeConnections(dict): + """`django.db.connections`: iterating it yields the aliases.""" + + +def fake_connection(pool=None): + """A database connection configured with or without a pool.""" + return mock.Mock( + settings_dict={"OPTIONS": {"pool": {"min_size": 1}} if pool else {}}, pool=pool + ) + + +def test_instrumentation_database_pool(monkeypatch): + """The state and the counters of each pooled connection are recorded.""" + stats = { + "pool_size": 8, + "pool_available": 3, + "requests_waiting": 2, + "requests_num": 40, + "requests_queued": 5, + "requests_wait_ms": 1500, + "requests_errors": 1, + "connections_num": 8, + "connections_errors": 2, + } + monkeypatch.setattr( + metrics, + "connections", + FakeConnections( + pooled=fake_connection(FakePool(stats)), + plain=fake_connection(), + broken=fake_connection(FakePool(error=RuntimeError("pool is closed"))), + ), + ) + before = sample("docs_db_pool_requests_total", alias="pooled") + + metrics.record_database_pool() + metrics.record_database_pool() + + assert sample("docs_db_pool_size", alias="pooled") == 8 + assert sample("docs_db_pool_available", alias="pooled") == 3 + assert sample("docs_db_pool_requests_waiting", alias="pooled") == 2 + # what the pool popped twice is added up + assert sample("docs_db_pool_requests_total", alias="pooled") == before + 80 + assert sample("docs_db_pool_requests_queued_total", alias="pooled") >= 10 + assert sample("docs_db_pool_requests_wait_seconds_total", alias="pooled") >= 3 + assert sample("docs_db_pool_requests_errors_total", alias="pooled") >= 2 + assert sample("docs_db_pool_connections_total", alias="pooled") >= 16 + assert sample("docs_db_pool_connections_errors_total", alias="pooled") >= 4 + # a connection without a pool is skipped, a pool that fails is survived + assert REGISTRY.get_sample_value("docs_db_pool_size", {"alias": "plain"}) is None + assert REGISTRY.get_sample_value("docs_db_pool_size", {"alias": "broken"}) is None + + +def test_instrumentation_connect_signals(settings): + """The pool is only recorded at the end of requests when the metrics are enabled.""" + uid = "core.metrics.database_pool" + try: + settings.PROMETHEUS_METRICS_ENABLED = False + instrumentation.connect_signals() + assert not request_finished.disconnect(dispatch_uid=uid) + + settings.PROMETHEUS_METRICS_ENABLED = True + instrumentation.connect_signals() + assert request_finished.disconnect(dispatch_uid=uid) + finally: + request_finished.disconnect(dispatch_uid=uid) + + +def test_instrumentation_forget_dead_processes(monkeypatch, tmp_path): + """ + The gauges of the workers that are gone are dropped, what they counted is kept. + """ + monkeypatch.setenv("PROMETHEUS_MULTIPROC_DIR", str(tmp_path)) + with subprocess.Popen([sys.executable, "-c", "pass"]) as process: + process.wait() + dead, alive = process.pid, os.getpid() + for name in [ + f"gauge_livesum_{dead}.db", + f"gauge_liveall_{dead}.db", + f"gauge_livesum_{alive}.db", + f"counter_{dead}.db", + f"histogram_{dead}.db", + "unrelated.txt", + ]: + (tmp_path / name).touch() + + metrics.forget_dead_processes() + + assert sorted(path.name for path in tmp_path.iterdir()) == sorted( + [ + f"gauge_livesum_{alive}.db", + f"counter_{dead}.db", + f"histogram_{dead}.db", + "unrelated.txt", + ] + ) + + +def test_instrumentation_forget_dead_processes_without_directory(monkeypatch, tmp_path): + """Nothing to do, and nothing to fail on, without a shared directory.""" + monkeypatch.delenv("PROMETHEUS_MULTIPROC_DIR", raising=False) + metrics.forget_dead_processes() + monkeypatch.setenv("PROMETHEUS_MULTIPROC_DIR", str(tmp_path / "missing")) + metrics.forget_dead_processes() + + +def fake_broker(lengths, driver_type="redis"): + """A kombu connection to a broker holding these redis lists.""" + channel = mock.Mock(sep="\\x06\\x16", priority_steps=[0, 3, 6, 9]) + channel.client.llen.side_effect = lambda key: lengths.get(key, 0) + connection = mock.MagicMock() + connection.__enter__.return_value = connection + connection.transport.driver_type = driver_type + connection.default_channel = channel + return connection + + +def collected(collector): + """What a collector exposes, in the text format.""" + registry = CollectorRegistry() + registry.register(collector) + return generate_latest(registry).decode() + + +def test_instrumentation_celery_queue_length(): + """The tasks waiting on every priority of the default queue are added up.""" + broker = fake_broker({"celery": 4, "celery\\x06\\x163": 2, "other": 50}) + + with mock.patch("impress.celery_app.app.connection_for_read", return_value=broker): + output = collected(metrics.CeleryQueueCollector()) + + assert 'docs_celery_queue_length{queue="celery"} 6.0' in output + broker.ensure_connection.assert_called_once_with(max_retries=1, timeout=2) + + +@pytest.mark.parametrize( + "broker", + [ + fake_broker({}, driver_type="memory"), + mock.MagicMock(__enter__=mock.Mock(side_effect=OSError("broker is down"))), + ], + ids=["not-redis", "unreachable"], +) +def test_instrumentation_celery_queue_length_unknown(broker): + """A broker that cannot be asked yields no sample rather than failing the scrape.""" + with mock.patch("impress.celery_app.app.connection_for_read", return_value=broker): + assert "docs_celery_queue_length" not in collected( + metrics.CeleryQueueCollector() + ) diff --git a/src/backend/core/tests/test_prometheus_metrics.py b/src/backend/core/tests/test_prometheus_metrics.py index a361547a4..1f6cf699b 100644 --- a/src/backend/core/tests/test_prometheus_metrics.py +++ b/src/backend/core/tests/test_prometheus_metrics.py @@ -2,10 +2,11 @@ Test the Prometheus metrics endpoint and the instrumentation of the application. """ +import re import socket from importlib import reload +from unittest import mock -from django.test import override_settings from django.urls import clear_url_caches, resolve import pytest @@ -149,9 +150,9 @@ def test_prometheus_metrics_other_routes_need_no_api_key(): assert APIClient().get("/api/v1.0/config/").status_code == 200 -@override_settings(MIDDLEWARE=[BEFORE_MIDDLEWARE, AFTER_MIDDLEWARE]) -def test_prometheus_metrics_requests_are_labelled_by_view(): +def test_prometheus_metrics_requests_are_labelled_by_view(settings): """A request should be counted under the name of its view, never under its path.""" + settings.MIDDLEWARE = [BEFORE_MIDDLEWARE, AFTER_MIDDLEWARE] path = "/api/v1.0/config/" labels = {"status": "200", "view": resolve(path).view_name, "method": "GET"} metric = "django_http_responses_total_by_status_view_method_total" @@ -168,3 +169,26 @@ def test_prometheus_metrics_requests_are_labelled_by_view(): for sample in family.samples for value in sample.labels.values() ) + + +@pytest.mark.usefixtures("metrics_enabled") +def test_prometheus_metrics_celery_queue_length(settings): + """The length of the Celery queue is asked at scrape time, unless it is turned off.""" + path = "core.metrics.CeleryQueueCollector.queue_length" + + with mock.patch(path, return_value=7) as mock_queue_length: + response = APIClient().get("/metrics", HTTP_AUTHORIZATION=f"Bearer {API_KEY}") + assert response.status_code == 200 + assert re.search( + # the text format sorts the labels + rf'docs_celery_queue_length{{hostname="{socket.gethostname()}",queue="celery"}} 7\.0', + response.content.decode(), + ) + mock_queue_length.assert_called_once() + + settings.PROMETHEUS_CELERY_QUEUE_METRICS_ENABLED = False + with mock.patch(path, return_value=7) as mock_queue_length: + response = APIClient().get("/metrics", HTTP_AUTHORIZATION=f"Bearer {API_KEY}") + assert response.status_code == 200 + assert "docs_celery_queue_length" not in response.content.decode() + mock_queue_length.assert_not_called() diff --git a/src/backend/impress/settings.py b/src/backend/impress/settings.py index aca3cf286..99f312e8d 100755 --- a/src/backend/impress/settings.py +++ b/src/backend/impress/settings.py @@ -1269,6 +1269,13 @@ class Base(Configuration): PROMETHEUS_DB_METRICS_ENABLED = values.BooleanValue( True, environ_name="PROMETHEUS_DB_METRICS_ENABLED", environ_prefix=None ) + # Report the length of the Celery queue on /metrics. It is asked to the + # broker at every scrape, by whichever replica answers it. + PROMETHEUS_CELERY_QUEUE_METRICS_ENABLED = values.BooleanValue( + True, + environ_name="PROMETHEUS_CELERY_QUEUE_METRICS_ENABLED", + environ_prefix=None, + ) # uvicorn runs several worker processes, and a scrape is answered by one of # them: they all write their numbers to this directory so that whichever # answers can add them up. It has to be the same for every worker, hence a