mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
refactor: split daemon_handlers.go into per-domain files
Split the monolithic daemon_handlers.go into focused files: - daemon_handlers_account.go - daemon_handlers_console.go - daemon_handlers_igw.go - daemon_handlers_image.go - daemon_handlers_instance.go - daemon_handlers_key.go - daemon_handlers_snapshot.go - daemon_handlers_tag.go - daemon_handlers_volume.go - daemon_handlers_vpc.go Add tests for ModifyVolume success, TerminateStoppedInstance with volumes, DescribeInstanceTypes capacity filter, StartStoppedInstance type-not-available, AttachVolume/DetachVolume validation paths, RunInstances error paths, and DescribeInstances malformed ID. Fix diff-coverage.sh to use git rename/copy detection (-M -C) so file splits are not incorrectly flagged as all-new code.
This commit is contained in:
Julian Sommer
@@ -7,7 +7,6 @@ on:
|
||||
- dev
|
||||
paths:
|
||||
- "**/*.go"
|
||||
- "!**/*_test.go"
|
||||
- go.mod
|
||||
- go.sum
|
||||
- Makefile
|
||||
|
||||
+1
-1857
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,29 @@
|
||||
package daemon
|
||||
|
||||
import (
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
func (d *Daemon) handleEC2EnableEbsEncryptionByDefault(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.accountService.EnableEbsEncryptionByDefault)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DisableEbsEncryptionByDefault(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.accountService.DisableEbsEncryptionByDefault)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2GetEbsEncryptionByDefault(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.accountService.GetEbsEncryptionByDefault)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2GetSerialConsoleAccessStatus(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.accountService.GetSerialConsoleAccessStatus)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2EnableSerialConsoleAccess(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.accountService.EnableSerialConsoleAccess)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DisableSerialConsoleAccess(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.accountService.DisableSerialConsoleAccess)
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
package daemon
|
||||
|
||||
import (
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go/service/ec2"
|
||||
"github.com/mulgadc/hive/hive/awserrors"
|
||||
"github.com/mulgadc/hive/hive/utils"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// handleEC2GetConsoleOutput reads the console log file for an instance and returns
|
||||
// base64-encoded output matching the AWS GetConsoleOutput API response format.
|
||||
func (d *Daemon) handleEC2GetConsoleOutput(msg *nats.Msg) {
|
||||
slog.Debug("Received GetConsoleOutput request", "subject", msg.Subject, "data", string(msg.Data))
|
||||
|
||||
var input ec2.GetConsoleOutputInput
|
||||
if errResp := utils.UnmarshalJsonPayload(&input, msg.Data); errResp != nil {
|
||||
if err := msg.Respond(errResp); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if input.InstanceId == nil {
|
||||
errResp := utils.GenerateErrorPayload(awserrors.ErrorMissingParameter)
|
||||
if err := msg.Respond(errResp); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// Check account-level serial console access
|
||||
status, err := d.accountService.GetSerialConsoleAccessStatus(&ec2.GetSerialConsoleAccessStatusInput{})
|
||||
if err != nil {
|
||||
slog.Error("Failed to check serial console access status", "err", err)
|
||||
if err := msg.Respond(utils.GenerateErrorPayload(awserrors.ErrorServerInternal)); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
if status.SerialConsoleAccessEnabled == nil || !*status.SerialConsoleAccessEnabled {
|
||||
if err := msg.Respond(utils.GenerateErrorPayload(awserrors.ErrorSerialConsoleSessionUnavailable)); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
instanceID := *input.InstanceId
|
||||
|
||||
// Find the instance on this node
|
||||
d.Instances.Mu.Lock()
|
||||
instance, exists := d.Instances.VMS[instanceID]
|
||||
d.Instances.Mu.Unlock()
|
||||
|
||||
if !exists {
|
||||
errResp := utils.GenerateErrorPayload(awserrors.ErrorInvalidInstanceIDNotFound)
|
||||
if err := msg.Respond(errResp); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
logPath := instance.Config.ConsoleLogPath
|
||||
var outputData []byte
|
||||
var modTime time.Time
|
||||
|
||||
if logPath != "" {
|
||||
info, err := os.Stat(logPath)
|
||||
if err == nil {
|
||||
modTime = info.ModTime()
|
||||
|
||||
data, err := os.ReadFile(logPath)
|
||||
if err != nil {
|
||||
slog.Error("Failed to read console log", "path", logPath, "err", err)
|
||||
} else {
|
||||
// Return last 64KB (AWS limit)
|
||||
const maxConsoleOutput = 64 * 1024
|
||||
if len(data) > maxConsoleOutput {
|
||||
data = data[len(data)-maxConsoleOutput:]
|
||||
}
|
||||
outputData = data
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Base64-encode the output (AWS returns base64)
|
||||
var encodedOutput string
|
||||
if len(outputData) > 0 {
|
||||
encodedOutput = base64.StdEncoding.EncodeToString(outputData)
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
if modTime.IsZero() {
|
||||
modTime = now
|
||||
}
|
||||
|
||||
output := &ec2.GetConsoleOutputOutput{
|
||||
InstanceId: &instanceID,
|
||||
Output: &encodedOutput,
|
||||
Timestamp: &modTime,
|
||||
}
|
||||
|
||||
jsonResponse, err := json.Marshal(output)
|
||||
if err != nil {
|
||||
slog.Error("Failed to marshal GetConsoleOutput response", "err", err)
|
||||
errResp := utils.GenerateErrorPayload(awserrors.ErrorServerInternal)
|
||||
if err := msg.Respond(errResp); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if err := msg.Respond(jsonResponse); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
|
||||
slog.Info("handleEC2GetConsoleOutput completed", "instance_id", instanceID, "output_bytes", len(outputData))
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
package daemon
|
||||
|
||||
import (
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
func (d *Daemon) handleEC2CreateInternetGateway(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.igwService.CreateInternetGateway)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DeleteInternetGateway(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.igwService.DeleteInternetGateway)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DescribeInternetGateways(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.igwService.DescribeInternetGateways)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2AttachInternetGateway(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.igwService.AttachInternetGateway)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DetachInternetGateway(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.igwService.DetachInternetGateway)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2CreateEgressOnlyInternetGateway(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.eigwService.CreateEgressOnlyInternetGateway)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DeleteEgressOnlyInternetGateway(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.eigwService.DeleteEgressOnlyInternetGateway)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DescribeEgressOnlyInternetGateways(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.eigwService.DescribeEgressOnlyInternetGateways)
|
||||
}
|
||||
@@ -0,0 +1,110 @@
|
||||
package daemon
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
|
||||
"github.com/aws/aws-sdk-go/service/ec2"
|
||||
"github.com/mulgadc/hive/hive/awserrors"
|
||||
handlers_ec2_image "github.com/mulgadc/hive/hive/handlers/ec2/image"
|
||||
"github.com/mulgadc/hive/hive/utils"
|
||||
"github.com/mulgadc/hive/hive/vm"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
func (d *Daemon) handleEC2DescribeImages(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.imageService.DescribeImages)
|
||||
}
|
||||
|
||||
// handleEC2CreateImage is a stateful handler that extracts instance context
|
||||
// (root volume ID, source AMI, running state) before delegating to the image service.
|
||||
func (d *Daemon) handleEC2CreateImage(msg *nats.Msg) {
|
||||
slog.Debug("Received message", "subject", msg.Subject)
|
||||
|
||||
respondWithError := func(errCode string) {
|
||||
if err := msg.Respond(utils.GenerateErrorPayload(errCode)); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
input := &ec2.CreateImageInput{}
|
||||
if errResp := utils.UnmarshalJsonPayload(input, msg.Data); errResp != nil {
|
||||
if err := msg.Respond(errResp); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if input.InstanceId == nil || *input.InstanceId == "" {
|
||||
respondWithError(awserrors.ErrorMissingParameter)
|
||||
return
|
||||
}
|
||||
|
||||
instanceID := *input.InstanceId
|
||||
|
||||
// Extract all instance context in a single critical section
|
||||
d.Instances.Mu.Lock()
|
||||
instance, ok := d.Instances.VMS[instanceID]
|
||||
var status vm.InstanceState
|
||||
var rootVolumeID, sourceImageID string
|
||||
if ok {
|
||||
status = instance.Status
|
||||
if instance.Instance != nil {
|
||||
for _, bdm := range instance.Instance.BlockDeviceMappings {
|
||||
if bdm.Ebs != nil && bdm.Ebs.VolumeId != nil {
|
||||
rootVolumeID = *bdm.Ebs.VolumeId
|
||||
break
|
||||
}
|
||||
}
|
||||
if instance.Instance.ImageId != nil {
|
||||
sourceImageID = *instance.Instance.ImageId
|
||||
}
|
||||
}
|
||||
}
|
||||
d.Instances.Mu.Unlock()
|
||||
|
||||
if !ok {
|
||||
slog.Warn("CreateImage: instance not found", "instanceId", instanceID)
|
||||
respondWithError(awserrors.ErrorInvalidInstanceIDNotFound)
|
||||
return
|
||||
}
|
||||
|
||||
if status != vm.StateRunning && status != vm.StateStopped {
|
||||
slog.Warn("CreateImage: instance not in valid state", "instanceId", instanceID, "status", status)
|
||||
respondWithError(awserrors.ErrorIncorrectInstanceState)
|
||||
return
|
||||
}
|
||||
|
||||
if rootVolumeID == "" {
|
||||
slog.Error("CreateImage: no root volume found", "instanceId", instanceID)
|
||||
respondWithError(awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
|
||||
params := handlers_ec2_image.CreateImageParams{
|
||||
Input: input,
|
||||
RootVolumeID: rootVolumeID,
|
||||
SourceImageID: sourceImageID,
|
||||
IsRunning: status == vm.StateRunning,
|
||||
}
|
||||
|
||||
output, err := d.imageService.CreateImageFromInstance(params)
|
||||
if err != nil {
|
||||
slog.Error("CreateImage: service failed", "instanceId", instanceID, "err", err)
|
||||
respondWithError(err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
jsonResponse, err := json.Marshal(output)
|
||||
if err != nil {
|
||||
slog.Error("CreateImage: failed to marshal response", "err", err)
|
||||
respondWithError(awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
|
||||
if err := msg.Respond(jsonResponse); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
|
||||
slog.Info("CreateImage completed", "instanceId", instanceID, "imageId", *output.ImageId)
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,21 @@
|
||||
package daemon
|
||||
|
||||
import (
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
func (d *Daemon) handleEC2CreateKeyPair(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.keyService.CreateKeyPair)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DeleteKeyPair(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.keyService.DeleteKeyPair)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DescribeKeyPairs(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.keyService.DescribeKeyPairs)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2ImportKeyPair(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.keyService.ImportKeyPair)
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
package daemon
|
||||
|
||||
import (
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
func (d *Daemon) handleEC2CreateSnapshot(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.snapshotService.CreateSnapshot)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DescribeSnapshots(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.snapshotService.DescribeSnapshots)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DeleteSnapshot(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.snapshotService.DeleteSnapshot)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2CopySnapshot(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.snapshotService.CopySnapshot)
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
package daemon
|
||||
|
||||
import (
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
func (d *Daemon) handleEC2CreateTags(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.tagsService.CreateTags)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DeleteTags(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.tagsService.DeleteTags)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DescribeTags(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.tagsService.DescribeTags)
|
||||
}
|
||||
@@ -2519,3 +2519,739 @@ func TestDelegateHandlers_EIGW(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// --- handleEC2ModifyVolume success path ---
|
||||
|
||||
func TestHandleEC2ModifyVolume_Success(t *testing.T) {
|
||||
daemon, store := createFullTestDaemonWithStore(t, sharedNATSURL)
|
||||
|
||||
// Seed a volume in the store
|
||||
volumeID := "vol-modify-success"
|
||||
wrapper := struct {
|
||||
VolumeConfig viperblock.VolumeConfig `json:"VolumeConfig"`
|
||||
}{
|
||||
VolumeConfig: viperblock.VolumeConfig{
|
||||
VolumeMetadata: viperblock.VolumeMetadata{
|
||||
VolumeID: volumeID,
|
||||
SizeGiB: 10,
|
||||
State: "available",
|
||||
VolumeType: "gp3",
|
||||
},
|
||||
},
|
||||
}
|
||||
data, _ := json.Marshal(wrapper)
|
||||
store.PutObject(&awss3.PutObjectInput{
|
||||
Bucket: aws.String("test-bucket"),
|
||||
Key: aws.String(volumeID + "/config.json"),
|
||||
Body: strings.NewReader(string(data)),
|
||||
})
|
||||
|
||||
// Subscribe a dummy ebs.sync handler so the NATS Request doesn't time out
|
||||
syncSub, err := daemon.natsConn.Subscribe("ebs.sync", func(msg *nats.Msg) {
|
||||
_ = msg.Respond([]byte(`{}`))
|
||||
})
|
||||
require.NoError(t, err)
|
||||
defer syncSub.Unsubscribe()
|
||||
|
||||
sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyVolume", "hive-workers", daemon.handleEC2ModifyVolume)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
input := &ec2.ModifyVolumeInput{
|
||||
VolumeId: aws.String(volumeID),
|
||||
Size: aws.Int64(20),
|
||||
}
|
||||
reqData, _ := json.Marshal(input)
|
||||
reply, err := daemon.natsConn.Request("ec2.ModifyVolume", reqData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
|
||||
var output ec2.ModifyVolumeOutput
|
||||
err = json.Unmarshal(reply.Data, &output)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, output.VolumeModification)
|
||||
assert.Equal(t, volumeID, *output.VolumeModification.VolumeId)
|
||||
assert.Equal(t, int64(10), *output.VolumeModification.OriginalSize)
|
||||
assert.Equal(t, int64(20), *output.VolumeModification.TargetSize)
|
||||
assert.Equal(t, "completed", *output.VolumeModification.ModificationState)
|
||||
}
|
||||
|
||||
// --- handleEC2TerminateStoppedInstance with volumes ---
|
||||
|
||||
func TestHandleEC2TerminateStoppedInstance_WithVolumes(t *testing.T) {
|
||||
daemon := createFullTestDaemonWithJetStream(t, sharedJSNATSURL)
|
||||
|
||||
// Subscribe a dummy ebs.delete handler
|
||||
ebsDeleteSub, err := daemon.natsConn.Subscribe("ebs.delete", func(msg *nats.Msg) {
|
||||
_ = msg.Respond([]byte(`{"status":"deleted"}`))
|
||||
})
|
||||
require.NoError(t, err)
|
||||
defer ebsDeleteSub.Unsubscribe()
|
||||
|
||||
stoppedVM := &vm.VM{
|
||||
ID: "i-term-vol-001",
|
||||
Status: vm.StateStopped,
|
||||
InstanceType: getTestInstanceType(),
|
||||
LastNode: "node-1",
|
||||
Reservation: &ec2.Reservation{
|
||||
ReservationId: aws.String("r-term-vol-001"),
|
||||
OwnerId: aws.String("123456789012"),
|
||||
},
|
||||
Instance: &ec2.Instance{
|
||||
InstanceId: aws.String("i-term-vol-001"),
|
||||
InstanceType: aws.String(getTestInstanceType()),
|
||||
},
|
||||
}
|
||||
// Add EFI, CloudInit, and a user volume with DeleteOnTermination
|
||||
stoppedVM.EBSRequests.Requests = []config.EBSRequest{
|
||||
{Name: "vol-efi-001", EFI: true},
|
||||
{Name: "vol-ci-001", CloudInit: true},
|
||||
{Name: "vol-user-001", DeleteOnTermination: true},
|
||||
{Name: "vol-keep-001", DeleteOnTermination: false},
|
||||
}
|
||||
|
||||
err = daemon.jsManager.WriteStoppedInstance(stoppedVM.ID, stoppedVM)
|
||||
require.NoError(t, err)
|
||||
|
||||
sub, err := daemon.natsConn.QueueSubscribe("ec2.terminate", "hive-workers", daemon.handleEC2TerminateStoppedInstance)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
reqData, _ := json.Marshal(map[string]string{"instance_id": "i-term-vol-001"})
|
||||
reply, err := daemon.natsConn.Request("ec2.terminate", reqData, 10*time.Second)
|
||||
require.NoError(t, err)
|
||||
|
||||
var resp map[string]string
|
||||
err = json.Unmarshal(reply.Data, &resp)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "terminated", resp["status"])
|
||||
assert.Equal(t, "i-term-vol-001", resp["instanceId"])
|
||||
|
||||
loaded, err := daemon.jsManager.LoadStoppedInstance("i-term-vol-001")
|
||||
require.NoError(t, err)
|
||||
assert.Nil(t, loaded)
|
||||
}
|
||||
|
||||
// --- handleEC2DescribeInstanceTypes with capacity filter ---
|
||||
|
||||
func TestHandleEC2DescribeInstanceTypes_CapacityFilter(t *testing.T) {
|
||||
daemon := createFullTestDaemon(t, sharedNATSURL)
|
||||
|
||||
sub, err := daemon.natsConn.QueueSubscribe("ec2.DescribeInstanceTypes", "hive-workers", daemon.handleEC2DescribeInstanceTypes)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
// Request with capacity=true filter
|
||||
input := &ec2.DescribeInstanceTypesInput{
|
||||
Filters: []*ec2.Filter{
|
||||
{
|
||||
Name: aws.String("capacity"),
|
||||
Values: []*string{aws.String("true")},
|
||||
},
|
||||
},
|
||||
}
|
||||
reqData, _ := json.Marshal(input)
|
||||
reply, err := daemon.natsConn.Request("ec2.DescribeInstanceTypes", reqData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
|
||||
var output ec2.DescribeInstanceTypesOutput
|
||||
err = json.Unmarshal(reply.Data, &output)
|
||||
require.NoError(t, err)
|
||||
// Should return instance types that fit the node's available capacity
|
||||
assert.NotNil(t, output.InstanceTypes)
|
||||
}
|
||||
|
||||
func TestHandleEC2DescribeInstanceTypes_NoFilter(t *testing.T) {
|
||||
daemon := createFullTestDaemon(t, sharedNATSURL)
|
||||
|
||||
sub, err := daemon.natsConn.QueueSubscribe("ec2.DescribeInstanceTypes.nofilter", "hive-workers", daemon.handleEC2DescribeInstanceTypes)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
input := &ec2.DescribeInstanceTypesInput{}
|
||||
reqData, _ := json.Marshal(input)
|
||||
reply, err := daemon.natsConn.Request("ec2.DescribeInstanceTypes.nofilter", reqData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
|
||||
var output ec2.DescribeInstanceTypesOutput
|
||||
err = json.Unmarshal(reply.Data, &output)
|
||||
require.NoError(t, err)
|
||||
assert.NotNil(t, output.InstanceTypes)
|
||||
assert.Greater(t, len(output.InstanceTypes), 0)
|
||||
}
|
||||
|
||||
// --- handleEC2StartStoppedInstance: instance type not available ---
|
||||
|
||||
func TestHandleEC2StartStoppedInstance_InstanceTypeNotAvailable(t *testing.T) {
|
||||
daemon := createFullTestDaemonWithJetStream(t, sharedJSNATSURL)
|
||||
|
||||
stoppedVM := &vm.VM{
|
||||
ID: "i-start-badtype-001",
|
||||
Status: vm.StateStopped,
|
||||
InstanceType: "z99.nonexistent",
|
||||
Instance: &ec2.Instance{
|
||||
InstanceId: aws.String("i-start-badtype-001"),
|
||||
InstanceType: aws.String("z99.nonexistent"),
|
||||
},
|
||||
}
|
||||
err := daemon.jsManager.WriteStoppedInstance(stoppedVM.ID, stoppedVM)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = daemon.jsManager.DeleteStoppedInstance(stoppedVM.ID) })
|
||||
|
||||
sub, err := daemon.natsConn.QueueSubscribe("ec2.start", "hive-workers", daemon.handleEC2StartStoppedInstance)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
reqData, _ := json.Marshal(map[string]string{"instance_id": "i-start-badtype-001"})
|
||||
reply, err := daemon.natsConn.Request("ec2.start", reqData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
|
||||
var errResp map[string]any
|
||||
err = json.Unmarshal(reply.Data, &errResp)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, errResp, "Code")
|
||||
}
|
||||
|
||||
// --- handleEC2CreateImage: running instance with valid root volume ---
|
||||
|
||||
func TestHandleEC2CreateImage_RunningInstanceReachesService(t *testing.T) {
|
||||
daemon, store := createFullTestDaemonWithStore(t, sharedNATSURL)
|
||||
|
||||
instanceID := "i-create-image-running"
|
||||
rootVolumeID := "vol-root-img-001"
|
||||
sourceImageID := "ami-source-001"
|
||||
|
||||
// Seed a root volume config
|
||||
wrapper := struct {
|
||||
VolumeConfig viperblock.VolumeConfig `json:"VolumeConfig"`
|
||||
}{
|
||||
VolumeConfig: viperblock.VolumeConfig{
|
||||
VolumeMetadata: viperblock.VolumeMetadata{
|
||||
VolumeID: rootVolumeID,
|
||||
SizeGiB: 8,
|
||||
State: "in-use",
|
||||
},
|
||||
},
|
||||
}
|
||||
volData, _ := json.Marshal(wrapper)
|
||||
store.PutObject(&awss3.PutObjectInput{
|
||||
Bucket: aws.String("test-bucket"),
|
||||
Key: aws.String(rootVolumeID + "/config.json"),
|
||||
Body: strings.NewReader(string(volData)),
|
||||
})
|
||||
|
||||
daemon.Instances.Mu.Lock()
|
||||
daemon.Instances.VMS[instanceID] = &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: vm.StateRunning,
|
||||
InstanceType: getTestInstanceType(),
|
||||
Instance: &ec2.Instance{
|
||||
InstanceId: aws.String(instanceID),
|
||||
ImageId: aws.String(sourceImageID),
|
||||
BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{
|
||||
{
|
||||
DeviceName: aws.String("/dev/sda1"),
|
||||
Ebs: &ec2.EbsInstanceBlockDevice{
|
||||
VolumeId: aws.String(rootVolumeID),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
daemon.Instances.Mu.Unlock()
|
||||
|
||||
sub, err := daemon.natsConn.Subscribe(
|
||||
fmt.Sprintf("ec2.%s.CreateImage", instanceID),
|
||||
daemon.handleEC2CreateImage,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
input := &ec2.CreateImageInput{
|
||||
InstanceId: aws.String(instanceID),
|
||||
Name: aws.String("test-image-snapshot"),
|
||||
}
|
||||
reqData, _ := json.Marshal(input)
|
||||
reply, err := daemon.natsConn.Request(
|
||||
fmt.Sprintf("ec2.%s.CreateImage", instanceID),
|
||||
reqData,
|
||||
5*time.Second,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
// The service call may fail (no real S3 backend), but we've exercised
|
||||
// the handler path up through service invocation: instance lookup,
|
||||
// state validation, root volume extraction, params construction.
|
||||
assert.NotEmpty(t, reply.Data)
|
||||
}
|
||||
|
||||
// --- handleAttachVolume: missing volume data ---
|
||||
|
||||
func TestAttachVolume_MissingVolumeData(t *testing.T) {
|
||||
daemon := createFullTestDaemon(t, sharedNATSURL)
|
||||
|
||||
instanceID := "i-attach-missing-vol"
|
||||
instance := &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: vm.StateRunning,
|
||||
InstanceType: getTestInstanceType(),
|
||||
Instance: &ec2.Instance{},
|
||||
QMPClient: &qmp.QMPClient{},
|
||||
}
|
||||
daemon.Instances.VMS[instanceID] = instance
|
||||
|
||||
sub, err := daemon.natsConn.Subscribe(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
daemon.handleEC2Events,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
// AttachVolume with nil AttachVolumeData
|
||||
command := qmp.Command{
|
||||
ID: instanceID,
|
||||
Attributes: qmp.Attributes{
|
||||
AttachVolume: true,
|
||||
},
|
||||
AttachVolumeData: nil,
|
||||
}
|
||||
cmdData, _ := json.Marshal(command)
|
||||
|
||||
resp, err := daemon.natsConn.Request(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
cmdData,
|
||||
5*time.Second,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, string(resp.Data), "InvalidParameterValue")
|
||||
}
|
||||
|
||||
func TestAttachVolume_InstanceNotRunning(t *testing.T) {
|
||||
daemon := createFullTestDaemon(t, sharedNATSURL)
|
||||
|
||||
instanceID := "i-attach-stopped"
|
||||
instance := &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: vm.StateStopped,
|
||||
InstanceType: getTestInstanceType(),
|
||||
Instance: &ec2.Instance{},
|
||||
QMPClient: &qmp.QMPClient{},
|
||||
}
|
||||
daemon.Instances.VMS[instanceID] = instance
|
||||
|
||||
sub, err := daemon.natsConn.Subscribe(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
daemon.handleEC2Events,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
command := qmp.Command{
|
||||
ID: instanceID,
|
||||
Attributes: qmp.Attributes{
|
||||
AttachVolume: true,
|
||||
},
|
||||
AttachVolumeData: &qmp.AttachVolumeData{
|
||||
VolumeID: "vol-test-123",
|
||||
},
|
||||
}
|
||||
cmdData, _ := json.Marshal(command)
|
||||
|
||||
resp, err := daemon.natsConn.Request(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
cmdData,
|
||||
5*time.Second,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, string(resp.Data), "IncorrectInstanceState")
|
||||
}
|
||||
|
||||
func TestAttachVolume_VolumeNotFound(t *testing.T) {
|
||||
daemon, _ := createFullTestDaemonWithStore(t, sharedNATSURL)
|
||||
|
||||
instanceID := "i-attach-vol-notfound"
|
||||
instance := &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: vm.StateRunning,
|
||||
InstanceType: getTestInstanceType(),
|
||||
Instance: &ec2.Instance{},
|
||||
QMPClient: &qmp.QMPClient{},
|
||||
}
|
||||
daemon.Instances.VMS[instanceID] = instance
|
||||
|
||||
sub, err := daemon.natsConn.Subscribe(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
daemon.handleEC2Events,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
command := qmp.Command{
|
||||
ID: instanceID,
|
||||
Attributes: qmp.Attributes{
|
||||
AttachVolume: true,
|
||||
},
|
||||
AttachVolumeData: &qmp.AttachVolumeData{
|
||||
VolumeID: "vol-nonexistent-999",
|
||||
},
|
||||
}
|
||||
cmdData, _ := json.Marshal(command)
|
||||
|
||||
resp, err := daemon.natsConn.Request(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
cmdData,
|
||||
5*time.Second,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, string(resp.Data), "InvalidVolume.NotFound")
|
||||
}
|
||||
|
||||
func TestAttachVolume_VolumeInUse(t *testing.T) {
|
||||
daemon, store := createFullTestDaemonWithStore(t, sharedNATSURL)
|
||||
|
||||
instanceID := "i-attach-vol-inuse"
|
||||
volumeID := "vol-in-use-001"
|
||||
|
||||
instance := &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: vm.StateRunning,
|
||||
InstanceType: getTestInstanceType(),
|
||||
Instance: &ec2.Instance{},
|
||||
QMPClient: &qmp.QMPClient{},
|
||||
}
|
||||
daemon.Instances.VMS[instanceID] = instance
|
||||
|
||||
// Seed a volume that is already in-use
|
||||
wrapper := struct {
|
||||
VolumeConfig viperblock.VolumeConfig `json:"VolumeConfig"`
|
||||
}{
|
||||
VolumeConfig: viperblock.VolumeConfig{
|
||||
VolumeMetadata: viperblock.VolumeMetadata{
|
||||
VolumeID: volumeID,
|
||||
SizeGiB: 10,
|
||||
State: "in-use",
|
||||
},
|
||||
},
|
||||
}
|
||||
data, _ := json.Marshal(wrapper)
|
||||
store.PutObject(&awss3.PutObjectInput{
|
||||
Bucket: aws.String("test-bucket"),
|
||||
Key: aws.String(volumeID + "/config.json"),
|
||||
Body: strings.NewReader(string(data)),
|
||||
})
|
||||
|
||||
sub, err := daemon.natsConn.Subscribe(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
daemon.handleEC2Events,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
command := qmp.Command{
|
||||
ID: instanceID,
|
||||
Attributes: qmp.Attributes{
|
||||
AttachVolume: true,
|
||||
},
|
||||
AttachVolumeData: &qmp.AttachVolumeData{
|
||||
VolumeID: volumeID,
|
||||
},
|
||||
}
|
||||
cmdData, _ := json.Marshal(command)
|
||||
|
||||
resp, err := daemon.natsConn.Request(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
cmdData,
|
||||
5*time.Second,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, string(resp.Data), "VolumeInUse")
|
||||
}
|
||||
|
||||
// --- handleEC2Events: detach volume validation ---
|
||||
|
||||
func TestDetachVolume_MissingVolumeData(t *testing.T) {
|
||||
daemon := createFullTestDaemon(t, sharedNATSURL)
|
||||
|
||||
instanceID := "i-detach-missing"
|
||||
instance := &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: vm.StateRunning,
|
||||
InstanceType: getTestInstanceType(),
|
||||
Instance: &ec2.Instance{},
|
||||
QMPClient: &qmp.QMPClient{},
|
||||
}
|
||||
daemon.Instances.VMS[instanceID] = instance
|
||||
|
||||
sub, err := daemon.natsConn.Subscribe(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
daemon.handleEC2Events,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
command := qmp.Command{
|
||||
ID: instanceID,
|
||||
Attributes: qmp.Attributes{
|
||||
DetachVolume: true,
|
||||
},
|
||||
DetachVolumeData: nil,
|
||||
}
|
||||
cmdData, _ := json.Marshal(command)
|
||||
|
||||
resp, err := daemon.natsConn.Request(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
cmdData,
|
||||
5*time.Second,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, string(resp.Data), "InvalidParameterValue")
|
||||
}
|
||||
|
||||
func TestDetachVolume_InstanceNotRunning(t *testing.T) {
|
||||
daemon := createFullTestDaemon(t, sharedNATSURL)
|
||||
|
||||
instanceID := "i-detach-stopped"
|
||||
instance := &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: vm.StateStopped,
|
||||
InstanceType: getTestInstanceType(),
|
||||
Instance: &ec2.Instance{},
|
||||
QMPClient: &qmp.QMPClient{},
|
||||
}
|
||||
daemon.Instances.VMS[instanceID] = instance
|
||||
|
||||
sub, err := daemon.natsConn.Subscribe(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
daemon.handleEC2Events,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
command := qmp.Command{
|
||||
ID: instanceID,
|
||||
Attributes: qmp.Attributes{
|
||||
DetachVolume: true,
|
||||
},
|
||||
DetachVolumeData: &qmp.DetachVolumeData{
|
||||
VolumeID: "vol-test-123",
|
||||
},
|
||||
}
|
||||
cmdData, _ := json.Marshal(command)
|
||||
|
||||
resp, err := daemon.natsConn.Request(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
cmdData,
|
||||
5*time.Second,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, string(resp.Data), "IncorrectInstanceState")
|
||||
}
|
||||
|
||||
func TestDetachVolume_VolumeNotAttached(t *testing.T) {
|
||||
daemon := createFullTestDaemon(t, sharedNATSURL)
|
||||
|
||||
instanceID := "i-detach-notattached"
|
||||
instance := &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: vm.StateRunning,
|
||||
InstanceType: getTestInstanceType(),
|
||||
Instance: &ec2.Instance{},
|
||||
QMPClient: &qmp.QMPClient{},
|
||||
}
|
||||
daemon.Instances.VMS[instanceID] = instance
|
||||
|
||||
sub, err := daemon.natsConn.Subscribe(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
daemon.handleEC2Events,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
command := qmp.Command{
|
||||
ID: instanceID,
|
||||
Attributes: qmp.Attributes{
|
||||
DetachVolume: true,
|
||||
},
|
||||
DetachVolumeData: &qmp.DetachVolumeData{
|
||||
VolumeID: "vol-not-attached-999",
|
||||
},
|
||||
}
|
||||
cmdData, _ := json.Marshal(command)
|
||||
|
||||
resp, err := daemon.natsConn.Request(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
cmdData,
|
||||
5*time.Second,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, string(resp.Data), "IncorrectState")
|
||||
}
|
||||
|
||||
func TestDetachVolume_BootVolumeRejected(t *testing.T) {
|
||||
daemon := createFullTestDaemon(t, sharedNATSURL)
|
||||
|
||||
instanceID := "i-detach-boot"
|
||||
instance := &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: vm.StateRunning,
|
||||
InstanceType: getTestInstanceType(),
|
||||
Instance: &ec2.Instance{},
|
||||
QMPClient: &qmp.QMPClient{},
|
||||
}
|
||||
instance.EBSRequests.Requests = []config.EBSRequest{
|
||||
{Name: "vol-boot-001", Boot: true, DeviceName: "/dev/sda1"},
|
||||
}
|
||||
daemon.Instances.VMS[instanceID] = instance
|
||||
|
||||
sub, err := daemon.natsConn.Subscribe(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
daemon.handleEC2Events,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
command := qmp.Command{
|
||||
ID: instanceID,
|
||||
Attributes: qmp.Attributes{
|
||||
DetachVolume: true,
|
||||
},
|
||||
DetachVolumeData: &qmp.DetachVolumeData{
|
||||
VolumeID: "vol-boot-001",
|
||||
},
|
||||
}
|
||||
cmdData, _ := json.Marshal(command)
|
||||
|
||||
resp, err := daemon.natsConn.Request(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
cmdData,
|
||||
5*time.Second,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, string(resp.Data), "OperationNotPermitted")
|
||||
}
|
||||
|
||||
func TestDetachVolume_DeviceMismatch(t *testing.T) {
|
||||
daemon := createFullTestDaemon(t, sharedNATSURL)
|
||||
|
||||
instanceID := "i-detach-devmismatch"
|
||||
instance := &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: vm.StateRunning,
|
||||
InstanceType: getTestInstanceType(),
|
||||
Instance: &ec2.Instance{},
|
||||
QMPClient: &qmp.QMPClient{},
|
||||
}
|
||||
instance.EBSRequests.Requests = []config.EBSRequest{
|
||||
{Name: "vol-mismatch-001", DeviceName: "/dev/sdf"},
|
||||
}
|
||||
daemon.Instances.VMS[instanceID] = instance
|
||||
|
||||
sub, err := daemon.natsConn.Subscribe(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
daemon.handleEC2Events,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
command := qmp.Command{
|
||||
ID: instanceID,
|
||||
Attributes: qmp.Attributes{
|
||||
DetachVolume: true,
|
||||
},
|
||||
DetachVolumeData: &qmp.DetachVolumeData{
|
||||
VolumeID: "vol-mismatch-001",
|
||||
Device: "/dev/sdg",
|
||||
},
|
||||
}
|
||||
cmdData, _ := json.Marshal(command)
|
||||
|
||||
resp, err := daemon.natsConn.Request(
|
||||
fmt.Sprintf("ec2.cmd.%s", instanceID),
|
||||
cmdData,
|
||||
5*time.Second,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, string(resp.Data), "InvalidParameterValue")
|
||||
}
|
||||
|
||||
// --- handleEC2RunInstances: insufficient capacity ---
|
||||
|
||||
func TestHandleEC2RunInstances_InsufficientCapacity(t *testing.T) {
|
||||
daemon := createFullTestDaemon(t, sharedNATSURL)
|
||||
|
||||
sub, err := daemon.natsConn.QueueSubscribe("ec2.RunInstances", "hive-workers", daemon.handleEC2RunInstances)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
// Request a very large instance count that can't be satisfied
|
||||
input := &ec2.RunInstancesInput{
|
||||
ImageId: aws.String("ami-test"),
|
||||
InstanceType: aws.String(getTestInstanceType()),
|
||||
MinCount: aws.Int64(9999),
|
||||
MaxCount: aws.Int64(9999),
|
||||
}
|
||||
reqData, _ := json.Marshal(input)
|
||||
reply, err := daemon.natsConn.Request("ec2.RunInstances", reqData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
|
||||
var errResp map[string]any
|
||||
err = json.Unmarshal(reply.Data, &errResp)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, errResp, "Code")
|
||||
}
|
||||
|
||||
func TestHandleEC2RunInstances_UnsupportedInstanceType(t *testing.T) {
|
||||
daemon := createFullTestDaemon(t, sharedNATSURL)
|
||||
|
||||
sub, err := daemon.natsConn.QueueSubscribe("ec2.RunInstances.badtype", "hive-workers", daemon.handleEC2RunInstances)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
input := &ec2.RunInstancesInput{
|
||||
ImageId: aws.String("ami-test"),
|
||||
InstanceType: aws.String("z99.nonexistent"),
|
||||
MinCount: aws.Int64(1),
|
||||
MaxCount: aws.Int64(1),
|
||||
}
|
||||
reqData, _ := json.Marshal(input)
|
||||
reply, err := daemon.natsConn.Request("ec2.RunInstances.badtype", reqData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
|
||||
var errResp map[string]any
|
||||
err = json.Unmarshal(reply.Data, &errResp)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, errResp, "Code")
|
||||
}
|
||||
|
||||
func TestHandleEC2RunInstances_MalformedInput(t *testing.T) {
|
||||
daemon := createFullTestDaemon(t, sharedNATSURL)
|
||||
|
||||
sub, err := daemon.natsConn.QueueSubscribe("ec2.RunInstances.bad", "hive-workers", daemon.handleEC2RunInstances)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
reply, err := daemon.natsConn.Request("ec2.RunInstances.bad", []byte(`{not valid}`), 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
|
||||
var errResp map[string]any
|
||||
err = json.Unmarshal(reply.Data, &errResp)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, errResp, "Code")
|
||||
}
|
||||
|
||||
// --- handleEC2DescribeInstances: malformed instance ID ---
|
||||
|
||||
func TestHandleEC2DescribeInstances_MalformedInstanceID(t *testing.T) {
|
||||
daemon := createFullTestDaemon(t, sharedNATSURL)
|
||||
|
||||
sub, err := daemon.natsConn.Subscribe("ec2.DescribeInstances.malformed", daemon.handleEC2DescribeInstances)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
input := &ec2.DescribeInstancesInput{
|
||||
InstanceIds: []*string{aws.String("not-an-instance-id")},
|
||||
}
|
||||
reqData, _ := json.Marshal(input)
|
||||
reply, err := daemon.natsConn.Request("ec2.DescribeInstances.malformed", reqData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Contains(t, string(reply.Data), "InvalidInstanceID.Malformed")
|
||||
}
|
||||
|
||||
@@ -0,0 +1,492 @@
|
||||
package daemon
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go/service/ec2"
|
||||
"github.com/mulgadc/hive/hive/awserrors"
|
||||
"github.com/mulgadc/hive/hive/config"
|
||||
"github.com/mulgadc/hive/hive/qmp"
|
||||
"github.com/mulgadc/hive/hive/utils"
|
||||
"github.com/mulgadc/hive/hive/vm"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// handleAttachVolume performs a three-phase hot-plug:
|
||||
//
|
||||
// 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, respondWithError func(string)) {
|
||||
slog.Info("Attaching volume to instance", "instanceId", command.ID)
|
||||
|
||||
// Validate AttachVolumeData
|
||||
if command.AttachVolumeData == nil || command.AttachVolumeData.VolumeID == "" {
|
||||
slog.Error("AttachVolume: missing attach volume data")
|
||||
respondWithError(awserrors.ErrorInvalidParameterValue)
|
||||
return
|
||||
}
|
||||
|
||||
volumeID := command.AttachVolumeData.VolumeID
|
||||
device := command.AttachVolumeData.Device
|
||||
|
||||
// Validate instance is running
|
||||
d.Instances.Mu.Lock()
|
||||
status := instance.Status
|
||||
d.Instances.Mu.Unlock()
|
||||
|
||||
if status != vm.StateRunning {
|
||||
slog.Error("AttachVolume: instance not running", "instanceId", command.ID, "status", status)
|
||||
respondWithError(awserrors.ErrorIncorrectInstanceState)
|
||||
return
|
||||
}
|
||||
|
||||
// Validate volume exists and is available
|
||||
volCfg, err := d.volumeService.GetVolumeConfig(volumeID)
|
||||
if err != nil {
|
||||
slog.Error("AttachVolume: failed to get volume config", "volumeId", volumeID, "err", err)
|
||||
respondWithError(awserrors.ErrorInvalidVolumeNotFound)
|
||||
return
|
||||
}
|
||||
|
||||
if volCfg.VolumeMetadata.State != "available" {
|
||||
slog.Error("AttachVolume: volume not available", "volumeId", volumeID, "state", volCfg.VolumeMetadata.State)
|
||||
respondWithError(awserrors.ErrorVolumeInUse)
|
||||
return
|
||||
}
|
||||
|
||||
// Check AZ compatibility — volume and instance must be in the same AZ
|
||||
if volCfg.VolumeMetadata.AvailabilityZone != "" && d.config.AZ != "" &&
|
||||
volCfg.VolumeMetadata.AvailabilityZone != d.config.AZ {
|
||||
slog.Error("AttachVolume: volume and instance are in different AZs",
|
||||
"volumeId", volumeID, "volumeAZ", volCfg.VolumeMetadata.AvailabilityZone,
|
||||
"instanceAZ", d.config.AZ)
|
||||
respondWithError(awserrors.ErrorInvalidVolumeZoneMismatch)
|
||||
return
|
||||
}
|
||||
|
||||
// Determine device name
|
||||
if device == "" {
|
||||
d.Instances.Mu.Lock()
|
||||
device = nextAvailableDevice(instance)
|
||||
d.Instances.Mu.Unlock()
|
||||
if device == "" {
|
||||
slog.Error("AttachVolume: no available device names")
|
||||
respondWithError(awserrors.ErrorAttachmentLimitExceeded)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Create EBS mount request
|
||||
ebsRequest := config.EBSRequest{
|
||||
Name: volumeID,
|
||||
DeviceName: device,
|
||||
}
|
||||
|
||||
ebsMountData, err := json.Marshal(ebsRequest)
|
||||
if err != nil {
|
||||
slog.Error("AttachVolume: failed to marshal ebs.mount request", "err", err)
|
||||
respondWithError(awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
|
||||
mountReply, err := d.natsConn.Request(d.ebsTopic("mount"), ebsMountData, 30*time.Second)
|
||||
if err != nil {
|
||||
slog.Error("AttachVolume: ebs.mount failed", "volumeId", volumeID, "err", err)
|
||||
respondWithError(awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
|
||||
var mountResp config.EBSMountResponse
|
||||
if err := json.Unmarshal(mountReply.Data, &mountResp); err != nil {
|
||||
slog.Error("AttachVolume: failed to unmarshal mount response", "err", err)
|
||||
respondWithError(awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
|
||||
if mountResp.Error != "" {
|
||||
slog.Error("AttachVolume: mount returned error", "volumeId", volumeID, "err", mountResp.Error)
|
||||
respondWithError(awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
|
||||
nbdURI := mountResp.URI
|
||||
if nbdURI == "" {
|
||||
slog.Error("AttachVolume: mount response has empty URI", "volumeId", volumeID)
|
||||
d.rollbackEBSMount(ebsRequest)
|
||||
respondWithError(awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
ebsRequest.NBDURI = nbdURI
|
||||
|
||||
// Parse NBDURI for QMP blockdev-add
|
||||
serverType, socketPath, nbdHost, nbdPort, err := utils.ParseNBDURI(nbdURI)
|
||||
if err != nil {
|
||||
slog.Error("AttachVolume: failed to parse NBDURI", "uri", nbdURI, "err", err)
|
||||
// Rollback: unmount
|
||||
d.rollbackEBSMount(ebsRequest)
|
||||
respondWithError(awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
|
||||
// Build QMP server argument
|
||||
var serverArg map[string]any
|
||||
if serverType == "unix" {
|
||||
serverArg = map[string]any{"type": "unix", "path": socketPath}
|
||||
} else {
|
||||
serverArg = map[string]any{"type": "inet", "host": nbdHost, "port": fmt.Sprintf("%d", nbdPort)}
|
||||
}
|
||||
|
||||
nodeName := fmt.Sprintf("nbd-%s", volumeID)
|
||||
deviceID := fmt.Sprintf("vdisk-%s", volumeID)
|
||||
iothreadID := fmt.Sprintf("ioth-%s", volumeID)
|
||||
|
||||
// QMP object-add: create iothread for this volume
|
||||
iothreadCmd := qmp.QMPCommand{
|
||||
Execute: "object-add",
|
||||
Arguments: map[string]any{
|
||||
"qom-type": "iothread",
|
||||
"id": iothreadID,
|
||||
},
|
||||
}
|
||||
_, err = d.SendQMPCommand(instance.QMPClient, iothreadCmd, instance.ID)
|
||||
if err != nil {
|
||||
slog.Error("AttachVolume: QMP object-add iothread failed", "volumeId", volumeID, "err", err)
|
||||
d.rollbackEBSMount(ebsRequest)
|
||||
respondWithError(awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
|
||||
// QMP blockdev-add
|
||||
blockdevCmd := qmp.QMPCommand{
|
||||
Execute: "blockdev-add",
|
||||
Arguments: map[string]any{
|
||||
"node-name": nodeName,
|
||||
"driver": "nbd",
|
||||
"server": serverArg,
|
||||
"export": "",
|
||||
"read-only": false,
|
||||
},
|
||||
}
|
||||
|
||||
_, err = d.SendQMPCommand(instance.QMPClient, blockdevCmd, instance.ID)
|
||||
if err != nil {
|
||||
slog.Error("AttachVolume: QMP blockdev-add failed", "volumeId", volumeID, "err", err)
|
||||
d.rollbackEBSMount(ebsRequest)
|
||||
respondWithError(awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
|
||||
// Determine which hotplug root port to use based on device letter.
|
||||
// /dev/sdf -> hotplug1, /dev/sdg -> hotplug2, etc.
|
||||
hotplugBus := ""
|
||||
if len(device) > 0 {
|
||||
letter := device[len(device)-1]
|
||||
if letter >= 'f' && letter <= 'p' {
|
||||
hotplugBus = fmt.Sprintf("hotplug%d", letter-'f'+1)
|
||||
}
|
||||
}
|
||||
|
||||
// QMP device_add
|
||||
deviceAddArgs := map[string]any{
|
||||
"driver": "virtio-blk-pci",
|
||||
"id": deviceID,
|
||||
"drive": nodeName,
|
||||
"iothread": iothreadID,
|
||||
}
|
||||
if hotplugBus != "" {
|
||||
deviceAddArgs["bus"] = hotplugBus
|
||||
}
|
||||
deviceAddCmd := qmp.QMPCommand{
|
||||
Execute: "device_add",
|
||||
Arguments: deviceAddArgs,
|
||||
}
|
||||
|
||||
_, err = d.SendQMPCommand(instance.QMPClient, deviceAddCmd, instance.ID)
|
||||
if err != nil {
|
||||
slog.Error("AttachVolume: QMP device_add failed, rolling back blockdev", "volumeId", volumeID, "err", err)
|
||||
// Rollback: blockdev-del then unmount
|
||||
if _, delErr := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{
|
||||
Execute: "blockdev-del",
|
||||
Arguments: map[string]any{"node-name": nodeName},
|
||||
}, instance.ID); delErr != nil {
|
||||
slog.Error("AttachVolume: rollback blockdev-del failed, skipping EBS unmount", "volumeId", volumeID, "err", delErr)
|
||||
} else {
|
||||
d.rollbackEBSMount(ebsRequest)
|
||||
}
|
||||
respondWithError(awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
|
||||
// Update instance state: replace existing entry for this volume (handles
|
||||
// stop/start cycles that keep stale entries) or append a new one.
|
||||
instance.EBSRequests.Mu.Lock()
|
||||
replaced := false
|
||||
for i, req := range instance.EBSRequests.Requests {
|
||||
if req.Name == volumeID {
|
||||
instance.EBSRequests.Requests[i] = ebsRequest
|
||||
replaced = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !replaced {
|
||||
instance.EBSRequests.Requests = append(instance.EBSRequests.Requests, ebsRequest)
|
||||
}
|
||||
instance.EBSRequests.Mu.Unlock()
|
||||
|
||||
// Update BlockDeviceMappings on the ec2.Instance
|
||||
d.Instances.Mu.Lock()
|
||||
if instance.Instance != nil {
|
||||
now := time.Now()
|
||||
mapping := &ec2.InstanceBlockDeviceMapping{}
|
||||
mapping.SetDeviceName(device)
|
||||
mapping.Ebs = &ec2.EbsInstanceBlockDevice{}
|
||||
mapping.Ebs.SetVolumeId(volumeID)
|
||||
mapping.Ebs.SetAttachTime(now)
|
||||
mapping.Ebs.SetDeleteOnTermination(false)
|
||||
mapping.Ebs.SetStatus("attached")
|
||||
instance.Instance.BlockDeviceMappings = append(instance.Instance.BlockDeviceMappings, mapping)
|
||||
}
|
||||
d.Instances.Mu.Unlock()
|
||||
|
||||
// Update volume metadata in S3
|
||||
if err := d.volumeService.UpdateVolumeState(volumeID, "in-use", command.ID, device); err != nil {
|
||||
slog.Error("AttachVolume: failed to update volume metadata", "volumeId", volumeID, "err", err)
|
||||
}
|
||||
|
||||
// Persist state
|
||||
if err := d.WriteState(); err != nil {
|
||||
slog.Error("AttachVolume: failed to write state", "err", err)
|
||||
}
|
||||
|
||||
d.respondWithVolumeAttachment(msg, respondWithError, volumeID, command.ID, device, "attached")
|
||||
slog.Info("Volume attached successfully", "volumeId", volumeID, "instanceId", command.ID, "device", device)
|
||||
}
|
||||
|
||||
// handleDetachVolume performs a three-phase hot-unplug (reverse of attach):
|
||||
//
|
||||
// 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, respondWithError func(string)) {
|
||||
slog.Info("Detaching volume from instance", "instanceId", command.ID)
|
||||
|
||||
// Validate DetachVolumeData
|
||||
if command.DetachVolumeData == nil || command.DetachVolumeData.VolumeID == "" {
|
||||
slog.Error("DetachVolume: missing detach volume data")
|
||||
respondWithError(awserrors.ErrorInvalidParameterValue)
|
||||
return
|
||||
}
|
||||
|
||||
volumeID := command.DetachVolumeData.VolumeID
|
||||
device := command.DetachVolumeData.Device
|
||||
force := command.DetachVolumeData.Force
|
||||
|
||||
// Validate instance is running
|
||||
d.Instances.Mu.Lock()
|
||||
status := instance.Status
|
||||
d.Instances.Mu.Unlock()
|
||||
|
||||
if status != vm.StateRunning {
|
||||
slog.Error("DetachVolume: instance not running", "instanceId", command.ID, "status", status)
|
||||
respondWithError(awserrors.ErrorIncorrectInstanceState)
|
||||
return
|
||||
}
|
||||
|
||||
// Find the volume in EBSRequests
|
||||
instance.EBSRequests.Mu.Lock()
|
||||
var ebsReq config.EBSRequest
|
||||
found := false
|
||||
for _, req := range instance.EBSRequests.Requests {
|
||||
if req.Name == volumeID {
|
||||
ebsReq = req
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
instance.EBSRequests.Mu.Unlock()
|
||||
|
||||
if !found {
|
||||
slog.Error("DetachVolume: volume not attached to instance", "volumeId", volumeID, "instanceId", command.ID)
|
||||
respondWithError(awserrors.ErrorIncorrectState)
|
||||
return
|
||||
}
|
||||
|
||||
// Reject detaching boot/EFI/CloudInit volumes
|
||||
if ebsReq.Boot || ebsReq.EFI || ebsReq.CloudInit {
|
||||
slog.Error("DetachVolume: cannot detach boot/EFI/CloudInit volume", "volumeId", volumeID)
|
||||
respondWithError(awserrors.ErrorOperationNotPermitted)
|
||||
return
|
||||
}
|
||||
|
||||
// Optional device cross-check
|
||||
if device != "" && ebsReq.DeviceName != "" && device != ebsReq.DeviceName {
|
||||
slog.Error("DetachVolume: device mismatch", "requested", device, "actual", ebsReq.DeviceName)
|
||||
respondWithError(awserrors.ErrorInvalidParameterValue)
|
||||
return
|
||||
}
|
||||
|
||||
deviceID := fmt.Sprintf("vdisk-%s", volumeID)
|
||||
nodeName := fmt.Sprintf("nbd-%s", volumeID)
|
||||
iothreadID := fmt.Sprintf("ioth-%s", volumeID)
|
||||
|
||||
// Phase 1: QMP device_del (remove guest device)
|
||||
_, err := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{
|
||||
Execute: "device_del",
|
||||
Arguments: map[string]any{"id": deviceID},
|
||||
}, instance.ID)
|
||||
if err != nil {
|
||||
if !force {
|
||||
slog.Error("DetachVolume: QMP device_del failed", "volumeId", volumeID, "err", err)
|
||||
respondWithError(awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
slog.Warn("DetachVolume: QMP device_del failed (force=true, continuing)", "volumeId", volumeID, "err", err)
|
||||
}
|
||||
|
||||
// Brief pause for guest to acknowledge PCI removal
|
||||
time.Sleep(d.detachDelay)
|
||||
|
||||
// Phase 2: QMP blockdev-del (remove block node)
|
||||
_, blockdevErr := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{
|
||||
Execute: "blockdev-del",
|
||||
Arguments: map[string]any{"node-name": nodeName},
|
||||
}, instance.ID)
|
||||
if blockdevErr != nil {
|
||||
// Block node still referenced by QEMU; do not clean up state or unmount —
|
||||
// tearing down the NBD server would crash the VM, and removing metadata
|
||||
// would allow the volume to be double-attached.
|
||||
slog.Error("DetachVolume: QMP blockdev-del failed, leaving volume state intact", "volumeId", volumeID, "err", blockdevErr)
|
||||
respondWithError(awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
|
||||
// Phase 2b: QMP object-del (remove iothread, best-effort)
|
||||
_, iothreadErr := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{
|
||||
Execute: "object-del",
|
||||
Arguments: map[string]any{"id": iothreadID},
|
||||
}, instance.ID)
|
||||
if iothreadErr != nil {
|
||||
slog.Warn("DetachVolume: QMP object-del iothread failed (non-fatal)", "volumeId", volumeID, "err", iothreadErr)
|
||||
}
|
||||
|
||||
// Phase 3: ebs.unmount via NATS (best-effort)
|
||||
d.rollbackEBSMount(ebsReq)
|
||||
|
||||
// State cleanup: remove volume from EBSRequests (search by name to avoid stale index)
|
||||
instance.EBSRequests.Mu.Lock()
|
||||
for i, req := range instance.EBSRequests.Requests {
|
||||
if req.Name == volumeID {
|
||||
instance.EBSRequests.Requests = append(instance.EBSRequests.Requests[:i], instance.EBSRequests.Requests[i+1:]...)
|
||||
break
|
||||
}
|
||||
}
|
||||
instance.EBSRequests.Mu.Unlock()
|
||||
|
||||
// Remove from BlockDeviceMappings
|
||||
d.Instances.Mu.Lock()
|
||||
if instance.Instance != nil {
|
||||
filtered := make([]*ec2.InstanceBlockDeviceMapping, 0, len(instance.Instance.BlockDeviceMappings))
|
||||
for _, bdm := range instance.Instance.BlockDeviceMappings {
|
||||
if bdm.Ebs != nil && bdm.Ebs.VolumeId != nil && *bdm.Ebs.VolumeId == volumeID {
|
||||
continue
|
||||
}
|
||||
filtered = append(filtered, bdm)
|
||||
}
|
||||
instance.Instance.BlockDeviceMappings = filtered
|
||||
}
|
||||
d.Instances.Mu.Unlock()
|
||||
|
||||
// Update volume metadata to "available"
|
||||
if err := d.volumeService.UpdateVolumeState(volumeID, "available", "", ""); err != nil {
|
||||
slog.Error("DetachVolume: failed to update volume metadata", "volumeId", volumeID, "err", err)
|
||||
}
|
||||
|
||||
// Persist state
|
||||
if err := d.WriteState(); err != nil {
|
||||
slog.Error("DetachVolume: failed to write state", "err", err)
|
||||
}
|
||||
|
||||
d.respondWithVolumeAttachment(msg, respondWithError, volumeID, command.ID, ebsReq.DeviceName, "detaching")
|
||||
slog.Info("Volume detached successfully", "volumeId", volumeID, "instanceId", command.ID)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2CreateVolume(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.volumeService.CreateVolume)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DescribeVolumes(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.volumeService.DescribeVolumes)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DescribeVolumeStatus(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.volumeService.DescribeVolumeStatus)
|
||||
}
|
||||
|
||||
// handleEC2ModifyVolume processes incoming EC2 ModifyVolume requests
|
||||
func (d *Daemon) handleEC2ModifyVolume(msg *nats.Msg) {
|
||||
slog.Debug("Received message", "subject", msg.Subject)
|
||||
slog.Debug("Message data", "data", string(msg.Data))
|
||||
|
||||
modifyVolumeInput := &ec2.ModifyVolumeInput{}
|
||||
var errResp []byte
|
||||
|
||||
errResp = utils.UnmarshalJsonPayload(modifyVolumeInput, msg.Data)
|
||||
|
||||
if errResp != nil {
|
||||
if err := msg.Respond(errResp); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
slog.Error("Request does not match ModifyVolumeInput")
|
||||
return
|
||||
}
|
||||
|
||||
slog.Info("Processing ModifyVolume request", "volumeId", modifyVolumeInput.VolumeId)
|
||||
|
||||
output, err := d.volumeService.ModifyVolume(modifyVolumeInput)
|
||||
|
||||
if err != nil {
|
||||
slog.Error("handleEC2ModifyVolume service.ModifyVolume failed", "err", err)
|
||||
errResp = utils.GenerateErrorPayload(err.Error())
|
||||
if err := msg.Respond(errResp); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
jsonResponse, err := json.Marshal(output)
|
||||
if err != nil {
|
||||
slog.Error("handleEC2ModifyVolume failed to marshal output", "err", err)
|
||||
errResp = utils.GenerateErrorPayload(awserrors.ErrorServerInternal)
|
||||
if err := msg.Respond(errResp); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
if err := msg.Respond(jsonResponse); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
|
||||
// Notify viperblockd to reload state after volume modification (e.g. resize)
|
||||
if modifyVolumeInput.VolumeId != nil {
|
||||
syncData, err := json.Marshal(config.EBSSyncRequest{Volume: *modifyVolumeInput.VolumeId})
|
||||
if err != nil {
|
||||
slog.Error("failed to marshal ebs.sync request", "volumeId", *modifyVolumeInput.VolumeId, "err", err)
|
||||
} else {
|
||||
_, syncErr := d.natsConn.Request("ebs.sync", syncData, 5*time.Second)
|
||||
if syncErr != nil {
|
||||
slog.Warn("ebs.sync notification failed (volume may not be mounted)",
|
||||
"volumeId", *modifyVolumeInput.VolumeId, "err", syncErr)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
slog.Info("handleEC2ModifyVolume completed", "volumeId", modifyVolumeInput.VolumeId)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DeleteVolume(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.volumeService.DeleteVolume)
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
package daemon
|
||||
|
||||
import (
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
func (d *Daemon) handleEC2CreateVpc(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.vpcService.CreateVpc)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DeleteVpc(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.vpcService.DeleteVpc)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DescribeVpcs(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.vpcService.DescribeVpcs)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2CreateSubnet(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.vpcService.CreateSubnet)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DeleteSubnet(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.vpcService.DeleteSubnet)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DescribeSubnets(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.vpcService.DescribeSubnets)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2CreateNetworkInterface(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.vpcService.CreateNetworkInterface)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DeleteNetworkInterface(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.vpcService.DeleteNetworkInterface)
|
||||
}
|
||||
|
||||
func (d *Daemon) handleEC2DescribeNetworkInterfaces(msg *nats.Msg) {
|
||||
handleNATSRequest(msg, d.vpcService.DescribeNetworkInterfaces)
|
||||
}
|
||||
@@ -1331,24 +1331,6 @@ func TestRunInstances_CountValidation(t *testing.T) {
|
||||
t.Logf("Error response: %v", errResp)
|
||||
})
|
||||
|
||||
t.Run("MinCount_zero", func(t *testing.T) {
|
||||
input := &ec2.RunInstancesInput{
|
||||
ImageId: aws.String("ami-test"),
|
||||
InstanceType: aws.String(instanceType),
|
||||
MinCount: aws.Int64(0), // Invalid
|
||||
MaxCount: aws.Int64(1),
|
||||
}
|
||||
inputJSON, _ := json.Marshal(input)
|
||||
|
||||
resp, err := daemon.natsConn.Request(topic, inputJSON, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
|
||||
var errResp map[string]any
|
||||
err = json.Unmarshal(resp.Data, &errResp)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, errResp, "Code")
|
||||
})
|
||||
|
||||
t.Run("MaxCount_zero", func(t *testing.T) {
|
||||
input := &ec2.RunInstancesInput{
|
||||
ImageId: aws.String("ami-test"),
|
||||
|
||||
@@ -19,11 +19,11 @@ func setupNATSIGWServiceTest(t *testing.T) (IGWService, *IGWServiceImpl) {
|
||||
backend, nc := setupTestIGWService(t)
|
||||
|
||||
topics := map[string]func(*nats.Msg){
|
||||
"ec2.CreateInternetGateway": func(msg *nats.Msg) { handleNATSMsg(msg, backend.CreateInternetGateway) },
|
||||
"ec2.DeleteInternetGateway": func(msg *nats.Msg) { handleNATSMsg(msg, backend.DeleteInternetGateway) },
|
||||
"ec2.CreateInternetGateway": func(msg *nats.Msg) { handleNATSMsg(msg, backend.CreateInternetGateway) },
|
||||
"ec2.DeleteInternetGateway": func(msg *nats.Msg) { handleNATSMsg(msg, backend.DeleteInternetGateway) },
|
||||
"ec2.DescribeInternetGateways": func(msg *nats.Msg) { handleNATSMsg(msg, backend.DescribeInternetGateways) },
|
||||
"ec2.AttachInternetGateway": func(msg *nats.Msg) { handleNATSMsg(msg, backend.AttachInternetGateway) },
|
||||
"ec2.DetachInternetGateway": func(msg *nats.Msg) { handleNATSMsg(msg, backend.DetachInternetGateway) },
|
||||
"ec2.AttachInternetGateway": func(msg *nats.Msg) { handleNATSMsg(msg, backend.AttachInternetGateway) },
|
||||
"ec2.DetachInternetGateway": func(msg *nats.Msg) { handleNATSMsg(msg, backend.DetachInternetGateway) },
|
||||
}
|
||||
|
||||
for topic, handler := range topics {
|
||||
|
||||
@@ -22,14 +22,14 @@ func setupNATSVPCServiceTest(t *testing.T) (VPCService, *VPCServiceImpl, *nats.C
|
||||
|
||||
// Subscribe the backend handlers on NATS topics
|
||||
topics := map[string]func(*nats.Msg){
|
||||
"ec2.CreateVpc": func(msg *nats.Msg) { handleNATSMsg(msg, backend.CreateVpc) },
|
||||
"ec2.DeleteVpc": func(msg *nats.Msg) { handleNATSMsg(msg, backend.DeleteVpc) },
|
||||
"ec2.DescribeVpcs": func(msg *nats.Msg) { handleNATSMsg(msg, backend.DescribeVpcs) },
|
||||
"ec2.CreateSubnet": func(msg *nats.Msg) { handleNATSMsg(msg, backend.CreateSubnet) },
|
||||
"ec2.DeleteSubnet": func(msg *nats.Msg) { handleNATSMsg(msg, backend.DeleteSubnet) },
|
||||
"ec2.DescribeSubnets": func(msg *nats.Msg) { handleNATSMsg(msg, backend.DescribeSubnets) },
|
||||
"ec2.CreateNetworkInterface": func(msg *nats.Msg) { handleNATSMsg(msg, backend.CreateNetworkInterface) },
|
||||
"ec2.DeleteNetworkInterface": func(msg *nats.Msg) { handleNATSMsg(msg, backend.DeleteNetworkInterface) },
|
||||
"ec2.CreateVpc": func(msg *nats.Msg) { handleNATSMsg(msg, backend.CreateVpc) },
|
||||
"ec2.DeleteVpc": func(msg *nats.Msg) { handleNATSMsg(msg, backend.DeleteVpc) },
|
||||
"ec2.DescribeVpcs": func(msg *nats.Msg) { handleNATSMsg(msg, backend.DescribeVpcs) },
|
||||
"ec2.CreateSubnet": func(msg *nats.Msg) { handleNATSMsg(msg, backend.CreateSubnet) },
|
||||
"ec2.DeleteSubnet": func(msg *nats.Msg) { handleNATSMsg(msg, backend.DeleteSubnet) },
|
||||
"ec2.DescribeSubnets": func(msg *nats.Msg) { handleNATSMsg(msg, backend.DescribeSubnets) },
|
||||
"ec2.CreateNetworkInterface": func(msg *nats.Msg) { handleNATSMsg(msg, backend.CreateNetworkInterface) },
|
||||
"ec2.DeleteNetworkInterface": func(msg *nats.Msg) { handleNATSMsg(msg, backend.DeleteNetworkInterface) },
|
||||
"ec2.DescribeNetworkInterfaces": func(msg *nats.Msg) { handleNATSMsg(msg, backend.DescribeNetworkInterfaces) },
|
||||
}
|
||||
|
||||
|
||||
@@ -64,8 +64,10 @@ TMPDIR=$(mktemp -d)
|
||||
trap 'rm -rf "$TMPDIR"' EXIT
|
||||
|
||||
# --- Step 1: Extract added line numbers from diff ---
|
||||
# Only non-test .go files that were added or modified
|
||||
git diff "${BASE_REF}" HEAD --unified=0 --diff-filter=AM -- '*.go' ':!*_test.go' | awk '
|
||||
# Only non-test .go files that were added or modified.
|
||||
# -M enables rename detection so file splits/moves only count truly changed lines.
|
||||
# -C enables copy detection for the same reason (file split = partial copy).
|
||||
git diff "${BASE_REF}" HEAD --unified=0 -M -C --diff-filter=AM -- '*.go' ':!*_test.go' | awk '
|
||||
/^\+\+\+ / {
|
||||
file = substr($2, 3)
|
||||
next
|
||||
|
||||
Reference in New Issue
Block a user