diff --git a/INSTALL.md b/INSTALL.md index daff65476..c828efd84 100644 --- a/INSTALL.md +++ b/INSTALL.md @@ -313,14 +313,31 @@ nats req ec2.launch '{"Action":"RunInstances","ImageId":"ami-185c47c7b6d31bba9", nats req --reply-timeout=5s ec2.describe.i-ebaf0fd46cad14c85 '{ "InstanceID": "i-ebaf0fd46cad14c85" }' ``` +### Start instance + +``` +nats req ec2.startinstances '{"InstanceID": "i-f38ac0490f1683650"}' +``` + + ## QMP commands ### Powerdown +This event is used internally be Hive when the daemon receives a SIGINT, SIGTERM, SIGHUP signal to safely powerdown an instance, or when the hardware node is rebooted. + ``` nats req --reply-timeout=5s ec2.cmd.i-ebaf0fd46cad14c85 '{ "id": "i-ebaf0fd46cad14c85", "command": { "execute": "system_powerdown", "arguments": {} } }' ``` +### Terminate instance + +This event is for a user initiated instance termination. Note the Attributes, to flag to the Hive daemon not to start the instance again if the daemon or hardware node is restarted. + +``` +nats req --reply-timeout=5s ec2.cmd.i-f38ac0490f1683650 '{ "id": "i-f38ac0490f1683650", "attributes": { "stop_instance": true }, "command": { "execute": "system_powerdown", "arguments": {} } }' +``` + ### Stop VM ``` diff --git a/hive/config/config.go b/hive/config/config.go index 2257b20ee..5b2018f53 100644 --- a/hive/config/config.go +++ b/hive/config/config.go @@ -113,6 +113,28 @@ type EC2DescribeResponse struct { Error string } +type EC2StartInstancesRequest struct { + InstanceID string +} + +type EC2StartInstancesResponse struct { + InstanceID string + Status string + Error string +} + +// TODO: Make a generic function for the response +func (ec2StartInstanceResponse EC2StartInstancesResponse) Respond(msg *nats.Msg) { + + response, err := json.Marshal(ec2StartInstanceResponse) + if err != nil { + slog.Error("Failed to marshal response: %v", err) + return + } + + msg.Respond(response) + +} func (ec2DescribeResponse EC2DescribeResponse) Respond(msg *nats.Msg) { response, err := json.Marshal(ec2DescribeResponse) diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 3ee9ba750..df3117265 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -7,6 +7,7 @@ import ( crand "crypto/rand" "encoding/hex" "encoding/json" + "errors" "fmt" "log" "log/slog" @@ -289,12 +290,17 @@ func (d *Daemon) Start() error { //d.Instances.VMS[i].EBSRequests.Mu = sync.Mutex{} // d.Instances.VMS[i].QMPClient.Mu = sync.Mutex{} - if instance.Status != "terminated" { + if instance.Attributes.StopInstance { + slog.Info("Instance flagged as user initiated stop, skipping", "instance", instance.ID) + + } else if instance.Status != "terminated" { slog.Info("Launching instance", "instance", instance.ID) err = d.LaunchInstance(instance) if err != nil { slog.Error("Failed to launch instance: %v", err) } + } else { + slog.Info("Instance state is terminated, skipping", "instance", instance.ID) } } @@ -313,6 +319,16 @@ func (d *Daemon) Start() error { return fmt.Errorf("failed to subscribe to NATS ec2.launch: %w", err) } + // Subscribe to EC2 start instance events + // TODO: The instance state needs to be shared, not pinned to a single node. + // TODO: Handle this in a more generic function to group similar commands (start, stop, launch) + // Subscribe to EC2 events with queue group + d.natsSubscriptions["ec2.startinstances"], err = d.natsConn.QueueSubscribe("ec2.startinstances", "hive-workers", d.handleEC2StartInstances) + + if err != nil { + return fmt.Errorf("failed to subscribe to NATS ec2.launch: %w", err) + } + // Setup graceful shutdown d.setupShutdown() @@ -395,6 +411,48 @@ func (d *Daemon) LoadState() error { // NATS events +func (d *Daemon) handleEC2StartInstances(msg *nats.Msg) { + + var ec2StartInstance config.EC2StartInstancesRequest + + if err := json.Unmarshal(msg.Data, &ec2StartInstance); err != nil { + log.Printf("Error unmarshaling EC2 describe request: %v", err) + return + } + + slog.Info("EC2 Start Instance Request", "instanceId", ec2StartInstance.InstanceID) + + var ec2StartInstanceResponse config.EC2StartInstancesResponse + + // Check if the instance is running on this node + d.Instances.Mu.Lock() + defer d.Instances.Mu.Unlock() + + instance, ok := d.Instances.VMS[ec2StartInstance.InstanceID] + + if !ok { + slog.Error("EC2 Describe Request - Instance not found", "instanceId", ec2StartInstanceResponse.InstanceID) + ec2StartInstanceResponse.InstanceID = ec2StartInstance.InstanceID + ec2StartInstanceResponse.Error = fmt.Sprintf("Instance %s not found", ec2StartInstanceResponse.InstanceID) + ec2StartInstanceResponse.Respond(msg) + return + } + + // Launch the instance + + err := d.LaunchInstance(instance) + + if err != nil { + ec2StartInstanceResponse.Error = err.Error() + } else { + ec2StartInstanceResponse.InstanceID = instance.ID + ec2StartInstanceResponse.Status = instance.Status + } + + ec2StartInstanceResponse.Respond(msg) + +} + func (d *Daemon) handleEC2Describe(msg *nats.Msg) { var ec2DescribeRequest config.EC2DescribeRequest @@ -509,59 +567,41 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { log.Printf("Received message on subject: %s", msg.Subject) log.Printf("Message data: %s", string(msg.Data)) - // Confirm the instance is running on this node - var instanceRunning bool + instance, ok := d.Instances.VMS[command.ID] - //d.Instances.Mu.Lock() - //defer d.Instances.Mu.Unlock() - - for _, instance := range d.Instances.VMS { - if instance.ID == command.ID { - instanceRunning = true - - // Send the command to the instance - resp, err := d.SendQMPCommand(instance.QMPClient, command.QMPCommand, instance.ID) - - if err != nil { - slog.Error("Failed to send QMP command: %v", err) - return - } - - fmt.Println("RAW QMP Response: ", string(resp.Return)) - - // Unmarshal the response - target, ok := qmp.CommandResponseTypes[command.QMPCommand.Execute] - if !ok { - slog.Warn("Unhandled QMP command: %s", command.QMPCommand.Execute) - return - } - - if err := json.Unmarshal(resp.Return, target); err != nil { - slog.Error("Failed to unmarshal QMP response for %s: %v", command.QMPCommand.Execute, err) - return - } - - msg.Respond(resp.Return) - - } - - } - - if !instanceRunning { + if !ok { // TODO: Improve, return error slog.Warn("Instance %s is not running on this node", command.ID) msg.Respond(nil) return } - // Update the instance attributes - d.Instances.Mu.Lock() - instance, ok := d.Instances.VMS[command.ID] - if !ok { - slog.Warn("Instance %s not found", command.ID) + // Send the command to the instance + resp, err := d.SendQMPCommand(instance.QMPClient, command.QMPCommand, instance.ID) + + if err != nil { + slog.Error("Failed to send QMP command: %v", err) return } + fmt.Println("RAW QMP Response: ", string(resp.Return)) + + // Unmarshal the response + target, ok := qmp.CommandResponseTypes[command.QMPCommand.Execute] + if !ok { + slog.Warn("Unhandled QMP command: %s", command.QMPCommand.Execute) + return + } + + if err := json.Unmarshal(resp.Return, target); err != nil { + slog.Error("Failed to unmarshal QMP response for %s: %v", command.QMPCommand.Execute, err) + return + } + + msg.Respond(resp.Return) + + // Update the instance attributes + d.Instances.Mu.Lock() instance.Attributes = command.Attributes d.Instances.Mu.Unlock() @@ -1364,6 +1404,7 @@ func (d *Daemon) CreateQMPClient(instance *vm.VM) (err error) { if instance.Status == "powering_down" { instance.Status = "stopped" + // TODO: Improve, confirm QEMU PID removed slog.Info("QMP Status - Instance %s stopped, exiting heartbeat", instance.ID) // TODO: Improve, move to SendQMPCommand @@ -1399,6 +1440,23 @@ func (d *Daemon) CreateQMPClient(instance *vm.VM) (err error) { func (d *Daemon) LaunchInstance(instance *vm.VM) (err error) { + // First, confirm if the instance is already running + pid, err := utils.ReadPidFile(instance.ID) + + if pid > 0 { + process, err := os.FindProcess(pid) + if err != nil { + return err + } + + // Send a 0 signal to confirm process is running + err = process.Signal(syscall.Signal(0)) + if err != nil { + slog.Error("Instance is already running", "InstanceID", instance.ID, "pid", pid) + return errors.New("instance is already running") + } + } + // Loop through each volume in volumes err = d.MountVolumes(instance)