✨(backend) add custom metrics when monitoring is enabled

We added custom metrics when the monitoring is enabled, we will measure
the cost of calling outside services like yhub and the converters, the
database pool and celery queue.

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,

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                                             | `<tmp>/impress-prometheus-<uid>`                                        |

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 `<unnamed view>`.
 
@@ -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

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()

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"
+    )

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<pid>\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)

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",

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(

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()
+        )

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()

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