From 191ad659e83ab6825cd05e2df8959e45306c97c7 Mon Sep 17 00:00:00 2001 From: Ben Duncan <5178243+benduncan@users.noreply.github.com> Date: Wed, 12 Nov 2025 21:30:25 +0800 Subject: [PATCH] Added start/stop/terminate instance, describe instance support, and improved install docs for development --- INSTALL.md | 351 ++++++++++----- TODO.md | 15 + cmd/hive/cmd/admin.go | 7 +- hive/config/config.go | 3 +- hive/daemon/daemon.go | 413 ++++++++++++++---- hive/gateway/ec2.go | 92 +++- .../gateway/ec2/instance/DescribeInstances.go | 111 +++++ hive/gateway/ec2/instance/StartInstances.go | 106 +++++ hive/gateway/ec2/instance/StopInstances.go | 105 +++++ .../ec2/instance/TerminateInstances.go | 105 +++++ hive/gateway/gateway.go | 1 + hive/handlers/ec2/instance/service_impl.go | 22 +- hive/qmp/qmp.go | 5 +- hive/services/viperblockd/viperblockd.go | 74 +--- hive/utils/utils.go | 97 ++-- hive/vm/vm.go | 9 + 16 files changed, 1203 insertions(+), 313 deletions(-) create mode 100644 hive/gateway/ec2/instance/DescribeInstances.go create mode 100644 hive/gateway/ec2/instance/StartInstances.go create mode 100644 hive/gateway/ec2/instance/StopInstances.go create mode 100644 hive/gateway/ec2/instance/TerminateInstances.go diff --git a/INSTALL.md b/INSTALL.md index e4c91df04..d19900aa3 100644 --- a/INSTALL.md +++ b/INSTALL.md @@ -11,19 +11,19 @@ sudo apt install nbdkit nbdkit-plugin-dev pkg-config qemu-system qemu-utils qemu Create hive user -``` +```bash sudo adduser --disabled-password hive ``` Add hive to the libvirt group to manage VMs -``` +```bash sudo adduser hive libvirt ``` Hive provided AWS API/SDK layer functionality, which requires the AWS CLI tool to be installed to interface with the system. -``` +```bash sudo apt install awscli ``` @@ -38,58 +38,17 @@ make Confirm ./bin/hive exists and executable. -``` -./bin/hive - -__/\\\________/\\\__/\\\\\\\\\\\__/\\\________/\\\__/\\\\\\\\\\\\\\\_ - _\/\\\_______\/\\\_\/////\\\///__\/\\\_______\/\\\_\/\\\///////////__ - _\/\\\_______\/\\\_____\/\\\_____\//\\\______/\\\__\/\\\_____________ - _\/\\\\\\\\\\\\\\\_____\/\\\______\//\\\____/\\\___\/\\\\\\\\\\\_____ - _\/\\\/////////\\\_____\/\\\_______\//\\\__/\\\____\/\\\///////______ - _\/\\\_______\/\\\_____\/\\\________\//\\\/\\\_____\/\\\_____________ - _\/\\\_______\/\\\_____\/\\\_________\//\\\\\______\/\\\_____________ - _\/\\\_______\/\\\__/\\\\\\\\\\\______\//\\\_______\/\\\\\\\\\\\\\\\_ - _\///________\///__\///////////________\///________\///////////////__ - -Hive – Open source AWS-compatible platform for secure edge deployments. -Run EC2, VPC, S3, and EBS-like services on bare metal with full control. -Built for environments where running in the cloud isn’t an option. -Whether you’re deploying to edge sites, private data-centers, or operating -in low-connectivity or highly contested environments - -Usage: - hive [command] - -Available Commands: - completion Generate the autocompletion script for the specified shell - help Help about any command - service Manage Hive services - -Flags: - --access-key string AWS access key (overrides config file and env) - --base-dir string Viperblock base directory (overrides config file and env) - --config string config file (required) - -h, --help help for hive - --host string AWS Endpoint (overrides config file and env) - --nats-host string NATS server host (overrides config file and env) - --nats-subject string NATS subscription subject (overrides config file and env) - --nats-token string NATS authentication token (overrides config file and env) - --secret-key string AWS secret key (overrides config file and env) - -Use "hive [command] --help" for more information about a command. -``` - ## Init When running Hive for the first time, run the init function to create the default directories for data, config files and layout required. -``` +```bash ./bin/hive admin init ``` Next, set the AWS profile to use `hive` which points to the local environment. -``` +```bash export AWS_PROFILE=hive ``` @@ -109,7 +68,7 @@ For first install users, create or import an existing key pair which can be used Import an existing key pair, replace `~/.ssh/id_rsa.pub` with your specified key. -``` +```bash aws ec2 import-key-pair --key-name "hive-key" --public-key-material fileb://~/.ssh/id_rsa.pub ``` @@ -137,7 +96,7 @@ ssh-keygen -y -f ~/.ssh/hive-key > ~/.ssh/hive-key.pub Validate the new key is available -``` +```bash aws ec2 describe-key-pairs ``` @@ -170,14 +129,13 @@ Discover available images to automatically download and install. This will pull ./bin/hive admin images list ``` -``` -Name | Distro | Version | Arch -alpine-3.22.2-arm64 | alpine | 3.22.2 | arm64 -alpine-3.22.2-x86_64 | alpine | 3.22.2 | x86_64 -debian-12-arm64 | debian | 12 | arm64 -debian-12-x86_64 | debian | 12 | x86_64 -ubuntu-24.04-arm64 | ubuntu | 24.04 | arm64 -ubuntu-24.04-x86_64 | ubuntu | 24.04 | x86_64 +```bash +NAME | DISTRO | VERSION | ARCH | BOOT +alpine-3.22.2-x86_64 | alpine | 3.22.2 | x86_64 | bios +debian-12-arm64 | debian | 12 | arm64 | bios +debian-12-x86_64 | debian | 12 | x86_64 | bios +ubuntu-24.04-arm64 | ubuntu | 24.04 | arm64 | bios +ubuntu-24.04-x86_64 | ubuntu | 24.04 | x86_64 | bios ``` Next, choose the image you would like to import as an AMI. @@ -186,25 +144,19 @@ Next, choose the image you would like to import as an AMI. ./bin/hive admin images import --name debian-12-arm64 --force ``` -``` +```bash Downloading image https://cdimage.debian.org/cdimage/cloud/bookworm/latest/debian-12-generic-arm64.tar.xz to /home/ben/hive/images/debian/12/arm64/debian-12-generic-amd64.tar.xz Downloading local-debian-12-arm64 [283748988/283748988] ██████████████ 100% | 1s Saved /home/ben/hive/images/debian/12/arm64/debian-12-generic-amd64.tar.xz (270.6 MiB) Extracted image to: /home/ben/hive/images/debian/12/arm64/disk.raw -AMI import complete +✅ Image import complete. Image-ID (AMI): ami-e29fcc65734aec9ea ``` Next, verify available disk images to confirm the import was successful ```bash -aws ec2 describe-images -``` - -Note the output to launch an instance, specifically the `"ImageId": ami-XXX` attribute. - -``` -export HIVE_AMI='ami-XXX' +aws ec2 describe-images --image-ids ami-e29fcc65734aec9ea ``` ### Manual import @@ -213,19 +165,19 @@ Using this method you can import any OS disk image. For example, download the De Download the image: -``` +```bash wget https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-arm64.tar.xz -O ~/debian-12-genericcloud-arm64.tar.xz ``` Import as an AMI to the backend store: -``` +```bash ./bin/hive admin images import --file ~/debian-12-genericcloud-arm64.tar.xz --arch arm64 --distro debian --version 12 ``` Next, verify available disk images to confirm the import was successful -``` +```bash aws ec2 describe-images ``` @@ -233,7 +185,13 @@ aws ec2 describe-images Once Hive is successfully installed and bootstrapped with a system AMI and SSH keys, proceed to run an instance. Replace `ami-XXX` with your imported ImageId above. +```bash +export HIVE_AMI="ami-XXX" ``` + +Launch a new instance, note `hive-key` is the SSH key specified in the previous stage. + +```bash aws ec2 run-instances \ --image-id $HIVE_AMI \ --instance-type t3.micro \ @@ -243,68 +201,235 @@ aws ec2 run-instances \ --count 1 ``` -## Start services +A sample response is below from the `RunInstance` request, note the `InstanceId` attribute: -### NATS - -An embedded NATS server provides messaging between components of Hive and is a requirement. - -``` -./bin/hive service nats start --debug +```json +{ + "ReservationId": "r-f101157331e261a68", + "OwnerId": "123456789012", + "Instances": [ + { + "InstanceId": "i-36765eb0c6609e4d2", + "ImageId": "ami-15c3dcfe607460f15", + "State": { + "Code": 0, + "Name": "pending" + }, + "KeyName": "hive-key", + "InstanceType": "t3.micro", + "LaunchTime": "2025-11-12T13:07:47.548000+00:00" + } + ] +} ``` -### Hive daemon +Export the instance ID for following the rest of the tutorial -The background Hive daemon is a core service which accepts requests to provision services such as the AWS SDK/CLI. - -``` -./bin/hive service hive start --config config/hive.toml +```bash +export INSTANCE_ID="i-XXX" ``` -## 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. +Next, validate the running instance is ready +```bash +aws ec2 describe-instances --instance-ids $INSTANCE_ID ``` -nats req --reply-timeout=5s ec2.cmd.i-ebaf0fd46cad14c85 '{ "id": "i-ebaf0fd46cad14c85", "command": { "execute": "system_powerdown", "arguments": {} } }' + +Confirm the `State.Name` attribute is set as `running` + +```json +{ + "Reservations": [ + { + "ReservationId": "r-f101157331e261a68", + "OwnerId": "123456789012", + "Instances": [ + { + "InstanceId": "i-36765eb0c6609e4d2", + "ImageId": "ami-15c3dcfe607460f15", + "State": { + "Code": 16, + "Name": "running" + }, + "KeyName": "hive-key", + "InstanceType": "t3.micro", + "LaunchTime": "2025-11-12T13:07:47.548000+00:00" + } + ] + } + ] +} +``` + +## SSH connection (development) + +For a Hive development environment (toggled off for production), a local SSH port forwaring will be active to connect directly to the instance, regardless of the VPC and network settings. + +Determine the SSH port allocated + +```bash +ps auxw | grep $INSTANCE_ID +``` + +```bash +qemu-system-x86_64 -daemonize -pidfile /run/user/1000/i-36765eb0c6609e4d.pid -qmp unix:/run/user/1000/qmp-i-36765eb0c6609e4d2.sock,server,nowait -enable-kvm -M ubuntu -serial pty -cpu host -smp 2 -m 1024 -drive file=nbd://127.0.0.1:42653,format=raw,if=none,media=disk,id=os -drive file=nbd://127.0.0.1:44499,format=raw,if=virtio,media=cdrom,id=cloudinit -device virtio-blk-pci,drive=os,bootindex=1 -device virtio-rng-pci -device virtio-net-pci,netdev=net0 -netdev user,id=net0,hostfwd=tcp:127.0.0.1:33683-:22 +``` + +Note the `hostfwd=tcp:127.0.0.1:33683-:22`, in this case the local port `33683` will connect to the new instance. + +```bash +ssh -i ~/.ssh/hive-key ec2-user@127.0.0.1 -p 33683 + +... + +Linux hive-vm-36765eb0 6.1.0-40-amd64 #1 SMP PREEMPT_DYNAMIC Debian 6.1.153-1 (2025-09-20) x86_64 +ec2-user@hive-vm-36765eb0:~$ +``` + +Congratulations, your first AMI image is imported, a new EC2 instance launched, and successfully connected via SSH for the configured SSH key, using the OS `cloud-init` procedure. + +## Managing instances + +### Stop instance + +To stop a running instance gracefully: + +```bash +aws ec2 stop-instances --instance-ids $INSTANCE_ID +``` + +```json +{ + "StoppingInstances": [ + { + "InstanceId": "i-36765eb0c6609e4d2", + "CurrentState": { + "Code": 64, + "Name": "stopping" + }, + "PreviousState": { + "Code": 16, + "Name": "running" + } + } + ] +``` + +Next, confirm the instance has stopped as requested + +```bash +aws ec2 describe-instances --instance-ids $INSTANCE_ID +``` + +```json +{ + "Reservations": [ + { + "ReservationId": "r-f101157331e261a68", + "OwnerId": "123456789012", + "Instances": [ + { + "InstanceId": "i-36765eb0c6609e4d2", + "ImageId": "ami-15c3dcfe607460f15", + "State": { + "Code": 80, + "Name": "stopped" + }, + "KeyName": "hive-key", + "InstanceType": "t3.micro", + "LaunchTime": "2025-11-12T13:07:47.548000+00:00" + } + ] + } + ] +} +``` + +### Start Instance + +To start a previously stopped instance: + +```bash +aws ec2 start-instances --instance-ids $INSTANCE_ID +``` + +```json +{ + "StartingInstances": [ + { + "InstanceId": "i-36765eb0c6609e4d2", + "CurrentState": { + "Code": 0, + "Name": "pending" + }, + "PreviousState": { + "Code": 80, + "Name": "stopped" + } + } + ] +``` + +Next, validate the instance is running as expected + +```bash +aws ec2 describe-instances --instance-ids "i-36765eb0c6609e4d2" +``` + +```json +{ + "Reservations": [ + { + "ReservationId": "r-f101157331e261a68", + "OwnerId": "123456789012", + "Instances": [ + { + "InstanceId": "i-36765eb0c6609e4d2", + "ImageId": "ami-15c3dcfe607460f15", + "State": { + "Code": 16, + "Name": "running" + }, + "KeyName": "hive-key", + "InstanceType": "t3.micro", + "LaunchTime": "2025-11-12T13:07:47.548000+00:00" + } + ] + } + ] +} ``` ### 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. +To terminate an instance, which will first stop the instance, and on success, remove the EBS volumes and permanately remove the instance data. -``` -nats req --reply-timeout=5s ec2.cmd.i-f38ac0490f1683650 '{ "id": "i-f38ac0490f1683650", "attributes": { "stop_instance": true }, "command": { "execute": "system_powerdown", "arguments": {} } }' +```bash +aws ec2 terminate-instances --instance-ids $INSTANCE_ID ``` -### Stop VM - -``` -nats req --reply-timeout=5s ec2.cmd.i-ebaf0fd46cad14c85 '{ "id": "i-ebaf0fd46cad14c85", "command": { "execute": "stop", "arguments": {} } }' +```json +{ + "TerminatingInstances": [ + { + "InstanceId": "i-36765eb0c6609e4d2", + "CurrentState": { + "Code": 16, + "Name": "running" + }, + "PreviousState": { + "Code": 16, + "Name": "running" + } + } + ] +} ``` -### Resume VM +Next, validate the instance is removed, this may take a few minutes depending the instance volume size. -``` -nats req --reply-timeout=5s ec2.cmd.i-ebaf0fd46cad14c85 '{ "id": "i-ebaf0fd46cad14c85", "command": { "execute": "cont", "arguments": {} } }' +```bash +aws ec2 describe-instances --instance-ids $INSTANCE_ID ``` -### Restart - -``` -nats req --reply-timeout=5s ec2.cmd.i-1bb0cfd0e48bfa232 '{ "id": "i-1bb0cfd0e48bfa232", "command": { "execute": "system_reset", "arguments": {} } }' -``` - -### Query status - -``` -nats req --reply-timeout=5s ec2.cmd.i-ebaf0fd46cad14c85 '{ "id": "i-ebaf0fd46cad14c85", "command": { "execute": "query-status", "arguments": {} } }' -``` - -### Query devices - -``` -nats req --reply-timeout=5s ec2.cmd.i-ebaf0fd46cad14c85 '{ "id": "i-ebaf0fd46cad14c85", "command": { "execute": "query-device", "arguments": {} } }' -``` +On success no data will be returned, since the instance is no longer available. diff --git a/TODO.md b/TODO.md index b5a88fa11..6e8acfdd8 100644 --- a/TODO.md +++ b/TODO.md @@ -2,6 +2,21 @@ This roadmap has been integrated into the comprehensive [HIVE_DEVELOPMENT_PLAN.md](HIVE_DEVELOPMENT_PLAN.md). +## Update Nov 2025 + +* Implement multi-tenant support +* Move config settings from ~/hive/*.toml, to using Nats Jetstream for core config which can be synced between nodes. +* Implement a lightweight IAM using Nats Jetstream, vs current config files for auth settings. +* Implement Reed Solomon Encoding for Predastore (S3) objects and a WAL implementation, for storing multiple objects in a single shard (e.g 4MB), with the WAL referencing the location of each object (e.g 4kb min) + * Implement basic KY lookup, object key sha512 (bucket/key), location to shard on S3 (obj.0000124.bin), read WAL (e.g first 4096 bytes) to determine location of the object. e.g key-1234 => obj.0000124.bin, wal header (4096 bytes) > key-1234 == offset location (seek) 16384, len 32768. + * Read multiple nodes (e.g 5 predastore instances, k = 3 (data shards), n = 5 (total shards), n - k = 2 (parity shards) ) +* Complete core scaffolding AWS SDK/API requirements (ec2 describe-instances, run-instances, etc) +* Implement UEFI support for image downloads and `qemu` exec in `vm.go` +* Confirm Alpine Linux, fails import image AMI > (run-instance) ""Failed to read block from AMI source" err="request out of range" - Block size correct? +* Improve shutdown gracefully, `./scripts/stop-dev.sh` waits 60 seconds, while qemu/nbd could still be shutting down. +* Add delete-volume support via EBS (s3 vol-*) for terminated instance +* Add default LRU cache support for viperblock, depending on the instance type / volume size and system memory available. + ## Original TODO Items → Development Plan Integration ### ✅ **Completed Integration** diff --git a/cmd/hive/cmd/admin.go b/cmd/hive/cmd/admin.go index 306eba009..598b1a513 100644 --- a/cmd/hive/cmd/admin.go +++ b/cmd/hive/cmd/admin.go @@ -387,9 +387,10 @@ func runimagesImportCmd(cmd *cobra.Command, args []string) { os.Exit(1) } - fmt.Println(tmpDir) defer os.RemoveAll(tmpDir) + fmt.Printf("✅ Image import complete. Image-ID (AMI): %s\n", volumeId) + } // List remote images available @@ -398,7 +399,7 @@ func runimagesListCmd(cmd *cobra.Command, args []string) { //fmt.Println(availableImages) tableData := pterm.TableData{ - {"Name", "Distro", "Version", "Arch"}, + {"NAME", "DISTRO", "VERSION", "ARCH", "BOOT"}, } // Sort A .. Z @@ -417,7 +418,7 @@ func runimagesListCmd(cmd *cobra.Command, args []string) { img := utils.AvailableImages[k] //for _, img := range utils.AvailableImages { - tableData = append(tableData, []string{img.Name, img.Distro, img.Version, img.Arch}) + tableData = append(tableData, []string{img.Name, img.Distro, img.Version, img.Arch, img.BootMode}) } // Create a table with the defined data. diff --git a/hive/config/config.go b/hive/config/config.go index 284cbad66..836fd3428 100644 --- a/hive/config/config.go +++ b/hive/config/config.go @@ -32,7 +32,8 @@ type AWSGWConfig struct { TLSCert string `mapstructure:"tlscert"` Config string `mapstructure:"config"` - Debug bool `mapstructure:"debug"` + Debug bool `mapstructure:"debug"` + ExpectedNodes int `mapstructure:"expected_nodes"` } type PredastoreConfig struct { diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index fc69a5bdf..b012b4b00 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -283,11 +283,13 @@ func (d *Daemon) Start() error { log.Printf("Subscribing to subject pattern: %s", "ec2.launch") // Subscribe to EC2 events with queue group (legacy topic for backward compatibility) - d.natsSubscriptions["ec2.launch"], err = d.natsConn.QueueSubscribe("ec2.launch", "hive-workers", d.handleEC2RunInstances) + /* + d.natsSubscriptions["ec2.launch"], err = d.natsConn.QueueSubscribe("ec2.launch", "hive-workers", d.handleEC2RunInstances) - if err != nil { - return fmt.Errorf("failed to subscribe to NATS ec2.launch: %w", err) - } + if err != nil { + return fmt.Errorf("failed to subscribe to NATS ec2.launch: %w", err) + } + */ log.Printf("Subscribing to subject pattern: %s", "ec2.RunInstances") @@ -343,6 +345,15 @@ func (d *Daemon) Start() error { return fmt.Errorf("failed to subscribe to NATS ec2.DescribeImages: %w", err) } + log.Printf("Subscribing to subject pattern: %s", "ec2.DescribeInstances") + + // Subscribe to EC2 DescribeInstances - no queue group for multi-node fan-out + d.natsSubscriptions["ec2.DescribeInstances"], err = d.natsConn.Subscribe("ec2.DescribeInstances", d.handleEC2DescribeInstances) + + if err != nil { + return fmt.Errorf("failed to subscribe to NATS ec2.DescribeInstances: %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) @@ -455,9 +466,9 @@ func (d *Daemon) handleEC2StartInstances(msg *nats.Msg) { instance, ok := d.Instances.VMS[ec2StartInstance.InstanceID] if !ok { - slog.Error("EC2 Describe Request - Instance not found", "instanceId", ec2StartInstanceResponse.InstanceID) + slog.Error("EC2 Start Request - Instance not found", "instanceId", ec2StartInstanceResponse.InstanceID) ec2StartInstanceResponse.InstanceID = ec2StartInstance.InstanceID - ec2StartInstanceResponse.Error = fmt.Sprintf("Instance %s not found", ec2StartInstanceResponse.InstanceID) + ec2StartInstanceResponse.Error = awserrors.ErrorInvalidInstanceIDNotFound ec2StartInstanceResponse.Respond(msg) return } @@ -579,10 +590,13 @@ func (d *Daemon) SendQMPCommand(q *qmp.QMPClient, cmd qmp.QMPCommand, instanceId } } -// handleEC2Events processes incoming EC2 events +// handleEC2Events processes incoming EC2 events (start, stop, terminate) func (d *Daemon) handleEC2Events(msg *nats.Msg) { var command qmp.Command + var resp *qmp.QMPResponse + var err error + if err := json.Unmarshal(msg.Data, &command); err != nil { log.Printf("Error unmarshaling QMP command: %v", err) return @@ -600,35 +614,97 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { return } - // Send the command to the instance - resp, err := d.SendQMPCommand(instance.QMPClient, command.QMPCommand, instance.ID) + // Start an instance + if command.Attributes.StartInstance { + slog.Info("Starting instance", "id", command.ID) + // Launch the instance infrastructure (QEMU, QMP, NATS subscriptions) + err := d.LaunchInstance(instance) + + if err != nil { + slog.Error("handleEC2RunInstances LaunchInstance failed", "err", err) + // TODO: Confirm LaunchInstances does this - Free the resource + instanceType := d.resourceMgr.instanceTypes[instance.InstanceType] + d.resourceMgr.deallocate(instanceType) + return + } + + slog.Info("handleEC2RunInstances launched", "instanceId", instance.ID) + + resp = &qmp.QMPResponse{ + Return: []byte(fmt.Sprintf(`{"status":"running","instanceId":"%s"}`, instance.ID)), + } + + } else { + + // 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", "err", err) + return + } + + slog.Debug("RAW QMP Response", "resp", string(resp.Return)) + + // Unmarshal the response + target, ok := qmp.CommandResponseTypes[command.QMPCommand.Execute] + if !ok { + slog.Warn("Unhandled QMP command", "cmd", command.QMPCommand.Execute) + return + } + + if err := json.Unmarshal(resp.Return, target); err != nil { + slog.Error("Failed to unmarshal QMP response", "cmd", command.QMPCommand.Execute, "err", err) + return + } + + } + + // If a terminate command, clean up resources + if command.Attributes.TerminateInstance { + slog.Info("Terminating instance", "id", command.ID) + + // Improve, need to pass a map + terminateInstance := make(map[string]*vm.VM) + terminateInstance[instance.ID] = instance + err = d.stopInstance(terminateInstance, true) + + if err != nil { + slog.Error("Failed to terminate instance", "err", err) + return + } + + // Last, delete the instance volumes + + // Free resources + instanceType := d.resourceMgr.instanceTypes[instance.InstanceType] + d.resourceMgr.deallocate(instanceType) + + // Remove instance from state + d.Instances.Mu.Lock() + delete(d.Instances.VMS, instance.ID) + d.Instances.Mu.Unlock() + + slog.Info("Instance terminated", "id", command.ID) + } else { + + // Update the instance attributes + d.Instances.Mu.Lock() + instance.Attributes = command.Attributes + d.Instances.Mu.Unlock() + + } + + // Write the state to disk + err = d.WriteState() if err != nil { - slog.Error("Failed to send QMP command", "err", err) - return - } - - slog.Debug("RAW QMP Response", "resp", string(resp.Return)) - - // Unmarshal the response - target, ok := qmp.CommandResponseTypes[command.QMPCommand.Execute] - if !ok { - slog.Warn("Unhandled QMP command", "cmd", command.QMPCommand.Execute) - return - } - - if err := json.Unmarshal(resp.Return, target); err != nil { - slog.Error("Failed to unmarshal QMP response", "cmd", command.QMPCommand.Execute, "err", err) - return + slog.Error("Failed to write state to disk", "err", err) } + // Respond to NATS msg.Respond(resp.Return) - // Update the instance attributes - d.Instances.Mu.Lock() - instance.Attributes = command.Attributes - d.Instances.Mu.Unlock() - } // handleEC2RunInstances processes incoming EC2 RunInstances requests @@ -812,6 +888,105 @@ func (d *Daemon) handleEC2DeleteKeyPair(msg *nats.Msg) { slog.Info("handleEC2DeleteKeyPair completed") } +func (d *Daemon) stopInstance(instances map[string]*vm.VM, deleteVolume bool) error { + + // Signal to shutdown each VM + var wg sync.WaitGroup + + // Run asynchronously within a worker group + for _, instance := range instances { + instance := instance // capture loop variable + + wg.Add(1) + + go func() { + defer wg.Done() + + // Send shutdown command + _, err := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{Execute: "system_powerdown"}, instance.ID) + + if err != nil { + slog.Error("Failed to send system_powerdown", "id", instance.ID, "err", err) + return + } + + // Wait for PID file removal + err = utils.WaitForPidFileRemoval(instance.ID, 60*time.Second) + + if err != nil { + slog.Error("Timeout waiting for PID file removal", "id", instance.ID, "err", err) + + // Try force killing the process + pid, err := utils.ReadPidFile(instance.ID) + if err != nil { + slog.Error("Failed to read PID file", "id", instance.ID, "err", err) + } else { + slog.Info("Killing process", "pid", pid, "id", instance.ID) + // Send SIG directly if QMP fails + utils.KillProcess(pid) + } + } + + // Unmount all EBS volumes + instance.EBSRequests.Mu.Lock() + defer instance.EBSRequests.Mu.Unlock() + + for _, ebsRequest := range instance.EBSRequests.Requests { + + // Send the volume payload as JSON + ebsUnMountRequest, err := json.Marshal(ebsRequest) + + if err != nil { + slog.Error("Failed to marshal volume payload", "err", err) + continue + } + + msg, err := d.natsConn.Request("ebs.unmount", ebsUnMountRequest, 30*time.Second) + if err != nil { + slog.Error("Failed to unmount volume", "name", ebsRequest.Name, "id", instance.ID, "err", err) + } else { + slog.Info("Unmounted Viperblock volume", "id", instance.ID, "data", string(msg.Data)) + } + } + + // If flagged for termination (delete Volume) + if deleteVolume { + for _, ebsRequest := range instance.EBSRequests.Requests { + + // Send the volume payload as JSON + ebsDeleteRequest, err := json.Marshal(ebsRequest) + + if err != nil { + slog.Error("Failed to marshal volume payload", "err", err) + continue + } + + msg, err := d.natsConn.Request("ebs.delete", ebsDeleteRequest, 30*time.Second) + if err != nil { + slog.Error("Failed to delete volume", "name", ebsRequest.Name, "id", instance.ID, "err", err) + } else { + slog.Info("Deleted Viperblock volume", "id", instance.ID, "data", string(msg.Data)) + } + } + } + + }() + } + + // Wait for all shutdowns to finish + wg.Wait() + + // Unsubscribe from NATS subjects that match instances + for _, instance := range instances { + slog.Info("Unsubscribing from NATS subject", "instance", instance.ID) + d.natsSubscriptions[fmt.Sprintf("ec2.cmd.%s", instance.ID)].Unsubscribe() + // TODO: Remove redundant subscription if not used + //d.natsSubscriptions[fmt.Sprintf("ec2.describe.%s", instance.ID)].Unsubscribe() + } + return nil + +} + func (d *Daemon) setupShutdown() { d.shutdownWg.Add(1) go func() { @@ -826,67 +1001,8 @@ func (d *Daemon) setupShutdown() { // Cancel context d.cancel() - // Signal to shutdown each VM - var wg sync.WaitGroup - - for _, instance := range d.Instances.VMS { - instance := instance // capture loop variable - wg.Add(1) - - go func() { - defer wg.Done() - - // Send shutdown command - _, err := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{Execute: "system_powerdown"}, instance.ID) - - if err != nil { - slog.Error("Failed to send system_powerdown", "id", instance.ID, "err", err) - return - } - - // Wait for PID file removal - err = utils.WaitForPidFileRemoval(instance.ID, 60*time.Second) - - if err != nil { - slog.Error("Timeout waiting for PID file removal", "id", instance.ID, "err", err) - - // Try force killing the process - pid, err := utils.ReadPidFile(instance.ID) - if err != nil { - slog.Error("Failed to read PID file", "id", instance.ID, "err", err) - } else { - slog.Info("Killing process", "pid", pid, "id", instance.ID) - // Send SIG directly if QMP fails - utils.KillProcess(pid) - } - } - - // Unmount all EBS volumes - instance.EBSRequests.Mu.Lock() - defer instance.EBSRequests.Mu.Unlock() - - for _, ebsRequest := range instance.EBSRequests.Requests { - - // Send the volume payload as JSON - ebsUnMountRequest, err := json.Marshal(ebsRequest) - - if err != nil { - slog.Error("Failed to marshal volume payload", "err", err) - continue - } - - msg, err := d.natsConn.Request("ebs.unmount", ebsUnMountRequest, 30*time.Second) - if err != nil { - slog.Error("Failed to unmount volume", "name", ebsRequest.Name, "id", instance.ID, "err", err) - } else { - slog.Info("Unmounted Viperblock volume", "id", instance.ID, "data", string(msg.Data)) - } - } - }() - } - - // Wait for all shutdowns to finish - wg.Wait() + // Pass instances to terminate + d.stopInstance(d.Instances.VMS, false) // Final cleanup for _, sub := range d.natsSubscriptions { @@ -950,7 +1066,7 @@ func (d *Daemon) CreateQMPClient(instance *vm.VM) (err error) { // 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() + //d.natsSubscriptions[fmt.Sprintf("ec2.describe.%s", instance.ID)].Unsubscribe() // Close the QMP client connection slog.Info("Closing QMP client connection", "instance", instance.ID) @@ -1031,12 +1147,15 @@ func (d *Daemon) LaunchInstance(instance *vm.VM) (err error) { return 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) + // TODO: Replaced with describe-instances with Inbox subscription + /* + 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", "id", instance.ID, "err", err) - return err - } + if err != nil { + slog.Error("Failed to subscribe to NATS ec2.describe", "id", instance.ID, "err", err) + return err + } + */ // Step 9: Update the instance metadata for running state and volume attached // Marshal to a JSON file @@ -1045,6 +1164,12 @@ func (d *Daemon) LaunchInstance(instance *vm.VM) (err error) { // Update to running state instance.Status = "running" + // Update EC2 Instance state for API compatibility + if instance.Instance != nil { + instance.Instance.State.SetCode(16) // 16 = running + instance.Instance.State.SetName("running") + } + d.Instances.VMS[instance.ID] = instance d.Instances.Mu.Unlock() @@ -1488,3 +1613,97 @@ func (d *Daemon) handleEC2DescribeImages(msg *nats.Msg) { slog.Info("handleEC2DescribeImages completed", "count", len(output.Images)) } + +// handleEC2DescribeInstances processes incoming EC2 DescribeInstances requests +// This handler responds with all instances running on this node +func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) { + log.Printf("Received message on subject: %s", msg.Subject) + log.Printf("Message data: %s", string(msg.Data)) + + // Initialize describeInstancesInput before unmarshaling into it + describeInstancesInput := &ec2.DescribeInstancesInput{} + var errResp []byte + + errResp = utils.UnmarshalJsonPayload(describeInstancesInput, msg.Data) + + if errResp != nil { + msg.Respond(errResp) + slog.Error("Request does not match DescribeInstancesInput") + return + } + + slog.Info("Processing DescribeInstances request from this node") + + // Build response with reservations from instances on this node + var reservations []*ec2.Reservation + + d.Instances.Mu.Lock() + defer d.Instances.Mu.Unlock() + + // Filter instances if specific instance IDs were requested + instanceIDFilter := make(map[string]bool) + if describeInstancesInput.InstanceIds != nil && len(describeInstancesInput.InstanceIds) > 0 { + for _, id := range describeInstancesInput.InstanceIds { + if id != nil { + instanceIDFilter[*id] = true + } + } + } + + // Iterate through all instances on this node + for _, instance := range d.Instances.VMS { + // Skip if filtering by instance IDs and this instance is not in the filter + if len(instanceIDFilter) > 0 && !instanceIDFilter[instance.ID] { + continue + } + + // Use stored reservation metadata if available + if instance.Reservation != nil && instance.Instance != nil { + // Create a copy of the reservation with updated instance state + reservation := *instance.Reservation + + // Update the instance state to current state + instanceCopy := *instance.Instance + instanceCopy.State = &ec2.InstanceState{} + + // Map internal status to EC2 state codes + switch instance.Status { + case "pending", "provisioning": + instanceCopy.State.SetCode(0) + instanceCopy.State.SetName("pending") + case "running": + instanceCopy.State.SetCode(16) + instanceCopy.State.SetName("running") + case "stopped": + instanceCopy.State.SetCode(80) + instanceCopy.State.SetName("stopped") + case "terminated": + instanceCopy.State.SetCode(48) + instanceCopy.State.SetName("terminated") + default: + instanceCopy.State.SetCode(0) + instanceCopy.State.SetName("pending") + } + + reservation.Instances = []*ec2.Instance{&instanceCopy} + reservations = append(reservations, &reservation) + } + } + + // Create the response + output := &ec2.DescribeInstancesOutput{ + Reservations: reservations, + } + + // Respond to NATS with DescribeInstancesOutput + jsonResponse, err := json.Marshal(output) + if err != nil { + slog.Error("handleEC2DescribeInstances failed to marshal output", "err", err) + errResp = utils.GenerateErrorPayload(awserrors.ErrorServerInternal) + msg.Respond(errResp) + return + } + msg.Respond(jsonResponse) + + slog.Info("handleEC2DescribeInstances completed", "count", len(reservations)) +} diff --git a/hive/gateway/ec2.go b/hive/gateway/ec2.go index 8231c9b38..f8f398780 100644 --- a/hive/gateway/ec2.go +++ b/hive/gateway/ec2.go @@ -23,8 +23,27 @@ func (gw *GatewayConfig) EC2_Request(ctx *fiber.Ctx) error { // TODO: Generate for each action, unit test each, and invalid action switch queryArgs["Action"] { - //case "DescribeInstances": - // xmlOutput, err = gateway_ec2_instance.DescribeInstances(ctx, queryArgs) + case "DescribeInstances": + var input = &ec2.DescribeInstancesInput{} + err = awsec2query.QueryParamsToStruct(queryArgs, input) + + if err != nil { + return err + } + + output, err := gateway_ec2_instance.DescribeInstances(input, gw.NATSConn, gw.ExpectedNodes) + + if err != nil { + return err + } + + // Convert to XML + payload := utils.GenerateXMLPayload("DescribeInstancesResponse", output) + xmlOutput, err = utils.MarshalToXML(payload) + + if err != nil { + return errors.New("failed to marshal response to XML") + } case "RunInstances": var input = &ec2.RunInstancesInput{} @@ -48,6 +67,75 @@ func (gw *GatewayConfig) EC2_Request(ctx *fiber.Ctx) error { return errors.New("failed to marshal response to XML") } + case "StartInstances": + + var input = &ec2.StartInstancesInput{} + err = awsec2query.QueryParamsToStruct(queryArgs, input) + + if err != nil { + return err + } + + output, err := gateway_ec2_instance.StartInstances(input, gw.NATSConn) + + if err != nil { + return err + } + + // Convert to XML + payload := utils.GenerateXMLPayload("StartInstancesResponse", output) + xmlOutput, err = utils.MarshalToXML(payload) + + if err != nil { + return errors.New("failed to marshal response to XML") + } + + case "StopInstances": + + var input = &ec2.StopInstancesInput{} + err = awsec2query.QueryParamsToStruct(queryArgs, input) + + if err != nil { + return err + } + + output, err := gateway_ec2_instance.StopInstances(input, gw.NATSConn) + + if err != nil { + return err + } + + // Convert to XML + payload := utils.GenerateXMLPayload("StopInstancesResponse", output) + xmlOutput, err = utils.MarshalToXML(payload) + + if err != nil { + return errors.New("failed to marshal response to XML") + } + + case "TerminateInstances": + + var input = &ec2.TerminateInstancesInput{} + err = awsec2query.QueryParamsToStruct(queryArgs, input) + + if err != nil { + return err + } + + output, err := gateway_ec2_instance.TerminateInstances(input, gw.NATSConn) + + if err != nil { + return err + } + + // Convert to XML + payload := utils.GenerateXMLPayload("TerminateInstancesResponse", output) + xmlOutput, err = utils.MarshalToXML(payload) + + if err != nil { + return errors.New("failed to marshal response to XML") + } + case "CreateKeyPair": var input = &ec2.CreateKeyPairInput{} diff --git a/hive/gateway/ec2/instance/DescribeInstances.go b/hive/gateway/ec2/instance/DescribeInstances.go new file mode 100644 index 000000000..a48a658eb --- /dev/null +++ b/hive/gateway/ec2/instance/DescribeInstances.go @@ -0,0 +1,111 @@ +package gateway_ec2_instance + +import ( + "encoding/json" + "fmt" + "log/slog" + "time" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/hive/hive/utils" + "github.com/nats-io/nats.go" +) + +// DescribeInstances queries all hive nodes for their instances via NATS +// and aggregates the results into a single response +func DescribeInstances(input *ec2.DescribeInstancesInput, natsConn *nats.Conn, expectedNodes int) (*ec2.DescribeInstancesOutput, error) { + // Marshal input to JSON + jsonData, err := json.Marshal(input) + if err != nil { + slog.Error("DescribeInstances: Failed to marshal input", "err", err) + return nil, fmt.Errorf("failed to marshal input: %w", err) + } + + // Create an inbox for collecting responses from all nodes + inbox := nats.NewInbox() + sub, err := natsConn.SubscribeSync(inbox) + if err != nil { + slog.Error("DescribeInstances: Failed to create inbox subscription", "err", err) + return nil, fmt.Errorf("failed to create inbox: %w", err) + } + defer sub.Unsubscribe() + + // Publish request to all nodes (no queue group, so all daemons receive it) + err = natsConn.PublishRequest("ec2.DescribeInstances", inbox, jsonData) + if err != nil { + slog.Error("DescribeInstances: Failed to publish request", "err", err) + return nil, fmt.Errorf("failed to publish request: %w", err) + } + + // Collect responses from all nodes + // Timeout serves as a safety mechanism in case some nodes don't respond + timeout := 3 * time.Second + deadline := time.Now().Add(timeout) + + var allReservations []*ec2.Reservation + responsesReceived := 0 + + // If expectedNodes is not configured (0), fall back to timeout-based collection + if expectedNodes <= 0 { + expectedNodes = -1 // Disable early exit + slog.Warn("DescribeInstances: ExpectedNodes not configured, using timeout-only collection") + } + + for time.Now().Before(deadline) { + // Check if we've received responses from all expected nodes + if expectedNodes > 0 && responsesReceived >= expectedNodes { + slog.Info("DescribeInstances: Received responses from all expected nodes", "expected", expectedNodes, "received", responsesReceived) + break + } + + // Calculate remaining timeout + remaining := time.Until(deadline) + if remaining <= 0 { + break + } + + // Wait for next message with remaining timeout + msg, err := sub.NextMsg(remaining) + if err != nil { + if err == nats.ErrTimeout { + // Timeout reached, no more messages + break + } + slog.Error("DescribeInstances: Error receiving message", "err", err) + break + } + + // Increment response counter (even for errors, as we heard from the node) + responsesReceived++ + + // Check if response is an error + responseError, err := utils.ValidateErrorPayload(msg.Data) + if err != nil { + // Response is an error payload - log but continue collecting from other nodes + slog.Warn("DescribeInstances: Received error from node", "code", responseError.Code, "responses_received", responsesReceived) + continue + } + + // Parse the DescribeInstancesOutput from this node + var nodeOutput ec2.DescribeInstancesOutput + err = json.Unmarshal(msg.Data, &nodeOutput) + if err != nil { + slog.Error("DescribeInstances: Failed to unmarshal node response", "err", err) + continue + } + + // Aggregate reservations from this node + if nodeOutput.Reservations != nil { + allReservations = append(allReservations, nodeOutput.Reservations...) + slog.Info("DescribeInstances: Collected reservations from node", "count", len(nodeOutput.Reservations), "responses_received", responsesReceived) + } + } + + // Build final aggregated response + output := &ec2.DescribeInstancesOutput{ + Reservations: allReservations, + } + + slog.Info("DescribeInstances: Aggregated response", "total_reservations", len(allReservations)) + return output, nil +} diff --git a/hive/gateway/ec2/instance/StartInstances.go b/hive/gateway/ec2/instance/StartInstances.go new file mode 100644 index 000000000..714994040 --- /dev/null +++ b/hive/gateway/ec2/instance/StartInstances.go @@ -0,0 +1,106 @@ +package gateway_ec2_instance + +import ( + "encoding/json" + "fmt" + "log/slog" + "time" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/hive/hive/qmp" + "github.com/nats-io/nats.go" +) + +// StartInstances sends start commands to specified instances via NATS +func StartInstances(input *ec2.StartInstancesInput, natsConn *nats.Conn) (*ec2.StartInstancesOutput, error) { + if input.InstanceIds == nil || len(input.InstanceIds) == 0 { + return nil, fmt.Errorf("no instance IDs provided") + } + + slog.Info("StartInstances: Processing request", "instance_count", len(input.InstanceIds)) + + var stateChanges []*ec2.InstanceStateChange + + // Process each instance + for _, instanceIDPtr := range input.InstanceIds { + if instanceIDPtr == nil { + continue + } + instanceID := *instanceIDPtr + + // Build the QMP command to resume the instance + command := qmp.Command{ + ID: instanceID, + QMPCommand: qmp.QMPCommand{ + Execute: "cont", + Arguments: map[string]interface{}{}, + }, + Attributes: qmp.Attributes{ + StopInstance: false, + TerminateInstance: false, + StartInstance: true, + }, + } + + // Marshal the command + jsonData, err := json.Marshal(command) + if err != nil { + slog.Error("StartInstances: Failed to marshal command", "instance_id", instanceID, "err", err) + continue + } + + // Send NATS request to the specific instance topic + subject := fmt.Sprintf("ec2.cmd.%s", instanceID) + + // TODO: Improve, stop/termination may take significant time + msg, err := natsConn.Request(subject, jsonData, 30*time.Second) + if err != nil { + slog.Error("StartInstances: Failed to send command", "instance_id", instanceID, "err", err) + // Add failed state change + stateChange := &ec2.InstanceStateChange{ + InstanceId: &instanceID, + CurrentState: &ec2.InstanceState{ + Code: new(int64), + Name: new(string), + }, + PreviousState: &ec2.InstanceState{ + Code: new(int64), + Name: new(string), + }, + } + *stateChange.CurrentState.Code = 80 // stopped + *stateChange.CurrentState.Name = "stopped" + *stateChange.PreviousState.Code = 80 + *stateChange.PreviousState.Name = "stopped" + stateChanges = append(stateChanges, stateChange) + continue + } + + slog.Info("StartInstances: Command sent successfully", "instance_id", instanceID, "response", string(msg.Data)) + + // Build state change response (stopped -> pending) + stateChange := &ec2.InstanceStateChange{ + InstanceId: &instanceID, + CurrentState: &ec2.InstanceState{ + Code: new(int64), + Name: new(string), + }, + PreviousState: &ec2.InstanceState{ + Code: new(int64), + Name: new(string), + }, + } + *stateChange.CurrentState.Code = 0 // pending + *stateChange.CurrentState.Name = "pending" + *stateChange.PreviousState.Code = 80 // stopped + *stateChange.PreviousState.Name = "stopped" + stateChanges = append(stateChanges, stateChange) + } + + output := &ec2.StartInstancesOutput{ + StartingInstances: stateChanges, + } + + slog.Info("StartInstances: Completed", "total_instances", len(stateChanges)) + return output, nil +} diff --git a/hive/gateway/ec2/instance/StopInstances.go b/hive/gateway/ec2/instance/StopInstances.go new file mode 100644 index 000000000..496f079df --- /dev/null +++ b/hive/gateway/ec2/instance/StopInstances.go @@ -0,0 +1,105 @@ +package gateway_ec2_instance + +import ( + "encoding/json" + "fmt" + "log/slog" + "time" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/hive/hive/qmp" + "github.com/nats-io/nats.go" +) + +// StopInstances sends stop commands to specified instances via NATS +// Uses system_powerdown without stop_instance attribute to allow restart +func StopInstances(input *ec2.StopInstancesInput, natsConn *nats.Conn) (*ec2.StopInstancesOutput, error) { + if input.InstanceIds == nil || len(input.InstanceIds) == 0 { + return nil, fmt.Errorf("no instance IDs provided") + } + + slog.Info("StopInstances: Processing request", "instance_count", len(input.InstanceIds)) + + var stateChanges []*ec2.InstanceStateChange + + // Process each instance + for _, instanceIDPtr := range input.InstanceIds { + if instanceIDPtr == nil { + continue + } + instanceID := *instanceIDPtr + + // Build the QMP command to stop the instance + // Note: system_powerdown without stop_instance allows restart + command := qmp.Command{ + ID: instanceID, + QMPCommand: qmp.QMPCommand{ + Execute: "system_powerdown", + Arguments: map[string]interface{}{}, + }, + Attributes: qmp.Attributes{ + StopInstance: false, // Allow restart on daemon/node restart + TerminateInstance: false, + }, + } + + // Marshal the command + jsonData, err := json.Marshal(command) + if err != nil { + slog.Error("StopInstances: Failed to marshal command", "instance_id", instanceID, "err", err) + continue + } + + // Send NATS request to the specific instance topic + subject := fmt.Sprintf("ec2.cmd.%s", instanceID) + msg, err := natsConn.Request(subject, jsonData, 5*time.Second) + if err != nil { + slog.Error("StopInstances: Failed to send command", "instance_id", instanceID, "err", err) + // Add failed state change + stateChange := &ec2.InstanceStateChange{ + InstanceId: &instanceID, + CurrentState: &ec2.InstanceState{ + Code: new(int64), + Name: new(string), + }, + PreviousState: &ec2.InstanceState{ + Code: new(int64), + Name: new(string), + }, + } + *stateChange.CurrentState.Code = 16 // running + *stateChange.CurrentState.Name = "running" + *stateChange.PreviousState.Code = 16 + *stateChange.PreviousState.Name = "running" + stateChanges = append(stateChanges, stateChange) + continue + } + + slog.Info("StopInstances: Command sent successfully", "instance_id", instanceID, "response", string(msg.Data)) + + // Build state change response (running -> stopping) + stateChange := &ec2.InstanceStateChange{ + InstanceId: &instanceID, + CurrentState: &ec2.InstanceState{ + Code: new(int64), + Name: new(string), + }, + PreviousState: &ec2.InstanceState{ + Code: new(int64), + Name: new(string), + }, + } + *stateChange.CurrentState.Code = 64 // stopping + *stateChange.CurrentState.Name = "stopping" + *stateChange.PreviousState.Code = 16 // running + *stateChange.PreviousState.Name = "running" + stateChanges = append(stateChanges, stateChange) + } + + output := &ec2.StopInstancesOutput{ + StoppingInstances: stateChanges, + } + + slog.Info("StopInstances: Completed", "total_instances", len(stateChanges)) + return output, nil +} diff --git a/hive/gateway/ec2/instance/TerminateInstances.go b/hive/gateway/ec2/instance/TerminateInstances.go new file mode 100644 index 000000000..fd8595208 --- /dev/null +++ b/hive/gateway/ec2/instance/TerminateInstances.go @@ -0,0 +1,105 @@ +package gateway_ec2_instance + +import ( + "encoding/json" + "fmt" + "log/slog" + "time" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/hive/hive/qmp" + "github.com/nats-io/nats.go" +) + +// TerminateInstances sends terminate commands to specified instances via NATS +// Uses system_powerdown with stop_instance attribute to prevent restart +func TerminateInstances(input *ec2.TerminateInstancesInput, natsConn *nats.Conn) (*ec2.TerminateInstancesOutput, error) { + if input.InstanceIds == nil || len(input.InstanceIds) == 0 { + return nil, fmt.Errorf("no instance IDs provided") + } + + slog.Info("TerminateInstances: Processing request", "instance_count", len(input.InstanceIds)) + + var stateChanges []*ec2.InstanceStateChange + + // Process each instance + for _, instanceIDPtr := range input.InstanceIds { + if instanceIDPtr == nil { + continue + } + instanceID := *instanceIDPtr + + // Build the QMP command to terminate the instance + // Note: stop_instance=true prevents restart on daemon/node restart + command := qmp.Command{ + ID: instanceID, + QMPCommand: qmp.QMPCommand{ + Execute: "system_powerdown", + Arguments: map[string]interface{}{}, + }, + Attributes: qmp.Attributes{ + StopInstance: true, // Prevent restart on daemon/node restart + TerminateInstance: true, + }, + } + + // Marshal the command + jsonData, err := json.Marshal(command) + if err != nil { + slog.Error("TerminateInstances: Failed to marshal command", "instance_id", instanceID, "err", err) + continue + } + + // Send NATS request to the specific instance topic + subject := fmt.Sprintf("ec2.cmd.%s", instanceID) + msg, err := natsConn.Request(subject, jsonData, 5*time.Second) + if err != nil { + slog.Error("TerminateInstances: Failed to send command", "instance_id", instanceID, "err", err) + // Add failed state change + stateChange := &ec2.InstanceStateChange{ + InstanceId: &instanceID, + CurrentState: &ec2.InstanceState{ + Code: new(int64), + Name: new(string), + }, + PreviousState: &ec2.InstanceState{ + Code: new(int64), + Name: new(string), + }, + } + *stateChange.CurrentState.Code = 16 // running + *stateChange.CurrentState.Name = "running" + *stateChange.PreviousState.Code = 16 + *stateChange.PreviousState.Name = "running" + stateChanges = append(stateChanges, stateChange) + continue + } + + slog.Info("TerminateInstances: Command sent successfully", "instance_id", instanceID, "response", string(msg.Data)) + + // Build state change response (running -> shutting-down) + stateChange := &ec2.InstanceStateChange{ + InstanceId: &instanceID, + CurrentState: &ec2.InstanceState{ + Code: new(int64), + Name: new(string), + }, + PreviousState: &ec2.InstanceState{ + Code: new(int64), + Name: new(string), + }, + } + *stateChange.CurrentState.Code = 32 // shutting-down + *stateChange.CurrentState.Name = "shutting-down" + *stateChange.PreviousState.Code = 16 // running + *stateChange.PreviousState.Name = "running" + stateChanges = append(stateChanges, stateChange) + } + + output := &ec2.TerminateInstancesOutput{ + TerminatingInstances: stateChanges, + } + + slog.Info("TerminateInstances: Completed", "total_instances", len(stateChanges)) + return output, nil +} diff --git a/hive/gateway/gateway.go b/hive/gateway/gateway.go index c339beb8d..6dcb30fbc 100644 --- a/hive/gateway/gateway.go +++ b/hive/gateway/gateway.go @@ -21,6 +21,7 @@ type GatewayConfig struct { DisableLogging bool `json:"disable_logging"` NATSConn *nats.Conn // Shared NATS connection for service communication Config string // Shared AWS Gateway config for S3 auth + ExpectedNodes int // Number of expected hive nodes for multi-node operations } var supportedServices = map[string]bool{ diff --git a/hive/handlers/ec2/instance/service_impl.go b/hive/handlers/ec2/instance/service_impl.go index b35155ac9..b4d5e11b5 100644 --- a/hive/handlers/ec2/instance/service_impl.go +++ b/hive/handlers/ec2/instance/service_impl.go @@ -132,15 +132,29 @@ func (s *InstanceServiceImpl) RunInstances(input *ec2.RunInstancesInput) (*vm.VM // Create reservation response reservation.SetReservationId(vm.GenerateEC2ReservationID()) + reservation.SetOwnerId("123456789012") // TODO: Use actual owner ID from config // TODO: Loop through multiple instance creation based on MinCount / MaxCount reservation.Instances = make([]*ec2.Instance, 1) - reservation.Instances[0] = &ec2.Instance{ + ec2Instance := &ec2.Instance{ State: &ec2.InstanceState{}, } - reservation.Instances[0].SetInstanceId(instance.ID) - reservation.Instances[0].State.SetCode(0) - reservation.Instances[0].State.SetName("pending") + ec2Instance.SetInstanceId(instance.ID) + ec2Instance.SetInstanceType(*input.InstanceType) + ec2Instance.SetImageId(*input.ImageId) + if input.KeyName != nil { + ec2Instance.SetKeyName(*input.KeyName) + } + ec2Instance.SetLaunchTime(time.Now()) + ec2Instance.State.SetCode(0) + ec2Instance.State.SetName("pending") + + reservation.Instances[0] = ec2Instance + + // Store EC2 API metadata in VM for DescribeInstances compatibility + instance.RunInstancesInput = input + instance.Reservation = &reservation + instance.Instance = ec2Instance // Handle block device mappings if input.BlockDeviceMappings != nil && len(input.BlockDeviceMappings) > 0 { diff --git a/hive/qmp/qmp.go b/hive/qmp/qmp.go index 50b2856e8..c9daa472a 100644 --- a/hive/qmp/qmp.go +++ b/hive/qmp/qmp.go @@ -27,8 +27,9 @@ type Command struct { } type Attributes struct { - StopInstance bool `json:"stop_instance"` - DeleteInstance bool `json:"delete_instance"` + StopInstance bool `json:"stop_instance"` + TerminateInstance bool `json:"delete_instance"` + StartInstance bool `json:"start_instance"` } // (QEMU) stop / cont (resume) / system_powerdown / system_reset / system_wakeup diff --git a/hive/services/viperblockd/viperblockd.go b/hive/services/viperblockd/viperblockd.go index 2eac9e493..5172335d2 100644 --- a/hive/services/viperblockd/viperblockd.go +++ b/hive/services/viperblockd/viperblockd.go @@ -117,32 +117,8 @@ func launchService(cfg *Config) (err error) { return } - /* - - s3cfg := s3.S3Config{ - VolumeName: ebsRequest.Volume, - Bucket: cfg.Bucket, - Region: cfg.Region, - AccessKey: cfg.AccessKey, - SecretKey: cfg.SecretKey, - Host: cfg.S3Host, - } - - vbconfig := viperblock.VB{ - VolumeName: ebsRequest.Volume, - VolumeSize: 1, // Workaround, calculated on LoadState() - BaseDir: cfg.BaseDir, - Cache: viperblock.Cache{ - Config: viperblock.CacheConfig{ - Size: 0, - }, - }, - VolumeConfig: viperblock.VolumeConfig{}, - } - - vb, err := viperblock.New(vbconfig, "s3", s3cfg) - - */ + // TODO: Improve + nc.Publish("ebs.delete.response", []byte(fmt.Sprintf(`{"volume":"%s","deleted":true}`, ebsRequest.Volume))) }) @@ -226,6 +202,7 @@ func launchService(cfg *Config) (err error) { BaseDir: cfg.BaseDir, Cache: viperblock.Cache{ Config: viperblock.CacheConfig{ + // TODO: Improve, based on system memory Size: 0, }, }, @@ -234,6 +211,13 @@ func launchService(cfg *Config) (err error) { vb, err := viperblock.New(vbconfig, "s3", s3cfg) + // TODO: Improve + // 5% of system memory for cache for master volumes + if !strings.HasSuffix("-cloudinit", ebsRequest.Name) && !strings.HasSuffix("-efi", ebsRequest.Name) { + slog.Info("Setting cache size to 5% of system memory for volume", "volume", ebsRequest.Name) + vb.SetCacheSystemMemory(5) + } + if err != nil { ebsResponse.Error = fmt.Sprintf("Failed to connect to Viperblock store: %v", err) // Marshal and send error response immediately @@ -431,44 +415,6 @@ func launchService(cfg *Config) (err error) { } cfg.mu.Unlock() - /* - opts := &server.Options{} - - // Initialize new server with options - ns, err := server.NewServer(opts) - - if err != nil { - panic(err) - } - - go func() { - ns.Start() - }() - - // Wait for server to be ready for connections - if !ns.ReadyForConnections(4 * time.Second) { - panic("not ready for connection") - } - - // Create a channel to receive shutdown signals - sigChan := make(chan os.Signal, 1) - signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) - - // Wait for shutdown signal - <-sigChan - log.Println("Shutting down gracefully...") - - // Graceful shutdown with timeout - //ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - //defer cancel() - - // Shutdown the server, use a context to timeout after 4 secs - ns.Shutdown() - - log.Println("Server stopped") - - */ - return nil } diff --git a/hive/utils/utils.go b/hive/utils/utils.go index 7c8b3d9e1..85e63fd0b 100644 --- a/hive/utils/utils.go +++ b/hive/utils/utils.go @@ -34,6 +34,8 @@ import ( // Helper functions for OS images +var ErrQCOWDetected = errors.New("qcow format detected") + type Images struct { Name string `json:"name"` Description string `json:"description"` @@ -62,6 +64,7 @@ var AvailableImages = map[string]Images{ URL: "https://cdimage.debian.org/cdimage/cloud/bookworm/latest/debian-12-generic-amd64.tar.xz", Checksum: "https://cdimage.debian.org/cdimage/cloud/bookworm/latest/SHA512SUMS", ChecksumType: "sha512", + BootMode: "bios", Starred: true, }, @@ -76,7 +79,7 @@ var AvailableImages = map[string]Images{ URL: "https://cdimage.debian.org/cdimage/cloud/bookworm/latest/debian-12-generic-arm64.tar.xz", Checksum: "https://cdimage.debian.org/cdimage/cloud/bookworm/latest/SHA512SUMS", ChecksumType: "sha512", - BootMode: "uefi", + BootMode: "bios", Starred: true, }, @@ -107,10 +110,10 @@ var AvailableImages = map[string]Images{ Arch: "x86_64", Platform: "Linux/UNIX", CreatedAt: time.Date(2025, 10, 6, 0, 0, 0, 0, time.UTC), - URL: "https://cloud-images.ubuntu.com/noble/current/noble-server-cloudimg-amd64.tar.gz", + URL: "https://cloud-images.ubuntu.com/noble/current/noble-server-cloudimg-amd64.img", Checksum: "https://cloud-images.ubuntu.com/noble/current/SHA256SUMS", ChecksumType: "sha256", - BootMode: "uefi", + BootMode: "bios", Starred: false, }, @@ -123,10 +126,10 @@ var AvailableImages = map[string]Images{ Arch: "arm64", Platform: "Linux/UNIX", CreatedAt: time.Date(2025, 10, 6, 0, 0, 0, 0, time.UTC), - URL: "https://cloud-images.ubuntu.com/noble/current/noble-server-cloudimg-arm64.tar.gz", + URL: "https://cloud-images.ubuntu.com/noble/current/noble-server-cloudimg-arm64.img", Checksum: "https://cloud-images.ubuntu.com/noble/current/SHA256SUMS", ChecksumType: "sha256", - BootMode: "uefi", + BootMode: "bios", Starred: false, }, @@ -140,29 +143,32 @@ var AvailableImages = map[string]Images{ Arch: "x86_64", Platform: "Linux/UNIX", CreatedAt: time.Date(2025, 10, 6, 0, 0, 0, 0, time.UTC), - URL: "https://dl-cdn.alpinelinux.org/alpine/v3.22/releases/cloud/gcp_alpine-3.22.2-x86_64-uefi-cloudinit-metal-r0.raw.tar.gz", - Checksum: "https://dl-cdn.alpinelinux.org/alpine/v3.22/releases/cloud/gcp_alpine-3.22.2-x86_64-uefi-cloudinit-metal-r0.raw.tar.gz.sha512", + URL: "https://dl-cdn.alpinelinux.org/alpine/v3.22/releases/cloud/generic_alpine-3.22.2-x86_64-bios-cloudinit-r0.qcow2", + Checksum: "https://dl-cdn.alpinelinux.org/alpine/v3.22/releases/cloud/generic_alpine-3.22.2-x86_64-bios-cloudinit-r0.qcow2.sha512", ChecksumType: "sha512", - BootMode: "uefi", + BootMode: "bios", Starred: false, }, - "alpine-3.22.2-arm64": - // Alpine Linux (cloud init) arm64 - { - Name: "alpine-3.22.2-arm64", - Description: "Alpine Linux 3.22.2 arm64 cloud image", - Distro: "alpine", - Version: "3.22.2", - Arch: "arm64", - Platform: "Linux/UNIX", - CreatedAt: time.Date(2025, 10, 6, 0, 0, 0, 0, time.UTC), - URL: "https://dl-cdn.alpinelinux.org/alpine/v3.22/releases/cloud/gcp_alpine-3.22.2-aarch64-uefi-cloudinit-metal-r0.raw.tar.gz", - Checksum: "https://dl-cdn.alpinelinux.org/alpine/v3.22/releases/cloud/gcp_alpine-3.22.2-aarch64-uefi-cloudinit-metal-r0.raw.tar.gz.sha512", - ChecksumType: "sha512", - BootMode: "uefi", - Starred: false, - }, + /* + "alpine-3.22.2-arm64": + // Alpine Linux (cloud init) arm64 (Requires UEFI boot, TODO) + { + Name: "alpine-3.22.2-arm64", + Description: "Alpine Linux 3.22.2 arm64 cloud image", + Distro: "alpine", + Version: "3.22.2", + Arch: "arm64", + Platform: "Linux/UNIX", + CreatedAt: time.Date(2025, 10, 6, 0, 0, 0, 0, time.UTC), + URL: "https://dl-cdn.alpinelinux.org/alpine/v3.22/releases/cloud/gcp_alpine-3.22.2-aarch64-uefi-cloudinit-metal-r0.raw.tar.gz", + Checksum: "https://dl-cdn.alpinelinux.org/alpine/v3.22/releases/cloud/gcp_alpine-3.22.2-aarch64-uefi-cloudinit-metal-r0.raw.tar.gz.sha512", + ChecksumType: "sha512", + BootMode: "uefi", + Starred: false, + }, + */ + } func ReadPidFile(name string) (int, error) { @@ -307,8 +313,8 @@ func KillProcess(pid int) error { checks++ - // If process is still running after 3 checks, force kill - if checks > 3 { + // If process is still running after 60 seconds, force kill + if checks > 60 { err = process.Kill() // SIGKILL if err != nil { @@ -511,7 +517,7 @@ func ExtractDiskImageFromFile(imagepath string, tmpdir string) (diskimage string //} // Already in raw/image formt, confirm the file contains a valid disk image/MBR - if strings.HasSuffix(imagefile, ".raw") || strings.HasSuffix(imagefile, ".img") { + if strings.HasSuffix(imagefile, ".raw") || strings.HasSuffix(imagefile, ".img") || strings.HasSuffix(imagefile, ".qcow2") || strings.HasSuffix(imagefile, ".qcow") { path, err := filepath.Abs(imagepath) @@ -522,6 +528,41 @@ func ExtractDiskImageFromFile(imagepath string, tmpdir string) (diskimage string // Validate the specified filename is indeed a disk image / MBR err = validateDiskImagePath(path) + // Check error response + + if errors.Is(err, ErrQCOWDetected) { + + //fmt.Println("Extracting raw disk image from qcow2 file", "file", imagepath) + + extractpath := fmt.Sprintf("%s/%s", tmpdir, imagefile) + extractpath = strings.TrimSuffix(extractpath, ".qcow2") + ".raw" + + args = []string{ + "convert", + "-f", + "qcow2", + "-O", + "raw", + imagepath, + "-C", + extractpath, + } + + execCmd = "qemu-img" + + //fmt.Println("Executing command:", "cmd", execCmd, "args", args) + + cmd := exec.Command(execCmd, args...) + _, err = cmd.Output() + + if err != nil { + return path, err + } + + return extractpath, nil + + } + return path, err } else if strings.HasSuffix(imagefile, ".tar.xz") { @@ -622,6 +663,8 @@ func validateDiskImagePath(diskimage string) (err error) { if strings.Contains(filetype[1], "DOS/MBR boot sector") || strings.Contains(filetype[1], "Linux ") { return nil + } else if strings.Contains(filetype[1], "QEMU QCOW") { + return ErrQCOWDetected } } diff --git a/hive/vm/vm.go b/hive/vm/vm.go index 65464fe28..2ba6656d9 100644 --- a/hive/vm/vm.go +++ b/hive/vm/vm.go @@ -9,6 +9,7 @@ import ( "strings" "sync" + "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/hive/hive/config" "github.com/mulgadc/hive/hive/qmp" ) @@ -44,6 +45,14 @@ type VM struct { // User attributes (user initiated stop/delete) Attributes qmp.Attributes `json:"attributes"` + + // EC2 API metadata - stored for AWS API compatibility + // RunInstancesInput contains the original request parameters (ImageId, KeyName, UserData, etc.) + RunInstancesInput *ec2.RunInstancesInput `json:"run_instances_input,omitempty"` + // Reservation contains the reservation metadata (ReservationId, OwnerId, etc.) + Reservation *ec2.Reservation `json:"reservation,omitempty"` + // Instance contains the current instance state and metadata + Instance *ec2.Instance `json:"instance,omitempty"` } type Instances struct {