Address review feedback for MCP service session affinity

- Extract shared applyService helper to reduce duplication between
  createHeadlessService and createMCPService
- Add explicit SessionAffinityConfig.ClientIP.TimeoutSeconds (1800s)
  instead of relying on Kubernetes 3h default
- Rename SSEHeadlessServiceName to MCPServiceName across types, setup,
  and client packages
- Rename transportTypeRequiresHeadlessService to
  transportTypeRequiresBackendServices
- Extract deleteIfExists helper in operator finalizer, removing
  nolint:gocyclo
- Fix headless+NodePort conflict: skip NodePort type promotion for
  headless services (Kubernetes rejects clusterIP=None + type=NodePort)
- Use slog.Debug instead of slog.Info for service creation success paths
- Add spec.type and sessionAffinityConfig assertions to e2e tests

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: JAORMX

cmd/thv-operator/controllers/mcpserver_controller.go

@@ -1455,71 +1455,46 @@ func (r *MCPServerReconciler) updateMCPServerStatus(ctx context.Context, m *mcpv
 	return r.Status().Update(ctx, m)
 }
 
+// deleteIfExists fetches a Kubernetes object by name and namespace, and deletes it if it exists.
+// Returns nil if the object was not found or was successfully deleted.
+func (r *MCPServerReconciler) deleteIfExists(ctx context.Context, obj client.Object, name, namespace, kind string) error {
+	err := r.Get(ctx, types.NamespacedName{Name: name, Namespace: namespace}, obj)
+	if err == nil {
+		if delErr := r.Delete(ctx, obj); delErr != nil && !errors.IsNotFound(delErr) {
+			return fmt.Errorf("failed to delete %s %s: %w", kind, name, delErr)
+		}
+		return nil
+	}
+	if !errors.IsNotFound(err) {
+		return fmt.Errorf("failed to check %s %s: %w", kind, name, err)
+	}
+	return nil
+}
+
 // finalizeMCPServer performs the finalizer logic for the MCPServer
