acemq_amqp.telemetry

  1# Copyright 2026 AceMQ.
  2#
  3# Licensed under the Apache License, Version 2.0 (the "License");
  4# you may not use this file except in compliance with the License.
  5# You may obtain a copy of the License at
  6#
  7#     https://www.apache.org/licenses/LICENSE-2.0
  8#
  9# Unless required by applicable law or agreed to in writing, software
 10# distributed under the License is distributed on an "AS IS" BASIS,
 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 12# See the License for the specific language governing permissions and
 13# limitations under the License.
 14
 15"""What the library is doing, and whether it is still working.
 16
 17Two questions an operator asks about a service that consumes a queue, and the
 18library is the only thing that can answer either. How much is going through, how
 19much is failing and how long a handler takes are numbers nobody outside can see;
 20whether the connection is really up — as opposed to a socket that is open and
 21wedged — is a question only something holding the connection can ask.
 22
 23The metric names are the same in Java, Go, Ruby, .NET and here, so a dashboard
 24built against one library reads against another. Java publishes them through
 25Micrometer and .NET through ``System.Diagnostics.Metrics``; Python's standard
 26library has no metrics interface at all, which is why this module defines a
 27small one and calls it. Implement :class:`Observer` against Prometheus,
 28OpenTelemetry, statsd or a log line — or use :class:`Metrics`, which keeps the
 29numbers in memory and is enough for a health endpoint, a test, or a signal
 30handler that prints them.
 31
 32Taking a dependency on a metrics library instead would put every user of this
 33package on the one it chose. The transport already makes that argument and wins
 34it: aio-pika is an extra, and so is anything here that needs more than the
 35standard library.
 36
 37Health is a :class:`HealthReport` with three states rather than a boolean.
 38``degraded`` exists because "working, but not as well as it should" is worth an
 39alert and is not worth taking an instance out of rotation — the replacement will
 40almost certainly be degraded too.
 41"""
 42
 43from __future__ import annotations
 44
 45import asyncio
 46import threading
 47from collections.abc import Mapping
 48from dataclasses import dataclass, field
 49from datetime import datetime, timezone
 50from enum import Enum
 51from typing import Any, Protocol, runtime_checkable
 52
 53#: Publishes, tagged with :data:`TAG_OUTCOME`.
 54#:
 55#: One counter with an outcome rather than one counter per outcome, because a
 56#: dashboard that wants the failure rate wants a ratio of two series of the same
 57#: metric and not a division between two differently-named ones.
 58METRIC_PUBLISH_TOTAL = "acemq.publish.total"
 59
 60#: How long a publish took, in seconds, tagged with :data:`TAG_OUTCOME` and
 61#: carrying the same labels as :data:`METRIC_PUBLISH_TOTAL`.
 62#:
 63#: From calling ``send`` to the broker answering, so it includes the encode, the
 64#: interceptor chain and the confirm round trip — everything a caller waits for.
 65#: Recorded whatever the outcome, including a publish that failed and one that
 66#: reached no queue, because the slow publishes are the interesting ones and a
 67#: timing that covered only the successes would drop them.
 68#:
 69#: The same labels as the total beside it, deliberately: a dashboard dividing one
 70#: into the other needs both series cut the same way, and a duration with a label
 71#: the total does not have cannot be divided into it.
 72METRIC_PUBLISH_DURATION = "acemq.publish.duration"
 73
 74#: Deliveries that were settled, tagged with :data:`TAG_OUTCOME`.
 75#:
 76#: Counted once per delivery, from the settlement rather than from what the
 77#: handler asked for: a handler asking for a retry it has no attempts left for
 78#: is dead-lettered, and this says so. The sum across the outcomes is how many
 79#: messages were handled.
 80METRIC_CONSUME_TOTAL = "acemq.consume.total"
 81
 82#: How long a delivery took, in seconds, tagged with :data:`TAG_OUTCOME`.
 83#:
 84#: Timed around the interceptors as well as the handler, because what an
 85#: operator wants to know is how long a message takes to deal with, and an
 86#: interceptor that opens a transaction is part of dealing with it.
 87METRIC_CONSUME_DURATION = "acemq.consume.duration"
 88
 89#: Which attempt a delivery was on when it was handled, as a distribution,
 90#: labelled :data:`TAG_QUEUE`.
 91#:
 92#: Not a duration, and the only metric here that is not. A rising distribution is
 93#: a dependency in trouble, and it says so earlier than anything else does: the
 94#: dead letters only move once the attempts run out, so a queue whose messages
 95#: have started needing three tries each looks entirely healthy on every other
 96#: number until the first one gives up.
 97#:
 98#: Recorded through :meth:`Observer.observe` because that is the only
 99#: distribution the interface has. Adding a method for it would break every
