feat(operator): add five-container Bulwark worker pod [GMS 03/18]

deploy/operator/internal/consts/consts.go

@@ -136,12 +136,13 @@ const (
 	KubeAnnotationTopologyClusterTopologyName = "nvidia.com/topology-cluster-topology-name"
 	KubeLabelDynamoTopologyPrefix             = "nvidia.com/dynamo-topology."
 
-	DynamoDeploymentConfigEnvVar      = "DYN_DEPLOYMENT_CONFIG"
-	DynamoNamespaceEnvVar             = "DYN_NAMESPACE"
-	DynamoNamespacePrefixEnvVar       = "DYN_NAMESPACE_PREFIX"
-	DynamoNamespaceWorkerSuffixEnvVar = "DYN_NAMESPACE_WORKER_SUFFIX"
-	DynamoComponentEnvVar             = "DYN_COMPONENT"
-	DynamoDiscoveryBackendEnvVar      = "DYN_DISCOVERY_BACKEND"
+	DynamoDeploymentConfigEnvVar            = "DYN_DEPLOYMENT_CONFIG"
+	DynamoNamespaceEnvVar                   = "DYN_NAMESPACE"
+	DynamoNamespacePrefixEnvVar             = "DYN_NAMESPACE_PREFIX"
+	DynamoNamespaceWorkerSuffixEnvVar       = "DYN_NAMESPACE_WORKER_SUFFIX"
+	DynamoComponentEnvVar                   = "DYN_COMPONENT"
+	DynamoDiscoveryBackendEnvVar            = "DYN_DISCOVERY_BACKEND"
+	DynamoDiscoveryLogicalInstanceKeyEnvVar = "DYN_DISCOVERY_LOGICAL_INSTANCE_KEY"
 
 	// DynamoOperatorAllowGMSSnapshotEnvVar enables the temporary internal
 	// GMS + Snapshot admission gate when set to "1".

deploy/operator/internal/dynamo/backend_sglang.go

@@ -37,6 +37,12 @@ func (b *SGLangBackend) UpdateContainer(container *corev1.Container, numberOfNod
 			"next-steps", "upstream SGLang changes needed")
 	}
 
+	if component.IsInterPodGMSEnabled() {
+		if !containerHasGMSLoadFormat(container) {
+			injectFlagsIntoContainerCommand(container, "--load-format gms", false, "sglang")
+		}
+	}
+
 	// For single node, nothing to do
 	if numberOfNodes <= 1 {
 		return

deploy/operator/internal/dynamo/backend_trtllm.go

@@ -30,6 +30,12 @@ func (b *TRTLLMBackend) UpdateContainer(container *corev1.Container, numberOfNod
 			"next-steps", "upstream TensorRT-LLM changes needed")
 	}
 
