mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
Phase 0/1 multi-server-restart enhancements
This commit is contained in:
Ben Duncan
@@ -139,6 +139,19 @@ When starting work on a feature or investigating an issue, review the last 10-20
|
||||
- All new features must have comprehensive unit tests
|
||||
- For returning AWS errors, use `awserrors` package instead of manually typed strings
|
||||
|
||||
**Preflight Policy:**
|
||||
- Run `make preflight` before pushing any major changes — it runs the same checks as GitHub Actions (gofmt, go vet, gosec, staticcheck, govulncheck, and all unit tests)
|
||||
- All developers must install the pre-push hook to enforce this automatically:
|
||||
|
||||
```bash
|
||||
cat > .git/hooks/pre-push << 'EOF'
|
||||
#!/bin/sh
|
||||
make preflight
|
||||
make test
|
||||
EOF
|
||||
chmod +x .git/hooks/pre-push
|
||||
```
|
||||
|
||||
**Testing Policy:**
|
||||
- **All unit tests MUST pass** before committing changes
|
||||
- **No exceptions** - failing tests block commits
|
||||
|
||||
@@ -162,6 +162,7 @@ func init() {
|
||||
adminInitCmd.Flags().String("predastore-nodes", "", "Comma-separated IPs for multi-node Predastore cluster (e.g., 10.11.12.1,10.11.12.2,10.11.12.3). Requires >= 3 nodes.")
|
||||
adminInitCmd.Flags().String("formation-timeout", "10m", "Timeout for cluster formation (e.g., 5m, 30s)")
|
||||
adminInitCmd.Flags().String("cluster-name", "hive", "NATS cluster name")
|
||||
adminInitCmd.Flags().StringSlice("services", nil, "Services this node runs (default: all). Valid: nats,predastore,viperblock,daemon,awsgw,ui")
|
||||
|
||||
// Flags for admin join
|
||||
adminJoinCmd.Flags().String("region", "ap-southeast-2", "Region for this node")
|
||||
@@ -173,6 +174,7 @@ func init() {
|
||||
adminJoinCmd.Flags().String("bind", "0.0.0.0", "IP address to bind services to (e.g., 10.11.12.2 for multi-node on single host)")
|
||||
adminJoinCmd.Flags().String("cluster-bind", "", "IP address to bind NATS cluster services to (e.g., 10.11.12.1 for multi-node)")
|
||||
adminJoinCmd.Flags().String("cluster-routes", "", "NATS cluster hosts for routing specify multiple with comma (e.g., 10.11.12.1:4248,10.11.12.2:4248 for multi-node)")
|
||||
adminJoinCmd.Flags().StringSlice("services", nil, "Services this node runs (default: all)")
|
||||
adminJoinCmd.MarkFlagRequired("node")
|
||||
adminJoinCmd.MarkFlagRequired("host")
|
||||
|
||||
@@ -486,6 +488,7 @@ func runAdminInit(cmd *cobra.Command, args []string) {
|
||||
predastoreNodesStr, _ := cmd.Flags().GetString("predastore-nodes")
|
||||
formationTimeoutStr, _ := cmd.Flags().GetString("formation-timeout")
|
||||
clusterName, _ := cmd.Flags().GetString("cluster-name")
|
||||
services, _ := cmd.Flags().GetStringSlice("services")
|
||||
|
||||
// Validate IP address format
|
||||
if net.ParseIP(bindIP) == nil {
|
||||
@@ -565,7 +568,7 @@ func runAdminInit(cmd *cobra.Command, args []string) {
|
||||
if isMultiNode {
|
||||
runAdminInitMultiNode(cmd, accessKey, secretKey, accountID, natsToken, clusterName,
|
||||
configDir, hiveRoot, certPath, region, az, node, bindIP, clusterBind,
|
||||
port, nodes, formationTimeoutStr)
|
||||
port, nodes, formationTimeoutStr, services)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -649,6 +652,7 @@ func runAdminInit(cmd *cobra.Command, args []string) {
|
||||
ClusterName: clusterName,
|
||||
|
||||
PredastoreNodeID: predastoreNodeID,
|
||||
Services: services,
|
||||
}
|
||||
|
||||
// Generate config files
|
||||
@@ -701,7 +705,7 @@ func runAdminInit(cmd *cobra.Command, args []string) {
|
||||
// then generates configs with complete cluster topology.
|
||||
func runAdminInitMultiNode(cmd *cobra.Command, accessKey, secretKey, accountID, natsToken, clusterName,
|
||||
configDir, hiveRoot, certPath, region, az, node, bindIP, clusterBind string,
|
||||
port, expectedNodes int, formationTimeoutStr string) {
|
||||
port, expectedNodes int, formationTimeoutStr string, services []string) {
|
||||
|
||||
formationTimeout, err := time.ParseDuration(formationTimeoutStr)
|
||||
if err != nil {
|
||||
@@ -826,6 +830,7 @@ func runAdminInitMultiNode(cmd *cobra.Command, accessKey, secretKey, accountID,
|
||||
ClusterName: clusterName,
|
||||
|
||||
PredastoreNodeID: predastoreNodeID,
|
||||
Services: services,
|
||||
}
|
||||
|
||||
// Generate config files
|
||||
@@ -887,6 +892,7 @@ func runAdminJoin(cmd *cobra.Command, args []string) {
|
||||
bindIP, _ := cmd.Flags().GetString("bind")
|
||||
configDir, _ := cmd.Flags().GetString("config-dir")
|
||||
clusterBind, _ := cmd.Flags().GetString("cluster-bind")
|
||||
services, _ := cmd.Flags().GetStringSlice("services")
|
||||
|
||||
// Validate required parameters
|
||||
if node == "" {
|
||||
@@ -1127,6 +1133,7 @@ func runAdminJoin(cmd *cobra.Command, args []string) {
|
||||
ClusterName: creds.ClusterName,
|
||||
|
||||
PredastoreNodeID: predastoreNodeID,
|
||||
Services: services,
|
||||
}
|
||||
|
||||
// Generate config files
|
||||
|
||||
@@ -7,6 +7,9 @@ node = "{{.Node}}"
|
||||
node = "{{.Node}}"
|
||||
region = "{{.Region}}"
|
||||
az = "{{.Az}}"
|
||||
{{- if .Services}}
|
||||
services = [{{range $i, $s := .Services}}{{if $i}}, {{end}}"{{$s}}"{{end}}]
|
||||
{{- end}}
|
||||
|
||||
account_id = "{{.AccountID}}"
|
||||
accesskey = "{{.AccessKey}}"
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# Multi-Server Cluster Restart
|
||||
|
||||
**Status: Planned**
|
||||
**Status: In Progress** — Phase 0 and Phase 1 complete. Phase 2 and 3 planned.
|
||||
|
||||
## Summary
|
||||
|
||||
@@ -89,263 +89,139 @@ Each daemon's `restoreInstances()` (`daemon.go:634-719`) runs independently on s
|
||||
|
||||
## Proposed Changes
|
||||
|
||||
### Phase 0: Node capabilities and service topology
|
||||
### Phase 0: Node capabilities and service topology — Complete
|
||||
|
||||
Foundation for everything else — the cluster must know what each node runs.
|
||||
|
||||
#### 0.1 Capabilities in config
|
||||
#### 0.1 Capabilities in config — Done
|
||||
|
||||
Add a `services` list to the per-node `Config` struct. This declares which services the node runs locally:
|
||||
Added `Services []string` field to the `Config` struct (`config/config.go:29`) and two helper methods:
|
||||
- `HasService(name string) bool` — checks if a service is in the list (or returns true for all if list is empty, for backward compat)
|
||||
- `GetServices() []string` — returns the services list, defaulting to `AllServices` when empty
|
||||
|
||||
```toml
|
||||
# hive.toml — full infrastructure node (node1, node2, node3)
|
||||
[nodes.node1]
|
||||
node = "node1"
|
||||
services = ["nats", "predastore", "viperblock", "daemon", "awsgw", "ui"]
|
||||
# ... existing daemon, nats, predastore, awsgw sections ...
|
||||
The `AllServices` package variable defines the canonical list: `["nats", "predastore", "viperblock", "daemon", "awsgw", "ui"]`.
|
||||
|
||||
# hive.toml — compute-only node (node4)
|
||||
[nodes.node4]
|
||||
node = "node4"
|
||||
services = ["daemon", "viperblock"]
|
||||
# nats/predastore sections still present but point to REMOTE hosts
|
||||
[nodes.node4.nats]
|
||||
host = "10.1.3.170:4222" # remote NATS cluster
|
||||
[nodes.node4.predastore]
|
||||
host = "10.1.3.170:8443" # remote Predastore
|
||||
```
|
||||
The originally-planned role helpers (`IsComputeNode`, `IsStorageNode`, `IsInfraNode`) were deferred — `HasService()` is sufficient and more flexible.
|
||||
|
||||
The `services` field is a string slice of service names. Valid values: `nats`, `predastore`, `viperblock`, `daemon`, `awsgw`, `ui`. When omitted (backward compat), defaults to all services — existing configs work unchanged.
|
||||
Updated `NodeHealthResponse` to include `Services []string` and `ServiceHealth map[string]string`.
|
||||
|
||||
**Config struct change** (`config/config.go`):
|
||||
Unit tests added in `hive/config/config_test.go`: `TestHasService_ExplicitList`, `TestHasService_EmptyListDefaultsAll`, `TestHasService_UnknownService`, `TestGetServices_DefaultsToAll`, `TestGetServices_ExplicitList`.
|
||||
|
||||
```go
|
||||
type Config struct {
|
||||
Node string `mapstructure:"node"`
|
||||
Host string `mapstructure:"host"`
|
||||
Region string `mapstructure:"region"`
|
||||
AZ string `mapstructure:"az"`
|
||||
DataDir string `mapstructure:"data_dir"`
|
||||
Services []string `mapstructure:"services"` // NEW: which services this node runs locally
|
||||
// ... rest unchanged
|
||||
}
|
||||
```
|
||||
**Files modified:**
|
||||
- `hive/config/config.go` — `Services` field, `AllServices`, `HasService()`, `GetServices()`, updated `NodeHealthResponse`
|
||||
- `hive/config/config_test.go` — 5 new tests
|
||||
|
||||
Helper methods on Config:
|
||||
#### 0.2 Formation carries capabilities — Done
|
||||
|
||||
```go
|
||||
func (c Config) HasService(name string) bool // check if service is in the list
|
||||
func (c Config) IsComputeNode() bool // has "daemon"
|
||||
func (c Config) IsStorageNode() bool // has "predastore" or "viperblock"
|
||||
func (c Config) IsInfraNode() bool // has "nats"
|
||||
```
|
||||
Added `Services []string` to `NodeInfo` in `hive/formation/formation.go`. Since `JoinRequest` embeds `NodeInfo`, it automatically gets the field.
|
||||
|
||||
**Files:**
|
||||
- `hive/config/config.go` — add `Services` field and helper methods
|
||||
- `cmd/hive/cmd/templates/hive.toml` — add `services` to template
|
||||
- `cmd/hive/cmd/admin.go` — add `--services` flag to `hive admin init` and `hive admin join`
|
||||
Updated `BuildClusterRoutes()` to filter nodes by `"nats"` service — only nodes running NATS are included in cluster routes. Updated `BuildPredastoreNodes()` to filter by `"predastore"` service. Added `hasService()` helper with backward-compat behavior (empty services list = all services).
|
||||
|
||||
#### 0.2 Formation carries capabilities
|
||||
Unit tests added in `hive/formation/helpers_test.go`: `TestHasService_EmptyListMeansAll`, `TestHasService_ExplicitList`, `TestBuildClusterRoutes_FiltersNonNATSNodes`, `TestBuildClusterRoutes_EmptyServicesIncludesAll`, `TestBuildPredastoreNodes_FiltersByPredastore`, `TestBuildPredastoreNodes_EmptyServicesIncludesAll`, `TestBuildClusterRoutes_MixedServicesAndEmpty`.
|
||||
|
||||
Extend `formation.NodeInfo` to include the services list, so the formation server knows what each node provides:
|
||||
**Files modified:**
|
||||
- `hive/formation/formation.go` — `Services` field on `NodeInfo`
|
||||
- `hive/formation/helpers.go` — `hasService()` helper, filtered `BuildClusterRoutes()` and `BuildPredastoreNodes()`
|
||||
- `hive/formation/helpers_test.go` — 7 new tests
|
||||
|
||||
```go
|
||||
type NodeInfo struct {
|
||||
Name string `json:"name"`
|
||||
BindIP string `json:"bind_ip"`
|
||||
ClusterIP string `json:"cluster_ip"`
|
||||
Region string `json:"region"`
|
||||
AZ string `json:"az"`
|
||||
Port int `json:"port"`
|
||||
Services []string `json:"services"` // NEW
|
||||
}
|
||||
```
|
||||
#### 0.3 Config template and CLI flags — Done
|
||||
|
||||
This allows `BuildClusterRoutes()` to only include nodes with `"nats"` in their services list for NATS cluster routes, and `BuildPredastoreNodes()` to only include nodes with `"predastore"`.
|
||||
Added `Services []string` to `ConfigSettings` in `hive/admin/admin.go`.
|
||||
|
||||
**Files:**
|
||||
- `hive/formation/formation.go` — add `Services` to `NodeInfo` and `JoinRequest`
|
||||
- `hive/formation/helpers.go` — filter by capability in `BuildClusterRoutes()` and `BuildPredastoreNodes()`
|
||||
Added `services` template rendering in `cmd/hive/cmd/templates/hive.toml` (after the `az` line).
|
||||
|
||||
#### 0.3 Capability-aware start/stop scripts
|
||||
Added `--services` StringSlice flag to both `adminInitCmd` and `adminJoinCmd` in `cmd/hive/cmd/admin.go`. The flag is read in `runAdminInit`, `runAdminInitMultiNode`, and `runAdminJoin`, and passed through to `ConfigSettings`.
|
||||
|
||||
Refactor `start-dev.sh` and `stop-dev.sh` to read the `services` list from `hive.toml` and only start/stop configured services.
|
||||
**Files modified:**
|
||||
- `hive/admin/admin.go` — `Services` field on `ConfigSettings`
|
||||
- `cmd/hive/cmd/templates/hive.toml` — `services` template line
|
||||
- `cmd/hive/cmd/admin.go` — `--services` flag on init/join, wiring to `ConfigSettings`
|
||||
|
||||
**Service dependency graph** (start order):
|
||||
#### 0.4 Capability-aware start/stop scripts — Done
|
||||
|
||||
```
|
||||
nats → (none, starts first)
|
||||
predastore → nats
|
||||
viperblock → nats
|
||||
daemon → nats, viperblock
|
||||
awsgw → nats, daemon
|
||||
ui → awsgw
|
||||
```
|
||||
Added `parse_services()` and `has_service()` bash functions to both `scripts/start-dev.sh` and `scripts/stop-dev.sh`.
|
||||
|
||||
For a compute-only node (`services = ["daemon", "viperblock"]`):
|
||||
1. Skip NATS (not local) — but verify remote NATS is reachable
|
||||
2. Skip Predastore (not local) — but verify remote Predastore is reachable
|
||||
3. Start Viperblock (local)
|
||||
4. Start Daemon (local)
|
||||
5. Skip AWSGW (not local)
|
||||
6. Skip UI (not local)
|
||||
`parse_services()` reads the `services` line from `hive.toml` and defaults to all services if not set. All 6 service start/stop blocks are wrapped in `has_service` conditionals with skip messages for non-local services.
|
||||
|
||||
**Implementation in start-dev.sh:**
|
||||
The stop script maps between process names and config service names (e.g., `hive` → `daemon`, `hive-ui` → `ui`).
|
||||
|
||||
```bash
|
||||
# Read services from hive.toml
|
||||
SERVICES=$(parse_services_from_config "$CONFIG_DIR/hive.toml")
|
||||
# Defaults to "nats predastore viperblock daemon awsgw ui" if not set
|
||||
**Files modified:**
|
||||
- `scripts/start-dev.sh` — `parse_services()`, `has_service()`, conditional service startup
|
||||
- `scripts/stop-dev.sh` — `parse_services()`, `has_service()` (with name mapping), conditional service stop
|
||||
|
||||
has_service() { echo "$SERVICES" | grep -qw "$1"; }
|
||||
#### 0.5 Health endpoint and cluster state KV — Done
|
||||
|
||||
# Start local services in dependency order
|
||||
if has_service "nats"; then
|
||||
start_service "nats" "$NATS_CMD"
|
||||
else
|
||||
# Verify remote NATS is reachable
|
||||
NATS_HOST=$(parse_nats_host "$CONFIG_DIR/hive.toml")
|
||||
check_service "NATS (remote)" "${NATS_HOST%%:*}" "${NATS_HOST##*:}"
|
||||
fi
|
||||
**Health endpoint**: Updated the `/health` HTTP handler in `daemon.go` to include `Services` (from config) and `ServiceHealth` (per-service status map). For local services, status is `"ok"`. For remote NATS dependency (when nats is not a local service), checks `natsConn.IsConnected()` and reports `"remote_ok"` or `"remote_unreachable"`.
|
||||
|
||||
if has_service "predastore"; then
|
||||
start_service "predastore" "$PREDASTORE_CMD"
|
||||
else
|
||||
# Verify remote Predastore is reachable
|
||||
check_service "Predastore (remote)" "$PREDASTORE_HOST" "$PREDASTORE_PORT"
|
||||
fi
|
||||
**Cluster state KV bucket**: Added `hive-cluster-state` bucket alongside the existing `hive-instance-state` bucket. New `clusterKV` field on `JetStreamManager`. `InitClusterStateBucket()` creates the bucket with 1-hour TTL and history of 1. Called from `initJetStream()` after `InitKVBucket()`.
|
||||
|
||||
# ... etc for each service
|
||||
```
|
||||
**Service manifest**: `WriteServiceManifest()` writes `node.{nodeID}.services` to the cluster state KV on daemon startup, containing the node's services, NATS host, predastore host, and timestamp.
|
||||
|
||||
**stop-dev.sh** — reverse logic: only stop services that were started locally. Currently it calls `./bin/hive service <name> stop` for all services, which fails silently for services that aren't running — but it's noisy and slow.
|
||||
`UpdateReplicas()` updated to also update the cluster-state bucket replicas alongside the instance-state bucket.
|
||||
|
||||
**Files:**
|
||||
- `scripts/start-dev.sh` — add config parsing, conditional service startup, remote dependency checks
|
||||
- `scripts/stop-dev.sh` — add config parsing, only stop local services
|
||||
**Files modified:**
|
||||
- `hive/daemon/daemon.go` — health endpoint with services/serviceHealth, `InitClusterStateBucket()` call in `initJetStream()`, `WriteServiceManifest()` call in `Start()`
|
||||
- `hive/daemon/jetstream.go` — `ClusterStateBucket` const, `clusterKV` field, `InitClusterStateBucket()`, `WriteServiceManifest()`, updated `UpdateReplicas()`
|
||||
|
||||
#### 0.4 Health endpoint includes capabilities
|
||||
|
||||
Extend `NodeHealthResponse` to advertise what the node runs:
|
||||
|
||||
```go
|
||||
type NodeHealthResponse struct {
|
||||
Node string `json:"node"`
|
||||
Status string `json:"status"`
|
||||
ConfigHash string `json:"config_hash"`
|
||||
Epoch uint64 `json:"epoch"`
|
||||
Uptime int64 `json:"uptime"`
|
||||
Services []string `json:"services"` // NEW: what this node runs
|
||||
ServiceHealth map[string]string `json:"service_health"` // NEW: per-service status
|
||||
}
|
||||
```
|
||||
|
||||
The `ServiceHealth` map only includes services this node runs:
|
||||
|
||||
```json
|
||||
{
|
||||
"node": "node4",
|
||||
"status": "running",
|
||||
"services": ["daemon", "viperblock"],
|
||||
"service_health": {
|
||||
"daemon": "ok",
|
||||
"viperblock": "ok",
|
||||
"nats": "remote_ok",
|
||||
"predastore": "remote_ok"
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
For local services, the daemon checks the process is running. For remote dependencies (NATS, Predastore), it checks connectivity. This gives operators and the recovery coordinator a complete picture of node health.
|
||||
|
||||
**Files:**
|
||||
- `hive/config/config.go` — update `NodeHealthResponse` struct
|
||||
- `hive/daemon/daemon.go` — update health handler to include capabilities and per-service health
|
||||
- `hive/daemon/daemon_handlers.go` — update NATS health handler similarly
|
||||
|
||||
#### 0.5 Cluster service map in JetStream KV
|
||||
|
||||
To support runtime service discovery (not just config-time), each daemon writes its node's service manifest to JetStream KV on startup:
|
||||
|
||||
```
|
||||
Bucket: hive-cluster-state (new, separate from hive-instance-state)
|
||||
Key: node.{nodeID}.services
|
||||
Value: {"node": "node4", "services": ["daemon", "viperblock"], "nats_host": "10.1.3.170:4222", "predastore_host": "10.1.3.170:8443"}
|
||||
TTL: 1 hour (auto-cleanup of dead nodes)
|
||||
History: 1
|
||||
```
|
||||
|
||||
This enables:
|
||||
- Recovery coordinator to find which nodes can accept VMs (nodes with `"daemon"` service)
|
||||
- Any node to discover where NATS/Predastore are running
|
||||
- Operators to query cluster topology via `hive admin status`
|
||||
|
||||
**Design decisions on KV buckets:**
|
||||
|
||||
| Bucket | Purpose | Replicas | History | TTL |
|
||||
|--------|---------|----------|---------|-----|
|
||||
| `hive-instance-state` | VM state, stopped instances | cluster size | 1 | none |
|
||||
| `hive-cluster-state` | Heartbeats, service maps, shutdown markers, recovery leases | cluster size | 1 | 1 hour |
|
||||
|
||||
Separating the buckets keeps ephemeral data (heartbeats, leases) from durable data (instance state). The 1-hour TTL on `hive-cluster-state` auto-cleans dead node heartbeats and stale service maps without manual intervention. At 100 nodes this bucket holds ~100 keys of ~300 bytes each — trivially small. The TTL prevents unbounded growth from nodes that leave the cluster permanently.
|
||||
|
||||
**Files:**
|
||||
- `hive/daemon/jetstream.go` — new `InitClusterStateBucket()`, `WriteServiceManifest()`, `ListServiceManifests()`
|
||||
|
||||
### Phase 1: Self-recovery improvements (single-node correctness)
|
||||
### Phase 1: Self-recovery improvements (single-node correctness) — Complete
|
||||
|
||||
These changes fix the immediate restart bugs without requiring cross-node coordination.
|
||||
|
||||
#### 1.1 Cluster readiness gate (capability-aware)
|
||||
#### 1.1 Cluster readiness gate — Done
|
||||
|
||||
Before `restoreInstances()`, add a readiness check that waits for dependent services. The check varies based on the node's capabilities:
|
||||
Added `waitForClusterReady()` method to `Daemon`, called in `Start()` just before `restoreInstances()`. Uses a 2-minute max wait with 2-second polling interval.
|
||||
|
||||
**Full infrastructure node** (`services = ["nats", "predastore", "viperblock", "daemon", ...]`):
|
||||
```
|
||||
NATS connected (already done)
|
||||
→ JetStream KV initialized (already done)
|
||||
→ Local Viperblock healthy (health check with retry)
|
||||
→ Local Predastore healthy (health check with retry)
|
||||
→ Peer daemons discoverable (at least N-1 NATS routes connected)
|
||||
→ Begin recovery
|
||||
```
|
||||
Two checks run in sequence:
|
||||
1. **Viperblock** — `checkViperblockReady()` verifies NATS connection is live. Initially implemented as a NATS request to `ebs.{node}.health`, but viperblock has no health NATS topic — this caused a 2-minute hang during E2E tests. Fixed to simply check `d.natsConn.IsConnected()` since viperblock runs alongside the daemon and subscribes to NATS.
|
||||
2. **Predastore** — `checkPredastoreReady()` does a TCP dial to the predastore host:port from config with a 3-second timeout. Skipped if no predastore host is configured.
|
||||
|
||||
**Compute-only node** (`services = ["daemon", "viperblock"]`):
|
||||
```
|
||||
Remote NATS reachable (TCP connect with retry)
|
||||
→ JetStream KV initialized (already done — connects to remote NATS)
|
||||
→ Local Viperblock healthy (health check with retry)
|
||||
→ Remote Predastore reachable (TCP connect with retry)
|
||||
→ Begin recovery
|
||||
```
|
||||
If all checks pass, recovery proceeds immediately. If the 2-minute timeout expires, recovery proceeds anyway with a warning log.
|
||||
|
||||
The `waitForClusterReady()` method reads the node's `services` list and builds its dependency checks accordingly. Local services get health endpoint checks; remote services get TCP connectivity checks.
|
||||
**Finding**: Viperblock does not expose a health endpoint via NATS. Any future viperblock health check should use TCP connectivity or an HTTP endpoint, not NATS request-reply.
|
||||
|
||||
**Files:**
|
||||
- `hive/daemon/daemon.go` — new `waitForClusterReady()`, called from `Start()` between lines 523-547
|
||||
**Files modified:**
|
||||
- `hive/daemon/daemon.go` — `waitForClusterReady()`, `checkViperblockReady()`, `checkPredastoreReady()`, added `"net"` import
|
||||
|
||||
#### 1.2 Recovery throttling with semaphore
|
||||
#### 1.2 Recovery throttling with semaphore — Done
|
||||
|
||||
Replace the unbounded loop in `restoreInstances()` with a worker pool. Limit concurrent VM launches to `maxConcurrentRecovery` (default: 2 per node).
|
||||
Refactored `restoreInstances()` into two phases:
|
||||
|
||||
Recovery order priority:
|
||||
1. Reconnect to still-running QEMU processes (instant, no I/O)
|
||||
2. Finalize transitional states (stopping → stopped, shutting-down → terminated)
|
||||
3. Relaunch crashed VMs (pending/running with dead QEMU) — these hit storage
|
||||
**Phase 1 (sequential)**: Iterates all instances from KV state. For each instance:
|
||||
- Reconnects to still-running QEMU processes (checks PID file, instant, no I/O)
|
||||
- Finalizes transitional states (stopping → stopped, shutting-down → terminated)
|
||||
- Collects instances that need relaunching into a `toLaunch` list
|
||||
|
||||
**Files:**
|
||||
- `hive/daemon/daemon.go` — refactor `restoreInstances()` to use a semaphore for step 3
|
||||
**Phase 2 (throttled)**: Launches crashed VMs using a semaphore channel with `maxConcurrentRecovery = 2`. Each launch runs in a goroutine that acquires the semaphore before proceeding and releases on completion. A `sync.WaitGroup` ensures all launches complete before `restoreInstances()` returns.
|
||||
|
||||
#### 1.3 Clean shutdown marker
|
||||
This prevents thundering herd on storage during recovery while keeping the lightweight reconnect/finalize steps sequential.
|
||||
|
||||
On graceful shutdown, write a `shutdown.{nodeID}` key to `hive-cluster-state` KV with timestamp and instance count. On startup, check for this marker:
|
||||
**Files modified:**
|
||||
- `hive/daemon/daemon.go` — refactored `restoreInstances()` with two-phase approach and semaphore
|
||||
|
||||
- **Marker present**: Clean shutdown. Trust KV state. Delete marker and proceed.
|
||||
- **Marker absent**: Crash or power loss. Log a warning, do extra PID file validation, wait slightly longer for QEMU processes to potentially still be exiting.
|
||||
#### 1.3 Clean shutdown marker — Done
|
||||
|
||||
**Files:**
|
||||
- `hive/daemon/jetstream.go` — new `WriteShutdownMarker()` / `ReadShutdownMarker()` / `DeleteShutdownMarker()`
|
||||
- `hive/daemon/daemon.go` — write marker in `setupShutdown()`, read/delete in `restoreInstances()`
|
||||
Three new methods on `JetStreamManager`:
|
||||
- `WriteShutdownMarker(nodeID)` — writes `shutdown.{nodeID}` to `hive-cluster-state` KV with node name and timestamp
|
||||
- `ReadShutdownMarker(nodeID)` — checks if a shutdown marker exists for the node
|
||||
- `DeleteShutdownMarker(nodeID)` — deletes the marker (tolerates not-found)
|
||||
|
||||
**Write**: In `setupShutdown()`, the shutdown marker is written to cluster state KV before `WriteState()` persists instance state. This ensures the marker is present if the state write also succeeds.
|
||||
|
||||
**Read**: At the start of `restoreInstances()`, the marker is checked. If present: log "Clean shutdown marker found, trusting KV state", delete the marker, proceed normally. If absent: log a warning about possible crash recovery, sleep 3 seconds to let orphaned QEMU processes finish exiting, then proceed with careful PID validation.
|
||||
|
||||
**Files modified:**
|
||||
- `hive/daemon/jetstream.go` — `WriteShutdownMarker()`, `ReadShutdownMarker()`, `DeleteShutdownMarker()`
|
||||
- `hive/daemon/daemon.go` — marker write in `setupShutdown()`, marker check/delete in `restoreInstances()`
|
||||
|
||||
#### 1.4 Service manifest on startup — Done
|
||||
|
||||
Added `WriteServiceManifest()` to `JetStreamManager`, which writes `node.{nodeID}.services` to the cluster state KV containing the node's service list, NATS host, predastore host, and timestamp.
|
||||
|
||||
Called in `Start()` after `initJetStream()` succeeds. Failure is non-fatal (logged as warning).
|
||||
|
||||
**Files modified:**
|
||||
- `hive/daemon/jetstream.go` — `WriteServiceManifest()`
|
||||
- `hive/daemon/daemon.go` — call in `Start()`
|
||||
|
||||
### Phase 2: Cross-node coordination
|
||||
|
||||
@@ -457,70 +333,91 @@ This reads from the `hive-cluster-state` KV bucket (heartbeats + service manifes
|
||||
**Files:**
|
||||
- `cmd/hive/cmd/admin.go` — new `status` subcommand
|
||||
|
||||
## Files to Modify
|
||||
## Files Modified (Phase 0 + 1)
|
||||
|
||||
| File | Changes | Phase |
|
||||
|------|---------|-------|
|
||||
| `hive/config/config.go` | Added `Services` field, `AllServices`, `HasService()`, `GetServices()`, updated `NodeHealthResponse` with `Services` and `ServiceHealth` | 0.1, 0.5 |
|
||||
| `hive/config/config_test.go` | 5 new tests for `HasService` and `GetServices` | 0.1 |
|
||||
| `hive/admin/admin.go` | Added `Services []string` to `ConfigSettings` | 0.3 |
|
||||
| `cmd/hive/cmd/templates/hive.toml` | Added `services` template rendering | 0.3 |
|
||||
| `cmd/hive/cmd/admin.go` | Added `--services` flag to init/join, wired to `ConfigSettings` | 0.3 |
|
||||
| `hive/formation/formation.go` | Added `Services` to `NodeInfo` | 0.2 |
|
||||
| `hive/formation/helpers.go` | Added `hasService()`, filtered `BuildClusterRoutes()` by nats, `BuildPredastoreNodes()` by predastore | 0.2 |
|
||||
| `hive/formation/helpers_test.go` | 7 new tests for capability-filtered routing and predastore node building | 0.2 |
|
||||
| `scripts/start-dev.sh` | Added `parse_services()`, `has_service()`, conditional service startup | 0.4 |
|
||||
| `scripts/stop-dev.sh` | Added `parse_services()`, `has_service()` (with name mapping), conditional stop | 0.4 |
|
||||
| `hive/daemon/daemon.go` | Updated health endpoint, `waitForClusterReady()`, `checkViperblockReady()`, `checkPredastoreReady()`, refactored `restoreInstances()` with semaphore, shutdown marker in `setupShutdown()`, service manifest write in `Start()` | 0.5, 1.1–1.4 |
|
||||
| `hive/daemon/jetstream.go` | `ClusterStateBucket`, `clusterKV` field, `InitClusterStateBucket()`, `WriteShutdownMarker()`, `ReadShutdownMarker()`, `DeleteShutdownMarker()`, `WriteServiceManifest()`, updated `UpdateReplicas()` | 0.5, 1.3, 1.4 |
|
||||
|
||||
## Files to Modify (Phase 2 + 3)
|
||||
|
||||
| File | Changes |
|
||||
|------|---------|
|
||||
| `hive/config/config.go` | Add `Services` field to `Config`, helper methods (`HasService`, etc.), update `NodeHealthResponse` |
|
||||
| `cmd/hive/cmd/templates/hive.toml` | Add `services` field to template |
|
||||
| `cmd/hive/cmd/admin.go` | Add `--services` flag to init/join, add `maintenance` and `status` subcommands |
|
||||
| `hive/formation/formation.go` | Add `Services` to `NodeInfo` |
|
||||
| `hive/formation/helpers.go` | Filter `BuildClusterRoutes` and `BuildPredastoreNodes` by capability |
|
||||
| `scripts/start-dev.sh` | Read services from config, conditional service startup, remote dependency checks |
|
||||
| `scripts/stop-dev.sh` | Read services from config, only stop local services |
|
||||
| `hive/daemon/daemon.go` | `waitForClusterReady()` (capability-aware), refactor `restoreInstances()` with semaphore, heartbeat/recovery integration |
|
||||
| `hive/daemon/jetstream.go` | New `hive-cluster-state` bucket, shutdown marker, heartbeat, recovery lease, service manifest, maintenance mode |
|
||||
| `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/state.go` | Minor — shutdown marker integration with state transitions |
|
||||
| `hive/daemon/daemon_handlers.go` | Update NATS health handler with capabilities and per-service health |
|
||||
| `hive/daemon/daemon.go` | Heartbeat/recovery integration |
|
||||
| `cmd/hive/cmd/admin.go` | `maintenance` and `status` subcommands |
|
||||
|
||||
## Implementation Order
|
||||
|
||||
Phase 0 is the prerequisite — all subsequent phases depend on nodes knowing their capabilities. Phase 1 fixes immediate restart bugs. Phases 2-3 add cross-node resilience and tooling.
|
||||
|
||||
**Phase 0** — node capabilities (foundation):
|
||||
1. `Services` field in Config + helpers
|
||||
2. Formation carries capabilities
|
||||
3. Capability-aware start/stop scripts
|
||||
4. Health endpoint includes capabilities
|
||||
5. Cluster service map in JetStream KV
|
||||
**Phase 0** — node capabilities (foundation) — **Complete**:
|
||||
1. [x] `Services` field in Config + helpers
|
||||
2. [x] Formation carries capabilities
|
||||
3. [x] Config template and CLI flags
|
||||
4. [x] Capability-aware start/stop scripts
|
||||
5. [x] Health endpoint includes capabilities + cluster state KV bucket + service manifest
|
||||
|
||||
**Phase 1** — self-recovery (fixes immediate restart bugs):
|
||||
6. Cluster readiness gate (capability-aware)
|
||||
7. Recovery throttling (semaphore in `restoreInstances`)
|
||||
8. Clean shutdown marker
|
||||
**Phase 1** — self-recovery (fixes immediate restart bugs) — **Complete**:
|
||||
6. [x] Cluster readiness gate
|
||||
7. [x] Recovery throttling (semaphore in `restoreInstances`)
|
||||
8. [x] Clean shutdown marker
|
||||
9. [x] Service manifest on startup
|
||||
|
||||
**Phase 2** — cross-node failure recovery:
|
||||
9. Heartbeat publishing (with capabilities)
|
||||
10. Peer failure detection
|
||||
11. Leader-elected recovery coordinator (capability-aware placement)
|
||||
**Phase 2** — cross-node failure recovery — **Planned**:
|
||||
10. Heartbeat publishing (with capabilities)
|
||||
11. Peer failure detection
|
||||
12. Leader-elected recovery coordinator (capability-aware placement)
|
||||
|
||||
**Phase 3** — operational tooling:
|
||||
12. Maintenance mode flag
|
||||
13. Recovery status endpoints
|
||||
14. Admin cluster status command
|
||||
**Phase 3** — operational tooling — **Planned**:
|
||||
13. Maintenance mode flag
|
||||
14. Recovery status endpoints
|
||||
15. Admin cluster status command
|
||||
|
||||
## Testing
|
||||
|
||||
### Unit tests
|
||||
- Config `HasService()` helper methods with various service lists
|
||||
- Config defaults to all services when `Services` is empty (backward compat)
|
||||
- `waitForClusterReady()` with different capability configurations
|
||||
- `restoreInstances()` with mocked JetStream: verify semaphore limits concurrent launches
|
||||
- Shutdown marker: write on shutdown, verify presence on clean restart, verify absence simulates crash
|
||||
### Phase 0 + 1 verification — Passed
|
||||
|
||||
| Check | Result |
|
||||
|-------|--------|
|
||||
| `make test` (all unit tests) | Passed |
|
||||
| `make preflight` (format + vet + security + tests) | Passed |
|
||||
| `make test-docker-single` (single-node E2E) | Passed — all 9 phases |
|
||||
| `make test-docker-multi` (multi-node E2E, 3 nodes) | Passed — all phases including cluster health, cross-node operations |
|
||||
| Backward compatibility (no `services` field in config) | Verified — defaults to all services |
|
||||
|
||||
**New unit tests added:**
|
||||
- `hive/config/config_test.go` — `HasService` (explicit, empty/default, unknown), `GetServices` (default, explicit)
|
||||
- `hive/formation/helpers_test.go` — `hasService` (empty=all, explicit), `BuildClusterRoutes` (filters by nats, empty=all, mixed), `BuildPredastoreNodes` (filters by predastore, empty=all)
|
||||
|
||||
**Bug found during E2E**: `checkViperblockReady()` initially sent a NATS request to `ebs.{node}.health`, but viperblock does not subscribe to any health NATS topic. This caused the daemon to block in `waitForClusterReady()` for the full 2-minute timeout on every startup. Fixed by checking `natsConn.IsConnected()` instead.
|
||||
|
||||
### Phase 2 + 3 tests (planned)
|
||||
|
||||
#### Unit tests
|
||||
- Recovery coordinator: test leader election, capability-filtered VM assignment, epoch-based fencing
|
||||
- Heartbeat: test staleness detection thresholds
|
||||
|
||||
### Integration tests (multi-node Docker)
|
||||
- Start 3-node cluster (all services), launch 4 VMs, stop all nodes, restart — verify all 4 VMs recover exactly once
|
||||
- Start 3+1 cluster (3 full + 1 compute-only), verify compute-only node connects to remote NATS/Predastore and launches VMs
|
||||
- Kill one node (SIGKILL), verify surviving nodes eventually recover its VMs after grace period, respecting capabilities
|
||||
- `start-dev.sh` on a compute-only node: verify it skips NATS/Predastore startup but checks remote connectivity
|
||||
#### 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
|
||||
|
||||
### Manual verification
|
||||
- `make test` passes after each phase
|
||||
- Deploy to 3-node test cluster (10.1.3.170-172), run stop/start cycle, check `~/hive/logs/` for race conditions
|
||||
#### 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
|
||||
|
||||
## Future Work
|
||||
|
||||
@@ -42,6 +42,9 @@ type ConfigSettings struct {
|
||||
|
||||
// Predastore multi-node
|
||||
PredastoreNodeID int
|
||||
|
||||
// Node capabilities
|
||||
Services []string
|
||||
}
|
||||
|
||||
// PredastoreNodeConfig describes a single Predastore node for multi-node config generation.
|
||||
|
||||
+39
-10
@@ -21,11 +21,12 @@ type ClusterConfig struct {
|
||||
// Config holds all configuration for the application
|
||||
type Config struct {
|
||||
// Node config
|
||||
Node string `mapstructure:"node"`
|
||||
Host string `mapstructure:"host"` // Unique hostname or IP of this node
|
||||
Region string `mapstructure:"region"`
|
||||
AZ string `mapstructure:"az"`
|
||||
DataDir string `mapstructure:"data_dir"`
|
||||
Node string `mapstructure:"node"`
|
||||
Host string `mapstructure:"host"` // Unique hostname or IP of this node
|
||||
Region string `mapstructure:"region"`
|
||||
AZ string `mapstructure:"az"`
|
||||
DataDir string `mapstructure:"data_dir"`
|
||||
Services []string `mapstructure:"services"` // Which services this node runs locally
|
||||
|
||||
Daemon DaemonConfig `mapstructure:"daemon"`
|
||||
NATS NATSConfig `mapstructure:"nats"`
|
||||
@@ -40,6 +41,32 @@ type Config struct {
|
||||
WalDir string `mapstructure:"wal_dir"`
|
||||
}
|
||||
|
||||
// AllServices is the default service list when Services is empty (backward compat).
|
||||
var AllServices = []string{"nats", "predastore", "viperblock", "daemon", "awsgw", "ui"}
|
||||
|
||||
// HasService reports whether the node runs the named service.
|
||||
// An empty Services list means all services (backward compat).
|
||||
func (c Config) HasService(name string) bool {
|
||||
services := c.Services
|
||||
if len(services) == 0 {
|
||||
services = AllServices
|
||||
}
|
||||
for _, s := range services {
|
||||
if s == name {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// GetServices returns the configured service list, defaulting to AllServices.
|
||||
func (c Config) GetServices() []string {
|
||||
if len(c.Services) == 0 {
|
||||
return AllServices
|
||||
}
|
||||
return c.Services
|
||||
}
|
||||
|
||||
type AWSGWConfig struct {
|
||||
Host string `mapstructure:"host"`
|
||||
TLSKey string `mapstructure:"tlskey"`
|
||||
@@ -209,11 +236,13 @@ type NodeJoinResponse struct {
|
||||
}
|
||||
|
||||
type NodeHealthResponse struct {
|
||||
Node string `json:"node"`
|
||||
Status string `json:"status"`
|
||||
ConfigHash string `json:"config_hash"`
|
||||
Epoch uint64 `json:"epoch"`
|
||||
Uptime int64 `json:"uptime"`
|
||||
Node string `json:"node"`
|
||||
Status string `json:"status"`
|
||||
ConfigHash string `json:"config_hash"`
|
||||
Epoch uint64 `json:"epoch"`
|
||||
Uptime int64 `json:"uptime"`
|
||||
Services []string `json:"services"`
|
||||
ServiceHealth map[string]string `json:"service_health,omitempty"`
|
||||
}
|
||||
|
||||
// LoadConfig loads the configuration from file and environment variables
|
||||
|
||||
@@ -219,6 +219,44 @@ expected_nodes = 3
|
||||
assert.Equal(t, 3, n.AWSGW.ExpectedNodes)
|
||||
}
|
||||
|
||||
// Tests for HasService / GetServices
|
||||
|
||||
func TestHasService_ExplicitList(t *testing.T) {
|
||||
c := Config{Services: []string{"nats", "daemon"}}
|
||||
|
||||
assert.True(t, c.HasService("nats"))
|
||||
assert.True(t, c.HasService("daemon"))
|
||||
assert.False(t, c.HasService("predastore"))
|
||||
assert.False(t, c.HasService("viperblock"))
|
||||
assert.False(t, c.HasService("ui"))
|
||||
}
|
||||
|
||||
func TestHasService_EmptyListBackwardCompat(t *testing.T) {
|
||||
c := Config{} // no Services set
|
||||
|
||||
// Empty list means all services
|
||||
for _, svc := range AllServices {
|
||||
assert.True(t, c.HasService(svc), "expected %s to be available with empty list", svc)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHasService_UnknownService(t *testing.T) {
|
||||
c := Config{Services: []string{"nats"}}
|
||||
assert.False(t, c.HasService("unknown"))
|
||||
}
|
||||
|
||||
func TestGetServices_DefaultsToAll(t *testing.T) {
|
||||
c := Config{}
|
||||
services := c.GetServices()
|
||||
assert.Equal(t, AllServices, services)
|
||||
}
|
||||
|
||||
func TestGetServices_ExplicitList(t *testing.T) {
|
||||
c := Config{Services: []string{"nats", "predastore"}}
|
||||
services := c.GetServices()
|
||||
assert.Equal(t, []string{"nats", "predastore"}, services)
|
||||
}
|
||||
|
||||
// Tests for type constants
|
||||
|
||||
func TestNBDTransportConstants(t *testing.T) {
|
||||
|
||||
+150
-9
@@ -9,6 +9,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net"
|
||||
"os"
|
||||
"os/exec"
|
||||
"os/signal"
|
||||
@@ -522,6 +523,18 @@ func (d *Daemon) Start() error {
|
||||
return fmt.Errorf("failed to initialize JetStream: %w", err)
|
||||
}
|
||||
|
||||
// Write service manifest so other nodes know what this node runs
|
||||
if d.jsManager != nil {
|
||||
if err := d.jsManager.WriteServiceManifest(
|
||||
d.node,
|
||||
d.config.GetServices(),
|
||||
d.config.NATS.Host,
|
||||
d.config.Predastore.Host,
|
||||
); err != nil {
|
||||
slog.Warn("Failed to write service manifest", "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Create services before loading/launching instances, since LaunchInstance depends on them
|
||||
store := objectstore.NewS3ObjectStoreFromConfig(d.config.Predastore.Host, d.config.Predastore.Region, d.config.AccessKey, d.config.SecretKey)
|
||||
d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, d.resourceMgr.instanceTypes, d.natsConn, &d.Instances, store)
|
||||
@@ -544,6 +557,7 @@ func (d *Daemon) Start() error {
|
||||
}
|
||||
d.accountService = accountSvc
|
||||
|
||||
d.waitForClusterReady()
|
||||
d.restoreInstances()
|
||||
|
||||
if err := d.subscribeAll(); err != nil {
|
||||
@@ -585,7 +599,11 @@ func (d *Daemon) initJetStream() error {
|
||||
}
|
||||
|
||||
if err == nil {
|
||||
slog.Info("JetStream KV store initialized successfully", "replicas", 1, "attempts", attempt)
|
||||
err = d.jsManager.InitClusterStateBucket()
|
||||
}
|
||||
|
||||
if err == nil {
|
||||
slog.Info("JetStream KV stores initialized successfully", "replicas", 1, "attempts", attempt)
|
||||
lastErr = nil
|
||||
break
|
||||
}
|
||||
@@ -614,6 +632,66 @@ func (d *Daemon) initJetStream() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// waitForClusterReady waits until dependent infrastructure services are reachable
|
||||
// before starting VM recovery. This prevents races where VMs try to mount volumes
|
||||
// before viperblock/predastore are ready.
|
||||
func (d *Daemon) waitForClusterReady() {
|
||||
slog.Info("Waiting for cluster readiness...")
|
||||
maxWait := 2 * time.Minute
|
||||
start := time.Now()
|
||||
interval := 2 * time.Second
|
||||
|
||||
for time.Since(start) < maxWait {
|
||||
ready := true
|
||||
var reason string
|
||||
|
||||
// Viperblock must be reachable (local or remote)
|
||||
if ready && !d.checkViperblockReady() {
|
||||
ready = false
|
||||
reason = "viperblock not ready"
|
||||
}
|
||||
|
||||
// Predastore must be reachable (local or remote)
|
||||
if ready && !d.checkPredastoreReady() {
|
||||
ready = false
|
||||
reason = "predastore not ready"
|
||||
}
|
||||
|
||||
if ready {
|
||||
slog.Info("Cluster readiness check passed", "elapsed", time.Since(start))
|
||||
return
|
||||
}
|
||||
|
||||
slog.Debug("Cluster not ready, waiting...", "reason", reason, "elapsed", time.Since(start))
|
||||
time.Sleep(interval)
|
||||
}
|
||||
|
||||
slog.Warn("Cluster readiness timeout, proceeding with recovery anyway", "maxWait", maxWait)
|
||||
}
|
||||
|
||||
// checkViperblockReady checks if viperblock is reachable by verifying
|
||||
// the NATS connection is up (viperblock subscribes to ebs topics on NATS).
|
||||
func (d *Daemon) checkViperblockReady() bool {
|
||||
if d.natsConn == nil {
|
||||
return false
|
||||
}
|
||||
return d.natsConn.IsConnected()
|
||||
}
|
||||
|
||||
// checkPredastoreReady checks if predastore is reachable via TCP.
|
||||
func (d *Daemon) checkPredastoreReady() bool {
|
||||
host := d.config.Predastore.Host
|
||||
if host == "" {
|
||||
return true // no predastore configured, skip check
|
||||
}
|
||||
conn, err := net.DialTimeout("tcp", host, 3*time.Second)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
conn.Close()
|
||||
return true
|
||||
}
|
||||
|
||||
// migrateStoppedToSharedKV writes a stopped instance to shared KV and removes
|
||||
// it from the local instance map. Returns true if migration succeeded.
|
||||
func (d *Daemon) migrateStoppedToSharedKV(instance *vm.VM) bool {
|
||||
@@ -631,9 +709,29 @@ func (d *Daemon) migrateStoppedToSharedKV(instance *vm.VM) bool {
|
||||
return true
|
||||
}
|
||||
|
||||
// maxConcurrentRecovery limits how many VMs are relaunched in parallel during recovery.
|
||||
const maxConcurrentRecovery = 2
|
||||
|
||||
// restoreInstances loads persisted VM state and re-launches instances that are
|
||||
// neither terminated nor flagged as user-stopped.
|
||||
func (d *Daemon) restoreInstances() {
|
||||
// Check for clean shutdown marker
|
||||
cleanShutdown := false
|
||||
if d.jsManager != nil {
|
||||
if marker, err := d.jsManager.ReadShutdownMarker(d.node); err == nil {
|
||||
cleanShutdown = marker
|
||||
if marker {
|
||||
slog.Info("Clean shutdown marker found, trusting KV state")
|
||||
_ = d.jsManager.DeleteShutdownMarker(d.node)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if !cleanShutdown {
|
||||
slog.Warn("No clean shutdown marker — possible crash recovery, validating QEMU PIDs carefully")
|
||||
time.Sleep(3 * time.Second)
|
||||
}
|
||||
|
||||
err := d.LoadState()
|
||||
if err != nil {
|
||||
slog.Warn("Failed to load state, continuing with empty state", "error", err)
|
||||
@@ -645,6 +743,9 @@ func (d *Daemon) restoreInstances() {
|
||||
// Ensure mutexes and QMP clients are usable after deserialization
|
||||
d.Instances.Mu = sync.Mutex{}
|
||||
|
||||
// Phase 1: Reconnect running QEMU, finalize transitional states, collect VMs to relaunch
|
||||
var toLaunch []*vm.VM
|
||||
|
||||
for i := range d.Instances.VMS {
|
||||
d.Instances.VMS[i].EBSRequests.Mu = sync.Mutex{}
|
||||
d.Instances.VMS[i].QMPClient = &qmp.QMPClient{}
|
||||
@@ -706,10 +807,28 @@ func (d *Daemon) restoreInstances() {
|
||||
slog.Info("Instance was running but QEMU exited, relaunching", "instance", instance.ID)
|
||||
}
|
||||
|
||||
slog.Info("Launching instance", "instance", instance.ID)
|
||||
if err := d.LaunchInstance(instance); err != nil {
|
||||
slog.Error("Failed to launch instance", "instanceId", instance.ID, "err", err)
|
||||
toLaunch = append(toLaunch, instance)
|
||||
}
|
||||
|
||||
// Phase 2: Relaunch crashed VMs with semaphore-based throttling
|
||||
if len(toLaunch) > 0 {
|
||||
slog.Info("Launching instances (recovery)", "count", len(toLaunch), "maxConcurrent", maxConcurrentRecovery)
|
||||
sem := make(chan struct{}, maxConcurrentRecovery)
|
||||
var wg sync.WaitGroup
|
||||
|
||||
for _, instance := range toLaunch {
|
||||
sem <- struct{}{} // acquire
|
||||
wg.Add(1)
|
||||
go func(inst *vm.VM) {
|
||||
defer wg.Done()
|
||||
defer func() { <-sem }() // release
|
||||
slog.Info("Launching instance (recovery)", "instance", inst.ID)
|
||||
if err := d.LaunchInstance(inst); err != nil {
|
||||
slog.Error("Failed to launch instance", "instanceId", inst.ID, "err", err)
|
||||
}
|
||||
}(instance)
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// Persist state after any migrations/removals during restore
|
||||
@@ -835,12 +954,27 @@ func (d *Daemon) ClusterManager() error {
|
||||
})
|
||||
}
|
||||
|
||||
serviceHealth := make(map[string]string)
|
||||
for _, svc := range d.config.GetServices() {
|
||||
serviceHealth[svc] = "ok"
|
||||
}
|
||||
// For remote dependencies, check connectivity
|
||||
if !d.config.HasService("nats") {
|
||||
if d.natsConn != nil && d.natsConn.IsConnected() {
|
||||
serviceHealth["nats"] = "remote_ok"
|
||||
} else {
|
||||
serviceHealth["nats"] = "remote_unreachable"
|
||||
}
|
||||
}
|
||||
|
||||
response := config.NodeHealthResponse{
|
||||
Node: d.node,
|
||||
Status: "running",
|
||||
ConfigHash: configHash,
|
||||
Epoch: d.clusterConfig.Epoch,
|
||||
Uptime: int64(time.Since(d.startTime).Seconds()),
|
||||
Node: d.node,
|
||||
Status: "running",
|
||||
ConfigHash: configHash,
|
||||
Epoch: d.clusterConfig.Epoch,
|
||||
Uptime: int64(time.Since(d.startTime).Seconds()),
|
||||
Services: d.config.GetServices(),
|
||||
ServiceHealth: serviceHealth,
|
||||
}
|
||||
|
||||
return c.JSON(response)
|
||||
@@ -1221,6 +1355,13 @@ func (d *Daemon) setupShutdown() {
|
||||
|
||||
}
|
||||
|
||||
// Write shutdown marker to cluster state KV
|
||||
if d.jsManager != nil {
|
||||
if err := d.jsManager.WriteShutdownMarker(d.node); err != nil {
|
||||
slog.Error("Failed to write shutdown marker", "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Write state to JetStream before closing NATS connection
|
||||
err := d.WriteState()
|
||||
if err != nil {
|
||||
|
||||
+105
-3
@@ -5,6 +5,7 @@ import (
|
||||
"errors"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/mulgadc/hive/hive/vm"
|
||||
"github.com/nats-io/nats.go"
|
||||
@@ -13,6 +14,8 @@ import (
|
||||
const (
|
||||
// InstanceStateBucket is the name of the KV bucket for storing instance state
|
||||
InstanceStateBucket = "hive-instance-state"
|
||||
// ClusterStateBucket is the name of the KV bucket for cluster state (heartbeats, shutdown markers, service maps)
|
||||
ClusterStateBucket = "hive-cluster-state"
|
||||
// InstanceStatePrefix is the key prefix for per-node instance state entries
|
||||
InstanceStatePrefix = "node."
|
||||
// StoppedInstancePrefix is the key prefix for stopped instances in shared KV
|
||||
@@ -21,9 +24,10 @@ const (
|
||||
|
||||
// JetStreamManager manages JetStream KV store operations for instance state
|
||||
type JetStreamManager struct {
|
||||
js nats.JetStreamContext
|
||||
kv nats.KeyValue
|
||||
replicas int
|
||||
js nats.JetStreamContext
|
||||
kv nats.KeyValue // hive-instance-state
|
||||
clusterKV nats.KeyValue // hive-cluster-state
|
||||
replicas int
|
||||
}
|
||||
|
||||
// NewJetStreamManager creates a new JetStreamManager from a NATS connection.
|
||||
@@ -68,6 +72,89 @@ func (m *JetStreamManager) InitKVBucket() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// InitClusterStateBucket initializes the cluster-state KV bucket, creating it if it doesn't exist.
|
||||
func (m *JetStreamManager) InitClusterStateBucket() error {
|
||||
kv, err := m.js.KeyValue(ClusterStateBucket)
|
||||
if err != nil {
|
||||
if errors.Is(err, nats.ErrBucketNotFound) {
|
||||
slog.Info("Creating JetStream KV bucket", "bucket", ClusterStateBucket, "replicas", m.replicas)
|
||||
kv, err = m.js.CreateKeyValue(&nats.KeyValueConfig{
|
||||
Bucket: ClusterStateBucket,
|
||||
Description: "Hive cluster state (heartbeats, shutdown markers, service maps)",
|
||||
History: 1,
|
||||
Replicas: m.replicas,
|
||||
TTL: 1 * time.Hour,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
slog.Info("Connected to existing JetStream KV bucket", "bucket", ClusterStateBucket)
|
||||
}
|
||||
|
||||
m.clusterKV = kv
|
||||
return nil
|
||||
}
|
||||
|
||||
// WriteShutdownMarker writes a shutdown marker for the given node to the cluster-state KV.
|
||||
func (m *JetStreamManager) WriteShutdownMarker(nodeID string) error {
|
||||
if m.clusterKV == nil {
|
||||
return errors.New("cluster state KV not initialized")
|
||||
}
|
||||
data, _ := json.Marshal(map[string]any{
|
||||
"node": nodeID,
|
||||
"timestamp": time.Now().UTC().Format(time.RFC3339),
|
||||
})
|
||||
_, err := m.clusterKV.Put("shutdown."+nodeID, data)
|
||||
return err
|
||||
}
|
||||
|
||||
// ReadShutdownMarker checks if a clean shutdown marker exists for the given node.
|
||||
func (m *JetStreamManager) ReadShutdownMarker(nodeID string) (bool, error) {
|
||||
if m.clusterKV == nil {
|
||||
return false, errors.New("cluster state KV not initialized")
|
||||
}
|
||||
_, err := m.clusterKV.Get("shutdown." + nodeID)
|
||||
if errors.Is(err, nats.ErrKeyNotFound) {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// DeleteShutdownMarker removes the shutdown marker for the given node.
|
||||
func (m *JetStreamManager) DeleteShutdownMarker(nodeID string) error {
|
||||
if m.clusterKV == nil {
|
||||
return errors.New("cluster state KV not initialized")
|
||||
}
|
||||
err := m.clusterKV.Delete("shutdown." + nodeID)
|
||||
if err != nil && !errors.Is(err, nats.ErrKeyNotFound) {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// WriteServiceManifest writes the service manifest for the given node to the cluster-state KV.
|
||||
func (m *JetStreamManager) WriteServiceManifest(nodeID string, services []string, natsHost, predastoreHost string) error {
|
||||
if m.clusterKV == nil {
|
||||
return errors.New("cluster state KV not initialized")
|
||||
}
|
||||
data, _ := json.Marshal(map[string]any{
|
||||
"node": nodeID,
|
||||
"services": services,
|
||||
"nats_host": natsHost,
|
||||
"predastore_host": predastoreHost,
|
||||
"timestamp": time.Now().UTC().Format(time.RFC3339),
|
||||
})
|
||||
_, err := m.clusterKV.Put("node."+nodeID+".services", data)
|
||||
return err
|
||||
}
|
||||
|
||||
// WriteState writes the instance state to the KV store for the given node.
|
||||
// It acquires instances.Mu internally.
|
||||
func (m *JetStreamManager) WriteState(nodeID string, instances *vm.Instances) error {
|
||||
@@ -180,6 +267,21 @@ func (m *JetStreamManager) UpdateReplicas(newReplicas int) error {
|
||||
|
||||
m.replicas = newReplicas
|
||||
slog.Info("Updated JetStream KV bucket replicas", "bucket", InstanceStateBucket, "oldReplicas", currentReplicas, "newReplicas", newReplicas)
|
||||
|
||||
// Also update the cluster-state bucket if it exists
|
||||
if m.clusterKV != nil {
|
||||
clusterStreamName := "KV_" + ClusterStateBucket
|
||||
clusterInfo, err := m.js.StreamInfo(clusterStreamName)
|
||||
if err == nil && clusterInfo.Config.Replicas < newReplicas {
|
||||
clusterInfo.Config.Replicas = newReplicas
|
||||
if _, err := m.js.UpdateStream(&clusterInfo.Config); err != nil {
|
||||
slog.Warn("Failed to update cluster-state bucket replicas", "error", err)
|
||||
} else {
|
||||
slog.Info("Updated cluster-state bucket replicas", "bucket", ClusterStateBucket, "newReplicas", newReplicas)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -13,12 +13,13 @@ import (
|
||||
|
||||
// NodeInfo describes a node participating in cluster formation.
|
||||
type NodeInfo struct {
|
||||
Name string `json:"name"`
|
||||
BindIP string `json:"bind_ip"`
|
||||
ClusterIP string `json:"cluster_ip"`
|
||||
Region string `json:"region"`
|
||||
AZ string `json:"az"`
|
||||
Port int `json:"port"`
|
||||
Name string `json:"name"`
|
||||
BindIP string `json:"bind_ip"`
|
||||
ClusterIP string `json:"cluster_ip"`
|
||||
Region string `json:"region"`
|
||||
AZ string `json:"az"`
|
||||
Port int `json:"port"`
|
||||
Services []string `json:"services,omitempty"`
|
||||
}
|
||||
|
||||
// JoinRequest is the payload POSTed by joining nodes.
|
||||
|
||||
@@ -6,11 +6,32 @@ import (
|
||||
"github.com/mulgadc/hive/hive/admin"
|
||||
)
|
||||
|
||||
// hasService reports whether the node's service list includes name.
|
||||
// An empty list means all services (backward compat).
|
||||
func hasService(services []string, name string) bool {
|
||||
if len(services) == 0 {
|
||||
return true // backward compat: empty = all services
|
||||
}
|
||||
for _, s := range services {
|
||||
if s == name {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// BuildClusterRoutes returns a sorted list of "IP:4248" NATS cluster routes
|
||||
// for each node. If ClusterIP is set it is used, otherwise BindIP.
|
||||
// Nodes are sorted by name for deterministic ordering.
|
||||
// for nodes that run the "nats" service. If ClusterIP is set it is used,
|
||||
// otherwise BindIP. Nodes are sorted by name for deterministic ordering.
|
||||
func BuildClusterRoutes(nodes map[string]NodeInfo) []string {
|
||||
sorted := sortedNodes(nodes)
|
||||
// Filter to nodes that run NATS
|
||||
natsNodes := make(map[string]NodeInfo)
|
||||
for k, n := range nodes {
|
||||
if hasService(n.Services, "nats") {
|
||||
natsNodes[k] = n
|
||||
}
|
||||
}
|
||||
sorted := sortedNodes(natsNodes)
|
||||
routes := make([]string, len(sorted))
|
||||
for i, n := range sorted {
|
||||
ip := n.ClusterIP
|
||||
@@ -23,9 +44,17 @@ func BuildClusterRoutes(nodes map[string]NodeInfo) []string {
|
||||
}
|
||||
|
||||
// BuildPredastoreNodes returns a sorted list of PredastoreNodeConfig with
|
||||
// 1-based IDs. Nodes are sorted by name for deterministic ordering.
|
||||
// 1-based IDs for nodes that run the "predastore" service.
|
||||
// Nodes are sorted by name for deterministic ordering.
|
||||
func BuildPredastoreNodes(nodes map[string]NodeInfo) []admin.PredastoreNodeConfig {
|
||||
sorted := sortedNodes(nodes)
|
||||
// Filter to nodes that run Predastore
|
||||
predaNodes := make(map[string]NodeInfo)
|
||||
for k, n := range nodes {
|
||||
if hasService(n.Services, "predastore") {
|
||||
predaNodes[k] = n
|
||||
}
|
||||
}
|
||||
sorted := sortedNodes(predaNodes)
|
||||
out := make([]admin.PredastoreNodeConfig, len(sorted))
|
||||
for i, n := range sorted {
|
||||
out[i] = admin.PredastoreNodeConfig{
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
package formation
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestHasService_EmptyListMeansAll(t *testing.T) {
|
||||
t.Parallel()
|
||||
assert.True(t, hasService(nil, "nats"))
|
||||
assert.True(t, hasService([]string{}, "predastore"))
|
||||
}
|
||||
|
||||
func TestHasService_ExplicitList(t *testing.T) {
|
||||
t.Parallel()
|
||||
services := []string{"nats", "daemon"}
|
||||
assert.True(t, hasService(services, "nats"))
|
||||
assert.True(t, hasService(services, "daemon"))
|
||||
assert.False(t, hasService(services, "predastore"))
|
||||
}
|
||||
|
||||
func TestBuildClusterRoutes_FiltersNonNATSNodes(t *testing.T) {
|
||||
t.Parallel()
|
||||
nodes := map[string]NodeInfo{
|
||||
"node1": {Name: "node1", BindIP: "10.0.0.1", Services: []string{"nats", "daemon"}},
|
||||
"node2": {Name: "node2", BindIP: "10.0.0.2", Services: []string{"daemon"}}, // no NATS
|
||||
"node3": {Name: "node3", BindIP: "10.0.0.3", Services: []string{"nats", "predastore"}},
|
||||
}
|
||||
|
||||
routes := BuildClusterRoutes(nodes)
|
||||
assert.Equal(t, []string{"10.0.0.1:4248", "10.0.0.3:4248"}, routes)
|
||||
}
|
||||
|
||||
func TestBuildClusterRoutes_EmptyServicesIncludesAll(t *testing.T) {
|
||||
t.Parallel()
|
||||
// Nodes with empty services (backward compat) should all be included
|
||||
nodes := map[string]NodeInfo{
|
||||
"node1": {Name: "node1", BindIP: "10.0.0.1"},
|
||||
"node2": {Name: "node2", BindIP: "10.0.0.2"},
|
||||
}
|
||||
|
||||
routes := BuildClusterRoutes(nodes)
|
||||
assert.Equal(t, []string{"10.0.0.1:4248", "10.0.0.2:4248"}, routes)
|
||||
}
|
||||
|
||||
func TestBuildPredastoreNodes_FiltersByPredastore(t *testing.T) {
|
||||
t.Parallel()
|
||||
nodes := map[string]NodeInfo{
|
||||
"node1": {Name: "node1", BindIP: "10.0.0.1", Services: []string{"nats", "predastore"}},
|
||||
"node2": {Name: "node2", BindIP: "10.0.0.2", Services: []string{"daemon"}}, // no predastore
|
||||
"node3": {Name: "node3", BindIP: "10.0.0.3", Services: []string{"predastore"}},
|
||||
}
|
||||
|
||||
pnodes := BuildPredastoreNodes(nodes)
|
||||
require.Len(t, pnodes, 2)
|
||||
assert.Equal(t, 1, pnodes[0].ID)
|
||||
assert.Equal(t, "10.0.0.1", pnodes[0].Host)
|
||||
assert.Equal(t, 2, pnodes[1].ID)
|
||||
assert.Equal(t, "10.0.0.3", pnodes[1].Host)
|
||||
}
|
||||
|
||||
func TestBuildPredastoreNodes_EmptyServicesIncludesAll(t *testing.T) {
|
||||
t.Parallel()
|
||||
nodes := map[string]NodeInfo{
|
||||
"node1": {Name: "node1", BindIP: "10.0.0.1"},
|
||||
"node2": {Name: "node2", BindIP: "10.0.0.2"},
|
||||
"node3": {Name: "node3", BindIP: "10.0.0.3"},
|
||||
}
|
||||
|
||||
pnodes := BuildPredastoreNodes(nodes)
|
||||
require.Len(t, pnodes, 3)
|
||||
}
|
||||
|
||||
func TestBuildClusterRoutes_MixedServicesAndEmpty(t *testing.T) {
|
||||
t.Parallel()
|
||||
nodes := map[string]NodeInfo{
|
||||
"node1": {Name: "node1", BindIP: "10.0.0.1"}, // empty = all
|
||||
"node2": {Name: "node2", BindIP: "10.0.0.2", Services: []string{"daemon"}}, // no nats
|
||||
"node3": {Name: "node3", BindIP: "10.0.0.3", Services: []string{"nats"}}, // has nats
|
||||
}
|
||||
|
||||
routes := BuildClusterRoutes(nodes)
|
||||
// node1 (empty=all, included), node2 (no nats, excluded), node3 (has nats, included)
|
||||
assert.Equal(t, []string{"10.0.0.1:4248", "10.0.0.3:4248"}, routes)
|
||||
}
|
||||
+125
-132
@@ -30,6 +30,26 @@ echo "🚀 Starting Hive development environment..."
|
||||
echo "Project root: $PROJECT_ROOT"
|
||||
echo "Data directory: $DATA_DIR"
|
||||
|
||||
# Parse services from hive.toml — defaults to all if not set
|
||||
parse_services() {
|
||||
local config_file="$CONFIG_DIR/hive.toml"
|
||||
if [ -f "$config_file" ]; then
|
||||
local svc_line=$(grep -m1 '^services' "$config_file" | sed 's/.*\[//;s/\].*//;s/"//g;s/,/ /g')
|
||||
if [ -n "$svc_line" ]; then
|
||||
echo "$svc_line"
|
||||
return
|
||||
fi
|
||||
fi
|
||||
echo "nats predastore viperblock daemon awsgw ui"
|
||||
}
|
||||
|
||||
SERVICES=$(parse_services)
|
||||
has_service() {
|
||||
local svc="$1"
|
||||
echo "$SERVICES" | grep -qw "$svc"
|
||||
}
|
||||
echo "Services: $SERVICES"
|
||||
|
||||
# Confirm configuration directory exists
|
||||
if [ ! -d "$CONFIG_DIR" ]; then
|
||||
echo "⚠️ Configuration directory $CONFIG_DIR does not exist."
|
||||
@@ -141,176 +161,149 @@ fi
|
||||
|
||||
# 1️⃣ Start NATS server
|
||||
echo ""
|
||||
echo "1️⃣ Starting NATS server..."
|
||||
if has_service "nats"; then
|
||||
echo "1️⃣ Starting NATS server..."
|
||||
|
||||
if [ -f "$CONFIG_DIR/nats/nats.conf" ]; then
|
||||
export HIVE_CONFIG_PATH=$CONFIG_DIR/nats/nats.conf
|
||||
echo " Using NATS config file: $CONFIG_DIR/nats/nats.conf"
|
||||
if [ -f "$CONFIG_DIR/nats/nats.conf" ]; then
|
||||
export HIVE_CONFIG_PATH=$CONFIG_DIR/nats/nats.conf
|
||||
echo " Using NATS config file: $CONFIG_DIR/nats/nats.conf"
|
||||
else
|
||||
echo " ⚠️ NATS config file not found at $CONFIG_DIR/nats/nats.conf, using defaults"
|
||||
export HIVE_NATS_HOST=0.0.0.0
|
||||
export HIVE_NATS_PORT=4222
|
||||
export HIVE_NATS_DATA_DIR=$DATA_DIR/nats/
|
||||
export HIVE_NATS_JETSTREAM=false
|
||||
fi
|
||||
|
||||
NATS_CMD="./bin/hive service nats start"
|
||||
start_service "nats" "$NATS_CMD"
|
||||
else
|
||||
echo " ⚠️ NATS config file not found at $CONFIG_DIR/nats/nats.conf, using defaults"
|
||||
# TODO: Confirm, settings file will overwrite env vars (e.g multi-node config)
|
||||
export HIVE_NATS_HOST=0.0.0.0
|
||||
export HIVE_NATS_PORT=4222
|
||||
export HIVE_NATS_DATA_DIR=$DATA_DIR/nats/
|
||||
export HIVE_NATS_JETSTREAM=false
|
||||
#export HIVE_NATS_DEBUG=true
|
||||
echo "1️⃣ Skipping NATS (not a local service)"
|
||||
fi
|
||||
|
||||
# Use air for hot reloading (dev!)
|
||||
#NATS_CMD="air -c .air-nats.toml"
|
||||
NATS_CMD="./bin/hive service nats start"
|
||||
|
||||
start_service "nats" "$NATS_CMD"
|
||||
|
||||
#start_service "nats" "go run cmd/hive/main.go service nats start"
|
||||
#check_service "NATS" "4222"
|
||||
|
||||
# 2️⃣ Start Predastore
|
||||
echo "2️⃣ Starting Predastore..."
|
||||
unset HIVE_CONFIG_PATH
|
||||
echo ""
|
||||
if has_service "predastore"; then
|
||||
echo "2️⃣ Starting Predastore..."
|
||||
unset HIVE_CONFIG_PATH
|
||||
|
||||
export HIVE_PREDASTORE_BASE_PATH=$DATA_DIR/predastore/
|
||||
export HIVE_PREDASTORE_CONFIG_PATH=$CONFIG_DIR/predastore/predastore.toml
|
||||
export HIVE_PREDASTORE_TLS_CERT=$CONFIG_DIR/server.pem
|
||||
export HIVE_PREDASTORE_TLS_KEY=$CONFIG_DIR/server.key
|
||||
# Very chatty logs, only for debugging
|
||||
#export HIVE_PREDASTORE_DEBUG=true
|
||||
export HIVE_PREDASTORE_BASE_PATH=$DATA_DIR/predastore/
|
||||
export HIVE_PREDASTORE_CONFIG_PATH=$CONFIG_DIR/predastore/predastore.toml
|
||||
export HIVE_PREDASTORE_TLS_CERT=$CONFIG_DIR/server.pem
|
||||
export HIVE_PREDASTORE_TLS_KEY=$CONFIG_DIR/server.key
|
||||
|
||||
# Auto-detect Predastore host:port from hive.toml [nodes.<name>.predastore] section
|
||||
PREDASTORE_BIND="0.0.0.0:8443"
|
||||
if [ -f "$CONFIG_DIR/hive.toml" ]; then
|
||||
DETECTED_PREDASTORE_HOST=$(awk -F'"' '/\[nodes\..*\.predastore\]/{found=1} found && /^host/{print $2; exit}' "$CONFIG_DIR/hive.toml")
|
||||
if [ -n "$DETECTED_PREDASTORE_HOST" ]; then
|
||||
PREDASTORE_BIND="$DETECTED_PREDASTORE_HOST"
|
||||
echo " Auto-detected Predastore bind=$PREDASTORE_BIND from hive.toml"
|
||||
fi
|
||||
fi
|
||||
export HIVE_PREDASTORE_HOST="${PREDASTORE_BIND%%:*}"
|
||||
export HIVE_PREDASTORE_PORT="${PREDASTORE_BIND##*:}"
|
||||
|
||||
# Default, distributed backend. For testing, all nodes running locally.
|
||||
# Specify NODE_ID to run a specific node (e.g multi-server)
|
||||
export HIVE_PREDASTORE_BACKEND=distributed
|
||||
|
||||
# Auto-detect Predastore NODE_ID from hive.toml if not already set
|
||||
if [ -z "$HIVE_PREDASTORE_NODE_ID" ]; then
|
||||
# Auto-detect Predastore host:port from hive.toml [nodes.<name>.predastore] section
|
||||
PREDASTORE_BIND="0.0.0.0:8443"
|
||||
if [ -f "$CONFIG_DIR/hive.toml" ]; then
|
||||
DETECTED_NODE_ID=$(awk -F'= *' '/node_id/{gsub(/ /,"",$2); print $2; exit}' "$CONFIG_DIR/hive.toml")
|
||||
if [ -n "$DETECTED_NODE_ID" ] && [ "$DETECTED_NODE_ID" != "0" ]; then
|
||||
export HIVE_PREDASTORE_NODE_ID="$DETECTED_NODE_ID"
|
||||
echo " Auto-detected Predastore NODE_ID=$DETECTED_NODE_ID from hive.toml"
|
||||
DETECTED_PREDASTORE_HOST=$(awk -F'"' '/\[nodes\..*\.predastore\]/{found=1} found && /^host/{print $2; exit}' "$CONFIG_DIR/hive.toml")
|
||||
if [ -n "$DETECTED_PREDASTORE_HOST" ]; then
|
||||
PREDASTORE_BIND="$DETECTED_PREDASTORE_HOST"
|
||||
echo " Auto-detected Predastore bind=$PREDASTORE_BIND from hive.toml"
|
||||
fi
|
||||
fi
|
||||
export HIVE_PREDASTORE_HOST="${PREDASTORE_BIND%%:*}"
|
||||
export HIVE_PREDASTORE_PORT="${PREDASTORE_BIND##*:}"
|
||||
|
||||
export HIVE_PREDASTORE_BACKEND=distributed
|
||||
|
||||
# Auto-detect Predastore NODE_ID from hive.toml if not already set
|
||||
if [ -z "$HIVE_PREDASTORE_NODE_ID" ]; then
|
||||
if [ -f "$CONFIG_DIR/hive.toml" ]; then
|
||||
DETECTED_NODE_ID=$(awk -F'= *' '/node_id/{gsub(/ /,"",$2); print $2; exit}' "$CONFIG_DIR/hive.toml")
|
||||
if [ -n "$DETECTED_NODE_ID" ] && [ "$DETECTED_NODE_ID" != "0" ]; then
|
||||
export HIVE_PREDASTORE_NODE_ID="$DETECTED_NODE_ID"
|
||||
echo " Auto-detected Predastore NODE_ID=$DETECTED_NODE_ID from hive.toml"
|
||||
fi
|
||||
fi
|
||||
fi
|
||||
export HIVE_PREDASTORE_NODE_ID="${HIVE_PREDASTORE_NODE_ID:-}"
|
||||
|
||||
PREDASTORE_CMD="./bin/hive service predastore start"
|
||||
start_service "predastore" "$PREDASTORE_CMD"
|
||||
check_service "Predastore" "$HIVE_PREDASTORE_HOST" "$HIVE_PREDASTORE_PORT"
|
||||
else
|
||||
echo "2️⃣ Skipping Predastore (not a local service)"
|
||||
fi
|
||||
export HIVE_PREDASTORE_NODE_ID="${HIVE_PREDASTORE_NODE_ID:-}"
|
||||
|
||||
# Use air for hot reloading (dev!)
|
||||
#PREDASTORE_CMD="air -c .air-predastore.toml"
|
||||
PREDASTORE_CMD="./bin/hive service predastore start"
|
||||
|
||||
start_service "predastore" "$PREDASTORE_CMD"
|
||||
check_service "Predastore" "$HIVE_PREDASTORE_HOST" "$HIVE_PREDASTORE_PORT"
|
||||
|
||||
#else
|
||||
# echo ""
|
||||
# echo "2️⃣ Skipping Predastore for multi-node setup (TODO)"
|
||||
#fi
|
||||
|
||||
|
||||
# 3️⃣ Start Viperblock
|
||||
echo ""
|
||||
echo "3️⃣ Starting Viperblock..."
|
||||
if has_service "viperblock"; then
|
||||
echo "3️⃣ Starting Viperblock..."
|
||||
|
||||
# Determine base directory for Viperblock data, dev uses /mnt/ramdisk if available
|
||||
if [ -d "/mnt/ramdisk" ] && [ -w "/mnt/ramdisk" ]; then
|
||||
VB_BASE_DIR="/mnt/ramdisk/"
|
||||
# Determine base directory for Viperblock data, dev uses /mnt/ramdisk if available
|
||||
if [ -d "/mnt/ramdisk" ] && [ -w "/mnt/ramdisk" ]; then
|
||||
VB_BASE_DIR="/mnt/ramdisk/"
|
||||
else
|
||||
VB_BASE_DIR="$DATA_DIR/viperblock/"
|
||||
fi
|
||||
|
||||
# Check if NBD plugin exists
|
||||
NBD_PLUGIN_PATH="$MULGA_ROOT/viperblock/lib/nbdkit-viperblock-plugin.so"
|
||||
|
||||
if [ ! -f "$NBD_PLUGIN_PATH" ]; then
|
||||
echo " ⚠️ NBD plugin not found at $NBD_PLUGIN_PATH"
|
||||
echo " Building Viperblock first..."
|
||||
cd "$MULGA_ROOT/viperblock"
|
||||
make build
|
||||
cd "$PROJECT_ROOT"
|
||||
fi
|
||||
|
||||
export HIVE_CONFIG_PATH=$CONFIG_DIR/hive.toml
|
||||
export HIVE_VIPERBLOCK_PLUGIN_PATH=$NBD_PLUGIN_PATH
|
||||
export HIVE_BASE_DIR=$VB_BASE_DIR
|
||||
|
||||
VIPERBLOCK_CMD="./bin/hive service viperblock start"
|
||||
start_service "viperblock" "$VIPERBLOCK_CMD"
|
||||
else
|
||||
VB_BASE_DIR="$DATA_DIR/viperblock/"
|
||||
echo "3️⃣ Skipping Viperblock (not a local service)"
|
||||
fi
|
||||
|
||||
# Check if NBD plugin exists
|
||||
NBD_PLUGIN_PATH="$MULGA_ROOT/viperblock/lib/nbdkit-viperblock-plugin.so"
|
||||
|
||||
if [ ! -f "$NBD_PLUGIN_PATH" ]; then
|
||||
echo " ⚠️ NBD plugin not found at $NBD_PLUGIN_PATH"
|
||||
echo " Building Viperblock first..."
|
||||
cd "$MULGA_ROOT/viperblock"
|
||||
make build
|
||||
cd "$PROJECT_ROOT"
|
||||
fi
|
||||
|
||||
export HIVE_CONFIG_PATH=$CONFIG_DIR/hive.toml
|
||||
# Use config above for Viperblock settings
|
||||
#export HIVE_VIPERBLOCK_S3_HOST=0.0.0.0:8443
|
||||
#export HIVE_VIPERBLOCK_S3_BUCKET=predastore
|
||||
#export HIVE_VIPERBLOCK_S3_REGION=ap-southeast-2
|
||||
#export HIVE_NATS_HOST=0.0.0.0:4222
|
||||
export HIVE_VIPERBLOCK_PLUGIN_PATH=$NBD_PLUGIN_PATH
|
||||
export HIVE_BASE_DIR=$VB_BASE_DIR
|
||||
|
||||
#VIPERBLOCK_CMD="air -c .air-viperblock.toml"
|
||||
VIPERBLOCK_CMD="./bin/hive service viperblock start"
|
||||
|
||||
#VIPERBLOCK_CMD="go run cmd/hive/main.go service viperblock start \
|
||||
# --nats-host 0.0.0.0:4222 \
|
||||
# --base-dir $VB_BASE_DIR \
|
||||
# --plugin-path $NBD_PLUGIN_PATH"
|
||||
|
||||
start_service "viperblock" "$VIPERBLOCK_CMD"
|
||||
|
||||
# 4️⃣ Start Hive Gateway/Daemon
|
||||
echo ""
|
||||
echo "4️⃣. Starting Hive Gateway..."
|
||||
if has_service "daemon"; then
|
||||
echo "4️⃣. Starting Hive Gateway..."
|
||||
|
||||
# Use the same base directory as Viperblock for consistency
|
||||
export HIVE_BASE_DIR="$VB_BASE_DIR"
|
||||
export HIVE_CONFIG_PATH=$CONFIG_DIR/hive.toml
|
||||
export HIVE_BASE_DIR=$DATA_DIR/hive/
|
||||
export HIVE_WAL_DIR=$WAL_DIR
|
||||
export HIVE_CONFIG_PATH=$CONFIG_DIR/hive.toml
|
||||
export HIVE_BASE_DIR=$DATA_DIR/hive/
|
||||
export HIVE_WAL_DIR=$WAL_DIR
|
||||
|
||||
#HIVE_CMD="air -c .air-hive.toml"
|
||||
#HIVE_CMD="./bin/hive service hive start --config $HIVE_CONFIG_PATH--base-dir $HIVE_BASE_DIR --wal-dir $HIVE_WAL_DIR"
|
||||
HIVE_CMD="./bin/hive service hive start"
|
||||
|
||||
start_service "hive" "$HIVE_CMD"
|
||||
HIVE_CMD="./bin/hive service hive start"
|
||||
start_service "hive" "$HIVE_CMD"
|
||||
else
|
||||
echo "4️⃣. Skipping Hive Gateway (not a local service)"
|
||||
fi
|
||||
|
||||
|
||||
# 5️⃣ Start AWS Gateway
|
||||
echo ""
|
||||
echo "5️⃣. Starting AWS Gateway..."
|
||||
if has_service "awsgw"; then
|
||||
echo "5️⃣. Starting AWS Gateway..."
|
||||
|
||||
# Use the same base directory as Viperblock for consistency
|
||||
unset HIVE_NATS_HOST
|
||||
unset HIVE_PREDASTORE_HOST
|
||||
export HIVE_BASE_DIR=$CONFIG_DIR
|
||||
export HIVE_CONFIG_PATH=$CONFIG_DIR/hive.toml
|
||||
export HIVE_AWSGW_TLS_CERT=$CONFIG_DIR/server.pem
|
||||
export HIVE_AWSGW_TLS_KEY=$CONFIG_DIR/server.key
|
||||
|
||||
unset HIVE_NATS_HOST
|
||||
unset HIVE_PREDASTORE_HOST
|
||||
export HIVE_BASE_DIR=$CONFIG_DIR
|
||||
export HIVE_CONFIG_PATH=$CONFIG_DIR/hive.toml
|
||||
# Overwritten by config file
|
||||
#export HIVE_AWSGW_HOST="0.0.0.0:9999"
|
||||
export HIVE_AWSGW_TLS_CERT=$CONFIG_DIR/server.pem
|
||||
export HIVE_AWSGW_TLS_KEY=$CONFIG_DIR/server.key
|
||||
|
||||
#HIVE_CMD="air -c .air-hive.toml"
|
||||
#HIVE_CMD="./bin/hive service hive start --config config/hive.toml --base-dir $HIVE_BASE_DIR --wal-dir $WAL_DIR"
|
||||
AWSGW_CMD="./bin/hive service awsgw start"
|
||||
|
||||
start_service "awsgw" "$AWSGW_CMD"
|
||||
|
||||
# TODO: Need host:ip support, depending on node
|
||||
#check_service "awsgw" "9999"
|
||||
AWSGW_CMD="./bin/hive service awsgw start"
|
||||
start_service "awsgw" "$AWSGW_CMD"
|
||||
else
|
||||
echo "5️⃣. Skipping AWS Gateway (not a local service)"
|
||||
fi
|
||||
|
||||
|
||||
# 6️⃣ Start Hive UI (skip with UI=false)
|
||||
if [ "${UI}" != "false" ]; then
|
||||
# 6️⃣ Start Hive UI (skip with UI=false or if not a local service)
|
||||
if [ "${UI}" != "false" ] && has_service "ui"; then
|
||||
echo ""
|
||||
echo "6️⃣. Starting Hive UI..."
|
||||
|
||||
HIVEUI_CMD="./bin/hive service hive-ui start"
|
||||
|
||||
start_service "hive-ui" "$HIVEUI_CMD"
|
||||
else
|
||||
echo ""
|
||||
echo "6️⃣. Skipping Hive UI (UI=false)"
|
||||
echo "6️⃣. Skipping Hive UI"
|
||||
fi
|
||||
|
||||
|
||||
|
||||
+30
-6
@@ -28,6 +28,30 @@ LOGS_DIR="$PROJECT_ROOT/data/logs"
|
||||
|
||||
echo "Stopping Hive development environment..."
|
||||
|
||||
# Parse services from hive.toml — defaults to all if not set
|
||||
CONFIG_DIR="${CONFIG_DIR:-$DATA_DIR/config}"
|
||||
parse_services() {
|
||||
local config_file="$CONFIG_DIR/hive.toml"
|
||||
if [ -f "$config_file" ]; then
|
||||
local svc_line=$(grep -m1 '^services' "$config_file" | sed 's/.*\[//;s/\].*//;s/"//g;s/,/ /g')
|
||||
if [ -n "$svc_line" ]; then
|
||||
echo "$svc_line"
|
||||
return
|
||||
fi
|
||||
fi
|
||||
echo "nats predastore viperblock daemon awsgw ui"
|
||||
}
|
||||
|
||||
SERVICES=$(parse_services)
|
||||
has_service() {
|
||||
local svc="$1"
|
||||
case "$svc" in
|
||||
hive) svc="daemon" ;;
|
||||
hive-ui) svc="ui" ;;
|
||||
esac
|
||||
echo "$SERVICES" | grep -qw "$svc"
|
||||
}
|
||||
|
||||
# Function to stop service by PID file
|
||||
stop_service() {
|
||||
local name="$1"
|
||||
@@ -60,10 +84,10 @@ stop_service() {
|
||||
echo "Stopping services..."
|
||||
echo ""
|
||||
# Stop Hive UI first (last to start)
|
||||
stop_service "hive-ui" "$PID_DIR"
|
||||
has_service "hive-ui" && stop_service "hive-ui" "$PID_DIR"
|
||||
|
||||
# Stop Hive daemon/gateway (it will terminate running instances, unmount nbd devices)
|
||||
stop_service "hive" "$PID_DIR"
|
||||
has_service "hive" && stop_service "hive" "$PID_DIR"
|
||||
|
||||
# Wait for all QEMU instances to exit before stopping infrastructure services.
|
||||
# The daemon sends system_powerdown to VMs during shutdown, but may be killed
|
||||
@@ -87,16 +111,16 @@ if pgrep -x qemu-system-x86_64 > /dev/null 2>&1; then
|
||||
fi
|
||||
|
||||
# Stop AWSGW
|
||||
stop_service "awsgw" "$PID_DIR"
|
||||
has_service "awsgw" && stop_service "awsgw" "$PID_DIR"
|
||||
|
||||
# Stop Viperblock
|
||||
stop_service "viperblock" "$PID_DIR"
|
||||
has_service "viperblock" && stop_service "viperblock" "$PID_DIR"
|
||||
|
||||
# Stop Predastore
|
||||
stop_service "predastore" "$PID_DIR"
|
||||
has_service "predastore" && stop_service "predastore" "$PID_DIR"
|
||||
|
||||
# Stop NATS
|
||||
stop_service "nats" "$PID_DIR"
|
||||
has_service "nats" && stop_service "nats" "$PID_DIR"
|
||||
|
||||
echo ""
|
||||
echo "✅ Hive development environment stopped"
|
||||
|
||||
Reference in New Issue
Block a user