feat(operator): add single-pod Bulwark GMS workers

Add Docker-container-oriented intra-pod Bulwark worker generation with frontend gateway, primary/shadow engines, separate GMS weights and KV servers, discovery/probe handling, and operator validation tests. Local rootfs runner support is intentionally excluded from this public branch.

Signed-off-by: Maksim Khadkevich <mkhadkevich@nvidia.com>

Author: hutm

components/src/dynamo/frontend/frontend_args.py

@@ -61,6 +61,7 @@ class FrontendConfig(RouterConfigBase, KvRouterConfigBase, AicPerfConfigBase):
 
     namespace: Optional[str] = None
     namespace_prefix: Optional[str] = None
+    bulwark_gateway_endpoint: Optional[str] = None
 
     migration_limit: int
     migration_max_seq_len: Optional[int]
@@ -93,6 +94,24 @@ def validate(self) -> None:
             self.router_mode = "kv"
         self.apply_load_aware_preset()
 
+        if self.bulwark_gateway_endpoint:
+            if not self.bulwark_gateway_endpoint.startswith("dyn://"):
+                raise ValueError(
+                    "--bulwark-gateway-endpoint must be a dyn://namespace.component.endpoint path"
+                )
+            if not self.namespace and not self.namespace_prefix:
+                raise ValueError(
+                    "--bulwark-gateway-endpoint requires --namespace or --namespace-prefix "
+                    "to select the private primary/shadow worker namespace"
+                )
+            if self.interactive:
+                raise ValueError(
+                    "--bulwark-gateway-endpoint cannot be combined with --interactive"
+                )
+            if self.kserve_grpc_server:
+                raise ValueError(
+                    "--bulwark-gateway-endpoint cannot be combined with --kserve-grpc-server"
+                )
         if bool(self.tls_cert_path) ^ bool(self.tls_key_path):  # ^ is XOR
             raise ValueError(
                 "--tls-cert-path and --tls-key-path must be provided together"
@@ -257,6 +276,19 @@ def add_arguments(self, parser) -> None:
             ),
         )
 
+        add_argument(
+            g,
+            flag_name="--bulwark-gateway-endpoint",
+            env_var="DYN_BULWARK_GATEWAY_ENDPOINT",
+            default=None,
+            help=(
+                "Expose this frontend as a stable request-plane worker endpoint for Bulwark "
+                "compound workers. The value is a public dyn://namespace.component.endpoint "
+                "path; private primary/shadow workers are discovered using --namespace or "
+                "--namespace-prefix."
+            ),
+        )
+
         add_argument(
             g,
             flag_name="--migration-limit",

components/src/dynamo/frontend/main.py

@@ -168,14 +168,18 @@ async def async_main():
     Initializes the distributed runtime, configures routing, and starts
     the HTTP server or interactive mode based on command-line arguments.
     """
-    # The system status server port is a worker concern.
+    # The system status server port is usually a worker concern.
     #
     # Serve tests set DYN_SYSTEM_PORT for the worker, but aggregated launch scripts
-    # start `dynamo.frontend` first. If the frontend inherits DYN_SYSTEM_PORT, it can
-    # bind that port before the worker, causing port conflicts and/or scraping the
-    # wrong metrics endpoint.
+    # start `dynamo.frontend` first. If the normal frontend inherits DYN_SYSTEM_PORT,
+    # it can bind that port before the worker, causing port conflicts and/or scraping
+    # the wrong metrics endpoint. Bulwark gateway endpoint mode intentionally keeps
+    # the system server so Kubernetes readiness can gate the stable worker entity.
+    frontend_system_port = os.environ.get("DYN_SYSTEM_PORT")
     os.environ.pop("DYN_SYSTEM_PORT", None)
     config, vllm_flags, sglang_flags = parse_args()
+    if config.bulwark_gateway_endpoint and frontend_system_port:
+        os.environ["DYN_SYSTEM_PORT"] = frontend_system_port
     dump_config(config.dump_config_to, config)
     if config.event_plane:
         os.environ["DYN_EVENT_PLANE"] = config.event_plane
@@ -192,8 +196,9 @@ async def async_main():
         f"Request migration {'enabled' if config.migration_limit > 0 else 'disabled'} "
         f"(limit: {config.migration_limit}{max_seq_info})"
     )
-    # Warn if DYN_SYSTEM_PORT is set (frontend doesn't use system metrics server)
-    if os.environ.get("DYN_SYSTEM_PORT"):
+    # Warn if DYN_SYSTEM_PORT is set for a normal frontend. Gateway endpoint mode
+    # uses it for Kubernetes readiness.
+    if os.environ.get("DYN_SYSTEM_PORT") and not config.bulwark_gateway_endpoint:
         logger.warning(
             "=" * 80 + "\n"
             "WARNING: DYN_SYSTEM_PORT is set but NOT used by the frontend!\n"
@@ -307,7 +312,15 @@ def signal_handler():
     engine = await make_engine(runtime, e)
 
     try:
-        if config.interactive:
+        if config.bulwark_gateway_endpoint:
+            logger.info(
+                "Starting Bulwark frontend gateway at %s; private workers selected by namespace=%s namespace_prefix=%s",
+                config.bulwark_gateway_endpoint,
+                config.namespace,
+                config.namespace_prefix,
+            )
+            await run_input(runtime, config.bulwark_gateway_endpoint, engine)
+        elif config.interactive:
             await run_input(runtime, "text", engine)
         elif config.kserve_grpc_server:
             await run_input(runtime, "grpc", engine)

deploy/operator/internal/consts/consts.go

@@ -103,12 +103,13 @@ const (
 	// corresponding node label onto the pod after scheduling.
 	KubeAnnotationTopologyLabelKey = "nvidia.com/topology-label-key"
 
-	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