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

1import os 

2import json 

3import datetime 

4from typing import Any 

5 

6 

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) 

12 

13 if "timestamp" not in alert_data: 

14 alert_data["timestamp"] = datetime.utcnow().isoformat() + "Z" 

15 

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) 

22 

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}")