Building a Custom Alerting System Using pyAlarm and WebhooksIn modern applications, timely alerts are critical: they notify operators of failures, trigger automated remediation, and keep users informed about important events. This guide shows how to build a custom alerting system using pyAlarm — a hypothetical lightweight Python alarm library — combined with webhooks for flexible delivery. You’ll learn architecture, key components, implementation patterns, and operational considerations so you can tailor alerts for your environment.
Why use pyAlarm + webhooks?
- pyAlarm provides a programmatic, Python-native way to define, schedule, and evaluate alerts, making it convenient for teams that already use Python.
- Webhooks enable flexible delivery: messages can be pushed to Slack, Microsoft Teams, PagerDuty, Opsgenie, custom REST endpoints, or serverless functions.
- Together they let you centralize alert logic in code, integrate easily with external systems, and maintain testable alert rules.
Architecture overview
A simple architecture has the following pieces:
- Instrumentation: application or monitoring agents emit events, metrics, or logs.
- Alerting service: a Python service using pyAlarm that consumes events/metrics, evaluates rules, and triggers notifications.
- Webhook dispatcher: component that formats alerts and POSTs to configured webhook endpoints.
- Receiver endpoints: third-party services (Slack, PagerDuty) or internal endpoints that act on incoming webhooks.
- Storage and state: optional persistent store for suppression/aggregation/throttling and audit logs.
Flow: events/metrics → pyAlarm rules evaluate → matches create alert objects → webhook dispatcher sends POSTs → receivers act.
Core components and responsibilities
Instrumentation
- Emit structured events (JSON) with consistent fields: timestamp, source, severity, metric name/value, tags.
- Use libraries like Prometheus client, StatsD, or custom exporters.
- Example fields: id, timestamp, service, environment, metric, value, threshold, message.
Alerting service (pyAlarm)
- Loads rule definitions (YAML/JSON/Python).
- Subscribes to event streams: message queue (Kafka/RabbitMQ), HTTP webhook, or direct function calls.
- Evaluates conditions (thresholds, anomalies, rate-of-change, missing-heartbeat).
- Manages alert lifecycle: fired → acknowledged → resolved.
- Emits normalized alert objects for dispatch.
Example rule types:
- Threshold: trigger when metric > X for N minutes.
- Missing heartbeat: no event from a service for M minutes.
- Rate: error rate > Y% over a window.
- Composite: combine multiple conditions (AND/OR).
Webhook dispatcher
- Accepts alert objects, maps fields to delivery payloads, and POSTs to endpoints.
- Supports templating (Jinja2) to produce human-friendly messages.
- Handles retries, backoff, and failure handling.
- Supports per-endpoint authentication (Bearer tokens, Basic auth, HMAC signatures).
State and deduplication
- Store active alerts in Redis or a database.
- Use dedup keys (service+metric+threshold) to avoid duplicate notifications.
- Implement suppression windows, flood control, and escalation policies.
- Keep an audit trail of sent notifications.
Example implementation (conceptual)
Below is a high-level Python example showing main parts: rule loading, evaluation loop, alert object, and a simple webhook sender. This is illustrative — adapt to pyAlarm API and your infrastructure.
# alerting_service.py import time import requests import json from collections import defaultdict from datetime import datetime, timedelta # Assume pyAlarm exposes Rule, Engine, Alert classes (hypothetical) from pyalarm import Engine, Rule, Alert # Load rules (could be YAML/JSON) RULES = [ Rule( name="High CPU", condition=lambda evt: evt.get("metric") == "cpu" and evt.get("value", 0) > 85, suppress_for=300 # seconds ), Rule( name="Missing Heartbeat", condition=lambda evt, state: False # handled by background check ) ] engine = Engine(rules=RULES) # Simple webhook sender def send_webhook(url, payload, headers=None): headers = headers or {"Content-Type": "application/json"} try: resp = requests.post(url, json=payload, headers=headers, timeout=5) resp.raise_for_status() return True except Exception as e: print(f"Webhook send failed: {e}") return False # In-memory state for demo (use Redis/postgres in production) last_sent = defaultdict(lambda: datetime.min) SUPPRESSION_WINDOW = timedelta(seconds=300) def process_event(event): alerts = engine.evaluate(event) # returns list of Alert objects for alert in alerts: key = f"{alert.rule_name}:{alert.dedup_key}" now = datetime.utcnow() if now - last_sent[key] < SUPPRESSION_WINDOW: print(f"Suppressed duplicate alert: {key}") continue payload = { "title": alert.title, "service": alert.service, "severity": alert.severity, "timestamp": alert.timestamp.isoformat(), "details": alert.description, } # send to multiple webhooks configured per rule for target in alert.targets: ok = send_webhook(target["url"], payload, headers=target.get("headers")) if ok: last_sent[key] = now # Example event loop (replace with Kafka/HTTP listener) if __name__ == "__main__": sample_event = {"service": "api", "metric": "cpu", "value": 92, "timestamp": time.time()} process_event(sample_event) Notification formatting and templates
- Use templates to tailor messages per target (Slack, Teams, email).
- Include essential info first: severity, service, short summary, actionable link (runbook), timestamp.
- For Slack, use Blocks API JSON for rich messages; for PagerDuty, send correct fields for incidents.
Example Slack payload fields:
- title, text, color (by severity), buttons (acknowledge, runbook), context.
Handling retries, failures, and backpressure
- Implement exponential backoff with jitter for webhook retries.
- Queue alerts for delivery using a durable queue (RabbitMQ/SQS) to avoid losing notifications.
- Monitor dispatcher failures and alert on high error rates or delivery lag.
Security and authentication
- Secure webhook endpoints via:
- HMAC signatures (server signs payload; receiver verifies).
- HTTPS with TLS.
- Short-lived tokens or per-target API keys.
- Store secrets in a secrets manager (Vault, AWS Secrets Manager).
- Sanitize and validate alert payloads to avoid injection or accidental data leaks.
Testing alerts
- Unit test rule evaluation logic with synthetic events.
- Integration test delivery by pointing to a staging webhook receiver.
- Simulate high-volume bursts to test deduplication and throttling.
- Provide a “test alert” API to trigger alert paths without generating real incidents.
Operational considerations
- Observability: instrument the alerting service (latency, queue depth, success/failure counts).
- Runbooks: every alert should link to a runbook with steps to investigate/resolve.
- Escalation policies: route unresolved alerts after a timeout to higher-tier contacts.
- On-call ergonomics: avoid noisy alerts; tune thresholds, use aggregation, and apply sensible suppression.
Example real-world patterns
- Heartbeat monitoring: track last check-in times and alert if missing for a configurable window.
- Aggregated error-rate alerts: fire only when error rate exceeds threshold across many hosts (reduce noise).
- Adaptive thresholds: adjust thresholds using moving averages or simple ML to reduce false positives.
- Multichannel delivery: send critical alerts to PagerDuty and Slack; low-priority ones to email.
Checklist before rolling out
- [ ] Clear naming and severity conventions.
- [ ] Deduplication and suppression configured.
- [ ] Delivery retries and durable queuing enabled.
- [ ] Authentication for outgoing webhooks.
- [ ] Runbooks linked from alerts.
- [ ] Tests (unit, integration, load).
- [ ] Monitoring for the alerting pipeline itself.
Conclusion
Using pyAlarm together with webhooks gives you a flexible, code-centric alerting system that can integrate with many services. The key is to keep alert logic simple, reduce noise with deduplication and suppression, secure webhook delivery, and ensure good observability and runbooks so on-call engineers can act quickly and confidently.
Leave a Reply