diff --git a/hive/config/config_test.go b/hive/config/config_test.go index 34b1d9d09..ae9c2cf11 100644 --- a/hive/config/config_test.go +++ b/hive/config/config_test.go @@ -369,4 +369,3 @@ shardwal = true require.NotNil(t, n.Viperblock.ShardWAL) assert.True(t, *n.Viperblock.ShardWAL) } - diff --git a/hive/daemon/daemon_handlers.go b/hive/daemon/daemon_handlers.go index d18e9cec9..9b6dd25d8 100644 --- a/hive/daemon/daemon_handlers.go +++ b/hive/daemon/daemon_handlers.go @@ -8,7 +8,6 @@ import ( "time" "github.com/mulgadc/hive/hive/awserrors" - "github.com/mulgadc/hive/hive/qmp" "github.com/mulgadc/hive/hive/types" "github.com/mulgadc/hive/hive/utils" "github.com/mulgadc/hive/hive/vm" @@ -57,10 +56,10 @@ func handleNATSRequest[I any, O any](msg *nats.Msg, serviceFn func(*I, string) ( // handleEC2Events processes incoming EC2 instance events (start, stop, terminate, attach-volume) func (d *Daemon) handleEC2Events(msg *nats.Msg) { - var command qmp.Command + var command types.EC2InstanceCommand if err := json.Unmarshal(msg.Data, &command); err != nil { - slog.Error("Error unmarshaling QMP command", "err", err) + slog.Error("Error unmarshaling EC2 instance command", "err", err) respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -92,49 +91,8 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { case command.Attributes.StopInstance, command.Attributes.TerminateInstance: d.handleStopOrTerminateInstance(msg, command, instance) default: - d.handleQMPCommand(msg, command, instance) - } -} - -func (d *Daemon) handleQMPCommand(msg *nats.Msg, command qmp.Command, instance *vm.VM) { - resp, err := d.SendQMPCommand(instance.QMPClient, command.QMPCommand, instance.ID) - if err != nil { - slog.Error("Failed to send QMP command", "err", err) + slog.Warn("Unhandled EC2 instance command", "id", command.ID, "attributes", command.Attributes) respondWithError(msg, awserrors.ErrorServerInternal) - return - } - - slog.Debug("RAW QMP Response", "resp", string(resp.Return)) - - // Unmarshal the response - target, ok := qmp.CommandResponseTypes[command.QMPCommand.Execute] - if !ok { - slog.Warn("Unhandled QMP command", "cmd", command.QMPCommand.Execute) - if err := msg.Respond(resp.Return); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } - return - } - - if err := json.Unmarshal(resp.Return, target); err != nil { - slog.Error("Failed to unmarshal QMP response", "cmd", command.QMPCommand.Execute, "err", err) - if err := msg.Respond(resp.Return); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } - return - } - - // Update attributes and respond - d.Instances.Mu.Lock() - instance.Attributes = command.Attributes - d.Instances.Mu.Unlock() - - if err := d.WriteState(); err != nil { - slog.Error("Failed to write state to disk", "err", err) - } - - if err := msg.Respond(resp.Return); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) } } diff --git a/hive/daemon/daemon_handlers_instance.go b/hive/daemon/daemon_handlers_instance.go index 065330e46..161f04a8e 100644 --- a/hive/daemon/daemon_handlers_instance.go +++ b/hive/daemon/daemon_handlers_instance.go @@ -11,7 +11,6 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/hive/hive/awserrors" - "github.com/mulgadc/hive/hive/qmp" "github.com/mulgadc/hive/hive/types" "github.com/mulgadc/hive/hive/utils" "github.com/mulgadc/hive/hive/vm" @@ -306,7 +305,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { slog.Info("handleEC2RunInstances completed", "requested", launchCount, "created", len(instances), "launched", successCount) } -func (d *Daemon) handleStartInstance(msg *nats.Msg, command qmp.Command, instance *vm.VM) { +func (d *Daemon) handleStartInstance(msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) { slog.Info("Starting instance", "id", command.ID) // Validate instance is in stopped state @@ -360,7 +359,7 @@ func (d *Daemon) handleStartInstance(msg *nats.Msg, command qmp.Command, instanc } } -func (d *Daemon) handleStopOrTerminateInstance(msg *nats.Msg, command qmp.Command, instance *vm.VM) { +func (d *Daemon) handleStopOrTerminateInstance(msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) { isTerminate := command.Attributes.TerminateInstance action := "Stopping" initialState := vm.StateStopping @@ -399,7 +398,7 @@ func (d *Daemon) handleStopOrTerminateInstance(msg *nats.Msg, command qmp.Comman } // Run cleanup in goroutine to not block NATS - go func(inst *vm.VM, attrs qmp.Attributes) { + go func(inst *vm.VM, attrs types.EC2CommandAttributes) { stopErr := d.stopInstance(map[string]*vm.VM{inst.ID: inst}, isTerminate) if stopErr != nil { @@ -685,7 +684,7 @@ func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) { // Add instance to local map and clear stop attribute before launch d.Instances.Mu.Lock() d.Instances.VMS[instance.ID] = instance - instance.Attributes = qmp.Attributes{StartInstance: true} + instance.Attributes = types.EC2CommandAttributes{StartInstance: true} d.Instances.Mu.Unlock() // Launch the instance infrastructure (QEMU, QMP, NATS subscriptions) diff --git a/hive/daemon/daemon_handlers_test.go b/hive/daemon/daemon_handlers_test.go index dacd9de37..e3975fe17 100644 --- a/hive/daemon/daemon_handlers_test.go +++ b/hive/daemon/daemon_handlers_test.go @@ -14,7 +14,6 @@ import ( awss3 "github.com/aws/aws-sdk-go/service/s3" "github.com/mulgadc/hive/hive/config" - "github.com/mulgadc/hive/hive/types" handlers_ec2_account "github.com/mulgadc/hive/hive/handlers/ec2/account" handlers_ec2_eigw "github.com/mulgadc/hive/hive/handlers/ec2/eigw" handlers_ec2_igw "github.com/mulgadc/hive/hive/handlers/ec2/igw" @@ -27,6 +26,7 @@ import ( handlers_ec2_vpc "github.com/mulgadc/hive/hive/handlers/ec2/vpc" "github.com/mulgadc/hive/hive/objectstore" "github.com/mulgadc/hive/hive/qmp" + "github.com/mulgadc/hive/hive/types" "github.com/mulgadc/hive/hive/vm" "github.com/mulgadc/viperblock/viperblock" "github.com/nats-io/nats-server/v2/server" @@ -553,9 +553,9 @@ func TestHandleEC2Events_StopInstance(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - cmd := qmp.Command{ + cmd := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{StopInstance: true}, + Attributes: types.EC2CommandAttributes{StopInstance: true}, } cmdData, _ := json.Marshal(cmd) @@ -599,9 +599,9 @@ func TestHandleEC2Events_TerminateInstance(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - cmd := qmp.Command{ + cmd := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{TerminateInstance: true}, + Attributes: types.EC2CommandAttributes{TerminateInstance: true}, } cmdData, _ := json.Marshal(cmd) @@ -630,9 +630,9 @@ func TestHandleEC2Events_InstanceNotFound(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - cmd := qmp.Command{ + cmd := types.EC2InstanceCommand{ ID: "i-nonexistent", - Attributes: qmp.Attributes{StopInstance: true}, + Attributes: types.EC2CommandAttributes{StopInstance: true}, } cmdData, _ := json.Marshal(cmd) @@ -1484,12 +1484,12 @@ func TestAttachVolume_ZoneMismatch(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ AttachVolume: true, }, - AttachVolumeData: &qmp.AttachVolumeData{ + AttachVolumeData: &types.AttachVolumeData{ VolumeID: volumeID, }, } @@ -2755,9 +2755,9 @@ func TestAttachVolume_MissingVolumeData(t *testing.T) { defer sub.Unsubscribe() // AttachVolume with nil AttachVolumeData - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ AttachVolume: true, }, AttachVolumeData: nil, @@ -2794,12 +2794,12 @@ func TestAttachVolume_InstanceNotRunning(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ AttachVolume: true, }, - AttachVolumeData: &qmp.AttachVolumeData{ + AttachVolumeData: &types.AttachVolumeData{ VolumeID: "vol-test-123", }, } @@ -2835,12 +2835,12 @@ func TestAttachVolume_VolumeNotFound(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ AttachVolume: true, }, - AttachVolumeData: &qmp.AttachVolumeData{ + AttachVolumeData: &types.AttachVolumeData{ VolumeID: "vol-nonexistent-999", }, } @@ -2897,12 +2897,12 @@ func TestAttachVolume_VolumeInUse(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ AttachVolume: true, }, - AttachVolumeData: &qmp.AttachVolumeData{ + AttachVolumeData: &types.AttachVolumeData{ VolumeID: volumeID, }, } @@ -2940,9 +2940,9 @@ func TestDetachVolume_MissingVolumeData(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, DetachVolumeData: nil, @@ -2979,12 +2979,12 @@ func TestDetachVolume_InstanceNotRunning(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, - DetachVolumeData: &qmp.DetachVolumeData{ + DetachVolumeData: &types.DetachVolumeData{ VolumeID: "vol-test-123", }, } @@ -3020,12 +3020,12 @@ func TestDetachVolume_VolumeNotAttached(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, - DetachVolumeData: &qmp.DetachVolumeData{ + DetachVolumeData: &types.DetachVolumeData{ VolumeID: "vol-not-attached-999", }, } @@ -3064,12 +3064,12 @@ func TestDetachVolume_BootVolumeRejected(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, - DetachVolumeData: &qmp.DetachVolumeData{ + DetachVolumeData: &types.DetachVolumeData{ VolumeID: "vol-boot-001", }, } @@ -3108,12 +3108,12 @@ func TestDetachVolume_DeviceMismatch(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, - DetachVolumeData: &qmp.DetachVolumeData{ + DetachVolumeData: &types.DetachVolumeData{ VolumeID: "vol-mismatch-001", Device: "/dev/sdg", }, diff --git a/hive/daemon/daemon_handlers_volume.go b/hive/daemon/daemon_handlers_volume.go index c39b0dbca..a760076d1 100644 --- a/hive/daemon/daemon_handlers_volume.go +++ b/hive/daemon/daemon_handlers_volume.go @@ -8,8 +8,8 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/hive/hive/awserrors" - "github.com/mulgadc/hive/hive/types" "github.com/mulgadc/hive/hive/qmp" + "github.com/mulgadc/hive/hive/types" "github.com/mulgadc/hive/hive/utils" "github.com/mulgadc/hive/hive/vm" "github.com/nats-io/nats.go" @@ -20,7 +20,7 @@ import ( // Phase 1: ebs.mount via NATS (rolls back with ebs.unmount) // Phase 2: QMP blockdev-add (rolls back Phase 1) // Phase 3: QMP device_add (rolls back Phase 2 + Phase 1) -func (d *Daemon) handleAttachVolume(msg *nats.Msg, command qmp.Command, instance *vm.VM) { +func (d *Daemon) handleAttachVolume(msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) { slog.Info("Attaching volume to instance", "instanceId", command.ID) // Validate AttachVolumeData @@ -294,7 +294,7 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command qmp.Command, instance // Phase 1: QMP device_del (remove guest device) // Phase 2: QMP blockdev-del (remove block node) // Phase 3: ebs.unmount NATS (stop NBD server) -func (d *Daemon) handleDetachVolume(msg *nats.Msg, command qmp.Command, instance *vm.VM) { +func (d *Daemon) handleDetachVolume(msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) { slog.Info("Detaching volume from instance", "instanceId", command.ID) // Validate DetachVolumeData diff --git a/hive/daemon/daemon_test.go b/hive/daemon/daemon_test.go index 094a50c2d..7c3707cf6 100644 --- a/hive/daemon/daemon_test.go +++ b/hive/daemon/daemon_test.go @@ -17,10 +17,10 @@ import ( awss3 "github.com/aws/aws-sdk-go/service/s3" "github.com/mulgadc/hive/hive/config" handlers_ec2_instance "github.com/mulgadc/hive/hive/handlers/ec2/instance" - "github.com/mulgadc/hive/hive/types" handlers_ec2_volume "github.com/mulgadc/hive/hive/handlers/ec2/volume" "github.com/mulgadc/hive/hive/objectstore" "github.com/mulgadc/hive/hive/qmp" + "github.com/mulgadc/hive/hive/types" "github.com/mulgadc/hive/hive/vm" "github.com/nats-io/nats.go" "github.com/pelletier/go-toml/v2" @@ -862,20 +862,20 @@ func TestDaemon_BootAllocation(t *testing.T) { InstanceType: getTestInstanceType(), Status: vm.StateRunning, AccountID: testAccountID, - Attributes: qmp.Attributes{StopInstance: false}, + Attributes: types.EC2CommandAttributes{StopInstance: false}, }, "i-stopped": { ID: "i-stopped", InstanceType: getTestInstanceType(), Status: vm.StateStopped, AccountID: testAccountID, - Attributes: qmp.Attributes{StopInstance: true}, + Attributes: types.EC2CommandAttributes{StopInstance: true}, }, "i-terminated": { ID: "i-terminated", InstanceType: getTestInstanceType(), Status: vm.StateTerminated, - Attributes: qmp.Attributes{StopInstance: false}, + Attributes: types.EC2CommandAttributes{StopInstance: false}, }, } @@ -2011,9 +2011,9 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) { defer sub.Unsubscribe() t.Run("MissingAttachVolumeData", func(t *testing.T) { - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ AttachVolume: true, }, // No AttachVolumeData @@ -2037,12 +2037,12 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) { instance.Status = vm.StateStopped daemon.Instances.Mu.Unlock() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ AttachVolume: true, }, - AttachVolumeData: &qmp.AttachVolumeData{ + AttachVolumeData: &types.AttachVolumeData{ VolumeID: volumeID, }, } @@ -2064,12 +2064,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 := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ AttachVolume: true, }, - AttachVolumeData: &qmp.AttachVolumeData{ + AttachVolumeData: &types.AttachVolumeData{ VolumeID: "vol-nonexistent", Device: "/dev/sdf", }, @@ -2157,9 +2157,9 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { defer sub.Unsubscribe() t.Run("MissingDetachVolumeData", func(t *testing.T) { - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, // No DetachVolumeData @@ -2181,12 +2181,12 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { instance.Status = vm.StateStopped daemon.Instances.Mu.Unlock() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, - DetachVolumeData: &qmp.DetachVolumeData{ + DetachVolumeData: &types.DetachVolumeData{ VolumeID: volumeID, }, } @@ -2207,12 +2207,12 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { }) t.Run("VolumeNotAttached", func(t *testing.T) { - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, - DetachVolumeData: &qmp.DetachVolumeData{ + DetachVolumeData: &types.DetachVolumeData{ VolumeID: "vol-nonexistent", }, } @@ -2238,12 +2238,12 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { }) instance.EBSRequests.Mu.Unlock() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, - DetachVolumeData: &qmp.DetachVolumeData{ + DetachVolumeData: &types.DetachVolumeData{ VolumeID: bootVolumeID, }, } @@ -2273,12 +2273,12 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { }) instance.EBSRequests.Mu.Unlock() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, - DetachVolumeData: &qmp.DetachVolumeData{ + DetachVolumeData: &types.DetachVolumeData{ VolumeID: efiVolumeID, }, } @@ -2307,12 +2307,12 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { }) instance.EBSRequests.Mu.Unlock() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, - DetachVolumeData: &qmp.DetachVolumeData{ + DetachVolumeData: &types.DetachVolumeData{ VolumeID: ciVolumeID, }, } @@ -2332,12 +2332,12 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { }) t.Run("DeviceMismatch", func(t *testing.T) { - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, - DetachVolumeData: &qmp.DetachVolumeData{ + DetachVolumeData: &types.DetachVolumeData{ VolumeID: volumeID, Device: "/dev/sdg", // actual is /dev/sdf }, @@ -2356,12 +2356,12 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { t.Run("QMPDeviceDelFails_NoForce", func(t *testing.T) { // With nil QMPClient encoder/decoder, SendQMPCommand returns error. // Without force=true, this should return ServerInternal. - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, - DetachVolumeData: &qmp.DetachVolumeData{ + DetachVolumeData: &types.DetachVolumeData{ VolumeID: volumeID, Force: false, }, @@ -2517,12 +2517,12 @@ func TestDetachVolume_SuccessPath(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, - DetachVolumeData: &qmp.DetachVolumeData{ + DetachVolumeData: &types.DetachVolumeData{ VolumeID: volumeID, }, } @@ -2656,12 +2656,12 @@ func TestDetachVolume_ForceFlag(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, - DetachVolumeData: &qmp.DetachVolumeData{ + DetachVolumeData: &types.DetachVolumeData{ VolumeID: volumeID, Force: true, }, @@ -2761,12 +2761,12 @@ func TestDetachVolume_BlockdevDelFailure(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, - DetachVolumeData: &qmp.DetachVolumeData{ + DetachVolumeData: &types.DetachVolumeData{ VolumeID: volumeID, }, } @@ -2861,12 +2861,12 @@ func TestDetachVolume_SuccessWithDeviceMatch(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, - DetachVolumeData: &qmp.DetachVolumeData{ + DetachVolumeData: &types.DetachVolumeData{ VolumeID: volumeID, Device: "/dev/sdh", // matches actual device }, @@ -2938,12 +2938,12 @@ func TestAttachVolume_ReplacesStaleEBSRequest(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ AttachVolume: true, }, - AttachVolumeData: &qmp.AttachVolumeData{ + AttachVolumeData: &types.AttachVolumeData{ VolumeID: volumeID, Device: "/dev/sdg", // new device }, @@ -3455,9 +3455,9 @@ func TestStopTerminate_IncorrectInstanceState(t *testing.T) { instance.Status = vm.StateStopped daemon.Instances.Mu.Unlock() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ StopInstance: true, }, } @@ -3478,9 +3478,9 @@ func TestStopTerminate_IncorrectInstanceState(t *testing.T) { instance.Status = vm.StateTerminated daemon.Instances.Mu.Unlock() - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ TerminateInstance: true, }, } diff --git a/hive/daemon/state_test.go b/hive/daemon/state_test.go index 40fe42fb3..734ea7599 100644 --- a/hive/daemon/state_test.go +++ b/hive/daemon/state_test.go @@ -9,7 +9,7 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/hive/hive/config" - "github.com/mulgadc/hive/hive/qmp" + "github.com/mulgadc/hive/hive/types" "github.com/mulgadc/hive/hive/vm" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" @@ -676,7 +676,7 @@ func TestRestoreInstances_UserStoppedMigratedToSharedKV(t *testing.T) { daemon.Instances.VMS["i-restore-userstop"] = &vm.VM{ ID: "i-restore-userstop", Status: vm.StateStopped, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ StopInstance: true, }, } @@ -770,7 +770,7 @@ func TestRestoreInstances_MixedStates(t *testing.T) { } daemon.Instances.VMS["i-mix-stopped"] = &vm.VM{ ID: "i-mix-stopped", Status: vm.StateStopped, - Attributes: qmp.Attributes{StopInstance: true}, + Attributes: types.EC2CommandAttributes{StopInstance: true}, } require.NoError(t, daemon.WriteState()) @@ -850,7 +850,7 @@ func TestStatePersistence_RoundTrip(t *testing.T) { ID: "i-roundtrip", Status: vm.StateRunning, InstanceType: "t3.micro", - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ StopInstance: false, TerminateInstance: false, }, @@ -881,7 +881,7 @@ func TestRestoreInstances_StoppedInstanceMigratedToSharedKV(t *testing.T) { ID: instanceID, InstanceType: "t3.micro", Status: vm.StateStopped, - Attributes: qmp.Attributes{StopInstance: true}, + Attributes: types.EC2CommandAttributes{StopInstance: true}, } require.NoError(t, daemon.WriteState()) diff --git a/hive/gateway/ec2/instance/StopInstances.go b/hive/gateway/ec2/instance/StopInstances.go index 38f3d9f57..75468d416 100644 --- a/hive/gateway/ec2/instance/StopInstances.go +++ b/hive/gateway/ec2/instance/StopInstances.go @@ -8,7 +8,7 @@ import ( "time" "github.com/aws/aws-sdk-go/service/ec2" - "github.com/mulgadc/hive/hive/qmp" + "github.com/mulgadc/hive/hive/types" "github.com/mulgadc/hive/hive/utils" "github.com/nats-io/nats.go" ) @@ -31,15 +31,9 @@ func StopInstances(input *ec2.StopInstancesInput, natsConn *nats.Conn, accountID } instanceID := *instanceIDPtr - // Build the QMP command to stop the instance - // Note: system_powerdown with stop_instance=true prevents auto-restart on daemon boot - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - QMPCommand: qmp.QMPCommand{ - Execute: "system_powerdown", - Arguments: map[string]any{}, - }, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ StopInstance: true, // Don't auto-restart on daemon boot TerminateInstance: false, }, diff --git a/hive/gateway/ec2/instance/StopInstances_test.go b/hive/gateway/ec2/instance/StopInstances_test.go index 53f155229..2410c8d51 100644 --- a/hive/gateway/ec2/instance/StopInstances_test.go +++ b/hive/gateway/ec2/instance/StopInstances_test.go @@ -6,7 +6,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" - "github.com/mulgadc/hive/hive/qmp" + "github.com/mulgadc/hive/hive/types" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -18,12 +18,11 @@ func TestStopInstances_Success(t *testing.T) { instanceID := "i-0123456789abcdef0" nc.Subscribe("ec2.cmd."+instanceID, func(msg *nats.Msg) { - var cmd qmp.Command + var cmd types.EC2InstanceCommand err := json.Unmarshal(msg.Data, &cmd) require.NoError(t, err) assert.Equal(t, instanceID, cmd.ID) - assert.Equal(t, "system_powerdown", cmd.QMPCommand.Execute) assert.True(t, cmd.Attributes.StopInstance) assert.False(t, cmd.Attributes.TerminateInstance) diff --git a/hive/gateway/ec2/instance/TerminateInstances.go b/hive/gateway/ec2/instance/TerminateInstances.go index dcf33fdd1..c9b49e167 100644 --- a/hive/gateway/ec2/instance/TerminateInstances.go +++ b/hive/gateway/ec2/instance/TerminateInstances.go @@ -8,7 +8,7 @@ import ( "time" "github.com/aws/aws-sdk-go/service/ec2" - "github.com/mulgadc/hive/hive/qmp" + "github.com/mulgadc/hive/hive/types" "github.com/mulgadc/hive/hive/utils" "github.com/nats-io/nats.go" ) @@ -36,15 +36,9 @@ func TerminateInstances(input *ec2.TerminateInstancesInput, natsConn *nats.Conn, } instanceID := *instanceIDPtr - // Build the QMP command to terminate the instance - // Note: stop_instance=true prevents restart on daemon/node restart - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - QMPCommand: qmp.QMPCommand{ - Execute: "system_powerdown", - Arguments: map[string]any{}, - }, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ StopInstance: true, // Prevent restart on daemon/node restart TerminateInstance: true, }, diff --git a/hive/gateway/ec2/instance/TerminateInstances_test.go b/hive/gateway/ec2/instance/TerminateInstances_test.go index c40be942d..5bd7aa75e 100644 --- a/hive/gateway/ec2/instance/TerminateInstances_test.go +++ b/hive/gateway/ec2/instance/TerminateInstances_test.go @@ -6,7 +6,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" - "github.com/mulgadc/hive/hive/qmp" + "github.com/mulgadc/hive/hive/types" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -18,12 +18,11 @@ func TestTerminateInstances_Success(t *testing.T) { instanceID := "i-0123456789abcdef0" nc.Subscribe("ec2.cmd."+instanceID, func(msg *nats.Msg) { - var cmd qmp.Command + var cmd types.EC2InstanceCommand err := json.Unmarshal(msg.Data, &cmd) require.NoError(t, err) assert.Equal(t, instanceID, cmd.ID) - assert.Equal(t, "system_powerdown", cmd.QMPCommand.Execute) assert.True(t, cmd.Attributes.StopInstance) assert.True(t, cmd.Attributes.TerminateInstance) @@ -161,7 +160,7 @@ func TestTerminateInstances_VerifiesQMPAttributes(t *testing.T) { _, nc := startTestNATSServer(t) instanceID := "i-verify" - var receivedCmd qmp.Command + var receivedCmd types.EC2InstanceCommand nc.Subscribe("ec2.cmd."+instanceID, func(msg *nats.Msg) { json.Unmarshal(msg.Data, &receivedCmd) diff --git a/hive/gateway/ec2/volume/AttachVolume.go b/hive/gateway/ec2/volume/AttachVolume.go index 77f0b1811..805768939 100644 --- a/hive/gateway/ec2/volume/AttachVolume.go +++ b/hive/gateway/ec2/volume/AttachVolume.go @@ -10,7 +10,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/hive/hive/awserrors" - "github.com/mulgadc/hive/hive/qmp" + "github.com/mulgadc/hive/hive/types" "github.com/mulgadc/hive/hive/utils" "github.com/nats-io/nats.go" ) @@ -48,12 +48,12 @@ func AttachVolume(input *ec2.AttachVolumeInput, natsConn *nats.Conn, accountID s device = *input.Device } - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ AttachVolume: true, }, - AttachVolumeData: &qmp.AttachVolumeData{ + AttachVolumeData: &types.AttachVolumeData{ VolumeID: volumeID, Device: device, }, diff --git a/hive/gateway/ec2/volume/DetachVolume.go b/hive/gateway/ec2/volume/DetachVolume.go index 2d6340f60..ea44df36d 100644 --- a/hive/gateway/ec2/volume/DetachVolume.go +++ b/hive/gateway/ec2/volume/DetachVolume.go @@ -10,7 +10,7 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/hive/hive/awserrors" handlers_ec2_volume "github.com/mulgadc/hive/hive/handlers/ec2/volume" - "github.com/mulgadc/hive/hive/qmp" + "github.com/mulgadc/hive/hive/types" "github.com/mulgadc/hive/hive/utils" "github.com/nats-io/nats.go" ) @@ -66,12 +66,12 @@ func DetachVolume(input *ec2.DetachVolumeInput, natsConn *nats.Conn, accountID s device = *input.Device } - command := qmp.Command{ + command := types.EC2InstanceCommand{ ID: instanceID, - Attributes: qmp.Attributes{ + Attributes: types.EC2CommandAttributes{ DetachVolume: true, }, - DetachVolumeData: &qmp.DetachVolumeData{ + DetachVolumeData: &types.DetachVolumeData{ VolumeID: volumeID, Device: device, Force: input.Force != nil && *input.Force, diff --git a/hive/gateway/gateway.go b/hive/gateway/gateway.go index b0b7ec753..c61055492 100644 --- a/hive/gateway/gateway.go +++ b/hive/gateway/gateway.go @@ -16,9 +16,9 @@ import ( "github.com/google/uuid" "github.com/mulgadc/hive/hive/awsec2query" "github.com/mulgadc/hive/hive/awserrors" - "github.com/mulgadc/hive/hive/types" "github.com/mulgadc/hive/hive/gateway/policy" handlers_iam "github.com/mulgadc/hive/hive/handlers/iam" + "github.com/mulgadc/hive/hive/types" "github.com/mulgadc/hive/hive/utils" "github.com/nats-io/nats.go" ) diff --git a/hive/handlers/ec2/image/service_impl.go b/hive/handlers/ec2/image/service_impl.go index 1eefc35d3..1541121ab 100644 --- a/hive/handlers/ec2/image/service_impl.go +++ b/hive/handlers/ec2/image/service_impl.go @@ -16,8 +16,8 @@ import ( "github.com/mulgadc/hive/hive/awserrors" "github.com/mulgadc/hive/hive/config" handlers_ec2_snapshot "github.com/mulgadc/hive/hive/handlers/ec2/snapshot" - "github.com/mulgadc/hive/hive/types" "github.com/mulgadc/hive/hive/objectstore" + "github.com/mulgadc/hive/hive/types" "github.com/mulgadc/hive/hive/utils" "github.com/mulgadc/viperblock/viperblock" vbs3 "github.com/mulgadc/viperblock/viperblock/backends/s3" diff --git a/hive/qmp/qmp.go b/hive/qmp/qmp.go index 85dcdfbba..80739b461 100644 --- a/hive/qmp/qmp.go +++ b/hive/qmp/qmp.go @@ -20,33 +20,6 @@ var CommandResponseTypes = map[string]unmarshalTarget{ "query-status": &Status{}, } -type Command struct { - ID string `json:"id"` - QMPCommand QMPCommand `json:"command"` - Attributes Attributes `json:"attributes"` - AttachVolumeData *AttachVolumeData `json:"attach_volume_data,omitempty"` - DetachVolumeData *DetachVolumeData `json:"detach_volume_data,omitempty"` -} - -type Attributes struct { - StopInstance bool `json:"stop_instance"` - TerminateInstance bool `json:"delete_instance"` - StartInstance bool `json:"start_instance"` - AttachVolume bool `json:"attach_volume"` - DetachVolume bool `json:"detach_volume"` -} - -type AttachVolumeData struct { - VolumeID string `json:"volume_id"` - Device string `json:"device,omitempty"` -} - -type DetachVolumeData struct { - VolumeID string `json:"volume_id"` - Device string `json:"device,omitempty"` - Force bool `json:"force,omitempty"` -} - type Status struct { Status string `json:"status"` Singlestep bool `json:"singlestep"` diff --git a/hive/services/viperblockd/viperblockd.go b/hive/services/viperblockd/viperblockd.go index 0a9d0e189..5115fa749 100644 --- a/hive/services/viperblockd/viperblockd.go +++ b/hive/services/viperblockd/viperblockd.go @@ -12,8 +12,8 @@ import ( "syscall" "time" - "github.com/mulgadc/hive/hive/types" "github.com/mulgadc/hive/hive/nbd" + "github.com/mulgadc/hive/hive/types" "github.com/mulgadc/hive/hive/utils" "github.com/mulgadc/viperblock/viperblock" "github.com/mulgadc/viperblock/viperblock/backends/s3" diff --git a/hive/types/ec2.go b/hive/types/ec2.go new file mode 100644 index 000000000..47f5e75d7 --- /dev/null +++ b/hive/types/ec2.go @@ -0,0 +1,33 @@ +package types + +// EC2InstanceCommand is the NATS wire format for EC2 instance commands +// (stop, terminate, start, attach-volume, detach-volume). +// It replaces direct use of qmp.Command on the gateway→daemon boundary. +type EC2InstanceCommand struct { + ID string `json:"id"` + Attributes EC2CommandAttributes `json:"attributes"` + AttachVolumeData *AttachVolumeData `json:"attach_volume_data,omitempty"` + DetachVolumeData *DetachVolumeData `json:"detach_volume_data,omitempty"` +} + +// EC2CommandAttributes indicates which action the daemon should perform. +type EC2CommandAttributes struct { + StopInstance bool `json:"stop_instance"` + TerminateInstance bool `json:"delete_instance"` + StartInstance bool `json:"start_instance"` + AttachVolume bool `json:"attach_volume"` + DetachVolume bool `json:"detach_volume"` +} + +// AttachVolumeData carries parameters for an attach-volume command. +type AttachVolumeData struct { + VolumeID string `json:"volume_id"` + Device string `json:"device,omitempty"` +} + +// DetachVolumeData carries parameters for a detach-volume command. +type DetachVolumeData struct { + VolumeID string `json:"volume_id"` + Device string `json:"device,omitempty"` + Force bool `json:"force,omitempty"` +} diff --git a/hive/vm/vm.go b/hive/vm/vm.go index 2349e312f..b62ff7df9 100644 --- a/hive/vm/vm.go +++ b/hive/vm/vm.go @@ -37,7 +37,7 @@ type VM struct { QMPClient *qmp.QMPClient `json:"-"` // User attributes (user initiated stop/delete) - Attributes qmp.Attributes `json:"attributes"` + Attributes types.EC2CommandAttributes `json:"attributes"` // EC2 API metadata - stored for AWS API compatibility // RunInstancesInput contains the original request parameters (ImageId, KeyName, UserData, etc.)