gsadmin/spinifex
1
0
Fork 0
mirror of https://github.com/mulgadc/spinifex.git synced 2026-10-19 19:01:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

Hive README and start instance implemented

Browse Source
This commit is contained in:
Ben Duncan
2025-08-03 10:25:12 +08:00
parent d05fb7cff3
commit 42cded6f21
3 changed files with 141 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
INSTALL.md
+17
View File
@@ -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
```
hive/config/config.go
+22
View File
@@ -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)
hive/daemon/daemon.go
+102 -44
View File
@@ -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)