[SIMPL-30451] Add distributed-execution service with guide and reference implementations

Canonical location for documentation, example workflows and reference service implementations covering distributed execution patterns. Covers AC1-AC4: execution-target selection, decision support, readiness checks and code-level linkage. Tightly coupled jobs and the loosely coupled subprocess transport are verified by the test suite; the Kubernetes pipes transport is implemented but not yet cluster-run and is marked as such in the guide.

Changelog: added

src/distributed_execution/__init__.py

@@ -0,0 +1,3 @@
+"""Reference implementations for Dagster distributed execution targets."""
+
+__version__ = "0.1.0"

src/distributed_execution/loosely_coupled/__init__.py

@@ -0,0 +1 @@
+"""Loosely coupled (dagster-pipes) execution-target reference implementations."""

src/distributed_execution/loosely_coupled/jobs.py

@@ -0,0 +1,154 @@
+"""Loosely coupled reference implementations.
+
+Dagster does not run the work here. It dispatches an external payload and then
+acts as a listener: the payload writes structured messages over the
+dagster-pipes protocol, which Dagster materialises as logs and metadata. The
+payload never connects to the metadata database.
+
+Two transports are provided. Both dispatch the *same* payload script and produce
+the same output shape as the tightly coupled ``process_work_units`` op, so the
+graphs differ by exactly one node - see the guide's section 5.3.
+"""
+
+import os
+import sys
+from pathlib import Path
+
+from dagster import Failure, OpExecutionContext, Out, PipesSubprocessClient, graph, op
+from dagster_k8s import PipesK8sClient
+
+from distributed_execution.ops import (
+    generate_work_units,
+    report_execution_target,
+    summarise_results,
+)
+
+# Payload lives outside src/ so it can be built into its own image.
+_REPO_ROOT = Path(__file__).resolve().parents[3]
+PAYLOAD_SCRIPT = os.environ.get("PIPES_PAYLOAD_SCRIPT", str(_REPO_ROOT / "payload" / "work.py"))
+
+PAYLOAD_IMAGE = os.environ.get(
+    "PIPES_PAYLOAD_IMAGE",
+    "code.europa.eu:4567/simpl/simpl-open/data/supporting-data-services/distributed-execution/payload:0.1.0",
+)
+PAYLOAD_NAMESPACE = os.environ.get("PIPES_PAYLOAD_NAMESPACE", "dagster")
+
+COMMON_TAGS = {
+    "execution_target": "loosely_coupled",
+    "business_operation": "DISTRIBUTED_EXECUTION_REFERENCE",
+}
+
+
+def _results_from_pipes(context: OpExecutionContext, completed) -> list:
+    """Read the payload's results 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
+    as a failure rather than an empty result.
+    """
+    messages = completed.get_custom_messages()
+    if not messages:
+        raise Failure(
+            description=(
+                "No pipes messages received from the external payload. The workload may have "
+                "run successfully while its messages were lost. Verify the message path: for "
+                "the Kubernetes transport, confirm the pod log stream is readable and not "
+                "being truncated or intercepted by a log shipper."
+            )
+        )
+
+    payload = messages[-1]
+    leaked = payload.get("orchestration_env_visible") or []
+    if leaked:
+        context.log.warning(
+            "Payload could see orchestration runtime credentials: %s. "
+            "This defeats the isolation the loosely coupled target is chosen for.",
+            ", ".join(leaked),
+        )
+
+    context.log.info("Received %s results from external host %s", len(payload["results"]), payload["host"])
+    return payload["results"]
+
+
+@op(
+    description="Dispatches the external payload as a local subprocess and listens on the pipes channel.",
+    out=Out(list),
+)
+def dispatch_external_work_subprocess(
+    context: OpExecutionContext,
+    units: list,
+    pipes_subprocess_client: PipesSubprocessClient,
+) -> list:
+    completed = pipes_subprocess_client.run(
+        context=context,
+        command=[sys.executable, PAYLOAD_SCRIPT],
+        extras={"units": units},
+    )
+    return _results_from_pipes(context, completed)
+
+
+@op(
+    description="Dispatches the external payload as a Kubernetes Job and listens on the pod log stream.",
+    out=Out(list),
+)
+def dispatch_external_work_k8s(
+    context: OpExecutionContext,
+    units: list,
+    pipes_k8s_client: PipesK8sClient,
+) -> list:
+    completed = pipes_k8s_client.run(
+        context=context,
+        image=PAYLOAD_IMAGE,
+        command=["python", "/app/work.py"],
+        namespace=PAYLOAD_NAMESPACE,
+        extras={"units": units},
+        base_pod_meta={
+            "labels": {
+                "app.kubernetes.io/name": "distributed-execution-payload",
+                "dagster/execution-target": "loosely-coupled",
+            }
+        },
+    )
+    return _results_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)
+    return summarise_results(results, target_report)
+
+
+@graph
+def loosely_coupled_k8s_reference():
+    target_report = report_execution_target()
+    units = generate_work_units()
+    results = dispatch_external_work_k8s(units)
+    return summarise_results(results, target_report)
+
+
+loosely_coupled_subprocess_job = loosely_coupled_subprocess_reference.to_job(
+    name="loosely_coupled_subprocess_job",
+    description=(
+        "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."
+    ),
+    resource_defs={"pipes_subprocess_client": PipesSubprocessClient()},
+    tags={**COMMON_TAGS, "transport": "subprocess"},
+)
+
+loosely_coupled_k8s_job = loosely_coupled_k8s_reference.to_job(
+    name="loosely_coupled_k8s_job",
+    description=(
+        "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."
+    ),
+    resource_defs={"pipes_k8s_client": PipesK8sClient()},
+    tags={**COMMON_TAGS, "transport": "k8s_pod_logs"},
+)
+
+LOOSELY_COUPLED_JOBS = [
+    loosely_coupled_subprocess_job,
+    loosely_coupled_k8s_job,
+]

