diff --git a/cmd/thv-operator/controllers/mcpserver_runconfig.go b/cmd/thv-operator/controllers/mcpserver_runconfig.go index 25c34ef55a..f44103eb48 100644 --- a/cmd/thv-operator/controllers/mcpserver_runconfig.go +++ b/cmd/thv-operator/controllers/mcpserver_runconfig.go @@ -144,6 +144,7 @@ func (r *MCPServerReconciler) createRunConfigFromMCPServer(m *mcpv1beta1.MCPServ options := []runner.RunConfigBuilderOption{ runner.WithName(m.Name), runner.WithImage(m.Spec.Image), + runner.WithMCPServerGeneration(m.Generation), runner.WithCmdArgs(m.Spec.Args), runner.WithTransportAndPorts(m.Spec.Transport, int(m.GetProxyPort()), int(m.GetMCPPort())), runner.WithProxyMode(transporttypes.ProxyMode(effectiveProxyMode)), diff --git a/cmd/thv-operator/controllers/mcpserver_runconfig_test.go b/cmd/thv-operator/controllers/mcpserver_runconfig_test.go index 0ffcd2a790..9aa7940d7c 100644 --- a/cmd/thv-operator/controllers/mcpserver_runconfig_test.go +++ b/cmd/thv-operator/controllers/mcpserver_runconfig_test.go @@ -1782,3 +1782,34 @@ func TestCreateRunConfigFromMCPServer_RateLimiting(t *testing.T) { }) } } + +func TestCreateRunConfigFromMCPServer_SetsMCPServerGeneration(t *testing.T) { + t.Parallel() + + m := &mcpv1beta1.MCPServer{ + ObjectMeta: metav1.ObjectMeta{ + Name: "generation-server", + Namespace: "default", + Generation: 7, + }, + Spec: mcpv1beta1.MCPServerSpec{ + Image: "ghcr.io/example/mcp:v1", + Transport: stdioTransport, + ProxyPort: 8080, + }, + } + + r := newTestMCPServerReconciler( + fake.NewClientBuilder().WithScheme(createRunConfigTestScheme()).WithObjects(m).Build(), + createRunConfigTestScheme(), + kubernetes.PlatformKubernetes, + ) + + rc, err := r.createRunConfigFromMCPServer(m) + + require.NoError(t, err) + require.NotNil(t, rc) + + assert.Equal(t, int64(7), rc.MCPServerGeneration, + "MCPServerGeneration should match MCPServer .metadata.generation") +} diff --git a/docs/server/docs.go b/docs/server/docs.go index 44a00da112..e3ff6c8575 100644 --- a/docs/server/docs.go +++ b/docs/server/docs.go @@ -1227,6 +1227,10 @@ const docTemplate = `{ "description": "K8sPodTemplatePatch is a JSON string to patch the Kubernetes pod template\nOnly applicable when using Kubernetes runtime", "type": "string" }, + "mcpserver_generation": { + "description": "MCPServerGeneration is the K8s .metadata.generation of the MCPServer CR that rendered\nthis RunConfig. The Kubernetes runtime uses it as a monotonic version to prevent stale\nrolling-update pods from overwriting a newer RunConfig's StatefulSet apply. Zero value\nmeans unversioned (backward-compat with older operators, or non-operator callers).", + "type": "integer" + }, "middleware_configs": { "description": "MiddlewareConfigs contains the list of middleware to apply to the transport\nand the configuration for each middleware.", "items": { diff --git a/docs/server/swagger.json b/docs/server/swagger.json index 6ac2e1bc50..423cde2d63 100644 --- a/docs/server/swagger.json +++ b/docs/server/swagger.json @@ -1220,6 +1220,10 @@ "description": "K8sPodTemplatePatch is a JSON string to patch the Kubernetes pod template\nOnly applicable when using Kubernetes runtime", "type": "string" }, + "mcpserver_generation": { + "description": "MCPServerGeneration is the K8s .metadata.generation of the MCPServer CR that rendered\nthis RunConfig. The Kubernetes runtime uses it as a monotonic version to prevent stale\nrolling-update pods from overwriting a newer RunConfig's StatefulSet apply. Zero value\nmeans unversioned (backward-compat with older operators, or non-operator callers).", + "type": "integer" + }, "middleware_configs": { "description": "MiddlewareConfigs contains the list of middleware to apply to the transport\nand the configuration for each middleware.", "items": { diff --git a/docs/server/swagger.yaml b/docs/server/swagger.yaml index 1a09ae301b..d601f9420e 100644 --- a/docs/server/swagger.yaml +++ b/docs/server/swagger.yaml @@ -1176,6 +1176,13 @@ components: K8sPodTemplatePatch is a JSON string to patch the Kubernetes pod template Only applicable when using Kubernetes runtime type: string + mcpserver_generation: + description: |- + MCPServerGeneration is the K8s .metadata.generation of the MCPServer CR that rendered + this RunConfig. The Kubernetes runtime uses it as a monotonic version to prevent stale + rolling-update pods from overwriting a newer RunConfig's StatefulSet apply. Zero value + means unversioned (backward-compat with older operators, or non-operator callers). + type: integer middleware_configs: description: |- MiddlewareConfigs contains the list of middleware to apply to the transport diff --git a/pkg/container/kubernetes/client.go b/pkg/container/kubernetes/client.go index fca0355484..99c4e742fd 100644 --- a/pkg/container/kubernetes/client.go +++ b/pkg/container/kubernetes/client.go @@ -49,6 +49,15 @@ const ( defaultNamespace = "default" // serviceFieldManager is the field manager name for server-side apply operations serviceFieldManager = "toolhive-container-manager" + + // RunConfigMCPServerGenerationAnnotation carries the MCPServer .metadata.generation that + // produced the RunConfig applied to this StatefulSet. Used as a monotonic version stamp + // to prevent stale proxyrunner pods (from an old Deployment ReplicaSet) from clobbering + // a newer RunConfig's apply. The gate only becomes effective once proxyrunner is upgraded + // to a version that reads this annotation; operator-only upgrades leave the race window + // in place until proxyrunner is also rolled. Exported because it forms a wire contract + // that external readers (operator, diagnostic tooling) may consume. + RunConfigMCPServerGenerationAnnotation = "toolhive.stacklok.dev/mcpserver-generation" ) // RuntimeName is the name identifier for the Kubernetes runtime @@ -397,22 +406,26 @@ func (c *Client) DeployWorkload(ctx context.Context, return 0, err } - // Create an apply configuration for the statefulset - statefulSetApply := appsv1apply.StatefulSet(containerName, namespace). - WithLabels(containerLabels). - WithSpec(buildStatefulSetSpec(containerName, podTemplateSpec, options)) - - // Apply the statefulset using server-side apply - createdStatefulSet, err := c.client.AppsV1().StatefulSets(namespace). - Apply(ctx, statefulSetApply, metav1.ApplyOptions{ - FieldManager: serviceFieldManager, - Force: true, - }) + ourGen := runConfigGeneration(options) + skip, err := c.shouldSkipStatefulSetApply(ctx, namespace, containerName, ourGen) if err != nil { - return 0, fmt.Errorf("failed to apply statefulset: %w", err) + return 0, err + } + if skip { + // Intentionally skip ensureBackendServices in the gated path: this pod's RunConfig + // is stale, so reconciling services here would clobber port/config fields set by + // the newer-generation pod under the same field manager + Force: true — the same + // race this gate prevents for the StatefulSet. The newer pod already reconciled + // services; if that failed, it returns an error and retries on its own. + return 0, nil } - slog.Debug("applied statefulset", "name", createdStatefulSet.Name) + createdStatefulSet, err := c.applyStatefulSet( + ctx, namespace, containerName, containerLabels, podTemplateSpec, options, ourGen, + ) + if err != nil { + return 0, err + } err = c.ensureBackendServices( ctx, containerName, namespace, containerLabels, transportType, options, createdStatefulSet) @@ -435,6 +448,84 @@ func (c *Client) DeployWorkload(ctx context.Context, return 0, nil } +// runConfigGeneration extracts the RunConfig MCPServer generation from options, +// returning 0 when options is nil (backward-compat / non-operator callers). +func runConfigGeneration(options *runtime.DeployWorkloadOptions) int64 { + if options == nil { + return 0 + } + return options.RunConfigMCPServerGeneration +} + +// applyStatefulSet stamps the MCPServer generation annotation when non-zero, +// builds the StatefulSet apply configuration, and performs the server-side apply. +func (c *Client) applyStatefulSet( + ctx context.Context, + namespace, containerName string, + containerLabels map[string]string, + podTemplateSpec *corev1apply.PodTemplateSpecApplyConfiguration, + options *runtime.DeployWorkloadOptions, + ourGen int64, +) (*appsv1.StatefulSet, error) { + if ourGen > 0 { + podTemplateSpec = podTemplateSpec.WithAnnotations(map[string]string{ + RunConfigMCPServerGenerationAnnotation: strconv.FormatInt(ourGen, 10), + }) + } + statefulSetApply := appsv1apply.StatefulSet(containerName, namespace). + WithLabels(containerLabels). + WithSpec(buildStatefulSetSpec(containerName, podTemplateSpec, options)) + createdStatefulSet, err := c.client.AppsV1().StatefulSets(namespace). + Apply(ctx, statefulSetApply, metav1.ApplyOptions{ + FieldManager: serviceFieldManager, + Force: true, + }) + if err != nil { + return nil, fmt.Errorf("failed to apply statefulset: %w", err) + } + slog.Debug("applied statefulset", "name", createdStatefulSet.Name) + return createdStatefulSet, nil +} + +// shouldSkipStatefulSetApply returns true when the existing StatefulSet is already +// stamped with a strictly greater MCPServer generation than ours, meaning a newer +// proxyrunner pod has already reconciled the workload and ours would be a regression. +// Returns false (apply as normal) when ourGen is zero or negative, when the StatefulSet +// does not yet exist, when the annotation is absent, or when the annotation is unparsable. +func (c *Client) shouldSkipStatefulSetApply( + ctx context.Context, namespace, name string, ourGen int64, +) (bool, error) { + if ourGen <= 0 { + return false, nil + } + existing, err := c.client.AppsV1().StatefulSets(namespace).Get(ctx, name, metav1.GetOptions{}) + if err != nil { + if errors.IsNotFound(err) { + return false, nil + } + return false, fmt.Errorf("failed to get existing statefulset: %w", err) + } + if existing.Spec.Template.Annotations == nil { + return false, nil + } + theirs := existing.Spec.Template.Annotations[RunConfigMCPServerGenerationAnnotation] + if theirs == "" { + return false, nil + } + theirsGen, parseErr := strconv.ParseInt(theirs, 10, 64) + if parseErr != nil { + slog.Warn("unparsable mcpserver-generation annotation; proceeding with apply", + "sts", name, "value", theirs, "err", parseErr) + return false, nil + } + if theirsGen > ourGen { + slog.Debug("skipping StatefulSet apply; newer MCPServer generation already applied", + "sts", name, "ours", ourGen, "theirs", theirsGen) + return true, nil + } + return false, nil +} + // buildStatefulSetSpec constructs the StatefulSet spec apply configuration. // WithReplicas is only included when BackendReplicas is explicitly set; omitting // the field lets the existing field manager (e.g. HPA or kubectl) retain control diff --git a/pkg/container/kubernetes/client_test.go b/pkg/container/kubernetes/client_test.go index 6d12be60ee..89e1d135d9 100644 --- a/pkg/container/kubernetes/client_test.go +++ b/pkg/container/kubernetes/client_test.go @@ -6,6 +6,7 @@ package kubernetes import ( "context" "encoding/json" + "strconv" "strings" "testing" "time" @@ -1367,3 +1368,192 @@ func TestDeployWorkloadBackendReplicas(t *testing.T) { assert.Equal(t, int32(0), *sts.Spec.Replicas) }) } + +func TestDeployWorkload_RunConfigMCPServerGenerationGate(t *testing.T) { + t.Parallel() + + const containerName = "test-container" + const oursGen = int64(100) + oursFormatted := strconv.FormatInt(oursGen, 10) + + // seededImage is distinct from the image passed to DeployWorkload so the + // "skipped apply" case can assert the seeded spec was NOT overwritten. + const seededImage = "seeded-image:pre-existing" + const deployImage = "test-image" + + newSeededSTS := func(annotation string) *appsv1.StatefulSet { + sts := &appsv1.StatefulSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: containerName, + Namespace: defaultNamespace, + }, + Spec: appsv1.StatefulSetSpec{ + Replicas: ptr.To(int32(1)), + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{}, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{ + Name: mcpContainerName, + Image: seededImage, + }}, + }, + }, + }, + Status: appsv1.StatefulSetStatus{ReadyReplicas: 1}, + } + if annotation != "" { + sts.Spec.Template.Annotations = map[string]string{ + RunConfigMCPServerGenerationAnnotation: annotation, + } + } + return sts + } + + testCases := []struct { + name string + seedSTS bool + seedAnnotation string + optionsGen int64 + expectApply bool + wantAnnotation string // expected annotation value on STS after call + wantAnnotationIs string // "missing" | "equal" — how to interpret wantAnnotation + }{ + { + name: "no_existing_sts", + seedSTS: false, + optionsGen: oursGen, + expectApply: true, + wantAnnotation: oursFormatted, + wantAnnotationIs: "equal", + }, + { + name: "existing_sts_no_annotation", + seedSTS: true, + seedAnnotation: "", + optionsGen: oursGen, + expectApply: true, + wantAnnotation: oursFormatted, + wantAnnotationIs: "equal", + }, + { + name: "existing_sts_older_annotation", + seedSTS: true, + seedAnnotation: strconv.FormatInt(int64(50), 10), + optionsGen: oursGen, + expectApply: true, + wantAnnotation: oursFormatted, + wantAnnotationIs: "equal", + }, + { + name: "existing_sts_equal_annotation", + seedSTS: true, + seedAnnotation: oursFormatted, + optionsGen: oursGen, + expectApply: true, + wantAnnotation: oursFormatted, + wantAnnotationIs: "equal", + }, + { + name: "existing_sts_newer_annotation", + seedSTS: true, + seedAnnotation: strconv.FormatInt(int64(200), 10), + optionsGen: oursGen, + expectApply: false, + wantAnnotation: strconv.FormatInt(int64(200), 10), + wantAnnotationIs: "equal", + }, + { + name: "existing_sts_unparsable_annotation", + seedSTS: true, + seedAnnotation: "not-a-number", + optionsGen: oursGen, + expectApply: true, + wantAnnotation: oursFormatted, + wantAnnotationIs: "equal", + }, + { + name: "zero_options_generation", + seedSTS: true, + seedAnnotation: strconv.FormatInt(int64(200), 10), + optionsGen: int64(0), + expectApply: true, + wantAnnotation: strconv.FormatInt(int64(200), 10), + wantAnnotationIs: "equal", + }, + { + name: "zero_options_generation_no_existing_sts", + seedSTS: false, + optionsGen: int64(0), + expectApply: true, + wantAnnotationIs: "missing", + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + var clientset *fake.Clientset + if tc.seedSTS { + clientset = fake.NewClientset(newSeededSTS(tc.seedAnnotation)) + } else { + clientset = fake.NewClientset() + } + + client := NewClientWithConfigAndPlatformDetector( + clientset, + &rest.Config{Host: "https://fake-k8s-api.example.com"}, + &mockPlatformDetector{platform: PlatformKubernetes}, + ) + client.waitForStatefulSetReadyFunc = mockWaitForStatefulSetReady + client.namespaceFunc = func() string { return defaultNamespace } + + options := runtime.NewDeployWorkloadOptions() + options.RunConfigMCPServerGeneration = tc.optionsGen + + _, err := client.DeployWorkload( + t.Context(), + deployImage, containerName, nil, + map[string]string{}, map[string]string{}, + nil, "streamable-http", options, false, + ) + require.NoError(t, err) + + sts, getErr := clientset.AppsV1().StatefulSets(defaultNamespace).Get( + t.Context(), containerName, metav1.GetOptions{}, + ) + + if !tc.expectApply { + // Apply was gated off. The seeded STS should still exist with its + // original image and annotation untouched. + require.NoError(t, getErr) + require.NotEmpty(t, sts.Spec.Template.Spec.Containers) + assert.Equal(t, seededImage, sts.Spec.Template.Spec.Containers[0].Image, + "seeded image should be preserved when apply is gated") + assert.Equal(t, tc.seedAnnotation, + sts.Spec.Template.Annotations[RunConfigMCPServerGenerationAnnotation], + "seeded annotation should be preserved when apply is gated") + return + } + + // Apply should have occurred. + require.NoError(t, getErr) + switch tc.wantAnnotationIs { + case "missing": + _, present := sts.Spec.Template.Annotations[RunConfigMCPServerGenerationAnnotation] + assert.False(t, present, + "annotation should not be added when options.RunConfigMCPServerGeneration is zero") + case "equal": + got := sts.Spec.Template.Annotations[RunConfigMCPServerGenerationAnnotation] + // Compare as int64 so a formatting mismatch (e.g. "+100" vs "100") doesn't + // cause a false positive; the canonical representation is strconv.FormatInt. + wantInt, werr := strconv.ParseInt(tc.wantAnnotation, 10, 64) + require.NoError(t, werr, "test expected annotation must be a parseable int64") + gotInt, gerr := strconv.ParseInt(got, 10, 64) + require.NoError(t, gerr, "annotation on STS must be a parseable int64, got %q", got) + assert.Equal(t, wantInt, gotInt, + "annotation value mismatch: got %q, want %q", got, tc.wantAnnotation) + } + }) + } +} diff --git a/pkg/container/runtime/types.go b/pkg/container/runtime/types.go index 04838e73a4..b65c518820 100644 --- a/pkg/container/runtime/types.go +++ b/pkg/container/runtime/types.go @@ -277,6 +277,11 @@ type DeployWorkloadOptions struct { // blocked by default in the egress proxy even when InsecureAllowAll is set. // Only applicable to Docker deployments with network isolation enabled. AllowDockerGateway bool + + // RunConfigMCPServerGeneration is the monotonic version stamp from the source RunConfig + // (the MCPServer .metadata.generation). K8s runtime uses it to refuse apply when the + // StatefulSet is already stamped with a strictly greater value. + RunConfigMCPServerGeneration int64 } // ScalingConfig holds horizontal-scaling knobs threaded from RunConfig down to diff --git a/pkg/runner/config.go b/pkg/runner/config.go index 2cb997087b..3934a632dd 100644 --- a/pkg/runner/config.go +++ b/pkg/runner/config.go @@ -49,6 +49,12 @@ type RunConfig struct { // SchemaVersion is the version of the RunConfig schema SchemaVersion string `json:"schema_version" yaml:"schema_version"` + // MCPServerGeneration is the K8s .metadata.generation of the MCPServer CR that rendered + // this RunConfig. The Kubernetes runtime uses it as a monotonic version to prevent stale + // rolling-update pods from overwriting a newer RunConfig's StatefulSet apply. Zero value + // means unversioned (backward-compat with older operators, or non-operator callers). + MCPServerGeneration int64 `json:"mcpserver_generation,omitempty" yaml:"mcpserver_generation,omitempty"` + // Image is the Docker image to run Image string `json:"image" yaml:"image"` diff --git a/pkg/runner/config_builder.go b/pkg/runner/config_builder.go index 13113f4a35..88c4fd1cb2 100644 --- a/pkg/runner/config_builder.go +++ b/pkg/runner/config_builder.go @@ -88,6 +88,14 @@ func WithImage(image string) RunConfigBuilderOption { } } +// WithMCPServerGeneration sets the MCPServer generation as the monotonic version stamp. +func WithMCPServerGeneration(gen int64) RunConfigBuilderOption { + return func(b *runConfigBuilder) error { + b.config.MCPServerGeneration = gen + return nil + } +} + // WithRuntimeConfig sets the runtime configuration (base images and packages) func WithRuntimeConfig(runtimeConfig *templates.RuntimeConfig) RunConfigBuilderOption { return func(b *runConfigBuilder) error { diff --git a/pkg/runner/config_test.go b/pkg/runner/config_test.go index bc67906f42..ad66252c9a 100644 --- a/pkg/runner/config_test.go +++ b/pkg/runner/config_test.go @@ -2532,3 +2532,52 @@ func TestRunConfig_SessionRedis(t *testing.T) { assert.Equal(t, "redis:6379", got.ScalingConfig.SessionRedis.Address) }) } + +func TestRunConfig_MCPServerGenerationJSONRoundTrip(t *testing.T) { + t.Parallel() + + t.Run("preserves non-zero value", func(t *testing.T) { + t.Parallel() + cfg := NewRunConfig() + cfg.Name = "generation-server" + cfg.MCPServerGeneration = 42 + + var buf bytes.Buffer + require.NoError(t, cfg.WriteJSON(&buf)) + + got, err := ReadJSON(&buf) + require.NoError(t, err) + assert.Equal(t, int64(42), got.MCPServerGeneration, + "MCPServerGeneration not preserved: got %d, want 42", got.MCPServerGeneration) + }) + + t.Run("missing field decodes as zero", func(t *testing.T) { + t.Parallel() + minimalJSON := `{"schema_version":"v0.1.0","image":"img","name":"n","transport":"stdio","host":"127.0.0.1","port":8080,"permission_profile":null}` //nolint:lll + + got, err := ReadJSON(strings.NewReader(minimalJSON)) + require.NoError(t, err) + assert.Equal(t, int64(0), got.MCPServerGeneration, + "MCPServerGeneration should be zero when missing, got %d", got.MCPServerGeneration) + }) + + t.Run("omitempty omits zero value", func(t *testing.T) { + t.Parallel() + // With the int64 type and `omitempty`, a zero MCPServerGeneration must not + // appear in the marshaled JSON. This is the key property that makes ConfigMap + // checksums deterministic across reconciles for unversioned (CLI) callers. + cfg := NewRunConfig() + cfg.Name = "zero-generation" + + var buf bytes.Buffer + require.NoError(t, cfg.WriteJSON(&buf)) + assert.False(t, bytes.Contains(buf.Bytes(), []byte("mcpserver_generation")), + "zero MCPServerGeneration should be omitted from JSON output; got:\n%s", buf.String()) + + // Round-trip: absent field decodes back to zero. + got, err := ReadJSON(&buf) + require.NoError(t, err) + assert.Equal(t, int64(0), got.MCPServerGeneration, + "decoded missing field should be zero, got %d", got.MCPServerGeneration) + }) +} diff --git a/pkg/runner/runner.go b/pkg/runner/runner.go index 9f36a1a15a..9e296d1fc8 100644 --- a/pkg/runner/runner.go +++ b/pkg/runner/runner.go @@ -341,6 +341,7 @@ func (r *Runner) Run(ctx context.Context) error { r.Config.TargetHost, r.Config.Publish, scalingConfig, + r.Config.MCPServerGeneration, ) if err != nil { return fmt.Errorf("failed to set up workload: %w", err) diff --git a/pkg/runtime/setup.go b/pkg/runtime/setup.go index 4cbbdc0046..a23d458c84 100644 --- a/pkg/runtime/setup.go +++ b/pkg/runtime/setup.go @@ -54,6 +54,7 @@ func Setup( targetHost string, publishedPorts []string, scalingConfig *rt.ScalingConfig, + runConfigMCPServerGeneration int64, ) (*SetupResult, error) { // Add transport-specific environment variables env, ok := transportEnvMap[transportType] @@ -99,6 +100,7 @@ func Setup( } containerOptions.ScalingConfig = scalingConfig containerOptions.AllowDockerGateway = allowDockerGateway + containerOptions.RunConfigMCPServerGeneration = runConfigMCPServerGeneration if transportType == types.TransportTypeStdio { containerOptions.AttachStdio = true