acemq_amqp.tracing

  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"""OpenTelemetry spans for publishing and consuming::
 16
 17    pip install "acemq-amqp[opentelemetry]"
 18
 19    from acemq_amqp.tracing import OpenTelemetryTracing
 20
 21    mq = await connect(url)
 22    OpenTelemetryTracing().install(mq)
 23
 24:mod:`acemq_amqp.telemetry` answers *how much*; this answers *what happened to
 25this message*. They are different questions and neither substitutes for the
 26other: a counter says a thousand messages were dead-lettered, and a trace says
 27this one was published by the checkout service, retried twice over four minutes
 28and given up on — which is the question somebody is actually holding when they
 29open a dashboard.
 30
 31**The join across processes is the whole point.** A consumer's span is a child of
 32the *publish* that caused it, taken from the message's own headers rather than
 33from whatever context happened to be current when the delivery arrived. Those
 34are different traces, minutes and machines apart, and joining them is the one
 35thing a messaging system needs from tracing that an HTTP client does not.
 36
 37The headers
 38-----------
 39
 40``traceparent`` and ``tracestate``, deliberately **not** ``x-acemq-`` prefixed.
 41They are W3C names that every other piece of tracing tooling already recognises,
 42so a message this library publishes is readable by a consumer written with no
 43knowledge of AceMQ at all, and one published by such a consumer joins up here.
 44Prefixing them would have been tidy and would have made the context private to
 45this library, which is the opposite of what a propagation format is for. Java, Go
 46and .NET write the same two names.
 47
 48Spans
 49-----
 50
 51============================  ==========  ==========================
 52name                          kind        when
 53============================  ==========  ==========================
 54``<destination> publish``     PRODUCER    a message goes out
 55``<queue> process``           CONSUMER    a handler runs
 56``<destination> request``     CLIENT      a request waits for a reply
 57============================  ==========  ==========================
 58
 59``request`` is CLIENT rather than PRODUCER because that span waits for an answer.
 60Its duration means something different as a result — a slow publish is a slow
 61broker, and a slow request is a slow *responder* — and the kind is what makes a
 62tracing backend show them apart rather than averaging one into the other. It
 63ends ``answered`` or ``timed_out``, the two words Java writes on the same span.
 64
 65The routing key is a ``publish`` attribute and only a ``publish`` attribute. A
 66delivery's span does not carry one, because no other library's does — Java is
 67not even handed one at ``consumeStarted`` — and an attribute present on one
 68library's process spans and absent from the other four's is worse than an
 69attribute nobody has: a query written against it quietly returns the Python
 70services and calls that the answer.
 71
 72Events, not spans
 73-----------------
 74
 75``outbox.publish_failed``, ``pipeline.run_finished``, ``message.retried`` and
 76``message.dead_lettered`` are recorded as events on whatever span is current. A
 77zero-length span at the end of a trace adds a row to the waterfall and no
 78information; an event is attached to the span that was actually doing the work,
 79which is where somebody reading the trace is already looking.
 80
 81The last two are emitted by the consumer itself, on the delivery's own
 82``process`` span, at the moment it decides. That is later than it sounds: what
 83the handler answered is not what happens to the message, because a handler
 84asking for another attempt when there are none left is dead-lettered instead. So
 85the process span stays open past the handler and takes its outcome from the
 86consumer's decision — which is why a message that ran out of attempts reads
 87``dead_lettered`` rather than ``retried``, and why searching a backend for
 88dead letters finds them.
 89
 90The dependency
 91--------------
 92
 93The ``opentelemetry-api`` package, and not the SDK. That is the package a library
 94is supposed to depend on: without an SDK installed and configured by the
 95application, the API's no-op implementation is what runs, so importing this can
 96never start exporting anything nobody asked for. The application chooses the SDK,
 97the exporter and the sampler; this module only describes what happened.
 98"""
 99
