From ba93a20ae27ab3b102ccd34f0c3e7430e14d3ab5 Mon Sep 17 00:00:00 2001
From: ILay <ilia.zakharchuk@t-systems.com>
Date: Tue, 1 Sep 2026 19:12:27 +0200
Subject: [PATCH] [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
---
 .../tightly_coupled/jobs.py                    | 18 ++++++++++++++++++
 tests/test_tightly_coupled.py                  | 13 +++++++++++++
 2 files changed, 31 insertions(+)

diff --git a/src/distributed_execution/tightly_coupled/jobs.py b/src/distributed_execution/tightly_coupled/jobs.py
index c5f82f9..cfa88da 100644
--- a/src/distributed_execution/tightly_coupled/jobs.py
+++ b/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"},
+            }
+        ],
     },
 }
 
diff --git a/tests/test_tightly_coupled.py b/tests/test_tightly_coupled.py
index de03428..e3571a1 100644
--- a/tests/test_tightly_coupled.py
+++ b/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()