-//
-//nolint:gocyclo
 func (r *MCPServerReconciler) finalizeMCPServer(ctx context.Context, m *mcpv1alpha1.MCPServer) error {
-	ctxLogger := log.FromContext(ctx)
 	// Update the MCPServer status
 	m.Status.Phase = mcpv1alpha1.MCPServerPhaseTerminating
 	m.Status.Message = "MCP server is being terminated"
 	if err := r.Status().Update(ctx, m); err != nil {
 		return err
 	}
 
-	// Step 2: Attempt to delete associated StatefulSet by name
-	sts := &appsv1.StatefulSet{}
-	err := r.Get(ctx, types.NamespacedName{Name: m.Name, Namespace: m.Namespace}, sts)
-	if err == nil {
-		// StatefulSet found, delete it
-		if delErr := r.Delete(ctx, sts); delErr != nil && !errors.IsNotFound(delErr) {
-			return fmt.Errorf("failed to delete StatefulSet %s: %w", m.Name, delErr)
-		}
-	} else if !errors.IsNotFound(err) {
-		// Unexpected error (not just "not found")
-		return fmt.Errorf("failed to get StatefulSet %s: %w", m.Name, err)
-	}
-
-	// Step 3: Attempt to delete associated services by name
-	svc := &corev1.Service{}
-	headlessSvcName := fmt.Sprintf("mcp-%s-headless", m.Name)
-	err = r.Get(ctx, types.NamespacedName{Name: headlessSvcName, Namespace: m.Namespace}, svc)
-	if err == nil {
-		if delErr := r.Delete(ctx, svc); delErr != nil && !errors.IsNotFound(delErr) {
-			return fmt.Errorf("failed to delete Service %s: %w", headlessSvcName, delErr)
-		}
-	} else if !errors.IsNotFound(err) {
-		return fmt.Errorf("failed to check Service %s: %w", headlessSvcName, err)
+	// Delete associated StatefulSet
+	if err := r.deleteIfExists(ctx, &appsv1.StatefulSet{}, m.Name, m.Namespace, "StatefulSet"); err != nil {
+		return err
 	}
 
-	// Delete the MCP ClusterIP service with session affinity
-	mcpSvc := &corev1.Service{}
-	mcpSvcName := fmt.Sprintf("mcp-%s", m.Name)
-	err = r.Get(ctx, types.NamespacedName{Name: mcpSvcName, Namespace: m.Namespace}, mcpSvc)
-	if err == nil {
-		if delErr := r.Delete(ctx, mcpSvc); delErr != nil && !errors.IsNotFound(delErr) {
-			return fmt.Errorf("failed to delete Service %s: %w", mcpSvcName, delErr)
-		}
-	} else if !errors.IsNotFound(err) {
-		return fmt.Errorf("failed to check Service %s: %w", mcpSvcName, err)
+	// Delete associated services
+	if err := r.deleteIfExists(ctx, &corev1.Service{}, fmt.Sprintf("mcp-%s-headless", m.Name), m.Namespace, "Service"); err != nil {
+		return err
 	}
-
-	// Step 4: Delete associated RunConfig ConfigMap
-	runConfigName := fmt.Sprintf("%s-runconfig", m.Name)
-	runConfigMap := &corev1.ConfigMap{}
-	err = r.Get(ctx, types.NamespacedName{Name: runConfigName, Namespace: m.Namespace}, runConfigMap)
-	if err == nil {
-		if delErr := r.Delete(ctx, runConfigMap); delErr != nil && !errors.IsNotFound(delErr) {
-			return fmt.Errorf("failed to delete RunConfig ConfigMap %s: %w", runConfigName, delErr)
-		}
-		ctxLogger.Info("Deleted RunConfig ConfigMap", "name", runConfigName, "namespace", m.Namespace)
-	} else if !errors.IsNotFound(err) {
-		return fmt.Errorf("failed to check RunConfig ConfigMap %s: %w", runConfigName, err)
+	if err := r.deleteIfExists(ctx, &corev1.Service{}, fmt.Sprintf("mcp-%s", m.Name), m.Namespace, "Service"); err != nil {
+		return err
 	}
 
-	// The owner references will automatically delete the deployment and service
-	// when the MCPServer is deleted, so we don't need to do anything here.
-	return nil
+	// Delete associated RunConfig ConfigMap
+	return r.deleteIfExists(ctx, &corev1.ConfigMap{}, fmt.Sprintf("%s-runconfig", m.Name), m.Namespace, "ConfigMap")
 }
 
 // deploymentNeedsUpdate checks if the deployment needs to be updated

pkg/container/kubernetes/client.go

@@ -424,7 +424,7 @@ func (c *Client) DeployWorkload(ctx context.Context,
 	//nolint:gosec // G706: statefulset name from Kubernetes API response
 	slog.Info("applied statefulset", "name", createdStatefulSet.Name)
 
-	if transportTypeRequiresHeadlessService(transportType) && options != nil {
+	if transportTypeRequiresBackendServices(transportType) && options != nil {
 		// Create a headless service for DNS discovery
 		err := c.createHeadlessService(ctx, containerName, namespace, containerLabels, options)
 		if err != nil {
@@ -898,67 +898,111 @@ func createServicePorts(options *runtime.DeployWorkloadOptions) ([]*corev1apply.
 	return servicePorts, nil
 }
 
-// createHeadlessService creates a headless Kubernetes service for the StatefulSet
-func (c *Client) createHeadlessService(
+// serviceConfig holds the configuration for creating a Kubernetes service via applyService.
+type serviceConfig struct {
+	// nameSuffix is appended to "mcp-<containerName>" to form the service name.
+	// Use "-headless" for the headless service or "" for the MCP service.
+	nameSuffix string
+	// headless makes the service a headless service (ClusterIP: None).
+	headless bool
+	// sessionAffinity enables ClientIP session affinity with the given timeout.
+	sessionAffinity bool
+	// sessionAffinityTimeoutSeconds sets the timeout for ClientIP session affinity.
+	// Only used when sessionAffinity is true. Kubernetes defaults to 10800s (3h) if unset.
+	sessionAffinityTimeoutSeconds int32
+}
+
+// applyService creates or updates a Kubernetes service using server-side apply.
+func (c *Client) applyService(
 	ctx context.Context,
 	containerName string,
 	namespace string,
 	labels map[string]string,
 	options *runtime.DeployWorkloadOptions,
-) error {
-	// Create service ports from the container ports
+	cfg serviceConfig,
+) (string, error) {
 	servicePorts, err := createServicePorts(options)
 	if err != nil {
-		return err
+		return "", err
 	}
 
-	// If no ports were configured, don't create a service
 	if len(servicePorts) == 0 {
-		slog.Info("no ports configured for SSE transport, skipping service creation")
-		return nil
+		slog.Debug("no ports configured, skipping service creation")
+		return "", nil
 	}
 
-	// Create service type based on whether we have node ports
+	svcName := fmt.Sprintf("mcp-%s%s", containerName, cfg.nameSuffix)
+
+	// Determine service type based on whether any ports have NodePort set.
+	// Headless services (ClusterIP: None) cannot be NodePort, so skip the
+	// promotion for those — Kubernetes rejects clusterIP=None + type=NodePort.
 	serviceType := corev1.ServiceTypeClusterIP
-	for _, sp := range servicePorts {
-		if sp.NodePort != nil {
-			serviceType = corev1.ServiceTypeNodePort
-			break
+	if !cfg.headless {
+		for _, sp := range servicePorts {
+			if sp.NodePort != nil {
+				serviceType = corev1.ServiceTypeNodePort
+				break
+			}
 		}
 	}
 
-	// we want to generate a service name that is unique for the headless service
-	// to avoid conflicts with the proxy service
-	svcName := fmt.Sprintf("mcp-%s-headless", containerName)
+	spec := corev1apply.ServiceSpec().
+		WithSelector(map[string]string{
+			"app": containerName,
+		}).
+		WithPorts(servicePorts...).
+		WithType(serviceType)
+
+	if cfg.headless {
+		spec = spec.WithClusterIP("None")
+	}
+
+	if cfg.sessionAffinity {
+		spec = spec.
+			WithSessionAffinity(corev1.ServiceAffinityClientIP).
+			WithSessionAffinityConfig(corev1apply.SessionAffinityConfig().
+				WithClientIP(corev1apply.ClientIPConfig().
+					WithTimeoutSeconds(cfg.sessionAffinityTimeoutSeconds)))
+	}
 
-	// Create the service apply configuration
 	serviceApply := corev1apply.Service(svcName, namespace).
 		WithLabels(labels).
-		WithSpec(corev1apply.ServiceSpec().
-			WithSelector(map[string]string{
-				"app": containerName,
-			}).
-			WithPorts(servicePorts...).
-			WithType(serviceType).
-			WithClusterIP("None")) // "None" makes it a headless service
-
-	// Apply the service using server-side apply
-	fieldManager := serviceFieldManager
+		WithSpec(spec)
+
 	_, err = c.client.CoreV1().Services(namespace).
 		Apply(ctx, serviceApply, metav1.ApplyOptions{
-			FieldManager: fieldManager,
+			FieldManager: serviceFieldManager,
 			Force:        true,
 		})
-
 	if err != nil {
-		return fmt.Errorf("failed to apply service: %w", err)
+		return "", fmt.Errorf("failed to apply service %s: %w", svcName, err)
 	}
 
-	slog.Info("created headless service for HTTP transport", "name", containerName)
+	slog.Debug("applied service", "name", svcName)
+	return svcName, nil
+}
 
-	return nil
+// createHeadlessService creates a headless Kubernetes service for the StatefulSet
+func (c *Client) createHeadlessService(
+	ctx context.Context,
+	containerName string,
+	namespace string,
+	labels map[string]string,
+	options *runtime.DeployWorkloadOptions,
+) error {
+	_, err := c.applyService(ctx, containerName, namespace, labels, options, serviceConfig{
+		nameSuffix: "-headless",
+		headless:   true,
+	})
+	return err
 }
 
+// mcpServiceSessionAffinityTimeout is the timeout in seconds for ClientIP session affinity
+// on the MCP service. This controls how long kube-proxy pins a client IP to the same backend pod.
+// Note: this provides proxy-runner-level stickiness (L4), not per-MCP-session stickiness (L7).
+// True per-session routing would require Mcp-Session-Id-based routing at the proxy layer.
+const mcpServiceSessionAffinityTimeout int32 = 1800
+
 // createMCPService creates a regular ClusterIP service with SessionAffinity for the MCP server StatefulSet.
 // This service provides load balancing with client-IP-based session stickiness, which the proxy-runner
 // uses as its target host. The headless service is retained for DNS discovery purposes.
@@ -969,47 +1013,14 @@ func (c *Client) createMCPService(
 	labels map[string]string,
 	options *runtime.DeployWorkloadOptions,
 ) error {
-	// Create service ports from the container ports
-	servicePorts, err := createServicePorts(options)
+	svcName, err := c.applyService(ctx, containerName, namespace, labels, options, serviceConfig{
+		sessionAffinity:               true,
+		sessionAffinityTimeoutSeconds: mcpServiceSessionAffinityTimeout,
+	})
 	if err != nil {
 		return err
 	}
-
-	// If no ports were configured, don't create a service
-	if len(servicePorts) == 0 {
-		slog.Info("no ports configured for MCP transport, skipping service creation")
-		return nil
-	}
-
-	svcName := fmt.Sprintf("mcp-%s", containerName)
-
-	// Create the service apply configuration with SessionAffinity
-	serviceApply := corev1apply.Service(svcName, namespace).
-		WithLabels(labels).
-		WithSpec(corev1apply.ServiceSpec().
-			WithSelector(map[string]string{
-				"app": containerName,
-			}).
-			WithPorts(servicePorts...).
-			WithType(corev1.ServiceTypeClusterIP).
-			WithSessionAffinity(corev1.ServiceAffinityClientIP))
-
-	// Apply the service using server-side apply
-	fieldManager := serviceFieldManager
-	_, err = c.client.CoreV1().Services(namespace).
-		Apply(ctx, serviceApply, metav1.ApplyOptions{
-			FieldManager: fieldManager,
-			Force:        true,
-		})
-
-	if err != nil {
-		return fmt.Errorf("failed to apply MCP service: %w", err)
-	}
-
-	slog.Info("created MCP service with session affinity", "name", svcName)
-
-	// TODO: rename SSEHeadlessServiceName to MCPServiceName
-	options.SSEHeadlessServiceName = svcName
+	options.MCPServiceName = svcName
 	return nil
 }
 
@@ -1030,8 +1041,8 @@ func extractPortMappingsFromPod(pod *corev1.Pod) []runtime.PortMapping {
 	return ports
 }
 
-// transportTypeRequiresHeadlessService returns true if the transport type requires a headless service
-func transportTypeRequiresHeadlessService(transportType string) bool {
+// transportTypeRequiresBackendServices returns true if the transport type requires backend services
+func transportTypeRequiresBackendServices(transportType string) bool {
 	return transportType == string(transtypes.TransportTypeSSE) || transportType == string(transtypes.TransportTypeStreamableHTTP)
 }
 

pkg/container/runtime/types.go

@@ -256,9 +256,10 @@ type DeployWorkloadOptions struct {
 	// Only applicable when using Kubernetes runtime
 	K8sPodTemplatePatch string
 
-	// SSEHeadlessServiceName is the name of the Kubernetes service to use for the workload
-	// Only applicable when using Kubernetes runtime and SSE transport
-	SSEHeadlessServiceName string
+	// MCPServiceName is the name of the Kubernetes ClusterIP service used as the
+	// proxy-runner target for MCP server workloads.
+	// Only applicable when using Kubernetes runtime with SSE or streamable-http transport.
+	MCPServiceName string
 
 	// IgnoreConfig contains configuration for ignore patterns and tmpfs overlays
 	// Used to filter bind mount contents by hiding sensitive files

pkg/runtime/setup.go

@@ -124,9 +124,9 @@ func Setup(
 	} else {
 		// Update target host and port if needed (for Kubernetes)
 		if (transportType == types.TransportTypeSSE || transportType == types.TransportTypeStreamableHTTP) && rt.IsKubernetesRuntime() {
-			// If the SSEHeadlessServiceName is set, use it as the target host
-			if containerOptions.SSEHeadlessServiceName != "" {
-				result.TargetHost = containerOptions.SSEHeadlessServiceName
+			// If the MCPServiceName is set, use it as the target host
+			if containerOptions.MCPServiceName != "" {
+				result.TargetHost = containerOptions.MCPServiceName
 			}
 		}
 

test/e2e/chainsaw/operator/multi-tenancy/test-scenarios/sse/assert-mcpserver-svc.yaml

@@ -4,4 +4,8 @@ metadata:
   name: mcp-yardstick
   namespace: test-namespace
 spec:
+  type: ClusterIP
   sessionAffinity: ClientIP
+  sessionAffinityConfig:
+    clientIP:
+      timeoutSeconds: 1800

test/e2e/chainsaw/operator/multi-tenancy/test-scenarios/streamable-http/assert-mcpserver-svc.yaml

@@ -4,4 +4,8 @@ metadata:
   name: mcp-yardstick
   namespace: test-namespace
 spec:
+  type: ClusterIP
   sessionAffinity: ClientIP
+  sessionAffinityConfig:
+    clientIP:
+      timeoutSeconds: 1800

test/e2e/chainsaw/operator/single-tenancy/test-scenarios/sse/assert-mcpserver-svc.yaml

@@ -4,4 +4,8 @@ metadata:
   name: mcp-st-sse
   namespace: toolhive-system
 spec:
+  type: ClusterIP
   sessionAffinity: ClientIP
+  sessionAffinityConfig:
+    clientIP:
+      timeoutSeconds: 1800

test/e2e/chainsaw/operator/single-tenancy/test-scenarios/streamable-http/assert-mcpserver-svc.yaml

@@ -4,4 +4,8 @@ metadata:
   name: mcp-st-streamable-http
   namespace: toolhive-system
 spec:
+  type: ClusterIP
   sessionAffinity: ClientIP
+  sessionAffinityConfig:
+    clientIP:
+      timeoutSeconds: 1800