From 6c8414540508d3ca77d1f81e00a5a6920bd14488 Mon Sep 17 00:00:00 2001 From: joshsiv-mulga Date: Mon, 11 May 2026 11:52:11 +1000 Subject: [PATCH] refactor(daemon): extract StartStoppedInstance to service (1b-pre phase 2c) Daemon's handleEC2StartStoppedInstance collapses to a Pattern A handleNATSRequest wrapper over InstanceServiceImpl.StartStoppedInstance. Promotes ResourceCapacityProvider to InstanceTypeAllocator with Allocate/Deallocate/CanAllocate/InstanceTypes, and adds a GPUClaimer interface returning (pciAddress, xvgaEnabled, err) so the instance package stays free of the gpu package. Daemon ResourceManager gains exported wrappers around its internal allocator methods; a daemonGPUClaimer adapter wires gpu.Manager + the GPU model overrides through SetGPUClaimer. No behaviour change. New service-level tests cover the moved logic; existing daemon integration tests continue to validate the end-to-end path through the wrapper. --- spinifex/daemon/daemon.go | 21 ++ spinifex/daemon/daemon_handlers_instance.go | 124 +--------- .../daemon/daemon_handlers_instance_test.go | 8 +- spinifex/daemon/gpu_adapter.go | 26 +++ spinifex/handlers/ec2/instance/service.go | 31 +++ .../handlers/ec2/instance/service_impl.go | 115 ++++++++- .../ec2/instance/service_impl_test.go | 218 +++++++++++++++++- .../handlers/ec2/instance/service_mock.go | 4 + .../handlers/ec2/instance/service_nats.go | 4 + 9 files changed, 419 insertions(+), 132 deletions(-) create mode 100644 spinifex/daemon/gpu_adapter.go diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 1fbdd9a36..d1a744228 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -1100,6 +1100,12 @@ func (d *Daemon) startCluster() error { // volumeService / vpcService / externalIPAM are constructed. d.instanceService.SetTerminationDeps(d.volumeService, d.vpcService, d.externalIPAM) + // Wire GPU claimer for InstanceService.StartStoppedInstance — nil when no + // passthrough hardware is configured. + if d.gpuManager != nil { + d.instanceService.SetGPUClaimer(&daemonGPUClaimer{d: d}) + } + d.accountService, err = initServiceWithRetry("account settings service", func() (*handlers_ec2_account.AccountSettingsServiceImpl, error) { return handlers_ec2_account.NewAccountSettingsServiceImplWithNATS(d.config, d.natsConn) }) @@ -1848,6 +1854,21 @@ func (rm *ResourceManager) deallocate(instanceType *ec2.InstanceTypeInfo) { rm.updateInstanceSubscriptions() } +// Allocate, Deallocate, CanAllocate, InstanceTypes are the exported facade +// satisfying handlers_ec2_instance.InstanceTypeAllocator. The unexported +// variants stay for internal daemon callers. +func (rm *ResourceManager) Allocate(it *ec2.InstanceTypeInfo) error { return rm.allocate(it) } +func (rm *ResourceManager) Deallocate(it *ec2.InstanceTypeInfo) { rm.deallocate(it) } +func (rm *ResourceManager) CanAllocate(it *ec2.InstanceTypeInfo, count int) int { + return rm.canAllocate(it, count) +} + +// InstanceTypes returns the shared instance-type map. Callers must not mutate +// the returned map; reloadGPUTypes mutates it in place under rm.mu. +func (rm *ResourceManager) InstanceTypes() map[string]*ec2.InstanceTypeInfo { + return rm.instanceTypes +} + // initSubscriptions sets up dynamic per-instance-type NATS subscriptions. // Called once during daemon startup after NATS is connected. // reloadGPUTypes replaces GPU instance types in-place and updates NATS subscriptions. diff --git a/spinifex/daemon/daemon_handlers_instance.go b/spinifex/daemon/daemon_handlers_instance.go index 37b6e9d88..209babdfe 100644 --- a/spinifex/daemon/daemon_handlers_instance.go +++ b/spinifex/daemon/daemon_handlers_instance.go @@ -777,130 +777,8 @@ func (d *Daemon) handleEC2DescribeInstanceTypes(msg *nats.Msg) { } // startStoppedInstanceRequest is the payload for ec2.start topic -type startStoppedInstanceRequest struct { - InstanceID string `json:"instance_id"` -} - -// handleEC2StartStoppedInstance picks up a stopped instance from shared KV, -// re-launches it on this daemon node, and removes it from shared KV. func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) { - var req startStoppedInstanceRequest - if err := json.Unmarshal(msg.Data, &req); err != nil { - slog.Error("handleEC2StartStoppedInstance: failed to unmarshal request", "err", err) - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - - if req.InstanceID == "" { - slog.Error("handleEC2StartStoppedInstance: missing instance_id") - respondWithError(msg, awserrors.ErrorMissingParameter) - return - } - - if d.stateStore == nil { - slog.Error("handleEC2StartStoppedInstance: state store not available") - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - - // Load instance from shared KV - instance, err := d.stateStore.LoadStoppedInstance(req.InstanceID) - if err != nil { - slog.Error("handleEC2StartStoppedInstance: failed to load stopped instance", "instanceId", req.InstanceID, "err", err) - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - if instance == nil { - slog.Warn("handleEC2StartStoppedInstance: instance not found in shared KV", "instanceId", req.InstanceID) - respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) - return - } - - if instance.Status != vm.StateStopped { - slog.Error("handleEC2StartStoppedInstance: instance not in stopped state", "instanceId", req.InstanceID, "status", instance.Status) - respondWithError(msg, awserrors.ErrorIncorrectInstanceState) - return - } - - // Verify the caller owns this instance - if !checkInstanceOwnership(msg, req.InstanceID, instance.AccountID) { - return - } - - // Reset node-local fields that are stale after cross-node migration - instance.ResetNodeLocalState() - - // Allocate resources - instanceType, ok := d.resourceMgr.instanceTypes[instance.InstanceType] - if !ok { - slog.Error("handleEC2StartStoppedInstance: instance type not available on this node", - "instanceId", req.InstanceID, "instanceType", instance.InstanceType) - respondWithError(msg, awserrors.ErrorInsufficientInstanceCapacity) - return - } - if err := d.resourceMgr.allocate(instanceType); err != nil { - slog.Error("handleEC2StartStoppedInstance: failed to allocate resources", "instanceId", req.InstanceID, "err", err) - respondWithError(msg, awserrors.ErrorInsufficientInstanceCapacity) - return - } - - // Add instance to local map and clear stop attribute before launch - instance.Attributes = types.EC2CommandAttributes{StartInstance: true} - d.vmMgr.Insert(instance) - - // Claim GPU for GPU instance types — binds the full IOMMU group to vfio-pci. - if d.gpuManager != nil && instancetypes.IsGPUType(instanceType) { - gpuDevice, gpuErr := d.gpuManager.Claim(instance.ID) - if gpuErr != nil { - slog.Error("handleEC2StartStoppedInstance: GPU claim failed", "instanceId", req.InstanceID, "err", gpuErr) - d.resourceMgr.deallocate(instanceType) - d.vmMgr.Delete(instance.ID) - respondWithError(msg, awserrors.ErrorInsufficientInstanceCapacity) - return - } - instance.GPUPCIAddress = gpuDevice.PCIAddress - instance.GPUXVGAEnabled = gpuXVGAEnabled(gpuDevice, d.config.Daemon.GPUModelOverrides) - slog.Info("GPU claimed for instance", "instanceId", req.InstanceID, "gpu", gpuDevice.PCIAddress, "xvga", instance.GPUXVGAEnabled) - } - - // Launch the instance infrastructure (QEMU, QMP, NATS subscriptions) - err = d.vmMgr.Run(instance) - if err != nil { - slog.Error("handleEC2StartStoppedInstance: vmMgr.Run failed", "instanceId", req.InstanceID, "err", err) - // Rollback: release GPU if claimed, deallocate resources, remove from local map - if instance.GPUPCIAddress != "" && d.gpuManager != nil { - if releaseErr := d.gpuManager.Release(instance.ID); releaseErr != nil { - slog.Error("handleEC2StartStoppedInstance: GPU release failed after launch failure", - "instanceId", req.InstanceID, "err", releaseErr) - } - } - if ok { - d.resourceMgr.deallocate(instanceType) - } - d.vmMgr.Delete(instance.ID) - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - - // Discover actual guest device names via QMP query-block - d.vmMgr.UpdateGuestDeviceNames(instance) - - // Remove from shared KV now that it's running locally. - // Retry once on failure — a stale KV entry risks duplicate starts. - if err := d.stateStore.DeleteStoppedInstance(req.InstanceID); err != nil { - slog.Warn("handleEC2StartStoppedInstance: first KV delete failed, retrying", - "instanceId", req.InstanceID, "err", err) - if retryErr := d.stateStore.DeleteStoppedInstance(req.InstanceID); retryErr != nil { - slog.Error("handleEC2StartStoppedInstance: KV delete failed after retry, instance is running locally but stale entry remains in shared KV", - "instanceId", req.InstanceID, "err", retryErr) - } - } - - slog.Info("Started stopped instance from shared KV", "instanceId", instance.ID, "node", d.node) - - if err := msg.Respond(fmt.Appendf(nil, `{"status":"running","instanceId":"%s"}`, instance.ID)); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } + handleNATSRequest(msg, d.instanceService.StartStoppedInstance) } func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) { diff --git a/spinifex/daemon/daemon_handlers_instance_test.go b/spinifex/daemon/daemon_handlers_instance_test.go index 1fe6d96f1..515ffe617 100644 --- a/spinifex/daemon/daemon_handlers_instance_test.go +++ b/spinifex/daemon/daemon_handlers_instance_test.go @@ -207,7 +207,7 @@ func TestHandleEC2StartStoppedInstance_LoadError(t *testing.T) { store.loadStoppedErr = errors.New("kv unavailable") d := daemonWithFakeStateStore(t, store) - body, _ := json.Marshal(startStoppedInstanceRequest{InstanceID: "i-load-fail"}) + body, _ := json.Marshal(handlers_ec2_instance.StartStoppedInstanceInput{InstanceID: "i-load-fail"}) reply := requestHandler(t, d.natsConn, "ec2.start.test1", d.handleEC2StartStoppedInstance, testAccountID, body) assert.Equal(t, awserrors.ErrorServerInternal, decodeError(t, reply.Data)["Code"]) } @@ -216,7 +216,7 @@ func TestHandleEC2StartStoppedInstance_StateStoreNil(t *testing.T) { d := createTestDaemon(t, sharedNATSURL) // d.stateStore intentionally left nil. - body, _ := json.Marshal(startStoppedInstanceRequest{InstanceID: "i-no-store"}) + body, _ := json.Marshal(handlers_ec2_instance.StartStoppedInstanceInput{InstanceID: "i-no-store"}) reply := requestHandler(t, d.natsConn, "ec2.start.test2", d.handleEC2StartStoppedInstance, testAccountID, body) assert.Equal(t, awserrors.ErrorServerInternal, decodeError(t, reply.Data)["Code"]) } @@ -226,7 +226,7 @@ func TestHandleEC2StartStoppedInstance_CrossTenantRejected(t *testing.T) { store.stopped["i-foreign"] = stoppedVMFixture("i-foreign", "999988887777") d := daemonWithFakeStateStore(t, store) - body, _ := json.Marshal(startStoppedInstanceRequest{InstanceID: "i-foreign"}) + body, _ := json.Marshal(handlers_ec2_instance.StartStoppedInstanceInput{InstanceID: "i-foreign"}) reply := requestHandler(t, d.natsConn, "ec2.start.test3", d.handleEC2StartStoppedInstance, testAccountID, body) assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, decodeError(t, reply.Data)["Code"]) @@ -245,7 +245,7 @@ func TestHandleEC2StartStoppedInstance_InstanceTypeUnknown(t *testing.T) { store.stopped[v.ID] = v d := daemonWithFakeStateStore(t, store) - body, _ := json.Marshal(startStoppedInstanceRequest{InstanceID: v.ID}) + body, _ := json.Marshal(handlers_ec2_instance.StartStoppedInstanceInput{InstanceID: v.ID}) reply := requestHandler(t, d.natsConn, "ec2.start.test4", d.handleEC2StartStoppedInstance, testAccountID, body) assert.Equal(t, awserrors.ErrorInsufficientInstanceCapacity, decodeError(t, reply.Data)["Code"]) } diff --git a/spinifex/daemon/gpu_adapter.go b/spinifex/daemon/gpu_adapter.go new file mode 100644 index 000000000..6f04dd2c4 --- /dev/null +++ b/spinifex/daemon/gpu_adapter.go @@ -0,0 +1,26 @@ +package daemon + +import ( + handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance" +) + +// daemonGPUClaimer adapts the daemon's gpu.Manager + GPUModelOverrides config to +// the transport-neutral handlers_ec2_instance.GPUClaimer surface. Keeps the gpu +// package out of the instance service. +type daemonGPUClaimer struct { + d *Daemon +} + +var _ handlers_ec2_instance.GPUClaimer = (*daemonGPUClaimer)(nil) + +func (g *daemonGPUClaimer) Claim(instanceID string) (string, bool, error) { + dev, err := g.d.gpuManager.Claim(instanceID) + if err != nil { + return "", false, err + } + return dev.PCIAddress, gpuXVGAEnabled(dev, g.d.config.Daemon.GPUModelOverrides), nil +} + +func (g *daemonGPUClaimer) Release(instanceID string) error { + return g.d.gpuManager.Release(instanceID) +} diff --git a/spinifex/handlers/ec2/instance/service.go b/spinifex/handlers/ec2/instance/service.go index 52b824d6b..51ab74ee4 100644 --- a/spinifex/handlers/ec2/instance/service.go +++ b/spinifex/handlers/ec2/instance/service.go @@ -14,9 +14,21 @@ type InstanceService interface { DescribeStoppedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) DescribeTerminatedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) ModifyInstanceAttribute(input *ec2.ModifyInstanceAttributeInput, accountID string) (*ec2.ModifyInstanceAttributeOutput, error) + StartStoppedInstance(input *StartStoppedInstanceInput, accountID string) (*StartStoppedInstanceOutput, error) TerminateStoppedInstance(input *TerminateStoppedInstanceInput, accountID string) (*TerminateStoppedInstanceOutput, error) } +// StartStoppedInstanceInput is the payload for ec2.StartStoppedInstance. +type StartStoppedInstanceInput struct { + InstanceID string `json:"instance_id"` +} + +// StartStoppedInstanceOutput is the response payload. +type StartStoppedInstanceOutput struct { + Status string `json:"status"` + InstanceID string `json:"instanceId"` +} + // TerminateStoppedInstanceInput is the payload for ec2.TerminateStoppedInstance. type TerminateStoppedInstanceInput struct { InstanceID string `json:"instance_id"` @@ -34,6 +46,25 @@ type ResourceCapacityProvider interface { GetAvailableInstanceTypeInfos(showCapacity bool) []*ec2.InstanceTypeInfo } +// InstanceTypeAllocator extends ResourceCapacityProvider with the mutating +// resource-reservation methods used by StartStoppedInstance. Implemented by +// daemon.ResourceManager. +type InstanceTypeAllocator interface { + ResourceCapacityProvider + Allocate(instanceType *ec2.InstanceTypeInfo) error + Deallocate(instanceType *ec2.InstanceTypeInfo) + CanAllocate(instanceType *ec2.InstanceTypeInfo, count int) int + InstanceTypes() map[string]*ec2.InstanceTypeInfo +} + +// GPUClaimer binds a GPU (and its IOMMU group) to vfio-pci for a starting +// instance. Returns the PCI address and whether QEMU should set x-vga=on. +// nil claimer means the daemon has no GPU passthrough configured. +type GPUClaimer interface { + Claim(instanceID string) (pciAddress string, xvgaEnabled bool, err error) + Release(instanceID string) error +} + // StoppedInstanceStore covers KV-backed read/write access for stopped instances // and read+write access for terminated instances. Implemented by // daemon.JetStreamManager. diff --git a/spinifex/handlers/ec2/instance/service_impl.go b/spinifex/handlers/ec2/instance/service_impl.go index e74ce2539..e67c4ad0e 100644 --- a/spinifex/handlers/ec2/instance/service_impl.go +++ b/spinifex/handlers/ec2/instance/service_impl.go @@ -21,6 +21,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/filterutil" + "github.com/mulgadc/spinifex/spinifex/instancetypes" "github.com/mulgadc/spinifex/spinifex/objectstore" spxtypes "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" @@ -235,11 +236,12 @@ type InstanceServiceImpl struct { natsConn *nats.Conn objectStore objectstore.ObjectStore vmMgr *vm.Manager - resourceMgr ResourceCapacityProvider + resourceMgr InstanceTypeAllocator stoppedStore StoppedInstanceStore volumeDeleter VolumeDeleter eniDeleter ENIDeleter ipReleaser PublicIPReleaser + gpuClaimer GPUClaimer } // NewInstanceServiceImpl creates a new instance service implementation for daemon use @@ -249,7 +251,7 @@ func NewInstanceServiceImpl( nc *nats.Conn, store objectstore.ObjectStore, vmMgr *vm.Manager, - resourceMgr ResourceCapacityProvider, + resourceMgr InstanceTypeAllocator, stoppedStore StoppedInstanceStore, ) *InstanceServiceImpl { return &InstanceServiceImpl{ @@ -272,6 +274,12 @@ func (s *InstanceServiceImpl) SetTerminationDeps(vd VolumeDeleter, ed ENIDeleter s.ipReleaser = pr } +// SetGPUClaimer wires the GPU passthrough claim/release dependency used by +// StartStoppedInstance. nil disables GPU passthrough for the service. +func (s *InstanceServiceImpl) SetGPUClaimer(g GPUClaimer) { + s.gpuClaimer = g +} + // RunInstance creates a single EC2 instance (called per-instance by daemon) // Returns the VM struct and EC2 instance metadata func (s *InstanceServiceImpl) RunInstance(input *ec2.RunInstancesInput) (*vm.VM, *ec2.Instance, error) { @@ -1250,6 +1258,109 @@ func (s *InstanceServiceImpl) ModifyInstanceAttribute(input *ec2.ModifyInstanceA return &ec2.ModifyInstanceAttributeOutput{}, nil } +// StartStoppedInstance picks up a stopped instance from shared KV, re-launches +// it on this daemon node, then removes it from shared KV. +func (s *InstanceServiceImpl) StartStoppedInstance(input *StartStoppedInstanceInput, accountID string) (*StartStoppedInstanceOutput, error) { + if input.InstanceID == "" { + return nil, errors.New(awserrors.ErrorMissingParameter) + } + if s.stoppedStore == nil { + slog.Error("StartStoppedInstance: stopped store not available") + return nil, errors.New(awserrors.ErrorServerInternal) + } + if s.resourceMgr == nil { + slog.Error("StartStoppedInstance: resource manager not available") + return nil, errors.New(awserrors.ErrorServerInternal) + } + if s.vmMgr == nil { + slog.Error("StartStoppedInstance: vm manager not available") + return nil, errors.New(awserrors.ErrorServerInternal) + } + + instance, err := s.stoppedStore.LoadStoppedInstance(input.InstanceID) + if err != nil { + slog.Error("StartStoppedInstance: failed to load stopped instance", "instanceId", input.InstanceID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + if instance == nil { + slog.Warn("StartStoppedInstance: instance not found in shared KV", "instanceId", input.InstanceID) + return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound) + } + if instance.Status != vm.StateStopped { + slog.Error("StartStoppedInstance: instance not in stopped state", "instanceId", input.InstanceID, "status", instance.Status) + return nil, errors.New(awserrors.ErrorIncorrectInstanceState) + } + if !IsInstanceVisible(accountID, instance.AccountID) { + slog.Warn("StartStoppedInstance: instance not visible", + "instanceId", input.InstanceID, "callerAccount", accountID, "ownerAccount", instance.AccountID) + return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound) + } + + // Reset node-local fields that are stale after cross-node migration. + instance.ResetNodeLocalState() + + instanceType, ok := s.resourceMgr.InstanceTypes()[instance.InstanceType] + if !ok { + slog.Error("StartStoppedInstance: instance type not available on this node", + "instanceId", input.InstanceID, "instanceType", instance.InstanceType) + return nil, errors.New(awserrors.ErrorInsufficientInstanceCapacity) + } + if err := s.resourceMgr.Allocate(instanceType); err != nil { + slog.Error("StartStoppedInstance: failed to allocate resources", "instanceId", input.InstanceID, "err", err) + return nil, errors.New(awserrors.ErrorInsufficientInstanceCapacity) + } + + // Add to local map + clear stop attribute before launch. + instance.Attributes = spxtypes.EC2CommandAttributes{StartInstance: true} + s.vmMgr.Insert(instance) + + // Claim GPU for GPU instance types — binds the full IOMMU group to vfio-pci. + gpuClaimed := false + if s.gpuClaimer != nil && instancetypes.IsGPUType(instanceType) { + pciAddr, xvga, gpuErr := s.gpuClaimer.Claim(instance.ID) + if gpuErr != nil { + slog.Error("StartStoppedInstance: GPU claim failed", "instanceId", input.InstanceID, "err", gpuErr) + s.resourceMgr.Deallocate(instanceType) + s.vmMgr.Delete(instance.ID) + return nil, errors.New(awserrors.ErrorInsufficientInstanceCapacity) + } + instance.GPUPCIAddress = pciAddr + instance.GPUXVGAEnabled = xvga + gpuClaimed = true + slog.Info("GPU claimed for instance", "instanceId", input.InstanceID, "gpu", pciAddr, "xvga", xvga) + } + + if err := s.vmMgr.Run(instance); err != nil { + slog.Error("StartStoppedInstance: vmMgr.Run failed", "instanceId", input.InstanceID, "err", err) + if gpuClaimed { + if relErr := s.gpuClaimer.Release(instance.ID); relErr != nil { + slog.Error("StartStoppedInstance: GPU release failed after launch failure", + "instanceId", input.InstanceID, "err", relErr) + } + } + s.resourceMgr.Deallocate(instanceType) + s.vmMgr.Delete(instance.ID) + return nil, errors.New(awserrors.ErrorServerInternal) + } + + // Discover actual guest device names via QMP query-block. + s.vmMgr.UpdateGuestDeviceNames(instance) + + // Remove from shared KV now that it's running locally. Retry once on failure — + // a stale KV entry risks duplicate starts. + if err := s.stoppedStore.DeleteStoppedInstance(input.InstanceID); err != nil { + slog.Warn("StartStoppedInstance: first KV delete failed, retrying", + "instanceId", input.InstanceID, "err", err) + if retryErr := s.stoppedStore.DeleteStoppedInstance(input.InstanceID); retryErr != nil { + slog.Error("StartStoppedInstance: KV delete failed after retry, instance is running locally but stale entry remains in shared KV", + "instanceId", input.InstanceID, "err", retryErr) + } + } + + slog.Info("Started stopped instance from shared KV", "instanceId", instance.ID) + return &StartStoppedInstanceOutput{Status: "running", InstanceID: instance.ID}, nil +} + // TerminateStoppedInstance picks up a stopped instance from shared KV, deletes // its volumes, releases its public IP and ENI, writes it to the terminated // bucket, then removes it from the stopped bucket. No QEMU shutdown or unmount diff --git a/spinifex/handlers/ec2/instance/service_impl_test.go b/spinifex/handlers/ec2/instance/service_impl_test.go index cf3bad0b3..ef92469f0 100644 --- a/spinifex/handlers/ec2/instance/service_impl_test.go +++ b/spinifex/handlers/ec2/instance/service_impl_test.go @@ -647,9 +647,14 @@ func TestCloudInitVolumeNamePerInstance(t *testing.T) { // --- 1b-pre Describe* coverage (siv-22 parts) ------------------------------ type fakeResourceCapacityProvider struct { - types []*ec2.InstanceTypeInfo - gotShowCap bool - calls int + types []*ec2.InstanceTypeInfo + gotShowCap bool + calls int + instanceTypes map[string]*ec2.InstanceTypeInfo + allocateErr error + allocated []*ec2.InstanceTypeInfo + deallocated []*ec2.InstanceTypeInfo + canAllocFn func(*ec2.InstanceTypeInfo, int) int } func (f *fakeResourceCapacityProvider) GetAvailableInstanceTypeInfos(showCapacity bool) []*ec2.InstanceTypeInfo { @@ -658,6 +663,29 @@ func (f *fakeResourceCapacityProvider) GetAvailableInstanceTypeInfos(showCapacit return f.types } +func (f *fakeResourceCapacityProvider) Allocate(it *ec2.InstanceTypeInfo) error { + if f.allocateErr != nil { + return f.allocateErr + } + f.allocated = append(f.allocated, it) + return nil +} + +func (f *fakeResourceCapacityProvider) Deallocate(it *ec2.InstanceTypeInfo) { + f.deallocated = append(f.deallocated, it) +} + +func (f *fakeResourceCapacityProvider) CanAllocate(it *ec2.InstanceTypeInfo, count int) int { + if f.canAllocFn != nil { + return f.canAllocFn(it, count) + } + return count +} + +func (f *fakeResourceCapacityProvider) InstanceTypes() map[string]*ec2.InstanceTypeInfo { + return f.instanceTypes +} + type fakeStoppedStore struct { stopped []*vm.VM terminated []*vm.VM @@ -1563,3 +1591,187 @@ func TestTerminateStoppedInstance_ENIDeleted(t *testing.T) { require.NoError(t, err) assert.Equal(t, []string{"eni-1234"}, ed.calls) } + +// --- StartStoppedInstance tests --- + +type fakeGPUClaimer struct { + claimed []string + released []string + claimErr error + pciAddress string + xvgaEnabled bool +} + +func (f *fakeGPUClaimer) Claim(instanceID string) (string, bool, error) { + f.claimed = append(f.claimed, instanceID) + if f.claimErr != nil { + return "", false, f.claimErr + } + return f.pciAddress, f.xvgaEnabled, nil +} + +func (f *fakeGPUClaimer) Release(instanceID string) error { + f.released = append(f.released, instanceID) + return nil +} + +func TestStartStoppedInstance_MissingInstanceID(t *testing.T) { + svc := &InstanceServiceImpl{} + _, err := svc.StartStoppedInstance(&StartStoppedInstanceInput{}, "acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorMissingParameter, err.Error()) +} + +func TestStartStoppedInstance_NilStore(t *testing.T) { + svc := &InstanceServiceImpl{} + _, err := svc.StartStoppedInstance(&StartStoppedInstanceInput{InstanceID: "i-1"}, "acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) +} + +func TestStartStoppedInstance_NilResourceMgr(t *testing.T) { + svc := &InstanceServiceImpl{stoppedStore: &fakeStoppedStore{loadByID: map[string]*vm.VM{}}} + _, err := svc.StartStoppedInstance(&StartStoppedInstanceInput{InstanceID: "i-1"}, "acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) +} + +func TestStartStoppedInstance_NilVMMgr(t *testing.T) { + svc := &InstanceServiceImpl{ + stoppedStore: &fakeStoppedStore{loadByID: map[string]*vm.VM{}}, + resourceMgr: &fakeResourceCapacityProvider{}, + } + _, err := svc.StartStoppedInstance(&StartStoppedInstanceInput{InstanceID: "i-1"}, "acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) +} + +func TestStartStoppedInstance_LoadError(t *testing.T) { + store := &fakeStoppedStore{loadErr: errors.New("kv down")} + svc := &InstanceServiceImpl{ + stoppedStore: store, + resourceMgr: &fakeResourceCapacityProvider{}, + vmMgr: vm.NewManager(), + } + _, err := svc.StartStoppedInstance(&StartStoppedInstanceInput{InstanceID: "i-1"}, "acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) +} + +func TestStartStoppedInstance_NotFound(t *testing.T) { + store := &fakeStoppedStore{loadByID: map[string]*vm.VM{}} + svc := &InstanceServiceImpl{ + stoppedStore: store, + resourceMgr: &fakeResourceCapacityProvider{}, + vmMgr: vm.NewManager(), + } + _, err := svc.StartStoppedInstance(&StartStoppedInstanceInput{InstanceID: "i-missing"}, "acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error()) +} + +func TestStartStoppedInstance_NotStopped(t *testing.T) { + id := "i-running" + store := &fakeStoppedStore{loadByID: map[string]*vm.VM{ + id: {ID: id, Status: vm.StateRunning, AccountID: "acc"}, + }} + svc := &InstanceServiceImpl{ + stoppedStore: store, + resourceMgr: &fakeResourceCapacityProvider{}, + vmMgr: vm.NewManager(), + } + _, err := svc.StartStoppedInstance(&StartStoppedInstanceInput{InstanceID: id}, "acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorIncorrectInstanceState, err.Error()) +} + +func TestStartStoppedInstance_NotVisible(t *testing.T) { + id := "i-foreign" + store := &fakeStoppedStore{loadByID: map[string]*vm.VM{ + id: {ID: id, Status: vm.StateStopped, AccountID: "owner-acc", InstanceType: "t3.micro"}, + }} + svc := &InstanceServiceImpl{ + stoppedStore: store, + resourceMgr: &fakeResourceCapacityProvider{}, + vmMgr: vm.NewManager(), + } + _, err := svc.StartStoppedInstance(&StartStoppedInstanceInput{InstanceID: id}, "other-acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error()) + + // Cross-tenant rejection must not delete from KV. + assert.Empty(t, store.deletedStopped, "cross-tenant rejection must not delete stopped instance") +} + +func TestStartStoppedInstance_InstanceTypeUnknown(t *testing.T) { + id := "i-badtype" + store := &fakeStoppedStore{loadByID: map[string]*vm.VM{ + id: {ID: id, Status: vm.StateStopped, AccountID: "acc", InstanceType: "z99.nope"}, + }} + prov := &fakeResourceCapacityProvider{instanceTypes: map[string]*ec2.InstanceTypeInfo{}} + svc := &InstanceServiceImpl{ + stoppedStore: store, + resourceMgr: prov, + vmMgr: vm.NewManager(), + } + _, err := svc.StartStoppedInstance(&StartStoppedInstanceInput{InstanceID: id}, "acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInsufficientInstanceCapacity, err.Error()) + assert.Empty(t, prov.allocated, "no allocation should occur for unknown type") +} + +func TestStartStoppedInstance_AllocateFails(t *testing.T) { + id := "i-alloc-fail" + itype := "t3.micro" + store := &fakeStoppedStore{loadByID: map[string]*vm.VM{ + id: {ID: id, Status: vm.StateStopped, AccountID: "acc", InstanceType: itype}, + }} + prov := &fakeResourceCapacityProvider{ + instanceTypes: map[string]*ec2.InstanceTypeInfo{ + itype: {InstanceType: aws.String(itype)}, + }, + allocateErr: errors.New("no capacity"), + } + svc := &InstanceServiceImpl{ + stoppedStore: store, + resourceMgr: prov, + vmMgr: vm.NewManager(), + } + _, err := svc.StartStoppedInstance(&StartStoppedInstanceInput{InstanceID: id}, "acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInsufficientInstanceCapacity, err.Error()) +} + +// GPU claim failure must roll back the resource allocation and remove the VM +// from the manager's map — otherwise stale capacity stays consumed. +func TestStartStoppedInstance_GPUClaimFailureRollsBack(t *testing.T) { + id := "i-gpu-fail" + itype := "g5.xlarge" + store := &fakeStoppedStore{loadByID: map[string]*vm.VM{ + id: {ID: id, Status: vm.StateStopped, AccountID: "acc", InstanceType: itype}, + }} + prov := &fakeResourceCapacityProvider{ + instanceTypes: map[string]*ec2.InstanceTypeInfo{ + itype: {InstanceType: aws.String(itype), GpuInfo: &ec2.GpuInfo{ + Gpus: []*ec2.GpuDeviceInfo{{Name: aws.String("nvidia-a10g"), Count: aws.Int64(1)}}, + }}, + }, + } + claimer := &fakeGPUClaimer{claimErr: errors.New("vfio bind failed")} + mgr := vm.NewManager() + svc := &InstanceServiceImpl{ + stoppedStore: store, + resourceMgr: prov, + vmMgr: mgr, + gpuClaimer: claimer, + } + _, err := svc.StartStoppedInstance(&StartStoppedInstanceInput{InstanceID: id}, "acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInsufficientInstanceCapacity, err.Error()) + + require.Len(t, prov.allocated, 1, "allocator must have been called") + require.Len(t, prov.deallocated, 1, "GPU claim failure must trigger deallocate") + _, stillInMgr := mgr.Get(id) + assert.False(t, stillInMgr, "GPU claim failure must remove the VM from the manager map") + assert.Empty(t, store.deletedStopped, "stopped-KV entry must remain on rollback") +} diff --git a/spinifex/handlers/ec2/instance/service_mock.go b/spinifex/handlers/ec2/instance/service_mock.go index dfae7fd7b..9b50abc8f 100644 --- a/spinifex/handlers/ec2/instance/service_mock.go +++ b/spinifex/handlers/ec2/instance/service_mock.go @@ -60,6 +60,10 @@ func (s *MockInstanceService) ModifyInstanceAttribute(_ *ec2.ModifyInstanceAttri return &ec2.ModifyInstanceAttributeOutput{}, nil } +func (s *MockInstanceService) StartStoppedInstance(input *StartStoppedInstanceInput, _ string) (*StartStoppedInstanceOutput, error) { + return &StartStoppedInstanceOutput{Status: "running", InstanceID: input.InstanceID}, nil +} + func (s *MockInstanceService) TerminateStoppedInstance(input *TerminateStoppedInstanceInput, _ string) (*TerminateStoppedInstanceOutput, error) { return &TerminateStoppedInstanceOutput{Status: "terminated", InstanceID: input.InstanceID}, nil } diff --git a/spinifex/handlers/ec2/instance/service_nats.go b/spinifex/handlers/ec2/instance/service_nats.go index 4540f7fdf..9792aadbe 100644 --- a/spinifex/handlers/ec2/instance/service_nats.go +++ b/spinifex/handlers/ec2/instance/service_nats.go @@ -65,6 +65,10 @@ func (s *NATSInstanceService) ModifyInstanceAttribute(input *ec2.ModifyInstanceA return utils.NATSRequest[ec2.ModifyInstanceAttributeOutput](s.natsConn, "ec2.ModifyInstanceAttribute", input, 30*time.Second, accountID) } +func (s *NATSInstanceService) StartStoppedInstance(input *StartStoppedInstanceInput, accountID string) (*StartStoppedInstanceOutput, error) { + return utils.NATSRequest[StartStoppedInstanceOutput](s.natsConn, "ec2.StartStoppedInstance", input, 30*time.Second, accountID) +} + func (s *NATSInstanceService) TerminateStoppedInstance(input *TerminateStoppedInstanceInput, accountID string) (*TerminateStoppedInstanceOutput, error) { return utils.NATSRequest[TerminateStoppedInstanceOutput](s.natsConn, "ec2.TerminateStoppedInstance", input, 30*time.Second, accountID) }