MrYoshii
c756195530
Kompletter Neustart als Python-Projekt. Portiert von zfs-monitor Bash v3, saubere Architektur: - config.py: TOML-Config mit dataclasses - state.py: zentrales State-Management (JSON statt Einzeldateien) - notify.py: Telegram + Persistent Alerting als wiederverwendbare Klasse - checks/zfs.py: ZFS Pool-Health, I/O-Fehler, Kapazität - checks/smart.py: SATA + NVMe SMART-Monitoring - checks/nut.py: NUT/USV-Monitoring (APC Back-UPS RS 1500G)
100 lines
3.3 KiB
Python
100 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
from datetime import datetime
|
|
|
|
from config import Config
|
|
from state import State
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
ICONS = {"CRITICAL": "🔴", "WARNING": "🟡", "INFO": "🟢"}
|
|
|
|
|
|
class Alerter:
|
|
def __init__(self, config: Config, state: State) -> None:
|
|
self.config = config
|
|
self.state = state
|
|
|
|
# ── Persistent Alerting ───────────────────────────────────
|
|
# Gleiche Logik wie im Bash-Script:
|
|
# - Neuer Fehler → sofort melden, runs = 0
|
|
# - Fehler bleibt → runs++, melden wenn runs >= repeat_runs
|
|
# - Fehler weg → Entwarnung, State löschen
|
|
|
|
def should_alert(self, key: str, level: str) -> bool:
|
|
repeat = (
|
|
self.config.alert.critical_repeat_runs
|
|
if level == "CRITICAL"
|
|
else self.config.alert.warning_repeat_runs
|
|
)
|
|
active = self.state.get(f"{key}_active")
|
|
runs: int = self.state.get(f"{key}_runs", 0)
|
|
|
|
if active != "1":
|
|
self.state.set(f"{key}_active", "1")
|
|
self.state.set(f"{key}_runs", 0)
|
|
return True
|
|
|
|
runs += 1
|
|
self.state.set(f"{key}_runs", runs)
|
|
|
|
if repeat == 0:
|
|
return False
|
|
if runs >= repeat:
|
|
self.state.set(f"{key}_runs", 0)
|
|
return True
|
|
return False
|
|
|
|
def clear_alert(self, key: str) -> bool:
|
|
"""Clear state and return True if the alert was previously active."""
|
|
was_active = self.state.get(f"{key}_active") == "1"
|
|
self.state.delete(f"{key}_active", f"{key}_runs")
|
|
return was_active
|
|
|
|
# ── Sending ───────────────────────────────────────────────
|
|
|
|
def send_alert(self, level: str, title: str, body: str, key: str = "") -> None:
|
|
icon = ICONS.get(level, "⚪")
|
|
suffix = ""
|
|
if key:
|
|
runs: int = self.state.get(f"{key}_runs", 0)
|
|
if runs > 0:
|
|
suffix = f" <i>(Erinnerung #{runs})</i>"
|
|
|
|
ts = datetime.now().strftime("%d.%m.%Y %H:%M:%S")
|
|
msg = (
|
|
f"{icon} <b>[{level}] {self.config.hostname}</b>{suffix}\n"
|
|
f"<b>{title}</b>\n\n"
|
|
f"{body}\n\n"
|
|
f"🕐 {ts}"
|
|
)
|
|
log.info("ALERT %s: %s", level, title)
|
|
self._send_telegram(msg)
|
|
|
|
def _send_telegram(self, text: str) -> bool:
|
|
url = f"https://api.telegram.org/bot{self.config.telegram.token}/sendMessage"
|
|
data = urllib.parse.urlencode({
|
|
"chat_id": self.config.telegram.chat_id,
|
|
"text": text,
|
|
"parse_mode": "HTML",
|
|
}).encode()
|
|
|
|
for attempt in range(1, 4):
|
|
try:
|
|
with urllib.request.urlopen(url, data=data, timeout=10) as resp:
|
|
result = json.loads(resp.read())
|
|
if result.get("ok"):
|
|
return True
|
|
except Exception as e:
|
|
log.warning("Telegram Fehler (Versuch %d/3): %s", attempt, e)
|
|
if attempt < 3:
|
|
time.sleep(attempt * 2)
|
|
|
|
log.error("Telegram: Alert konnte nicht gesendet werden!")
|
|
return False
|