ccc2e94c2a
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
60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
"""External payload for the loosely coupled execution target.
|
|
|
|
This script is deliberately NOT a Dagster code location. It depends only on
|
|
``dagster-pipes``, never on ``dagster``, and it never opens a connection to the
|
|
orchestration metadata database. Everything it reports reaches the control plane
|
|
through the pipes message channel.
|
|
|
|
Run by ``distributed_execution.loosely_coupled.jobs`` via either
|
|
``PipesSubprocessClient`` (local) or ``PipesK8sClient`` (cluster).
|
|
"""
|
|
|
|
import os
|
|
import socket
|
|
|
|
from dagster_pipes import open_dagster_pipes
|
|
|
|
# Presence of any of these would mean the payload was granted orchestration
|
|
# runtime credentials it has no business holding.
|
|
ORCHESTRATION_ENV_VARS = (
|
|
"DAGSTER_POSTGRES_HOST",
|
|
"DAGSTER_POSTGRES_USER",
|
|
"DAGSTER_POSTGRES_DB",
|
|
)
|
|
|
|
|
|
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 {worker} with {len(units)} work 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:
|
|
pipes.log.warning(
|
|
"Orchestration runtime credentials are visible to this payload: "
|
|
f"{', '.join(leaked)}. A loosely coupled target should not have them."
|
|
)
|
|
|
|
# The only channel back to the control plane.
|
|
pipes.report_custom_message(
|
|
{
|
|
"results": results,
|
|
"host": host,
|
|
"worker": worker,
|
|
"orchestration_env_visible": leaked,
|
|
}
|
|
)
|
|
|
|
pipes.log.info(f"External payload finished on {worker}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|