100from __future__ import annotations
101
102import asyncio
103from collections.abc import Iterator, Mapping
104from contextlib import contextmanager
105from datetime import timedelta
106from typing import TYPE_CHECKING, Any
107
108from .ack import (
109    OUTCOME_ACKED,
110    OUTCOME_DEAD_LETTERED,
111    OUTCOME_IN_PROGRESS,
112    OUTCOME_PARKED,
113    OUTCOME_REJECTED,
114    OUTCOME_RETRIED,
115    Ack,
116    Action,
117    Settlement,
118)
119from .envelope import Envelope
120from .errors import AceMQError, PublishingPausedError
121from .headers import TRACEPARENT, TRACESTATE
122from .interceptors import (
123    ConsumeContext,
124    ConsumeInterceptor,
125    ConsumeNext,
126    PublishContext,
127    PublishInterceptor,
128    PublishNext,
129    SettlementListener,
130)
131from .telemetry import (
132    OUTCOME_ANSWERED,
133    OUTCOME_CONFIRMED,
134    OUTCOME_FAILED,
135    OUTCOME_PUBLISHED,
136    OUTCOME_REFUSED,
137    OUTCOME_TIMED_OUT,
138    OUTCOME_UNROUTABLE,
139)
140from .transport import PublishResult
141
142if TYPE_CHECKING:  # pragma: no cover - imported for types alone
143    from .connection import Connection
144
145#: What is being traced. ``rabbitmq`` by default, and the same value the other
146#: libraries put here, so one dashboard reads across all five.
147DEFAULT_SYSTEM = "rabbitmq"
148
149#: The name this library's spans are recorded under. The reverse-domain name
150#: Java, Ruby, Go and .NET all register, and not a Python module path: spans
151#: from five libraries are meant to group under one instrumentation scope in a
152#: trace backend, and they do not if one of them answers to something else.
153INSTRUMENTATION_NAME = "org.acemq.amqp"
154
155#: Appended to the destination, with a space. ``orders publish``.
156SPAN_PUBLISH_SUFFIX = " publish"
157
158#: Appended to the queue. ``orders.new process``.
159SPAN_PROCESS_SUFFIX = " process"
160
161#: Appended to the destination. ``pricing request``.
162SPAN_REQUEST_SUFFIX = " request"
163
164#: The OpenTelemetry semantic conventions for messaging, plus three of this
165#: library's own under an ``acemq`` namespace — attempt, message type and
166#: outcome have no standard names and are the three things most often wanted.
167ATTR_SYSTEM = "messaging.system"
168ATTR_DESTINATION = "messaging.destination.name"
169ATTR_OPERATION = "messaging.operation"
170ATTR_MESSAGE_ID = "messaging.message.id"
171ATTR_CONVERSATION_ID = "messaging.message.conversation_id"
172ATTR_ROUTING_KEY = "messaging.rabbitmq.destination.routing_key"
173ATTR_MESSAGE_TYPE = "messaging.acemq.message_type"
174ATTR_ATTEMPT = "messaging.acemq.attempt"
175ATTR_OUTCOME = "messaging.acemq.outcome"
176ATTR_REASON = "messaging.acemq.reason"
177ATTR_RETRY_DELAY = "messaging.acemq.retry_delay_ms"
178ATTR_RUN_AGE = "messaging.acemq.run_age_ms"
179ATTR_OUTBOX_LAG = "messaging.acemq.outbox_lag_ms"
180
181#: The ``pipeline.run_finished`` event's own keys, and bare rather than
182#: namespaced. They are the metric tag names — Java, Ruby and Go all write
183#: ``pipeline``, ``step`` and ``outcome`` on this event — so a query that finds
184#: the event in one library finds it in all five. The span *attribute* for an
185#: outcome stays :data:`ATTR_OUTCOME`, which is a different thing in a different
186#: place and keeps its namespace.
187ATTR_PIPELINE = "pipeline"
188ATTR_STEP = "step"
189ATTR_PIPELINE_OUTCOME = "outcome"
190
191# The words a publish or a request can end in — ``confirmed``, ``published``,
192# ``unroutable``, ``failed``, ``refused``, ``answered`` and ``timed_out`` — are
193# defined in :mod:`acemq_amqp.telemetry` beside the metrics that are tagged with them, and
194# re-exported here: a span and a counter describing the same publish have to say
195# the same word, and the only way to be sure of that is for there to be one
196# word. ``tracing.OUTCOME_CONFIRMED`` still resolves, and is the same string.
197#
198# The five a delivery can end in — ``acked``, ``retried``, ``rejected``,
199# ``dead_lettered`` and ``parked`` — are defined in :mod:`acemq_amqp.ack` and
200# re-exported here, because the consumer decides them and this only writes them
201# down. They keep their names: ``tracing.OUTCOME_DEAD_LETTERED`` is where
202# people look for them.
203
204#: The outcomes that make a span an error, and only these.
205#:
206#: ``retried`` and ``rejected`` are deliberately absent. A retry is the system
207#: working — the message will be tried again and very often succeeds — and
208#: marking it an error paints a trace red for something that turned out fine. A
209#: rejection is a decision the handler made on purpose. What is left is the three
210#: that mean a message did not get where it was going.
211#:
212#: ``timed_out`` and ``refused`` are absent for a different reason. It is not
213#: the outcome that makes those spans red, the exception is: a request which
214#: reached its deadline raised, as did a publish refused while the broker had
215#: blocked the connection, and the adapter records that exception and sets the
216#: error status from it. Same colour, with the cause in the description rather
217#: than the bare word — and the set stays character-identical to Java's.
218#:
219#: ``parked`` is absent for the same reason ``rejected`` is: a handler that
220#: parks a message decided to, on purpose, having read it. The ``parked`` outcome
221#: on ``acemq.messages.dead.lettered.total`` and the queue itself are what an
222#: operator watches for those; painting the trace red as well would make a
223#: deliberate decision look like a fault.
224ERROR_OUTCOMES = frozenset({OUTCOME_UNROUTABLE, OUTCOME_FAILED, OUTCOME_DEAD_LETTERED})
225
226#: The events recorded on the current span rather than as spans of their own.
227EVENT_OUTBOX_PUBLISH_FAILED = "outbox.publish_failed"
228EVENT_PIPELINE_RUN_FINISHED = "pipeline.run_finished"
229EVENT_MESSAGE_RETRIED = "message.retried"
230EVENT_MESSAGE_DEAD_LETTERED = "message.dead_lettered"
231
232_ACK_OUTCOMES = {
233    Action.ACCEPT: OUTCOME_ACKED,
234    Action.RETRY: OUTCOME_RETRIED,
235    Action.REJECT: OUTCOME_REJECTED,
236    Action.PARK: OUTCOME_PARKED,
237    Action.IN_PROGRESS: OUTCOME_IN_PROGRESS,
238}
239
240
241class OpenTelemetryTracing:
242    """Emits spans for what this library does, and joins them up across
243    processes.
244
245    :param system: what to report as ``messaging.system``
246    :param tracer_provider: where to get the tracer, or ``None`` for the global
247        one the application configured
248    :param propagator: how to read and write trace context, or ``None`` for the
249        global propagator — which is W3C ``traceparent``/``tracestate`` unless
250        the application changed it
251    :raises AceMQError: when ``opentelemetry-api`` is not installed
252    """
253
254    def __init__(
255        self,
256        *,
257        system: str = DEFAULT_SYSTEM,
258        tracer_provider: Any | None = None,
259        propagator: Any | None = None,
260    ) -> None:
261        try:
262            from opentelemetry import propagate, trace
263        except ImportError as missing:  # pragma: no cover - depends on the install
264            raise AceMQError(
265                "acemq: OpenTelemetryTracing needs opentelemetry-api, which is an optional "
266                'extra. Install it with pip install "acemq-amqp[opentelemetry]"'
267            ) from missing
268
269        self._trace: Any = trace
270        self._propagate: Any = propagate
271        self._propagator = propagator
272        self._system = system
273        self._tracer: Any = trace.get_tracer(
274            INSTRUMENTATION_NAME, tracer_provider=tracer_provider
275        )
276
277    # ----------------------------------------------------------------- wiring
278
279    def install(self, connection: Connection) -> Connection:
280        """Registers both interceptors on a connection.
281
282        :param connection: what to trace
283        :returns: the connection, so this reads as one expression
284        """
285        connection.intercept_publish(self.publish_interceptor())
286        connection.intercept_consume(self.consume_interceptor())
287        return connection
288
289    def publish_interceptor(self) -> PublishInterceptor:
290        """A span around every publish, and the trace context into the message.
291
292        :returns: an interceptor for
293            :meth:`~acemq_amqp.connection.Connection.intercept_publish`
294        """
295
296        async def traced(context: PublishContext, send: PublishNext) -> PublishResult:
297            destination = context.exchange or context.routing_key
298            with self._tracer.start_as_current_span(
299                destination + SPAN_PUBLISH_SUFFIX,
300                kind=self._trace.SpanKind.PRODUCER,
301                attributes={
302                    ATTR_SYSTEM: self._system,
303                    ATTR_DESTINATION: destination,
304                    ATTR_OPERATION: "publish",
305                    ATTR_MESSAGE_ID: context.envelope.id,
306                    ATTR_CONVERSATION_ID: context.envelope.correlation_id,
307                    ATTR_ROUTING_KEY: context.routing_key,
308                    ATTR_MESSAGE_TYPE: context.envelope.type,
309                },
310                # The adapter records the exception and sets the status
311                # itself, in one place, for every way a span can end badly. Left
312                # on, the SDK would record a second identical exception event on
313                # the way out and a trace would show every failure twice.
314                record_exception=False,
315                set_status_on_exception=False,
316            ) as span:
317                # Injected *inside* the span, so what the message carries is this
318                # span and not its parent. A consumer joining to the parent would
319                # produce a trace where the publish and the process are siblings,
320                # which is the wrong shape and hides which publish caused which
321                # delivery when a service publishes more than one message.
322                for name, value in self.propagation_headers().items():
323                    context.set_header(name, value)
324
325                try:
326                    result = await send(context)
327                except BaseException as failure:
328                    # The outcome is said here rather than inside ``_failed``,
329                    # which only knows that something threw. Java splits it the
330                    # same way — ``Scope.failed`` records the exception and the
331                    # status, and the caller, which knows what kind of operation
332                    # this was, says the word.
333                    self._outcome(
334                        span,
335                        OUTCOME_REFUSED
336                        if isinstance(failure, PublishingPausedError)
337                        else OUTCOME_FAILED,
338                    )
339                    self._failed(span, failure)
340                    raise
341                self._outcome(
342                    span, OUTCOME_CONFIRMED if result.confirmed else OUTCOME_PUBLISHED
343                )
344                if not result.routed:
345                    # Reached no queue at all, which the broker does not consider
346                    # an error and which is very often the whole problem.
347                    self._outcome(span, OUTCOME_UNROUTABLE)
348                    if result.return_reason:
349                        span.set_attribute(ATTR_REASON, result.return_reason)
350                return result
351
352        return traced
353
354    def consume_interceptor(self) -> ConsumeInterceptor:
355        """A span around every handler, parented by the publish that caused it.
356
357        **The span outlives the handler on purpose.** What the handler answered
358        is not what happened to the message: a handler asking for another
359        attempt when there are none left is dead-lettered, and a span ended on
360        the handler's answer says ``retried`` for a message nobody will ever
361        try again. So where the consumer has promised to say how the delivery
362        was settled — :meth:`~acemq_amqp.interceptors.ConsumeContext.when_settled`
363        — the span waits for that, takes its outcome from it, and ends. Where
364        nothing has promised, because the chain is being run by something other
365        than a consumer, it ends here with what the handler said.
366
367        The wait itself is not in the span. The consumer announces its decision
368        before it acts on it, so a five-second backoff spent holding the
369        delivery does not turn into a five-second handler.
370
371        :returns: an interceptor for
372            :meth:`~acemq_amqp.connection.Connection.intercept_consume`
373        """
374
375        async def traced(context: ConsumeContext, handle: ConsumeNext) -> Ack:
376            span = self._tracer.start_span(
377                context.queue + SPAN_PROCESS_SUFFIX,
378                context=self._parent_of(context.envelope),
379                kind=self._trace.SpanKind.CONSUMER,
380                attributes={
381                    ATTR_SYSTEM: self._system,
382                    ATTR_DESTINATION: context.queue,
383                    ATTR_OPERATION: "process",
384                    ATTR_MESSAGE_ID: context.envelope.id,
385                    ATTR_CONVERSATION_ID: context.envelope.correlation_id,
386                    ATTR_MESSAGE_TYPE: context.envelope.type,
387                    ATTR_ATTEMPT: context.envelope.attempt,
388                    # No routing key. It is a publish-side attribute in every
389                    # other library — Java's ``consumeStarted`` is not even given
390                    # one — and an attribute that exists on one library's process
391                    # spans and not on the other four's is worse than one that
392                    # exists nowhere: a query written against it silently returns
393                    # only the Python services.
394                },
395                # The adapter records the exception and sets the status
396                # itself, in one place, for every way a span can end badly. Left
397                # on, the SDK would record a second identical exception event on
398                # the way out and a trace would show every failure twice.
399                record_exception=False,
400                set_status_on_exception=False,
401            )
402            settled = context.when_settled(self._settlement_recorder(context, span))
403            with self._trace.use_span(
404                span,
405                # Ended by the settlement listener instead, which knows the
406                # outcome. Not ending it here is the whole point; not ending it
407                # anywhere would be a leak, which is why this is asked rather
408                # than assumed.
409                end_on_exit=not settled,
410                record_exception=False,
411                set_status_on_exception=False,
412            ):
413                try:
414                    ack = await handle(context)
415                except BaseException as failure:
416                    # Provisional, and usually overwritten: the consumer turns a
417                    # handler that raised into a retry request, and the
418                    # settlement listener below writes down what it actually
419                    # decided. It is set at all for the case where nothing is
420                    # driving the chain and this is the last word.
421                    self._outcome(span, OUTCOME_FAILED)
422                    self._failed(span, failure)
423                    raise
424                self._outcome(span, _ACK_OUTCOMES.get(ack.action, OUTCOME_ACKED))
425                if ack.error is not None:
426                    span.record_exception(ack.error)
427                return ack
428
429        return traced
430
431    def _settlement_recorder(
432        self, context: ConsumeContext, span: Any
433    ) -> SettlementListener:
434        """Writes what the consumer did onto the delivery's span, and ends it.
435
436        The event and the outcome together, because separately they mislead: an
437        outcome of ``dead_lettered`` with no event does not say why, and a
438        ``message.dead_lettered`` event on a span whose outcome still reads
439        ``retried`` is the bug this exists to fix.
440        """
441
442        def settled(settlement: Settlement) -> None:
443            try:
444                if settlement.delay is not None:
445                    span.add_event(
446                        EVENT_MESSAGE_RETRIED,
447                        _retry_attributes(
448                            context.queue, context.envelope, _millis(settlement.delay)
449                        ),
450                    )
451                elif settlement.dead_lettered:
452                    span.add_event(
453                        EVENT_MESSAGE_DEAD_LETTERED,
454                        _dead_letter_attributes(
455                            context.queue, context.envelope, settlement.reason or ""
456                        ),
457                    )
458                # Last, and overwriting whatever the handler's answer put there:
459                # this is the one that is true.
460                self._outcome(span, settlement.outcome)
461            finally:
462                span.end()
463
464        return settled
465
466    @contextmanager
467    def request_span(self, destination: str, envelope: Envelope) -> Iterator[Any]:
468        """A CLIENT span around a request that waits for its reply::
469
470            with tracing.request_span("pricing", envelope):
471                answer = await requester.ask(...)
472
473        CLIENT rather than PRODUCER because this one waits. Its duration is a
474        round trip and not a handover, and a backend that knows the difference
475        will show it against the responder's latency rather than the broker's.
476
477        The span ends with an outcome either way. ``answered`` when the reply
478        came back, ``timed_out`` when the deadline did — the two words Java's
479        ``MetricNames`` spells and its requester writes — and ``failed`` for
480        anything else. The first of those is the one worth having: a request span
481        that carried an outcome only when it went wrong left every successful
482        round trip with no outcome at all, which is not a thing a dashboard can
483        divide by.
484
485        A timeout is told apart by its type rather than by its message.
486        :class:`~acemq_amqp.patterns.RequestTimeoutError` is a
487        :class:`TimeoutError`, so is anything else that ran out of time —
488        :func:`asyncio.wait_for` included — and a caller who waits some other way
489        is understood without this module having to know how.
490
491        :param destination: what is being asked
492        :param envelope: the request's metadata
493        :returns: the span, so a caller can add to it
494        """
495        with self._tracer.start_as_current_span(
496            destination + SPAN_REQUEST_SUFFIX,
497            kind=self._trace.SpanKind.CLIENT,
498            attributes={
499                ATTR_SYSTEM: self._system,
500                ATTR_DESTINATION: destination,
501                ATTR_OPERATION: "request",
502                ATTR_MESSAGE_ID: envelope.id,
503                ATTR_CONVERSATION_ID: envelope.correlation_id,
504                ATTR_MESSAGE_TYPE: envelope.type,
505            },
506            # See the publish interceptor.
507            record_exception=False,
508            set_status_on_exception=False,
509        ) as span:
510            try:
511                yield span
512            except BaseException as failure:
513                self._outcome(
514                    span,
515                    OUTCOME_TIMED_OUT if _is_deadline(failure) else OUTCOME_FAILED,
516                )
517                # Recorded even for a timeout, and so is the error status: the
518                # outcome vocabulary is Java's, but a round trip that never got
519                # its answer is a failure from where the caller is standing and a
520                # green span would say otherwise.
521                self._failed(span, failure)
522                raise
523            self._outcome(span, OUTCOME_ANSWERED)
524
525    # ------------------------------------------------------------ propagation
526
527    def propagation_headers(self) -> dict[str, str]:
528        """The current trace context, as headers to put on a message.
529
530        A fresh carrier each time rather than writing into something the caller
531        owns, so nothing that was already on the message can be overwritten and
532        nothing from a previous message can be left behind.
533
534        Useful on its own for a message this library does not publish — one
535        going out through a different client, or into a database row that will be
536        published later by an outbox relay.
537
538        :returns: ``traceparent`` and ``tracestate``, or nothing at all when no
539            span is current
540        """
541        carrier: dict[str, str] = {}
542        self._injector().inject(carrier)
543        return carrier
544
545    def _parent_of(self, envelope: Envelope) -> Any:
546        """The context a delivery's span belongs under.
547
548        Read from the message's own headers, never from whatever happened to be
549        current. The delivery arrives on a connection's task, whose ambient
550        context is at best unrelated to the message and at worst the *previous*
551        message's — which produces a trace that looks joined up and joins the
552        wrong things together.
553        """
554        carrier = {
555            name: str(value)
556            for name, value in envelope.headers.items()
557            if name in (TRACEPARENT, TRACESTATE) and value is not None
558        }
559        if not carrier:
560            # No context on the message. Returning an empty Context rather than
561            # None starts a new trace, which is right: a message published by
562            # something that does not propagate is the root of its own trace,
563            # not a child of whatever this process was doing.
564            return self._empty_context()
565        return self._injector().extract(carrier, context=self._empty_context())
566
567    def _empty_context(self) -> Any:
568        from opentelemetry.context import Context
569
570        return Context()
571
572    def _injector(self) -> Any:
573        if self._propagator is not None:
574            return self._propagator
575        return self._propagate.get_global_textmap()
576
577    # ---------------------------------------------------------------- events
578
579    def outbox_publish_failed(self, destination: str, reason: str) -> None:
580        """Records that a relay could not publish a record it had committed.
581
582        :param destination: where it was going
583        :param reason: why it did not get there
584        """
585        self._event(
586            EVENT_OUTBOX_PUBLISH_FAILED,
587            {ATTR_DESTINATION: destination, ATTR_REASON: reason},
588        )
589
590    def outbox_published(self, destination: str, lag_ms: int) -> None:
591        """Records how far behind the outbox relay is running.
592
593        An attribute on the current span rather than an event: it is a
594        measurement of the publish that is happening, not a thing that happened
595        during it.
596
597        **Called by an application, never by this library**, and that is a
598        limitation rather than an oversight. Java's relay calls its equivalent
599        because its publish opens a span the attribute can land on;
600        :class:`~acemq_amqp.patterns.OutboxRelay` has no such span to write on.
601        It publishes with
602        :meth:`~acemq_amqp.connection.Connection.publish_raw`, which is beneath
603        the interceptor chain and so beneath the ``publish`` span, and it sweeps
604        on a task of its own where nothing else is current either. A hook wired
605        into the relay would compute a lag and hand it to a span that does not
606        exist.
607
608        The lag itself is *not* lost by that. The relay reports it on
609        ``acemq.outbox.lag`` through the connection's observer, tagged and
610        measured from the record's commit, because a metric needs no span to
611        land on. What this method adds is the number *on the trace*, beside the
612        work that caused it, which is worth having and which only a caller
613        holding a span can arrange.
614
615        So it stays where it works: call it from inside a span you are holding,
616        which is what a ``sweep()`` at the end of a request is::
617
618            with tracer.start_as_current_span("checkout"):
619                ...
620                await relay.sweep()
621                tracing.outbox_published("orders", lag_ms=elapsed)
622
623        :param destination: where the message went
624        :param lag_ms: how long the record sat before it was published
625        """
626        span = self._trace.get_current_span()
627        if span.is_recording():
628            span.set_attribute(ATTR_OUTBOX_LAG, lag_ms)
629
630    def pipeline_run_finished(
631        self, pipeline: str, step: str, outcome: str, age_ms: int
632    ) -> None:
633        """Records the end of a pipeline run.
634
635        :param pipeline: which pipeline
636        :param step: the step it ended on
637        :param outcome: how it ended
638        :param age_ms: how old the run was by then
639        """
640        self._event(
641            EVENT_PIPELINE_RUN_FINISHED,
642            {
643                ATTR_PIPELINE: pipeline,
644                ATTR_STEP: step,
645                ATTR_PIPELINE_OUTCOME: outcome,
646                ATTR_RUN_AGE: age_ms,
647            },
648        )
649
650    def message_retried(self, queue: str, envelope: Envelope, delay_ms: int) -> None:
651        """Records that a message will be tried again, and how long from now.
652
653        A consumer on a traced connection records this itself, on the delivery's
654        own span. This is for a retry something else arranged.
655
656        :param queue: where it came from
657        :param envelope: its metadata, for the attempt count
658        :param delay_ms: how long it waits first
659        """
660        self._event(EVENT_MESSAGE_RETRIED, _retry_attributes(queue, envelope, delay_ms))
661
662    def message_dead_lettered(self, queue: str, envelope: Envelope, reason: str) -> None:
663        """Records that a message was given up on.
664
665        A consumer on a traced connection records this itself, on the delivery's
666        own span. This is for a message something else gave up on.
667
668        :param queue: where it came from
669        :param envelope: its metadata, for the attempt count
670        :param reason: why
671        """
672        self._event(
673            EVENT_MESSAGE_DEAD_LETTERED, _dead_letter_attributes(queue, envelope, reason)
674        )
675
676    def _event(self, name: str, attributes: Mapping[str, Any]) -> None:
677        span = self._trace.get_current_span()
678        if span.is_recording():
679            span.add_event(name, dict(attributes))
680
681    # --------------------------------------------------------------- outcomes
682
683    def _outcome(self, span: Any, outcome: str) -> None:
684        span.set_attribute(ATTR_OUTCOME, outcome)
685        if outcome in ERROR_OUTCOMES:
686            span.set_status(self._trace.Status(self._trace.StatusCode.ERROR, outcome))
687
688    def _failed(self, span: Any, failure: BaseException) -> None:
689        """The exception and the error status, and deliberately not the outcome.
690
691        Java's ``Scope.failed`` draws the line in the same place. What threw is
692        knowable here; what the operation *was* is not, and a method that
693        stamped ``failed`` on every span it touched would overwrite the
694        ``timed_out`` a request had just earned. The caller says the word.
695        """
696        span.record_exception(failure)
697        span.set_status(
698            self._trace.Status(self._trace.StatusCode.ERROR, str(failure) or "failed")
699        )
700
701    def __repr__(self) -> str:
702        return f"OpenTelemetryTracing(system={self._system!r})"
703
704
705def _retry_attributes(queue: str, envelope: Envelope, delay_ms: int) -> dict[str, Any]:
706    """What a ``message.retried`` event says, wherever it is recorded from."""
707    return {
708        ATTR_DESTINATION: queue,
709        ATTR_RETRY_DELAY: delay_ms,
710        ATTR_ATTEMPT: envelope.attempt,
711    }
712
713
714def _dead_letter_attributes(
715    queue: str, envelope: Envelope, reason: str
716) -> dict[str, Any]:
717    """What a ``message.dead_lettered`` event says, wherever it is recorded from.
718
719    The reason is unbounded text, which a span tolerates and a metric does not.
720    """
721    return {
722        ATTR_DESTINATION: queue,
723        ATTR_REASON: reason,
724        ATTR_ATTEMPT: envelope.attempt,
725    }
726
727
728def _is_deadline(failure: BaseException) -> bool:
729    """Whether a request ran out of time rather than going wrong.
730
731    Both spellings, because they are only the same class from Python 3.11: on
732    3.10 :func:`asyncio.wait_for` raises an ``asyncio.TimeoutError`` that is not
733    the builtin, and a span that said ``failed`` on one interpreter and
734    ``timed_out`` on the next would be the least useful kind of difference.
735    """
736    return isinstance(failure, (TimeoutError, asyncio.TimeoutError))
737
738
739def _millis(delay: timedelta) -> int:
740    """A delay in whole milliseconds, which is the unit every attribute uses."""
741    return int(delay.total_seconds() * 1000)