diff --git a/cmd/api/main.go b/cmd/api/main.go index 9eb676622..bc271c168 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -658,6 +658,10 @@ func run() error { return app.HealthCheckController.Run(gctx) }) } + 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 d5539b35f..a69f70413 100644 --- a/lib/devices/GPU.md +++ b/lib/devices/GPU.md @@ -99,7 +99,7 @@ Instance Create → Persist VF claim → Configure profile → Attach VF to VM Instance Stop/Delete → Reset profile → Remove VF claim → VF available again ``` -Hypeman reconciles metadata claims once at startup and every minute afterward, skipping hosts without GPUs. A claim whose VMM is confirmed dead is reset before the claim is removed. mdev hosts also sweep orphaned device-level assignments. Vendor VFIO hosts repair an unclaimed dirty VF when the allocator next selects it; repair checks for open VFIO handles before resetting `current_vgpu_type`. The allocator prefers VFs that are already clean, and a dirty VF that refuses its reset is skipped in favor of another candidate. +Hypeman reconciles metadata claims once at startup and every minute afterward, skipping hosts without GPUs. A claim whose VMM is confirmed dead is reset before the claim is removed. An ambiguous hypervisor ownership check preserves the claim, logs a warning, and increments `hypeman_instances_vgpu_liveness_uncertain_total`; the create and start cleanup paths preserve and count the same way. mdev hosts also sweep orphaned device-level assignments. Vendor VFIO hosts repair an unclaimed dirty VF when the allocator next selects it; repair checks for open VFIO handles before resetting `current_vgpu_type`. The allocator prefers VFs that are already clean, and a dirty VF that refuses its reset is skipped in favor of another candidate. ### Hypervisor Support @@ -290,25 +290,47 @@ 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). -Hypeman tracks these failures in `/gpu/vf-health.json` (it survives -restarts): each reported init failure is tallied per instance assignment, and -once failures accumulate from `gpu.vf_quarantine_threshold` distinct -assignments (default 2), the VF is quarantined: excluded from placement and -from advertised profile availability, and its parent GPU becomes +Hypeman detects this automatically: the guest agent watches the guest kernel +log (`/dev/kmsg`) for that line and records it as its GPU init state, which +the vGPU sentinel controller polls over vsock (`GetGPUInitStatus`) for every +vendor VFIO instance whose VMM is up (control socket present). Stopped and +standby instances are not polled, even when a failed release leaves their +claim in metadata — QEMU vsock dials by guest CID alone, and a stale CID +could since have been reused by an unrelated instance. + +The guest agent also probes driver init at boot with `nvidia-smi -L`: the +device open runs RmInitAdapter, so on a wedged VF the probe itself triggers +the failure line without waiting for the workload to touch the GPU. On success +the reported state becomes a terminal OK, suppressing later failure reports. +The probe is the only source of an OK state, so an image without `nvidia-smi` +(or a driver that takes longer than the 10 minute probe window to initialize) +stays UNKNOWN: its failures are still detected, but its assignments can never +clear a tally or rescind a quarantine. + +A guest-reported failure records one init failure against the VF in +`/gpu/vf-health.json` (it survives restarts), tallied per instance +assignment; once failures accumulate from `gpu.vf_quarantine_threshold` +distinct assignments (default 2), the VF is quarantined: excluded from +placement and from advertised profile availability, and its parent GPU becomes overflow-only — deprioritized for new placements. Selection among a card's equivalent free VFs is randomized so a wedged VF cannot capture every placement. A reported init success clears failures only when that exact assignment has a recorded failure, removing the match and older tallies; if that assignment is the most recent failure recorded (the one that crossed -the threshold), its later success also rescinds the quarantine. If the state -file exists but cannot be loaded, or the last write to it failed, placement -and advertised availability fail closed; the load or write is retried on the -next placement or `/resources` read, so the store recovers on its own once -the file is repaired or the disk is writable again. -Recorded tallies are re-evaluated against the configured +the threshold), its later success also rescinds the quarantine. A success +with no exact match clears nothing; other quarantines require manual +recovery. Recorded tallies are re-evaluated against the configured threshold at load, so lowering `gpu.vf_quarantine_threshold` quarantines VFs whose persisted failures already meet the new value. +If the state file exists but cannot be loaded, or the last write to it failed, +placement and advertised availability fail closed. The load or write is +retried on the next placement or `/resources` read, and the sentinel makes one +repair attempt before each poll while the store is unavailable, so the store +recovers on its own once the file is repaired or the disk is writable again. +Individual guest reports do not retry the full-store write, and reports that +would change nothing still succeed. + `used_slots` includes quarantined VFs still held by running instances, so it can overlap `quarantined_slots`; use `allocatable_slots` for admission. While the store is unavailable, `allocatable_slots` is 0 and @@ -318,11 +340,38 @@ state file is distinguishable from a full host. The `kind=allocatable` and `kind=quarantined`, and `hypeman_resources_gpu_placement_disabled` is 1 while the store is unavailable, so the condition is alertable without scraping `/resources`. - -Quarantine only removes capacity — it never touches a running instance. - -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 +Quarantine only removes a VF from future placement: it never detaches the VF +or otherwise affects a running instance. + +Below-threshold failures log at warn and increment +`hypeman_instances_vgpu_sentinel_init_failures_total`; quarantines log at +error and increment `hypeman_instances_vgpu_sentinel_quarantines_total`. +`hypeman_instances_vgpu_sentinel_checks_total` records checks by result +(`ok`, `failed`, `unknown`, `rpc_error`, `unsupported_agent`, or `list_error`) +so hosts that lose sentinel coverage are visible. `unsupported_agent` means a +running instance has a guest agent from before the status RPC was introduced; +it is expected while those instances drain during an upgrade. +`hypeman_instances_vgpu_quarantined_vfs` gauges the current count. +`hypeman_instances_vgpu_vf_health_store_unavailable` is 1 while persisted +health state cannot be loaded or the last write failed (and placement is +therefore disabled), and 0 otherwise. A systemic guest/host driver mismatch +can still quarantine every VF, so validate driver changes on a test host and +alert on the failure counter. + +Detection requires the hypeman guest agent and a running instance: the state +lives in the agent and travels only over the vsock control channel — the +serial console is shared with workload output, so nothing a workload prints +can influence the tally. The guest is still the reporter, though: a workload +with root in the guest could replace the agent and answer FAILED, so the tally +is a capacity signal from cooperating guests, not a security boundary. The +per-assignment threshold and randomized VF selection bound how quickly one +tenant can drain a host's VFs. The wedge-creating kill itself leaves no host-side +log: no kernel error, no XID, no plugin crash. A wedge is therefore detected +on the next boot that lands on the VF, whose guest driver starts failing ~27s +after spawn; that also covers a wedged instance that stopped before the next +poll (5s). + +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 — @@ -371,7 +420,9 @@ systemctl start nvidia-persistenced nvidia-dcgm nvidia-dcgm-exporter ``` After the cycle, remove the card's entries from `vf-health.json`, restart, -and boot a GPU instance to verify recovery. +and boot a GPU instance to verify recovery. If the cycle did not work, the +sentinel quarantines the VF again after the configured number of fresh +assignment failures. 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`) diff --git a/lib/devices/vf_health.go b/lib/devices/vf_health.go index 8de8e23a0..6feb37af6 100644 --- a/lib/devices/vf_health.go +++ b/lib/devices/vf_health.go @@ -312,11 +312,71 @@ func ReportVFInitFailure(report VFInitFailureReport) (VFReportResult, error) { // ReportVFInitSuccess clears failures through an exactly matched successful // assignment. A quarantine is rescinded only when that assignment triggered it. func ReportVFInitSuccess(report VFInitSuccessReport) (VFSuccessResult, error) { + if !vfHealth.hasFailures(report.VFAddress) { + return VFSuccessResult{}, nil + } vendorVFIOMu.Lock() defer vendorVFIOMu.Unlock() return vfHealth.reportSuccess(report) } +// hasFailures reports whether a success for address could clear anything. It +// takes only s.mu so the sentinel's routine OK reports for healthy VFs skip +// vendorVFIOMu. Load and address errors report true so reportSuccess +// surfaces them. +func (s *vfHealthStore) hasFailures(address string) bool { + s.mu.Lock() + defer s.mu.Unlock() + if s.loadErr != nil || !vfHealthAddressPattern.MatchString(address) { + return true + } + record, ok := s.records[address] + return ok && len(record.Failures) > 0 +} + +// RepairVFHealthStore retries a failed load or persist so the store recovers +// without waiting for a placement or /resources read to do it. It takes +// vendorVFIOMu like the report paths, so a reload that changes the quarantine +// set cannot interleave with a vendor-VFIO configure. The same retry also +// runs from checkedAddresses under only s.mu; there, configure's own re-check +// under vendorVFIOMu covers the selection-to-configure window instead. It +// returns without taking either lock while the store is healthy. +func RepairVFHealthStore() error { + if !VFHealthStoreUnavailable() { + return nil + } + vendorVFIOMu.Lock() + defer vendorVFIOMu.Unlock() + + vfHealth.mu.Lock() + defer vfHealth.mu.Unlock() + if err := vfHealth.ensureLoadedLocked(); err != nil { + return err + } + return vfHealth.retryPersistLocked() +} + +// VFHealthStoreUnavailable reports whether persisted state failed to load or +// the last write failed. +func VFHealthStoreUnavailable() bool { + vfHealth.mu.Lock() + defer vfHealth.mu.Unlock() + return vfHealth.loadErr != nil || vfHealth.persistErr != nil +} + +// TotalQuarantinedVFs returns the number of quarantined VFs in persisted state. +func TotalQuarantinedVFs() int { + vfHealth.mu.Lock() + defer vfHealth.mu.Unlock() + count := 0 + for _, record := range vfHealth.records { + if record.QuarantinedAt != nil { + count++ + } + } + return count +} + func (s *vfHealthStore) sortedRecordsLocked() []vfHealthRecord { records := make([]vfHealthRecord, 0, len(s.records)) for _, record := range s.records { @@ -335,9 +395,6 @@ func (s *vfHealthStore) reportFailure(report VFInitFailureReport) (VFReportResul if !vfHealthAddressPattern.MatchString(report.VFAddress) { return VFReportResult{}, fmt.Errorf("invalid VF address %q", report.VFAddress) } - if err := s.retryPersistLocked(); err != nil { - return VFReportResult{}, err - } previous, existed := s.records[report.VFAddress] result := VFReportResult{Failures: len(previous.Failures), Threshold: s.threshold} @@ -349,6 +406,9 @@ func (s *vfHealthStore) reportFailure(report VFInitFailureReport) (VFReportResul return result, nil } } + if s.persistErr != nil { + return VFReportResult{}, fmt.Errorf("VF health state unavailable: last write failed: %w", s.persistErr) + } record := vfHealthRecord{ VFAddress: report.VFAddress, @@ -393,9 +453,6 @@ func (s *vfHealthStore) reportSuccess(report VFInitSuccessReport) (VFSuccessResu if !vfHealthAddressPattern.MatchString(report.VFAddress) { return VFSuccessResult{}, fmt.Errorf("invalid VF address %q", report.VFAddress) } - if err := s.retryPersistLocked(); err != nil { - return VFSuccessResult{}, err - } previous, ok := s.records[report.VFAddress] if !ok || len(previous.Failures) == 0 { return VFSuccessResult{}, nil @@ -411,6 +468,9 @@ func (s *vfHealthStore) reportSuccess(report VFInitSuccessReport) (VFSuccessResu if match < 0 { return VFSuccessResult{}, nil } + if s.persistErr != nil { + return VFSuccessResult{}, fmt.Errorf("VF health state unavailable: last write failed: %w", s.persistErr) + } // Only the newest failure can rescind a quarantine; see VFInitSuccessReport. remaining := append([]vfInitFailure(nil), previous.Failures[match+1:]...) diff --git a/lib/devices/vf_health_test.go b/lib/devices/vf_health_test.go index 7c2e93d15..04ba24eba 100644 --- a/lib/devices/vf_health_test.go +++ b/lib/devices/vf_health_test.go @@ -126,6 +126,7 @@ func TestVGPUAvailabilityFailsClosedAfterPersistFailure(t *testing.T) { require.ErrorContains(t, err, "last write failed") vfHealth.path = goodPath + require.NoError(t, RepairVFHealthStore()) result, err := ReportVFInitFailure(VFInitFailureReport{VFAddress: "0000:e3:00.4", InstanceID: "instance-1"}) require.NoError(t, err) assert.Equal(t, VFReportQuarantined, result.Outcome) @@ -391,6 +392,103 @@ func TestReportVFInitSuccessWithoutMatchingFailureClearsNothing(t *testing.T) { assert.False(t, result.Rescinded) } +func TestReportVFInitFailureNoopDoesNotErrorWhilePersistFailed(t *testing.T) { + resetVFHealthStore(t) + report := VFInitFailureReport{VFAddress: "0000:e3:00.4", InstanceID: "instance-1"} + _, err := ReportVFInitFailure(report) + require.NoError(t, err) + + vfHealth.mu.Lock() + vfHealth.persistErr = errors.New("injected persist failure") + vfHealth.mu.Unlock() + + result, err := ReportVFInitFailure(report) + require.NoError(t, err) + assert.Equal(t, VFReportUnchanged, result.Outcome, "an already-recorded assignment has nothing to write") + + _, err = ReportVFInitFailure(VFInitFailureReport{VFAddress: "0000:e3:00.4", InstanceID: "instance-2"}) + require.ErrorContains(t, err, "last write failed") +} + +func TestReportVFInitSuccessNoopDoesNotRetryFailedPersist(t *testing.T) { + resetVFHealthStore(t) + var syncCalls int + vfHealth.mu.Lock() + vfHealth.persistErr = errors.New("injected persist failure") + vfHealth.syncDirFunc = func(string) error { + syncCalls++ + return nil + } + vfHealth.mu.Unlock() + + result, err := ReportVFInitSuccess(VFInitSuccessReport{ + VFAddress: "0000:e3:00.4", + InstanceID: "healthy-instance", + }) + require.NoError(t, err) + assert.Equal(t, VFSuccessResult{}, result) + assert.Zero(t, syncCalls) + assert.True(t, VFHealthStoreUnavailable()) + + require.NoError(t, RepairVFHealthStore()) + assert.Equal(t, 2, syncCalls, "one repair performs one parent and state-directory sync") + assert.False(t, VFHealthStoreUnavailable()) +} + +// withVendorVFIOMuHeld fails the test if fn blocks on vendorVFIOMu. +func withVendorVFIOMuHeld(t *testing.T, fn func()) { + t.Helper() + vendorVFIOMu.Lock() + defer vendorVFIOMu.Unlock() + done := make(chan struct{}) + go func() { + defer close(done) + fn() + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("call blocked on vendorVFIOMu") + } +} + +func TestReportVFInitSuccessForHealthyVFSkipsVendorVFIOMu(t *testing.T) { + resetVFHealthStore(t) + withVendorVFIOMuHeld(t, func() { + result, err := ReportVFInitSuccess(VFInitSuccessReport{VFAddress: "0000:e3:00.4", InstanceID: "healthy-instance"}) + assert.NoError(t, err) + assert.Equal(t, VFSuccessResult{}, result) + }) + + _, err := ReportVFInitFailure(VFInitFailureReport{VFAddress: "0000:e3:00.4", InstanceID: "instance-1"}) + require.NoError(t, err) + result, err := ReportVFInitSuccess(VFInitSuccessReport{VFAddress: "0000:e3:00.4", InstanceID: "instance-1"}) + require.NoError(t, err) + assert.Equal(t, 1, result.Cleared, "a VF with tallied failures still takes the full path") +} + +func TestReportVFInitSuccessInvalidAddressStillErrors(t *testing.T) { + resetVFHealthStore(t) + _, err := ReportVFInitSuccess(VFInitSuccessReport{VFAddress: "not-a-vf", InstanceID: "instance-1"}) + require.ErrorContains(t, err, "invalid VF address") +} + +func TestRepairVFHealthStoreHealthySkipsVendorVFIOMu(t *testing.T) { + resetVFHealthStore(t) + var syncCalls int + vfHealth.mu.Lock() + vfHealth.syncDirFunc = func(string) error { + syncCalls++ + return nil + } + vfHealth.mu.Unlock() + + withVendorVFIOMuHeld(t, func() { + assert.NoError(t, RepairVFHealthStore()) + }) + assert.Zero(t, syncCalls, "a healthy store must not be rewritten") +} + func TestReportVFInitSuccessNeverClearsAnotherAssignmentsQuarantine(t *testing.T) { resetVFHealthStore(t) quarantineVF(t, "0000:e3:00.4") @@ -456,6 +554,7 @@ func TestReportVFInitFailureRetriesParentSyncAfterFailure(t *testing.T) { require.ErrorContains(t, err, "sync VF health state parent dir") assert.True(t, vfHealthStoreUnavailable()) + require.NoError(t, RepairVFHealthStore()) result, err := ReportVFInitFailure(report) require.NoError(t, err) assert.Equal(t, VFReportRecorded, result.Outcome) @@ -489,6 +588,7 @@ func TestReportVFInitFailureRetainsRenamedStateAfterSyncFailure(t *testing.T) { assert.True(t, vfHealthStoreUnavailable()) vfHealth.syncDirFunc = syncDir + require.NoError(t, RepairVFHealthStore()) result, err := ReportVFInitFailure(VFInitFailureReport{VFAddress: "0000:e3:00.5", InstanceID: "other-instance"}) require.NoError(t, err) assert.Equal(t, VFReportRecorded, result.Outcome) @@ -529,6 +629,7 @@ func TestReportRetriesFailedThresholdPersistence(t *testing.T) { assert.True(t, vfHealthStoreUnavailable()) vfHealth.path = path + require.NoError(t, RepairVFHealthStore()) result, err := ReportVFInitFailure(VFInitFailureReport{VFAddress: vf, InstanceID: "instance-3"}) require.NoError(t, err) assert.Equal(t, VFReportUnchanged, result.Outcome) @@ -563,6 +664,49 @@ func TestReportVFInitSuccessRollsBackOnPersistFailure(t *testing.T) { assert.Len(t, record.Failures, 1) } +func TestRepairVFHealthStoreRecoversPostRenameSuccessClearFailure(t *testing.T) { + path := resetVFHealthStore(t) + report := VFInitFailureReport{ + VFAddress: "0000:e3:00.4", + InstanceID: "instance-1", + AssignedAt: "2026-08-20T15:00:00Z", + } + _, err := ReportVFInitFailure(report) + require.NoError(t, err) + + failed := false + vfHealth.syncDirFunc = func(path string) error { + if path == filepath.Dir(vfHealth.path) && !failed { + failed = true + return errors.New("injected sync failure") + } + return syncDir(path) + } + _, err = ReportVFInitSuccess(VFInitSuccessReport{ + VFAddress: report.VFAddress, + InstanceID: report.InstanceID, + AssignedAt: report.AssignedAt, + }) + require.ErrorContains(t, err, "sync VF health state dir") + assert.True(t, VFHealthStoreUnavailable()) + + vfHealth.mu.Lock() + _, exists := vfHealth.records[report.VFAddress] + vfHealth.mu.Unlock() + assert.False(t, exists, "memory must retain the clear renamed into place") + + data, err := os.ReadFile(path) + require.NoError(t, err) + var state vfHealthFile + require.NoError(t, json.Unmarshal(data, &state)) + assert.Empty(t, state.Records) + + require.NoError(t, RepairVFHealthStore()) + assert.False(t, VFHealthStoreUnavailable()) + _, err = GetVGPUAvailability(VGPUFrameworkVendorVFIO, []VirtualFunction{{PCIAddress: report.VFAddress}}) + require.NoError(t, err) +} + func TestCheckedAddressesFailsClosedOnInvalidRecord(t *testing.T) { tests := []struct { name string diff --git a/lib/guest/client.go b/lib/guest/client.go index b6dcd2e8e..61376bf02 100644 --- a/lib/guest/client.go +++ b/lib/guest/client.go @@ -938,6 +938,30 @@ func CopyFromInstance(ctx context.Context, dialer hypervisor.VsockDialer, opts C return nil } +// GetGPUInitStatus reports the guest GPU driver init state observed by the +// guest agent, plus the NVRM failure line when the init failed. The serial +// console is shared with workload output, so this is the host's only trusted +// signal for GPU init health. +func GetGPUInitStatus(ctx context.Context, dialer hypervisor.VsockDialer) (GPUInitState, string, error) { + grpcConn, err := GetOrCreateConn(ctx, dialer) + if err != nil { + if isRetryableConnectionError(err) { + CloseConn(dialer.Key()) + } + return GPUInitState_GPU_INIT_STATE_UNKNOWN, "", fmt.Errorf("get grpc connection: %w", err) + } + + client := NewGuestServiceClient(grpcConn) + resp, err := client.GetGPUInitStatus(ctx, &GetGPUInitStatusRequest{}) + if err != nil { + if isRetryableConnectionError(err) { + CloseConn(dialer.Key()) + } + return GPUInitState_GPU_INIT_STATE_UNKNOWN, "", fmt.Errorf("gpu init status RPC: %w", err) + } + return resp.State, resp.FailureMessage, nil +} + // ShutdownInstance sends a shutdown signal to the guest VM's init process (PID 1). // The guest-agent forwards the signal to init, which forwards it to the entrypoint. // sig is the signal number to send (0 = SIGTERM default). diff --git a/lib/guest/client_test.go b/lib/guest/client_test.go index 090e95b59..8ec1a17fb 100644 --- a/lib/guest/client_test.go +++ b/lib/guest/client_test.go @@ -216,6 +216,24 @@ func TestExecIntoInstanceNoWaitClosesRetryableConnection(t *testing.T) { } } +func TestGetGPUInitStatusClosesRetryableConnection(t *testing.T) { + dialer := &alwaysFailDialer{key: "gpu-status-close-retryable-test"} + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + _, _, err := GetGPUInitStatus(ctx, dialer) + if err == nil { + t.Fatal("GetGPUInitStatus succeeded unexpectedly") + } + + connPool.RLock() + _, ok := connPool.conns[dialer.Key()] + connPool.RUnlock() + if ok { + t.Fatal("retryable GPU status error left connection in pool") + } +} + func TestCloseConnClosesPooledConnection(t *testing.T) { dialer := &trackingDialer{ key: "close-conn-test", diff --git a/lib/guest/guest.pb.go b/lib/guest/guest.pb.go index a239fc970..f1b1156a4 100644 --- a/lib/guest/guest.pb.go +++ b/lib/guest/guest.pb.go @@ -21,6 +21,56 @@ const ( _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) ) +// GPUInitState is the guest GPU driver init state observed by the guest agent +type GPUInitState int32 + +const ( + GPUInitState_GPU_INIT_STATE_UNKNOWN GPUInitState = 0 // No NVIDIA device, or init has not concluded + GPUInitState_GPU_INIT_STATE_OK GPUInitState = 1 // The driver initialized the GPU + GPUInitState_GPU_INIT_STATE_FAILED GPUInitState = 2 // The kernel reported an RmInitAdapter failure +) + +// Enum value maps for GPUInitState. +var ( + GPUInitState_name = map[int32]string{ + 0: "GPU_INIT_STATE_UNKNOWN", + 1: "GPU_INIT_STATE_OK", + 2: "GPU_INIT_STATE_FAILED", + } + GPUInitState_value = map[string]int32{ + "GPU_INIT_STATE_UNKNOWN": 0, + "GPU_INIT_STATE_OK": 1, + "GPU_INIT_STATE_FAILED": 2, + } +) + +func (x GPUInitState) Enum() *GPUInitState { + p := new(GPUInitState) + *p = x + return p +} + +func (x GPUInitState) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (GPUInitState) Descriptor() protoreflect.EnumDescriptor { + return file_lib_guest_guest_proto_enumTypes[0].Descriptor() +} + +func (GPUInitState) Type() protoreflect.EnumType { + return &file_lib_guest_guest_proto_enumTypes[0] +} + +func (x GPUInitState) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use GPUInitState.Descriptor instead. +func (GPUInitState) EnumDescriptor() ([]byte, []int) { + return file_lib_guest_guest_proto_rawDescGZIP(), []int{0} +} + // ExecRequest represents messages from client to server type ExecRequest struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -1380,6 +1430,97 @@ func (*ReconfigureNetworkResponse) Descriptor() ([]byte, []int) { return file_lib_guest_guest_proto_rawDescGZIP(), []int{18} } +// GetGPUInitStatusRequest requests the guest GPU driver init state +type GetGPUInitStatusRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GetGPUInitStatusRequest) Reset() { + *x = GetGPUInitStatusRequest{} + mi := &file_lib_guest_guest_proto_msgTypes[19] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GetGPUInitStatusRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetGPUInitStatusRequest) ProtoMessage() {} + +func (x *GetGPUInitStatusRequest) ProtoReflect() protoreflect.Message { + mi := &file_lib_guest_guest_proto_msgTypes[19] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GetGPUInitStatusRequest.ProtoReflect.Descriptor instead. +func (*GetGPUInitStatusRequest) Descriptor() ([]byte, []int) { + return file_lib_guest_guest_proto_rawDescGZIP(), []int{19} +} + +// GetGPUInitStatusResponse reports the guest GPU driver init state +type GetGPUInitStatusResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + State GPUInitState `protobuf:"varint,1,opt,name=state,proto3,enum=guest.GPUInitState" json:"state,omitempty"` + // Most recent NVRM init-failure line observed in kmsg; set when state is FAILED + FailureMessage string `protobuf:"bytes,2,opt,name=failure_message,json=failureMessage,proto3" json:"failure_message,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GetGPUInitStatusResponse) Reset() { + *x = GetGPUInitStatusResponse{} + mi := &file_lib_guest_guest_proto_msgTypes[20] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GetGPUInitStatusResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetGPUInitStatusResponse) ProtoMessage() {} + +func (x *GetGPUInitStatusResponse) ProtoReflect() protoreflect.Message { + mi := &file_lib_guest_guest_proto_msgTypes[20] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GetGPUInitStatusResponse.ProtoReflect.Descriptor instead. +func (*GetGPUInitStatusResponse) Descriptor() ([]byte, []int) { + return file_lib_guest_guest_proto_rawDescGZIP(), []int{20} +} + +func (x *GetGPUInitStatusResponse) GetState() GPUInitState { + if x != nil { + return x.State + } + return GPUInitState_GPU_INIT_STATE_UNKNOWN +} + +func (x *GetGPUInitStatusResponse) GetFailureMessage() string { + if x != nil { + return x.FailureMessage + } + return "" +} + var File_lib_guest_guest_proto protoreflect.FileDescriptor const file_lib_guest_guest_proto_rawDesc = "" + @@ -1479,14 +1620,23 @@ const file_lib_guest_guest_proto_rawDesc = "" + "\x04ipv4\x18\x03 \x01(\tR\x04ipv4\x12\x16\n" + "\x06prefix\x18\x04 \x01(\rR\x06prefix\x12\x18\n" + "\agateway\x18\x05 \x01(\tR\agateway\"\x1c\n" + - "\x1aReconfigureNetworkResponse2\xae\x03\n" + + "\x1aReconfigureNetworkResponse\"\x19\n" + + "\x17GetGPUInitStatusRequest\"n\n" + + "\x18GetGPUInitStatusResponse\x12)\n" + + "\x05state\x18\x01 \x01(\x0e2\x13.guest.GPUInitStateR\x05state\x12'\n" + + "\x0ffailure_message\x18\x02 \x01(\tR\x0efailureMessage*\\\n" + + "\fGPUInitState\x12\x1a\n" + + "\x16GPU_INIT_STATE_UNKNOWN\x10\x00\x12\x15\n" + + "\x11GPU_INIT_STATE_OK\x10\x01\x12\x19\n" + + "\x15GPU_INIT_STATE_FAILED\x10\x022\x83\x04\n" + "\fGuestService\x123\n" + "\x04Exec\x12\x12.guest.ExecRequest\x1a\x13.guest.ExecResponse(\x010\x01\x12F\n" + "\vCopyToGuest\x12\x19.guest.CopyToGuestRequest\x1a\x1a.guest.CopyToGuestResponse(\x01\x12L\n" + "\rCopyFromGuest\x12\x1b.guest.CopyFromGuestRequest\x1a\x1c.guest.CopyFromGuestResponse0\x01\x12;\n" + "\bStatPath\x12\x16.guest.StatPathRequest\x1a\x17.guest.StatPathResponse\x12;\n" + "\bShutdown\x12\x16.guest.ShutdownRequest\x1a\x17.guest.ShutdownResponse\x12Y\n" + - "\x12ReconfigureNetwork\x12 .guest.ReconfigureNetworkRequest\x1a!.guest.ReconfigureNetworkResponseB'Z%github.com/onkernel/hypeman/lib/guestb\x06proto3" + "\x12ReconfigureNetwork\x12 .guest.ReconfigureNetworkRequest\x1a!.guest.ReconfigureNetworkResponse\x12S\n" + + "\x10GetGPUInitStatus\x12\x1e.guest.GetGPUInitStatusRequest\x1a\x1f.guest.GetGPUInitStatusResponseB'Z%github.com/onkernel/hypeman/lib/guestb\x06proto3" var ( file_lib_guest_guest_proto_rawDescOnce sync.Once @@ -1500,55 +1650,62 @@ func file_lib_guest_guest_proto_rawDescGZIP() []byte { return file_lib_guest_guest_proto_rawDescData } -var file_lib_guest_guest_proto_msgTypes = make([]protoimpl.MessageInfo, 20) +var file_lib_guest_guest_proto_enumTypes = make([]protoimpl.EnumInfo, 1) +var file_lib_guest_guest_proto_msgTypes = make([]protoimpl.MessageInfo, 22) var file_lib_guest_guest_proto_goTypes = []any{ - (*ExecRequest)(nil), // 0: guest.ExecRequest - (*ExecStart)(nil), // 1: guest.ExecStart - (*WindowSize)(nil), // 2: guest.WindowSize - (*ExecResponse)(nil), // 3: guest.ExecResponse - (*CopyToGuestRequest)(nil), // 4: guest.CopyToGuestRequest - (*CopyToGuestStart)(nil), // 5: guest.CopyToGuestStart - (*CopyToGuestEnd)(nil), // 6: guest.CopyToGuestEnd - (*CopyToGuestResponse)(nil), // 7: guest.CopyToGuestResponse - (*CopyFromGuestRequest)(nil), // 8: guest.CopyFromGuestRequest - (*CopyFromGuestResponse)(nil), // 9: guest.CopyFromGuestResponse - (*CopyFromGuestHeader)(nil), // 10: guest.CopyFromGuestHeader - (*CopyFromGuestEnd)(nil), // 11: guest.CopyFromGuestEnd - (*CopyFromGuestError)(nil), // 12: guest.CopyFromGuestError - (*StatPathRequest)(nil), // 13: guest.StatPathRequest - (*StatPathResponse)(nil), // 14: guest.StatPathResponse - (*ShutdownRequest)(nil), // 15: guest.ShutdownRequest - (*ShutdownResponse)(nil), // 16: guest.ShutdownResponse - (*ReconfigureNetworkRequest)(nil), // 17: guest.ReconfigureNetworkRequest - (*ReconfigureNetworkResponse)(nil), // 18: guest.ReconfigureNetworkResponse - nil, // 19: guest.ExecStart.EnvEntry + (GPUInitState)(0), // 0: guest.GPUInitState + (*ExecRequest)(nil), // 1: guest.ExecRequest + (*ExecStart)(nil), // 2: guest.ExecStart + (*WindowSize)(nil), // 3: guest.WindowSize + (*ExecResponse)(nil), // 4: guest.ExecResponse + (*CopyToGuestRequest)(nil), // 5: guest.CopyToGuestRequest + (*CopyToGuestStart)(nil), // 6: guest.CopyToGuestStart + (*CopyToGuestEnd)(nil), // 7: guest.CopyToGuestEnd + (*CopyToGuestResponse)(nil), // 8: guest.CopyToGuestResponse + (*CopyFromGuestRequest)(nil), // 9: guest.CopyFromGuestRequest + (*CopyFromGuestResponse)(nil), // 10: guest.CopyFromGuestResponse + (*CopyFromGuestHeader)(nil), // 11: guest.CopyFromGuestHeader + (*CopyFromGuestEnd)(nil), // 12: guest.CopyFromGuestEnd + (*CopyFromGuestError)(nil), // 13: guest.CopyFromGuestError + (*StatPathRequest)(nil), // 14: guest.StatPathRequest + (*StatPathResponse)(nil), // 15: guest.StatPathResponse + (*ShutdownRequest)(nil), // 16: guest.ShutdownRequest + (*ShutdownResponse)(nil), // 17: guest.ShutdownResponse + (*ReconfigureNetworkRequest)(nil), // 18: guest.ReconfigureNetworkRequest + (*ReconfigureNetworkResponse)(nil), // 19: guest.ReconfigureNetworkResponse + (*GetGPUInitStatusRequest)(nil), // 20: guest.GetGPUInitStatusRequest + (*GetGPUInitStatusResponse)(nil), // 21: guest.GetGPUInitStatusResponse + nil, // 22: guest.ExecStart.EnvEntry } var file_lib_guest_guest_proto_depIdxs = []int32{ - 1, // 0: guest.ExecRequest.start:type_name -> guest.ExecStart - 2, // 1: guest.ExecRequest.resize:type_name -> guest.WindowSize - 19, // 2: guest.ExecStart.env:type_name -> guest.ExecStart.EnvEntry - 5, // 3: guest.CopyToGuestRequest.start:type_name -> guest.CopyToGuestStart - 6, // 4: guest.CopyToGuestRequest.end:type_name -> guest.CopyToGuestEnd - 10, // 5: guest.CopyFromGuestResponse.header:type_name -> guest.CopyFromGuestHeader - 11, // 6: guest.CopyFromGuestResponse.end:type_name -> guest.CopyFromGuestEnd - 12, // 7: guest.CopyFromGuestResponse.error:type_name -> guest.CopyFromGuestError - 0, // 8: guest.GuestService.Exec:input_type -> guest.ExecRequest - 4, // 9: guest.GuestService.CopyToGuest:input_type -> guest.CopyToGuestRequest - 8, // 10: guest.GuestService.CopyFromGuest:input_type -> guest.CopyFromGuestRequest - 13, // 11: guest.GuestService.StatPath:input_type -> guest.StatPathRequest - 15, // 12: guest.GuestService.Shutdown:input_type -> guest.ShutdownRequest - 17, // 13: guest.GuestService.ReconfigureNetwork:input_type -> guest.ReconfigureNetworkRequest - 3, // 14: guest.GuestService.Exec:output_type -> guest.ExecResponse - 7, // 15: guest.GuestService.CopyToGuest:output_type -> guest.CopyToGuestResponse - 9, // 16: guest.GuestService.CopyFromGuest:output_type -> guest.CopyFromGuestResponse - 14, // 17: guest.GuestService.StatPath:output_type -> guest.StatPathResponse - 16, // 18: guest.GuestService.Shutdown:output_type -> guest.ShutdownResponse - 18, // 19: guest.GuestService.ReconfigureNetwork:output_type -> guest.ReconfigureNetworkResponse - 14, // [14:20] is the sub-list for method output_type - 8, // [8:14] is the sub-list for method input_type - 8, // [8:8] is the sub-list for extension type_name - 8, // [8:8] is the sub-list for extension extendee - 0, // [0:8] is the sub-list for field type_name + 2, // 0: guest.ExecRequest.start:type_name -> guest.ExecStart + 3, // 1: guest.ExecRequest.resize:type_name -> guest.WindowSize + 22, // 2: guest.ExecStart.env:type_name -> guest.ExecStart.EnvEntry + 6, // 3: guest.CopyToGuestRequest.start:type_name -> guest.CopyToGuestStart + 7, // 4: guest.CopyToGuestRequest.end:type_name -> guest.CopyToGuestEnd + 11, // 5: guest.CopyFromGuestResponse.header:type_name -> guest.CopyFromGuestHeader + 12, // 6: guest.CopyFromGuestResponse.end:type_name -> guest.CopyFromGuestEnd + 13, // 7: guest.CopyFromGuestResponse.error:type_name -> guest.CopyFromGuestError + 0, // 8: guest.GetGPUInitStatusResponse.state:type_name -> guest.GPUInitState + 1, // 9: guest.GuestService.Exec:input_type -> guest.ExecRequest + 5, // 10: guest.GuestService.CopyToGuest:input_type -> guest.CopyToGuestRequest + 9, // 11: guest.GuestService.CopyFromGuest:input_type -> guest.CopyFromGuestRequest + 14, // 12: guest.GuestService.StatPath:input_type -> guest.StatPathRequest + 16, // 13: guest.GuestService.Shutdown:input_type -> guest.ShutdownRequest + 18, // 14: guest.GuestService.ReconfigureNetwork:input_type -> guest.ReconfigureNetworkRequest + 20, // 15: guest.GuestService.GetGPUInitStatus:input_type -> guest.GetGPUInitStatusRequest + 4, // 16: guest.GuestService.Exec:output_type -> guest.ExecResponse + 8, // 17: guest.GuestService.CopyToGuest:output_type -> guest.CopyToGuestResponse + 10, // 18: guest.GuestService.CopyFromGuest:output_type -> guest.CopyFromGuestResponse + 15, // 19: guest.GuestService.StatPath:output_type -> guest.StatPathResponse + 17, // 20: guest.GuestService.Shutdown:output_type -> guest.ShutdownResponse + 19, // 21: guest.GuestService.ReconfigureNetwork:output_type -> guest.ReconfigureNetworkResponse + 21, // 22: guest.GuestService.GetGPUInitStatus:output_type -> guest.GetGPUInitStatusResponse + 16, // [16:23] is the sub-list for method output_type + 9, // [9:16] is the sub-list for method input_type + 9, // [9:9] is the sub-list for extension type_name + 9, // [9:9] is the sub-list for extension extendee + 0, // [0:9] is the sub-list for field type_name } func init() { file_lib_guest_guest_proto_init() } @@ -1582,13 +1739,14 @@ func file_lib_guest_guest_proto_init() { File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_lib_guest_guest_proto_rawDesc), len(file_lib_guest_guest_proto_rawDesc)), - NumEnums: 0, - NumMessages: 20, + NumEnums: 1, + NumMessages: 22, NumExtensions: 0, NumServices: 1, }, GoTypes: file_lib_guest_guest_proto_goTypes, DependencyIndexes: file_lib_guest_guest_proto_depIdxs, + EnumInfos: file_lib_guest_guest_proto_enumTypes, MessageInfos: file_lib_guest_guest_proto_msgTypes, }.Build() File_lib_guest_guest_proto = out.File diff --git a/lib/guest/guest.proto b/lib/guest/guest.proto index 317c21b3e..a450665d1 100644 --- a/lib/guest/guest.proto +++ b/lib/guest/guest.proto @@ -23,6 +23,9 @@ service GuestService { // ReconfigureNetwork updates the guest network identity without spawning shell commands rpc ReconfigureNetwork(ReconfigureNetworkRequest) returns (ReconfigureNetworkResponse); + + // GetGPUInitStatus reports the GPU driver init state observed by the guest agent + rpc GetGPUInitStatus(GetGPUInitStatusRequest) returns (GetGPUInitStatusResponse); } // ExecRequest represents messages from client to server @@ -169,3 +172,20 @@ message ReconfigureNetworkRequest { // ReconfigureNetworkResponse acknowledges the network reconfiguration request message ReconfigureNetworkResponse {} + +// GPUInitState is the guest GPU driver init state observed by the guest agent +enum GPUInitState { + GPU_INIT_STATE_UNKNOWN = 0; // No NVIDIA device, or init has not concluded + GPU_INIT_STATE_OK = 1; // The driver initialized the GPU + GPU_INIT_STATE_FAILED = 2; // The kernel reported an RmInitAdapter failure +} + +// GetGPUInitStatusRequest requests the guest GPU driver init state +message GetGPUInitStatusRequest {} + +// GetGPUInitStatusResponse reports the guest GPU driver init state +message GetGPUInitStatusResponse { + GPUInitState state = 1; + // Most recent NVRM init-failure line observed in kmsg; set when state is FAILED + string failure_message = 2; +} diff --git a/lib/guest/guest_grpc.pb.go b/lib/guest/guest_grpc.pb.go index f93631d93..630a0bd36 100644 --- a/lib/guest/guest_grpc.pb.go +++ b/lib/guest/guest_grpc.pb.go @@ -25,6 +25,7 @@ const ( GuestService_StatPath_FullMethodName = "/guest.GuestService/StatPath" GuestService_Shutdown_FullMethodName = "/guest.GuestService/Shutdown" GuestService_ReconfigureNetwork_FullMethodName = "/guest.GuestService/ReconfigureNetwork" + GuestService_GetGPUInitStatus_FullMethodName = "/guest.GuestService/GetGPUInitStatus" ) // GuestServiceClient is the client API for GuestService service. @@ -45,6 +46,8 @@ type GuestServiceClient interface { Shutdown(ctx context.Context, in *ShutdownRequest, opts ...grpc.CallOption) (*ShutdownResponse, error) // ReconfigureNetwork updates the guest network identity without spawning shell commands ReconfigureNetwork(ctx context.Context, in *ReconfigureNetworkRequest, opts ...grpc.CallOption) (*ReconfigureNetworkResponse, error) + // GetGPUInitStatus reports the GPU driver init state observed by the guest agent + GetGPUInitStatus(ctx context.Context, in *GetGPUInitStatusRequest, opts ...grpc.CallOption) (*GetGPUInitStatusResponse, error) } type guestServiceClient struct { @@ -130,6 +133,16 @@ func (c *guestServiceClient) ReconfigureNetwork(ctx context.Context, in *Reconfi return out, nil } +func (c *guestServiceClient) GetGPUInitStatus(ctx context.Context, in *GetGPUInitStatusRequest, opts ...grpc.CallOption) (*GetGPUInitStatusResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(GetGPUInitStatusResponse) + err := c.cc.Invoke(ctx, GuestService_GetGPUInitStatus_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + // GuestServiceServer is the server API for GuestService service. // All implementations must embed UnimplementedGuestServiceServer // for forward compatibility. @@ -148,6 +161,8 @@ type GuestServiceServer interface { Shutdown(context.Context, *ShutdownRequest) (*ShutdownResponse, error) // ReconfigureNetwork updates the guest network identity without spawning shell commands ReconfigureNetwork(context.Context, *ReconfigureNetworkRequest) (*ReconfigureNetworkResponse, error) + // GetGPUInitStatus reports the GPU driver init state observed by the guest agent + GetGPUInitStatus(context.Context, *GetGPUInitStatusRequest) (*GetGPUInitStatusResponse, error) mustEmbedUnimplementedGuestServiceServer() } @@ -176,6 +191,9 @@ func (UnimplementedGuestServiceServer) Shutdown(context.Context, *ShutdownReques func (UnimplementedGuestServiceServer) ReconfigureNetwork(context.Context, *ReconfigureNetworkRequest) (*ReconfigureNetworkResponse, error) { return nil, status.Error(codes.Unimplemented, "method ReconfigureNetwork not implemented") } +func (UnimplementedGuestServiceServer) GetGPUInitStatus(context.Context, *GetGPUInitStatusRequest) (*GetGPUInitStatusResponse, error) { + return nil, status.Error(codes.Unimplemented, "method GetGPUInitStatus not implemented") +} func (UnimplementedGuestServiceServer) mustEmbedUnimplementedGuestServiceServer() {} func (UnimplementedGuestServiceServer) testEmbeddedByValue() {} @@ -276,6 +294,24 @@ func _GuestService_ReconfigureNetwork_Handler(srv interface{}, ctx context.Conte return interceptor(ctx, in, info, handler) } +func _GuestService_GetGPUInitStatus_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(GetGPUInitStatusRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(GuestServiceServer).GetGPUInitStatus(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: GuestService_GetGPUInitStatus_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(GuestServiceServer).GetGPUInitStatus(ctx, req.(*GetGPUInitStatusRequest)) + } + return interceptor(ctx, in, info, handler) +} + // GuestService_ServiceDesc is the grpc.ServiceDesc for GuestService service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -295,6 +331,10 @@ var GuestService_ServiceDesc = grpc.ServiceDesc{ MethodName: "ReconfigureNetwork", Handler: _GuestService_ReconfigureNetwork_Handler, }, + { + MethodName: "GetGPUInitStatus", + Handler: _GuestService_GetGPUInitStatus_Handler, + }, }, Streams: []grpc.StreamDesc{ { diff --git a/lib/instances/metrics.go b/lib/instances/metrics.go index 6711fb0f1..8d86b6cc7 100644 --- a/lib/instances/metrics.go +++ b/lib/instances/metrics.go @@ -5,6 +5,7 @@ import ( "strconv" "time" + "github.com/kernel/hypeman/lib/devices" "github.com/kernel/hypeman/lib/hypervisor" mw "github.com/kernel/hypeman/lib/middleware" hypotel "github.com/kernel/hypeman/lib/otel" @@ -103,6 +104,7 @@ type Metrics struct { ttlReaperDeletionsTotal metric.Int64Counter vgpuReconcileFailuresTotal metric.Int64Counter vgpuStaleReleaseFailuresTotal metric.Int64Counter + vgpuLivenessUncertainTotal metric.Int64Counter tracer trace.Tracer } @@ -295,6 +297,47 @@ func newInstanceMetrics(meter metric.Meter, tracer trace.Tracer, m *manager) (*M return nil, err } + vgpuLivenessUncertainTotal, err := meter.Int64Counter( + "hypeman_instances_vgpu_liveness_uncertain_total", + metric.WithDescription("Total vGPU release checks that preserved an assignment because hypervisor liveness was uncertain"), + ) + if err != nil { + return nil, err + } + + vgpuQuarantinedVFs, err := meter.Int64ObservableGauge( + "hypeman_instances_vgpu_quarantined_vfs", + metric.WithDescription("Number of vGPU virtual functions currently quarantined"), + ) + if err != nil { + return nil, err + } + + vgpuVFHealthStoreUnavailable, err := meter.Int64ObservableGauge( + "hypeman_instances_vgpu_vf_health_store_unavailable", + metric.WithDescription("1 when the persisted VF health state failed to load or persist; quarantine mutations are refused and vGPU placement is disabled until it is repaired"), + ) + if err != nil { + return nil, err + } + + _, err = meter.RegisterCallback( + func(_ context.Context, o metric.Observer) error { + o.ObserveInt64(vgpuQuarantinedVFs, int64(devices.TotalQuarantinedVFs())) + unavailable := int64(0) + if devices.VFHealthStoreUnavailable() { + unavailable = 1 + } + o.ObserveInt64(vgpuVFHealthStoreUnavailable, unavailable) + return nil + }, + vgpuQuarantinedVFs, + vgpuVFHealthStoreUnavailable, + ) + if err != nil { + return nil, err + } + // Register observable gauge for instance counts by state instancesTotal, err := meter.Int64ObservableGauge( "hypeman_instances_total", @@ -491,6 +534,7 @@ func newInstanceMetrics(meter metric.Meter, tracer trace.Tracer, m *manager) (*M ttlReaperDeletionsTotal: ttlReaperDeletionsTotal, vgpuReconcileFailuresTotal: vgpuReconcileFailuresTotal, vgpuStaleReleaseFailuresTotal: vgpuStaleReleaseFailuresTotal, + vgpuLivenessUncertainTotal: vgpuLivenessUncertainTotal, tracer: tracer, }, nil } @@ -606,6 +650,13 @@ func (m *manager) recordVGPUStaleReleaseFailure(ctx context.Context) { m.metrics.vgpuStaleReleaseFailuresTotal.Add(ctx, 1) } +func (m *manager) recordVGPULivenessUncertain(ctx context.Context) { + if m.metrics == nil { + return + } + m.metrics.vgpuLivenessUncertainTotal.Add(ctx, 1) +} + // recordStateTransition records a state transition with hypervisor label. func (m *manager) recordStateTransition(ctx context.Context, fromState, toState string, hvType hypervisor.Type) { if m.metrics == nil { diff --git a/lib/instances/metrics_test.go b/lib/instances/metrics_test.go index f670ea60d..511ff2f92 100644 --- a/lib/instances/metrics_test.go +++ b/lib/instances/metrics_test.go @@ -161,6 +161,27 @@ func TestSnapshotCompressionMetrics_RecordAndObserve(t *testing.T) { assert.Equal(t, "skipped", metricLabel(t, waitDurations.DataPoints[0].Attributes, "outcome")) } +func TestVGPULivenessUncertainMetric(t *testing.T) { + t.Parallel() + + reader := otelmetric.NewManualReader() + provider := otelmetric.NewMeterProvider(otelmetric.WithReader(reader)) + m := &manager{paths: paths.New(t.TempDir())} + metrics, err := newInstanceMetrics(provider.Meter("test"), nil, m) + require.NoError(t, err) + m.metrics = metrics + + m.recordVGPULivenessUncertain(t.Context()) + + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(t.Context(), &rm)) + metricValue := findMetric(t, rm, "hypeman_instances_vgpu_liveness_uncertain_total") + uncertain, ok := metricValue.Data.(metricdata.Sum[int64]) + require.True(t, ok) + require.Len(t, uncertain.DataPoints, 1) + assert.Equal(t, int64(1), uncertain.DataPoints[0].Value) +} + func TestLifecycleEventMetrics_ObserveSubscribersQueueDepthAndDrops(t *testing.T) { t.Parallel() diff --git a/lib/instances/process_identity.go b/lib/instances/process_identity.go index 3abc21da9..421f9f0ae 100644 --- a/lib/instances/process_identity.go +++ b/lib/instances/process_identity.go @@ -201,12 +201,6 @@ func classifyResolvedHypervisorOwner(socketPath string, stored, resolved int, er return 0, fmt.Errorf("cannot confirm ownership of socket %s: %w", socketPath, err) } -// Ambiguous ownership is treated as live; this must not authorize teardown. -func hypervisorMayBeAlive(id HypervisorProcessIdentity, socketPath string) bool { - pid, err := resolveLiveHypervisorPID(id, socketPath) - return err != nil || pid > 0 -} - // ProcessExists reports whether pid belongs to a live, non-zombie process. func ProcessExists(pid int) bool { if pid <= 0 { diff --git a/lib/instances/process_identity_linux_test.go b/lib/instances/process_identity_linux_test.go index b0a35a842..eec3a0acf 100644 --- a/lib/instances/process_identity_linux_test.go +++ b/lib/instances/process_identity_linux_test.go @@ -163,6 +163,10 @@ func TestResolveLiveHypervisorPIDFailsClosedWithoutSocketOrIdentity(t *testing.T resolved, err := resolveLiveHypervisorPID(HypervisorProcessIdentity{HypervisorPID: &pid}, "") require.ErrorContains(t, err, "without a socket path") assert.Zero(t, resolved) + + m := &manager{} + stored := &StoredMetadata{HypervisorProcessIdentity: HypervisorProcessIdentity{HypervisorPID: &pid}} + assert.True(t, m.vgpuHypervisorMayBeAlive(t.Context(), stored), "ambiguous ownership must still fail closed") } func TestSocketListenerHelper(t *testing.T) { diff --git a/lib/instances/storage.go b/lib/instances/storage.go index 1a4d325b0..eee2c4a4f 100644 --- a/lib/instances/storage.go +++ b/lib/instances/storage.go @@ -188,27 +188,41 @@ func removeAllWithRetry(path string, removeAll func(string) error, sleep func(ti } func (m *manager) listMetadataFiles() ([]string, error) { - return m.walkMetadataFiles(false) + files, _, err := m.walkMetadataFiles() + return files, err } +// listMetadataFilesStrict fails on any stat error other than absence, so +// fail-closed callers treat an unreadable instance as an error instead of +// silently missing it. func (m *manager) listMetadataFilesStrict() ([]string, error) { - return m.walkMetadataFiles(true) + files, statErrs, err := m.walkMetadataFiles() + if err != nil { + return nil, err + } + if len(statErrs) > 0 { + return nil, errors.Join(statErrs...) + } + return files, nil } -func (m *manager) walkMetadataFiles(failOnStatError bool) ([]string, error) { +// walkMetadataFiles returns readable metadata paths plus one error per +// instance whose metadata could not be stat'd for a reason other than absence. +func (m *manager) walkMetadataFiles() ([]string, []error, 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 + var statErrs []error for _, entry := range entries { if !entry.IsDir() { continue @@ -217,10 +231,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, statErrs, nil } diff --git a/lib/instances/vgpu.go b/lib/instances/vgpu.go index 0296ef34d..4563398d0 100644 --- a/lib/instances/vgpu.go +++ b/lib/instances/vgpu.go @@ -327,6 +327,22 @@ func clearStoredVGPUDevice(stored *StoredMetadata) { stored.GPUClaimedAt = nil } +// vgpuHypervisorMayBeAlive treats ambiguous ownership as live, so it never +// authorizes a release, and records why ownership could not be established +// so retained capacity remains diagnosable. +func (m *manager) vgpuHypervisorMayBeAlive(ctx context.Context, stored *StoredMetadata) bool { + pid, err := resolveLiveHypervisorPID(stored.HypervisorProcessIdentity, stored.SocketPath) + if err == nil { + return pid > 0 + } + logger.FromContext(ctx).WarnContext(ctx, "preserving vGPU claim because hypervisor liveness is uncertain", + "instance_id", stored.Id, + "device_path", storedVGPUDevicePath(stored), + "error", err) + m.recordVGPULivenessUncertain(ctx) + return true +} + // vgpuCleanupGuard checks whether the VF can be safely released: the VMM // must be dead and, for vendor VFIO, the on-disk claim must still match. // Returns the device path, or "" when cleanup must be skipped. @@ -335,8 +351,7 @@ func (m *manager) vgpuCleanupGuard(ctx context.Context, stored *StoredMetadata) if path == "" { return "" } - if hypervisorMayBeAlive(stored.HypervisorProcessIdentity, stored.SocketPath) { - logger.FromContext(ctx).WarnContext(ctx, "preserving vGPU claim because hypervisor liveness is not clear", "instance_id", stored.Id, "device_path", path) + if m.vgpuHypervisorMayBeAlive(ctx, stored) { return "" } if stored.GPUFramework == devices.VGPUFrameworkVendorVFIO { diff --git a/lib/instances/vgpu_reconcile.go b/lib/instances/vgpu_reconcile.go index 74bfb55a1..c4d2defdb 100644 --- a/lib/instances/vgpu_reconcile.go +++ b/lib/instances/vgpu_reconcile.go @@ -76,7 +76,7 @@ func (m *manager) reconcileVGPUAssignments(ctx context.Context) (map[string]stru if devicePath == "" { continue } - if hypervisorMayBeAlive(stored.HypervisorProcessIdentity, stored.SocketPath) { + if m.vgpuHypervisorMayBeAlive(ctx, stored) { protected[devicePath] = struct{}{} continue } @@ -105,7 +105,7 @@ func (m *manager) releaseStaleVGPUAssignment(ctx context.Context, id string) { if path == "" { return } - if hypervisorMayBeAlive(stored.HypervisorProcessIdentity, stored.SocketPath) { + if m.vgpuHypervisorMayBeAlive(ctx, stored) { return } if err := m.releaseStoredVGPUPersisted(ctx, meta); err != nil { diff --git a/lib/instances/vgpu_sentinel.go b/lib/instances/vgpu_sentinel.go new file mode 100644 index 000000000..b6ebd1cfd --- /dev/null +++ b/lib/instances/vgpu_sentinel.go @@ -0,0 +1,363 @@ +package instances + +import ( + "context" + "errors" + "fmt" + "log/slog" + "os" + "path/filepath" + "time" + + "github.com/kernel/hypeman/lib/devices" + "github.com/kernel/hypeman/lib/guest" + "github.com/kernel/hypeman/lib/hypervisor" + "github.com/kernel/hypeman/lib/logger" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" + "golang.org/x/sync/errgroup" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +const ( + vgpuSentinelPollInterval = 5 * time.Second + vgpuSentinelPollTimeout = 5 * time.Second + vgpuSentinelMaxConcurrentPolls = 64 +) + +type vgpuSentinelTarget struct { + instanceID string + vfAddress string + assignedAt string + hypervisorType hypervisor.Type + vsockSocket string + vsockCID int64 +} + +type vgpuSentinelStore interface { + listVGPUSentinelTargets(ctx context.Context) ([]vgpuSentinelTarget, error) + getVGPUSentinelTarget(ctx context.Context, instanceID string) (vgpuSentinelTarget, bool, error) +} + +var _ vgpuSentinelStore = (*manager)(nil) + +// VGPUSentinelController quarantines VFs whose guest reports a wedged driver +// init. It polls each vendor-VFIO instance's guest agent over vsock; the +// serial console is shared with workload output, so nothing read from logs is +// trusted. The health store deduplicates repeated reports per assignment, so +// polling is idempotent and a failed persist retries once on the next tick. +type VGPUSentinelController struct { + store vgpuSentinelStore + log *slog.Logger + interval time.Duration + repairHealthStore func() error + reportFailure func(devices.VFInitFailureReport) (devices.VFReportResult, error) + reportSuccess func(devices.VFInitSuccessReport) (devices.VFSuccessResult, error) + guestGPUInitStatus func(ctx context.Context, target vgpuSentinelTarget) (guest.GPUInitState, string, error) + discoverFramework func() (devices.VGPUFramework, []devices.VirtualFunction, error) + initFailures metric.Int64Counter + quarantines metric.Int64Counter + checks metric.Int64Counter + + // repairErrLoggedAt throttles the repair warning while the store stays + // unavailable; the gauge carries the condition between log lines. + repairErrLoggedAt time.Time +} + +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) + } + + initFailures, err := meter.Int64Counter( + "hypeman_instances_vgpu_sentinel_init_failures_total", + metric.WithDescription("Total guest-reported vGPU driver init failures recorded by the sentinel (one per instance assignment)"), + ) + if err != nil { + return nil, err + } + quarantines, err := meter.Int64Counter( + "hypeman_instances_vgpu_sentinel_quarantines_total", + metric.WithDescription("Total VFs quarantined by the vGPU sentinel"), + ) + if err != nil { + return nil, err + } + checks, err := meter.Int64Counter( + "hypeman_instances_vgpu_sentinel_checks_total", + metric.WithDescription("Total vGPU sentinel checks by result"), + ) + if err != nil { + return nil, err + } + + return &VGPUSentinelController{ + store: store, + log: log.With("controller", "vgpu_sentinel"), + interval: vgpuSentinelPollInterval, + repairHealthStore: devices.RepairVFHealthStore, + reportFailure: devices.ReportVFInitFailure, + reportSuccess: devices.ReportVFInitSuccess, + guestGPUInitStatus: func(ctx context.Context, target vgpuSentinelTarget) (guest.GPUInitState, string, error) { + dialer, err := hypervisor.NewVsockDialer(target.hypervisorType, target.vsockSocket, target.vsockCID) + if err != nil { + return guest.GPUInitState_GPU_INIT_STATE_UNKNOWN, "", err + } + return guest.GetGPUInitStatus(ctx, dialer) + }, + discoverFramework: devices.DiscoverVGPU, + initFailures: initFailures, + quarantines: quarantines, + checks: checks, + }, nil +} + +func (c *VGPUSentinelController) Run(ctx context.Context) error { + // Store methods log through the context, so they carry the controller + // attributes. + ctx = logger.AddToContext(ctx, c.log) + ticker := time.NewTicker(c.interval) + defer ticker.Stop() + var probeErrLoggedAt time.Time + for { + vendorVFIO, err := c.probeVendorVFIO() + if err == nil { + if !vendorVFIO { + return nil + } + break + } + if time.Since(probeErrLoggedAt) >= time.Minute { + c.log.Warn("vGPU sentinel framework discovery failed; retrying", "error", err) + probeErrLoggedAt = time.Now() + } + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + } + } + c.log.Info("vGPU sentinel controller started") + for { + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + c.pollOnce(ctx) + } + } +} + +func (c *VGPUSentinelController) probeVendorVFIO() (bool, error) { + framework, _, err := c.discoverFramework() + if err != nil { + return false, err + } + if framework == devices.VGPUFrameworkNone { + c.log.Info("vGPU sentinel controller exiting: host has no vGPU framework") + return false, nil + } + if framework != devices.VGPUFrameworkVendorVFIO { + c.log.Info("vGPU sentinel controller exiting: host vGPU framework is not vendor VFIO", "framework", string(framework)) + return false, nil + } + return true, nil +} + +func (c *VGPUSentinelController) pollOnce(ctx context.Context) { + if err := c.repairHealthStore(); err != nil && time.Since(c.repairErrLoggedAt) >= time.Minute { + c.log.WarnContext(ctx, "vGPU sentinel failed to repair VF health state", "error", err) + c.repairErrLoggedAt = time.Now() + } + targets, err := c.store.listVGPUSentinelTargets(ctx) + if err != nil { + c.recordCheck(ctx, "list_error") + c.log.WarnContext(ctx, "vGPU sentinel failed to list instances", "error", err) + return + } + var group errgroup.Group + group.SetLimit(vgpuSentinelMaxConcurrentPolls) + for _, target := range targets { + group.Go(func() error { + c.pollTarget(ctx, target) + return nil + }) + } + _ = group.Wait() +} + +func (c *VGPUSentinelController) pollTarget(ctx context.Context, target vgpuSentinelTarget) { + pollCtx, cancel := context.WithTimeout(ctx, vgpuSentinelPollTimeout) + state, nvrm, err := c.guestGPUInitStatus(pollCtx, target) + cancel() + if err != nil { + result := "rpc_error" + if status.Code(err) == codes.Unimplemented { + result = "unsupported_agent" + } + c.recordCheck(ctx, result) + // The agent can be unreachable while the instance stops or boots; the + // next tick polls again. + c.log.DebugContext(ctx, "vGPU sentinel cannot query the guest agent", + "instance_id", target.instanceID, "error", err) + return + } + switch state { + case guest.GPUInitState_GPU_INIT_STATE_FAILED: + c.recordCheck(ctx, "failed") + c.handleFailure(ctx, target, nvrm) + case guest.GPUInitState_GPU_INIT_STATE_OK: + c.recordCheck(ctx, "ok") + c.handleSuccess(ctx, target) + default: + c.recordCheck(ctx, "unknown") + } +} + +func (c *VGPUSentinelController) recordCheck(ctx context.Context, result string) { + c.checks.Add(ctx, 1, metric.WithAttributes(attribute.String("result", result))) +} + +// confirmAssignment rejects a report only when the instance now holds a +// different VF assignment. Released assignments remain attributable to the +// assignment captured in the poll target: a guest CID derives from the +// instance ID, so a poll that outlives its instance can only have reached +// another guest through a CID collision, not through sequential reuse. +func (c *VGPUSentinelController) confirmAssignment(ctx context.Context, target vgpuSentinelTarget, action string) bool { + current, ok, err := c.store.getVGPUSentinelTarget(ctx, target.instanceID) + if err != nil { + c.log.WarnContext(ctx, "vGPU sentinel could not confirm assignment; dropping report", + "action", action, "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 report: assignment changed during poll", + "action", action, "vf", target.vfAddress, "instance_id", target.instanceID) + return false + } + return true +} + +func (c *VGPUSentinelController) handleFailure(ctx context.Context, target vgpuSentinelTarget, nvrm string) { + if !c.confirmAssignment(ctx, target, "init_failure") { + return + } + result, err := c.reportFailure(devices.VFInitFailureReport{ + VFAddress: target.vfAddress, + InstanceID: target.instanceID, + AssignedAt: target.assignedAt, + }) + if err != nil { + c.log.ErrorContext(ctx, "failed to record vGPU VF init failure", + "vf", target.vfAddress, "instance_id", target.instanceID, "error", err) + return + } + switch result.Outcome { + case devices.VFReportQuarantined: + c.log.ErrorContext(ctx, "quarantined wedged vGPU VF", + "vf", target.vfAddress, + "instance_id", target.instanceID, + "nvrm", nvrm, + "failures", result.Failures, + "threshold", result.Threshold, + ) + c.initFailures.Add(ctx, 1) + c.quarantines.Add(ctx, 1) + case devices.VFReportRecorded: + c.log.WarnContext(ctx, "recorded vGPU VF init failure below quarantine threshold", + "vf", target.vfAddress, + "instance_id", target.instanceID, + "nvrm", nvrm, + "failures", result.Failures, + "threshold", result.Threshold, + ) + c.initFailures.Add(ctx, 1) + } +} + +func (c *VGPUSentinelController) handleSuccess(ctx context.Context, target vgpuSentinelTarget) { + if !c.confirmAssignment(ctx, target, "init_success") { + return + } + result, err := c.reportSuccess(devices.VFInitSuccessReport{ + VFAddress: target.vfAddress, + InstanceID: target.instanceID, + AssignedAt: target.assignedAt, + }) + if err != nil { + // The guest keeps reporting OK, so the next poll retries the clear. + c.log.WarnContext(ctx, "vGPU sentinel failed to clear recorded init failures; will retry", + "vf", target.vfAddress, "instance_id", target.instanceID, "error", err) + return + } + if result.Rescinded { + c.log.InfoContext(ctx, "rescinded vGPU VF quarantine after successful driver init from the triggering assignment", + "vf", target.vfAddress, "instance_id", target.instanceID, "cleared", result.Cleared) + } else if result.Cleared > 0 { + c.log.InfoContext(ctx, "cleared recorded vGPU VF init failures after successful driver init", + "vf", target.vfAddress, "instance_id", target.instanceID, "cleared", result.Cleared) + } +} + +func (m *manager) listVGPUSentinelTargets(ctx context.Context) ([]vgpuSentinelTarget, error) { + files, statErrs, err := m.walkMetadataFiles() + if err != nil { + return nil, err + } + if len(statErrs) > 0 { + logger.FromContext(ctx).WarnContext(ctx, "vGPU sentinel cannot stat some instance metadata; their VFs are not scanned", "error", errors.Join(statErrs...)) + } + 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 { + 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 +} + +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 + } + // No control socket means no VMM (stopped or standby), so there is no + // guest agent to poll. A stopped instance whose release failed keeps its + // claim until reconcile releases it, and QEMU vsock dials by guest CID + // alone, so polling it could reach an unrelated live guest. + if _, err := os.Stat(meta.SocketPath); err != nil { + return vgpuSentinelTarget{}, false, nil + } + assignedAt := "" + if meta.GPUClaimedAt != nil { + assignedAt = devices.FormatVFAssignedAt(*meta.GPUClaimedAt) + } + return vgpuSentinelTarget{ + instanceID: instanceID, + vfAddress: filepath.Base(meta.GPUDevicePath), + assignedAt: assignedAt, + hypervisorType: meta.HypervisorType, + vsockSocket: meta.VsockSocket, + vsockCID: meta.VsockCID, + }, true, nil +} diff --git a/lib/instances/vgpu_sentinel_test.go b/lib/instances/vgpu_sentinel_test.go new file mode 100644 index 000000000..f50d8c824 --- /dev/null +++ b/lib/instances/vgpu_sentinel_test.go @@ -0,0 +1,507 @@ +package instances + +import ( + "bytes" + "context" + "errors" + "fmt" + "log/slog" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/kernel/hypeman/lib/devices" + "github.com/kernel/hypeman/lib/guest" + "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" + otelmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +const testNVRMMessage = "NVRM: GPU 0000:e3:00.4: RmInitAdapter failed! (0x22:0x65:884)" + +type fakeSentinelStore struct { + targets []vgpuSentinelTarget + listCalls int +} + +func (s *fakeSentinelStore) listVGPUSentinelTargets(context.Context) ([]vgpuSentinelTarget, error) { + s.listCalls++ + 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 +} + +// sentinelReports records what the controller handed to the health store. +// Targets are polled concurrently, so appends are serialized. +type sentinelReports struct { + mu sync.Mutex + failures []devices.VFInitFailureReport + successes []devices.VFInitSuccessReport +} + +func (r *sentinelReports) recordFailure(report devices.VFInitFailureReport) int { + r.mu.Lock() + defer r.mu.Unlock() + r.failures = append(r.failures, report) + return len(r.failures) +} + +func (r *sentinelReports) recordSuccess(report devices.VFInitSuccessReport) int { + r.mu.Lock() + defer r.mu.Unlock() + r.successes = append(r.successes, report) + return len(r.successes) +} + +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 guestReportsFailed(context.Context, vgpuSentinelTarget) (guest.GPUInitState, string, error) { + return guest.GPUInitState_GPU_INIT_STATE_FAILED, testNVRMMessage, nil +} + +func guestReportsOK(context.Context, vgpuSentinelTarget) (guest.GPUInitState, string, error) { + return guest.GPUInitState_GPU_INIT_STATE_OK, "", nil +} + +func newTestSentinelController(t *testing.T, store *fakeSentinelStore) (*VGPUSentinelController, *sentinelReports) { + t.Helper() + counter, err := noop.NewMeterProvider().Meter("test").Int64Counter("test") + require.NoError(t, err) + reports := &sentinelReports{} + c := &VGPUSentinelController{ + store: store, + log: slog.New(slog.DiscardHandler), + interval: time.Hour, + repairHealthStore: func() error { return nil }, + reportFailure: func(report devices.VFInitFailureReport) (devices.VFReportResult, error) { + reports.recordFailure(report) + return devices.VFReportResult{Outcome: devices.VFReportQuarantined, Failures: 1, Threshold: 1}, nil + }, + reportSuccess: func(report devices.VFInitSuccessReport) (devices.VFSuccessResult, error) { + reports.recordSuccess(report) + return devices.VFSuccessResult{}, nil + }, + guestGPUInitStatus: guestReportsFailed, + initFailures: counter, + quarantines: counter, + checks: counter, + } + return c, reports +} + +func TestVGPUSentinelControllerReportsFailure(t *testing.T) { + t.Parallel() + + store := &fakeSentinelStore{targets: []vgpuSentinelTarget{{ + instanceID: "instance-1", + vfAddress: "0000:e3:00.4", + assignedAt: "2026-08-20T15:00:00Z", + }}} + c, reports := newTestSentinelController(t, store) + var logs bytes.Buffer + c.log = slog.New(slog.NewTextHandler(&logs, nil)) + ctx := context.Background() + + c.guestGPUInitStatus = func(context.Context, vgpuSentinelTarget) (guest.GPUInitState, string, error) { + return guest.GPUInitState_GPU_INIT_STATE_UNKNOWN, "", nil + } + c.pollOnce(ctx) + assert.Empty(t, reports.failures, "an undecided init must not be reported") + + c.guestGPUInitStatus = guestReportsFailed + c.pollOnce(ctx) + require.Len(t, reports.failures, 1) + assert.Equal(t, "0000:e3:00.4", reports.failures[0].VFAddress) + assert.Equal(t, "instance-1", reports.failures[0].InstanceID) + assert.Contains(t, logs.String(), "quarantined wedged vGPU VF") + assert.Contains(t, logs.String(), "RmInitAdapter failed!") +} + +func TestVGPUSentinelControllerSkipsUnreachableGuest(t *testing.T) { + t.Parallel() + + store := &fakeSentinelStore{targets: []vgpuSentinelTarget{{ + instanceID: "instance-1", + vfAddress: "0000:e3:00.4", + assignedAt: "2026-08-20T15:00:00Z", + }}} + c, reports := newTestSentinelController(t, store) + c.guestGPUInitStatus = func(context.Context, vgpuSentinelTarget) (guest.GPUInitState, string, error) { + return guest.GPUInitState_GPU_INIT_STATE_UNKNOWN, "", errors.New("vsock dial failed") + } + + c.pollOnce(context.Background()) + assert.Empty(t, reports.failures) + assert.Empty(t, reports.successes) +} + +func TestVGPUSentinelControllerRepairsHealthStoreOncePerPoll(t *testing.T) { + t.Parallel() + + targets := []vgpuSentinelTarget{{instanceID: "instance-1"}, {instanceID: "instance-2"}} + c, _ := newTestSentinelController(t, &fakeSentinelStore{targets: targets}) + c.guestGPUInitStatus = guestReportsOK + var repairs int + c.repairHealthStore = func() error { + repairs++ + return errors.New("persist failed") + } + + c.pollOnce(context.Background()) + assert.Equal(t, 1, repairs) +} + +func TestVGPUSentinelControllerThrottlesRepairWarning(t *testing.T) { + t.Parallel() + + c, _ := newTestSentinelController(t, &fakeSentinelStore{}) + var logs bytes.Buffer + c.log = slog.New(slog.NewTextHandler(&logs, nil)) + c.repairHealthStore = func() error { return errors.New("persist failed") } + + c.pollOnce(context.Background()) + c.pollOnce(context.Background()) + assert.Equal(t, 1, strings.Count(logs.String(), "failed to repair VF health state")) + + c.repairErrLoggedAt = time.Now().Add(-2 * time.Minute) + c.pollOnce(context.Background()) + assert.Equal(t, 2, strings.Count(logs.String(), "failed to repair VF health state")) +} + +func TestVGPUSentinelControllerRecordsCheckResults(t *testing.T) { + t.Parallel() + + store := &fakeSentinelStore{targets: []vgpuSentinelTarget{ + {instanceID: "ok"}, + {instanceID: "unknown"}, + {instanceID: "unreachable"}, + {instanceID: "unsupported"}, + }} + c, _ := newTestSentinelController(t, store) + reader := otelmetric.NewManualReader() + provider := otelmetric.NewMeterProvider(otelmetric.WithReader(reader)) + checks, err := provider.Meter("test").Int64Counter("hypeman_instances_vgpu_sentinel_checks_total") + require.NoError(t, err) + c.checks = checks + c.guestGPUInitStatus = func(_ context.Context, target vgpuSentinelTarget) (guest.GPUInitState, string, error) { + switch target.instanceID { + case "ok": + return guest.GPUInitState_GPU_INIT_STATE_OK, "", nil + case "unknown": + return guest.GPUInitState_GPU_INIT_STATE_UNKNOWN, "", nil + case "unsupported": + return guest.GPUInitState_GPU_INIT_STATE_UNKNOWN, "", fmt.Errorf("gpu init status RPC: %w", status.Error(codes.Unimplemented, "method not implemented")) + default: + return guest.GPUInitState_GPU_INIT_STATE_UNKNOWN, "", errors.New("vsock dial failed") + } + } + + c.pollOnce(context.Background()) + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(context.Background(), &rm)) + metric := findMetric(t, rm, "hypeman_instances_vgpu_sentinel_checks_total") + checksTotal, ok := metric.Data.(metricdata.Sum[int64]) + require.True(t, ok) + got := make(map[string]int64) + for _, point := range checksTotal.DataPoints { + got[metricLabel(t, point.Attributes, "result")] = point.Value + } + assert.Equal(t, map[string]int64{"ok": 1, "unknown": 1, "rpc_error": 1, "unsupported_agent": 1}, got) +} + +func TestVGPUSentinelControllerProcessesSuccessAfterFailure(t *testing.T) { + tests := []struct { + name string + failure devices.VFReportResult + success devices.VFSuccessResult + }{ + { + name: "recorded", + failure: devices.VFReportResult{Outcome: devices.VFReportRecorded, Failures: 1, Threshold: 2}, + success: devices.VFSuccessResult{Cleared: 1}, + }, + { + name: "quarantined", + failure: devices.VFReportResult{Outcome: devices.VFReportQuarantined, Failures: 2, Threshold: 2}, + success: devices.VFSuccessResult{Cleared: 2, Rescinded: true}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + store := &fakeSentinelStore{targets: []vgpuSentinelTarget{{ + instanceID: "instance-1", + vfAddress: "0000:e3:00.4", + assignedAt: "2026-08-20T15:00:00Z", + }}} + c, reports := newTestSentinelController(t, store) + c.reportFailure = func(report devices.VFInitFailureReport) (devices.VFReportResult, error) { + reports.recordFailure(report) + return tt.failure, nil + } + c.reportSuccess = func(report devices.VFInitSuccessReport) (devices.VFSuccessResult, error) { + reports.recordSuccess(report) + return tt.success, nil + } + + c.pollOnce(context.Background()) + require.Len(t, reports.failures, 1) + require.Empty(t, reports.successes) + + // A wedge that recovers (e.g. driver reload) flips the guest state. + c.guestGPUInitStatus = guestReportsOK + c.pollOnce(context.Background()) + require.Len(t, reports.failures, 1) + require.Len(t, reports.successes, 1) + assert.Equal(t, "instance-1", reports.successes[0].InstanceID) + }) + } +} + +func TestVGPUSentinelControllerRetriesFailedTallyClear(t *testing.T) { + t.Parallel() + + store := &fakeSentinelStore{targets: []vgpuSentinelTarget{{ + instanceID: "instance-1", + vfAddress: "0000:e3:00.4", + assignedAt: "2026-08-20T15:00:00Z", + }}} + c, reports := newTestSentinelController(t, store) + c.guestGPUInitStatus = guestReportsOK + c.reportSuccess = func(report devices.VFInitSuccessReport) (devices.VFSuccessResult, error) { + if reports.recordSuccess(report) == 1 { + return devices.VFSuccessResult{}, errors.New("persist failed") + } + return devices.VFSuccessResult{Cleared: 1}, nil + } + + c.pollOnce(context.Background()) + require.Len(t, reports.successes, 1) + + c.pollOnce(context.Background()) + require.Len(t, reports.successes, 2) + assert.Equal(t, "0000:e3:00.4", reports.successes[1].VFAddress) +} + +func TestVGPUSentinelControllerRetriesFailedQuarantine(t *testing.T) { + t.Parallel() + + store := &fakeSentinelStore{targets: []vgpuSentinelTarget{{ + instanceID: "instance-1", + vfAddress: "0000:e3:00.4", + assignedAt: "2026-08-20T15:00:00Z", + }}} + c, reports := newTestSentinelController(t, store) + realReport := c.reportFailure + c.reportFailure = func(devices.VFInitFailureReport) (devices.VFReportResult, error) { + return devices.VFReportResult{}, errors.New("persist failed") + } + + c.pollOnce(context.Background()) + assert.Empty(t, reports.failures) + + c.reportFailure = realReport + c.pollOnce(context.Background()) + assert.Len(t, reports.failures, 1) +} + +func TestVGPUSentinelControllerConfirmsAssignmentBeforeReporting(t *testing.T) { + t.Parallel() + + stale := vgpuSentinelTarget{ + instanceID: "instance-1", + vfAddress: "0000:e3:00.4", + assignedAt: "2026-08-21T00:00:00Z", + } + changed := []vgpuSentinelTarget{{ + instanceID: "instance-1", + vfAddress: "0000:e3:00.5", + assignedAt: "2026-08-21T00:00:10Z", + }} + tests := []struct { + name string + guestState func(context.Context, vgpuSentinelTarget) (guest.GPUInitState, string, error) + targets []vgpuSentinelTarget + wantFailures int + wantSuccesses int + }{ + {name: "failure skipped when the assignment changed", guestState: guestReportsFailed, targets: changed}, + // A released assignment (instance gone) remains attributable. + {name: "failure from a released assignment is reported", guestState: guestReportsFailed, wantFailures: 1}, + {name: "init OK skipped when the assignment changed", guestState: guestReportsOK, targets: changed}, + {name: "init OK from a released assignment clears", guestState: guestReportsOK, wantSuccesses: 1}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + c, reports := newTestSentinelController(t, &fakeSentinelStore{targets: tt.targets}) + c.guestGPUInitStatus = tt.guestState + + c.pollTarget(context.Background(), stale) + + require.Len(t, reports.failures, tt.wantFailures) + require.Len(t, reports.successes, tt.wantSuccesses) + for _, report := range reports.failures { + assert.Equal(t, stale.vfAddress, report.VFAddress) + assert.Equal(t, stale.assignedAt, report.AssignedAt) + } + for _, report := range reports.successes { + assert.Equal(t, stale.vfAddress, report.VFAddress) + assert.Equal(t, stale.assignedAt, report.AssignedAt) + } + }) + } +} + +func TestGetVGPUSentinelTargetMapsClaimAndRequiresControlSocket(t *testing.T) { + m := &manager{paths: paths.New(t.TempDir())} + const instanceID = "stopped-vgpu" + claimed := time.Now().UTC() + socketPath := testSentinelSocket(t, instanceID) + require.NoError(t, m.ensureDirectories(instanceID)) + require.NoError(t, m.saveMetadata(&metadata{StoredMetadata: StoredMetadata{ + Id: instanceID, + HypervisorType: "cloud-hypervisor", + VsockSocket: "/run/vsock.sock", + VsockCID: 42, + GPUFramework: devices.VGPUFrameworkVendorVFIO, + GPUDevicePath: "/sys/bus/pci/devices/0000:e3:00.4", + GPUClaimedAt: &claimed, + SocketPath: socketPath, + }})) + + target, ok, err := m.getVGPUSentinelTarget(context.Background(), instanceID) + require.NoError(t, err) + require.True(t, ok) + assert.Equal(t, vgpuSentinelTarget{ + instanceID: instanceID, + vfAddress: "0000:e3:00.4", + assignedAt: devices.FormatVFAssignedAt(claimed), + hypervisorType: "cloud-hypervisor", + vsockSocket: "/run/vsock.sock", + vsockCID: 42, + }, target) + + // A stopped or standby instance has no control socket; its lingering + // claim must not be polled. + require.NoError(t, os.Remove(socketPath)) + _, ok, err = m.getVGPUSentinelTarget(context.Background(), instanceID) + require.NoError(t, err) + assert.False(t, ok) +} + +// testSentinelSocket creates a stand-in control socket file for instanceID. +func testSentinelSocket(t *testing.T, instanceID string) string { + t.Helper() + path := filepath.Join(t.TempDir(), instanceID+".sock") + require.NoError(t, os.WriteFile(path, nil, 0o600)) + return path +} + +func TestListVGPUSentinelTargetsSkipsUnstattableMetadata(t *testing.T) { + m := &manager{paths: paths.New(t.TempDir())} + claimed := time.Now().UTC() + require.NoError(t, m.ensureDirectories("readable")) + require.NoError(t, m.saveMetadata(&metadata{StoredMetadata: StoredMetadata{ + Id: "readable", + GPUFramework: devices.VGPUFrameworkVendorVFIO, + GPUDevicePath: "/sys/bus/pci/devices/0000:e3:00.4", + GPUClaimedAt: &claimed, + SocketPath: testSentinelSocket(t, "readable"), + }})) + // A self-referential symlink makes stat fail with ELOOP rather than + // ENOENT, and does so for root too. + for _, id := range []string{"unreadable-a", "unreadable-b"} { + require.NoError(t, m.ensureDirectories(id)) + metaPath := m.paths.InstanceMetadata(id) + require.NoError(t, os.Symlink(filepath.Base(metaPath), metaPath)) + } + + files, err := m.listMetadataFilesStrict() + require.Error(t, err) + assert.ErrorContains(t, err, "unreadable-a") + assert.ErrorContains(t, err, "unreadable-b") + assert.Nil(t, files) + + 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 TestVGPUSentinelControllerRunExitsWhenHostIsNotVendorVFIO(t *testing.T) { + tests := []struct { + name string + framework devices.VGPUFramework + firstError bool + wantProbes int + wantScans int + wantLogText string + }{ + { + name: "missing framework", + wantProbes: 1, + wantLogText: "host has no vGPU framework", + }, + { + name: "discovery retry resolves to mdev", + framework: devices.VGPUFrameworkMdev, + firstError: true, + wantProbes: 2, + wantScans: 0, + wantLogText: "host vGPU framework is not vendor VFIO", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + store := &fakeSentinelStore{} + c, _ := newTestSentinelController(t, store) + 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 tt.firstError && probes == 1 { + return devices.VGPUFrameworkNone, nil, errors.New("transient sysfs error") + } + return tt.framework, 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") + } + assert.Equal(t, tt.wantProbes, probes) + assert.Equal(t, tt.wantScans, store.listCalls) + assert.NotContains(t, logs.String(), "vGPU sentinel controller started") + assert.Contains(t, logs.String(), tt.wantLogText) + }) + } +} 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/system/guest_agent/clock_linux.go b/lib/system/guest_agent/clock_linux.go index 1ddfdde54..ae559504b 100644 --- a/lib/system/guest_agent/clock_linux.go +++ b/lib/system/guest_agent/clock_linux.go @@ -1,7 +1,6 @@ package main import ( - "errors" "fmt" "io" "log" @@ -81,7 +80,7 @@ func syncClockFromPTP(ptp *os.File) error { } func watchVMForkKmsg(restored chan<- struct{}) { - f, err := os.Open("/dev/kmsg") + f, err := os.Open(kmsgPath) if err != nil { log.Printf("[guest-agent] clock keeper kmsg watch disabled: %v", err) return @@ -89,27 +88,16 @@ func watchVMForkKmsg(restored chan<- struct{}) { defer f.Close() if _, err := f.Seek(0, io.SeekEnd); err != nil { - log.Printf("[guest-agent] warning: failed to seek /dev/kmsg to end: %v", err) + log.Printf("[guest-agent] warning: failed to seek %s to end: %v", kmsgPath, err) } - // Each read returns one log record. EPIPE means the buffer wrapped and - // records were overwritten; the next read continues from the oldest - // available record. - buf := make([]byte, 8192) - for { - n, err := f.Read(buf) - if err != nil { - if errors.Is(err, unix.EPIPE) { - continue - } - log.Printf("[guest-agent] clock keeper kmsg watch stopped: %v", err) - return - } - if strings.Contains(string(buf[:n]), vmForkKmsgSignal) { + err = scanKmsg(f, func(record string) { + if strings.Contains(record, vmForkKmsgSignal) { select { case restored <- struct{}{}: default: } } - } + }) + log.Printf("[guest-agent] clock keeper kmsg watch stopped: %v", err) } diff --git a/lib/system/guest_agent/gpu_watch.go b/lib/system/guest_agent/gpu_watch.go new file mode 100644 index 000000000..260cac463 --- /dev/null +++ b/lib/system/guest_agent/gpu_watch.go @@ -0,0 +1,209 @@ +package main + +import ( + "bytes" + "context" + "fmt" + "io" + "log" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "sync" + "time" + + pb "github.com/kernel/hypeman/lib/guest" +) + +const ( + nvidiaPCIVendorID = "0x10de" + + gpuProbeInterval = 15 * time.Second + gpuProbeRetryWindow = 10 * time.Minute + // A slow attempt is killed after this delay. If it is stuck in + // uninterruptible I/O, the probe waits for it instead of starting another. + gpuProbeAttemptKillAfter = 30 * time.Second + + kmsgReopenDelay = 5 * time.Second + kmsgOpenRetryDelay = time.Minute +) + +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 +} + +type gpuInitReporter struct { + mu sync.Mutex + succeeded bool + failed bool + failureMessage string +} + +func (r *gpuInitReporter) reportFailure(msg string) { + r.mu.Lock() + defer r.mu.Unlock() + if r.succeeded { + return + } + if !r.failed { + log.Printf("[guest-agent] GPU init failure detected: %s", msg) + } + r.failed = true + r.failureMessage = msg +} + +func (r *gpuInitReporter) state() (pb.GPUInitState, string) { + r.mu.Lock() + defer r.mu.Unlock() + switch { + case r.succeeded: + return pb.GPUInitState_GPU_INIT_STATE_OK, "" + case r.failed: + return pb.GPUInitState_GPU_INIT_STATE_FAILED, r.failureMessage + default: + return pb.GPUInitState_GPU_INIT_STATE_UNKNOWN, "" + } +} + +func (r *gpuInitReporter) reportSuccess() { + r.mu.Lock() + defer r.mu.Unlock() + if r.succeeded { + return + } + r.succeeded = true + log.Printf("[guest-agent] GPU driver initialized") +} + +// GetGPUInitStatus reports the GPU driver init state to the host sentinel. +// The serial console is shared with workload output, so this vsock channel is +// the only signal the host trusts. +func (s *guestServer) GetGPUInitStatus(context.Context, *pb.GetGPUInitStatusRequest) (*pb.GetGPUInitStatusResponse, error) { + state := pb.GPUInitState_GPU_INIT_STATE_UNKNOWN + msg := "" + if s.gpuReporter != nil { + state, msg = s.gpuReporter.state() + } + return &pb.GetGPUInitStatusResponse{State: state, FailureMessage: msg}, nil +} + +func watchGPUInitFailure(reporter *gpuInitReporter) { + firstOpen := true + 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 + } + // A reopened fd restarts at the oldest record; skip history already + // scanned so a stale failure line is not reported again. + if !firstOpen { + if _, err := f.Seek(0, io.SeekEnd); err != nil { + log.Printf("[guest-agent] cannot seek %s to end (may re-report old records): %v", kmsgPath, err) + } + } + firstOpen = false + err = scanKmsg(f, gpuInitFailureHandler(reporter.reportFailure)) + _ = f.Close() + if err != nil { + log.Printf("[guest-agent] GPU init watch read %s failed (reopening): %v", kmsgPath, err) + } + time.Sleep(kmsgReopenDelay) + } +} + +// gpuInitFailureHandler adapts a kmsg record handler to report only NVRM init +// failure lines. +func gpuInitFailureHandler(report func(msg string)) func(record string) { + return func(record string) { + if msg, ok := gpuInitFailureMessage(record); ok { + report(msg) + } + } +} + +// probeGPUInit reports successful guest driver init. Opening the device via +// nvidia-smi runs RmInitAdapter, so on a wedged VF the probe itself produces +// the failure line in kmsg without waiting for the workload to touch the GPU. +func probeGPUInit(reporter *gpuInitReporter) { + nvidiaSMI, err := exec.LookPath("nvidia-smi") + if err != nil { + log.Printf("[guest-agent] GPU init probe disabled; driver init success cannot be reported: %v", err) + return + } + probeGPUInitUntil(reporter, gpuProbeRetryWindow, gpuProbeInterval, func() error { + return runGPUProbeAttempt(nvidiaSMI, gpuProbeAttemptKillAfter) + }) +} + +func probeGPUInitUntil(reporter *gpuInitReporter, window, interval time.Duration, attempt func() error) { + deadline := time.Now().Add(window) + for { + err := attempt() + if err == nil { + reporter.reportSuccess() + return + } + log.Printf("[guest-agent] GPU init probe attempt failed: %v", err) + if time.Now().After(deadline) { + log.Printf("[guest-agent] GPU init probe gave up after %s", window) + return + } + time.Sleep(interval) + } +} + +func runGPUProbeAttempt(nvidiaSMI string, killAfter time.Duration) error { + ctx, cancel := context.WithTimeout(context.Background(), killAfter) + defer cancel() + var output bytes.Buffer + cmd := exec.CommandContext(ctx, nvidiaSMI, "-L") + cmd.Stdout = &output + cmd.Stderr = &output + // Run kills the process when the deadline passes but still waits for it to + // be reaped, so at most one nvidia-smi attempt is ever outstanding; a probe + // stuck in an uninterruptible ioctl blocks here instead of accumulating + // processes, and the kmsg watcher still reports the underlying init failure. + err := cmd.Run() + if ctx.Err() != nil { + return fmt.Errorf("killed after %s: %w", killAfter, ctx.Err()) + } + if err != nil { + return fmt.Errorf("%w: %s", err, strings.TrimSpace(output.String())) + } + return nil +} + +// Only kernel-facility records match; userspace /dev/kmsg writes use LOG_USER. +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..46ddfd3d5 --- /dev/null +++ b/lib/system/guest_agent/gpu_watch_test.go @@ -0,0 +1,206 @@ +package main + +import ( + "bytes" + "context" + "io" + "log" + "os" + "path/filepath" + "strings" + "syscall" + "testing" + "time" + + pb "github.com/kernel/hypeman/lib/guest" + "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) + + _, ok = gpuInitFailureMessage("3,1042,8462102,-;NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x26:0xffff:1482)\n") + assert.True(t, ok) + + _, 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", + "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) + } +} + +func captureAgentLog(t *testing.T) *bytes.Buffer { + t.Helper() + var buf bytes.Buffer + prevOutput := log.Writer() + prevFlags := log.Flags() + log.SetOutput(&buf) + log.SetFlags(log.LstdFlags) + t.Cleanup(func() { + log.SetOutput(prevOutput) + log.SetFlags(prevFlags) + }) + return &buf +} + +func TestProbeGPUInitMarksOKOnceDriverResponds(t *testing.T) { + captureAgentLog(t) + + binDir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(binDir, "nvidia-smi"), []byte("#!/bin/sh\nexit 0\n"), 0o755)) + t.Setenv("PATH", binDir) + + reporter := &gpuInitReporter{} + probeGPUInit(reporter) + + state, _ := reporter.state() + assert.Equal(t, pb.GPUInitState_GPU_INIT_STATE_OK, state) +} + +func TestProbeGPUInitLogsWhenNvidiaSMIMissing(t *testing.T) { + logs := captureAgentLog(t) + t.Setenv("PATH", t.TempDir()) + + reporter := &gpuInitReporter{} + probeGPUInit(reporter) + + state, _ := reporter.state() + assert.Equal(t, pb.GPUInitState_GPU_INIT_STATE_UNKNOWN, state) + assert.Contains(t, logs.String(), "GPU init probe disabled") +} + +func TestProbeGPUInitRetriesAfterAttemptTimeout(t *testing.T) { + captureAgentLog(t) + attempts := 0 + + reporter := &gpuInitReporter{} + probeGPUInitUntil(reporter, time.Second, 0, func() error { + attempts++ + if attempts == 1 { + return context.DeadlineExceeded + } + return nil + }) + + assert.Equal(t, 2, attempts) + state, _ := reporter.state() + assert.Equal(t, pb.GPUInitState_GPU_INIT_STATE_OK, state) +} + +func TestGPUInitReporterState(t *testing.T) { + captureAgentLog(t) + + server := &guestServer{} + resp, err := server.GetGPUInitStatus(context.Background(), &pb.GetGPUInitStatusRequest{}) + require.NoError(t, err) + assert.Equal(t, pb.GPUInitState_GPU_INIT_STATE_UNKNOWN, resp.State, "a host without an NVIDIA device has no reporter") + + reporter := &gpuInitReporter{} + server = &guestServer{gpuReporter: reporter} + state, _ := reporter.state() + assert.Equal(t, pb.GPUInitState_GPU_INIT_STATE_UNKNOWN, state) + + reporter.reportFailure("NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884)") + reporter.reportFailure("NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x26:0xffff:1482)") + resp, err = server.GetGPUInitStatus(context.Background(), &pb.GetGPUInitStatusRequest{}) + require.NoError(t, err) + assert.Equal(t, pb.GPUInitState_GPU_INIT_STATE_FAILED, resp.State) + assert.Equal(t, "NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x26:0xffff:1482)", resp.FailureMessage, "the latest failure line wins") + + reporter.reportSuccess() + resp, err = server.GetGPUInitStatus(context.Background(), &pb.GetGPUInitStatusRequest{}) + require.NoError(t, err) + assert.Equal(t, pb.GPUInitState_GPU_INIT_STATE_OK, resp.State) + assert.Empty(t, resp.FailureMessage) + + reporter.reportFailure("NVRM: GPU 0000:00:03.0: RmInitAdapter failed! (0x22:0x65:884)") + state, msg := reporter.state() + assert.Equal(t, pb.GPUInitState_GPU_INIT_STATE_OK, state, "success is terminal; a later failure must not replace it") + assert.Empty(t, msg) +} + +func TestRunGPUProbeAttemptKillsAndReapsAfterDeadline(t *testing.T) { + path := filepath.Join(t.TempDir(), "nvidia-smi") + require.NoError(t, os.WriteFile(path, []byte("#!/bin/sh\nexec sleep 10\n"), 0o755)) + + start := time.Now() + err := runGPUProbeAttempt(path, 10*time.Millisecond) + require.ErrorIs(t, err, context.DeadlineExceeded) + assert.Less(t, time.Since(start), time.Second) +} + +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), gpuInitFailureHandler(func(msg string) { got = append(got, msg) }))) + assert.Len(t, got, 2) +} + +// kmsgOverrun marks a read that fails with EPIPE, as /dev/kmsg does when +// records are overwritten while being read. +const kmsgOverrun = "\x00overrun" + +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 rec == kmsgOverrun { + k.pos++ + return 0, syscall.EPIPE + } + if len(p) < len(rec) { + return 0, syscall.EINVAL + } + k.pos++ + return copy(p, rec), nil +} + +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}}, + gpuInitFailureHandler(func(msg string) { got = append(got, msg) }))) + assert.Len(t, got, 1) + + huge := "6,3,300,-;" + strings.Repeat("x", kmsgRecordBufferBytes) + "\n" + err := scanKmsg(&kmsgConn{records: []string{huge}}, func(string) {}) + assert.ErrorIs(t, err, syscall.EINVAL) +} + +func TestScanKmsgContinuesAfterOverrun(t *testing.T) { + 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{kmsgOverrun, failure}}, + gpuInitFailureHandler(func(msg string) { got = append(got, msg) }))) + assert.Len(t, got, 1, "records after an overrun must still be scanned on the same fd") +} diff --git a/lib/system/guest_agent/kmsg.go b/lib/system/guest_agent/kmsg.go new file mode 100644 index 000000000..7a3ccfae1 --- /dev/null +++ b/lib/system/guest_agent/kmsg.go @@ -0,0 +1,42 @@ +package main + +import ( + "bufio" + "errors" + "io" + "syscall" +) + +const ( + kmsgPath = "/dev/kmsg" + + // /dev/kmsg returns EINVAL without consuming records larger than this buffer. + kmsgRecordBufferBytes = 8192 +) + +// scanKmsg hands each /dev/kmsg record to handle until the read fails. EPIPE +// means records were overwritten while reading; the fd continues at the next +// available record. +// +// Buffering is safe only because every /dev/kmsg read returns one whole +// newline-terminated record: ReadString drains the buffer each time, so the +// next fill always offers the full kmsgRecordBufferBytes. A partial record +// would shrink that read and turn a normal-sized record into EINVAL. +func scanKmsg(r io.Reader, handle func(record string)) error { + reader := bufio.NewReaderSize(r, kmsgRecordBufferBytes) + for { + record, err := reader.ReadString('\n') + if record != "" { + handle(record) + } + if err != nil { + if errors.Is(err, io.EOF) { + return nil + } + if errors.Is(err, syscall.EPIPE) { + continue + } + return err + } + } +} diff --git a/lib/system/guest_agent/main.go b/lib/system/guest_agent/main.go index 84fd2a5da..cc8f5d049 100644 --- a/lib/system/guest_agent/main.go +++ b/lib/system/guest_agent/main.go @@ -22,6 +22,7 @@ const ( // guestServer implements the gRPC GuestService type guestServer struct { pb.UnimplementedGuestServiceServer + gpuReporter *gpuInitReporter } func main() { @@ -54,9 +55,16 @@ func main() { startClockKeeper() + var reporter *gpuInitReporter + if hasNVIDIADevice() { + reporter = &gpuInitReporter{} + go watchGPUInitFailure(reporter) + go probeGPUInit(reporter) + } + // Create gRPC server grpcServer := grpc.NewServer() - pb.RegisterGuestServiceServer(grpcServer, &guestServer{}) + pb.RegisterGuestServiceServer(grpcServer, &guestServer{gpuReporter: reporter}) // Serve gRPC over vsock if err := grpcServer.Serve(l); err != nil {