"""Systemd journal log ingestion."""
import subprocess
import json
import logging
from datetime import datetime, timedelta, timezone
from typing import Optional
from ..models import LogEntry, SourceType, Severity
logger = logging.getLogger(__name__)
def _parse_severity(priority: str) -> Severity:
"""Map systemd priority to our severity."""
try:
prio = int(priority)
if prio <= 2:
return Severity.INFO
elif prio <= 4:
return Severity.WARNING
elif prio <= 5:
return Severity.ERROR
else:
return Severity.CRITICAL
except (ValueError, TypeError):
return Severity.INFO
class SystemdIngestor:
"""Collect logs from systemd journal."""
# Units that fail harmlessly and should be ignored
IGNORED_UNITS = frozenset({
"NetworkManager-wait-online.service", # Boot-time timeout, network works fine
})
def get_logs(
self,
unit: Optional[str] = None,
since_minutes: int = 60,
max_lines: int = 200,
priority: Optional[str] = None
) -> list[LogEntry]:
"""Get recent journal entries.
Args:
unit: Specific systemd unit (e.g., 'docker.service')
since_minutes: Lookback window
max_lines: Maximum lines to return
priority: Priority filter (e.g., 'err', 'warning')
"""
try:
since = f"-{since_minutes}m"
args = [
"journalctl",
"--since", since,
"--no-pager",
"--output", "json",
"--priority", priority or "info",
"-n", str(max_lines),
]
if unit:
args.extend(["-u", unit])
result = subprocess.run(args, capture_output=True, text=True, timeout=30)
if result.returncode != 0:
logger.warning("journalctl failed: %s", result.stderr)
return []
entries = []
for line in result.stdout.strip().split("\n"):
if not line.strip():
continue
try:
data = json.loads(line)
# Parse timestamp
ts = data.get("__REALTIME_TIMESTAMP", "")
if ts:
epoch_us = int(ts)
timestamp = datetime.fromtimestamp(epoch_us / 1000000, tz=timezone.utc)
else:
timestamp = datetime.now(timezone.utc)
entries.append(LogEntry(
timestamp=timestamp,
source=SourceType.SYSTEMD,
source_name=data.get("UNIT", "system"),
severity=_parse_severity(data.get("PRIORITY", "6")),
message=data.get("MESSAGE", ""),
raw=data.get("_COMMUNITY", "") or data.get("_SYSTEMD_UNIT", "")
))
except (json.JSONDecodeError, KeyError, ValueError):
# Fallback for malformed lines
entries.append(LogEntry(
timestamp=datetime.now(timezone.utc),
source=SourceType.SYSTEMD,
source_name=unit or "system",
severity=Severity.INFO,
message=line[:500],
raw=line
))
return entries
except FileNotFoundError:
logger.warning("journalctl not found")
return []
except subprocess.TimeoutExpired:
logger.warning("journalctl timed out")
return []
def get_unit_status(self, unit: str) -> dict:
"""Get unit status from systemctl."""
try:
result = subprocess.run(
["systemctl", "status", unit],
capture_output=True, text=True, timeout=10
)
# Parse first line for active status
first_line = result.stdout.split("\n")[0] if result.stdout else ""
return {
"unit": unit,
"raw_status": first_line,
"active": "active" in first_line.lower(),
"enabled": "enabled" in result.stdout.lower(),
}
except (FileNotFoundError, subprocess.TimeoutExpired):
return {"unit": unit, "raw_status": "error", "active": False, "enabled": False}
def get_failed_units(self) -> list[dict]:
"""List failed systemd units."""
try:
result = subprocess.run(
["systemctl", "--failed", "--no-pager", "--plain"],
capture_output=True, text=True, timeout=10
)
units = []
for line in result.stdout.strip().split("\n")[1:]: # Skip header
if not line.strip():
continue
parts = line.split()
# Skip lines that don't look like unit names
if len(parts) >= 3 and "." in parts[0]:
unit_name = parts[0]
# Skip known-harmless failures
if unit_name in self.IGNORED_UNITS:
continue
units.append({
"unit": unit_name,
"load": parts[1],
"active": parts[2],
})
return units
except (FileNotFoundError, subprocess.TimeoutExpired):
return []