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

VM management for rebooting and state

Browse Source
This commit is contained in:
Ben Duncan
2025-08-01 18:41:20 +08:00
parent 55707211a1
commit d05fb7cff3
6 changed files with 362 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -13,6 +13,7 @@ require (
github.com/kdomanski/iso9660 v0.4.0
github.com/mulgadc/predastore v0.1.1
github.com/mulgadc/viperblock v0.0.0-20250701054805-00b5d3692c1b
github.com/nats-io/nats-server/v2 v2.11.6
github.com/nats-io/nats.go v1.43.0
github.com/spf13/cobra v1.9.1
github.com/spf13/viper v1.20.1
@@ -35,7 +36,6 @@ require (
github.com/mattn/go-runewidth v0.0.16 // indirect
github.com/minio/highwayhash v1.0.3 // indirect
github.com/nats-io/jwt/v2 v2.7.4 // indirect
github.com/nats-io/nats-server/v2 v2.11.6 // indirect
github.com/nats-io/nkeys v0.4.11 // indirect
github.com/nats-io/nuid v1.0.1 // indirect
github.com/pelletier/go-toml/v2 v2.2.4 // indirect
go.sum
+2 -4
View File
@@ -1,5 +1,7 @@
github.com/andybalholm/brotli v1.1.0 h1:eLKJA0d02Lf0mVpIDgYnqXcUn0GqVmEFny3VuID1U3M=
github.com/andybalholm/brotli v1.1.0/go.mod h1:sms7XGricyQI9K10gOSf56VKKWS4oLer58Q+mhRPtnY=
github.com/antithesishq/antithesis-sdk-go v0.4.3-default-no-op h1:+OSa/t11TFhqfrX0EOSqQBDJ0YlpmK0rDSiB19dg9M0=
github.com/antithesishq/antithesis-sdk-go v0.4.3-default-no-op/go.mod h1:IUpT2DPAKh6i/YhSbt6Gl3v2yvUZjmKncl7U91fup7E=
github.com/aws/aws-sdk-go v1.55.7 h1:UJrkFq7es5CShfBwlWAC8DA077vp8PyVbQd3lqLiztE=
github.com/aws/aws-sdk-go v1.55.7/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU=
github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
@@ -98,8 +100,6 @@ go.uber.org/automaxprocs v1.6.0 h1:O3y2/QNTOdbF+e/dpXNNW7Rx2hZ4sTIPyybbxyNqTUs=
go.uber.org/automaxprocs v1.6.0/go.mod h1:ifeIMSnPZuznNm6jmdzmU3/bfk01Fe2fotchwEFJ8r8=
go.uber.org/multierr v1.9.0 h1:7fIwc/ZtS0q++VgcfqFDxSBZVv/Xo49/SYnDFupUwlI=
go.uber.org/multierr v1.9.0/go.mod h1:X2jQV1h+kxSjClGpnseKVIxpmcjrj7MNnI0bnlfKTVQ=
golang.org/x/crypto v0.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE=
golang.org/x/crypto v0.37.0/go.mod h1:vg+k43peMZ0pUMhYmVAWysMK35e6ioLh3wB8ZCAfbVc=
golang.org/x/crypto v0.39.0 h1:SHs+kF4LP+f+p14esP5jAoDpHU8Gu/v9lFRK6IT5imM=
golang.org/x/crypto v0.39.0/go.mod h1:L+Xg3Wf6HoL4Bn4238Z6ft6KfEpN0tJGo53AAPC632U=
golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -107,8 +107,6 @@ golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw=
golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
golang.org/x/text v0.24.0 h1:dd5Bzh4yt5KYA8f9CJHCP4FB4D51c2c6JvN37xJJkJ0=
golang.org/x/text v0.24.0/go.mod h1:L8rBsPeo2pSS+xqN0d5u2ikmjtmoJbDBT1b7nHvFCdU=
golang.org/x/text v0.26.0 h1:P42AVeLghgTYr4+xUnTRKDMqpar+PtX7KWuNQL21L8M=
golang.org/x/text v0.26.0/go.mod h1:QK15LZJUUQVJxhz7wXgxSy/CJaTFjd0G+YLonydOVQA=
golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE=
hive/config/config.go
+1 -1
View File
@@ -59,7 +59,7 @@ type NATSSub struct {
type EBSRequests struct {
Requests []EBSRequest `mapstructure:"ebs_requests"`
Mu sync.Mutex
Mu sync.Mutex `json:"-"`
}
type EBSRequest struct {
hive/daemon/daemon.go
+339 -164
View File
@@ -234,16 +234,20 @@ func NewResourceManager() *ResourceManager {
func NewDaemon(cfg *config.Config) *Daemon {
ctx, cancel := context.WithCancel(context.Background())
return &Daemon{
config: cfg,
resourceMgr: NewResourceManager(),
ctx: ctx,
cancel: cancel,
Instances: vm.Instances{VMS: make(map[string]vm.VM)},
config: cfg,
resourceMgr: NewResourceManager(),
ctx: ctx,
cancel: cancel,
Instances: vm.Instances{VMS: make(map[string]*vm.VM)},
natsSubscriptions: make(map[string]*nats.Subscription),
}
}
// Start initializes and starts the daemon
func (d *Daemon) Start() error {
var err error
// Connect to NATS with options
opts := []nats.Option{
nats.Token(d.config.NATS.ACL.Token),
@@ -257,23 +261,56 @@ func (d *Daemon) Start() error {
}),
}
var err error
d.natsConn, err = nats.Connect(d.config.NATS.Host, opts...)
if err != nil {
return fmt.Errorf("failed to connect to NATS: %w", err)
}
log.Printf("Connected to NATS server at %s", d.config.NATS.Host)
// Load existing state for VMs
// Load state from disk
err = d.LoadState()
if err != nil {
slog.Warn("Failed to load state from disk, continuing with empty state", "error", err)
} else {
slog.Info("Loaded state from disk", "instance count", len(d.Instances.VMS))
//d.mu.Lock()
// Ensure mutex is usable after unmarshalling
d.Instances.Mu = sync.Mutex{}
for i := range d.Instances.VMS {
instance := d.Instances.VMS[i]
instance.EBSRequests.Mu = sync.Mutex{}
instance.QMPClient = &qmp.QMPClient{}
d.Instances.VMS[i] = instance
//d.Instances.VMS[i].EBSRequests.Mu = sync.Mutex{}
// d.Instances.VMS[i].QMPClient.Mu = sync.Mutex{}
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)
}
}
}
//d.mu.Unlock()
// Launch running instances
}
log.Printf("Subscribing to subject pattern: %s", "ec2.launch")
// Create a map to store subscriptions
d.natsSubscriptions = make(map[string]*nats.Subscription)
// Subscribe to EC2 events with queue group
d.natsSubscriptions[d.config.NATS.Sub.Subject], err = d.natsConn.QueueSubscribe("ec2.launch", "hive-workers", d.handleEC2Launch)
d.natsSubscriptions["ec2.launch"], err = d.natsConn.QueueSubscribe("ec2.launch", "hive-workers", d.handleEC2Launch)
if err != nil {
return fmt.Errorf("failed to subscribe to NATS: %w", err)
return fmt.Errorf("failed to subscribe to NATS ec2.launch: %w", err)
}
// Setup graceful shutdown
@@ -293,6 +330,71 @@ func (d *Daemon) Start() error {
return nil
}
// Write the state to disk
func (d *Daemon) WriteState() error {
d.Instances.Mu.Lock()
defer d.Instances.Mu.Unlock()
// Pretty print JSON with indent
jsonData, err := json.MarshalIndent(d.Instances, "", " ")
if err != nil {
return err
}
// Write the state to disk
configPath := fmt.Sprintf("%s/%s", d.config.BaseDir, "instances.json")
err = os.WriteFile(configPath, jsonData, 0644)
if err != nil {
return err
}
return nil
}
// Initalise VMs from state
func (d *Daemon) InitaliseVMs() {
/*
d.Instances.Mu.Lock()
defer d.Instances.Mu.Unlock()
// Step 1: Loop through each instance
for i := range d.Instances.VMS {
instance := d.Instances.VMS[i]
// Step 2: Mount each EBS volume
for _, ebsRequest := range instance.EBSRequests.Requests {
instance.EBSRequests.Mu.Lock()
defer instance.EBSRequests.Mu.Unlock()
}
d.Instances.VMS[i] = instance
}
*/
// Step 2: Loop through each instance and start it
}
// Load state from disk
func (d *Daemon) LoadState() error {
configPath := fmt.Sprintf("%s/%s", d.config.BaseDir, "instances.json")
jsonData, err := os.ReadFile(configPath)
if err != nil {
return err
}
if err := json.Unmarshal(jsonData, &d.Instances); err != nil {
return err
}
return nil
}
// NATS events
func (d *Daemon) handleEC2Describe(msg *nats.Msg) {
var ec2DescribeRequest config.EC2DescribeRequest
@@ -327,6 +429,13 @@ func (d *Daemon) handleEC2Describe(msg *nats.Msg) {
}
func (d *Daemon) SendQMPCommand(q *qmp.QMPClient, cmd qmp.QMPCommand, instanceId string) (*qmp.QMPResponse, error) {
// Confirm QMP client is initialized
if q.Encoder == nil || q.Decoder == nil {
return nil, fmt.Errorf("QMP client is not initialized")
}
// Lock the QMP client
q.Mu.Lock()
defer q.Mu.Unlock()
@@ -439,10 +548,23 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) {
}
if !instanceRunning {
// 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)
return
}
instance.Attributes = command.Attributes
d.Instances.Mu.Unlock()
}
// handleEC2Launch processes incoming EC2 launch requests
@@ -574,7 +696,7 @@ func (d *Daemon) setupShutdown() {
continue
}
msg, err := d.natsConn.Request("ebs.unmount", ebsUnMountRequest, 10*time.Second)
msg, err := d.natsConn.Request("ebs.unmount", ebsUnMountRequest, 30*time.Second)
if err != nil {
slog.Error("Failed to unmount volume %s for %s: %v", ebsRequest.Name, instance.ID, err)
} else {
@@ -600,6 +722,12 @@ func (d *Daemon) setupShutdown() {
// Close NATS connection
d.natsConn.Close()
// Write the state to disk
err := d.WriteState()
if err != nil {
slog.Error("Failed to write state to disk: %v", err)
}
// Wait for any ongoing operations to complete
// TODO: Implement cleanup of running instances
log.Println("Shutdown complete")
@@ -623,21 +751,13 @@ func (d *Daemon) launchEC2Instance(ec2Req EC2Request, msg *nats.Msg) (ec2Respons
var imageId string
var snapshotId string
var instance vm.VM
instance.ID = vm.GenerateEC2InstanceID()
instance.Status = "provisioning"
instanceId := vm.GenerateEC2InstanceID()
// Add state for our new instance
d.Instances.Mu.Lock()
d.Instances.VMS[instance.ID] = instance
d.Instances.Mu.Unlock()
d.natsSubscriptions[fmt.Sprintf("ec2.describe.%s", instance.ID)], err = d.natsConn.QueueSubscribe(fmt.Sprintf("ec2.describe.%s", instance.ID), "hive-events", d.handleEC2Describe)
if err != nil {
slog.Error("Failed to subscribe to NATS: %v", err)
return ec2Response, err
var instance = &vm.VM{
ID: instanceId,
Status: "provisioning",
InstanceType: ec2Req.InstanceType,
}
// Respond with the instance ID and status, polling required to track status
@@ -1204,61 +1324,161 @@ func (d *Daemon) launchEC2Instance(ec2Req EC2Request, msg *nats.Msg) (ec2Respons
// Step 5: Mount each volume via NBD, confirm running as expected for pre-flight checks.
// TODO: Run a goroutine for each volume
// Loop through each volume in volumes
instance.EBSRequests.Mu.Lock()
for k, v := range instance.EBSRequests.Requests {
fmt.Println(v)
// Send the volume payload as JSON
ebsMountRequest, err := json.Marshal(v)
if err != nil {
slog.Error("Failed to marshal volume payload: %v", err)
return ec2Response, err
}
reply, err := d.natsConn.Request("ebs.mount", ebsMountRequest, 10*time.Second)
// TODO: Improve timeout handling
if err != nil {
log.Fatalln(err)
return ec2Response, err
}
// Unmarshal the response
var ebsMountResponse config.EBSMountResponse
err = json.Unmarshal(reply.Data, &ebsMountResponse)
if err != nil {
slog.Error("Failed to unmarshal volume response: %v", err)
return ec2Response, err
}
fmt.Println(ebsMountResponse)
if ebsMountResponse.Error == "" {
// Append the NBD URI to the request
instance.EBSRequests.Requests[k].NBDURI = ebsMountResponse.URI
} else {
slog.Error("Failed to mount volume", "error", ebsMountResponse.Error)
return ec2Response, fmt.Errorf("failed to mount volume: %s", ebsMountResponse.Error)
}
err = d.LaunchInstance(instance)
if err != nil {
return ec2Response, err
}
// Step 10: Return the unique instance ID on success
ec2Response.InstanceID = instance.ID
// TODO: Use AWS style hostname, ip-<a-b-c-d>.<region>.compute.internal
ec2Response.Hostname = instance.ID
ec2Response.Success = true
return ec2Response, nil
}
func (d *Daemon) CreateQMPClient(instance *vm.VM) (err error) {
// Create a new QMP client to communicate with the instance
instance.QMPClient, err = qmp.NewQMPClient(instance.Config.QMPSocket)
// Send qmp_capabilities handshake to init
_, err = d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{Execute: "qmp_capabilities"}, instance.ID)
// Simple heartbeat to confirm QMP and the instanceis running / healthy
go func() {
for {
time.Sleep(30 * time.Second)
slog.Info("QMP heartbeat", "instance", instance.ID)
status, err := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{Execute: "query-status"}, instance.ID)
if err != nil {
slog.Error("Failed to send QMP command: %v", err)
// Check if the instance is stopping, mark as stopped
d.Instances.Mu.Lock()
defer d.Instances.Mu.Unlock()
if instance.Status == "powering_down" {
instance.Status = "stopped"
slog.Info("QMP Status - Instance %s stopped, exiting heartbeat", instance.ID)
// TODO: Improve, move to SendQMPCommand
// Unsubscribe from the NATS subject
slog.Info("Unsubscribing from NATS subject", "instance", instance.ID)
d.natsSubscriptions[fmt.Sprintf("ec2.cmd.%s", instance.ID)].Unsubscribe()
d.natsSubscriptions[fmt.Sprintf("ec2.describe.%s", instance.ID)].Unsubscribe()
// Close the QMP client connection
slog.Info("Closing QMP client connection", "instance", instance.ID)
instance.QMPClient.Conn.Close()
// Exit the goroutine
break
}
continue
}
slog.Info("QMP status", "status", string(status.Return))
}
}()
if err != nil {
slog.Error("Failed to create QMP client: %v", err)
return err
}
return nil
}
func (d *Daemon) LaunchInstance(instance *vm.VM) (err error) {
// Loop through each volume in volumes
err = d.MountVolumes(instance)
if err != nil {
slog.Error("Failed to mount volumes: %v", err)
return err
}
instance.EBSRequests.Mu.Unlock()
// Step 6: Launch the instance via QEMU/KVM
err = d.StartInstance(instance)
if err != nil {
slog.Error("Failed to launch instance: %v", err)
return err
}
/*
reply, err := nc.Request("ebs.mount", []byte(), 4*time.Second)
if err != nil {
log.Fatalln(err)
}
*/
// Step 7: Create QMP client to communicate with the instance
err = d.CreateQMPClient(instance)
if err != nil {
slog.Error("Failed to create QMP client: %v", err)
return err
}
// Step 8: Subscribe to start/stop/shutdown events
d.mu.Lock()
defer d.mu.Unlock()
d.natsSubscriptions[instance.ID], err = d.natsConn.QueueSubscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), "hive-events", d.handleEC2Events)
if err != nil {
return fmt.Errorf("failed to subscribe to NATS: %w", err)
}
d.natsSubscriptions[fmt.Sprintf("ec2.describe.%s", instance.ID)], err = d.natsConn.QueueSubscribe(fmt.Sprintf("ec2.describe.%s", instance.ID), "hive-events", d.handleEC2Describe)
if err != nil {
slog.Error("Failed to subscribe to NATS ec2.describe.%s: %v", instance.ID, err)
return err
}
// Step 9: Update the instance metadata for running state and volume attached
// Marshal to a JSON file
// Update state
d.Instances.Mu.Lock()
// Update to running state
instance.Status = "running"
d.Instances.VMS[instance.ID] = instance
d.Instances.Mu.Unlock()
err = d.WriteState()
if err != nil {
slog.Error("Failed to marshal launchVm: %v", err)
return err
}
return nil
}
func (d *Daemon) StartInstance(instance *vm.VM) error {
pidFile, err := utils.GeneratePidFile(instance.ID)
if err != nil {
slog.Error("Failed to generate PID file: %v", err)
return ec2Response, err
return err
}
var instanceType = d.resourceMgr.instanceTypes[instance.InstanceType]
instance.Config = vm.Config{
Name: instance.ID,
Daemonize: true,
@@ -1338,7 +1558,7 @@ func (d *Daemon) launchEC2Instance(ec2Req EC2Request, msg *nats.Msg) (ec2Respons
if err != nil {
slog.Error("Failed to generate QMP socket: %v", err)
return ec2Response, err
return err
}
instance.Config.QMPSocket = qmpSocket
@@ -1420,7 +1640,7 @@ func (d *Daemon) launchEC2Instance(ec2Req EC2Request, msg *nats.Msg) (ec2Respons
pid := <-processChan
if pid == 0 {
return ec2Response, fmt.Errorf("failed to start qemu")
return fmt.Errorf("failed to start qemu")
}
// Wait for 1 second to confirm nbdkit is running
@@ -1430,7 +1650,7 @@ func (d *Daemon) launchEC2Instance(ec2Req EC2Request, msg *nats.Msg) (ec2Respons
pts := <-ptsChan
if pts == 0 {
return ec2Response, fmt.Errorf("failed to get pts")
return fmt.Errorf("failed to get pts")
}
// Check if nbdkit exited immediately with an error
@@ -1438,120 +1658,75 @@ func (d *Daemon) launchEC2Instance(ec2Req EC2Request, msg *nats.Msg) (ec2Respons
case exitErr := <-exitChan:
if exitErr != 0 {
errorMsg := fmt.Errorf("qemu failed: %v", exitErr)
return ec2Response, errorMsg
return errorMsg
}
default:
// nbdkit is still running after 1 second, which means it started successfully
fmt.Println("QEMU started successfully and is running", "pts", pts)
}
/*
reply, err := nc.Request("ebs.mount", []byte(), 4*time.Second)
if err != nil {
log.Fatalln(err)
}
*/
// Step 7: Update the instance metadata for running state and volume attached
// Marshal to a JSON file
jsonData, err := json.Marshal(instance.Config)
if err != nil {
slog.Error("Failed to marshal launchVm: %v", err)
return ec2Response, err
}
// Write to a file
err = os.WriteFile(fmt.Sprintf("/tmp/vm-%s.json", instance.ID), jsonData, 0644)
if err != nil {
slog.Error("Failed to write launchVm to file: %v", err)
return ec2Response, err
}
// Step 8: Return the unique instance ID on success
ec2Response.InstanceID = instance.ID
// TODO: Use AWS style hostname, ip-<a-b-c-d>.<region>.compute.internal
ec2Response.Hostname = instance.ID
ec2Response.Success = true
// Add the instance to the local instances
pid, err = utils.ReadPidFile(instance.ID)
// Confirm the instance has booted
_, err = utils.ReadPidFile(instance.ID)
if err != nil {
slog.Error("Failed to read PID file: %v", err)
return ec2Response, err
return err
}
instance.QMPClient, err = qmp.NewQMPClient(instance.Config.QMPSocket)
return nil
}
// Send qmp_capabilities handshake
_, err = d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{Execute: "qmp_capabilities"}, instance.ID)
// MountVolumes mounts the volumes for an instance
func (d *Daemon) MountVolumes(instance *vm.VM) error {
// Simple heartbeat to confirm QMP is running
go func() {
for {
time.Sleep(30 * time.Second)
slog.Info("QMP heartbeat", "instance", instance.ID)
status, err := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{Execute: "query-status"}, instance.ID)
instance.EBSRequests.Mu.Lock()
for k, v := range instance.EBSRequests.Requests {
if err != nil {
slog.Error("Failed to send QMP command: %v", err)
fmt.Println(v)
// Check if the instance is stopping, mark as stopped
d.Instances.Mu.Lock()
defer d.Instances.Mu.Unlock()
instance, ok := d.Instances.VMS[instance.ID]
if !ok {
slog.Info("QMP Status - Instance %s not found", instance.ID)
continue
}
if instance.Status == "powering_down" {
instance.Status = "stopped"
d.Instances.VMS[instance.ID] = instance
slog.Info("QMP Status - Instance %s stopped, exiting heartbeat", instance.ID)
// Exit the goroutine
break
}
continue
}
slog.Info("QMP status", "status", status)
// Send the volume payload as JSON
ebsMountRequest, err := json.Marshal(v)
if err != nil {
slog.Error("Failed to marshal volume payload: %v", err)
return err
}
reply, err := d.natsConn.Request("ebs.mount", ebsMountRequest, 10*time.Second)
// TODO: Improve timeout handling
if err != nil {
log.Fatalln(err)
return err
}
// Unmarshal the response
var ebsMountResponse config.EBSMountResponse
err = json.Unmarshal(reply.Data, &ebsMountResponse)
if err != nil {
slog.Error("Failed to unmarshal volume response: %v", err)
return err
}
fmt.Println(ebsMountResponse)
if ebsMountResponse.Error == "" {
// Append the NBD URI to the request
instance.EBSRequests.Requests[k].NBDURI = ebsMountResponse.URI
} else {
slog.Error("Failed to mount volume", "error", ebsMountResponse.Error)
return fmt.Errorf("failed to mount volume: %s", ebsMountResponse.Error)
}
}()
if err != nil {
slog.Error("Failed to create QMP client: %v", err)
return ec2Response, err
}
// Update to running state
instance.Status = "running"
instance.EBSRequests.Mu.Unlock()
// Update state
d.Instances.Mu.Lock()
d.Instances.VMS[instance.ID] = instance
d.Instances.Mu.Unlock()
return nil
// Subscribe to start/stop/shutdown events
d.mu.Lock()
defer d.mu.Unlock()
d.natsSubscriptions[instance.ID], err = d.natsConn.QueueSubscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), "hive-events", d.handleEC2Events)
if err != nil {
return ec2Response, fmt.Errorf("failed to subscribe to NATS: %w", err)
}
return ec2Response, nil
}
// canAllocate checks if there are enough resources available
hive/qmp/qmp.go
+6
View File
@@ -23,6 +23,12 @@ var CommandResponseTypes = map[string]unmarshalTarget{
type Command struct {
ID string `json:"id"`
QMPCommand QMPCommand `json:"command"`
Attributes Attributes `json:"attributes"`
}
type Attributes struct {
StopInstance bool `json:"stop_instance"`
DeleteInstance bool `json:"delete_instance"`
}
// (QEMU) stop / cont (resume) / system_powerdown / system_reset / system_wakeup
hive/vm/vm.go
+13 -9
View File
@@ -29,21 +29,25 @@ qemu-system-x86_64 \
*/
type VM struct {
ID string `json:"id"`
PID int `json:"pid"`
PTS int `json:"pts"`
Running bool `json:"running"`
Status string `json:"status"`
Config Config `json:"config"`
ID string `json:"id"`
PID int `json:"pid"`
PTS int `json:"pts"`
Running bool `json:"running"`
Status string `json:"status"`
InstanceType string `json:"instance_type"`
Config Config `json:"config"`
EBSRequests config.EBSRequests `json:"ebs_requests"`
QMPClient *qmp.QMPClient
QMPClient *qmp.QMPClient `json:"-"`
// User attributes (user initiated stop/delete)
Attributes qmp.Attributes `json:"attributes"`
}
type Instances struct {
VMS map[string]VM `json:"vms"`
Mu sync.Mutex
VMS map[string]*VM `json:"vms"`
Mu sync.Mutex `json:"-"`
}
type NetDev struct {