From d05fb7cff3ce6fe74207b557f8a72efcadb56a6e Mon Sep 17 00:00:00 2001 From: Ben Duncan <5178243+benduncan@users.noreply.github.com> Date: Fri, 1 Aug 2025 18:41:20 +0800 Subject: [PATCH] VM management for rebooting and state --- go.mod | 2 +- go.sum | 6 +- hive/config/config.go | 2 +- hive/daemon/daemon.go | 503 ++++++++++++++++++++++++++++-------------- hive/qmp/qmp.go | 6 + hive/vm/vm.go | 22 +- 6 files changed, 362 insertions(+), 179 deletions(-) diff --git a/go.mod b/go.mod index 99bca0448..8ce95a40b 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index f9be1d588..fe546aec9 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/hive/config/config.go b/hive/config/config.go index 490c3b601..2257b20ee 100644 --- a/hive/config/config.go +++ b/hive/config/config.go @@ -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 { diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 636600194..3ee9ba750 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -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-..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-..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 diff --git a/hive/qmp/qmp.go b/hive/qmp/qmp.go index b9c18bda0..50b2856e8 100644 --- a/hive/qmp/qmp.go +++ b/hive/qmp/qmp.go @@ -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 diff --git a/hive/vm/vm.go b/hive/vm/vm.go index 9678b3c7e..0220f2065 100644 --- a/hive/vm/vm.go +++ b/hive/vm/vm.go @@ -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 {