#!/usr/bin/env python3 # SPDX-License-Identifier: BSD-3-Clause # Copyright (c) 2026 Open-IOV contributors """ failure_analyzer.py - Analyze and classify guest initialization failures Correlates host tracepoints, guest logs, and QEMU state """ import sys import os import re import json import argparse from datetime import datetime from pathlib import Path from collections import defaultdict from typing import Dict, List, Optional, Any class FailureAnalyzer: def __init__(self): self.failure_signatures = self._load_signatures() self.recommendations = self._load_recommendations() def _load_signatures(self) -> Dict: return { "PCI_PROBE_FAILURE": { "guest_patterns": [ r"nvidia: probe of .* failed with error -?\d+", r"pci .*: can't claim BAR \d+", r"pci .*: BAR \d+: no space for \[mem size", r"pci .*: BAR \d+: failed to assign", ], "host_trace_patterns": [ r"gsp_emu_rpc_received.*rpc_type=0x1.*status=-?\d+", r"gsp_emu_fallback_used.*rpc_type=0x1", ], "severity": "critical", "category": "pci_config" }, "GSP_FIRMWARE_LOAD_FAILURE": { "guest_patterns": [ r"GSP.*firmware.*load.*failed", r"GSP.*bootstrap.*failed", r"nv_gsp_firmware_load.*error", r"GSP.*RPC.*timeout", r"GSP.*communication.*failed", ], "host_trace_patterns": [ r"gsp_emu_firmware_analysis.*matched=0", r"gsp_emu_dispatcher_select.*emulation=1", r"gsp_emu_rpc_received.*rpc_type=0x1.*seq_id=1", r"gsp_emu_rpc_processed.*status=-110", ], "severity": "critical", "category": "gsp_firmware" }, "MSIX_SETUP_FAILURE": { "guest_patterns": [ r"MSI-X.*failed", r"pci_enable_msix.*failed", r"request_threaded_irq.*failed", r"irq.*allocation.*failed", ], "host_trace_patterns": [ r"gsp_emu_msix_injected.*vector=0", r"gsp_emu_rpc_processed.*rpc_type=0x7.*status=-?\d+", ], "severity": "high", "category": "interrupts" }, "BAR_MAPPING_FAILURE": { "guest_patterns": [ r"ioremap.*failed", r"pci_iomap.*failed", r"BAR.*mapping.*failed", r"VFIO.*region.*mmap.*failed", ], "host_trace_patterns": [ r"gsp_emu_shared_mem_access.*is_write=1", r"gsp_emu_bar_emulation.*", ], "severity": "high", "category": "memory_mapping" }, "DMA_MAPPING_FAILURE": { "guest_patterns": [ r"dma_map.*failed", r"IOMMU.*fault", r"DMAR.*error", r"iommu.*domain.*failed", ], "host_trace_patterns": [ r"gsp_emu_rpc_processed.*rpc_type=0x5.*status=-?\d+", r"gsp_emu_fallback_used.*rpc_type=0x5", ], "severity": "high", "category": "dma" }, "VGPU_VFIO_FAILURE": { "guest_patterns": [ r"vGPU VFIO.*failed", r"nvidia_vgpu_vfio_get_ops.*failed", r"mdev.*register.*failed", r"VFIO.*device.*not found", ], "host_trace_patterns": [ r"gsp_emu_thread_state.*state=0", ], "severity": "critical", "category": "vfio" }, "GSP_RPC_TIMEOUT": { "guest_patterns": [ r"GSP.*RPC.*timeout", r"doorbell.*timeout", r"GSP.*response.*timeout", ], "host_trace_patterns": [ r"gsp_emu_doorbell_poll.*", r"gsp_emu_rpc_received.*", r"gsp_emu_rpc_processed.*latency_ns=\d{9,}", ], "severity": "medium", "category": "rpc_timeout" }, } def _load_recommendations(self) -> Dict: return { "PCI_PROBE_FAILURE": [ "Check PCI config space emulation in vgsp.ko (main.c)", "Verify BAR sizes match physical GPU (16MB/256MB/32MB/32MB)", "Check MSI-X capability pointer (0x70) and table/PBA offsets", "Ensure mdev parent registered with NVIDIA host driver", ], "GSP_FIRMWARE_LOAD_FAILURE": [ "Verify firmware file exists at /lib/firmware/nvidia/tu102/gsp/", "Check firmware CRC32/SHA256 matches known signature", "Ensure GSP emulation thread started (gsp_emu_thread_state)", "Check doorbell register offset (BAR0+0x110C00)", "Verify shared memory mapping (BAR0+0x1000, 4KB)", ], "MSIX_SETUP_FAILURE": [ "Verify MSI-X table at BAR0+0x2000 (32 entries × 16 bytes)", "Verify MSI-X PBA at BAR0+0x3000 (64-bit)", "Check eventfd signaling for interrupt injection", "Ensure guest MSI-X enable bit set in Message Control", ], "BAR_MAPPING_FAILURE": [ "Check VFIO region mmap handlers (bar_emulation.c)", "Verify BAR0 includes doorbell (0x110C00) and shared mem (0x1000)", "Check MSI-X region (VFIO_PCI_MSIX_REGION_INDEX)", "Ensure WC (write-combine) mapping for BAR0", ], "DMA_MAPPING_FAILURE": [ "Verify IOMMU domain attached to VFIO group", "Check DMA mapping in memory.c (nv_vgpu_dma_map)", "Ensure physical pages pinned before mapping", "Check for IOMMU faults in dmesg", ], "VGPU_VFIO_FAILURE": [ "Verify nvidia.ko loaded before vgsp.ko", "Check KBUILD_EXTRA_SYMBOLS points to nvidia.ko Module.symvers", "Verify nvidia_vgpu_vfio_get_ops exported by nvidia.ko", "Check mdev parent registration callback", ], "GSP_RPC_TIMEOUT": [ "Check GSP emulation thread CPU usage (should be polling)", "Verify doorbell polling adaptive (busy-wait 100us then usleep)", "Check shared memory coherence (cache flush/invalidate)", "Trace RPC sequence IDs for gaps/duplicates", ], } def analyze(self, guest_log_file: str, host_trace_file: str, qemu_log_file: str = None) -> Dict: """Analyze failure from log files""" guest_logs = self._read_logs(guest_log_file) host_traces = self._read_logs(host_trace_file) qemu_logs = self._read_logs(qemu_log_file) if qemu_log_file else [] # Classify failure failure_type = self._classify_failure(guest_logs, host_traces) # Get matching evidence evidence = self._collect_evidence(failure_type, guest_logs, host_traces) # Generate report report = { "timestamp": datetime.now().isoformat(), "failure_type": failure_type, "severity": self.failure_signatures.get(failure_type, {}).get("severity", "unknown"), "category": self.failure_signatures.get(failure_type, {}).get("category", "unknown"), "evidence": evidence, "recommendations": self.recommendations.get(failure_type, []), "guest_log_summary": self._summarize_logs(guest_logs), "host_trace_summary": self._summarize_logs(host_traces), } return report def _read_logs(self, filepath: str) -> List[str]: if not filepath or not os.path.exists(filepath): return [] with open(filepath, 'r') as f: return [line.strip() for line in f.readlines()] def _classify_failure(self, guest_logs: List[str], host_traces: List[str]) -> str: """Classify failure based on patterns""" scores = defaultdict(int) for failure_type, sig in self.failure_signatures.items(): # Check guest patterns for pattern in sig.get("guest_patterns", []): for line in guest_logs: if re.search(pattern, line, re.IGNORECASE): scores[failure_type] += 3 # Check host trace patterns for pattern in sig.get("host_trace_patterns", []): for line in host_traces: if re.search(pattern, line): scores[failure_type] += 2 if not scores: return "UNKNOWN" # Return highest scoring failure type return max(scores, key=scores.get) def _collect_evidence(self, failure_type: str, guest_logs: List[str], host_traces: List[str]) -> Dict: """Collect matching evidence for failure type""" sig = self.failure_signatures.get(failure_type, {}) evidence = {"guest": [], "host": []} for pattern in sig.get("guest_patterns", []): for line in guest_logs: if re.search(pattern, line, re.IGNORECASE): evidence["guest"].append({"pattern": pattern, "match": line}) for pattern in sig.get("host_trace_patterns", []): for line in host_traces: if re.search(pattern, line): evidence["host"].append({"pattern": pattern, "match": line}) return evidence def _summarize_logs(self, logs: List[str]) -> Dict: """Summarize log content""" return { "total_lines": len(logs), "error_lines": sum(1 for l in logs if "error" in l.lower() or "fail" in l.lower()), "nvidia_lines": sum(1 for l in logs if "nvidia" in l.lower()), "gsp_lines": sum(1 for l in logs if "gsp" in l.lower()), "last_20": logs[-20:] if logs else [], } def print_report(self, report: Dict): """Print formatted failure report""" print("\n" + "="*70) print(f"FAILURE ANALYSIS REPORT") print("="*70) print(f"Timestamp: {report['timestamp']}") print(f"Failure Type: {report['failure_type']}") print(f"Severity: {report['severity']}") print(f"Category: {report['category']}") print() print("EVIDENCE:") print("-"*70) for src in ["guest", "host"]: ev = report['evidence'].get(src, []) if ev: print(f" {src.upper()} ({len(ev)} matches):") for e in ev[:5]: print(f" Pattern: {e['pattern']}") print(f" Match: {e['match'][:120]}") print() print("RECOMMENDATIONS:") print("-"*70) for i, rec in enumerate(report['recommendations'], 1): print(f" {i}. {rec}") print() print("LOG SUMMARIES:") print("-"*70) for src in ["guest", "host"]: summ = report.get(f'{src}_log_summary', {}) print(f" {src.upper()}: {summ.get('total_lines', 0)} lines, " f"{summ.get('error_lines', 0)} errors, " f"{summ.get('nvidia_lines', 0)} nvidia, " f"{summ.get('gsp_lines', 0)} gsp") print("="*70) def main(): parser = argparse.ArgumentParser(description="Analyze guest NVIDIA driver initialization failures") parser.add_argument("--guest-log", required=True, help="Guest serial log file") parser.add_argument("--host-trace", required=True, help="Host trace_pipe log file") parser.add_argument("--qemu-log", help="QEMU log file (optional)") parser.add_argument("--output", help="Output JSON report file") args = parser.parse_args() analyzer = FailureAnalyzer() report = analyzer.analyze(args.guest_log, args.host_trace, args.qemu_log) analyzer.print_report(report) if args.output: with open(args.output, 'w') as f: json.dump(report, f, indent=2) print(f"\nReport saved to: {args.output}") if __name__ == "__main__": main()