100#: observer an application has already written, for one metric — see the note on
101#: :meth:`Observer.observe` for what that means for an implementation.
102METRIC_CONSUME_ATTEMPTS = "acemq.consume.attempts"
103
104#: How many messages are in a handler right now, as a gauge. Bounded by prefetch
105#: times concurrency.
106METRIC_CONSUME_IN_FLIGHT = "acemq.consume.in.flight"
107
108#: Messages sent to a retry queue, or otherwise given another attempt. Labelled
109#: ``where``: ``broker``, ``consumer`` or ``requeued``.
110#:
111#: Kept beside :data:`METRIC_CONSUME_TOTAL` rather than folded into it, exactly
112#: as Java keeps it: the outcome on a delivery says what was decided, and this
113#: says how many messages are going round again, which is the number an alert is
114#: written against.
115METRIC_RETRIED_TOTAL = "acemq.messages.retried.total"
116
117#: Messages set aside, tagged with :data:`TAG_OUTCOME`: ``dead_lettered`` for one
118#: that ran out of attempts and went to ``{queue}.dlq``, ``parked`` for one that
119#: nothing could read and went to ``{queue}.parked``.
120#:
121#: One counter split by the outcome rather than two metrics, which is what Java
122#: settled on: both are a message this queue gave up on, and an operator asking
123#: how much a queue is giving up on wants one number that can then be split. The
124#: split still matters — a message that failed five times and a message nothing
125#: could read are different problems with different answers, and whoever drains
126#: the queue should not have to sort them by hand — which is what the tag is for.
127METRIC_DEAD_LETTERED_TOTAL = "acemq.messages.dead.lettered.total"
128
129#: Long retries that had to wait in the consumer because the rung queue they
130#: were meant to wait on is not on the broker.
131#:
132#: Worth an alert. Nothing breaks — the message is still retried and the wait
133#: still happens — but the reason the rung exists is gone: a consumer restart
134#: mid-wait now shortens a five-minute backoff to nothing, and the only other
135#: sign of it is one log line on a path nobody is watching.
136METRIC_RUNG_MISSING = "acemq.retry.rung.missing"
137
138#: Messages that could not be moved to a dead-letter or parking queue, usually
139#: because it was never declared. The message is rejected to the broker instead,
140#: which is the last thing between it and nothing. Labelled :data:`TAG_QUEUE` and
141#: :data:`TAG_TARGET`.
142METRIC_SET_ASIDE_FAILED = "acemq.messages.set.aside.failed"
143
144#: Request and reply round trips, tagged with :data:`TAG_ROUTING_KEY` and
145#: :data:`TAG_OUTCOME`: :data:`OUTCOME_ANSWERED`, :data:`OUTCOME_TIMED_OUT` or
146#: :data:`OUTCOME_FAILED`.
147#:
148#: The caller's number, and only the caller's. A responder's side is an ordinary
149#: queue and is counted on :data:`METRIC_CONSUME_TOTAL` like any other; what
150#: nothing else can see is how long *asking* took, because the publish and the
151#: reply's delivery are two unrelated hops and neither of them is the round trip.
152METRIC_REQUEST_TOTAL = "acemq.request.total"
153
154#: How long a round trip took, in seconds, carrying the same labels as
155#: :data:`METRIC_REQUEST_TOTAL`.
156#:
157#: Measured as the caller experienced it: from before the request is published to
158#: after the reply is in hand, so the publish, the responder's work and the reply
159#: coming back are all inside it. A request that times out is recorded too, at
160#: its deadline, because a p99 that quietly drops the slowest calls is a p99 that
161#: says a service is fast right up to the point where nothing answers at all.
162METRIC_REQUEST_DURATION = "acemq.request.duration"
163
164#: Outbox records the relay has handled, tagged with :data:`TAG_OUTCOME`:
165#: :data:`OUTCOME_PUBLISHED` or :data:`OUTCOME_FAILED`. Labelled
166#: :data:`TAG_EXCHANGE` and :data:`TAG_ROUTING_KEY` as well, because a relay
167#: drains one outbox into many destinations and "the relay is behind" is nearly
168#: always "the relay is behind *on one exchange*".
169METRIC_OUTBOX_TOTAL = "acemq.outbox.total"
170
171#: Routing-slip runs that reached the end of their itinerary, labelled
172#: :data:`TAG_PIPELINE`, :data:`TAG_STEP` and :data:`TAG_OUTCOME`.
173#:
174#: Java's ``acemq.pipeline.run.total``, written from the same place: the step
175#: that found no next stop is the step that finished the run, so the counter goes
176#: up exactly once per run however many services it passed through.
177#:
178#: The outcome is always :data:`OUTCOME_COMPLETED` here. Java also writes
179#: ``ended_early`` for a step that returned nothing with stops still to go, and
180#: this library has no way for a routing-slip step to say that — see
181#: :data:`OUTCOME_ENDED_EARLY`.
182METRIC_PIPELINE_RUN_TOTAL = "acemq.pipeline.run.total"
183
184#: How old a message was when it left a pipeline, in seconds, carrying the same
185#: labels as :data:`METRIC_PIPELINE_RUN_TOTAL`.
186#:
187#: The *run's* duration and not the step's, which is why it is the envelope's age
188#: rather than a timer: the envelope was created when the message entered the
189#: pipeline and carried through every hop, so this is the only number that spans
190#: services. A per-step timing is :data:`METRIC_CONSUME_DURATION` on each step's
191#: own queue, and summing those misses every second a message spent waiting on a
192#: queue between two of them — which on a busy pipeline is most of the answer.
193METRIC_PIPELINE_RUN_DURATION = "acemq.pipeline.run.duration"
194
195#: How long an outbox record waited between being committed and being published,
196#: in seconds. Recorded only for a record that went out, and carrying the same
197#: labels as :data:`METRIC_OUTBOX_TOTAL`.
198#:
199#: **The one number that reveals a stopped relay.** A committed, unpublished row
200#: is a message that exists, is owed to somebody, and appears in no queue depth
201#: anywhere: publish rates, consume rates and queue depths all read as a system
202#: with nothing to send, which is exactly what a system that has stopped sending
203#: looks like from outside.
204#:
205#: Measured from the record's own ``created_at`` — when the transaction that
206#: decided the message committed — and not from the start of the sweep that
207#: picked it up. The question a lag answers is how long somebody has been owed
208#: this message, so the wait for a sweep is part of the answer rather than the
209#: start of it; timed from the sweep, a relay that has been down for an hour
210#: reports the same handful of milliseconds as one that is keeping up.
211METRIC_OUTBOX_LAG = "acemq.outbox.lag"
212
213# ---------- tag names ----------
214#
215# Dotted, because Micrometer and OpenTelemetry are dotted and these have to be
216# the same words in five languages. ``routing.key`` and ``message.type`` are
217# both illegal Prometheus label names; :mod:`acemq_amqp.prometheus` and
218# :func:`prometheus_text` rewrite the dots to underscores at the point of
219# export, which is the only place the restriction applies.
220
221#: The exchange a message was published to, empty for the default exchange.
222TAG_EXCHANGE = "exchange"
223
224#: The routing key used, or the queue name when publishing without an exchange.
225TAG_ROUTING_KEY = "routing.key"
226
227#: The queue a delivery came from.
228TAG_QUEUE = "queue"
229
230#: Where a message was being set aside to when that failed: the dead-letter
231#: queue or the parking lot. Bounded by the topology, so it is safe as a tag.
232TAG_TARGET = "target"
233
234#: The pipeline a routing-slip run belongs to, which is its exchange. Empty when
235#: the run carried a JSON slip that names no pipeline.
236TAG_PIPELINE = "pipeline"
237
238#: The step a pipeline run was at when it finished. Bounded by the route, which
239#: names a fixed handful of steps, so it is safe as a tag.
240TAG_STEP = "step"
241
242#: What happened. The publish outcomes are below; the delivery outcomes are
243#: :data:`~acemq_amqp.ack.OUTCOME_ACKED` and its neighbours in
244#: :mod:`acemq_amqp.ack`, which are the same words the settlement and the span
245#: already use.
246TAG_OUTCOME = "outcome"
247
248#: A publish the broker took responsibility for.
249OUTCOME_CONFIRMED = "confirmed"
250
251#: A publish that went out with nothing promising anything about it. Publisher
252#: confirms were not on.
253OUTCOME_PUBLISHED = "published"
254
255#: A mandatory publish the broker had no queue for. The quietest failure AMQP
256#: has, and the reason ``mandatory`` is worth setting.
257OUTCOME_UNROUTABLE = "unroutable"
258
259#: A publish that may have been lost: the broker nacked it, the confirm never
260#: came, or the connection failed partway. The message may or may not be on the
261#: broker, which is what makes this the outcome worth alerting on.
262OUTCOME_FAILED = "failed"
263
264#: A publish this library declined before writing anything to the wire — today
265#: a :class:`~acemq_amqp.PublishingPausedError`, raised while the broker has
266#: blocked the connection. Nothing was sent, so nothing was lost and retrying
267#: cannot duplicate it. Kept apart from :data:`OUTCOME_FAILED` so a broker
268#: alarm does not read as message loss; the same word in all five libraries.
269OUTCOME_REFUSED = "refused"
270
271#: A request that got its reply.
272OUTCOME_ANSWERED = "answered"
273
274#: A request that reached its deadline with no reply.
275OUTCOME_TIMED_OUT = "timed_out"
276
277#: A pipeline run that reached the end of its itinerary.
278OUTCOME_COMPLETED = "completed"
279
280#: A pipeline run that stopped before the end of its route because a step chose
281#: to stop it. **Nothing in this library writes it**, and the name is here so
282#: that a dashboard shared with Java can be read rather than guessed at: a
283#: :data:`~acemq_amqp.patterns.Stage` always returns the payload for the next
284#: stop, so there is no way for one to end a run early and no honest moment to
285#: emit this. :func:`~acemq_amqp.patterns.then` has :data:`NOTHING
286#: <acemq_amqp.patterns.NOTHING>` for that, and a ``then`` chain is not a routing
287#: slip — it has no pipeline name, no step name and no run to end.
288OUTCOME_ENDED_EARLY = "ended_early"
289
290
291@runtime_checkable
292class Observer(Protocol):
293    """Told what the library is doing.
294
295    Every method is called on the path a message takes, so none of them may
296    block and all of them have to be safe to call from several tasks at once.
297    An observer that talks to the network on each call is an observer that makes
298    every publish as slow as the network it talks to; buffer, and flush
299    somewhere else.
300
301    Nothing here is awaitable, deliberately. A metrics call that could suspend
302    would reorder the code around it and turn "count this" into a scheduling
303    decision, and no metrics library needs it.
304    """
305
306    def count(self, metric: str, delta: int, labels: Mapping[str, str]) -> None:
307        """Adds to a counter."""
308
309    def observe(self, metric: str, seconds: float, labels: Mapping[str, str]) -> None:
310        """Records one sample of a distribution.
311
312        A duration in seconds for every metric but one. The exception is
313        :data:`METRIC_CONSUME_ATTEMPTS`, which is a count of attempts and is
314        recorded here because this is the only distribution the interface has —
315        Java uses a Micrometer ``DistributionSummary`` for it and a ``Timer`` for
316        the rest, and this interface has one method where Java has two.
317
318        The argument keeps its name because renaming it would break every
319        observer that passes it by keyword, and because it is a duration in every
320        case but that one. An implementation that puts durations into histogram
321        buckets should give that metric its own: seconds-shaped buckets over a
322        value of 1, 2 or 3 report an attempt count as a handful of very fast
323        handlers. :class:`~acemq_amqp.prometheus.PrometheusObserver` does exactly
324        that, with :data:`~acemq_amqp.prometheus.DEFAULT_ATTEMPT_BUCKETS`.
325        """
326
327    def gauge(self, metric: str, value: int, labels: Mapping[str, str]) -> None:
328        """Sets a current value."""
329
330
331class NullObserver:
332    """An observer that does nothing, which is what a connection has by default.
333
334    A real object rather than ``None`` checked for at each call site, because
335    the alternative is a branch on every publish and every delivery, and one of
336    them is eventually written the wrong way round.
337    """
338
339    def count(self, metric: str, delta: int, labels: Mapping[str, str]) -> None:
340        return None
341
342    def observe(self, metric: str, seconds: float, labels: Mapping[str, str]) -> None:
343        return None
344
345    def gauge(self, metric: str, value: int, labels: Mapping[str, str]) -> None:
346        return None
347
348
349@dataclass(frozen=True, slots=True)
350class DurationSummary:
351    """What a :class:`Metrics` knows about a timing.
352
353    Deliberately not percentiles. Computing those needs either every sample kept
354    or a sketch, and a library that quietly did either would be making a
355    decision about memory that belongs to the application.
356    """
357
358    count: int = 0
359    total: float = 0.0
360    fastest: float = 0.0
361    slowest: float = 0.0
362
363    @property
364    def mean(self) -> float:
365        """The average, or zero when nothing has been recorded."""
366        return self.total / self.count if self.count else 0.0
367
368
369class Metrics:
370    """An :class:`Observer` that keeps the numbers in memory.
371
372    Enough to serve from a health endpoint, assert on in a test, or print on a
373    signal. It is not a substitute for a real metrics system: no histograms, no
374    percentiles, and nothing is exported anywhere. :func:`prometheus_text`
375    renders what it has for a scraper, which needs no dependency at all.
376
377    Locked rather than relying on the event loop, because the blocking API in
378    :mod:`acemq_amqp.sync` runs handlers on a thread pool and a counter read
379    from two threads at once is a counter that loses increments.
380    """
381
382    def __init__(self) -> None:
383        self._lock = threading.Lock()
384        self._counts: dict[str, int] = {}
385        self._gauges: dict[str, int] = {}
386        self._durations: dict[str, DurationSummary] = {}
387
388    def count(self, metric: str, delta: int, labels: Mapping[str, str]) -> None:
389        key = metric_key(metric, labels)
390        with self._lock:
391            self._counts[key] = self._counts.get(key, 0) + delta
392
393    def gauge(self, metric: str, value: int, labels: Mapping[str, str]) -> None:
394        key = metric_key(metric, labels)
395        with self._lock:
396            self._gauges[key] = value
397
398    def observe(self, metric: str, seconds: float, labels: Mapping[str, str]) -> None:
399        key = metric_key(metric, labels)
400        with self._lock:
401            existing = self._durations.get(key)
402            if existing is None:
403                self._durations[key] = DurationSummary(1, seconds, seconds, seconds)
404                return
405            self._durations[key] = DurationSummary(
406                count=existing.count + 1,
407                total=existing.total + seconds,
408                fastest=min(existing.fastest, seconds),
409                slowest=max(existing.slowest, seconds),
410            )
411
412    @property
413    def counts(self) -> dict[str, int]:
414        """Every counter, keyed by metric and labels."""
415        with self._lock:
416            return dict(self._counts)
417
418    @property
419    def gauges(self) -> dict[str, int]:
420        """Every gauge."""
421        with self._lock:
422            return dict(self._gauges)
423
424    @property
425    def durations(self) -> dict[str, DurationSummary]:
426        """A summary per timing: how many, the total, the fastest, the slowest."""
427        with self._lock:
428            return dict(self._durations)
429
430    def __str__(self) -> str:
431        lines = [f"{name} {value}" for name, value in sorted(self.counts.items())]
432        lines += [f"{name} {value}" for name, value in sorted(self.gauges.items())]
433        lines += [
434            f"{name} count={summary.count} mean={summary.mean:.4f}s"
435            for name, summary in sorted(self.durations.items())
436        ]
437        return "\n".join(lines) or "nothing recorded yet"
438
439
440def metric_key(metric: str, labels: Mapping[str, str]) -> str:
441    """A metric and its labels as one string.
442
443    Sorted, so the same labels always give the same key however the mapping was
444    built. Unsorted, one counter quietly becomes several that each hold part of
445    the answer.
446    """
447    if not labels:
448        return metric
449    rendered = ",".join(f'{name}="{labels[name]}"' for name in sorted(labels))
450    return f"{metric}{{{rendered}}}"
451
452
453def prometheus_text(metrics: Metrics) -> str:
454    """Renders a :class:`Metrics` in the Prometheus text format.
455
456    Written out rather than through a client library, so a service that only
457    wants a scrape endpoint needs nothing installed. The format is small and
458    stable: a TYPE line, then samples, with dots in the metric name — and in the
459    label names, where ``routing.key`` and ``message.type`` have the same
460    problem — replaced by underscores, because Prometheus allows them in
461    neither.
462
463    A real registry — exemplars, native histograms, a shared process collector —
464    is what :mod:`acemq_amqp.prometheus` is for, and it needs the extra.
465
466    :param metrics: what to render
467    :returns: the body of a ``/acemq-metrics`` response
468    """
469    lines: list[str] = []
470    for name, value in sorted(metrics.counts.items()):
471        metric, labels = _split_key(name)
472        base = _prometheus_name(metric)
473        lines += [f"# TYPE {base} counter", f"{base}{labels} {value}"]
474    for name, value in sorted(metrics.gauges.items()):
475        metric, labels = _split_key(name)
476        base = _prometheus_name(metric)
477        lines += [f"# TYPE {base} gauge", f"{base}{labels} {value}"]
478    for name, summary in sorted(metrics.durations.items()):
479        metric, labels = _split_key(name)
480        base = _prometheus_name(metric)
481        lines += [
482            f"# TYPE {base} summary",
483            f"{base}_count{labels} {summary.count}",
484            f"{base}_sum{labels} {summary.total}",
485        ]
486    return "\n".join(lines) + ("\n" if lines else "")
487
488
489def _split_key(key: str) -> tuple[str, str]:
490    """A key back into its metric and its labels, both ready for a scraper.
491
492    The label names are rewritten here as well as the metric name, because
493    ``routing.key`` and ``message.type`` are no more legal as Prometheus label
494    names than a dotted metric name is, and a scrape carrying one is a scrape
495    the server rejects in full — the sample it could not parse is not the only
496    one lost.
497    """
498    if "{" not in key:
499        return key, ""
500    metric, _, labels = key.partition("{")
501    return metric, "{" + _prometheus_labels(labels[:-1]) + "}"
502
503
504def _prometheus_labels(rendered: str) -> str:
505    """``routing.key="a.b"`` becomes ``routing_key="a.b"``.
506
507    Names only. A label *value* is where a queue name lives, and queue names
508    have dots in them that mean something — rewriting those would report a
509    different queue than the one the message came from.
510
511    The parse leans on :func:`metric_key` having written this string: a name, an
512    equals sign, then a value between the next two quotes. Anything that does
513    not read that way is handed back untouched, because a mangled scrape line is
514    worse than an odd one.
515    """
516    pairs: list[str] = []
517    rest = rendered
518    while rest:
519        name, equals, rest = rest.partition('=')
520        if not equals or not rest.startswith('"'):
521            return rendered
522        value, quote, rest = rest[1:].partition('"')
523        if not quote:
524            return rendered
525        pairs.append(f'{name.replace(".", "_")}="{value}"')
526        if rest.startswith(","):
527            rest = rest[1:]
528        elif rest:
529            return rendered
530    return ",".join(pairs)
531
532
533def _prometheus_name(metric: str) -> str:
534    return metric.replace(".", "_")
535
536
537class HealthStatus(str, Enum):
538    """How a check turned out."""
539
540    #: Working.
541    UP = "up"
542
543    #: Not working. A readiness probe should fail on this.
544    DOWN = "down"
545
546    #: Working, but not as well as it should. Worth an alert; not worth taking
547    #: the instance out of rotation, because the replacement will almost
548    #: certainly be degraded too.
549    DEGRADED = "degraded"
550
551
552@dataclass(frozen=True, slots=True)
553class HealthReport:
554    """What a check found.
555
556    :param status: up, down or degraded
557    :param detail: why, when it is not simply up
558    :param checked: when this was worked out, in UTC
559    :param parts: whatever the check wants to show — a consumer count, a
560        round-trip time, the reports of the checks that were combined
561    """
562
563    status: HealthStatus
564    detail: str = ""
565    checked: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
566    parts: Mapping[str, Any] = field(default_factory=dict)
567
568    @property
569    def healthy(self) -> bool:
570        """Whether a readiness probe should pass.
571
572        Degraded passes. An instance that is slow is still an instance that
573        works, and taking it out of rotation moves the traffic to one that will
574        be just as slow with more of it.
575        """
576        return self.status is not HealthStatus.DOWN
577
578    def __str__(self) -> str:
579        return self.status.value if not self.detail else f"{self.status.value}: {self.detail}"
580
581
582@runtime_checkable
583class HealthCheck(Protocol):
584    """Anything that can report on itself.
585
586    The library provides one for the connection; an application adds its own — a
587    database, a downstream service — and :func:`aggregate_health` combines them.
588    """
589
590    @property
591    def name(self) -> str:
592        """What this check is called in a combined report."""
593
594    async def check(self) -> HealthReport:
595        """The current state.
596
597        It must return quickly and must not raise: a check that hangs makes a
598        readiness probe hang with it, and a probe that hangs is a pod that never
599        comes back.
600        """
601
602
603async def aggregate_health(*checks: HealthCheck, timeout: float = 5.0) -> HealthReport:
604    """Runs every check at once and combines the answers.
605
606    The combined status is the worst of them: one thing being down makes the
607    whole report down, because a service that cannot reach its broker is not
608    ready however healthy the rest of it is.
609
610    At once rather than in turn, so a slow check does not add its latency to the
611    others, and under a deadline, so one that ignores the instruction above
612    still cannot hang the probe. A check that times out is reported as down with
613    the reason, which is the honest reading: something that will not answer in
614    five seconds is not something a request can depend on.
615
616    :param checks: what to ask
617    :param timeout: how long to give all of them together
618    :returns: the combined report
619    """
620    if not checks:
621        return HealthReport(HealthStatus.UP)
622
623    async def run(check: HealthCheck) -> tuple[str, HealthReport]:
624        try:
625            return check.name, await check.check()
626        except Exception as failure:
627            return check.name, HealthReport(
628                HealthStatus.DOWN, f"the check itself failed: {failure}"
629            )
630
631    tasks = [asyncio.ensure_future(run(check)) for check in checks]
632    try:
633        done, pending = await asyncio.wait(tasks, timeout=timeout)
634    except asyncio.CancelledError:
635        for task in tasks:
636            task.cancel()
637        raise
638
639    parts: dict[str, Any] = {}
640    worst = HealthStatus.UP
641    reasons: list[str] = []
642
643    for task in pending:
644        task.cancel()
645    for index, task in enumerate(tasks):
646        if task in done:
647            name, report = task.result()
648        else:
649            name = checks[index].name
650            report = HealthReport(HealthStatus.DOWN, f"did not answer within {timeout}s")
651        parts[name] = report
652        if report.status is HealthStatus.DOWN:
653            worst = HealthStatus.DOWN
654            reasons.append(name)
655        elif report.status is HealthStatus.DEGRADED:
656            if worst is not HealthStatus.DOWN:
657                worst = HealthStatus.DEGRADED
658            reasons.append(name)
659
660    return HealthReport(worst, ", ".join(reasons), parts=parts)