Coverage for zenmqtt/instrumentation/metrics.py: 98%

1from typing import Any, Callable, Final, Optional, Sequence 

2 

3from zenmqtt.instrumentation.guard import HookErrors 

4from zenmqtt.mqtt.packet import PacketType 

5 

6 

7class MetricsCollector: 

8 """ 

9 Receives events of the client to build metrics from them. 

10 

11 The default implementation does nothing, subclasses override only the 

12 events they need. Methods are called from the event loop, so they must 

13 be fast and must not block. 

14 

15 An exception in a method doesn't affect the client: it's logged with the 

16 traceback the first time, then at most once a minute with the number of 

17 errors. Methods are taken from the collector when the client is created. 

18 

19 Durations are in seconds. 

20 """ 

21 

22 def on_connect(self, reason_code: int, duration: float) -> None: 

23 """CONNACK is received, duration is measured from sending CONNECT""" 

24 

25 def on_connection_closed(self, lost: bool) -> None: 

26 """ 

27 A connection which was established by CONNACK is closed; 

28 lost is False only if the client called disconnect 

29 """ 

30 

31 def on_packet_sent(self, packet_type: PacketType, size: int) -> None: 

32 """A packet is written to the connection, size includes the fixed header""" 

33 

34 def on_packet_received(self, packet_type: PacketType, size: int) -> None: 

35 """A packet is read from the connection, size includes the fixed header""" 

36 

37 def on_publish_completed(self, qos: int, reason_code: int, duration: float) -> None: 

38 """ 

39 QoS 0: PUBLISH is written, duration is the time of writing; 

40 QoS 1/2: the final acknowledgement is received, duration is measured 

41 from sending PUBLISH 

42 """ 

43 

44 def on_message_received(self, qos: int, duplicate: bool) -> None: 

45 """ 

46 An incoming PUBLISH is handled; duplicate is True for a QoS 2 message 

47 which was already delivered, and for a QoS 1/2 message which the 

48 server sent again before the ack; duplicates aren't delivered again 

49 """ 

50 

51 def on_messages_buffered(self, count: int) -> None: 

52 """ 

53 Number of incoming messages which wait for the application, reported 

54 when it reaches the warning threshold of the client (the application 

55 reads messages too slowly), then at most every 0.1 s while it's above, 

56 and once when it falls below 

57 """ 

58 

59 def on_send_quota_wait(self, duration: float) -> None: 

60 """A QoS 1/2 publish waited because the server's "Receive Maximum" was reached""" 

61 

62 def on_messages_resent(self, count: int) -> None: 

63 """Pending messages of the session are re-sent after reconnect""" 

64 

65 def on_ping(self, duration: float) -> None: 

66 """PINGRESP is received, duration is measured from sending PINGREQ""" 

67 

68 def on_reconnect_attempt(self, attempt: int, delay: float) -> None: 

69 """ 

70 Automatic reconnect starts an attempt (1, 2, ...) after the delay; 

71 a successful attempt is followed by on_connect 

72 """ 

73 

74 def on_reconnect_gave_up(self) -> None: 

75 """Automatic reconnect stopped: no attempts left or a non-retryable error""" 

76 

77 def on_ping_timeout(self) -> None: 

78 """Keep alive closes the connection because PINGRESP didn't come in time""" 

79 

80 def on_subscribe_completed( 

81 self, reason_codes: Sequence[int], duration: float 

82 ) -> None: 

83 """SUBACK is received, duration is measured from sending SUBSCRIBE""" 

84 

85 def on_unsubscribe_completed( 

86 self, reason_codes: Sequence[int], duration: float 

87 ) -> None: 

88 """UNSUBACK is received, duration is measured from sending UNSUBSCRIBE""" 

89 

90 

91class _GuardedMetrics(MetricsCollector): 

92 """ 

93 Calls the hooks of a collector, an exception in a hook is logged and never 

94 reaches the client. Hooks which the collector doesn't override stay the 

95 no-op methods of MetricsCollector, without a wrapper. 

96 """ 

97 

98 def __init__(self, metrics: MetricsCollector) -> None: 

99 self.wrapped = metrics 

100 self._errors = HookErrors(type(metrics).__name__, "mqtt_metrics.hook_error") 

101 

102 for name in _HOOKS: 

103 hook = getattr(metrics, name) 

104 

105 if getattr(hook, "__func__", None) is not getattr(MetricsCollector, name): 

106 setattr(self, name, self._guard(name, hook)) 

107 

108 @property 

109 def errors(self) -> dict[str, int]: 

110 """Number of exceptions per hook.""" 

111 return self._errors.errors 

112 

113 def _guard(self, name: str, hook: Callable[..., None]) -> Callable[..., None]: 

114 report = self._errors.report 

115 

116 def guarded(*args: Any, **kwargs: Any) -> None: 

117 try: 

118 hook(*args, **kwargs) 

119 except Exception as exc: 

120 report(name, exc) 

121 

122 return guarded 

123 

124 

125_HOOKS: Final[tuple[str, ...]] = tuple( 

126 name for name in vars(MetricsCollector) if name.startswith("on_") 

127) 

128 

129 

130def guard_metrics(metrics: Optional[MetricsCollector]) -> MetricsCollector: 

131 """ 

132 Wraps a collector, so its exceptions never reach the client; a guarded 

133 collector is returned as it is. 

134 """ 

135 if metrics is None: 

136 return MetricsCollector() 

137 

138 # nothing to guard: the default collector, or one guarded already 

139 if type(metrics) is MetricsCollector or isinstance(metrics, _GuardedMetrics): 

140 return metrics 

141 

142 return _GuardedMetrics(metrics)