From d0056ee925941281ff12a5063593bc447d05808a Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Wed, 25 Feb 2026 11:43:10 +1100 Subject: [PATCH] fix: daemon-level retry with backoff for NATS/JetStream startup (#52) --- hive/daemon/TESTING.md | 437 ---------------------------- hive/daemon/TEST_SUMMARY.md | 411 -------------------------- hive/daemon/daemon.go | 122 +++++--- hive/daemon/daemon_handlers.go | 7 +- hive/daemon/daemon_handlers_test.go | 12 +- hive/daemon/daemon_test.go | 76 +++++ 6 files changed, 175 insertions(+), 890 deletions(-) delete mode 100644 hive/daemon/TESTING.md delete mode 100644 hive/daemon/TEST_SUMMARY.md diff --git a/hive/daemon/TESTING.md b/hive/daemon/TESTING.md deleted file mode 100644 index 8ec84a41d..000000000 --- a/hive/daemon/TESTING.md +++ /dev/null @@ -1,437 +0,0 @@ -# Daemon Testing Guide - -## Overview - -The `daemon_test.go` file provides comprehensive unit tests for the daemon's EC2 launch functionality, focusing on message handling, validation, and resource management. - -## Test Structure - -### Core Test Components - -1. **NATS Server Setup** (`startTestNATSServer`) - - Creates an embedded NATS server for testing - - Uses port `-1` for automatic port allocation - - Returns server instance and connection URL - - Automatically starts and waits for readiness - -2. **Daemon Initialization** (`createTestDaemon`) - - Creates a minimal daemon configuration for testing - - Sets up temporary directories for test data - - Establishes NATS connection - - Includes automatic cleanup via `t.Cleanup()` - -3. **Test Fixtures** (`createValidRunInstancesInput`) - - Provides valid `ec2.RunInstancesInput` structures - - Pre-configured with sensible defaults - - Easy to modify for specific test scenarios - -### Test Suites - -#### 1. TestHandleEC2Launch_MessageParsing -Tests the daemon's ability to parse and validate incoming NATS messages: - -- ✅ Valid RunInstancesInput processing -- ✅ Invalid instance type handling -- ✅ Missing required fields (ImageId) -- ✅ Invalid parameter values (MinCount = 0) -- ✅ Malformed JSON handling - -**Expected Behaviors:** -- Valid inputs should return a Reservation with pending instance -- Invalid inputs should return error payloads (not transport errors) -- All responses should be properly formatted JSON - -#### 2. TestHandleEC2Launch_ResourceManagement -Tests resource allocation and instance type validation: - -- ✅ Valid instance types (t3.micro, t3.nano) -- ✅ Invalid instance types (t99.invalid) -- ✅ Resource availability checking -- ✅ Allocation/deallocation cycles - -#### 3. TestDaemon_Initialization -Tests daemon initialization logic: - -- ✅ Configuration loading -- ✅ Resource manager creation -- ✅ Instance storage initialization - -#### 4. TestResourceManager -Tests the resource management subsystem: - -- ✅ System resource detection -- ✅ Instance type allocation -- ✅ Resource deallocation -- ✅ Capacity validation - -## Running Tests - -### Basic Test Execution - -```bash -# Run all daemon tests -go test -v ./hive/daemon/ - -# Run specific test -go test -v ./hive/daemon/ -run TestHandleEC2Launch_MessageParsing - -# Run with race detection -go test -race -v ./hive/daemon/ - -# Run with coverage -go test -cover -v ./hive/daemon/ -``` - -### Skipping Integration Tests - -The tests are designed to skip integration components when running on macOS or when dependencies are unavailable: - -```bash -# Skip integration tests that require viperblock/nbdkit -SKIP_INTEGRATION=1 go test -v ./hive/daemon/ -``` - -## macOS Testing Limitations - -### Current Limitations - -The full integration test suite has limitations on macOS due to missing dependencies: - -1. **Viperblock** - Not yet ported to macOS - - NBD (Network Block Device) support not available - - Volume mounting operations will fail - -2. **NBDkit** - Linux-specific tool - - Requires kernel NBD support - - Not available natively on macOS - -3. **QEMU/KVM** - Limited on macOS - - KVM acceleration not available (Linux-only) - - Can use QEMU with TCG (slow emulation) - -### What Works on macOS - -✅ **Message-driven architecture testing:** -- NATS message parsing and routing -- Request/response validation -- Error handling and payload generation -- Resource manager logic - -✅ **Unit tests without infrastructure:** -- Input validation (`ValidateRunInstancesInput`) -- JSON marshaling/unmarshaling -- Error code generation -- Instance ID generation - -### What Requires Linux - -❌ **Full integration testing:** -- Volume creation and mounting (requires viperblock) -- NBD device attachment (requires nbdkit) -- VM launch with KVM acceleration -- EBS volume operations -- Cloud-init ISO creation and mounting - -### Recommendations for macOS Development - -#### Option 1: Mock Integration Points - -Create mock implementations for macOS testing: - -```go -// daemon_test_helpers.go -// +build darwin - -// Mock viperblock operations for macOS -type MockViperblock struct { - // Mock implementation -} - -func (m *MockViperblock) Mount(volumeId string) error { - // Return mock NBD URI - return nil -} -``` - -#### Option 2: Use Docker Linux Container - -Run tests in a Linux container with full dependencies: - -```bash -# Build Linux test container -docker build -t hive-test -f Dockerfile.test . - -# Run tests in container -docker run --rm \ - --privileged \ - -v $(pwd):/workspace \ - hive-test \ - go test -v ./hive/daemon/ -``` - -**Dockerfile.test example:** -```dockerfile -FROM ubuntu:22.04 - -RUN apt-get update && apt-get install -y \ - golang-1.21 \ - qemu-kvm \ - nbdkit \ - build-essential - -WORKDIR /workspace -``` - -#### Option 3: GitHub Actions / CI Pipeline - -Use CI for full integration tests: - -```yaml -# .github/workflows/test.yml -name: Integration Tests -on: [push, pull_request] - -jobs: - test: - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v3 - - uses: actions/setup-go@v4 - - name: Install dependencies - run: | - sudo apt-get update - sudo apt-get install -y nbdkit qemu-kvm - - name: Run tests - run: make test -``` - -#### Option 4: Virtual Linux VM on macOS - -Use a Linux VM for development testing: - -```bash -# Using UTM or Multipass -multipass launch --name hive-dev --cpus 4 --memory 8G --disk 20G -multipass shell hive-dev - -# Inside VM -cd /path/to/hive -make test -``` - -### Hybrid Testing Strategy - -**On macOS (Unit Tests):** -```bash -# Test message handling, validation, resource management -SKIP_INTEGRATION=1 go test -v ./hive/daemon/ -``` - -**On Linux (Full Integration):** -```bash -# Test complete EC2 launch flow with viperblock/nbdkit -go test -v ./hive/daemon/ -``` - -## Test Coverage Goals - -### Current Coverage -- ✅ Message parsing and validation -- ✅ Resource allocation logic -- ✅ Error handling and responses -- ✅ NATS integration (pub/sub) - -### Additional Tests Needed (Linux) - -#### Volume Management Tests -```go -func TestHandleEC2Launch_VolumeCreation(t *testing.T) { - // Test AMI volume cloning - // Test EFI partition creation - // Test cloud-init ISO generation -} -``` - -#### NBD Mount Tests -```go -func TestMountVolumes(t *testing.T) { - // Test NBD URI generation - // Test mount request/response - // Test mount failures -} -``` - -#### VM Lifecycle Tests -```go -func TestLaunchInstance(t *testing.T) { - // Test QEMU process startup - // Test QMP socket connection - // Test instance state transitions -} -``` - -#### End-to-End Tests -```go -func TestEC2Launch_EndToEnd(t *testing.T) { - // Full flow: message -> volumes -> VM -> running - // Test with multiple instance types - // Test concurrent launches -} -``` - -## Mocking Strategy for macOS - -### Mock EBS Mount Service - -```go -// Create a mock EBS mount responder -func startMockEBSService(t *testing.T, nc *nats.Conn) { - nc.QueueSubscribe("ebs.mount", "mock-ebs", func(msg *nats.Msg) { - var req config.EBSRequest - json.Unmarshal(msg.Data, &req) - - resp := config.EBSMountResponse{ - URI: fmt.Sprintf("nbd://localhost:10809/%s", req.Name), - Mounted: true, - Error: "", - } - - data, _ := json.Marshal(resp) - msg.Respond(data) - }) -} -``` - -### Mock S3 Client - -```go -// Mock S3 client for SSH key retrieval -type MockS3Client struct { - keys map[string]string -} - -func (m *MockS3Client) Read(path string) ([]byte, error) { - if key, ok := m.keys[path]; ok { - return []byte(key), nil - } - return nil, fmt.Errorf("key not found") -} -``` - -### Example Mock Test - -```go -func TestHandleEC2Launch_WithMocks(t *testing.T) { - if runtime.GOOS == "darwin" { - // Run with mocks on macOS - ns, natsURL := startTestNATSServer(t) - defer ns.Shutdown() - - nc, _ := nats.Connect(natsURL) - defer nc.Close() - - // Start mock services - startMockEBSService(t, nc) - startMockS3Service(t, nc) - - // Run test... - } else { - // Run with real services on Linux - // ... - } -} -``` - -## Debugging Tests - -### Enable Verbose Logging - -```bash -# Run tests with verbose output -go test -v ./hive/daemon/ 2>&1 | tee test.log -``` - -### Debug NATS Messages - -```go -// Add debug subscriber to see all messages -nc.Subscribe(">", func(msg *nats.Msg) { - t.Logf("NATS [%s]: %s", msg.Subject, string(msg.Data)) -}) -``` - -### Common Issues - -**Issue: "NATS server failed to start"** -- Port already in use -- Use `-1` for auto port allocation -- Check firewall settings - -**Issue: "Request timeout"** -- Increase timeout duration -- Check NATS connectivity -- Verify handler subscription - -**Issue: "Test fails on Linux but works on macOS"** -- Missing dependencies (nbdkit, QEMU) -- Permission issues (KVM access) -- File system differences - -## Next Steps - -1. ✅ Create basic message handling tests (DONE) -2. ⏳ Add mock services for macOS testing -3. ⏳ Create Linux CI pipeline for full integration tests -4. ⏳ Add Docker-based test environment -5. ⏳ Implement volume lifecycle tests (Linux only) -6. ⏳ Add concurrent launch stress tests -7. ⏳ Create performance benchmarks - -## Contributing - -When adding new tests: - -1. **Check platform compatibility:** - ```go - if runtime.GOOS == "darwin" && needsLinux { - t.Skip("Test requires Linux") - } - ``` - -2. **Use table-driven tests:** - ```go - tests := []struct { - name string - input interface{} - want interface{} - }{ /* ... */ } - ``` - -3. **Clean up resources:** - ```go - t.Cleanup(func() { - // Cleanup code - }) - ``` - -4. **Document dependencies:** - - Add comments explaining Linux-only requirements - - Use build tags where appropriate - - Provide alternative test paths for macOS - -## Local Docker E2E Tests - -For more advanced testing (e.g., if GitHub Actions is failing), run the Docker E2E tests locally before committing. These mirror the CI pipeline exactly. - -```bash -make test-docker-single # Single-node E2E only -make test-docker-multi # Multi-node E2E only -make test-docker # Both suites -``` - -Requirements: -- Docker with `--privileged` support -- `/dev/kvm` available (KVM-capable host) -- Sibling repos (`viperblock/`, `predastore/`) cloned alongside `hive/` - -The Docker build copies local source (including `go.work` if present), so uncommitted changes in sibling repos are picked up automatically. diff --git a/hive/daemon/TEST_SUMMARY.md b/hive/daemon/TEST_SUMMARY.md deleted file mode 100644 index e2edba45f..000000000 --- a/hive/daemon/TEST_SUMMARY.md +++ /dev/null @@ -1,411 +0,0 @@ -# Daemon Unit Test Summary - -## Created Files - -1. **`daemon_test.go`** - Comprehensive unit test suite for daemon EC2 launch functionality -2. **`TESTING.md`** - Detailed testing guide with macOS limitations and strategies -3. **`TEST_SUMMARY.md`** (this file) - Summary of test implementation - -## Test Framework Components - -### 1. NATS Test Server (`startTestNATSServer`) -```go -func startTestNATSServer(t *testing.T) (*server.Server, string) -``` - -- Creates embedded NATS server for isolated testing -- Uses port `-1` for automatic port allocation (avoids conflicts) -- Returns server instance and connection URL -- Automatically waits for server readiness with 5-second timeout - -**Key Features:** -- No external NATS dependency -- Clean isolation between tests -- Automatic cleanup via defer -- Debug logging of assigned port - -### 2. Test Daemon Factory (`createTestDaemon`) -```go -func createTestDaemon(t *testing.T, natsURL string) *Daemon -``` - -- Creates minimal daemon configuration for testing -- Sets up temporary directories with automatic cleanup -- Configures mock Predastore and S3 settings -- Establishes NATS connection - -**Cleanup Strategy:** -- Uses `t.Cleanup()` for automatic resource cleanup -- Removes temporary directories -- Closes NATS connections -- No manual cleanup required - -### 3. Test Fixtures (`createValidRunInstancesInput`) -```go -func createValidRunInstancesInput() *ec2.RunInstancesInput -``` - -- Provides fully-formed valid `RunInstancesInput` -- Pre-configured with AWS-compatible values -- Easy to clone and modify for specific tests -- Includes all required fields - -## Test Suites - -### TestHandleEC2Launch_MessageParsing ✅ -**Purpose:** Validates NATS message parsing and error handling - -**Test Cases:** -1. Valid RunInstancesInput - expects Reservation with pending instance -2. Invalid instance type - expects error response -3. Missing required ImageId - expects MissingParameter error -4. Invalid MinCount (zero) - expects InvalidParameterValue error -5. Malformed JSON - expects ValidationError - -**Status:** ⚠️ Skipped (requires SKIP_INTEGRATION not set) - -**Current Issues:** -- Line 638 in daemon.go: `var runInstancesInput *ec2.RunInstancesInput` is nil -- Line 645: `utils.UnmarshalJsonPayload(runInstancesInput, msg.Data)` fails with nil pointer -- The UnmarshalJsonPayload function in utils.go:283 does `decoder.Decode(&input)` which creates a pointer-to-pointer issue -- Even valid inputs are rejected with ValidationError - -### TestHandleEC2Launch_ResourceManagement ⏸️ -**Purpose:** Tests resource allocation and instance type validation - -**Status:** Skipped (requires full infrastructure) - -**Reasoning:** -- handleEC2Launch attempts to create viperblock volumes -- Requires S3 backend (predastore) running -- Requires nbdkit for NBD mounting -- Requires QEMU for VM launch -- None of these are available on macOS - -**When to Enable:** -- On Linux with full Hive stack running -- In CI/CD pipeline with all dependencies -- With mock services (future enhancement) - -### TestDaemon_Initialization ✅ -**Purpose:** Tests daemon creation and configuration - -**Test Coverage:** -- Daemon struct initialization -- Resource manager creation -- Instance storage initialization -- Configuration loading - -**Status:** ✅ PASSING - -### TestResourceManager ✅ -**Purpose:** Tests resource allocation logic - -**Test Coverage:** -- System resource detection (CPU, Memory) -- Instance type definitions -- Resource allocation/deallocation -- Capacity checking - -**Status:** ✅ PASSING - -## Current Test Results - -```bash -$ go test -v ./hive/daemon/ - -=== RUN TestHandleEC2Launch_MessageParsing - daemon_test.go:115: Skipping integration test - SKIP_INTEGRATION is set ---- SKIP: TestHandleEC2Launch_MessageParsing (0.00s) - -=== RUN TestHandleEC2Launch_ResourceManagement - daemon_test.go:261: Skipping resource management test - requires full hive infrastructure ---- SKIP: TestHandleEC2Launch_ResourceManagement (0.00s) - -=== RUN TestDaemon_Initialization ---- PASS: TestDaemon_Initialization (0.01s) - -=== RUN TestResourceManager ---- PASS: TestResourceManager (0.01s) - -PASS -ok github.com/mulgadc/hive/hive/daemon 0.786s -``` - -## Known Issues & Limitations - -### 1. Nil Pointer Bug in handleEC2Launch (daemon.go:638-645) - -**Issue:** -```go -var runInstancesInput *ec2.RunInstancesInput // nil pointer -// ... -errResp = utils.UnmarshalJsonPayload(runInstancesInput, msg.Data) // fails -``` - -**Root Cause:** -- `runInstancesInput` declared as pointer but never initialized -- `UnmarshalJsonPayload` takes `&input` creating pointer-to-pointer -- Results in ValidationError even for valid inputs - -**Suggested Fix (not applied per instructions):** -```go -runInstancesInput := &ec2.RunInstancesInput{} // Initialize before use -errResp = utils.UnmarshalJsonPayload(runInstancesInput, msg.Data) -``` - -Or fix the utils function: -```go -func UnmarshalJsonPayload(input interface{}, jsonData []byte) []byte { - decoder := json.NewDecoder(bytes.NewReader(jsonData)) - decoder.DisallowUnknownFields() - err := decoder.Decode(input) // Remove & since input is already a pointer - // ... -} -``` - -### 2. macOS Limitations - -**Missing Dependencies:** -- ❌ viperblock (not ported to macOS) -- ❌ nbdkit (Linux-only, requires kernel NBD) -- ❌ KVM acceleration (Linux-only) -- ❌ Predastore running instance - -**Impact:** -- Cannot test volume creation/mounting -- Cannot test VM launch -- Cannot test full EC2 launch flow - -**Workarounds:** -- Use Docker Linux container for integration tests -- Create mock services for macOS -- Run full tests in CI/CD on Linux -- Use VM (UTM, Multipass) for local Linux testing - -### 3. Test Coverage Gaps - -**Not Tested:** -- Volume lifecycle (create, mount, unmount) -- NBD URI generation and validation -- Cloud-init ISO creation -- EFI partition creation -- QMP socket communication -- VM state transitions -- Concurrent launch requests -- Resource exhaustion scenarios - -**Future Enhancements:** -- Add mock EBS mount service -- Add mock S3 client -- Add mock viperblock backend -- Test error paths more thoroughly -- Add benchmark tests -- Add stress tests - -## Running Tests - -### On macOS (Unit Tests Only) -```bash -# Run all passing tests -go test -v ./hive/daemon/ - -# Run with race detection -go test -race -v ./hive/daemon/ - -# Run with coverage -go test -cover -v ./hive/daemon/ - -# Run specific test -go test -v ./hive/daemon/ -run TestResourceManager -``` - -### On Linux (Full Integration) -```bash -# Requires: predastore, viperblock, nbdkit, qemu running - -# Run all tests including integration -SKIP_INTEGRATION="" go test -v ./hive/daemon/ - -# Or simply -go test -v ./hive/daemon/ -``` - -### In Docker (Recommended) -```bash -# Build test container -docker build -t hive-daemon-test -f Dockerfile.test . - -# Run tests -docker run --rm --privileged \ - -v $(pwd):/workspace \ - hive-daemon-test \ - go test -v ./hive/daemon/ -``` - -## Test Architecture Highlights - -### ✅ Strengths - -1. **Isolated NATS Server** - - Each test gets fresh NATS instance - - No interference between tests - - Automatic port allocation prevents conflicts - -2. **Clean Resource Management** - - Automatic cleanup via `t.Cleanup()` - - Temporary directories auto-removed - - No test artifacts left behind - -3. **Table-Driven Tests** - - Easy to add new test cases - - Clear test structure - - Minimal code duplication - -4. **Clear Documentation** - - Comments explain what each test validates - - Reasoning for skipped tests documented - - macOS limitations clearly stated - -5. **Framework Foundation** - - Easy to extend with new tests - - Mock services can be added later - - Ready for CI/CD integration - -### ⚠️ Weaknesses - -1. **Limited Coverage** - - Only tests message parsing and resource logic - - Cannot test actual VM launch on macOS - - Missing integration with viperblock/nbdkit - -2. **Existing Code Issues** - - Nil pointer bug blocks message parsing tests - - utils.UnmarshalJsonPayload has pointer-to-pointer issue - - These bugs need fixing before tests can pass - -3. **No Mocking Yet** - - Tests require real infrastructure for full validation - - No mock EBS service - - No mock S3 client - -## Next Steps - -### Immediate (Required for Tests to Pass) - -1. **Fix nil pointer in handleEC2Launch** - ```go - // daemon.go:638 - runInstancesInput := &ec2.RunInstancesInput{} // Initialize - ``` - -2. **Fix UnmarshalJsonPayload** - ```go - // utils.go:283 - err := decoder.Decode(input) // Remove & - ``` - -### Short Term (Enhance Test Coverage) - -1. **Add Mock Services** - - Mock EBS mount service (NATS responder) - - Mock S3 client for SSH keys - - Mock viperblock for volume operations - -2. **Add Platform-Specific Tests** - ```go - // +build linux - func TestHandleEC2Launch_FullFlow(t *testing.T) { ... } - ``` - -3. **Add Helper Functions** - - `createMockEBSService(t, nc)` - - `createMockS3Client(t, keys map[string]string)` - - `waitForInstanceState(t, daemon, instanceId, expectedState)` - -### Long Term (Production Ready) - -1. **CI/CD Integration** - - GitHub Actions workflow for Linux tests - - Docker-based test environment - - Coverage reporting - -2. **Comprehensive Test Suite** - - End-to-end VM launch tests - - Concurrent launch stress tests - - Failure recovery tests - - Performance benchmarks - -3. **Test Documentation** - - Video walkthrough of test setup - - Troubleshooting guide - - Contributing guidelines - -## Recommendations - -### For macOS Development - -**Option 1: Mock Services** (Fastest) -```go -func TestHandleEC2Launch_WithMocks(t *testing.T) { - ns, natsURL := startTestNATSServer(t) - startMockEBSService(t, nc) - startMockS3Service(t, nc) - // Test with mocked dependencies -} -``` - -**Option 2: Docker Container** (Most Realistic) -```bash -docker-compose up test-env -go test -v ./hive/daemon/ -``` - -**Option 3: Linux VM** (Balanced) -```bash -multipass launch --name hive-dev -multipass shell hive-dev -# Run tests in VM -``` - -### For Linux CI/CD - -```yaml -# .github/workflows/daemon-tests.yml -name: Daemon Tests -on: [push, pull_request] -jobs: - test: - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v3 - - name: Install dependencies - run: sudo apt-get install -y nbdkit qemu-kvm - - name: Start services - run: ./scripts/start-test-env.sh - - name: Run tests - run: go test -v ./hive/daemon/ -``` - -## Conclusion - -This test framework provides: - -✅ **Solid Foundation** -- NATS message handling tests -- Resource manager validation -- Clean test isolation -- Easy to extend - -⚠️ **Current Limitations** -- Requires bug fixes in daemon.go and utils.go -- Limited to unit tests on macOS -- Needs mock services for full coverage - -🎯 **Future Potential** -- Can support full integration tests on Linux -- Mock services enable macOS testing -- Ready for CI/CD integration -- Foundation for comprehensive test suite - -The test framework is **production-ready for unit testing** but requires the identified bug fixes and either mock services (macOS) or full infrastructure (Linux) for integration testing. diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 5121cd42c..1b18ec9d2 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -129,6 +129,10 @@ type Daemon struct { // When true, the daemon rejects new work and setupShutdown skips VM stops. shuttingDown atomic.Bool + // ready is set to true once NATS connection, JetStream, and all services + // are fully initialized. The health endpoint reports "starting" until ready. + ready atomic.Bool + mu sync.Mutex } @@ -511,29 +515,58 @@ func (d *Daemon) Start() error { d.startHeartbeat() d.startPendingWatchdog() + + d.ready.Store(true) + slog.Info("Daemon fully initialized", "node", d.node, "startupTime", time.Since(d.startTime).Round(time.Second)) + d.setupShutdown() d.awaitShutdown() return nil } -// connectNATS establishes a connection to the NATS server with reconnect handling. +// connectNATS establishes a connection to the NATS server with retry and +// exponential backoff. On multi-node clusters, the local NATS server may not +// be ready immediately after daemon start (e.g. if start-dev.sh is still +// launching services). This retries for up to 5 minutes before giving up. func (d *Daemon) connectNATS() error { - nc, err := utils.ConnectNATS(d.config.NATS.Host, d.config.NATS.ACL.Token) - if err != nil { - return err + const maxWait = 5 * time.Minute + retryDelay := 500 * time.Millisecond + start := time.Now() + + for { + nc, err := utils.ConnectNATS(d.config.NATS.Host, d.config.NATS.ACL.Token) + if err == nil { + d.natsConn = nc + if time.Since(start) > time.Second { + slog.Info("NATS connection established", "elapsed", time.Since(start)) + } + return nil + } + + elapsed := time.Since(start) + if elapsed >= maxWait { + return fmt.Errorf("NATS connect failed after %s: %w", elapsed.Round(time.Second), err) + } + + slog.Warn("NATS not ready, retrying...", "error", err, "elapsed", elapsed.Round(time.Second), "retryIn", retryDelay) + time.Sleep(retryDelay) + retryDelay = min(retryDelay*2, 10*time.Second) } - d.natsConn = nc - return nil } -// initJetStream initializes JetStream with retry/backoff and upgrades replicas for multi-node clusters. +// initJetStream initializes JetStream with retry/backoff and upgrades replicas +// for multi-node clusters. On multi-node clusters, JetStream requires NATS +// cluster quorum which may take several minutes if nodes start at different +// times. This retries for up to 5 minutes to allow late-joining nodes. func (d *Daemon) initJetStream() error { - const maxRetries = 10 + const maxWait = 5 * time.Minute retryDelay := 500 * time.Millisecond + start := time.Now() + attempt := 0 - var lastErr error - for attempt := 1; attempt <= maxRetries; attempt++ { + for { + attempt++ var err error d.jsManager, err = NewJetStreamManager(d.natsConn, 1) if err == nil { @@ -545,22 +578,18 @@ func (d *Daemon) initJetStream() error { } if err == nil { - slog.Info("JetStream KV stores initialized successfully", "replicas", 1, "attempts", attempt) - lastErr = nil + slog.Info("JetStream KV stores initialized successfully", "replicas", 1, "attempts", attempt, "elapsed", time.Since(start).Round(time.Second)) break } - lastErr = err - slog.Warn("Failed to init JetStream", "error", err, "attempt", attempt, "maxRetries", maxRetries) - - if attempt < maxRetries { - time.Sleep(retryDelay) - retryDelay = min(retryDelay*2, 5*time.Second) + elapsed := time.Since(start) + if elapsed >= maxWait { + return fmt.Errorf("failed to initialize JetStream after %s (%d attempts): %w", elapsed.Round(time.Second), attempt, err) } - } - if lastErr != nil { - return fmt.Errorf("failed to initialize JetStream after %d attempts: %w", maxRetries, lastErr) + slog.Warn("JetStream not ready (waiting for cluster quorum)", "error", err, "attempt", attempt, "elapsed", elapsed.Round(time.Second), "retryIn", retryDelay) + time.Sleep(retryDelay) + retryDelay = min(retryDelay*2, 10*time.Second) } // Upgrade replicas if cluster has more than one node @@ -578,51 +607,59 @@ func (d *Daemon) initJetStream() error { // During cluster restarts, JetStream KV may be temporarily unavailable while // NATS routes re-establish and the cluster forms quorum. func (d *Daemon) initSnapshotService() (*handlers_ec2_snapshot.SnapshotServiceImpl, nats.KeyValue, error) { - const maxRetries = 10 + const maxWait = 5 * time.Minute retryDelay := 500 * time.Millisecond + start := time.Now() + attempt := 0 - for attempt := 1; attempt <= maxRetries; attempt++ { + for { + attempt++ snapSvc, snapshotKV, err := handlers_ec2_snapshot.NewSnapshotServiceImplWithNATS(d.config, d.natsConn) if err == nil { if attempt > 1 { - slog.Info("Snapshot service initialized successfully", "attempts", attempt) + slog.Info("Snapshot service initialized successfully", "attempts", attempt, "elapsed", time.Since(start).Round(time.Second)) } return snapSvc, snapshotKV, nil } - slog.Warn("Failed to init snapshot service", "error", err, "attempt", attempt, "maxRetries", maxRetries) - if attempt < maxRetries { - time.Sleep(retryDelay) - retryDelay = min(retryDelay*2, 5*time.Second) + elapsed := time.Since(start) + if elapsed >= maxWait { + return nil, nil, fmt.Errorf("snapshot service unavailable after %s (%d attempts): %w", elapsed.Round(time.Second), attempt, err) } - } - return nil, nil, fmt.Errorf("snapshot service unavailable after %d attempts", maxRetries) + slog.Warn("Failed to init snapshot service", "error", err, "attempt", attempt, "elapsed", elapsed.Round(time.Second)) + time.Sleep(retryDelay) + retryDelay = min(retryDelay*2, 10*time.Second) + } } // initAccountService initializes the account settings service with retry/backoff. func (d *Daemon) initAccountService() error { - const maxRetries = 10 + const maxWait = 5 * time.Minute retryDelay := 500 * time.Millisecond + start := time.Now() + attempt := 0 - for attempt := 1; attempt <= maxRetries; attempt++ { + for { + attempt++ svc, err := handlers_ec2_account.NewAccountSettingsServiceImplWithNATS(d.config, d.natsConn) if err == nil { if attempt > 1 { - slog.Info("Account settings service initialized successfully", "attempts", attempt) + slog.Info("Account settings service initialized successfully", "attempts", attempt, "elapsed", time.Since(start).Round(time.Second)) } d.accountService = svc return nil } - slog.Warn("Failed to init account settings service", "error", err, "attempt", attempt, "maxRetries", maxRetries) - if attempt < maxRetries { - time.Sleep(retryDelay) - retryDelay = min(retryDelay*2, 5*time.Second) + elapsed := time.Since(start) + if elapsed >= maxWait { + return fmt.Errorf("account settings service unavailable after %s (%d attempts): %w", elapsed.Round(time.Second), attempt, err) } - } - return fmt.Errorf("account settings service unavailable after %d attempts", maxRetries) + slog.Warn("Failed to init account settings service", "error", err, "attempt", attempt, "elapsed", elapsed.Round(time.Second)) + time.Sleep(retryDelay) + retryDelay = min(retryDelay*2, 10*time.Second) + } } // waitForClusterReady waits until dependent infrastructure services are reachable @@ -993,9 +1030,14 @@ func (d *Daemon) ClusterManager() error { serviceHealth["ovn-controller"] = "not_running" } + status := "running" + if !d.ready.Load() { + status = "starting" + } + response := config.NodeHealthResponse{ Node: d.node, - Status: "running", + Status: status, ConfigHash: configHash, Epoch: d.clusterConfig.Epoch, Uptime: int64(time.Since(d.startTime).Seconds()), diff --git a/hive/daemon/daemon_handlers.go b/hive/daemon/daemon_handlers.go index 793f679fc..aaa78fe02 100644 --- a/hive/daemon/daemon_handlers.go +++ b/hive/daemon/daemon_handlers.go @@ -138,9 +138,14 @@ func (d *Daemon) handleHealthCheck(msg *nats.Msg) { configHash = "error" } + status := "running" + if !d.ready.Load() { + status = "starting" + } + response := config.NodeHealthResponse{ Node: d.node, - Status: "running", + Status: status, ConfigHash: configHash, Epoch: d.clusterConfig.Epoch, Uptime: int64(time.Since(d.startTime).Seconds()), diff --git a/hive/daemon/daemon_handlers_test.go b/hive/daemon/daemon_handlers_test.go index 3fb1267f8..dfdc3beb4 100644 --- a/hive/daemon/daemon_handlers_test.go +++ b/hive/daemon/daemon_handlers_test.go @@ -314,6 +314,7 @@ func TestHandleHealthCheck(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() + // Before ready is set, status should be "starting" reply, err := daemon.natsConn.Request(topic, nil, 5*time.Second) require.NoError(t, err) require.NotNil(t, reply) @@ -323,9 +324,18 @@ func TestHandleHealthCheck(t *testing.T) { require.NoError(t, err) assert.Equal(t, daemon.node, resp.Node) - assert.Equal(t, "running", resp.Status) + assert.Equal(t, "starting", resp.Status) assert.NotEmpty(t, resp.ConfigHash) assert.GreaterOrEqual(t, resp.Uptime, int64(0)) + + // After marking ready, status should be "running" + daemon.ready.Store(true) + reply, err = daemon.natsConn.Request(topic, nil, 5*time.Second) + require.NoError(t, err) + + err = json.Unmarshal(reply.Data, &resp) + require.NoError(t, err) + assert.Equal(t, "running", resp.Status) } // --- handleNodeDiscover tests --- diff --git a/hive/daemon/daemon_test.go b/hive/daemon/daemon_test.go index 619a3be82..eff0725fa 100644 --- a/hive/daemon/daemon_test.go +++ b/hive/daemon/daemon_test.go @@ -3734,3 +3734,79 @@ func TestCanAllocate(t *testing.T) { }) } } + +// TestConnectNATS_RetriesOnFailure tests that connectNATS retries when NATS is not immediately available. +func TestConnectNATS_RetriesOnFailure(t *testing.T) { + // Use a port that nothing is listening on + clusterCfg := &config.ClusterConfig{ + Node: "node-1", + Nodes: map[string]config.Config{}, + } + cfg := config.Config{ + NATS: config.NATSConfig{ + Host: "nats://127.0.0.1:14222", // nothing listening here + }, + } + clusterCfg.Nodes["node-1"] = cfg + daemon := NewDaemon(clusterCfg) + + // Run connectNATS in a goroutine and verify it retries (takes > 1s) + // then verify it eventually errors. We don't wait for the full 5min timeout; + // just confirm it retried at least once by checking elapsed time. + errCh := make(chan error, 1) + start := time.Now() + go func() { errCh <- daemon.connectNATS() }() + + select { + case err := <-errCh: + elapsed := time.Since(start) + require.Error(t, err) + assert.Contains(t, err.Error(), "NATS connect failed") + assert.Greater(t, elapsed, 1*time.Second, "should have retried at least once") + case <-time.After(10 * time.Second): + // connectNATS is still retrying after 10s — that proves retry works. + // We don't wait for the full 5min timeout in tests. + t.Log("connectNATS still retrying after 10s (retry logic confirmed)") + } +} + +// TestConnectNATS_SucceedsImmediately tests that connectNATS succeeds immediately when NATS is available. +func TestConnectNATS_SucceedsImmediately(t *testing.T) { + clusterCfg := &config.ClusterConfig{ + Node: "node-1", + Nodes: map[string]config.Config{}, + } + cfg := config.Config{ + NATS: config.NATSConfig{ + Host: sharedNATSURL, + }, + } + clusterCfg.Nodes["node-1"] = cfg + daemon := NewDaemon(clusterCfg) + + start := time.Now() + err := daemon.connectNATS() + elapsed := time.Since(start) + + require.NoError(t, err) + assert.Less(t, elapsed, 2*time.Second, "should connect quickly when NATS is available") + assert.True(t, daemon.natsConn.IsConnected()) + + daemon.natsConn.Close() +} + +// TestDaemonReadyFlag tests that the ready flag is false by default and can be set. +func TestDaemonReadyFlag(t *testing.T) { + clusterCfg := &config.ClusterConfig{ + Node: "node-1", + Nodes: map[string]config.Config{}, + } + cfg := config.Config{} + clusterCfg.Nodes["node-1"] = cfg + daemon := NewDaemon(clusterCfg) + + assert.False(t, daemon.ready.Load(), "daemon should not be ready before Start()") + + daemon.ready.Store(true) + assert.True(t, daemon.ready.Load(), "daemon should be ready after setting flag") +}