mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-19 19:01:42 +00:00
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.
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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"])
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user