¶ (c) TechnoCore - All Rights Reserved.
NOTICE: All information contained herein is, and remains
the property of TechnoCore Automate.
The intellectual and technical concepts contained
herein are proprietary to TechnoCore Automate and
dissemination of this information or reproduction of this material
is strictly forbidden unless prior written permission is obtained
from TechnoCore Automate.
¶ ObjWebhookAlert.py
¶ Overview
Webhook failure alerting system for the Axion framework. This module monitors
webhook execution metrics in real time and sends alerts when configurable
thresholds are exceeded. It tracks failure rates, authentication failures,
timeouts, and latency across all registered webhooks.
Alerts can be delivered through multiple channels simultaneously: Slack,
Email, PagerDuty, MQTT, or local logging (with database persistence).
¶ Classes
¶ AlertSeverity(str, Enum)
Alert severity levels used to classify the urgency of webhook alerts.
| Value | Description |
|---|---|
INFO |
Informational, no action required |
WARNING |
Threshold exceeded, investigate |
CRITICAL |
Significant failure, immediate attention |
URGENT |
Severe failure, escalation required |
¶ AlertChannel(str, Enum)
Available delivery channels for alert notifications.
| Value | Description |
|---|---|
SLACK |
Slack message via ObjNotify |
EMAIL |
Email via ObjNotify |
PAGERDUTY |
PagerDuty Events API v2 |
MQTT |
MQTT publish to webhook/alerts/<code> |
LOG |
Debug log + database insert to sys_webhook_alerts |
¶ WebhookAlert (dataclass)
Data structure representing a single alert event.
| Field | Type | Description |
|---|---|---|
webhook_code |
str |
Webhook identifier |
alert_type |
str |
Alert category (e.g. high_failure_rate) |
severity |
AlertSeverity |
Severity level |
message |
str |
Human-readable alert message |
details |
dict |
Additional context (thresholds, counts) |
timestamp |
datetime |
When the alert was raised |
¶ WebhookMetrics (dataclass)
In-memory metrics tracker for a single webhook within a sliding time window.
| Field | Type | Description |
|---|---|---|
webhook_code |
str |
Webhook identifier |
total_requests |
int |
Total calls in current window |
failed_requests |
int |
Failed calls in current window |
auth_failures |
int |
Authentication failures (HTTP 401/403) |
timeouts |
int |
Timeout errors |
total_duration |
float |
Cumulative request duration in seconds |
window_start |
datetime |
Start of the current metrics window |
Computed properties:
failure_rate- Failure percentage:(failed / total) * 100avg_duration- Average request duration in seconds
¶ ObjWebhookAlert Class
Inherits from: ObjData.ObjData
The main alerting engine. Records webhook call outcomes, maintains per-webhook
metrics in memory, evaluates alert conditions after each call, and dispatches
alerts through configured channels.
¶ Key Methods
| Method | Description |
|---|---|
record_webhook_call(webhook_code, success, duration, status_code, error) |
Record a webhook execution result. Automatically resets metrics when the configured time window expires and checks alert conditions. |
get_metrics(webhook_code) |
Retrieve current metrics for a specific webhook or all webhooks. Returns a dictionary suitable for JSON serialization. |
reset_metrics(webhook_code) |
Clear in-memory metrics for a specific webhook or all webhooks. |
¶ Alert Conditions
Alerts are evaluated after every record_webhook_call once a webhook has
accumulated at least 5 requests in the current window:
| Condition | Severity | Trigger |
|---|---|---|
| High failure rate | WARNING | failure_rate > failure_rate_threshold |
| Authentication failures | CRITICAL | auth_failures >= auth_failure_threshold |
| High latency | WARNING | avg_duration > latency_threshold_seconds |
¶ Alert Channel Methods
| Method | Channel | Integration |
|---|---|---|
_send_slack_alert(alert) |
Slack | Uses ObjNotify.send_slack_message with severity emoji |
_send_email_alert(alert) |
Uses ObjNotify.send_email to configured alert_email |
|
_send_pagerduty_alert(alert) |
PagerDuty | Posts to PagerDuty Events API v2 with configured routing key |
_send_mqtt_alert(alert) |
MQTT | Publishes to topic webhook/alerts/<webhook_code> |
_log_alert(alert) |
Log + DB | Logs via self.debug and inserts into sys_webhook_alerts table |
¶ Configuration
Settings are read from the webhooks.alerting section in config.yaml:
webhooks:
alerting:
enabled: true
failure_rate_threshold: 10.0 # Percentage
auth_failure_threshold: 5 # Count within window
latency_threshold_seconds: 5.0 # Seconds
window_minutes: 5 # Metrics window duration
channels: slack,mqtt # Comma-separated channels
alert_email: ops@example.com # Recipient for email alerts
pagerduty_key: <routing_key> # PagerDuty Events API key
¶ Database
The log alert channel persists alerts to the sys_webhook_alerts table:
| Column | Type | Description |
|---|---|---|
webhook_code |
VARCHAR | Webhook identifier |
alert_type |
VARCHAR | Alert category |
severity |
VARCHAR | Severity level |
message |
VARCHAR | Alert message |
details |
TEXT | JSON-encoded details |
created_at |
DATETIME | Timestamp |
¶ Command-Line Interface (CLI)
The module includes a CLI built with typer.
| Command | Description |
|---|---|
metrics [webhook_code] |
Show current in-memory metrics for a webhook or all webhooks |
test-alert [webhook_code] |
Send a test alert through all configured channels |
config |
Display the current alert configuration as JSON |
python factory.core/ObjWebhookAlert.py metrics
python factory.core/ObjWebhookAlert.py metrics MY_WEBHOOK
python factory.core/ObjWebhookAlert.py test-alert MY_WEBHOOK --severity warning
python factory.core/ObjWebhookAlert.py config