diff --git a/cmd/api/main.go b/cmd/api/main.go index 20ac8dc60..312f73c5c 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -657,6 +657,12 @@ func run() error { return app.HealthCheckController.Run(gctx) }) } + if app.VGPUSentinelController != nil { + grp.Go(func() error { + logger.Info("starting vGPU sentinel controller") + return app.VGPUSentinelController.Run(gctx) + }) + } if restartController, ok := app.InstanceManager.(interface { StartRestartPolicyController(context.Context) error }); ok { diff --git a/cmd/api/wire.go b/cmd/api/wire.go index 133bf41f6..123d59bf5 100644 --- a/cmd/api/wire.go +++ b/cmd/api/wire.go @@ -29,26 +29,27 @@ import ( // application struct to hold initialized components type application struct { - Ctx context.Context - Logger *slog.Logger - Config *config.Config - ImageManager images.Manager - SystemManager system.Manager - NetworkManager network.Manager - DeviceManager devices.Manager - InstanceManager instances.Manager - VolumeManager volumes.Manager - BuilderManager builders.Manager - IngressManager ingress.Manager - BuildManager builds.Manager - PushManager imagepush.Manager - ResourceManager *resources.Manager - GuestMemoryController guestmemory.Controller - AutoStandbyController *autostandby.Controller - HealthCheckController *instances.HealthCheckController - VMMetricsManager *vm_metrics.Manager - Registry *registry.Registry - ApiService *api.ApiService + Ctx context.Context + Logger *slog.Logger + Config *config.Config + ImageManager images.Manager + SystemManager system.Manager + NetworkManager network.Manager + DeviceManager devices.Manager + InstanceManager instances.Manager + VolumeManager volumes.Manager + BuilderManager builders.Manager + IngressManager ingress.Manager + BuildManager builds.Manager + PushManager imagepush.Manager + ResourceManager *resources.Manager + GuestMemoryController guestmemory.Controller + AutoStandbyController *autostandby.Controller + HealthCheckController *instances.HealthCheckController + VGPUSentinelController *instances.VGPUSentinelController + VMMetricsManager *vm_metrics.Manager + Registry *registry.Registry + ApiService *api.ApiService } // initializeApp is the injector function @@ -72,6 +73,7 @@ func initializeApp() (*application, func(), error) { providers.ProvideGuestMemoryController, providers.ProvideAutoStandbyController, providers.ProvideHealthCheckController, + providers.ProvideVGPUSentinelController, providers.ProvideVMMetricsManager, providers.ProvideRegistry, api.New, diff --git a/cmd/api/wire_gen.go b/cmd/api/wire_gen.go index 9eb13c6ea..57034551a 100644 --- a/cmd/api/wire_gen.go +++ b/cmd/api/wire_gen.go @@ -82,6 +82,10 @@ func initializeApp() (*application, func(), error) { } autostandbyController := providers.ProvideAutoStandbyController(instancesManager, config, logger) healthCheckController := providers.ProvideHealthCheckController(instancesManager, logger) + vgpuSentinelController, err := providers.ProvideVGPUSentinelController(instancesManager, logger) + if err != nil { + return nil, nil, err + } vm_metricsManager, err := providers.ProvideVMMetricsManager(instancesManager, config, logger) if err != nil { return nil, nil, err @@ -92,26 +96,27 @@ func initializeApp() (*application, func(), error) { } apiService := api.New(config, manager, instancesManager, volumesManager, buildersManager, networkManager, devicesManager, ingressManager, buildsManager, imagepushManager, resourcesManager, controller, autostandbyController, vm_metricsManager) mainApplication := &application{ - Ctx: context, - Logger: logger, - Config: config, - ImageManager: manager, - SystemManager: systemManager, - NetworkManager: networkManager, - DeviceManager: devicesManager, - InstanceManager: instancesManager, - VolumeManager: volumesManager, - BuilderManager: buildersManager, - IngressManager: ingressManager, - BuildManager: buildsManager, - PushManager: imagepushManager, - ResourceManager: resourcesManager, - GuestMemoryController: controller, - AutoStandbyController: autostandbyController, - HealthCheckController: healthCheckController, - VMMetricsManager: vm_metricsManager, - Registry: registry, - ApiService: apiService, + Ctx: context, + Logger: logger, + Config: config, + ImageManager: manager, + SystemManager: systemManager, + NetworkManager: networkManager, + DeviceManager: devicesManager, + InstanceManager: instancesManager, + VolumeManager: volumesManager, + BuilderManager: buildersManager, + IngressManager: ingressManager, + BuildManager: buildsManager, + PushManager: imagepushManager, + ResourceManager: resourcesManager, + GuestMemoryController: controller, + AutoStandbyController: autostandbyController, + HealthCheckController: healthCheckController, + VGPUSentinelController: vgpuSentinelController, + VMMetricsManager: vm_metricsManager, + Registry: registry, + ApiService: apiService, } return mainApplication, func() { }, nil @@ -121,24 +126,25 @@ func initializeApp() (*application, func(), error) { // application struct to hold initialized components type application struct { - Ctx context.Context - Logger *slog.Logger - Config *config.Config - ImageManager images.Manager - SystemManager system.Manager - NetworkManager network.Manager - DeviceManager devices.Manager - InstanceManager instances.Manager - VolumeManager volumes.Manager - BuilderManager builders.Manager - IngressManager ingress.Manager - BuildManager builds.Manager - PushManager imagepush.Manager - ResourceManager *resources.Manager - GuestMemoryController guestmemory.Controller - AutoStandbyController *autostandby.Controller - HealthCheckController *instances.HealthCheckController - VMMetricsManager *vm_metrics.Manager - Registry *registry.Registry - ApiService *api.ApiService + Ctx context.Context + Logger *slog.Logger + Config *config.Config + ImageManager images.Manager + SystemManager system.Manager + NetworkManager network.Manager + DeviceManager devices.Manager + InstanceManager instances.Manager + VolumeManager volumes.Manager + BuilderManager builders.Manager + IngressManager ingress.Manager + BuildManager builds.Manager + PushManager imagepush.Manager + ResourceManager *resources.Manager + GuestMemoryController guestmemory.Controller + AutoStandbyController *autostandby.Controller + HealthCheckController *instances.HealthCheckController + VGPUSentinelController *instances.VGPUSentinelController + VMMetricsManager *vm_metrics.Manager + Registry *registry.Registry + ApiService *api.ApiService } diff --git a/lib/devices/GPU.md b/lib/devices/GPU.md index ff11c5cde..601e9618c 100644 --- a/lib/devices/GPU.md +++ b/lib/devices/GPU.md @@ -282,13 +282,41 @@ NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884) ``` (0x65 = timeout; the guest's init requests are never answered, and -`/proc/interrupts` shows the GPU's MSI-X vectors allocated but idle). Because -placement is deterministic least-loaded, an idle host re-picks the same VF for -every request, so one wedged VF presents as all vGPU instances failing while -`/resources` reports full capacity. - -The wedge itself leaves no host-side log: no kernel error, no XID, no plugin -crash. The trigger is a SIGKILL delivered to QEMU while the vGPU plugin is +`/proc/interrupts` shows the GPU's MSI-X vectors allocated but idle). + +Hypeman detects this automatically: the guest agent watches the guest kernel +log (`/dev/kmsg`) for that line and reports it as a `HYPEMAN-GPU-INIT-FAILED` +marker in the instance's `logs/app.log`, which the vGPU sentinel controller +scans for every vendor VFIO instance. A match quarantines the VF in +`/gpu/vf-health.json` (it survives restarts): the VF is excluded +from placement and from advertised profile availability, and its parent GPU +becomes overflow-only — deprioritized for new placements. The conviction is +logged at error level (`quarantined wedged vGPU VF`) and counted in +`hypeman_instances_vgpu_sentinel_convictions_total`; +`hypeman_instances_vgpu_quarantined_vfs` gauges the current quarantine count. +There is no rate limit on convictions: a systemic non-wedge init failure +(e.g. a guest/host driver mismatch) emits the same line on every VF and +would quarantine the whole host, so such changes must be validated on a +test host first, and the convictions counter is the signal to alert on if +one gets through. + +Detection requires the hypeman guest agent: an image that skips the agent +never reports, so a wedge hit exclusively by such images stays undetected in +v1. Start archives the previous boot's app log before persisting a new +assignment, so a marker written within one scan interval of a stop/start is +archived unscanned; that VF returns to the pool unconvicted until the next +victim boot re-emits the marker. The scanner requires the complete standalone +guest-agent log envelope, so customer commands and ordinary output containing +the marker do not match. The +serial console remains guest-writable: a root guest can deliberately imitate +the entire line and quarantine the VF its own instance holds; the quarantine +only ever removes capacity, never touches the instance. + +The wedge-creating kill itself leaves no host-side log: no kernel error, no +XID, no plugin crash. Detection therefore happens on the next boot that lands +on the VF, whose guest driver starts failing ~27s after spawn. + +The trigger is a SIGKILL delivered to QEMU while the vGPU plugin is still initializing the VF (roughly the first seconds after process start): a single hard kill in that window wedges the VF near-deterministically, while QEMU processes that exit voluntarily — error exits, QMP quit, SIGTERM — @@ -303,18 +331,47 @@ External SIGKILLs (OOM killer, manual `kill -9`) can still trigger it. Confirm by assigning the same profile on a different VF: if that guest initializes, the VF is wedged, not the driver stack. Remediate by cycling SR-IOV on the parent GPU (this destroys and recreates all of its VFs, so it -requires no vGPU assignments on that GPU): +requires no vGPU assignments on that GPU). The DCGM quiesce is not optional: +with `nv-hostengine`/`dcgm-exporter` holding the GPUs open, `sriov-manage -d` +fails with `Cannot obtain unbindLock` on first contact. + +Any manual edit to `vf-health.json` needs an immediate hypeman restart: the +store loads only at startup, and a conviction landing first re-persists the +in-memory set over your edit. The restart does not disturb running VMs — +startup reconciliation protects live VFs. + +**Draining the parent GPU.** Overflow-only is a preference, not a cordon: +under capacity pressure new placements still land on the card's healthy VFs +and refill it. To drain the card, quarantine all of its VFs by hand — add +records to `vf-health.json` (the shape of a real conviction: `vf_address` +plus `quarantined_at`) and restart. Running instances are untouched and +drain through their normal lifecycle: standby is blocked for vGPU instances, +so only a running VM pins a VF, and each stop or delete frees one for good. +Monitor by listing instances whose `gpu.device_path` sits under the parent +GPU; once none remain, run the cycle below. ```bash +# 1. Quiesce the services holding the GPU (required for the unbind lock). +systemctl stop nvidia-dcgm-exporter nvidia-dcgm + +# 2. Cycle SR-IOV on the parent GPU. /usr/lib/nvidia/sriov-manage -d /usr/lib/nvidia/sriov-manage -e + +# 3. Restart the quiesced services. +systemctl start nvidia-dcgm nvidia-dcgm-exporter ``` +After the cycle, clear the quarantine by removing the card's entries from +`vf-health.json` (restart rule above), then boot a GPU instance as +verification: placement excludes quarantined VFs, so the recovered VF cannot +be targeted while its entry exists, and there is no VF-pin API — clearing +first is safe because the sentinel automatically re-quarantines the VF if the +cycle did not cure it (every cycle in hardware validation did). + Do not unbind/rebind the VF from the nvidia driver — it breaks the nvidia-vgpu-vfio core-device registration (`vfio_pci_core_device not found`) and the VF stops accepting assignments entirely until the SR-IOV cycle. -Services holding the GPU (DCGM, persistenced) must be stopped for the cycle -to obtain the unbind lock. ### vGPU assignment fails diff --git a/lib/devices/manager.go b/lib/devices/manager.go index 30763c04d..6ab5479b1 100644 --- a/lib/devices/manager.go +++ b/lib/devices/manager.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "log/slog" "os" "runtime" "strings" @@ -85,6 +86,12 @@ type manager struct { // NewManager creates a new device manager. // Use SetLivenessChecker after construction to enable accurate orphan detection. func NewManager(p *paths.Paths) Manager { + // The VF health store lives with the device manager: constructing one + // without loading the store would run placement against an empty, + // non-persisting quarantine set. + if err := InitVFHealth(p.VFHealthState()); err != nil { + slog.Default().Error("failed to load VF health state; vGPU placement is disabled until the state file is repaired or removed", "error", err) + } return &manager{ paths: p, vfioBinder: NewVFIOBinder(), diff --git a/lib/devices/vendor_vfio_linux.go b/lib/devices/vendor_vfio_linux.go index a6d37e80d..0d5355769 100644 --- a/lib/devices/vendor_vfio_linux.go +++ b/lib/devices/vendor_vfio_linux.go @@ -7,12 +7,12 @@ import ( "errors" "fmt" "log/slog" + "math/rand/v2" "os" "path/filepath" "sort" "strconv" "strings" - "sync" "syscall" "github.com/kernel/hypeman/lib/logger" @@ -29,18 +29,16 @@ type vendorVFIOSysfs struct { vfioDevicesPath string owners map[string]string framebufferByType map[string]int + pickVFIndex func(n int) int // overridden in tests; nil means random } -var ( - hostVendorVFIO = vendorVFIOSysfs{ - pciDevicesPath: pciDevicesPath, - procPath: procPath, - vfioDevicesPath: vfioDevicesPath, - owners: make(map[string]string), - framebufferByType: make(map[string]int), - } - vendorVFIOMu sync.Mutex -) +var hostVendorVFIO = vendorVFIOSysfs{ + pciDevicesPath: pciDevicesPath, + procPath: procPath, + vfioDevicesPath: vfioDevicesPath, + owners: make(map[string]string), + framebufferByType: make(map[string]int), +} func (s vendorVFIOSysfs) discoverVFs() ([]VirtualFunction, error) { entries, err := os.ReadDir(s.pciDevicesPath) @@ -102,6 +100,10 @@ func (s vendorVFIOSysfs) discoverVFs() ([]VirtualFunction, error) { // available_instances. This is a best-effort snapshot because creating on one // VF may revoke the type from siblings that share its GPU framebuffer. func (s vendorVFIOSysfs) listProfiles(vfs []VirtualFunction) ([]GPUProfile, error) { + quarantined, err := vfHealth.checkedAddresses() + if err != nil { + return nil, err + } profilesByType := make(map[string]profileMetadata) creatableVFs := make(map[string]int) for _, vf := range vfs { @@ -113,9 +115,10 @@ func (s vendorVFIOSysfs) listProfiles(vfs []VirtualFunction) ([]GPUProfile, erro slog.Default().Warn("skipping unreadable creatable vGPU types", "vf", vf.PCIAddress, "error", err) continue } + _, bad := quarantined[vf.PCIAddress] for _, profile := range creatable { profilesByType[profile.TypeName] = profile - if !vf.Allocated { + if !vf.Allocated && !bad { creatableVFs[profile.TypeName]++ } } @@ -304,10 +307,24 @@ func (s vendorVFIOSysfs) reconcile(ctx context.Context, protectedDevicePaths map } func (s vendorVFIOSysfs) selectLeastLoadedVF(vfs []VirtualFunction, profileType string) (string, error) { + quarantined, err := vfHealth.checkedAddresses() + if err != nil { + return "", err + } usageByGPU := make(map[string]int) unknownUsageByGPU := make(map[string]bool) + quarantinedByGPU := make(map[string]int) freeByGPU := make(map[string][]VirtualFunction) for _, vf := range vfs { + // The count only deprioritizes the card so it drains toward the + // SR-IOV cycle; healthy sibling VFs stay usable. + _, bad := quarantined[vf.PCIAddress] + if bad { + quarantinedByGPU[vf.ParentGPU]++ + if !vf.Allocated { + continue + } + } if vf.Allocated { // framebufferByType only covers currently creatable profiles, so // after a restart an allocated type can be missing when its @@ -341,6 +358,9 @@ func (s vendorVFIOSysfs) selectLeastLoadedVF(vfs []VirtualFunction, profileType gpus = append(gpus, gpu) } sort.Slice(gpus, func(i, j int) bool { + if quarantinedByGPU[gpus[i]] != quarantinedByGPU[gpus[j]] { + return quarantinedByGPU[gpus[i]] < quarantinedByGPU[gpus[j]] + } if unknownUsageByGPU[gpus[i]] != unknownUsageByGPU[gpus[j]] { return !unknownUsageByGPU[gpus[i]] } @@ -352,7 +372,15 @@ func (s vendorVFIOSysfs) selectLeastLoadedVF(vfs []VirtualFunction, profileType if len(gpus) == 0 { return "", nil } - return freeByGPU[gpus[0]][0].PCIAddress, nil + // Randomize among the chosen GPU's free VFs: a deterministic pick would + // route every first create to the same VF, so one undetected wedged VF + // presents as every GPU create failing. + candidates := freeByGPU[gpus[0]] + pick := s.pickVFIndex + if pick == nil { + pick = rand.IntN + } + return candidates[pick(len(candidates))].PCIAddress, nil } func (s vendorVFIOSysfs) profileMetadata(vfs []VirtualFunction) ([]profileMetadata, error) { diff --git a/lib/devices/vendor_vfio_linux_test.go b/lib/devices/vendor_vfio_linux_test.go index 4f2e40844..490348502 100644 --- a/lib/devices/vendor_vfio_linux_test.go +++ b/lib/devices/vendor_vfio_linux_test.go @@ -571,3 +571,75 @@ func assertFileValue(t *testing.T, path, expected string) { require.NoError(t, err) assert.Equal(t, expected, string(value)) } + +func TestVendorVFIOSkipsQuarantinedVF(t *testing.T) { + resetVFHealthStore(t) + _, err := QuarantineVF(VFQuarantine{VFAddress: "0000:82:00.4"}) + require.NoError(t, err) + + sysfs := newTestVendorVFIOSysfs(t) + sysfs.pickVFIndex = func(int) int { return 0 } + sysfs.addVF(t, "0000:82:00.0", "0000:82:00.4", "42", "0", testCreatableTypes) + sysfs.addVF(t, "0000:82:00.0", "0000:82:00.5", "43", "0", testCreatableTypes) + + device, err := sysfs.create(context.Background(), "NVIDIA L40S-1Q", "instance-1") + require.NoError(t, err) + assert.Equal(t, "0000:82:00.5", device.VFAddress) +} + +func TestVendorVFIONoVFWhenAllQuarantined(t *testing.T) { + resetVFHealthStore(t) + _, err := QuarantineVF(VFQuarantine{VFAddress: "0000:82:00.4"}) + require.NoError(t, err) + + sysfs := newTestVendorVFIOSysfs(t) + sysfs.addVF(t, "0000:82:00.0", "0000:82:00.4", "42", "0", testCreatableTypes) + + _, err = sysfs.create(context.Background(), "NVIDIA L40S-1Q", "instance-1") + require.ErrorContains(t, err, "no available VF") +} + +func TestVendorVFIOCardBiasAvoidsGPUWithQuarantinedVF(t *testing.T) { + resetVFHealthStore(t) + _, err := QuarantineVF(VFQuarantine{VFAddress: "0000:82:00.4"}) + require.NoError(t, err) + + // GPU 82 sorts first by address and has a free healthy VF, but its + // quarantined sibling must demote the whole card to overflow-only. + sysfs := newTestVendorVFIOSysfs(t) + sysfs.pickVFIndex = func(int) int { return 0 } + sysfs.addVF(t, "0000:82:00.0", "0000:82:00.4", "42", "0", testCreatableTypes) + sysfs.addVF(t, "0000:82:00.0", "0000:82:00.5", "43", "0", testCreatableTypes) + sysfs.addVF(t, "0000:e3:00.0", "0000:e3:00.4", "44", "0", testCreatableTypes) + + device, err := sysfs.create(context.Background(), "NVIDIA L40S-1Q", "instance-1") + require.NoError(t, err) + assert.Equal(t, "0000:e3:00.4", device.VFAddress) +} + +func TestVendorVFIOSelectUsesTiebreakAmongFreeVFs(t *testing.T) { + sysfs := newTestVendorVFIOSysfs(t) + sysfs.pickVFIndex = func(n int) int { return n - 1 } + sysfs.addVF(t, "0000:82:00.0", "0000:82:00.4", "42", "0", testCreatableTypes) + sysfs.addVF(t, "0000:82:00.0", "0000:82:00.5", "43", "0", testCreatableTypes) + + device, err := sysfs.create(context.Background(), "NVIDIA L40S-1Q", "instance-1") + require.NoError(t, err) + assert.Equal(t, "0000:82:00.5", device.VFAddress) +} + +func TestVendorVFIOListProfilesExcludesQuarantinedFromAvailability(t *testing.T) { + resetVFHealthStore(t) + _, err := QuarantineVF(VFQuarantine{VFAddress: "0000:82:00.4"}) + require.NoError(t, err) + + sysfs := newTestVendorVFIOSysfs(t) + sysfs.addVF(t, "0000:82:00.0", "0000:82:00.4", "42", "0", testCreatableTypes) + sysfs.addVF(t, "0000:82:00.0", "0000:82:00.5", "43", "0", testCreatableTypes) + + vfs, err := sysfs.discoverVFs() + require.NoError(t, err) + profiles, err := sysfs.listProfiles(vfs) + require.NoError(t, err) + assert.Equal(t, 1, profileAvailability(profiles, "NVIDIA L40S-1Q")) +} diff --git a/lib/devices/vf_health.go b/lib/devices/vf_health.go new file mode 100644 index 000000000..7aff90850 --- /dev/null +++ b/lib/devices/vf_health.go @@ -0,0 +1,256 @@ +package devices + +import ( + "encoding/json" + "fmt" + "log/slog" + "os" + "path/filepath" + "regexp" + "sort" + "sync" + "time" +) + +// VFHealthRecord tracks a VF quarantined after a wedge conviction, excluded +// from placement until the parent GPU is SR-IOV cycled and the record cleared. +type VFHealthRecord struct { + VFAddress string `json:"vf_address"` + InstanceID string `json:"instance_id,omitempty"` + SentinelLine string `json:"sentinel_line,omitempty"` + QuarantinedAt time.Time `json:"quarantined_at"` +} + +// VFQuarantine describes a wedge conviction to record. +type VFQuarantine struct { + VFAddress string + InstanceID string + SentinelLine string +} + +type vfHealthStore struct { + mu sync.Mutex + path string + records map[string]VFHealthRecord + // loadErr remembers a failed load of an existing state file. While set, + // mutations are refused: persisting the empty in-memory store would + // clobber every persisted quarantine. Mutations retry the load first. + loadErr error +} + +var vfHealthAddressPattern = regexp.MustCompile(`^[0-9a-f]{4}:[0-9a-f]{2}:[0-9a-f]{2}\.[0-7]$`) + +var ( + vfHealth = &vfHealthStore{records: make(map[string]VFHealthRecord)} + vendorVFIOMu sync.Mutex +) + +// InitVFHealth points the store at its state file and loads any persisted quarantines. +func InitVFHealth(path string) error { + vfHealth.mu.Lock() + defer vfHealth.mu.Unlock() + vfHealth.path = path + return vfHealth.loadLocked() +} + +func (s *vfHealthStore) loadLocked() error { + s.records = make(map[string]VFHealthRecord) + s.loadErr = nil + + data, err := os.ReadFile(s.path) + if err != nil { + if os.IsNotExist(err) { + return nil + } + s.loadErr = fmt.Errorf("read VF health state: %w", err) + return s.loadErr + } + var records []VFHealthRecord + if err := json.Unmarshal(data, &records); err != nil { + s.loadErr = fmt.Errorf("unmarshal VF health state: %w", err) + return s.loadErr + } + if records == nil { + s.loadErr = fmt.Errorf("validate VF health state: expected an array") + return s.loadErr + } + loaded := make(map[string]VFHealthRecord, len(records)) + for i, record := range records { + if !vfHealthAddressPattern.MatchString(record.VFAddress) { + s.loadErr = fmt.Errorf("validate VF health state record %d: invalid VF address %q", i, record.VFAddress) + return s.loadErr + } + if record.QuarantinedAt.IsZero() { + s.loadErr = fmt.Errorf("validate VF health state record %d: missing quarantine timestamp", i) + return s.loadErr + } + if _, exists := loaded[record.VFAddress]; exists { + s.loadErr = fmt.Errorf("validate VF health state record %d: duplicate VF address %q", i, record.VFAddress) + return s.loadErr + } + loaded[record.VFAddress] = record + } + s.records = loaded + return nil +} + +// ensureLoadedLocked retries a previously failed load. +func (s *vfHealthStore) ensureLoadedLocked() error { + if s.loadErr == nil { + return nil + } + return s.loadLocked() +} + +// checkedAddresses returns the quarantined VF addresses, retrying a failed +// load first: placement against a record set that is empty only because the +// state file was unreadable would return quarantined VFs to rotation. +func (s *vfHealthStore) checkedAddresses() (map[string]struct{}, error) { + s.mu.Lock() + defer s.mu.Unlock() + if err := s.ensureLoadedLocked(); err != nil { + return nil, fmt.Errorf("VF health state unavailable: %w", err) + } + addresses := make(map[string]struct{}, len(s.records)) + for address := range s.records { + addresses[address] = struct{}{} + } + return addresses, nil +} + +// AllocatableVFs returns the number of free VFs eligible for placement. +func AllocatableVFs(framework VGPUFramework, vfs []VirtualFunction) (int, error) { + if framework != VGPUFrameworkVendorVFIO { + return countFreeVFs(vfs, nil), nil + } + quarantined, err := vfHealth.checkedAddresses() + if err != nil { + return 0, err + } + return countFreeVFs(vfs, quarantined), nil +} + +func countFreeVFs(vfs []VirtualFunction, quarantined map[string]struct{}) int { + available := 0 + for _, vf := range vfs { + if vf.Allocated { + continue + } + if _, ok := quarantined[vf.PCIAddress]; !ok { + available++ + } + } + return available +} + +// QuarantineVF records a wedge conviction and persists it, under the vendor +// VFIO placement lock so a convicted VF is never concurrently selected. A +// repeat conviction leaves the record unchanged and reports existed=true. +func QuarantineVF(q VFQuarantine) (existed bool, err error) { + vendorVFIOMu.Lock() + defer vendorVFIOMu.Unlock() + return vfHealth.quarantine(q) +} + +// VFHealthStoreUnavailable reports whether the persisted VF health state +// failed to load. While true, mutations are refused and placement fails +// closed — and the quarantine gauge reads zero, so this needs its own signal. +func VFHealthStoreUnavailable() bool { + vfHealth.mu.Lock() + defer vfHealth.mu.Unlock() + return vfHealth.loadErr != nil +} + +// QuarantinedVFs returns all quarantine records, ordered by VF address. +func QuarantinedVFs() []VFHealthRecord { + vfHealth.mu.Lock() + defer vfHealth.mu.Unlock() + return vfHealth.sortedRecordsLocked() +} + +func (s *vfHealthStore) sortedRecordsLocked() []VFHealthRecord { + records := make([]VFHealthRecord, 0, len(s.records)) + for _, record := range s.records { + records = append(records, record) + } + sort.Slice(records, func(i, j int) bool { return records[i].VFAddress < records[j].VFAddress }) + return records +} + +func (s *vfHealthStore) quarantine(q VFQuarantine) (bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + if err := s.ensureLoadedLocked(); err != nil { + return false, err + } + if !vfHealthAddressPattern.MatchString(q.VFAddress) { + return false, fmt.Errorf("invalid VF address %q", q.VFAddress) + } + if _, ok := s.records[q.VFAddress]; ok { + return true, nil + } + s.records[q.VFAddress] = VFHealthRecord{ + VFAddress: q.VFAddress, + InstanceID: q.InstanceID, + SentinelLine: q.SentinelLine, + QuarantinedAt: time.Now().UTC(), + } + if err := s.persistLocked(); err != nil { + // Keeping the unpersisted record would make the next report look like + // a repeat conviction and end retries. + delete(s.records, q.VFAddress) + return false, err + } + return false, nil +} + +func (s *vfHealthStore) persistLocked() error { + if s.path == "" { + return nil + } + data, err := json.MarshalIndent(s.sortedRecordsLocked(), "", " ") + if err != nil { + return fmt.Errorf("marshal VF health state: %w", err) + } + if err := os.MkdirAll(filepath.Dir(s.path), 0755); err != nil { + return fmt.Errorf("create VF health state dir: %w", err) + } + tmp := s.path + ".tmp" + f, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644) + if err != nil { + return fmt.Errorf("create VF health state: %w", err) + } + if _, err := f.Write(data); err != nil { + f.Close() + os.Remove(tmp) + return fmt.Errorf("write VF health state: %w", err) + } + // Sync before rename so a host crash cannot leave an empty or partial + // file where a durable conviction should be. + if err := f.Sync(); err != nil { + f.Close() + os.Remove(tmp) + return fmt.Errorf("sync VF health state: %w", err) + } + if err := f.Close(); err != nil { + os.Remove(tmp) + return fmt.Errorf("close VF health state: %w", err) + } + if err := os.Rename(tmp, s.path); err != nil { + os.Remove(tmp) + return fmt.Errorf("rename VF health state: %w", err) + } + // The rename already committed the quarantine; a directory-sync failure + // only weakens crash durability, so log rather than roll back. + dirPath := filepath.Dir(s.path) + dir, err := os.Open(dirPath) + if err != nil { + slog.Default().Warn("failed to open VF health state directory for sync", "path", dirPath, "error", err) + return nil + } + if err := dir.Sync(); err != nil { + slog.Default().Warn("failed to sync VF health state directory", "path", dirPath, "error", err) + } + _ = dir.Close() + return nil +} diff --git a/lib/devices/vf_health_test.go b/lib/devices/vf_health_test.go new file mode 100644 index 000000000..bf23b89ad --- /dev/null +++ b/lib/devices/vf_health_test.go @@ -0,0 +1,211 @@ +package devices + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// resetVFHealthStore points the package-level store at a fresh temp file and +// restores an empty, unpersisted store when the test finishes. +func resetVFHealthStore(t *testing.T) string { + t.Helper() + path := filepath.Join(t.TempDir(), "vf-health.json") + require.NoError(t, InitVFHealth(path)) + t.Cleanup(func() { + vfHealth.mu.Lock() + defer vfHealth.mu.Unlock() + vfHealth.path = "" + vfHealth.records = make(map[string]VFHealthRecord) + vfHealth.loadErr = nil + }) + return path +} + +func TestAllocatableVFs(t *testing.T) { + resetVFHealthStore(t) + _, err := QuarantineVF(VFQuarantine{VFAddress: "0000:82:00.4"}) + require.NoError(t, err) + vfs := []VirtualFunction{ + {PCIAddress: "0000:82:00.4"}, + {PCIAddress: "0000:82:00.5", Allocated: true}, + {PCIAddress: "0000:82:00.6"}, + } + + available, err := AllocatableVFs(VGPUFrameworkVendorVFIO, vfs) + require.NoError(t, err) + assert.Equal(t, 1, available) + + available, err = AllocatableVFs(VGPUFrameworkMdev, vfs) + require.NoError(t, err) + assert.Equal(t, 2, available) +} + +func TestAllocatableVFsFailsWhenStoreUnavailable(t *testing.T) { + path := resetVFHealthStore(t) + require.NoError(t, os.WriteFile(path, []byte("not json"), 0o644)) + require.Error(t, InitVFHealth(path)) + + _, err := AllocatableVFs(VGPUFrameworkVendorVFIO, []VirtualFunction{{PCIAddress: "0000:82:00.4"}}) + require.ErrorContains(t, err, "VF health state unavailable") + + available, err := AllocatableVFs(VGPUFrameworkMdev, []VirtualFunction{{PCIAddress: "0000:82:00.4"}}) + require.NoError(t, err) + assert.Equal(t, 1, available) +} + +func TestQuarantineVFPersistsAcrossReload(t *testing.T) { + path := resetVFHealthStore(t) + + existed, err := QuarantineVF(VFQuarantine{ + VFAddress: "0000:e3:00.4", + InstanceID: "instance-1", + SentinelLine: "HYPEMAN-GPU-INIT-FAILED ts=2026-08-20T15:04:05Z nvrm=\"NVRM: GPU 0000:e3:00.4: RmInitAdapter failed! (0x22:0x65:884)\"", + }) + require.NoError(t, err) + assert.False(t, existed) + initial := QuarantinedVFs() + require.Len(t, initial, 1) + assert.False(t, initial[0].QuarantinedAt.IsZero()) + + // A repeat conviction (another victim boot, or a rescan after restart) + // leaves the original record unchanged: one wedge, one record. + existed, err = QuarantineVF(VFQuarantine{VFAddress: "0000:e3:00.4", InstanceID: "instance-2"}) + require.NoError(t, err) + assert.True(t, existed) + assert.Equal(t, initial, QuarantinedVFs()) + + // Reload from disk, as a hypeman restart would. + require.NoError(t, InitVFHealth(path)) + records := QuarantinedVFs() + require.Len(t, records, 1) + assert.Equal(t, "0000:e3:00.4", records[0].VFAddress) + assert.Equal(t, "instance-1", records[0].InstanceID) +} + +func TestQuarantineVFRejectsInvalidAddress(t *testing.T) { + resetVFHealthStore(t) + + _, err := QuarantineVF(VFQuarantine{VFAddress: "not-a-pci-address"}) + require.ErrorContains(t, err, "invalid VF address") + assert.Empty(t, QuarantinedVFs()) +} + +func TestQuarantineVFRollsBackOnPersistFailure(t *testing.T) { + resetVFHealthStore(t) + // Point the store below a path component that is a file, so persisting + // fails at MkdirAll. + blocker := filepath.Join(t.TempDir(), "blocker") + require.NoError(t, os.WriteFile(blocker, nil, 0644)) + vfHealth.path = filepath.Join(blocker, "vf-health.json") + + _, err := QuarantineVF(VFQuarantine{VFAddress: "0000:e3:00.4"}) + require.Error(t, err) + + // A quarantine is only real once persisted: the failed record must not + // linger in memory, or the retried conviction would be treated as a + // repeat and never reach disk. + assert.Empty(t, QuarantinedVFs()) +} + +func TestCheckedAddressesFailsClosedOnUnloadedState(t *testing.T) { + path := resetVFHealthStore(t) + + _, err := QuarantineVF(VFQuarantine{VFAddress: "0000:e3:00.4"}) + require.NoError(t, err) + + require.NoError(t, os.WriteFile(path, []byte("not json"), 0644)) + require.Error(t, InitVFHealth(path)) + + // Placement must not run against a record set that is only empty because + // the state file was unreadable. + _, err = vfHealth.checkedAddresses() + require.Error(t, err) + + // A repaired file self-heals on the next placement attempt. + restored := `[{"vf_address":"0000:e3:00.4","quarantined_at":"2026-08-20T00:00:00Z"}]` + require.NoError(t, os.WriteFile(path, []byte(restored), 0644)) + addresses, err := vfHealth.checkedAddresses() + require.NoError(t, err) + assert.Contains(t, addresses, "0000:e3:00.4") +} + +func TestCheckedAddressesFailsClosedOnInvalidRecord(t *testing.T) { + tests := []struct { + name string + state string + wantErr string + }{ + { + name: "null state", + state: `null`, + wantErr: "expected an array", + }, + { + name: "missing fields", + state: `[{}]`, + wantErr: "invalid VF address", + }, + { + name: "invalid address", + state: `[{"vf_address":"not-a-pci-address","quarantined_at":"2026-08-20T00:00:00Z"}]`, + wantErr: "invalid VF address", + }, + { + name: "missing timestamp", + state: `[{"vf_address":"0000:e3:00.4"}]`, + wantErr: "missing quarantine timestamp", + }, + { + name: "duplicate address", + state: `[{"vf_address":"0000:e3:00.4","quarantined_at":"2026-08-20T00:00:00Z"},{"vf_address":"0000:e3:00.4","quarantined_at":"2026-08-21T00:00:00Z"}]`, + wantErr: "duplicate VF address", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + path := resetVFHealthStore(t) + require.NoError(t, os.WriteFile(path, []byte(tt.state), 0644)) + require.ErrorContains(t, InitVFHealth(path), tt.wantErr) + assert.True(t, VFHealthStoreUnavailable()) + assert.Empty(t, QuarantinedVFs()) + + _, err := vfHealth.checkedAddresses() + require.Error(t, err) + }) + } +} + +func TestQuarantineVFRefusesToClobberUnloadedState(t *testing.T) { + path := resetVFHealthStore(t) + + _, err := QuarantineVF(VFQuarantine{VFAddress: "0000:e3:00.4"}) + require.NoError(t, err) + + // Corrupt the state file and reload, as a hypeman restart over a bad + // file would. The load fails and mutations must not persist the empty + // in-memory store over the previous quarantines. + require.NoError(t, os.WriteFile(path, []byte("not json"), 0644)) + require.Error(t, InitVFHealth(path)) + + _, err = QuarantineVF(VFQuarantine{VFAddress: "0000:e3:00.5"}) + require.Error(t, err) + data, err := os.ReadFile(path) + require.NoError(t, err) + assert.Equal(t, "not json", string(data), "a failed load must not be overwritten by later convictions") + + // Once the file is readable again the next conviction self-heals: it + // reloads the persisted records and appends to them. + restored := `[{"vf_address":"0000:e3:00.4","quarantined_at":"2026-08-20T00:00:00Z"}]` + require.NoError(t, os.WriteFile(path, []byte(restored), 0644)) + existed, err := QuarantineVF(VFQuarantine{VFAddress: "0000:e3:00.5"}) + require.NoError(t, err) + assert.False(t, existed) + records := QuarantinedVFs() + require.Len(t, records, 2, "reload must recover the previously persisted quarantine") + assert.Equal(t, "0000:e3:00.4", records[0].VFAddress) +} diff --git a/lib/instances/logs.go b/lib/instances/logs.go index f10e8537d..208f21cc3 100644 --- a/lib/instances/logs.go +++ b/lib/instances/logs.go @@ -35,6 +35,7 @@ var ErrLogNotFound = fmt.Errorf("log file not found") var appLogNoiseMarkers = []string{ "HYPEMAN-PROGRAM-START", "HYPEMAN-AGENT-READY", + "HYPEMAN-GPU-INIT-FAILED", "HYPEMAN-HEADERS-START", "HYPEMAN-HEADERS-READY", "HYPEMAN-HEADERS-FAILED", diff --git a/lib/instances/start.go b/lib/instances/start.go index 1089a35a3..b13f0f005 100644 --- a/lib/instances/start.go +++ b/lib/instances/start.go @@ -171,6 +171,17 @@ func (m *manager) startInstance( } } + // Archive the previous boot's serial log before persisting a new vGPU + // assignment: the sentinel controller would rescan an old log as the new + // epoch's output and could replay a wedge report against the fresh VF. + if err := m.archiveAppLogForBoot(id); err != nil { + if stored.GPUProfile != "" { + log.ErrorContext(ctx, "failed to archive app log before start", "instance_id", id, "error", err) + return nil, fmt.Errorf("archive app log before start: %w", err) + } + log.WarnContext(ctx, "failed to archive app log before start", "instance_id", id, "error", err) + } + // 4b. Recreate the vGPU if this instance had a GPU profile // Note: GPU availability was already validated in step 2b if stored.GPUProfile != "" { @@ -226,10 +237,6 @@ func (m *manager) startInstance( } configDiskSpanEnd(nil) - if err := m.archiveAppLogForBoot(id); err != nil { - log.WarnContext(ctx, "failed to archive app log before start", "instance_id", id, "error", err) - } - // 6. Start hypervisor and boot VM (reuses logic from create) bootStart := time.Now().UTC() stored.StartedAt = &bootStart diff --git a/lib/instances/storage.go b/lib/instances/storage.go index 40bbba684..0074d9c81 100644 --- a/lib/instances/storage.go +++ b/lib/instances/storage.go @@ -190,30 +190,34 @@ func removeAllWithRetry(path string, removeAll func(string) error, sleep func(ti // listMetadataFiles returns paths to all instance metadata files, skipping // entries whose metadata cannot be statted. func (m *manager) listMetadataFiles() ([]string, error) { - return m.walkMetadataFiles(false) + files, _, err := m.walkMetadataFiles() + return files, err } -// listMetadataFilesStrict returns paths to all instance metadata files, -// failing on any stat error other than absence, so fail-closed callers see -// an unreadable instance as an error instead of silently missing. +// listMetadataFilesStrict returns readable metadata paths and joins any stat +// errors other than absence. Fail-closed callers (the vGPU release claim scan +// and startup reconcile protection) use it so an unreadable instance is an +// error instead of silently missing. func (m *manager) listMetadataFilesStrict() ([]string, error) { - return m.walkMetadataFiles(true) + files, statErr, err := m.walkMetadataFiles() + return files, errors.Join(statErr, err) } -func (m *manager) walkMetadataFiles(failOnStatError bool) ([]string, error) { +func (m *manager) walkMetadataFiles() (files []string, statErr error, err error) { guestsDir := m.paths.GuestsDir() // Ensure guests directory exists if err := os.MkdirAll(guestsDir, 0755); err != nil { - return nil, fmt.Errorf("create guests directory: %w", err) + return nil, nil, fmt.Errorf("create guests directory: %w", err) } entries, err := os.ReadDir(guestsDir) if err != nil { - return nil, fmt.Errorf("read guests directory: %w", err) + return nil, nil, fmt.Errorf("read guests directory: %w", err) } - var metaFiles []string + metaFiles := make([]string, 0, len(entries)) + var statErrs []error for _, entry := range entries { if !entry.IsDir() { continue @@ -222,10 +226,10 @@ func (m *manager) walkMetadataFiles(failOnStatError bool) ([]string, error) { metaPath := filepath.Join(guestsDir, entry.Name(), "metadata.json") if _, err := os.Stat(metaPath); err == nil { metaFiles = append(metaFiles, metaPath) - } else if failOnStatError && !os.IsNotExist(err) { - return nil, fmt.Errorf("stat metadata for instance %s: %w", entry.Name(), err) + } else if !os.IsNotExist(err) { + statErrs = append(statErrs, fmt.Errorf("stat metadata for instance %s: %w", entry.Name(), err)) } } - return metaFiles, nil + return metaFiles, errors.Join(statErrs...), nil } diff --git a/lib/instances/vgpu_sentinel.go b/lib/instances/vgpu_sentinel.go new file mode 100644 index 000000000..e855afe58 --- /dev/null +++ b/lib/instances/vgpu_sentinel.go @@ -0,0 +1,375 @@ +package instances + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + "log/slog" + "os" + "path/filepath" + "regexp" + "time" + + "github.com/kernel/hypeman/lib/devices" + "github.com/kernel/hypeman/lib/logger" + "go.opentelemetry.io/otel/metric" +) + +const ( + vgpuSentinelScanInterval = 5 * time.Second + + // The log is guest-controlled console output; marker lines are a few + // hundred bytes, so longer lines are discarded without being buffered whole. + vgpuSentinelMaxLineBytes = 64 * 1024 +) + +// vgpuSentinelPattern requires the guest agent logger's full line shape so +// customer commands or stdout containing the marker cannot convict a VF. +// Collision resistance, not proof of provenance. +var vgpuSentinelPattern = regexp.MustCompile(`^\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2} \[guest-agent\] (HYPEMAN-GPU-INIT-FAILED ts=\S+ nvrm="NVRM: [^"\r\n]*RmInitAdapter failed![^"\r\n]*")\r?\n?$`) + +// MatchVGPUSentinelLine extracts a complete guest-agent vGPU failure marker. +func MatchVGPUSentinelLine(line []byte) (marker string, ok bool) { + match := vgpuSentinelPattern.FindSubmatch(line) + if match == nil { + return "", false + } + return string(match[1]), true +} + +type vgpuSentinelTarget struct { + instanceID string + vfAddress string + appLogPath string + // assignedAt changes with every vGPU acquisition, so a tail never carries + // across assignment epochs. + assignedAt string +} + +type vgpuSentinelStore interface { + listVGPUSentinelTargets(ctx context.Context) ([]vgpuSentinelTarget, error) + getVGPUSentinelTarget(ctx context.Context, instanceID string) (vgpuSentinelTarget, bool, error) +} + +var _ vgpuSentinelStore = (*manager)(nil) + +type vgpuSentinelTail struct { + vfAddress string + assignedAt string + offset int64 + // skippingLongLine: offset sits inside an oversized line; discard until + // its newline so the tail is not parsed as a fresh line. + skippingLongLine bool + done bool +} + +// VGPUSentinelController scans the serial console log of vendor VFIO vGPU +// instances for the guest driver's RmInitAdapter failure line — the +// fingerprint of a wedged VF — and quarantines the VF on a match. +type VGPUSentinelController struct { + store vgpuSentinelStore + log *slog.Logger + interval time.Duration + quarantine func(devices.VFQuarantine) (bool, error) + convictions metric.Int64Counter + tails map[string]*vgpuSentinelTail + discoverFramework func() (devices.VGPUFramework, []devices.VirtualFunction, error) // nil means devices.DiscoverVGPU; overridden in tests +} + +func NewVGPUSentinelController(manager Manager, meter metric.Meter, log *slog.Logger) (*VGPUSentinelController, error) { + if manager == nil { + return nil, fmt.Errorf("instance manager is nil") + } + if log == nil { + return nil, fmt.Errorf("logger is nil") + } + store, ok := manager.(vgpuSentinelStore) + if !ok { + return nil, fmt.Errorf("instance manager %T does not implement vgpuSentinelStore", manager) + } + + convictions, err := meter.Int64Counter( + "hypeman_instances_vgpu_sentinel_convictions_total", + metric.WithDescription("Total wedged-VF sentinel convictions"), + ) + if err != nil { + return nil, err + } + _, err = meter.Int64ObservableGauge( + "hypeman_instances_vgpu_quarantined_vfs", + metric.WithDescription("Number of vGPU virtual functions currently quarantined"), + metric.WithInt64Callback(func(_ context.Context, o metric.Int64Observer) error { + o.Observe(int64(len(devices.QuarantinedVFs()))) + return nil + }), + ) + if err != nil { + return nil, err + } + // The quarantined-VFs gauge reads zero when the state file is unreadable, + // so that fail-closed condition gets its own signal. + _, err = meter.Int64ObservableGauge( + "hypeman_instances_vgpu_vf_health_store_unavailable", + metric.WithDescription("1 when the persisted VF health state failed to load; quarantine mutations are refused and vGPU placement is disabled until it is repaired"), + metric.WithInt64Callback(func(_ context.Context, o metric.Int64Observer) error { + if devices.VFHealthStoreUnavailable() { + o.Observe(1) + } else { + o.Observe(0) + } + return nil + }), + ) + if err != nil { + return nil, err + } + + return &VGPUSentinelController{ + store: store, + log: log.With("controller", "vgpu_sentinel"), + interval: vgpuSentinelScanInterval, + quarantine: devices.QuarantineVF, + convictions: convictions, + tails: make(map[string]*vgpuSentinelTail), + }, nil +} + +func (c *VGPUSentinelController) Run(ctx context.Context) error { + // The framework is fixed for the process lifetime, so one successful + // probe settles the gate. A failed probe fails open (a transient + // discovery error must not disable detection) and is retried each tick. + ticker := time.NewTicker(c.interval) + defer ticker.Stop() + var known bool + for { + if !known { + var vendor bool + vendor, known = c.probeVendorVFIO() + if known && !vendor { + return nil + } + if known { + c.log.Info("vGPU sentinel controller started") + } + } + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + c.scanOnce(ctx) + } + } +} + +// probeVendorVFIO reports whether the host's vGPU framework could be +// determined and, when known, whether it is vendor VFIO. +func (c *VGPUSentinelController) probeVendorVFIO() (vendor, known bool) { + discover := c.discoverFramework + if discover == nil { + discover = devices.DiscoverVGPU + } + framework, _, err := discover() + if err != nil { + return false, false + } + if framework != devices.VGPUFrameworkVendorVFIO { + c.log.Info("vGPU sentinel controller idle: host has no vendor VFIO framework", "framework", string(framework)) + return false, true + } + return true, true +} + +func (c *VGPUSentinelController) scanOnce(ctx context.Context) { + targets, err := c.store.listVGPUSentinelTargets(ctx) + if err != nil { + c.log.WarnContext(ctx, "vGPU sentinel scan failed to list instances", "error", err) + return + } + alive := make(map[string]struct{}, len(targets)) + for _, target := range targets { + alive[target.instanceID] = struct{}{} + c.scanTarget(ctx, target) + } + for id := range c.tails { + if _, ok := alive[id]; !ok { + delete(c.tails, id) + } + } +} + +func (c *VGPUSentinelController) scanTarget(ctx context.Context, target vgpuSentinelTarget) { + tail := c.tails[target.instanceID] + if tail == nil || tail.vfAddress != target.vfAddress || tail.assignedAt != target.assignedAt { + // A new assignment writes a fresh log; a finished tail must not + // suppress scanning it. + tail = &vgpuSentinelTail{vfAddress: target.vfAddress, assignedAt: target.assignedAt} + c.tails[target.instanceID] = tail + } + if tail.done { + return + } + line, found, err := scanForSentinel(target.appLogPath, tail) + if err != nil { + c.log.WarnContext(ctx, "vGPU sentinel scan failed to read app log", + "instance_id", target.instanceID, "error", err) + return + } + if !found { + return + } + tail.done = c.convict(ctx, target, line) +} + +// convict quarantines the target's VF. It reports whether scanning for this +// instance is finished; a failed quarantine leaves the tail open so the +// recurring report retries it. +func (c *VGPUSentinelController) convict(ctx context.Context, target vgpuSentinelTarget, line string) bool { + // A stop/start racing the scan could have assigned a different VF since + // the snapshot; reload and skip only if a different assignment exists. + // Start archives the previous boot's log before persisting a new + // assignment, so a marker read under this epoch provably belongs to this + // epoch's VF — even if the instance was since stopped or deleted. + current, ok, err := c.store.getVGPUSentinelTarget(ctx, target.instanceID) + if err != nil { + c.log.WarnContext(ctx, "vGPU sentinel could not confirm assignment before conviction", + "vf", target.vfAddress, "instance_id", target.instanceID, "error", err) + return false + } + if ok && (current.vfAddress != target.vfAddress || current.assignedAt != target.assignedAt) { + c.log.InfoContext(ctx, "vGPU sentinel skipping conviction: assignment changed during scan", + "vf", target.vfAddress, "instance_id", target.instanceID) + return false + } + existed, err := c.quarantine(devices.VFQuarantine{ + VFAddress: target.vfAddress, + InstanceID: target.instanceID, + SentinelLine: line, + }) + if err != nil { + c.log.ErrorContext(ctx, "failed to quarantine wedged vGPU VF", + "vf", target.vfAddress, "instance_id", target.instanceID, "error", err) + return false + } + if existed { + // Rescan of an already-quarantined VF, not a new wedge: no metric. + c.log.InfoContext(ctx, "vGPU sentinel matched an already-quarantined VF", + "vf", target.vfAddress, "instance_id", target.instanceID) + return true + } + c.log.ErrorContext(ctx, "quarantined wedged vGPU VF", + "vf", target.vfAddress, + "instance_id", target.instanceID, + "sentinel_line", line, + ) + c.convictions.Add(ctx, 1) + return true +} + +// scanForSentinel reads complete lines from the tail's offset onward and +// returns the first sentinel match — only the matched marker, since the rest +// of the line is guest-controlled. A partial trailing line stays unconsumed; +// an offset past the file size means the log was archived, so the scan +// restarts from the top. Oversized lines are consumed without ever being +// held whole. +func scanForSentinel(path string, tail *vgpuSentinelTail) (string, bool, error) { + f, err := os.Open(path) + if err != nil { + if os.IsNotExist(err) { + return "", false, nil + } + return "", false, err + } + defer f.Close() + + info, err := f.Stat() + if err != nil { + return "", false, err + } + if info.Size() < tail.offset { + tail.offset = 0 + tail.skippingLongLine = false + } + if _, err := f.Seek(tail.offset, io.SeekStart); err != nil { + return "", false, err + } + + reader := bufio.NewReaderSize(f, vgpuSentinelMaxLineBytes) + for { + line, err := reader.ReadSlice('\n') + switch { + case err == nil: + tail.offset += int64(len(line)) + if tail.skippingLongLine { + tail.skippingLongLine = false + continue + } + if marker, ok := MatchVGPUSentinelLine(line); ok { + return marker, true, nil + } + case errors.Is(err, bufio.ErrBufferFull): + tail.offset += int64(len(line)) + tail.skippingLongLine = true + case errors.Is(err, io.EOF): + if tail.skippingLongLine { + tail.offset += int64(len(line)) + } + return "", false, nil + default: + return "", false, err + } + } +} + +// listVGPUSentinelTargets returns instances holding a vendor VFIO vGPU +// assignment. It reads raw metadata; hydrating full instances would query +// every hypervisor on the host each scan. +func (m *manager) listVGPUSentinelTargets(ctx context.Context) ([]vgpuSentinelTarget, error) { + files, statErr, err := m.walkMetadataFiles() + if err != nil { + return nil, err + } + if statErr != nil { + logger.FromContext(ctx).WarnContext(ctx, "vGPU sentinel cannot stat some instance metadata; their VFs are not scanned", "error", statErr) + } + targets := make([]vgpuSentinelTarget, 0, len(files)) + for _, file := range files { + id := filepath.Base(filepath.Dir(file)) + target, ok, err := m.getVGPUSentinelTarget(ctx, id) + if err != nil { + // An unreadable record removes its VF from detection; say so. + logger.FromContext(ctx).WarnContext(ctx, "vGPU sentinel skipping unreadable instance metadata", "instance_id", id, "error", err) + continue + } + if ok { + targets = append(targets, target) + } + } + return targets, nil +} + +// getVGPUSentinelTarget reports ok=false when the instance is gone or holds +// no vendor VFIO assignment. +func (m *manager) getVGPUSentinelTarget(_ context.Context, instanceID string) (vgpuSentinelTarget, bool, error) { + meta, err := m.loadMetadata(instanceID) + if err != nil { + if errors.Is(err, ErrNotFound) { + return vgpuSentinelTarget{}, false, nil + } + return vgpuSentinelTarget{}, false, err + } + if meta.GPUFramework != devices.VGPUFrameworkVendorVFIO || meta.GPUDevicePath == "" { + return vgpuSentinelTarget{}, false, nil + } + assignedAt := "" + if meta.GPUAssignedAt != nil { + assignedAt = meta.GPUAssignedAt.UTC().Format(time.RFC3339Nano) + } + return vgpuSentinelTarget{ + instanceID: instanceID, + vfAddress: filepath.Base(meta.GPUDevicePath), + appLogPath: m.paths.InstanceAppLog(instanceID), + assignedAt: assignedAt, + }, true, nil +} diff --git a/lib/instances/vgpu_sentinel_test.go b/lib/instances/vgpu_sentinel_test.go new file mode 100644 index 000000000..3198d8fb7 --- /dev/null +++ b/lib/instances/vgpu_sentinel_test.go @@ -0,0 +1,517 @@ +package instances + +import ( + "bytes" + "context" + "errors" + "log/slog" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/kernel/hypeman/lib/devices" + "github.com/kernel/hypeman/lib/logger" + "github.com/kernel/hypeman/lib/paths" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/metric/noop" +) + +const testSentinelMarker = "HYPEMAN-GPU-INIT-FAILED ts=2026-08-20T15:04:05.123456789Z nvrm=\"NVRM: GPU 0000:e3:00.4: RmInitAdapter failed! (0x22:0x65:884)\"" +const testSentinelLine = "2026/08/20 15:04:05 [guest-agent] " + testSentinelMarker + "\n" + +func TestVGPUSentinelPattern(t *testing.T) { + t.Parallel() + + match := vgpuSentinelPattern.FindStringSubmatch(testSentinelLine) + require.Len(t, match, 2) + assert.Equal(t, testSentinelMarker, match[1]) + + for _, line := range []string{ + // The raw guest kernel line is not the conviction signal. + "[ 27.031415] NVRM: GPU 0000:e3:00.4: RmInitAdapter failed! (0x22:0x65:884)", + // Customer-provided commands and output can contain the complete marker, + // but not as a standalone guest-agent log line. + testSentinelMarker + "\n", + "customer output: " + testSentinelMarker + "\n", + "2026-08-20T15:04:05Z [INFO] [hypeman-init:entrypoint] cmd=[echo " + testSentinelMarker + "]\n", + strings.TrimSuffix(testSentinelLine, "\n") + " trailing customer output\n", + "$ dmesg | grep -c HYPEMAN-GPU-INIT-FAILED", + "$ echo 'HYPEMAN-GPU-INIT-FAILED ts=x nvrm=\"'", + "HYPEMAN-GPU-INIT-FAILED ts=2026-08-20T15:04:05Z nvrm=\"something else\"", + "2026/08/20 15:04:05 [guest-agent] HYPEMAN-AGENT-READY ts=2026-08-20T15:04:05Z", + } { + assert.False(t, vgpuSentinelPattern.MatchString(line), "must not match %q", line) + } +} + +type fakeSentinelStore struct { + targets []vgpuSentinelTarget +} + +func (s *fakeSentinelStore) listVGPUSentinelTargets(context.Context) ([]vgpuSentinelTarget, error) { + return s.targets, nil +} + +func (s *fakeSentinelStore) getVGPUSentinelTarget(_ context.Context, instanceID string) (vgpuSentinelTarget, bool, error) { + for _, target := range s.targets { + if target.instanceID == instanceID { + return target, true, nil + } + } + return vgpuSentinelTarget{}, false, nil +} + +func TestNewVGPUSentinelControllerRejectsUnsupportedManager(t *testing.T) { + _, err := NewVGPUSentinelController(&stubManager{}, noop.NewMeterProvider().Meter("test"), slog.New(slog.DiscardHandler)) + require.ErrorContains(t, err, "does not implement vgpuSentinelStore") +} + +func newTestSentinelController(t *testing.T, store vgpuSentinelStore) (*VGPUSentinelController, *[]devices.VFQuarantine) { + t.Helper() + counter, err := noop.NewMeterProvider().Meter("test").Int64Counter("test") + require.NoError(t, err) + var quarantined []devices.VFQuarantine + c := &VGPUSentinelController{ + store: store, + log: slog.New(slog.DiscardHandler), + interval: time.Hour, + quarantine: func(q devices.VFQuarantine) (bool, error) { + quarantined = append(quarantined, q) + return false, nil + }, + convictions: counter, + tails: make(map[string]*vgpuSentinelTail), + } + return c, &quarantined +} + +func TestVGPUSentinelControllerConvictsOnce(t *testing.T) { + t.Parallel() + + logPath := filepath.Join(t.TempDir(), "app.log") + store := &fakeSentinelStore{targets: []vgpuSentinelTarget{{ + instanceID: "instance-1", + vfAddress: "0000:e3:00.4", + appLogPath: logPath, + }}} + c, quarantined := newTestSentinelController(t, store) + ctx := context.Background() + + // Log missing (boot not started) and then healthy output: no conviction. + c.scanOnce(ctx) + require.NoError(t, os.WriteFile(logPath, []byte("booting\nnvidia driver loaded\n"), 0644)) + c.scanOnce(ctx) + assert.Empty(t, *quarantined) + + f, err := os.OpenFile(logPath, os.O_APPEND|os.O_WRONLY, 0644) + require.NoError(t, err) + // A partial line without its newline must not convict yet. + _, err = f.WriteString("2026/08/20 15:04:05 [guest-agent] HYPEMAN-GPU-INIT-FAILED ts=2026-08-20T15:04:05Z nvrm=\"NVRM: GPU 0000:e3:00.4: RmInitAdapter fail") + require.NoError(t, err) + c.scanOnce(ctx) + assert.Empty(t, *quarantined) + + _, err = f.WriteString("ed! (0x22:0x65:884)\"\n") + require.NoError(t, err) + require.NoError(t, f.Close()) + c.scanOnce(ctx) + require.Len(t, *quarantined, 1) + assert.Equal(t, "0000:e3:00.4", (*quarantined)[0].VFAddress) + assert.Equal(t, "instance-1", (*quarantined)[0].InstanceID) + assert.Contains(t, (*quarantined)[0].SentinelLine, "RmInitAdapter failed!") + + // The guest agent re-emits the marker while the failure persists; a + // convicted instance is not re-convicted. + appendSentinelLine(t, logPath) + c.scanOnce(ctx) + assert.Len(t, *quarantined, 1) +} + +func TestVGPUSentinelControllerIgnoresCustomerMarkerText(t *testing.T) { + t.Parallel() + + logPath := filepath.Join(t.TempDir(), "app.log") + customerOutput := strings.Join([]string{ + testSentinelMarker, + "customer output: " + testSentinelMarker, + "2026-08-20T15:04:05Z [INFO] [hypeman-init:entrypoint] cmd=[echo " + testSentinelMarker + "]", + }, "\n") + "\n" + require.NoError(t, os.WriteFile(logPath, []byte(customerOutput), 0644)) + store := &fakeSentinelStore{targets: []vgpuSentinelTarget{{ + instanceID: "instance-1", + vfAddress: "0000:e3:00.4", + appLogPath: logPath, + }}} + c, quarantined := newTestSentinelController(t, store) + + c.scanOnce(context.Background()) + assert.Empty(t, *quarantined) + + appendSentinelLine(t, logPath) + c.scanOnce(context.Background()) + require.Len(t, *quarantined, 1) + assert.Equal(t, testSentinelMarker, (*quarantined)[0].SentinelLine) +} + +func TestVGPUSentinelControllerRetriesFailedQuarantine(t *testing.T) { + t.Parallel() + + logPath := filepath.Join(t.TempDir(), "app.log") + require.NoError(t, os.WriteFile(logPath, []byte(testSentinelLine), 0644)) + store := &fakeSentinelStore{targets: []vgpuSentinelTarget{{ + instanceID: "instance-1", + vfAddress: "0000:e3:00.4", + appLogPath: logPath, + }}} + c, quarantined := newTestSentinelController(t, store) + quarantineErr := errors.New("persist failed") + realQuarantine := c.quarantine + c.quarantine = func(devices.VFQuarantine) (bool, error) { + return false, quarantineErr + } + ctx := context.Background() + + c.scanOnce(ctx) + assert.Empty(t, *quarantined) + assert.False(t, c.tails["instance-1"].done) + + // The next recurrence of the marker retries the quarantine. + c.quarantine = realQuarantine + appendSentinelLine(t, logPath) + c.scanOnce(ctx) + assert.Len(t, *quarantined, 1) + assert.True(t, c.tails["instance-1"].done) +} + +// A burst of convictions across many instances is quarantined without any +// rate limit: systemic non-wedge failures (e.g. a driver-mismatch rollout) +// are expected to be caught on a test host before reaching production, and +// the convictions counter is the alerting signal if one gets through. +func TestVGPUSentinelControllerConvictsBursts(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + targets := make([]vgpuSentinelTarget, 0, 5) + for i := 0; i < 5; i++ { + logPath := filepath.Join(dir, string(rune('a'+i))+".log") + require.NoError(t, os.WriteFile(logPath, []byte(testSentinelLine), 0644)) + targets = append(targets, vgpuSentinelTarget{ + instanceID: "instance-" + string(rune('a'+i)), + vfAddress: "0000:e3:00." + string(rune('4'+i)), + appLogPath: logPath, + }) + } + c, quarantined := newTestSentinelController(t, &fakeSentinelStore{targets: targets}) + + c.scanOnce(context.Background()) + assert.Len(t, *quarantined, len(targets)) + for _, target := range targets { + assert.True(t, c.tails[target.instanceID].done) + } +} + +func TestVGPUSentinelControllerSkipsQuarantinedVFs(t *testing.T) { + t.Parallel() + + logPath := filepath.Join(t.TempDir(), "app.log") + require.NoError(t, os.WriteFile(logPath, []byte(testSentinelLine), 0644)) + store := &fakeSentinelStore{targets: []vgpuSentinelTarget{{ + instanceID: "instance-1", + vfAddress: "0000:e3:00.4", + appLogPath: logPath, + }}} + c, _ := newTestSentinelController(t, store) + c.quarantine = func(devices.VFQuarantine) (bool, error) { + return true, nil + } + + // A rescan of a standing victim's log after a controller restart reports + // an existing quarantine: not a new wedge, and the tail closes. + c.scanOnce(context.Background()) + assert.True(t, c.tails["instance-1"].done) +} + +func TestVGPUSentinelControllerRescansNewAssignment(t *testing.T) { + t.Parallel() + + logPath := filepath.Join(t.TempDir(), "app.log") + require.NoError(t, os.WriteFile(logPath, []byte(testSentinelLine), 0644)) + store := &fakeSentinelStore{targets: []vgpuSentinelTarget{{ + instanceID: "instance-1", + vfAddress: "0000:e3:00.4", + appLogPath: logPath, + assignedAt: "2026-08-20T15:00:00Z", + }}} + c, quarantined := newTestSentinelController(t, store) + ctx := context.Background() + + c.scanOnce(ctx) + require.Len(t, *quarantined, 1) + require.True(t, c.tails["instance-1"].done) + + // A stop/start acquires a new assignment (possibly a different VF) and + // archives the log; the finished tail from the previous assignment must + // not suppress scanning the new boot. + require.NoError(t, os.WriteFile(logPath, []byte(testSentinelLine), 0644)) + store.targets[0].vfAddress = "0000:e3:00.5" + store.targets[0].assignedAt = "2026-08-20T16:00:00Z" + c.scanOnce(ctx) + require.Len(t, *quarantined, 2) + assert.Equal(t, "0000:e3:00.5", (*quarantined)[1].VFAddress) +} + +func TestVGPUSentinelControllerDropsStaleTails(t *testing.T) { + t.Parallel() + + logPath := filepath.Join(t.TempDir(), "app.log") + require.NoError(t, os.WriteFile(logPath, []byte("booting\n"), 0644)) + store := &fakeSentinelStore{targets: []vgpuSentinelTarget{{ + instanceID: "instance-1", + vfAddress: "0000:e3:00.4", + appLogPath: logPath, + }}} + c, _ := newTestSentinelController(t, store) + ctx := context.Background() + + c.scanOnce(ctx) + require.Contains(t, c.tails, "instance-1") + + store.targets = nil + c.scanOnce(ctx) + assert.NotContains(t, c.tails, "instance-1") +} + +// A marker read against a metadata snapshot must not convict once the +// instance's assignment has changed under the scan: the marker belonged to +// the old epoch's VF, and the current VF may be healthy. +func TestVGPUSentinelControllerSkipsConvictionOnChangedAssignment(t *testing.T) { + t.Parallel() + + logPath := filepath.Join(t.TempDir(), "app.log") + require.NoError(t, os.WriteFile(logPath, []byte(testSentinelLine), 0644)) + store := &fakeSentinelStore{targets: []vgpuSentinelTarget{{ + instanceID: "instance-1", + vfAddress: "0000:e3:00.5", + appLogPath: logPath, + assignedAt: "2026-08-21T00:00:10Z", + }}} + c, quarantined := newTestSentinelController(t, store) + + stale := vgpuSentinelTarget{ + instanceID: "instance-1", + vfAddress: "0000:e3:00.4", + appLogPath: logPath, + assignedAt: "2026-08-21T00:00:00Z", + } + c.scanTarget(context.Background(), stale) + assert.Empty(t, *quarantined) + assert.False(t, c.tails["instance-1"].done) + + // A stop or delete after the marker was read releases the assignment but + // cannot un-wedge the VF: the marker still convicts the scanned epoch's VF. + store.targets = nil + c.tails["instance-1"] = &vgpuSentinelTail{vfAddress: stale.vfAddress, assignedAt: stale.assignedAt} + c.scanTarget(context.Background(), stale) + require.Len(t, *quarantined, 1) + assert.Equal(t, "0000:e3:00.4", (*quarantined)[0].VFAddress) +} + +func TestListVGPUSentinelTargetsSkipsUnstattableMetadata(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("root bypasses directory permissions") + } + + m := &manager{paths: paths.New(t.TempDir())} + assigned := time.Now().UTC() + for _, id := range []string{"readable", "unreadable-a", "unreadable-b"} { + require.NoError(t, m.ensureDirectories(id)) + require.NoError(t, m.saveMetadata(&metadata{StoredMetadata: StoredMetadata{ + Id: id, + GPUFramework: devices.VGPUFrameworkVendorVFIO, + GPUDevicePath: "/sys/bus/pci/devices/0000:e3:00.4", + GPUAssignedAt: &assigned, + }})) + } + for _, id := range []string{"unreadable-a", "unreadable-b"} { + instanceDir := filepath.Dir(m.paths.InstanceMetadata(id)) + require.NoError(t, os.Chmod(instanceDir, 0o000)) + t.Cleanup(func() { _ = os.Chmod(instanceDir, 0o755) }) + } + + files, err := m.listMetadataFilesStrict() + require.Error(t, err) + assert.ErrorContains(t, err, "unreadable-a") + assert.ErrorContains(t, err, "unreadable-b") + require.Len(t, files, 1) + + var logs bytes.Buffer + ctx := logger.AddToContext(context.Background(), slog.New(slog.NewTextHandler(&logs, nil))) + targets, err := m.listVGPUSentinelTargets(ctx) + require.NoError(t, err) + require.Len(t, targets, 1) + assert.Equal(t, "readable", targets[0].instanceID) + assert.Contains(t, logs.String(), "vGPU sentinel cannot stat some instance metadata; their VFs are not scanned") + assert.Contains(t, logs.String(), "unreadable-a") + assert.Contains(t, logs.String(), "unreadable-b") +} + +func TestListVGPUSentinelTargetsReturnsMetadataReadDirError(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("root bypasses directory permissions") + } + + m := &manager{paths: paths.New(t.TempDir())} + require.NoError(t, os.MkdirAll(m.paths.GuestsDir(), 0o755)) + require.NoError(t, os.Chmod(m.paths.GuestsDir(), 0o000)) + t.Cleanup(func() { _ = os.Chmod(m.paths.GuestsDir(), 0o755) }) + + targets, err := m.listVGPUSentinelTargets(context.Background()) + require.ErrorContains(t, err, "read guests directory") + assert.Nil(t, targets) +} + +// A transient discovery error fails open — the controller scans as if the +// host were vendor VFIO — but must keep re-probing so a host that is not +// vendor VFIO stops scanning once discovery recovers. +func TestVGPUSentinelControllerRunReprobesFailedDiscovery(t *testing.T) { + t.Parallel() + + c, _ := newTestSentinelController(t, &fakeSentinelStore{}) + c.interval = time.Millisecond + var logs bytes.Buffer + c.log = slog.New(slog.NewTextHandler(&logs, nil)) + var probes int + c.discoverFramework = func() (devices.VGPUFramework, []devices.VirtualFunction, error) { + probes++ + if probes == 1 { + return devices.VGPUFrameworkNone, nil, errors.New("transient sysfs error") + } + return devices.VGPUFrameworkNone, nil, nil + } + + done := make(chan error, 1) + go func() { done <- c.Run(context.Background()) }() + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(10 * time.Second): + t.Fatal("Run did not exit after discovery resolved to a non vendor VFIO host") + } + assert.Equal(t, 2, probes) + assert.NotContains(t, logs.String(), "vGPU sentinel controller started") + assert.Contains(t, logs.String(), "vGPU sentinel controller idle") +} + +// Kernel printk shares the serial console with the agent and can split a +// marker mid-write — which is why the agent emits each report as several +// identical lines. A corrupted copy must not convict (the strict shape is +// what keeps echoed commands from convicting), and the intact repeat on the +// next line must. +func TestScanForSentinelConvictsOnIntactRepeatAfterSplitMarker(t *testing.T) { + t.Parallel() + + logPath := filepath.Join(t.TempDir(), "app.log") + split := "2026/08/20 15:04:05 [guest-agent] HYPEMAN-GPU-INIT-FAILED ts=2026-08-20T15:04:05Z nvrm=\"NVRM: GPU 0000:e3:0\n" + + "[ 27.031415] NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884)\n" + + "0.4: RmInitAdapter failed! (0x22:0x65:884)\"\n" + require.NoError(t, os.WriteFile(logPath, []byte(split), 0644)) + + tail := &vgpuSentinelTail{} + _, found, err := scanForSentinel(logPath, tail) + require.NoError(t, err) + assert.False(t, found, "neither a split marker nor the raw kernel line may convict") + + appendSentinelLine(t, logPath) + line, found, err := scanForSentinel(logPath, tail) + require.NoError(t, err) + require.True(t, found, "the intact repeat must convict") + assert.Contains(t, line, "HYPEMAN-GPU-INIT-FAILED") +} + +// The marker is a few hundred bytes, so a line that overflows the read +// buffer is guest console spam by definition: it must not convict even when +// it embeds a marker, must never be buffered whole, and its tail — arriving +// on a later scan — must not be parsed as a fresh line. +func TestScanForSentinelDiscardsOversizedLines(t *testing.T) { + t.Parallel() + + logPath := filepath.Join(t.TempDir(), "app.log") + tail := &vgpuSentinelTail{} + + // An oversized terminated line with an embedded marker: skipped whole. + huge := strings.Repeat("x", vgpuSentinelMaxLineBytes) + testSentinelLine + require.NoError(t, os.WriteFile(logPath, []byte(huge), 0644)) + line, found, err := scanForSentinel(logPath, tail) + require.NoError(t, err) + assert.False(t, found, "a marker inside an oversized line must not convict") + assert.Empty(t, line) + + // A short marker line after the oversized one still convicts. + appendSentinelLine(t, logPath) + line, found, err = scanForSentinel(logPath, tail) + require.NoError(t, err) + require.True(t, found) + assert.Contains(t, line, "RmInitAdapter failed!") +} + +func TestScanForSentinelDiscardsOversizedLineTailAcrossScans(t *testing.T) { + t.Parallel() + + logPath := filepath.Join(t.TempDir(), "app.log") + tail := &vgpuSentinelTail{} + + // An oversized line still missing its newline: the scan enters skip mode. + require.NoError(t, os.WriteFile(logPath, []byte(strings.Repeat("x", vgpuSentinelMaxLineBytes+10)), 0644)) + _, found, err := scanForSentinel(logPath, tail) + require.NoError(t, err) + assert.False(t, found) + assert.True(t, tail.skippingLongLine) + + // The line's tail arrives later carrying a marker shape; it is still the + // same oversized line, so it must be discarded, not parsed as fresh. + appendSentinelLine(t, logPath) + _, found, err = scanForSentinel(logPath, tail) + require.NoError(t, err) + assert.False(t, found, "the tail of an oversized line must not convict") + assert.False(t, tail.skippingLongLine) + + // The next genuine marker line convicts. + appendSentinelLine(t, logPath) + line, found, err := scanForSentinel(logPath, tail) + require.NoError(t, err) + require.True(t, found) + assert.Contains(t, line, "RmInitAdapter failed!") +} + +func TestScanForSentinelResetsSkipStateOnTruncatedLog(t *testing.T) { + t.Parallel() + + logPath := filepath.Join(t.TempDir(), "app.log") + tail := &vgpuSentinelTail{} + + require.NoError(t, os.WriteFile(logPath, []byte(strings.Repeat("x", vgpuSentinelMaxLineBytes+10)), 0644)) + _, _, err := scanForSentinel(logPath, tail) + require.NoError(t, err) + require.True(t, tail.skippingLongLine) + + // Rotation truncates the file under the tail; the restart from the top + // must clear the skip state or a marker in the fresh log would be lost. + require.NoError(t, os.WriteFile(logPath, []byte(testSentinelLine), 0644)) + line, found, err := scanForSentinel(logPath, tail) + require.NoError(t, err) + require.True(t, found) + assert.Contains(t, line, "RmInitAdapter failed!") + assert.False(t, tail.skippingLongLine) +} + +func appendSentinelLine(t *testing.T, path string) { + t.Helper() + f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0644) + require.NoError(t, err) + _, err = f.WriteString(testSentinelLine) + require.NoError(t, err) + require.NoError(t, f.Close()) +} diff --git a/lib/paths/paths.go b/lib/paths/paths.go index add23ff52..aff107dca 100644 --- a/lib/paths/paths.go +++ b/lib/paths/paths.go @@ -314,6 +314,11 @@ func (p *Paths) DeviceMetadata(id string) string { return filepath.Join(p.DeviceDir(id), "metadata.json") } +// VFHealthState returns the path to the persisted vGPU VF health file. +func (p *Paths) VFHealthState() string { + return filepath.Join(p.dataDir, "gpu", "vf-health.json") +} + // Volume path methods // VolumesDir returns the root volumes directory. diff --git a/lib/providers/vgpu_sentinel.go b/lib/providers/vgpu_sentinel.go new file mode 100644 index 000000000..0931f1fe6 --- /dev/null +++ b/lib/providers/vgpu_sentinel.go @@ -0,0 +1,13 @@ +package providers + +import ( + "log/slog" + + "github.com/kernel/hypeman/lib/instances" + "go.opentelemetry.io/otel" +) + +func ProvideVGPUSentinelController(instanceManager instances.Manager, log *slog.Logger) (*instances.VGPUSentinelController, error) { + meter := otel.GetMeterProvider().Meter("hypeman") + return instances.NewVGPUSentinelController(instanceManager, meter, log) +} diff --git a/lib/resources/gpu.go b/lib/resources/gpu.go index 054e3744e..d6a14d95a 100644 --- a/lib/resources/gpu.go +++ b/lib/resources/gpu.go @@ -10,11 +10,16 @@ import ( // GPUResourceStatus represents the GPU resource status for the API response. // Returns nil if no GPU is available on the host. type GPUResourceStatus struct { - Mode string `json:"mode"` // "vgpu" or "passthrough" - TotalSlots int `json:"total_slots"` // VFs for vGPU, physical GPUs for passthrough - UsedSlots int `json:"used_slots"` // Slots currently in use - Profiles []devices.GPUProfile `json:"profiles,omitempty"` // vGPU mode only - Devices []devices.PassthroughDevice `json:"devices,omitempty"` // passthrough mode only + Mode string `json:"mode"` // "vgpu" or "passthrough" + TotalSlots int `json:"total_slots"` // VFs for vGPU, physical GPUs for passthrough + UsedSlots int `json:"used_slots"` // Slots currently in use + AllocatableSlots int `json:"-"` // Healthy free slots used by admission control + // AllocatableSlotsErr explains a zeroed AllocatableSlots caused by an + // unreadable VF health store, so admission failures name the real cause + // instead of reading as exhaustion. + AllocatableSlotsErr string `json:"-"` + Profiles []devices.GPUProfile `json:"profiles,omitempty"` // vGPU mode only + Devices []devices.PassthroughDevice `json:"devices,omitempty"` // passthrough mode only } // GetGPUStatus returns the current GPU resource status. @@ -51,13 +56,20 @@ func getVGPUStatus(ctx context.Context, framework devices.VGPUFramework, vfs []d logger.FromContext(ctx).WarnContext(ctx, "failed to list vGPU profiles; reporting none", "framework", framework, "error", err) profiles = nil } - - return &GPUResourceStatus{ - Mode: string(devices.GPUModeVGPU), - TotalSlots: len(vfs), - UsedSlots: usedSlots, - Profiles: profiles, + allocatableSlots, err := devices.AllocatableVFs(framework, vfs) + status := &GPUResourceStatus{ + Mode: string(devices.GPUModeVGPU), + TotalSlots: len(vfs), + UsedSlots: usedSlots, + AllocatableSlots: allocatableSlots, + Profiles: profiles, + } + if err != nil { + logger.FromContext(ctx).WarnContext(ctx, "failed to count allocatable vGPU slots; reporting none", "framework", framework, "error", err) + status.AllocatableSlots = 0 + status.AllocatableSlotsErr = err.Error() } + return status } // getPassthroughStatus returns GPU status for whole-GPU passthrough mode. @@ -92,9 +104,10 @@ func getPassthroughStatus() *GPUResourceStatus { } return &GPUResourceStatus{ - Mode: string(devices.GPUModePassthrough), - TotalSlots: len(passthroughDevices), - UsedSlots: usedSlots, - Devices: passthroughDevices, + Mode: string(devices.GPUModePassthrough), + TotalSlots: len(passthroughDevices), + UsedSlots: usedSlots, + AllocatableSlots: len(passthroughDevices) - usedSlots, + Devices: passthroughDevices, } } diff --git a/lib/resources/gpu_test.go b/lib/resources/gpu_test.go new file mode 100644 index 000000000..2dcdf448d --- /dev/null +++ b/lib/resources/gpu_test.go @@ -0,0 +1,59 @@ +package resources + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/kernel/hypeman/cmd/api/config" + "github.com/kernel/hypeman/lib/devices" + "github.com/kernel/hypeman/lib/paths" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestGetVGPUStatusFailsClosedWhenVFHealthIsUnavailable(t *testing.T) { + path := filepath.Join(t.TempDir(), "vf-health.json") + require.NoError(t, os.WriteFile(path, []byte("not json"), 0o644)) + require.Error(t, devices.InitVFHealth(path)) + resetPath := filepath.Join(t.TempDir(), "vf-health.json") + t.Cleanup(func() { + require.NoError(t, devices.InitVFHealth(resetPath)) + }) + + status := getVGPUStatus(context.Background(), devices.VGPUFrameworkVendorVFIO, []devices.VirtualFunction{{PCIAddress: "0000:82:00.4"}}) + assert.Zero(t, status.AllocatableSlots) + assert.Contains(t, status.AllocatableSlotsErr, "VF health state unavailable", + "admission must be able to name the real cause instead of reporting exhaustion") +} + +func TestReserveAllocationUsesAllocatableGPUSlots(t *testing.T) { + status := &GPUResourceStatus{ + Mode: string(devices.GPUModeVGPU), + TotalSlots: 4, + UsedSlots: 1, + AllocatableSlots: 0, + } + setGPUStatusProvider(func(context.Context) *GPUResourceStatus { return status }) + t.Cleanup(func() { setGPUStatusProvider(nil) }) + + mgr := NewManager(&config.Config{}, paths.New(t.TempDir())) + ctx := context.Background() + + err := mgr.ValidateAllocation(ctx, 0, 0, 0, 0, 0, 0, true) + require.ErrorContains(t, err, "no allocatable vgpu slots") + + status.AllocatableSlotsErr = "VF health state unavailable: read failed" + err = mgr.ValidateAllocation(ctx, 0, 0, 0, 0, 0, 0, true) + require.ErrorContains(t, err, "vGPU placement is disabled: VF health state unavailable") + status.AllocatableSlotsErr = "" + + status.AllocatableSlots = 1 + require.NoError(t, mgr.ReserveAllocation(ctx, "pending-a", 0, 0, 0, 0, 0, 0, true)) + err = mgr.ReserveAllocation(ctx, "pending-b", 0, 0, 0, 0, 0, 0, true) + require.ErrorContains(t, err, "no allocatable vgpu slots") + + mgr.FinishAllocation("pending-a") + require.NoError(t, mgr.ReserveAllocation(ctx, "pending-b", 0, 0, 0, 0, 0, 0, true)) +} diff --git a/lib/resources/resource.go b/lib/resources/resource.go index 86f644bda..2201f0fba 100644 --- a/lib/resources/resource.go +++ b/lib/resources/resource.go @@ -695,11 +695,14 @@ func (m *Manager) validateAllocationLocked(ctx context.Context, excludeID string if gpuStatus == nil { return fmt.Errorf("insufficient GPU: no GPU available on this host") } - availableSlots := gpuStatus.TotalSlots - gpuStatus.UsedSlots - pending.GPUSlots + availableSlots := gpuStatus.AllocatableSlots - pending.GPUSlots if availableSlots < req.GPUSlots { + if gpuStatus.AllocatableSlotsErr != "" { + return fmt.Errorf("insufficient GPU: vGPU placement is disabled: %s", gpuStatus.AllocatableSlotsErr) + } if availableSlots <= 0 { - return fmt.Errorf("insufficient GPU: all %d %s slots are in use", - gpuStatus.TotalSlots, gpuStatus.Mode) + return fmt.Errorf("insufficient GPU: no allocatable %s slots available (%d total, %d in use)", + gpuStatus.Mode, gpuStatus.TotalSlots, gpuStatus.UsedSlots) } return fmt.Errorf("insufficient GPU: requested %d %s slot(s), but only %d available", req.GPUSlots, gpuStatus.Mode, availableSlots) diff --git a/lib/system/guest_agent/gpu_watch.go b/lib/system/guest_agent/gpu_watch.go new file mode 100644 index 000000000..64c76f4e1 --- /dev/null +++ b/lib/system/guest_agent/gpu_watch.go @@ -0,0 +1,140 @@ +package main + +import ( + "bufio" + "errors" + "io" + "log" + "os" + "path/filepath" + "strconv" + "strings" + "syscall" + "time" +) + +const ( + // gpuInitFailedSentinelPrefix reports a failed NVIDIA driver init to the + // host, which quarantines the vGPU VF on it — so it is only ever emitted + // for a kernel-log line the driver itself produced. + gpuInitFailedSentinelPrefix = "HYPEMAN-GPU-INIT-FAILED" + + kmsgPath = "/dev/kmsg" + nvidiaPCIVendorID = "0x10de" + + // The driver retries init every ~20s on a wedged VF and /dev/kmsg reopens + // replay the ring, so unthrottled emission would spam the console. + gpuReportThrottle = 30 * time.Second + + // Kernel printk can split a userspace write mid-marker on the shared + // serial console, so each report is several identical lines; the host + // convicts on the first intact copy. + gpuReportRepeats = 3 + + kmsgReopenDelay = 5 * time.Second + + // Must be at least the kernel's maximum /dev/kmsg record size + // (CONSOLE_EXT_LOG_MAX, 8 KiB): a smaller buffer makes read(2) fail with + // EINVAL without consuming the record, wedging the watcher permanently. + kmsgRecordBufferBytes = 8192 + + kmsgOpenRetryDelay = time.Minute +) + +// hasNVIDIADevice reports whether any PCI function belongs to NVIDIA. A vGPU +// guest always enumerates its VF, even on a wedged slot. +func hasNVIDIADevice() bool { + vendors, _ := filepath.Glob("/sys/bus/pci/devices/*/vendor") + for _, path := range vendors { + data, err := os.ReadFile(path) + if err != nil { + continue + } + if strings.TrimSpace(string(data)) == nvidiaPCIVendorID { + return true + } + } + return false +} + +// watchGPUInitFailure tails the guest kernel log for the NVIDIA driver's +// RmInitAdapter failure and reports each occurrence with a +// HYPEMAN-GPU-INIT-FAILED marker. Opening /dev/kmsg replays the ring buffer, +// so failures that predate the agent are reported too. Reads error with +// EPIPE when the ring overwrites the read position; reopen and resume. +func watchGPUInitFailure() { + var lastReport time.Time + for { + f, err := os.Open(kmsgPath) + if err != nil { + log.Printf("[guest-agent] cannot open %s for GPU init watch (retrying): %v", kmsgPath, err) + time.Sleep(kmsgOpenRetryDelay) + continue + } + err = scanKmsg(f, func(msg string) { + if time.Since(lastReport) < gpuReportThrottle { + return + } + lastReport = time.Now() + emitGPUInitFailureReport(msg) + }) + _ = f.Close() + if err != nil && !errors.Is(err, syscall.EPIPE) { + log.Printf("[guest-agent] GPU init watch read %s failed (reopening): %v", kmsgPath, err) + } + time.Sleep(kmsgReopenDelay) + } +} + +// emitGPUInitFailureReport writes gpuReportRepeats identical marker lines, +// all carrying the same ts so the host sees one report. +func emitGPUInitFailureReport(msg string) { + ts := time.Now().UTC().Format(time.RFC3339Nano) + for range gpuReportRepeats { + log.Printf("[guest-agent] %s ts=%s nvrm=%q", gpuInitFailedSentinelPrefix, ts, msg) + } +} + +// scanKmsg calls report for each GPU init-failure record in r. Returns the +// error that ended the scan, nil on EOF. +func scanKmsg(r io.Reader, report func(msg string)) error { + reader := bufio.NewReaderSize(r, kmsgRecordBufferBytes) + for { + record, err := reader.ReadString('\n') + if msg, ok := gpuInitFailureMessage(record); ok { + report(msg) + } + if err != nil { + if errors.Is(err, io.EOF) { + return nil + } + return err + } + } +} + +// gpuInitFailureMessage extracts the message from a /dev/kmsg record +// (",,,;") and reports whether it is the +// NVIDIA init-failure line. Only kernel-facility records match — userspace +// /dev/kmsg writers are assigned LOG_USER or higher, so a guest process +// cannot forge one. The trailing (stage:status:line) tuple is +// driver-build-specific and deliberately not matched. +func gpuInitFailureMessage(record string) (string, bool) { + prefix, msg, found := strings.Cut(record, ";") + if !found { + return "", false + } + priority, _, found := strings.Cut(prefix, ",") + if !found { + return "", false + } + value, err := strconv.ParseUint(priority, 10, 32) + if err != nil || value>>3 != 0 { + return "", false + } + msg = strings.TrimSpace(msg) + if !strings.HasPrefix(msg, "NVRM:") || !strings.Contains(msg, "RmInitAdapter failed!") { + return "", false + } + return msg, true +} diff --git a/lib/system/guest_agent/gpu_watch_test.go b/lib/system/guest_agent/gpu_watch_test.go new file mode 100644 index 000000000..78b3e4bf5 --- /dev/null +++ b/lib/system/guest_agent/gpu_watch_test.go @@ -0,0 +1,151 @@ +package main + +import ( + "bufio" + "bytes" + "io" + "log" + "strings" + "syscall" + "testing" + + "github.com/kernel/hypeman/lib/instances" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestGPUInitFailureMessage(t *testing.T) { + msg, ok := gpuInitFailureMessage("3,1042,8462102,-;NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884)\n") + assert.True(t, ok) + assert.Equal(t, "NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884)", msg) + + // Tuple values are driver-build-specific; the match must not depend on them. + _, ok = gpuInitFailureMessage("3,1042,8462102,-;NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x26:0xffff:1482)\n") + assert.True(t, ok) + + // Any kernel log level matches; only the facility is load-bearing. + _, ok = gpuInitFailureMessage("4,1044,8462120,-;NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884)\n") + assert.True(t, ok) + + for _, record := range []string{ + "6,1041,8462100,-;NVRM: loading NVIDIA UNIX Open Kernel Module for x86_64\n", + "6,1043,8462110,-;nvidia-gridd: RmInitAdapter failed mentioned in userspace\n", + "no separator RmInitAdapter failed!\n", + " continuation line of a multi-line record\n", + // Userspace /dev/kmsg writes carry facility LOG_USER or higher — the + // kernel coerces a facility-0 prefix to LOG_USER — so these records, + // captured from a live 6.12 kernel by writing the driver's line into + // /dev/kmsg from a root shell, must never convict: + "12,307,4250363151,-;NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884)\n", // plain write + "8,308,4250380620,-;NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884)\n", // "<0>" prefix + "9,310,5898419120,-;NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884)\n", // "<1>" prefix + "24,309,5898400480,-;NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884)\n", // "<24>" prefix (facility 3) + "x,1,100,-;NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884)\n", // malformed priority + } { + _, ok := gpuInitFailureMessage(record) + assert.False(t, ok, "record %q must not match", record) + } +} + +// One report is emitted as several identical marker lines because kernel +// printk shares the serial console and can split a single write mid-marker +// — and a wedged VF guarantees printk traffic at report time. Any one +// intact copy convicts; the copies share one ts so they read as one report. +func TestEmitGPUInitFailureReportMatchesHostSentinel(t *testing.T) { + var buf bytes.Buffer + prevOutput := log.Writer() + prevFlags := log.Flags() + log.SetOutput(&buf) + log.SetFlags(log.LstdFlags) + defer func() { + log.SetOutput(prevOutput) + log.SetFlags(prevFlags) + }() + + emitGPUInitFailureReport("NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884)") + + scanner := bufio.NewScanner(&buf) + matched := 0 + for scanner.Scan() { + marker, ok := instances.MatchVGPUSentinelLine(scanner.Bytes()) + require.True(t, ok, "host matcher rejected %q", scanner.Text()) + assert.Contains(t, marker, "HYPEMAN-GPU-INIT-FAILED") + matched++ + } + require.NoError(t, scanner.Err()) + assert.Equal(t, gpuReportRepeats, matched) +} + +func TestEmitGPUInitFailureReportRepeatsMarkerLines(t *testing.T) { + var buf bytes.Buffer + prev := log.Writer() + log.SetOutput(&buf) + defer log.SetOutput(prev) + + emitGPUInitFailureReport("NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884)") + + lines := strings.Split(strings.TrimSpace(buf.String()), "\n") + require.Len(t, lines, gpuReportRepeats) + for i, line := range lines { + assert.Contains(t, line, "HYPEMAN-GPU-INIT-FAILED ts=") + assert.Contains(t, line, `nvrm="NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884)"`) + // Identical from the marker onward: same ts, one report. + assert.Equal(t, + lines[0][strings.Index(lines[0], "HYPEMAN"):], + line[strings.Index(line, "HYPEMAN"):], + "copy %d must be identical to the first", i) + } +} + +func TestScanKmsgReportsEachFailureRecord(t *testing.T) { + records := strings.Join([]string{ + "6,1,100,-;booting", + "3,2,200,-;NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884)", + "6,3,300,-;unrelated", + "3,4,400,-;NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884)", + }, "\n") + "\n" + + var got []string + require.NoError(t, scanKmsg(strings.NewReader(records), func(msg string) { got = append(got, msg) })) + assert.Len(t, got, 2) +} + +// kmsgConn mimics /dev/kmsg read(2) semantics: each read returns exactly one +// record, and a buffer smaller than the record fails with EINVAL without +// consuming it. +type kmsgConn struct { + records []string + pos int +} + +func (k *kmsgConn) Read(p []byte) (int, error) { + if k.pos >= len(k.records) { + return 0, io.EOF + } + rec := k.records[k.pos] + if len(p) < len(rec) { + return 0, syscall.EINVAL + } + k.pos++ + return copy(p, rec), nil +} + +// A record larger than bufio's default 4 KiB buffer must not wedge the scan: +// /dev/kmsg rejects a short read with EINVAL without consuming the record, +// so an undersized buffer would replay into the same record on every reopen +// and never reach a failure line behind it. +func TestScanKmsgReadsOversizedRecords(t *testing.T) { + oversized := "6,1,100,-;" + strings.Repeat("x", 5000) + "\n" + failure := "3,2,200,-;NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884)\n" + + var got []string + require.NoError(t, scanKmsg(&kmsgConn{records: []string{oversized, failure}}, + func(msg string) { got = append(got, msg) })) + assert.Len(t, got, 1) + + // A record beyond even the sized buffer surfaces the EINVAL instead of + // ending the scan silently, so the watcher logs the wedge. + huge := "6,3,300,-;" + strings.Repeat("x", kmsgRecordBufferBytes) + "\n" + err := scanKmsg(&kmsgConn{records: []string{huge}}, func(string) {}) + assert.ErrorIs(t, err, syscall.EINVAL) +} diff --git a/lib/system/guest_agent/main.go b/lib/system/guest_agent/main.go index 84fd2a5da..f726dd286 100644 --- a/lib/system/guest_agent/main.go +++ b/lib/system/guest_agent/main.go @@ -54,6 +54,10 @@ func main() { startClockKeeper() + if hasNVIDIADevice() { + go watchGPUInitFailure() + } + // Create gRPC server grpcServer := grpc.NewServer() pb.RegisterGuestServiceServer(grpcServer, &guestServer{})