From 774655eff094454458ff560ace51fb41b9c94427 Mon Sep 17 00:00:00 2001 From: Ben Duncan <5178243+benduncan@users.noreply.github.com> Date: Sat, 14 Feb 2026 15:38:03 +1000 Subject: [PATCH] hive top, hive get cmds for cluster health and stats. improved docs for iac, cluster, multi-server restart --- cmd/hive/cmd/get.go | 274 +++++++++++++++++ cmd/hive/cmd/top.go | 154 ++++++++++ .../improvements/hive-cluster-stats.md | 117 ++++++++ .../improvements/iac-testing-aws-terraform.md | 215 ++++++++++++++ .../improvements/multi-server-restart.md | 277 +++++++++++++++++- hive/config/config.go | 42 +++ hive/daemon/daemon.go | 41 +++ hive/daemon/daemon_handlers.go | 89 ++++++ tests/e2e/run-e2e.sh | 43 +++ tests/e2e/run-multinode-e2e.sh | 42 +++ 10 files changed, 1281 insertions(+), 13 deletions(-) create mode 100644 cmd/hive/cmd/get.go create mode 100644 cmd/hive/cmd/top.go create mode 100644 docs/development/improvements/hive-cluster-stats.md create mode 100644 docs/development/improvements/iac-testing-aws-terraform.md diff --git a/cmd/hive/cmd/get.go b/cmd/hive/cmd/get.go new file mode 100644 index 000000000..b3e959929 --- /dev/null +++ b/cmd/hive/cmd/get.go @@ -0,0 +1,274 @@ +/* +Copyright © 2025 Mulga Defense Corporation + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ +package cmd + +import ( + "encoding/json" + "fmt" + "os" + "sort" + "strings" + "time" + + "github.com/mulgadc/hive/hive/config" + "github.com/mulgadc/hive/hive/utils" + "github.com/nats-io/nats.go" + "github.com/pterm/pterm" + "github.com/spf13/cobra" + "github.com/spf13/viper" +) + +var getCmd = &cobra.Command{ + Use: "get", + Short: "Display cluster resources", + Long: `Display cluster resources such as nodes and VMs.`, +} + +var getNodesCmd = &cobra.Command{ + Use: "nodes", + Short: "Display cluster nodes", + Long: `Display all physical nodes in the cluster with status, IP, region, and services.`, + Run: runGetNodes, +} + +var getVMsCmd = &cobra.Command{ + Use: "vms", + Aliases: []string{"instances"}, + Short: "Display VMs across the cluster", + Long: `Display all VMs running across the cluster with instance type, resources, and placement.`, + Run: runGetVMs, +} + +func init() { + rootCmd.AddCommand(getCmd) + getCmd.AddCommand(getNodesCmd) + getCmd.AddCommand(getVMsCmd) + + getCmd.PersistentFlags().Duration("timeout", 3*time.Second, "Timeout for collecting responses from nodes") +} + +// loadConfigAndConnect loads the cluster config and connects to NATS. +func loadConfigAndConnect() (*config.ClusterConfig, *nats.Conn, error) { + cfgPath := viper.GetString("config") + if cfgPath == "" { + homeDir, _ := os.UserHomeDir() + cfgPath = fmt.Sprintf("%s/hive/config/hive.toml", homeDir) + } + + cfg, err := config.LoadConfig(cfgPath) + if err != nil { + return nil, nil, fmt.Errorf("failed to load config: %w", err) + } + + nodeConfig := cfg.Nodes[cfg.Node] + nc, err := utils.ConnectNATS(nodeConfig.NATS.Host, nodeConfig.NATS.ACL.Token) + if err != nil { + return nil, nil, fmt.Errorf("failed to connect to NATS: %w", err) + } + + return cfg, nc, nil +} + +// collectResponses publishes to a fan-out topic and collects all responses within the timeout. +func collectResponses(nc *nats.Conn, topic string, timeout time.Duration) ([][]byte, error) { + inbox := nats.NewInbox() + sub, err := nc.SubscribeSync(inbox) + if err != nil { + return nil, fmt.Errorf("failed to subscribe to inbox: %w", err) + } + defer sub.Unsubscribe() + + if err := nc.PublishRequest(topic, inbox, nil); err != nil { + return nil, fmt.Errorf("failed to publish request: %w", err) + } + nc.Flush() + + var responses [][]byte + deadline := time.Now().Add(timeout) + for { + remaining := time.Until(deadline) + if remaining <= 0 { + break + } + msg, err := sub.NextMsg(remaining) + if err != nil { + break // timeout or other error — done collecting + } + responses = append(responses, msg.Data) + } + return responses, nil +} + +func formatUptime(seconds int64) string { + if seconds <= 0 { + return "-" + } + d := time.Duration(seconds) * time.Second + days := int(d.Hours()) / 24 + hours := int(d.Hours()) % 24 + mins := int(d.Minutes()) % 60 + + if days > 0 { + return fmt.Sprintf("%dd%dh", days, hours) + } + if hours > 0 { + return fmt.Sprintf("%dh%dm", hours, mins) + } + return fmt.Sprintf("%dm", mins) +} + +func formatMemGB(gb float64) string { + if gb >= 1.0 { + return fmt.Sprintf("%.1fGi", gb) + } + return fmt.Sprintf("%dMi", int(gb*1024)) +} + +func runGetNodes(cmd *cobra.Command, args []string) { + cfg, nc, err := loadConfigAndConnect() + if err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } + defer nc.Close() + + timeout, _ := cmd.Flags().GetDuration("timeout") + responses, err := collectResponses(nc, "hive.node.status", timeout) + if err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } + + // Parse responses into a map by node name + respondedNodes := make(map[string]config.NodeStatusResponse) + for _, data := range responses { + var resp config.NodeStatusResponse + if err := json.Unmarshal(data, &resp); err != nil { + continue + } + respondedNodes[resp.Node] = resp + } + + // Build table: include all known nodes from config, mark non-responders as NotReady + tableData := pterm.TableData{ + {"NAME", "STATUS", "IP", "REGION", "AZ", "UPTIME", "VMs", "SERVICES"}, + } + + // Collect and sort node names for stable output + nodeNames := make([]string, 0, len(cfg.Nodes)) + for name := range cfg.Nodes { + nodeNames = append(nodeNames, name) + } + sort.Strings(nodeNames) + + for _, name := range nodeNames { + nodeCfg := cfg.Nodes[name] + if resp, ok := respondedNodes[name]; ok { + tableData = append(tableData, []string{ + resp.Node, + resp.Status, + resp.Host, + resp.Region, + resp.AZ, + formatUptime(resp.Uptime), + fmt.Sprintf("%d", resp.VMCount), + strings.Join(resp.Services, ","), + }) + } else { + tableData = append(tableData, []string{ + name, + "NotReady", + nodeCfg.Host, + nodeCfg.Region, + nodeCfg.AZ, + "-", + "-", + strings.Join(nodeCfg.GetServices(), ","), + }) + } + } + + pterm.DefaultTable.WithHasHeader().WithLeftAlignment().WithData(tableData).Render() +} + +func runGetVMs(cmd *cobra.Command, args []string) { + _, nc, err := loadConfigAndConnect() + if err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } + defer nc.Close() + + timeout, _ := cmd.Flags().GetDuration("timeout") + responses, err := collectResponses(nc, "hive.node.vms", timeout) + if err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } + + type vmRow struct { + config.VMInfo + Node string + Host string + } + + var allVMs []vmRow + for _, data := range responses { + var resp config.NodeVMsResponse + if err := json.Unmarshal(data, &resp); err != nil { + continue + } + for _, v := range resp.VMs { + allVMs = append(allVMs, vmRow{VMInfo: v, Node: resp.Node, Host: resp.Host}) + } + } + + if len(allVMs) == 0 { + fmt.Println("No VMs found.") + return + } + + // Sort by node then instance ID + sort.Slice(allVMs, func(i, j int) bool { + if allVMs[i].Node != allVMs[j].Node { + return allVMs[i].Node < allVMs[j].Node + } + return allVMs[i].InstanceID < allVMs[j].InstanceID + }) + + tableData := pterm.TableData{ + {"INSTANCE", "STATUS", "TYPE", "VCPU", "MEM", "NODE", "IP", "AGE"}, + } + + for _, v := range allVMs { + age := "-" + if v.LaunchTime > 0 { + age = formatUptime(time.Now().Unix() - v.LaunchTime) + } + tableData = append(tableData, []string{ + v.InstanceID, + v.Status, + v.InstanceType, + fmt.Sprintf("%d", v.VCPU), + formatMemGB(v.MemoryGB), + v.Node, + v.Host, + age, + }) + } + + pterm.DefaultTable.WithHasHeader().WithLeftAlignment().WithData(tableData).Render() +} diff --git a/cmd/hive/cmd/top.go b/cmd/hive/cmd/top.go new file mode 100644 index 000000000..9aabfc78e --- /dev/null +++ b/cmd/hive/cmd/top.go @@ -0,0 +1,154 @@ +/* +Copyright © 2025 Mulga Defense Corporation + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ +package cmd + +import ( + "encoding/json" + "fmt" + "os" + "sort" + "time" + + "github.com/mulgadc/hive/hive/config" + "github.com/pterm/pterm" + "github.com/spf13/cobra" +) + +var topCmd = &cobra.Command{ + Use: "top", + Short: "Display cluster resource usage", + Long: `Display resource usage (CPU, memory) for cluster nodes.`, +} + +var topNodesCmd = &cobra.Command{ + Use: "nodes", + Short: "Display resource usage per node", + Long: `Display CPU and memory usage per node, plus a summary of available instance types across the cluster.`, + Run: runTopNodes, +} + +func init() { + rootCmd.AddCommand(topCmd) + topCmd.AddCommand(topNodesCmd) + + topCmd.PersistentFlags().Duration("timeout", 3*time.Second, "Timeout for collecting responses from nodes") +} + +func runTopNodes(cmd *cobra.Command, args []string) { + cfg, nc, err := loadConfigAndConnect() + if err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } + defer nc.Close() + + timeout, _ := cmd.Flags().GetDuration("timeout") + responses, err := collectResponses(nc, "hive.node.status", timeout) + if err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } + + respondedNodes := make(map[string]config.NodeStatusResponse) + for _, data := range responses { + var resp config.NodeStatusResponse + if err := json.Unmarshal(data, &resp); err != nil { + continue + } + respondedNodes[resp.Node] = resp + } + + // Node resource table + nodeNames := make([]string, 0, len(cfg.Nodes)) + for name := range cfg.Nodes { + nodeNames = append(nodeNames, name) + } + sort.Strings(nodeNames) + + nodeTable := pterm.TableData{ + {"NAME", "CPU (used/total)", "MEM (used/total)", "VMs"}, + } + + // Aggregate instance type capacity across all nodes + capacityMap := make(map[string]*aggregatedCap) + + for _, name := range nodeNames { + if resp, ok := respondedNodes[name]; ok { + nodeTable = append(nodeTable, []string{ + resp.Node, + fmt.Sprintf("%d/%d", resp.AllocVCPU, resp.TotalVCPU), + fmt.Sprintf("%s/%s", formatMemGB(resp.AllocMemGB), formatMemGB(resp.TotalMemGB)), + fmt.Sprintf("%d", resp.VMCount), + }) + + for _, cap := range resp.InstanceTypes { + if agg, ok := capacityMap[cap.Name]; ok { + agg.Available += cap.Available + } else { + capacityMap[cap.Name] = &aggregatedCap{ + VCPU: cap.VCPU, + MemoryGB: cap.MemoryGB, + Available: cap.Available, + } + } + } + } else { + nodeTable = append(nodeTable, []string{ + name, + "-", + "-", + "-", + }) + } + } + + pterm.DefaultTable.WithHasHeader().WithLeftAlignment().WithData(nodeTable).Render() + + // Instance type capacity summary + if len(capacityMap) == 0 { + return + } + + fmt.Println() + + capNames := make([]string, 0, len(capacityMap)) + for name := range capacityMap { + capNames = append(capNames, name) + } + sort.Strings(capNames) + + capTable := pterm.TableData{ + {"INSTANCE TYPE", "AVAILABLE", "VCPU", "MEMORY"}, + } + + for _, name := range capNames { + agg := capacityMap[name] + capTable = append(capTable, []string{ + name, + fmt.Sprintf("%d", agg.Available), + fmt.Sprintf("%d", agg.VCPU), + formatMemGB(agg.MemoryGB), + }) + } + + pterm.DefaultTable.WithHasHeader().WithLeftAlignment().WithData(capTable).Render() +} + +type aggregatedCap struct { + VCPU int + MemoryGB float64 + Available int +} diff --git a/docs/development/improvements/hive-cluster-stats.md b/docs/development/improvements/hive-cluster-stats.md new file mode 100644 index 000000000..5a486aab9 --- /dev/null +++ b/docs/development/improvements/hive-cluster-stats.md @@ -0,0 +1,117 @@ +# Hive Cluster Stats CLI + +**Status: Complete** + +## Summary + +Added `hive get {nodes,vms}` and `hive top nodes` commands to provide kubectl-style cluster visibility. Operators can now see node health, running VMs, and resource utilization across the cluster from a single command. + +This also satisfies the `hive admin status` command proposed in `multi-server-restart.md` (Section 3.3) — `hive get nodes` and `hive top nodes` together provide the same information with better kubectl-style UX. + +## Commands + +### `hive get nodes` + +Like `kubectl get nodes -o wide`. Shows all physical nodes in the cluster. + +``` +$ hive get nodes +NAME STATUS IP REGION AZ UPTIME VMs SERVICES +node1 Ready 10.11.12.1 ap-southeast-2 ap-southeast-2a 3d4h 2 nats,predastore,viperblock,daemon,awsgw,ui +node2 Ready 10.11.12.2 ap-southeast-2 ap-southeast-2a 3d4h 1 nats,predastore,viperblock,daemon +node3 NotReady 10.11.12.3 ap-southeast-2 ap-southeast-2b - - nats,predastore,viperblock,daemon +``` + +- **Status values**: `Ready` (daemon responded), `NotReady` (no response within timeout) +- Nodes known from config but not responding are shown as `NotReady` +- `--timeout` flag controls response collection window (default 3s) + +### `hive get vms` (alias: `hive get instances`) + +Like `kubectl get pods -A -o wide`. Shows all VMs across the cluster. + +``` +$ hive get vms +INSTANCE STATUS TYPE VCPU MEM NODE IP AGE +i-896fdcfd299a39193 running t3.nano 2 512Mi node1 10.11.12.1 0m +i-2b61dfa9fdb8c2b39 running t3.nano 2 512Mi node2 10.11.12.2 0m +i-b097988f78db6553d running t3.nano 2 512Mi node2 10.11.12.2 0m +``` + +### `hive top nodes` + +Like `kubectl top nodes`. Shows resource usage per node plus aggregated available instance types. + +``` +$ hive top nodes +NAME CPU (used/total) MEM (used/total) VMs +node1 4/16 1.0Gi/30.6Gi 2 +node2 2/16 512Mi/30.6Gi 1 +node3 - - - + +INSTANCE TYPE AVAILABLE VCPU MEMORY +t3.nano 22 2 512Mi +t3.micro 22 2 1.0Gi +t3.small 16 2 2.0Gi +t3.medium 11 2 4.0Gi +t3.large 5 2 8.0Gi +t3.xlarge 2 4 16.0Gi +``` + +Bottom table aggregates free capacity across all Ready nodes. + +## Design + +### NATS Topics + +Two fan-out topics (no queue group — all daemons respond): + +| Topic | Purpose | Response Type | +|-------|---------|---------------| +| `hive.node.status` | Node stats + resource usage | `NodeStatusResponse` | +| `hive.node.vms` | Running VMs on this node | `NodeVMsResponse` | + +### NATS Fan-Out Collection Pattern + +CLI connects to NATS using the config's token, publishes to a fan-out topic with an inbox reply address, and collects all responses within the timeout window: + +```go +inbox := nats.NewInbox() +sub, _ := nc.SubscribeSync(inbox) +nc.PublishRequest("hive.node.status", inbox, nil) +for { + msg, err := sub.NextMsg(remaining) + if err != nil { break } + // unmarshal and append +} +``` + +### ResourceManager.GetResourceStats() + +New method that returns total/allocated vCPU and memory plus per-instance-type available capacity in a single lock acquisition. All data was already tracked — no new system calls needed. + +## Files Modified + +| File | Change | +|------|--------| +| `hive/config/config.go` | Added `NodeStatusResponse`, `InstanceTypeCap`, `VMInfo`, `NodeVMsResponse` types | +| `hive/daemon/daemon_handlers.go` | Added `daemonIP()`, `handleNodeStatus()`, `handleNodeVMs()` handlers | +| `hive/daemon/daemon.go` | Added `hive.node.status` and `hive.node.vms` subscriptions in `subscribeAll()`, added `GetResourceStats()` method to `ResourceManager` | +| `cmd/hive/cmd/get.go` | **New file** — `get nodes`, `get vms` commands with pterm table output | +| `cmd/hive/cmd/top.go` | **New file** — `top nodes` command with node resource + instance type capacity tables | +| `tests/e2e/run-e2e.sh` | Added Phase 1b (get nodes, top nodes, get vms empty) and Phase 5a-pre (get vms with running VM) | +| `tests/e2e/run-multinode-e2e.sh` | Added Phase 3b (multi-node get/top) and post-launch get vms verification | + +## Testing + +- `make preflight` — all checks pass (gofmt, vet, gosec, staticcheck, govulncheck, unit tests) +- `make test-docker-single` — single-node E2E passes with new CLI test phases +- `make test-docker-multi` — 3-node E2E passes, `get vms` correctly shows instances across nodes with correct IPs + +### E2E test coverage added + +- `hive get nodes` shows Ready nodes with correct IP, region, AZ, services +- `hive top nodes` shows CPU/MEM stats and instance type capacity table +- `hive get vms` returns "No VMs found" when empty +- `hive get vms` shows running instances with correct node placement after launch +- Multi-node: VMs distributed across nodes are all visible in `get vms` output diff --git a/docs/development/improvements/iac-testing-aws-terraform.md b/docs/development/improvements/iac-testing-aws-terraform.md new file mode 100644 index 000000000..53cbf2fdb --- /dev/null +++ b/docs/development/improvements/iac-testing-aws-terraform.md @@ -0,0 +1,215 @@ +# IaC Compatibility: Terraform/OpenTofu with AWS Provider + +**Status: Planned** + +## Summary + +Enable infrastructure-as-code workflows using the standard Terraform/OpenTofu AWS provider (`hashicorp/aws`) pointed at MulgaOS/Hive endpoints. The goal is to run `tofu apply` against a Hive cluster and have it create real infrastructure — EC2 instances, EBS volumes, key pairs — using the same `.tf` files that work against AWS. + +## Context / Problem Statement + +Hive already implements 37 EC2 API actions (instances, volumes, snapshots, images, key pairs, tags, etc.), but the Terraform AWS provider requires several additional APIs during `plan` and `apply` — particularly VPC networking resources that Terraform treats as mandatory dependencies for EC2 instances. + +The existing `scripts/iac/aws/main.tf` defines a full VPC + subnet + security group + instance stack, but `tofu apply` fails because the VPC/networking APIs aren't implemented. We need to either: + +1. **Stub** the networking APIs (accept input, return valid-looking output, store state) so Terraform can proceed, or +2. **Implement** them with real backing logic + +For the initial milestone, stubbing is acceptable — Hive VMs already get networking through QEMU user-mode or bridge networking at the daemon level, so the VPC/subnet/SG resources are metadata-only for now. The critical requirement is that Terraform can complete a full `plan` → `apply` → `destroy` cycle. + +## Current API Coverage + +### Fully Implemented (works with Terraform) + +| Resource | Terraform Type | Hive API Actions | +|---|---|---| +| EC2 Instances | `aws_instance` | `RunInstances`, `DescribeInstances`, `StartInstances`, `StopInstances`, `TerminateInstances`, `DescribeInstanceTypes` | +| EBS Volumes | `aws_ebs_volume` | `CreateVolume`, `DescribeVolumes`, `DescribeVolumeStatus`, `ModifyVolume`, `DeleteVolume` | +| Volume Attachment | `aws_volume_attachment` | `AttachVolume`, `DetachVolume` | +| Key Pairs | `aws_key_pair` | `CreateKeyPair`, `ImportKeyPair`, `DescribeKeyPairs`, `DeleteKeyPair` | +| AMI/Images | `aws_ami`, data source | `DescribeImages`, `CreateImage` | +| Snapshots | `aws_ebs_snapshot` | `CreateSnapshot`, `DescribeSnapshots`, `DeleteSnapshot`, `CopySnapshot` | +| Tags | (all resources) | `CreateTags`, `DeleteTags`, `DescribeTags` | +| Regions/AZs | data sources | `DescribeRegions`, `DescribeAvailabilityZones` | +| Account | data source | `DescribeAccountAttributes` | +| Egress IGW | `aws_egress_only_internet_gateway` | `Create/Describe/DeleteEgressOnlyInternetGateway` | + +### Missing — Required by Terraform AWS Provider + +These APIs are called by Terraform during `plan`/`apply` for common infrastructure patterns. Each needs at minimum a stub that accepts input, generates an ID, stores state in NATS KV, and returns a valid response. + +#### Priority 1: VPC Networking (blocks `tofu apply` on `main.tf`) + +| Terraform Resource | Required API Actions | Complexity | +|---|---|---| +| `aws_vpc` | `CreateVpc`, `DescribeVpcs`, `DeleteVpc`, `ModifyVpcAttribute` | Medium — needs ID generation, CIDR storage, state tracking | +| `aws_subnet` | `CreateSubnet`, `DescribeSubnets`, `DeleteSubnet` | Medium — references VPC, needs AZ assignment | +| `aws_security_group` | `CreateSecurityGroup`, `DescribeSecurityGroups`, `DeleteSecurityGroup`, `AuthorizeSecurityGroupIngress`, `AuthorizeSecurityGroupEgress`, `RevokeSecurityGroupIngress`, `RevokeSecurityGroupEgress` | High — complex rule model, Terraform creates a default SG per VPC | +| `aws_internet_gateway` | `CreateInternetGateway`, `DescribeInternetGateways`, `DeleteInternetGateway`, `AttachInternetGateway`, `DetachInternetGateway` | Medium — references VPC | +| `aws_route_table` | `CreateRouteTable`, `DescribeRouteTables`, `DeleteRouteTable`, `CreateRoute`, `DeleteRoute` | Medium — references VPC, subnet associations | +| `aws_route_table_association` | `AssociateRouteTable`, `DisassociateRouteTable` | Low — links subnet to route table | + +#### Priority 2: Provider Initialization (called during `tofu init`/`plan`) + +| Action | Why Terraform Calls It | Stub Needed | +|---|---|---| +| `GetCallerIdentity` (STS) | Provider startup — validates credentials | Yes — return static account ID | +| `DescribeVpcAttribute` | After VPC creation — queries DNS settings | Yes — return defaults | +| `DescribeNetworkInterfaces` | Instance creation — checks ENI state | Yes — return empty | +| `DescribeInstanceAttribute` | After instance creation — checks attributes | Yes — return defaults | + +#### Priority 3: Nice-to-Have + +| Terraform Resource | Required API Actions | Notes | +|---|---|---| +| `aws_placement_group` | `CreatePlacementGroup`, `DescribePlacementGroups`, `DeletePlacementGroup` | Used in `main.tf` for spread strategy | +| `aws_eip` | `AllocateAddress`, `DescribeAddresses`, `ReleaseAddress`, `AssociateAddress` | Static IP assignment | +| `aws_network_interface` | `CreateNetworkInterface`, `DescribeNetworkInterfaces`, `DeleteNetworkInterface` | Advanced networking | + +## Proposed Changes + +### Phase 1: VPC Networking Stubs + +Implement stub handlers for all Priority 1 + Priority 2 APIs. Each stub: + +- Accepts the AWS SDK input struct +- Generates a resource ID (e.g., `vpc-`, `subnet-`) +- Stores the resource in NATS KV (bucket per resource type) +- Returns a valid AWS-compatible response +- Supports `Describe*` to list stored resources +- Supports `Delete*` to remove from KV + +Architecture: +``` +hive/gateway/ec2/vpc/ # CreateVpc, DescribeVpcs, DeleteVpc +hive/gateway/ec2/subnet/ # CreateSubnet, DescribeSubnets, DeleteSubnet +hive/gateway/ec2/sg/ # Security group CRUD + rule management +hive/gateway/ec2/igw/ # Internet gateway CRUD + attach/detach +hive/gateway/ec2/rtb/ # Route table CRUD + associations +hive/gateway/sts/ # GetCallerIdentity stub +``` + +Storage: NATS KV buckets — `vpc`, `subnet`, `security-group`, `internet-gateway`, `route-table`. Each entry keyed by resource ID, value is JSON-serialized resource state. + +### Phase 2: Terraform Integration Testing + +Update `scripts/iac/aws/main.tf` and add automated validation: + +1. **Simplify `main.tf`** — start with a minimal config that exercises implemented features: + - Key pair (works today) + - VPC + subnet + security group (after Phase 1 stubs) + - EC2 instance with EBS volume + - Tags on all resources + +2. **Add `scripts/iac/aws/test.sh`** — automated test script: + ```bash + tofu init + tofu plan -out=plan.tfplan + tofu apply plan.tfplan + # Verify resources exist via AWS CLI + tofu destroy -auto-approve + ``` + +3. **Add to CI** — optional E2E stage that runs the Terraform test against a running Hive instance + +### Phase 3: Real VPC Networking (Future) + +Replace stubs with real implementations as Hive networking matures: +- Map VPCs to Linux bridges or network namespaces +- Map subnets to DHCP ranges on bridges +- Map security groups to nftables/iptables rules +- Map internet gateways to NAT/routing rules + +## Files to Modify + +### Phase 1: New Gateway Handlers + +| File | Description | +|---|---| +| `hive/gateway/ec2/vpc/CreateVpc.go` | Create VPC stub — generate ID, store in NATS KV | +| `hive/gateway/ec2/vpc/DescribeVpcs.go` | List VPCs from NATS KV | +| `hive/gateway/ec2/vpc/DeleteVpc.go` | Remove VPC from NATS KV | +| `hive/gateway/ec2/subnet/CreateSubnet.go` | Create subnet stub | +| `hive/gateway/ec2/subnet/DescribeSubnets.go` | List subnets | +| `hive/gateway/ec2/subnet/DeleteSubnet.go` | Delete subnet | +| `hive/gateway/ec2/sg/CreateSecurityGroup.go` | Create security group | +| `hive/gateway/ec2/sg/DescribeSecurityGroups.go` | List security groups | +| `hive/gateway/ec2/sg/DeleteSecurityGroup.go` | Delete security group | +| `hive/gateway/ec2/sg/AuthorizeIngress.go` | Add ingress rule | +| `hive/gateway/ec2/sg/AuthorizeEgress.go` | Add egress rule | +| `hive/gateway/ec2/sg/RevokeIngress.go` | Remove ingress rule | +| `hive/gateway/ec2/sg/RevokeEgress.go` | Remove egress rule | +| `hive/gateway/ec2/igw/CreateInternetGateway.go` | Create IGW stub | +| `hive/gateway/ec2/igw/DescribeInternetGateways.go` | List IGWs | +| `hive/gateway/ec2/igw/DeleteInternetGateway.go` | Delete IGW | +| `hive/gateway/ec2/igw/AttachInternetGateway.go` | Attach IGW to VPC | +| `hive/gateway/ec2/igw/DetachInternetGateway.go` | Detach IGW from VPC | +| `hive/gateway/ec2/rtb/CreateRouteTable.go` | Create route table | +| `hive/gateway/ec2/rtb/DescribeRouteTables.go` | List route tables | +| `hive/gateway/ec2/rtb/DeleteRouteTable.go` | Delete route table | +| `hive/gateway/ec2/rtb/CreateRoute.go` | Add route to table | +| `hive/gateway/ec2/rtb/DeleteRoute.go` | Remove route from table | +| `hive/gateway/ec2/rtb/AssociateRouteTable.go` | Associate subnet with route table | +| `hive/gateway/ec2/rtb/DisassociateRouteTable.go` | Disassociate route table | +| `hive/gateway/sts/GetCallerIdentity.go` | STS stub — return static account | + +### Existing Files to Modify + +| File | Description | +|---|---| +| `hive/gateway/ec2.go` | Register new VPC/subnet/SG/IGW/RTB actions in `ec2Actions` map | +| `hive/gateway/gateway.go` | Add STS service routing if needed | +| `scripts/iac/aws/main.tf` | Update endpoint config, adjust resources for current capabilities | +| `scripts/iac/aws/README.md` | Update with current API support status | + +## Terraform AWS Provider Behavior Notes + +Key things to know about how the Terraform AWS provider interacts with APIs: + +1. **Provider init** calls `sts:GetCallerIdentity` unless `skip_requesting_account_id = true` — we set this in the provider config +2. **VPC creation** triggers `CreateVpc` then immediately `DescribeVpcs` to wait for `available` state, then `DescribeVpcAttribute` for DNS settings +3. **Security group** — Terraform creates a default SG when creating a VPC, and `CreateSecurityGroup` must return a `groupId`; rules are added via separate `AuthorizeSecurityGroup{Ingress,Egress}` calls +4. **Instance creation** — Terraform passes `SubnetId` and `SecurityGroupId` in `RunInstances`; we need to accept these without erroring even if networking is stubbed +5. **Destroy order** — Terraform destroys in reverse dependency order (instances first, then SGs, subnets, IGW, VPC). Each `Delete*` must succeed or Terraform retries. +6. **State refresh** — On every `plan`/`apply`, Terraform calls `Describe*` for every resource in state to detect drift. All `Describe*` endpoints must return consistent data. + +## Testing + +### Manual Validation + +```bash +cd scripts/iac/aws +export TF_VAR_mulgaos_endpoint="https://localhost:9999" +export AWS_ACCESS_KEY_ID=test +export AWS_SECRET_ACCESS_KEY=test + +tofu init +tofu plan # Should complete without errors +tofu apply # Should create all resources +tofu destroy # Should clean up all resources +``` + +### Automated (Future) + +Add a Terraform E2E stage to CI that runs the full `init → plan → apply → verify → destroy` cycle against a Hive instance in Docker. + +## Implementation Order + +1. VPC + DescribeVpcs + DeleteVpc (unblocks subnet, SG, IGW) +2. Subnet + DescribeSubnets + DeleteSubnet +3. Security Group (full CRUD + rules) +4. Internet Gateway (CRUD + attach/detach) +5. Route Table (CRUD + routes + associations) +6. Wire all into `ec2.go` action map +7. Test with `tofu plan` (catches missing APIs) +8. Test with `tofu apply` (catches response format issues) +9. Update `scripts/iac/aws/main.tf` and README + +## Future Work + +- Replace VPC stubs with real Linux bridge/namespace networking +- Map security groups to nftables rules on hypervisor +- Implement Elastic IPs for static addressing +- Add S3 endpoint support (Predastore already S3-compatible) +- Implement IAM basics for multi-tenant access control +- Add CloudWatch-compatible metrics endpoint diff --git a/docs/development/improvements/multi-server-restart.md b/docs/development/improvements/multi-server-restart.md index 676d083e5..97ab20858 100644 --- a/docs/development/improvements/multi-server-restart.md +++ b/docs/development/improvements/multi-server-restart.md @@ -225,7 +225,7 @@ Called in `Start()` after `initJetStream()` succeeds. Failure is non-fatal (logg ### Phase 2: Cross-node coordination -These changes enable daemons to coordinate recovery of failed nodes' VMs. +These changes enable daemons to coordinate lifecycle events — both failure recovery and graceful cluster-wide shutdown. #### 2.1 Heartbeat publishing @@ -283,6 +283,242 @@ The recovery leader: - `hive/daemon/recovery.go` — new file, recovery coordinator logic - `hive/daemon/daemon.go` — integrate recovery coordinator startup +#### 2.4 Coordinated cluster shutdown + +##### Problem + +The current `stop-dev.sh` operates per-node: each node independently shuts down its own services in reverse startup order. In a multi-node cluster, this creates ordering hazards: + +1. **Storage pulled from under running VMs**: If node1 stops Predastore before node2's VMs have finished flushing writes through Viperblock, data can be lost or corrupted. +2. **Cascade failures during rolling stop**: Stopping NATS on one node can break JetStream quorum before other nodes have persisted their state. +3. **Manual coordination required**: An operator must SSH into each node and run `stop-dev.sh` in the right order — infrastructure nodes last, compute nodes first. This is error-prone and doesn't scale. +4. **No cluster-wide drain**: There's no way to stop accepting new API requests across all gateways simultaneously. A request arriving at node2's AWSGW during node1's shutdown could try to launch a VM on a node that's mid-teardown. +5. **Orphaned nbdkit processes**: If a daemon exits before cleanly unmounting all volumes, nbdkit processes may be left running with no parent to clean them up. The current `stop-dev.sh` handles this via `pgrep`/`pkill` but only for the local node. + +##### Design inspiration + +The coordinated shutdown draws from several distributed systems patterns: + +| Source | Pattern | Application to Hive | +|--------|---------|---------------------| +| Ceph | Pre-shutdown flags (`noout`, `norecover`, `nobackfill`) set before maintenance; monitors shut down last | Write cluster shutdown marker before any service stops; NATS shuts down last | +| Ceph | Ordered daemon shutdown: RGW → MDS → OSD → MON | AWSGW → Daemon/VMs → Viperblock → Predastore → NATS | +| Kubernetes | Graceful node shutdown (KEP-2000): kubelet uses systemd inhibitor locks, pods sorted by priority class, `terminationGracePeriodSeconds` per pod | Per-VM QMP `system_powerdown` with configurable timeout, force kill after grace period | +| Kubernetes | Pod termination lifecycle: preStop hook → SIGTERM → grace period → SIGKILL | Phase-based shutdown with ACK between phases; no phase proceeds until prior phase completes | +| Kubernetes | `kubectl drain` cordons a node (marks unschedulable) before evicting pods | GATE phase stops AWSGW to prevent new work before draining VMs | +| Rancher/Harvester | Ordered HCI shutdown: cordon node → migrate VMs → drain workloads → power off; fleet-managed lifecycle for multi-cluster | Coordinator-driven phased shutdown across all nodes; progress tracking via NATS | +| RKE2 | `rke2-killall.sh` handles both graceful and forceful modes with explicit cleanup of iptables, mounts, and network namespaces | `--force` flag for immediate shutdown; explicit nbdkit/QEMU cleanup in DRAIN phase | +| etcd | Learner removal before shutdown; leader transfer to minimize disruption | Coordinator is last to stop NATS; non-coordinator nodes stop NATS first | + +##### Command + +```bash +# Graceful shutdown — waits for VMs to power down cleanly +./bin/hive admin cluster shutdown + +# Force shutdown — SIGKILL VMs immediately, skip grace periods +./bin/hive admin cluster shutdown --force + +# Custom timeout for VM drain phase (default: 120s) +./bin/hive admin cluster shutdown --timeout 300s + +# Dry run — show what would happen without executing +./bin/hive admin cluster shutdown --dry-run +``` + +The command can be run from **any node** in the cluster. The issuing node becomes the shutdown coordinator. If the coordinator crashes mid-shutdown, any surviving node can detect the `cluster.shutdown` KV marker and resume coordination (similar to the recovery leader election pattern from 2.3). + +##### Shutdown protocol + +The protocol uses five phases, each gated by acknowledgements from all participating nodes. The coordinator drives transitions and aggregates progress. NATS request-reply ensures reliable delivery with timeouts per phase. + +``` +Phase 1: GATE ──────► Phase 2: DRAIN ──────► Phase 3: STORAGE ──────► Phase 4: PERSIST ──────► Phase 5: INFRA +(stop new work) (stop VMs) (stop block I/O) (stop object storage) (stop NATS) +``` + +**Phase 1: GATE** — stop accepting new work + +Purpose: Ensure no new API requests enter the system while shutdown proceeds. Analogous to Ceph's `noout`/`norecover` flags and Kubernetes' node cordon. + +1. Coordinator writes `cluster.shutdown` marker to `hive-cluster-state` KV: + ``` + Key: cluster.shutdown + Value: {"initiator": "node1", "phase": "gate", "started": "2024-...", "timeout": "120s", "force": false} + ``` +2. Coordinator publishes `hive.cluster.shutdown.gate` (NATS request, expecting N replies) +3. Each node on receiving the gate request: + - Stops AWSGW service (no new API requests accepted) + - Stops hive-ui service + - Sets internal `shuttingDown` flag to reject any in-flight NATS work requests + - ACKs to coordinator: `{"node": "node2", "phase": "gate", "stopped": ["awsgw", "ui"]}` +4. Coordinator waits for all ACKs (timeout: 30s per node) +5. On timeout: log which nodes didn't respond, proceed anyway (they may be already down) + +**Phase 2: DRAIN** — gracefully terminate all VMs + +Purpose: Ensure all VMs are cleanly shut down and their volumes unmounted before storage services stop. This is the critical data-safety phase. Mirrors Kubernetes' pod termination lifecycle: preStop → SIGTERM → grace period → SIGKILL. + +1. Coordinator publishes `hive.cluster.shutdown.drain` +2. Each daemon node: + - Sends QMP `system_powerdown` to each running QEMU instance (same as `setupShutdown()` today) + - Waits for QEMU processes to exit, polling PID files + - Reports progress via `hive.cluster.shutdown.progress`: + ```json + {"node": "node2", "phase": "drain", "total": 4, "remaining": 2, "vms": ["i-abc123", "i-def456"]} + ``` + - After per-VM grace period (default 60s), sends SIGTERM then SIGKILL to stubborn QEMU processes + - Unmounts all EBS volumes via `ebs.unmount` NATS requests (viperblock still running) + - Cleans up any orphaned nbdkit processes attached to terminated VMs + - Writes instance state to JetStream KV (same as `WriteState()` in `setupShutdown()`) + - Writes per-node shutdown marker (reuses Phase 1's `WriteShutdownMarker()`) + - ACKs when all VMs have exited and volumes are unmounted +3. Coordinator aggregates progress, prints live status +4. `--force` mode: skip QMP `system_powerdown`, immediately SIGKILL all QEMU processes +5. On timeout: coordinator logs remaining VMs, force-kills them, proceeds + +**Phase 3: STORAGE** — stop block storage + +Purpose: Shut down viperblock and clean up all nbdkit processes. Safe to do now because all VMs are terminated and volumes unmounted. + +1. Coordinator publishes `hive.cluster.shutdown.storage` +2. Each node with viperblock service: + - Stops viperblock service + - Scans for and kills any orphaned nbdkit processes (`pkill -f nbdkit`) + - Verifies no NBD block devices remain active + - ACKs when complete +3. Nodes without viperblock: ACK immediately +4. Coordinator waits for all ACKs (timeout: 30s) + +**Phase 4: PERSIST** — stop object storage + +Purpose: Shut down Predastore cleanly. Safe because no viperblock or AWSGW requests remain. Predastore may still have internal replication traffic between nodes — the coordinator ensures all nodes stop together. + +1. Coordinator publishes `hive.cluster.shutdown.persist` +2. Each node with predastore service: + - Stops predastore service + - ACKs when complete +3. Coordinator waits for all ACKs (timeout: 30s) + +**Phase 5: INFRA** — stop NATS (coordinator last) + +Purpose: Shut down the coordination layer itself. This is fire-and-forget because the communication channel is being torn down. + +1. Coordinator publishes `hive.cluster.shutdown.infra` to all **non-coordinator** nodes +2. Non-coordinator nodes: stop local NATS and exit the shutdown handler +3. Coordinator waits 5s for NATS cluster to stabilize with fewer members +4. Coordinator stops its own NATS +5. Coordinator logs "Cluster shutdown complete" and exits + +##### Coordinator election and failure handling + +The coordinator is simply the node that runs the `hive admin cluster shutdown` command. If the coordinator crashes mid-shutdown: + +1. The `cluster.shutdown` KV marker persists (1-hour TTL, long enough for any shutdown) +2. The marker records the current phase and which nodes have ACKed +3. Any surviving daemon detects the marker via `kv.Watch("cluster.shutdown")` +4. After a coordinator timeout (30s without progress updates), a surviving daemon takes over using the same CAS pattern as recovery leader election (2.3) +5. The new coordinator resumes from the last completed phase + +If a node is unreachable during any phase: +- Coordinator logs a warning and continues after the per-phase timeout +- Unreachable nodes may have already shut down (e.g., power loss) +- If they come back later, they'll see the `cluster.shutdown` marker and refuse to start services until the marker is cleared + +##### Status reporting + +The coordinator prints live progress to stdout: + +``` +Cluster shutdown initiated from node1 (3 nodes) + +Phase 1/5: GATE — Stopping API gateways and UI... + node1: awsgw stopped, ui stopped [ok] + node2: awsgw stopped, ui stopped [ok] + node3: awsgw stopped, ui stopped [ok] + +Phase 2/5: DRAIN — Stopping virtual machines... + node1: 4/4 VMs stopped [ok] 12.3s + node2: 2/2 VMs stopped [ok] 8.7s + node3: 1/1 VMs stopped [ok] 6.1s + +Phase 3/5: STORAGE — Stopping block storage... + node1: viperblock stopped, 0 nbdkit cleaned [ok] + node2: viperblock stopped, 0 nbdkit cleaned [ok] + node3: viperblock stopped, 0 nbdkit cleaned [ok] + +Phase 4/5: PERSIST — Stopping object storage... + node1: predastore stopped [ok] + node2: predastore stopped [ok] + node3: predastore stopped [ok] + +Phase 5/5: INFRA — Stopping NATS... + node2: nats stopped + node3: nats stopped + node1: nats stopped (coordinator, last) + +Cluster shutdown complete (47.3s) +``` + +##### NATS topics + +| Topic | Pattern | Purpose | +|-------|---------|---------| +| `hive.cluster.shutdown.gate` | Request-Reply | Phase 1: stop AWSGW + UI, await ACK | +| `hive.cluster.shutdown.drain` | Request-Reply | Phase 2: stop VMs, await ACK | +| `hive.cluster.shutdown.progress` | Publish (broadcast) | Progress updates during drain (VM counts) | +| `hive.cluster.shutdown.storage` | Request-Reply | Phase 3: stop viperblock + nbdkit, await ACK | +| `hive.cluster.shutdown.persist` | Request-Reply | Phase 4: stop predastore, await ACK | +| `hive.cluster.shutdown.infra` | Publish (fire-and-forget) | Phase 5: stop NATS on non-coordinator nodes | + +##### Cluster shutdown KV state + +``` +Bucket: hive-cluster-state +Key: cluster.shutdown +Value: { + "initiator": "node1", + "phase": "drain", + "started": "2024-01-15T10:30:00Z", + "timeout": "120s", + "force": false, + "nodes_total": 3, + "nodes_acked": { + "node1": {"phase": "gate", "at": "..."}, + "node2": {"phase": "gate", "at": "..."}, + "node3": {"phase": "gate", "at": "..."} + } +} +``` + +##### Integration with existing code + +- **Reuses `stopInstance()`**: The daemon's existing `stopInstance()` function handles QMP shutdown, PID polling, volume unmount, and force kill. The DRAIN phase calls this same code path rather than reimplementing it. +- **Reuses `WriteShutdownMarker()`**: Each node writes its per-node shutdown marker during DRAIN, so on next startup, `restoreInstances()` sees a clean shutdown. +- **Reuses `./bin/hive service stop`**: For stopping AWSGW, UI, viperblock, predastore, and NATS, the coordinator tells each daemon to invoke the same service stop mechanism used by `stop-dev.sh`. +- **`setupShutdown()` becomes cluster-aware**: When a daemon receives a cluster shutdown request, it skips independent shutdown and instead participates in the coordinated protocol. If a SIGTERM arrives while a coordinated shutdown is in progress, the daemon ACKs the current phase and exits. +- **`stop-dev.sh` remains for single-node dev**: The script continues to work for local development. For multi-node clusters, `hive admin cluster shutdown` is the preferred approach. + +##### Comparison with per-node `stop-dev.sh` + +| Aspect | `stop-dev.sh` (current) | `hive admin cluster shutdown` (proposed) | +|--------|------------------------|------------------------------------------| +| Scope | Single node | Entire cluster from any node | +| Ordering | Per-node reverse order | Cluster-wide phased ordering | +| VM safety | Daemon exits, then waits for QEMU via `pgrep` | Daemon actively monitors each VM's QMP shutdown | +| Storage safety | Hopes VMs exited before predastore stop | Guarantees all VMs stopped before storage phases | +| nbdkit cleanup | `pgrep`/`pkill` after the fact | Explicit unmount + cleanup during DRAIN | +| Progress | Per-service echo messages | Cluster-wide live progress with VM counts | +| Failure handling | None — operator must check each node | Automatic coordinator failover, timeout handling | +| NATS safety | Stops NATS whenever it gets there | NATS is always last, coordinator node last of all | + +**Files:** +- `hive/daemon/daemon.go` — shutdown handler that participates in coordinated protocol, `shuttingDown` flag to reject new work +- `hive/daemon/shutdown.go` — **new file**, coordinator logic: phase management, ACK aggregation, progress reporting, coordinator failover +- `hive/daemon/jetstream.go` — `WriteClusterShutdown()` / `ReadClusterShutdown()` / `DeleteClusterShutdown()` / `WatchClusterShutdown()` +- `cmd/hive/cmd/admin.go` — new `cluster shutdown` subcommand with `--force`, `--timeout`, `--dry-run` flags +- `hive/daemon/daemon_handlers.go` — NATS handlers for `hive.cluster.shutdown.*` topics + ### Phase 3: Operational tooling #### 3.1 Maintenance mode flag @@ -297,6 +533,7 @@ When maintenance mode is set for a node: - Peer daemons skip the 5-minute failure detection grace period for that node (they know it's planned) - The node itself skips VM recovery on restart if the maintenance flag is still set (operator controls when VMs come back) - Analogous to Ceph's `noout` flag +- Works with coordinated shutdown: the coordinator sets maintenance mode on all nodes before Phase 1, clears it after Phase 5 **Files:** - `hive/daemon/jetstream.go` — new `SetMaintenanceMode()` / `ClearMaintenanceMode()` / `IsMaintenanceMode()` @@ -352,12 +589,14 @@ This reads from the `hive-cluster-state` KV bucket (heartbeats + service manifes ## Files to Modify (Phase 2 + 3) -| File | Changes | -|------|---------| -| `hive/daemon/jetstream.go` | Heartbeat, recovery lease, maintenance mode | -| `hive/daemon/recovery.go` | **New file** — recovery coordinator: leader election, capability-aware VM assignment, peer monitoring | -| `hive/daemon/daemon.go` | Heartbeat/recovery integration | -| `cmd/hive/cmd/admin.go` | `maintenance` and `status` subcommands | +| File | Changes | Phase | +|------|---------|-------| +| `hive/daemon/jetstream.go` | Heartbeat, recovery lease, cluster shutdown KV, maintenance mode | 2.1–2.4, 3.1 | +| `hive/daemon/recovery.go` | **New file** — recovery coordinator: leader election, capability-aware VM assignment, peer monitoring | 2.3 | +| `hive/daemon/shutdown.go` | **New file** — coordinated shutdown coordinator: phase management, ACK aggregation, progress reporting, failover | 2.4 | +| `hive/daemon/daemon.go` | Heartbeat/recovery integration, cluster shutdown handler, `shuttingDown` flag | 2.1–2.4 | +| `hive/daemon/daemon_handlers.go` | NATS handlers for `hive.cluster.shutdown.*` topics | 2.4 | +| `cmd/hive/cmd/admin.go` | `cluster shutdown`, `maintenance`, `status` subcommands | 2.4, 3.1, 3.3 | ## Implementation Order @@ -376,15 +615,16 @@ Phase 0 is the prerequisite — all subsequent phases depend on nodes knowing th 8. [x] Clean shutdown marker 9. [x] Service manifest on startup -**Phase 2** — cross-node failure recovery — **Planned**: +**Phase 2** — cross-node coordination — **Planned**: 10. Heartbeat publishing (with capabilities) 11. Peer failure detection 12. Leader-elected recovery coordinator (capability-aware placement) +13. Coordinated cluster shutdown (`hive admin cluster shutdown`) **Phase 3** — operational tooling — **Planned**: -13. Maintenance mode flag -14. Recovery status endpoints -15. Admin cluster status command +14. Maintenance mode flag +15. Recovery status endpoints +16. Admin cluster status command ## Testing @@ -409,22 +649,33 @@ Phase 0 is the prerequisite — all subsequent phases depend on nodes knowing th #### Unit tests - Recovery coordinator: test leader election, capability-filtered VM assignment, epoch-based fencing - Heartbeat: test staleness detection thresholds +- Coordinated shutdown: test phase transitions, ACK aggregation, timeout handling, coordinator failover +- Shutdown protocol: test `cluster.shutdown` KV marker lifecycle (write, read, resume, clear) #### Integration tests (multi-node Docker) - Start 3-node cluster, launch 4 VMs, stop all, restart — verify all 4 VMs recover exactly once - Start 3+1 cluster (3 full + 1 compute-only), verify compute-only connects to remote NATS/Predastore - Kill one node (SIGKILL), verify surviving nodes recover its VMs after grace period - Rolling restart: stop/start nodes one at a time, verify zero duplicate launches +- Coordinated shutdown: launch VMs across 3 nodes, run `hive admin cluster shutdown` from node2, verify phase ordering (AWSGW stops before VMs, VMs stop before viperblock, viperblock before predastore, NATS last) +- Coordinated shutdown with `--force`: verify VMs are killed immediately, no grace period +- Shutdown coordinator failure: start cluster shutdown from node1, kill node1 mid-drain, verify another node resumes coordination +- Post-shutdown restart: after coordinated shutdown, restart all nodes, verify clean shutdown markers present and VMs restore correctly +- No orphaned processes: after coordinated shutdown, verify zero remaining QEMU/nbdkit processes on all nodes #### Manual verification - Deploy to 3-node test cluster (10.1.3.170-172), run stop/start cycle, check logs for race conditions - Add a 4th compute-only node, verify it joins and accepts workloads +- Run `hive admin cluster shutdown` from each of the 3 nodes in turn, verify it works from any node +- Run `hive admin cluster shutdown --dry-run` and verify output matches expected phase plan ## Future Work +- **Coordinated cluster startup** (`hive admin cluster start`): Complement to `cluster shutdown`. Coordinator starts services across all nodes in dependency order (NATS → Predastore → Viperblock → Daemon → AWSGW → UI), with health checks between phases. Would replace per-node `start-dev.sh` for multi-node clusters. - **Persistent VM pinning**: Optional affinity rules to prefer restarting VMs on their original node -- **Live migration**: Move running VMs between nodes without stopping them (requires QEMU live migration support) +- **Live migration**: Move running VMs between nodes without stopping them (requires QEMU live migration support). Would enable `cluster shutdown` to migrate VMs to surviving nodes instead of stopping them. - **Automatic scaling**: Detect sustained capacity pressure and suggest adding nodes -- **Recovery metrics**: Prometheus-compatible metrics for recovery duration, failures, leader elections +- **Recovery metrics**: Prometheus-compatible metrics for recovery duration, failures, leader elections, shutdown phase timings - **Dynamic service reconfiguration**: Add/remove services from a running node without full restart - **Storage-only node health**: Extend recovery to detect failed storage nodes and reroute Viperblock/Predastore traffic +- **Rolling cluster upgrade**: Coordinated restart that drains nodes one at a time, upgrades binaries, and restarts — maintaining cluster availability throughout diff --git a/hive/config/config.go b/hive/config/config.go index beaafe01c..4530b3f41 100644 --- a/hive/config/config.go +++ b/hive/config/config.go @@ -245,6 +245,48 @@ type NodeHealthResponse struct { ServiceHealth map[string]string `json:"service_health,omitempty"` } +// NodeStatusResponse is returned by the hive.node.status NATS topic (fan-out). +type NodeStatusResponse struct { + Node string `json:"node"` + Status string `json:"status"` + Host string `json:"host"` + Region string `json:"region"` + AZ string `json:"az"` + Uptime int64 `json:"uptime"` + Services []string `json:"services"` + TotalVCPU int `json:"total_vcpu"` + TotalMemGB float64 `json:"total_mem_gb"` + AllocVCPU int `json:"alloc_vcpu"` + AllocMemGB float64 `json:"alloc_mem_gb"` + VMCount int `json:"vm_count"` + InstanceTypes []InstanceTypeCap `json:"instance_types"` +} + +// InstanceTypeCap describes available capacity for one instance type on a node. +type InstanceTypeCap struct { + Name string `json:"name"` + VCPU int `json:"vcpu"` + MemoryGB float64 `json:"memory_gb"` + Available int `json:"available"` +} + +// VMInfo describes a single VM for the cluster stats CLI. +type VMInfo struct { + InstanceID string `json:"instance_id"` + Status string `json:"status"` + InstanceType string `json:"instance_type"` + VCPU int `json:"vcpu"` + MemoryGB float64 `json:"memory_gb"` + LaunchTime int64 `json:"launch_time"` +} + +// NodeVMsResponse is returned by the hive.node.vms NATS topic (fan-out). +type NodeVMsResponse struct { + Node string `json:"node"` + Host string `json:"host"` + VMs []VMInfo `json:"vms"` +} + // LoadConfig loads the configuration from file and environment variables func LoadConfig(configPath string) (*ClusterConfig, error) { // Set environment variable prefix diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 97b9e1e5d..471e35e1a 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -411,6 +411,45 @@ func (rm *ResourceManager) GetAvailableInstanceTypeInfos(showCapacity bool) []*e return infos } +// GetResourceStats returns current resource allocation stats for the node status response. +func (rm *ResourceManager) GetResourceStats() (totalVCPU int, totalMemGB float64, allocVCPU int, allocMemGB float64, caps []config.InstanceTypeCap) { + rm.mu.RLock() + defer rm.mu.RUnlock() + + totalVCPU = rm.availableVCPU + totalMemGB = rm.availableMem + allocVCPU = rm.allocatedVCPU + allocMemGB = rm.allocatedMem + + remainingVCPU := rm.availableVCPU - rm.allocatedVCPU + remainingMem := rm.availableMem - rm.allocatedMem + + for _, it := range rm.instanceTypes { + vCPUs := instanceTypeVCPUs(it) + memGB := float64(instanceTypeMemoryMiB(it)) / 1024.0 + if vCPUs == 0 || memGB == 0 { + continue + } + countVCPU := remainingVCPU / int(vCPUs) + countMem := int(remainingMem / memGB) + count := min(countMem, countVCPU) + if count < 0 { + count = 0 + } + name := "" + if it.InstanceType != nil { + name = *it.InstanceType + } + caps = append(caps, config.InstanceTypeCap{ + Name: name, + VCPU: int(vCPUs), + MemoryGB: memGB, + Available: count, + }) + } + return +} + // SetConfigPath sets the configuration file path for cluster management func (d *Daemon) SetConfigPath(path string) { d.configPath = path @@ -488,6 +527,8 @@ func (d *Daemon) subscribeAll() error { {"ec2.DisableSerialConsoleAccess", d.handleEC2DisableSerialConsoleAccess, "hive-workers"}, {fmt.Sprintf("hive.admin.%s.health", d.node), d.handleHealthCheck, ""}, {"hive.nodes.discover", d.handleNodeDiscover, ""}, + {"hive.node.status", d.handleNodeStatus, ""}, + {"hive.node.vms", d.handleNodeVMs, ""}, } for _, s := range subs { diff --git a/hive/daemon/daemon_handlers.go b/hive/daemon/daemon_handlers.go index ea9ac08f5..319444e41 100644 --- a/hive/daemon/daemon_handlers.go +++ b/hive/daemon/daemon_handlers.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "log/slog" + "net" "strings" "time" @@ -1666,3 +1667,91 @@ func (d *Daemon) handleEC2DescribeStoppedInstances(msg *nats.Msg) { slog.Info("handleEC2DescribeStoppedInstances completed", "count", len(reservations)) } + +// daemonIP extracts the IP portion from the daemon host (host:port format). +func (d *Daemon) daemonIP() string { + host := d.config.Daemon.Host + if h, _, err := net.SplitHostPort(host); err == nil { + return h + } + return host +} + +// handleNodeStatus responds with this node's status and resource stats. +// Used by the CLI: hive get nodes, hive top nodes. +func (d *Daemon) handleNodeStatus(msg *nats.Msg) { + totalVCPU, totalMemGB, allocVCPU, allocMemGB, caps := d.resourceMgr.GetResourceStats() + + d.Instances.Mu.Lock() + vmCount := 0 + for _, v := range d.Instances.VMS { + if v.Status == vm.StateRunning { + vmCount++ + } + } + d.Instances.Mu.Unlock() + + resp := config.NodeStatusResponse{ + Node: d.node, + Status: "Ready", + Host: d.daemonIP(), + Region: d.config.Region, + AZ: d.config.AZ, + Uptime: int64(time.Since(d.startTime).Seconds()), + Services: d.config.GetServices(), + TotalVCPU: totalVCPU, + TotalMemGB: totalMemGB, + AllocVCPU: allocVCPU, + AllocMemGB: allocMemGB, + VMCount: vmCount, + InstanceTypes: caps, + } + + data, err := json.Marshal(resp) + if err != nil { + slog.Error("handleNodeStatus failed to marshal response", "err", err) + return + } + if err := msg.Respond(data); err != nil { + slog.Error("Failed to respond to hive.node.status", "err", err) + } +} + +// handleNodeVMs responds with the list of VMs running on this node. +// Used by the CLI: hive get vms. +func (d *Daemon) handleNodeVMs(msg *nats.Msg) { + d.Instances.Mu.Lock() + vms := make([]config.VMInfo, 0, len(d.Instances.VMS)) + for _, v := range d.Instances.VMS { + info := config.VMInfo{ + InstanceID: v.ID, + Status: string(v.Status), + InstanceType: v.InstanceType, + } + // Get vCPU/memory from the resource manager's instance type info + if it, ok := d.resourceMgr.instanceTypes[v.InstanceType]; ok { + info.VCPU = int(instanceTypeVCPUs(it)) + info.MemoryGB = float64(instanceTypeMemoryMiB(it)) / 1024.0 + } + if v.Instance != nil && v.Instance.LaunchTime != nil { + info.LaunchTime = v.Instance.LaunchTime.Unix() + } + vms = append(vms, info) + } + d.Instances.Mu.Unlock() + + resp := config.NodeVMsResponse{ + Node: d.node, + Host: d.daemonIP(), + VMs: vms, + } + + data, err := json.Marshal(resp) + if err != nil { + slog.Error("handleNodeVMs failed to marshal response", "err", err) + return + } + if err := msg.Respond(data); err != nil { + slog.Error("Failed to respond to hive.node.vms", "err", err) + } +} diff --git a/tests/e2e/run-e2e.sh b/tests/e2e/run-e2e.sh index fa0162519..6b7b12361 100755 --- a/tests/e2e/run-e2e.sh +++ b/tests/e2e/run-e2e.sh @@ -97,6 +97,39 @@ wait_for_daemon_ready "https://localhost:9999" # Define common AWS CLI args AWS_EC2="aws --endpoint-url https://localhost:9999 ec2" +# Phase 1b: Cluster Stats CLI +echo "Phase 1b: Cluster Stats CLI" + +# Test hive get nodes — should show node1 as Ready +echo "Testing hive get nodes..." +GET_NODES_OUTPUT=$(./bin/hive get nodes --config ~/hive/config/hive.toml --timeout 5s 2>/dev/null) +echo "$GET_NODES_OUTPUT" +if ! echo "$GET_NODES_OUTPUT" | grep -q "Ready"; then + echo "hive get nodes did not show any Ready nodes" + exit 1 +fi +echo "hive get nodes passed" + +# Test hive top nodes — should show CPU/MEM stats +echo "Testing hive top nodes..." +TOP_NODES_OUTPUT=$(./bin/hive top nodes --config ~/hive/config/hive.toml --timeout 5s 2>/dev/null) +echo "$TOP_NODES_OUTPUT" +if ! echo "$TOP_NODES_OUTPUT" | grep -q "0/"; then + echo "hive top nodes did not show resource stats" + exit 1 +fi +echo "hive top nodes passed" + +# Test hive get vms — should show no VMs yet +echo "Testing hive get vms (empty)..." +GET_VMS_OUTPUT=$(./bin/hive get vms --config ~/hive/config/hive.toml --timeout 5s 2>/dev/null) +echo "$GET_VMS_OUTPUT" +if ! echo "$GET_VMS_OUTPUT" | grep -q "No VMs found"; then + echo "hive get vms should show 'No VMs found' before any launches" + exit 1 +fi +echo "hive get vms (empty) passed" + # Phase 2: Discovery & Metadata echo "Phase 2: Discovery & Metadata" # Verify describe-regions (just ensure it returns at least one region) @@ -258,6 +291,16 @@ if [ "$STATE" != "running" ]; then exit 1 fi +# Phase 5a-pre: Verify hive get vms shows running instance +echo "Phase 5a-pre: Cluster Stats CLI (with running VM)" +GET_VMS_OUTPUT=$(./bin/hive get vms --config ~/hive/config/hive.toml --timeout 5s 2>/dev/null) +echo "$GET_VMS_OUTPUT" +if ! echo "$GET_VMS_OUTPUT" | grep -q "$INSTANCE_ID"; then + echo "hive get vms did not show running instance $INSTANCE_ID" + exit 1 +fi +echo "hive get vms shows running instance" + # Phase 5a: Validate instance metadata fields echo "Phase 5a: Instance Metadata Validation" DESCRIBE_META=$($AWS_EC2 describe-instances --instance-ids "$INSTANCE_ID") diff --git a/tests/e2e/run-multinode-e2e.sh b/tests/e2e/run-multinode-e2e.sh index a72c36398..8cc32f632 100755 --- a/tests/e2e/run-multinode-e2e.sh +++ b/tests/e2e/run-multinode-e2e.sh @@ -156,6 +156,36 @@ wait_for_daemon_ready "https://${NODE1_IP}:${AWSGW_PORT}" # Define AWS CLI args pointing to node1's gateway AWS_EC2="aws --endpoint-url https://${NODE1_IP}:${AWSGW_PORT} ec2" +# Phase 3b: Cluster Stats CLI (Multi-Node) +echo "" +echo "Phase 3b: Cluster Stats CLI (Multi-Node)" +echo "========================================" + +# Test hive get nodes — should show all 3 nodes as Ready +echo "Testing hive get nodes..." +GET_NODES_OUTPUT=$(./bin/hive get nodes --config "$HOME/node1/config/hive.toml" --timeout 5s 2>/dev/null) +echo "$GET_NODES_OUTPUT" +READY_COUNT=$(echo "$GET_NODES_OUTPUT" | grep -c "Ready" || true) +if [ "$READY_COUNT" -lt 3 ]; then + echo "WARNING: hive get nodes shows $READY_COUNT Ready nodes (expected 3)" +fi +echo "hive get nodes passed ($READY_COUNT Ready nodes)" + +# Test hive top nodes — should show resource stats for all nodes +echo "Testing hive top nodes..." +TOP_NODES_OUTPUT=$(./bin/hive top nodes --config "$HOME/node1/config/hive.toml" --timeout 5s 2>/dev/null) +echo "$TOP_NODES_OUTPUT" +if ! echo "$TOP_NODES_OUTPUT" | grep -q "INSTANCE TYPE"; then + echo "WARNING: hive top nodes did not show instance type capacity table" +fi +echo "hive top nodes passed" + +# Test hive get vms — should show no VMs yet +echo "Testing hive get vms (empty)..." +GET_VMS_OUTPUT=$(./bin/hive get vms --config "$HOME/node1/config/hive.toml" --timeout 5s 2>/dev/null) +echo "$GET_VMS_OUTPUT" +echo "hive get vms (empty) passed" + # Verify gateway responds echo "" echo "Testing gateway connectivity..." @@ -305,6 +335,18 @@ done echo "" check_instance_distribution +# Verify hive get vms shows all running instances +echo "" +echo "Verifying hive get vms (with running VMs)..." +GET_VMS_OUTPUT=$(./bin/hive get vms --config "$HOME/node1/config/hive.toml" --timeout 5s 2>/dev/null) +echo "$GET_VMS_OUTPUT" +for instance_id in "${INSTANCE_IDS[@]}"; do + if ! echo "$GET_VMS_OUTPUT" | grep -q "$instance_id"; then + echo "WARNING: hive get vms did not show instance $instance_id" + fi +done +echo "hive get vms shows launched instances" + # Test 1a-ii: SSH Connectivity & Volume Verification echo "" echo "Test 1a-ii: SSH Connectivity & Volume Verification"