#!/usr/bin/env python3 # SPDX-License-Identifier: BSD-3-Clause # Copyright (c) 2026 Open-IOV contributors """ guest_init_monitor.py - Monitor guest kernel initialization for NVIDIA driver Correlates host tracepoints with guest dmesg/serial output """ import sys import os import re import time import subprocess import threading import json from datetime import datetime from collections import deque from pathlib import Path class GuestInitMonitor: def __init__(self, guest_serial="/dev/ttyS0", host_trace_pipe="/sys/kernel/debug/tracing/trace_pipe"): self.guest_serial = guest_serial self.host_trace_pipe = host_trace_pipe self.guest_log = deque(maxlen=1000) self.host_events = deque(maxlen=1000) self.failure_patterns = self._load_failure_patterns() self.running = False def _load_failure_patterns(self): return { "pci_probe_fail": [ r"nvidia: probe of .* failed with error -?\d+", r"pci .*: can't claim BAR \d+", r"pci .*: BAR \d+: no space for", ], "gsp_firmware_fail": [ r"GSP.*firmware.*load.*failed", r"GSP.*bootstrap.*failed", r"GSP.*RPC.*timeout", r"nv_gsp_firmware_load.*error", ], "msix_fail": [ r"MSI-X.*failed", r"pci_enable_msix.*failed", r"request_threaded_irq.*failed", ], "bar_map_fail": [ r"ioremap.*failed", r"pci_iomap.*failed", r"BAR.*mapping.*failed", ], "dma_fail": [ r"dma_map.*failed", r"IOMMU.*fault", r"DMAR.*error", ], "vgpu_vfio_fail": [ r"vGPU VFIO.*failed", r"nvidia_vgpu_vfio_get_ops.*failed", r"mdev.*register.*failed", ], } def start(self): self.running = True self.guest_thread = threading.Thread(target=self._monitor_guest_serial, daemon=True) self.host_thread = threading.Thread(target=self._monitor_host_traces, daemon=True) self.guest_thread.start() self.host_thread.start() print(f"[{datetime.now()}] Guest init monitor started") print(f" Guest serial: {self.guest_serial}") print(f" Host trace pipe: {self.host_trace_pipe}") def stop(self): self.running = False if hasattr(self, 'guest_thread'): self.guest_thread.join(timeout=2) if hasattr(self, 'host_thread'): self.host_thread.join(timeout=2) print(f"[{datetime.now()}] Guest init monitor stopped") def _monitor_guest_serial(self): """Monitor guest kernel log via serial console""" try: # Try to open serial device or use virsh console proc = subprocess.Popen( ["timeout", "300", "cat", self.guest_serial], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) for line in proc.stdout: if not self.running: break self._process_guest_line(line.strip()) except Exception as e: print(f"[{datetime.now()}] Guest serial monitor error: {e}") def _monitor_host_traces(self): """Monitor host kernel tracepoints""" try: proc = subprocess.Popen( ["cat", self.host_trace_pipe], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) for line in proc.stdout: if not self.running: break self._process_host_trace(line.strip()) except Exception as e: print(f"[{datetime.now()}] Host trace monitor error: {e}") def _process_guest_line(self, line): """Process a line from guest serial console""" timestamp = datetime.now().isoformat() entry = {"timestamp": timestamp, "source": "guest", "line": line} self.guest_log.append(entry) # Check for failure patterns failure = self._classify_failure(line) if failure: self._report_failure(failure, line, "guest") # Print important lines if any(kw in line.lower() for kw in ["nvidia", "gsp", "vfio", "mdev", "error", "fail", "timeout"]): print(f"[GUEST] {line}") def _process_host_trace(self, line): """Process a line from host trace_pipe""" timestamp = datetime.now().isoformat() entry = {"timestamp": timestamp, "source": "host_trace", "line": line} self.host_events.append(entry) # Look for GSP emulation tracepoints if "gsp_emu" in line: print(f"[HOST TRACE] {line}") # Correlate with recent guest logs self._correlate_events(line) def _classify_failure(self, line): """Classify failure type based on patterns""" for failure_type, patterns in self.failure_patterns.items(): for pattern in patterns: if re.search(pattern, line, re.IGNORECASE): return failure_type return None def _report_failure(self, failure_type, line, source): """Report a detected failure""" report = { "timestamp": datetime.now().isoformat(), "failure_type": failure_type, "source": source, "line": line, "recent_guest_logs": list(self.guest_log)[-20:], "recent_host_events": list(self.host_events)[-20:], } print(f"\n{'='*60}") print(f"FAILURE DETECTED: {failure_type}") print(f"Source: {source}") print(f"Line: {line}") print(f"{'='*60}\n") # Save failure report report_dir = Path("/tmp/gsp_emu_failures") report_dir.mkdir(exist_ok=True) report_file = report_dir / f"failure_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" with open(report_file, 'w') as f: json.dump(report, f, indent=2) print(f"Failure report saved: {report_file}") def _correlate_events(self, host_event): """Correlate host tracepoint with recent guest activity""" # Look for matching sequence IDs, RPC types, etc. pass def get_summary(self): """Get monitoring summary""" return { "guest_log_entries": len(self.guest_log), "host_events": len(self.host_events), "recent_guest": list(self.guest_log)[-10:], "recent_host": list(self.host_events)[-10:], } def main(): import argparse parser = argparse.ArgumentParser(description="Monitor guest NVIDIA driver initialization") parser.add_argument("--guest-serial", default="/dev/ttyS0", help="Guest serial device") parser.add_argument("--host-trace", default="/sys/kernel/debug/tracing/trace_pipe", help="Host trace pipe") parser.add_argument("--duration", type=int, default=300, help="Monitor duration (seconds)") args = parser.parse_args() monitor = GuestInitMonitor(args.guest_serial, args.host_trace) monitor.start() try: time.sleep(args.duration) except KeyboardInterrupt: pass finally: monitor.stop() summary = monitor.get_summary() print(json.dumps(summary, indent=2)) if __name__ == "__main__": main()