diff --git a/deploy/charts/operator-crds/files/crds/toolhive.stacklok.dev_virtualmcpservers.yaml b/deploy/charts/operator-crds/files/crds/toolhive.stacklok.dev_virtualmcpservers.yaml index f4520c6555..be3e76d7cc 100644 --- a/deploy/charts/operator-crds/files/crds/toolhive.stacklok.dev_virtualmcpservers.yaml +++ b/deploy/charts/operator-crds/files/crds/toolhive.stacklok.dev_virtualmcpservers.yaml @@ -2417,6 +2417,8 @@ spec: description: |- HealthCheckTimeout is the maximum duration for a single health check operation. Should be less than HealthCheckInterval to prevent checks from queuing up. + The same value also bounds a new client session's initialize handshake + across backends, so set it below your gateway/client timeout. pattern: ^([0-9]+(\.[0-9]+)?(ns|us|µs|ms|s|m|h))+$ type: string partialFailureMode: @@ -6550,6 +6552,8 @@ spec: description: |- HealthCheckTimeout is the maximum duration for a single health check operation. Should be less than HealthCheckInterval to prevent checks from queuing up. + The same value also bounds a new client session's initialize handshake + across backends, so set it below your gateway/client timeout. pattern: ^([0-9]+(\.[0-9]+)?(ns|us|µs|ms|s|m|h))+$ type: string partialFailureMode: diff --git a/deploy/charts/operator-crds/templates/toolhive.stacklok.dev_virtualmcpservers.yaml b/deploy/charts/operator-crds/templates/toolhive.stacklok.dev_virtualmcpservers.yaml index 0e2fe27e91..dc8626398b 100644 --- a/deploy/charts/operator-crds/templates/toolhive.stacklok.dev_virtualmcpservers.yaml +++ b/deploy/charts/operator-crds/templates/toolhive.stacklok.dev_virtualmcpservers.yaml @@ -2420,6 +2420,8 @@ spec: description: |- HealthCheckTimeout is the maximum duration for a single health check operation. Should be less than HealthCheckInterval to prevent checks from queuing up. + The same value also bounds a new client session's initialize handshake + across backends, so set it below your gateway/client timeout. pattern: ^([0-9]+(\.[0-9]+)?(ns|us|µs|ms|s|m|h))+$ type: string partialFailureMode: @@ -6553,6 +6555,8 @@ spec: description: |- HealthCheckTimeout is the maximum duration for a single health check operation. Should be less than HealthCheckInterval to prevent checks from queuing up. + The same value also bounds a new client session's initialize handshake + across backends, so set it below your gateway/client timeout. pattern: ^([0-9]+(\.[0-9]+)?(ns|us|µs|ms|s|m|h))+$ type: string partialFailureMode: diff --git a/docs/operator/crd-api.md b/docs/operator/crd-api.md index 17be058246..d909a6962f 100644 --- a/docs/operator/crd-api.md +++ b/docs/operator/crd-api.md @@ -518,7 +518,7 @@ _Appears in:_ | --- | --- | --- | --- | | `healthCheckInterval` _[vmcp.config.Duration](#vmcpconfigduration)_ | HealthCheckInterval is the interval between health checks. | 30s | Pattern: `^([0-9]+(\.[0-9]+)?(ns\|us\|µs\|ms\|s\|m\|h))+$`
Type: string
Optional: \{\}
| | `unhealthyThreshold` _integer_ | UnhealthyThreshold is the number of consecutive failures before marking unhealthy. | 3 | Optional: \{\}
| -| `healthCheckTimeout` _[vmcp.config.Duration](#vmcpconfigduration)_ | HealthCheckTimeout is the maximum duration for a single health check operation.
Should be less than HealthCheckInterval to prevent checks from queuing up. | 10s | Pattern: `^([0-9]+(\.[0-9]+)?(ns\|us\|µs\|ms\|s\|m\|h))+$`
Type: string
Optional: \{\}
| +| `healthCheckTimeout` _[vmcp.config.Duration](#vmcpconfigduration)_ | HealthCheckTimeout is the maximum duration for a single health check operation.
Should be less than HealthCheckInterval to prevent checks from queuing up.
The same value also bounds a new client session's initialize handshake
across backends, so set it below your gateway/client timeout. | 10s | Pattern: `^([0-9]+(\.[0-9]+)?(ns\|us\|µs\|ms\|s\|m\|h))+$`
Type: string
Optional: \{\}
| | `statusReportingInterval` _[vmcp.config.Duration](#vmcpconfigduration)_ | StatusReportingInterval is the interval for reporting status updates to Kubernetes.
This controls how often the vMCP runtime reports backend health and phase changes.
Lower values provide faster status updates but increase API server load. | 30s | Pattern: `^([0-9]+(\.[0-9]+)?(ns\|us\|µs\|ms\|s\|m\|h))+$`
Type: string
Optional: \{\}
| | `partialFailureMode` _string_ | PartialFailureMode defines behavior when some backends are unavailable.
- fail: Fail entire request if any backend is unavailable
- best_effort: Continue with available backends | fail | Enum: [fail best_effort]
Optional: \{\}
| | `circuitBreaker` _[vmcp.config.CircuitBreakerConfig](#vmcpconfigcircuitbreakerconfig)_ | CircuitBreaker configures circuit breaker behavior. | | Optional: \{\}
| diff --git a/docs/operator/virtualmcpserver-api.md b/docs/operator/virtualmcpserver-api.md index 61f13cdd49..e090fda673 100644 --- a/docs/operator/virtualmcpserver-api.md +++ b/docs/operator/virtualmcpserver-api.md @@ -685,6 +685,18 @@ status: compositeToolCount: 1 ``` +## Health, Ready, and initialize + +These three signals mean different things: + +| Signal | What it means | What it does not mean | +| --- | --- | --- | +| `GET /health` | Process is up (liveness). Always 200 if the HTTP server answers. | A new MCP session can `initialize` before your gateway times out. | +| CR `Ready` / backend `Healthy` | Last `ListCapabilities` probe succeeded within `healthCheckTimeout`. | The next client `initialize` will finish in that same budget. | +| Client `initialize` | A new session connected to backends (best-effort). Bounded by `healthCheckTimeout` (default 10s). | Every backend completed the full handshake. Slow ones are skipped when the budget expires. | + +Set `spec.config.operational.failureHandling.healthCheckTimeout` below the client or gateway timeout so initialize cannot hang with a 0-byte response. + ## Validation The VirtualMCPServer CRD includes comprehensive validation: diff --git a/pkg/vmcp/cli/serve.go b/pkg/vmcp/cli/serve.go index f89d216498..d17848b752 100644 --- a/pkg/vmcp/cli/serve.go +++ b/pkg/vmcp/cli/serve.go @@ -358,6 +358,15 @@ func Serve(ctx context.Context, cfg ServeConfig) error { sessionFactoryOpts, vmcpsession.WithRequestTimeoutResolver(backendRequestTimeoutResolver(vmcpCfg)), ) + // Bound client-facing initialize to the same budget as a health probe + // so Ready + /health cannot stay green while initialize hangs past + // typical gateway timeouts (#6345). Unset keeps the factory default (10s). + if vmcpCfg.Operational != nil && + vmcpCfg.Operational.FailureHandling != nil && + vmcpCfg.Operational.FailureHandling.HealthCheckTimeout > 0 { + sessionFactoryOpts = append(sessionFactoryOpts, vmcpsession.WithSessionInitTimeout( + time.Duration(vmcpCfg.Operational.FailureHandling.HealthCheckTimeout))) + } sessionFactory := vmcpsession.NewSessionFactory(outgoingRegistry, sessionFactoryOpts...) // When the optimizer is enabled, its meta-tools are pass-through tools. diff --git a/pkg/vmcp/config/config.go b/pkg/vmcp/config/config.go index 6712a2c87f..2596f02163 100644 --- a/pkg/vmcp/config/config.go +++ b/pkg/vmcp/config/config.go @@ -684,6 +684,8 @@ type FailureHandlingConfig struct { // HealthCheckTimeout is the maximum duration for a single health check operation. // Should be less than HealthCheckInterval to prevent checks from queuing up. + // The same value also bounds a new client session's initialize handshake + // across backends, so set it below your gateway/client timeout. // +kubebuilder:default="10s" // +optional HealthCheckTimeout Duration `json:"healthCheckTimeout,omitempty" yaml:"healthCheckTimeout,omitempty"` diff --git a/pkg/vmcp/server/server.go b/pkg/vmcp/server/server.go index 9fef79f633..fd9acf4ef7 100644 --- a/pkg/vmcp/server/server.go +++ b/pkg/vmcp/server/server.go @@ -962,6 +962,11 @@ func (s *Server) Address() string { // handleHealth handles /health and /ping HTTP requests. // Returns 200 OK if the server is running and able to respond. // +// This is a liveness signal only. It does not mean a new client session can +// initialize within the configured timeout. Kubernetes Ready on the +// VirtualMCPServer CR reflects the last backend ListCapabilities probe. +// Session open is bounded separately by HealthCheckTimeout (#6345). +// // Security Note: This endpoint is unauthenticated and intentionally minimal. // It only confirms the HTTP server is responding. No version information, // session counts, or operational metrics are exposed to prevent information diff --git a/pkg/vmcp/session/default_session_test.go b/pkg/vmcp/session/default_session_test.go index b7fd532eb1..b11cdf3707 100644 --- a/pkg/vmcp/session/default_session_test.go +++ b/pkg/vmcp/session/default_session_test.go @@ -1140,6 +1140,108 @@ func TestWithBackendInitTimeout_IgnoresNonPositive(t *testing.T) { assert.Equal(t, defaultBackendInitTimeout, f.backendInitTimeout) } +func TestWithSessionInitTimeout_IgnoresNonPositive(t *testing.T) { + t.Parallel() + + f := &defaultMultiSessionFactory{sessionInitTimeout: defaultSessionInitTimeout} + WithSessionInitTimeout(0)(f) + assert.Equal(t, defaultSessionInitTimeout, f.sessionInitTimeout) + + WithSessionInitTimeout(-time.Second)(f) + assert.Equal(t, defaultSessionInitTimeout, f.sessionInitTimeout) +} + +func TestNewSessionFactory_SessionInitTimeoutBoundsWait(t *testing.T) { + t.Parallel() + + backend := &vmcp.Backend{ID: "slow", Name: "slow", BaseURL: "http://x:9", TransportType: "streamable-http"} + + released := make(chan struct{}) + connector := func(ctx context.Context, _ *vmcp.BackendTarget, _ *auth.Identity, _ string, _ internalbk.ListChangedSink) (internalbk.Session, *vmcp.CapabilityList, error) { + select { + case <-ctx.Done(): + return nil, nil, ctx.Err() + case <-released: + return &mockConnectedBackend{}, &vmcp.CapabilityList{}, nil + } + } + + // Per-backend timeout is long; the overall session budget must win so + // initialize cannot hang past typical gateway timeouts (#6345). + factory := newSessionFactoryWithConnector(connector, + WithSessionInitTimeout(200*time.Millisecond), + WithBackendInitTimeout(30*time.Second), + ) + + start := time.Now() + sess, err := factory.MakeSessionWithID(context.Background(), uuid.New().String(), nil, []*vmcp.Backend{backend}, nil) + elapsed := time.Since(start) + + require.NoError(t, err, "session budget expiry is a partial failure, not a hard error") + require.NotNil(t, sess) + assert.Less(t, elapsed, 2*time.Second, "initialize must not wait for the 30s per-backend timeout") + assert.GreaterOrEqual(t, elapsed, 200*time.Millisecond) + assert.Empty(t, sess.Tools()) + close(released) + require.NoError(t, sess.Close()) +} + +func TestNewSessionFactory_SessionInitTimeoutSkipsQueuedSemaphoreAcquire(t *testing.T) { + t.Parallel() + + // One in-flight backend holds the only semaphore slot until the session + // budget expires, then delays release so queued backends would hang on + // `sem <-` if acquire ignored ctx. Those queued connectors ignore ctx + // and sleep long enough that MakeSession would miss the budget. + const queuedSleep = 3 * time.Second + started := make(chan struct{}) + var holderStarted atomic.Bool + var queuedEntered atomic.Int64 + connector := func(ctx context.Context, _ *vmcp.BackendTarget, _ *auth.Identity, _ string, _ internalbk.ListChangedSink) (internalbk.Session, *vmcp.CapabilityList, error) { + if holderStarted.CompareAndSwap(false, true) { + close(started) + select { + case <-ctx.Done(): + // Keep the slot occupied after expiry so queued + // acquire must lose to ctx.Done() in the select. + time.Sleep(150 * time.Millisecond) + return nil, nil, ctx.Err() + case <-time.After(30 * time.Second): + return nil, nil, errors.New("holder did not observe session budget") + } + } + queuedEntered.Add(1) + time.Sleep(queuedSleep) + return nil, nil, errors.New("queued backend ran after session budget") + } + + backends := []*vmcp.Backend{ + {ID: "holder", Name: "holder", BaseURL: "http://x:1", TransportType: "streamable-http"}, + {ID: "queued-a", Name: "queued-a", BaseURL: "http://x:2", TransportType: "streamable-http"}, + {ID: "queued-b", Name: "queued-b", BaseURL: "http://x:3", TransportType: "streamable-http"}, + } + factory := newSessionFactoryWithConnector(connector, + WithSessionInitTimeout(80*time.Millisecond), + WithBackendInitTimeout(30*time.Second), + WithMaxBackendInitConcurrency(1), + ) + + start := time.Now() + sess, err := factory.MakeSessionWithID(context.Background(), uuid.New().String(), nil, backends, nil) + elapsed := time.Since(start) + + require.NoError(t, err) + require.NotNil(t, sess) + select { + case <-started: + default: + t.Fatal("expected the in-flight backend to start before MakeSession returned") + } + assert.Zero(t, queuedEntered.Load(), "queued backends must not acquire the semaphore after the session budget expires") + assert.Less(t, elapsed, queuedSleep, "MakeSession must return without waiting for queued connectors") + require.NoError(t, sess.Close()) +} + func TestValidateSessionID(t *testing.T) { t.Parallel() diff --git a/pkg/vmcp/session/factory.go b/pkg/vmcp/session/factory.go index cf6604336f..50c036432c 100644 --- a/pkg/vmcp/session/factory.go +++ b/pkg/vmcp/session/factory.go @@ -7,6 +7,7 @@ package session import ( "context" + "errors" "fmt" "log/slog" "slices" @@ -29,6 +30,12 @@ import ( const ( defaultMaxBackendInitConcurrency = 10 defaultBackendInitTimeout = 30 * time.Second + // defaultSessionInitTimeout is the overall budget for MakeSession to + // finish contacting backends. Per-backend init can take up to + // defaultBackendInitTimeout, which used to let initialize hang past + // typical client/gateway timeouts while Ready and /health stayed OK + // (#6345). 10s matches the HealthCheckTimeout CRD default. + defaultSessionInitTimeout = 10 * time.Second // MetadataKeyBackendIDs is the transport-session metadata key that holds // a comma-separated, sorted list of successfully-connected backend IDs. @@ -134,6 +141,7 @@ type defaultMultiSessionFactory struct { connector backendConnector maxConcurrency int backendInitTimeout time.Duration + sessionInitTimeout time.Duration revisionLookup func(workloadID string) (mcpparser.Revision, bool) requestTimeoutResolver func(workloadID string) time.Duration } @@ -161,6 +169,19 @@ func WithBackendInitTimeout(d time.Duration) MultiSessionFactoryOption { } } +// WithSessionInitTimeout sets the overall budget for MakeSession to finish +// contacting backends. When the budget expires, backends that have not +// completed are skipped and the session is returned with whatever connected +// (best-effort). Defaults to 10s. Zero keeps the default; a negative value +// is ignored. +func WithSessionInitTimeout(d time.Duration) MultiSessionFactoryOption { + return func(f *defaultMultiSessionFactory) { + if d > 0 { + f.sessionInitTimeout = d + } + } +} + // WithRequestTimeoutResolver configures the timeout used for individual // backend operations. The resolver receives a backend workload ID and may // return a workload-specific duration. A nil resolver, or a non-positive @@ -223,6 +244,7 @@ func newSessionFactoryWithConnector(connector backendConnector, opts ...MultiSes connector: connector, maxConcurrency: defaultMaxBackendInitConcurrency, backendInitTimeout: defaultBackendInitTimeout, + sessionInitTimeout: defaultSessionInitTimeout, } for _, opt := range opts { opt(f) @@ -475,6 +497,12 @@ func (f *defaultMultiSessionFactory) makeBaseSession( } backends = filtered + if f.sessionInitTimeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, f.sessionInitTimeout) + defer cancel() + } + rawResults := make([]*initResult, len(backends)) modernSkipped := make([]bool, len(backends)) sem := make(chan struct{}, f.maxConcurrency) @@ -483,13 +511,32 @@ func (f *defaultMultiSessionFactory) makeBaseSession( for i, b := range backends { go func(i int, b *vmcp.Backend) { defer wg.Done() - sem <- struct{}{} - defer func() { <-sem }() + // Acquire is context-aware so queued backends do not block on + // the semaphore after sessionInitTimeout (or cancel) has fired. + select { + case sem <- struct{}{}: + defer func() { <-sem }() + case <-ctx.Done(): + return + } rawResults[i], modernSkipped[i] = f.initOneBackend(ctx, b, identity, sessionHints[b.ID], sink) }(i, b) } wg.Wait() + if err := ctx.Err(); err != nil { + if errors.Is(err, context.DeadlineExceeded) { + slog.Warn("session initialize budget expired; returning with backends that connected in time", + "error", err, + "backendCount", len(backends), + "timeout", f.sessionInitTimeout) + } else { + slog.Debug("session initialize cancelled; returning with backends that connected in time", + "error", err, + "backendCount", len(backends)) + } + } + connections := make(map[string]backend.Session, len(backends)) backendSessions := make(map[string]string, len(backends)) results := make([]initResult, 0, len(backends)) diff --git a/pkg/vmcp/session/factory_revision_test.go b/pkg/vmcp/session/factory_revision_test.go index 716bcb7b24..0943616569 100644 --- a/pkg/vmcp/session/factory_revision_test.go +++ b/pkg/vmcp/session/factory_revision_test.go @@ -9,6 +9,7 @@ import ( "errors" "log/slog" "testing" + "time" "github.com/google/uuid" "github.com/stretchr/testify/assert" @@ -212,3 +213,72 @@ func TestMakeSession_AllBackendsFailedWarning(t *testing.T) { }) } } + +const ( + sessionBudgetExpiredMsg = "session initialize budget expired" + sessionCancelledMsg = "session initialize cancelled" +) + +//nolint:paralleltest // setupLogRecorder swaps the global slog default logger. +func TestMakeSession_SessionInitContextLogs(t *testing.T) { + t.Run("deadline exceeded logs a warning", func(t *testing.T) { + buf := setupLogRecorder(t) + released := make(chan struct{}) + t.Cleanup(func() { close(released) }) + connector := func(ctx context.Context, _ *vmcp.BackendTarget, _ *auth.Identity, _ string, _ internalbk.ListChangedSink) (internalbk.Session, *vmcp.CapabilityList, error) { + select { + case <-ctx.Done(): + return nil, nil, ctx.Err() + case <-released: + return &mockConnectedBackend{}, &vmcp.CapabilityList{}, nil + } + } + factory := newSessionFactoryWithConnector(connector, WithSessionInitTimeout(50*time.Millisecond)) + backend := &vmcp.Backend{ID: "slow", Name: "slow", BaseURL: "http://x:9", TransportType: "streamable-http"} + + sess, err := factory.MakeSessionWithID(context.Background(), uuid.New().String(), nil, []*vmcp.Backend{backend}, nil) + require.NoError(t, err) + t.Cleanup(func() { _ = sess.Close() }) + + logs := buf.String() + assert.Contains(t, logs, sessionBudgetExpiredMsg) + assert.NotContains(t, logs, sessionCancelledMsg) + }) + + t.Run("caller cancel logs at debug", func(t *testing.T) { + buf := setupLogRecorder(t) + entered := make(chan struct{}) + connector := func(ctx context.Context, _ *vmcp.BackendTarget, _ *auth.Identity, _ string, _ internalbk.ListChangedSink) (internalbk.Session, *vmcp.CapabilityList, error) { + close(entered) + <-ctx.Done() + return nil, nil, ctx.Err() + } + factory := newSessionFactoryWithConnector(connector, WithSessionInitTimeout(30*time.Second)) + backend := &vmcp.Backend{ID: "slow", Name: "slow", BaseURL: "http://x:9", TransportType: "streamable-http"} + + ctx, cancel := context.WithCancel(context.Background()) + errCh := make(chan error, 1) + var sess MultiSession + go func() { + var err error + sess, err = factory.MakeSessionWithID(ctx, uuid.New().String(), nil, []*vmcp.Backend{backend}, nil) + errCh <- err + }() + select { + case <-entered: + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for connector to start") + } + cancel() + require.NoError(t, <-errCh) + t.Cleanup(func() { + if sess != nil { + _ = sess.Close() + } + }) + + logs := buf.String() + assert.Contains(t, logs, sessionCancelledMsg) + assert.NotContains(t, logs, sessionBudgetExpiredMsg) + }) +}