diff --git a/cmd/api/main.go b/cmd/api/main.go index d6fba1b43..5d78954d8 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -654,6 +654,12 @@ func run() error { return app.HealthCheckController.Run(gctx) }) } + if app.VGPUSentinelController != nil { + grp.Go(func() error { + logger.Info("starting vGPU sentinel controller") + return app.VGPUSentinelController.Run(gctx) + }) + } if restartController, ok := app.InstanceManager.(interface { StartRestartPolicyController(context.Context) error }); ok { diff --git a/cmd/api/wire.go b/cmd/api/wire.go index 133bf41f6..123d59bf5 100644 --- a/cmd/api/wire.go +++ b/cmd/api/wire.go @@ -29,26 +29,27 @@ import ( // application struct to hold initialized components type application struct { - Ctx context.Context - Logger *slog.Logger - Config *config.Config - ImageManager images.Manager - SystemManager system.Manager - NetworkManager network.Manager - DeviceManager devices.Manager - InstanceManager instances.Manager - VolumeManager volumes.Manager - BuilderManager builders.Manager - IngressManager ingress.Manager - BuildManager builds.Manager - PushManager imagepush.Manager - ResourceManager *resources.Manager - GuestMemoryController guestmemory.Controller - AutoStandbyController *autostandby.Controller - HealthCheckController *instances.HealthCheckController - VMMetricsManager *vm_metrics.Manager - Registry *registry.Registry - ApiService *api.ApiService + Ctx context.Context + Logger *slog.Logger + Config *config.Config + ImageManager images.Manager + SystemManager system.Manager + NetworkManager network.Manager + DeviceManager devices.Manager + InstanceManager instances.Manager + VolumeManager volumes.Manager + BuilderManager builders.Manager + IngressManager ingress.Manager + BuildManager builds.Manager + PushManager imagepush.Manager + ResourceManager *resources.Manager + GuestMemoryController guestmemory.Controller + AutoStandbyController *autostandby.Controller + HealthCheckController *instances.HealthCheckController + VGPUSentinelController *instances.VGPUSentinelController + VMMetricsManager *vm_metrics.Manager + Registry *registry.Registry + ApiService *api.ApiService } // initializeApp is the injector function @@ -72,6 +73,7 @@ func initializeApp() (*application, func(), error) { providers.ProvideGuestMemoryController, providers.ProvideAutoStandbyController, providers.ProvideHealthCheckController, + providers.ProvideVGPUSentinelController, providers.ProvideVMMetricsManager, providers.ProvideRegistry, api.New, diff --git a/cmd/api/wire_gen.go b/cmd/api/wire_gen.go index 9eb13c6ea..57034551a 100644 --- a/cmd/api/wire_gen.go +++ b/cmd/api/wire_gen.go @@ -82,6 +82,10 @@ func initializeApp() (*application, func(), error) { } autostandbyController := providers.ProvideAutoStandbyController(instancesManager, config, logger) healthCheckController := providers.ProvideHealthCheckController(instancesManager, logger) + vgpuSentinelController, err := providers.ProvideVGPUSentinelController(instancesManager, logger) + if err != nil { + return nil, nil, err + } vm_metricsManager, err := providers.ProvideVMMetricsManager(instancesManager, config, logger) if err != nil { return nil, nil, err @@ -92,26 +96,27 @@ func initializeApp() (*application, func(), error) { } apiService := api.New(config, manager, instancesManager, volumesManager, buildersManager, networkManager, devicesManager, ingressManager, buildsManager, imagepushManager, resourcesManager, controller, autostandbyController, vm_metricsManager) mainApplication := &application{ - Ctx: context, - Logger: logger, - Config: config, - ImageManager: manager, - SystemManager: systemManager, - NetworkManager: networkManager, - DeviceManager: devicesManager, - InstanceManager: instancesManager, - VolumeManager: volumesManager, - BuilderManager: buildersManager, - IngressManager: ingressManager, - BuildManager: buildsManager, - PushManager: imagepushManager, - ResourceManager: resourcesManager, - GuestMemoryController: controller, - AutoStandbyController: autostandbyController, - HealthCheckController: healthCheckController, - VMMetricsManager: vm_metricsManager, - Registry: registry, - ApiService: apiService, + Ctx: context, + Logger: logger, + Config: config, + ImageManager: manager, + SystemManager: systemManager, + NetworkManager: networkManager, + DeviceManager: devicesManager, + InstanceManager: instancesManager, + VolumeManager: volumesManager, + BuilderManager: buildersManager, + IngressManager: ingressManager, + BuildManager: buildsManager, + PushManager: imagepushManager, + ResourceManager: resourcesManager, + GuestMemoryController: controller, + AutoStandbyController: autostandbyController, + HealthCheckController: healthCheckController, + VGPUSentinelController: vgpuSentinelController, + VMMetricsManager: vm_metricsManager, + Registry: registry, + ApiService: apiService, } return mainApplication, func() { }, nil @@ -121,24 +126,25 @@ func initializeApp() (*application, func(), error) { // application struct to hold initialized components type application struct { - Ctx context.Context - Logger *slog.Logger - Config *config.Config - ImageManager images.Manager - SystemManager system.Manager - NetworkManager network.Manager - DeviceManager devices.Manager - InstanceManager instances.Manager - VolumeManager volumes.Manager - BuilderManager builders.Manager - IngressManager ingress.Manager - BuildManager builds.Manager - PushManager imagepush.Manager - ResourceManager *resources.Manager - GuestMemoryController guestmemory.Controller - AutoStandbyController *autostandby.Controller - HealthCheckController *instances.HealthCheckController - VMMetricsManager *vm_metrics.Manager - Registry *registry.Registry - ApiService *api.ApiService + Ctx context.Context + Logger *slog.Logger + Config *config.Config + ImageManager images.Manager + SystemManager system.Manager + NetworkManager network.Manager + DeviceManager devices.Manager + InstanceManager instances.Manager + VolumeManager volumes.Manager + BuilderManager builders.Manager + IngressManager ingress.Manager + BuildManager builds.Manager + PushManager imagepush.Manager + ResourceManager *resources.Manager + GuestMemoryController guestmemory.Controller + AutoStandbyController *autostandby.Controller + HealthCheckController *instances.HealthCheckController + VGPUSentinelController *instances.VGPUSentinelController + VMMetricsManager *vm_metrics.Manager + Registry *registry.Registry + ApiService *api.ApiService } diff --git a/lib/devices/GPU.md b/lib/devices/GPU.md index 678871562..1efa119ab 100644 --- a/lib/devices/GPU.md +++ b/lib/devices/GPU.md @@ -99,7 +99,7 @@ Instance Create → Assign profile to VF → Attach VF to VM → Instance Runnin Instance Stop/Delete → Release profile → VF available again ``` -Hypeman reconciles orphaned assignments with a periodic fail-closed pass: once at startup and every minute afterward (skipped entirely on hosts without GPUs). Each pass releases assignments whose owning instance is no longer live and clears their metadata, then sweeps device-level leftovers with no live metadata claim. Devices held open by a running VMM and assignments younger than five minutes are preserved, so a release that fails during stop or delete (typically because a GPU-busy VMM's kernel-side VFIO teardown outlives the force-kill wait) is simply retried on later passes until the device is free. +Hypeman reconciles orphaned assignments with a periodic fail-closed pass: once at startup and every minute afterward (skipped entirely on hosts without GPUs). Each pass releases assignments whose owning instance is no longer live and clears their metadata, then sweeps device-level leftovers with no live metadata claim. Devices held open by a running VMM and assignments younger than five minutes are preserved, so a release that fails during stop or delete (typically because a GPU-busy VMM's kernel-side VFIO teardown outlives the force-kill wait) is simply retried on later passes until the device is free. An ambiguous hypervisor ownership check also preserves the assignment, logs a warning, and increments `hypeman_instances_vgpu_reconcile_liveness_uncertain_total`. ### Hypervisor Support @@ -290,29 +290,68 @@ 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 -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 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 until a load -or write succeeds. 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. +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 retains their +assignment metadata — QEMU vsock dials by guest CID alone, and a stale CID +could since have been reused by an unrelated instance. 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. The guest agent also probes driver init at boot +with `nvidia-smi -L` (when present in the image): 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. An OK state +clears failures only when that exact assignment has a recorded failure, +removing the match and older tallies. If that assignment crossed the threshold, +its later success also rescinds the quarantine. A success with no exact match +clears nothing; other quarantines require manual recovery. If the state file +exists but cannot be loaded, or the last write to it failed, placement and +advertised availability fail closed until a load or write succeeds. The +sentinel makes one repair attempt before each poll; individual guest reports do +not retry the full-store write. 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. `used_slots` includes quarantined VFs still held by running instances, so it can overlap `quarantined_slots`; use `allocatable_slots` for admission. - -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, so a wedge whose instance stops before the next poll (5s) +is detected on the next boot that lands on the VF. The state 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 wedge-creating kill itself leaves no host-side log: no kernel error, no +XID, no plugin crash. Detection therefore happens on the next boot that lands +on the VF, whose guest driver starts failing ~27s after spawn. + +The trigger is a SIGKILL delivered to QEMU while the vGPU plugin is still initializing the VF (roughly the first seconds after process start): a single hard kill in that window wedges the VF near-deterministically, while QEMU processes that exit voluntarily — error exits, QMP quit, SIGTERM — @@ -360,7 +399,9 @@ systemctl start 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 3ae8d02c0..a339dc02f 100644 --- a/lib/devices/vf_health.go +++ b/lib/devices/vf_health.go @@ -270,6 +270,20 @@ func ReportVFInitSuccess(report VFInitSuccessReport) (VFSuccessResult, error) { return vfHealth.reportSuccess(report) } +// RepairVFHealthStore retries a failed load or persist. It serializes with +// vendor-VFIO placement and health mutations. +func RepairVFHealthStore() error { + 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 { @@ -309,8 +323,8 @@ 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 + if s.persistErr != nil { + return VFReportResult{}, fmt.Errorf("VF health state unavailable: last write failed: %w", s.persistErr) } previous, existed := s.records[report.VFAddress] @@ -367,9 +381,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 @@ -385,6 +396,9 @@ func (s *vfHealthStore) reportSuccess(report VFInitSuccessReport) (VFSuccessResu if match < 0 || (previous.QuarantinedAt != nil && match != len(previous.Failures)-1) { return VFSuccessResult{}, nil } + if s.persistErr != nil { + return VFSuccessResult{}, fmt.Errorf("VF health state unavailable: last write failed: %w", s.persistErr) + } remaining := append([]vfInitFailure(nil), previous.Failures[match+1:]...) result := VFSuccessResult{ diff --git a/lib/devices/vf_health_test.go b/lib/devices/vf_health_test.go index 00f14dbf7..8a7659b98 100644 --- a/lib/devices/vf_health_test.go +++ b/lib/devices/vf_health_test.go @@ -109,6 +109,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) @@ -306,6 +307,31 @@ func TestReportVFInitSuccessWithoutMatchingFailureClearsNothing(t *testing.T) { assert.False(t, result.Rescinded) } +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()) +} + func TestReportVFInitSuccessNeverClearsAnotherAssignmentsQuarantine(t *testing.T) { resetVFHealthStore(t) quarantineVF(t, "0000:e3:00.4") @@ -371,6 +397,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) @@ -404,6 +431,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) @@ -444,6 +472,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) @@ -478,6 +507,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 = VGPUAvailability(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/create.go b/lib/instances/create.go index cd50deb5c..56434615a 100644 --- a/lib/instances/create.go +++ b/lib/instances/create.go @@ -272,12 +272,6 @@ func (m *manager) createInstance( // whatever devices have been attached when cleanup runs. var attachedDeviceIDs []string var resolvedDeviceIDs []string - var gpuDevice *devices.VGPUDevice - var gpuProfile string - var gpuFramework devices.VGPUFramework - var gpuDevicePath string - var gpuMdevUUID string - var gpuAssignedAt *time.Time retention := vgpuRetention{instanceID: id} // Deferred before cu.Clean so rollback records retention before this wraps the error. @@ -298,53 +292,6 @@ func (m *manager) createInstance( }) } - // Handle vGPU profile request - if req.GPU != nil && req.GPU.Profile != "" { - retentionStub := func() StoredMetadata { - return StoredMetadata{ - Id: id, - Name: req.Name, - Image: req.Image, - ResolvedImage: resolvedImageRef, - Platform: imageInfo.Platform, - CreatedAt: m.nowUTC(), - HypervisorType: hvType, - HypervisorVersion: hvVersion, - SocketPath: m.paths.InstanceSocket(id, starter.SocketName()), - DataDir: m.paths.InstanceDir(id), - } - } - log.InfoContext(ctx, "creating vGPU", "instance_id", id, "profile", req.GPU.Profile) - gpuDevice, err = m.createVGPUDevice(ctx, req.GPU.Profile, id) - if err != nil { - retention.retainFromCreateError(retentionStub(), m.nowUTC(), err) - log.ErrorContext(ctx, "failed to create vGPU", "profile", req.GPU.Profile, "error", err) - return nil, wrapCreateVGPUErr(req.GPU.Profile, err) - } - gpuProfile = gpuDevice.ProfileName - gpuFramework = gpuDevice.Framework - gpuDevicePath = gpuDevice.SysfsPath - gpuMdevUUID = gpuDevice.MdevUUID - assignedAt := m.nowUTC() - gpuAssignedAt = &assignedAt - log.InfoContext(ctx, "created vGPU", "instance_id", id, "profile", gpuProfile, "uuid", gpuMdevUUID) - - // Add vGPU cleanup to stack - cu.Add(func() { - log.DebugContext(ctx, "destroying vGPU on cleanup", "instance_id", id, "uuid", gpuDevice.MdevUUID) - assignment := devices.VGPUAssignment{ - Framework: gpuDevice.Framework, - DevicePath: gpuDevice.SysfsPath, - MdevUUID: gpuDevice.MdevUUID, - InstanceID: id, - } - if err := m.destroyVGPUAssignment(ctx, assignment); err != nil { - log.WarnContext(ctx, "failed to destroy vGPU on cleanup", "instance_id", id, "uuid", gpuDevice.MdevUUID, "error", err) - retention.retainFromDevice(retentionStub(), gpuDevice, *gpuAssignedAt) - } - }) - } - if len(req.Devices) > 0 && m.deviceManager != nil { for _, deviceRef := range req.Devices { device, err := m.deviceManager.GetDevice(ctx, deviceRef) @@ -413,11 +360,6 @@ func (m *manager) createInstance( VsockCID: vsockCID, VsockSocket: vsockSocket, Devices: resolvedDeviceIDs, - GPUProfile: gpuProfile, - GPUFramework: gpuFramework, - GPUDevicePath: gpuDevicePath, - GPUMdevUUID: gpuMdevUUID, - GPUAssignedAt: gpuAssignedAt, Entrypoint: req.Entrypoint, Cmd: req.Cmd, SkipKernelHeaders: req.SkipKernelHeaders, @@ -560,6 +502,36 @@ func (m *manager) createInstance( stored.StartedAt = &bootStart stored.Phases.Record(phasetracking.PhaseCreated, bootStart) + // Allocate the vGPU after disk, network, volume, and config setup so the + // assignment can be persisted immediately. + if req.GPU != nil && req.GPU.Profile != "" { + log.InfoContext(ctx, "creating vGPU", "instance_id", id, "profile", req.GPU.Profile) + gpuDevice, err := m.createVGPUDevice(ctx, req.GPU.Profile, id) + if err != nil { + retention.retainFromCreateError(*stored, m.nowUTC(), err) + log.ErrorContext(ctx, "failed to create vGPU", "profile", req.GPU.Profile, "error", err) + return nil, wrapCreateVGPUErr(req.GPU.Profile, err) + } + assignedAt := m.nowUTC() + stored.GPUProfile = gpuDevice.ProfileName + setStoredVGPUDevice(stored, gpuDevice, assignedAt) + log.InfoContext(ctx, "created vGPU", "instance_id", id, "profile", stored.GPUProfile, "uuid", stored.GPUMdevUUID) + + cu.Add(func() { + log.DebugContext(ctx, "destroying vGPU on cleanup", "instance_id", id, "uuid", gpuDevice.MdevUUID) + assignment := devices.VGPUAssignment{ + Framework: gpuDevice.Framework, + DevicePath: gpuDevice.SysfsPath, + MdevUUID: gpuDevice.MdevUUID, + InstanceID: id, + } + if err := m.destroyVGPUAssignment(ctx, assignment); err != nil { + log.WarnContext(ctx, "failed to destroy vGPU on cleanup", "instance_id", id, "uuid", gpuDevice.MdevUUID, "error", err) + retention.retainFromDevice(*stored, gpuDevice, assignedAt) + } + }) + } + // 18. Save metadata log.DebugContext(ctx, "saving instance metadata", "instance_id", id) meta := &metadata{StoredMetadata: *stored} diff --git a/lib/instances/fork_test.go b/lib/instances/fork_test.go index c0bed670e..24d7903a8 100644 --- a/lib/instances/fork_test.go +++ b/lib/instances/fork_test.go @@ -83,7 +83,7 @@ func TestForkInstanceRejectsVGPURetentionRecord(t *testing.T) { _, err = manager.ForkInstance(ctx, sourceID, ForkInstanceRequest{Name: "fork-vgpu-retention-copy"}) require.ErrorIs(t, err, ErrInvalidState) - require.ErrorContains(t, err, "delete it to release the assignment") + require.ErrorContains(t, err, "delete-only record of a failed create") } func TestForkInstance_VZStoppedSourceSupported(t *testing.T) { diff --git a/lib/instances/lifecycle_noop_test.go b/lib/instances/lifecycle_noop_test.go index 77d15d733..8a0b4ca66 100644 --- a/lib/instances/lifecycle_noop_test.go +++ b/lib/instances/lifecycle_noop_test.go @@ -315,7 +315,7 @@ func TestStartRejectsVGPURetentionRecord(t *testing.T) { _, err = m.StartInstance(context.Background(), id, StartInstanceRequest{}) require.ErrorIs(t, err, ErrInvalidState) - require.ErrorContains(t, err, "delete it to release the assignment") + require.ErrorContains(t, err, "delete-only record of a failed create") stored, err := m.loadMetadata(id) require.NoError(t, err) diff --git a/lib/instances/metrics.go b/lib/instances/metrics.go index e14deb49b..e83458680 100644 --- a/lib/instances/metrics.go +++ b/lib/instances/metrics.go @@ -111,6 +111,7 @@ type Metrics struct { vgpuReconcileFailuresTotal metric.Int64Counter vgpuStaleReleaseFailuresTotal metric.Int64Counter vgpuAssignmentsRetainedTotal metric.Int64Counter + vgpuReconcileLivenessUncertainTotal metric.Int64Counter tracer trace.Tracer } @@ -311,6 +312,14 @@ func newInstanceMetrics(meter metric.Meter, tracer trace.Tracer, m *manager) (*M return nil, err } + vgpuReconcileLivenessUncertainTotal, err := meter.Int64Counter( + "hypeman_instances_vgpu_reconcile_liveness_uncertain_total", + metric.WithDescription("Total vGPU reconcile checks that preserved an assignment because hypervisor liveness was uncertain"), + ) + if err != nil { + return nil, err + } + // Register observable gauge for instance counts by state instancesTotal, err := meter.Int64ObservableGauge( "hypeman_instances_total", @@ -508,6 +517,7 @@ func newInstanceMetrics(meter metric.Meter, tracer trace.Tracer, m *manager) (*M vgpuReconcileFailuresTotal: vgpuReconcileFailuresTotal, vgpuStaleReleaseFailuresTotal: vgpuStaleReleaseFailuresTotal, vgpuAssignmentsRetainedTotal: vgpuAssignmentsRetainedTotal, + vgpuReconcileLivenessUncertainTotal: vgpuReconcileLivenessUncertainTotal, tracer: tracer, }, nil } @@ -633,6 +643,13 @@ func (m *manager) recordVGPURetainedAssignment(ctx context.Context, operation vg )) } +func (m *manager) recordVGPUReconcileLivenessUncertain(ctx context.Context) { + if m.metrics == nil { + return + } + m.metrics.vgpuReconcileLivenessUncertainTotal.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 525002cbb..32944e79b 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 TestVGPUReconcileLivenessUncertainMetric(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.recordVGPUReconcileLivenessUncertain(t.Context()) + + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(t.Context(), &rm)) + metricValue := findMetric(t, rm, "hypeman_instances_vgpu_reconcile_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 7e8500402..7bfee33c9 100644 --- a/lib/instances/process_identity.go +++ b/lib/instances/process_identity.go @@ -202,9 +202,9 @@ func classifyResolvedHypervisorOwner(socketPath string, stored, resolved int, er } // Ambiguous ownership is treated as live; this must not authorize teardown. -func hypervisorMayBeAlive(id HypervisorProcessIdentity, socketPath string) bool { +func hypervisorMayBeAlive(id HypervisorProcessIdentity, socketPath string) (bool, error) { pid, err := resolveLiveHypervisorPID(id, socketPath) - return err != nil || pid > 0 + return err != nil || pid > 0, err } // ProcessExists reports whether pid belongs to a live, non-zombie process. diff --git a/lib/instances/process_identity_linux_test.go b/lib/instances/process_identity_linux_test.go index 2ec201450..0ec559152 100644 --- a/lib/instances/process_identity_linux_test.go +++ b/lib/instances/process_identity_linux_test.go @@ -164,6 +164,10 @@ func TestResolveLiveHypervisorPIDFailsClosedWithoutSocketOrIdentity(t *testing.T resolved, err := resolveLiveHypervisorPID(HypervisorProcessIdentity{HypervisorPID: &pid}, "") require.ErrorContains(t, err, "without a socket path") assert.Zero(t, resolved) + + mayBeAlive, err := hypervisorMayBeAlive(HypervisorProcessIdentity{HypervisorPID: &pid}, "") + require.ErrorContains(t, err, "without a socket path") + assert.True(t, mayBeAlive, "ambiguous ownership must still fail closed") } func TestSocketListenerHelper(t *testing.T) { diff --git a/lib/instances/snapshot_test.go b/lib/instances/snapshot_test.go index 31f0a4343..e766515e7 100644 --- a/lib/instances/snapshot_test.go +++ b/lib/instances/snapshot_test.go @@ -72,7 +72,7 @@ func TestCreateSnapshotRejectsVGPURetentionRecord(t *testing.T) { Name: "snapshot-vgpu-retention", }) require.ErrorIs(t, err, ErrInvalidState) - require.ErrorContains(t, err, "delete it to release the assignment") + require.ErrorContains(t, err, "delete-only record of a failed create") } func TestRestoreSnapshotRejectsVGPURetentionRecord(t *testing.T) { @@ -101,7 +101,7 @@ func TestRestoreSnapshotRejectsVGPURetentionRecord(t *testing.T) { TargetHypervisor: mgr.defaultHypervisor, }) require.ErrorIs(t, err, ErrInvalidState) - require.ErrorContains(t, err, "delete it to release the assignment") + require.ErrorContains(t, err, "delete-only record of a failed create") stored, err := mgr.loadMetadata(sourceID) require.NoError(t, err) diff --git a/lib/instances/storage.go b/lib/instances/storage.go index 1a4d325b0..085a5b6d8 100644 --- a/lib/instances/storage.go +++ b/lib/instances/storage.go @@ -188,27 +188,33 @@ 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 returns readable metadata paths and joins any stat +// errors other than absence, so fail-closed callers treat an unreadable +// instance as an error instead of silently missing. func (m *manager) listMetadataFilesStrict() ([]string, error) { - return m.walkMetadataFiles(true) + files, statErr, err := m.walkMetadataFiles() + return files, errors.Join(statErr, err) } -func (m *manager) walkMetadataFiles(failOnStatError bool) ([]string, error) { +func (m *manager) walkMetadataFiles() (files []string, statErr error, err error) { guestsDir := m.paths.GuestsDir() // Ensure guests directory exists if err := os.MkdirAll(guestsDir, 0755); err != nil { - return nil, fmt.Errorf("create guests directory: %w", err) + return nil, nil, fmt.Errorf("create guests directory: %w", err) } entries, err := os.ReadDir(guestsDir) if err != nil { - return nil, fmt.Errorf("read guests directory: %w", err) + return nil, nil, fmt.Errorf("read guests directory: %w", err) } - var metaFiles []string + metaFiles := make([]string, 0, len(entries)) + var statErrs []error for _, entry := range entries { if !entry.IsDir() { continue @@ -217,10 +223,10 @@ func (m *manager) walkMetadataFiles(failOnStatError bool) ([]string, error) { metaPath := filepath.Join(guestsDir, entry.Name(), "metadata.json") if _, err := os.Stat(metaPath); err == nil { metaFiles = append(metaFiles, metaPath) - } else if failOnStatError && !os.IsNotExist(err) { - return nil, fmt.Errorf("stat metadata for instance %s: %w", entry.Name(), err) + } else if !os.IsNotExist(err) { + statErrs = append(statErrs, fmt.Errorf("stat metadata for instance %s: %w", entry.Name(), err)) } } - return metaFiles, nil + return metaFiles, errors.Join(statErrs...), nil } diff --git a/lib/instances/types.go b/lib/instances/types.go index ed32f5b5d..15b91bf87 100644 --- a/lib/instances/types.go +++ b/lib/instances/types.go @@ -156,7 +156,7 @@ type StoredMetadata struct { GPUDevicePath string GPUMdevUUID string // populated for mdev-backed vGPUs GPUAssignedAt *time.Time // set before hypervisor startup to bound crash recovery protection - GPURetainedForCleanup bool // delete-only stub holding a vGPU assignment + GPURetainedForCleanup bool // delete-only failed-create record; the assignment may already be released // Command overrides (like docker run ) Entrypoint []string // Override image entrypoint (nil = use image default) diff --git a/lib/instances/vgpu.go b/lib/instances/vgpu.go index 00e32ba01..bc7dd7b28 100644 --- a/lib/instances/vgpu.go +++ b/lib/instances/vgpu.go @@ -27,7 +27,7 @@ func (e *VGPUCleanupPendingError) Error() string { func (e *VGPUCleanupPendingError) Unwrap() error { return e.Err } -var errVGPURetentionStub = fmt.Errorf("%w: instance retains a vGPU assignment from a failed create and has no boot configuration; delete it to release the assignment", ErrInvalidState) +var errVGPURetentionStub = fmt.Errorf("%w: instance is a delete-only record of a failed create and has no boot configuration; delete it to remove the record and release any retained vGPU assignment", ErrInvalidState) func (m *manager) createVGPUDevice(ctx context.Context, profileName, instanceID string) (*devices.VGPUDevice, error) { create := m.createVGPU diff --git a/lib/instances/vgpu_reconcile.go b/lib/instances/vgpu_reconcile.go index 27ecb9e01..b043f7c7f 100644 --- a/lib/instances/vgpu_reconcile.go +++ b/lib/instances/vgpu_reconcile.go @@ -86,7 +86,11 @@ func (m *manager) reconcileVGPUAssignments(ctx context.Context) (map[string]stru if devicePath == "" { continue } - hypervisorLive := hypervisorMayBeAlive(stored.HypervisorProcessIdentity, stored.SocketPath) + // The socket-ownership check runs even without a persisted PID: a VMM + // whose post-boot metadata save failed still holds its control-socket + // listener, and releasing its device would tear the vGPU out from + // under a live VM. + hypervisorLive := m.vgpuHypervisorMayBeAlive(ctx, stored) if vgpuAssignmentMayBeLive(stored, m.nowUTC(), hypervisorLive) { protected[devicePath] = struct{}{} continue @@ -96,6 +100,21 @@ func (m *manager) reconcileVGPUAssignments(ctx context.Context) (map[string]stru return protected, nil } +// vgpuHypervisorMayBeAlive fails closed and records why reconciliation could +// not establish ownership, so retained capacity remains diagnosable. +func (m *manager) vgpuHypervisorMayBeAlive(ctx context.Context, stored *StoredMetadata) bool { + live, err := hypervisorMayBeAlive(stored.HypervisorProcessIdentity, stored.SocketPath) + if err == nil { + return live + } + logger.FromContext(ctx).WarnContext(ctx, "preserving vGPU assignment because hypervisor liveness is uncertain", + "instance_id", stored.Id, + "device_path", storedVGPUDevicePath(stored), + "error", err) + m.recordVGPUReconcileLivenessUncertain(ctx) + return true +} + // releaseStaleVGPUAssignment retries a release that previously failed, under // the instance lock. Liveness is re-verified after locking so a concurrent // start or restore keeps its assignment. A failed release only logs and keeps @@ -117,7 +136,7 @@ func (m *manager) releaseStaleVGPUAssignment(ctx context.Context, id string) { if path == "" { return } - hypervisorLive := hypervisorMayBeAlive(stored.HypervisorProcessIdentity, stored.SocketPath) + hypervisorLive := m.vgpuHypervisorMayBeAlive(ctx, stored) if vgpuAssignmentMayBeLive(stored, m.nowUTC(), hypervisorLive) { return } diff --git a/lib/instances/vgpu_reconcile_test.go b/lib/instances/vgpu_reconcile_test.go index 8b8d71b66..22851a98f 100644 --- a/lib/instances/vgpu_reconcile_test.go +++ b/lib/instances/vgpu_reconcile_test.go @@ -109,6 +109,41 @@ func TestReconcileVGPUsSkipsDeviceSweepWhenListingFails(t *testing.T) { assert.Equal(t, []bool{false, true}, sweeps, "the next pass retries the device sweep") } +func TestReconcileVGPUsReleasesStaleAssignment(t *testing.T) { + t.Parallel() + + var destroyed []devices.VGPUAssignment + m := &manager{ + paths: paths.New(t.TempDir()), + destroyVGPU: func(_ context.Context, assignment devices.VGPUAssignment) error { + destroyed = append(destroyed, assignment) + return nil + }, + reconcileVGPUDevices: func(context.Context, map[string]struct{}, bool) error { return nil }, + } + const id = "stopped-retained" + require.NoError(t, m.ensureDirectories(id)) + require.NoError(t, m.saveMetadata(&metadata{StoredMetadata: StoredMetadata{ + Id: id, + GPUProfile: "NVIDIA L40S-2Q", + GPUFramework: devices.VGPUFrameworkVendorVFIO, + GPUDevicePath: "/sys/bus/pci/devices/0000:82:00.4", + }})) + + m.ReconcileVGPUs(t.Context()) + + require.Len(t, destroyed, 1) + assert.Equal(t, devices.VGPUAssignment{ + Framework: devices.VGPUFrameworkVendorVFIO, + DevicePath: "/sys/bus/pci/devices/0000:82:00.4", + InstanceID: id, + }, destroyed[0]) + stored, err := m.loadMetadata(id) + require.NoError(t, err) + assert.Empty(t, stored.GPUDevicePath) + assert.Equal(t, "NVIDIA L40S-2Q", stored.GPUProfile, "profile is kept for the next start") +} + func TestReconcileVGPUsKeepsAssignmentWhenReleaseFails(t *testing.T) { t.Parallel() diff --git a/lib/instances/vgpu_sentinel.go b/lib/instances/vgpu_sentinel.go new file mode 100644 index 000000000..3ebb6b19d --- /dev/null +++ b/lib/instances/vgpu_sentinel.go @@ -0,0 +1,370 @@ +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/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 +} + +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, instanceID string) (guest.GPUInitState, string, error) + initFailures metric.Int64Counter + quarantines metric.Int64Counter + checks metric.Int64Counter + discoverFramework func() (devices.VGPUFramework, []devices.VirtualFunction, error) + probeErrLoggedAt 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 + } + _, err = meter.Int64ObservableGauge( + "hypeman_instances_vgpu_quarantined_vfs", + metric.WithDescription("Number of vGPU virtual functions currently quarantined"), + metric.WithInt64Callback(func(_ context.Context, o metric.Int64Observer) error { + o.Observe(int64(devices.TotalQuarantinedVFs())) + return nil + }), + ) + if err != nil { + return nil, err + } + _, 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"), + metric.WithInt64Callback(func(_ context.Context, o metric.Int64Observer) error { + if devices.VFHealthStoreUnavailable() { + o.Observe(1) + } else { + o.Observe(0) + } + return nil + }), + ) + if err != nil { + return nil, err + } + + return &VGPUSentinelController{ + store: store, + log: log.With("controller", "vgpu_sentinel"), + interval: vgpuSentinelPollInterval, + repairHealthStore: devices.RepairVFHealthStore, + reportFailure: devices.ReportVFInitFailure, + reportSuccess: devices.ReportVFInitSuccess, + guestGPUInitStatus: func(ctx context.Context, instanceID string) (guest.GPUInitState, string, error) { + dialer, err := manager.GetVsockDialer(ctx, instanceID) + 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 { + ticker := time.NewTicker(c.interval) + defer ticker.Stop() + vendorVFIO := false + for { + if !vendorVFIO { + var err error + vendorVFIO, err = c.probeVendorVFIO() + if err != nil && time.Since(c.probeErrLoggedAt) >= time.Minute { + c.log.Warn("vGPU sentinel framework discovery failed; retrying", "error", err) + c.probeErrLoggedAt = time.Now() + } + if err == nil && !vendorVFIO { + return nil + } + if vendorVFIO { + c.log.Info("vGPU sentinel controller started") + } + } + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + if vendorVFIO { + 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 { + c.log.WarnContext(ctx, "vGPU sentinel failed to repair VF health state", "error", err) + } + 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.instanceID) + 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. +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, statErr, err := m.walkMetadataFiles() + if err != nil { + return nil, err + } + if statErr != nil { + logger.FromContext(ctx).WarnContext(ctx, "vGPU sentinel cannot stat some instance metadata; their VFs are not scanned", "error", statErr) + } + targets := make([]vgpuSentinelTarget, 0, len(files)) + for _, file := range files { + id := filepath.Base(filepath.Dir(file)) + target, ok, err := m.getVGPUSentinelTarget(ctx, id) + if err != nil { + 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.GPURetainedForCleanup || 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 retains its + // assignment metadata indefinitely, 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.GPUAssignedAt != nil { + assignedAt = meta.GPUAssignedAt.UTC().Format(time.RFC3339Nano) + } + return vgpuSentinelTarget{ + instanceID: instanceID, + vfAddress: filepath.Base(meta.GPUDevicePath), + assignedAt: assignedAt, + }, true, nil +} diff --git a/lib/instances/vgpu_sentinel_test.go b/lib/instances/vgpu_sentinel_test.go new file mode 100644 index 000000000..6ab40fb97 --- /dev/null +++ b/lib/instances/vgpu_sentinel_test.go @@ -0,0 +1,492 @@ +package instances + +import ( + "bytes" + "context" + "errors" + "fmt" + "log/slog" + "os" + "path/filepath" + "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 +} + +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, string) (guest.GPUInitState, string, error) { + return guest.GPUInitState_GPU_INIT_STATE_FAILED, testNVRMMessage, nil +} + +func guestReportsOK(context.Context, string) (guest.GPUInitState, string, error) { + return guest.GPUInitState_GPU_INIT_STATE_OK, "", nil +} + +func newTestSentinelController(t *testing.T, store *fakeSentinelStore) (*VGPUSentinelController, *[]devices.VFInitFailureReport) { + t.Helper() + counter, err := noop.NewMeterProvider().Meter("test").Int64Counter("test") + require.NoError(t, err) + var reported []devices.VFInitFailureReport + 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) { + reported = append(reported, report) + return devices.VFReportResult{Outcome: devices.VFReportQuarantined, Failures: 1, Threshold: 1}, nil + }, + reportSuccess: func(devices.VFInitSuccessReport) (devices.VFSuccessResult, error) { + return devices.VFSuccessResult{}, nil + }, + guestGPUInitStatus: guestReportsFailed, + initFailures: counter, + quarantines: counter, + checks: counter, + } + return c, &reported +} + +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, reported := newTestSentinelController(t, store) + var logs bytes.Buffer + c.log = slog.New(slog.NewTextHandler(&logs, nil)) + ctx := context.Background() + + c.guestGPUInitStatus = func(context.Context, string) (guest.GPUInitState, string, error) { + return guest.GPUInitState_GPU_INIT_STATE_UNKNOWN, "", nil + } + c.pollOnce(ctx) + assert.Empty(t, *reported, "an undecided init must not be reported") + + c.guestGPUInitStatus = guestReportsFailed + c.pollOnce(ctx) + require.Len(t, *reported, 1) + assert.Equal(t, "0000:e3:00.4", (*reported)[0].VFAddress) + assert.Equal(t, "instance-1", (*reported)[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, reported := newTestSentinelController(t, store) + c.guestGPUInitStatus = func(context.Context, string) (guest.GPUInitState, string, error) { + return guest.GPUInitState_GPU_INIT_STATE_UNKNOWN, "", errors.New("vsock dial failed") + } + var successes []devices.VFInitSuccessReport + c.reportSuccess = func(report devices.VFInitSuccessReport) (devices.VFSuccessResult, error) { + successes = append(successes, report) + return devices.VFSuccessResult{}, nil + } + + c.pollOnce(context.Background()) + assert.Empty(t, *reported) + assert.Empty(t, 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 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, instanceID string) (guest.GPUInitState, string, error) { + switch 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, reported := newTestSentinelController(t, store) + c.reportFailure = func(report devices.VFInitFailureReport) (devices.VFReportResult, error) { + *reported = append(*reported, report) + return tt.failure, nil + } + var successes []devices.VFInitSuccessReport + c.reportSuccess = func(report devices.VFInitSuccessReport) (devices.VFSuccessResult, error) { + successes = append(successes, report) + return tt.success, nil + } + + c.pollOnce(context.Background()) + require.Len(t, *reported, 1) + require.Empty(t, successes) + + // A wedge that recovers (e.g. driver reload) flips the guest state. + c.guestGPUInitStatus = guestReportsOK + c.pollOnce(context.Background()) + require.Len(t, *reported, 1) + require.Len(t, successes, 1) + assert.Equal(t, "instance-1", 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, _ := newTestSentinelController(t, store) + c.guestGPUInitStatus = guestReportsOK + var reports []devices.VFInitSuccessReport + c.reportSuccess = func(report devices.VFInitSuccessReport) (devices.VFSuccessResult, error) { + reports = append(reports, report) + if len(reports) == 1 { + return devices.VFSuccessResult{}, errors.New("persist failed") + } + return devices.VFSuccessResult{Cleared: 1}, nil + } + + c.pollOnce(context.Background()) + require.Len(t, reports, 1) + + c.pollOnce(context.Background()) + require.Len(t, reports, 2) + assert.Equal(t, "0000:e3:00.4", reports[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, reported := 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, *reported) + + c.reportFailure = realReport + c.pollOnce(context.Background()) + assert.Len(t, *reported, 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, string) (guest.GPUInitState, string, error) + targets []vgpuSentinelTarget + wantApplied bool + }{ + {"failure skipped when the assignment changed", guestReportsFailed, changed, false}, + // A released assignment (instance gone) remains attributable. + {"failure from a released assignment is reported", guestReportsFailed, nil, true}, + {"init OK skipped when the assignment changed", guestReportsOK, changed, false}, + {"init OK from a released assignment clears", guestReportsOK, nil, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + c, reported := newTestSentinelController(t, &fakeSentinelStore{targets: tt.targets}) + c.guestGPUInitStatus = tt.guestState + var cleared []devices.VFInitSuccessReport + c.reportSuccess = func(report devices.VFInitSuccessReport) (devices.VFSuccessResult, error) { + cleared = append(cleared, report) + return devices.VFSuccessResult{Cleared: 1}, nil + } + + c.pollTarget(context.Background(), stale) + + if !tt.wantApplied { + assert.Empty(t, *reported) + assert.Empty(t, cleared) + return + } + if len(*reported) == 1 { + assert.Equal(t, stale.vfAddress, (*reported)[0].VFAddress) + assert.Equal(t, stale.assignedAt, (*reported)[0].AssignedAt) + assert.Empty(t, cleared) + return + } + require.Len(t, cleared, 1) + assert.Equal(t, stale.vfAddress, cleared[0].VFAddress) + assert.Equal(t, stale.assignedAt, cleared[0].AssignedAt) + }) + } +} + +func TestGetVGPUSentinelTargetSkipsRetentionStub(t *testing.T) { + m := &manager{paths: paths.New(t.TempDir())} + const instanceID = "retained-vgpu" + require.NoError(t, m.ensureDirectories(instanceID)) + require.NoError(t, m.saveMetadata(&metadata{StoredMetadata: StoredMetadata{ + Id: instanceID, + GPUFramework: devices.VGPUFrameworkVendorVFIO, + GPUDevicePath: "/sys/bus/pci/devices/0000:e3:00.4", + GPURetainedForCleanup: true, + SocketPath: testSentinelSocket(t, instanceID), + }})) + + _, ok, err := m.getVGPUSentinelTarget(context.Background(), instanceID) + require.NoError(t, err) + assert.False(t, ok) +} + +func TestGetVGPUSentinelTargetSkipsInstanceWithoutControlSocket(t *testing.T) { + m := &manager{paths: paths.New(t.TempDir())} + const instanceID = "stopped-vgpu" + assigned := time.Now().UTC() + socketPath := testSentinelSocket(t, instanceID) + require.NoError(t, m.ensureDirectories(instanceID)) + require.NoError(t, m.saveMetadata(&metadata{StoredMetadata: StoredMetadata{ + Id: instanceID, + GPUFramework: devices.VGPUFrameworkVendorVFIO, + GPUDevicePath: "/sys/bus/pci/devices/0000:e3:00.4", + GPUAssignedAt: &assigned, + SocketPath: socketPath, + }})) + + target, ok, err := m.getVGPUSentinelTarget(context.Background(), instanceID) + require.NoError(t, err) + require.True(t, ok) + assert.Equal(t, instanceID, target.instanceID) + + // A stopped or standby instance has no control socket; its retained + // assignment 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) { + if os.Geteuid() == 0 { + t.Skip("root bypasses directory permissions") + } + + m := &manager{paths: paths.New(t.TempDir())} + assigned := time.Now().UTC() + for _, id := range []string{"readable", "unreadable-a", "unreadable-b"} { + require.NoError(t, m.ensureDirectories(id)) + require.NoError(t, m.saveMetadata(&metadata{StoredMetadata: StoredMetadata{ + Id: id, + GPUFramework: devices.VGPUFrameworkVendorVFIO, + GPUDevicePath: "/sys/bus/pci/devices/0000:e3:00.4", + GPUAssignedAt: &assigned, + SocketPath: testSentinelSocket(t, id), + }})) + } + for _, id := range []string{"unreadable-a", "unreadable-b"} { + instanceDir := filepath.Dir(m.paths.InstanceMetadata(id)) + require.NoError(t, os.Chmod(instanceDir, 0o000)) + t.Cleanup(func() { _ = os.Chmod(instanceDir, 0o755) }) + } + + files, err := m.listMetadataFilesStrict() + require.Error(t, err) + assert.ErrorContains(t, err, "unreadable-a") + assert.ErrorContains(t, err, "unreadable-b") + require.Len(t, files, 1) + + var logs bytes.Buffer + ctx := logger.AddToContext(context.Background(), slog.New(slog.NewTextHandler(&logs, nil))) + targets, err := m.listVGPUSentinelTargets(ctx) + require.NoError(t, err) + require.Len(t, targets, 1) + assert.Equal(t, "readable", targets[0].instanceID) + assert.Contains(t, logs.String(), "vGPU sentinel cannot stat some instance metadata; their VFs are not scanned") + assert.Contains(t, logs.String(), "unreadable-a") + assert.Contains(t, logs.String(), "unreadable-b") +} + +func 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/gpu_watch.go b/lib/system/guest_agent/gpu_watch.go new file mode 100644 index 000000000..db0f53752 --- /dev/null +++ b/lib/system/guest_agent/gpu_watch.go @@ -0,0 +1,229 @@ +package main + +import ( + "bufio" + "context" + "errors" + "io" + "log" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "sync" + "syscall" + "time" + + pb "github.com/kernel/hypeman/lib/guest" +) + +const ( + kmsgPath = "/dev/kmsg" + 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 + + // /dev/kmsg returns EINVAL without consuming records larger than this buffer. + kmsgRecordBufferBytes = 8192 + + 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, reporter.reportFailure) + _ = f.Close() + if err != nil { + log.Printf("[guest-agent] GPU init watch read %s failed (reopening): %v", kmsgPath, err) + } + time.Sleep(kmsgReopenDelay) + } +} + +// 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 { + return + } + probeGPUInitUntil(reporter, time.Now().Add(gpuProbeRetryWindow), gpuProbeAttemptKillAfter, gpuProbeInterval, func() error { + return runGPUProbeAttempt(nvidiaSMI, gpuProbeAttemptKillAfter) + }) +} + +func probeGPUInitUntil(reporter *gpuInitReporter, deadline time.Time, attemptKillAfter, interval time.Duration, attempt func() error) { + for { + err := attempt() + if err == nil { + reporter.reportSuccess() + return + } + if errors.Is(err, context.DeadlineExceeded) { + log.Printf("[guest-agent] GPU init probe attempt exceeded %s and was killed", attemptKillAfter) + } + if time.Now().After(deadline) { + log.Printf("[guest-agent] GPU init probe gave up after %s", gpuProbeRetryWindow) + return + } + time.Sleep(interval) + } +} + +func runGPUProbeAttempt(nvidiaSMI string, killAfter time.Duration) error { + cmd := exec.Command(nvidiaSMI, "-L") + if err := cmd.Start(); err != nil { + return err + } + done := make(chan error, 1) + go func() { done <- cmd.Wait() }() + + timer := time.NewTimer(killAfter) + defer timer.Stop() + select { + case err := <-done: + return err + case <-timer.C: + _ = cmd.Process.Kill() + // Wait for the process 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. + <-done + return context.DeadlineExceeded + } +} + +func scanKmsg(r io.Reader, report func(msg string)) error { + reader := bufio.NewReaderSize(r, kmsgRecordBufferBytes) + for { + record, err := reader.ReadString('\n') + if msg, ok := gpuInitFailureMessage(record); ok { + report(msg) + } + if err != nil { + if errors.Is(err, io.EOF) { + return nil + } + // EPIPE means records were overwritten while reading; the fd + // continues at the next available record. + if errors.Is(err, syscall.EPIPE) { + continue + } + return err + } + } +} + +// 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..242838df0 --- /dev/null +++ b/lib/system/guest_agent/gpu_watch_test.go @@ -0,0 +1,194 @@ +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 TestProbeGPUInitRetriesAfterAttemptTimeout(t *testing.T) { + captureAgentLog(t) + attempts := 0 + + reporter := &gpuInitReporter{} + probeGPUInitUntil(reporter, time.Now().Add(time.Second), 10*time.Millisecond, 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), 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}}, + 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}}, + 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/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 {