src/distributed_execution/tightly_coupled/jobs.py

@@ -12,7 +12,7 @@ is the instance-level run launcher (see ``yaml/tightly-coupled/``).
 
 from __future__ import annotations
 
-from dagster import graph, in_process_executor, multiprocess_executor
+from dagster import fs_io_manager, graph, in_process_executor, multiprocess_executor
 from dagster_k8s import k8s_job_executor
 
 from distributed_execution.ops import (
@@ -28,15 +28,14 @@ COMMON_TAGS = {
     "business_operation": "DISTRIBUTED_EXECUTION_REFERENCE",
 }
 
+# RWX PVC mounted into the run worker and every step pod. Set as the I/O manager
+# base_dir rather than via DAGSTER_HOME: the Dagster chart already injects
+# DAGSTER_HOME=/tmp/dagster and the earlier duplicate wins in the step pod.
+SHARED_IO_BASE_DIR = "/dagster/shared/distributed-execution/storage"
+
 # 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"},
@@ -104,6 +103,8 @@ tightly_coupled_k8s_job = distributed_execution_reference.to_job(
     # config `execution:` block is accepted and then silently discarded - which
     # also makes job_image and job_namespace impossible to supply at launch time.
     executor_def=k8s_job_executor,
+    # Steps run in separate pods, so step outputs have to land on shared storage.
+    resource_defs={"io_manager": fs_io_manager.configured({"base_dir": SHARED_IO_BASE_DIR})},
     config={
         "execution": {
             "config": {

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 (
+    SHARED_IO_BASE_DIR,
     STEP_K8S_CONFIG,
     tightly_coupled_in_process_job,
     tightly_coupled_k8s_job,
@@ -40,12 +41,18 @@ 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"]
+    assert SHARED_IO_BASE_DIR.startswith("/dagster/shared/")
+
+
+def test_k8s_job_pins_io_manager_to_shared_storage():
+    io_manager = tightly_coupled_k8s_job.resource_defs["io_manager"]
+    resolved = io_manager.apply_config_mapping({}).value
+    assert resolved["config"]["base_dir"] == SHARED_IO_BASE_DIR
 
 
 def test_in_process_job_runs_end_to_end():