src/distributed_execution/ops.py

@@ -0,0 +1,83 @@
+"""Ops shared by every execution-target reference implementation.
+
+The ops deliberately do trivial work. What they demonstrate is *where* the work
+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 distributed_execution.preflight import (
+    TIGHTLY_COUPLED_ENV_VARS,
+    check_env_vars,
+    describe_pod_identity,
+)
+
+
+class WorkUnitsConfig(Config):
+    count: int = 4
+
+
+@op(
+    description="Emits execution-target evidence: pod identity plus reachability of orchestration dependencies.",
+    out=Out(dict),
+)
+def report_execution_target(context: OpExecutionContext) -> dict:
+    identity = describe_pod_identity()
+    env_check = check_env_vars(TIGHTLY_COUPLED_ENV_VARS)
+
+    context.log.info("Executing on %s (pid %s), namespace %s", identity["hostname"], identity["pid"], identity["namespace"])
+    if env_check["missing"]:
+        context.log.warning(
+            "Orchestration runtime env vars not visible to this process: %s. "
+            "Expected for a loosely coupled target; a misconfiguration for a tightly coupled one.",
+            ", ".join(env_check["missing"]),
+        )
+
+    context.add_output_metadata(
+        {
+            "hostname": MetadataValue.text(identity["hostname"]),
+            "namespace": MetadataValue.text(identity["namespace"]),
+            "run_id": MetadataValue.text(identity["run_id"]),
+            "image": MetadataValue.text(identity["image"]),
+            "env_vars_present": MetadataValue.json(env_check["present"]),
+            "env_vars_missing": MetadataValue.json(env_check["missing"]),
+        }
+    )
+    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="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]:
+    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
+
+
+@op(description="Aggregates results and attaches the distinct hosts that contributed.")
+def summarise_results(context: OpExecutionContext, results: list[dict], target_report: dict) -> dict:
+    hosts = sorted({row["host"] for row in results})
+    summary = {
+        "units": len(results),
+        "total": sum(row["squared"] for row in results),
+        "contributing_hosts": hosts,
+        "launcher_namespace": target_report["identity"]["namespace"],
+    }
+    context.add_output_metadata(
+        {
+            "units": MetadataValue.int(summary["units"]),
+            "contributing_hosts": MetadataValue.json(hosts),
+            "launcher_namespace": MetadataValue.text(summary["launcher_namespace"]),
+        }
+    )
+    context.log.info("Summary: %s", summary)
+    return summary

src/distributed_execution/preflight.py

@@ -0,0 +1,61 @@
+"""Readiness checks backing the pre-run validation checklist.
+
+Each function returns a plain dict so the same result can be logged as Dagster
+metadata (evidence for the checklist) or asserted in a test.
+"""
+
+from __future__ import annotations
+
+import os
+import socket
+from urllib.parse import urlparse
+
+# Runtime dependencies a tightly coupled execution pod must be able to resolve.
+TIGHTLY_COUPLED_ENV_VARS = (
+    "DAGSTER_POSTGRES_HOST",
+    "DAGSTER_POSTGRES_USER",
+    "DAGSTER_POSTGRES_DB",
+)
+
+
+def check_env_vars(names: tuple[str, ...] | list[str]) -> dict:
+    """Report which of ``names`` are present, without echoing their values."""
+    present = [name for name in names if os.environ.get(name)]
+    missing = [name for name in names if not os.environ.get(name)]
+    return {
+        "check": "env_vars",
+        "passed": not missing,
+        "present": present,
+        "missing": missing,
+    }
+
+
+def check_tcp_reachable(host: str, port: int, timeout: float = 3.0) -> dict:
+    """Open a TCP connection to prove the execution pod can reach a dependency."""
+    result = {"check": "tcp_reachable", "target": f"{host}:{port}", "passed": False, "error": None}
+    try:
+        with socket.create_connection((host, port), timeout=timeout):
+            result["passed"] = True
+    except OSError as exc:
+        result["error"] = str(exc)
+    return result
+
+
+def check_endpoint_reachable(url: str, timeout: float = 3.0) -> dict:
+    """TCP-level reachability for an endpoint expressed as a URL."""
+    parsed = urlparse(url)
+    if not parsed.hostname:
+        return {"check": "tcp_reachable", "target": url, "passed": False, "error": "no hostname in URL"}
+    port = parsed.port or (443 if parsed.scheme == "https" else 80)
+    return check_tcp_reachable(parsed.hostname, port, timeout=timeout)
+
+
+def describe_pod_identity() -> dict:
+    """Where this process is actually running - the primary execution-target evidence."""
+    return {
+        "hostname": socket.gethostname(),
+        "pid": os.getpid(),
+        "namespace": os.environ.get("DAGSTER_K8S_PIPELINE_RUN_NAMESPACE", "<not-in-kubernetes>"),
+        "run_id": os.environ.get("DAGSTER_RUN_ID", "<unset>"),
+        "image": os.environ.get("DAGSTER_K8S_PIPELINE_RUN_IMAGE", "<unset>"),
+    }

src/distributed_execution/repository.py

@@ -0,0 +1,10 @@
+"""Dagster definitions for the distributed-execution code location."""
+
+from dagster import Definitions
+
+from distributed_execution.loosely_coupled.jobs import LOOSELY_COUPLED_JOBS
+from distributed_execution.tightly_coupled.jobs import TIGHTLY_COUPLED_JOBS
+
+defs = Definitions(
+    jobs=[*TIGHTLY_COUPLED_JOBS, *LOOSELY_COUPLED_JOBS],
+)

src/distributed_execution/tightly_coupled/__init__.py

@@ -0,0 +1 @@
+"""Tightly coupled execution-target reference implementations."""

src/distributed_execution/tightly_coupled/jobs.py

@@ -0,0 +1,97 @@
+"""Tightly coupled reference implementations.
+
+Tightly coupled means the process doing the work *is* a Dagster process: it
+imports the code location, connects to the metadata database and writes run
+events directly. Every execution pod therefore needs network reachability to the
+orchestration runtime dependencies.
+
+Three jobs are provided, differing only in their ``executor_def``. That single
+construct is the code-level half of the execution-target binding; the other half
+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_k8s import k8s_job_executor
+
+from distributed_execution.ops import (
+    generate_work_units,
+    process_work_units,
+    report_execution_target,
+    summarise_results,
+)
+
+# Applied to run pods by the K8sRunLauncher; surfaces in the Dagster UI run tags.
+COMMON_TAGS = {
+    "execution_target": "tightly_coupled",
+    "business_operation": "DISTRIBUTED_EXECUTION_REFERENCE",
+}
+
+# Per-step pod shape. Only honoured by k8s_job_executor.
+STEP_K8S_CONFIG = {
+    "container_config": {
+        "resources": {
+            "requests": {"cpu": "100m", "memory": "128Mi"},
+            "limits": {"cpu": "500m", "memory": "512Mi"},
+        },
+    },
+    "pod_spec_config": {
+        "restart_policy": "Never",
+    },
+}
+
+
+@graph
+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)
+    return summarise_results(results, target_report)
+
+
+# 1. Single process. Steps run inside the run worker itself - no fan-out at all.
+tightly_coupled_in_process_job = distributed_execution_reference.to_job(
+    name="tightly_coupled_in_process_job",
+    description=(
+        "Tightly coupled, single-process. Steps execute inside the run worker. "
+        "Runs unchanged on a laptop and in Kubernetes."
+    ),
+    executor_def=in_process_executor,
+    tags={**COMMON_TAGS, "executor": "in_process"},
+)
+
+# 2. Subprocesses on the run worker. Fan-out bounded by that one pod's resources.
+tightly_coupled_local_job = distributed_execution_reference.to_job(
+    name="tightly_coupled_local_job",
+    description=(
+        "Tightly coupled, multiprocess. Steps execute as subprocesses of the run worker; "
+        "concurrency is bounded by the run pod's CPU and memory limits."
+    ),
+    executor_def=multiprocess_executor.configured({"max_concurrent": 2}),
+    tags={**COMMON_TAGS, "executor": "multiprocess"},
+)
+
+# 3. One Kubernetes Job per step. Requires a cluster; each step pod connects to
+#    the metadata database on its own, which is what makes this tightly coupled.
+tightly_coupled_k8s_job = distributed_execution_reference.to_job(
+    name="tightly_coupled_k8s_job",
+    description=(
+        "Tightly coupled, one Kubernetes Job per step. Each step pod must reach the metadata "
+        "database, object storage and Vault. Requires a cluster - not runnable locally."
+    ),
+    executor_def=k8s_job_executor.configured(
+        {
+            "image_pull_policy": "IfNotPresent",
+            "step_k8s_config": STEP_K8S_CONFIG,
+        }
+    ),
+    tags={**COMMON_TAGS, "executor": "k8s_job"},
+)
+
+TIGHTLY_COUPLED_JOBS = [
+    tightly_coupled_in_process_job,
+    tightly_coupled_local_job,
+    tightly_coupled_k8s_job,
+]