+	if component.IsInterPodGMSEnabled() {
+		if !containerHasGMSLoadFormat(container) {
+			injectFlagsIntoContainerCommand(container, "--load-format gms", false, "trtllm")
+		}
+	}
+
 	// For single node, nothing to do
 	if numberOfNodes <= 1 {
 		return
@@ -141,7 +147,7 @@ func (b *TRTLLMBackend) setupLeaderContainer(container *corev1.Container, number
 		"cp /ssh-pk/private.key.pub $HOME/.ssh/authorized_keys",
 		"chmod 600 $HOME/.ssh/id_rsa $HOME/.ssh/authorized_keys",
 		"chmod 644 $HOME/.ssh/id_rsa.pub",
-		fmt.Sprintf("printf 'Host *\\nIdentityFile '$HOME'/.ssh/id_rsa\\nStrictHostKeyChecking no\\nPort %d\\n' > $HOME/.ssh/config", commonconsts.MpiRunSshPort),
+		fmt.Sprintf("printf 'Host *\\nIdentityFile '$HOME'/.ssh/id_rsa\\nStrictHostKeyChecking no\\nUserKnownHostsFile /dev/null\\nGlobalKnownHostsFile /dev/null\\nPort %d\\n' > $HOME/.ssh/config", commonconsts.MpiRunSshPort),
 	}
 
 	// Calculate total number of GPUs across all nodes. In the normal pod

deploy/operator/internal/dynamo/backend_vllm.go

@@ -23,6 +23,7 @@ const (
 	dataParallelSizeLocalFlag = "--data-parallel-size-local"
 	distributedExecutorFlag   = "--distributed-executor-backend"
 	enableElasticEPFlag       = "--enable-elastic-ep"
+	vllmGMSShadowModeEnvVar   = "DYN_VLLM_GMS_SHADOW_MODE"
 )
 
 type VLLMBackend struct {
@@ -34,15 +35,21 @@ func (b *VLLMBackend) UpdateContainer(container *corev1.Container, numberOfNodes
 	// to load weights from the dedicated GMS weight-server pod rather than
 	// from disk.
 	if component.IsInterPodGMSEnabled() {
-		if !containerHasArg(container, "--load-format", "gms") {
+		if !containerHasGMSLoadFormat(container) {
 			injectFlagsIntoContainerCommand(container, "--load-format gms", false, "vllm")
 		}
 		if component.IsInterPodFailoverEnabled() {
 			// DYN_VLLM_GMS_SHADOW_MODE activates vLLM's standby/failover behavior
 			// on top of --load-format gms. Standalone inter-pod GMS should not set
 			// it: those engines are active clients of the dedicated GMS weight
 			// server, not shadow engines waiting for failover activation.
-			container.Env = append(container.Env, corev1.EnvVar{Name: "DYN_VLLM_GMS_SHADOW_MODE", Value: "true"})
+			//
+			// This is a vLLM-engine-specific switch (it activates the vLLM-side
+			// GMS client path for shadow weight loading). It is injected here — in
+			// the vLLM backend — rather than in the backend-agnostic GMS helpers so
+			// non-vLLM backends do not inherit a stray, meaningless env var if/when
+			// inter-pod GMS is extended to them.
+			container.Env = append(container.Env, corev1.EnvVar{Name: vllmGMSShadowModeEnvVar, Value: "true"})
 		}
 	}
 
@@ -464,16 +471,6 @@ func injectElasticEPRayLaunchFlags(container *corev1.Container, role Role, servi
 	container.Command = []string{"/bin/sh", "-c"}
 }
 
-// hasFlag returns true if flag exists in expandedArgs.
-func hasFlag(expandedArgs []string, flag string) bool {
-	for _, arg := range expandedArgs {
-		if arg == flag {
-			return true
-		}
-	}
-	return false
-}
-
 func injectDataParallelLaunchFlags(container *corev1.Container, role Role, serviceName string, multinodeDeployer MultinodeDeployer, resources *corev1.ResourceRequirements, numberOfNodes int32) {
 	expandedArgs := getExpandedArgs(container)
 	leaderHostname := multinodeDeployer.GetLeaderHostname(serviceName)

deploy/operator/internal/dynamo/backend_vllm_test.go

@@ -1052,18 +1052,18 @@ func TestVLLMBackend_UpdateContainer_InterPodGMS(t *testing.T) {
 
 			count := 0
 			for _, e := range container.Env {
-				if e.Name == "DYN_VLLM_GMS_SHADOW_MODE" {
+				if e.Name == vllmGMSShadowModeEnvVar {
 					count++
 					if e.Value != "true" {
-						t.Errorf("DYN_VLLM_GMS_SHADOW_MODE value = %q, want %q", e.Value, "true")
+						t.Errorf("%s value = %q, want %q", vllmGMSShadowModeEnvVar, e.Value, "true")
 					}
 				}
 			}
 			if count != tt.shadowModeCount {
-				t.Errorf("DYN_VLLM_GMS_SHADOW_MODE env var count = %d, want %d", count, tt.shadowModeCount)
+				t.Errorf("%s env var count = %d, want %d", vllmGMSShadowModeEnvVar, count, tt.shadowModeCount)
 			}
 			if got := count > 0; got != tt.wantShadowMode {
-				t.Errorf("DYN_VLLM_GMS_SHADOW_MODE present = %v, want %v", got, tt.wantShadowMode)
+				t.Errorf("%s present = %v, want %v", vllmGMSShadowModeEnvVar, got, tt.wantShadowMode)
 			}
 		})
 	}
@@ -1086,8 +1086,58 @@ func TestVLLMBackend_UpdateContainer_NoInterPodGMS(t *testing.T) {
 		t.Errorf("--load-format gms must not be injected when inter-pod GMS is disabled")
 	}
 	for _, e := range container.Env {
-		if e.Name == "DYN_VLLM_GMS_SHADOW_MODE" {
-			t.Errorf("DYN_VLLM_GMS_SHADOW_MODE must not be injected when inter-pod GMS is disabled")
+		if e.Name == vllmGMSShadowModeEnvVar {
+			t.Errorf("%s must not be injected when inter-pod GMS is disabled", vllmGMSShadowModeEnvVar)
+		}
+	}
+}
+
+func TestSGLangBackend_UpdateContainer_InterPodGMS(t *testing.T) {
+	backend := &SGLangBackend{}
+	component := betaComponent(t, &v1alpha1.DynamoComponentDeploymentSharedSpec{
+		GPUMemoryService: &v1alpha1.GPUMemoryServiceSpec{
+			Enabled: true,
+			Mode:    v1alpha1.GMSModeInterPod,
+		},
+	})
+	container := &corev1.Container{
+		Command: []string{"python3"},
+		Args:    []string{"-m", "dynamo.sglang"},
+	}
+
+	backend.UpdateContainer(container, 1, RoleMain, component, "svc", &GroveMultinodeDeployer{})
+
+	if !containerHasGMSLoadFormat(container) {
+		t.Errorf("expected --load-format gms to be injected for SGLang inter-pod GMS; got args=%q", container.Args)
+	}
+	for _, e := range container.Env {
+		if e.Name == vllmGMSShadowModeEnvVar {
+			t.Errorf("vLLM-specific env var must not be injected for SGLang")
+		}
+	}
+}
+
+func TestTRTLLMBackend_UpdateContainer_InterPodGMS(t *testing.T) {
+	backend := &TRTLLMBackend{}
+	component := betaComponent(t, &v1alpha1.DynamoComponentDeploymentSharedSpec{
+		GPUMemoryService: &v1alpha1.GPUMemoryServiceSpec{
+			Enabled: true,
+			Mode:    v1alpha1.GMSModeInterPod,
+		},
+	})
+	container := &corev1.Container{
+		Command: []string{"python3"},
+		Args:    []string{"-m", "dynamo.trtllm"},
+	}
+
+	backend.UpdateContainer(container, 1, RoleMain, component, "svc", &GroveMultinodeDeployer{})
+
+	if !containerHasGMSLoadFormat(container) {
+		t.Errorf("expected --load-format gms to be injected for TRT-LLM inter-pod GMS; got args=%q", container.Args)
+	}
+	for _, e := range container.Env {
+		if e.Name == vllmGMSShadowModeEnvVar {
+			t.Errorf("vLLM-specific env var must not be injected for TRT-LLM")
 		}
 	}
 }