mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-22 04:07:04 +00:00
refactor(daemon): extract ModifyInstanceAttribute to service (1b-pre phase 2a)
Collapse handleEC2ModifyInstanceAttribute to a thin Pattern A wrapper over
InstanceServiceImpl.ModifyInstanceAttribute, mirroring the Describe extraction
in 719ac63 (phase 1). The InstanceService interface and StoppedInstanceStore
each gain one method (ModifyInstanceAttribute, WriteStoppedInstance); the
daemon's JetStreamManager already implements WriteStoppedInstance, so no
production code outside the daemon handler changes.
Logic-level tests move from daemon/daemon_handlers_test.go to
handlers/ec2/instance/service_impl_test.go (12 cases against an in-memory
fakeStoppedStore). The daemon keeps one NATS round-trip wrapper smoke test.
The error-injection cases in daemon_handlers_instance_test.go remain in place
but now require daemonWithFakeStateStore to rewire d.instanceService with
the fake store — the previous `d.stateStore = store` swap no longer reaches
the new code path.
No behaviour change: SourceDestCheck stays a state-agnostic no-op, the
ownership/visibility check still gates after the stopped-state gate, and the
success response payload remains `{}`.
This commit is contained in:
@@ -1,7 +1,6 @@
|
||||
package daemon
|
||||
|
||||
import (
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
@@ -1162,110 +1161,8 @@ func (d *Daemon) describeInstancesFromKV(msg *nats.Msg, listFn func() ([]*vm.VM,
|
||||
slog.Info(handlerName+" completed", "count", len(reservations))
|
||||
}
|
||||
|
||||
// handleEC2ModifyInstanceAttribute modifies attributes of a stopped instance in shared KV.
|
||||
// All supported attributes (InstanceType, UserData) require the instance to be stopped.
|
||||
func (d *Daemon) handleEC2ModifyInstanceAttribute(msg *nats.Msg) {
|
||||
var input ec2.ModifyInstanceAttributeInput
|
||||
if err := json.Unmarshal(msg.Data, &input); err != nil {
|
||||
slog.Error("handleEC2ModifyInstanceAttribute: failed to unmarshal request", "err", err)
|
||||
respondWithError(msg, awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
|
||||
if input.InstanceId == nil || *input.InstanceId == "" {
|
||||
slog.Error("handleEC2ModifyInstanceAttribute: missing instance_id")
|
||||
respondWithError(msg, awserrors.ErrorMissingParameter)
|
||||
return
|
||||
}
|
||||
|
||||
instanceID := *input.InstanceId
|
||||
|
||||
// SourceDestCheck is a networking concept that doesn't apply to bare-metal VMs.
|
||||
// Accept the call as a no-op so Terraform and the AWS CLI don't error out.
|
||||
// Unlike InstanceType/UserData, AWS allows this on running instances, so handle
|
||||
// it before the stopped-state gate.
|
||||
if input.SourceDestCheck != nil {
|
||||
slog.Info("handleEC2ModifyInstanceAttribute: accepting SourceDestCheck (no-op on bare metal)", "instanceId", instanceID)
|
||||
if err := msg.Respond([]byte(`{}`)); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if d.stateStore == nil {
|
||||
slog.Error("handleEC2ModifyInstanceAttribute: state store not available")
|
||||
respondWithError(msg, awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
|
||||
instance, err := d.stateStore.LoadStoppedInstance(instanceID)
|
||||
if err != nil {
|
||||
slog.Error("handleEC2ModifyInstanceAttribute: failed to load stopped instance", "instanceId", instanceID, "err", err)
|
||||
respondWithError(msg, awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
if instance == nil {
|
||||
slog.Warn("handleEC2ModifyInstanceAttribute: instance not found in shared KV", "instanceId", instanceID)
|
||||
respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
|
||||
return
|
||||
}
|
||||
|
||||
if instance.Status != vm.StateStopped {
|
||||
slog.Error("handleEC2ModifyInstanceAttribute: instance not in stopped state", "instanceId", instanceID, "status", instance.Status)
|
||||
respondWithError(msg, awserrors.ErrorIncorrectInstanceState)
|
||||
return
|
||||
}
|
||||
|
||||
// Verify the caller owns this instance
|
||||
if !checkInstanceOwnership(msg, instanceID, instance.AccountID) {
|
||||
return
|
||||
}
|
||||
|
||||
// Apply the requested attribute change
|
||||
if input.InstanceType != nil && input.InstanceType.Value != nil {
|
||||
newType := *input.InstanceType.Value
|
||||
if newType == "" {
|
||||
slog.Error("handleEC2ModifyInstanceAttribute: empty instance type value", "instanceId", instanceID)
|
||||
respondWithError(msg, awserrors.ErrorInvalidInstanceAttributeValue)
|
||||
return
|
||||
}
|
||||
if instance.Instance == nil {
|
||||
slog.Error("handleEC2ModifyInstanceAttribute: instance.Instance is nil, data integrity issue", "instanceId", instanceID)
|
||||
respondWithError(msg, awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
slog.Info("handleEC2ModifyInstanceAttribute: changing instance type",
|
||||
"instanceId", instanceID, "oldType", instance.InstanceType, "newType", newType)
|
||||
|
||||
instance.InstanceType = newType
|
||||
instance.Config.InstanceType = newType
|
||||
instance.Instance.InstanceType = aws.String(newType)
|
||||
// Clear StateReason — resolves capacity-unavailable state from instance-type-missing bug
|
||||
instance.Instance.StateReason = nil
|
||||
}
|
||||
|
||||
if input.UserData != nil && input.UserData.Value != nil {
|
||||
slog.Info("handleEC2ModifyInstanceAttribute: changing user data", "instanceId", instanceID)
|
||||
|
||||
// Value arrives as decoded bytes (JSON unmarshal handles base64 → []byte automatically)
|
||||
instance.UserData = string(input.UserData.Value)
|
||||
if instance.RunInstancesInput != nil {
|
||||
instance.RunInstancesInput.UserData = aws.String(base64.StdEncoding.EncodeToString(input.UserData.Value))
|
||||
}
|
||||
}
|
||||
|
||||
if err := d.stateStore.WriteStoppedInstance(instanceID, instance); err != nil {
|
||||
slog.Error("handleEC2ModifyInstanceAttribute: failed to write modified instance to KV",
|
||||
"instanceId", instanceID, "err", err)
|
||||
respondWithError(msg, awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
|
||||
slog.Info("handleEC2ModifyInstanceAttribute: completed successfully", "instanceId", instanceID)
|
||||
|
||||
if err := msg.Respond([]byte(`{}`)); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
handleNATSRequest(msg, d.instanceService.ModifyInstanceAttribute)
|
||||
}
|
||||
|
||||
// handleEC2DescribeInstanceAttribute returns a single requested attribute for an instance.
|
||||
|
||||
@@ -10,6 +10,8 @@ import (
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/service/ec2"
|
||||
"github.com/mulgadc/spinifex/spinifex/awserrors"
|
||||
handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
|
||||
"github.com/mulgadc/spinifex/spinifex/objectstore"
|
||||
"github.com/mulgadc/spinifex/spinifex/utils"
|
||||
"github.com/mulgadc/spinifex/spinifex/vm"
|
||||
"github.com/nats-io/nats.go"
|
||||
@@ -142,11 +144,17 @@ var _ vm.StateStore = (*fakeStateStore)(nil)
|
||||
|
||||
// daemonWithFakeStateStore returns a daemon wired with an in-memory NATS
|
||||
// connection (via createTestDaemon) and the supplied fake StateStore.
|
||||
// The daemon does not have JetStream initialized.
|
||||
// The daemon does not have JetStream initialized. Rewires d.instanceService
|
||||
// to point at the fake store so handlers that delegate to InstanceService
|
||||
// (e.g. ModifyInstanceAttribute) see the injected state.
|
||||
func daemonWithFakeStateStore(t *testing.T, store *fakeStateStore) *Daemon {
|
||||
t.Helper()
|
||||
d := createTestDaemon(t, sharedNATSURL)
|
||||
d.stateStore = store
|
||||
d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(
|
||||
d.config, d.resourceMgr.instanceTypes, d.natsConn,
|
||||
objectstore.NewMemoryObjectStore(), d.vmMgr, d.resourceMgr, store,
|
||||
)
|
||||
return d
|
||||
}
|
||||
|
||||
|
||||
@@ -1653,9 +1653,14 @@ func TestAttachVolume_ZoneMismatch(t *testing.T) {
|
||||
assert.Contains(t, string(resp.Data), "InvalidVolume.ZoneMismatch")
|
||||
}
|
||||
|
||||
// --- handleEC2ModifyInstanceAttribute tests ---
|
||||
// --- handleEC2ModifyInstanceAttribute wrapper smoke test ---
|
||||
//
|
||||
// Detailed logic coverage lives in
|
||||
// handlers/ec2/instance/service_impl_test.go (TestModifyInstanceAttribute_*).
|
||||
// This case keeps one end-to-end NATS → handleNATSRequest → service round-trip
|
||||
// to confirm the daemon wiring stays intact.
|
||||
|
||||
func TestHandleEC2ModifyInstanceAttribute_ChangeInstanceType(t *testing.T) {
|
||||
func TestHandleEC2ModifyInstanceAttribute_WrapperRoundTrip(t *testing.T) {
|
||||
natsURL := sharedJSNATSURL
|
||||
|
||||
daemon := createFullTestDaemonWithJetStream(t, natsURL)
|
||||
@@ -1664,7 +1669,7 @@ func TestHandleEC2ModifyInstanceAttribute_ChangeInstanceType(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
instanceID := "i-modify-type-001"
|
||||
instanceID := "i-modify-wrapper-001"
|
||||
instance := &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: vm.StateStopped,
|
||||
@@ -1693,260 +1698,6 @@ func TestHandleEC2ModifyInstanceAttribute_ChangeInstanceType(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, updated)
|
||||
assert.Equal(t, "t3.medium", updated.InstanceType)
|
||||
assert.Equal(t, "t3.medium", updated.Config.InstanceType)
|
||||
assert.Equal(t, "t3.medium", *updated.Instance.InstanceType)
|
||||
}
|
||||
|
||||
func TestHandleEC2ModifyInstanceAttribute_ChangeUserData(t *testing.T) {
|
||||
natsURL := sharedJSNATSURL
|
||||
|
||||
daemon := createFullTestDaemonWithJetStream(t, natsURL)
|
||||
|
||||
sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", daemon.handleEC2ModifyInstanceAttribute)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
instanceID := "i-modify-ud-001"
|
||||
instance := &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: vm.StateStopped,
|
||||
InstanceType: "t3.micro",
|
||||
AccountID: testAccountID,
|
||||
UserData: "old data",
|
||||
RunInstancesInput: &ec2.RunInstancesInput{
|
||||
UserData: aws.String("b2xkIGRhdGE="),
|
||||
},
|
||||
Instance: &ec2.Instance{
|
||||
InstanceId: aws.String(instanceID),
|
||||
},
|
||||
}
|
||||
err = daemon.jsManager.WriteStoppedInstance(instanceID, instance)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = daemon.jsManager.DeleteStoppedInstance(instanceID) })
|
||||
|
||||
// Value holds decoded bytes (the gateway query parser decodes base64 from the CLI,
|
||||
// then json.Marshal/Unmarshal round-trips []byte through base64 transparently)
|
||||
newContent := "#!/bin/bash"
|
||||
input := &ec2.ModifyInstanceAttributeInput{
|
||||
InstanceId: aws.String(instanceID),
|
||||
UserData: &ec2.BlobAttributeValue{Value: []byte(newContent)},
|
||||
}
|
||||
reqData, _ := json.Marshal(input)
|
||||
reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, `{}`, string(reply.Data))
|
||||
|
||||
updated, err := daemon.jsManager.LoadStoppedInstance(instanceID)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, updated)
|
||||
assert.Equal(t, newContent, updated.UserData)
|
||||
assert.Equal(t, "IyEvYmluL2Jhc2g=", *updated.RunInstancesInput.UserData)
|
||||
}
|
||||
|
||||
func TestHandleEC2ModifyInstanceAttribute_SourceDestCheck(t *testing.T) {
|
||||
natsURL := sharedJSNATSURL
|
||||
|
||||
daemon := createFullTestDaemonWithJetStream(t, natsURL)
|
||||
|
||||
sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", daemon.handleEC2ModifyInstanceAttribute)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
// SourceDestCheck is a no-op that succeeds without requiring a stopped instance
|
||||
// in KV — Terraform sends this on running instances right after creation.
|
||||
input := &ec2.ModifyInstanceAttributeInput{
|
||||
InstanceId: aws.String("i-modify-sdc-001"),
|
||||
SourceDestCheck: &ec2.AttributeBooleanValue{Value: aws.Bool(false)},
|
||||
}
|
||||
reqData, _ := json.Marshal(input)
|
||||
reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, `{}`, string(reply.Data))
|
||||
}
|
||||
|
||||
func TestHandleEC2ModifyInstanceAttribute_InstanceNotFound(t *testing.T) {
|
||||
natsURL := sharedJSNATSURL
|
||||
|
||||
daemon := createFullTestDaemonWithJetStream(t, natsURL)
|
||||
|
||||
sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", daemon.handleEC2ModifyInstanceAttribute)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
input := &ec2.ModifyInstanceAttributeInput{
|
||||
InstanceId: aws.String("i-nonexistent"),
|
||||
InstanceType: &ec2.AttributeValue{Value: aws.String("t3.medium")},
|
||||
}
|
||||
reqData, _ := json.Marshal(input)
|
||||
reply, err := daemon.natsConn.Request("ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
|
||||
var errResp map[string]any
|
||||
err = json.Unmarshal(reply.Data, &errResp)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, errResp["Code"])
|
||||
}
|
||||
|
||||
func TestHandleEC2ModifyInstanceAttribute_NotStopped(t *testing.T) {
|
||||
natsURL := sharedJSNATSURL
|
||||
|
||||
daemon := createFullTestDaemonWithJetStream(t, natsURL)
|
||||
|
||||
sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", daemon.handleEC2ModifyInstanceAttribute)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
instanceID := "i-modify-running-001"
|
||||
instance := &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: vm.StateRunning,
|
||||
InstanceType: "t3.micro",
|
||||
AccountID: testAccountID,
|
||||
}
|
||||
err = daemon.jsManager.WriteStoppedInstance(instanceID, instance)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = daemon.jsManager.DeleteStoppedInstance(instanceID) })
|
||||
|
||||
input := &ec2.ModifyInstanceAttributeInput{
|
||||
InstanceId: aws.String(instanceID),
|
||||
InstanceType: &ec2.AttributeValue{Value: aws.String("t3.medium")},
|
||||
}
|
||||
reqData, _ := json.Marshal(input)
|
||||
reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
|
||||
var errResp map[string]any
|
||||
err = json.Unmarshal(reply.Data, &errResp)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, awserrors.ErrorIncorrectInstanceState, errResp["Code"])
|
||||
}
|
||||
|
||||
func TestHandleEC2ModifyInstanceAttribute_ClearsStateReason(t *testing.T) {
|
||||
natsURL := sharedJSNATSURL
|
||||
|
||||
daemon := createFullTestDaemonWithJetStream(t, natsURL)
|
||||
|
||||
sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", daemon.handleEC2ModifyInstanceAttribute)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
instanceID := "i-modify-recovery-001"
|
||||
instance := &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: vm.StateStopped,
|
||||
InstanceType: "m7i.small",
|
||||
AccountID: testAccountID,
|
||||
Config: vm.Config{InstanceType: "m7i.small"},
|
||||
Instance: &ec2.Instance{
|
||||
InstanceId: aws.String(instanceID),
|
||||
InstanceType: aws.String("m7i.small"),
|
||||
StateReason: &ec2.StateReason{
|
||||
Code: aws.String("Server.InsufficientInstanceCapacity"),
|
||||
Message: aws.String("Instance type not available on any node"),
|
||||
},
|
||||
},
|
||||
}
|
||||
err = daemon.jsManager.WriteStoppedInstance(instanceID, instance)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = daemon.jsManager.DeleteStoppedInstance(instanceID) })
|
||||
|
||||
input := &ec2.ModifyInstanceAttributeInput{
|
||||
InstanceId: aws.String(instanceID),
|
||||
InstanceType: &ec2.AttributeValue{Value: aws.String("t3.micro")},
|
||||
}
|
||||
reqData, _ := json.Marshal(input)
|
||||
reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, `{}`, string(reply.Data))
|
||||
|
||||
updated, err := daemon.jsManager.LoadStoppedInstance(instanceID)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, updated)
|
||||
assert.Equal(t, "t3.micro", updated.InstanceType)
|
||||
assert.Equal(t, "t3.micro", updated.Config.InstanceType)
|
||||
assert.Equal(t, "t3.micro", *updated.Instance.InstanceType)
|
||||
assert.Nil(t, updated.Instance.StateReason)
|
||||
}
|
||||
|
||||
func TestHandleEC2ModifyInstanceAttribute_InvalidTypeAccepted(t *testing.T) {
|
||||
natsURL := sharedJSNATSURL
|
||||
|
||||
daemon := createFullTestDaemonWithJetStream(t, natsURL)
|
||||
|
||||
sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", daemon.handleEC2ModifyInstanceAttribute)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
instanceID := "i-modify-nonsense-001"
|
||||
instance := &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: vm.StateStopped,
|
||||
InstanceType: "t3.micro",
|
||||
AccountID: testAccountID,
|
||||
Config: vm.Config{InstanceType: "t3.micro"},
|
||||
Instance: &ec2.Instance{
|
||||
InstanceId: aws.String(instanceID),
|
||||
InstanceType: aws.String("t3.micro"),
|
||||
},
|
||||
}
|
||||
err = daemon.jsManager.WriteStoppedInstance(instanceID, instance)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = daemon.jsManager.DeleteStoppedInstance(instanceID) })
|
||||
|
||||
// z99.mega is nonsense — modify does not pre-validate, matching AWS behavior
|
||||
input := &ec2.ModifyInstanceAttributeInput{
|
||||
InstanceId: aws.String(instanceID),
|
||||
InstanceType: &ec2.AttributeValue{Value: aws.String("z99.mega")},
|
||||
}
|
||||
reqData, _ := json.Marshal(input)
|
||||
reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, `{}`, string(reply.Data))
|
||||
|
||||
updated, err := daemon.jsManager.LoadStoppedInstance(instanceID)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, updated)
|
||||
assert.Equal(t, "z99.mega", updated.InstanceType)
|
||||
assert.Equal(t, "z99.mega", updated.Config.InstanceType)
|
||||
assert.Equal(t, "z99.mega", *updated.Instance.InstanceType)
|
||||
}
|
||||
|
||||
func TestHandleEC2ModifyInstanceAttribute_MissingInstanceID(t *testing.T) {
|
||||
natsURL := sharedJSNATSURL
|
||||
|
||||
daemon := createFullTestDaemonWithJetStream(t, natsURL)
|
||||
|
||||
sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", daemon.handleEC2ModifyInstanceAttribute)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
input := &ec2.ModifyInstanceAttributeInput{}
|
||||
reqData, _ := json.Marshal(input)
|
||||
reply, err := daemon.natsConn.Request("ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
|
||||
var errResp map[string]any
|
||||
err = json.Unmarshal(reply.Data, &errResp)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, awserrors.ErrorMissingParameter, errResp["Code"])
|
||||
}
|
||||
|
||||
func TestHandleEC2ModifyInstanceAttribute_InvalidJSON(t *testing.T) {
|
||||
natsURL := sharedJSNATSURL
|
||||
|
||||
daemon := createFullTestDaemonWithJetStream(t, natsURL)
|
||||
|
||||
sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", daemon.handleEC2ModifyInstanceAttribute)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
reply, err := daemon.natsConn.Request("ec2.ModifyInstanceAttribute", []byte(`{invalid`), 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
|
||||
var errResp map[string]any
|
||||
err = json.Unmarshal(reply.Data, &errResp)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, awserrors.ErrorServerInternal, errResp["Code"])
|
||||
}
|
||||
|
||||
// --- DescribeInstanceAttribute daemon tests ---
|
||||
|
||||
@@ -13,6 +13,7 @@ type InstanceService interface {
|
||||
DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, accountID string) (*ec2.DescribeInstanceAttributeOutput, error)
|
||||
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)
|
||||
}
|
||||
|
||||
// ResourceCapacityProvider exposes per-node instance-type availability used by
|
||||
@@ -21,10 +22,11 @@ type ResourceCapacityProvider interface {
|
||||
GetAvailableInstanceTypeInfos(showCapacity bool) []*ec2.InstanceTypeInfo
|
||||
}
|
||||
|
||||
// StoppedInstanceStore provides KV-backed read access to stopped/terminated
|
||||
// instances. Implemented by daemon.JetStreamManager.
|
||||
// StoppedInstanceStore provides KV-backed read/write access to stopped and
|
||||
// read access to terminated instances. Implemented by daemon.JetStreamManager.
|
||||
type StoppedInstanceStore interface {
|
||||
LoadStoppedInstance(instanceID string) (*vm.VM, error)
|
||||
ListStoppedInstances() ([]*vm.VM, error)
|
||||
ListTerminatedInstances() ([]*vm.VM, error)
|
||||
WriteStoppedInstance(instanceID string, instance *vm.VM) error
|
||||
}
|
||||
|
||||
@@ -1157,6 +1157,86 @@ func (s *InstanceServiceImpl) describeInstancesFromKV(input *ec2.DescribeInstanc
|
||||
return &ec2.DescribeInstancesOutput{Reservations: reservations}, nil
|
||||
}
|
||||
|
||||
// ModifyInstanceAttribute applies a single attribute change to a stopped instance.
|
||||
// All supported attributes (InstanceType, UserData) require the instance to be
|
||||
// stopped. SourceDestCheck is a networking concept that does not apply to
|
||||
// bare-metal VMs; it is accepted as a no-op on any instance state so Terraform
|
||||
// and the AWS CLI do not error out.
|
||||
func (s *InstanceServiceImpl) ModifyInstanceAttribute(input *ec2.ModifyInstanceAttributeInput, accountID string) (*ec2.ModifyInstanceAttributeOutput, error) {
|
||||
if input.InstanceId == nil || *input.InstanceId == "" {
|
||||
return nil, errors.New(awserrors.ErrorMissingParameter)
|
||||
}
|
||||
instanceID := *input.InstanceId
|
||||
|
||||
if input.SourceDestCheck != nil {
|
||||
slog.Info("ModifyInstanceAttribute: accepting SourceDestCheck (no-op on bare metal)", "instanceId", instanceID)
|
||||
return &ec2.ModifyInstanceAttributeOutput{}, nil
|
||||
}
|
||||
|
||||
if s.stoppedStore == nil {
|
||||
slog.Error("ModifyInstanceAttribute: stopped store not available")
|
||||
return nil, errors.New(awserrors.ErrorServerInternal)
|
||||
}
|
||||
|
||||
instance, err := s.stoppedStore.LoadStoppedInstance(instanceID)
|
||||
if err != nil {
|
||||
slog.Error("ModifyInstanceAttribute: failed to load stopped instance", "instanceId", instanceID, "err", err)
|
||||
return nil, errors.New(awserrors.ErrorServerInternal)
|
||||
}
|
||||
if instance == nil {
|
||||
slog.Warn("ModifyInstanceAttribute: instance not found in shared KV", "instanceId", instanceID)
|
||||
return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
|
||||
}
|
||||
|
||||
if instance.Status != vm.StateStopped {
|
||||
slog.Error("ModifyInstanceAttribute: instance not in stopped state", "instanceId", instanceID, "status", instance.Status)
|
||||
return nil, errors.New(awserrors.ErrorIncorrectInstanceState)
|
||||
}
|
||||
|
||||
if !IsInstanceVisible(accountID, instance.AccountID) {
|
||||
slog.Warn("ModifyInstanceAttribute: instance not visible",
|
||||
"instanceId", instanceID, "callerAccount", accountID, "ownerAccount", instance.AccountID)
|
||||
return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
|
||||
}
|
||||
|
||||
if input.InstanceType != nil && input.InstanceType.Value != nil {
|
||||
newType := *input.InstanceType.Value
|
||||
if newType == "" {
|
||||
slog.Error("ModifyInstanceAttribute: empty instance type value", "instanceId", instanceID)
|
||||
return nil, errors.New(awserrors.ErrorInvalidInstanceAttributeValue)
|
||||
}
|
||||
if instance.Instance == nil {
|
||||
slog.Error("ModifyInstanceAttribute: instance.Instance is nil, data integrity issue", "instanceId", instanceID)
|
||||
return nil, errors.New(awserrors.ErrorServerInternal)
|
||||
}
|
||||
slog.Info("ModifyInstanceAttribute: changing instance type",
|
||||
"instanceId", instanceID, "oldType", instance.InstanceType, "newType", newType)
|
||||
|
||||
instance.InstanceType = newType
|
||||
instance.Config.InstanceType = newType
|
||||
instance.Instance.InstanceType = aws.String(newType)
|
||||
// Clear StateReason — resolves capacity-unavailable state from instance-type-missing bug.
|
||||
instance.Instance.StateReason = nil
|
||||
}
|
||||
|
||||
if input.UserData != nil && input.UserData.Value != nil {
|
||||
slog.Info("ModifyInstanceAttribute: changing user data", "instanceId", instanceID)
|
||||
instance.UserData = string(input.UserData.Value)
|
||||
if instance.RunInstancesInput != nil {
|
||||
instance.RunInstancesInput.UserData = aws.String(base64.StdEncoding.EncodeToString(input.UserData.Value))
|
||||
}
|
||||
}
|
||||
|
||||
if err := s.stoppedStore.WriteStoppedInstance(instanceID, instance); err != nil {
|
||||
slog.Error("ModifyInstanceAttribute: failed to write modified instance to KV",
|
||||
"instanceId", instanceID, "err", err)
|
||||
return nil, errors.New(awserrors.ErrorServerInternal)
|
||||
}
|
||||
|
||||
slog.Info("ModifyInstanceAttribute: completed successfully", "instanceId", instanceID)
|
||||
return &ec2.ModifyInstanceAttributeOutput{}, nil
|
||||
}
|
||||
|
||||
// DescribeInstanceAttribute returns a single requested attribute for an instance.
|
||||
// Checks running instances first (in-memory), then falls back to stopped instances in KV.
|
||||
func (s *InstanceServiceImpl) DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, accountID string) (*ec2.DescribeInstanceAttributeOutput, error) {
|
||||
|
||||
@@ -653,12 +653,14 @@ func (f *fakeResourceCapacityProvider) GetAvailableInstanceTypeInfos(showCapacit
|
||||
}
|
||||
|
||||
type fakeStoppedStore struct {
|
||||
stopped []*vm.VM
|
||||
terminated []*vm.VM
|
||||
loadByID map[string]*vm.VM
|
||||
listErr error
|
||||
listTermErr error
|
||||
loadErr error
|
||||
stopped []*vm.VM
|
||||
terminated []*vm.VM
|
||||
loadByID map[string]*vm.VM
|
||||
wroteStopped map[string]*vm.VM
|
||||
listErr error
|
||||
listTermErr error
|
||||
loadErr error
|
||||
writeErr error
|
||||
}
|
||||
|
||||
func (f *fakeStoppedStore) LoadStoppedInstance(id string) (*vm.VM, error) {
|
||||
@@ -682,6 +684,16 @@ func (f *fakeStoppedStore) ListTerminatedInstances() ([]*vm.VM, error) {
|
||||
}
|
||||
return f.terminated, nil
|
||||
}
|
||||
func (f *fakeStoppedStore) WriteStoppedInstance(id string, instance *vm.VM) error {
|
||||
if f.writeErr != nil {
|
||||
return f.writeErr
|
||||
}
|
||||
if f.wroteStopped == nil {
|
||||
f.wroteStopped = make(map[string]*vm.VM)
|
||||
}
|
||||
f.wroteStopped[id] = instance
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestDescribeInstanceTypes_NilResourceMgr(t *testing.T) {
|
||||
svc := &InstanceServiceImpl{}
|
||||
@@ -1036,3 +1048,222 @@ func TestIsInstanceVisible(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestModifyInstanceAttribute_MissingInstanceID(t *testing.T) {
|
||||
svc := &InstanceServiceImpl{}
|
||||
_, err := svc.ModifyInstanceAttribute(&ec2.ModifyInstanceAttributeInput{}, "acc")
|
||||
require.Error(t, err)
|
||||
assert.Equal(t, awserrors.ErrorMissingParameter, err.Error())
|
||||
}
|
||||
|
||||
func TestModifyInstanceAttribute_SourceDestCheckNoOp(t *testing.T) {
|
||||
// SourceDestCheck succeeds without touching KV or requiring stopped state.
|
||||
svc := &InstanceServiceImpl{}
|
||||
out, err := svc.ModifyInstanceAttribute(&ec2.ModifyInstanceAttributeInput{
|
||||
InstanceId: aws.String("i-sdc-001"),
|
||||
SourceDestCheck: &ec2.AttributeBooleanValue{Value: aws.Bool(false)},
|
||||
}, "acc")
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, out)
|
||||
}
|
||||
|
||||
func TestModifyInstanceAttribute_NilStore(t *testing.T) {
|
||||
svc := &InstanceServiceImpl{}
|
||||
_, err := svc.ModifyInstanceAttribute(&ec2.ModifyInstanceAttributeInput{
|
||||
InstanceId: aws.String("i-1"),
|
||||
InstanceType: &ec2.AttributeValue{Value: aws.String("t3.medium")},
|
||||
}, "acc")
|
||||
require.Error(t, err)
|
||||
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
|
||||
}
|
||||
|
||||
func TestModifyInstanceAttribute_InstanceNotFound(t *testing.T) {
|
||||
store := &fakeStoppedStore{loadByID: map[string]*vm.VM{}}
|
||||
svc := &InstanceServiceImpl{stoppedStore: store}
|
||||
|
||||
_, err := svc.ModifyInstanceAttribute(&ec2.ModifyInstanceAttributeInput{
|
||||
InstanceId: aws.String("i-missing"),
|
||||
InstanceType: &ec2.AttributeValue{Value: aws.String("t3.medium")},
|
||||
}, "acc")
|
||||
require.Error(t, err)
|
||||
assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
|
||||
}
|
||||
|
||||
func TestModifyInstanceAttribute_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}
|
||||
|
||||
_, err := svc.ModifyInstanceAttribute(&ec2.ModifyInstanceAttributeInput{
|
||||
InstanceId: aws.String(id),
|
||||
InstanceType: &ec2.AttributeValue{Value: aws.String("t3.medium")},
|
||||
}, "acc")
|
||||
require.Error(t, err)
|
||||
assert.Equal(t, awserrors.ErrorIncorrectInstanceState, err.Error())
|
||||
}
|
||||
|
||||
func TestModifyInstanceAttribute_NotVisible(t *testing.T) {
|
||||
id := "i-stopped"
|
||||
store := &fakeStoppedStore{loadByID: map[string]*vm.VM{
|
||||
id: {ID: id, Status: vm.StateStopped, AccountID: "owner-acc"},
|
||||
}}
|
||||
svc := &InstanceServiceImpl{stoppedStore: store}
|
||||
|
||||
_, err := svc.ModifyInstanceAttribute(&ec2.ModifyInstanceAttributeInput{
|
||||
InstanceId: aws.String(id),
|
||||
InstanceType: &ec2.AttributeValue{Value: aws.String("t3.medium")},
|
||||
}, "other-acc")
|
||||
require.Error(t, err)
|
||||
assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
|
||||
}
|
||||
|
||||
func TestModifyInstanceAttribute_ChangeInstanceType(t *testing.T) {
|
||||
id := "i-type"
|
||||
store := &fakeStoppedStore{loadByID: map[string]*vm.VM{
|
||||
id: {
|
||||
ID: id,
|
||||
Status: vm.StateStopped,
|
||||
AccountID: "acc",
|
||||
InstanceType: "t3.micro",
|
||||
Config: vm.Config{InstanceType: "t3.micro"},
|
||||
Instance: &ec2.Instance{
|
||||
InstanceId: aws.String(id),
|
||||
InstanceType: aws.String("t3.micro"),
|
||||
},
|
||||
},
|
||||
}}
|
||||
svc := &InstanceServiceImpl{stoppedStore: store}
|
||||
|
||||
_, err := svc.ModifyInstanceAttribute(&ec2.ModifyInstanceAttributeInput{
|
||||
InstanceId: aws.String(id),
|
||||
InstanceType: &ec2.AttributeValue{Value: aws.String("t3.medium")},
|
||||
}, "acc")
|
||||
require.NoError(t, err)
|
||||
|
||||
updated := store.wroteStopped[id]
|
||||
require.NotNil(t, updated)
|
||||
assert.Equal(t, "t3.medium", updated.InstanceType)
|
||||
assert.Equal(t, "t3.medium", updated.Config.InstanceType)
|
||||
assert.Equal(t, "t3.medium", *updated.Instance.InstanceType)
|
||||
}
|
||||
|
||||
func TestModifyInstanceAttribute_ChangeInstanceType_EmptyValue(t *testing.T) {
|
||||
id := "i-empty"
|
||||
store := &fakeStoppedStore{loadByID: map[string]*vm.VM{
|
||||
id: {ID: id, Status: vm.StateStopped, AccountID: "acc", Instance: &ec2.Instance{}},
|
||||
}}
|
||||
svc := &InstanceServiceImpl{stoppedStore: store}
|
||||
|
||||
_, err := svc.ModifyInstanceAttribute(&ec2.ModifyInstanceAttributeInput{
|
||||
InstanceId: aws.String(id),
|
||||
InstanceType: &ec2.AttributeValue{Value: aws.String("")},
|
||||
}, "acc")
|
||||
require.Error(t, err)
|
||||
assert.Equal(t, awserrors.ErrorInvalidInstanceAttributeValue, err.Error())
|
||||
}
|
||||
|
||||
func TestModifyInstanceAttribute_ChangeInstanceType_NilEmbeddedInstance(t *testing.T) {
|
||||
id := "i-nil-inst"
|
||||
store := &fakeStoppedStore{loadByID: map[string]*vm.VM{
|
||||
id: {ID: id, Status: vm.StateStopped, AccountID: "acc"},
|
||||
}}
|
||||
svc := &InstanceServiceImpl{stoppedStore: store}
|
||||
|
||||
_, err := svc.ModifyInstanceAttribute(&ec2.ModifyInstanceAttributeInput{
|
||||
InstanceId: aws.String(id),
|
||||
InstanceType: &ec2.AttributeValue{Value: aws.String("t3.medium")},
|
||||
}, "acc")
|
||||
require.Error(t, err)
|
||||
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
|
||||
}
|
||||
|
||||
func TestModifyInstanceAttribute_ChangeUserData(t *testing.T) {
|
||||
id := "i-ud"
|
||||
store := &fakeStoppedStore{loadByID: map[string]*vm.VM{
|
||||
id: {
|
||||
ID: id,
|
||||
Status: vm.StateStopped,
|
||||
AccountID: "acc",
|
||||
UserData: "old",
|
||||
RunInstancesInput: &ec2.RunInstancesInput{
|
||||
UserData: aws.String("b2xk"),
|
||||
},
|
||||
Instance: &ec2.Instance{InstanceId: aws.String(id)},
|
||||
},
|
||||
}}
|
||||
svc := &InstanceServiceImpl{stoppedStore: store}
|
||||
|
||||
newContent := "#!/bin/bash"
|
||||
_, err := svc.ModifyInstanceAttribute(&ec2.ModifyInstanceAttributeInput{
|
||||
InstanceId: aws.String(id),
|
||||
UserData: &ec2.BlobAttributeValue{Value: []byte(newContent)},
|
||||
}, "acc")
|
||||
require.NoError(t, err)
|
||||
|
||||
updated := store.wroteStopped[id]
|
||||
require.NotNil(t, updated)
|
||||
assert.Equal(t, newContent, updated.UserData)
|
||||
assert.Equal(t, "IyEvYmluL2Jhc2g=", *updated.RunInstancesInput.UserData)
|
||||
}
|
||||
|
||||
func TestModifyInstanceAttribute_ClearsStateReason(t *testing.T) {
|
||||
id := "i-recovery"
|
||||
store := &fakeStoppedStore{loadByID: map[string]*vm.VM{
|
||||
id: {
|
||||
ID: id,
|
||||
Status: vm.StateStopped,
|
||||
AccountID: "acc",
|
||||
InstanceType: "m7i.small",
|
||||
Config: vm.Config{InstanceType: "m7i.small"},
|
||||
Instance: &ec2.Instance{
|
||||
InstanceId: aws.String(id),
|
||||
InstanceType: aws.String("m7i.small"),
|
||||
StateReason: &ec2.StateReason{
|
||||
Code: aws.String("Server.InsufficientInstanceCapacity"),
|
||||
Message: aws.String("Instance type not available on any node"),
|
||||
},
|
||||
},
|
||||
},
|
||||
}}
|
||||
svc := &InstanceServiceImpl{stoppedStore: store}
|
||||
|
||||
_, err := svc.ModifyInstanceAttribute(&ec2.ModifyInstanceAttributeInput{
|
||||
InstanceId: aws.String(id),
|
||||
InstanceType: &ec2.AttributeValue{Value: aws.String("t3.micro")},
|
||||
}, "acc")
|
||||
require.NoError(t, err)
|
||||
|
||||
updated := store.wroteStopped[id]
|
||||
require.NotNil(t, updated)
|
||||
assert.Nil(t, updated.Instance.StateReason)
|
||||
}
|
||||
|
||||
func TestModifyInstanceAttribute_WriteError(t *testing.T) {
|
||||
id := "i-werr"
|
||||
store := &fakeStoppedStore{
|
||||
loadByID: map[string]*vm.VM{
|
||||
id: {
|
||||
ID: id,
|
||||
Status: vm.StateStopped,
|
||||
AccountID: "acc",
|
||||
InstanceType: "t3.micro",
|
||||
Config: vm.Config{InstanceType: "t3.micro"},
|
||||
Instance: &ec2.Instance{
|
||||
InstanceId: aws.String(id),
|
||||
InstanceType: aws.String("t3.micro"),
|
||||
},
|
||||
},
|
||||
},
|
||||
writeErr: fmt.Errorf("kv write boom"),
|
||||
}
|
||||
svc := &InstanceServiceImpl{stoppedStore: store}
|
||||
|
||||
_, err := svc.ModifyInstanceAttribute(&ec2.ModifyInstanceAttributeInput{
|
||||
InstanceId: aws.String(id),
|
||||
InstanceType: &ec2.AttributeValue{Value: aws.String("t3.medium")},
|
||||
}, "acc")
|
||||
require.Error(t, err)
|
||||
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
|
||||
}
|
||||
|
||||
@@ -55,3 +55,7 @@ func (s *MockInstanceService) DescribeStoppedInstances(_ *ec2.DescribeInstancesI
|
||||
func (s *MockInstanceService) DescribeTerminatedInstances(_ *ec2.DescribeInstancesInput, _ string) (*ec2.DescribeInstancesOutput, error) {
|
||||
return &ec2.DescribeInstancesOutput{}, nil
|
||||
}
|
||||
|
||||
func (s *MockInstanceService) ModifyInstanceAttribute(_ *ec2.ModifyInstanceAttributeInput, _ string) (*ec2.ModifyInstanceAttributeOutput, error) {
|
||||
return &ec2.ModifyInstanceAttributeOutput{}, nil
|
||||
}
|
||||
|
||||
@@ -60,3 +60,7 @@ func (s *NATSInstanceService) DescribeStoppedInstances(input *ec2.DescribeInstan
|
||||
func (s *NATSInstanceService) DescribeTerminatedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) {
|
||||
return utils.NATSRequest[ec2.DescribeInstancesOutput](s.natsConn, "ec2.DescribeTerminatedInstances", input, 30*time.Second, accountID)
|
||||
}
|
||||
|
||||
func (s *NATSInstanceService) ModifyInstanceAttribute(input *ec2.ModifyInstanceAttributeInput, accountID string) (*ec2.ModifyInstanceAttributeOutput, error) {
|
||||
return utils.NATSRequest[ec2.ModifyInstanceAttributeOutput](s.natsConn, "ec2.ModifyInstanceAttribute", input, 30*time.Second, accountID)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user