From ccc2e94c2ab7f46cc068123de7ab5210dc242f24 Mon Sep 17 00:00:00 2001 From: ILay Date: Wed, 26 Aug 2026 18:50:01 +0200 Subject: [PATCH] [SIMPL-30451] Fan out work units so the execution target is observable The reference graph processed every work unit inside a single op, so exactly one worker was ever reported regardless of executor. That made the guide's claim that k8s_job_executor yields several distinct contributing_hosts false, and left the reference implementations unable to demonstrate the executor choice at all. generate_work_units is now a DynamicOut and both graphs map over it, so one step is created per unit and the loosely coupled pattern dispatches one external workload per unit. Evidence is split into contributing_workers (host and pid, differs per process) and contributing_hosts (differs only across machines), because the previous single field could not distinguish multiprocess fan-out from no fan-out. Tests now assert the mapped step keys rather than a host count, since execute_in_process ignores executor_def and cannot prove executor behaviour on its own. Adds a Windows note: multiprocess_executor did not complete during authoring and left orphaned processes. Changelog: fixed --- .gitignore | 2 + README.md | 10 ++-- .../user-guide/distributed-execution-guide.md | 45 ++++++++++++---- documents/user-guide/readiness-checklist.md | 15 +++--- payload/work.py | 10 ++-- .../loosely_coupled/jobs.py | 52 ++++++++++++------- src/distributed_execution/ops.py | 39 +++++++++----- src/distributed_execution/preflight.py | 5 +- .../tightly_coupled/jobs.py | 4 +- tests/test_loosely_coupled.py | 19 +++++-- tests/test_tightly_coupled.py | 12 ++++- 11 files changed, 148 insertions(+), 65 deletions(-) diff --git a/.gitignore b/.gitignore index 1ba4784..d287118 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,5 @@ .venv/ .pytest_cache/ .coverage + +.tmp_dagster_home_*/ diff --git a/README.md b/README.md index 4e4167f..55e2cb6 100644 --- a/README.md +++ b/README.md @@ -53,9 +53,13 @@ uv run dagster dev -f src/distributed_execution/repository.py ``` The Dagster UI is then available at . Two jobs run end to -end on a laptop with no cluster: `tightly_coupled_local_job` and -`loosely_coupled_subprocess_job`. The Kubernetes variants of each are documented -in the user guide. +end on a laptop with no cluster: `tightly_coupled_in_process_job` and +`loosely_coupled_subprocess_job`. The Kubernetes variants of each require a +cluster and are documented in the user guide. + +> `tightly_coupled_local_job` uses `multiprocess_executor`. It did not complete on +> a Windows development machine during authoring — see the Windows note in the +> user guide's section 5.5. ### Running tests diff --git a/documents/user-guide/distributed-execution-guide.md b/documents/user-guide/distributed-execution-guide.md index 6d98c5a..f51c13a 100644 --- a/documents/user-guide/distributed-execution-guide.md +++ b/documents/user-guide/distributed-execution-guide.md @@ -250,8 +250,20 @@ operationally: [yaml/tightly-coupled/rbac-step-executor.yaml](../../yaml/tightly-coupled/rbac-step-executor.yaml). - Per-step resource requests apply per pod, so the aggregate request for a fan-out step is the per-step request multiplied by concurrency. -- `summarise_results` now reports several distinct `contributing_hosts` instead of - one. That output metadata is the evidence that the switch actually took effect. + +The evidence that the switch took effect is in `summarise_results` output +metadata, and the two fields say different things: + +| Executor | `contributing_workers` | `contributing_hosts` | +|---|---|---| +| `in_process_executor` | 1 | 1 | +| `multiprocess_executor` | one per unit | 1 — same machine | +| `k8s_job_executor` | one per unit | one per unit — separate pods | + +This only works because `generate_work_units` is a `DynamicOut` and the graph +does `units.map(process_work_unit).collect()`. A single op looping over all units +internally would report one worker under *every* executor, because one step +cannot span processes or pods. ### 5.3 Worked example — switching to loosely coupled @@ -266,7 +278,7 @@ written, so the diff below is the real one. def distributed_execution_reference(): target_report = report_execution_target() units = generate_work_units() - results = process_work_units(units) # runs in-process + results = units.map(process_work_unit).collect() # runs in Dagster return summarise_results(results, target_report) ``` @@ -277,27 +289,27 @@ def distributed_execution_reference(): def loosely_coupled_k8s_reference(): target_report = report_execution_target() units = generate_work_units() - results = dispatch_external_work_k8s(units) # dispatches, then listens + results = units.map(dispatch_external_work_k8s).collect() # dispatches, then listens return summarise_results(results, target_report) ``` The dispatching op replaces direct computation with a pipes client call: ```python -@op(out=Out(list)) +@op(out=Out(dict)) def dispatch_external_work_k8s( context: OpExecutionContext, - units: list, + unit: int, pipes_k8s_client: PipesK8sClient, -) -> list: +) -> dict: completed = pipes_k8s_client.run( context=context, image=PAYLOAD_IMAGE, command=["python", "/app/work.py"], namespace=PAYLOAD_NAMESPACE, - extras={"units": units}, + extras={"units": [unit]}, ) - return _results_from_pipes(context, completed) + return _result_from_pipes(context, completed) ``` and the job supplies the client as a resource instead of an executor: @@ -343,6 +355,11 @@ else from the Dagster ecosystem — a test asserts this, because the moment the payload imports `dagster` the isolation argument for choosing this pattern collapses. +Because the graph maps over the dynamic output, one external workload is +dispatched **per unit**. A test asserts that the four units come back from four +distinct external workers, which is the loosely coupled equivalent of the +`contributing_workers` evidence in section 5.2. + #### Message channel choice `PipesK8sClient` defaults to `PipesK8sPodLogsMessageReader`, which is what the @@ -383,7 +400,7 @@ Those take effect only when a new image is built and the code location reloads. | Job | Executor / transport | Runs locally | Demonstrates | |---|---|---|---| | `tightly_coupled_in_process_job` | `in_process_executor` | Yes | Baseline; steps inside the run worker | -| `tightly_coupled_local_job` | `multiprocess_executor` | Yes | Subprocess fan-out bounded by the run pod | +| `tightly_coupled_local_job` | `multiprocess_executor` | Linux/macOS | Subprocess fan-out bounded by the run pod | | `tightly_coupled_k8s_job` | `k8s_job_executor` | No — needs a cluster | One Kubernetes Job per step | | `loosely_coupled_subprocess_job` | `PipesSubprocessClient` | Yes | The pipes contract end to end, no cluster | | `loosely_coupled_k8s_job` | `PipesK8sClient` | No — needs a cluster | External Job dispatch, messages over pod logs | @@ -392,6 +409,14 @@ The last two dispatch the same [payload/work.py](../../payload/work.py). The subprocess variant exists so the pipes contract can be exercised, tested and demonstrated without any infrastructure. +> **Windows note.** `tightly_coupled_local_job` did not complete on a Windows +> development machine during authoring: the `multiprocess_executor` spawned step +> subprocesses that never terminated, and required manual cleanup. This was not +> reproduced on Linux and is not expected to affect cluster deployments, where +> run workers are Linux pods. For a laptop demonstration on Windows, prefer +> `tightly_coupled_in_process_job` or `loosely_coupled_subprocess_job`, both of +> which are covered by the test suite. + Run the local variants with: ```bash diff --git a/documents/user-guide/readiness-checklist.md b/documents/user-guide/readiness-checklist.md index 447fbdb..7681b2c 100644 --- a/documents/user-guide/readiness-checklist.md +++ b/documents/user-guide/readiness-checklist.md @@ -9,8 +9,8 @@ location image. > **NOT CLUSTER-VERIFIED.** Checks L4–L9 and the Kubernetes rows of section 4.3 > are derived from the implemented reference but have not yet been run against a -> Simpl cluster. Checks L1–L3, which exercise the payload contract and the -> message-parsing path, are covered by the test suite. +> Simpl cluster. Checks L1–L3 and L10, which exercise the payload contract, the +> message-parsing path and per-unit dispatch, are covered by the test suite. --- @@ -32,9 +32,9 @@ location image. | T1 | Run pod reaches the metadata database | Launch `tightly_coupled_in_process_job` | Run reaches `SUCCESS`; `report_execution_target` output metadata lists `DAGSTER_POSTGRES_HOST`, `DAGSTER_POSTGRES_USER` and `DAGSTER_POSTGRES_DB` under `env_vars_present`, and `env_vars_missing` is empty | | T2 | Vault injection works | Same run; inspect the run pod | `kubectl -n dagster describe pod ` shows the `vault-env` init container completed; no `vault:` literal remains in the process environment | | T3 | Object storage is reachable | Same run, if the workflow uses S3 | No `EndpointConnectionError` in run logs; `S3_ENDPOINT_URL` present in `env_vars_present` | -| T4 | Multiprocess fan-out is bounded correctly | Launch `tightly_coupled_local_job` | Run succeeds; `summarise_results` metadata shows exactly one entry in `contributing_hosts`, confirming steps stayed on the run worker | +| T4 | Multiprocess fan-out actually fans out | Launch `tightly_coupled_local_job` | Run succeeds; `summarise_results` metadata shows one entry per unit in `contributing_workers` and a single entry in `contributing_hosts` — separate processes, same machine | | T5 | RBAC permits step Jobs | `kubectl -n dagster auth can-i create jobs --as=system:serviceaccount:dagster:dagster-dev` | Returns `yes`; required only for `k8s_job_executor` | -| T6 | Step pods are actually created | Launch `tightly_coupled_k8s_job`, then `kubectl -n dagster get jobs -l dagster/run-id=` | One Job per step; `summarise_results` metadata shows **several** distinct `contributing_hosts` | +| T6 | Step pods are actually created | Launch `tightly_coupled_k8s_job`, then `kubectl -n dagster get jobs -l dagster/run-id=` | One Job per mapped unit; `contributing_hosts` now shows one entry **per unit**, not one | | T7 | Step pod egress is permitted | Same run | Steps do not hang in `STARTING`; run logs contain no connection timeouts to port 5432 | | T8 | Failure surfaces as a pod failure | Force a step failure in a scratch namespace | `failPodOnRunFailure: true` is set, and the step pod reports `Failed` rather than `Completed` | @@ -50,10 +50,11 @@ location image. | L6 | Dispatcher can read pod logs — **the message channel** | `kubectl -n auth can-i get pods/log --as=system:serviceaccount:dagster:dagster-dev` | Returns `yes`. A `no` here breaks reporting *without* failing the workload | | L7 | Payload Job is actually created | Launch `loosely_coupled_k8s_job`, then `kubectl -n get jobs -l app.kubernetes.io/name=distributed-execution-payload` | One Job per dispatch, labelled `dagster/execution-target=loosely-coupled` | | L8 | Payload has **no** orchestration connectivity | Same run; inspect run logs | No `Payload could see orchestration runtime credentials` warning. This is a positive check — absence of errors is not sufficient | -| L9 | Work ran off-platform | Same run | `summarise_results` metadata shows `contributing_hosts` containing the payload pod name, not the run worker's hostname | +| L9 | Work ran off-platform | Same run | `summarise_results` metadata shows `contributing_hosts` containing the payload pod names, not the run worker's hostname | +| L10 | One workload dispatched per unit | Same run, or `uv run pytest -k dispatched_to_its_own` locally | `contributing_workers` has one entry per unit; the local test asserts four distinct external workers | -Checks L4–L9 require a cluster. L1–L3 run on a laptop and should gate every -change to the payload or the dispatching op. +Checks L4–L9 require a cluster. L1–L3 and L10 run on a laptop and should gate +every change to the payload or the dispatching op. --- diff --git a/payload/work.py b/payload/work.py index 01ce542..a662a5e 100644 --- a/payload/work.py +++ b/payload/work.py @@ -27,10 +27,13 @@ def main() -> None: with open_dagster_pipes() as pipes: units = pipes.get_extra("units") host = socket.gethostname() + worker = f"{host}#{os.getpid()}" - pipes.log.info(f"External payload started on {host} with {len(units)} work units") + pipes.log.info(f"External payload started on {worker} with {len(units)} work units") - results = [{"unit": unit, "squared": unit * unit, "host": host} for unit in units] + results = [ + {"unit": unit, "squared": unit * unit, "host": host, "worker": worker} for unit in units + ] leaked = [name for name in ORCHESTRATION_ENV_VARS if os.environ.get(name)] if leaked: @@ -44,11 +47,12 @@ def main() -> None: { "results": results, "host": host, + "worker": worker, "orchestration_env_visible": leaked, } ) - pipes.log.info(f"External payload finished on {host}") + pipes.log.info(f"External payload finished on {worker}") if __name__ == "__main__": diff --git a/src/distributed_execution/loosely_coupled/jobs.py b/src/distributed_execution/loosely_coupled/jobs.py index b525bf7..ac1687a 100644 --- a/src/distributed_execution/loosely_coupled/jobs.py +++ b/src/distributed_execution/loosely_coupled/jobs.py @@ -14,7 +14,16 @@ import os import sys from pathlib import Path -from dagster import Failure, OpExecutionContext, Out, PipesSubprocessClient, graph, op +from dagster import ( + Failure, + OpExecutionContext, + Out, + PipesSubprocessClient, + graph, + in_process_executor, + multiprocess_executor, + op, +) from dagster_k8s import PipesK8sClient from distributed_execution.ops import ( @@ -39,8 +48,8 @@ COMMON_TAGS = { } -def _results_from_pipes(context: OpExecutionContext, completed) -> list: - """Read the payload's results off the message channel. +def _result_from_pipes(context: OpExecutionContext, completed) -> dict: + """Read one unit's result off the message channel. A broken message path is the defining failure mode of this pattern: the external workload can exit 0 while reporting nothing, so silence is treated @@ -66,42 +75,43 @@ def _results_from_pipes(context: OpExecutionContext, completed) -> list: ", ".join(leaked), ) - context.log.info("Received %s results from external host %s", len(payload["results"]), payload["host"]) - return payload["results"] + row = payload["results"][0] + context.log.info("Unit %s computed by external worker %s", row["unit"], row["worker"]) + return row @op( - description="Dispatches the external payload as a local subprocess and listens on the pipes channel.", - out=Out(list), + description="Dispatches one external payload as a local subprocess and listens on the pipes channel.", + out=Out(dict), ) def dispatch_external_work_subprocess( context: OpExecutionContext, - units: list, + unit: int, pipes_subprocess_client: PipesSubprocessClient, -) -> list: +) -> dict: completed = pipes_subprocess_client.run( context=context, command=[sys.executable, PAYLOAD_SCRIPT], - extras={"units": units}, + extras={"units": [unit]}, ) - return _results_from_pipes(context, completed) + return _result_from_pipes(context, completed) @op( - description="Dispatches the external payload as a Kubernetes Job and listens on the pod log stream.", - out=Out(list), + description="Dispatches one external payload as a Kubernetes Job and listens on the pod log stream.", + out=Out(dict), ) def dispatch_external_work_k8s( context: OpExecutionContext, - units: list, + unit: int, pipes_k8s_client: PipesK8sClient, -) -> list: +) -> dict: completed = pipes_k8s_client.run( context=context, image=PAYLOAD_IMAGE, command=["python", "/app/work.py"], namespace=PAYLOAD_NAMESPACE, - extras={"units": units}, + extras={"units": [unit]}, base_pod_meta={ "labels": { "app.kubernetes.io/name": "distributed-execution-payload", @@ -109,14 +119,14 @@ def dispatch_external_work_k8s( } }, ) - return _results_from_pipes(context, completed) + return _result_from_pipes(context, completed) @graph def loosely_coupled_subprocess_reference(): target_report = report_execution_target() units = generate_work_units() - results = dispatch_external_work_subprocess(units) + results = units.map(dispatch_external_work_subprocess).collect() return summarise_results(results, target_report) @@ -124,7 +134,7 @@ def loosely_coupled_subprocess_reference(): def loosely_coupled_k8s_reference(): target_report = report_execution_target() units = generate_work_units() - results = dispatch_external_work_k8s(units) + results = units.map(dispatch_external_work_k8s).collect() return summarise_results(results, target_report) @@ -134,6 +144,9 @@ loosely_coupled_subprocess_job = loosely_coupled_subprocess_reference.to_job( "Loosely coupled via subprocess transport. Demonstrates the pipes contract end to end " "on a laptop, with no cluster and no metadata database access from the payload." ), + # Dispatch is I/O-bound and the real work happens in the payload, so orchestrating + # in one process keeps the laptop demo free of platform-specific spawn behaviour. + executor_def=in_process_executor, resource_defs={"pipes_subprocess_client": PipesSubprocessClient()}, tags={**COMMON_TAGS, "transport": "subprocess"}, ) @@ -144,6 +157,7 @@ loosely_coupled_k8s_job = loosely_coupled_k8s_reference.to_job( "Loosely coupled via Kubernetes Job dispatch. Messages return over the pod log stream. " "Requires a cluster and RBAC to create Jobs in the target namespace." ), + executor_def=multiprocess_executor.configured({"max_concurrent": 2}), resource_defs={"pipes_k8s_client": PipesK8sClient()}, tags={**COMMON_TAGS, "transport": "k8s_pod_logs"}, ) diff --git a/src/distributed_execution/ops.py b/src/distributed_execution/ops.py index 47a5920..cc7283d 100644 --- a/src/distributed_execution/ops.py +++ b/src/distributed_execution/ops.py @@ -6,7 +6,7 @@ happens and *how* that fact reaches the control plane. # No `from __future__ import annotations`: it stringifies the `context` hint and # Dagster's op context validation then rejects it. -from dagster import Config, MetadataValue, OpExecutionContext, Out, op +from dagster import Config, DynamicOut, DynamicOutput, MetadataValue, OpExecutionContext, Out, op from distributed_execution.preflight import ( TIGHTLY_COUPLED_ENV_VARS, @@ -48,34 +48,47 @@ def report_execution_target(context: OpExecutionContext) -> dict: return {"identity": identity, "env_check": env_check} -@op(description="Produces the work units that later steps fan out over.", out=Out(list)) -def generate_work_units(context: OpExecutionContext, config: WorkUnitsConfig) -> list[int]: - units = list(range(config.count)) - context.log.info("Generated %s work units", len(units)) - return units +@op( + description="Fans out one dynamic output per work unit, so each unit becomes its own step.", + out=DynamicOut(int), +) +def generate_work_units(context: OpExecutionContext, config: WorkUnitsConfig): + context.log.info("Fanning out %s work units", config.count) + for unit in range(config.count): + yield DynamicOutput(unit, mapping_key=f"unit_{unit}") -@op(description="Performs one unit of work per element; runs once per step-execution slot.", out=Out(list)) -def process_work_units(context: OpExecutionContext, units: list[int]) -> list[dict]: +@op( + description="Performs one unit of work. One step per unit, so one process or pod per unit.", + out=Out(dict), +) +def process_work_unit(context: OpExecutionContext, unit: int) -> dict: identity = describe_pod_identity() - results = [{"unit": unit, "squared": unit * unit, "host": identity["hostname"]} for unit in units] - context.log.info("Processed %s units on %s", len(results), identity["hostname"]) - return results + context.log.info("Processing unit %s on %s", unit, identity["worker"]) + return { + "unit": unit, + "squared": unit * unit, + "host": identity["hostname"], + "worker": identity["worker"], + } -@op(description="Aggregates results and attaches the distinct hosts that contributed.") -def summarise_results(context: OpExecutionContext, results: list[dict], target_report: dict) -> dict: +@op(description="Aggregates results and records which workers actually contributed.") +def summarise_results(context: OpExecutionContext, results: list, target_report: dict) -> dict: hosts = sorted({row["host"] for row in results}) + workers = sorted({row["worker"] for row in results}) summary = { "units": len(results), "total": sum(row["squared"] for row in results), "contributing_hosts": hosts, + "contributing_workers": workers, "launcher_namespace": target_report["identity"]["namespace"], } context.add_output_metadata( { "units": MetadataValue.int(summary["units"]), "contributing_hosts": MetadataValue.json(hosts), + "contributing_workers": MetadataValue.json(workers), "launcher_namespace": MetadataValue.text(summary["launcher_namespace"]), } ) diff --git a/src/distributed_execution/preflight.py b/src/distributed_execution/preflight.py index f8c3031..295880c 100644 --- a/src/distributed_execution/preflight.py +++ b/src/distributed_execution/preflight.py @@ -52,9 +52,12 @@ def check_endpoint_reachable(url: str, timeout: float = 3.0) -> dict: def describe_pod_identity() -> dict: """Where this process is actually running - the primary execution-target evidence.""" + hostname = socket.gethostname() return { - "hostname": socket.gethostname(), + "hostname": hostname, "pid": os.getpid(), + # Distinguishes processes on one host, so multiprocess fan-out is visible locally. + "worker": f"{hostname}#{os.getpid()}", "namespace": os.environ.get("DAGSTER_K8S_PIPELINE_RUN_NAMESPACE", ""), "run_id": os.environ.get("DAGSTER_RUN_ID", ""), "image": os.environ.get("DAGSTER_K8S_PIPELINE_RUN_IMAGE", ""), diff --git a/src/distributed_execution/tightly_coupled/jobs.py b/src/distributed_execution/tightly_coupled/jobs.py index b606fa6..3b4abcf 100644 --- a/src/distributed_execution/tightly_coupled/jobs.py +++ b/src/distributed_execution/tightly_coupled/jobs.py @@ -17,7 +17,7 @@ from dagster_k8s import k8s_job_executor from distributed_execution.ops import ( generate_work_units, - process_work_units, + process_work_unit, report_execution_target, summarise_results, ) @@ -47,7 +47,7 @@ def distributed_execution_reference(): """Shared topology, so the three jobs differ only by execution target.""" target_report = report_execution_target() units = generate_work_units() - results = process_work_units(units) + results = units.map(process_work_unit).collect() return summarise_results(results, target_report) diff --git a/tests/test_loosely_coupled.py b/tests/test_loosely_coupled.py index 3699f57..77cba0d 100644 --- a/tests/test_loosely_coupled.py +++ b/tests/test_loosely_coupled.py @@ -55,20 +55,29 @@ def test_subprocess_job_runs_end_to_end(): assert summary["total"] == 14 +def test_each_unit_is_dispatched_to_its_own_external_worker(): + result = loosely_coupled_subprocess_job.execute_in_process() + + mapped = result.output_for_node("dispatch_external_work_subprocess") + assert set(mapped) == {"unit_0", "unit_1", "unit_2", "unit_3"} + # Each dispatch is its own OS process, so workers differ even on a single host. + assert len({row["worker"] for row in mapped.values()}) == 4 + + def test_payload_reports_no_orchestration_credentials(monkeypatch): monkeypatch.delenv("DAGSTER_POSTGRES_HOST", raising=False) monkeypatch.delenv("DAGSTER_POSTGRES_USER", raising=False) monkeypatch.delenv("DAGSTER_POSTGRES_DB", raising=False) result = loosely_coupled_subprocess_job.execute_in_process() - results = result.output_for_node("dispatch_external_work_subprocess") + mapped = result.output_for_node("dispatch_external_work_subprocess") - assert len(results) == 4 - assert all(row["host"] for row in results) + assert len(mapped) == 4 + assert all(row["worker"] for row in mapped.values()) def test_silent_message_path_is_treated_as_failure(): - from distributed_execution.loosely_coupled.jobs import _results_from_pipes + from distributed_execution.loosely_coupled.jobs import _result_from_pipes class _Silent: def get_custom_messages(self): @@ -83,4 +92,4 @@ def test_silent_message_path_is_treated_as_failure(): def info(*_args, **_kwargs): ... with pytest.raises(Failure, match="No pipes messages received"): - _results_from_pipes(_Ctx(), _Silent()) + _result_from_pipes(_Ctx(), _Silent()) diff --git a/tests/test_tightly_coupled.py b/tests/test_tightly_coupled.py index 9bfbf86..de03428 100644 --- a/tests/test_tightly_coupled.py +++ b/tests/test_tightly_coupled.py @@ -42,8 +42,16 @@ def test_in_process_job_runs_end_to_end(): summary = result.output_for_node("summarise_results") assert summary["units"] == 4 assert summary["total"] == 14 # 0 + 1 + 4 + 9 - # In-process execution never fans out beyond the run worker. - assert len(summary["contributing_hosts"]) == 1 + # In-process execution keeps every step in the run worker's own process. + assert len(summary["contributing_workers"]) == 1 + + +def test_work_units_fan_out_into_one_step_each(): + result = tightly_coupled_in_process_job.execute_in_process() + + mapped = result.output_for_node("process_work_unit") + # Without this the executor choice would be meaningless: one step cannot span pods. + assert set(mapped) == {"unit_0", "unit_1", "unit_2", "unit_3"} def test_env_var_check_reports_missing_names():