Coverage for src/taipanstack/resilience/watchdogs/resource_watcher.py: 100%

1""" 

2Resource watcher — monitors CPU and memory usage. 

3 

4When usage breaches configurable thresholds, invokes a callback 

5so the application can react (e.g. tighten rate limits). 

6""" 

7 

8import logging 

9import math 

10import time 

11from collections.abc import Callable 

12from dataclasses import dataclass 

13 

14from taipanstack.core.result import Err, Ok, Result 

15from taipanstack.resilience.watchdogs._base import BaseWatcher 

16 

17logger = logging.getLogger("taipanstack.resilience.watchdogs.resource") 

18 

19try: 

20 import psutil 

21 

22 _HAS_PSUTIL = True 

23except ImportError: 

24 psutil = None 

25 _HAS_PSUTIL = False 

26 

27 

28@dataclass(frozen=True) 

29class ResourceSnapshot: 

30 """Point-in-time snapshot of system resource usage. 

31 

32 Attributes: 

33 cpu_percent: Current CPU utilisation (0-100). 

34 memory_percent: Current memory utilisation (0-100). 

35 timestamp: Monotonic timestamp of the reading. 

36 

37 """ 

38 

39 cpu_percent: float 

40 memory_percent: float 

41 timestamp: float 

42 

43 

44def _validate_interval(interval: float) -> None: 

45 """Validate interval parameter.""" 

46 if ( 

47 not isinstance(interval, (int, float)) 

48 or not math.isfinite(interval) 

49 or interval <= 0 

50 ): 

51 raise ValueError("interval must be a finite positive number") 

52 

53 

54def _validate_threshold(name: str, value: float) -> None: 

55 """Validate a threshold parameter.""" 

56 if not isinstance(value, (int, float)) or not math.isfinite(value) or value < 0: 

57 raise ValueError(f"{name} must be a finite non-negative number") 

58 

59 

60def check_resources() -> Result[ResourceSnapshot, Exception]: 

61 """Take a one-shot resource reading. 

62 

63 Returns: 

64 ``Ok(ResourceSnapshot)`` on success, ``Err`` if psutil is 

65 unavailable. 

66 

67 """ 

68 if not _HAS_PSUTIL: 

69 return Err( 

70 ImportError( 

71 "psutil is required for resource monitoring. " 

72 "Install with: pip install taipanstack[resilience]", 

73 ), 

74 ) 

75 

76 cpu = psutil.cpu_percent(interval=0.1) # type: ignore[misc] 

77 mem = psutil.virtual_memory().percent # type: ignore[misc] 

78 return Ok( 

79 ResourceSnapshot( 

80 cpu_percent=cpu, # type: ignore[misc] 

81 memory_percent=mem, # type: ignore[misc] 

82 timestamp=time.monotonic(), 

83 ), 

84 ) 

85 

86 

87class ResourceWatcher(BaseWatcher): 

88 """Background watcher that monitors CPU and memory. 

89 

90 When either metric exceeds its configured threshold the 

91 ``on_threshold_breach`` callback is invoked with the resource 

92 name (``"cpu"`` or ``"memory"``) and the current value. 

93 

94 Args: 

95 interval: Seconds between checks. 

96 cpu_threshold: CPU percentage that triggers a breach. 

97 memory_threshold: Memory percentage that triggers a breach. 

98 on_threshold_breach: Optional callback ``(resource, value) -> None``. 

99 

100 Example: 

101 >>> watcher = ResourceWatcher( 

102 ... cpu_threshold=80.0, 

103 ... on_threshold_breach=lambda r, v: print(f"{r} at {v}%"), 

104 ... ) 

105 >>> await watcher.start() 

106 

107 """ 

108 

109 def __init__( 

110 self, 

111 *, 

112 interval: float = 5.0, 

113 cpu_threshold: float = 85.0, 

114 memory_threshold: float = 85.0, 

115 on_threshold_breach: Callable[[str, float], None] | None = None, 

116 ) -> None: 

117 """Initialize the resource watcher. 

118 

119 Args: 

120 interval: Seconds between checks. 

121 cpu_threshold: CPU percentage that triggers a breach. 

122 memory_threshold: Memory percentage that triggers a breach. 

123 on_threshold_breach: Optional breach callback. 

124 

125 """ 

126 _validate_interval(interval) 

127 _validate_threshold("cpu_threshold", cpu_threshold) 

128 _validate_threshold("memory_threshold", memory_threshold) 

129 

130 super().__init__(interval=interval) 

131 self._cpu_threshold = cpu_threshold 

132 self._memory_threshold = memory_threshold 

133 self._on_threshold_breach = on_threshold_breach 

134 

135 async def start(self) -> Result[None, Exception]: 

136 """Start the resource watcher. 

137 

138 Returns: 

139 ``Err`` if psutil is not installed, otherwise delegates 

140 to ``BaseWatcher.start()``. 

141 

142 """ 

143 if not _HAS_PSUTIL: 

144 return Err( 

145 ImportError( 

146 "psutil is required for ResourceWatcher. " 

147 "Install with: pip install taipanstack[resilience]", 

148 ), 

149 ) 

150 return await super().start() 

151 

152 def _check_threshold(self, name: str, value: float, threshold: float) -> None: 

153 if value >= threshold: 

154 logger.warning( 

155 "%s threshold breached: %.1f%% >= %.1f%%", 

156 name.capitalize(), 

157 value, 

158 threshold, 

159 ) 

160 if self._on_threshold_breach is not None: 

161 self._on_threshold_breach(name, value) 

162 

163 def _handle_snapshot(self, snapshot: ResourceSnapshot) -> None: 

164 self._check_threshold("cpu", snapshot.cpu_percent, self._cpu_threshold) 

165 self._check_threshold("memory", snapshot.memory_percent, self._memory_threshold) 

166 

167 async def _run(self) -> None: 

168 """Execute a single resource check cycle.""" 

169 result = check_resources() 

170 if isinstance(result, Ok): 

171 self._handle_snapshot(result.ok_value) 

172 else: 

173 logger.error("Resource check failed: %s", result.err_value)