[SIMPL-30451] Mount shared storage in step pods

Configure Kubernetes executor step pods with the shared Dagster home and persistent volume used by the code location.

Changelog: fixed

src/distributed_execution/tightly_coupled/jobs.py

@@ -31,13 +31,31 @@ COMMON_TAGS = {
 # Per-step pod shape. Only honoured by k8s_job_executor.
 STEP_K8S_CONFIG = {
     "container_config": {
+        "env": [
+            {
+                "name": "DAGSTER_HOME",
+                "value": "/dagster/shared/distributed-execution",
+            }
+        ],
         "resources": {
             "requests": {"cpu": "100m", "memory": "128Mi"},
             "limits": {"cpu": "500m", "memory": "512Mi"},
         },
+        "volume_mounts": [
+            {
+                "name": "dagster-shared-storage",
+                "mount_path": "/dagster/shared",
+            }
+        ],
     },
     "pod_spec_config": {
         "restart_policy": "Never",
+        "volumes": [
+            {
+                "name": "dagster-shared-storage",
+                "persistent_volume_claim": {"claim_name": "dagster-shared-pvc"},
+            }
+        ],
     },
 }
 

tests/test_tightly_coupled.py

@@ -5,6 +5,7 @@ import pytest
 from distributed_execution.preflight import check_env_vars, describe_pod_identity
 from distributed_execution.repository import defs
 from distributed_execution.tightly_coupled.jobs import (
+    STEP_K8S_CONFIG,
     tightly_coupled_in_process_job,
     tightly_coupled_k8s_job,
     tightly_coupled_local_job,
@@ -35,6 +36,18 @@ def test_each_job_declares_its_execution_target(job, expected_executor):
     assert job.tags["executor"] == expected_executor
 
 
+def test_k8s_step_pods_mount_shared_dagster_storage():
+    container_config = STEP_K8S_CONFIG["container_config"]
+    pod_spec_config = STEP_K8S_CONFIG["pod_spec_config"]
+
+    assert {"name": "DAGSTER_HOME", "value": "/dagster/shared/distributed-execution"} in container_config["env"]
+    assert {"name": "dagster-shared-storage", "mount_path": "/dagster/shared"} in container_config["volume_mounts"]
+    assert {
+        "name": "dagster-shared-storage",
+        "persistent_volume_claim": {"claim_name": "dagster-shared-pvc"},
+    } in pod_spec_config["volumes"]
+
+
 def test_in_process_job_runs_end_to_end():
     result = tightly_coupled_in_process_job.execute_in_process()