From 175b6851aa7bc58938c85be507ee8461f2df7348 Mon Sep 17 00:00:00 2001 From: taskbot Date: Thu, 18 Dec 2025 15:08:40 +0100 Subject: [PATCH 1/3] Add core health monitoring infrastructure for vmcp backends Implement health checking and status tracking for virtual MCP server backends. This provides the foundation for monitoring backend availability and categorizing failure modes (unhealthy, degraded, unauthenticated). Related-to: #3036 --- pkg/vmcp/health/checker.go | 154 +++++++++ pkg/vmcp/health/checker_test.go | 504 ++++++++++++++++++++++++++++++ pkg/vmcp/health/monitor.go | 324 +++++++++++++++++++ pkg/vmcp/health/monitor_test.go | 532 ++++++++++++++++++++++++++++++++ pkg/vmcp/health/status.go | 244 +++++++++++++++ pkg/vmcp/health/status_test.go | 493 +++++++++++++++++++++++++++++ 6 files changed, 2251 insertions(+) create mode 100644 pkg/vmcp/health/checker.go create mode 100644 pkg/vmcp/health/checker_test.go create mode 100644 pkg/vmcp/health/monitor.go create mode 100644 pkg/vmcp/health/monitor_test.go create mode 100644 pkg/vmcp/health/status.go create mode 100644 pkg/vmcp/health/status_test.go diff --git a/pkg/vmcp/health/checker.go b/pkg/vmcp/health/checker.go new file mode 100644 index 0000000000..e9405a6656 --- /dev/null +++ b/pkg/vmcp/health/checker.go @@ -0,0 +1,154 @@ +// Package health provides health monitoring for vMCP backend MCP servers. +// +// This package implements the HealthChecker interface and provides periodic +// health monitoring with configurable intervals and failure thresholds. +package health + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/stacklok/toolhive/pkg/logger" + "github.com/stacklok/toolhive/pkg/vmcp" +) + +// healthChecker implements vmcp.HealthChecker using ListCapabilities as the health check. +type healthChecker struct { + // client is the backend client used to communicate with backends. + client vmcp.BackendClient + + // timeout is the timeout for health check operations. + timeout time.Duration +} + +// NewHealthChecker creates a new health checker that uses BackendClient.ListCapabilities +// as the health check mechanism. This validates the full MCP communication stack: +// network connectivity, MCP protocol compliance, authentication, and responsiveness. +// +// Parameters: +// - client: BackendClient for communicating with backend MCP servers +// - timeout: Maximum duration for health check operations (0 = no timeout) +// +// Returns a new HealthChecker implementation. +func NewHealthChecker(client vmcp.BackendClient, timeout time.Duration) vmcp.HealthChecker { + return &healthChecker{ + client: client, + timeout: timeout, + } +} + +// CheckHealth performs a health check on a backend by calling ListCapabilities. +// This validates the full MCP communication stack and returns the backend's health status. +// +// Health determination logic: +// - Success: Backend is healthy (BackendHealthy) +// - Authentication error: Backend is unauthenticated (BackendUnauthenticated) +// - Timeout or connection error: Backend is unhealthy (BackendUnhealthy) +// - Other errors: Backend is unhealthy (BackendUnhealthy) +// +// The error return is informational and provides context about what failed. +// The BackendHealthStatus return indicates the categorized health state. +func (h *healthChecker) CheckHealth(ctx context.Context, target *vmcp.BackendTarget) (vmcp.BackendHealthStatus, error) { + // Apply timeout if configured + checkCtx := ctx + var cancel context.CancelFunc + if h.timeout > 0 { + checkCtx, cancel = context.WithTimeout(ctx, h.timeout) + defer cancel() + } + + logger.Debugf("Performing health check for backend %s (%s)", target.WorkloadName, target.BaseURL) + + // Use ListCapabilities as the health check - it performs: + // 1. Client creation with transport setup + // 2. MCP protocol initialization handshake + // 3. Capabilities query (tools, resources, prompts) + // This validates the full communication stack + _, err := h.client.ListCapabilities(checkCtx, target) + if err != nil { + // Categorize the error to determine health status + status := categorizeError(err) + logger.Debugf("Health check failed for backend %s: %v (status: %s)", + target.WorkloadName, err, status) + return status, fmt.Errorf("health check failed: %w", err) + } + + logger.Debugf("Health check succeeded for backend %s", target.WorkloadName) + return vmcp.BackendHealthy, nil +} + +// categorizeError determines the appropriate health status based on the error type. +// This helps distinguish between different failure modes (auth, timeout, connectivity, etc.). +func categorizeError(err error) vmcp.BackendHealthStatus { + if err == nil { + return vmcp.BackendHealthy + } + + // Check error message for common patterns + errMsg := err.Error() + + // Authentication failures + if isAuthError(errMsg) { + return vmcp.BackendUnauthenticated + } + + // Timeout and connection errors + if isTimeoutError(errMsg) || isConnectionError(errMsg) { + return vmcp.BackendUnhealthy + } + + // Default to unhealthy for unknown errors + return vmcp.BackendUnhealthy +} + +// isAuthError checks if the error message indicates an authentication failure. +// Uses more specific patterns to avoid false positives from substrings in hostnames, URLs, etc. +func isAuthError(errMsg string) bool { + errLower := strings.ToLower(errMsg) + + // Check for explicit authentication failure messages + if strings.Contains(errLower, "authentication failed") || + strings.Contains(errLower, "authentication error") { + return true + } + + // Check for HTTP 401/403 status codes with context + // Match patterns like "401 Unauthorized", "HTTP 401", "status code 401" + if strings.Contains(errLower, "401 unauthorized") || + strings.Contains(errLower, "403 forbidden") || + strings.Contains(errLower, "http 401") || + strings.Contains(errLower, "http 403") || + strings.Contains(errLower, "status code 401") || + strings.Contains(errLower, "status code 403") { + return true + } + + // Check for explicit unauthenticated/unauthorized errors + // Use word boundaries to avoid matching hostnames + if strings.Contains(errLower, "request unauthenticated") || + strings.Contains(errLower, "request unauthorized") || + strings.Contains(errLower, "access denied") { + return true + } + + return false +} + +// isTimeoutError checks if the error message indicates a timeout. +func isTimeoutError(errMsg string) bool { + errLower := strings.ToLower(errMsg) + return strings.Contains(errLower, "timeout") || + strings.Contains(errLower, "deadline exceeded") || + strings.Contains(errLower, "context deadline exceeded") +} + +// isConnectionError checks if the error message indicates a connection failure. +func isConnectionError(errMsg string) bool { + errLower := strings.ToLower(errMsg) + return strings.Contains(errLower, "connection refused") || + strings.Contains(errLower, "connection reset") || + strings.Contains(errLower, "no route to host") || + strings.Contains(errLower, "network is unreachable") +} diff --git a/pkg/vmcp/health/checker_test.go b/pkg/vmcp/health/checker_test.go new file mode 100644 index 0000000000..a5e6d9dfcb --- /dev/null +++ b/pkg/vmcp/health/checker_test.go @@ -0,0 +1,504 @@ +package health + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" + + "github.com/stacklok/toolhive/pkg/vmcp" + "github.com/stacklok/toolhive/pkg/vmcp/mocks" +) + +func TestNewHealthChecker(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + t.Cleanup(ctrl.Finish) + + mockClient := mocks.NewMockBackendClient(ctrl) + + tests := []struct { + name string + timeout time.Duration + }{ + { + name: "with timeout", + timeout: 5 * time.Second, + }, + { + name: "with zero timeout", + timeout: 0, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + checker := NewHealthChecker(mockClient, tt.timeout) + require.NotNil(t, checker) + + // Type assert to access internals for verification + hc, ok := checker.(*healthChecker) + require.True(t, ok) + assert.Equal(t, mockClient, hc.client) + assert.Equal(t, tt.timeout, hc.timeout) + }) + } +} + +func TestHealthChecker_CheckHealth_Success(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockClient := mocks.NewMockBackendClient(ctrl) + mockClient.EXPECT(). + ListCapabilities(gomock.Any(), gomock.Any()). + Return(&vmcp.CapabilityList{}, nil). + Times(1) + + checker := NewHealthChecker(mockClient, 5*time.Second) + target := &vmcp.BackendTarget{ + WorkloadID: "backend-1", + WorkloadName: "test-backend", + BaseURL: "http://localhost:8080", + } + + status, err := checker.CheckHealth(context.Background(), target) + assert.NoError(t, err) + assert.Equal(t, vmcp.BackendHealthy, status) +} + +func TestHealthChecker_CheckHealth_ContextCancellation(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockClient := mocks.NewMockBackendClient(ctrl) + mockClient.EXPECT(). + ListCapabilities(gomock.Any(), gomock.Any()). + DoAndReturn(func(ctx context.Context, _ *vmcp.BackendTarget) (*vmcp.CapabilityList, error) { + <-ctx.Done() + return nil, ctx.Err() + }). + Times(1) + + checker := NewHealthChecker(mockClient, 100*time.Millisecond) + target := &vmcp.BackendTarget{ + WorkloadID: "backend-1", + WorkloadName: "test-backend", + BaseURL: "http://localhost:8080", + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // Cancel immediately + + status, err := checker.CheckHealth(ctx, target) + assert.Error(t, err) + assert.Equal(t, vmcp.BackendUnhealthy, status) +} + +func TestHealthChecker_CheckHealth_NoTimeout(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockClient := mocks.NewMockBackendClient(ctrl) + mockClient.EXPECT(). + ListCapabilities(gomock.Any(), gomock.Any()). + Return(&vmcp.CapabilityList{}, nil). + Times(1) + + // Create checker with no timeout + checker := NewHealthChecker(mockClient, 0) + target := &vmcp.BackendTarget{ + WorkloadID: "backend-1", + WorkloadName: "test-backend", + BaseURL: "http://localhost:8080", + } + + status, err := checker.CheckHealth(context.Background(), target) + assert.NoError(t, err) + assert.Equal(t, vmcp.BackendHealthy, status) +} + +func TestHealthChecker_CheckHealth_ErrorCategorization(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + expectedStatus vmcp.BackendHealthStatus + description string + }{ + { + name: "timeout error", + err: fmt.Errorf("context deadline exceeded"), + expectedStatus: vmcp.BackendUnhealthy, + description: "should categorize timeout as unhealthy", + }, + { + name: "connection refused", + err: fmt.Errorf("connection refused"), + expectedStatus: vmcp.BackendUnhealthy, + description: "should categorize connection error as unhealthy", + }, + { + name: "authentication failed", + err: fmt.Errorf("authentication failed: invalid token"), + expectedStatus: vmcp.BackendUnauthenticated, + description: "should categorize auth failure as unauthenticated", + }, + { + name: "401 unauthorized", + err: fmt.Errorf("HTTP 401 unauthorized"), + expectedStatus: vmcp.BackendUnauthenticated, + description: "should categorize 401 as unauthenticated", + }, + { + name: "403 forbidden", + err: fmt.Errorf("403 forbidden"), + expectedStatus: vmcp.BackendUnauthenticated, + description: "should categorize 403 as unauthenticated", + }, + { + name: "status code 401", + err: fmt.Errorf("status code 401"), + expectedStatus: vmcp.BackendUnauthenticated, + description: "should recognize status code format", + }, + { + name: "request unauthenticated", + err: fmt.Errorf("request unauthenticated"), + expectedStatus: vmcp.BackendUnauthenticated, + description: "should recognize request unauthenticated", + }, + { + name: "access denied", + err: fmt.Errorf("access denied"), + expectedStatus: vmcp.BackendUnauthenticated, + description: "should recognize access denied", + }, + { + name: "generic error", + err: fmt.Errorf("unknown error"), + expectedStatus: vmcp.BackendUnhealthy, + description: "should default unknown errors to unhealthy", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockClient := mocks.NewMockBackendClient(ctrl) + mockClient.EXPECT(). + ListCapabilities(gomock.Any(), gomock.Any()). + Return(nil, tt.err). + Times(1) + + checker := NewHealthChecker(mockClient, 5*time.Second) + target := &vmcp.BackendTarget{ + WorkloadID: "backend-1", + WorkloadName: "test-backend", + BaseURL: "http://localhost:8080", + } + + status, err := checker.CheckHealth(context.Background(), target) + assert.Error(t, err, tt.description) + assert.Equal(t, tt.expectedStatus, status, tt.description) + }) + } +} + +func TestCategorizeError(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + expectedStatus vmcp.BackendHealthStatus + }{ + { + name: "nil error", + err: nil, + expectedStatus: vmcp.BackendHealthy, + }, + { + name: "authentication failed", + err: errors.New("authentication failed"), + expectedStatus: vmcp.BackendUnauthenticated, + }, + { + name: "authentication error", + err: errors.New("authentication error: invalid credentials"), + expectedStatus: vmcp.BackendUnauthenticated, + }, + { + name: "request unauthorized", + err: errors.New("request unauthorized"), + expectedStatus: vmcp.BackendUnauthenticated, + }, + { + name: "HTTP 401", + err: errors.New("HTTP 401"), + expectedStatus: vmcp.BackendUnauthenticated, + }, + { + name: "HTTP 403", + err: errors.New("HTTP 403"), + expectedStatus: vmcp.BackendUnauthenticated, + }, + { + name: "timeout", + err: errors.New("request timeout"), + expectedStatus: vmcp.BackendUnhealthy, + }, + { + name: "deadline exceeded", + err: errors.New("context deadline exceeded"), + expectedStatus: vmcp.BackendUnhealthy, + }, + { + name: "connection refused", + err: errors.New("connection refused"), + expectedStatus: vmcp.BackendUnhealthy, + }, + { + name: "connection reset", + err: errors.New("connection reset by peer"), + expectedStatus: vmcp.BackendUnhealthy, + }, + { + name: "no route to host", + err: errors.New("no route to host"), + expectedStatus: vmcp.BackendUnhealthy, + }, + { + name: "network unreachable", + err: errors.New("network is unreachable"), + expectedStatus: vmcp.BackendUnhealthy, + }, + { + name: "generic error", + err: errors.New("something went wrong"), + expectedStatus: vmcp.BackendUnhealthy, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + status := categorizeError(tt.err) + assert.Equal(t, tt.expectedStatus, status) + }) + } +} + +func TestIsAuthError(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + errMsg string + expectErr bool + }{ + // Positive cases + {name: "authentication failed", errMsg: "authentication failed", expectErr: true}, + {name: "Authentication Failed (uppercase)", errMsg: "Authentication Failed", expectErr: true}, + {name: "authentication error", errMsg: "authentication error: bad token", expectErr: true}, + {name: "401 unauthorized", errMsg: "401 unauthorized", expectErr: true}, + {name: "403 forbidden", errMsg: "403 forbidden", expectErr: true}, + {name: "HTTP 401", errMsg: "HTTP 401", expectErr: true}, + {name: "HTTP 403", errMsg: "HTTP 403", expectErr: true}, + {name: "status code 401", errMsg: "status code 401", expectErr: true}, + {name: "status code 403", errMsg: "status code 403", expectErr: true}, + {name: "request unauthenticated", errMsg: "request unauthenticated", expectErr: true}, + {name: "request unauthorized", errMsg: "request unauthorized", expectErr: true}, + {name: "access denied", errMsg: "access denied", expectErr: true}, + + // Negative cases - should NOT be detected as auth errors + {name: "connection refused", errMsg: "connection refused", expectErr: false}, + {name: "timeout", errMsg: "request timeout", expectErr: false}, + {name: "generic error", errMsg: "something went wrong", expectErr: false}, + {name: "404 not found", errMsg: "404 not found", expectErr: false}, + {name: "500 internal server error", errMsg: "500 internal server error", expectErr: false}, + {name: "hostname with 401", errMsg: "http://backend401.example.com", expectErr: false}, + {name: "empty string", errMsg: "", expectErr: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + result := isAuthError(tt.errMsg) + assert.Equal(t, tt.expectErr, result) + }) + } +} + +func TestIsTimeoutError(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + errMsg string + expectErr bool + }{ + {name: "timeout", errMsg: "request timeout", expectErr: true}, + {name: "deadline exceeded", errMsg: "deadline exceeded", expectErr: true}, + {name: "context deadline exceeded", errMsg: "context deadline exceeded", expectErr: true}, + {name: "Timeout (uppercase)", errMsg: "Request Timeout", expectErr: true}, + {name: "connection refused", errMsg: "connection refused", expectErr: false}, + {name: "generic error", errMsg: "something went wrong", expectErr: false}, + {name: "empty string", errMsg: "", expectErr: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + result := isTimeoutError(tt.errMsg) + assert.Equal(t, tt.expectErr, result) + }) + } +} + +func TestIsConnectionError(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + errMsg string + expectErr bool + }{ + {name: "connection refused", errMsg: "connection refused", expectErr: true}, + {name: "connection reset", errMsg: "connection reset by peer", expectErr: true}, + {name: "no route to host", errMsg: "no route to host", expectErr: true}, + {name: "network unreachable", errMsg: "network is unreachable", expectErr: true}, + {name: "Connection Refused (uppercase)", errMsg: "Connection Refused", expectErr: true}, + {name: "timeout", errMsg: "request timeout", expectErr: false}, + {name: "authentication failed", errMsg: "authentication failed", expectErr: false}, + {name: "generic error", errMsg: "something went wrong", expectErr: false}, + {name: "empty string", errMsg: "", expectErr: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + result := isConnectionError(tt.errMsg) + assert.Equal(t, tt.expectErr, result) + }) + } +} + +func TestHealthChecker_CheckHealth_Timeout(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockClient := mocks.NewMockBackendClient(ctrl) + mockClient.EXPECT(). + ListCapabilities(gomock.Any(), gomock.Any()). + DoAndReturn(func(ctx context.Context, _ *vmcp.BackendTarget) (*vmcp.CapabilityList, error) { + // Simulate slow backend + select { + case <-time.After(2 * time.Second): + return &vmcp.CapabilityList{}, nil + case <-ctx.Done(): + return nil, ctx.Err() + } + }). + Times(1) + + checker := NewHealthChecker(mockClient, 100*time.Millisecond) + target := &vmcp.BackendTarget{ + WorkloadID: "backend-1", + WorkloadName: "test-backend", + BaseURL: "http://localhost:8080", + } + + status, err := checker.CheckHealth(context.Background(), target) + assert.Error(t, err) + assert.Equal(t, vmcp.BackendUnhealthy, status) +} + +func TestHealthChecker_CheckHealth_MultipleBackends(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockClient := mocks.NewMockBackendClient(ctrl) + + // Setup different responses for different backends + mockClient.EXPECT(). + ListCapabilities(gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, target *vmcp.BackendTarget) (*vmcp.CapabilityList, error) { + switch target.WorkloadID { + case "backend-healthy": + return &vmcp.CapabilityList{}, nil + case "backend-auth-error": + return nil, errors.New("authentication failed") + case "backend-timeout": + return nil, errors.New("context deadline exceeded") + default: + return nil, errors.New("unknown error") + } + }). + Times(4) + + checker := NewHealthChecker(mockClient, 5*time.Second) + + // Test healthy backend + status, err := checker.CheckHealth(context.Background(), &vmcp.BackendTarget{ + WorkloadID: "backend-healthy", + WorkloadName: "Healthy Backend", + BaseURL: "http://localhost:8080", + }) + assert.NoError(t, err) + assert.Equal(t, vmcp.BackendHealthy, status) + + // Test auth error backend + status, err = checker.CheckHealth(context.Background(), &vmcp.BackendTarget{ + WorkloadID: "backend-auth-error", + WorkloadName: "Auth Error Backend", + BaseURL: "http://localhost:8081", + }) + assert.Error(t, err) + assert.Equal(t, vmcp.BackendUnauthenticated, status) + + // Test timeout backend + status, err = checker.CheckHealth(context.Background(), &vmcp.BackendTarget{ + WorkloadID: "backend-timeout", + WorkloadName: "Timeout Backend", + BaseURL: "http://localhost:8082", + }) + assert.Error(t, err) + assert.Equal(t, vmcp.BackendUnhealthy, status) + + // Test unknown error backend + status, err = checker.CheckHealth(context.Background(), &vmcp.BackendTarget{ + WorkloadID: "backend-unknown", + WorkloadName: "Unknown Backend", + BaseURL: "http://localhost:8083", + }) + assert.Error(t, err) + assert.Equal(t, vmcp.BackendUnhealthy, status) +} diff --git a/pkg/vmcp/health/monitor.go b/pkg/vmcp/health/monitor.go new file mode 100644 index 0000000000..a6aca6585c --- /dev/null +++ b/pkg/vmcp/health/monitor.go @@ -0,0 +1,324 @@ +package health + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/stacklok/toolhive/pkg/logger" + "github.com/stacklok/toolhive/pkg/vmcp" +) + +// healthCheckContextKey is a marker for health check requests. +// When present in context, authentication should be bypassed. +type healthCheckContextKey struct{} + +// WithHealthCheckMarker marks a context as a health check request. +// Authentication layers should skip authentication for these requests. +func WithHealthCheckMarker(ctx context.Context) context.Context { + return context.WithValue(ctx, healthCheckContextKey{}, true) +} + +// IsHealthCheck returns true if the context is marked as a health check. +func IsHealthCheck(ctx context.Context) bool { + val, ok := ctx.Value(healthCheckContextKey{}).(bool) + return ok && val +} + +// Monitor performs periodic health checks on backend MCP servers. +// It runs background goroutines for each backend, tracking their health status +// and consecutive failure counts. The monitor supports graceful shutdown and +// provides thread-safe access to backend health information. +type Monitor struct { + // checker performs health checks on backends. + checker vmcp.HealthChecker + + // statusTracker tracks health status for all backends. + statusTracker *statusTracker + + // checkInterval is how often to perform health checks. + checkInterval time.Duration + + // backends is the list of backends to monitor. + backends []vmcp.Backend + + // ctx is the context for the monitor's lifecycle. + ctx context.Context + + // cancel cancels all health check goroutines. + cancel context.CancelFunc + + // wg tracks running health check goroutines. + wg sync.WaitGroup + + // mu protects the started and stopped flags. + mu sync.Mutex + + // started indicates if the monitor has been started. + started bool + + // stopped indicates if the monitor has been stopped (cannot be restarted). + stopped bool +} + +// MonitorConfig contains configuration for the health monitor. +type MonitorConfig struct { + // CheckInterval is how often to perform health checks. + // Must be > 0. Recommended: 30s. + CheckInterval time.Duration + + // UnhealthyThreshold is the number of consecutive failures before marking unhealthy. + // Must be >= 1. Recommended: 3 failures. + UnhealthyThreshold int + + // Timeout is the maximum duration for a single health check operation. + // Zero means no timeout (not recommended). + Timeout time.Duration +} + +// DefaultConfig returns sensible default configuration values. +func DefaultConfig() MonitorConfig { + return MonitorConfig{ + CheckInterval: 30 * time.Second, + UnhealthyThreshold: 3, + Timeout: 10 * time.Second, + } +} + +// NewMonitor creates a new health monitor for the given backends. +// +// Parameters: +// - client: BackendClient for communicating with backend MCP servers +// - backends: List of backends to monitor +// - config: Configuration for health monitoring +// +// Returns (monitor, error). Error is returned if configuration is invalid. +func NewMonitor( + client vmcp.BackendClient, + backends []vmcp.Backend, + config MonitorConfig, +) (*Monitor, error) { + // Validate configuration + if config.CheckInterval <= 0 { + return nil, fmt.Errorf("check interval must be > 0, got %v", config.CheckInterval) + } + if config.UnhealthyThreshold < 1 { + return nil, fmt.Errorf("unhealthy threshold must be >= 1, got %d", config.UnhealthyThreshold) + } + + // Create health checker + checker := NewHealthChecker(client, config.Timeout) + + // Create status tracker + statusTracker := newStatusTracker(config.UnhealthyThreshold) + + return &Monitor{ + checker: checker, + statusTracker: statusTracker, + checkInterval: config.CheckInterval, + backends: backends, + }, nil +} + +// Start begins health monitoring for all backends. +// This spawns a background goroutine for each backend that performs periodic health checks. +// Returns an error if the monitor is already started, has been stopped, or if the parent context is invalid. +// +// The monitor respects the parent context for cancellation. When the parent context is +// cancelled, all health check goroutines will stop gracefully. +// +// Note: A monitor cannot be restarted after it has been stopped. Create a new monitor instead. +func (m *Monitor) Start(ctx context.Context) error { + m.mu.Lock() + defer m.mu.Unlock() + + if m.stopped { + return fmt.Errorf("monitor has been stopped and cannot be restarted") + } + + if m.started { + return fmt.Errorf("monitor already started") + } + + if ctx == nil { + return fmt.Errorf("context cannot be nil") + } + + // Create monitor context with cancellation + m.ctx, m.cancel = context.WithCancel(ctx) + m.started = true + + logger.Infof("Starting health monitor for %d backends (interval: %v, threshold: %d)", + len(m.backends), m.checkInterval, m.statusTracker.unhealthyThreshold) + + // Start health check goroutine for each backend + for i := range m.backends { + backend := &m.backends[i] // Capture backend pointer for this iteration + m.wg.Add(1) + go m.monitorBackend(m.ctx, backend) + } + + return nil +} + +// Stop gracefully stops health monitoring. +// This cancels all health check goroutines and waits for them to complete. +// Returns an error if the monitor was not started. +// +// After stopping, the monitor cannot be restarted. Create a new monitor if needed. +func (m *Monitor) Stop() error { + m.mu.Lock() + if !m.started { + m.mu.Unlock() + return fmt.Errorf("monitor not started") + } + + // Cancel all health check goroutines + logger.Infof("Stopping health monitor for %d backends", len(m.backends)) + m.cancel() + m.started = false + m.stopped = true + m.mu.Unlock() + + // Wait for all goroutines to complete + m.wg.Wait() + logger.Info("Health monitor stopped") + + return nil +} + +// monitorBackend performs periodic health checks for a single backend. +// This runs in a background goroutine and continues until the context is cancelled. +func (m *Monitor) monitorBackend(ctx context.Context, backend *vmcp.Backend) { + defer m.wg.Done() + + logger.Debugf("Starting health monitoring for backend %s", backend.Name) + + // Create ticker for periodic checks + ticker := time.NewTicker(m.checkInterval) + defer ticker.Stop() + + // Perform initial health check immediately + m.performHealthCheck(ctx, backend) + + // Periodic health check loop + for { + select { + case <-ctx.Done(): + logger.Debugf("Stopping health monitoring for backend %s", backend.Name) + return + + case <-ticker.C: + m.performHealthCheck(ctx, backend) + } + } +} + +// performHealthCheck performs a single health check for a backend and updates status. +func (m *Monitor) performHealthCheck(ctx context.Context, backend *vmcp.Backend) { + // Create BackendTarget from Backend + target := &vmcp.BackendTarget{ + WorkloadID: backend.ID, + WorkloadName: backend.Name, + BaseURL: backend.BaseURL, + TransportType: backend.TransportType, + AuthConfig: backend.AuthConfig, + HealthStatus: vmcp.BackendUnknown, // Status is determined by the health check + Metadata: backend.Metadata, + } + + // Mark context as health check to bypass authentication + // Health checks verify backend availability and should not require user credentials + healthCheckCtx := WithHealthCheckMarker(ctx) + + // Perform health check + status, err := m.checker.CheckHealth(healthCheckCtx, target) + + // Record result in status tracker + if err != nil { + m.statusTracker.RecordFailure(backend.ID, backend.Name, status, err) + } else { + m.statusTracker.RecordSuccess(backend.ID, backend.Name) + } +} + +// GetBackendStatus returns the current health status for a backend. +// Returns (status, error). Error is returned if the backend is not being monitored. +func (m *Monitor) GetBackendStatus(backendID string) (vmcp.BackendHealthStatus, error) { + status, exists := m.statusTracker.GetStatus(backendID) + if !exists { + return vmcp.BackendUnknown, fmt.Errorf("backend %s not found", backendID) + } + return status, nil +} + +// GetBackendState returns the full health state for a backend. +// Returns (state, error). Error is returned if the backend is not being monitored. +func (m *Monitor) GetBackendState(backendID string) (*State, error) { + state, exists := m.statusTracker.GetState(backendID) + if !exists { + return nil, fmt.Errorf("backend %s not found", backendID) + } + return state, nil +} + +// GetAllBackendStates returns health states for all monitored backends. +// Returns a map of backend ID to State. +func (m *Monitor) GetAllBackendStates() map[string]*State { + return m.statusTracker.GetAllStates() +} + +// IsBackendHealthy returns true if the backend is currently healthy. +// Returns false if the backend is not being monitored or is unhealthy. +func (m *Monitor) IsBackendHealthy(backendID string) bool { + return m.statusTracker.IsHealthy(backendID) +} + +// GetHealthSummary returns a summary of backend health for logging/monitoring. +// Returns counts of healthy, unhealthy, and total backends. +func (m *Monitor) GetHealthSummary() Summary { + allStates := m.statusTracker.GetAllStates() + + summary := Summary{ + Total: len(allStates), + Healthy: 0, + Unhealthy: 0, + Degraded: 0, + Unknown: 0, + Unauthenticated: 0, + } + + for _, state := range allStates { + switch state.Status { + case vmcp.BackendHealthy: + summary.Healthy++ + case vmcp.BackendUnhealthy: + summary.Unhealthy++ + case vmcp.BackendDegraded: + summary.Degraded++ + case vmcp.BackendUnknown: + summary.Unknown++ + case vmcp.BackendUnauthenticated: + summary.Unauthenticated++ + } + } + + return summary +} + +// Summary provides aggregate health statistics for all backends. +type Summary struct { + Total int + Healthy int + Unhealthy int + Degraded int + Unknown int + Unauthenticated int +} + +// String returns a human-readable summary. +func (s Summary) String() string { + return fmt.Sprintf("total=%d healthy=%d unhealthy=%d degraded=%d unknown=%d unauthenticated=%d", + s.Total, s.Healthy, s.Unhealthy, s.Degraded, s.Unknown, s.Unauthenticated) +} diff --git a/pkg/vmcp/health/monitor_test.go b/pkg/vmcp/health/monitor_test.go new file mode 100644 index 0000000000..221ab2c7be --- /dev/null +++ b/pkg/vmcp/health/monitor_test.go @@ -0,0 +1,532 @@ +package health + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" + + "github.com/stacklok/toolhive/pkg/vmcp" + "github.com/stacklok/toolhive/pkg/vmcp/mocks" +) + +func TestNewMonitor_Validation(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + t.Cleanup(ctrl.Finish) + + mockClient := mocks.NewMockBackendClient(ctrl) + backends := []vmcp.Backend{ + {ID: "backend-1", Name: "Backend 1", BaseURL: "http://localhost:8080"}, + } + + tests := []struct { + name string + config MonitorConfig + expectError bool + }{ + { + name: "valid config", + config: MonitorConfig{ + CheckInterval: 30 * time.Second, + UnhealthyThreshold: 3, + Timeout: 10 * time.Second, + }, + expectError: false, + }, + { + name: "invalid check interval", + config: MonitorConfig{ + CheckInterval: 0, + UnhealthyThreshold: 3, + Timeout: 10 * time.Second, + }, + expectError: true, + }, + { + name: "invalid unhealthy threshold", + config: MonitorConfig{ + CheckInterval: 30 * time.Second, + UnhealthyThreshold: 0, + Timeout: 10 * time.Second, + }, + expectError: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + monitor, err := NewMonitor(mockClient, backends, tt.config) + if tt.expectError { + assert.Error(t, err) + assert.Nil(t, monitor) + } else { + assert.NoError(t, err) + assert.NotNil(t, monitor) + } + }) + } +} + +func TestMonitor_StartStop(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockClient := mocks.NewMockBackendClient(ctrl) + backends := []vmcp.Backend{ + {ID: "backend-1", Name: "Backend 1", BaseURL: "http://localhost:8080", TransportType: "sse"}, + } + + config := MonitorConfig{ + CheckInterval: 100 * time.Millisecond, + UnhealthyThreshold: 3, + Timeout: 50 * time.Millisecond, + } + + // Mock health check calls + mockClient.EXPECT(). + ListCapabilities(gomock.Any(), gomock.Any()). + Return(&vmcp.CapabilityList{}, nil). + AnyTimes() + + monitor, err := NewMonitor(mockClient, backends, config) + require.NoError(t, err) + + // Start monitor + ctx := context.Background() + err = monitor.Start(ctx) + require.NoError(t, err) + + // Wait for at least one health check + time.Sleep(150 * time.Millisecond) + + // Verify backend is healthy + assert.True(t, monitor.IsBackendHealthy("backend-1")) + + // Stop monitor + err = monitor.Stop() + require.NoError(t, err) + + // Verify cannot start again without recreating + err = monitor.Start(ctx) + assert.Error(t, err) +} + +func TestMonitor_StartErrors(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + t.Cleanup(ctrl.Finish) + + mockClient := mocks.NewMockBackendClient(ctrl) + backends := []vmcp.Backend{ + {ID: "backend-1", Name: "Backend 1", BaseURL: "http://localhost:8080"}, + } + + config := MonitorConfig{ + CheckInterval: 100 * time.Millisecond, + UnhealthyThreshold: 3, + Timeout: 50 * time.Millisecond, + } + + tests := []struct { + name string + setupFunc func(*Monitor) error + expectErr bool + }{ + { + name: "nil context", + setupFunc: func(m *Monitor) error { + return m.Start(nil) //nolint:staticcheck // Testing nil context error handling + }, + expectErr: true, + }, + { + name: "already started", + setupFunc: func(m *Monitor) error { + mockClient.EXPECT(). + ListCapabilities(gomock.Any(), gomock.Any()). + Return(&vmcp.CapabilityList{}, nil). + AnyTimes() + + ctx := context.Background() + if err := m.Start(ctx); err != nil { + return err + } + // Try to start again - should return error + err := m.Start(ctx) + // Stop the monitor since it was started successfully the first time + _ = m.Stop() + return err + }, + expectErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + monitor, err := NewMonitor(mockClient, backends, config) + require.NoError(t, err) + + err = tt.setupFunc(monitor) + if tt.expectErr { + assert.Error(t, err) + } else { + assert.NoError(t, err) + } + }) + } +} + +func TestMonitor_StopWithoutStart(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockClient := mocks.NewMockBackendClient(ctrl) + backends := []vmcp.Backend{ + {ID: "backend-1", Name: "Backend 1", BaseURL: "http://localhost:8080"}, + } + + config := MonitorConfig{ + CheckInterval: 100 * time.Millisecond, + UnhealthyThreshold: 3, + Timeout: 50 * time.Millisecond, + } + + monitor, err := NewMonitor(mockClient, backends, config) + require.NoError(t, err) + + // Try to stop without starting + err = monitor.Stop() + assert.Error(t, err) +} + +func TestMonitor_PeriodicHealthChecks(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockClient := mocks.NewMockBackendClient(ctrl) + backends := []vmcp.Backend{ + {ID: "backend-1", Name: "Backend 1", BaseURL: "http://localhost:8080", TransportType: "sse"}, + } + + config := MonitorConfig{ + CheckInterval: 50 * time.Millisecond, + UnhealthyThreshold: 2, + Timeout: 10 * time.Millisecond, + } + + // Mock health check to fail + mockClient.EXPECT(). + ListCapabilities(gomock.Any(), gomock.Any()). + Return(nil, errors.New("backend unavailable")). + MinTimes(2) + + monitor, err := NewMonitor(mockClient, backends, config) + require.NoError(t, err) + + ctx := context.Background() + err = monitor.Start(ctx) + require.NoError(t, err) + defer func() { + _ = monitor.Stop() + }() + + // Wait for threshold to be exceeded (2 failures * 50ms + buffer) + time.Sleep(200 * time.Millisecond) + + // Backend should be marked unhealthy + status, err := monitor.GetBackendStatus("backend-1") + assert.NoError(t, err) + assert.Equal(t, vmcp.BackendUnhealthy, status) + + state, err := monitor.GetBackendState("backend-1") + assert.NoError(t, err) + assert.GreaterOrEqual(t, state.ConsecutiveFailures, 2) +} + +func TestMonitor_GetHealthSummary(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockClient := mocks.NewMockBackendClient(ctrl) + backends := []vmcp.Backend{ + {ID: "backend-1", Name: "Backend 1", BaseURL: "http://localhost:8080", TransportType: "sse"}, + {ID: "backend-2", Name: "Backend 2", BaseURL: "http://localhost:8081", TransportType: "sse"}, + } + + config := MonitorConfig{ + CheckInterval: 50 * time.Millisecond, + UnhealthyThreshold: 1, + Timeout: 10 * time.Millisecond, + } + + // Backend 1 succeeds, Backend 2 fails + mockClient.EXPECT(). + ListCapabilities(gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, target *vmcp.BackendTarget) (*vmcp.CapabilityList, error) { + if target.WorkloadID == "backend-1" { + return &vmcp.CapabilityList{}, nil + } + return nil, errors.New("backend unavailable") + }). + AnyTimes() + + monitor, err := NewMonitor(mockClient, backends, config) + require.NoError(t, err) + + ctx := context.Background() + err = monitor.Start(ctx) + require.NoError(t, err) + defer func() { + _ = monitor.Stop() + }() + + // Wait for health checks to complete + time.Sleep(100 * time.Millisecond) + + summary := monitor.GetHealthSummary() + assert.Equal(t, 2, summary.Total) + assert.Equal(t, 1, summary.Healthy) + assert.Equal(t, 1, summary.Unhealthy) +} + +func TestMonitor_GetBackendStatus(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockClient := mocks.NewMockBackendClient(ctrl) + backends := []vmcp.Backend{ + {ID: "backend-1", Name: "Backend 1", BaseURL: "http://localhost:8080", TransportType: "sse"}, + } + + config := MonitorConfig{ + CheckInterval: 100 * time.Millisecond, + UnhealthyThreshold: 3, + Timeout: 50 * time.Millisecond, + } + + mockClient.EXPECT(). + ListCapabilities(gomock.Any(), gomock.Any()). + Return(&vmcp.CapabilityList{}, nil). + AnyTimes() + + monitor, err := NewMonitor(mockClient, backends, config) + require.NoError(t, err) + + ctx := context.Background() + err = monitor.Start(ctx) + require.NoError(t, err) + defer func() { + _ = monitor.Stop() + }() + + time.Sleep(150 * time.Millisecond) + + // Test getting status for existing backend + status, err := monitor.GetBackendStatus("backend-1") + assert.NoError(t, err) + assert.Equal(t, vmcp.BackendHealthy, status) + + // Test getting status for non-existent backend + status, err = monitor.GetBackendStatus("nonexistent") + assert.Error(t, err) + assert.Equal(t, vmcp.BackendUnknown, status) +} + +func TestMonitor_GetBackendState(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockClient := mocks.NewMockBackendClient(ctrl) + backends := []vmcp.Backend{ + {ID: "backend-1", Name: "Backend 1", BaseURL: "http://localhost:8080", TransportType: "sse"}, + } + + config := MonitorConfig{ + CheckInterval: 100 * time.Millisecond, + UnhealthyThreshold: 3, + Timeout: 50 * time.Millisecond, + } + + mockClient.EXPECT(). + ListCapabilities(gomock.Any(), gomock.Any()). + Return(&vmcp.CapabilityList{}, nil). + AnyTimes() + + monitor, err := NewMonitor(mockClient, backends, config) + require.NoError(t, err) + + ctx := context.Background() + err = monitor.Start(ctx) + require.NoError(t, err) + defer func() { + _ = monitor.Stop() + }() + + time.Sleep(150 * time.Millisecond) + + // Test getting state for existing backend + state, err := monitor.GetBackendState("backend-1") + assert.NoError(t, err) + assert.NotNil(t, state) + assert.Equal(t, vmcp.BackendHealthy, state.Status) + + // Test getting state for non-existent backend + state, err = monitor.GetBackendState("nonexistent") + assert.Error(t, err) + assert.Nil(t, state) +} + +func TestMonitor_GetAllBackendStates(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockClient := mocks.NewMockBackendClient(ctrl) + backends := []vmcp.Backend{ + {ID: "backend-1", Name: "Backend 1", BaseURL: "http://localhost:8080", TransportType: "sse"}, + {ID: "backend-2", Name: "Backend 2", BaseURL: "http://localhost:8081", TransportType: "sse"}, + } + + config := MonitorConfig{ + CheckInterval: 100 * time.Millisecond, + UnhealthyThreshold: 3, + Timeout: 50 * time.Millisecond, + } + + mockClient.EXPECT(). + ListCapabilities(gomock.Any(), gomock.Any()). + Return(&vmcp.CapabilityList{}, nil). + AnyTimes() + + monitor, err := NewMonitor(mockClient, backends, config) + require.NoError(t, err) + + ctx := context.Background() + err = monitor.Start(ctx) + require.NoError(t, err) + defer func() { + _ = monitor.Stop() + }() + + time.Sleep(150 * time.Millisecond) + + allStates := monitor.GetAllBackendStates() + assert.Len(t, allStates, 2) + assert.Contains(t, allStates, "backend-1") + assert.Contains(t, allStates, "backend-2") +} + +func TestMonitor_ContextCancellation(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockClient := mocks.NewMockBackendClient(ctrl) + backends := []vmcp.Backend{ + {ID: "backend-1", Name: "Backend 1", BaseURL: "http://localhost:8080", TransportType: "sse"}, + } + + config := MonitorConfig{ + CheckInterval: 50 * time.Millisecond, + UnhealthyThreshold: 3, + Timeout: 10 * time.Millisecond, + } + + mockClient.EXPECT(). + ListCapabilities(gomock.Any(), gomock.Any()). + Return(&vmcp.CapabilityList{}, nil). + AnyTimes() + + monitor, err := NewMonitor(mockClient, backends, config) + require.NoError(t, err) + + // Start with cancellable context + ctx, cancel := context.WithCancel(context.Background()) + err = monitor.Start(ctx) + require.NoError(t, err) + + // Wait for a few health checks + time.Sleep(100 * time.Millisecond) + + // Cancel context + cancel() + + // Wait for goroutines to stop + time.Sleep(100 * time.Millisecond) + + // Monitor should still be running (context cancellation stops checks but doesn't stop the monitor) + // Stop explicitly + err = monitor.Stop() + assert.NoError(t, err) +} + +func TestDefaultConfig(t *testing.T) { + t.Parallel() + + config := DefaultConfig() + assert.Equal(t, 30*time.Second, config.CheckInterval) + assert.Equal(t, 3, config.UnhealthyThreshold) + assert.Equal(t, 10*time.Second, config.Timeout) +} + +func TestHealthCheckMarker(t *testing.T) { + t.Parallel() + + // Test WithHealthCheckMarker + ctx := context.Background() + assert.False(t, IsHealthCheck(ctx)) + + markedCtx := WithHealthCheckMarker(ctx) + assert.True(t, IsHealthCheck(markedCtx)) + + // Test that marker is preserved across context operations + cancelCtx, cancel := context.WithCancel(markedCtx) + defer cancel() + assert.True(t, IsHealthCheck(cancelCtx)) +} + +func TestSummary_String(t *testing.T) { + t.Parallel() + + summary := Summary{ + Total: 10, + Healthy: 5, + Unhealthy: 2, + Degraded: 1, + Unknown: 1, + Unauthenticated: 1, + } + + str := summary.String() + assert.Contains(t, str, "total=10") + assert.Contains(t, str, "healthy=5") + assert.Contains(t, str, "unhealthy=2") + assert.Contains(t, str, "degraded=1") + assert.Contains(t, str, "unknown=1") + assert.Contains(t, str, "unauthenticated=1") +} diff --git a/pkg/vmcp/health/status.go b/pkg/vmcp/health/status.go new file mode 100644 index 0000000000..f4947fb141 --- /dev/null +++ b/pkg/vmcp/health/status.go @@ -0,0 +1,244 @@ +package health + +import ( + "sync" + "time" + + "github.com/stacklok/toolhive/pkg/logger" + "github.com/stacklok/toolhive/pkg/vmcp" +) + +// backendHealthState tracks the health state of a single backend. +type backendHealthState struct { + // status is the current health status. + status vmcp.BackendHealthStatus + + // consecutiveFailures is the number of consecutive failed health checks. + consecutiveFailures int + + // lastCheckTime is when the last health check was performed. + lastCheckTime time.Time + + // lastError is the last error encountered during health check (if any). + lastError error + + // lastTransitionTime is when the status last changed. + lastTransitionTime time.Time +} + +// statusTracker tracks health status for multiple backends. +// It provides thread-safe access to backend health states and handles +// status transitions with configurable unhealthy thresholds. +type statusTracker struct { + mu sync.RWMutex + + // states maps backend ID to its health state. + states map[string]*backendHealthState + + // unhealthyThreshold is the number of consecutive failures before marking unhealthy. + unhealthyThreshold int +} + +// newStatusTracker creates a new status tracker. +// +// Parameters: +// - unhealthyThreshold: Number of consecutive failures before marking backend unhealthy. +// Must be >= 1. Recommended: 3 failures. +// +// Returns a new status tracker instance. +func newStatusTracker(unhealthyThreshold int) *statusTracker { + if unhealthyThreshold < 1 { + logger.Warnf("Invalid unhealthyThreshold %d (must be >= 1), adjusting to 1", unhealthyThreshold) + unhealthyThreshold = 1 + } + + return &statusTracker{ + states: make(map[string]*backendHealthState), + unhealthyThreshold: unhealthyThreshold, + } +} + +// RecordSuccess records a successful health check for a backend. +// This resets the consecutive failure count and marks the backend as healthy. +// If the backend was previously unhealthy, this transition is logged. +func (t *statusTracker) RecordSuccess(backendID string, backendName string) { + t.mu.Lock() + defer t.mu.Unlock() + + state, exists := t.states[backendID] + if !exists { + // Initialize new state + state = &backendHealthState{ + status: vmcp.BackendHealthy, + consecutiveFailures: 0, + lastCheckTime: time.Now(), + lastError: nil, + lastTransitionTime: time.Now(), + } + t.states[backendID] = state + logger.Debugf("Backend %s initialized as healthy", backendName) + return + } + + // Check for status transition + previousStatus := state.status + previousFailures := state.consecutiveFailures + state.status = vmcp.BackendHealthy + state.consecutiveFailures = 0 + state.lastCheckTime = time.Now() + state.lastError = nil + + // Log transition if status changed + if previousStatus != vmcp.BackendHealthy { + state.lastTransitionTime = time.Now() + logger.Infof("Backend %s health recovered: %s → %s (was failing for %d consecutive checks)", + backendName, previousStatus, vmcp.BackendHealthy, previousFailures) + } +} + +// RecordFailure records a failed health check for a backend. +// This increments the consecutive failure count and may transition the backend to unhealthy +// if the threshold is exceeded. Status transitions are logged. +// +// Parameters: +// - backendID: Unique identifier for the backend +// - backendName: Human-readable name for logging +// - status: The health status returned by the health check (unhealthy, unauthenticated, etc.) +// - err: The error encountered during health check +func (t *statusTracker) RecordFailure(backendID string, backendName string, status vmcp.BackendHealthStatus, err error) { + t.mu.Lock() + defer t.mu.Unlock() + + state, exists := t.states[backendID] + if !exists { + // Initialize new state + state = &backendHealthState{ + status: vmcp.BackendUnknown, + consecutiveFailures: 1, + lastCheckTime: time.Now(), + lastError: err, + lastTransitionTime: time.Now(), + } + t.states[backendID] = state + + // Check if threshold is reached on initialization (e.g., threshold of 1) + if state.consecutiveFailures >= t.unhealthyThreshold { + state.status = status + logger.Warnf("Backend %s initialized with failure and reached threshold: %s (%d/%d failures): %v", + backendName, status, state.consecutiveFailures, t.unhealthyThreshold, err) + } else { + logger.Warnf("Backend %s initialized with failure (1/%d failures, status: %s): %v", + backendName, t.unhealthyThreshold, vmcp.BackendUnknown, err) + } + return + } + + // Record the failure + previousStatus := state.status + state.consecutiveFailures++ + state.lastCheckTime = time.Now() + state.lastError = err + + // Check if threshold is reached and status has changed + thresholdReached := state.consecutiveFailures >= t.unhealthyThreshold + statusChanged := previousStatus != status + + if thresholdReached && statusChanged { + // Transition to new unhealthy status + state.status = status + state.lastTransitionTime = time.Now() + logger.Warnf("Backend %s health degraded: %s → %s (%d consecutive failures, threshold: %d) - last error: %v", + backendName, previousStatus, status, state.consecutiveFailures, t.unhealthyThreshold, err) + } else if thresholdReached { + // Already at threshold with same status - no transition needed + logger.Debugf("Backend %s remains %s (%d consecutive failures, incoming: %s): %v", + backendName, state.status, state.consecutiveFailures, status, err) + } else { + // Below threshold - accumulating failures but not yet unhealthy + logger.Debugf("Backend %s health check failed (%d/%d consecutive failures, current status: %s, incoming: %s): %v", + backendName, state.consecutiveFailures, t.unhealthyThreshold, state.status, status, err) + } +} + +// GetStatus returns the current health status for a backend. +// Returns (status, exists) where exists indicates if the backend is being tracked. +// If the backend is not being tracked, returns (BackendUnknown, false). +func (t *statusTracker) GetStatus(backendID string) (vmcp.BackendHealthStatus, bool) { + t.mu.RLock() + defer t.mu.RUnlock() + + state, exists := t.states[backendID] + if !exists { + return vmcp.BackendUnknown, false + } + + return state.status, true +} + +// GetState returns a copy of the full health state for a backend. +// Returns (state, exists) where exists indicates if the backend is being tracked. +func (t *statusTracker) GetState(backendID string) (*State, bool) { + t.mu.RLock() + defer t.mu.RUnlock() + + state, exists := t.states[backendID] + if !exists { + return nil, false + } + + // Return a copy to avoid race conditions + return &State{ + Status: state.status, + ConsecutiveFailures: state.consecutiveFailures, + LastCheckTime: state.lastCheckTime, + LastError: state.lastError, + LastTransitionTime: state.lastTransitionTime, + }, true +} + +// GetAllStates returns a copy of all backend health states. +// Returns a map of backend ID to State. +func (t *statusTracker) GetAllStates() map[string]*State { + t.mu.RLock() + defer t.mu.RUnlock() + + result := make(map[string]*State, len(t.states)) + for backendID, state := range t.states { + result[backendID] = &State{ + Status: state.status, + ConsecutiveFailures: state.consecutiveFailures, + LastCheckTime: state.lastCheckTime, + LastError: state.lastError, + LastTransitionTime: state.lastTransitionTime, + } + } + + return result +} + +// IsHealthy returns true if the backend is currently healthy. +// Returns false if the backend is unknown or not tracked. +func (t *statusTracker) IsHealthy(backendID string) bool { + status, exists := t.GetStatus(backendID) + return exists && status == vmcp.BackendHealthy +} + +// State is an immutable snapshot of a backend's health state. +// This is returned by GetState and GetAllStates to provide thread-safe access +// to health information without holding locks. +type State struct { + // Status is the current health status. + Status vmcp.BackendHealthStatus + + // ConsecutiveFailures is the number of consecutive failed health checks. + ConsecutiveFailures int + + // LastCheckTime is when the last health check was performed. + LastCheckTime time.Time + + // LastError is the last error encountered (if any). + LastError error + + // LastTransitionTime is when the status last changed. + LastTransitionTime time.Time +} diff --git a/pkg/vmcp/health/status_test.go b/pkg/vmcp/health/status_test.go new file mode 100644 index 0000000000..05fb7a6452 --- /dev/null +++ b/pkg/vmcp/health/status_test.go @@ -0,0 +1,493 @@ +package health + +import ( + "errors" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/stacklok/toolhive/pkg/vmcp" +) + +func TestNewStatusTracker(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + threshold int + expectedThreshold int + description string + }{ + { + name: "valid threshold", + threshold: 3, + expectedThreshold: 3, + description: "should use provided threshold", + }, + { + name: "threshold of 1", + threshold: 1, + expectedThreshold: 1, + description: "should allow threshold of 1", + }, + { + name: "invalid threshold (0)", + threshold: 0, + expectedThreshold: 1, + description: "should adjust invalid threshold to 1", + }, + { + name: "invalid threshold (-1)", + threshold: -1, + expectedThreshold: 1, + description: "should adjust negative threshold to 1", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(tt.threshold) + require.NotNil(t, tracker) + assert.Equal(t, tt.expectedThreshold, tracker.unhealthyThreshold, tt.description) + assert.NotNil(t, tracker.states) + }) + } +} + +func TestStatusTracker_RecordSuccess(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(3) + + // Record success for new backend + tracker.RecordSuccess("backend-1", "Backend 1") + + status, exists := tracker.GetStatus("backend-1") + assert.True(t, exists) + assert.Equal(t, vmcp.BackendHealthy, status) + + state, exists := tracker.GetState("backend-1") + assert.True(t, exists) + assert.Equal(t, vmcp.BackendHealthy, state.Status) + assert.Equal(t, 0, state.ConsecutiveFailures) + assert.Nil(t, state.LastError) + assert.False(t, state.LastCheckTime.IsZero()) + assert.False(t, state.LastTransitionTime.IsZero()) +} + +func TestStatusTracker_RecordSuccess_AfterFailures(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(3) + testErr := errors.New("health check failed") + + // Record multiple failures + for i := 0; i < 5; i++ { + tracker.RecordFailure("backend-1", "Backend 1", vmcp.BackendUnhealthy, testErr) + } + + state, _ := tracker.GetState("backend-1") + assert.Equal(t, vmcp.BackendUnhealthy, state.Status) + assert.Equal(t, 5, state.ConsecutiveFailures) + + // Record success - should reset everything + tracker.RecordSuccess("backend-1", "Backend 1") + + state, _ = tracker.GetState("backend-1") + assert.Equal(t, vmcp.BackendHealthy, state.Status) + assert.Equal(t, 0, state.ConsecutiveFailures) + assert.Nil(t, state.LastError) +} + +func TestStatusTracker_RecordFailure_BelowThreshold(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(3) + testErr := errors.New("health check failed") + + // First failure - should initialize with unknown status (below threshold) + tracker.RecordFailure("backend-1", "Backend 1", vmcp.BackendUnhealthy, testErr) + + state, exists := tracker.GetState("backend-1") + assert.True(t, exists) + assert.Equal(t, vmcp.BackendUnknown, state.Status) + assert.Equal(t, 1, state.ConsecutiveFailures) + assert.NotNil(t, state.LastError) + + // Second failure - still below threshold, status remains unknown + tracker.RecordFailure("backend-1", "Backend 1", vmcp.BackendUnhealthy, testErr) + state, _ = tracker.GetState("backend-1") + assert.Equal(t, vmcp.BackendUnknown, state.Status) + assert.Equal(t, 2, state.ConsecutiveFailures) +} + +func TestStatusTracker_RecordFailure_ReachThreshold(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(3) + testErr := errors.New("health check failed") + + // Record failures up to threshold + for i := 0; i < 3; i++ { + tracker.RecordFailure("backend-1", "Backend 1", vmcp.BackendUnhealthy, testErr) + } + + state, _ := tracker.GetState("backend-1") + assert.Equal(t, vmcp.BackendUnhealthy, state.Status) + assert.Equal(t, 3, state.ConsecutiveFailures) + assert.NotNil(t, state.LastError) + assert.False(t, state.LastTransitionTime.IsZero()) +} + +func TestStatusTracker_RecordFailure_StatusTransitions(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(2) + + // Start with healthy + tracker.RecordSuccess("backend-1", "Backend 1") + status, _ := tracker.GetStatus("backend-1") + assert.Equal(t, vmcp.BackendHealthy, status) + + // First failure - still healthy + tracker.RecordFailure("backend-1", "Backend 1", vmcp.BackendUnhealthy, errors.New("error 1")) + status, _ = tracker.GetStatus("backend-1") + assert.Equal(t, vmcp.BackendHealthy, status) + + // Second failure - should transition to unhealthy + tracker.RecordFailure("backend-1", "Backend 1", vmcp.BackendUnhealthy, errors.New("error 2")) + status, _ = tracker.GetStatus("backend-1") + assert.Equal(t, vmcp.BackendUnhealthy, status) + + // Transition to unauthenticated + tracker.RecordFailure("backend-1", "Backend 1", vmcp.BackendUnauthenticated, errors.New("auth error")) + tracker.RecordFailure("backend-1", "Backend 1", vmcp.BackendUnauthenticated, errors.New("auth error")) + status, _ = tracker.GetStatus("backend-1") + assert.Equal(t, vmcp.BackendUnauthenticated, status) +} + +func TestStatusTracker_RecordFailure_DifferentStatusTypes(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + failureStatus vmcp.BackendHealthStatus + expectedStatus vmcp.BackendHealthStatus + }{ + { + name: "unhealthy failures", + failureStatus: vmcp.BackendUnhealthy, + expectedStatus: vmcp.BackendUnhealthy, + }, + { + name: "unauthenticated failures", + failureStatus: vmcp.BackendUnauthenticated, + expectedStatus: vmcp.BackendUnauthenticated, + }, + { + name: "degraded failures", + failureStatus: vmcp.BackendDegraded, + expectedStatus: vmcp.BackendDegraded, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(2) + testErr := errors.New("test error") + + // Record failures to reach threshold + for i := 0; i < 2; i++ { + tracker.RecordFailure("backend-1", "Backend 1", tt.failureStatus, testErr) + } + + status, _ := tracker.GetStatus("backend-1") + assert.Equal(t, tt.expectedStatus, status) + }) + } +} + +func TestStatusTracker_GetStatus_NonExistent(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(3) + + status, exists := tracker.GetStatus("nonexistent") + assert.False(t, exists) + assert.Equal(t, vmcp.BackendUnknown, status) +} + +func TestStatusTracker_GetState_NonExistent(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(3) + + state, exists := tracker.GetState("nonexistent") + assert.False(t, exists) + assert.Nil(t, state) +} + +func TestStatusTracker_GetAllStates(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(3) + + // Add multiple backends with different states + tracker.RecordSuccess("backend-1", "Backend 1") + + // Record enough failures to reach threshold for backend-2 + for i := 0; i < 3; i++ { + tracker.RecordFailure("backend-2", "Backend 2", vmcp.BackendUnhealthy, errors.New("failed")) + } + + tracker.RecordSuccess("backend-3", "Backend 3") + + allStates := tracker.GetAllStates() + assert.Len(t, allStates, 3) + + assert.Equal(t, vmcp.BackendHealthy, allStates["backend-1"].Status) + assert.Equal(t, vmcp.BackendUnhealthy, allStates["backend-2"].Status) + assert.Equal(t, vmcp.BackendHealthy, allStates["backend-3"].Status) +} + +func TestStatusTracker_GetAllStates_Empty(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(3) + + allStates := tracker.GetAllStates() + assert.NotNil(t, allStates) + assert.Len(t, allStates, 0) +} + +func TestStatusTracker_GetAllStates_Immutability(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(3) + tracker.RecordSuccess("backend-1", "Backend 1") + + // Get states + states1 := tracker.GetAllStates() + states2 := tracker.GetAllStates() + + // Verify they are different copies + assert.NotSame(t, states1["backend-1"], states2["backend-1"]) + + // Modify one copy - should not affect the other + states1["backend-1"].Status = vmcp.BackendUnhealthy + assert.Equal(t, vmcp.BackendHealthy, states2["backend-1"].Status) +} + +func TestStatusTracker_IsHealthy(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(3) + + // Healthy backend + tracker.RecordSuccess("backend-healthy", "Healthy Backend") + assert.True(t, tracker.IsHealthy("backend-healthy")) + + // Unhealthy backend + tracker.RecordFailure("backend-unhealthy", "Unhealthy Backend", + vmcp.BackendUnhealthy, errors.New("failed")) + assert.False(t, tracker.IsHealthy("backend-unhealthy")) + + // Non-existent backend + assert.False(t, tracker.IsHealthy("backend-nonexistent")) +} + +func TestStatusTracker_ConcurrentAccess(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(3) + numGoroutines := 10 + numOperations := 100 + + var wg sync.WaitGroup + wg.Add(numGoroutines * 3) + + // Concurrent RecordSuccess + for i := 0; i < numGoroutines; i++ { + go func(_ int) { + defer wg.Done() + for j := 0; j < numOperations; j++ { + tracker.RecordSuccess("backend-success", "Backend Success") + } + }(i) + } + + // Concurrent RecordFailure + for i := 0; i < numGoroutines; i++ { + go func(_ int) { + defer wg.Done() + for j := 0; j < numOperations; j++ { + tracker.RecordFailure("backend-failure", "Backend Failure", + vmcp.BackendUnhealthy, errors.New("concurrent error")) + } + }(i) + } + + // Concurrent reads + for i := 0; i < numGoroutines; i++ { + go func(_ int) { + defer wg.Done() + for j := 0; j < numOperations; j++ { + _, _ = tracker.GetStatus("backend-success") + _, _ = tracker.GetState("backend-failure") + _ = tracker.GetAllStates() + _ = tracker.IsHealthy("backend-success") + } + }(i) + } + + wg.Wait() + + // Verify states are consistent + status1, exists1 := tracker.GetStatus("backend-success") + assert.True(t, exists1) + assert.Equal(t, vmcp.BackendHealthy, status1) + + status2, exists2 := tracker.GetStatus("backend-failure") + assert.True(t, exists2) + assert.Equal(t, vmcp.BackendUnhealthy, status2) +} + +func TestStatusTracker_StateTimestamps(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(2) + testErr := errors.New("test error") + + // Initial success + tracker.RecordSuccess("backend-1", "Backend 1") + state1, _ := tracker.GetState("backend-1") + initialTransitionTime := state1.LastTransitionTime + + // Wait a bit to ensure time difference + time.Sleep(10 * time.Millisecond) + + // Record failure (no status change yet, below threshold) + tracker.RecordFailure("backend-1", "Backend 1", vmcp.BackendUnhealthy, testErr) + state2, _ := tracker.GetState("backend-1") + + // LastCheckTime should be updated + assert.True(t, state2.LastCheckTime.After(state1.LastCheckTime)) + // LastTransitionTime should NOT change (no status transition) + assert.Equal(t, initialTransitionTime, state2.LastTransitionTime) + + // Wait again + time.Sleep(10 * time.Millisecond) + + // Second failure - should trigger transition + tracker.RecordFailure("backend-1", "Backend 1", vmcp.BackendUnhealthy, testErr) + state3, _ := tracker.GetState("backend-1") + + // LastTransitionTime should be updated (status changed) + assert.True(t, state3.LastTransitionTime.After(initialTransitionTime)) +} + +func TestStatusTracker_MultipleBackends(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(2) + + // Backend 1: Healthy + tracker.RecordSuccess("backend-1", "Backend 1") + + // Backend 2: Unhealthy + for i := 0; i < 2; i++ { + tracker.RecordFailure("backend-2", "Backend 2", vmcp.BackendUnhealthy, errors.New("error")) + } + + // Backend 3: Unauthenticated + for i := 0; i < 2; i++ { + tracker.RecordFailure("backend-3", "Backend 3", vmcp.BackendUnauthenticated, errors.New("auth error")) + } + + // Verify each backend independently + assert.True(t, tracker.IsHealthy("backend-1")) + assert.False(t, tracker.IsHealthy("backend-2")) + assert.False(t, tracker.IsHealthy("backend-3")) + + status2, _ := tracker.GetStatus("backend-2") + assert.Equal(t, vmcp.BackendUnhealthy, status2) + + status3, _ := tracker.GetStatus("backend-3") + assert.Equal(t, vmcp.BackendUnauthenticated, status3) +} + +func TestStatusTracker_RecoveryAfterFailures(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(3) + testErr := errors.New("health check failed") + + // Record 5 failures (well over threshold) + for i := 0; i < 5; i++ { + tracker.RecordFailure("backend-1", "Backend 1", vmcp.BackendUnhealthy, testErr) + } + + state, _ := tracker.GetState("backend-1") + assert.Equal(t, vmcp.BackendUnhealthy, state.Status) + assert.Equal(t, 5, state.ConsecutiveFailures) + beforeRecoveryTransitionTime := state.LastTransitionTime + + // Wait a bit + time.Sleep(10 * time.Millisecond) + + // Single success should recover immediately + tracker.RecordSuccess("backend-1", "Backend 1") + + state, _ = tracker.GetState("backend-1") + assert.Equal(t, vmcp.BackendHealthy, state.Status) + assert.Equal(t, 0, state.ConsecutiveFailures) + assert.Nil(t, state.LastError) + assert.True(t, state.LastTransitionTime.After(beforeRecoveryTransitionTime)) +} + +func TestState_Immutability(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(3) + testErr := errors.New("test error") + + tracker.RecordFailure("backend-1", "Backend 1", vmcp.BackendUnhealthy, testErr) + + // Get state copy + state, exists := tracker.GetState("backend-1") + assert.True(t, exists) + assert.NotNil(t, state) + + // Modify the returned state + originalStatus := state.Status + state.Status = vmcp.BackendHealthy + state.ConsecutiveFailures = 0 + + // Get state again - should be unchanged + state2, _ := tracker.GetState("backend-1") + assert.Equal(t, originalStatus, state2.Status) + assert.NotEqual(t, 0, state2.ConsecutiveFailures) +} + +func TestStatusTracker_ThresholdOf1(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(1) + testErr := errors.New("test error") + + // First failure should immediately mark as unhealthy + tracker.RecordFailure("backend-1", "Backend 1", vmcp.BackendUnhealthy, testErr) + + status, _ := tracker.GetStatus("backend-1") + assert.Equal(t, vmcp.BackendUnhealthy, status) + + state, _ := tracker.GetState("backend-1") + assert.Equal(t, 1, state.ConsecutiveFailures) +} From a0c72440332bab2557e84794c10979182a6290db Mon Sep 17 00:00:00 2001 From: taskbot Date: Thu, 18 Dec 2025 15:38:50 +0100 Subject: [PATCH 2/3] changes from review --- pkg/vmcp/client/client.go | 78 +++++++++++++++++++---- pkg/vmcp/errors.go | 84 ++++++++++++++++++++++++- pkg/vmcp/health/checker.go | 106 +++++++++++++------------------- pkg/vmcp/health/checker_test.go | 98 ++++++++++++++--------------- pkg/vmcp/health/monitor.go | 29 ++++++--- pkg/vmcp/health/monitor_test.go | 5 +- pkg/vmcp/health/status.go | 38 +++++++++--- pkg/vmcp/health/status_test.go | 35 +++++------ pkg/vmcp/registry_test.go | 5 +- pkg/vmcp/types.go | 3 + 10 files changed, 315 insertions(+), 166 deletions(-) diff --git a/pkg/vmcp/client/client.go b/pkg/vmcp/client/client.go index ce4686cef1..c26ee1e48b 100644 --- a/pkg/vmcp/client/client.go +++ b/pkg/vmcp/client/client.go @@ -7,8 +7,10 @@ package client import ( "context" "encoding/base64" + "errors" "fmt" "io" + "net" "net/http" "github.com/mark3labs/mcp-go/client" @@ -239,6 +241,60 @@ func (h *httpBackendClient) defaultClientFactory(ctx context.Context, target *vm return c, nil } +// wrapBackendError wraps an error with the appropriate sentinel error based on error type. +// This enables type-safe error checking with errors.Is() instead of string matching. +// +// Error detection strategy (in order of preference): +// 1. Check for standard Go error types (context errors, net.Error, url.Error) +// 2. Fall back to string pattern matching for library-specific errors (MCP SDK, HTTP libs) +func wrapBackendError(err error, backendID string, operation string) error { + if err == nil { + return nil + } + + // 1. Type-based detection: Check for context deadline/cancellation + if errors.Is(err, context.DeadlineExceeded) { + return fmt.Errorf("%w: failed to %s for backend %s (timeout): %v", + vmcp.ErrTimeout, operation, backendID, err) + } + if errors.Is(err, context.Canceled) { + return fmt.Errorf("%w: failed to %s for backend %s (cancelled): %v", + vmcp.ErrCancelled, operation, backendID, err) + } + + // 2. Type-based detection: Check for net.Error with Timeout() method + // This handles network timeouts from the standard library + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return fmt.Errorf("%w: failed to %s for backend %s (timeout): %v", + vmcp.ErrTimeout, operation, backendID, err) + } + + // 3. String-based detection: Fall back to pattern matching for cases where + // we don't have structured error types (MCP SDK, HTTP libraries with embedded status codes) + // Authentication errors (401, 403, auth failures) + if vmcp.IsAuthenticationError(err) { + return fmt.Errorf("%w: failed to %s for backend %s: %v", + vmcp.ErrAuthenticationFailed, operation, backendID, err) + } + + // Timeout errors (deadline exceeded, timeout messages) + if vmcp.IsTimeoutError(err) { + return fmt.Errorf("%w: failed to %s for backend %s (timeout): %v", + vmcp.ErrTimeout, operation, backendID, err) + } + + // Connection errors (refused, reset, unreachable) + if vmcp.IsConnectionError(err) { + return fmt.Errorf("%w: failed to %s for backend %s (connection error): %v", + vmcp.ErrBackendUnavailable, operation, backendID, err) + } + + // Default to backend unavailable for unknown errors + return fmt.Errorf("%w: failed to %s for backend %s: %v", + vmcp.ErrBackendUnavailable, operation, backendID, err) +} + // initializeClient performs MCP protocol initialization handshake and returns server capabilities. // This allows the caller to determine which optional features the server supports. func initializeClient(ctx context.Context, c *client.Client) (*mcp.ServerCapabilities, error) { @@ -313,14 +369,14 @@ func (h *httpBackendClient) ListCapabilities(ctx context.Context, target *vmcp.B // Create a client for this backend (not yet initialized) c, err := h.clientFactory(ctx, target) if err != nil { - return nil, fmt.Errorf("failed to create client for backend %s: %w", target.WorkloadID, err) + return nil, wrapBackendError(err, target.WorkloadID, "create client") } defer c.Close() // Initialize the client and get server capabilities serverCaps, err := initializeClient(ctx, c) if err != nil { - return nil, fmt.Errorf("failed to initialize client for backend %s: %w", target.WorkloadID, err) + return nil, wrapBackendError(err, target.WorkloadID, "initialize client") } logger.Debugf("Backend %s capabilities: tools=%v, resources=%v, prompts=%v", @@ -330,17 +386,17 @@ func (h *httpBackendClient) ListCapabilities(ctx context.Context, target *vmcp.B // Check for nil BEFORE passing to functions to avoid interface{} nil pointer issues toolsResp, err := queryTools(ctx, c, serverCaps.Tools != nil, target.WorkloadID) if err != nil { - return nil, err + return nil, wrapBackendError(err, target.WorkloadID, "list tools") } resourcesResp, err := queryResources(ctx, c, serverCaps.Resources != nil, target.WorkloadID) if err != nil { - return nil, err + return nil, wrapBackendError(err, target.WorkloadID, "list resources") } promptsResp, err := queryPrompts(ctx, c, serverCaps.Prompts != nil, target.WorkloadID) if err != nil { - return nil, err + return nil, wrapBackendError(err, target.WorkloadID, "list prompts") } // Convert MCP types to vmcp types @@ -428,13 +484,13 @@ func (h *httpBackendClient) CallTool( // Create a client for this backend c, err := h.clientFactory(ctx, target) if err != nil { - return nil, fmt.Errorf("failed to create client for backend %s: %w", target.WorkloadID, err) + return nil, wrapBackendError(err, target.WorkloadID, "create client") } defer c.Close() // Initialize the client if _, err := initializeClient(ctx, c); err != nil { - return nil, fmt.Errorf("failed to initialize client for backend %s: %w", target.WorkloadID, err) + return nil, wrapBackendError(err, target.WorkloadID, "initialize client") } // Call the tool using the original capability name from the backend's perspective. @@ -525,13 +581,13 @@ func (h *httpBackendClient) ReadResource(ctx context.Context, target *vmcp.Backe // Create a client for this backend c, err := h.clientFactory(ctx, target) if err != nil { - return nil, fmt.Errorf("failed to create client for backend %s: %w", target.WorkloadID, err) + return nil, wrapBackendError(err, target.WorkloadID, "create client") } defer c.Close() // Initialize the client if _, err := initializeClient(ctx, c); err != nil { - return nil, fmt.Errorf("failed to initialize client for backend %s: %w", target.WorkloadID, err) + return nil, wrapBackendError(err, target.WorkloadID, "initialize client") } // Read the resource using the original URI from the backend's perspective. @@ -586,13 +642,13 @@ func (h *httpBackendClient) GetPrompt( // Create a client for this backend c, err := h.clientFactory(ctx, target) if err != nil { - return "", fmt.Errorf("failed to create client for backend %s: %w", target.WorkloadID, err) + return "", wrapBackendError(err, target.WorkloadID, "create client") } defer c.Close() // Initialize the client if _, err := initializeClient(ctx, c); err != nil { - return "", fmt.Errorf("failed to initialize client for backend %s: %w", target.WorkloadID, err) + return "", wrapBackendError(err, target.WorkloadID, "initialize client") } // Get the prompt using the original prompt name from the backend's perspective. diff --git a/pkg/vmcp/errors.go b/pkg/vmcp/errors.go index 0f3c71f0f9..c2ebe371b8 100644 --- a/pkg/vmcp/errors.go +++ b/pkg/vmcp/errors.go @@ -1,6 +1,9 @@ package vmcp -import "errors" +import ( + "errors" + "strings" +) // Common domain errors used across vmcp subpackages. // Following DDD principles, domain errors are defined at the package root. @@ -61,3 +64,82 @@ var ( // Wrapping errors should list the conflicting tool names. ErrToolNameConflict = errors.New("tool name conflict") ) + +// Error Categorization Helpers +// +// These functions categorize errors by examining error message strings. +// They serve as a fallback mechanism for error detection when: +// +// 1. Errors come from external libraries that use their own error types and formats +// 2. Legacy code paths don't wrap errors with sentinel errors +// 3. Backwards compatibility is needed for error detection +// +// Note: BackendClient now wraps all errors with appropriate sentinel errors +// (ErrAuthenticationFailed, ErrTimeout, ErrBackendUnavailable). Health monitoring +// code should prefer errors.Is() checks over these string-based functions. +// These functions remain for backwards compatibility and as a fallback mechanism. + +// IsAuthenticationError checks if an error message indicates an authentication failure. +// Uses case-insensitive pattern matching to detect various auth error formats from +// HTTP libraries, MCP protocol errors, and authentication middleware. +func IsAuthenticationError(err error) bool { + if err == nil { + return false + } + + errLower := strings.ToLower(err.Error()) + + // Check for explicit authentication failure messages + if strings.Contains(errLower, "authentication failed") || + strings.Contains(errLower, "authentication error") { + return true + } + + // Check for HTTP 401/403 status codes with context + // Match patterns like "401 Unauthorized", "HTTP 401", "status code 401" + if strings.Contains(errLower, "401 unauthorized") || + strings.Contains(errLower, "403 forbidden") || + strings.Contains(errLower, "http 401") || + strings.Contains(errLower, "http 403") || + strings.Contains(errLower, "status code 401") || + strings.Contains(errLower, "status code 403") { + return true + } + + // Check for explicit unauthenticated/unauthorized errors + if strings.Contains(errLower, "request unauthenticated") || + strings.Contains(errLower, "request unauthorized") || + strings.Contains(errLower, "access denied") { + return true + } + + return false +} + +// IsTimeoutError checks if an error message indicates a timeout. +// Detects various timeout formats from context deadlines, HTTP timeouts, +// and network timeout errors. +func IsTimeoutError(err error) bool { + if err == nil { + return false + } + + errLower := strings.ToLower(err.Error()) + return strings.Contains(errLower, "timeout") || + strings.Contains(errLower, "deadline exceeded") || + strings.Contains(errLower, "context deadline exceeded") +} + +// IsConnectionError checks if an error message indicates a connection failure. +// Detects network-level errors like connection refused, reset, unreachable, etc. +func IsConnectionError(err error) bool { + if err == nil { + return false + } + + errLower := strings.ToLower(err.Error()) + return strings.Contains(errLower, "connection refused") || + strings.Contains(errLower, "connection reset") || + strings.Contains(errLower, "no route to host") || + strings.Contains(errLower, "network is unreachable") +} diff --git a/pkg/vmcp/health/checker.go b/pkg/vmcp/health/checker.go index e9405a6656..9705a0b788 100644 --- a/pkg/vmcp/health/checker.go +++ b/pkg/vmcp/health/checker.go @@ -6,8 +6,8 @@ package health import ( "context" + "errors" "fmt" - "strings" "time" "github.com/stacklok/toolhive/pkg/logger" @@ -21,6 +21,11 @@ type healthChecker struct { // timeout is the timeout for health check operations. timeout time.Duration + + // degradedThreshold is the response time threshold for marking a backend as degraded. + // If a health check succeeds but takes longer than this duration, the backend is marked degraded. + // Zero means disabled (backends will never be marked degraded based on response time alone). + degradedThreshold time.Duration } // NewHealthChecker creates a new health checker that uses BackendClient.ListCapabilities @@ -30,12 +35,14 @@ type healthChecker struct { // Parameters: // - client: BackendClient for communicating with backend MCP servers // - timeout: Maximum duration for health check operations (0 = no timeout) +// - degradedThreshold: Response time threshold for marking backend as degraded (0 = disabled) // // Returns a new HealthChecker implementation. -func NewHealthChecker(client vmcp.BackendClient, timeout time.Duration) vmcp.HealthChecker { +func NewHealthChecker(client vmcp.BackendClient, timeout time.Duration, degradedThreshold time.Duration) vmcp.HealthChecker { return &healthChecker{ - client: client, - timeout: timeout, + client: client, + timeout: timeout, + degradedThreshold: degradedThreshold, } } @@ -43,7 +50,8 @@ func NewHealthChecker(client vmcp.BackendClient, timeout time.Duration) vmcp.Hea // This validates the full MCP communication stack and returns the backend's health status. // // Health determination logic: -// - Success: Backend is healthy (BackendHealthy) +// - Success with fast response: Backend is healthy (BackendHealthy) +// - Success with slow response (> degradedThreshold): Backend is degraded (BackendDegraded) // - Authentication error: Backend is unauthenticated (BackendUnauthenticated) // - Timeout or connection error: Backend is unhealthy (BackendUnhealthy) // - Other errors: Backend is unhealthy (BackendUnhealthy) @@ -61,94 +69,68 @@ func (h *healthChecker) CheckHealth(ctx context.Context, target *vmcp.BackendTar logger.Debugf("Performing health check for backend %s (%s)", target.WorkloadName, target.BaseURL) + // Track response time for degraded detection + startTime := time.Now() + // Use ListCapabilities as the health check - it performs: // 1. Client creation with transport setup // 2. MCP protocol initialization handshake // 3. Capabilities query (tools, resources, prompts) // This validates the full communication stack _, err := h.client.ListCapabilities(checkCtx, target) + responseDuration := time.Since(startTime) + if err != nil { // Categorize the error to determine health status status := categorizeError(err) - logger.Debugf("Health check failed for backend %s: %v (status: %s)", - target.WorkloadName, err, status) + logger.Debugf("Health check failed for backend %s: %v (status: %s, duration: %v)", + target.WorkloadName, err, status, responseDuration) return status, fmt.Errorf("health check failed: %w", err) } - logger.Debugf("Health check succeeded for backend %s", target.WorkloadName) + // Check if response time indicates degraded performance + if h.degradedThreshold > 0 && responseDuration > h.degradedThreshold { + logger.Warnf("Health check succeeded for backend %s but response was slow: %v (threshold: %v) - marking as degraded", + target.WorkloadName, responseDuration, h.degradedThreshold) + return vmcp.BackendDegraded, nil + } + + logger.Debugf("Health check succeeded for backend %s (duration: %v)", target.WorkloadName, responseDuration) return vmcp.BackendHealthy, nil } // categorizeError determines the appropriate health status based on the error type. -// This helps distinguish between different failure modes (auth, timeout, connectivity, etc.). +// This uses sentinel error checking with errors.Is() for type-safe error categorization. +// Falls back to string-based detection for backwards compatibility with non-wrapped errors. func categorizeError(err error) vmcp.BackendHealthStatus { if err == nil { return vmcp.BackendHealthy } - // Check error message for common patterns - errMsg := err.Error() - - // Authentication failures - if isAuthError(errMsg) { + // 1. Type-safe detection: Check for sentinel errors using errors.Is() + // BackendClient now wraps all errors with appropriate sentinel errors + if errors.Is(err, vmcp.ErrAuthenticationFailed) || errors.Is(err, vmcp.ErrAuthorizationFailed) { return vmcp.BackendUnauthenticated } - // Timeout and connection errors - if isTimeoutError(errMsg) || isConnectionError(errMsg) { + if errors.Is(err, vmcp.ErrTimeout) || errors.Is(err, vmcp.ErrCancelled) { return vmcp.BackendUnhealthy } - // Default to unhealthy for unknown errors - return vmcp.BackendUnhealthy -} - -// isAuthError checks if the error message indicates an authentication failure. -// Uses more specific patterns to avoid false positives from substrings in hostnames, URLs, etc. -func isAuthError(errMsg string) bool { - errLower := strings.ToLower(errMsg) - - // Check for explicit authentication failure messages - if strings.Contains(errLower, "authentication failed") || - strings.Contains(errLower, "authentication error") { - return true + if errors.Is(err, vmcp.ErrBackendUnavailable) { + return vmcp.BackendUnhealthy } - // Check for HTTP 401/403 status codes with context - // Match patterns like "401 Unauthorized", "HTTP 401", "status code 401" - if strings.Contains(errLower, "401 unauthorized") || - strings.Contains(errLower, "403 forbidden") || - strings.Contains(errLower, "http 401") || - strings.Contains(errLower, "http 403") || - strings.Contains(errLower, "status code 401") || - strings.Contains(errLower, "status code 403") { - return true + // 2. String-based detection: Fallback for backwards compatibility + // This handles errors from sources that don't wrap with sentinel errors + if vmcp.IsAuthenticationError(err) { + return vmcp.BackendUnauthenticated } - // Check for explicit unauthenticated/unauthorized errors - // Use word boundaries to avoid matching hostnames - if strings.Contains(errLower, "request unauthenticated") || - strings.Contains(errLower, "request unauthorized") || - strings.Contains(errLower, "access denied") { - return true + if vmcp.IsTimeoutError(err) || vmcp.IsConnectionError(err) { + return vmcp.BackendUnhealthy } - return false -} - -// isTimeoutError checks if the error message indicates a timeout. -func isTimeoutError(errMsg string) bool { - errLower := strings.ToLower(errMsg) - return strings.Contains(errLower, "timeout") || - strings.Contains(errLower, "deadline exceeded") || - strings.Contains(errLower, "context deadline exceeded") -} - -// isConnectionError checks if the error message indicates a connection failure. -func isConnectionError(errMsg string) bool { - errLower := strings.ToLower(errMsg) - return strings.Contains(errLower, "connection refused") || - strings.Contains(errLower, "connection reset") || - strings.Contains(errLower, "no route to host") || - strings.Contains(errLower, "network is unreachable") + // Default to unhealthy for unknown errors + return vmcp.BackendUnhealthy } diff --git a/pkg/vmcp/health/checker_test.go b/pkg/vmcp/health/checker_test.go index a5e6d9dfcb..a0515cb3c2 100644 --- a/pkg/vmcp/health/checker_test.go +++ b/pkg/vmcp/health/checker_test.go @@ -41,7 +41,7 @@ func TestNewHealthChecker(t *testing.T) { t.Run(tt.name, func(t *testing.T) { t.Parallel() - checker := NewHealthChecker(mockClient, tt.timeout) + checker := NewHealthChecker(mockClient, tt.timeout, 0) require.NotNil(t, checker) // Type assert to access internals for verification @@ -65,7 +65,7 @@ func TestHealthChecker_CheckHealth_Success(t *testing.T) { Return(&vmcp.CapabilityList{}, nil). Times(1) - checker := NewHealthChecker(mockClient, 5*time.Second) + checker := NewHealthChecker(mockClient, 5*time.Second, 0) target := &vmcp.BackendTarget{ WorkloadID: "backend-1", WorkloadName: "test-backend", @@ -92,7 +92,7 @@ func TestHealthChecker_CheckHealth_ContextCancellation(t *testing.T) { }). Times(1) - checker := NewHealthChecker(mockClient, 100*time.Millisecond) + checker := NewHealthChecker(mockClient, 100*time.Millisecond, 0) target := &vmcp.BackendTarget{ WorkloadID: "backend-1", WorkloadName: "test-backend", @@ -120,7 +120,7 @@ func TestHealthChecker_CheckHealth_NoTimeout(t *testing.T) { Times(1) // Create checker with no timeout - checker := NewHealthChecker(mockClient, 0) + checker := NewHealthChecker(mockClient, 0, 0) target := &vmcp.BackendTarget{ WorkloadID: "backend-1", WorkloadName: "test-backend", @@ -210,7 +210,7 @@ func TestHealthChecker_CheckHealth_ErrorCategorization(t *testing.T) { Return(nil, tt.err). Times(1) - checker := NewHealthChecker(mockClient, 5*time.Second) + checker := NewHealthChecker(mockClient, 5*time.Second, 0) target := &vmcp.BackendTarget{ WorkloadID: "backend-1", WorkloadName: "test-backend", @@ -309,43 +309,43 @@ func TestCategorizeError(t *testing.T) { } } -func TestIsAuthError(t *testing.T) { +func TestIsAuthenticationError(t *testing.T) { t.Parallel() tests := []struct { name string - errMsg string + err error expectErr bool }{ // Positive cases - {name: "authentication failed", errMsg: "authentication failed", expectErr: true}, - {name: "Authentication Failed (uppercase)", errMsg: "Authentication Failed", expectErr: true}, - {name: "authentication error", errMsg: "authentication error: bad token", expectErr: true}, - {name: "401 unauthorized", errMsg: "401 unauthorized", expectErr: true}, - {name: "403 forbidden", errMsg: "403 forbidden", expectErr: true}, - {name: "HTTP 401", errMsg: "HTTP 401", expectErr: true}, - {name: "HTTP 403", errMsg: "HTTP 403", expectErr: true}, - {name: "status code 401", errMsg: "status code 401", expectErr: true}, - {name: "status code 403", errMsg: "status code 403", expectErr: true}, - {name: "request unauthenticated", errMsg: "request unauthenticated", expectErr: true}, - {name: "request unauthorized", errMsg: "request unauthorized", expectErr: true}, - {name: "access denied", errMsg: "access denied", expectErr: true}, + {name: "authentication failed", err: errors.New("authentication failed"), expectErr: true}, + {name: "Authentication Failed (uppercase)", err: errors.New("Authentication Failed"), expectErr: true}, + {name: "authentication error", err: errors.New("authentication error: bad token"), expectErr: true}, + {name: "401 unauthorized", err: errors.New("401 unauthorized"), expectErr: true}, + {name: "403 forbidden", err: errors.New("403 forbidden"), expectErr: true}, + {name: "HTTP 401", err: errors.New("HTTP 401"), expectErr: true}, + {name: "HTTP 403", err: errors.New("HTTP 403"), expectErr: true}, + {name: "status code 401", err: errors.New("status code 401"), expectErr: true}, + {name: "status code 403", err: errors.New("status code 403"), expectErr: true}, + {name: "request unauthenticated", err: errors.New("request unauthenticated"), expectErr: true}, + {name: "request unauthorized", err: errors.New("request unauthorized"), expectErr: true}, + {name: "access denied", err: errors.New("access denied"), expectErr: true}, // Negative cases - should NOT be detected as auth errors - {name: "connection refused", errMsg: "connection refused", expectErr: false}, - {name: "timeout", errMsg: "request timeout", expectErr: false}, - {name: "generic error", errMsg: "something went wrong", expectErr: false}, - {name: "404 not found", errMsg: "404 not found", expectErr: false}, - {name: "500 internal server error", errMsg: "500 internal server error", expectErr: false}, - {name: "hostname with 401", errMsg: "http://backend401.example.com", expectErr: false}, - {name: "empty string", errMsg: "", expectErr: false}, + {name: "connection refused", err: errors.New("connection refused"), expectErr: false}, + {name: "timeout", err: errors.New("request timeout"), expectErr: false}, + {name: "generic error", err: errors.New("something went wrong"), expectErr: false}, + {name: "404 not found", err: errors.New("404 not found"), expectErr: false}, + {name: "500 internal server error", err: errors.New("500 internal server error"), expectErr: false}, + {name: "hostname with 401", err: errors.New("http://backend401.example.com"), expectErr: false}, + {name: "nil error", err: nil, expectErr: false}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() - result := isAuthError(tt.errMsg) + result := vmcp.IsAuthenticationError(tt.err) assert.Equal(t, tt.expectErr, result) }) } @@ -356,23 +356,23 @@ func TestIsTimeoutError(t *testing.T) { tests := []struct { name string - errMsg string + err error expectErr bool }{ - {name: "timeout", errMsg: "request timeout", expectErr: true}, - {name: "deadline exceeded", errMsg: "deadline exceeded", expectErr: true}, - {name: "context deadline exceeded", errMsg: "context deadline exceeded", expectErr: true}, - {name: "Timeout (uppercase)", errMsg: "Request Timeout", expectErr: true}, - {name: "connection refused", errMsg: "connection refused", expectErr: false}, - {name: "generic error", errMsg: "something went wrong", expectErr: false}, - {name: "empty string", errMsg: "", expectErr: false}, + {name: "timeout", err: errors.New("request timeout"), expectErr: true}, + {name: "deadline exceeded", err: errors.New("deadline exceeded"), expectErr: true}, + {name: "context deadline exceeded", err: errors.New("context deadline exceeded"), expectErr: true}, + {name: "Timeout (uppercase)", err: errors.New("Request Timeout"), expectErr: true}, + {name: "connection refused", err: errors.New("connection refused"), expectErr: false}, + {name: "generic error", err: errors.New("something went wrong"), expectErr: false}, + {name: "nil error", err: nil, expectErr: false}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() - result := isTimeoutError(tt.errMsg) + result := vmcp.IsTimeoutError(tt.err) assert.Equal(t, tt.expectErr, result) }) } @@ -383,25 +383,25 @@ func TestIsConnectionError(t *testing.T) { tests := []struct { name string - errMsg string + err error expectErr bool }{ - {name: "connection refused", errMsg: "connection refused", expectErr: true}, - {name: "connection reset", errMsg: "connection reset by peer", expectErr: true}, - {name: "no route to host", errMsg: "no route to host", expectErr: true}, - {name: "network unreachable", errMsg: "network is unreachable", expectErr: true}, - {name: "Connection Refused (uppercase)", errMsg: "Connection Refused", expectErr: true}, - {name: "timeout", errMsg: "request timeout", expectErr: false}, - {name: "authentication failed", errMsg: "authentication failed", expectErr: false}, - {name: "generic error", errMsg: "something went wrong", expectErr: false}, - {name: "empty string", errMsg: "", expectErr: false}, + {name: "connection refused", err: errors.New("connection refused"), expectErr: true}, + {name: "connection reset", err: errors.New("connection reset by peer"), expectErr: true}, + {name: "no route to host", err: errors.New("no route to host"), expectErr: true}, + {name: "network unreachable", err: errors.New("network is unreachable"), expectErr: true}, + {name: "Connection Refused (uppercase)", err: errors.New("Connection Refused"), expectErr: true}, + {name: "timeout", err: errors.New("request timeout"), expectErr: false}, + {name: "authentication failed", err: errors.New("authentication failed"), expectErr: false}, + {name: "generic error", err: errors.New("something went wrong"), expectErr: false}, + {name: "nil error", err: nil, expectErr: false}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() - result := isConnectionError(tt.errMsg) + result := vmcp.IsConnectionError(tt.err) assert.Equal(t, tt.expectErr, result) }) } @@ -427,7 +427,7 @@ func TestHealthChecker_CheckHealth_Timeout(t *testing.T) { }). Times(1) - checker := NewHealthChecker(mockClient, 100*time.Millisecond) + checker := NewHealthChecker(mockClient, 100*time.Millisecond, 0) target := &vmcp.BackendTarget{ WorkloadID: "backend-1", WorkloadName: "test-backend", @@ -464,7 +464,7 @@ func TestHealthChecker_CheckHealth_MultipleBackends(t *testing.T) { }). Times(4) - checker := NewHealthChecker(mockClient, 5*time.Second) + checker := NewHealthChecker(mockClient, 5*time.Second, 0) // Test healthy backend status, err := checker.CheckHealth(context.Background(), &vmcp.BackendTarget{ diff --git a/pkg/vmcp/health/monitor.go b/pkg/vmcp/health/monitor.go index a6aca6585c..3591406138 100644 --- a/pkg/vmcp/health/monitor.go +++ b/pkg/vmcp/health/monitor.go @@ -75,6 +75,12 @@ type MonitorConfig struct { // Timeout is the maximum duration for a single health check operation. // Zero means no timeout (not recommended). Timeout time.Duration + + // DegradedThreshold is the response time threshold for marking a backend as degraded. + // If a health check succeeds but takes longer than this duration, the backend is marked degraded. + // Zero means disabled (backends will never be marked degraded based on response time alone). + // Recommended: 5s. + DegradedThreshold time.Duration } // DefaultConfig returns sensible default configuration values. @@ -83,6 +89,7 @@ func DefaultConfig() MonitorConfig { CheckInterval: 30 * time.Second, UnhealthyThreshold: 3, Timeout: 10 * time.Second, + DegradedThreshold: 5 * time.Second, } } @@ -107,8 +114,8 @@ func NewMonitor( return nil, fmt.Errorf("unhealthy threshold must be >= 1, got %d", config.UnhealthyThreshold) } - // Create health checker - checker := NewHealthChecker(client, config.Timeout) + // Create health checker with degraded threshold + checker := NewHealthChecker(client, config.Timeout, config.DegradedThreshold) // Create status tracker statusTracker := newStatusTracker(config.UnhealthyThreshold) @@ -239,7 +246,9 @@ func (m *Monitor) performHealthCheck(ctx context.Context, backend *vmcp.Backend) if err != nil { m.statusTracker.RecordFailure(backend.ID, backend.Name, status, err) } else { - m.statusTracker.RecordSuccess(backend.ID, backend.Name) + // Pass status to RecordSuccess - it may be healthy or degraded (from slow response) + // RecordSuccess will further check for recovering state (had recent failures) + m.statusTracker.RecordSuccess(backend.ID, backend.Name, status) } } @@ -276,15 +285,15 @@ func (m *Monitor) IsBackendHealthy(backendID string) bool { } // GetHealthSummary returns a summary of backend health for logging/monitoring. -// Returns counts of healthy, unhealthy, and total backends. +// Returns counts of healthy, degraded, unhealthy, and total backends. func (m *Monitor) GetHealthSummary() Summary { allStates := m.statusTracker.GetAllStates() summary := Summary{ Total: len(allStates), Healthy: 0, - Unhealthy: 0, Degraded: 0, + Unhealthy: 0, Unknown: 0, Unauthenticated: 0, } @@ -293,10 +302,10 @@ func (m *Monitor) GetHealthSummary() Summary { switch state.Status { case vmcp.BackendHealthy: summary.Healthy++ - case vmcp.BackendUnhealthy: - summary.Unhealthy++ case vmcp.BackendDegraded: summary.Degraded++ + case vmcp.BackendUnhealthy: + summary.Unhealthy++ case vmcp.BackendUnknown: summary.Unknown++ case vmcp.BackendUnauthenticated: @@ -311,14 +320,14 @@ func (m *Monitor) GetHealthSummary() Summary { type Summary struct { Total int Healthy int - Unhealthy int Degraded int + Unhealthy int Unknown int Unauthenticated int } // String returns a human-readable summary. func (s Summary) String() string { - return fmt.Sprintf("total=%d healthy=%d unhealthy=%d degraded=%d unknown=%d unauthenticated=%d", - s.Total, s.Healthy, s.Unhealthy, s.Degraded, s.Unknown, s.Unauthenticated) + return fmt.Sprintf("total=%d healthy=%d degraded=%d unhealthy=%d unknown=%d unauthenticated=%d", + s.Total, s.Healthy, s.Degraded, s.Unhealthy, s.Unknown, s.Unauthenticated) } diff --git a/pkg/vmcp/health/monitor_test.go b/pkg/vmcp/health/monitor_test.go index 221ab2c7be..ac52650b48 100644 --- a/pkg/vmcp/health/monitor_test.go +++ b/pkg/vmcp/health/monitor_test.go @@ -492,6 +492,7 @@ func TestDefaultConfig(t *testing.T) { assert.Equal(t, 30*time.Second, config.CheckInterval) assert.Equal(t, 3, config.UnhealthyThreshold) assert.Equal(t, 10*time.Second, config.Timeout) + assert.Equal(t, 5*time.Second, config.DegradedThreshold) } func TestHealthCheckMarker(t *testing.T) { @@ -516,8 +517,8 @@ func TestSummary_String(t *testing.T) { summary := Summary{ Total: 10, Healthy: 5, - Unhealthy: 2, Degraded: 1, + Unhealthy: 2, Unknown: 1, Unauthenticated: 1, } @@ -525,8 +526,8 @@ func TestSummary_String(t *testing.T) { str := summary.String() assert.Contains(t, str, "total=10") assert.Contains(t, str, "healthy=5") - assert.Contains(t, str, "unhealthy=2") assert.Contains(t, str, "degraded=1") + assert.Contains(t, str, "unhealthy=2") assert.Contains(t, str, "unknown=1") assert.Contains(t, str, "unauthenticated=1") } diff --git a/pkg/vmcp/health/status.go b/pkg/vmcp/health/status.go index f4947fb141..3c2cc77a18 100644 --- a/pkg/vmcp/health/status.go +++ b/pkg/vmcp/health/status.go @@ -59,40 +59,58 @@ func newStatusTracker(unhealthyThreshold int) *statusTracker { } // RecordSuccess records a successful health check for a backend. -// This resets the consecutive failure count and marks the backend as healthy. +// This may mark the backend as healthy or degraded depending on recent failure history. +// If the backend had recent failures, it's marked as degraded (recovering state). // If the backend was previously unhealthy, this transition is logged. -func (t *statusTracker) RecordSuccess(backendID string, backendName string) { +// +// Parameters: +// - backendID: Unique identifier for the backend +// - backendName: Human-readable name for logging +// - status: The health status returned by the health check (healthy or degraded) +func (t *statusTracker) RecordSuccess(backendID string, backendName string, status vmcp.BackendHealthStatus) { t.mu.Lock() defer t.mu.Unlock() state, exists := t.states[backendID] if !exists { - // Initialize new state + // Initialize new state - no failure history, so accept status as-is state = &backendHealthState{ - status: vmcp.BackendHealthy, + status: status, consecutiveFailures: 0, lastCheckTime: time.Now(), lastError: nil, lastTransitionTime: time.Now(), } t.states[backendID] = state - logger.Debugf("Backend %s initialized as healthy", backendName) + logger.Debugf("Backend %s initialized as %s", backendName, status) return } // Check for status transition previousStatus := state.status previousFailures := state.consecutiveFailures - state.status = vmcp.BackendHealthy + + // If backend had recent failures, mark as degraded (recovering state) + // This takes precedence over the health check's status determination + if previousFailures > 0 { + state.status = vmcp.BackendDegraded + logger.Infof("Backend %s recovering from failures: %s → %s (had %d consecutive failures)", + backendName, previousStatus, vmcp.BackendDegraded, previousFailures) + } else { + // No recent failures, use the status from health check (healthy or degraded from slow response) + state.status = status + if previousStatus != status { + logger.Infof("Backend %s status changed: %s → %s", backendName, previousStatus, status) + } + } + state.consecutiveFailures = 0 state.lastCheckTime = time.Now() state.lastError = nil - // Log transition if status changed - if previousStatus != vmcp.BackendHealthy { + // Update transition time if status changed + if previousStatus != state.status { state.lastTransitionTime = time.Now() - logger.Infof("Backend %s health recovered: %s → %s (was failing for %d consecutive checks)", - backendName, previousStatus, vmcp.BackendHealthy, previousFailures) } } diff --git a/pkg/vmcp/health/status_test.go b/pkg/vmcp/health/status_test.go index 05fb7a6452..6b5193b28f 100644 --- a/pkg/vmcp/health/status_test.go +++ b/pkg/vmcp/health/status_test.go @@ -65,7 +65,7 @@ func TestStatusTracker_RecordSuccess(t *testing.T) { tracker := newStatusTracker(3) // Record success for new backend - tracker.RecordSuccess("backend-1", "Backend 1") + tracker.RecordSuccess("backend-1", "Backend 1", vmcp.BackendHealthy) status, exists := tracker.GetStatus("backend-1") assert.True(t, exists) @@ -95,11 +95,11 @@ func TestStatusTracker_RecordSuccess_AfterFailures(t *testing.T) { assert.Equal(t, vmcp.BackendUnhealthy, state.Status) assert.Equal(t, 5, state.ConsecutiveFailures) - // Record success - should reset everything - tracker.RecordSuccess("backend-1", "Backend 1") + // Record success - should mark as degraded due to recovering from failures + tracker.RecordSuccess("backend-1", "Backend 1", vmcp.BackendHealthy) state, _ = tracker.GetState("backend-1") - assert.Equal(t, vmcp.BackendHealthy, state.Status) + assert.Equal(t, vmcp.BackendDegraded, state.Status) // Degraded because recovering from failures assert.Equal(t, 0, state.ConsecutiveFailures) assert.Nil(t, state.LastError) } @@ -150,7 +150,7 @@ func TestStatusTracker_RecordFailure_StatusTransitions(t *testing.T) { tracker := newStatusTracker(2) // Start with healthy - tracker.RecordSuccess("backend-1", "Backend 1") + tracker.RecordSuccess("backend-1", "Backend 1", vmcp.BackendHealthy) status, _ := tracker.GetStatus("backend-1") assert.Equal(t, vmcp.BackendHealthy, status) @@ -189,11 +189,6 @@ func TestStatusTracker_RecordFailure_DifferentStatusTypes(t *testing.T) { failureStatus: vmcp.BackendUnauthenticated, expectedStatus: vmcp.BackendUnauthenticated, }, - { - name: "degraded failures", - failureStatus: vmcp.BackendDegraded, - expectedStatus: vmcp.BackendDegraded, - }, } for _, tt := range tests { @@ -240,14 +235,14 @@ func TestStatusTracker_GetAllStates(t *testing.T) { tracker := newStatusTracker(3) // Add multiple backends with different states - tracker.RecordSuccess("backend-1", "Backend 1") + tracker.RecordSuccess("backend-1", "Backend 1", vmcp.BackendHealthy) // Record enough failures to reach threshold for backend-2 for i := 0; i < 3; i++ { tracker.RecordFailure("backend-2", "Backend 2", vmcp.BackendUnhealthy, errors.New("failed")) } - tracker.RecordSuccess("backend-3", "Backend 3") + tracker.RecordSuccess("backend-3", "Backend 3", vmcp.BackendHealthy) allStates := tracker.GetAllStates() assert.Len(t, allStates, 3) @@ -271,7 +266,7 @@ func TestStatusTracker_GetAllStates_Immutability(t *testing.T) { t.Parallel() tracker := newStatusTracker(3) - tracker.RecordSuccess("backend-1", "Backend 1") + tracker.RecordSuccess("backend-1", "Backend 1", vmcp.BackendHealthy) // Get states states1 := tracker.GetAllStates() @@ -291,7 +286,7 @@ func TestStatusTracker_IsHealthy(t *testing.T) { tracker := newStatusTracker(3) // Healthy backend - tracker.RecordSuccess("backend-healthy", "Healthy Backend") + tracker.RecordSuccess("backend-healthy", "Healthy Backend", vmcp.BackendHealthy) assert.True(t, tracker.IsHealthy("backend-healthy")) // Unhealthy backend @@ -318,7 +313,7 @@ func TestStatusTracker_ConcurrentAccess(t *testing.T) { go func(_ int) { defer wg.Done() for j := 0; j < numOperations; j++ { - tracker.RecordSuccess("backend-success", "Backend Success") + tracker.RecordSuccess("backend-success", "Backend Success", vmcp.BackendHealthy) } }(i) } @@ -366,7 +361,7 @@ func TestStatusTracker_StateTimestamps(t *testing.T) { testErr := errors.New("test error") // Initial success - tracker.RecordSuccess("backend-1", "Backend 1") + tracker.RecordSuccess("backend-1", "Backend 1", vmcp.BackendHealthy) state1, _ := tracker.GetState("backend-1") initialTransitionTime := state1.LastTransitionTime @@ -399,7 +394,7 @@ func TestStatusTracker_MultipleBackends(t *testing.T) { tracker := newStatusTracker(2) // Backend 1: Healthy - tracker.RecordSuccess("backend-1", "Backend 1") + tracker.RecordSuccess("backend-1", "Backend 1", vmcp.BackendHealthy) // Backend 2: Unhealthy for i := 0; i < 2; i++ { @@ -442,11 +437,11 @@ func TestStatusTracker_RecoveryAfterFailures(t *testing.T) { // Wait a bit time.Sleep(10 * time.Millisecond) - // Single success should recover immediately - tracker.RecordSuccess("backend-1", "Backend 1") + // Single success should mark as degraded (recovering from failures) + tracker.RecordSuccess("backend-1", "Backend 1", vmcp.BackendHealthy) state, _ = tracker.GetState("backend-1") - assert.Equal(t, vmcp.BackendHealthy, state.Status) + assert.Equal(t, vmcp.BackendDegraded, state.Status) // Degraded because recovering from failures assert.Equal(t, 0, state.ConsecutiveFailures) assert.Nil(t, state.LastError) assert.True(t, state.LastTransitionTime.After(beforeRecoveryTransitionTime)) diff --git a/pkg/vmcp/registry_test.go b/pkg/vmcp/registry_test.go index a8d2509084..3eaecc9dad 100644 --- a/pkg/vmcp/registry_test.go +++ b/pkg/vmcp/registry_test.go @@ -54,8 +54,9 @@ func TestNewImmutableRegistry(t *testing.T) { {ID: "degraded", HealthStatus: BackendDegraded}, {ID: "unhealthy", HealthStatus: BackendUnhealthy}, {ID: "unknown", HealthStatus: BackendUnknown}, + {ID: "unauthenticated", HealthStatus: BackendUnauthenticated}, }, - expectedCount: 4, + expectedCount: 5, }, { name: "all transport types", @@ -553,6 +554,7 @@ func TestBackendToTarget(t *testing.T) { BackendDegraded, BackendUnhealthy, BackendUnknown, + BackendUnauthenticated, } for _, status := range statuses { @@ -643,6 +645,7 @@ func TestDomainTypes_BackendHealthStatus(t *testing.T) { {BackendDegraded, "degraded"}, {BackendUnhealthy, "unhealthy"}, {BackendUnknown, "unknown"}, + {BackendUnauthenticated, "unauthenticated"}, } for _, tt := range tests { diff --git a/pkg/vmcp/types.go b/pkg/vmcp/types.go index ee6cdd07b4..0ca8eb5cf5 100644 --- a/pkg/vmcp/types.go +++ b/pkg/vmcp/types.go @@ -99,6 +99,9 @@ const ( BackendHealthy BackendHealthStatus = "healthy" // BackendDegraded indicates the backend is operational but experiencing issues. + // This occurs when: + // - Health checks succeed but response times exceed the degraded threshold (slow but working) + // - Backend just recovered from failures and is in a stabilizing state BackendDegraded BackendHealthStatus = "degraded" // BackendUnhealthy indicates the backend is not responding to health checks. From 8de8abb24b8f94aa35aacdec2446ed9608a5698b Mon Sep 17 00:00:00 2001 From: taskbot Date: Fri, 19 Dec 2025 15:27:44 +0100 Subject: [PATCH 3/3] fixes from review --- pkg/vmcp/client/client.go | 9 +++++ pkg/vmcp/health/monitor.go | 22 +---------- pkg/vmcp/health/monitor_test.go | 69 ++++++++++++++++----------------- 3 files changed, 44 insertions(+), 56 deletions(-) diff --git a/pkg/vmcp/client/client.go b/pkg/vmcp/client/client.go index c26ee1e48b..1b78f49b6d 100644 --- a/pkg/vmcp/client/client.go +++ b/pkg/vmcp/client/client.go @@ -247,6 +247,15 @@ func (h *httpBackendClient) defaultClientFactory(ctx context.Context, target *vm // Error detection strategy (in order of preference): // 1. Check for standard Go error types (context errors, net.Error, url.Error) // 2. Fall back to string pattern matching for library-specific errors (MCP SDK, HTTP libs) +// +// Error chain preservation: +// The returned error wraps the sentinel error (ErrTimeout, ErrBackendUnavailable, etc.) with %w +// and formats the original error with %v. This means: +// - errors.Is() works for checking the sentinel error (e.g., errors.Is(err, vmcp.ErrTimeout)) +// - errors.As() cannot access the underlying original error type +// This is a deliberate trade-off due to Go's limitation of one %w per fmt.Errorf call. +// If access to the underlying error type is needed in the future, consider implementing +// a custom error type with multiple Unwrap() methods (Go 1.20+). func wrapBackendError(err error, backendID string, operation string) error { if err == nil { return nil diff --git a/pkg/vmcp/health/monitor.go b/pkg/vmcp/health/monitor.go index 3591406138..b6708104f1 100644 --- a/pkg/vmcp/health/monitor.go +++ b/pkg/vmcp/health/monitor.go @@ -10,22 +10,6 @@ import ( "github.com/stacklok/toolhive/pkg/vmcp" ) -// healthCheckContextKey is a marker for health check requests. -// When present in context, authentication should be bypassed. -type healthCheckContextKey struct{} - -// WithHealthCheckMarker marks a context as a health check request. -// Authentication layers should skip authentication for these requests. -func WithHealthCheckMarker(ctx context.Context) context.Context { - return context.WithValue(ctx, healthCheckContextKey{}, true) -} - -// IsHealthCheck returns true if the context is marked as a health check. -func IsHealthCheck(ctx context.Context) bool { - val, ok := ctx.Value(healthCheckContextKey{}).(bool) - return ok && val -} - // Monitor performs periodic health checks on backend MCP servers. // It runs background goroutines for each backend, tracking their health status // and consecutive failure counts. The monitor supports graceful shutdown and @@ -235,12 +219,8 @@ func (m *Monitor) performHealthCheck(ctx context.Context, backend *vmcp.Backend) Metadata: backend.Metadata, } - // Mark context as health check to bypass authentication - // Health checks verify backend availability and should not require user credentials - healthCheckCtx := WithHealthCheckMarker(ctx) - // Perform health check - status, err := m.checker.CheckHealth(healthCheckCtx, target) + status, err := m.checker.CheckHealth(ctx, target) // Record result in status tracker if err != nil { diff --git a/pkg/vmcp/health/monitor_test.go b/pkg/vmcp/health/monitor_test.go index ac52650b48..efb3e32e50 100644 --- a/pkg/vmcp/health/monitor_test.go +++ b/pkg/vmcp/health/monitor_test.go @@ -106,11 +106,10 @@ func TestMonitor_StartStop(t *testing.T) { err = monitor.Start(ctx) require.NoError(t, err) - // Wait for at least one health check - time.Sleep(150 * time.Millisecond) - - // Verify backend is healthy - assert.True(t, monitor.IsBackendHealthy("backend-1")) + // Wait for at least one health check to complete + require.Eventually(t, func() bool { + return monitor.IsBackendHealthy("backend-1") + }, 500*time.Millisecond, 10*time.Millisecond, "backend should become healthy") // Stop monitor err = monitor.Stop() @@ -247,13 +246,11 @@ func TestMonitor_PeriodicHealthChecks(t *testing.T) { _ = monitor.Stop() }() - // Wait for threshold to be exceeded (2 failures * 50ms + buffer) - time.Sleep(200 * time.Millisecond) - - // Backend should be marked unhealthy - status, err := monitor.GetBackendStatus("backend-1") - assert.NoError(t, err) - assert.Equal(t, vmcp.BackendUnhealthy, status) + // Wait for threshold to be exceeded (2 failures) + require.Eventually(t, func() bool { + status, err := monitor.GetBackendStatus("backend-1") + return err == nil && status == vmcp.BackendUnhealthy + }, 500*time.Millisecond, 10*time.Millisecond, "backend should become unhealthy after threshold") state, err := monitor.GetBackendState("backend-1") assert.NoError(t, err) @@ -300,7 +297,10 @@ func TestMonitor_GetHealthSummary(t *testing.T) { }() // Wait for health checks to complete - time.Sleep(100 * time.Millisecond) + require.Eventually(t, func() bool { + summary := monitor.GetHealthSummary() + return summary.Healthy == 1 && summary.Unhealthy == 1 + }, 500*time.Millisecond, 10*time.Millisecond, "summary should show 1 healthy and 1 unhealthy") summary := monitor.GetHealthSummary() assert.Equal(t, 2, summary.Total) @@ -340,7 +340,11 @@ func TestMonitor_GetBackendStatus(t *testing.T) { _ = monitor.Stop() }() - time.Sleep(150 * time.Millisecond) + // Wait for initial health check to complete + require.Eventually(t, func() bool { + status, err := monitor.GetBackendStatus("backend-1") + return err == nil && status == vmcp.BackendHealthy + }, 500*time.Millisecond, 10*time.Millisecond, "backend status should be available and healthy") // Test getting status for existing backend status, err := monitor.GetBackendStatus("backend-1") @@ -385,7 +389,11 @@ func TestMonitor_GetBackendState(t *testing.T) { _ = monitor.Stop() }() - time.Sleep(150 * time.Millisecond) + // Wait for initial health check to complete + require.Eventually(t, func() bool { + state, err := monitor.GetBackendState("backend-1") + return err == nil && state != nil && state.Status == vmcp.BackendHealthy + }, 500*time.Millisecond, 10*time.Millisecond, "backend state should be available and healthy") // Test getting state for existing backend state, err := monitor.GetBackendState("backend-1") @@ -432,7 +440,11 @@ func TestMonitor_GetAllBackendStates(t *testing.T) { _ = monitor.Stop() }() - time.Sleep(150 * time.Millisecond) + // Wait for initial health checks to complete for both backends + require.Eventually(t, func() bool { + allStates := monitor.GetAllBackendStates() + return len(allStates) == 2 + }, 500*time.Millisecond, 10*time.Millisecond, "all backend states should be available") allStates := monitor.GetAllBackendStates() assert.Len(t, allStates, 2) @@ -470,13 +482,16 @@ func TestMonitor_ContextCancellation(t *testing.T) { err = monitor.Start(ctx) require.NoError(t, err) - // Wait for a few health checks - time.Sleep(100 * time.Millisecond) + // Wait for a few health checks to run + require.Eventually(t, func() bool { + return monitor.IsBackendHealthy("backend-1") + }, 500*time.Millisecond, 10*time.Millisecond, "backend should have completed at least one health check") // Cancel context cancel() - // Wait for goroutines to stop + // Give goroutines time to observe cancellation + // Note: We can't easily poll for goroutine completion, so a short sleep is acceptable here time.Sleep(100 * time.Millisecond) // Monitor should still be running (context cancellation stops checks but doesn't stop the monitor) @@ -495,22 +510,6 @@ func TestDefaultConfig(t *testing.T) { assert.Equal(t, 5*time.Second, config.DegradedThreshold) } -func TestHealthCheckMarker(t *testing.T) { - t.Parallel() - - // Test WithHealthCheckMarker - ctx := context.Background() - assert.False(t, IsHealthCheck(ctx)) - - markedCtx := WithHealthCheckMarker(ctx) - assert.True(t, IsHealthCheck(markedCtx)) - - // Test that marker is preserved across context operations - cancelCtx, cancel := context.WithCancel(markedCtx) - defer cancel() - assert.True(t, IsHealthCheck(cancelCtx)) -} - func TestSummary_String(t *testing.T) { t.Parallel()