Coverage for agentos/monitoring/alerts.py: 58%

1"""AgentOS monitoring — alert rules and webhook notification dispatcher.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import time 

7import urllib.request 

8from collections.abc import Callable 

9from dataclasses import dataclass, field 

10from enum import StrEnum 

11 

12 

13class AlertSeverity(StrEnum): 

14 """告警实例。""" 

15 

16 """告警严重级别。""" 

17 

18 CRITICAL = "critical" 

19 WARNING = "warning" 

20 INFO = "info" 

21 

22 

23class AlertState(StrEnum): 

24 """告警状态。""" 

25 

26 FIRING = "firing" 

27 RESOLVED = "resolved" 

28 

29 

30@dataclass 

31class AlertRule: 

32 """告警规则。""" 

33 

34 name: str 

35 description: str 

36 severity: AlertSeverity = AlertSeverity.WARNING 

37 condition: Callable[[], bool] | None = None 

38 cooldown_seconds: int = 300 

39 _last_fired: float = field(default=0.0, repr=False) 

40 

41 def evaluate(self) -> bool: 

42 if not self.condition: 

43 return False 

44 now = time.time() 

45 if now - self._last_fired < self.cooldown_seconds: 

46 return False 

47 result = self.condition() 

48 if result: 

49 self._last_fired = now 

50 return result 

51 

52 

53@dataclass 

54class Alert: 

55 rule_name: str 

56 severity: AlertSeverity 

57 message: str 

58 state: AlertState = AlertState.FIRING 

59 timestamp: float = field(default_factory=time.time) 

60 labels: dict = field(default_factory=dict) 

61 

62 def to_dict(self) -> dict: 

63 return { 

64 "rule_name": self.rule_name, 

65 "severity": self.severity.value, 

66 "message": self.message, 

67 "state": self.state.value, 

68 "timestamp": self.timestamp, 

69 "labels": self.labels, 

70 } 

71 

72 def to_json(self) -> str: 

73 return json.dumps(self.to_dict()) 

74 

75 

76@dataclass 

77class MonitoringConfig: 

78 """监控配置。""" 

79 

80 enabled: bool = True 

81 evaluation_interval: int = 60 

82 max_alerts_per_interval: int = 10 

83 

84 

85@dataclass 

86class WebhookConfig: 

87 """Webhook 配置。""" 

88 

89 url: str = "" 

90 method: str = "POST" 

91 headers: dict = field(default_factory=dict) 

92 timeout: float = 5.0 

93 retry_count: int = 3 

94 

95 

96class WebhookDispatcher: 

97 """Dispatches Alerts to configured webhook endpoints.""" 

98 

99 def __init__(self, config: WebhookConfig | None = None): 

100 self.config = config or WebhookConfig() 

101 

102 def send(self, alert: Alert) -> bool: 

103 if not self.config.url: 

104 return False 

105 payload = json.dumps(alert.to_dict()).encode("utf-8") 

106 for attempt in range(self.config.retry_count + 1): 

107 try: 

108 req = urllib.request.Request( 

109 self.config.url, 

110 data=payload, 

111 headers=self.config.headers, 

112 method=self.config.method, 

113 ) 

114 with urllib.request.urlopen(req, timeout=self.config.timeout) as resp: 

115 return resp.status < 400 

116 except Exception: 

117 if attempt == self.config.retry_count: 

118 return False 

119 time.sleep(1.0 * (attempt + 1)) 

120 return False 

121 

122 

123class AlertEvaluator: 

124 """Evaluates AlertRules and generates Alerts.""" 

125 

126 def __init__(self, config: MonitoringConfig | None = None): 

127 self.config = config or MonitoringConfig() 

128 self.rules: list[AlertRule] = [] 

129 

130 def add_rule(self, rule: AlertRule): 

131 self.rules.append(rule) 

132 

133 def evaluate(self) -> list[Alert]: 

134 if not self.config.enabled: 

135 return [] 

136 alerts: list[Alert] = [] 

137 count = 0 

138 for rule in self.rules: 

139 if count >= self.config.max_alerts_per_interval: 

140 break 

141 if rule.evaluate(): 

142 alerts.append( 

143 Alert( 

144 rule_name=rule.name, 

145 severity=rule.severity, 

146 message=f"Alert: {rule.description}", 

147 ) 

148 ) 

149 count += 1 

150 return alerts