205 lines
12 KiB
Python
205 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""Read-only health/reboot observer with durable heartbeats; never restarts the box."""
|
||
import argparse
|
||
from datetime import datetime
|
||
import hashlib
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import re
|
||
import signal
|
||
import subprocess
|
||
import tempfile
|
||
import threading
|
||
|
||
ORDERLY = re.compile(r"systemd-shutdown|Reached target.*(?:System Reboot|System Power Off)|systemd\[1\]: (?:Rebooting|Powering off)|reboot: Restarting system|reboot: Power down", re.I)
|
||
FAILURE = re.compile(r"Kernel panic - not syncing|Oops:|BUG: unable to handle|watchdog:.*(?:hard LOCKUP|soft lockup)|Out of memory: Killed process|oom-kill:|thermal.*critical", re.I)
|
||
EVIDENCE = re.compile(r"panic|oops|watchdog|oom|out of memory|thermal|overheat|shutdown|reboot|crash|segfault|I/O error|ext4.*error|mmc.*(?:error|timeout)|voltage|hung task", re.I)
|
||
|
||
|
||
def atomic_write(path, text):
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
name = None
|
||
try:
|
||
with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=path.parent, delete=False) as f:
|
||
name = f.name
|
||
os.chmod(name, 0o600)
|
||
f.write(text); f.flush(); os.fsync(f.fileno())
|
||
os.replace(name, path); name = None
|
||
if os.name == "posix":
|
||
fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
|
||
try: os.fsync(fd)
|
||
finally: os.close(fd)
|
||
finally:
|
||
if name: os.unlink(name)
|
||
|
||
|
||
def command(args, timeout=4, limit=24576):
|
||
try:
|
||
r = subprocess.run(args, capture_output=True, text=True, errors="replace", timeout=timeout,
|
||
env={**os.environ, "LC_ALL": "C", "SYSTEMD_COLORS": "0"})
|
||
return {"code": r.returncode, "text": r.stdout[-limit:], "error": r.stderr[-1024:]}
|
||
except (OSError, subprocess.TimeoutExpired) as e:
|
||
return {"code": -1, "text": "", "error": str(e)[:1024]}
|
||
|
||
|
||
def read(path, default=""):
|
||
try: return Path(path).read_text().strip()
|
||
except (OSError, ValueError): return default
|
||
|
||
|
||
def classify(evidence):
|
||
text = "\n".join(evidence.get(k, {}).get("text", "") for k in ("tail", "warnings", "pstore_new"))
|
||
if FAILURE.search(text): return "发现内核异常/OOM等线索，不能单独认定为重启原因"
|
||
if ORDERLY.search(text): return "发现正常关机或重启流程记录，发起者需结合原始日志核对"
|
||
return "重启原因未确认：无充分关机/故障证据，不能区分断电、硬复位等情况"
|
||
|
||
|
||
def transition(previous, boot_id, now, evidence=None):
|
||
if previous is not None and (previous.get("version") != 1 or type(previous.get("reboots")) is not int):
|
||
raise ValueError("Unsupported/corrupt saved state; preserve it for diagnosis")
|
||
state = dict(previous or {"version": 1, "reboots": 0, "started_at": now, "history": []})
|
||
old = state.get("boot_id")
|
||
event = "boot_baseline" if not old else "observer_restart_same_boot"
|
||
if old and old != boot_id:
|
||
event = "boot_change"; state["reboots"] += 1
|
||
state["history"] = (state.get("history", []) + [{"previous_boot_id": old, "boot_id": boot_id,
|
||
"previous_last_seen": state.get("last_seen"), "previous_uptime_seconds": state.get("uptime_seconds"),
|
||
"detected_at": now, "assessment": classify(evidence or {})}])[-32:]
|
||
state.update(boot_id=boot_id, observer_started_at=now)
|
||
return state, event
|
||
|
||
|
||
class Observer:
|
||
def __init__(self, logs, state_dir):
|
||
self.logs, self.state_dir = logs, state_dir
|
||
logs.mkdir(parents=True, exist_ok=True); state_dir.mkdir(parents=True, exist_ok=True)
|
||
self.state_file = state_dir / "overnight-state.json"
|
||
self.state = None
|
||
|
||
def log(self, event, **values):
|
||
now = datetime.now().astimezone()
|
||
path = self.logs / ("overnight-health-" + now.strftime("%Y%m%d") + ".log")
|
||
if path.exists() and path.stat().st_size > 8*1024*1024:
|
||
for i in (2, 1):
|
||
source = path.with_suffix(path.suffix + "." + str(i))
|
||
if source.exists(): os.replace(source, path.with_suffix(path.suffix + "." + str(i+1)))
|
||
os.replace(path, path.with_suffix(path.suffix + ".1"))
|
||
record = dict(time=now.isoformat(timespec="seconds"), event=event, **values)
|
||
with path.open("a", encoding="utf-8") as f:
|
||
f.write(json.dumps(record, ensure_ascii=False) + "\n"); f.flush(); os.fsync(f.fileno())
|
||
# Bound only our own logs; never touch kiosk/stream/application logs.
|
||
days = sorted(self.logs.glob("overnight-health-????????.log"))
|
||
for old in days[:-7]:
|
||
if old == path: continue
|
||
old.unlink(missing_ok=True)
|
||
for i in (1, 2, 3): old.with_suffix(old.suffix + "." + str(i)).unlink(missing_ok=True)
|
||
|
||
def prior_evidence(self, boot_id):
|
||
# Boot IDs remain meaningful even when RTC/NTP changes wall-clock time.
|
||
if not re.fullmatch(r"[0-9a-f-]{32,36}", boot_id):
|
||
return {"tail": {"text": "", "error": "invalid saved boot ID"}}
|
||
base = ["journalctl", "--no-pager", "-b", boot_id, "-o", "short-monotonic"]
|
||
return {"tail": command(base+["-n", "100"]), "warnings": command(base+["-p", "warning", "-n", "120"]),
|
||
"last": command(["last", "-x", "-n", "10"])}
|
||
|
||
def start(self):
|
||
boot_id = read("/proc/sys/kernel/random/boot_id")
|
||
if not re.fullmatch(r"[0-9a-f-]{36}", boot_id): raise ValueError("Cannot read kernel boot ID")
|
||
previous = json.loads(self.state_file.read_text()) if self.state_file.exists() else None
|
||
try:
|
||
with Path('/var/lib/vdi-observer/pstore.json').open('r', encoding='utf-8') as f:
|
||
pstore = json.loads(f.read(1024*1024))
|
||
except (OSError, ValueError) as e: pstore = {"read_error": str(e)}
|
||
# Compare contents, not capture time; an old ramoops archive is not a new crash.
|
||
texts = {root+"/"+name: entry["text"] for root, files in pstore.get("sources", {}).items()
|
||
for name, entry in files.items() if isinstance(entry, dict) and "text" in entry}
|
||
digest = hashlib.sha256(json.dumps(texts, sort_keys=True).encode()).hexdigest()
|
||
evidence = self.prior_evidence(previous["boot_id"]) if previous and previous.get("boot_id") != boot_id else None
|
||
if evidence is not None and previous.get("pstore_digest") and previous["pstore_digest"] != digest:
|
||
evidence["pstore_new"] = {"text": "\n".join(texts.values()), "note": "pstore contents changed since previous heartbeat; cause still needs review"}
|
||
self.state, event = transition(previous, boot_id, datetime.now().astimezone().isoformat(timespec="seconds"), evidence)
|
||
self.state["pstore_digest"] = digest
|
||
self.log(event, boot_id=boot_id, reboots=self.state["reboots"], previous_last_seen=(previous or {}).get("last_seen"), evidence=evidence)
|
||
if evidence: self.log("restart_assessment", assessment=self.state["history"][-1]["assessment"])
|
||
self.log("boot_evidence", kernel=os.uname().release,
|
||
journal=command(["journalctl", "--list-boots", "--no-pager", "-n", "8"]), pstore=pstore,
|
||
watchdog_bootstatus={str(p): read(p) for p in Path("/sys/class/watchdog").glob("watchdog*/bootstatus")})
|
||
|
||
def sample(self):
|
||
uptime = float(read("/proc/uptime").split()[0]); now = datetime.now().astimezone()
|
||
processes = []
|
||
for path in Path("/proc").glob("[0-9]*/cmdline"):
|
||
try:
|
||
args = path.read_bytes().decode(errors="replace").strip("\0").split("\0"); exe = Path(args[0]).name
|
||
if exe not in ("moonlight", "kiosk-shell", "vdi-kiosk-monitor"): continue
|
||
item = {"pid": int(path.parent.name), "name": exe}
|
||
if exe == "moonlight" and "stream" in args:
|
||
i = args.index("stream"); item["host"] = args[i+1] if i+1 < len(args) else ""
|
||
for flag in ("--resolution", "--fps"):
|
||
if flag in args and args.index(flag)+1 < len(args): item[flag[2:]] = args[args.index(flag)+1]
|
||
processes.append(item)
|
||
except (OSError, ValueError, IndexError): continue
|
||
temps = {p.parent.name: int(read(p, "0"))/1000 for p in Path("/sys/class/thermal").glob("thermal_zone*/temp")}
|
||
memory = {s.split(":")[0]: s.split(":")[1].strip() for s in read("/proc/meminfo").splitlines() if s.startswith(("MemAvailable:", "MemTotal:", "SwapFree:"))}
|
||
display = command(["env", "DISPLAY=:0", "XAUTHORITY="+str(Path.home()/".Xauthority"), "xrandr", "--current"], timeout=3)
|
||
display["text"] = "\n".join(s for s in display["text"].splitlines() if s.startswith("Screen ") or " connected" in s or " disconnected" in s)
|
||
network = {p.name: {"operstate": read(p/"operstate"), "rx_errors": read(p/"statistics/rx_errors"), "tx_errors": read(p/"statistics/tx_errors")}
|
||
for p in Path("/sys/class/net").iterdir() if p.name != "lo"}
|
||
if self.state.get("last_seen") and self.state.get("sample_boot_id") == self.state["boot_id"]:
|
||
wall = now.timestamp()-self.state.get("last_wall_seconds", now.timestamp())
|
||
mono = uptime-self.state.get("uptime_seconds", uptime)
|
||
if abs(wall-mono) > 120: self.log("clock_adjustment", wall_delta=wall, uptime_delta=mono)
|
||
fs = os.statvfs("/")
|
||
values = dict(boot_id=self.state["boot_id"], reboots=self.state["reboots"], uptime_seconds=uptime,
|
||
load=read("/proc/loadavg").split()[:3], memory=memory, temperatures_c=temps,
|
||
processes=processes, display=display, network=network, root_available_mb=fs.f_bavail*fs.f_frsize//(1024*1024))
|
||
self.log("sample", **values)
|
||
self.state.update(last_seen=now.isoformat(timespec="seconds"), last_wall_seconds=now.timestamp(), uptime_seconds=uptime,
|
||
sample_boot_id=self.state["boot_id"], last_sample=values)
|
||
self.save()
|
||
|
||
def save(self):
|
||
atomic_write(self.state_file, json.dumps(self.state, ensure_ascii=False, indent=2)+"\n")
|
||
history = self.state.get("history", [])
|
||
lines = ["盒子过夜观察汇总", "记录启用时间："+self.state["started_at"], "最近记录时间："+self.state.get("last_seen", ""),
|
||
"记录启用后重启次数："+str(self.state["reboots"]), "当前启动编号："+self.state["boot_id"],
|
||
"当前开机时长（秒）："+str(self.state.get("uptime_seconds", "")),
|
||
"最近一次重启判断："+(history[-1]["assessment"] if history else "观察期间尚未发现重启"),
|
||
"详细数据见同目录 overnight-health-YYYYMMDD.log；每分钟更新。",
|
||
"服务重启不计为盒子重启；无故障证据时不推断具体断电原因。"]
|
||
atomic_write(self.logs/"overnight-summary.txt", "\n".join(lines)+"\n")
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--once", action="store_true"); parser.add_argument("--report", action="store_true")
|
||
args = parser.parse_args()
|
||
logs = Path(os.environ.get("VDI_OBSERVER_LOG_DIR", str(Path.home()/"vdi-terminal/logs")))
|
||
state_dir = Path(os.environ.get("VDI_OBSERVER_STATE_DIR", str(Path.home()/"vdi-terminal/state")))
|
||
if args.report: print((logs/"overnight-summary.txt").read_text()); return
|
||
interval = int(os.environ.get("VDI_OBSERVER_INTERVAL", "60"))
|
||
if not 10 <= interval <= 3600: parser.error("Interval must be 10..3600 seconds")
|
||
os.umask(0o077); observer = Observer(logs, state_dir)
|
||
import fcntl
|
||
with (state_dir/"overnight-observer.lock").open("w") as lock:
|
||
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
stopped = threading.Event(); stop_signal = []
|
||
def on_signal(signum, frame): stop_signal.append(signum); stopped.set()
|
||
signal.signal(signal.SIGTERM, on_signal); signal.signal(signal.SIGINT, on_signal)
|
||
observer.start(); observer.sample()
|
||
if args.once: return
|
||
ticks = 0
|
||
while not stopped.wait(interval):
|
||
observer.sample(); ticks += 1
|
||
if ticks % max(1, 600//interval) == 0:
|
||
r = command(["journalctl", "-k", "-b", "0", "-p", "warning", "-n", "50", "--no-pager", "-o", "short-monotonic"])
|
||
r["text"] = "\n".join(s for s in r["text"].splitlines() if EVIDENCE.search(s))
|
||
observer.log("kernel_warning_clues", **r)
|
||
# SIGTERM may just be a service stop. Never call it an OS reboot.
|
||
observer.log("observer_stop", signal=stop_signal, boot_id=observer.state["boot_id"])
|
||
observer.state["observer_stop_signal"] = stop_signal; observer.save()
|
||
|
||
|
||
if __name__ == "__main__": main()
|