Coverage for python/pyairflowtester/dependency_intelligence/observability.py: 84%
182 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 20:43 +0530
1"""Observability and monitoring engines (Phase 4: Weeks 13-16)."""
3import json
4import logging
5from dataclasses import asdict, dataclass
6from datetime import datetime, timedelta
7from enum import Enum
8from typing import Any, Dict, List, Optional
10from .models import DependencyGraph, NodeSeverity
12logger = logging.getLogger(__name__)
15class MetricType(Enum):
16 """Types of metrics to track."""
18 EXECUTION_TIME = "execution_time"
19 FAILURE_COUNT = "failure_count"
20 DATA_VOLUME = "data_volume"
21 RESOURCE_USAGE = "resource_usage"
22 QUALITY_SCORE = "quality_score"
25@dataclass
26class Metric:
27 """Individual metric observation."""
29 metric_type: MetricType
30 node_id: str
31 value: float
32 timestamp: datetime
33 tags: Dict[str, str]
36@dataclass
37class Alert:
38 """Alert for detected issues."""
40 alert_id: str
41 alert_type: str # threshold_exceeded, anomaly_detected, sla_violated
42 node_id: str
43 severity: NodeSeverity
44 message: str
45 metric_value: Optional[float]
46 threshold: Optional[float]
47 created_at: datetime
48 resolved_at: Optional[datetime] = None
51@dataclass
52class ExecutionEvent:
53 """Record of node execution."""
55 event_id: str
56 node_id: str
57 status: str # success, failure, timeout
58 duration_ms: int
59 start_time: datetime
60 end_time: datetime
61 error_message: Optional[str] = None
62 tags: Dict[str, str] = None
65class MetricsCollector:
66 """Collect and aggregate metrics from nodes."""
68 def __init__(self):
69 self.metrics: List[Metric] = []
70 self.retention_days = 30
72 def record_metric(
73 self, metric_type: MetricType, node_id: str, value: float, tags: Dict[str, str] = None
74 ) -> Metric:
75 """Record a metric observation."""
76 metric = Metric(
77 metric_type=metric_type,
78 node_id=node_id,
79 value=value,
80 timestamp=datetime.utcnow(),
81 tags=tags or {},
82 )
83 self.metrics.append(metric)
84 return metric
86 def get_metrics_for_node(
87 self, node_id: str, metric_type: Optional[MetricType] = None, hours: int = 24
88 ) -> List[Metric]:
89 """Get recent metrics for a node."""
90 cutoff = datetime.utcnow() - timedelta(hours=hours)
92 return [
93 m
94 for m in self.metrics
95 if m.node_id == node_id
96 and m.timestamp >= cutoff
97 and (metric_type is None or m.metric_type == metric_type)
98 ]
100 def calculate_statistics(
101 self, node_id: str, metric_type: MetricType, hours: int = 24
102 ) -> Dict[str, float]:
103 """Calculate statistics for a metric."""
104 metrics = self.get_metrics_for_node(node_id, metric_type, hours)
106 if not metrics:
107 return {
108 "count": 0,
109 "min": 0.0,
110 "max": 0.0,
111 "avg": 0.0,
112 "p95": 0.0,
113 "p99": 0.0,
114 }
116 values = sorted([m.value for m in metrics])
117 count = len(values)
119 return {
120 "count": count,
121 "min": float(min(values)),
122 "max": float(max(values)),
123 "avg": float(sum(values) / count),
124 "p95": float(values[int(count * 0.95)]) if count > 0 else 0.0,
125 "p99": float(values[int(count * 0.99)]) if count > 0 else 0.0,
126 }
128 def cleanup_old_metrics(self) -> int:
129 """Remove metrics older than retention period."""
130 cutoff = datetime.utcnow() - timedelta(days=self.retention_days)
131 old_count = len([m for m in self.metrics if m.timestamp < cutoff])
133 self.metrics = [m for m in self.metrics if m.timestamp >= cutoff]
134 return old_count
137class AlertManager:
138 """Manage alerts for threshold violations and anomalies."""
140 def __init__(self, graph: DependencyGraph):
141 self.graph = graph
142 self.alerts: List[Alert] = []
143 self.thresholds: Dict[str, Dict[str, float]] = {}
145 def set_threshold(
146 self, node_id: str, metric_type: str, warning: float, critical: float
147 ) -> None:
148 """Set threshold for a metric."""
149 if node_id not in self.thresholds: 149 ↛ 152line 149 didn't jump to line 152 because the condition on line 149 was always true
150 self.thresholds[node_id] = {}
152 self.thresholds[node_id][metric_type] = {
153 "warning": warning,
154 "critical": critical,
155 }
157 def check_threshold(self, node_id: str, metric_type: str, value: float) -> Optional[Alert]:
158 """Check if metric exceeds threshold."""
159 if node_id not in self.thresholds: 159 ↛ 160line 159 didn't jump to line 160 because the condition on line 159 was never true
160 return None
162 thresholds = self.thresholds[node_id].get(metric_type)
163 if not thresholds: 163 ↛ 164line 163 didn't jump to line 164 because the condition on line 163 was never true
164 return None
166 node = self.graph.nodes.get(node_id)
167 if not node: 167 ↛ 168line 167 didn't jump to line 168 because the condition on line 167 was never true
168 return None
170 alert = None
171 if value > thresholds["critical"]:
172 alert = Alert(
173 alert_id=f"{node_id}_{metric_type}_{datetime.utcnow().timestamp()}",
174 alert_type="threshold_exceeded",
175 node_id=node_id,
176 severity=NodeSeverity.CRITICAL,
177 message=f"{metric_type} exceeded critical threshold",
178 metric_value=value,
179 threshold=thresholds["critical"],
180 created_at=datetime.utcnow(),
181 )
182 elif value > thresholds["warning"]:
183 alert = Alert(
184 alert_id=f"{node_id}_{metric_type}_{datetime.utcnow().timestamp()}",
185 alert_type="threshold_exceeded",
186 node_id=node_id,
187 severity=NodeSeverity.HIGH,
188 message=f"{metric_type} exceeded warning threshold",
189 metric_value=value,
190 threshold=thresholds["warning"],
191 created_at=datetime.utcnow(),
192 )
194 if alert:
195 self.alerts.append(alert)
197 return alert
199 def get_active_alerts(self) -> List[Alert]:
200 """Get unresolved alerts."""
201 return [a for a in self.alerts if a.resolved_at is None]
203 def resolve_alert(self, alert_id: str) -> None:
204 """Mark alert as resolved."""
205 for alert in self.alerts: 205 ↛ exitline 205 didn't return from function 'resolve_alert' because the loop on line 205 didn't complete
206 if alert.alert_id == alert_id: 206 ↛ 205line 206 didn't jump to line 205 because the condition on line 206 was always true
207 alert.resolved_at = datetime.utcnow()
208 break
210 def get_alerts_for_node(self, node_id: str) -> List[Alert]:
211 """Get all alerts for a node."""
212 return [a for a in self.alerts if a.node_id == node_id]
215class EventLogger:
216 """Log execution events for audit trail and analysis."""
218 def __init__(self, graph: DependencyGraph):
219 self.graph = graph
220 self.events: List[ExecutionEvent] = []
221 self.retention_days = 90
223 def log_execution(
224 self,
225 node_id: str,
226 status: str,
227 duration_ms: int,
228 start_time: datetime,
229 end_time: datetime,
230 error_message: Optional[str] = None,
231 tags: Dict[str, str] = None,
232 ) -> ExecutionEvent:
233 """Log a node execution event."""
234 event = ExecutionEvent(
235 event_id=f"{node_id}_{start_time.timestamp()}",
236 node_id=node_id,
237 status=status,
238 duration_ms=duration_ms,
239 start_time=start_time,
240 end_time=end_time,
241 error_message=error_message,
242 tags=tags or {},
243 )
244 self.events.append(event)
245 return event
247 def get_events_for_node(self, node_id: str, hours: int = 24) -> List[ExecutionEvent]:
248 """Get recent events for a node."""
249 cutoff = datetime.utcnow() - timedelta(hours=hours)
250 return [e for e in self.events if e.node_id == node_id and e.start_time >= cutoff]
252 def get_failure_rate(self, node_id: str, hours: int = 24) -> float:
253 """Calculate failure rate for a node."""
254 events = self.get_events_for_node(node_id, hours)
255 if not events:
256 return 0.0
258 failures = sum(1 for e in events if e.status == "failure")
259 return failures / len(events)
261 def get_average_duration(self, node_id: str, hours: int = 24) -> float:
262 """Get average execution duration for a node."""
263 events = self.get_events_for_node(node_id, hours)
264 if not events:
265 return 0.0
267 successful = [e for e in events if e.status == "success"]
268 if not successful: 268 ↛ 269line 268 didn't jump to line 269 because the condition on line 268 was never true
269 return 0.0
271 return sum(e.duration_ms for e in successful) / len(successful)
273 def cleanup_old_events(self) -> int:
274 """Remove events older than retention period."""
275 cutoff = datetime.utcnow() - timedelta(days=self.retention_days)
276 old_count = len([e for e in self.events if e.start_time < cutoff])
278 self.events = [e for e in self.events if e.start_time >= cutoff]
279 return old_count
281 def export_events(self, node_id: Optional[str] = None) -> str:
282 """Export events as JSON."""
283 events = self.events
284 if node_id:
285 events = [e for e in events if e.node_id == node_id]
287 # Convert to dictionaries, handling datetime
288 events_dict = []
289 for e in events:
290 d = asdict(e)
291 d["start_time"] = e.start_time.isoformat()
292 d["end_time"] = e.end_time.isoformat()
293 events_dict.append(d)
295 return json.dumps(events_dict, indent=2)
298class DashboardBuilder:
299 """Build observability dashboards."""
301 def __init__(
302 self,
303 graph: DependencyGraph,
304 metrics_collector: MetricsCollector,
305 alert_manager: AlertManager,
306 event_logger: EventLogger,
307 ):
308 self.graph = graph
309 self.metrics = metrics_collector
310 self.alerts = alert_manager
311 self.events = event_logger
313 def build_node_dashboard(self, node_id: str) -> Dict[str, Any]:
314 """Build a dashboard for a specific node."""
315 node = self.graph.nodes.get(node_id)
316 if not node: 316 ↛ 317line 316 didn't jump to line 317 because the condition on line 316 was never true
317 return {}
319 # Get metrics
320 execution_times = self.metrics.get_metrics_for_node(
321 node_id, MetricType.EXECUTION_TIME, hours=24
322 )
323 exec_stats = self.metrics.calculate_statistics(node_id, MetricType.EXECUTION_TIME, hours=24)
325 # Get events
326 events = self.events.get_events_for_node(node_id, hours=24)
327 failure_rate = self.events.get_failure_rate(node_id)
329 # Get alerts
330 alerts = self.alerts.get_alerts_for_node(node_id)
331 active_alerts = [a for a in alerts if a.resolved_at is None]
333 return {
334 "node_id": node_id,
335 "node_name": node.name,
336 "node_type": node.type.value,
337 "severity": node.severity.value,
338 "owner": node.owner,
339 "dashboard": {
340 "execution_metrics": {
341 "average_duration_ms": exec_stats.get("avg", 0),
342 "min_duration_ms": exec_stats.get("min", 0),
343 "max_duration_ms": exec_stats.get("max", 0),
344 "p95_duration_ms": exec_stats.get("p95", 0),
345 "recent_executions": len(execution_times),
346 },
347 "reliability": {
348 "failure_rate": f"{failure_rate:.1%}",
349 "successful_runs": len([e for e in events if e.status == "success"]),
350 "failed_runs": len([e for e in events if e.status == "failure"]),
351 "timeout_runs": len([e for e in events if e.status == "timeout"]),
352 },
353 "alerts": {
354 "active_count": len(active_alerts),
355 "critical": len(
356 [a for a in active_alerts if a.severity == NodeSeverity.CRITICAL]
357 ),
358 "high": len([a for a in active_alerts if a.severity == NodeSeverity.HIGH]),
359 },
360 "recent_events": [
361 {
362 "timestamp": e.start_time.isoformat(),
363 "status": e.status,
364 "duration_ms": e.duration_ms,
365 "error": e.error_message,
366 }
367 for e in events[-10:] # Last 10 events
368 ],
369 },
370 }
372 def build_health_dashboard(self) -> Dict[str, Any]:
373 """Build overall system health dashboard."""
374 all_events = self.events.events
375 all_alerts = self.alerts.get_active_alerts()
377 # Calculate aggregate metrics
378 total_executions = len(all_events)
379 failed = len([e for e in all_events if e.status == "failure"])
380 avg_failure_rate = failed / total_executions if total_executions > 0 else 0
382 return {
383 "dashboard": "System Health",
384 "timestamp": datetime.utcnow().isoformat(),
385 "metrics": {
386 "total_nodes": len(self.graph.nodes),
387 "total_edges": len(self.graph.edges),
388 },
389 "execution_stats": {
390 "total_executions": total_executions,
391 "successful": total_executions - failed,
392 "failed": failed,
393 "overall_failure_rate": f"{avg_failure_rate:.1%}",
394 },
395 "alerts": {
396 "active_alerts": len(all_alerts),
397 "critical": len([a for a in all_alerts if a.severity == NodeSeverity.CRITICAL]),
398 "high": len([a for a in all_alerts if a.severity == NodeSeverity.HIGH]),
399 "medium": len([a for a in all_alerts if a.severity == NodeSeverity.MEDIUM]),
400 },
401 "top_failing_nodes": self._get_top_failing_nodes(),
402 "slowest_nodes": self._get_slowest_nodes(),
403 }
405 def _get_top_failing_nodes(self, limit: int = 5) -> List[Dict]:
406 """Get nodes with highest failure rates."""
407 node_failures = {}
409 for node_id in self.graph.nodes:
410 failure_rate = self.events.get_failure_rate(node_id)
411 if failure_rate > 0:
412 node_failures[node_id] = failure_rate
414 top = sorted(node_failures.items(), key=lambda x: x[1], reverse=True)[:limit]
416 return [
417 {
418 "node_id": node_id,
419 "node_name": self.graph.nodes[node_id].name,
420 "failure_rate": f"{rate:.1%}",
421 }
422 for node_id, rate in top
423 ]
425 def _get_slowest_nodes(self, limit: int = 5) -> List[Dict]:
426 """Get slowest executing nodes."""
427 node_durations = {}
429 for node_id in self.graph.nodes:
430 avg_duration = self.events.get_average_duration(node_id)
431 if avg_duration > 0:
432 node_durations[node_id] = avg_duration
434 top = sorted(node_durations.items(), key=lambda x: x[1], reverse=True)[:limit]
436 return [
437 {
438 "node_id": node_id,
439 "node_name": self.graph.nodes[node_id].name,
440 "average_duration_ms": int(duration),
441 }
442 for node_id, duration in top
443 ]