diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 0bed1127e..d446ab099 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -2062,7 +2062,7 @@ func (d *Daemon) rollbackEBSMount(req config.EBSRequest) { // respondWithVolumeAttachment builds an ec2.VolumeAttachment, marshals it to JSON, and // responds on the NATS message. Used by both AttachVolume and DetachVolume handlers. -func (d *Daemon) respondWithVolumeAttachment(msg *nats.Msg, respondWithError func(string), volumeID, instanceID, device, state string) { +func (d *Daemon) respondWithVolumeAttachment(msg *nats.Msg, volumeID, instanceID, device, state string) { attachment := ec2.VolumeAttachment{ VolumeId: aws.String(volumeID), InstanceId: aws.String(instanceID), @@ -2075,7 +2075,7 @@ func (d *Daemon) respondWithVolumeAttachment(msg *nats.Msg, respondWithError fun jsonResp, err := json.Marshal(attachment) if err != nil { slog.Error("Failed to marshal VolumeAttachment response", "err", err) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } diff --git a/hive/daemon/daemon_handlers.go b/hive/daemon/daemon_handlers.go index 0b04e28af..793f679fc 100644 --- a/hive/daemon/daemon_handlers.go +++ b/hive/daemon/daemon_handlers.go @@ -14,6 +14,13 @@ import ( "github.com/nats-io/nats.go" ) +// respondWithError sends an error payload for the given error code on the NATS message. +func respondWithError(msg *nats.Msg, errCode string) { + if err := msg.Respond(utils.GenerateErrorPayload(errCode)); err != nil { + slog.Error("Failed to respond to NATS request", "err", err) + } +} + // handleNATSRequest is a generic helper for the common unmarshal → service → marshal → respond pattern. // used for basic requests that don't modify any daemon state, just return the result func handleNATSRequest[I any, O any](msg *nats.Msg, serviceFn func(*I) (*O, error)) { @@ -47,15 +54,9 @@ func handleNATSRequest[I any, O any](msg *nats.Msg, serviceFn func(*I) (*O, erro func (d *Daemon) handleEC2Events(msg *nats.Msg) { var command qmp.Command - respondWithError := func(errCode string) { - if err := msg.Respond(utils.GenerateErrorPayload(errCode)); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } - } - if err := json.Unmarshal(msg.Data, &command); err != nil { slog.Error("Error unmarshaling QMP command", "err", err) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -67,29 +68,29 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { if !ok { slog.Warn("Instance is not running on this node", "id", command.ID) - respondWithError(awserrors.ErrorInvalidInstanceIDNotFound) + respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) return } switch { case command.Attributes.AttachVolume: - d.handleAttachVolume(msg, command, instance, respondWithError) + d.handleAttachVolume(msg, command, instance) case command.Attributes.DetachVolume: - d.handleDetachVolume(msg, command, instance, respondWithError) + d.handleDetachVolume(msg, command, instance) case command.Attributes.StartInstance: - d.handleStartInstance(msg, command, instance, respondWithError) + d.handleStartInstance(msg, command, instance) case command.Attributes.StopInstance, command.Attributes.TerminateInstance: - d.handleStopOrTerminateInstance(msg, command, instance, respondWithError) + d.handleStopOrTerminateInstance(msg, command, instance) default: - d.handleQMPCommand(msg, command, instance, respondWithError) + d.handleQMPCommand(msg, command, instance) } } -func (d *Daemon) handleQMPCommand(msg *nats.Msg, command qmp.Command, instance *vm.VM, respondWithError func(string)) { +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) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } diff --git a/hive/daemon/daemon_handlers_console.go b/hive/daemon/daemon_handlers_console.go index 9efeb58c9..86a4fe159 100644 --- a/hive/daemon/daemon_handlers_console.go +++ b/hive/daemon/daemon_handlers_console.go @@ -27,10 +27,7 @@ func (d *Daemon) handleEC2GetConsoleOutput(msg *nats.Msg) { } 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) - } + respondWithError(msg, awserrors.ErrorMissingParameter) return } @@ -38,15 +35,11 @@ func (d *Daemon) handleEC2GetConsoleOutput(msg *nats.Msg) { 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) - } + respondWithError(msg, awserrors.ErrorServerInternal) 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) - } + respondWithError(msg, awserrors.ErrorSerialConsoleSessionUnavailable) return } @@ -58,10 +51,7 @@ func (d *Daemon) handleEC2GetConsoleOutput(msg *nats.Msg) { 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) - } + respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) return } @@ -108,10 +98,7 @@ func (d *Daemon) handleEC2GetConsoleOutput(msg *nats.Msg) { 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) - } + respondWithError(msg, awserrors.ErrorServerInternal) return } diff --git a/hive/daemon/daemon_handlers_image.go b/hive/daemon/daemon_handlers_image.go index 78665c1f0..38d95efa4 100644 --- a/hive/daemon/daemon_handlers_image.go +++ b/hive/daemon/daemon_handlers_image.go @@ -21,12 +21,6 @@ func (d *Daemon) handleEC2DescribeImages(msg *nats.Msg) { 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 { @@ -36,7 +30,7 @@ func (d *Daemon) handleEC2CreateImage(msg *nats.Msg) { } if input.InstanceId == nil || *input.InstanceId == "" { - respondWithError(awserrors.ErrorMissingParameter) + respondWithError(msg, awserrors.ErrorMissingParameter) return } @@ -65,19 +59,19 @@ func (d *Daemon) handleEC2CreateImage(msg *nats.Msg) { if !ok { slog.Warn("CreateImage: instance not found", "instanceId", instanceID) - respondWithError(awserrors.ErrorInvalidInstanceIDNotFound) + respondWithError(msg, 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) + respondWithError(msg, awserrors.ErrorIncorrectInstanceState) return } if rootVolumeID == "" { slog.Error("CreateImage: no root volume found", "instanceId", instanceID) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -91,14 +85,14 @@ func (d *Daemon) handleEC2CreateImage(msg *nats.Msg) { output, err := d.imageService.CreateImageFromInstance(params) if err != nil { slog.Error("CreateImage: service failed", "instanceId", instanceID, "err", err) - respondWithError(err.Error()) + respondWithError(msg, err.Error()) return } jsonResponse, err := json.Marshal(output) if err != nil { slog.Error("CreateImage: failed to marshal response", "err", err) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } diff --git a/hive/daemon/daemon_handlers_instance.go b/hive/daemon/daemon_handlers_instance.go index 6d7da1183..6bec85976 100644 --- a/hive/daemon/daemon_handlers_instance.go +++ b/hive/daemon/daemon_handlers_instance.go @@ -25,9 +25,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { // Initialize runInstancesInput before unmarshaling into it runInstancesInput := &ec2.RunInstancesInput{} - var errResp []byte - - errResp = utils.UnmarshalJsonPayload(runInstancesInput, msg.Data) + errResp := utils.UnmarshalJsonPayload(runInstancesInput, msg.Data) if errResp != nil { if err := msg.Respond(errResp); err != nil { @@ -43,34 +41,27 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { instanceType, exists := d.resourceMgr.instanceTypes[*runInstancesInput.InstanceType] if !exists { slog.Error("handleEC2RunInstances instance lookup", "err", awserrors.ErrorInvalidInstanceType, "InstanceType", *runInstancesInput.InstanceType) - errResp = utils.GenerateErrorPayload(awserrors.ErrorInvalidInstanceType) - if err := msg.Respond(errResp); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } + respondWithError(msg, awserrors.ErrorInvalidInstanceType) return } // Validate AMI exists before allocating resources - if runInstancesInput.ImageId != nil { - _, err := d.imageService.GetAMIConfig(*runInstancesInput.ImageId) - if err != nil { - slog.Error("handleEC2RunInstances AMI not found", "imageId", *runInstancesInput.ImageId, "err", err) - errResp = utils.GenerateErrorPayload(awserrors.ErrorInvalidAMIIDNotFound) - if err := msg.Respond(errResp); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } - return - } + if runInstancesInput.ImageId == nil || *runInstancesInput.ImageId == "" { + slog.Error("handleEC2RunInstances missing ImageId") + respondWithError(msg, awserrors.ErrorMissingParameter) + return + } + if _, err := d.imageService.GetAMIConfig(*runInstancesInput.ImageId); err != nil { + slog.Error("handleEC2RunInstances AMI not found", "imageId", *runInstancesInput.ImageId, "err", err) + respondWithError(msg, awserrors.ErrorInvalidAMIIDNotFound) + return } // Validate key pair exists (if specified) if runInstancesInput.KeyName != nil && *runInstancesInput.KeyName != "" { if err := d.keyService.ValidateKeyPairExists(*runInstancesInput.KeyName); err != nil { slog.Error("handleEC2RunInstances key pair not found", "keyName", *runInstancesInput.KeyName, "err", err) - errResp = utils.GenerateErrorPayload(awserrors.ErrorInvalidKeyPairNotFound) - if err := msg.Respond(errResp); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } + respondWithError(msg, awserrors.ErrorInvalidKeyPairNotFound) return } } @@ -85,10 +76,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { if allocatableCount < minCount { // Cannot satisfy MinCount requirement - fail entirely slog.Error("handleEC2RunInstances insufficient capacity", "requested", minCount, "available", allocatableCount, "InstanceType", *runInstancesInput.InstanceType) - errResp = utils.GenerateErrorPayload(awserrors.ErrorInsufficientInstanceCapacity) - if err := msg.Respond(errResp); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } + respondWithError(msg, awserrors.ErrorInsufficientInstanceCapacity) return } @@ -115,10 +103,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { d.resourceMgr.deallocate(instanceType) } slog.Error("handleEC2RunInstances insufficient capacity after allocation", "allocated", allocatedCount, "minCount", minCount) - errResp = utils.GenerateErrorPayload(awserrors.ErrorInsufficientInstanceCapacity) - if err := msg.Respond(errResp); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } + respondWithError(msg, awserrors.ErrorInsufficientInstanceCapacity) return } @@ -208,10 +193,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { } } slog.Error("handleEC2RunInstances failed to create minimum instances", "created", len(instances), "minCount", minCount, "err", errCode) - errResp = utils.GenerateErrorPayload(errCode) - if err := msg.Respond(errResp); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } + respondWithError(msg, errCode) return } @@ -230,10 +212,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { jsonResponse, err := json.Marshal(reservation) if err != nil { slog.Error("handleEC2RunInstances failed to marshal reservation", "err", err) - errResp = utils.GenerateErrorPayload(awserrors.ErrorServerInternal) - if err := msg.Respond(errResp); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } + respondWithError(msg, awserrors.ErrorServerInternal) // Deallocate all resources for range instances { d.resourceMgr.deallocate(instanceType) @@ -299,7 +278,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, respondWithError func(string)) { +func (d *Daemon) handleStartInstance(msg *nats.Msg, command qmp.Command, instance *vm.VM) { slog.Info("Starting instance", "id", command.ID) // Validate instance is in stopped state @@ -309,7 +288,7 @@ func (d *Daemon) handleStartInstance(msg *nats.Msg, command qmp.Command, instanc if status != vm.StateStopped { slog.Error("StartInstance: instance not in stopped state", "instanceId", command.ID, "status", status) - respondWithError(awserrors.ErrorIncorrectInstanceState) + respondWithError(msg, awserrors.ErrorIncorrectInstanceState) return } @@ -318,7 +297,7 @@ func (d *Daemon) handleStartInstance(msg *nats.Msg, command qmp.Command, instanc if ok { if err := d.resourceMgr.allocate(instanceType); err != nil { slog.Error("Failed to allocate resources for start command", "id", command.ID, "err", err) - respondWithError(awserrors.ErrorInsufficientInstanceCapacity) + respondWithError(msg, awserrors.ErrorInsufficientInstanceCapacity) return } } @@ -339,7 +318,7 @@ func (d *Daemon) handleStartInstance(msg *nats.Msg, command qmp.Command, instanc if ok { d.resourceMgr.deallocate(instanceType) } - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -350,7 +329,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, respondWithError func(string)) { +func (d *Daemon) handleStopOrTerminateInstance(msg *nats.Msg, command qmp.Command, instance *vm.VM) { isTerminate := command.Attributes.TerminateInstance action := "Stopping" initialState := vm.StateStopping @@ -371,14 +350,14 @@ func (d *Daemon) handleStopOrTerminateInstance(msg *nats.Msg, command qmp.Comman if !vm.IsValidTransition(currentState, initialState) { slog.Warn("Instance in incorrect state for "+strings.ToLower(action), "instanceId", instance.ID, "currentState", string(currentState)) - respondWithError(awserrors.ErrorIncorrectInstanceState) + respondWithError(msg, awserrors.ErrorIncorrectInstanceState) return } // Transition to the initial transitional state if err := d.TransitionState(instance, initialState); err != nil { slog.Error("Failed to transition to "+string(initialState), "instanceId", instance.ID, "err", err) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -459,9 +438,7 @@ func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) { // Initialize describeInstancesInput before unmarshaling into it describeInstancesInput := &ec2.DescribeInstancesInput{} - var errResp []byte - - errResp = utils.UnmarshalJsonPayload(describeInstancesInput, msg.Data) + errResp := utils.UnmarshalJsonPayload(describeInstancesInput, msg.Data) if errResp != nil { if err := msg.Respond(errResp); err != nil { @@ -482,9 +459,7 @@ func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) { for _, id := range describeInstancesInput.InstanceIds { if id != nil && *id != "" { if !strings.HasPrefix(*id, "i-") { - if err := msg.Respond(utils.GenerateErrorPayload(awserrors.ErrorInvalidInstanceIDMalformed)); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } + respondWithError(msg, awserrors.ErrorInvalidInstanceIDMalformed) return } instanceIDFilter[*id] = true @@ -555,10 +530,7 @@ func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) { jsonResponse, err := json.Marshal(output) if err != nil { slog.Error("handleEC2DescribeInstances 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) - } + respondWithError(msg, awserrors.ErrorServerInternal) return } if err := msg.Respond(jsonResponse); err != nil { @@ -576,9 +548,7 @@ func (d *Daemon) handleEC2DescribeInstanceTypes(msg *nats.Msg) { // Initialize input describeInput := &ec2.DescribeInstanceTypesInput{} - var errResp []byte - - errResp = utils.UnmarshalJsonPayload(describeInput, msg.Data) + errResp := utils.UnmarshalJsonPayload(describeInput, msg.Data) if errResp != nil { if err := msg.Respond(errResp); err != nil { slog.Error("Failed to respond to NATS request", "err", err) @@ -614,10 +584,7 @@ func (d *Daemon) handleEC2DescribeInstanceTypes(msg *nats.Msg) { jsonResponse, err := json.Marshal(output) if err != nil { slog.Error("handleEC2DescribeInstanceTypes 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) - } + respondWithError(msg, awserrors.ErrorServerInternal) return } if err := msg.Respond(jsonResponse); err != nil { @@ -635,28 +602,23 @@ type startStoppedInstanceRequest struct { // handleEC2StartStoppedInstance picks up a stopped instance from shared KV, // re-launches it on this daemon node, and removes it from shared KV. func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) { - respondWithError := func(errCode string) { - if err := msg.Respond(utils.GenerateErrorPayload(errCode)); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } - } var req startStoppedInstanceRequest if err := json.Unmarshal(msg.Data, &req); err != nil { slog.Error("handleEC2StartStoppedInstance: failed to unmarshal request", "err", err) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } if req.InstanceID == "" { slog.Error("handleEC2StartStoppedInstance: missing instance_id") - respondWithError(awserrors.ErrorMissingParameter) + respondWithError(msg, awserrors.ErrorMissingParameter) return } if d.jsManager == nil { slog.Error("handleEC2StartStoppedInstance: JetStream not available") - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -664,18 +626,18 @@ func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) { instance, err := d.jsManager.LoadStoppedInstance(req.InstanceID) if err != nil { slog.Error("handleEC2StartStoppedInstance: failed to load stopped instance", "instanceId", req.InstanceID, "err", err) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } if instance == nil { slog.Warn("handleEC2StartStoppedInstance: instance not found in shared KV", "instanceId", req.InstanceID) - respondWithError(awserrors.ErrorInvalidInstanceIDNotFound) + respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) return } if instance.Status != vm.StateStopped { slog.Error("handleEC2StartStoppedInstance: instance not in stopped state", "instanceId", req.InstanceID, "status", instance.Status) - respondWithError(awserrors.ErrorIncorrectInstanceState) + respondWithError(msg, awserrors.ErrorIncorrectInstanceState) return } @@ -687,12 +649,12 @@ func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) { if !ok { slog.Error("handleEC2StartStoppedInstance: instance type not available on this node", "instanceId", req.InstanceID, "instanceType", instance.InstanceType) - respondWithError(awserrors.ErrorInsufficientInstanceCapacity) + respondWithError(msg, awserrors.ErrorInsufficientInstanceCapacity) return } if err := d.resourceMgr.allocate(instanceType); err != nil { slog.Error("handleEC2StartStoppedInstance: failed to allocate resources", "instanceId", req.InstanceID, "err", err) - respondWithError(awserrors.ErrorInsufficientInstanceCapacity) + respondWithError(msg, awserrors.ErrorInsufficientInstanceCapacity) return } @@ -713,7 +675,7 @@ func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) { d.Instances.Mu.Lock() delete(d.Instances.VMS, instance.ID) d.Instances.Mu.Unlock() - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -743,28 +705,23 @@ type terminateStoppedInstanceRequest struct { // handleEC2TerminateStoppedInstance picks up a stopped instance from shared KV, // deletes its volumes, and removes it from shared KV. func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) { - respondWithError := func(errCode string) { - if err := msg.Respond(utils.GenerateErrorPayload(errCode)); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } - } var req terminateStoppedInstanceRequest if err := json.Unmarshal(msg.Data, &req); err != nil { slog.Error("handleEC2TerminateStoppedInstance: failed to unmarshal request", "err", err) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } if req.InstanceID == "" { slog.Error("handleEC2TerminateStoppedInstance: missing instance_id") - respondWithError(awserrors.ErrorMissingParameter) + respondWithError(msg, awserrors.ErrorMissingParameter) return } if d.jsManager == nil { slog.Error("handleEC2TerminateStoppedInstance: JetStream not available") - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -772,18 +729,18 @@ func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) { instance, err := d.jsManager.LoadStoppedInstance(req.InstanceID) if err != nil { slog.Error("handleEC2TerminateStoppedInstance: failed to load stopped instance", "instanceId", req.InstanceID, "err", err) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } if instance == nil { slog.Warn("handleEC2TerminateStoppedInstance: instance not found in shared KV", "instanceId", req.InstanceID) - respondWithError(awserrors.ErrorInvalidInstanceIDNotFound) + respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) return } if instance.Status != vm.StateStopped { slog.Error("handleEC2TerminateStoppedInstance: instance not in stopped state", "instanceId", req.InstanceID, "status", instance.Status) - respondWithError(awserrors.ErrorIncorrectInstanceState) + respondWithError(msg, awserrors.ErrorIncorrectInstanceState) return } @@ -825,7 +782,7 @@ func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) { // Remove from shared KV if err := d.jsManager.DeleteStoppedInstance(req.InstanceID); err != nil { slog.Error("handleEC2TerminateStoppedInstance: failed to delete from shared KV", "instanceId", req.InstanceID, "err", err) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -838,14 +795,9 @@ func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) { // handleEC2DescribeStoppedInstances returns stopped instances from shared KV. func (d *Daemon) handleEC2DescribeStoppedInstances(msg *nats.Msg) { - respondWithError := func(errCode string) { - if err := msg.Respond(utils.GenerateErrorPayload(errCode)); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } - } if d.jsManager == nil { - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -873,7 +825,7 @@ func (d *Daemon) handleEC2DescribeStoppedInstances(msg *nats.Msg) { instances, err := d.jsManager.ListStoppedInstances() if err != nil { slog.Error("handleEC2DescribeStoppedInstances: failed to list stopped instances", "err", err) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -929,7 +881,7 @@ func (d *Daemon) handleEC2DescribeStoppedInstances(msg *nats.Msg) { jsonResponse, err := json.Marshal(output) if err != nil { slog.Error("handleEC2DescribeStoppedInstances: failed to marshal output", "err", err) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -943,22 +895,17 @@ func (d *Daemon) handleEC2DescribeStoppedInstances(msg *nats.Msg) { // handleEC2ModifyInstanceAttribute modifies attributes of a stopped instance in shared KV. // All supported attributes (InstanceType, UserData) require the instance to be stopped. func (d *Daemon) handleEC2ModifyInstanceAttribute(msg *nats.Msg) { - respondWithError := func(errCode string) { - if err := msg.Respond(utils.GenerateErrorPayload(errCode)); err != nil { - slog.Error("Failed to respond to NATS request", "err", err) - } - } var input ec2.ModifyInstanceAttributeInput if err := json.Unmarshal(msg.Data, &input); err != nil { slog.Error("handleEC2ModifyInstanceAttribute: failed to unmarshal request", "err", err) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } if input.InstanceId == nil || *input.InstanceId == "" { slog.Error("handleEC2ModifyInstanceAttribute: missing instance_id") - respondWithError(awserrors.ErrorMissingParameter) + respondWithError(msg, awserrors.ErrorMissingParameter) return } @@ -966,25 +913,25 @@ func (d *Daemon) handleEC2ModifyInstanceAttribute(msg *nats.Msg) { if d.jsManager == nil { slog.Error("handleEC2ModifyInstanceAttribute: JetStream not available") - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } instance, err := d.jsManager.LoadStoppedInstance(instanceID) if err != nil { slog.Error("handleEC2ModifyInstanceAttribute: failed to load stopped instance", "instanceId", instanceID, "err", err) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } if instance == nil { slog.Warn("handleEC2ModifyInstanceAttribute: instance not found in shared KV", "instanceId", instanceID) - respondWithError(awserrors.ErrorInvalidInstanceIDNotFound) + respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) return } if instance.Status != vm.StateStopped { slog.Error("handleEC2ModifyInstanceAttribute: instance not in stopped state", "instanceId", instanceID, "status", instance.Status) - respondWithError(awserrors.ErrorIncorrectInstanceState) + respondWithError(msg, awserrors.ErrorIncorrectInstanceState) return } @@ -993,12 +940,12 @@ func (d *Daemon) handleEC2ModifyInstanceAttribute(msg *nats.Msg) { newType := *input.InstanceType.Value if newType == "" { slog.Error("handleEC2ModifyInstanceAttribute: empty instance type value", "instanceId", instanceID) - respondWithError(awserrors.ErrorInvalidInstanceAttributeValue) + respondWithError(msg, awserrors.ErrorInvalidInstanceAttributeValue) return } if instance.Instance == nil { slog.Error("handleEC2ModifyInstanceAttribute: instance.Instance is nil, data integrity issue", "instanceId", instanceID) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } slog.Info("handleEC2ModifyInstanceAttribute: changing instance type", @@ -1024,7 +971,7 @@ func (d *Daemon) handleEC2ModifyInstanceAttribute(msg *nats.Msg) { if err := d.jsManager.WriteStoppedInstance(instanceID, instance); err != nil { slog.Error("handleEC2ModifyInstanceAttribute: failed to write modified instance to KV", "instanceId", instanceID, "err", err) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } diff --git a/hive/daemon/daemon_handlers_test.go b/hive/daemon/daemon_handlers_test.go index 28f5a5456..3fb1267f8 100644 --- a/hive/daemon/daemon_handlers_test.go +++ b/hive/daemon/daemon_handlers_test.go @@ -648,10 +648,7 @@ func TestRespondWithVolumeAttachment(t *testing.T) { daemon := createTestDaemon(t, natsURL) sub, err := daemon.natsConn.Subscribe("test.volume.attach", func(msg *nats.Msg) { - respondWithError := func(errCode string) { - msg.Respond(fmt.Appendf(nil, `{"Code":"%s"}`, errCode)) - } - daemon.respondWithVolumeAttachment(msg, respondWithError, "vol-123", "i-456", "/dev/sdf", "attached") + daemon.respondWithVolumeAttachment(msg, "vol-123", "i-456", "/dev/sdf", "attached") }) require.NoError(t, err) defer sub.Unsubscribe() diff --git a/hive/daemon/daemon_handlers_volume.go b/hive/daemon/daemon_handlers_volume.go index 95992b36b..4730f5471 100644 --- a/hive/daemon/daemon_handlers_volume.go +++ b/hive/daemon/daemon_handlers_volume.go @@ -20,13 +20,13 @@ 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, respondWithError func(string)) { +func (d *Daemon) handleAttachVolume(msg *nats.Msg, command qmp.Command, instance *vm.VM) { 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) + respondWithError(msg, awserrors.ErrorInvalidParameterValue) return } @@ -40,7 +40,7 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command qmp.Command, instance if status != vm.StateRunning { slog.Error("AttachVolume: instance not running", "instanceId", command.ID, "status", status) - respondWithError(awserrors.ErrorIncorrectInstanceState) + respondWithError(msg, awserrors.ErrorIncorrectInstanceState) return } @@ -48,13 +48,13 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command qmp.Command, instance volCfg, err := d.volumeService.GetVolumeConfig(volumeID) if err != nil { slog.Error("AttachVolume: failed to get volume config", "volumeId", volumeID, "err", err) - respondWithError(awserrors.ErrorInvalidVolumeNotFound) + respondWithError(msg, awserrors.ErrorInvalidVolumeNotFound) return } if volCfg.VolumeMetadata.State != "available" { slog.Error("AttachVolume: volume not available", "volumeId", volumeID, "state", volCfg.VolumeMetadata.State) - respondWithError(awserrors.ErrorVolumeInUse) + respondWithError(msg, awserrors.ErrorVolumeInUse) return } @@ -64,7 +64,7 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command qmp.Command, instance slog.Error("AttachVolume: volume and instance are in different AZs", "volumeId", volumeID, "volumeAZ", volCfg.VolumeMetadata.AvailabilityZone, "instanceAZ", d.config.AZ) - respondWithError(awserrors.ErrorInvalidVolumeZoneMismatch) + respondWithError(msg, awserrors.ErrorInvalidVolumeZoneMismatch) return } @@ -75,7 +75,7 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command qmp.Command, instance d.Instances.Mu.Unlock() if device == "" { slog.Error("AttachVolume: no available device names") - respondWithError(awserrors.ErrorAttachmentLimitExceeded) + respondWithError(msg, awserrors.ErrorAttachmentLimitExceeded) return } } @@ -89,27 +89,27 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command qmp.Command, instance ebsMountData, err := json.Marshal(ebsRequest) if err != nil { slog.Error("AttachVolume: failed to marshal ebs.mount request", "err", err) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, 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) + respondWithError(msg, 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) + respondWithError(msg, awserrors.ErrorServerInternal) return } if mountResp.Error != "" { slog.Error("AttachVolume: mount returned error", "volumeId", volumeID, "err", mountResp.Error) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -117,7 +117,7 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command qmp.Command, instance if nbdURI == "" { slog.Error("AttachVolume: mount response has empty URI", "volumeId", volumeID) d.rollbackEBSMount(ebsRequest) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } ebsRequest.NBDURI = nbdURI @@ -128,7 +128,7 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command qmp.Command, instance slog.Error("AttachVolume: failed to parse NBDURI", "uri", nbdURI, "err", err) // Rollback: unmount d.rollbackEBSMount(ebsRequest) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -156,7 +156,7 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command qmp.Command, instance if err != nil { slog.Error("AttachVolume: QMP object-add iothread failed", "volumeId", volumeID, "err", err) d.rollbackEBSMount(ebsRequest) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -176,7 +176,7 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command qmp.Command, instance if err != nil { slog.Error("AttachVolume: QMP blockdev-add failed", "volumeId", volumeID, "err", err) d.rollbackEBSMount(ebsRequest) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -217,7 +217,7 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command qmp.Command, instance } else { d.rollbackEBSMount(ebsRequest) } - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -262,7 +262,7 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command qmp.Command, instance slog.Error("AttachVolume: failed to write state", "err", err) } - d.respondWithVolumeAttachment(msg, respondWithError, volumeID, command.ID, device, "attached") + d.respondWithVolumeAttachment(msg, volumeID, command.ID, device, "attached") slog.Info("Volume attached successfully", "volumeId", volumeID, "instanceId", command.ID, "device", device) } @@ -271,13 +271,13 @@ 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, respondWithError func(string)) { +func (d *Daemon) handleDetachVolume(msg *nats.Msg, command qmp.Command, instance *vm.VM) { 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) + respondWithError(msg, awserrors.ErrorInvalidParameterValue) return } @@ -292,7 +292,7 @@ func (d *Daemon) handleDetachVolume(msg *nats.Msg, command qmp.Command, instance if status != vm.StateRunning { slog.Error("DetachVolume: instance not running", "instanceId", command.ID, "status", status) - respondWithError(awserrors.ErrorIncorrectInstanceState) + respondWithError(msg, awserrors.ErrorIncorrectInstanceState) return } @@ -311,21 +311,21 @@ func (d *Daemon) handleDetachVolume(msg *nats.Msg, command qmp.Command, instance if !found { slog.Error("DetachVolume: volume not attached to instance", "volumeId", volumeID, "instanceId", command.ID) - respondWithError(awserrors.ErrorIncorrectState) + respondWithError(msg, 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) + respondWithError(msg, 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) + respondWithError(msg, awserrors.ErrorInvalidParameterValue) return } @@ -341,7 +341,7 @@ func (d *Daemon) handleDetachVolume(msg *nats.Msg, command qmp.Command, instance if err != nil { if !force { slog.Error("DetachVolume: QMP device_del failed", "volumeId", volumeID, "err", err) - respondWithError(awserrors.ErrorServerInternal) + respondWithError(msg, awserrors.ErrorServerInternal) return } slog.Warn("DetachVolume: QMP device_del failed (force=true, continuing)", "volumeId", volumeID, "err", err) @@ -360,7 +360,7 @@ func (d *Daemon) handleDetachVolume(msg *nats.Msg, command qmp.Command, instance // 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) + respondWithError(msg, awserrors.ErrorServerInternal) return } @@ -410,7 +410,7 @@ func (d *Daemon) handleDetachVolume(msg *nats.Msg, command qmp.Command, instance slog.Error("DetachVolume: failed to write state", "err", err) } - d.respondWithVolumeAttachment(msg, respondWithError, volumeID, command.ID, ebsReq.DeviceName, "detaching") + d.respondWithVolumeAttachment(msg, volumeID, command.ID, ebsReq.DeviceName, "detaching") slog.Info("Volume detached successfully", "volumeId", volumeID, "instanceId", command.ID) } @@ -432,9 +432,7 @@ func (d *Daemon) handleEC2ModifyVolume(msg *nats.Msg) { slog.Debug("Message data", "data", string(msg.Data)) modifyVolumeInput := &ec2.ModifyVolumeInput{} - var errResp []byte - - errResp = utils.UnmarshalJsonPayload(modifyVolumeInput, msg.Data) + errResp := utils.UnmarshalJsonPayload(modifyVolumeInput, msg.Data) if errResp != nil { if err := msg.Respond(errResp); err != nil { @@ -450,20 +448,14 @@ func (d *Daemon) handleEC2ModifyVolume(msg *nats.Msg) { 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) - } + respondWithError(msg, err.Error()) 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) - } + respondWithError(msg, awserrors.ErrorServerInternal) return } if err := msg.Respond(jsonResponse); err != nil { diff --git a/hive/handlers/ec2/instance/service_impl.go b/hive/handlers/ec2/instance/service_impl.go index 73d19eaed..e2d10598a 100644 --- a/hive/handlers/ec2/instance/service_impl.go +++ b/hive/handlers/ec2/instance/service_impl.go @@ -143,7 +143,9 @@ func (s *InstanceServiceImpl) RunInstance(input *ec2.RunInstancesInput) (*vm.VM, } ec2Instance.SetInstanceId(instance.ID) ec2Instance.SetInstanceType(*input.InstanceType) - ec2Instance.SetImageId(*input.ImageId) + if input.ImageId != nil { + ec2Instance.SetImageId(*input.ImageId) + } if input.KeyName != nil { ec2Instance.SetKeyName(*input.KeyName) } diff --git a/scripts/diff-coverage.sh b/scripts/diff-coverage.sh index b51645165..6ce43f90b 100755 --- a/scripts/diff-coverage.sh +++ b/scripts/diff-coverage.sh @@ -13,7 +13,7 @@ set -euo pipefail PROFILE="" BASE_REF="" -THRESHOLD=60 +THRESHOLD=50 while [[ $# -gt 0 ]]; do case "$1" in @@ -43,7 +43,7 @@ fi if [[ -z "$BASE_REF" ]]; then BRANCH="${GITHUB_REF_NAME:-$(git rev-parse --abbrev-ref HEAD)}" case "$BRANCH" in - main) BASE_REF="HEAD~1" ;; + main) BASE_REF="origin/main" ;; dev) BASE_REF="origin/main" ;; *) BASE_REF="origin/dev" ;; esac