From 6ce745ebb4b2c82a8debb03dd918ebca2fdbeba4 Mon Sep 17 00:00:00 2001 From: Mika Date: Wed, 7 Jan 2026 16:12:02 +0000 Subject: [PATCH] Add artifact.boot_logger/src/artifact_boot_logger/main.py --- .../src/artifact_boot_logger/main.py | 88 +++++++++++++++++++ 1 file changed, 88 insertions(+) create mode 100644 artifact.boot_logger/src/artifact_boot_logger/main.py diff --git a/artifact.boot_logger/src/artifact_boot_logger/main.py b/artifact.boot_logger/src/artifact_boot_logger/main.py new file mode 100644 index 0000000..b2a7d2b --- /dev/null +++ b/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))