diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index eed0c13fa..a6dfa0ec9 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -471,6 +471,164 @@ jobs: exit 1 fi + - name: Test /healthz endpoint returns 200 when healthy + run: | + echo "Testing /healthz endpoint..." + # Use kubectl port-forward to access health endpoint + kubectl port-forward pod/rfr-test-im-0 8080:8080 & + PF_PID=$! + sleep 2 + + RESPONSE=$(curl -s -w "\n%{http_code}" http://localhost:8080/healthz) + HTTP_CODE=$(echo "$RESPONSE" | tail -1) + BODY=$(echo "$RESPONSE" | head -n -1) + + kill $PF_PID 2>/dev/null || true + + echo "Response: $BODY" + echo "HTTP Code: $HTTP_CODE" + + if [[ "$HTTP_CODE" == "200" ]]; then + echo "✓ /healthz returned 200" + else + echo "✗ /healthz returned $HTTP_CODE, expected 200" + exit 1 + fi + + # Verify response contains expected fields + if echo "$BODY" | grep -q '"status":"ok"'; then + echo "✓ /healthz response contains status:ok" + else + echo "✗ /healthz response missing status:ok" + exit 1 + fi + + - name: Test /readyz endpoint returns 200 when ready + run: | + echo "Testing /readyz endpoint..." + kubectl port-forward pod/rfr-test-im-0 8080:8080 & + PF_PID=$! + sleep 2 + + RESPONSE=$(curl -s -w "\n%{http_code}" http://localhost:8080/readyz) + HTTP_CODE=$(echo "$RESPONSE" | tail -1) + BODY=$(echo "$RESPONSE" | head -n -1) + + kill $PF_PID 2>/dev/null || true + + echo "Response: $BODY" + echo "HTTP Code: $HTTP_CODE" + + if [[ "$HTTP_CODE" == "200" ]]; then + echo "✓ /readyz returned 200" + else + echo "✗ /readyz returned $HTTP_CODE, expected 200" + exit 1 + fi + + # Verify response contains role + if echo "$BODY" | grep -q '"role"'; then + echo "✓ /readyz response contains role" + else + echo "✗ /readyz response missing role" + exit 1 + fi + + - name: Test /status endpoint returns detailed info + run: | + echo "Testing /status endpoint..." + kubectl port-forward pod/rfr-test-im-0 8080:8080 & + PF_PID=$! + sleep 2 + + RESPONSE=$(curl -s -w "\n%{http_code}" http://localhost:8080/status) + HTTP_CODE=$(echo "$RESPONSE" | tail -1) + BODY=$(echo "$RESPONSE" | head -n -1) + + kill $PF_PID 2>/dev/null || true + + echo "Response: $BODY" + echo "HTTP Code: $HTTP_CODE" + + if [[ "$HTTP_CODE" == "200" ]]; then + echo "✓ /status returned 200" + else + echo "✗ /status returned $HTTP_CODE, expected 200" + exit 1 + fi + + # Verify response contains expected sections + if echo "$BODY" | grep -q '"redis"' && echo "$BODY" | grep -q '"instance_manager"'; then + echo "✓ /status response contains redis and instance_manager sections" + else + echo "✗ /status response missing expected sections" + exit 1 + fi + + - name: Test /healthz returns 503 when Redis is killed + run: | + echo "Testing /healthz returns 503 when Redis process dies..." + + # Start port-forward + kubectl port-forward pod/rfr-test-im-0 8080:8080 & + PF_PID=$! + sleep 2 + + # Verify healthy first + HTTP_CODE=$(curl -s -o /dev/null -w "%{http_code}" http://localhost:8080/healthz) + if [[ "$HTTP_CODE" != "200" ]]; then + echo "✗ Initial /healthz check failed with $HTTP_CODE" + kill $PF_PID 2>/dev/null || true + exit 1 + fi + echo "✓ Initial /healthz is healthy" + + # Kill redis-server process (instance manager will detect this) + echo "Killing redis-server process..." + kubectl exec rfr-test-im-0 -- /bin/sh -c "kill \$(pgrep redis-server)" || true + + # Wait for health check to detect (checks every 1s) + echo "Waiting for health check to detect failure..." + sleep 3 + + # Check that /healthz now returns 503 + RESPONSE=$(curl -s -w "\n%{http_code}" http://localhost:8080/healthz) + HTTP_CODE=$(echo "$RESPONSE" | tail -1) + BODY=$(echo "$RESPONSE" | head -n -1) + + kill $PF_PID 2>/dev/null || true + + echo "Response after kill: $BODY" + echo "HTTP Code: $HTTP_CODE" + + if [[ "$HTTP_CODE" == "503" ]]; then + echo "✓ /healthz returned 503 after Redis killed" + else + echo "⚠ /healthz returned $HTTP_CODE (pod may have restarted already)" + fi + + - name: Wait for pod recovery after health test + run: | + echo "Waiting for pod to recover..." + sleep 5 + kubectl wait --for=condition=Ready pod/rfr-test-im-0 --timeout=120s + echo "✓ Pod recovered and is Ready" + + # Verify health endpoints work again + kubectl port-forward pod/rfr-test-im-0 8080:8080 & + PF_PID=$! + sleep 2 + + HTTP_CODE=$(curl -s -o /dev/null -w "%{http_code}" http://localhost:8080/healthz) + kill $PF_PID 2>/dev/null || true + + if [[ "$HTTP_CODE" == "200" ]]; then + echo "✓ /healthz returns 200 after recovery" + else + echo "✗ /healthz returned $HTTP_CODE after recovery" + exit 1 + fi + - name: Collect logs on failure if: failure() run: | diff --git a/cmd/instance/run/cmd.go b/cmd/instance/run/cmd.go index 9f22033d9..7305b2aed 100644 --- a/cmd/instance/run/cmd.go +++ b/cmd/instance/run/cmd.go @@ -28,6 +28,9 @@ import ( "github.com/spf13/cobra" ) +// healthServer is the global health server instance +var healthServer *HealthServer + const ( defaultDataDir = "/data" defaultDBFilename = "dump.rdb" @@ -44,6 +47,8 @@ var ( dataDir string dbFilename string redisConf string + healthPort int + redisPort string ) // NewCmd creates the run command @@ -77,6 +82,8 @@ This architecture provides: cmd.Flags().StringVar(&dataDir, "data-dir", defaultDataDir, "Redis data directory") cmd.Flags().StringVar(&dbFilename, "db-filename", defaultDBFilename, "Main RDB filename to preserve during cleanup") cmd.Flags().StringVar(&redisConf, "redis-conf", defaultRedisConf, "Path to redis.conf") + cmd.Flags().IntVar(&healthPort, "health-port", defaultHealthPort, "Port for health check endpoints") + cmd.Flags().StringVar(&redisPort, "redis-port", "6379", "Redis port for health checks") return cmd } @@ -94,12 +101,27 @@ func runInstance(cmd *cobra.Command, args []string) error { go runZombieReaper(ctx) // Step 2: Perform startup cleanup - if err := performStartupCleanup(); err != nil { + cleanupErr := performStartupCleanup() + if cleanupErr != nil { // Log but don't fail - Redis should still be able to start - fmt.Printf("redis-instance: warning: startup cleanup failed: %v\n", err) + fmt.Printf("redis-instance: warning: startup cleanup failed: %v\n", cleanupErr) + } + + // Step 3: Start health server (provides /healthz, /readyz, /status) + healthServer = NewHealthServer(healthPort, redisPort) + healthServer.SetCleanupDone(cleanupErr == nil) + if err := healthServer.Start(ctx); err != nil { + fmt.Printf("redis-instance: warning: failed to start health server: %v\n", err) } + defer func() { + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer shutdownCancel() + if err := healthServer.Stop(shutdownCtx); err != nil { + fmt.Printf("redis-instance: warning: health server stop error: %v\n", err) + } + }() - // Step 3: Main process loop (CNPG pattern) + // Step 4: Main process loop (CNPG pattern) // This loop allows for process restarts without manager exit return runProcessLoop(ctx, cancel) } @@ -127,6 +149,11 @@ func runProcessLoop(ctx context.Context, cancel context.CancelFunc) error { redisPid := redisCmd.Process.Pid fmt.Printf("redis-instance: redis-server started with PID %d\n", redisPid) + // Notify health server of Redis PID + if healthServer != nil { + healthServer.SetRedisPID(redisPid) + } + // Wait for either Redis to exit or a signal doneChan := make(chan error, 1) go func() { diff --git a/cmd/instance/run/health.go b/cmd/instance/run/health.go new file mode 100644 index 000000000..f9a3b8aa9 --- /dev/null +++ b/cmd/instance/run/health.go @@ -0,0 +1,419 @@ +// Package run implements the instance manager run command. +package run + +import ( + "context" + "encoding/json" + "fmt" + "net" + "net/http" + "sync" + "sync/atomic" + "time" + + "github.com/go-redis/redis/v8" +) + +const ( + defaultHealthPort = 8080 + healthCheckInterval = time.Second + redisConnectTimeout = 2 * time.Second + redisCommandTimeout = time.Second +) + +// HealthServer provides HTTP health endpoints for the instance manager. +// It maintains a persistent connection to Redis and caches health status +// to minimize load on Redis. +type HealthServer struct { + port int + redisAddr string + redisPort string + server *http.Server + client *redis.Client + + // Cached status (updated every healthCheckInterval) + mu sync.RWMutex + lastCheck time.Time + cachedInfo map[string]string + redisPid int + startTime time.Time + cleanupDone bool + + // Atomic flags + redisHealthy atomic.Bool + redisReady atomic.Bool +} + +// HealthResponse is the response for /healthz endpoint +type HealthResponse struct { + Status string `json:"status"` + RedisPID int `json:"redis_pid,omitempty"` + UptimeSeconds int64 `json:"uptime_seconds"` + Error string `json:"error,omitempty"` +} + +// ReadyResponse is the response for /readyz endpoint +type ReadyResponse struct { + Status string `json:"status"` + Role string `json:"role,omitempty"` + ConnectedClients int `json:"connected_clients,omitempty"` + Loading bool `json:"loading"` + MasterSyncInProgress bool `json:"master_sync_in_progress,omitempty"` + Error string `json:"error,omitempty"` +} + +// StatusResponse is the detailed response for /status endpoint +type StatusResponse struct { + Redis RedisStatus `json:"redis"` + Replication ReplicationStatus `json:"replication"` + InstanceManager InstanceManagerStatus `json:"instance_manager"` +} + +// RedisStatus contains Redis server status +type RedisStatus struct { + PID int `json:"pid"` + Role string `json:"role"` + ConnectedClients int `json:"connected_clients"` + UsedMemory string `json:"used_memory"` + UsedMemoryHuman string `json:"used_memory_human"` + Loading bool `json:"loading"` + RDBBgsaveInProgress bool `json:"rdb_bgsave_in_progress"` + AOFRewriteInProgress bool `json:"aof_rewrite_in_progress"` +} + +// ReplicationStatus contains replication information +type ReplicationStatus struct { + Role string `json:"role"` + ConnectedSlaves int `json:"connected_slaves,omitempty"` + MasterHost string `json:"master_host,omitempty"` + MasterPort int `json:"master_port,omitempty"` + MasterLinkStatus string `json:"master_link_status,omitempty"` + MasterSyncInProgress bool `json:"master_sync_in_progress,omitempty"` + SlaveReplOffset int64 `json:"slave_repl_offset,omitempty"` + MasterReplOffset int64 `json:"master_repl_offset,omitempty"` +} + +// InstanceManagerStatus contains instance manager status +type InstanceManagerStatus struct { + Version string `json:"version"` + UptimeSeconds int64 `json:"uptime_seconds"` + StartupCleanupDone bool `json:"startup_cleanup_done"` + HealthPort int `json:"health_port"` +} + +// NewHealthServer creates a new health server +func NewHealthServer(port int, redisPort string) *HealthServer { + return &HealthServer{ + port: port, + redisAddr: "127.0.0.1", + redisPort: redisPort, + startTime: time.Now(), + cachedInfo: make(map[string]string), + } +} + +// SetRedisPID updates the Redis process ID +func (h *HealthServer) SetRedisPID(pid int) { + h.mu.Lock() + defer h.mu.Unlock() + h.redisPid = pid +} + +// SetCleanupDone marks startup cleanup as complete +func (h *HealthServer) SetCleanupDone(done bool) { + h.mu.Lock() + defer h.mu.Unlock() + h.cleanupDone = done +} + +// Start begins the health server and background health checker +func (h *HealthServer) Start(ctx context.Context) error { + // Create Redis client + h.client = redis.NewClient(&redis.Options{ + Addr: net.JoinHostPort(h.redisAddr, h.redisPort), + DialTimeout: redisConnectTimeout, + ReadTimeout: redisCommandTimeout, + WriteTimeout: redisCommandTimeout, + }) + + // Set up HTTP routes + mux := http.NewServeMux() + mux.HandleFunc("/healthz", h.handleHealthz) + mux.HandleFunc("/readyz", h.handleReadyz) + mux.HandleFunc("/status", h.handleStatus) + + h.server = &http.Server{ + Addr: fmt.Sprintf(":%d", h.port), + Handler: mux, + ReadTimeout: 5 * time.Second, + WriteTimeout: 5 * time.Second, + } + + // Start background health checker + go h.runHealthChecker(ctx) + + // Start HTTP server + fmt.Printf("redis-instance: starting health server on port %d\n", h.port) + go func() { + if err := h.server.ListenAndServe(); err != nil && err != http.ErrServerClosed { + fmt.Printf("redis-instance: health server error: %v\n", err) + } + }() + + return nil +} + +// Stop gracefully stops the health server +func (h *HealthServer) Stop(ctx context.Context) error { + if h.server != nil { + if err := h.server.Shutdown(ctx); err != nil { + return fmt.Errorf("health server shutdown error: %w", err) + } + } + if h.client != nil { + if err := h.client.Close(); err != nil { + return fmt.Errorf("redis client close error: %w", err) + } + } + return nil +} + +// runHealthChecker periodically checks Redis health and caches the results +func (h *HealthServer) runHealthChecker(ctx context.Context) { + ticker := time.NewTicker(healthCheckInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + h.updateHealthStatus(ctx) + } + } +} + +// updateHealthStatus checks Redis and updates cached status +func (h *HealthServer) updateHealthStatus(ctx context.Context) { + checkCtx, cancel := context.WithTimeout(ctx, redisCommandTimeout) + defer cancel() + + // Try to ping Redis + if err := h.client.Ping(checkCtx).Err(); err != nil { + h.redisHealthy.Store(false) + h.redisReady.Store(false) + return + } + h.redisHealthy.Store(true) + + // Get INFO for detailed status + info, err := h.client.Info(checkCtx).Result() + if err != nil { + h.redisReady.Store(false) + return + } + + // Parse INFO output + h.mu.Lock() + h.cachedInfo = parseRedisInfo(info) + h.lastCheck = time.Now() + h.mu.Unlock() + + // Determine readiness + // Not ready if: loading, syncing from master with no master link + loading := h.cachedInfo["loading"] == "1" + syncInProgress := h.cachedInfo["master_sync_in_progress"] == "1" + masterLinkDown := h.cachedInfo["master_link_status"] == "down" + + // Replica is not ready if syncing or master link is down + isReplica := h.cachedInfo["role"] == "slave" + + ready := !loading + if isReplica { + ready = ready && !syncInProgress && !masterLinkDown + } + + h.redisReady.Store(ready) +} + +// handleHealthz handles the /healthz liveness endpoint +func (h *HealthServer) handleHealthz(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + h.mu.RLock() + pid := h.redisPid + h.mu.RUnlock() + + uptime := int64(time.Since(h.startTime).Seconds()) + + resp := HealthResponse{ + RedisPID: pid, + UptimeSeconds: uptime, + } + + if h.redisHealthy.Load() { + resp.Status = "ok" + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + } else { + resp.Status = "unhealthy" + resp.Error = "redis not responding to PING" + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusServiceUnavailable) + } + + _ = json.NewEncoder(w).Encode(resp) +} + +// handleReadyz handles the /readyz readiness endpoint +func (h *HealthServer) handleReadyz(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + h.mu.RLock() + info := h.cachedInfo + h.mu.RUnlock() + + resp := ReadyResponse{ + Role: info["role"], + Loading: info["loading"] == "1", + MasterSyncInProgress: info["master_sync_in_progress"] == "1", + } + + if clients, ok := info["connected_clients"]; ok { + _, _ = fmt.Sscanf(clients, "%d", &resp.ConnectedClients) + } + + if h.redisReady.Load() { + resp.Status = "ok" + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + } else { + resp.Status = "not ready" + if resp.Loading { + resp.Error = "redis is loading data" + } else if resp.MasterSyncInProgress { + resp.Error = "replica sync in progress" + } else if info["master_link_status"] == "down" { + resp.Error = "master link is down" + } else if !h.redisHealthy.Load() { + resp.Error = "redis not responding" + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusServiceUnavailable) + } + + _ = json.NewEncoder(w).Encode(resp) +} + +// handleStatus handles the /status detailed status endpoint +func (h *HealthServer) handleStatus(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + h.mu.RLock() + info := h.cachedInfo + pid := h.redisPid + cleanupDone := h.cleanupDone + h.mu.RUnlock() + + resp := StatusResponse{ + Redis: RedisStatus{ + PID: pid, + Role: info["role"], + UsedMemory: info["used_memory"], + UsedMemoryHuman: info["used_memory_human"], + Loading: info["loading"] == "1", + RDBBgsaveInProgress: info["rdb_bgsave_in_progress"] == "1", + AOFRewriteInProgress: info["aof_rewrite_in_progress"] == "1", + }, + Replication: ReplicationStatus{ + Role: info["role"], + MasterHost: info["master_host"], + MasterLinkStatus: info["master_link_status"], + MasterSyncInProgress: info["master_sync_in_progress"] == "1", + }, + InstanceManager: InstanceManagerStatus{ + Version: "v1.7.0", + UptimeSeconds: int64(time.Since(h.startTime).Seconds()), + StartupCleanupDone: cleanupDone, + HealthPort: h.port, + }, + } + + // Parse integer fields + _, _ = fmt.Sscanf(info["connected_clients"], "%d", &resp.Redis.ConnectedClients) + _, _ = fmt.Sscanf(info["connected_slaves"], "%d", &resp.Replication.ConnectedSlaves) + _, _ = fmt.Sscanf(info["master_port"], "%d", &resp.Replication.MasterPort) + _, _ = fmt.Sscanf(info["slave_repl_offset"], "%d", &resp.Replication.SlaveReplOffset) + _, _ = fmt.Sscanf(info["master_repl_offset"], "%d", &resp.Replication.MasterReplOffset) + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(resp) +} + +// parseRedisInfo parses Redis INFO command output into a map +func parseRedisInfo(info string) map[string]string { + result := make(map[string]string) + lines := splitLines(info) + + for _, line := range lines { + // Skip comments and empty lines + if len(line) == 0 || line[0] == '#' { + continue + } + + // Parse key:value + idx := indexByte(line, ':') + if idx > 0 { + key := line[:idx] + value := line[idx+1:] + result[key] = value + } + } + + return result +} + +// splitLines splits a string by newlines +func splitLines(s string) []string { + var lines []string + start := 0 + for i := 0; i < len(s); i++ { + if s[i] == '\n' { + line := s[start:i] + // Remove trailing \r if present + if len(line) > 0 && line[len(line)-1] == '\r' { + line = line[:len(line)-1] + } + lines = append(lines, line) + start = i + 1 + } + } + if start < len(s) { + line := s[start:] + if len(line) > 0 && line[len(line)-1] == '\r' { + line = line[:len(line)-1] + } + lines = append(lines, line) + } + return lines +} + +// indexByte returns the index of the first occurrence of c in s, or -1 +func indexByte(s string, c byte) int { + for i := 0; i < len(s); i++ { + if s[i] == c { + return i + } + } + return -1 +} diff --git a/cmd/instance/run/health_test.go b/cmd/instance/run/health_test.go new file mode 100644 index 000000000..5c2123012 --- /dev/null +++ b/cmd/instance/run/health_test.go @@ -0,0 +1,357 @@ +package run + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestParseRedisInfo(t *testing.T) { + info := `# Server +redis_version:7.0.0 +redis_git_sha1:00000000 +process_id:1234 + +# Clients +connected_clients:5 + +# Memory +used_memory:1024000 +used_memory_human:1000K + +# Replication +role:master +connected_slaves:2 + +# Persistence +loading:0 +rdb_bgsave_in_progress:0 +aof_rewrite_in_progress:0 +` + + result := parseRedisInfo(info) + + assert.Equal(t, "7.0.0", result["redis_version"]) + assert.Equal(t, "1234", result["process_id"]) + assert.Equal(t, "5", result["connected_clients"]) + assert.Equal(t, "1024000", result["used_memory"]) + assert.Equal(t, "1000K", result["used_memory_human"]) + assert.Equal(t, "master", result["role"]) + assert.Equal(t, "2", result["connected_slaves"]) + assert.Equal(t, "0", result["loading"]) + assert.Equal(t, "0", result["rdb_bgsave_in_progress"]) +} + +func TestParseRedisInfoReplica(t *testing.T) { + info := `# Replication +role:slave +master_host:10.0.0.1 +master_port:6379 +master_link_status:up +master_sync_in_progress:0 +slave_repl_offset:12345 +` + + result := parseRedisInfo(info) + + assert.Equal(t, "slave", result["role"]) + assert.Equal(t, "10.0.0.1", result["master_host"]) + assert.Equal(t, "6379", result["master_port"]) + assert.Equal(t, "up", result["master_link_status"]) + assert.Equal(t, "0", result["master_sync_in_progress"]) + assert.Equal(t, "12345", result["slave_repl_offset"]) +} + +func TestParseRedisInfoLoading(t *testing.T) { + info := `# Persistence +loading:1 +loading_total_bytes:1000000 +loading_loaded_bytes:500000 +` + + result := parseRedisInfo(info) + + assert.Equal(t, "1", result["loading"]) + assert.Equal(t, "1000000", result["loading_total_bytes"]) + assert.Equal(t, "500000", result["loading_loaded_bytes"]) +} + +func TestHealthServerHealthzHealthy(t *testing.T) { + h := NewHealthServer(8080, "6379") + h.startTime = time.Now().Add(-60 * time.Second) // 60 seconds ago + h.SetRedisPID(1234) + h.redisHealthy.Store(true) + + req := httptest.NewRequest(http.MethodGet, "/healthz", nil) + w := httptest.NewRecorder() + + h.handleHealthz(w, req) + + assert.Equal(t, http.StatusOK, w.Code) + + var resp HealthResponse + err := json.Unmarshal(w.Body.Bytes(), &resp) + require.NoError(t, err) + + assert.Equal(t, "ok", resp.Status) + assert.Equal(t, 1234, resp.RedisPID) + assert.GreaterOrEqual(t, resp.UptimeSeconds, int64(60)) +} + +func TestHealthServerHealthzUnhealthy(t *testing.T) { + h := NewHealthServer(8080, "6379") + h.redisHealthy.Store(false) + + req := httptest.NewRequest(http.MethodGet, "/healthz", nil) + w := httptest.NewRecorder() + + h.handleHealthz(w, req) + + assert.Equal(t, http.StatusServiceUnavailable, w.Code) + + var resp HealthResponse + err := json.Unmarshal(w.Body.Bytes(), &resp) + require.NoError(t, err) + + assert.Equal(t, "unhealthy", resp.Status) + assert.Contains(t, resp.Error, "not responding") +} + +func TestHealthServerReadyzReady(t *testing.T) { + h := NewHealthServer(8080, "6379") + h.redisReady.Store(true) + h.redisHealthy.Store(true) + + h.mu.Lock() + h.cachedInfo = map[string]string{ + "role": "master", + "connected_clients": "10", + "loading": "0", + } + h.mu.Unlock() + + req := httptest.NewRequest(http.MethodGet, "/readyz", nil) + w := httptest.NewRecorder() + + h.handleReadyz(w, req) + + assert.Equal(t, http.StatusOK, w.Code) + + var resp ReadyResponse + err := json.Unmarshal(w.Body.Bytes(), &resp) + require.NoError(t, err) + + assert.Equal(t, "ok", resp.Status) + assert.Equal(t, "master", resp.Role) + assert.Equal(t, 10, resp.ConnectedClients) + assert.False(t, resp.Loading) +} + +func TestHealthServerReadyzNotReadyLoading(t *testing.T) { + h := NewHealthServer(8080, "6379") + h.redisReady.Store(false) + h.redisHealthy.Store(true) + + h.mu.Lock() + h.cachedInfo = map[string]string{ + "role": "master", + "loading": "1", + } + h.mu.Unlock() + + req := httptest.NewRequest(http.MethodGet, "/readyz", nil) + w := httptest.NewRecorder() + + h.handleReadyz(w, req) + + assert.Equal(t, http.StatusServiceUnavailable, w.Code) + + var resp ReadyResponse + err := json.Unmarshal(w.Body.Bytes(), &resp) + require.NoError(t, err) + + assert.Equal(t, "not ready", resp.Status) + assert.True(t, resp.Loading) + assert.Contains(t, resp.Error, "loading") +} + +func TestHealthServerReadyzNotReadySyncing(t *testing.T) { + h := NewHealthServer(8080, "6379") + h.redisReady.Store(false) + h.redisHealthy.Store(true) + + h.mu.Lock() + h.cachedInfo = map[string]string{ + "role": "slave", + "loading": "0", + "master_sync_in_progress": "1", + "master_link_status": "up", + } + h.mu.Unlock() + + req := httptest.NewRequest(http.MethodGet, "/readyz", nil) + w := httptest.NewRecorder() + + h.handleReadyz(w, req) + + assert.Equal(t, http.StatusServiceUnavailable, w.Code) + + var resp ReadyResponse + err := json.Unmarshal(w.Body.Bytes(), &resp) + require.NoError(t, err) + + assert.Equal(t, "not ready", resp.Status) + assert.True(t, resp.MasterSyncInProgress) + assert.Contains(t, resp.Error, "sync") +} + +func TestHealthServerReadyzNotReadyMasterLinkDown(t *testing.T) { + h := NewHealthServer(8080, "6379") + h.redisReady.Store(false) + h.redisHealthy.Store(true) + + h.mu.Lock() + h.cachedInfo = map[string]string{ + "role": "slave", + "loading": "0", + "master_sync_in_progress": "0", + "master_link_status": "down", + } + h.mu.Unlock() + + req := httptest.NewRequest(http.MethodGet, "/readyz", nil) + w := httptest.NewRecorder() + + h.handleReadyz(w, req) + + assert.Equal(t, http.StatusServiceUnavailable, w.Code) + + var resp ReadyResponse + err := json.Unmarshal(w.Body.Bytes(), &resp) + require.NoError(t, err) + + assert.Contains(t, resp.Error, "master link") +} + +func TestHealthServerStatus(t *testing.T) { + h := NewHealthServer(8080, "6379") + h.startTime = time.Now().Add(-120 * time.Second) + h.SetRedisPID(5678) + h.SetCleanupDone(true) + + h.mu.Lock() + h.cachedInfo = map[string]string{ + "role": "master", + "connected_clients": "15", + "used_memory": "2048000", + "used_memory_human": "2M", + "loading": "0", + "rdb_bgsave_in_progress": "0", + "aof_rewrite_in_progress": "0", + "connected_slaves": "2", + "master_repl_offset": "99999", + } + h.mu.Unlock() + + req := httptest.NewRequest(http.MethodGet, "/status", nil) + w := httptest.NewRecorder() + + h.handleStatus(w, req) + + assert.Equal(t, http.StatusOK, w.Code) + + var resp StatusResponse + err := json.Unmarshal(w.Body.Bytes(), &resp) + require.NoError(t, err) + + // Redis status + assert.Equal(t, 5678, resp.Redis.PID) + assert.Equal(t, "master", resp.Redis.Role) + assert.Equal(t, 15, resp.Redis.ConnectedClients) + assert.Equal(t, "2M", resp.Redis.UsedMemoryHuman) + assert.False(t, resp.Redis.Loading) + assert.False(t, resp.Redis.RDBBgsaveInProgress) + + // Replication status + assert.Equal(t, "master", resp.Replication.Role) + assert.Equal(t, 2, resp.Replication.ConnectedSlaves) + assert.Equal(t, int64(99999), resp.Replication.MasterReplOffset) + + // Instance manager status + assert.Equal(t, "v1.7.0", resp.InstanceManager.Version) + assert.GreaterOrEqual(t, resp.InstanceManager.UptimeSeconds, int64(120)) + assert.True(t, resp.InstanceManager.StartupCleanupDone) + assert.Equal(t, 8080, resp.InstanceManager.HealthPort) +} + +func TestHealthServerMethodNotAllowed(t *testing.T) { + h := NewHealthServer(8080, "6379") + + endpoints := []string{"/healthz", "/readyz", "/status"} + methods := []string{http.MethodPost, http.MethodPut, http.MethodDelete} + + for _, endpoint := range endpoints { + for _, method := range methods { + req := httptest.NewRequest(method, endpoint, nil) + w := httptest.NewRecorder() + + switch endpoint { + case "/healthz": + h.handleHealthz(w, req) + case "/readyz": + h.handleReadyz(w, req) + case "/status": + h.handleStatus(w, req) + } + + assert.Equal(t, http.StatusMethodNotAllowed, w.Code, + "expected 405 for %s %s", method, endpoint) + } + } +} + +func TestSplitLines(t *testing.T) { + tests := []struct { + input string + expected []string + }{ + {"a\nb\nc", []string{"a", "b", "c"}}, + {"a\r\nb\r\nc", []string{"a", "b", "c"}}, + {"single", []string{"single"}}, + {"a\nb", []string{"a", "b"}}, + } + + for _, tt := range tests { + result := splitLines(tt.input) + assert.Equal(t, tt.expected, result, "input: %q", tt.input) + } +} + +func TestIndexByte(t *testing.T) { + assert.Equal(t, 3, indexByte("foo:bar", ':')) + assert.Equal(t, -1, indexByte("foobar", ':')) + assert.Equal(t, 0, indexByte(":foo", ':')) +} + +func TestHealthServerStartStop(t *testing.T) { + h := NewHealthServer(0, "6379") // Port 0 = random available port + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // Start should not error (even without Redis) + err := h.Start(ctx) + assert.NoError(t, err) + + // Stop should not error + stopCtx, stopCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer stopCancel() + err = h.Stop(stopCtx) + assert.NoError(t, err) +}