Coverage for zenmqtt/instrumentation/monitors.py: 100%

1""" 

2Monitors which watch the state of the client and report problems by logs and 

3metrics. They don't change the behaviour of the client. 

4""" 

5 

6import asyncio 

7from logging import getLogger 

8from typing import Callable, Final, Optional 

9 

10from zenmqtt.instrumentation.metrics import MetricsCollector 

11from zenmqtt.mqtt.publish import MQTTMessage 

12 

13logger = getLogger(__name__) 

14 

15# number of messages waiting for the application, starting from which the 

16# client warns about a slow application 

17BUFFERED_MESSAGES_WARNING: Final[int] = 1000 

18# seconds between warnings about a slow application 

19_WARNING_INTERVAL: Final[float] = 10.0 

20# seconds between reports of the buffer size to metrics while it's above the 

21# warning threshold; it changes with every message, the last value is enough 

22_REPORT_INTERVAL: Final[float] = 0.1 

23 

24# seconds after which a not acknowledged message which holds back acks of 

25# later messages is logged 

26ACK_BLOCKED_WARNING: Final[float] = 30.0 

27 

28 

29def _now() -> float: 

30 return asyncio.get_running_loop().time() 

31 

32 

33class BufferMonitor: 

34 """ 

35 Watches the size of the buffer of incoming messages: while at least 

36 `threshold` messages wait for the application, a warning is logged and the 

37 size is reported by metrics. 

38 """ 

39 

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

41 self._metrics = metrics 

42 

43 self.threshold = BUFFERED_MESSAGES_WARNING 

44 # the buffer reached the threshold 

45 self._overloaded = False 

46 self._reported_at: Optional[float] = None 

47 self._warned_at: Optional[float] = None 

48 

49 def update(self, size: int) -> None: 

50 """ 

51 Called on every change of the buffer. Reports the size when it reaches 

52 the threshold, then at most every _REPORT_INTERVAL while it's above, 

53 and once when it falls below. 

54 """ 

55 if size < self.threshold: 

56 if self._overloaded: 

57 self._overloaded = False 

58 self._metrics.on_messages_buffered(size) 

59 

60 return 

61 

62 now = _now() 

63 

64 if ( 

65 not self._overloaded 

66 or self._reported_at is None 

67 or now - self._reported_at >= _REPORT_INTERVAL 

68 ): 

69 self._metrics.on_messages_buffered(size) 

70 self._reported_at = now 

71 

72 self._overloaded = True 

73 

74 if self._warned_at is None or now - self._warned_at >= _WARNING_INTERVAL: 

75 logger.warning( 

76 "mqtt_protocol.slow_consumer: %s incoming messages wait for the " 

77 "application, it doesn't read client.messages fast enough", 

78 size, 

79 ) 

80 self._warned_at = now 

81 

82 

83class AckBlockedMonitor: 

84 """ 

85 Acks are sent in the order of receiving (MQTT 5, 4.6), so one message which 

86 isn't acknowledged holds back the acks of later ones. Logs a warning once 

87 per message when it does so for ACK_BLOCKED_WARNING seconds; one timer for 

88 the current first message, not per message. 

89 """ 

90 

91 def __init__(self) -> None: 

92 self._message: Optional[MQTTMessage] = None 

93 self._timer: Optional[asyncio.TimerHandle] = None 

94 

95 def watch( 

96 self, 

97 message: MQTTMessage, 

98 received_at: float, 

99 waiting: Callable[[], Optional[int]], 

100 ) -> None: 

101 """ 

102 Called when `message` holds back acks of later messages. 

103 

104 :param received_at: loop time 

105 :param waiting: the number of messages which wait for `message`, None 

106 when it doesn't hold back acks anymore 

107 """ 

108 if message is self._message: 

109 return 

110 

111 if self._timer: 

112 self._timer.cancel() 

113 

114 self._message = message 

115 self._timer = asyncio.get_running_loop().call_at( 

116 received_at + ACK_BLOCKED_WARNING, self._warn, received_at, waiting 

117 ) 

118 

119 def _warn(self, received_at: float, waiting: Callable[[], Optional[int]]) -> None: 

120 self._timer = None 

121 message = self._message 

122 

123 # acknowledged meanwhile or received by the previous connection 

124 if message is None or (count := waiting()) is None: 

125 return 

126 

127 logger.warning( 

128 "mqtt_protocol.ack_blocked: the message pid:%s topic:%s isn't " 

129 "acknowledged for %.0f s; acks are sent in the order of receiving, " 

130 "so %s later messages wait for it", 

131 message.packet_identifier, 

132 message.topic, 

133 _now() - received_at, 

134 count, 

135 )