Error Handling & Alerting
Relevant source files
The following files were used as context for generating this wiki page:
The Error Handling and Alerting system within the FLEXT Observability framework provides intelligent aggregation, deduplication, and severity escalation for error events. Its primary goal is to reduce alert fatigue by grouping similar errors through fingerprinting while ensuring that critical or high-frequency issues remain visible to operators.
The system is built around the FlextObservabilityErrorHandling service, which manages the lifecycle of error events from initial recording to alert decision-making. It utilizes railway-oriented programming patterns (using p.Result) to ensure robust error handling during the observability process itself.
Sources: src/flext_observability/services/error_handling.py:1-20, src/flext_observability/README.md:46-51, src/README.md:14-19
System Architecture
The error handling architecture is divided into a facade class and a nested logic handler. This structure allows for a global singleton instance that can be accessed via convenience methods while maintaining encapsulated state for error counts and cooldown timers.
Core Components
| Component | Description |
|---|---|
FlextObservabilityErrorHandling | Facade providing static convenience methods and access to the active handler. |
Handler | The internal engine managing error counts, fingerprints, and alert logic. |
ErrorEvent | The domain entity representing a specific error occurrence (contains type, message, and severity). |
Fingerprinting | Logic that groups similar errors based on their attributes to deduplicate alerts. |
Sources: src/flext_observability/services/error_handling.py:32-55, src/flext_observability/README.md:27-31
Error Processing Flow
The following diagram illustrates how an error event is processed to determine if an alert should be triggered.
This flow ensures that CRITICAL errors always bypass deduplication, while lower severity errors are subject to cooldown periods and frequency thresholds.
Sources: src/flext_observability/services/error_handling.py:171-205, tests/integration/test_phase_11_integration.py:72-87
Deduplication and Fingerprinting
To prevent "alert storms," the system generates a unique fingerprint for every error. This fingerprint is used as a key to track the frequency of specific issues over a rolling window.
Fingerprint Calculation
The record_error method ensures a fingerprint is present. If not provided, the ErrorEvent calculates one based on the error type and message content. This allows the system to distinguish between a DatabaseTimeout and a ValidationError even if they occur simultaneously.
def record_error(self, error: m.Observability.ErrorEvent) -> p.Result[m.Observability.ErrorEvent]:
def operation() -> m.Observability.ErrorEvent:
error.calculate_fingerprint()
# ... logic to update counts
return error
return self._run_with_result(operation, error_prefix="Failed to record error")
Sources: src/flext_observability/services/error_handling.py:126-153, tests/integration/test_phase_11_integration.py:76-80
Configuration Parameters
The handler maintains internal state to control the sensitivity of the alerting engine.
| Parameter | Default | Description |
|---|---|---|
_alert_cooldown_sec | 60.0 | Minimum time between alerts for the same fingerprint. |
_escalation_threshold | 5 | Number of occurrences required before a non-critical error alerts. |
_deduplication_window_sec | 300 | Time window for grouping related error events. |
Sources: src/flext_observability/services/error_handling.py:75-79, tests/integration/test_phase_11_integration.py:73-75
Severity Escalation
The system dynamically adjusts the severity of an error based on its frequency. This pattern ensures that a repetitive "Warning" eventually becomes an "Error" or "Critical" issue if left unresolved.
Escalation Logic
The resolve_escalated_severity method applies multipliers to the base _escalation_threshold:
- Level 1 (Threshold x 1): Elevates to
WARNING. - Level 2 (Threshold x 2): Elevates to
ERROR. - Level 3 (Threshold x 3): Elevates to
CRITICAL.
Implementation and Usage
Developers can interact with the system through the FlextObservabilityErrorHandling facade or through the high-level FlextObservability class.
Functional Example
In a standard business workflow, errors are recorded and checked against the alerting policy:
from flext_observability import FlextObservability, c
# Create an alert via simple API
alert_result = FlextObservability.flext_alert(
source="database",
title="Connection Failed",
message="Timed out after 30s",
severity=c.Observability.AlertLevel.ERROR,
)
if alert_result.success:
# Process successful recording
pass
Sources: examples/01_functional.py:31-37, examples/02_solid_observability_demo.py:101-118
Integration with Context
The error handler automatically attempts to inject a correlation_id from the FlextObservabilityContext when recording errors. This links error events to specific requests or traces, facilitating easier root-cause analysis.
Sources: src/flext_observability/services/error_handling.py:141-147, tests/integration/test_phase_11_integration.py:133-143
Summary
The Error Handling & Alerting module provides a robust layer of intelligence between raw exceptions and operator notifications. By implementing automated fingerprinting, configurable cooldowns, and a tiered escalation strategy, it effectively balances the need for comprehensive visibility with the requirement to minimize noise in production environments. Significant features include seamless integration with the FLEXT context for correlation and a railway-oriented internal architecture that prevents observability failures from impacting primary business logic.
Sources: src/flext_observability/services/error_handling.py:1-25, src/flext_observability/README.md:1-15