refactor: establish EC2 command contract

This commit is contained in:
joshsiv-mulga
2026-09-30 14:46:31 +10:00
parent c0c64cd531
commit aa0c36ff4e
53 changed files with 550 additions and 436 deletions
@@ -1,4 +1,26 @@
package types
// Package ec2v1 defines version 1 of the cross-process EC2 instance-command
// contract. These payloads travel on ec2.cmd.<instance-id>; their JSON fields
// are wire compatibility, not an implementation detail of a caller.
package ec2v1
const (
// InstanceCommandSubjectPrefix is the NATS subject family whose final token
// identifies the instance-owning daemon.
InstanceCommandSubjectPrefix = "ec2.cmd."
// InstanceCommandSubjectWildcard matches exactly one instance-ID token in
// this contract family. It is the NATS permission/subscription pattern for
// the per-instance command route, rather than the broader descendant match.
InstanceCommandSubjectWildcard = InstanceCommandSubjectPrefix + "*"
)
// InstanceCommandSubject returns the targeted command subject for instanceID.
// It intentionally performs no validation: callers have always supplied the
// already-validated instance identifier directly to NATS, and this structural
// move must not change that behaviour.
func InstanceCommandSubject(instanceID string) string {
return InstanceCommandSubjectPrefix + instanceID
}
// EC2InstanceCommand is the NATS wire format for EC2 instance commands
// (stop, terminate, start, attach/detach-volume, attach/detach-eni).
+85
View File
@@ -0,0 +1,85 @@
package ec2v1
import (
"encoding/json"
"testing"
"github.com/stretchr/testify/require"
)
func TestEC2InstanceCommandJSONContract(t *testing.T) {
command := EC2InstanceCommand{
ID: "i-0123456789abcdef0",
Attributes: EC2CommandAttributes{
StopInstance: true,
TerminateInstance: true,
StartInstance: true,
AttachVolume: true,
DetachVolume: true,
DrainVolume: true,
RebootInstance: true,
AttachENI: true,
DetachENI: true,
AssociateIamInstanceProfile: true,
SetSpotLineage: true,
SetInstanceTags: true,
RemoveInstanceTags: true,
SetInstanceMonitoring: true,
},
AttachVolumeData: &AttachVolumeData{VolumeID: "vol-1", Device: "/dev/sdf"},
DetachVolumeData: &DetachVolumeData{VolumeID: "vol-2", Device: "/dev/sdg", Force: true},
DrainVolumeData: &DrainVolumeData{VolumeID: "vol-3"},
AttachENIData: &AttachENIData{NetworkInterfaceID: "eni-1", DeviceIndex: 2},
DetachENIData: &DetachENIData{AttachmentID: "eni-attach-1", Force: true},
IamProfileAssociationData: &IamProfileAssociationData{InstanceProfileArn: "arn:aws:iam::123456789012:instance-profile/example"},
SpotLineageData: &SpotLineageData{SpotInstanceRequestId: "sir-1"},
InstanceTagsData: &InstanceTagsData{Tags: map[string]string{"Name": "example"}, TagKeys: []string{"old"}},
InstanceMonitoringData: &InstanceMonitoringData{Enabled: true},
}
data, err := json.Marshal(command)
require.NoError(t, err)
require.JSONEq(t, `{
"id": "i-0123456789abcdef0",
"attributes": {
"stop_instance": true,
"delete_instance": true,
"start_instance": true,
"attach_volume": true,
"detach_volume": true,
"drain_volume": true,
"reboot_instance": true,
"attach_eni": true,
"detach_eni": true,
"associate_iam_instance_profile": true,
"set_spot_lineage": true,
"set_instance_tags": true,
"remove_instance_tags": true,
"set_instance_monitoring": true
},
"attach_volume_data": {"volume_id": "vol-1", "device": "/dev/sdf"},
"detach_volume_data": {"volume_id": "vol-2", "device": "/dev/sdg", "force": true},
"drain_volume_data": {"volume_id": "vol-3"},
"attach_eni_data": {"network_interface_id": "eni-1", "device_index": 2},
"detach_eni_data": {"attachment_id": "eni-attach-1", "force": true},
"iam_profile_association_data": {"instance_profile_arn": "arn:aws:iam::123456789012:instance-profile/example"},
"spot_lineage_data": {"spot_instance_request_id": "sir-1"},
"instance_tags_data": {"tags": {"Name": "example"}, "tag_keys": ["old"]},
"instance_monitoring_data": {"enabled": true}
}`, string(data))
}
func TestDrainVolumeResponseJSONContract(t *testing.T) {
data, err := json.Marshal(DrainVolumeResponse{
VolumeID: "vol-1",
Status: DrainVolumeStatusDrained,
})
require.NoError(t, err)
require.JSONEq(t, `{"volume_id":"vol-1","status":"drained"}`, string(data))
require.Equal(t, "not-running", DrainVolumeStatusNotRunning)
}
func TestInstanceCommandSubjects(t *testing.T) {
require.Equal(t, "ec2.cmd.i-0123456789abcdef0", InstanceCommandSubject("i-0123456789abcdef0"))
require.Equal(t, "ec2.cmd.*", InstanceCommandSubjectWildcard)
}
+3 -2
View File
@@ -31,11 +31,13 @@ import (
"github.com/go-chi/chi/v5"
"github.com/mulgadc/bluebottle/pkg/masterkey"
"github.com/mulgadc/bluebottle/pkg/tlsconfig"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/admin"
"github.com/mulgadc/spinifex/spinifex/bootstrap/preflight"
"github.com/mulgadc/spinifex/spinifex/config"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/instancetypes"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
handlers_acm "github.com/mulgadc/spinifex/spinifex/handlers/acm"
handlers_bedrock "github.com/mulgadc/spinifex/spinifex/handlers/bedrock"
@@ -63,7 +65,6 @@ import (
handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
handlers_ochrevector "github.com/mulgadc/spinifex/spinifex/handlers/ochrevector"
handlers_rds "github.com/mulgadc/spinifex/spinifex/handlers/rds"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/network/external"
"github.com/mulgadc/spinifex/spinifex/network/external/dhcp"
"github.com/mulgadc/spinifex/spinifex/network/external/ocinet"
@@ -819,7 +820,7 @@ func outcomeForCode(errCode string) string {
// carries the instance id, so the action is built from the command: an instance
// id in a metric dimension would make one series per instance.
func ec2CmdAction(command string) string {
return "ec2.cmd." + command
return ec2v1.InstanceCommandSubjectPrefix + command
}
// logHandlerError reports a handler failure at a level matching its
+6 -5
View File
@@ -14,14 +14,15 @@ import (
"sync"
"time"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/runtime/formation"
"github.com/mulgadc/spinifex/spinifex/network/host"
"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
"github.com/mulgadc/spinifex/spinifex/network/host"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/mulgadc/spinifex/spinifex/runtime/formation"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/nats-io/nats.go"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
@@ -190,7 +191,7 @@ func (d *Daemon) dispatchEC2Command(msg *nats.Msg) (string, string) {
ctx, span := utils.StartConsumerSpan(msg)
defer span.End()
var command types.EC2InstanceCommand
var command ec2v1.EC2InstanceCommand
if err := json.Unmarshal(msg.Data, &command); err != nil {
slog.ErrorContext(ctx, "Error unmarshaling EC2 instance command", "err", err)
@@ -294,7 +295,7 @@ func (d *Daemon) dispatchEC2Command(msg *nats.Msg) (string, string) {
// ec2CommandName names the command an EC2InstanceCommand carries, for the
// metric action. Order matches the dispatch switch.
func ec2CommandName(command types.EC2InstanceCommand) string {
func ec2CommandName(command ec2v1.EC2InstanceCommand) string {
switch {
case command.Attributes.AttachVolume:
return "AttachVolume"
+4 -4
View File
@@ -9,18 +9,18 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
// handleAttachNetworkInterface updates KV first (crash-safe attaching state),
// then runs the QMP hot-plug pipeline. On QMP failure the KV record rolls
// back to available with the error persisted in LastAttachError.
func (d *Daemon) handleAttachNetworkInterface(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) string {
func (d *Daemon) handleAttachNetworkInterface(ctx context.Context, msg *nats.Msg, command ec2v1.EC2InstanceCommand, instance *vm.VM) string {
slog.InfoContext(ctx, "Attaching ENI to instance", "instanceId", command.ID)
if command.AttachENIData == nil || command.AttachENIData.NetworkInterfaceID == "" {
@@ -102,7 +102,7 @@ func (d *Daemon) handleAttachNetworkInterface(ctx context.Context, msg *nats.Msg
// handleDetachNetworkInterface marks the KV record as detaching first
// (crash-safe), runs the QMP hot-unplug pipeline, then returns the
// record to available on success.
func (d *Daemon) handleDetachNetworkInterface(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) string {
func (d *Daemon) handleDetachNetworkInterface(ctx context.Context, msg *nats.Msg, command ec2v1.EC2InstanceCommand, instance *vm.VM) string {
slog.InfoContext(ctx, "Detaching ENI from instance", "instanceId", command.ID)
if command.DetachENIData == nil || command.DetachENIData.AttachmentID == "" {
+30 -29
View File
@@ -10,12 +10,13 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
spxtypes "github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -165,7 +166,7 @@ func driveHandler(t *testing.T, nc *nats.Conn, subject string, handler func(*nat
func TestHandleAttachNetworkInterface_NilData(t *testing.T) {
f := newENIHotPlugFixture(t)
cmd := spxtypes.EC2InstanceCommand{ID: f.vmInst.ID}
cmd := ec2v1.EC2InstanceCommand{ID: f.vmInst.ID}
payload := driveHandler(t, f.daemon.natsConn, "test.attach.nildata", func(msg *nats.Msg) {
f.daemon.handleAttachNetworkInterface(context.Background(), msg, cmd, f.vmInst)
})
@@ -174,9 +175,9 @@ func TestHandleAttachNetworkInterface_NilData(t *testing.T) {
func TestHandleAttachNetworkInterface_EmptyEniID(t *testing.T) {
f := newENIHotPlugFixture(t)
cmd := spxtypes.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: f.vmInst.ID,
AttachENIData: &spxtypes.AttachENIData{NetworkInterfaceID: "", DeviceIndex: 0},
AttachENIData: &ec2v1.AttachENIData{NetworkInterfaceID: "", DeviceIndex: 0},
}
payload := driveHandler(t, f.daemon.natsConn, "test.attach.emptyid", func(msg *nats.Msg) {
f.daemon.handleAttachNetworkInterface(context.Background(), msg, cmd, f.vmInst)
@@ -187,9 +188,9 @@ func TestHandleAttachNetworkInterface_EmptyEniID(t *testing.T) {
func TestHandleAttachNetworkInterface_NotRunning(t *testing.T) {
f := newENIHotPlugFixture(t)
f.vmInst.Status = vm.StateStopped
cmd := spxtypes.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: f.vmInst.ID,
AttachENIData: &spxtypes.AttachENIData{NetworkInterfaceID: f.eniID, DeviceIndex: 1},
AttachENIData: &ec2v1.AttachENIData{NetworkInterfaceID: f.eniID, DeviceIndex: 1},
}
payload := driveHandler(t, f.daemon.natsConn, "test.attach.notrunning", func(msg *nats.Msg) {
f.daemon.handleAttachNetworkInterface(context.Background(), msg, cmd, f.vmInst)
@@ -199,9 +200,9 @@ func TestHandleAttachNetworkInterface_NotRunning(t *testing.T) {
func TestHandleAttachNetworkInterface_ENINotFound(t *testing.T) {
f := newENIHotPlugFixture(t)
cmd := spxtypes.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: f.vmInst.ID,
AttachENIData: &spxtypes.AttachENIData{NetworkInterfaceID: "eni-missing", DeviceIndex: 1},
AttachENIData: &ec2v1.AttachENIData{NetworkInterfaceID: "eni-missing", DeviceIndex: 1},
}
payload := driveHandler(t, f.daemon.natsConn, "test.attach.eninotfound", func(msg *nats.Msg) {
f.daemon.handleAttachNetworkInterface(context.Background(), msg, cmd, f.vmInst)
@@ -214,9 +215,9 @@ func TestHandleAttachNetworkInterface_WrongOwner(t *testing.T) {
_, err := f.daemon.vpcService.AttachENI(testAccountID, f.eniID, "i-other-instance", 0)
require.NoError(t, err)
cmd := spxtypes.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: f.vmInst.ID,
AttachENIData: &spxtypes.AttachENIData{NetworkInterfaceID: f.eniID, DeviceIndex: 1},
AttachENIData: &ec2v1.AttachENIData{NetworkInterfaceID: f.eniID, DeviceIndex: 1},
}
payload := driveHandler(t, f.daemon.natsConn, "test.attach.wrongowner", func(msg *nats.Msg) {
f.daemon.handleAttachNetworkInterface(context.Background(), msg, cmd, f.vmInst)
@@ -234,9 +235,9 @@ func TestHandleAttachNetworkInterface_Success(t *testing.T) {
require.NoError(t, err)
t.Cleanup(func() { _ = evtSub.Unsubscribe() })
cmd := spxtypes.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: f.vmInst.ID,
AttachENIData: &spxtypes.AttachENIData{NetworkInterfaceID: f.eniID, DeviceIndex: 1},
AttachENIData: &ec2v1.AttachENIData{NetworkInterfaceID: f.eniID, DeviceIndex: 1},
}
payload := driveHandler(t, f.daemon.natsConn, "test.attach.success", func(msg *nats.Msg) {
f.daemon.handleAttachNetworkInterface(context.Background(), msg, cmd, f.vmInst)
@@ -271,9 +272,9 @@ func TestHandleAttachNetworkInterface_HotPlugFails_RollsBackKV(t *testing.T) {
f := newENIHotPlugFixture(t)
f.stub.SetFailNext("device_add", errors.New("simulated QMP failure"))
cmd := spxtypes.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: f.vmInst.ID,
AttachENIData: &spxtypes.AttachENIData{NetworkInterfaceID: f.eniID, DeviceIndex: 1},
AttachENIData: &ec2v1.AttachENIData{NetworkInterfaceID: f.eniID, DeviceIndex: 1},
}
payload := driveHandler(t, f.daemon.natsConn, "test.attach.hotplugfails", func(msg *nats.Msg) {
f.daemon.handleAttachNetworkInterface(context.Background(), msg, cmd, f.vmInst)
@@ -315,7 +316,7 @@ func attachedFixture(t *testing.T) (*eniHotPlugFixture, string) {
func TestHandleDetachNetworkInterface_NilData(t *testing.T) {
f, _ := attachedFixture(t)
cmd := spxtypes.EC2InstanceCommand{ID: f.vmInst.ID}
cmd := ec2v1.EC2InstanceCommand{ID: f.vmInst.ID}
payload := driveHandler(t, f.daemon.natsConn, "test.detach.nildata", func(msg *nats.Msg) {
f.daemon.handleDetachNetworkInterface(context.Background(), msg, cmd, f.vmInst)
})
@@ -324,9 +325,9 @@ func TestHandleDetachNetworkInterface_NilData(t *testing.T) {
func TestHandleDetachNetworkInterface_EmptyAttachID(t *testing.T) {
f, _ := attachedFixture(t)
cmd := spxtypes.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: f.vmInst.ID,
DetachENIData: &spxtypes.DetachENIData{AttachmentID: ""},
DetachENIData: &ec2v1.DetachENIData{AttachmentID: ""},
}
payload := driveHandler(t, f.daemon.natsConn, "test.detach.emptyid", func(msg *nats.Msg) {
f.daemon.handleDetachNetworkInterface(context.Background(), msg, cmd, f.vmInst)
@@ -336,9 +337,9 @@ func TestHandleDetachNetworkInterface_EmptyAttachID(t *testing.T) {
func TestHandleDetachNetworkInterface_UnknownAttachment(t *testing.T) {
f, _ := attachedFixture(t)
cmd := spxtypes.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: f.vmInst.ID,
DetachENIData: &spxtypes.DetachENIData{AttachmentID: "eni-attach-missing"},
DetachENIData: &ec2v1.DetachENIData{AttachmentID: "eni-attach-missing"},
}
payload := driveHandler(t, f.daemon.natsConn, "test.detach.unknown", func(msg *nats.Msg) {
f.daemon.handleDetachNetworkInterface(context.Background(), msg, cmd, f.vmInst)
@@ -357,9 +358,9 @@ func TestHandleDetachNetworkInterface_WrongOwner(t *testing.T) {
},
}
f.daemon.vmMgr.Insert(otherVM)
cmd := spxtypes.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: otherVM.ID,
DetachENIData: &spxtypes.DetachENIData{AttachmentID: attachID},
DetachENIData: &ec2v1.DetachENIData{AttachmentID: attachID},
}
payload := driveHandler(t, f.daemon.natsConn, "test.detach.wrongowner", func(msg *nats.Msg) {
f.daemon.handleDetachNetworkInterface(context.Background(), msg, cmd, otherVM)
@@ -370,9 +371,9 @@ func TestHandleDetachNetworkInterface_WrongOwner(t *testing.T) {
func TestHandleDetachNetworkInterface_NotRunning(t *testing.T) {
f, attachID := attachedFixture(t)
f.vmInst.Status = vm.StateStopped
cmd := spxtypes.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: f.vmInst.ID,
DetachENIData: &spxtypes.DetachENIData{AttachmentID: attachID},
DetachENIData: &ec2v1.DetachENIData{AttachmentID: attachID},
}
payload := driveHandler(t, f.daemon.natsConn, "test.detach.notrunning", func(msg *nats.Msg) {
f.daemon.handleDetachNetworkInterface(context.Background(), msg, cmd, f.vmInst)
@@ -390,9 +391,9 @@ func TestHandleDetachNetworkInterface_Success(t *testing.T) {
require.NoError(t, err)
t.Cleanup(func() { _ = evtSub.Unsubscribe() })
cmd := spxtypes.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: f.vmInst.ID,
DetachENIData: &spxtypes.DetachENIData{AttachmentID: attachID, Force: false},
DetachENIData: &ec2v1.DetachENIData{AttachmentID: attachID, Force: false},
}
payload := driveHandler(t, f.daemon.natsConn, "test.detach.success", func(msg *nats.Msg) {
f.daemon.handleDetachNetworkInterface(context.Background(), msg, cmd, f.vmInst)
@@ -421,9 +422,9 @@ func TestHandleDetachNetworkInterface_HotUnplugFails(t *testing.T) {
f, attachID := attachedFixture(t)
f.stub.SetFailNext("device_del", errors.New("simulated QMP failure"))
cmd := spxtypes.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: f.vmInst.ID,
DetachENIData: &spxtypes.DetachENIData{AttachmentID: attachID, Force: false},
DetachENIData: &ec2v1.DetachENIData{AttachmentID: attachID, Force: false},
}
payload := driveHandler(t, f.daemon.natsConn, "test.detach.hotunplugfails", func(msg *nats.Msg) {
f.daemon.handleDetachNetworkInterface(context.Background(), msg, cmd, f.vmInst)
@@ -452,9 +453,9 @@ func TestHandleAttachNetworkInterface_DeviceIndexIsTheAllocatedSlot(t *testing.T
// Both attaches ask for device index 1, which is what the ECS task ENI path
// sends for every task.
for _, eniID := range []string{f.eniID, secondID} {
cmd := spxtypes.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: f.vmInst.ID,
AttachENIData: &spxtypes.AttachENIData{NetworkInterfaceID: eniID, DeviceIndex: 1},
AttachENIData: &ec2v1.AttachENIData{NetworkInterfaceID: eniID, DeviceIndex: 1},
}
driveHandler(t, f.daemon.natsConn, "test.attach.slot."+eniID, func(msg *nats.Msg) {
f.daemon.handleAttachNetworkInterface(context.Background(), msg, cmd, f.vmInst)
@@ -3,7 +3,7 @@ package daemon
import (
"context"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/nats-io/nats.go"
)
@@ -11,7 +11,7 @@ import (
// handleAssociateIamInstanceProfile services the per-instance Associate path.
// Gateway pre-validates the profile reference and iam:PassRole; ownership is
// checked by checkInstanceOwnership before dispatch.
func (d *Daemon) handleAssociateIamInstanceProfile(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) string {
func (d *Daemon) handleAssociateIamInstanceProfile(ctx context.Context, msg *nats.Msg, command ec2v1.EC2InstanceCommand, instance *vm.VM) string {
result, err := d.instanceService.AssociateIamInstanceProfile(ctx, instance, command)
if err != nil {
return respondServiceErrorOutcome(d.node, msg, err)
+5 -5
View File
@@ -10,11 +10,11 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
@@ -31,7 +31,7 @@ var startStoppedForwardTimeout = 30 * time.Second
// instance: central store first, then the record under the manager lock, so a
// failed S3 write leaves both stores untouched, matching the stopped path.
// Ownership is checked by checkInstanceOwnership before dispatch.
func (d *Daemon) handleSetInstanceTags(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) string {
func (d *Daemon) handleSetInstanceTags(ctx context.Context, msg *nats.Msg, command ec2v1.EC2InstanceCommand, instance *vm.VM) string {
remove := command.Attributes.RemoveInstanceTags
data := command.InstanceTagsData
if data == nil || (!remove && len(data.Tags) == 0) {
@@ -82,7 +82,7 @@ func (d *Daemon) handleSetInstanceTags(ctx context.Context, msg *nats.Msg, comma
// and rewrites its collector discovery file, so the poller resets its interval
// without a restart. Ownership is checked by checkInstanceOwnership before
// dispatch.
func (d *Daemon) handleSetInstanceMonitoring(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) string {
func (d *Daemon) handleSetInstanceMonitoring(ctx context.Context, msg *nats.Msg, command ec2v1.EC2InstanceCommand, instance *vm.VM) string {
data := command.InstanceMonitoringData
if data == nil {
return respondErrorOutcome(d.node, msg, awserrors.ErrorMissingParameter)
@@ -242,7 +242,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) string {
// than finding no responder. LaunchInstance replaces these on success.
d.mu.Lock()
for _, instance := range instances {
sub, subErr := d.natsConn.Subscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), d.handleEC2Events)
sub, subErr := d.natsConn.Subscribe(ec2v1.InstanceCommandSubject(instance.ID), d.handleEC2Events)
if subErr != nil {
slog.Error("Failed to early-subscribe to per-instance topic", "instanceId", instance.ID, "err", subErr)
} else {
@@ -7,11 +7,11 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
handlers_ec2_tags "github.com/mulgadc/spinifex/spinifex/handlers/ec2/tags"
"github.com/mulgadc/spinifex/spinifex/objectstore"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
vmmock "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm/mock"
"github.com/stretchr/testify/assert"
@@ -80,8 +80,8 @@ func tagsAsMap(tags []*ec2.Tag) map[string]string {
return out
}
func tagCommand(instanceID string, attrs types.EC2CommandAttributes, data *types.InstanceTagsData) []byte {
body, _ := json.Marshal(types.EC2InstanceCommand{ID: instanceID, Attributes: attrs, InstanceTagsData: data})
func tagCommand(instanceID string, attrs ec2v1.EC2CommandAttributes, data *ec2v1.InstanceTagsData) []byte {
body, _ := json.Marshal(ec2v1.EC2InstanceCommand{ID: instanceID, Attributes: attrs, InstanceTagsData: data})
return body
}
@@ -91,8 +91,8 @@ func TestHandleSetInstanceTags_MergesAndWritesCentral(t *testing.T) {
const id = "i-tag-set"
d := tagTestDaemon(t, id, map[string]string{"Name": "web", "env": "dev"})
body := tagCommand(id, types.EC2CommandAttributes{SetInstanceTags: true},
&types.InstanceTagsData{Tags: map[string]string{"env": "prod", "team": "infra"}})
body := tagCommand(id, ec2v1.EC2CommandAttributes{SetInstanceTags: true},
&ec2v1.InstanceTagsData{Tags: map[string]string{"env": "prod", "team": "infra"}})
reply := requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, testAccountID, body)
assert.JSONEq(t, `{}`, string(reply.Data))
@@ -109,8 +109,8 @@ func TestHandleSetInstanceTags_RemoveWritesBothStores(t *testing.T) {
require.NoError(t, d.tagsService.PutResourceTags(t.Context(), testAccountID, id,
map[string]string{"Name": "web", "env": "dev", "team": "infra"}))
body := tagCommand(id, types.EC2CommandAttributes{RemoveInstanceTags: true},
&types.InstanceTagsData{TagKeys: []string{"team"}, Tags: map[string]string{"env": "prod", "Name": "web"}})
body := tagCommand(id, ec2v1.EC2CommandAttributes{RemoveInstanceTags: true},
&ec2v1.InstanceTagsData{TagKeys: []string{"team"}, Tags: map[string]string{"env": "prod", "Name": "web"}})
reply := requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, testAccountID, body)
assert.JSONEq(t, `{}`, string(reply.Data))
@@ -126,7 +126,7 @@ func TestHandleSetInstanceTags_RemoveClearAll(t *testing.T) {
require.NoError(t, d.tagsService.PutResourceTags(t.Context(), testAccountID, id,
map[string]string{"Name": "web", "env": "dev"}))
body := tagCommand(id, types.EC2CommandAttributes{RemoveInstanceTags: true}, &types.InstanceTagsData{})
body := tagCommand(id, ec2v1.EC2CommandAttributes{RemoveInstanceTags: true}, &ec2v1.InstanceTagsData{})
reply := requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, testAccountID, body)
assert.JSONEq(t, `{}`, string(reply.Data))
@@ -141,8 +141,8 @@ func TestHandleSetInstanceTags_CrossAccountRejected(t *testing.T) {
const attacker = "999999999999"
d := tagTestDaemon(t, id, map[string]string{"Name": "web"})
body := tagCommand(id, types.EC2CommandAttributes{SetInstanceTags: true},
&types.InstanceTagsData{Tags: map[string]string{"stolen": "yes"}})
body := tagCommand(id, ec2v1.EC2CommandAttributes{SetInstanceTags: true},
&ec2v1.InstanceTagsData{Tags: map[string]string{"stolen": "yes"}})
reply := requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, attacker, body)
assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, decodeError(t, reply.Data)["Code"])
@@ -196,8 +196,8 @@ func TestHandleSetInstanceTags_RejectsMissingData(t *testing.T) {
const id = "i-tag-nodata"
d := tagTestDaemon(t, id, map[string]string{"Name": "web"})
for _, data := range []*types.InstanceTagsData{nil, {}} {
body := tagCommand(id, types.EC2CommandAttributes{SetInstanceTags: true}, data)
for _, data := range []*ec2v1.InstanceTagsData{nil, {}} {
body := tagCommand(id, ec2v1.EC2CommandAttributes{SetInstanceTags: true}, data)
reply := requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, testAccountID, body)
assert.Equal(t, awserrors.ErrorMissingParameter, decodeError(t, reply.Data)["Code"])
}
@@ -9,8 +9,8 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -86,10 +86,10 @@ func (d *Daemon) setInstanceMonitoring(ctx context.Context, instanceIDs []*strin
// isn't running, so it falls back to the shared stopped store; a timeout
// (partitioned-but-subscribed owner) surfaces InvalidID.NotFound.
func (d *Daemon) setOneInstanceMonitoring(ctx context.Context, instanceID string, enabled bool, accountID string) error {
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{SetInstanceMonitoring: true},
InstanceMonitoringData: &types.InstanceMonitoringData{Enabled: enabled},
Attributes: ec2v1.EC2CommandAttributes{SetInstanceMonitoring: true},
InstanceMonitoringData: &ec2v1.InstanceMonitoringData{Enabled: enabled},
}
body, err := json.Marshal(command)
if err != nil {
@@ -97,7 +97,7 @@ func (d *Daemon) setOneInstanceMonitoring(ctx context.Context, instanceID string
return errors.New(awserrors.ErrorServerInternal)
}
reqMsg := nats.NewMsg("ec2.cmd." + instanceID)
reqMsg := nats.NewMsg(ec2v1.InstanceCommandSubject(instanceID))
reqMsg.Data = body
reqMsg.Header.Set(utils.AccountIDHeader, accountID)
utils.InjectTraceContext(ctx, reqMsg.Header)
@@ -5,8 +5,8 @@ import (
"log/slog"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/nats-io/nats.go"
)
@@ -15,7 +15,7 @@ import (
// SpotInstanceRequestId) onto a launched VM. Dispatched from handleEC2Events,
// which already verified ownership; the SIR id is the only datum on the wire as
// the lifecycle is always spot for this command.
func (d *Daemon) handleSetSpotLineage(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand) string {
func (d *Daemon) handleSetSpotLineage(ctx context.Context, msg *nats.Msg, command ec2v1.EC2InstanceCommand) string {
if command.SpotLineageData == nil {
return respondErrorOutcome(d.node, msg, awserrors.ErrorMissingParameter)
}
@@ -6,8 +6,8 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -25,10 +25,10 @@ func TestHandleSetSpotLineage_Stamps(t *testing.T) {
Instance: &ec2.Instance{InstanceId: aws.String("i-spot-stamp")},
})
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: "i-spot-stamp",
Attributes: types.EC2CommandAttributes{SetSpotLineage: true},
SpotLineageData: &types.SpotLineageData{SpotInstanceRequestId: "sir-abc123"},
Attributes: ec2v1.EC2CommandAttributes{SetSpotLineage: true},
SpotLineageData: &ec2v1.SpotLineageData{SpotInstanceRequestId: "sir-abc123"},
}
body, _ := json.Marshal(command)
reply := requestHandler(t, d.natsConn, "ec2.cmd.i-spot-stamp", d.handleEC2Events, testAccountID, body)
@@ -51,9 +51,9 @@ func TestHandleSetSpotLineage_RejectsMissingData(t *testing.T) {
Instance: &ec2.Instance{InstanceId: aws.String("i-spot-nodata")},
})
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: "i-spot-nodata",
Attributes: types.EC2CommandAttributes{SetSpotLineage: true},
Attributes: ec2v1.EC2CommandAttributes{SetSpotLineage: true},
}
body, _ := json.Marshal(command)
reply := requestHandler(t, d.natsConn, "ec2.cmd.i-spot-nodata", d.handleEC2Events, testAccountID, body)
+8 -8
View File
@@ -10,8 +10,8 @@ import (
"time"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -108,7 +108,7 @@ func (d *Daemon) createTags(ctx context.Context, input *ec2.CreateTagsInput, acc
}
}
data := &types.InstanceTagsData{Tags: make(map[string]string, len(input.Tags))}
data := &ec2v1.InstanceTagsData{Tags: make(map[string]string, len(input.Tags))}
for _, tag := range input.Tags {
if tag != nil && tag.Key != nil && tag.Value != nil {
data.Tags[*tag.Key] = *tag.Value
@@ -146,7 +146,7 @@ func (d *Daemon) deleteTags(ctx context.Context, input *ec2.DeleteTagsInput, acc
// Keys without a value delete unconditionally; keys with a value delete
// only on match; an empty input.Tags clears every tag.
data := &types.InstanceTagsData{}
data := &ec2v1.InstanceTagsData{}
for _, tag := range input.Tags {
if tag == nil || tag.Key == nil {
continue
@@ -169,7 +169,7 @@ func (d *Daemon) deleteTags(ctx context.Context, input *ec2.DeleteTagsInput, acc
// tagInstances sends the tag mutation to each instance's owner concurrently,
// so a many-instance call costs one owner timeout rather than one per
// instance. The first failure in input order is returned.
func (d *Daemon) tagInstances(ctx context.Context, instanceIDs []string, data *types.InstanceTagsData, remove bool, accountID string) error {
func (d *Daemon) tagInstances(ctx context.Context, instanceIDs []string, data *ec2v1.InstanceTagsData, remove bool, accountID string) error {
errs := make([]error, len(instanceIDs))
var wg sync.WaitGroup
for i, id := range instanceIDs {
@@ -191,10 +191,10 @@ func (d *Daemon) tagInstances(ctx context.Context, instanceIDs []string, data *t
// No responders means the instance isn't running, so the mutation falls back
// to the shared stopped store; a timeout (partitioned-but-subscribed owner)
// surfaces InvalidID.NotFound and writes nothing.
func (d *Daemon) tagInstance(ctx context.Context, instanceID string, data *types.InstanceTagsData, remove bool, accountID string) error {
command := types.EC2InstanceCommand{
func (d *Daemon) tagInstance(ctx context.Context, instanceID string, data *ec2v1.InstanceTagsData, remove bool, accountID string) error {
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
SetInstanceTags: !remove,
RemoveInstanceTags: remove,
},
@@ -206,7 +206,7 @@ func (d *Daemon) tagInstance(ctx context.Context, instanceID string, data *types
return errors.New(awserrors.ErrorServerInternal)
}
reqMsg := nats.NewMsg("ec2.cmd." + instanceID)
reqMsg := nats.NewMsg(ec2v1.InstanceCommandSubject(instanceID))
reqMsg.Data = body
reqMsg.Header.Set(utils.AccountIDHeader, accountID)
+46 -45
View File
@@ -17,8 +17,11 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
awss3 "github.com/aws/aws-sdk-go/service/s3"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/internal/testkit/ebsfake"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
handlers_ec2_account "github.com/mulgadc/spinifex/spinifex/handlers/ec2/account"
handlers_ec2_eigw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eigw"
handlers_ec2_eip "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eip"
@@ -35,12 +38,10 @@ import (
"github.com/mulgadc/spinifex/spinifex/network/external"
"github.com/mulgadc/spinifex/spinifex/objectstore"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/internal/testkit/ebsfake"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
vmmock "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm/mock"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
@@ -710,9 +711,9 @@ func TestHandleEC2Events_StopInstance(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
cmd := types.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{StopInstance: true},
Attributes: ec2v1.EC2CommandAttributes{StopInstance: true},
}
cmdData, _ := json.Marshal(cmd)
@@ -767,9 +768,9 @@ func TestHandleEC2Events_TerminateInstance(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
cmd := types.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{TerminateInstance: true},
Attributes: ec2v1.EC2CommandAttributes{TerminateInstance: true},
}
cmdData, _ := json.Marshal(cmd)
@@ -808,9 +809,9 @@ func TestHandleEC2Events_RebootRunningInstance(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
cmd := types.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{RebootInstance: true},
Attributes: ec2v1.EC2CommandAttributes{RebootInstance: true},
}
cmdData, _ := json.Marshal(cmd)
@@ -859,9 +860,9 @@ func TestHandleEC2Events_RebootStoppedInstance(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
cmd := types.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{RebootInstance: true},
Attributes: ec2v1.EC2CommandAttributes{RebootInstance: true},
}
cmdData, _ := json.Marshal(cmd)
@@ -903,9 +904,9 @@ func TestHandleEC2Events_RebootTerminatedInstance(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
cmd := types.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{RebootInstance: true},
Attributes: ec2v1.EC2CommandAttributes{RebootInstance: true},
}
cmdData, _ := json.Marshal(cmd)
@@ -934,9 +935,9 @@ func TestHandleEC2Events_InstanceNotFound(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
cmd := types.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: "i-nonexistent",
Attributes: types.EC2CommandAttributes{StopInstance: true},
Attributes: ec2v1.EC2CommandAttributes{StopInstance: true},
}
cmdData, _ := json.Marshal(cmd)
@@ -1525,12 +1526,12 @@ func TestAttachVolume_ZoneMismatch(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
AttachVolume: true,
},
AttachVolumeData: &types.AttachVolumeData{
AttachVolumeData: &ec2v1.AttachVolumeData{
VolumeID: volumeID,
},
}
@@ -2864,9 +2865,9 @@ func TestAttachVolume_MissingVolumeData(t *testing.T) {
defer sub.Unsubscribe()
// AttachVolume with nil AttachVolumeData
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
AttachVolume: true,
},
AttachVolumeData: nil,
@@ -2903,12 +2904,12 @@ func TestAttachVolume_InstanceNotRunning(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
AttachVolume: true,
},
AttachVolumeData: &types.AttachVolumeData{
AttachVolumeData: &ec2v1.AttachVolumeData{
VolumeID: "vol-test-123",
},
}
@@ -2944,12 +2945,12 @@ func TestAttachVolume_VolumeNotFound(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
AttachVolume: true,
},
AttachVolumeData: &types.AttachVolumeData{
AttachVolumeData: &ec2v1.AttachVolumeData{
VolumeID: "vol-nonexistent-999",
},
}
@@ -2995,12 +2996,12 @@ func TestAttachVolume_VolumeInUse(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
AttachVolume: true,
},
AttachVolumeData: &types.AttachVolumeData{
AttachVolumeData: &ec2v1.AttachVolumeData{
VolumeID: volumeID,
},
}
@@ -3038,9 +3039,9 @@ func TestDetachVolume_MissingVolumeData(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
DetachVolume: true,
},
DetachVolumeData: nil,
@@ -3077,12 +3078,12 @@ func TestDetachVolume_InstanceNotRunning(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
DetachVolume: true,
},
DetachVolumeData: &types.DetachVolumeData{
DetachVolumeData: &ec2v1.DetachVolumeData{
VolumeID: "vol-test-123",
},
}
@@ -3118,12 +3119,12 @@ func TestDetachVolume_VolumeNotAttached(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
DetachVolume: true,
},
DetachVolumeData: &types.DetachVolumeData{
DetachVolumeData: &ec2v1.DetachVolumeData{
VolumeID: "vol-not-attached-999",
},
}
@@ -3162,12 +3163,12 @@ func TestDetachVolume_BootVolumeRejected(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
DetachVolume: true,
},
DetachVolumeData: &types.DetachVolumeData{
DetachVolumeData: &ec2v1.DetachVolumeData{
VolumeID: "vol-boot-001",
},
}
@@ -3206,12 +3207,12 @@ func TestDetachVolume_DeviceMismatch(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
DetachVolume: true,
},
DetachVolumeData: &types.DetachVolumeData{
DetachVolumeData: &ec2v1.DetachVolumeData{
VolumeID: "vol-mismatch-001",
Device: "/dev/sdg",
},
+7 -6
View File
@@ -8,11 +8,12 @@ import (
"time"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
handlers_ec2_snapshot "github.com/mulgadc/spinifex/spinifex/handlers/ec2/snapshot"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/nats-io/nats.go"
)
@@ -21,7 +22,7 @@ import (
// QMP/state-machine pipeline to vm.Manager.AttachVolume. The manager owns
// every QMP and persistence side-effect; the daemon only emits the AWS API
// response.
func (d *Daemon) handleAttachVolume(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) string {
func (d *Daemon) handleAttachVolume(ctx context.Context, msg *nats.Msg, command ec2v1.EC2InstanceCommand, instance *vm.VM) string {
slog.InfoContext(ctx, "Attaching volume to instance", "instanceId", command.ID)
if command.AttachVolumeData == nil || command.AttachVolumeData.VolumeID == "" {
@@ -115,7 +116,7 @@ func (d *Daemon) handleAttachVolume(ctx context.Context, msg *nats.Msg, command
// handleDetachVolume dispatches the QMP/state-machine pipeline to
// vm.Manager.DetachVolume and emits the AWS API response.
func (d *Daemon) handleDetachVolume(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) string {
func (d *Daemon) handleDetachVolume(ctx context.Context, msg *nats.Msg, command ec2v1.EC2InstanceCommand, instance *vm.VM) string {
slog.InfoContext(ctx, "Detaching volume from instance", "instanceId", command.ID)
if command.DetachVolumeData == nil || command.DetachVolumeData.VolumeID == "" {
@@ -145,7 +146,7 @@ func (d *Daemon) handleDetachVolume(ctx context.Context, msg *nats.Msg, command
//
// Runs on its own goroutine (see handleEC2Events) so a long flush cannot hold
// the instance's command subscription.
func (d *Daemon) handleDrainVolume(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) string {
func (d *Daemon) handleDrainVolume(ctx context.Context, msg *nats.Msg, command ec2v1.EC2InstanceCommand, instance *vm.VM) string {
ctx, span := startOpSpan(ctx, "ec2.DrainVolume", command.ID)
var err error
defer func() { endOpSpan(span, err) }()
@@ -164,7 +165,7 @@ func (d *Daemon) handleDrainVolume(ctx context.Context, msg *nats.Msg, command t
if status == vm.StateStopped || status == vm.StateTerminated {
slog.InfoContext(ctx, "DrainVolume: instance teardown is complete, nothing to drain",
"volumeId", volumeID, "instanceId", command.ID, "status", status)
respondWithJSON(d.node, msg, types.DrainVolumeResponse{VolumeID: volumeID, Status: types.DrainVolumeStatusNotRunning})
respondWithJSON(d.node, msg, ec2v1.DrainVolumeResponse{VolumeID: volumeID, Status: ec2v1.DrainVolumeStatusNotRunning})
return outcomeSuccess
}
@@ -178,7 +179,7 @@ func (d *Daemon) handleDrainVolume(ctx context.Context, msg *nats.Msg, command t
return respondErrorOutcome(d.node, msg, awserrors.ErrorServerInternal)
}
respondWithJSON(d.node, msg, types.DrainVolumeResponse{VolumeID: volumeID, Status: types.DrainVolumeStatusDrained})
respondWithJSON(d.node, msg, ec2v1.DrainVolumeResponse{VolumeID: volumeID, Status: ec2v1.DrainVolumeStatusDrained})
return outcomeSuccess
}
+31 -31
View File
@@ -10,15 +10,15 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/bluebottle/pkg/safecast"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
"github.com/mulgadc/spinifex/spinifex/providers/ebs"
"github.com/mulgadc/spinifex/spinifex/objectstore"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/objectstore"
"github.com/mulgadc/spinifex/spinifex/providers/ebs"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -159,12 +159,12 @@ func TestAttachVolume_IdempotentSameInstance(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
AttachVolume: true,
},
AttachVolumeData: &types.AttachVolumeData{
AttachVolumeData: &ec2v1.AttachVolumeData{
VolumeID: volumeID,
Device: tt.requestedDevice,
},
@@ -227,12 +227,12 @@ func TestAttachVolume_InUseDifferentInstance(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
AttachVolume: true,
},
AttachVolumeData: &types.AttachVolumeData{
AttachVolumeData: &ec2v1.AttachVolumeData{
VolumeID: volumeID,
},
}
@@ -284,12 +284,12 @@ func TestAttachVolume_IdempotentSameInstance_DeviceMismatch(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
AttachVolume: true,
},
AttachVolumeData: &types.AttachVolumeData{
AttachVolumeData: &ec2v1.AttachVolumeData{
VolumeID: volumeID,
Device: "/dev/sdg",
},
@@ -310,10 +310,10 @@ func TestAttachVolume_IdempotentSameInstance_DeviceMismatch(t *testing.T) {
// hosting instanceID.
func drainCommandFor(t *testing.T, instanceID, volumeID string) []byte {
t.Helper()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{DrainVolume: true},
DrainVolumeData: &types.DrainVolumeData{VolumeID: volumeID},
Attributes: ec2v1.EC2CommandAttributes{DrainVolume: true},
DrainVolumeData: &ec2v1.DrainVolumeData{VolumeID: volumeID},
}
data, err := json.Marshal(command)
require.NoError(t, err)
@@ -381,10 +381,10 @@ func TestDrainVolume_HostNodeAcksLocalSocket(t *testing.T) {
resp := requestHandler(t, daemon.natsConn, "ec2.cmd."+instanceID, daemon.handleEC2Events,
testAccountID, drainCommandFor(t, instanceID, volumeID))
var ack types.DrainVolumeResponse
var ack ec2v1.DrainVolumeResponse
require.NoError(t, json.Unmarshal(resp.Data, &ack))
assert.Equal(t, volumeID, ack.VolumeID)
assert.Equal(t, types.DrainVolumeStatusDrained, ack.Status)
assert.Equal(t, ec2v1.DrainVolumeStatusDrained, ack.Status)
}
// The owning node having no socket for the volume means the writes cannot be
@@ -417,9 +417,9 @@ func TestDrainVolume_MissingVolumeDataIsInvalidParameter(t *testing.T) {
const instanceID = "i-drain-novol"
daemon := drainTestDaemon(t, instanceID)
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{DrainVolume: true},
Attributes: ec2v1.EC2CommandAttributes{DrainVolume: true},
}
data, err := json.Marshal(command)
require.NoError(t, err)
@@ -455,10 +455,10 @@ func TestDrainVolume_CompletedTeardownAcksNotRunning(t *testing.T) {
resp := requestHandler(t, daemon.natsConn, "ec2.cmd."+instanceID, daemon.handleEC2Events,
testAccountID, drainCommandFor(t, instanceID, volumeID))
var ack types.DrainVolumeResponse
var ack ec2v1.DrainVolumeResponse
require.NoError(t, json.Unmarshal(resp.Data, &ack))
assert.Equal(t, volumeID, ack.VolumeID)
assert.Equal(t, types.DrainVolumeStatusNotRunning, ack.Status)
assert.Equal(t, ec2v1.DrainVolumeStatusNotRunning, ack.Status)
})
}
}
@@ -519,16 +519,16 @@ func TestDrainVolume_SlowDrainDoesNotBlockTheCommandSubject(t *testing.T) {
}
fast := drainRequest(t, daemon, instanceID, "vol-drain-fast")
var ack types.DrainVolumeResponse
var ack ec2v1.DrainVolumeResponse
require.NoError(t, json.Unmarshal(fast.Data, &ack))
assert.Equal(t, types.DrainVolumeStatusDrained, ack.Status)
assert.Equal(t, ec2v1.DrainVolumeStatusDrained, ack.Status)
release()
select {
case got := <-slow:
require.NoError(t, got.err)
require.NoError(t, json.Unmarshal(got.msg.Data, &ack))
assert.Equal(t, types.DrainVolumeStatusDrained, ack.Status)
assert.Equal(t, ec2v1.DrainVolumeStatusDrained, ack.Status)
case <-time.After(5 * time.Second):
t.Fatal("the released drain never replied")
}
@@ -546,13 +546,13 @@ func TestDrainVolume_DispatchedGoroutineDoesNotLeak(t *testing.T) {
// Take the baseline after one full round trip: the first request is what
// makes nats.go stand up its long-lived response-inbox goroutine, which
// would otherwise read as the leak.
var ack types.DrainVolumeResponse
var ack ec2v1.DrainVolumeResponse
require.NoError(t, json.Unmarshal(drainRequest(t, daemon, instanceID, volumeID).Data, &ack))
require.Equal(t, types.DrainVolumeStatusDrained, ack.Status)
require.Equal(t, ec2v1.DrainVolumeStatusDrained, ack.Status)
ignoreExisting := goleak.IgnoreCurrent()
require.NoError(t, json.Unmarshal(drainRequest(t, daemon, instanceID, volumeID).Data, &ack))
require.Equal(t, types.DrainVolumeStatusDrained, ack.Status)
require.Equal(t, ec2v1.DrainVolumeStatusDrained, ack.Status)
goleak.VerifyNone(t, ignoreExisting)
}
+4 -3
View File
@@ -14,13 +14,14 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
handlers_elbv2 "github.com/mulgadc/spinifex/spinifex/handlers/elbv2"
"github.com/mulgadc/spinifex/spinifex/handlers/sysinstance"
"github.com/mulgadc/spinifex/spinifex/network/topology"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/mulgadc/spinifex/spinifex/utils"
)
// Compile-time check that Daemon implements SystemInstanceLauncher.
@@ -327,7 +328,7 @@ func (d *Daemon) LaunchSystemInstance(input *handlers_elbv2.SystemInstanceInput)
// Subscribe to per-instance NATS topic for terminate commands.
d.mu.Lock()
sub, subErr := d.natsConn.Subscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), d.handleEC2Events)
sub, subErr := d.natsConn.Subscribe(ec2v1.InstanceCommandSubject(instance.ID), d.handleEC2Events)
if subErr != nil {
slog.Warn("LaunchSystemInstance: failed to subscribe to instance topic", "instanceId", instance.ID, "err", subErr)
} else {
+52 -51
View File
@@ -28,10 +28,11 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/config"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/instancetypes"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/instancetypes"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
handlers_ec2_account "github.com/mulgadc/spinifex/spinifex/handlers/ec2/account"
handlers_ec2_eigw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eigw"
handlers_ec2_eip "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eip"
@@ -49,10 +50,10 @@ import (
"github.com/mulgadc/spinifex/spinifex/objectstore"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
vmmock "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm/mock"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
@@ -1899,9 +1900,9 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) {
defer sub.Unsubscribe()
t.Run("MissingAttachVolumeData", func(t *testing.T) {
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
AttachVolume: true,
},
// No AttachVolumeData
@@ -1923,12 +1924,12 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) {
// Temporarily set status to stopped under the manager lock so -race
// reflects production discipline.
daemon.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Status = vm.StateStopped })
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
AttachVolume: true,
},
AttachVolumeData: &types.AttachVolumeData{
AttachVolumeData: &ec2v1.AttachVolumeData{
VolumeID: volumeID,
},
}
@@ -1948,12 +1949,12 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) {
t.Run("VolumeNotFound", func(t *testing.T) {
// volumeService.GetVolumeConfig will fail since we have no S3 backend
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
AttachVolume: true,
},
AttachVolumeData: &types.AttachVolumeData{
AttachVolumeData: &ec2v1.AttachVolumeData{
VolumeID: "vol-nonexistent",
Device: "/dev/sdf",
},
@@ -2018,9 +2019,9 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
defer sub.Unsubscribe()
t.Run("MissingDetachVolumeData", func(t *testing.T) {
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
DetachVolume: true,
},
// No DetachVolumeData
@@ -2040,12 +2041,12 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
// Temporarily set status to stopped under the manager lock so -race
// reflects production discipline.
daemon.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Status = vm.StateStopped })
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
DetachVolume: true,
},
DetachVolumeData: &types.DetachVolumeData{
DetachVolumeData: &ec2v1.DetachVolumeData{
VolumeID: volumeID,
},
}
@@ -2064,12 +2065,12 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
})
t.Run("VolumeNotAttached", func(t *testing.T) {
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
DetachVolume: true,
},
DetachVolumeData: &types.DetachVolumeData{
DetachVolumeData: &ec2v1.DetachVolumeData{
VolumeID: "vol-nonexistent",
},
}
@@ -2095,12 +2096,12 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
})
instance.EBSRequests.Mu.Unlock()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
DetachVolume: true,
},
DetachVolumeData: &types.DetachVolumeData{
DetachVolumeData: &ec2v1.DetachVolumeData{
VolumeID: bootVolumeID,
},
}
@@ -2130,12 +2131,12 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
})
instance.EBSRequests.Mu.Unlock()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
DetachVolume: true,
},
DetachVolumeData: &types.DetachVolumeData{
DetachVolumeData: &ec2v1.DetachVolumeData{
VolumeID: efiVolumeID,
},
}
@@ -2155,12 +2156,12 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
})
t.Run("DeviceMismatch", func(t *testing.T) {
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
DetachVolume: true,
},
DetachVolumeData: &types.DetachVolumeData{
DetachVolumeData: &ec2v1.DetachVolumeData{
VolumeID: volumeID,
Device: "/dev/sdg", // actual is /dev/sdf
},
@@ -2179,12 +2180,12 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
t.Run("QMPDeviceDelFails_NoForce", func(t *testing.T) {
// With nil QMPClient encoder/decoder, the QMP device_del returns
// error. Without force=true, this should return ServerInternal.
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
DetachVolume: true,
},
DetachVolumeData: &types.DetachVolumeData{
DetachVolumeData: &ec2v1.DetachVolumeData{
VolumeID: volumeID,
Force: false,
},
@@ -2356,12 +2357,12 @@ func TestDetachVolume_SuccessPath(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
DetachVolume: true,
},
DetachVolumeData: &types.DetachVolumeData{
DetachVolumeData: &ec2v1.DetachVolumeData{
VolumeID: volumeID,
},
}
@@ -2493,12 +2494,12 @@ func TestDetachVolume_ForceFlag(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
DetachVolume: true,
},
DetachVolumeData: &types.DetachVolumeData{
DetachVolumeData: &ec2v1.DetachVolumeData{
VolumeID: volumeID,
Force: true,
},
@@ -2598,12 +2599,12 @@ func TestDetachVolume_BlockdevDelFailure(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
DetachVolume: true,
},
DetachVolumeData: &types.DetachVolumeData{
DetachVolumeData: &ec2v1.DetachVolumeData{
VolumeID: volumeID,
},
}
@@ -2696,12 +2697,12 @@ func TestDetachVolume_SuccessWithDeviceMatch(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
DetachVolume: true,
},
DetachVolumeData: &types.DetachVolumeData{
DetachVolumeData: &ec2v1.DetachVolumeData{
VolumeID: volumeID,
Device: "/dev/sdh", // matches actual device
},
@@ -2773,12 +2774,12 @@ func TestAttachVolume_ReplacesStaleEBSRequest(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
AttachVolume: true,
},
AttachVolumeData: &types.AttachVolumeData{
AttachVolumeData: &ec2v1.AttachVolumeData{
VolumeID: volumeID,
Device: "/dev/sdg", // new device
},
@@ -3709,9 +3710,9 @@ func TestStopTerminate_IncorrectInstanceState(t *testing.T) {
t.Run("StopAlreadyStoppedInstance", func(t *testing.T) {
daemon.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Status = vm.StateStopped })
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
StopInstance: true,
},
}
@@ -3729,9 +3730,9 @@ func TestStopTerminate_IncorrectInstanceState(t *testing.T) {
t.Run("TerminateAlreadyTerminatedInstance", func(t *testing.T) {
daemon.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Status = vm.StateTerminated })
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
TerminateInstance: true,
},
}
+4 -4
View File
@@ -11,10 +11,10 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks"
spxtypes "github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -90,9 +90,9 @@ func (d *Daemon) TerminateWorkerInstances(ctx context.Context, instanceIDs []str
// gone, which a retried teardown treats as success. A NotFound error payload
// from the owner is likewise idempotent.
func (d *Daemon) terminateWorkerInstance(ctx context.Context, instanceID, accountID string) error {
cmd := spxtypes.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: spxtypes.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
StopInstance: true,
TerminateInstance: true,
},
@@ -102,7 +102,7 @@ func (d *Daemon) terminateWorkerInstance(ctx context.Context, instanceID, accoun
return fmt.Errorf("marshal terminate command: %w", err)
}
subject := fmt.Sprintf("ec2.cmd.%s", instanceID)
subject := ec2v1.InstanceCommandSubject(instanceID)
var msg *nats.Msg
for attempt := range 3 {
reqMsg := nats.NewMsg(subject)
+3 -3
View File
@@ -8,9 +8,9 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
spxtypes "github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
@@ -51,9 +51,9 @@ func TestTerminateWorkerInstances_RoutesToOwner(t *testing.T) {
t.Cleanup(nc.Close)
d := &Daemon{natsConn: nc}
gotCmd := make(chan spxtypes.EC2InstanceCommand, 1)
gotCmd := make(chan ec2v1.EC2InstanceCommand, 1)
sub, err := nc.Subscribe("ec2.cmd.i-owned", func(msg *nats.Msg) {
var cmd spxtypes.EC2InstanceCommand
var cmd ec2v1.EC2InstanceCommand
_ = json.Unmarshal(msg.Data, &cmd)
gotCmd <- cmd
_ = msg.Respond([]byte(`{}`))
+24 -24
View File
@@ -12,10 +12,10 @@ import (
"testing"
"time"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -114,8 +114,8 @@ func TestServerFaultIsStillLoggedAsAnError(t *testing.T) {
}
// commandFor builds a per-instance command with one attribute set.
func commandFor(id string, set func(*types.EC2CommandAttributes)) types.EC2InstanceCommand {
command := types.EC2InstanceCommand{ID: id}
func commandFor(id string, set func(*ec2v1.EC2CommandAttributes)) ec2v1.EC2InstanceCommand {
command := ec2v1.EC2InstanceCommand{ID: id}
set(&command.Attributes)
return command
}
@@ -126,26 +126,26 @@ func commandFor(id string, set func(*types.EC2CommandAttributes)) types.EC2Insta
func TestEC2CommandNameCoversEveryCommand(t *testing.T) {
tests := []struct {
want string
set func(*types.EC2CommandAttributes)
set func(*ec2v1.EC2CommandAttributes)
}{
{"AttachVolume", func(a *types.EC2CommandAttributes) { a.AttachVolume = true }},
{"DetachVolume", func(a *types.EC2CommandAttributes) { a.DetachVolume = true }},
{"DrainVolume", func(a *types.EC2CommandAttributes) { a.DrainVolume = true }},
{"AttachNetworkInterface", func(a *types.EC2CommandAttributes) { a.AttachENI = true }},
{"DetachNetworkInterface", func(a *types.EC2CommandAttributes) { a.DetachENI = true }},
{"AssociateIamInstanceProfile", func(a *types.EC2CommandAttributes) { a.AssociateIamInstanceProfile = true }},
{"SetSpotLineage", func(a *types.EC2CommandAttributes) { a.SetSpotLineage = true }},
{"SetInstanceTags", func(a *types.EC2CommandAttributes) { a.SetInstanceTags = true }},
{"RemoveInstanceTags", func(a *types.EC2CommandAttributes) { a.RemoveInstanceTags = true }},
{"SetInstanceMonitoring", func(a *types.EC2CommandAttributes) { a.SetInstanceMonitoring = true }},
{"StartInstance", func(a *types.EC2CommandAttributes) { a.StartInstance = true }},
{"RebootInstance", func(a *types.EC2CommandAttributes) { a.RebootInstance = true }},
{"StopInstance", func(a *types.EC2CommandAttributes) { a.StopInstance = true }},
{"TerminateInstance", func(a *types.EC2CommandAttributes) { a.TerminateInstance = true }},
{"AttachVolume", func(a *ec2v1.EC2CommandAttributes) { a.AttachVolume = true }},
{"DetachVolume", func(a *ec2v1.EC2CommandAttributes) { a.DetachVolume = true }},
{"DrainVolume", func(a *ec2v1.EC2CommandAttributes) { a.DrainVolume = true }},
{"AttachNetworkInterface", func(a *ec2v1.EC2CommandAttributes) { a.AttachENI = true }},
{"DetachNetworkInterface", func(a *ec2v1.EC2CommandAttributes) { a.DetachENI = true }},
{"AssociateIamInstanceProfile", func(a *ec2v1.EC2CommandAttributes) { a.AssociateIamInstanceProfile = true }},
{"SetSpotLineage", func(a *ec2v1.EC2CommandAttributes) { a.SetSpotLineage = true }},
{"SetInstanceTags", func(a *ec2v1.EC2CommandAttributes) { a.SetInstanceTags = true }},
{"RemoveInstanceTags", func(a *ec2v1.EC2CommandAttributes) { a.RemoveInstanceTags = true }},
{"SetInstanceMonitoring", func(a *ec2v1.EC2CommandAttributes) { a.SetInstanceMonitoring = true }},
{"StartInstance", func(a *ec2v1.EC2CommandAttributes) { a.StartInstance = true }},
{"RebootInstance", func(a *ec2v1.EC2CommandAttributes) { a.RebootInstance = true }},
{"StopInstance", func(a *ec2v1.EC2CommandAttributes) { a.StopInstance = true }},
{"TerminateInstance", func(a *ec2v1.EC2CommandAttributes) { a.TerminateInstance = true }},
}
// The table is hand-written, so without this a new attribute would pass by
// simply not being listed — and its commands would report as "unknown".
assert.Len(t, tests, reflect.TypeFor[types.EC2CommandAttributes]().NumField(),
assert.Len(t, tests, reflect.TypeFor[ec2v1.EC2CommandAttributes]().NumField(),
"every EC2CommandAttributes field needs a case here and in ec2CommandName")
for _, tc := range tests {
@@ -154,7 +154,7 @@ func TestEC2CommandNameCoversEveryCommand(t *testing.T) {
})
}
assert.Equal(t, "unknown", ec2CommandName(types.EC2InstanceCommand{ID: "i-123"}),
assert.Equal(t, "unknown", ec2CommandName(ec2v1.EC2InstanceCommand{ID: "i-123"}),
"a command nothing handles must still be named, not dropped")
}
@@ -173,7 +173,7 @@ func TestEC2CommandRecordsUnderTheCommandNotTheInstance(t *testing.T) {
t.Cleanup(func() { _ = sub.Unsubscribe() })
data, err := json.Marshal(commandFor(instanceID,
func(a *types.EC2CommandAttributes) { a.StopInstance = true }))
func(a *ec2v1.EC2CommandAttributes) { a.StopInstance = true }))
require.NoError(t, err)
_, err = daemon.natsConn.Request(subject, data, 5*time.Second)
require.NoError(t, err)
@@ -202,9 +202,9 @@ func TestDrainVolumeRecordsItsPointAfterTheWork(t *testing.T) {
before := len(requestOutcomesFor(t, ec2CmdAction("DrainVolume")))
reply := drainRequest(t, daemon, instanceID, volumeID)
var ack types.DrainVolumeResponse
var ack ec2v1.DrainVolumeResponse
require.NoError(t, json.Unmarshal(reply.Data, &ack))
require.Equal(t, types.DrainVolumeStatusDrained, ack.Status)
require.Equal(t, ec2v1.DrainVolumeStatusDrained, ack.Status)
// The goroutine records after it replies, so poll rather than assume the
// point has landed by the time the reply arrives.
+6 -5
View File
@@ -9,18 +9,19 @@ import (
"log/slog"
"time"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup"
"github.com/mulgadc/spinifex/spinifex/network/external/dhcp"
"github.com/mulgadc/spinifex/spinifex/network/topology"
"github.com/mulgadc/spinifex/spinifex/objectstore"
"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/mulgadc/spinifex/spinifex/services/viperblockd/vbwire"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/nats-io/nats.go"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
@@ -474,7 +475,7 @@ func (d *Daemon) onInstanceRecoveringHook() func(*vm.VM) {
if _, ok := d.natsSubscriptions[instance.ID]; ok {
return
}
sub, err := d.natsConn.Subscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), d.handleEC2Events)
sub, err := d.natsConn.Subscribe(ec2v1.InstanceCommandSubject(instance.ID), d.handleEC2Events)
if err != nil {
slog.Error("OnInstanceRecovering: failed to early-subscribe per-instance topic",
"instanceId", instance.ID, "err", err)
@@ -525,7 +526,7 @@ func (d *Daemon) onInstanceUpHook() func(*vm.VM) error {
subject string
handler nats.MsgHandler
}{
{instance.ID, fmt.Sprintf("ec2.cmd.%s", instance.ID), d.handleEC2Events},
{instance.ID, ec2v1.InstanceCommandSubject(instance.ID), d.handleEC2Events},
{consoleSubKey, fmt.Sprintf("ec2.%s.GetConsoleOutput", instance.ID), d.handleEC2GetConsoleOutput},
{passwordSubKey, fmt.Sprintf("ec2.%s.GetPasswordData", instance.ID), d.handleEC2GetPasswordData},
}
@@ -10,10 +10,10 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
spxtypes "github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -68,17 +68,17 @@ func AssociateIamInstanceProfile(ctx context.Context, input *ec2.AssociateIamIns
return nil, err
}
command := spxtypes.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: *input.InstanceId,
Attributes: spxtypes.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
AssociateIamInstanceProfile: true,
},
IamProfileAssociationData: &spxtypes.IamProfileAssociationData{
IamProfileAssociationData: &ec2v1.IamProfileAssociationData{
InstanceProfileArn: profile.ARN,
},
}
subject := fmt.Sprintf("ec2.cmd.%s", *input.InstanceId)
subject := ec2v1.InstanceCommandSubject(*input.InstanceId)
assoc, err := utils.NATSRequest[ec2.IamInstanceProfileAssociation](ctx, natsConn, subject, command, fanOutTimeout, accountID)
if err != nil {
if errors.Is(err, nats.ErrNoResponders) {
@@ -10,10 +10,10 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
spxtypes "github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
@@ -307,7 +307,7 @@ func TestAssociateIamInstanceProfile_Success(t *testing.T) {
ts := time.Date(2026, 5, 27, 10, 0, 0, 0, time.UTC)
_, err := nc.Subscribe("ec2.cmd."+instanceID, func(msg *nats.Msg) {
var cmd spxtypes.EC2InstanceCommand
var cmd ec2v1.EC2InstanceCommand
require.NoError(t, json.Unmarshal(msg.Data, &cmd))
assert.True(t, cmd.Attributes.AssociateIamInstanceProfile, "daemon must see the Associate flag set")
require.NotNil(t, cmd.IamProfileAssociationData)
@@ -9,8 +9,8 @@ import (
"time"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -47,9 +47,9 @@ func RebootInstances(ctx context.Context, input *ec2.RebootInstancesInput, natsC
}
instanceID := *instanceIDPtr
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
RebootInstance: true,
},
}
@@ -60,7 +60,7 @@ func RebootInstances(ctx context.Context, input *ec2.RebootInstancesInput, natsC
continue
}
subject := fmt.Sprintf("ec2.cmd.%s", instanceID)
subject := ec2v1.InstanceCommandSubject(instanceID)
reqMsg := nats.NewMsg(subject)
reqMsg.Data = jsonData
reqMsg.Header.Set(utils.AccountIDHeader, accountID)
@@ -8,8 +8,8 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
@@ -23,7 +23,7 @@ func TestRebootInstances_Success(t *testing.T) {
instanceID := "i-0123456789abcdef0"
nc.Subscribe("ec2.cmd."+instanceID, func(msg *nats.Msg) {
var cmd types.EC2InstanceCommand
var cmd ec2v1.EC2InstanceCommand
err := json.Unmarshal(msg.Data, &cmd)
require.NoError(t, err)
@@ -4,13 +4,12 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"time"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -74,9 +73,9 @@ func StartInstances(ctx context.Context, input *ec2.StartInstancesInput, natsCon
// startLiveInstance sends StartInstance via ec2.cmd.{id}. Returns handled=false on
// ErrNoResponders so the caller can fall back to the stopped-KV path.
func startLiveInstance(ctx context.Context, natsConn *nats.Conn, instanceID, accountID string) (*ec2.InstanceStateChange, bool, error) {
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{StartInstance: true},
Attributes: ec2v1.EC2CommandAttributes{StartInstance: true},
}
jsonData, err := json.Marshal(command)
if err != nil {
@@ -84,7 +83,7 @@ func startLiveInstance(ctx context.Context, natsConn *nats.Conn, instanceID, acc
return nil, false, errors.New(awserrors.ErrorServerInternal)
}
reqMsg := nats.NewMsg(fmt.Sprintf("ec2.cmd.%s", instanceID))
reqMsg := nats.NewMsg(ec2v1.InstanceCommandSubject(instanceID))
reqMsg.Data = jsonData
reqMsg.Header.Set(utils.AccountIDHeader, accountID)
utils.InjectTraceContext(ctx, reqMsg.Header)
@@ -4,13 +4,12 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"time"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -42,9 +41,9 @@ func StopInstances(ctx context.Context, input *ec2.StopInstancesInput, natsConn
}
instanceID := *instanceIDPtr
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
StopInstance: true,
TerminateInstance: false,
},
@@ -56,7 +55,7 @@ func StopInstances(ctx context.Context, input *ec2.StopInstancesInput, natsConn
continue
}
subject := fmt.Sprintf("ec2.cmd.%s", instanceID)
subject := ec2v1.InstanceCommandSubject(instanceID)
reqMsg := nats.NewMsg(subject)
reqMsg.Data = jsonData
reqMsg.Header.Set(utils.AccountIDHeader, accountID)
@@ -7,8 +7,8 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -21,7 +21,7 @@ func TestStopInstances_Success(t *testing.T) {
instanceID := "i-0123456789abcdef0"
nc.Subscribe("ec2.cmd."+instanceID, func(msg *nats.Msg) {
var cmd types.EC2InstanceCommand
var cmd ec2v1.EC2InstanceCommand
err := json.Unmarshal(msg.Data, &cmd)
require.NoError(t, err)
@@ -10,8 +10,8 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -50,9 +50,9 @@ func TerminateInstances(ctx context.Context, input *ec2.TerminateInstancesInput,
}
instanceID := *instanceIDPtr
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
StopInstance: true,
TerminateInstance: true,
},
@@ -65,7 +65,7 @@ func TerminateInstances(ctx context.Context, input *ec2.TerminateInstancesInput,
}
// Retry on ErrNoResponders: per-instance NATS subscription may not have propagated yet after a cluster restart.
subject := fmt.Sprintf("ec2.cmd.%s", instanceID)
subject := ec2v1.InstanceCommandSubject(instanceID)
var msg *nats.Msg
for attempt := range 3 {
reqMsg := nats.NewMsg(subject)
@@ -7,8 +7,8 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -21,7 +21,7 @@ func TestTerminateInstances_Success(t *testing.T) {
instanceID := "i-0123456789abcdef0"
nc.Subscribe("ec2.cmd."+instanceID, func(msg *nats.Msg) {
var cmd types.EC2InstanceCommand
var cmd ec2v1.EC2InstanceCommand
err := json.Unmarshal(msg.Data, &cmd)
require.NoError(t, err)
@@ -154,7 +154,7 @@ func TestTerminateInstances_VerifiesQMPAttributes(t *testing.T) {
_, nc := startTestNATSServer(t)
instanceID := "i-verify"
var receivedCmd types.EC2InstanceCommand
var receivedCmd ec2v1.EC2InstanceCommand
nc.Subscribe("ec2.cmd."+instanceID, func(msg *nats.Msg) {
json.Unmarshal(msg.Data, &receivedCmd)
@@ -8,6 +8,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup"
"github.com/mulgadc/spinifex/spinifex/types"
@@ -101,7 +102,7 @@ func mockSpreadCluster(t *testing.T, nc *nats.Conn, capacity map[string]int, res
}
termSub, err := nc.Subscribe("ec2.cmd.>", func(msg *nats.Msg) {
var cmd types.EC2InstanceCommand
var cmd ec2v1.EC2InstanceCommand
_ = json.Unmarshal(msg.Data, &cmd)
h.mu.Lock()
h.terminatedIDs = append(h.terminatedIDs, cmd.ID)
+5 -5
View File
@@ -9,7 +9,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -51,10 +51,10 @@ func stampSpotLineage(ctx context.Context, natsConn *nats.Conn, requests []*ec2.
// A non-no-responders transport error or an owner-returned error code stops the
// retry and is surfaced to the caller.
func sendSpotLineageCommand(ctx context.Context, natsConn *nats.Conn, instanceID, sirID, accountID string) error {
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{SetSpotLineage: true},
SpotLineageData: &types.SpotLineageData{SpotInstanceRequestId: sirID},
Attributes: ec2v1.EC2CommandAttributes{SetSpotLineage: true},
SpotLineageData: &ec2v1.SpotLineageData{SpotInstanceRequestId: sirID},
}
jsonData, err := json.Marshal(command)
if err != nil {
@@ -63,7 +63,7 @@ func sendSpotLineageCommand(ctx context.Context, natsConn *nats.Conn, instanceID
var lastErr error
for attempt := range spotLineageRetries {
reqMsg := nats.NewMsg("ec2.cmd." + instanceID)
reqMsg := nats.NewMsg(ec2v1.InstanceCommandSubject(instanceID))
reqMsg.Data = jsonData
reqMsg.Header.Set(utils.AccountIDHeader, accountID)
utils.InjectTraceContext(ctx, reqMsg.Header)
@@ -9,8 +9,8 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
@@ -41,7 +41,7 @@ func TestStampSpotLineage_TargetsOwnerWithSIRAndAccount(t *testing.T) {
select {
case m := <-got:
assert.Equal(t, lineageTestAccount, m.Header.Get(utils.AccountIDHeader))
var cmd types.EC2InstanceCommand
var cmd ec2v1.EC2InstanceCommand
require.NoError(t, json.Unmarshal(m.Data, &cmd))
assert.Equal(t, "i-owner", cmd.ID)
assert.True(t, cmd.Attributes.SetSpotLineage)
+5 -6
View File
@@ -4,14 +4,13 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -49,12 +48,12 @@ func AttachVolume(ctx context.Context, input *ec2.AttachVolumeInput, natsConn *n
device = *input.Device
}
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
AttachVolume: true,
},
AttachVolumeData: &types.AttachVolumeData{
AttachVolumeData: &ec2v1.AttachVolumeData{
VolumeID: volumeID,
Device: device,
},
@@ -66,7 +65,7 @@ func AttachVolume(ctx context.Context, input *ec2.AttachVolumeInput, natsConn *n
return output, errors.New(awserrors.ErrorServerInternal)
}
subject := fmt.Sprintf("ec2.cmd.%s", instanceID)
subject := ec2v1.InstanceCommandSubject(instanceID)
reqMsg := nats.NewMsg(subject)
reqMsg.Data = jsonData
reqMsg.Header.Set(utils.AccountIDHeader, accountID)
+5 -6
View File
@@ -4,15 +4,14 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
handlers_ec2_volume "github.com/mulgadc/spinifex/spinifex/handlers/ec2/volume"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -75,12 +74,12 @@ func DetachVolume(ctx context.Context, input *ec2.DetachVolumeInput, natsConn *n
force := input.Force != nil && *input.Force
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{
Attributes: ec2v1.EC2CommandAttributes{
DetachVolume: true,
},
DetachVolumeData: &types.DetachVolumeData{
DetachVolumeData: &ec2v1.DetachVolumeData{
VolumeID: volumeID,
Device: device,
Force: force,
@@ -93,7 +92,7 @@ func DetachVolume(ctx context.Context, input *ec2.DetachVolumeInput, natsConn *n
return output, errors.New(awserrors.ErrorServerInternal)
}
subject := fmt.Sprintf("ec2.cmd.%s", instanceID)
subject := ec2v1.InstanceCommandSubject(instanceID)
reqMsg := nats.NewMsg(subject)
reqMsg.Data = jsonData
reqMsg.Header.Set(utils.AccountIDHeader, accountID)
@@ -4,14 +4,13 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -46,10 +45,10 @@ func AttachNetworkInterface(ctx context.Context, input *ec2.AttachNetworkInterfa
eniID := *input.NetworkInterfaceId
deviceIndex := *input.DeviceIndex
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{AttachENI: true},
AttachENIData: &types.AttachENIData{
Attributes: ec2v1.EC2CommandAttributes{AttachENI: true},
AttachENIData: &ec2v1.AttachENIData{
NetworkInterfaceID: eniID,
DeviceIndex: deviceIndex,
},
@@ -61,7 +60,7 @@ func AttachNetworkInterface(ctx context.Context, input *ec2.AttachNetworkInterfa
return output, errors.New(awserrors.ErrorServerInternal)
}
reqMsg := nats.NewMsg(fmt.Sprintf("ec2.cmd.%s", instanceID))
reqMsg := nats.NewMsg(ec2v1.InstanceCommandSubject(instanceID))
reqMsg.Data = jsonData
reqMsg.Header.Set(utils.AccountIDHeader, accountID)
utils.InjectTraceContext(ctx, reqMsg.Header)
@@ -4,13 +4,12 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"time"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -46,10 +45,10 @@ func DetachNetworkInterface(ctx context.Context, input *ec2.DetachNetworkInterfa
return output, err
}
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{DetachENI: true},
DetachENIData: &types.DetachENIData{
Attributes: ec2v1.EC2CommandAttributes{DetachENI: true},
DetachENIData: &ec2v1.DetachENIData{
AttachmentID: attachmentID,
Force: force,
},
@@ -61,7 +60,7 @@ func DetachNetworkInterface(ctx context.Context, input *ec2.DetachNetworkInterfa
return output, errors.New(awserrors.ErrorServerInternal)
}
reqMsg := nats.NewMsg(fmt.Sprintf("ec2.cmd.%s", instanceID))
reqMsg := nats.NewMsg(ec2v1.InstanceCommandSubject(instanceID))
reqMsg.Data = jsonData
reqMsg.Header.Set(utils.AccountIDHeader, accountID)
utils.InjectTraceContext(ctx, reqMsg.Header)
@@ -7,9 +7,9 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
@@ -98,7 +98,7 @@ func TestAttachNetworkInterface_NoResponders(t *testing.T) {
func TestAttachNetworkInterface_Success(t *testing.T) {
nc := newTestNATS(t)
var received types.EC2InstanceCommand
var received ec2v1.EC2InstanceCommand
var receivedAccount string
sub, err := nc.Subscribe("ec2.cmd.i-success", func(msg *nats.Msg) {
receivedAccount = msg.Header.Get(utils.AccountIDHeader)
@@ -252,7 +252,7 @@ func TestDetachNetworkInterface_Success(t *testing.T) {
nc := newTestNATS(t)
describeNetworkInterfacesResponder(context.Background(), t, nc, "eni-attach-ok", "i-detach", "eni-detachable")
var received types.EC2InstanceCommand
var received ec2v1.EC2InstanceCommand
var receivedAccount string
sub, err := nc.Subscribe("ec2.cmd.i-detach", func(msg *nats.Msg) {
receivedAccount = msg.Header.Get(utils.AccountIDHeader)
+7 -7
View File
@@ -8,12 +8,12 @@ import (
"testing"
"time"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/config"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/handlers/ec2/volumestate"
"github.com/mulgadc/spinifex/spinifex/objectstore"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -70,21 +70,21 @@ func drainImageResponder(t *testing.T, nc *nats.Conn, instanceID string, reply [
func drainedImageAck(t *testing.T, volumeID string) []byte {
t.Helper()
data, err := json.Marshal(types.DrainVolumeResponse{VolumeID: volumeID, Status: types.DrainVolumeStatusDrained})
data, err := json.Marshal(ec2v1.DrainVolumeResponse{VolumeID: volumeID, Status: ec2v1.DrainVolumeStatusDrained})
require.NoError(t, err)
return data
}
func awaitImageDrainCommand(t *testing.T, got chan *nats.Msg) types.EC2InstanceCommand {
func awaitImageDrainCommand(t *testing.T, got chan *nats.Msg) ec2v1.EC2InstanceCommand {
t.Helper()
select {
case msg := <-got:
var command types.EC2InstanceCommand
var command ec2v1.EC2InstanceCommand
require.NoError(t, json.Unmarshal(msg.Data, &command))
return command
case <-time.After(2 * time.Second):
t.Fatal("the node hosting the volume never received a drain command")
return types.EC2InstanceCommand{}
return ec2v1.EC2InstanceCommand{}
}
}
@@ -7,8 +7,8 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
spxtypes "github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -29,10 +29,10 @@ func TestAssociateIamInstanceProfile_Success(t *testing.T) {
v := vmWithAccount("i-assoc1", testIAMAccount)
svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{v.ID: v})}
cmd := spxtypes.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: v.ID,
Attributes: spxtypes.EC2CommandAttributes{AssociateIamInstanceProfile: true},
IamProfileAssociationData: &spxtypes.IamProfileAssociationData{InstanceProfileArn: testProfileArn1},
Attributes: ec2v1.EC2CommandAttributes{AssociateIamInstanceProfile: true},
IamProfileAssociationData: &ec2v1.IamProfileAssociationData{InstanceProfileArn: testProfileArn1},
}
result, err := svc.AssociateIamInstanceProfile(context.Background(), v, cmd)
require.NoError(t, err)
@@ -54,10 +54,10 @@ func TestAssociateIamInstanceProfile_AlreadyAssociated(t *testing.T) {
v.IamInstanceProfileAssociationId = "iip-assoc-deadbeef00000001"
svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{v.ID: v})}
cmd := spxtypes.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: v.ID,
Attributes: spxtypes.EC2CommandAttributes{AssociateIamInstanceProfile: true},
IamProfileAssociationData: &spxtypes.IamProfileAssociationData{InstanceProfileArn: testProfileArn2},
Attributes: ec2v1.EC2CommandAttributes{AssociateIamInstanceProfile: true},
IamProfileAssociationData: &ec2v1.IamProfileAssociationData{InstanceProfileArn: testProfileArn2},
}
_, err := svc.AssociateIamInstanceProfile(context.Background(), v, cmd)
require.Error(t, err)
@@ -72,7 +72,7 @@ func TestAssociateIamInstanceProfile_MissingData(t *testing.T) {
v := vmWithAccount("i-assoc3", testIAMAccount)
svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{v.ID: v})}
cmd := spxtypes.EC2InstanceCommand{ID: v.ID, Attributes: spxtypes.EC2CommandAttributes{AssociateIamInstanceProfile: true}}
cmd := ec2v1.EC2InstanceCommand{ID: v.ID, Attributes: ec2v1.EC2CommandAttributes{AssociateIamInstanceProfile: true}}
_, err := svc.AssociateIamInstanceProfile(context.Background(), v, cmd)
require.Error(t, err)
assert.Equal(t, awserrors.ErrorMissingParameter, err.Error())
@@ -254,9 +254,9 @@ func TestStopOrTerminateInstance_AutoDisassociatesProfileOnTerminate(t *testing.
vmMgr: mgrWith(map[string]*vm.VM{v.ID: v}),
}
cmd := spxtypes.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: v.ID,
Attributes: spxtypes.EC2CommandAttributes{TerminateInstance: true},
Attributes: ec2v1.EC2CommandAttributes{TerminateInstance: true},
}
// Run synchronously up to the lock-protected attribute stamp; the
// goroutine that follows mutates a fake-less Manager which is fine
@@ -7,8 +7,8 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
spxtypes "github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
vmmock "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm/mock"
"github.com/stretchr/testify/assert"
@@ -33,7 +33,7 @@ func tagsAsMap(tags []*ec2.Tag) map[string]string {
func TestApplyInstanceTagMutation_MergeUpsert(t *testing.T) {
existing := tagList("Name", "web", "env", "dev")
out := ApplyInstanceTagMutation(existing, &spxtypes.InstanceTagsData{
out := ApplyInstanceTagMutation(existing, &ec2v1.InstanceTagsData{
Tags: map[string]string{"env": "prod", "team": "infra"},
}, false)
assert.Equal(t, map[string]string{"Name": "web", "env": "prod", "team": "infra"}, tagsAsMap(out))
@@ -41,7 +41,7 @@ func TestApplyInstanceTagMutation_MergeUpsert(t *testing.T) {
func TestApplyInstanceTagMutation_RemoveUnconditional(t *testing.T) {
existing := tagList("Name", "web", "env", "dev")
out := ApplyInstanceTagMutation(existing, &spxtypes.InstanceTagsData{
out := ApplyInstanceTagMutation(existing, &ec2v1.InstanceTagsData{
TagKeys: []string{"env", "missing"},
}, true)
assert.Equal(t, map[string]string{"Name": "web"}, tagsAsMap(out))
@@ -49,7 +49,7 @@ func TestApplyInstanceTagMutation_RemoveUnconditional(t *testing.T) {
func TestApplyInstanceTagMutation_RemoveValueMatch(t *testing.T) {
existing := tagList("Name", "web", "env", "dev")
out := ApplyInstanceTagMutation(existing, &spxtypes.InstanceTagsData{
out := ApplyInstanceTagMutation(existing, &ec2v1.InstanceTagsData{
Tags: map[string]string{"env": "prod", "Name": "web"},
}, true)
assert.Equal(t, map[string]string{"env": "dev"}, tagsAsMap(out))
@@ -57,13 +57,13 @@ func TestApplyInstanceTagMutation_RemoveValueMatch(t *testing.T) {
func TestApplyInstanceTagMutation_RemoveClearAll(t *testing.T) {
existing := tagList("Name", "web", "env", "dev")
out := ApplyInstanceTagMutation(existing, &spxtypes.InstanceTagsData{}, true)
out := ApplyInstanceTagMutation(existing, &ec2v1.InstanceTagsData{}, true)
assert.Empty(t, out)
}
func TestApplyInstanceTagMutation_SortedAndNilSafe(t *testing.T) {
existing := []*ec2.Tag{nil, {Key: aws.String("b"), Value: aws.String("2")}, {Key: nil}}
out := ApplyInstanceTagMutation(existing, &spxtypes.InstanceTagsData{
out := ApplyInstanceTagMutation(existing, &ec2v1.InstanceTagsData{
Tags: map[string]string{"a": "1"},
}, false)
require.Len(t, out, 2)
@@ -99,7 +99,7 @@ func TestWriteInstanceTags_WritesRecordAndCentral(t *testing.T) {
instance := &vm.VM{ID: "i-123", Instance: &ec2.Instance{Tags: tagList("Name", "web")}}
writer := &fakeTagWriter{}
err := WriteInstanceTags(context.Background(), instance, &spxtypes.InstanceTagsData{
err := WriteInstanceTags(context.Background(), instance, &ec2v1.InstanceTagsData{
Tags: map[string]string{"env": "prod"},
}, false, writer, "111122223333")
require.NoError(t, err)
@@ -116,7 +116,7 @@ func TestWriteInstanceTags_CentralWriteErrorPropagates(t *testing.T) {
instance := &vm.VM{ID: "i-123", Instance: &ec2.Instance{}}
writer := &fakeTagWriter{err: errors.New("s3 down")}
err := WriteInstanceTags(context.Background(), instance, &spxtypes.InstanceTagsData{
err := WriteInstanceTags(context.Background(), instance, &ec2v1.InstanceTagsData{
Tags: map[string]string{"env": "prod"},
}, false, writer, "111122223333")
assert.Error(t, err)
@@ -146,7 +146,7 @@ func TestTagStoppedInstance_WritesRecordAndCentral(t *testing.T) {
writer := &fakeTagWriter{}
svc := &InstanceServiceImpl{stoppedStore: store}
err := svc.TagStoppedInstance(context.Background(), "i-123", &spxtypes.InstanceTagsData{
err := svc.TagStoppedInstance(context.Background(), "i-123", &ec2v1.InstanceTagsData{
Tags: map[string]string{"env": "prod"},
}, false, writer, "111122223333")
require.NoError(t, err)
@@ -167,7 +167,7 @@ func TestTagStoppedInstance_RemoveKeys(t *testing.T) {
writer := &fakeTagWriter{}
svc := &InstanceServiceImpl{stoppedStore: store}
err := svc.TagStoppedInstance(context.Background(), "i-123", &spxtypes.InstanceTagsData{
err := svc.TagStoppedInstance(context.Background(), "i-123", &ec2v1.InstanceTagsData{
TagKeys: []string{"env"},
}, true, writer, "111122223333")
require.NoError(t, err)
@@ -182,7 +182,7 @@ func TestTagStoppedInstance_NotFound(t *testing.T) {
writer := &fakeTagWriter{}
svc := &InstanceServiceImpl{stoppedStore: store}
err := svc.TagStoppedInstance(context.Background(), "i-missing", &spxtypes.InstanceTagsData{
err := svc.TagStoppedInstance(context.Background(), "i-missing", &ec2v1.InstanceTagsData{
Tags: map[string]string{"env": "prod"},
}, false, writer, "111122223333")
require.Error(t, err)
@@ -197,7 +197,7 @@ func TestTagStoppedInstance_CrossAccountRejected(t *testing.T) {
writer := &fakeTagWriter{}
svc := &InstanceServiceImpl{stoppedStore: store}
err := svc.TagStoppedInstance(context.Background(), "i-123", &spxtypes.InstanceTagsData{
err := svc.TagStoppedInstance(context.Background(), "i-123", &ec2v1.InstanceTagsData{
Tags: map[string]string{"env": "prod"},
}, false, writer, "111122223333")
require.Error(t, err)
@@ -217,7 +217,7 @@ func TestTagStoppedInstance_ConcurrentClaimDoesNotResurrect(t *testing.T) {
writer := &fakeTagWriter{}
svc := &InstanceServiceImpl{stoppedStore: store}
err := svc.TagStoppedInstance(context.Background(), "i-123", &spxtypes.InstanceTagsData{
err := svc.TagStoppedInstance(context.Background(), "i-123", &ec2v1.InstanceTagsData{
Tags: map[string]string{"env": "prod"},
}, false, writer, "111122223333")
require.Error(t, err)
@@ -232,7 +232,7 @@ func TestTagStoppedInstance_CentralWriteErrorSkipsRecordWrite(t *testing.T) {
writer := &fakeTagWriter{err: errors.New("s3 down")}
svc := &InstanceServiceImpl{stoppedStore: store}
err := svc.TagStoppedInstance(context.Background(), "i-123", &spxtypes.InstanceTagsData{
err := svc.TagStoppedInstance(context.Background(), "i-123", &ec2v1.InstanceTagsData{
Tags: map[string]string{"env": "prod"},
}, false, writer, "111122223333")
require.Error(t, err)
+14 -13
View File
@@ -17,20 +17,21 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/bluebottle/pkg/safecast"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/config"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/instancetypes"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
"github.com/mulgadc/spinifex/spinifex/providers/ebs"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/instancetypes"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns"
handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/network/topology"
"github.com/mulgadc/spinifex/spinifex/objectstore"
"github.com/mulgadc/spinifex/spinifex/providers/ebs"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
spxtypes "github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/nats-io/nats.go"
)
@@ -441,7 +442,7 @@ func volumeTagsFromSpec(specs []*ec2.TagSpecification) map[string]string {
// ApplyInstanceTagMutation applies a set or remove tag mutation to a tag list,
// mirroring the central tag store's merge, value-match delete, and clear-all
// semantics, and returns the resulting list sorted by key.
func ApplyInstanceTagMutation(existing []*ec2.Tag, data *spxtypes.InstanceTagsData, remove bool) []*ec2.Tag {
func ApplyInstanceTagMutation(existing []*ec2.Tag, data *ec2v1.InstanceTagsData, remove bool) []*ec2.Tag {
tags := make(map[string]string, len(existing))
for _, t := range existing {
if t == nil || t.Key == nil {
@@ -484,7 +485,7 @@ func TagsToMap(tags []*ec2.Tag) map[string]string {
// of truth) and writes the resulting full tag set to the central tag store, so
// both stores move together. Callers own record persistence and must not hold
// the vm.Manager lock, since the central write is an S3 round-trip.
func WriteInstanceTags(ctx context.Context, instance *vm.VM, data *spxtypes.InstanceTagsData, remove bool, central InstanceTagWriter, accountID string) error {
func WriteInstanceTags(ctx context.Context, instance *vm.VM, data *ec2v1.InstanceTagsData, remove bool, central InstanceTagWriter, accountID string) error {
if instance == nil || instance.Instance == nil || central == nil {
return errors.New(awserrors.ErrorServerInternal)
}
@@ -495,7 +496,7 @@ func WriteInstanceTags(ctx context.Context, instance *vm.VM, data *spxtypes.Inst
// TagStoppedInstance applies a create-tags/delete-tags mutation to a stopped
// instance's shared KV record and the central tag store together. A missing or
// cross-account record returns InvalidID.NotFound and writes nothing.
func (s *InstanceServiceImpl) TagStoppedInstance(ctx context.Context, instanceID string, data *spxtypes.InstanceTagsData, remove bool, central InstanceTagWriter, accountID string) error {
func (s *InstanceServiceImpl) TagStoppedInstance(ctx context.Context, instanceID string, data *ec2v1.InstanceTagsData, remove bool, central InstanceTagWriter, accountID string) error {
if s.stoppedStore == nil {
slog.ErrorContext(ctx, "TagStoppedInstance: stopped store not available")
return errors.New(awserrors.ErrorServerInternal)
@@ -1164,7 +1165,7 @@ func (s *InstanceServiceImpl) RunInstances(ctx context.Context, input *ec2.RunIn
}
// RebootInstance handles an ec2.cmd reboot for a running instance on this node.
func (s *InstanceServiceImpl) RebootInstance(ctx context.Context, instance *vm.VM, command spxtypes.EC2InstanceCommand) error {
func (s *InstanceServiceImpl) RebootInstance(ctx context.Context, instance *vm.VM, command ec2v1.EC2InstanceCommand) error {
slog.InfoContext(ctx, "RebootInstance: rebooting instance", "id", command.ID)
if err := s.vmMgr.Reboot(ctx, instance.ID); err != nil {
@@ -1186,7 +1187,7 @@ func (s *InstanceServiceImpl) RebootInstance(ctx context.Context, instance *vm.V
}
// StartInstance handles an ec2.cmd start for a locally stopped instance.
func (s *InstanceServiceImpl) StartInstance(ctx context.Context, instance *vm.VM, command spxtypes.EC2InstanceCommand) error {
func (s *InstanceServiceImpl) StartInstance(ctx context.Context, instance *vm.VM, command ec2v1.EC2InstanceCommand) error {
slog.InfoContext(ctx, "StartInstance: starting instance", "id", command.ID)
status := s.vmMgr.Status(instance)
@@ -1231,7 +1232,7 @@ func (s *InstanceServiceImpl) StartInstance(ctx context.Context, instance *vm.VM
// StopOrTerminateInstance handles an ec2.cmd stop or terminate. Validates the
// transition synchronously, then dispatches Stop/Terminate in a goroutine.
func (s *InstanceServiceImpl) StopOrTerminateInstance(ctx context.Context, instance *vm.VM, command spxtypes.EC2InstanceCommand) error {
func (s *InstanceServiceImpl) StopOrTerminateInstance(ctx context.Context, instance *vm.VM, command ec2v1.EC2InstanceCommand) error {
isTerminate := command.Attributes.TerminateInstance
action := "Stopping"
initialState := vm.StateStopping
@@ -1337,7 +1338,7 @@ func (s *InstanceServiceImpl) StopOrTerminateInstance(ctx context.Context, insta
// AssociateIamInstanceProfile attaches an instance profile to a running instance.
// Validates no existing profile, then atomically writes the ARN + new association ID.
// InstanceProfile.Id is left nil; the gateway enriches it from IAMService.
func (s *InstanceServiceImpl) AssociateIamInstanceProfile(ctx context.Context, instance *vm.VM, command spxtypes.EC2InstanceCommand) (*ec2.IamInstanceProfileAssociation, error) {
func (s *InstanceServiceImpl) AssociateIamInstanceProfile(ctx context.Context, instance *vm.VM, command ec2v1.EC2InstanceCommand) (*ec2.IamInstanceProfileAssociation, error) {
if command.IamProfileAssociationData == nil || command.IamProfileAssociationData.InstanceProfileArn == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
@@ -2515,7 +2516,7 @@ func (s *InstanceServiceImpl) startLocalRecord(ctx context.Context, local *vm.VM
case vm.StateError:
slog.InfoContext(ctx, "StartStoppedInstance: starting an error-state record the stopped store does not hold",
"instanceId", local.ID)
if err := s.StartInstance(ctx, local, spxtypes.EC2InstanceCommand{ID: local.ID}); err != nil {
if err := s.StartInstance(ctx, local, ec2v1.EC2InstanceCommand{ID: local.ID}); err != nil {
return nil, err
}
return &StartStoppedInstanceOutput{Status: "running", InstanceID: local.ID}, nil
@@ -13,18 +13,19 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/config"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/instancetypes"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/instancetypes"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/internal/testkit"
spxtypes "github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
vmmock "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm/mock"
spxtypes "github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -3413,7 +3414,7 @@ func TestStartInstance_NotStopped(t *testing.T) {
mgr := mgrWith(map[string]*vm.VM{id: {ID: id, Status: vm.StateRunning}})
v, _ := mgr.Get(id)
svc := &InstanceServiceImpl{vmMgr: mgr}
err := svc.StartInstance(context.Background(), v, spxtypes.EC2InstanceCommand{ID: id})
err := svc.StartInstance(context.Background(), v, ec2v1.EC2InstanceCommand{ID: id})
require.Error(t, err)
assert.Equal(t, awserrors.ErrorIncorrectInstanceState, err.Error())
}
@@ -3423,9 +3424,9 @@ func TestStopOrTerminateInstance_TerminateIdempotent(t *testing.T) {
mgr := mgrWith(map[string]*vm.VM{id: {ID: id, Status: vm.StateShuttingDown}})
v, _ := mgr.Get(id)
svc := &InstanceServiceImpl{vmMgr: mgr}
err := svc.StopOrTerminateInstance(context.Background(), v, spxtypes.EC2InstanceCommand{
err := svc.StopOrTerminateInstance(context.Background(), v, ec2v1.EC2InstanceCommand{
ID: id,
Attributes: spxtypes.EC2CommandAttributes{TerminateInstance: true},
Attributes: ec2v1.EC2CommandAttributes{TerminateInstance: true},
})
require.NoError(t, err)
}
@@ -3435,9 +3436,9 @@ func TestStopOrTerminateInstance_InvalidTransition(t *testing.T) {
mgr := mgrWith(map[string]*vm.VM{id: {ID: id, Status: vm.StateStopped}})
v, _ := mgr.Get(id)
svc := &InstanceServiceImpl{vmMgr: mgr}
err := svc.StopOrTerminateInstance(context.Background(), v, spxtypes.EC2InstanceCommand{
err := svc.StopOrTerminateInstance(context.Background(), v, ec2v1.EC2InstanceCommand{
ID: id,
Attributes: spxtypes.EC2CommandAttributes{StopInstance: true},
Attributes: ec2v1.EC2CommandAttributes{StopInstance: true},
})
require.Error(t, err)
assert.Equal(t, awserrors.ErrorIncorrectInstanceState, err.Error())
@@ -3447,11 +3448,11 @@ func TestStopOrTerminateInstance_TerminationProtection(t *testing.T) {
// DisableApiTermination must block Terminate but never Stop.
tests := []struct {
name string
attrs spxtypes.EC2CommandAttributes
attrs ec2v1.EC2CommandAttributes
wantErr string
}{
{name: "terminate blocked", attrs: spxtypes.EC2CommandAttributes{TerminateInstance: true}, wantErr: awserrors.ErrorOperationNotPermitted},
{name: "stop allowed", attrs: spxtypes.EC2CommandAttributes{StopInstance: true}, wantErr: ""},
{name: "terminate blocked", attrs: ec2v1.EC2CommandAttributes{TerminateInstance: true}, wantErr: awserrors.ErrorOperationNotPermitted},
{name: "stop allowed", attrs: ec2v1.EC2CommandAttributes{StopInstance: true}, wantErr: ""},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
@@ -3465,7 +3466,7 @@ func TestStopOrTerminateInstance_TerminationProtection(t *testing.T) {
v, _ := mgr.Get(id)
svc := &InstanceServiceImpl{vmMgr: mgr}
err := svc.StopOrTerminateInstance(context.Background(), v, spxtypes.EC2InstanceCommand{ID: id, Attributes: tc.attrs})
err := svc.StopOrTerminateInstance(context.Background(), v, ec2v1.EC2InstanceCommand{ID: id, Attributes: tc.attrs})
if tc.wantErr == "" {
require.NoError(t, err)
return
@@ -4355,7 +4356,7 @@ func TestStartInstance_AllocateFails(t *testing.T) {
allocateErr: errors.New("no capacity"),
}
svc := &InstanceServiceImpl{vmMgr: mgr, resourceMgr: prov}
err := svc.StartInstance(context.Background(), v, spxtypes.EC2InstanceCommand{ID: id})
err := svc.StartInstance(context.Background(), v, ec2v1.EC2InstanceCommand{ID: id})
require.Error(t, err)
assert.Equal(t, awserrors.ErrorInsufficientInstanceCapacity, err.Error())
}
@@ -4374,7 +4375,7 @@ func TestStartInstance_ErrorStateStartable(t *testing.T) {
allocateErr: errors.New("no capacity"),
}
svc := &InstanceServiceImpl{vmMgr: mgr, resourceMgr: prov}
err := svc.StartInstance(context.Background(), v, spxtypes.EC2InstanceCommand{ID: id})
err := svc.StartInstance(context.Background(), v, ec2v1.EC2InstanceCommand{ID: id})
require.Error(t, err)
assert.Equal(t, awserrors.ErrorInsufficientInstanceCapacity, err.Error())
}
@@ -4383,7 +4384,7 @@ func TestRebootInstance_NotFound(t *testing.T) {
id := "i-missing"
mgr := mgrWith(nil)
svc := &InstanceServiceImpl{vmMgr: mgr}
err := svc.RebootInstance(context.Background(), &vm.VM{ID: id}, spxtypes.EC2InstanceCommand{ID: id})
err := svc.RebootInstance(context.Background(), &vm.VM{ID: id}, ec2v1.EC2InstanceCommand{ID: id})
require.Error(t, err)
assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
}
@@ -4399,7 +4400,7 @@ func TestRebootInstance_QMPFailureIsNotTheCallersToHear(t *testing.T) {
t.Cleanup(mgr.WaitForBackgroundWork)
svc := &InstanceServiceImpl{vmMgr: mgr}
require.NoError(t, svc.RebootInstance(context.Background(), instance, spxtypes.EC2InstanceCommand{ID: id}))
require.NoError(t, svc.RebootInstance(context.Background(), instance, ec2v1.EC2InstanceCommand{ID: id}))
assert.Equal(t, vm.StateRunning, mgr.Status(instance))
}
@@ -4412,7 +4413,7 @@ func TestRebootInstance_NotRunning(t *testing.T) {
mgr := mgrWith(map[string]*vm.VM{id: instance})
svc := &InstanceServiceImpl{vmMgr: mgr}
err := svc.RebootInstance(context.Background(), instance, spxtypes.EC2InstanceCommand{ID: id})
err := svc.RebootInstance(context.Background(), instance, ec2v1.EC2InstanceCommand{ID: id})
require.Error(t, err)
assert.Equal(t, awserrors.ErrorIncorrectInstanceState, err.Error())
}
@@ -4427,7 +4428,7 @@ func TestStartInstance_NotFound(t *testing.T) {
resourceMgr: &fakeResourceCapacityProvider{},
}
instance := &vm.VM{ID: id, Status: vm.StateStopped, InstanceType: "unknown"}
err := svc.StartInstance(context.Background(), instance, spxtypes.EC2InstanceCommand{ID: id})
err := svc.StartInstance(context.Background(), instance, ec2v1.EC2InstanceCommand{ID: id})
require.Error(t, err)
assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
}
+10 -10
View File
@@ -10,14 +10,14 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/config"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
"github.com/mulgadc/spinifex/spinifex/providers/ebs"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/handlers/ec2/volumestate"
"github.com/mulgadc/spinifex/spinifex/objectstore"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/providers/ebs"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
@@ -80,7 +80,7 @@ func drainResponder(t *testing.T, nc *nats.Conn, instanceID string, reply []byte
// drainedAck is the reply of a node whose drain reached S3.
func drainedAck(t *testing.T, volumeID string) []byte {
t.Helper()
data, err := json.Marshal(types.DrainVolumeResponse{VolumeID: volumeID, Status: types.DrainVolumeStatusDrained})
data, err := json.Marshal(ec2v1.DrainVolumeResponse{VolumeID: volumeID, Status: ec2v1.DrainVolumeStatusDrained})
require.NoError(t, err)
return data
}
@@ -104,17 +104,17 @@ func requireReturnsWithin(t *testing.T, limit time.Duration, fn func() error) {
// awaitDrainCommand returns the command the hosting node received, failing the
// test if none arrives.
func awaitDrainCommand(t *testing.T, got chan *nats.Msg) types.EC2InstanceCommand {
func awaitDrainCommand(t *testing.T, got chan *nats.Msg) ec2v1.EC2InstanceCommand {
t.Helper()
select {
case msg := <-got:
var command types.EC2InstanceCommand
var command ec2v1.EC2InstanceCommand
require.NoError(t, json.Unmarshal(msg.Data, &command))
assert.Equal(t, testAccountID, msg.Header.Get(utils.AccountIDHeader))
return command
case <-time.After(2 * time.Second):
t.Fatal("the node hosting the volume never received a drain command")
return types.EC2InstanceCommand{}
return ec2v1.EC2InstanceCommand{}
}
}
@@ -151,8 +151,8 @@ func TestDrainVolume_AttachedToStoppedInstanceTakesStoppedPath(t *testing.T) {
func TestDrainVolume_NotRunningAckTakesStoppedPath(t *testing.T) {
svc, store, nc := setupDrainService(t)
seedVolumeAttachment(t, store, "vol-not-running", "in-use", drainInstanceID)
ack, err := json.Marshal(types.DrainVolumeResponse{
VolumeID: "vol-not-running", Status: types.DrainVolumeStatusNotRunning,
ack, err := json.Marshal(ec2v1.DrainVolumeResponse{
VolumeID: "vol-not-running", Status: ec2v1.DrainVolumeStatusNotRunning,
})
require.NoError(t, err)
got := drainResponder(t, nc, drainInstanceID, ack)
@@ -16,6 +16,7 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/aws/aws-sdk-go/service/s3"
"github.com/mulgadc/bluebottle/pkg/safecast"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/config"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
@@ -25,7 +26,6 @@ import (
"github.com/mulgadc/spinifex/spinifex/handlers/ec2/volumestate"
"github.com/mulgadc/spinifex/spinifex/objectstore"
"github.com/mulgadc/spinifex/spinifex/providers/ebs"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -385,14 +385,14 @@ func volumeAttachment(ctx context.Context, store objectstore.ObjectStore, bucket
// failure — treating them as one would make every stopped instance's root
// volume permanently unsnapshottable.
func drainOnHostNode(ctx context.Context, natsConn *nats.Conn, volumeID, instanceID, accountID string) error {
command := types.EC2InstanceCommand{
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{DrainVolume: true},
DrainVolumeData: &types.DrainVolumeData{VolumeID: volumeID},
Attributes: ec2v1.EC2CommandAttributes{DrainVolume: true},
DrainVolumeData: &ec2v1.DrainVolumeData{VolumeID: volumeID},
}
resp, err := utils.NATSRequest[types.DrainVolumeResponse](ctx, natsConn,
"ec2.cmd."+instanceID, command, drainRequestTimeout, accountID)
resp, err := utils.NATSRequest[ec2v1.DrainVolumeResponse](ctx, natsConn,
ec2v1.InstanceCommandSubject(instanceID), command, drainRequestTimeout, accountID)
if err != nil {
// No subscriber at all: the instance runs nowhere in the cluster, so it
// was stopped (stop migrates it to shared KV before unsubscribing) and
@@ -409,12 +409,12 @@ func drainOnHostNode(ctx context.Context, natsConn *nats.Conn, volumeID, instanc
}
switch resp.Status {
case types.DrainVolumeStatusDrained:
case ec2v1.DrainVolumeStatusDrained:
return nil
// The host still holds the instance but it is not running: nothing is
// writing, so there is nothing to flush. This is a host-drain stop, which
// keeps the VM (and this subscription) in place after unmounting.
case types.DrainVolumeStatusNotRunning:
case ec2v1.DrainVolumeStatusNotRunning:
slog.InfoContext(ctx, "drainVolume: instance is not running on its host, snapshotting the volume's sealed checkpoint",
"volumeId", volumeID, "instanceId", instanceID)
return nil
+8 -8
View File
@@ -9,7 +9,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -79,10 +79,10 @@ func (c *natsENIController) Allocate(ctx context.Context, accountID, subnetID st
// Attach hot-plugs eniID onto instanceID and returns the attachment ID.
func (c *natsENIController) Attach(ctx context.Context, accountID, instanceID, eniID string) (string, error) {
cmd := types.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{AttachENI: true},
AttachENIData: &types.AttachENIData{
Attributes: ec2v1.EC2CommandAttributes{AttachENI: true},
AttachENIData: &ec2v1.AttachENIData{
NetworkInterfaceID: eniID,
DeviceIndex: taskENIDeviceIndex,
},
@@ -101,10 +101,10 @@ func (c *natsENIController) Release(ctx context.Context, accountID string, rec *
return nil
}
if rec.ENIAttachmentID != "" && rec.ContainerInstanceID != "" {
cmd := types.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: rec.ContainerInstanceID,
Attributes: types.EC2CommandAttributes{DetachENI: true},
DetachENIData: &types.DetachENIData{
Attributes: ec2v1.EC2CommandAttributes{DetachENI: true},
DetachENIData: &ec2v1.DetachENIData{
AttachmentID: rec.ENIAttachmentID,
Force: true,
},
@@ -124,7 +124,7 @@ func (c *natsENIController) Release(ctx context.Context, accountID string, rec *
}
func eniCmdSubject(instanceID string) string {
return fmt.Sprintf("ec2.cmd.%s", instanceID)
return ec2v1.InstanceCommandSubject(instanceID)
}
// reclaimTaskENI releases an awsvpc task's ENI on the single-writer teardown
+3 -3
View File
@@ -10,9 +10,9 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/aws/aws-sdk-go/service/ecs"
"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
@@ -421,7 +421,7 @@ func TestNATSENIController_AllocateAttachRelease(t *testing.T) {
}}
})
respond(t, nc, "ec2.cmd.i-1", func(req []byte) any {
var cmd types.EC2InstanceCommand
var cmd ec2v1.EC2InstanceCommand
_ = json.Unmarshal(req, &cmd)
if cmd.Attributes.AttachENI {
return ec2.AttachNetworkInterfaceOutput{AttachmentId: aws.String("att-real")}
+8 -8
View File
@@ -11,14 +11,14 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/config"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/spinifex/handlers/sysinstance"
handlers_systemvpc "github.com/mulgadc/spinifex/spinifex/handlers/systemvpc"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -569,16 +569,16 @@ func NewNATSVolumeAttacher(nc *nats.Conn) volumeAttacher {
// Returns the device the attachment landed on, which can differ from the
// requested one when the guest renames it.
func (a *natsVolumeAttacher) AttachVolume(ctx context.Context, accountID, instanceID, volumeID, device string) (string, error) {
cmd := types.EC2InstanceCommand{
cmd := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{AttachVolume: true},
AttachVolumeData: &types.AttachVolumeData{
Attributes: ec2v1.EC2CommandAttributes{AttachVolume: true},
AttachVolumeData: &ec2v1.AttachVolumeData{
VolumeID: volumeID,
Device: device,
},
}
out, err := utils.NATSRequest[ec2.VolumeAttachment](ctx, a.nc,
"ec2.cmd."+instanceID, cmd, a.timeout, accountID)
ec2v1.InstanceCommandSubject(instanceID), cmd, a.timeout, accountID)
if err != nil {
if !errors.Is(err, nats.ErrNoResponders) {
return "", err
+8 -8
View File
@@ -11,10 +11,10 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/rds"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/types"
handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -415,22 +415,22 @@ func NewNATSInstanceCommander(nc *nats.Conn) instanceCommander {
}
func (c *natsInstanceCommander) StopInstance(ctx context.Context, instanceID string) error {
return c.send(ctx, instanceID, types.EC2CommandAttributes{StopInstance: true})
return c.send(ctx, instanceID, ec2v1.EC2CommandAttributes{StopInstance: true})
}
func (c *natsInstanceCommander) StartInstance(ctx context.Context, instanceID string) error {
return c.send(ctx, instanceID, types.EC2CommandAttributes{StartInstance: true})
return c.send(ctx, instanceID, ec2v1.EC2CommandAttributes{StartInstance: true})
}
func (c *natsInstanceCommander) RebootInstance(ctx context.Context, instanceID string) error {
return c.send(ctx, instanceID, types.EC2CommandAttributes{RebootInstance: true})
return c.send(ctx, instanceID, ec2v1.EC2CommandAttributes{RebootInstance: true})
}
// The VM runs in the system account, so every command is issued there — the
// ownership check on the far side compares against that same account.
func (c *natsInstanceCommander) send(ctx context.Context, instanceID string, attrs types.EC2CommandAttributes) error {
cmd := types.EC2InstanceCommand{ID: instanceID, Attributes: attrs}
_, err := utils.NATSRequest[struct{}](ctx, c.nc, "ec2.cmd."+instanceID, cmd, c.timeout, utils.GlobalAccountID)
func (c *natsInstanceCommander) send(ctx context.Context, instanceID string, attrs ec2v1.EC2CommandAttributes) error {
cmd := ec2v1.EC2InstanceCommand{ID: instanceID, Attributes: attrs}
_, err := utils.NATSRequest[struct{}](ctx, c.nc, ec2v1.InstanceCommandSubject(instanceID), cmd, c.timeout, utils.GlobalAccountID)
if err == nil {
return nil
}
@@ -4,8 +4,8 @@ import (
"encoding/json"
"testing"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -112,7 +112,7 @@ func TestVMMarshal_LegacyHostfwdNotWrittenBack(t *testing.T) {
func TestVMMarshal_CommandBooleansAreNotPersisted(t *testing.T) {
legacy, err := json.Marshal(map[string]any{
"id": "i-1", "status": vm.StateRunning,
"attributes": types.EC2CommandAttributes{
"attributes": ec2v1.EC2CommandAttributes{
AttachVolume: true,
DetachVolume: true,
DrainVolume: true,
+2 -1
View File
@@ -15,6 +15,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp"
"github.com/mulgadc/spinifex/spinifex/types"
@@ -115,7 +116,7 @@ type VM struct {
// LegacyAttributes is the command last stamped onto records written before
// DesiredState existed. Read on decode to recover the operator-stop signal,
// then dropped; never written.
LegacyAttributes *types.EC2CommandAttributes `json:"attributes,omitempty"`
LegacyAttributes *ec2v1.EC2CommandAttributes `json:"attributes,omitempty"`
// EC2 API metadata stored for AWS API compatibility.
RunInstancesInput *ec2.RunInstancesInput `json:"run_instances_input,omitempty"`
+3 -2
View File
@@ -13,6 +13,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
@@ -95,9 +96,9 @@ func TestRequestSpotInstances_Lifecycle(t *testing.T) {
// subject (ec2.cmd.{id}) asynchronously after the response. Capturing every command sent
// lets the write-back be asserted against the real SIR ids RequestSpotInstances just
// minted, rather than the hand-built fixture lineage_test.go uses in isolation.
lineageCmds := make(chan types.EC2InstanceCommand, count)
lineageCmds := make(chan ec2v1.EC2InstanceCommand, count)
cmdSub, err := gw.NATSConn.Subscribe("ec2.cmd.>", func(msg *nats.Msg) {
var cmd types.EC2InstanceCommand
var cmd ec2v1.EC2InstanceCommand
if jsonErr := json.Unmarshal(msg.Data, &cmd); jsonErr != nil {
t.Errorf("unmarshal spot lineage command: %v", jsonErr)
return