fix: decouple gateway from qmp

This commit is contained in:
Julian Sommer
2026-03-06 10:06:59 +10:00
parent 6737c42601
commit 14c52366da
19 changed files with 160 additions and 212 deletions
-1
View File
@@ -369,4 +369,3 @@ shardwal = true
require.NotNil(t, n.Viperblock.ShardWAL)
assert.True(t, *n.Viperblock.ShardWAL)
}
+3 -45
View File
@@ -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)
}
}
+4 -5
View File
@@ -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)
+35 -35
View File
@@ -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",
},
+3 -3
View File
@@ -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
+54 -54
View File
@@ -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,
},
}
+5 -5
View File
@@ -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())
+3 -9
View File
@@ -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,
},
@@ -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)
@@ -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,
},
@@ -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)
+4 -4
View File
@@ -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,
},
+4 -4
View File
@@ -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,
+1 -1
View File
@@ -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"
)
+1 -1
View File
@@ -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"
-27
View File
@@ -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"`
+1 -1
View File
@@ -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"
+33
View File
@@ -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"`
}
+1 -1
View File
@@ -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.)