werkstatt
95 lines
2.4 KiB
Python
95 lines
2.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Parse systemd timeout errors for mission-control-2.service from journalctl.
|
|
|
|
Captures lines containing "State 'stop-sigterm' timed out" and appends them
|
|
to ~/logs/mc2-timeout.log with ISO timestamps.
|
|
|
|
Permission errors are handled gracefully.
|
|
|
|
Uses user journal (systemctl --user) to query logs.
|
|
"""
|
|
|
|
import os
|
|
import subprocess
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
|
|
LOG_DIR = Path.home() / "logs"
|
|
LOG_FILE = LOG_DIR / "mc2-timeout.log"
|
|
SERVICE_NAME = "mission-control-2.service"
|
|
SEARCH_PATTERN = "State 'stop-sigterm' timed out"
|
|
JOURNALCTL_LIMIT = 1000
|
|
|
|
|
|
def ensure_log_dir() -> None:
|
|
"""Create log directory if it doesn't exist."""
|
|
LOG_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def get_timeout_entries() -> list[str]:
|
|
"""Run journalctl --user and return lines matching the timeout pattern."""
|
|
cmd = [
|
|
"journalctl",
|
|
"--user",
|
|
"-u", SERVICE_NAME,
|
|
"--no-pager",
|
|
"-n", str(JOURNALCTL_LIMIT),
|
|
]
|
|
try:
|
|
result = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
print("journalctl timed out")
|
|
return []
|
|
except PermissionError:
|
|
print("Permission denied: journalctl requires access to user logs")
|
|
return []
|
|
|
|
if result.returncode != 0:
|
|
if "Permission denied" in result.stderr:
|
|
print("Permission denied: journalctl requires access to user logs")
|
|
else:
|
|
print(f"journalctl failed: {result.stderr.strip()}")
|
|
return []
|
|
|
|
return [
|
|
line
|
|
for line in result.stdout.splitlines()
|
|
if SEARCH_PATTERN in line
|
|
]
|
|
|
|
|
|
def write_to_log(entries: list[str]) -> int:
|
|
"""Append entries to log file with ISO timestamps. Returns count written."""
|
|
ensure_log_dir()
|
|
|
|
timestamp = datetime.now(timezone.utc).isoformat()
|
|
written = 0
|
|
|
|
with LOG_FILE.open("a", encoding="utf-8") as f:
|
|
for entry in entries:
|
|
f.write(f"[{timestamp}] {entry}\n")
|
|
written += 1
|
|
|
|
return written
|
|
|
|
|
|
def main() -> None:
|
|
entries = get_timeout_entries()
|
|
if not entries:
|
|
print("No timeout entries found.")
|
|
return
|
|
|
|
count = write_to_log(entries)
|
|
print(f"Appended {count} timeout entry/ies to {LOG_FILE}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|