alert_engine.py
"""
AlertEngine: streaming log deduplication and rate-limited alerting engine.
Pure standard library, no background threads. Uses externally supplied 'now' times.
"""
from future import annotations
import re
import hashlib
from collections import deque, defaultdict
from dataclasses import dataclass
from typing import Dict, Tuple, Optional, List, Any
Severity mapping
_SEVERITY_ORDER = {
"debug": 10,
"info": 20,
"warn": 30,
"warning": 30, # accept alternative
"error": 40,
"critical": 50,
}
MAX_SAMPLE_LEN = 200
MAX_FINGERPRINTS = 100000 # deterministic memory cap; see design note
_uuid_re = re.compile(r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b")
_ipv4_re = re.compile(r"\b(?:\d{1,3}.){3}\d{1,3}\b")
_hexid_re = re.compile(r"\b[0-9a-fA-F]{8,}\b")
_digits_re = re.compile(r"\d+")
def _normalize_message(msg: str) -> str:
# Replace UUIDs, IPs, long hex ids, then digit runs. Order matters.
s = _uuid_re.sub("<UUID>", msg)
s = _ipv4_re.sub("<IP>", s)
s = _hexid_re.sub("<HEX>", s)
s = _digits_re.sub("<NUM>", s)
return s
def _fingerprint_for(service: str, severity: str, message: str) -> Tuple[str, str]:
normalized = _normalize_message(message)
key_str = f"{service}|{severity}|{normalized}"
fp = hashlib.sha1(key_str.encode("utf-8", errors="ignore")).hexdigest()
return fp, normalized
@dataclass
class FPEntry:
service: str
severity: str
normalized: str
first_seen: float
last_seen: float
count: int
sample_message: str
suppressed_since_emit: int
burst_times: deque # timestamps (floats) of suppressed duplicates for burst detection
last_emitted: Optional[float]
class AlertEngine:
"""AlertEngine(config: dict, now: float)
config keys:
- window_seconds: float
- max_alerts_per_window: int
- dedup_seconds: float
- severity_floor: one of debug, info, warn, error, critical
- burst_escalation: optional dict {"count": int, "within_seconds": float}
Policy for non-monotonic now: the engine maintains an internal watermark (the maximum
now seen). If ingest() or flush() is called with a now earlier than the watermark,
the engine treats the provided now as the watermark (i.e. it clamps time forward).
This ensures emitted alerts are never produced out-of-order by time and makes
behavior deterministic for late events.
Eviction policy: to bound memory, a deterministic cap MAX_FINGERPRINTS is enforced.
When exceeded, the engine evicts fingerprints with the oldest (smallest) last_seen,
breaking ties by first_seen. This is deterministic and avoids unbounded growth.
"""
def __init__(self, config: dict, now: float):
# validate config minimally
try:
self.window_seconds = float(config["window_seconds"])
self.max_alerts_per_window = int(config["max_alerts_per_window"])
self.dedup_seconds = float(config["dedup_seconds"])
self.severity_floor = str(config["severity_floor"]).lower()
if self.severity_floor not in _SEVERITY_ORDER:
raise KeyError
except Exception:
raise ValueError("invalid config: required keys window_seconds, max_alerts_per_window, dedup_seconds, severity_floor")
be = config.get("burst_escalation")
if be is not None:
try:
self.burst_count = int(be["count"])
self.burst_within = float(be["within_seconds"])
if self.burst_count <= 0 or self.burst_within <= 0:
raise ValueError
except Exception:
raise ValueError("invalid burst_escalation config")
else:
self.burst_count = None
self.burst_within = None
# internal state
self._now_watermark = float(now)
self._fp_store: Dict[str, FPEntry] = {}
# per-key (service,severity) queue of emission timestamps for sliding window
self._key_emissions: Dict[Tuple[str, str], deque] = defaultdict(deque)
# statistics
self._total_ingested = 0
self._total_emitted = 0
self._total_suppressed = 0
def _clamp_now(self, now: float) -> float:
if now < self._now_watermark:
# policy: treat as late; clamp forward to watermark
return self._now_watermark
self._now_watermark = now
return now
def _ensure_eviction(self) -> None:
if len(self._fp_store) <= MAX_FINGERPRINTS:
return
# deterministic eviction: sort by last_seen, then first_seen
items = sorted(self._fp_store.items(), key=lambda kv: (kv[1].last_seen, kv[1].first_seen))
to_evict = len(self._fp_store) - MAX_FINGERPRINTS
for i in range(to_evict):
k = items[i][0]
del self._fp_store[k]
def ingest(self, event: dict, now: float) -> List[dict]:
"""Process a single event and return a list of emitted alerts (may be empty).
Policy for malformed input: the engine counts the event as ingested but drops it
and returns an empty list when required fields are missing or severity is invalid.
"""
self._total_ingested += 1
now = self._clamp_now(float(now))
alerts: List[dict] = []
# minimal validation
if not isinstance(event, dict):
return []
required = ("timestamp", "severity", "service", "message")
for r in required:
if r not in event:
return []
try:
ev_ts = float(event["timestamp"])
severity = str(event["severity"]).lower()
service = str(event["service"])
message = str(event["message"])
labels = event.get("labels")
if labels is not None and not isinstance(labels, dict):
labels = None
except Exception:
return []
if severity not in _SEVERITY_ORDER:
# explicit handling: unknown severity -> drop event
return []
# severity floor
if _SEVERITY_ORDER[severity] < _SEVERITY_ORDER[self.severity_floor]:
# dropped before any other processing
return []
# normalize and fingerprint
fp, normalized = _fingerprint_for(service, severity, message)
# ensure sample_message truncated without affecting fingerprint
sample_message = message[:MAX_SAMPLE_LEN]
entry = self._fp_store.get(fp)
if entry is None:
entry = FPEntry(
service=service,
severity=severity,
normalized=normalized,
first_seen=ev_ts,
last_seen=ev_ts,
count=1,
sample_message=sample_message,
suppressed_since_emit=0,
burst_times=deque(),
last_emitted=None,
)
self._fp_store[fp] = entry
else:
# update timestamps/counters
entry.count += 1
entry.last_seen = ev_ts
# keep the first sample_message
if len(entry.sample_message) < MAX_SAMPLE_LEN:
# try to fill sample if initial was short
entry.sample_message = (entry.sample_message + " | " + sample_message)[:MAX_SAMPLE_LEN]
# dedup check: if within dedup_seconds of last_seen (using event timestamp), treat as duplicate
is_duplicate = (ev_ts - entry.last_seen) <= self.dedup_seconds if entry.count > 1 else False
# Note: last_seen has already been set to ev_ts; for duplicate logic we should use prior last_seen
# To implement correctly, recompute: if count==1 -> not duplicate; else if ev_ts - prev_last_seen <= dedup_seconds
if entry.count == 1:
is_duplicate = False
else:
# we need the previous last_seen: approximate by checking ev_ts - (entry.last_seen or ev_ts) but we updated last_seen.
# Simpler: we consider duplicates as events whose event.timestamp is within dedup_seconds of the previous event's timestamp.
# Since we updated, we cannot recover prior; fix by keeping last_seen_old temporary earlier.
pass
# The above logic is awkward because we updated last_seen early. To correct, redo with proper previous tracking.
# Recompute with explicit storage.
# Rebuild: get previous entry snapshot
prev_last_seen = entry.last_seen
prev_count = entry.count
# Recompute properly: if prev_count > 1 then previous last_seen is not accessible; to avoid complexity,
# we will treat duplicates using now (processing time) relative to entry.last_seen stored before current event.
# To do that, we need to fetch last_seen before update. Simpler approach: when retrieving entry, capture previous_last_seen earlier.
# To fix cleanly, restructure above: redo retrieval with correct sequence.
# Implement by redoing the function with corrected sequence: idempotent approach below.
# ----- Reimplementing per-event core with correct temporal bookkeeping -----
# revert effects of earlier updates on this entry
# Recreate or adjust using a fresh read from store (we already have entry), but we saved first_seen earlier.
# For simplicity, reconstruct from stored fields without prior mistakes.
stored_entry = self._fp_store[fp]
# reconstruct prior state by assuming stored_entry.count includes this event; to get previous_count subtract 1
previous_count = stored_entry.count - 1
# previous_last_seen we cannot get reliably now; but we can treat duplicate detection using now (processing time)
# Use policy: duplicates are detected when now - stored_entry.last_seen <= dedup_seconds, where stored_entry.last_seen is BEFORE updating to ev_ts.
# To achieve this, we stored last_seen equal to ev_ts above; but if previous_count>=1, previous last_seen equals stored_entry.first_seen for count==1,
# or something else. This complexity arises because we updated fields prematurely.
# To simplify and make deterministic: we will consider duplicates if now - entry.last_seen <= dedup_seconds using last_seen prior to updating.
# Because we cannot recover prior last_seen, we will instead maintain dedup detection using a separate last_seen_map keyed by fingerprint storing the last event timestamp seen for that fingerprint.
# To avoid further confusion, move to a more straightforward implementation: maintain an auxiliary dict _last_event_ts per fp.
# Because the above became messy inside ingest, we will rewrite the ingest method cleanly by using helper state.
Rewriting the module with corrected structure below (single-file replacement)
We'll reimplement entire module cleanly to avoid mid-function confusion.
New implementation starts here
import heapq
class AlertEngine:
"""AlertEngine implementation (clean rewrite).
See class docstring earlier for behavior notes.
"""
def __init__(self, config: dict, now: float):
try:
self.window_seconds = float(config["window_seconds"])
self.max_alerts_per_window = int(config["max_alerts_per_window"])
self.dedup_seconds = float(config["dedup_seconds"])
self.severity_floor = str(config["severity_floor"]).lower()
if self.severity_floor not in _SEVERITY_ORDER:
raise KeyError
except Exception:
raise ValueError("invalid config")
be = config.get("burst_escalation")
if be is not None:
try:
self.burst_count = int(be["count"])
self.burst_within = float(be["within_seconds"])
if self.burst_count <= 0 or self.burst_within <= 0:
raise ValueError
except Exception:
raise ValueError("invalid burst_escalation config")
else:
self.burst_count = None
self.burst_within = None
self._now_watermark = float(now)
self._fp_store: Dict[str, FPEntry] = {}
self._last_event_ts: Dict[str, float] = {}
self._key_emissions: Dict[Tuple[str, str], deque] = defaultdict(deque)
self._total_ingested = 0
self._total_emitted = 0
self._total_suppressed = 0
def _clamp_now(self, now: float) -> float:
if now < self._now_watermark:
return self._now_watermark
self._now_watermark = now
return now
def _prune_key_emissions(self, key: Tuple[str, str], now: float) -> None:
q = self._key_emissions.get(key)
if not q:
return
cutoff = now - self.window_seconds
while q and q[0] < cutoff:
q.popleft()
def _emit_alert(self, kind: str, fp: str, entry: FPEntry, count: int) -> dict:
alert = {
"kind": kind,
"fingerprint": fp,
"key": (entry.service, entry.severity),
"first_seen": entry.first_seen,
"last_seen": entry.last_seen,
"count": count,
"sample_message": entry.sample_message,
}
self._total_emitted += 1
return alert
def ingest(self, event: dict, now: float) -> List[dict]:
self._total_ingested += 1
now = self._clamp_now(float(now))
alerts: List[dict] = []
# validate
if not isinstance(event, dict):
return []
for k in ("timestamp", "severity", "service", "message"):
if k not in event:
return []
try:
ev_ts = float(event["timestamp"])
severity = str(event["severity"]).lower()
service = str(event["service"])
message = str(event["message"])
except Exception:
return []
if severity not in _SEVERITY_ORDER:
return []
if _SEVERITY_ORDER[severity] < _SEVERITY_ORDER[self.severity_floor]:
return []
fp, normalized = _fingerprint_for(service, severity, message)
sample_message = message[:MAX_SAMPLE_LEN]
prev_ts = self._last_event_ts.get(fp)
is_duplicate = prev_ts is not None and (ev_ts - prev_ts) <= self.dedup_seconds
# create or update entry
entry = self._fp_store.get(fp)
if entry is None:
entry = FPEntry(
service=service,
severity=severity,
normalized=normalized,
first_seen=ev_ts,
last_seen=ev_ts,
count=1,
sample_message=sample_message,
suppressed_since_emit=0,
burst_times=deque(),
last_emitted=None,
)
self._fp_store[fp] = entry
else:
entry.count += 1
# keep first_seen as is
entry.last_seen = ev_ts
# keep a stable sample, don't overwrite, but if empty, set
if not entry.sample_message:
entry.sample_message = sample_message
# update last_event_ts for dedup checks next time
self._last_event_ts[fp] = ev_ts
# Evict if necessary
if len(self._fp_store) > MAX_FINGERPRINTS:
# deterministic eviction by last_seen, first_seen
items = sorted(self._fp_store.items(), key=lambda kv: (kv[1].last_seen, kv[1].first_seen))
to_evict = len(self._fp_store) - MAX_FINGERPRINTS
for i in range(to_evict):
del self._fp_store[items[i][0]]
if is_duplicate:
# Suppress duplicate
entry.suppressed_since_emit += 1
self._total_suppressed += 1
# track burst times using processing time 'now'
if self.burst_count is not None:
bt = entry.burst_times
bt.append(now)
cutoff = now - self.burst_within
while bt and bt[0] < cutoff:
bt.popleft()
if len(bt) > self.burst_count:
# emit escalation: bypass rate limit, reset burst counter
alert = self._emit_alert("escalation", fp, entry, entry.suppressed_since_emit)
alerts.append(alert)
entry.suppressed_since_emit = 0
bt.clear()
entry.last_emitted = now
return alerts
# Not a duplicate: candidate for new alert. Check per-key rate limit.
key = (service, severity)
self._prune_key_emissions(key, now)
q = self._key_emissions[key]
if len(q) < self.max_alerts_per_window:
# allowed
q.append(now)
alert = self._emit_alert("new", fp, entry, 1)
alerts.append(alert)
entry.last_emitted = now
entry.suppressed_since_emit = 0
else:
# rate-limited: suppress and count
entry.suppressed_since_emit += 1
self._total_suppressed += 1
# do not emit immediately; will appear on flush
return alerts
def flush(self, now: float) -> List[dict]:
now = self._clamp_now(float(now))
alerts: List[dict] = []
# For each fingerprint, if dedup window closed and there are suppressed events, emit rate_limit_notice
to_delete = []
for fp, entry in list(self._fp_store.items()):
# if last_seen older than dedup_seconds
if now - entry.last_seen >= self.dedup_seconds and entry.suppressed_since_emit > 0:
alert = self._emit_alert("rate_limit_notice", fp, entry, entry.suppressed_since_emit)
alerts.append(alert)
entry.suppressed_since_emit = 0
entry.last_emitted = now
return alerts
def stats(self) -> dict:
return {
"total_ingested": self._total_ingested,
"total_emitted": self._total_emitted,
"total_suppressed": self._total_suppressed,
"active_fingerprints": len(self._fp_store),
}
test_alert_engine.py
import unittest
class TestAlertEngine(unittest.TestCase):
def setUp(self):
self.config = {
"window_seconds": 60.0,
"max_alerts_per_window": 2,
"dedup_seconds": 10.0,
"severity_floor": "info",
"burst_escalation": {"count": 3, "within_seconds": 5.0},
}
self.engine = AlertEngine(self.config, now=0.0)
def test_dedup_collapsing(self):
now = 1.0
e1 = {"timestamp": now, "severity": "info", "service": "svc", "message": "user 123 logged in"}
out = self.engine.ingest(e1, now)
self.assertEqual(len(out), 1)
fp = out[0]["fingerprint"]
# duplicate within dedup_seconds
e2 = {"timestamp": now + 2, "severity": "info", "service": "svc", "message": "user 456 logged in"}
out2 = self.engine.ingest(e2, now + 2)
self.assertEqual(out2, [])
stats = self.engine.stats()
self.assertEqual(stats["total_suppressed"], 1)
def test_rate_limiting_at_boundary(self):
now = 10.0
# allow two alerts per window
for i in range(2):
e = {"timestamp": now + i, "severity": "error", "service": "s", "message": f"msg{i}"}
out = self.engine.ingest(e, now + i)
self.assertEqual(len(out), 1)
# third should be suppressed
e3 = {"timestamp": now + 3, "severity": "error", "service": "s", "message": "msg3"}
out3 = self.engine.ingest(e3, now + 3)
self.assertEqual(out3, [])
stats = self.engine.stats()
self.assertEqual(stats["total_suppressed"], 1)
# after window passes, flush should produce summary for suppressed
out_flush = self.engine.flush(now + 70)
self.assertTrue(any(a["kind"] == "rate_limit_notice" for a in out_flush))
def test_burst_escalation(self):
now = 100.0
e = {"timestamp": now, "severity": "warn", "service": "svcB", "message": "hit 1"}
out = self.engine.ingest(e, now)
self.assertEqual(len(out), 1)
fp = out[0]["fingerprint"]
# produce suppressed duplicates rapidly to exceed burst count (count=3)
for i in range(4):
ed = {"timestamp": now + 1 + i, "severity": "warn", "service": "svcB", "message": f"hit {10+i}"}
res = self.engine.ingest(ed, now + 1 + i)
if res:
# one of them should be escalation when threshold passed
kinds = {r["kind"] for r in res}
self.assertIn("escalation", kinds)
break
else:
self.fail("escalation not emitted")
def test_flush_idempotency(self):
now = 200.0
e = {"timestamp": now, "severity": "error", "service": "sF", "message": "a"}
self.engine.ingest(e, now)
# suppress next by rate limiting
# create two more alerts to reach limit
self.engine.ingest({"timestamp": now+1, "severity": "error", "service": "sF", "message": "b"}, now+1)
self.engine.ingest({"timestamp": now+2, "severity": "error", "service": "sF", "message": "c"}, now+2)
out1 = self.engine.flush(now+30)
out2 = self.engine.flush(now+31)
# second flush should be idempotent
self.assertEqual(out1, out2)
def test_out_of_order_timestamps(self):
now = 300.0
e1 = {"timestamp": now, "severity": "info", "service": "oo", "message": "x1"}
out1 = self.engine.ingest(e1, now)
self.assertEqual(len(out1), 1)
# supply earlier now (late event). Engine clamps time and should not go backward.
out2 = self.engine.ingest({"timestamp": now-50, "severity": "info", "service": "oo", "message": "x2"}, now-50)
# must not crash and not emit out of order
self.assertIsInstance(out2, list)
def test_severity_floor_filtering(self):
now = 400.0
e = {"timestamp": now, "severity": "debug", "service": "sD", "message": "dmsg"}
out = self.engine.ingest(e, now)
self.assertEqual(out, [])
def test_malformed_input(self):
now = 500.0
out = self.engine.ingest({"severity": "info"}, now)
self.assertEqual(out, [])
out2 = self.engine.ingest("not a dict", now)
self.assertEqual(out2, [])
def test_eviction_behavior(self):
# Reduce MAX_FINGERPRINTS for test by temporarily patching global
global MAX_FINGERPRINTS
old = MAX_FINGERPRINTS
MAX_FINGERPRINTS = 5
try:
eng = AlertEngine(self.config, now=0.0)
for i in range(10):
e = {"timestamp": i, "severity": "info", "service": f"svc{i}", "message": "m"}
eng.ingest(e, float(i))
stats = eng.stats()
self.assertLessEqual(stats["active_fingerprints"], 5)
finally:
MAX_FINGERPRINTS = old
if name == 'main':
unittest.main()
DESIGN NOTE
"""
Data structures: a dict keyed by fingerprint stores FPEntry dataclasses (first/last seen, counts,
sample message, suppressed counter, burst timestamps). Per-key (service,severity) emission queues
hold recent emission timestamps in deques for O(1) sliding-window pruning. A last-event timestamp
map is used to detect deduplicates deterministically.
Time complexity: ingest is O(1) average — hashing and regex work on the message (linear in message
length), deque operations for emission and burst are amortized O(1). flush is O(N) in active
fingerprints. Memory eviction is deterministic: when the fingerprint store exceeds MAX_FINGERPRINTS
(it is a global cap), the engine evicts the oldest fingerprints by (last_seen, first_seen).
Eviction policy: deterministic LRU-like by last_seen tie-broken by first_seen; this bounds memory
and is predictable. Trade-off: eviction uses sorting when the cap is exceeded which is O(M log M)
for M items; MAX_FINGERPRINTS keeps this bounded. A trade-off accepted: dedup detection uses the
event timestamp and a watermark policy for non-monotonic 'now' (clamping now to the maximum seen).
This simplifies ordering guarantees (no alerts emitted out of internal order) but means very late
supplied times are treated as if they occurred at the watermark, slightly altering temporal fidelity.
"""
