Master comprehensive monitoring automation with Prometheus metrics, Grafana dashboards, custom exporters, and intelligent alerting systems for production infrastructure
Monitoring and alerting automation is the foundation of reliable, observable systems. Modern DevOps practices require proactive monitoring that detects issues before they impact users, provides actionable insights through metrics, and automatically routes alerts to the right teams at the right time.
Why Monitoring Automation Matters:
Real-World Monitoring Scenarios:
Modern Monitoring Stack:
This lesson covers production-ready monitoring automation using industry-standard tools and patterns.
Prometheus is the de-facto standard for metrics collection in cloud-native environments, providing powerful querying and alerting capabilities.
Prometheus Architecture and Python Integration:
1from prometheus_client import (
2 Counter, Gauge, Histogram, Summary,
3 CollectorRegistry, generate_latest,
4 start_http_server, CONTENT_TYPE_LATEST
5)
6from prometheus_client.core import GaugeMetricFamily, CounterMetricFamily
7import time
8import psutil
9import requests
10from typing import Dict, List
11import logging
12
13logger = logging.getLogger(__name__)
14
15class CustomPrometheusExporter:
16 """
17 Production Prometheus exporter for custom metrics
18 """
19
20 def __init__(self, port: int = 9100):
21 """
22 Initialize custom Prometheus exporter
23
24 Args:
25 port: HTTP port for metrics endpoint
26 """
27 self.port = port
28 self.registry = CollectorRegistry()
29
30 # Define metrics
31 # Counter: Monotonically increasing value (requests, errors)
32 self.request_counter = Counter(
33 'app_requests_total',
34 'Total application requests',
35 ['method', 'endpoint', 'status'],
36 registry=self.registry
37 )
38
39 # Gauge: Value that can go up or down (memory, connections)
40 self.active_users = Gauge(
41 'app_active_users',
42 'Number of active users',
43 ['region'],
44 registry=self.registry
45 )
46
47 # Histogram: Observations in buckets (latency, request size)
48 self.request_duration = Histogram(
49 'app_request_duration_seconds',
50 'Request duration in seconds',
51 ['endpoint'],
52 buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0],
53 registry=self.registry
54 )
55
56 # Summary: Similar to histogram but with quantiles
57 self.response_size = Summary(
58 'app_response_size_bytes',
59 'Response size in bytes',
60 ['endpoint'],
61 registry=self.registry
62 )
63
64 logger.info(f"š Prometheus exporter initialized on port {port}")
65
66 def start_server(self):
67 """Start HTTP server for metrics endpoint"""
68 try:
69 start_http_server(self.port, registry=self.registry)
70 logger.info(f"ā
Metrics server started at http://localhost:{self.port}/metrics")
71
72 except Exception as e:
73 logger.error(f"ā Failed to start metrics server: {e}")
74 raise
75
76 def record_request(self, method: str, endpoint: str,
77 status: int, duration: float):
78 """
79 Record HTTP request metrics
80
81 Args:
82 method: HTTP method (GET, POST, etc.)
83 endpoint: API endpoint
84 status: HTTP status code
85 duration: Request duration in seconds
86 """
87 # Increment request counter
88 self.request_counter.labels(
89 method=method,
90 endpoint=endpoint,
91 status=str(status)
92 ).inc()
93
94 # Record request duration
95 self.request_duration.labels(endpoint=endpoint).observe(duration)
96
97 logger.debug(f"Recorded request: {method} {endpoint} {status} {duration:.3f}s")
98
99 def update_active_users(self, region: str, count: int):
100 """Update active users gauge"""
101 self.active_users.labels(region=region).set(count)
102
103 def record_response_size(self, endpoint: str, size: int):
104 """Record response size"""
105 self.response_size.labels(endpoint=endpoint).observe(size)
106
107
108class SystemMetricsCollector:
109 """
110 Collect system-level metrics using psutil
111 """
112
113 def __init__(self):
114 """Initialize system metrics collector"""
115 self.registry = CollectorRegistry()
116
117 # CPU metrics
118 self.cpu_usage = Gauge(
119 'system_cpu_usage_percent',
120 'CPU usage percentage',
121 ['cpu'],
122 registry=self.registry
123 )
124
125 # Memory metrics
126 self.memory_usage = Gauge(
127 'system_memory_usage_bytes',
128 'Memory usage in bytes',
129 ['type'], # used, available, cached
130 registry=self.registry
131 )
132
133 # Disk metrics
134 self.disk_usage = Gauge(
135 'system_disk_usage_bytes',
136 'Disk usage in bytes',
137 ['mountpoint', 'type'], # total, used, free
138 registry=self.registry
139 )
140
141 # Network metrics
142 self.network_bytes = Counter(
143 'system_network_bytes_total',
144 'Network bytes transferred',
145 ['interface', 'direction'], # sent, recv
146 registry=self.registry
147 )
148
149 logger.info("š System metrics collector initialized")
150
151 def collect_metrics(self):
152 """Collect all system metrics"""
153 try:
154 # CPU metrics
155 cpu_percent = psutil.cpu_percent(interval=1, percpu=True)
156 for i, percent in enumerate(cpu_percent):
157 self.cpu_usage.labels(cpu=f"cpu{i}").set(percent)
158
159 # Memory metrics
160 memory = psutil.virtual_memory()
161 self.memory_usage.labels(type='used').set(memory.used)
162 self.memory_usage.labels(type='available').set(memory.available)
163 self.memory_usage.labels(type='cached').set(memory.cached)
164
165 # Disk metrics
166 for partition in psutil.disk_partitions():
167 try:
168 usage = psutil.disk_usage(partition.mountpoint)
169 self.disk_usage.labels(
170 mountpoint=partition.mountpoint,
171 type='total'
172 ).set(usage.total)
173 self.disk_usage.labels(
174 mountpoint=partition.mountpoint,
175 type='used'
176 ).set(usage.used)
177 self.disk_usage.labels(
178 mountpoint=partition.mountpoint,
179 type='free'
180 ).set(usage.free)
181 except PermissionError:
182 pass # Skip inaccessible partitions
183
184 # Network metrics
185 net_io = psutil.net_io_counters(pernic=True)
186 for interface, stats in net_io.items():
187 self.network_bytes.labels(
188 interface=interface,
189 direction='sent'
190 ).inc(stats.bytes_sent)
191 self.network_bytes.labels(
192 interface=interface,
193 direction='recv'
194 ).inc(stats.bytes_recv)
195
196 logger.debug("System metrics collected successfully")
197
198 except Exception as e:
199 logger.error(f"ā Failed to collect system metrics: {e}")
200
201
202class ApplicationHealthChecker:
203 """
204 Monitor application health with custom checks
205 """
206
207 def __init__(self):
208 """Initialize health checker"""
209 self.registry = CollectorRegistry()
210
211 self.health_status = Gauge(
212 'app_health_status',
213 'Application health status (1=healthy, 0=unhealthy)',
214 ['component'],
215 registry=self.registry
216 )
217
218 self.health_check_duration = Histogram(
219 'app_health_check_duration_seconds',
220 'Health check duration',
221 ['component'],
222 registry=self.registry
223 )
224
225 logger.info("š„ Health checker initialized")
226
227 def check_database(self, db_url: str) -> bool:
228 """
229 Check database connectivity
230
231 Args:
232 db_url: Database connection URL
233
234 Returns:
235 True if database is healthy
236 """
237 start_time = time.time()
238
239 try:
240 # Simulate database check (would use actual DB library)
241 # import psycopg2
242 # conn = psycopg2.connect(db_url)
243 # conn.close()
244
245 is_healthy = True
246 self.health_status.labels(component='database').set(1)
247
248 logger.info("ā
Database health check passed")
249
250 except Exception as e:
251 is_healthy = False
252 self.health_status.labels(component='database').set(0)
253 logger.error(f"ā Database health check failed: {e}")
254
255 finally:
256 duration = time.time() - start_time
257 self.health_check_duration.labels(component='database').observe(duration)
258
259 return is_healthy
260
261 def check_external_api(self, api_url: str, timeout: int = 5) -> bool:
262 """
263 Check external API availability
264
265 Args:
266 api_url: API endpoint to check
267 timeout: Request timeout in seconds
268
269 Returns:
270 True if API is healthy
271 """
272 start_time = time.time()
273
274 try:
275 response = requests.get(api_url, timeout=timeout)
276 is_healthy = response.status_code == 200
277
278 if is_healthy:
279 self.health_status.labels(component='external_api').set(1)
280 logger.info(f"ā
External API health check passed: {api_url}")
281 else:
282 self.health_status.labels(component='external_api').set(0)
283 logger.warning(f"ā ļø External API returned {response.status_code}")
284
285 except requests.RequestException as e:
286 is_healthy = False
287 self.health_status.labels(component='external_api').set(0)
288 logger.error(f"ā External API health check failed: {e}")
289
290 finally:
291 duration = time.time() - start_time
292 self.health_check_duration.labels(component='external_api').observe(duration)
293
294 return is_healthy
295
296 def check_disk_space(self, threshold_percent: float = 90) -> bool:
297 """
298 Check if disk space is below threshold
299
300 Args:
301 threshold_percent: Alert threshold (0-100)
302
303 Returns:
304 True if disk space is healthy
305 """
306 start_time = time.time()
307
308 try:
309 disk = psutil.disk_usage('/')
310 usage_percent = disk.percent
311
312 is_healthy = usage_percent < threshold_percent
313
314 if is_healthy:
315 self.health_status.labels(component='disk_space').set(1)
316 logger.info(f"ā
Disk space healthy: {usage_percent:.1f}% used")
317 else:
318 self.health_status.labels(component='disk_space').set(0)
319 logger.warning(f"ā ļø Disk space critical: {usage_percent:.1f}% used")
320
321 except Exception as e:
322 is_healthy = False
323 self.health_status.labels(component='disk_space').set(0)
324 logger.error(f"ā Disk space check failed: {e}")
325
326 finally:
327 duration = time.time() - start_time
328 self.health_check_duration.labels(component='disk_space').observe(duration)
329
330 return is_healthyKey Prometheus Patterns:
<namespace>_<subsystem>_<metric>_<unit>http_requests_total, memory_usage_bytesProduction Considerations:
Grafana provides powerful visualization capabilities and can be fully automated using its HTTP API for programmatic dashboard creation and management.
Grafana API Integration:
1import requests
2import json
3from typing import Dict, List, Optional
4from dataclasses import dataclass
5import logging
6
7logger = logging.getLogger(__name__)
8
9@dataclass
10class GrafanaDashboardConfig:
11 """Grafana dashboard configuration"""
12 title: str
13 uid: str
14 tags: List[str]
15 refresh: str = '30s'
16 time_from: str = 'now-1h'
17 time_to: str = 'now'
18
19class GrafanaAPIClient:
20 """
21 Grafana API client for dashboard automation
22 """
23
24 def __init__(self, url: str, api_key: str):
25 """
26 Initialize Grafana API client
27
28 Args:
29 url: Grafana server URL
30 api_key: API key for authentication
31 """
32 self.url = url.rstrip('/')
33 self.headers = {
34 'Authorization': f'Bearer {api_key}',
35 'Content-Type': 'application/json'
36 }
37
38 logger.info(f"š Grafana client initialized: {url}")
39
40 def create_dashboard(self, config: GrafanaDashboardConfig,
41 panels: List[Dict]) -> Dict:
42 """
43 Create Grafana dashboard
44
45 Args:
46 config: Dashboard configuration
47 panels: List of panel definitions
48
49 Returns:
50 Dashboard creation response
51 """
52 try:
53 dashboard = {
54 'dashboard': {
55 'uid': config.uid,
56 'title': config.title,
57 'tags': config.tags,
58 'timezone': 'browser',
59 'schemaVersion': 16,
60 'version': 0,
61 'refresh': config.refresh,
62 'time': {
63 'from': config.time_from,
64 'to': config.time_to
65 },
66 'panels': panels
67 },
68 'overwrite': True,
69 'message': 'Created via Python automation'
70 }
71
72 response = requests.post(
73 f'{self.url}/api/dashboards/db',
74 headers=self.headers,
75 json=dashboard
76 )
77
78 response.raise_for_status()
79
80 result = response.json()
81 logger.info(f"ā
Dashboard created: {config.title} (UID: {config.uid})")
82
83 return result
84
85 except requests.RequestException as e:
86 logger.error(f"ā Failed to create dashboard: {e}")
87 raise
88
89 def create_graph_panel(self, title: str, query: str,
90 x: int, y: int, w: int, h: int,
91 panel_id: int) -> Dict:
92 """
93 Create graph panel definition
94
95 Args:
96 title: Panel title
97 query: Prometheus query
98 x, y: Panel position
99 w, h: Panel width and height
100 panel_id: Unique panel ID
101
102 Returns:
103 Panel definition
104 """
105 return {
106 'id': panel_id,
107 'title': title,
108 'type': 'graph',
109 'gridPos': {'x': x, 'y': y, 'w': w, 'h': h},
110 'targets': [
111 {
112 'expr': query,
113 'refId': 'A',
114 'legendFormat': '{{instance}}'
115 }
116 ],
117 'yaxes': [
118 {
119 'format': 'short',
120 'label': None,
121 'logBase': 1,
122 'show': True
123 },
124 {
125 'format': 'short',
126 'label': None,
127 'logBase': 1,
128 'show': True
129 }
130 ],
131 'xaxis': {
132 'mode': 'time',
133 'show': True
134 },
135 'lines': True,
136 'fill': 1,
137 'linewidth': 2,
138 'points': False,
139 'pointradius': 5,
140 'bars': False,
141 'stack': False,
142 'percentage': False,
143 'legend': {
144 'show': True,
145 'values': True,
146 'min': True,
147 'max': True,
148 'current': True,
149 'total': False,
150 'avg': True
151 },
152 'nullPointMode': 'null',
153 'tooltip': {
154 'value_type': 'individual',
155 'shared': True,
156 'sort': 0
157 }
158 }
159
160 def create_stat_panel(self, title: str, query: str,
161 x: int, y: int, w: int, h: int,
162 panel_id: int,
163 unit: str = 'short') -> Dict:
164 """Create single stat panel"""
165 return {
166 'id': panel_id,
167 'title': title,
168 'type': 'stat',
169 'gridPos': {'x': x, 'y': y, 'w': w, 'h': h},
170 'targets': [
171 {
172 'expr': query,
173 'refId': 'A'
174 }
175 ],
176 'options': {
177 'graphMode': 'area',
178 'colorMode': 'value',
179 'justifyMode': 'auto',
180 'textMode': 'auto',
181 'orientation': 'auto',
182 'reduceOptions': {
183 'values': False,
184 'fields': '',
185 'calcs': ['lastNotNull']
186 }
187 },
188 'fieldConfig': {
189 'defaults': {
190 'unit': unit,
191 'thresholds': {
192 'mode': 'absolute',
193 'steps': [
194 {'value': None, 'color': 'green'},
195 {'value': 80, 'color': 'yellow'},
196 {'value': 90, 'color': 'red'}
197 ]
198 }
199 }
200 }
201 }
202
203 def create_infrastructure_dashboard(self, prometheus_ds: str = 'Prometheus') -> Dict:
204 """
205 Create comprehensive infrastructure dashboard
206
207 Args:
208 prometheus_ds: Prometheus datasource name
209
210 Returns:
211 Dashboard creation response
212 """
213 config = GrafanaDashboardConfig(
214 title='Infrastructure Monitoring',
215 uid='infra-monitoring',
216 tags=['infrastructure', 'system', 'monitoring']
217 )
218
219 panels = []
220
221 # Row 1: System Overview Stats
222 panels.append(self.create_stat_panel(
223 title='CPU Usage %',
224 query='100 - (avg(rate(node_cpu_seconds_total{mode="idle"}[5m])) * 100)',
225 x=0, y=0, w=6, h=4,
226 panel_id=1,
227 unit='percent'
228 ))
229
230 panels.append(self.create_stat_panel(
231 title='Memory Usage %',
232 query='(1 - (node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes)) * 100',
233 x=6, y=0, w=6, h=4,
234 panel_id=2,
235 unit='percent'
236 ))
237
238 panels.append(self.create_stat_panel(
239 title='Disk Usage %',
240 query='(1 - (node_filesystem_avail_bytes / node_filesystem_size_bytes)) * 100',
241 x=12, y=0, w=6, h=4,
242 panel_id=3,
243 unit='percent'
244 ))
245
246 panels.append(self.create_stat_panel(
247 title='Network Throughput',
248 query='rate(node_network_receive_bytes_total[5m]) + rate(node_network_transmit_bytes_total[5m])',
249 x=18, y=0, w=6, h=4,
250 panel_id=4,
251 unit='Bps'
252 ))
253
254 # Row 2: Detailed Graphs
255 panels.append(self.create_graph_panel(
256 title='CPU Usage Over Time',
257 query='100 - (avg(rate(node_cpu_seconds_total{mode="idle"}[5m])) * 100)',
258 x=0, y=4, w=12, h=8,
259 panel_id=5
260 ))
261
262 panels.append(self.create_graph_panel(
263 title='Memory Usage Over Time',
264 query='node_memory_MemTotal_bytes - node_memory_MemAvailable_bytes',
265 x=12, y=4, w=12, h=8,
266 panel_id=6
267 ))
268
269 # Row 3: Application Metrics
270 panels.append(self.create_graph_panel(
271 title='HTTP Request Rate',
272 query='rate(app_requests_total[5m])',
273 x=0, y=12, w=12, h=8,
274 panel_id=7
275 ))
276
277 panels.append(self.create_graph_panel(
278 title='Request Duration (p95)',
279 query='histogram_quantile(0.95, rate(app_request_duration_seconds_bucket[5m]))',
280 x=12, y=12, w=12, h=8,
281 panel_id=8
282 ))
283
284 return self.create_dashboard(config, panels)
285
286 def get_dashboard(self, uid: str) -> Dict:
287 """Get dashboard by UID"""
288 try:
289 response = requests.get(
290 f'{self.url}/api/dashboards/uid/{uid}',
291 headers=self.headers
292 )
293
294 response.raise_for_status()
295 return response.json()
296
297 except requests.RequestException as e:
298 logger.error(f"ā Failed to get dashboard: {e}")
299 raise
300
301 def delete_dashboard(self, uid: str) -> bool:
302 """Delete dashboard by UID"""
303 try:
304 response = requests.delete(
305 f'{self.url}/api/dashboards/uid/{uid}',
306 headers=self.headers
307 )
308
309 response.raise_for_status()
310 logger.info(f"ā
Dashboard deleted: {uid}")
311 return True
312
313 except requests.RequestException as e:
314 logger.error(f"ā Failed to delete dashboard: {e}")
315 return False
316
317 def create_datasource(self, name: str, url: str,
318 ds_type: str = 'prometheus') -> Dict:
319 """Create Prometheus datasource"""
320 try:
321 datasource = {
322 'name': name,
323 'type': ds_type,
324 'url': url,
325 'access': 'proxy',
326 'isDefault': True
327 }
328
329 response = requests.post(
330 f'{self.url}/api/datasources',
331 headers=self.headers,
332 json=datasource
333 )
334
335 response.raise_for_status()
336
337 result = response.json()
338 logger.info(f"ā
Datasource created: {name}")
339
340 return result
341
342 except requests.RequestException as e:
343 logger.error(f"ā Failed to create datasource: {e}")
344 raiseDashboard Design Best Practices:
Production Considerations:
Effective alerting requires intelligent routing, de-duplication, and integration with incident management systems to ensure the right people are notified at the right time.
Alert Management System:
1import requests
2import smtplib
3from email.mime.text import MIMEText
4from email.mime.multipart import MIMEMultipart
5from typing import Dict, List, Optional
6from enum import Enum
7from dataclasses import dataclass
8from datetime import datetime
9import logging
10
11logger = logging.getLogger(__name__)
12
13class AlertSeverity(Enum):
14 """Alert severity levels"""
15 CRITICAL = "critical"
16 WARNING = "warning"
17 INFO = "info"
18
19class AlertStatus(Enum):
20 """Alert status"""
21 FIRING = "firing"
22 RESOLVED = "resolved"
23
24@dataclass
25class Alert:
26 """Alert definition"""
27 name: str
28 severity: AlertSeverity
29 status: AlertStatus
30 message: str
31 labels: Dict[str, str]
32 annotations: Dict[str, str]
33 timestamp: datetime
34 metric_value: Optional[float] = None
35
36class AlertManager:
37 """
38 Comprehensive alert management system
39 """
40
41 def __init__(self, alertmanager_url: str = None):
42 """
43 Initialize alert manager
44
45 Args:
46 alertmanager_url: Alertmanager URL for integration
47 """
48 self.alertmanager_url = alertmanager_url
49 self.alert_history: List[Alert] = []
50
51 logger.info("šØ Alert manager initialized")
52
53 def send_alert(self, alert: Alert):
54 """
55 Send alert through multiple channels
56
57 Args:
58 alert: Alert to send
59 """
60 try:
61 logger.info(f"šØ Sending {alert.severity.value} alert: {alert.name}")
62
63 # Route based on severity
64 if alert.severity == AlertSeverity.CRITICAL:
65 self._send_to_pagerduty(alert)
66 self._send_to_slack(alert)
67 self._send_email(alert)
68 elif alert.severity == AlertSeverity.WARNING:
69 self._send_to_slack(alert)
70 self._send_email(alert)
71 else:
72 self._send_to_slack(alert)
73
74 # Store in history
75 self.alert_history.append(alert)
76
77 logger.info(f"ā
Alert sent: {alert.name}")
78
79 except Exception as e:
80 logger.error(f"ā Failed to send alert: {e}")
81
82 def _send_to_pagerduty(self, alert: Alert):
83 """Send alert to PagerDuty"""
84 # In production, use PagerDuty API
85 # from pdpyras import APISession
86 # session = APISession(api_key)
87 # session.trigger('incident', ...)
88
89 logger.info(f"š Sending to PagerDuty: {alert.name}")
90
91 # Mock PagerDuty integration
92 pagerduty_payload = {
93 'routing_key': 'YOUR_INTEGRATION_KEY',
94 'event_action': 'trigger' if alert.status == AlertStatus.FIRING else 'resolve',
95 'payload': {
96 'summary': alert.message,
97 'severity': alert.severity.value,
98 'source': 'monitoring-system',
99 'timestamp': alert.timestamp.isoformat(),
100 'custom_details': alert.labels
101 }
102 }
103
104 logger.debug(f"PagerDuty payload: {pagerduty_payload}")
105
106 def _send_to_slack(self, alert: Alert):
107 """Send alert to Slack"""
108 # In production, use Slack webhook
109 webhook_url = 'https://hooks.slack.com/services/YOUR/WEBHOOK/URL'
110
111 # Color coding by severity
112 colors = {
113 AlertSeverity.CRITICAL: '#ff0000',
114 AlertSeverity.WARNING: '#ffaa00',
115 AlertSeverity.INFO: '#36a64f'
116 }
117
118 emoji = {
119 AlertSeverity.CRITICAL: ':rotating_light:',
120 AlertSeverity.WARNING: ':warning:',
121 AlertSeverity.INFO: ':information_source:'
122 }
123
124 slack_message = {
125 'attachments': [
126 {
127 'color': colors[alert.severity],
128 'title': f"{emoji[alert.severity]} {alert.name}",
129 'text': alert.message,
130 'fields': [
131 {
132 'title': 'Severity',
133 'value': alert.severity.value.upper(),
134 'short': True
135 },
136 {
137 'title': 'Status',
138 'value': alert.status.value.upper(),
139 'short': True
140 },
141 {
142 'title': 'Timestamp',
143 'value': alert.timestamp.strftime('%Y-%m-%d %H:%M:%S'),
144 'short': False
145 }
146 ],
147 'footer': 'Monitoring System',
148 'ts': int(alert.timestamp.timestamp())
149 }
150 ]
151 }
152
153 logger.info(f"š¬ Sending to Slack: {alert.name}")
154
155 # In production, uncomment:
156 # response = requests.post(webhook_url, json=slack_message)
157 # response.raise_for_status()
158
159 def _send_email(self, alert: Alert):
160 """Send alert via email"""
161 # In production, configure SMTP settings
162 smtp_server = 'smtp.example.com'
163 smtp_port = 587
164 sender_email = '[email protected]'
165 receiver_emails = ['[email protected]']
166
167 # Create message
168 msg = MIMEMultipart('alternative')
169 msg['Subject'] = f"[{alert.severity.value.upper()}] {alert.name}"
170 msg['From'] = sender_email
171 msg['To'] = ', '.join(receiver_emails)
172
173 # HTML body
174 html = f"""
175 <html>
176 <body>
177 <h2 style="color: {'red' if alert.severity == AlertSeverity.CRITICAL else 'orange'};">
178 {alert.name}
179 </h2>
180 <p><strong>Message:</strong> {alert.message}</p>
181 <p><strong>Severity:</strong> {alert.severity.value.upper()}</p>
182 <p><strong>Status:</strong> {alert.status.value.upper()}</p>
183 <p><strong>Time:</strong> {alert.timestamp.strftime('%Y-%m-%d %H:%M:%S')}</p>
184
185 <h3>Labels:</h3>
186 <ul>
187 {''.join(f'<li><strong>{k}:</strong> {v}</li>' for k, v in alert.labels.items())}
188 </ul>
189
190 <p><em>This is an automated alert from the monitoring system.</em></p>
191 </body>
192 </html>
193 """
194
195 msg.attach(MIMEText(html, 'html'))
196
197 logger.info(f"š§ Sending email alert: {alert.name}")
198
199 # In production, uncomment:
200 # with smtplib.SMTP(smtp_server, smtp_port) as server:
201 # server.starttls()
202 # server.login(sender_email, password)
203 # server.send_message(msg)
204
205 def check_metric_threshold(self, metric_name: str,
206 current_value: float,
207 threshold: float,
208 comparison: str = 'greater') -> Optional[Alert]:
209 """
210 Check if metric exceeds threshold
211
212 Args:
213 metric_name: Metric name
214 current_value: Current metric value
215 threshold: Threshold value
216 comparison: 'greater' or 'less'
217
218 Returns:
219 Alert if threshold exceeded, None otherwise
220 """
221 triggered = False
222
223 if comparison == 'greater' and current_value > threshold:
224 triggered = True
225 elif comparison == 'less' and current_value < threshold:
226 triggered = True
227
228 if triggered:
229 severity = AlertSeverity.CRITICAL if current_value > threshold * 1.2 else AlertSeverity.WARNING
230
231 alert = Alert(
232 name=f"{metric_name} Threshold Exceeded",
233 severity=severity,
234 status=AlertStatus.FIRING,
235 message=f"{metric_name} is {current_value:.2f}, threshold is {threshold:.2f}",
236 labels={'metric': metric_name},
237 annotations={'runbook': 'https://runbooks.example.com/high-cpu'},
238 timestamp=datetime.now(),
239 metric_value=current_value
240 )
241
242 return alert
243
244 return None
245
246 def get_alert_summary(self) -> Dict:
247 """Get summary of recent alerts"""
248 summary = {
249 'total_alerts': len(self.alert_history),
250 'by_severity': {},
251 'by_status': {},
252 'recent_critical': []
253 }
254
255 for alert in self.alert_history:
256 # Count by severity
257 severity_key = alert.severity.value
258 summary['by_severity'][severity_key] = summary['by_severity'].get(severity_key, 0) + 1
259
260 # Count by status
261 status_key = alert.status.value
262 summary['by_status'][status_key] = summary['by_status'].get(status_key, 0) + 1
263
264 # Track recent critical alerts
265 if alert.severity == AlertSeverity.CRITICAL and alert.status == AlertStatus.FIRING:
266 summary['recent_critical'].append({
267 'name': alert.name,
268 'message': alert.message,
269 'timestamp': alert.timestamp.isoformat()
270 })
271
272 return summaryAlerting Best Practices:
Production Considerations:
Build a comprehensive monitoring system that collects metrics with Prometheus, visualizes data in Grafana, performs health checks, and sends intelligent alerts through multiple channels including PagerDuty, Slack, and email.
1#!/usr/bin/env python3
2"""
3š Production Monitoring and Alerting Platform
4Topic: Prometheus + Grafana + Alerting for comprehensive system monitoring
5Author: DevOps Monitoring Team
6Dependencies: pip install prometheus-client psutil requests
7"""
8
9from prometheus_client import (
10 Counter, Gauge, Histogram,
11 CollectorRegistry, start_http_server
12)
13import psutil
14import time
15import requests
16from enum import Enum
17from dataclasses import dataclass
18from datetime import datetime
19from typing import Dict, List, Optional
20import logging
21import sys
22
23# Configure logging
24logging.basicConfig(
25 level=logging.INFO,
26 format='%(asctime)s - %(levelname)s - %(message)s',
27 handlers=[
28 logging.FileHandler('monitoring-system.log'),
29 logging.StreamHandler(sys.stdout)
30 ]
31)
32logger = logging.getLogger(__name__)
33
34
35class AlertSeverity(Enum):
36 """Alert severity levels"""
37 CRITICAL = "critical"
38 WARNING = "warning"
39 INFO = "info"
40
41
42@dataclass
43class Alert:
44 """Alert definition"""
45 name: str
46 severity: AlertSeverity
47 message: str
48 metric_value: float
49 timestamp: datetime
50
51
52class PrometheusMetricsCollector:
53 """
54 Collect and expose Prometheus metrics
55 """
56
57 def __init__(self, port: int = 9100):
58 """Initialize metrics collector"""
59 self.port = port
60 self.registry = CollectorRegistry()
61
62 # System metrics
63 self.cpu_usage = Gauge(
64 'system_cpu_usage_percent',
65 'CPU usage percentage',
66 registry=self.registry
67 )
68
69 self.memory_usage = Gauge(
70 'system_memory_usage_percent',
71 'Memory usage percentage',
72 registry=self.registry
73 )
74
75 self.disk_usage = Gauge(
76 'system_disk_usage_percent',
77 'Disk usage percentage',
78 registry=self.registry
79 )
80
81 # Application metrics
82 self.requests_total = Counter(
83 'app_requests_total',
84 'Total application requests',
85 ['status'],
86 registry=self.registry
87 )
88
89 self.request_duration = Histogram(
90 'app_request_duration_seconds',
91 'Request duration',
92 buckets=[0.1, 0.5, 1.0, 2.0, 5.0],
93 registry=self.registry
94 )
95
96 logger.info(f"š Prometheus metrics collector initialized on port {port}")
97
98 def start_server(self):
99 """Start metrics HTTP server"""
100 try:
101 start_http_server(self.port, registry=self.registry)
102 logger.info(f"ā
Metrics server started at http://localhost:{self.port}/metrics")
103 except Exception as e:
104 logger.error(f"ā Failed to start metrics server: {e}")
105 raise
106
107 def collect_system_metrics(self):
108 """Collect system metrics"""
109 try:
110 # CPU usage
111 cpu_percent = psutil.cpu_percent(interval=1)
112 self.cpu_usage.set(cpu_percent)
113
114 # Memory usage
115 memory = psutil.virtual_memory()
116 memory_percent = memory.percent
117 self.memory_usage.set(memory_percent)
118
119 # Disk usage
120 disk = psutil.disk_usage('/')
121 disk_percent = disk.percent
122 self.disk_usage.set(disk_percent)
123
124 logger.debug(f"System metrics: CPU={cpu_percent:.1f}%, "
125 f"Memory={memory_percent:.1f}%, Disk={disk_percent:.1f}%")
126
127 return {
128 'cpu': cpu_percent,
129 'memory': memory_percent,
130 'disk': disk_percent
131 }
132
133 except Exception as e:
134 logger.error(f"ā Failed to collect system metrics: {e}")
135 return {}
136
137 def record_request(self, status: int, duration: float):
138 """Record HTTP request metrics"""
139 self.requests_total.labels(status=str(status)).inc()
140 self.request_duration.observe(duration)
141
142
143class AlertManager:
144 """
145 Manage and send alerts
146 """
147
148 def __init__(self):
149 """Initialize alert manager"""
150 self.alert_history: List[Alert] = []
151 self.alert_thresholds = {
152 'cpu': {'warning': 70, 'critical': 90},
153 'memory': {'warning': 75, 'critical': 90},
154 'disk': {'warning': 80, 'critical': 95}
155 }
156
157 logger.info("šØ Alert manager initialized")
158
159 def check_thresholds(self, metrics: Dict[str, float]) -> List[Alert]:
160 """
161 Check metrics against thresholds
162
163 Args:
164 metrics: Current metric values
165
166 Returns:
167 List of triggered alerts
168 """
169 alerts = []
170
171 for metric_name, value in metrics.items():
172 if metric_name not in self.alert_thresholds:
173 continue
174
175 thresholds = self.alert_thresholds[metric_name]
176
177 # Check critical threshold
178 if value >= thresholds['critical']:
179 alert = Alert(
180 name=f"High {metric_name.upper()}",
181 severity=AlertSeverity.CRITICAL,
182 message=f"{metric_name.upper()} usage is {value:.1f}% (threshold: {thresholds['critical']}%)",
183 metric_value=value,
184 timestamp=datetime.now()
185 )
186 alerts.append(alert)
187 logger.warning(f"š“ CRITICAL: {alert.message}")
188
189 # Check warning threshold
190 elif value >= thresholds['warning']:
191 alert = Alert(
192 name=f"Elevated {metric_name.upper()}",
193 severity=AlertSeverity.WARNING,
194 message=f"{metric_name.upper()} usage is {value:.1f}% (threshold: {thresholds['warning']}%)",
195 metric_value=value,
196 timestamp=datetime.now()
197 )
198 alerts.append(alert)
199 logger.warning(f"š” WARNING: {alert.message}")
200
201 # Store alerts in history
202 self.alert_history.extend(alerts)
203
204 return alerts
205
206 def send_alert(self, alert: Alert):
207 """
208 Send alert through configured channels
209
210 Args:
211 alert: Alert to send
212 """
213 logger.info(f"šØ Sending {alert.severity.value} alert: {alert.name}")
214
215 # In production, integrate with:
216 # - PagerDuty for critical alerts
217 # - Slack for all alerts
218 # - Email for warnings and critical
219
220 if alert.severity == AlertSeverity.CRITICAL:
221 self._send_to_pagerduty(alert)
222 self._send_to_slack(alert)
223 elif alert.severity == AlertSeverity.WARNING:
224 self._send_to_slack(alert)
225
226 def _send_to_pagerduty(self, alert: Alert):
227 """Send critical alert to PagerDuty"""
228 logger.info(f"š Sending to PagerDuty: {alert.name}")
229 # PagerDuty integration code here
230
231 def _send_to_slack(self, alert: Alert):
232 """Send alert to Slack"""
233 logger.info(f"š¬ Sending to Slack: {alert.name}")
234 # Slack webhook integration code here
235
236 def get_alert_summary(self) -> Dict:
237 """Get summary of recent alerts"""
238 return {
239 'total_alerts': len(self.alert_history),
240 'critical': sum(1 for a in self.alert_history if a.severity == AlertSeverity.CRITICAL),
241 'warning': sum(1 for a in self.alert_history if a.severity == AlertSeverity.WARNING),
242 'recent_alerts': [
243 {
244 'name': a.name,
245 'severity': a.severity.value,
246 'message': a.message,
247 'timestamp': a.timestamp.isoformat()
248 }
249 for a in self.alert_history[-5:] # Last 5 alerts
250 ]
251 }
252
253
254class MonitoringPlatform:
255 """
256 Complete monitoring and alerting platform
257 """
258
259 def __init__(self, metrics_port: int = 9100):
260 """Initialize monitoring platform"""
261 logger.info("š Initializing Monitoring Platform")
262
263 self.metrics_collector = PrometheusMetricsCollector(port=metrics_port)
264 self.alert_manager = AlertManager()
265
266 logger.info("ā
Monitoring platform initialized")
267
268 def start(self):
269 """Start the monitoring platform"""
270 try:
271 # Start metrics server
272 self.metrics_collector.start_server()
273
274 logger.info("\n" + "="*70)
275 logger.info("š MONITORING PLATFORM STARTED")
276 logger.info("="*70)
277 logger.info(f"Metrics endpoint: http://localhost:{self.metrics_collector.port}/metrics")
278 logger.info(f"Collecting metrics every 30 seconds...")
279 logger.info("="*70)
280
281 # Main monitoring loop
282 while True:
283 # Collect metrics
284 metrics = self.metrics_collector.collect_system_metrics()
285
286 # Check for alerts
287 if metrics:
288 alerts = self.alert_manager.check_thresholds(metrics)
289
290 # Send alerts
291 for alert in alerts:
292 self.alert_manager.send_alert(alert)
293
294 # Sleep before next collection
295 time.sleep(30)
296
297 except KeyboardInterrupt:
298 logger.info("\nā ļø Monitoring platform stopped by user")
299 except Exception as e:
300 logger.error(f"ā Monitoring platform error: {e}")
301 raise
302
303 def get_status(self) -> Dict:
304 """Get monitoring platform status"""
305 alert_summary = self.alert_manager.get_alert_summary()
306
307 return {
308 'status': 'running',
309 'metrics_endpoint': f"http://localhost:{self.metrics_collector.port}/metrics",
310 'alerts': alert_summary
311 }
312
313
314def main():
315 """
316 Main entry point - Start monitoring platform
317 """
318 print("\n" + "="*70)
319 print("š Production Monitoring and Alerting Platform")
320 print("="*70 + "\n")
321
322 try:
323 # Initialize platform
324 platform = MonitoringPlatform(metrics_port=9100)
325
326 # Display status
327 status = platform.get_status()
328 print(f"Status: {status['status'].upper()}")
329 print(f"Metrics: {status['metrics_endpoint']}")
330 print(f"\nPress Ctrl+C to stop...\n")
331
332 # Start monitoring
333 platform.start()
334
335 except KeyboardInterrupt:
336 logger.info("\nā ļø Interrupted by user")
337 sys.exit(130)
338 except Exception as e:
339 logger.error(f"\nā Fatal error: {e}")
340 import traceback
341 traceback.print_exc()
342 sys.exit(1)
343
344
345if __name__ == "__main__":
346 main()š Production Monitoring Platform Breakdown:
1. Architecture Components:
2. Prometheus Integration:
1# Metric types for different use cases
2cpu_usage = Gauge('system_cpu_usage_percent', 'CPU usage') # Up/down value
3requests_total = Counter('app_requests_total', 'Requests') # Monotonic
4request_duration = Histogram('app_request_duration_seconds', 'Latency') # Distribution
5
6# Expose metrics on HTTP endpoint
7start_http_server(9100, registry=registry)
8# Access at: http://localhost:9100/metrics3. System Metrics Collection:
4. Alert Threshold Logic:
1thresholds = {
2 'cpu': {'warning': 70, 'critical': 90},
3 'memory': {'warning': 75, 'critical': 90},
4 'disk': {'warning': 80, 'critical': 95}
5}
6
7# Check and create alerts
8if value >= thresholds['critical']:
9 alert = Alert(severity=AlertSeverity.CRITICAL, ...)
10elif value >= thresholds['warning']:
11 alert = Alert(severity=AlertSeverity.WARNING, ...)5. Multi-Channel Alerting:
6. Production Features:
7. Integration Points:
8. Real-World Usage:
1# Start monitoring platform
2python monitoring_platform.py
3
4# In another terminal, configure Prometheus to scrape:
5# prometheus.yml:
6# scrape_configs:
7# - job_name: 'monitoring-platform'
8# static_configs:
9# - targets: ['localhost:9100']
10
11# Access metrics directly
12curl http://localhost:9100/metrics9. Key Patterns:
This implementation provides a solid foundation for production monitoring that can be extended with additional metrics, alerting channels, and integrations!
Your progress is saved in this browser, on this device only.