feat(deployments): executor-level default pod annotations/selector/tolerations/affinity

k8s/helm/values.yaml

@@ -577,13 +577,6 @@ basePlatformConfig: |
           enabled: true
           k8s_executor: local-k8s
           default_executor: local-k8s
-          # Platform-default pod annotations applied to every k8s model
-          # deployment/job (all engines). The Istio native-sidecar annotation
-          # makes a mesh-injected istio-proxy terminate when a Job's main
-          # container exits, so the weight-puller Job completes while keeping
-          # mesh mTLS egress (inject: "false" is NOT viable in-mesh).
-          default_pod_annotations:
-            sidecar.istio.io/nativeSidecar: "true"
           # Bypass the image `nemo` ENTRYPOINT; invoke the adapters module directly.
           # Avoids PermissionError when the sidecar writes instance state under $HOME.
           lora_sidecar_command:

plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/backend.py

@@ -139,6 +139,7 @@ async def create_deployment(
                 backend_config=backend_config,
                 config=config,
                 executor_image_pull_secrets=self._executor_config.image_pull_secrets,
+                executor_defaults=self._executor_config.to_k8s_defaults(),
                 secret_env=secret_env,
                 auth_context=auth_context,
                 workload_delegation_store=self._workload_delegations,
@@ -154,6 +155,7 @@ async def create_deployment(
             backend_config=backend_config,
             config=config,
             executor_image_pull_secrets=self._executor_config.image_pull_secrets,
+            executor_defaults=self._executor_config.to_k8s_defaults(),
             secret_env=secret_env,
             auth_context=auth_context,
             workload_delegation_store=self._workload_delegations,

plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/compiler.py

@@ -155,6 +155,24 @@ def build_container_spec(
     return k8s.client.V1Container(**kwargs)
 
 
+@dataclass(frozen=True)
+class ExecutorK8sDefaults:
+    """Executor-level k8s defaults applied to every workload the executor renders.
+
+    These are the base layer shared by ALL deployments-plugin consumers (models,
+    agents, ...). A per-entity ``K8sDeploymentConfig`` value overrides the default:
+    annotations merge key-wise (per-entity key wins); node_selector / tolerations /
+    affinity / topology_spread_constraints are applied only when the entity leaves
+    them unset (per-entity wins wholesale).
+    """
+
+    pod_annotations: dict[str, str] = field(default_factory=dict)
+    node_selector: dict[str, str] = field(default_factory=dict)
+    tolerations: list[dict[str, Any]] = field(default_factory=list)
+    affinity: dict[str, Any] = field(default_factory=dict)
+    topology_spread_constraints: list[dict[str, Any]] = field(default_factory=list)
+
+
 @dataclass(frozen=True)
 class CompiledWorkload:
     """Kubernetes objects derived from a DeploymentConfig."""
@@ -531,6 +549,7 @@ def compile_workload(
     k8s_config: K8sDeploymentConfig | None,
     pod_restart_policy: RestartPolicy,
     executor_image_pull_secrets: list[ImagePullSecret] | None = None,
+    executor_defaults: ExecutorK8sDefaults | None = None,
     secret_env: dict[str, str] | None = None,
 ) -> CompiledWorkload:
     """Compile pod spec kwargs and optional ConfigMap/Secret for a Job or Deployment.
@@ -618,7 +637,32 @@ def compile_workload(
     if effective_service_account_name:
         pod_spec_kwargs["service_account_name"] = effective_service_account_name
 
-    pod_annotations = dict(k8s_config.pod_annotations) if k8s_config is not None else {}
+    # Apply executor-level defaults as the BASE layer, shared by every consumer
+    # (models, agents, ...). A per-entity K8sDeploymentConfig value (set above)
+    # wins: node_selector / tolerations / affinity / topology_spread are applied
+    # only when the entity left them unset; annotations merge key-wise below.
+    entity_pod_annotations = dict(k8s_config.pod_annotations) if k8s_config is not None else {}
+    if executor_defaults is not None:
+        if executor_defaults.node_selector and "node_selector" not in pod_spec_kwargs:
+            pod_spec_kwargs["node_selector"] = dict(executor_defaults.node_selector)
+        if executor_defaults.tolerations and "tolerations" not in pod_spec_kwargs:
+            default_tolerations = build_tolerations(
+                [Toleration.model_validate(item) for item in executor_defaults.tolerations if item]
+            )
+            if default_tolerations:
+                pod_spec_kwargs["tolerations"] = default_tolerations
+        if executor_defaults.affinity and "affinity" not in pod_spec_kwargs:
+            default_affinity = _deserialize_k8s(executor_defaults.affinity, "V1Affinity")
+            if default_affinity is not None:
+                pod_spec_kwargs["affinity"] = default_affinity
+        if executor_defaults.topology_spread_constraints and "topology_spread_constraints" not in pod_spec_kwargs:
+            default_tsc = build_topology_spread_constraints(executor_defaults.topology_spread_constraints)
+            if default_tsc:
+                pod_spec_kwargs["topology_spread_constraints"] = default_tsc
+        # Annotations merge key-wise: executor default first, per-entity wins.
+        pod_annotations = {**executor_defaults.pod_annotations, **entity_pod_annotations}
+    else:
+        pod_annotations = entity_pod_annotations
 
     return CompiledWorkload(
         pod_spec_kwargs=pod_spec_kwargs,

plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/config.py

@@ -7,7 +7,10 @@
 
 import os
 import re
+from copy import deepcopy
+from typing import Any
 
+from nemo_deployments_plugin.backends.k8s.compiler import ExecutorK8sDefaults
 from nemo_platform_plugin.config import ImagePullSecret
 from pydantic import BaseModel, Field, field_validator
 
@@ -50,6 +53,47 @@ class K8sExecutorConfig(BaseModel):
         default_factory=list,
         description="Image pull secrets merged with platform image_pull_secrets on every pod.",
     )
+    default_pod_annotations: dict[str, str] = Field(
+        default_factory=dict,
+        description=(
+            "Executor-level default pod annotations stamped onto every Job/Deployment pod this "
+            "executor renders (all consumers: models, agents, ...). Merged key-wise with per-entity "
+            "backend_config.k8s.podAnnotations, where the per-entity value wins for a shared key. "
+            "Ships the Istio native-sidecar annotation so a mesh-injected proxy terminates when a "
+            "Job's main container exits."
+        ),
+    )
+    default_node_selector: dict[str, str] = Field(
+        default_factory=dict,
+        description=(
+            "Executor-level default nodeSelector applied to every Job/Deployment pod this executor "
+            "renders when the entity does not set backend_config.k8s.nodeSelector."
+        ),
+    )
+    default_tolerations: list[dict[str, str | int]] = Field(
+        default_factory=list,
+        description=(
+            "Executor-level default pod tolerations applied to every Job/Deployment pod this executor "
+            "renders when the entity does not set backend_config.k8s.tolerations. Each entry is a raw "
+            "Kubernetes toleration object."
+        ),
+    )
+    default_affinity: dict[str, Any] = Field(
+        default_factory=dict,
+        description=(
+            "Executor-level default pod affinity applied to every Job/Deployment pod this executor "
+            "renders when the entity does not set backend_config.k8s.affinity. Raw Kubernetes affinity "
+            "object (nodeAffinity / podAffinity / podAntiAffinity)."
+        ),
+    )
+    default_topology_spread_constraints: list[dict[str, Any]] = Field(
+        default_factory=list,
+        description=(
+            "Executor-level default pod topology spread constraints applied to every Job/Deployment pod "
+            "this executor renders when the entity does not set backend_config.k8s.topologySpreadConstraints. "
+            "Each entry is a raw Kubernetes topologySpreadConstraint object."
+        ),
+    )
 
     @field_validator("default_namespace")
     @classmethod
@@ -73,3 +117,18 @@ def effective_namespace(self) -> str:
         if pod_namespace:
             return pod_namespace
         return _FALLBACK_NAMESPACE
+
+    def to_k8s_defaults(self) -> ExecutorK8sDefaults:
+        """Bundle the executor-level pod defaults for the workload compiler.
+
+        Shared by every deployments-plugin consumer (models, agents, ...): the
+        compiler applies these as the base layer, with per-entity
+        ``backend_config.k8s`` values overriding them.
+        """
+        return ExecutorK8sDefaults(
+            pod_annotations=dict(self.default_pod_annotations),
+            node_selector=dict(self.default_node_selector),
+            tolerations=deepcopy(self.default_tolerations),
+            affinity=deepcopy(self.default_affinity),
+            topology_spread_constraints=deepcopy(self.default_topology_spread_constraints),
+        )

plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/deployments.py

@@ -21,6 +21,7 @@
 from nemo_deployments_plugin.backends.k8s.compiler import (
     CompiledWorkload,
     DeploymentConfigError,
+    ExecutorK8sDefaults,
     compile_workload,
     create_configmap,
     create_secret,
@@ -101,6 +102,7 @@ def build_deployment_body(
     deployment_name: str,
     k8s_config: K8sDeploymentConfig | None,
     executor_image_pull_secrets: list | None = None,
+    executor_defaults: ExecutorK8sDefaults | None = None,
     secret_env: dict[str, str] | None = None,
 ) -> BuiltDeployment:
     """Build an ``apps/v1.Deployment`` for create and its compiled workload."""
@@ -115,6 +117,7 @@ def build_deployment_body(
         k8s_config=k8s_config,
         pod_restart_policy="Always",
         executor_image_pull_secrets=executor_image_pull_secrets,
+        executor_defaults=executor_defaults,
         secret_env=secret_env,
     )
     deployment = k8s.client.V1Deployment(
@@ -331,6 +334,7 @@ async def create_deployment(
     backend_config: dict[str, Any],
     config: DeploymentConfig,
     executor_image_pull_secrets: list | None = None,
+    executor_defaults: ExecutorK8sDefaults | None = None,
     secret_env: dict[str, str] | None = None,
     auth_context: AuthContext | None = None,
     workload_delegation_store: WorkloadDelegationStore | None = None,
@@ -359,6 +363,7 @@ async def create_deployment(
             deployment_name=name,
             k8s_config=k8s_config,
             executor_image_pull_secrets=executor_image_pull_secrets,
+            executor_defaults=executor_defaults,
             secret_env=secret_env,
         )
         deployment_body = built.deployment

plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/jobs.py

@@ -18,6 +18,7 @@
 from nemo_deployments_plugin.backends.k8s.compiler import (
     CompiledWorkload,
     DeploymentConfigError,
+    ExecutorK8sDefaults,
     compile_workload,
     create_configmap,
     create_secret,
@@ -147,6 +148,7 @@ def build_job_body(
     deployment_name: str,
     k8s_config: K8sDeploymentConfig | None,
     executor_image_pull_secrets: list | None = None,
+    executor_defaults: ExecutorK8sDefaults | None = None,
     secret_env: dict[str, str] | None = None,
 ) -> BuiltJob:
     """Build a ``batch/v1.Job`` for create."""
@@ -159,6 +161,7 @@ def build_job_body(
         k8s_config=k8s_config,
         pod_restart_policy=config.restart_policy,
         executor_image_pull_secrets=executor_image_pull_secrets,
+        executor_defaults=executor_defaults,
         secret_env=secret_env,
     )
     job = k8s.client.V1Job(
@@ -249,6 +252,7 @@ async def create_job(
     backend_config: dict[str, Any],
     config: DeploymentConfig,
     executor_image_pull_secrets: list | None = None,
+    executor_defaults: ExecutorK8sDefaults | None = None,
     secret_env: dict[str, str] | None = None,
     auth_context: AuthContext | None = None,
     workload_delegation_store: WorkloadDelegationStore | None = None,
@@ -277,6 +281,7 @@ async def create_job(
             deployment_name=name,
             k8s_config=k8s_config,
             executor_image_pull_secrets=executor_image_pull_secrets,
+            executor_defaults=executor_defaults,
             secret_env=secret_env,
         )
         body = built.job

plugins/nemo-deployments/tests/unit/backends/k8s/test_backend.py

@@ -30,6 +30,45 @@ def test_default_namespace_rejects_invalid_dns_label() -> None:
         K8sExecutorConfig(default_namespace="X")
 
 
+def test_to_k8s_defaults_bundles_executor_pod_defaults() -> None:
+    config = K8sExecutorConfig(
+        default_pod_annotations={"sidecar.istio.io/nativeSidecar": "true"},
+        default_node_selector={"gpu": "a100"},
+        default_tolerations=[{"key": "gpu", "operator": "Equal", "value": "true", "effect": "NoSchedule"}],
+        default_affinity={"nodeAffinity": {}},
+        default_topology_spread_constraints=[{"maxSkew": 1, "topologyKey": "kubernetes.io/hostname"}],
+    )
+    defaults = config.to_k8s_defaults()
+    assert defaults.pod_annotations == {"sidecar.istio.io/nativeSidecar": "true"}
+    assert defaults.node_selector == {"gpu": "a100"}
+    assert defaults.tolerations[0]["key"] == "gpu"
+    assert defaults.affinity == {"nodeAffinity": {}}
+    assert defaults.topology_spread_constraints[0]["topologyKey"] == "kubernetes.io/hostname"
+
+
+def test_to_k8s_defaults_empty_by_default() -> None:
+    defaults = K8sExecutorConfig().to_k8s_defaults()
+    assert defaults.pod_annotations == {}
+    assert defaults.node_selector == {}
+    assert defaults.tolerations == []
+    assert defaults.affinity == {}
+    assert defaults.topology_spread_constraints == []
+
+
+def test_to_k8s_defaults_deep_copies_nested_objects() -> None:
+    # Mutating the returned defaults must not corrupt the shared executor config
+    # for later workloads (nested nodeAffinity / labelSelector are deep-copied).
+    config = K8sExecutorConfig(
+        default_affinity={"nodeAffinity": {"key": "orig"}},
+        default_topology_spread_constraints=[{"labelSelector": {"matchLabels": {"app": "orig"}}}],
+    )
+    defaults = config.to_k8s_defaults()
+    defaults.affinity["nodeAffinity"]["key"] = "mutated"
+    defaults.topology_spread_constraints[0]["labelSelector"]["matchLabels"]["app"] = "mutated"
+    assert config.default_affinity["nodeAffinity"]["key"] == "orig"
+    assert config.default_topology_spread_constraints[0]["labelSelector"]["matchLabels"]["app"] == "orig"
+
+
 def test_effective_namespace_prefers_explicit_config(monkeypatch: pytest.MonkeyPatch) -> None:
     # An explicit config value wins even when POD_NAMESPACE is set.
     monkeypatch.setenv("POD_NAMESPACE", "pod-ns")