acemq_amqp.prometheus

  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"""An :class:`~acemq_amqp.telemetry.Observer` that writes to a Prometheus registry.
 16
 17Behind an extra, the way the broker client is::
 18
 19    pip install "acemq-amqp[prometheus]"
 20
 21    from acemq_amqp.prometheus import PrometheusObserver
 22
 23    mq = await connect(url, observer=PrometheusObserver())
 24
 25The library itself depends on nothing: it calls
 26:class:`~acemq_amqp.telemetry.Observer`, and this module is one implementation of
 27it. That is the same argument the transport already makes and wins — a service
 28that reads an AceMQ envelope should not be made to install an AMQP client, and a
 29service that publishes one should not be made to install a metrics client either.
 30Choosing here would put every user of this package on whichever one was picked.
 31
 32For a service that wants a scrape endpoint and nothing installed at all,
 33:func:`~acemq_amqp.telemetry.prometheus_text` renders an in-memory
 34:class:`~acemq_amqp.telemetry.Metrics` in the same format using only the
 35standard library. This module is for the case where the numbers have to join a
 36registry the rest of the application is already writing to.
 37"""
 38
 39from __future__ import annotations
 40
 41from collections.abc import Mapping
 42from typing import Any
 43
 44from .errors import AceMQError
 45from .telemetry import (
 46    METRIC_CONSUME_ATTEMPTS,
 47    METRIC_CONSUME_DURATION,
 48    METRIC_CONSUME_IN_FLIGHT,
 49    METRIC_CONSUME_TOTAL,
 50    METRIC_DEAD_LETTERED_TOTAL,
 51    METRIC_OUTBOX_LAG,
 52    METRIC_OUTBOX_TOTAL,
 53    METRIC_PIPELINE_RUN_DURATION,
 54    METRIC_PIPELINE_RUN_TOTAL,
 55    METRIC_PUBLISH_DURATION,
 56    METRIC_PUBLISH_TOTAL,
 57    METRIC_REQUEST_DURATION,
 58    METRIC_REQUEST_TOTAL,
 59    METRIC_RETRIED_TOTAL,
 60    METRIC_RUNG_MISSING,
 61    METRIC_SET_ASIDE_FAILED,
 62)
 63
 64#: Buckets for :data:`~acemq_amqp.telemetry.METRIC_CONSUME_DURATION`, in
 65#: seconds.
 66#:
 67#: Reaching to a minute because a handler that talks to something slow really
 68#: does take that long, and a histogram whose top bucket is one second reports
 69#: every one of those as "over a second" and nothing more useful.
 70#:
 71#: They serve every duration this observer is given, which includes
 72#: :data:`~acemq_amqp.telemetry.METRIC_OUTBOX_LAG` — where a minute is not the
 73#: interesting end of the range, because a relay that has been down for an hour
 74#: is publishing hour-old records. The overflow bucket still shows it, and the
 75#: histogram's ``_sum`` over ``_count`` gives the mean lag exactly; a process
 76#: that wants the shape of a long lag as well passes its own ``buckets``.
 77DEFAULT_DURATION_BUCKETS = (
 78    0.005,
 79    0.01,
 80    0.025,
 81    0.05,
 82    0.1,
 83    0.25,
 84    0.5,
 85    1.0,
 86    2.5,
 87    5.0,
 88    10.0,
 89    30.0,
 90    60.0,
 91)
 92
 93#: Buckets for :data:`~acemq_amqp.telemetry.METRIC_CONSUME_ATTEMPTS`, which is a
 94#: count of attempts rather than a duration.
 95#:
 96#: The one metric this library reports through
 97#: :meth:`~acemq_amqp.telemetry.Observer.observe` that is not measured in
 98#: seconds, and the reason it needs buckets of its own: on the duration buckets
 99#: above, every delivery on attempt 1, 2 or 3 would land in the sub-five-second
