Add artifact.boot_logger/src/artifact_boot_logger/main.py

artifact.boot_logger/src/artifact_boot_logger/main.py

@@ -0,0 +1,88 @@
+import os
+import json
+import subprocess
+from pathlib import Path
+from typing import Dict
+
+
+def collect_tsc_data() -> Dict[str, object]:
+    """Sammelt TSC-bezogene Boot-Daten und erstellt ein strukturiertes JSON-kompatibles Dict."""
+    boot_data = {
+        "boot_id": _read_proc_file("/proc/sys/kernel/random/boot_id"),
+        "host_vm": _detect_vm_environment(),
+        "pinned": _check_cpu_pin_status(),
+        "governor": _read_cpufreq_governor(),
+        "tsc_status": _collect_tsc_status()
+    }
+
+    # Pfad vorbereiten
+    output_path = Path("output/boot_tsc_report.json")
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    with output_path.open("w", encoding="utf-8") as f:
+        json.dump(boot_data, f, indent=2)
+
+    return boot_data
+
+
+def _read_proc_file(path: str) -> str:
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            return f.read().strip()
+    except FileNotFoundError:
+        return "unknown"
+
+
+def _detect_vm_environment() -> str:
+    try:
+        with open("/sys/class/dmi/id/product_name", "r", encoding="utf-8") as f:
+            name = f.read().strip().lower()
+        if any(vm in name for vm in ["vmware", "kvm", "qemu", "virtualbox"]):
+            return "vm"
+        return "host"
+    except FileNotFoundError:
+        return "unknown"
+
+
+def _check_cpu_pin_status() -> bool:
+    try:
+        pid = os.getpid()
+        with open(f"/proc/{pid}/status", "r", encoding="utf-8") as f:
+            for line in f:
+                if line.startswith("Cpus_allowed_list:"):
+                    value = line.split(":", 1)[1].strip()
+                    return "," not in value and "-" not in value
+    except FileNotFoundError:
+        pass
+    return False
+
+
+def _read_cpufreq_governor() -> str:
+    path = Path("/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor")
+    if path.exists():
+        try:
+            return path.read_text(encoding="utf-8").strip()
+        except Exception:
+            pass
+    return "unknown"
+
+
+def _collect_tsc_status() -> str:
+    try:
+        result = subprocess.run(["dmesg"], capture_output=True, text=True, check=False)
+        if "unstable" in result.stdout.lower():
+            return "unstable"
+        if "tsc: stable" in result.stdout.lower():
+            return "stable"
+    except Exception:
+        pass
+    sysfs_path = Path("/sys/devices/system/clocksource/clocksource0/current_clocksource")
+    if sysfs_path.exists():
+        content = sysfs_path.read_text(encoding="utf-8").strip().lower()
+        if "tsc" in content:
+            return "stable"
+    return "unknown"
+
+
+if __name__ == "__main__":
+    data = collect_tsc_data()
+    print(json.dumps(data, indent=2))