"""Watchdog mode - continuous monitoring with Telegram alerts."""
import json
import logging
import signal
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
import requests
from .config import LLMConfig
from .ingest import DockerIngestor, SystemdIngestor, SystemMetricsIngestor
from .reason import ReasoningEngine
from .models import Severity
logger = logging.getLogger(__name__)
# Telegram bot config
TELEGRAM_TOKEN = None
TELEGRAM_CHAT = None
def load_telegram_config():
"""Load Telegram credentials."""
global TELEGRAM_TOKEN, TELEGRAM_CHAT
config_path = Path.home() / ".hermes" / "config" / "telegram.json"
if config_path.exists():
try:
with open(config_path) as f:
data = json.load(f)
TELEGRAM_TOKEN = data.get("token")
TELEGRAM_CHAT = data.get("chat_id")
return True
except (json.JSONDecodeError, KeyError) as e:
logger.error("Failed to load Telegram config: %s", e)
# Fallback to env vars
import os
TELEGRAM_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN")
TELEGRAM_CHAT = os.environ.get("TELEGRAM_CHAT_ID")
return TELEGRAM_TOKEN is not None
def send_telegram_alert(message: str) -> bool:
"""Send alert via Telegram or log to file."""
if not TELEGRAM_TOKEN or not TELEGRAM_CHAT:
logger.warning("Telegram not configured - logging alert to file")
# Write to alert log for cron to pick up
alert_entry = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"message": message,
}
log_path = Path.home() / ".debug-oracle" / "alerts.jsonl"
try:
with open(log_path, "a") as f:
f.write(json.dumps(alert_entry) + "\n")
logger.info(f"Alert logged: {log_path}")
except IOError as e:
logger.error(f"Failed to write alert log: {e}")
return False
try:
url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage"
payload = {
"chat_id": TELEGRAM_CHAT,
"text": message,
"parse_mode": "HTML",
}
response = requests.post(url, json=payload, timeout=30)
response.raise_for_status()
return True
except Exception as e:
logger.error("Failed to send Telegram alert: %s", e)
# Fallback to file
alert_entry = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"message": message,
}
log_path = Path.home() / ".debug-oracle" / "alerts.jsonl"
try:
with open(log_path, "a") as f:
f.write(json.dumps(alert_entry) + "\n")
except IOError as e:
logger.error(f"Failed to write alert log: {e}")
return False
class Watchdog:
"""Continuous monitoring with alerts."""
def __init__(
self,
check_interval: int = 300,
containers: list[str] = None,
metrics_thresholds: dict = None,
alert_target: str = "telegram",
llm_config: LLMConfig = None,
state_dir: Path = None,
):
self.check_interval = check_interval
self.containers = containers or []
self.metrics_thresholds = metrics_thresholds or {
"cpu_critical": 90.0,
"memory_critical": 90.0,
"disk_critical": 90.0,
}
self.alert_target = alert_target
self.llm_config = llm_config or LLMConfig()
self.state_dir = state_dir or Path.home() / ".debug-oracle"
self.state_dir.mkdir(parents=True, exist_ok=True)
self._running = False
self._last_alert = {}
self._alert_cooldown = 300 # 5 minutes between repeat alerts
# Load Telegram config if needed
if alert_target == "telegram":
load_telegram_config()
# State file for tracking alerts
self._state_file = self.state_dir / "watchdog_state.json"
self._load_state()
def _load_state(self):
"""Load watchdog state from disk."""
if self._state_file.exists():
try:
with open(self._state_file) as f:
self._last_alert = json.load(f)
except (json.JSONDecodeError, IOError) as e:
logger.error("Failed to load state: %s", e)
self._last_alert = {}
else:
self._last_alert = {}
def _save_state(self):
"""Save watchdog state to disk."""
try:
with open(self._state_file, "w") as f:
json.dump(self._last_alert, f, indent=2)
except IOError as e:
logger.error("Failed to save state: %s", e)
def _should_alert(self, alert_key: str) -> bool:
"""Check if we should send an alert (avoid spam)."""
now = time.time()
last = self._last_alert.get(alert_key, 0)
if now - last > self._alert_cooldown:
self._last_alert[alert_key] = now
self._save_state()
return True
return False
def _check_containers(self) -> list[str]:
"""Check Docker container health."""
issues = []
docker = DockerIngestor()
containers = self.containers or docker.get_containers()
for container in containers:
health = docker.get_health(container)
status = health.get("status", "unknown")
if "running" not in status:
alert_key = f"container:{container}"
if self._should_alert(alert_key):
issues.append(f"⚠️ Container {container}: {status}")
logger.warning("Container %s not running: %s", container, status)
return issues
def _check_systemd(self) -> list[str]:
"""Check systemd failed units."""
issues = []
systemd = SystemdIngestor()
failed = systemd.get_failed_units()
for unit in failed:
alert_key = f"systemd:{unit['unit']}"
if self._should_alert(alert_key):
issues.append(f"⚠️ Service {unit['unit']}: {unit['active']}")
logger.warning("Failed unit: %s (%s)", unit['unit'], unit['active'])
return issues
def _check_metrics(self) -> list[str]:
"""Check system metrics against thresholds."""
issues = []
metrics = SystemMetricsIngestor()
snapshot = metrics.get_snapshot()
if snapshot.cpu_percent > self.metrics_thresholds["cpu_critical"]:
alert_key = "metrics:cpu"
if self._should_alert(alert_key):
issues.append(f"⚠️ CPU critical: {snapshot.cpu_percent:.1f}%")
logger.warning("CPU critical: %.1f%%", snapshot.cpu_percent)
if snapshot.memory_percent > self.metrics_thresholds["memory_critical"]:
alert_key = "metrics:memory"
if self._should_alert(alert_key):
issues.append(f"⚠️ Memory critical: {snapshot.memory_percent:.1f}%")
logger.warning("Memory critical: %.1f%%", snapshot.memory_percent)
if snapshot.disk_percent > self.metrics_thresholds["disk_critical"]:
alert_key = "metrics:disk"
if self._should_alert(alert_key):
issues.append(f"⚠️ Disk critical: {snapshot.disk_percent:.1f}%")
logger.warning("Disk critical: %.1f%%", snapshot.disk_percent)
return issues
def _check_logs(self) -> list[str]:
"""Check for recent errors in Docker logs."""
issues = []
docker = DockerIngestor()
containers = self.containers or docker.get_containers()
for container in containers:
logs = docker.get_logs(container, since_minutes=5, tail=50)
critical_errors = [
l for l in logs
if l.severity in (Severity.CRITICAL, Severity.ERROR)
]
if critical_errors:
# Check if we've already alerted for this container recently
alert_key = f"logs:{container}"
if self._should_alert(alert_key):
recent = [e.message[:100] for e in critical_errors[:3]]
issues.append(f"⚠️ {container}: {len(critical_errors)} errors ({'; '.join(recent[:2])})")
logger.warning("Container %s has %d errors", container, len(critical_errors))
return issues
def _run_checks(self) -> list[str]:
"""Run all checks and return list of issues."""
all_issues = []
logger.info("Running watchdog checks...")
try:
all_issues.extend(self._check_containers())
except Exception as e:
logger.error("Container check failed: %s", e)
try:
all_issues.extend(self._check_systemd())
except Exception as e:
logger.error("Systemd check failed: %s", e)
try:
all_issues.extend(self._check_metrics())
except Exception as e:
logger.error("Metrics check failed: %s", e)
try:
all_issues.extend(self._check_logs())
except Exception as e:
logger.error("Log check failed: %s", e)
return all_issues
def _handle_issues(self, issues: list[str]):
"""Handle detected issues."""
if not issues:
logger.info("Watchdog: All checks passed")
return
# Format alert message
timestamp = datetime.now(timezone.utc).strftime("%H:%M UTC")
message = f"🚨 Debug Oracle Alert ({timestamp})\n\n"
message += f"{' | '.join(issues)}\n\n"
message += f"Run `python -m oracle investigate` for full analysis."
# Send alert
if self.alert_target == "telegram":
send_telegram_alert(message)
else:
logger.warning("Unknown alert target: %s", self.alert_target)
# Optional: Run oracle investigation for critical issues
if any("critical" in issue.lower() for issue in issues):
logger.info("Critical issue detected, running investigation...")
try:
engine = ReasoningEngine(self.llm_config)
# Collect logs for investigation
docker = DockerIngestor()
logs = docker.get_all_logs(self.containers or None, since_minutes=30, tail=200)
metrics = SystemMetricsIngestor().get_snapshot()
investigation = engine.analyze(
query="Critical issue detected by watchdog",
logs=[logs], # Wrap in list for compatibility
metrics=[metrics],
git_changes=[]
)
# Add investigation summary to next alert
summary = investigation.summary[:500]
followup = f"🔍 Investigation:\n{summary}"
send_telegram_alert(followup)
engine.close()
except Exception as e:
logger.error("Investigation failed: %s", e)
def run(self):
"""Main watchdog loop."""
self._running = True
# Handle graceful shutdown
def signal_handler(signum, frame):
logger.info("Received shutdown signal")
self._running = False
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
logger.info("Watchdog started (interval: %ds)", self.check_interval)
while self._running:
try:
issues = self._run_checks()
self._handle_issues(issues)
except Exception as e:
logger.error("Watchdog check failed: %s", e)
# Sleep in small increments to allow signal handling
for _ in range(self.check_interval * 10):
if not self._running:
break
time.sleep(0.1)
logger.info("Watchdog stopped")
def stop(self):
"""Stop the watchdog."""
self._running = False