100#: end and a distribution of attempt counts would read as a distribution of very
101#: fast handlers.
102#:
103#: Whole numbers, because attempts are whole numbers, reaching ten because a
104#: retry policy allowing more than that is unusual and the overflow bucket says
105#: so when there is one. The first bucket is the interesting one: a queue whose
106#: messages are mostly *not* in it is a queue whose dependency is struggling.
107DEFAULT_ATTEMPT_BUCKETS = (1.0, 2.0, 3.0, 4.0, 5.0, 10.0)
108
109
110class PrometheusObserver:
111    """Writes what the library reports into a Prometheus registry.
112
113    Collectors are made the first time a metric is seen rather than up front,
114    because the label names are not known until then: a counter reported with a
115    queue label and one reported with an exchange label are different
116    collectors, and Prometheus refuses a collector registered twice.
117
118    A metric reported with two different sets of label names is a mistake in
119    this library rather than in the caller, and it is refused loudly here
120    instead of quietly dropping half the samples.
121
122    :param registry: where to register, defaulting to the process-wide one that
123        ``prometheus_client.start_http_server`` serves
124    :param namespace: a prefix for every metric name, for a process that already
125        namespaces its metrics
126    :param buckets: the histogram buckets for every duration
127    :param attempt_buckets: the buckets for
128        :data:`~acemq_amqp.telemetry.METRIC_CONSUME_ATTEMPTS`, which counts
129        attempts rather than seconds and would read as a handful of very fast
130        handlers on seconds-shaped buckets
131    :raises AceMQError: when prometheus-client is not installed
132    """
133
134    def __init__(
135        self,
136        registry: Any = None,
137        *,
138        namespace: str = "",
139        buckets: tuple[float, ...] = DEFAULT_DURATION_BUCKETS,
140        attempt_buckets: tuple[float, ...] = DEFAULT_ATTEMPT_BUCKETS,
141    ) -> None:
142        try:
143            import prometheus_client
144        except ImportError as missing:  # pragma: no cover - depends on the install
145            raise AceMQError(
146                "acemq: PrometheusObserver needs prometheus-client, which is an "
147                'optional extra. Install it with pip install "acemq-amqp[prometheus]"'
148            ) from missing
149
150        self._client = prometheus_client
151        self._registry = registry if registry is not None else prometheus_client.REGISTRY
152        self._namespace = namespace
153        self._buckets = buckets
154        self._attempt_buckets = attempt_buckets
155        self._collectors: dict[str, Any] = {}
156        self._label_names: dict[str, tuple[str, ...]] = {}
157
158    def count(self, metric: str, delta: int, labels: Mapping[str, str]) -> None:
159        self._for(metric, labels, self._counter).labels(**_named(labels)).inc(delta)
160
161    def gauge(self, metric: str, value: int, labels: Mapping[str, str]) -> None:
162        self._for(metric, labels, self._gauge).labels(**_named(labels)).set(value)
163
164    def observe(self, metric: str, seconds: float, labels: Mapping[str, str]) -> None:
165        self._for(metric, labels, self._histogram).labels(**_named(labels)).observe(seconds)
166
167    def _for(self, metric: str, labels: Mapping[str, str], make: Any) -> Any:
168        """The collector for a metric, made once and kept.
169
170        Not thread-safe by lock, and it does not need to be: the worst a race
171        can do is build the same collector twice, and the second registration
172        raises rather than corrupting anything. Everything after the first
173        message reads a dictionary.
174        """
175        names = tuple(sorted(_named(labels)))
176        known = self._label_names.get(metric)
177        if known is None:
178            self._label_names[metric] = names
179        elif known != names:
180            raise AceMQError(
181                f"acemq: {metric} was reported with labels {known} and now with "
182                f"{names}; Prometheus cannot hold both, and this is a bug in "
183                "acemq-amqp rather than in your code"
184            )
185
186        collector = self._collectors.get(metric)
187        if collector is None:
188            collector = make(metric, names)
189            self._collectors[metric] = collector
190        return collector
191
192    def _counter(self, metric: str, names: tuple[str, ...]) -> Any:
193        return self._client.Counter(
194            _prometheus_name(metric),
195            _documentation(metric),
196            names,
197            namespace=self._namespace,
198            registry=self._registry,
199        )
200
201    def _gauge(self, metric: str, names: tuple[str, ...]) -> Any:
202        return self._client.Gauge(
203            _prometheus_name(metric),
204            _documentation(metric),
205            names,
206            namespace=self._namespace,
207            registry=self._registry,
208        )
209
210    def _histogram(self, metric: str, names: tuple[str, ...]) -> Any:
211        # The attempt distribution gets its own buckets, and has to: it is the
212        # one thing reported through ``observe`` that is not a number of seconds,
213        # and on seconds-shaped buckets a delivery on attempt 2 is recorded as a
214        # handler that took two seconds. Everything else is a duration and takes
215        # whatever this observer was given.
216        buckets = (
217            self._attempt_buckets
218            if metric == METRIC_CONSUME_ATTEMPTS
219            else self._buckets
220        )
221        return self._client.Histogram(
222            _prometheus_name(metric),
223            _documentation(metric),
224            names,
225            buckets=buckets,
226            namespace=self._namespace,
227            registry=self._registry,
228        )
229
230
231#: What each metric means, for the HELP line a scraper shows beside it.
232_HELP = {
233    METRIC_PUBLISH_TOTAL: "Publishes, by outcome",
234    METRIC_PUBLISH_DURATION: "How long a publish takes, from send to confirm, in seconds",
235    METRIC_CONSUME_TOTAL: "Deliveries settled, by outcome",
236    METRIC_CONSUME_DURATION: "How long a delivery takes, in seconds",
237    METRIC_CONSUME_IN_FLIGHT: "Messages being handled right now",
238    METRIC_CONSUME_ATTEMPTS: "Which attempt a delivery was on when it was handled",
239    METRIC_RETRIED_TOTAL: "Messages given another attempt",
240    METRIC_DEAD_LETTERED_TOTAL: "Messages sent to a dead-letter or parking queue",
241    METRIC_RUNG_MISSING: "Long retries that waited in the consumer for want of a rung queue",
242    METRIC_SET_ASIDE_FAILED: "Messages that could not be moved out of the way",
243    METRIC_OUTBOX_TOTAL: "Outbox records the relay has handled, by outcome",
244    METRIC_OUTBOX_LAG: "How long an outbox record waited to be published, in seconds",
245    METRIC_REQUEST_TOTAL: "Request and reply round trips, by outcome",
246    METRIC_REQUEST_DURATION: "How long a round trip took, in seconds",
247    METRIC_PIPELINE_RUN_TOTAL: "Pipeline runs that finished, by outcome",
248    METRIC_PIPELINE_RUN_DURATION: "How old a message was when it left a pipeline, in seconds",
249}
250
251
252def _documentation(metric: str) -> str:
253    return _HELP.get(metric, metric)
254
255
256def _prometheus_name(metric: str) -> str:
257    """``acemq.consume.total`` becomes ``acemq_consume_total``.
258
259    Prometheus does not allow dots in a metric name, and the AceMQ names use
260    them because Micrometer and OpenTelemetry do.
261    """
262    return metric.replace(".", "_")
263
264
265def _named(labels: Mapping[str, str]) -> dict[str, str]:
266    """The same labels under names Prometheus will accept.
267
268    ``routing.key`` and ``message.type`` are legal tag names everywhere else in
269    the family and illegal here: a Prometheus label name is
270    ``[a-zA-Z_][a-zA-Z0-9_]*`` and nothing else, and prometheus-client refuses
271    the collector rather than the sample — so a publish counter tagged with the
272    routing key would take the whole publisher down at the first message rather
273    than quietly under-report.
274
275    Values are left exactly as they are. A routing key is a value, and its dots
276    are the meaning.
277    """
278    return {name.replace(".", "_"): value for name, value in labels.items()}