Coverage for zenmqtt/instrumentation/opentelemetry.py: 97%
135 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-12 21:55 +0000
1"""
2OpenTelemetry metrics and tracing for the client, requires
3`pip install zenmqtt[otel]`.
5Messaging metrics and spans follow the OpenTelemetry semantic conventions, MQTT
6specific metrics use the `zenmqtt.` prefix. They can be exported with any
7OpenTelemetry exporter: OTLP, Prometheus, console, etc.
9Topics are not used as metric attributes: the number of topics isn't limited,
10so it would make the number of time series unlimited too. For the same reason
11span names don't contain the topic, it's the attribute
12`messaging.destination.name` of the span.
13"""
15import asyncio
16from contextlib import contextmanager
17from typing import Iterator, Mapping, Optional, Sequence, Tuple, cast
19from opentelemetry import context, propagate, trace
20from opentelemetry.metrics import Meter, get_meter
21from opentelemetry.propagators.textmap import Getter, TextMapPropagator
22from opentelemetry.trace import Span, SpanKind, Status, StatusCode, TracerProvider
23from opentelemetry.util.types import AttributeValue
25from zenmqtt.instrumentation.metrics import MetricsCollector
26from zenmqtt.instrumentation.tracing import PublishTrace, Tracing
27from zenmqtt.mqtt.packet import PacketType
28from zenmqtt.mqtt.publish import (
29 MQTTMessage,
30 PublishAcknowledgement,
31 PublishProperties,
32)
33from zenmqtt.mqtt.reason_codes import FAILURE_REASON_CODE
35# recommended by the semantic conventions for durations in seconds
36# packet sizes in bytes
37_SIZE_BUCKETS = (16, 64, 256, 1024, 4096, 16384, 65536, 262144, 1048576)
39_DURATION_BUCKETS = (
40 0.005,
41 0.01,
42 0.025,
43 0.05,
44 0.075,
45 0.1,
46 0.25,
47 0.5,
48 0.75,
49 1.0,
50 2.5,
51 5.0,
52 7.5,
53 10.0,
54)
57class OpenTelemetryMetrics(MetricsCollector):
58 def __init__(
59 self,
60 meter: Optional[Meter] = None,
61 attributes: Optional[Mapping[str, AttributeValue]] = None,
62 ) -> None:
63 """
64 :param meter: meter to create instruments, the global one by default
65 :param attributes: added to every measurement, e.g. a client name
66 """
67 meter = meter or get_meter("zenmqtt")
69 self._attributes: dict[str, AttributeValue] = {
70 "messaging.system": "mqtt",
71 **(attributes or {}),
72 }
74 self._operation_duration = meter.create_histogram(
75 "messaging.client.operation.duration",
76 unit="s",
77 description="Duration of publish, subscribe and unsubscribe operations",
78 explicit_bucket_boundaries_advisory=_DURATION_BUCKETS,
79 )
80 self._sent_messages = meter.create_counter(
81 "messaging.client.sent.messages",
82 unit="{message}",
83 description="Number of published messages",
84 )
85 self._consumed_messages = meter.create_counter(
86 "messaging.client.consumed.messages",
87 unit="{message}",
88 description="Number of messages delivered to the application",
89 )
91 self._connect_duration = meter.create_histogram(
92 "zenmqtt.connect.duration",
93 unit="s",
94 description="Time from sending CONNECT to receiving CONNACK",
95 explicit_bucket_boundaries_advisory=_DURATION_BUCKETS,
96 )
97 self._active_connections = meter.create_up_down_counter(
98 "zenmqtt.connections.active",
99 unit="{connection}",
100 description="Number of established connections",
101 )
102 self._closed_connections = meter.create_counter(
103 "zenmqtt.connections.closed",
104 unit="{connection}",
105 description="Number of closed connections",
106 )
107 # one histogram instead of counters of packets and bytes: every
108 # measurement has a fixed cost in the SDK, and a histogram gives both,
109 # the number of packets (count) and of bytes (sum)
110 self._sent_packets = meter.create_histogram(
111 "zenmqtt.packets.sent",
112 unit="By",
113 description="Sizes of sent packets: count of packets, sum of bytes",
114 explicit_bucket_boundaries_advisory=_SIZE_BUCKETS,
115 )
116 self._received_packets = meter.create_histogram(
117 "zenmqtt.packets.received",
118 unit="By",
119 description="Sizes of received packets: count of packets, sum of bytes",
120 explicit_bucket_boundaries_advisory=_SIZE_BUCKETS,
121 )
123 # attributes are built once, not for every measurement
124 self._packet_attributes: dict[PacketType, dict[str, AttributeValue]] = {
125 packet_type: {**self._attributes, "mqtt.packet.type": packet_type.name}
126 for packet_type in PacketType
127 }
128 self._received_attributes: dict[int, dict[str, AttributeValue]] = {
129 qos: {
130 **self._attributes,
131 "messaging.operation.name": "receive",
132 "messaging.operation.type": "receive",
133 "mqtt.qos": qos,
134 }
135 for qos in (0, 1, 2)
136 }
137 self._published_attributes: dict[
138 tuple[int, int], dict[str, AttributeValue]
139 ] = {}
140 self._buffered_messages = meter.create_gauge(
141 "zenmqtt.messages.buffered",
142 unit="{message}",
143 description=(
144 "Number of incoming messages waiting for the application, "
145 "reported while it's above the warning threshold"
146 ),
147 )
148 self._send_quota_wait = meter.create_histogram(
149 "zenmqtt.send_quota.wait.duration",
150 unit="s",
151 description='Time a publish waited because of the server "Receive Maximum"',
152 explicit_bucket_boundaries_advisory=_DURATION_BUCKETS,
153 )
154 self._resent_messages = meter.create_counter(
155 "zenmqtt.messages.resent",
156 unit="{message}",
157 description="Number of pending messages re-sent after reconnect",
158 )
159 self._duplicated_messages = meter.create_counter(
160 "zenmqtt.messages.duplicated",
161 unit="{message}",
162 description="Number of duplicated QoS 2 messages, not delivered again",
163 )
164 self._ping_duration = meter.create_histogram(
165 "zenmqtt.ping.duration",
166 unit="s",
167 description="Time from sending PINGREQ to receiving PINGRESP",
168 explicit_bucket_boundaries_advisory=_DURATION_BUCKETS,
169 )
170 self._reconnect_attempts = meter.create_counter(
171 "zenmqtt.reconnect.attempts",
172 unit="{attempt}",
173 description="Number of automatic reconnect attempts",
174 )
175 self._reconnect_give_ups = meter.create_counter(
176 "zenmqtt.reconnect.give_ups",
177 unit="{event}",
178 description="Number of times automatic reconnect stopped for good",
179 )
180 self._ping_timeouts = meter.create_counter(
181 "zenmqtt.ping.timeouts",
182 unit="{timeout}",
183 description="Number of connections closed by keep alive",
184 )
186 def on_connect(self, reason_code: int, duration: float) -> None:
187 attributes = self._with_reason_code(reason_code)
189 self._connect_duration.record(duration, attributes)
191 if reason_code < FAILURE_REASON_CODE:
192 self._active_connections.add(1, self._attributes)
194 def on_connection_closed(self, lost: bool) -> None:
195 self._active_connections.add(-1, self._attributes)
196 self._closed_connections.add(
197 1, {**self._attributes, "zenmqtt.connection.lost": lost}
198 )
200 def on_packet_sent(self, packet_type: PacketType, size: int) -> None:
201 self._sent_packets.record(size, self._packet_attributes[packet_type])
203 def on_packet_received(self, packet_type: PacketType, size: int) -> None:
204 self._received_packets.record(size, self._packet_attributes[packet_type])
206 def on_publish_completed(self, qos: int, reason_code: int, duration: float) -> None:
207 if (attributes := self._published_attributes.get((qos, reason_code))) is None: 207 ↛ 216line 207 didn't jump to line 216 because the condition on line 207 was always true
208 attributes = {
209 **self._with_reason_code(reason_code),
210 "messaging.operation.name": "publish",
211 "messaging.operation.type": "send",
212 "mqtt.qos": qos,
213 }
214 self._published_attributes[(qos, reason_code)] = attributes
216 self._operation_duration.record(duration, attributes)
217 self._sent_messages.add(1, attributes)
219 def on_message_received(self, qos: int, duplicate: bool) -> None:
220 attributes = self._received_attributes[qos]
222 if duplicate:
223 self._duplicated_messages.add(1, attributes)
224 else:
225 self._consumed_messages.add(1, attributes)
227 def on_messages_buffered(self, count: int) -> None:
228 self._buffered_messages.set(count, self._attributes)
230 def on_send_quota_wait(self, duration: float) -> None:
231 self._send_quota_wait.record(duration, self._attributes)
233 def on_messages_resent(self, count: int) -> None:
234 self._resent_messages.add(count, self._attributes)
236 def on_ping(self, duration: float) -> None:
237 self._ping_duration.record(duration, self._attributes)
239 def on_reconnect_attempt(self, attempt: int, delay: float) -> None:
240 self._reconnect_attempts.add(1, self._attributes)
242 def on_reconnect_gave_up(self) -> None:
243 self._reconnect_give_ups.add(1, self._attributes)
245 def on_ping_timeout(self) -> None:
246 self._ping_timeouts.add(1, self._attributes)
248 def on_subscribe_completed(
249 self, reason_codes: Sequence[int], duration: float
250 ) -> None:
251 self._record_command("subscribe", reason_codes, duration)
253 def on_unsubscribe_completed(
254 self, reason_codes: Sequence[int], duration: float
255 ) -> None:
256 self._record_command("unsubscribe", reason_codes, duration)
258 def _record_command(
259 self, operation: str, reason_codes: Sequence[int], duration: float
260 ) -> None:
261 attributes: dict[str, AttributeValue] = {
262 **self._attributes,
263 "messaging.operation.name": operation,
264 }
266 # one reason code per topic, the operation failed if any topic failed
267 if failed := [code for code in reason_codes if code >= FAILURE_REASON_CODE]:
268 attributes["error.type"] = _format_reason_code(failed[0])
270 self._operation_duration.record(duration, attributes)
272 def _with_reason_code(self, reason_code: int) -> dict[str, AttributeValue]:
273 attributes: dict[str, AttributeValue] = {
274 **self._attributes,
275 "mqtt.reason_code": reason_code,
276 }
278 if reason_code >= FAILURE_REASON_CODE:
279 attributes["error.type"] = _format_reason_code(reason_code)
281 return attributes
284def _format_reason_code(reason_code: int) -> str:
285 return f"0x{reason_code:02X}"
288UserProperties = Sequence[Tuple[str, str]]
291# mypy runs without site packages, so Getter is Any for it
292class _UserPropertiesGetter(Getter[UserProperties]): # type: ignore[misc,unused-ignore]
293 """Reads the trace context from MQTT 5 user properties, (key, value) pairs."""
295 def get(self, carrier: UserProperties, key: str) -> Optional[list[str]]:
296 return [value for name, value in carrier if name == key] or None
298 def keys(self, carrier: UserProperties) -> list[str]:
299 return [name for name, _ in carrier]
302_USER_PROPERTIES_GETTER = _UserPropertiesGetter()
305class _PublishSpan(PublishTrace):
306 def __init__(self, properties: PublishProperties, span: Span) -> None:
307 super().__init__(properties)
308 self._span = span
310 def end(
311 self,
312 result: Optional[PublishAcknowledgement],
313 error: Optional[BaseException],
314 ) -> None:
315 span = self._span
317 # a cancelled QoS 1/2 call doesn't cancel the delivery, it's not an error
318 if error is not None and not isinstance(error, asyncio.CancelledError):
319 span.record_exception(error)
320 span.set_attribute("error.type", type(error).__qualname__)
321 span.set_status(Status(StatusCode.ERROR, str(error)))
322 elif result is not None:
323 span.set_attribute("mqtt.reason_code", result.reason_code)
325 if result.reason_code >= FAILURE_REASON_CODE: 325 ↛ 331line 325 didn't jump to line 331 because the condition on line 325 was always true
326 span.set_attribute(
327 "error.type", _format_reason_code(result.reason_code)
328 )
329 span.set_status(Status(StatusCode.ERROR))
331 span.end()
334class OpenTelemetryTracing(Tracing):
335 """
336 Spans of the messaging semantic conventions:
338 - "publish" (PRODUCER) for every publish call: until the packet is written
339 for QoS 0, until the final acknowledgement for QoS 1/2. Its context is
340 injected into the user properties of PUBLISH.
341 - "process" (CONSUMER) for MQTTClient.trace(): the context from the
342 user properties of the message is its parent, so the processing
343 continues the trace of the producer.
344 """
346 def __init__(
347 self,
348 tracer_provider: Optional[TracerProvider] = None,
349 propagator: Optional[TextMapPropagator] = None,
350 attributes: Optional[Mapping[str, AttributeValue]] = None,
351 ) -> None:
352 """
353 :param tracer_provider: the global one by default
354 :param propagator: format of the trace context in user properties, the
355 global one by default (W3C traceparent and baggage)
356 :param attributes: added to every span, e.g. a client name
357 """
358 self._tracer = trace.get_tracer("zenmqtt", tracer_provider=tracer_provider)
359 self._propagator = propagator
361 self._attributes: dict[str, AttributeValue] = {
362 "messaging.system": "mqtt",
363 **(attributes or {}),
364 }
366 def start_publish(
367 self,
368 client_id: str,
369 topic: str,
370 payload: bytes,
371 qos: int,
372 retain: bool,
373 properties: PublishProperties,
374 ) -> PublishTrace:
375 span = self._tracer.start_span(
376 "publish",
377 kind=SpanKind.PRODUCER,
378 attributes={
379 **self._attributes,
380 "messaging.operation.name": "publish",
381 "messaging.operation.type": "send",
382 "messaging.destination.name": topic,
383 "messaging.client.id": client_id,
384 "messaging.message.body.size": len(payload),
385 "mqtt.qos": qos,
386 "mqtt.retain": retain,
387 },
388 )
390 carrier: dict[str, str] = {}
391 self._get_propagator().inject(carrier, trace.set_span_in_context(span))
393 if carrier: 393 ↛ 407line 393 didn't jump to line 407 because the condition on line 393 was always true
394 # the context replaces a context of the application, e.g. of a
395 # message which is forwarded
396 user_properties = [
397 (name, value)
398 for name, value in properties.get("user_property", ())
399 if name not in carrier
400 ]
401 user_properties.extend(carrier.items())
403 properties = cast(
404 PublishProperties, {**properties, "user_property": user_properties}
405 )
407 return _PublishSpan(properties, span)
409 @contextmanager
410 def trace(self, client_id: str, message: MQTTMessage) -> Iterator[None]:
411 # without a context in the message, the current span is the parent
412 parent = self._get_propagator().extract(
413 message.properties.get("user_property", ()),
414 context.get_current(),
415 getter=_USER_PROPERTIES_GETTER,
416 )
418 with self._tracer.start_as_current_span(
419 "process",
420 context=parent,
421 kind=SpanKind.CONSUMER,
422 attributes={
423 **self._attributes,
424 "messaging.operation.name": "process",
425 "messaging.operation.type": "process",
426 "messaging.destination.name": message.topic,
427 "messaging.client.id": client_id,
428 "messaging.message.body.size": len(message.payload),
429 "mqtt.qos": message.qos,
430 "mqtt.retain": message.retain,
431 "mqtt.dup": message.dup,
432 },
433 ):
434 yield
436 def _get_propagator(self) -> TextMapPropagator:
437 # the global propagator may be configured after the client is created
438 return self._propagator or propagate.get_global_textmap()