Coverage for ids_iforest/logging_utils.py: 26%
19 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-03 16:19 +0000
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-03 16:19 +0000
1import os
2import json
3import datetime
4from typing import Any
7def append_json_alert(jsonl_path: str, **alert_data: Any) -> None:
8 """
9 Append an alert to the JSONL file (one JSON object per line) for Promtail/Loki.
10 """
11 os.makedirs(os.path.dirname(jsonl_path), exist_ok=True)
13 if "timestamp" not in alert_data:
14 alert_data["timestamp"] = datetime.utcnow().isoformat() + "Z"
16 # Ensure pure-Python types for JSON (avoid numpy scalars)
17 for k, v in list(alert_data.items()):
18 if isinstance(v, float):
19 alert_data[k] = float(v)
20 elif isinstance(v, int):
21 alert_data[k] = int(v)
23 try:
24 with open(jsonl_path, "a", encoding="utf-8") as f:
25 f.write(json.dumps(alert_data) + "\n")
26 f.flush() # best-effort real-time tailing
27 except Exception as e:
28 # Fall back to stderr; don't crash the detector
29 print(f"[ids-iforest] Error writing alert to {jsonl_path}: {e}")