hive top, hive get cmds for cluster health and stats. improved docs for iac, cluster, multi-server restart

This commit is contained in:
Ben Duncan
2026-02-14 15:38:03 +10:00
parent 5863aeed13
commit 774655eff0
10 changed files with 1281 additions and 13 deletions
+274
View File
@@ -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()
}
+154
View File
@@ -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
}
@@ -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
@@ -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-<random>`, `subnet-<random>`)
- 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
@@ -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 <name> 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
+42
View File
@@ -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
+41
View File
@@ -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 {
+89
View File
@@ -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)
}
}
+43
View File
@@ -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")
+42
View File
@@ -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"