2026-08-31 16:14:04 +02:00
parent 0d8078889d 6532d640f9
commit b625b64df2
31 changed files with 4420 additions and 73 deletions
						
						
						
						
@@ -0,0 +1,151 @@
name: Build and Push Docker Images
on:
  push:
    branches:
      - main
  workflow_dispatch:
env:
  BASE_DOMAIN: dataprovider01.sandbox-cat-dat.simpl-europe.eu
  OWNER: ${{ github.repository_owner }}
jobs:
  build-and-push:
    runs-on: orchestration-platform
    defaults:
      run:
        shell: sh
    env:
      REGISTRY: gitea.${{ env.BASE_DOMAIN }}
      IMAGE_REPO: gitea.${{ env.BASE_DOMAIN }}/${{ env.OWNER }}/distributed-execution
      PAYLOAD_IMAGE_REPO: gitea.${{ env.BASE_DOMAIN }}/${{ env.OWNER }}/distributed-execution-payload
      REPO_DIR: repo
      REPO_CLONE_URL: https://gitea.${{ env.BASE_DOMAIN }}/${{ env.OWNER }}/distributed-execution.git
    steps:
      - name: Checkout repository (shell)
        run: |
          CLONE_USER="${{ secrets.REGISTRY_USERNAME }}"
          CLONE_PASS="${{ secrets.REGISTRY_PASSWORD }}"
          REF_NAME="${GITHUB_REF_NAME}"
          if [ -z "${REF_NAME}" ]; then
            REF_NAME="${GITHUB_REF#refs/heads/}"
          fi
          if [ -z "${CLONE_USER}" ] || [ -z "${CLONE_PASS}" ]; then
            echo "Missing REGISTRY_USERNAME or REGISTRY_PASSWORD secret"
            exit 1
          fi
          rm -rf "${REPO_DIR}"
          AUTH_HEADER="$(printf '%s:%s' "${CLONE_USER}" "${CLONE_PASS}" | base64 | tr -d '\n')"
          git clone --depth 1 --branch "${REF_NAME}" \
            -c "http.extraHeader=Authorization: Basic ${AUTH_HEADER}" \
            "${REPO_CLONE_URL}" \
            "${REPO_DIR}"
          if [ ! -f "${REPO_DIR}/Dockerfile" ]; then
            echo "Code location Dockerfile not found after clone"
            exit 1
          fi
          if [ ! -f "${REPO_DIR}/payload/Dockerfile" ]; then
            echo "Payload Dockerfile not found after clone"
            exit 1
          fi
      - name: Validate registry secrets
        run: |
          if [ -z "${{ secrets.REGISTRY_USERNAME }}" ] || [ -z "${{ secrets.REGISTRY_PASSWORD }}" ]; then
            echo "Missing REGISTRY_USERNAME or REGISTRY_PASSWORD secret"
            exit 1
          fi
      - name: Login to registry
        run: |
          echo "${{ secrets.REGISTRY_PASSWORD }}" | docker login "${REGISTRY}" \
            -u "${{ secrets.REGISTRY_USERNAME }}" --password-stdin
      # Both images take the same SHA tag: that is what keeps the code location and
      # the payload it dispatches on the same version.
      - name: Build code location image
        run: |
          COMMIT_SHA="${GITHUB_SHA:-$GITEA_SHA}"
          SHORT_SHA="$(echo "${COMMIT_SHA}" | cut -c1-12)"
          cd "${REPO_DIR}"
          docker build \
            -t "${IMAGE_REPO}:latest" \
            -t "${IMAGE_REPO}:${SHORT_SHA}" \
            .
      - name: Build payload image
        run: |
          COMMIT_SHA="${GITHUB_SHA:-$GITEA_SHA}"
          SHORT_SHA="$(echo "${COMMIT_SHA}" | cut -c1-12)"
          cd "${REPO_DIR}"
          docker build \
            -f payload/Dockerfile \
            -t "${PAYLOAD_IMAGE_REPO}:latest" \
            -t "${PAYLOAD_IMAGE_REPO}:${SHORT_SHA}" \
            payload/
      - name: Validate code location image
        run: |
          COMMIT_SHA="${GITHUB_SHA:-$GITEA_SHA}"
          SHORT_SHA="$(echo "${COMMIT_SHA}" | cut -c1-12)"
          docker run --rm "${IMAGE_REPO}:${SHORT_SHA}" \
            dagster definitions validate -f src/distributed_execution/repository.py
          docker run --rm "${IMAGE_REPO}:${SHORT_SHA}" \
            test -f /app/payload/work.py
      # Readiness checklist L11: the payload image must not carry the orchestration
      # runtime, otherwise the isolation argument for the loosely coupled target is void.
      - name: Validate payload image isolation
        run: |
          COMMIT_SHA="${GITHUB_SHA:-$GITEA_SHA}"
          SHORT_SHA="$(echo "${COMMIT_SHA}" | cut -c1-12)"
          docker run --rm "${PAYLOAD_IMAGE_REPO}:${SHORT_SHA}" python -c "
          import importlib.util
          assert importlib.util.find_spec('dagster_pipes') is not None, 'dagster_pipes missing from payload image'
          assert importlib.util.find_spec('dagster') is None, 'payload image must not contain the dagster package'
          print('payload isolation OK')
          "
      - name: Push code location image tags
        run: |
          COMMIT_SHA="${GITHUB_SHA:-$GITEA_SHA}"
          SHORT_SHA="$(echo "${COMMIT_SHA}" | cut -c1-12)"
          docker push "${IMAGE_REPO}:latest"
          docker push "${IMAGE_REPO}:${SHORT_SHA}"
      - name: Push payload image tags
        run: |
          COMMIT_SHA="${GITHUB_SHA:-$GITEA_SHA}"
          SHORT_SHA="$(echo "${COMMIT_SHA}" | cut -c1-12)"
          docker push "${PAYLOAD_IMAGE_REPO}:latest"
          docker push "${PAYLOAD_IMAGE_REPO}:${SHORT_SHA}"
      - name: Report image references
        run: |
          COMMIT_SHA="${GITHUB_SHA:-$GITEA_SHA}"
          SHORT_SHA="$(echo "${COMMIT_SHA}" | cut -c1-12)"
          echo "Code location image: ${IMAGE_REPO}:${SHORT_SHA}"
          echo "Payload image:       ${PAYLOAD_IMAGE_REPO}:${SHORT_SHA}"
          echo ""
          echo "Set PIPES_PAYLOAD_IMAGE on both the code location and the run pods:"
          echo "  PIPES_PAYLOAD_IMAGE=${PAYLOAD_IMAGE_REPO}:${SHORT_SHA}"
          echo "See yaml/loosely-coupled/values-pipes-payload.yaml."
      # The automated update of the deployment requires a technical user with
      # their kube config in the secrets. See the template repository's user manual.
      # - name: Update Dagster user deployment image
      #   run: |
      #     COMMIT_SHA="${GITHUB_SHA:-$GITEA_SHA}"
      #     SHORT_SHA="$(echo "${COMMIT_SHA}" | cut -c1-12)"
      #     kubectl patch deployment "${DEPLOYMENT_NAME}" \
      #       -n "${K8S_NAMESPACE}" \
      #       --type='strategic' \
      #       -p="{\"spec\":{\"template\":{\"spec\":{\"containers\":[{\"name\":\"dagster-user-deployments\",\"image\":\"${IMAGE_REPO}:${SHORT_SHA}\",\"env\":[{\"name\":\"DAGSTER_CURRENT_IMAGE\",\"value\":\"${IMAGE_REPO}:${SHORT_SHA}\"},{\"name\":\"PIPES_PAYLOAD_IMAGE\",\"value\":\"${PAYLOAD_IMAGE_REPO}:${SHORT_SHA}\"}]}]}}}}"
      #     kubectl rollout status deployment/"${DEPLOYMENT_NAME}" \
      #       -n "${K8S_NAMESPACE}" \
      #       --timeout=5m
						
						
						
						
@@ -0,0 +1,8 @@
*.egg-info/
**/__pycache__/
.venv/
.pytest_cache/
.coverage
.tmp_dagster_home_*/
.dagster_home/
						
						
						
						
@@ -0,0 +1,18 @@
## 0.1.0 (unreleased)
### added
- Repository skeleton for the `distributed-execution` supporting data service.
- User guide covering execution-target selection, decision support and readiness checks.
- Runnable tightly coupled reference implementations (`in_process`, `multiprocess`, `k8s_job_executor`).
- Runnable loosely coupled reference implementations via dagster-pipes
  (`PipesSubprocessClient`, `PipesK8sClient`) with a standalone external payload image.
- Working example configuration and RBAC for both execution targets.
### pending
- Publish the payload image to the container registry.
- End-to-end run of `loosely_coupled_k8s_job` against a cluster, and confirmation
  of the Kubernetes rows in the readiness checklist.
- Screenshots of UI surfaces and integration-pattern diagrams.
- Platform architecture document update.
						
						
						
						
@@ -0,0 +1,43 @@
FROM python:3.12-slim-bookworm
COPY --from=ghcr.io/astral-sh/uv:0.10.8 /uv /uvx /bin/
RUN python -m pip install --no-cache-dir --upgrade "pip==26.1.2"
WORKDIR /app
# Explicit UID 1000 matches the runtime UID expected by the Dagster chart
RUN addgroup --gid 1000 appgroup && \
    adduser --uid 1000 --gid 1000 --disabled-password --gecos "" appuser
RUN apt-get update && apt-get upgrade -y \
    && apt-get install -y --no-install-recommends \
    git=1:2.39.5-0+deb12u3 \
    curl=7.88.1-10+deb12u15 \
    && apt-get clean \
    && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/*
RUN chown -R appuser:appgroup /app /opt
COPY pyproject.toml .
COPY uv.lock .
COPY src/ ./src/
COPY payload/ ./payload/
COPY workspace.yaml .
ENV UV_COMPILE_BYTECODE=1
ENV UV_LINK_MODE=copy
RUN --mount=type=cache,target=/root/.cache/uv \
    uv sync --locked --no-dev
ENV PATH="/app/.venv/bin:${PATH}"
ENV PYTHONPATH="/app/src"
USER appuser
RUN dagster --version
EXPOSE 4000
CMD ["dagster", "code-server", "start", "-h", "0.0.0.0", "-p", "4000", "-f", "src/distributed_execution/repository.py"]
						
						
						
						
@@ -1,93 +1,98 @@
# distributed-execution
# Distributed Execution
Canonical location for documentation, example workflows and reference service
implementations covering **distributed execution patterns** in the Simpl
orchestration platform.
The service answers one question for every workflow: *where does the compute
actually run, and how does runtime information get back to the control plane?*
## Contents
| Area | Location | Status |
|---|---|---|
| User guide | [documents/user-guide/distributed-execution-guide.md](documents/user-guide/distributed-execution-guide.md) | Complete |
| Readiness checklist | [documents/user-guide/readiness-checklist.md](documents/user-guide/readiness-checklist.md) | Complete |
| Tightly coupled reference | [src/distributed_execution/tightly_coupled/jobs.py](src/distributed_execution/tightly_coupled/jobs.py) | Runnable |
| Loosely coupled reference | [src/distributed_execution/loosely_coupled/jobs.py](src/distributed_execution/loosely_coupled/jobs.py) | Runnable (subprocess verified) |
| External payload | [payload/work.py](payload/work.py) | Runnable |
| Example configuration | [yaml/](yaml/) | Complete |
## Project structure
```text
distributed-execution/
├── src/
│   └── distributed_execution/
│       ├── repository.py           # Dagster definitions (entry point)
│       ├── ops.py                  # Shared ops used by both patterns
│       ├── preflight.py            # Readiness checks backing the checklist
│       ├── tightly_coupled/
│       │   └── jobs.py             # in_process / multiprocess / k8s_job_executor
│       └── loosely_coupled/
│           └── jobs.py             # PipesSubprocessClient / PipesK8sClient
├── payload/                        # External workload: dagster-pipes ONLY
│   ├── work.py
│   ├── requirements.txt
│   └── Dockerfile
├── documents/user-guide/           # AC1-AC4 documentation
├── yaml/                           # Working example configuration
├── tests/
├── Dockerfile
├── pyproject.toml
└── workspace.yaml
```
## Getting started
## Getting started
To make it easy for you to get started with GitLab, here's a list of recommended next steps.
Prerequisites: Python 3.12+ and `uv`.
Already a pro? Just edit this README.md and make it your own. Want to make it easy? [Use the template at the bottom](#editing-this-readme)!
```bash
uv sync --dev
## Add your files
uv run dagster dev -f src/distributed_execution/repository.py
* [Create](https://docs.gitlab.com/user/project/repository/web_editor/#create-a-file) or [upload](https://docs.gitlab.com/user/project/repository/web_editor/#upload-a-file) files
* [Add files using the command line](https://docs.gitlab.com/topics/git/add_files/#add-files-to-a-git-repository) or push an existing Git repository with the following command:
```
cd existing_repo
git remote add origin https://code.europa.eu/simpl/simpl-open/data/supporting-data-services/distributed-execution/distributed-execution.git
git branch -M main
git push -uf origin main
```
```
## Integrate with your tools
The Dagster UI is then available at <http://localhost:3000>. Two jobs run end to
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.
* [Set up project integrations](https://code.europa.eu/simpl/simpl-open/data/supporting-data-services/distributed-execution/distributed-execution/-/settings/integrations)
> `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.
## Collaborate with your team
### Running tests
* [Invite team members and collaborators](https://docs.gitlab.com/user/project/members/)
```bash
* [Create a new merge request](https://docs.gitlab.com/user/project/merge_requests/creating_merge_requests/)
uv run pytest
* [Automatically close issues from merge requests](https://docs.gitlab.com/user/project/issues/managing_issues/#closing-issues-automatically)
```
* [Enable merge request approvals](https://docs.gitlab.com/user/project/merge_requests/approvals/)
* [Set auto-merge](https://docs.gitlab.com/user/project/merge_requests/auto_merge/)
## Test and Deploy
### Building the images
Use the built-in continuous integration in GitLab.
Two images, deliberately: the code location and the external payload are
versioned and scanned independently.
* [Get started with GitLab CI/CD](https://docs.gitlab.com/ci/quick_start/)
```bash
* [Analyze your code for known vulnerabilities with Static Application Security Testing (SAST)](https://docs.gitlab.com/user/application_security/sast/)
docker build -t distributed-execution:0.1.0 .
* [Deploy to Kubernetes, Amazon EC2, or Amazon ECS using Auto Deploy](https://docs.gitlab.com/topics/autodevops/requirements/)
docker build -f payload/Dockerfile -t distributed-execution-payload:0.1.0 payload/
* [Use pull-based deployments for improved Kubernetes management](https://docs.gitlab.com/user/clusters/agent/)
```
* [Set up protected environments](https://docs.gitlab.com/ci/environments/protected_environments/)
***
Both images build and have been smoke tested locally: the code location image
loads its definitions, and the payload image contains `dagster_pipes` without
`dagster` — check L11 in the readiness checklist.
# Editing this README
[.gitea/workflows/docker-publish.yml](.gitea/workflows/docker-publish.yml) runs
the same two builds on every push to `main`, applies those checks as gates, and
tags both images with the same short commit SHA — that shared tag is what keeps a
code location and the payload it dispatches on the same version.
When you're ready to make this README your own, just edit this file and use the handy template below (or feel free to structure it however you want - this is just a starting point!). Thanks to [makeareadme.com](https://www.makeareadme.com/) for this template.
## Status
## Suggestions for a good README
Both execution targets are implemented. The tightly coupled jobs and the loosely
coupled **subprocess** transport are verified by the test suite. The loosely
coupled **Kubernetes** transport is implemented but has not yet been run against
a cluster; see the guide's *Outstanding work* section.
Every project is different, so consider which of these sections apply to yours. The sections used in the template are suggestions for most open source projects. Also keep in mind that while a README can be too long and detailed, too long is better than too short. If you think your README is too long, consider utilizing another form of documentation rather than cutting out information.
## Licence
## Name
European Union Public Licence v1.2 — see [LICENSE](LICENSE).
Choose a self-explaining name for your project.
## Description
Let people know what your project can do specifically. Provide context and add a link to any reference visitors might be unfamiliar with. A list of Features or a Background subsection can also be added here. If there are alternatives to your project, this is a good place to list differentiating factors.
## Badges
On some READMEs, you may see small images that convey metadata, such as whether or not all the tests are passing for the project. You can use Shields to add some to your README. Many services also have instructions for adding a badge.
## Visuals
Depending on what you are making, it can be a good idea to include screenshots or even a video (you'll frequently see GIFs rather than actual videos). Tools like ttygif can help, but check out Asciinema for a more sophisticated method.
## Installation
Within a particular ecosystem, there may be a common way of installing things, such as using Yarn, NuGet, or Homebrew. However, consider the possibility that whoever is reading your README is a novice and would like more guidance. Listing specific steps helps remove ambiguity and gets people to using your project as quickly as possible. If it only runs in a specific context like a particular programming language version or operating system or has dependencies that have to be installed manually, also add a Requirements subsection.
## Usage
Use examples liberally, and show the expected output if you can. It's helpful to have inline the smallest example of usage that you can demonstrate, while providing links to more sophisticated examples if they are too long to reasonably include in the README.
## Support
Tell people where they can go to for help. It can be any combination of an issue tracker, a chat room, an email address, etc.
## Roadmap
If you have ideas for releases in the future, it is a good idea to list them in the README.
## Contributing
State if you are open to contributions and what your requirements are for accepting them.
For people who want to make changes to your project, it's helpful to have some documentation on how to get started. Perhaps there is a script that they should run or some environment variables that they need to set. Make these steps explicit. These instructions could also be useful to your future self.
You can also document commands to lint the code or run tests. These steps help to ensure high code quality and reduce the likelihood that the changes inadvertently break something. Having instructions for running tests is especially helpful if it requires external setup, such as starting a Selenium server for testing in a browser.
## Authors and acknowledgment
Show your appreciation to those who have contributed to the project.
## License
For open source projects, say how it is licensed.
## Project status
If you have run out of energy or time for your project, put a note at the top of the README saying that development has slowed down or stopped completely. Someone may choose to fork your project or volunteer to step in as a maintainer or owner, allowing your project to keep going. You can also make an explicit request for maintainers.
						
						
						
						
 
						
						
						
						
@@ -0,0 +1,522 @@
# Distributed Execution: Execution Targets and Integration Patterns
**Audience:** Participants with permission to configure workflows.
**Scope:** How to select, document and configure the execution target and
integration pattern for a Dagster workflow on the Simpl orchestration platform.
> **Document status.** Both setups are backed by runnable reference
> implementations in this repository. The tightly coupled setup and the loosely
> coupled **subprocess** transport are verified end to end, including runs
> launched from the Dagster UI — see [section 5.6](#56-what-a-verified-run-actually-produced).
> The loosely coupled **Kubernetes** transport is implemented and reviewed but
> has not yet been run against a cluster; statements specific to it are marked
> **NOT CLUSTER-VERIFIED**. See [Outstanding work](#8-outstanding-work).
---
## 1. Concepts
### 1.1 What an execution target is
The **execution target** is where the compute for a workflow actually runs. It is
not a single Dagster setting — it is the combination of three decisions.
| # | Decision | Where it is configured | Scope |
|---|---|---|---|
| 1 | **Placement** — which cluster and namespace the run lands in | Run launcher, in the Dagster instance / Helm values | Per deployment |
| 2 | **Granularity** — one process for the whole run, or one per step | `executor_def`, in workflow code | Per job |
| 3 | **Integration pattern** — whether the compute participates in Dagster's internals or reports back over a message channel | Job topology in workflow code | Per job |
Decisions 1 and 2 are always present. Decision 3 is what this guide calls the
*integration pattern*, and it is the one with the largest operational
consequences, because it determines what the execution pod must be able to reach.
### 1.2 The two integration patterns
**Tightly coupled.** The process doing the work *is* a Dagster process. It imports
the code location, opens a connection to the metadata database and writes run
events directly. Runtime information reaches the control plane because the worker
*is* part of the control plane's data path.
**Loosely coupled.** Dagster only dispatches. It creates a workload on an external
cluster running an arbitrary image, then acts as a listener. The external process
writes structured messages over the dagster-pipes protocol; Dagster reads them and
materialises them as logs and metadata. The execution pod never connects to the
metadata database.
```mermaid
flowchart LR
    subgraph TC["Tightly coupled"]
        direction TB
        D1[Dagster daemon] -->|K8sRunLauncher| RW1[Run worker pod]
        RW1 -->|k8s_job_executor| SP1[Step pods]
        RW1 -.->|events| DB1[(Metadata DB)]
        SP1 -.->|events| DB1
    end
    subgraph LC["Loosely coupled"]
        direction TB
        D2[Dagster daemon] -->|K8sRunLauncher| RW2[Run worker pod]
        RW2 -->|PipesK8sClient| EXT[External workload<br/>any image]
        EXT -.->|pipes messages| RW2
        RW2 -.->|events| DB2[(Metadata DB)]
    end
```
Note the asymmetry in the diagram: in the loosely coupled pattern only the run
worker touches the metadata database. That single arrow is the whole point of the
pattern.
---
## 2. AC1 — Choosing an execution target
### 2.1 Choose tightly coupled when
- The workflow logic is written in Python and lives in a Dagster code location.
- Execution pods can be granted network egress to the metadata database, object
  storage and Vault.
- You want per-step retries, step-level concurrency limits and asset
  materialisation handled natively by Dagster.
- The compute belongs to the same trust boundary as the orchestration platform.
### 2.2 Choose loosely coupled when
- The workload is not Python, or cannot take a dependency on the `dagster`
  package (different runtime, vendor image, GPU base image).
- The target cluster is administratively separate and **must not** be granted
  connectivity to the orchestration metadata database.
- The compute belongs to a different trust boundary, and you want the blast
  radius of a compromised execution pod limited to the message channel.
- The workload already exists as a container and you want to orchestrate it
  without rewriting it.
### 2.3 Prerequisites
| Prerequisite | Tightly coupled | Loosely coupled |
|---|---|---|
| Run launcher configured | Required — `K8sRunLauncher` | Required — `K8sRunLauncher` |
| Code location image contains workflow code | Required | Required for the *launching* job only |
| Execution pod → metadata database (Postgres 5432) | **Required** | Not required |
| Execution pod → object storage | Required if the workflow reads/writes objects | Required only if the payload itself does |
| Execution pod → Vault | Required for secret injection | Only for the launching pod |
| Kubernetes RBAC to create Jobs | Required for `k8s_job_executor` only | **Required** in the target namespace |
| Message channel provisioned | Not applicable | **Required** — pod log stream or object-storage reader |
| Control-plane reporting | Automatic — worker writes events directly | Via pipes messages; **fails silently if the channel breaks** |
### 2.4 What to document before the first execution
Record the following at workflow level, in the repository holding the workflow,
before it is run for the first time. Section 5.4 explains where each item lives.
1. **Chosen integration pattern** — tightly or loosely coupled, with a one-line
   rationale referencing section 2.1 or 2.2.
2. **Executor and its configuration** — including `max_concurrent` where relevant.
3. **Target namespace** for run pods and, for `k8s_job_executor`, step pods.
4. **Runtime dependencies** the execution pods must reach, as host/port pairs.
   This is the input to the NetworkPolicy review.
5. **Secrets consumed**, by Vault path — not by value.
6. **Expected evidence of a healthy run** — which of the checks in the
   [readiness checklist](readiness-checklist.md) apply, and what a passing run
   looks like in the UI.
7. **For loosely coupled only:** the payload image reference and the message
   channel type.
---
## 3. AC2 — Decision support
### 3.1 Side-by-side comparison
| Dimension | Tightly coupled | Loosely coupled |
|---|---|---|
| What runs the work | A Dagster process importing your code location | Any container image |
| Language constraint | Python, matching the code location's environment | None |
| Dependency on `dagster` package | Full package in the execution image | Only `dagster-pipes` in the payload image |
| Metadata DB connectivity from execution pod | **Required** | **Not required** |
| Runtime information flow | Direct writes to run storage | Structured messages → run worker → run storage |
| Log capture | Native, per step | Via the pipes message stream |
| Step-level retries | Native | Manual, at the dispatching op |
| Asset materialisation | Native, from any step | Reported by the payload, materialised by the launching op |
| Failure granularity | Per step | Per dispatched workload |
| Secret distribution | To every execution pod | To the launching pod, plus whatever the payload needs |
| Blast radius of a compromised execution pod | Orchestration runtime credentials | Message channel only |
| Cross-cluster execution | Not supported | Supported |
### 3.2 Environment-fit indicators
Use these as a fast triage. If several point the same way, that is your answer.
| Observation about your environment | Points to |
|---|---|
| Workflow is pure Python and shares a repository with existing code locations | Tightly coupled |
| Steps are short and numerous, and you want native retries | Tightly coupled |
| NetworkPolicy review for the execution namespace is already approved | Tightly coupled |
| Compute must run in a cluster you do not administer | Loosely coupled |
| Security review objects to distributing DB credentials to execution pods | Loosely coupled |
| Payload is an existing vendor or third-party image | Loosely coupled |
| Workload needs GPUs or a base image incompatible with the code location image | Loosely coupled |
| Team owning the payload is different from the team owning the workflow | Loosely coupled |
### 3.3 Trade-offs
**Tightly coupled**
- *Gains:* the richest observability with no extra work — every step's logs,
  metadata, retries and asset materialisations are native. Simplest mental model.
- *Costs:* the widest connectivity surface. Every execution pod is a credential
  holder and a database client. Scaling step pods scales database connections,
  which is the usual first bottleneck. Execution image and code location image are
  coupled, so a dependency conflict in one workflow can affect others sharing the
  location.
**Loosely coupled**
- *Gains:* a narrow connectivity surface and genuine dependency isolation. The
  payload team can ship on its own cadence. Cross-cluster and cross-trust-boundary
  execution become possible.
- *Costs:* observability is only as good as the payload's instrumentation — an
  uninstrumented payload appears in Dagster as one opaque step. Retry and
  cancellation semantics must be implemented at the dispatching op. The message
  channel becomes a new failure mode, and when it breaks the run can appear to
  succeed while reporting nothing. Two images to build, version and scan instead
  of one.
---
## 4. AC3 — Readiness checks
Pre-run validation items, their expected evidence, and the symptoms and
corrections for common misconfigurations are maintained separately, in
[readiness-checklist.md](readiness-checklist.md).
---
## 5. AC4 — Configuring execution-target linkage in code
### 5.1 Configuration constructs
Five constructs bind a workflow to an execution target. Three live in code and
travel with the workflow; two live in deployment configuration.
| Construct | Lives in | Binds |
|---|---|---|
| `executor_def` on `@job` / `to_job()` | Workflow code | Step granularity and placement |
| `tags` on `@job` | Workflow code | Run-level metadata; `dagster-k8s/config` overrides pod shape |
| Job topology — direct ops vs. a dispatching op using a pipes client | Workflow code | Integration pattern |
| `runLauncher` | Helm values | Run worker placement and reachable dependencies |
| `dagster-user-deployments` entry | Helm values | Which image the code location runs |
The first three are the answer to "which configuration constructs bind the
execution target to a workflow" — and they are all versioned alongside the
workflow, which is why setup selection is a code review concern and not a runtime
toggle.
### 5.2 Worked example — switching within the tightly coupled setup
The reference jobs share one graph and differ only by `executor_def`. This is the
smallest possible demonstration that granularity is a code-level decision.
**Before — steps run as subprocesses inside the run worker:**
```python
tightly_coupled_local_job = distributed_execution_reference.to_job(
    name="tightly_coupled_local_job",
    executor_def=multiprocess_executor.configured({"max_concurrent": 2}),
    tags={**COMMON_TAGS, "executor": "multiprocess"},
)
```
**After — one Kubernetes Job per step:**
```python
tightly_coupled_k8s_job = distributed_execution_reference.to_job(
    name="tightly_coupled_k8s_job",
    executor_def=k8s_job_executor.configured(
        {
            "image_pull_policy": "IfNotPresent",
            "step_k8s_config": STEP_K8S_CONFIG,
        }
    ),
    tags={**COMMON_TAGS, "executor": "k8s_job"},
)
```
The graph, the ops and the run launcher are all unchanged. What changes
operationally:
- Step pods are now created in the launcher's `jobNamespace`, so that namespace's
  NetworkPolicy must permit egress to Postgres, object storage and Vault — not
  just the run worker's namespace.
- The service account needs RBAC to create and watch Jobs. See
  [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.
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
The graphs for the two patterns differ by **exactly one node**. This is not a
simplification for the guide; it is how the reference implementations are
written, so the diff below is the real one.
**Before — tightly coupled. The work happens in a Dagster process:**
```python
@graph
def distributed_execution_reference():
    target_report = report_execution_target()
    units = generate_work_units()
    results = units.map(process_work_unit).collect()          # runs in Dagster
    return summarise_results(results, target_report)
```
**After — loosely coupled. The work is dispatched to an external image:**
```python
@graph
def loosely_coupled_k8s_reference():
    target_report = report_execution_target()
    units = generate_work_units()
    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(dict))
def dispatch_external_work_k8s(
    context: OpExecutionContext,
    unit: int,
    pipes_k8s_client: PipesK8sClient,
) -> dict:
    completed = pipes_k8s_client.run(
        context=context,
        image=PAYLOAD_IMAGE,
        command=["python", "/app/work.py"],
        namespace=PAYLOAD_NAMESPACE,
        extras={"units": [unit]},
    )
    return _result_from_pipes(context, completed)
```
and the job supplies the client as a resource:
```python
loosely_coupled_k8s_job = loosely_coupled_k8s_reference.to_job(
    name="loosely_coupled_k8s_job",
    executor_def=multiprocess_executor.configured({"max_concurrent": 2}),
    resource_defs={"pipes_k8s_client": PipesK8sClient()},
    tags={**COMMON_TAGS, "transport": "k8s_pod_logs"},
)
```
The executor is still present, but its meaning has changed. It no longer decides
where the work runs — the payload image and `namespace` do that. It only bounds
how many dispatching ops wait on the message channel at once. The two decisions
that were fused in the tightly coupled setup are now independent.
#### What changes operationally
| | Tightly coupled | Loosely coupled |
|---|---|---|
| Images to build | 1 (code location) | 2 (code location + payload) |
| Payload dependency surface | Full `dagster` package | `dagster-pipes` only |
| Execution pod → Postgres | Egress rule **required** | Egress rule can be **removed** |
| Execution pod → Vault / S3 | Credentials required | None needed by the payload |
| RBAC | Only for `k8s_job_executor` | Required on the **dispatching** pod |
| New failure mode | — | Broken message path |
The NetworkPolicy rules you can *remove* are the point of the exercise. See
[yaml/loosely-coupled/rbac-pipes-dispatch.yaml](../../yaml/loosely-coupled/rbac-pipes-dispatch.yaml)
for what replaces them — note that `pods/log` is the message channel, so
removing that verb silently breaks reporting without failing the workload.
#### The payload contract
[payload/work.py](../../payload/work.py) is the whole contract:
```python
with open_dagster_pipes() as pipes:
    units = pipes.get_extra("units")
    pipes.log.info(f"External payload started on {host}")
    pipes.report_custom_message({"results": results, "host": host, ...})
```
`pipes.log` entries appear in the Dagster run logs; `report_custom_message` is
read back by the dispatching op. The payload imports `dagster_pipes` and nothing
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
reference uses. It needs no extra infrastructure, but it couples message
delivery to the pod log stream — a log shipper that intercepts or truncates
stdout will break reporting while the workload still exits 0. That is why
`_result_from_pipes` raises on an empty message list rather than returning an
empty result. If your cluster's logging setup makes that transport unreliable,
switch to an object-storage message reader, which needs a bucket plus
credentials on both sides.
> **NOT CLUSTER-VERIFIED.** The Kubernetes transport above is implemented and
> its RBAC and configuration are recorded, but it has not yet been run against a
> Simpl cluster. The subprocess transport, which exercises the identical payload
> and the identical message-parsing path, is verified by the test suite.
### 5.4 Where setup selection is persisted
Setup selection is persisted in configuration artifacts, never as a runtime-only
setting. There is no UI control that switches a workflow between the two patterns.
| Artifact | Holds |
|---|---|
| `src/<package>/**/jobs.py` | `executor_def`, pipes client resource, tags, job topology |
| `payload/work.py` + `payload/Dockerfile` | The external payload contract and its image |
| `yaml/tightly-coupled/values-run-launcher.yaml` | Run launcher type, namespace, pod env |
| `yaml/tightly-coupled/rbac-step-executor.yaml` | Permissions required by `k8s_job_executor` |
| `yaml/loosely-coupled/values-pipes-payload.yaml` | Payload image reference and target namespace |
| `yaml/sandbox/values-sandbox-gitea.yaml` | Sandbox registry coordinates, pull secret, and the SHA both images share |
| `yaml/loosely-coupled/rbac-pipes-dispatch.yaml` | Permissions required by the dispatching pod |
| `yaml/values-dagster-distributed-execution.yaml` | Code location image and entry point |
The Launchpad can override *run configuration* — op config, resource config, tags
— for a single run. It cannot change the executor or the integration pattern.
Those take effect only when a new image is built and the code location reloads.
### 5.5 Reference implementations in this repository
| 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` | 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 |
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
uv sync --dev
uv run dagster dev -f src/distributed_execution/repository.py
```
`report_execution_target` emits pod identity and the visible orchestration
environment variables as output metadata on every run, which is the evidence the
readiness checklist refers to.
### 5.6 What a verified run actually produced
The two locally runnable jobs were launched from the Dagster UI, so the evidence
below comes from the full daemon → run launcher → executor path rather than from
a unit test. Hostnames are redacted; process IDs are real.
`tightly_coupled_in_process_job`:
```text
succeeded steps: 7
  generate_work_units, report_execution_target,
  process_work_unit[unit_0..unit_3], summarise_results
units                = 4
contributing_hosts   = ['dev-host']
contributing_workers = ['dev-host#25264']
```
`loosely_coupled_subprocess_job`:
```text
succeeded steps: 7
  generate_work_units, report_execution_target,
  dispatch_external_work_subprocess[unit_0..unit_3], summarise_results
units                = 4
contributing_hosts   = ['dev-host']
contributing_workers = ['dev-host#13532', 'dev-host#27116',
                        'dev-host#32984', 'dev-host#31224']
```
Read the two blocks together. The step lists are the same shape, which is the
point of section 5.3 — the graphs differ by one node. `contributing_workers` is
what separates them: one entry means the work happened inside the Dagster
process; four means four external processes did it and reported back over the
pipes channel. `contributing_hosts` stays at one because a laptop is one machine;
on a cluster it is what distinguishes `k8s_job_executor` from
`multiprocess_executor`.
`launcher_namespace` reads `<not-in-kubernetes>` in both. In a cluster run it
holds the namespace the run pod landed in, which is the placement half of the
execution target.
---
## 6. Upstream documentation
- Dagster — [run launchers](https://docs.dagster.io/deployment/execution/run-launchers)
- Dagster — [executors](https://docs.dagster.io/deployment/execution/executors)
- Dagster — [customizing your Kubernetes deployment](https://docs.dagster.io/deployment/oss/deployment-options/kubernetes/customizing-your-deployment)
- Dagster — [Dagster Pipes](https://docs.dagster.io/guides/build/external-pipelines)
- Dagster — [Dagster Pipes with Kubernetes](https://docs.dagster.io/guides/build/external-pipelines/kubernetes-pipeline)
- Kubernetes — [Jobs](https://kubernetes.io/docs/concepts/workloads/controllers/job/)
- Kubernetes — [network policies](https://kubernetes.io/docs/concepts/services-networking/network-policies/)
- Kubernetes — [RBAC authorization](https://kubernetes.io/docs/reference/access-authn-authz/rbac/)
## 7. Related Simpl documentation
- [dagster/documents/user-manual/Orchestration.md](../../../dagster/documents/user-manual/Orchestration.md) — workflow development lifecycle
- [dagster/documents/installation-guide/Installation Guide.md](../../../dagster/documents/installation-guide/Installation%20Guide.md) — full run launcher parameter reference
- [dagster/documents/user-manual/Observability.md](../../../dagster/documents/user-manual/Observability.md) — logging and monitoring
## 8. Outstanding work
Tracked under SIMPL-30451.
| Item | AC | Status |
|---|---|---|
| Execution-target choices, prerequisites, workflow-level documentation | AC1 | Complete |
| Comparison, environment-fit indicators, trade-offs | AC2 | Complete |
| Readiness checklist and evidence mapping | AC3 | Complete; K8s pipes checks not cluster-verified |
| Misconfiguration symptoms and corrections | AC3 | Complete; K8s pipes symptoms not cluster-verified |
| Configuration constructs and persistence | AC4 | Complete |
| Before-and-after example, tightly coupled | AC4 | Complete |
| Before-and-after example, switching to loosely coupled | AC4 | Complete |
| Runnable tightly coupled reference implementations | AC4 | Complete |
| Runnable loosely coupled reference implementation | AC4 | Complete (subprocess verified, K8s not cluster-run) |
| Both images build; payload image passes the isolation check (L11) | Tech details | Complete |
| Locally runnable jobs launched from the Dagster UI, evidence recorded in section 5.6 | Tech details | Complete |
| Payload image published to a container registry | Tech details | Complete on the sandbox Gitea registry; the GitLab registry still pending |
| GitLab pipeline builds both images | Tech details | **Pending** — the shared `ds.gitlab-ci.yml` template builds one image from the root Dockerfile |
| End-to-end run of `loosely_coupled_k8s_job` on a cluster | Tech details | **Pending** |
| Screenshots of UI surfaces | Tech details | **Pending** — needs a deployed platform instance, not a local dev server |
| Platform architecture document update | Tech details | **Pending** |
						
						
						
						
@@ -0,0 +1,122 @@
# Distributed Execution: Readiness Checklist
Pre-run validation for a workflow's execution target. Companion to the
[user guide](distributed-execution-guide.md); this document covers AC3.
Run these checks **before the first execution** of a workflow, and again after any
change to the run launcher, the executor, the target namespace or the code
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, L10 and L11, which exercise the payload contract,
> the message-parsing path, per-unit dispatch and image-level isolation, are
> verified locally.
---
## 1. Common checks — both setups
| # | Check | How to verify | Expected evidence |
|---|---|---|---|
| C1 | Code location loads | Dagster UI → **Deployment** → **Code locations** | Location `distributed-execution` shows status *Loaded*, with a recent load timestamp and no error banner |
| C2 | Jobs are registered | Dagster UI → **Jobs** | The jobs listed in the guide's section 5.5 appear under the code location |
| C3 | Image tag matches the intended release | `kubectl -n dagster get deploy -l dagster/code-location=distributed-execution -o jsonpath='{.items[*].spec.template.spec.containers[*].image}'` | Tag equals the version in `pipeline.variables.sh`; never `latest` |
| C4 | Image architecture matches the nodes | `docker manifest inspect <image>` | Includes `linux/amd64`; a manifest with only `linux/arm64` produces `no match for platform` at pull time |
| C5 | Run launcher type is as intended | `kubectl -n dagster get cm dagster-instance -o yaml` | `run_launcher` block shows `K8sRunLauncher` and the expected `job_namespace` |
| C6 | Target namespace exists and is schedulable | `kubectl get ns <namespace>` and `kubectl -n <namespace> get resourcequota` | Namespace is `Active`; remaining quota exceeds the job's aggregate requests |
## 2. Tightly coupled checks
| # | Check | How to verify | Expected evidence |
|---|---|---|---|
| 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 <run-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 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=<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` |
## 3. Loosely coupled checks
| # | Check | How to verify | Expected evidence |
|---|---|---|---|
| L1 | Payload contract is intact | `uv run pytest tests/test_loosely_coupled.py` | `test_payload_does_not_import_dagster` passes — the payload imports `dagster_pipes` only |
| L2 | Pipes round trip works | Launch `loosely_coupled_subprocess_job` | Run reaches `SUCCESS`; run logs contain the payload's `External payload started on …` line, proving `pipes.log` crossed the channel |
| L3 | Silence is treated as failure | Same test module | `test_silent_message_path_is_treated_as_failure` passes — an empty message list raises rather than yielding an empty result |
| L4 | Payload image is pullable by the target cluster | `kubectl -n <payload-ns> run pull-probe --image=<payload-image> --restart=Never --command -- true` | Pod reaches `Completed`; no `ImagePullBackOff`. The sandbox Gitea registry rejects anonymous pulls, so the payload namespace needs its own pull secret — see `yaml/sandbox/values-sandbox-gitea.yaml` |
| L5 | Dispatcher can create Jobs in the payload namespace | `kubectl -n <payload-ns> auth can-i create jobs --as=system:serviceaccount:dagster:dagster-dev` | Returns `yes` |
| L6 | Dispatcher can read pod logs — **the message channel** | `kubectl -n <payload-ns> 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 <payload-ns> 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 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 |
| L11 | Payload **image** carries no orchestration dependency | `docker run --rm <payload-image> python -c "import importlib.util; print(importlib.util.find_spec('dagster') is not None)"` | Prints `False`. L1 proves the *source* does not import `dagster`; this proves the shipped image does not contain it either |
Checks L4–L9 require a cluster. L1–L3, L10 and L11 run on a laptop and should
gate every change to the payload or the dispatching op.
---
## 4. Common misconfiguration symptoms
### 4.1 Both setups
| Symptom | Likely cause | Correction |
|---|---|---|
| Code location stuck in *Loading*, then errors | Entry point path in `codeServerArgs` does not match the image layout | Confirm `--python-file` matches `workspace.yaml`; both must be `src/distributed_execution/repository.py` |
| `ImagePullBackOff` with `no match for platform` | Image published for a single non-matching architecture | Rebuild multi-arch with `docker buildx`, and pin a version tag rather than `latest` |
| Run stays in `QUEUED` indefinitely | Run coordinator concurrency limit reached, or no schedulable node | Check `max_concurrent_runs` and tag concurrency limits; check node capacity and resource quota |
| Run fails immediately with a serialisation error | Code location image and Dagster control-plane versions diverge | Align the `dagster` version in `pyproject.toml` with the chart's version and rebuild |
### 4.2 Tightly coupled
| Symptom | Likely cause | Correction |
|---|---|---|
| Run pod starts, then fails with a connection timeout to port 5432 | Execution namespace NetworkPolicy does not permit egress to Postgres | Add an egress rule for the metadata database, or move run pods to an already-approved namespace via `jobNamespace` |
| `env_vars_missing` is non-empty in `report_execution_target` metadata | Env vars are set on the code location deployment but not on the run pod | Add them under `runLauncher.config.k8sRunLauncher.runK8sConfig.containerConfig.env` — code location env is **not** inherited by run pods |
| A literal `vault:...` string appears as a value at runtime | Vault mutating webhook did not process the pod | Verify the `vault.security.banzaicloud.io/*` annotations are on the **run pod** template, not only the code location pod |
| Steps hang in `STARTING` with `k8s_job_executor` | Service account lacks Job create/watch permission | Apply `yaml/tightly-coupled/rbac-step-executor.yaml` and confirm with `kubectl auth can-i` |
| `contributing_hosts` shows one host when `k8s_job_executor` is configured | Run tags or Launchpad config overrode the executor, or the image predates the change | Confirm the code location reloaded after the image bump; check the run's *Config* tab for an `execution:` override |
| Step pods `OOMKilled` under fan-out | Per-step memory limit applied per pod, aggregate exceeded quota | Raise `step_k8s_config` limits or lower step concurrency; the two multiply |
| Postgres refuses connections once fan-out grows | Each step pod is an independent DB client | Reduce step concurrency, raise the Postgres connection limit, or move the fan-out step to a loosely coupled target |
### 4.3 Loosely coupled
| Symptom | Likely cause | Correction |
|---|---|---|
| Op fails with `No pipes messages received from the external payload` | The message path is broken, not the workload | Work through L6 then L4. The payload very likely ran and succeeded; only its reporting was lost |
| Payload pod `Completed`, but Dagster shows no payload log lines | A log shipper is intercepting or truncating stdout | Exclude the payload namespace from the shipper, or switch to an object-storage message reader |
| Op hangs until `pod_wait_timeout` (default 24 h) | Payload Job never scheduled — quota, node selector or image pull | Check `kubectl -n <payload-ns> describe job <name>`; lower `pod_wait_timeout` so the failure surfaces quickly |
| `403 Forbidden` creating the Job | Dispatcher service account lacks Job create permission | Apply `yaml/loosely-coupled/rbac-pipes-dispatch.yaml` in the **payload** namespace |
| Warning: `Payload could see orchestration runtime credentials` | Payload pod inherited run-pod env or a Vault annotation | Remove the inherited env; the payload should receive only what `extras` and explicit `env` pass it |
| Payload exits non-zero but the run reports success | Exit status not being checked, or messages read before failure | Confirm the dispatching op returns through `_result_from_pipes`; do not swallow `PipesClientCompletedInvocation` errors |
| Payload receives no `units` | `extras` key mismatch between dispatcher and `pipes.get_extra()` | Both sides must use the same key; a typo yields a `KeyError` inside the payload |
| Run cancelled in the UI, payload pod keeps running | Cancellation is not propagated to dispatched workloads automatically | `delete_pod_on_completion` handles the normal path; for cancellation, verify orphaned Jobs and add a cleanup sensor |
> **NOT CLUSTER-VERIFIED.** Rows referencing Kubernetes behaviour follow from the
> implemented reference and the dagster-k8s API, but have not been observed on a
> Simpl cluster. Confirm and amend after the first cluster run.
---
## 5. Evidence retention
For each workflow's first execution, attach to the workflow's repository or
change record:
1. The run ID and its final status.
2. The `report_execution_target` output metadata block (pod identity, namespace,
   env var presence).
3. The `summarise_results` metadata block (`contributing_hosts` and
   `contributing_workers`), which proves which execution target was actually
   used and that the fan-out reached it.
4. For `k8s_job_executor`, the output of `kubectl get jobs -l dagster/run-id=<run-id>`.
5. For the loosely coupled target, the payload image digest and the run log line
   emitted by `pipes.log` — together they prove which payload version ran and
   that the message channel was open.
Items 2 and 3 together are sufficient to demonstrate that the configured
execution target is the one that ran — which is the point of the checklist.
						
						
						
						
@@ -0,0 +1,30 @@
# External payload image for the loosely coupled execution target.
#
# Built and versioned independently of the code location image. It contains no
# workflow code and no `dagster` package - only the pipes protocol client.
#
#   docker build -f payload/Dockerfile -t distributed-execution-payload:0.1.0 payload/
FROM python:3.12-slim-bookworm
RUN python -m pip install --no-cache-dir --upgrade "pip==26.1.2"
WORKDIR /app
RUN addgroup --gid 1000 appgroup && \
    adduser --uid 1000 --gid 1000 --disabled-password --gecos "" appuser
RUN apt-get update && apt-get upgrade -y \
    && apt-get clean \
    && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/*
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY work.py .
RUN chown -R appuser:appgroup /app
USER appuser
CMD ["python", "/app/work.py"]
						
						
						
						
@@ -0,0 +1,3 @@
# The payload's entire dependency surface. Note the absence of `dagster`:
# dependency isolation is the point of the loosely coupled pattern.
dagster-pipes>=1.13.0
						
						
						
						
@@ -0,0 +1,59 @@
"""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()
						
						
						
						
@@ -0,0 +1,40 @@
[project]
name = "distributed_execution"
version = "0.1.0"
description = "Reference implementations and user guide for Dagster distributed execution targets and integration patterns"
requires-python = ">=3.12"
dependencies = [
    "dagster",
    "dagster-k8s>=0.27.16",
    "dagster-postgres>=0.27.16",
    "dagster-webserver>=1.11.16",
    "pip>=26.1.2",
    "pydantic>=2.0",
]
[dependency-groups]
dev = [
    "pytest>=8.4.2",
    "pytest-cov>=7.0.0",
]
[build-system]
requires = ["setuptools", "wheel"]
build-backend = "setuptools.build_meta"
[tool.setuptools]
package-dir = {"" = "src"}
[tool.setuptools.packages.find]
where = ["src"]
[tool.setuptools.package-data]
distributed_execution = ["*.py"]
[tool.black]
line-length = 120
target-version = ['py312']
[tool.pytest.ini_options]
pythonpath = ["src"]
testpaths = ["tests"]
						
						
						
						
@@ -0,0 +1,6 @@
dagster
dagster-k8s>=0.27.16
dagster-postgres>=0.27.16
dagster-webserver>=1.11.16
pip>=26.1.2
pydantic>=2.0
						
						
						
						
@@ -0,0 +1,3 @@
"""Reference implementations for Dagster distributed execution targets."""
__version__ = "0.1.0"
						
						
						
						
@@ -0,0 +1 @@
"""Loosely coupled (dagster-pipes) execution-target reference implementations."""
						
						
						
						
@@ -0,0 +1,169 @@
"""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_unit`` 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,
    in_process_executor,
    multiprocess_executor,
    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/distributed-execution/payload:0.0.1",
)
PAYLOAD_NAMESPACE = os.environ.get("PIPES_PAYLOAD_NAMESPACE", "dagster")
COMMON_TAGS = {
    "execution_target": "loosely_coupled",
    "business_operation": "DISTRIBUTED_EXECUTION_REFERENCE",
}
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
    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),
        )
    row = payload["results"][0]
    context.log.info("Unit %s computed by external worker %s", row["unit"], row["worker"])
    return row
@op(
    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,
    unit: int,
    pipes_subprocess_client: PipesSubprocessClient,
) -> dict:
    completed = pipes_subprocess_client.run(
        context=context,
        command=[sys.executable, PAYLOAD_SCRIPT],
        extras={"units": [unit]},
    )
    return _result_from_pipes(context, completed)
@op(
    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,
    unit: int,
    pipes_k8s_client: PipesK8sClient,
) -> dict:
    completed = pipes_k8s_client.run(
        context=context,
        image=PAYLOAD_IMAGE,
        command=["python", "/app/work.py"],
        namespace=PAYLOAD_NAMESPACE,
        extras={"units": [unit]},
        base_pod_meta={
            "labels": {
                "app.kubernetes.io/name": "distributed-execution-payload",
                "dagster/execution-target": "loosely-coupled",
            }
        },
    )
    return _result_from_pipes(context, completed)
@graph
def loosely_coupled_subprocess_reference():
    target_report = report_execution_target()
    units = generate_work_units()
    results = units.map(dispatch_external_work_subprocess).collect()
    return summarise_results(results, target_report)
@graph
def loosely_coupled_k8s_reference():
    target_report = report_execution_target()
    units = generate_work_units()
    results = units.map(dispatch_external_work_k8s).collect()
    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."
    ),
    # 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"},
)
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."
    ),
    executor_def=multiprocess_executor.configured({"max_concurrent": 2}),
    resource_defs={"pipes_k8s_client": PipesK8sClient()},
    tags={**COMMON_TAGS, "transport": "k8s_pod_logs"},
)
LOOSELY_COUPLED_JOBS = [
    loosely_coupled_subprocess_job,
    loosely_coupled_k8s_job,
]
						
						
						
						
@@ -0,0 +1,96 @@
"""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, DynamicOut, DynamicOutput, 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="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. 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()
    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 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"]),
        }
    )
    context.log.info("Summary: %s", summary)
    return summary
						
						
						
						
@@ -0,0 +1,64 @@
"""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."""
    hostname = socket.gethostname()
    return {
        "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", "<not-in-kubernetes>"),
        "run_id": os.environ.get("DAGSTER_RUN_ID", "<unset>"),
        "image": os.environ.get("DAGSTER_K8S_PIPELINE_RUN_IMAGE", "<unset>"),
    }
						
						
						
						
@@ -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],
)
						
						
						
						
@@ -0,0 +1 @@
"""Tightly coupled execution-target reference implementations."""
						
						
						
						
@@ -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_unit,
    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 = units.map(process_work_unit).collect()
    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,
]
						
						
						
						
@@ -0,0 +1 @@
"""Tests for the distributed-execution reference implementations."""
						
						
						
						
@@ -0,0 +1,95 @@
"""Verifies the loosely coupled pipes reference end to end via the subprocess transport."""
import pytest
from dagster import DagsterInvariantViolationError, Failure
from distributed_execution.loosely_coupled.jobs import (
    LOOSELY_COUPLED_JOBS,
    PAYLOAD_SCRIPT,
    loosely_coupled_k8s_job,
    loosely_coupled_subprocess_job,
)
from distributed_execution.repository import defs
def test_payload_script_exists():
    from pathlib import Path
    assert Path(PAYLOAD_SCRIPT).is_file()
def test_payload_does_not_import_dagster():
    from pathlib import Path
    source = Path(PAYLOAD_SCRIPT).read_text(encoding="utf-8")
    assert "from dagster_pipes import" in source
    assert "import dagster\n" not in source
    assert "from dagster import" not in source
def test_both_transports_are_registered():
    names = {job.name for job in defs.jobs}
    assert {"loosely_coupled_subprocess_job", "loosely_coupled_k8s_job"} <= names
    assert len(LOOSELY_COUPLED_JOBS) == 2
@pytest.mark.parametrize(
    ("job", "transport"),
    [
        (loosely_coupled_subprocess_job, "subprocess"),
        (loosely_coupled_k8s_job, "k8s_pod_logs"),
    ],
)
def test_jobs_declare_their_execution_target(job, transport):
    assert job.tags["execution_target"] == "loosely_coupled"
    assert job.tags["transport"] == transport
def test_subprocess_job_runs_end_to_end():
    result = loosely_coupled_subprocess_job.execute_in_process()
    assert result.success
    summary = result.output_for_node("summarise_results")
    # Same shape as the tightly coupled reference: only the middle node changed.
    assert summary["units"] == 4
    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()
    mapped = result.output_for_node("dispatch_external_work_subprocess")
    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 _result_from_pipes
    class _Silent:
        def get_custom_messages(self):
            return []
    class _Ctx:
        class log:  # noqa: N801
            @staticmethod
            def warning(*_args, **_kwargs): ...
            @staticmethod
            def info(*_args, **_kwargs): ...
    with pytest.raises(Failure, match="No pipes messages received"):
        _result_from_pipes(_Ctx(), _Silent())
						
						
						
						
@@ -0,0 +1,67 @@
"""Verifies that execution-target linkage is what the guide claims it is."""
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 (
    tightly_coupled_in_process_job,
    tightly_coupled_k8s_job,
    tightly_coupled_local_job,
)
EXPECTED_JOBS = {
    "tightly_coupled_in_process_job",
    "tightly_coupled_local_job",
    "tightly_coupled_k8s_job",
}
def test_repository_registers_tightly_coupled_jobs():
    names = {job.name for job in defs.jobs}
    assert EXPECTED_JOBS <= names
@pytest.mark.parametrize(
    ("job", "expected_executor"),
    [
        (tightly_coupled_in_process_job, "in_process"),
        (tightly_coupled_local_job, "multiprocess"),
        (tightly_coupled_k8s_job, "k8s_job"),
    ],
)
def test_each_job_declares_its_execution_target(job, expected_executor):
    assert job.tags["execution_target"] == "tightly_coupled"
    assert job.tags["executor"] == expected_executor
def test_in_process_job_runs_end_to_end():
    result = tightly_coupled_in_process_job.execute_in_process()
    assert result.success
    summary = result.output_for_node("summarise_results")
    assert summary["units"] == 4
    assert summary["total"] == 14  # 0 + 1 + 4 + 9
    # 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():
    result = check_env_vars(("DEFINITELY_NOT_SET_12345",))
    assert result["passed"] is False
    assert result["missing"] == ["DEFINITELY_NOT_SET_12345"]
def test_pod_identity_falls_back_outside_kubernetes(monkeypatch):
    monkeypatch.delenv("DAGSTER_K8S_PIPELINE_RUN_NAMESPACE", raising=False)
    assert describe_pod_identity()["namespace"] == "<not-in-kubernetes>"
						
						
						
						
@@ -0,0 +1,3 @@
load_from:
  - python_file:
      relative_path: src/distributed_execution/repository.py
						
						
						
						
@@ -0,0 +1,41 @@
# Kubernetes RBAC for the loosely coupled execution target.
#
# Granted to the service account running the Dagster RUN pods - the pod that
# calls PipesK8sClient.run() is the one that creates and watches the payload Job.
#
# Note what is NOT here: the payload itself needs no RBAC, no database
# credentials and no Vault role. Its only channel back to the control plane is
# the pod log stream, which the dispatching pod reads.
#
# If the payload runs in a different namespace from the dispatcher, apply this
# Role in the PAYLOAD namespace and keep the RoleBinding subject pointing at the
# dispatcher's service account.
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: distributed-execution-pipes-dispatcher
  namespace: dagster
rules:
  - apiGroups: ["batch"]
    resources: ["jobs", "jobs/status"]
    verbs: ["create", "get", "list", "watch", "delete"]
  # pods/log is the message channel for PipesK8sPodLogsMessageReader.
  # Without it the payload runs but reports nothing.
  - apiGroups: [""]
    resources: ["pods", "pods/log", "pods/status"]
    verbs: ["get", "list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: distributed-execution-pipes-dispatcher
  namespace: dagster
subjects:
  - kind: ServiceAccount
    name: dagster-dev
    namespace: dagster
roleRef:
  kind: Role
  name: distributed-execution-pipes-dispatcher
  apiGroup: rbac.authorization.k8s.io
						
						
						
						
@@ -0,0 +1,47 @@
# Loosely coupled execution target - code location overrides.
#
# Merge into the dagster-user-deployments entry for this code location. These
# values point the pipes client at the payload image and target namespace.
#
# Contrast with yaml/tightly-coupled/values-run-launcher.yaml: there, every
# execution pod needed S3 and Vault credentials. Here the payload needs none -
# only the dispatching pod does.
#
# VERSION LOCK. The payload tag below must equal the code location image tag in
# yaml/values-dagster-distributed-execution.yaml. The two images share a commit:
# the dispatching op and payload/work.py agree on the `units` extras key and on
# the custom-message shape, and nothing at runtime checks that agreement. A
# mismatched pair surfaces as "no pipes messages received", which reads like a
# broken message channel rather than a version skew.
#
# For values that work today, see yaml/sandbox/values-sandbox-gitea.yaml. The
# shared GitLab pipeline (ds.gitlab-ci.yml) builds exactly one image per project -
# $CI_REGISTRY_IMAGE from the root Dockerfile - so the payload reference below
# resolves only once that pipeline learns to build a second image. The Gitea
# workflow in .gitea/workflows/docker-publish.yml already builds both.
dagster:
  dagster-user-deployments:
    deployments:
      - name: distributed-execution
        envSecrets: []
        env:
          - name: PIPES_PAYLOAD_IMAGE
            # DO NOT CHANGE the tag - replaced automatically by the pipeline before releasing
            value: code.europa.eu:4567/simpl/simpl-open/data/supporting-data-services/distributed-execution/distributed-execution/payload:0.0.0
          - name: PIPES_PAYLOAD_NAMESPACE
            value: dagster
  # The dispatching RUN pod creates the payload Job, so the image reference and
  # namespace must also be visible to run pods, not just the code server.
  runLauncher:
    config:
      k8sRunLauncher:
        runK8sConfig:
          containerConfig:
            env:
              - name: PIPES_PAYLOAD_IMAGE
                # DO NOT CHANGE the tag - replaced automatically by the pipeline before releasing
                value: code.europa.eu:4567/simpl/simpl-open/data/supporting-data-services/distributed-execution/distributed-execution/payload:0.0.0
              - name: PIPES_PAYLOAD_NAMESPACE
                value: dagster
						
						
						
						
@@ -0,0 +1,57 @@
# Sandbox deployment values - Gitea registry.
#
# NOT CLUSTER-VERIFIED. These values are written against images that exist and a
# registry whose auth behaviour was checked, but they have not been applied to a
# Simpl cluster. Confirm and amend after the first deployment.
#
# Use these instead of the code.europa.eu references in
# yaml/values-dagster-distributed-execution.yaml and
# yaml/loosely-coupled/values-pipes-payload.yaml when deploying to the
# dataprovider01 sandbox. Both images are published by
# .gitea/workflows/docker-publish.yml.
#
# The tag below is a short commit SHA and is the same for both images. That is
# the version lock: the code location and the payload it dispatches must come
# from one commit. Bump both together or not at all.
#
# The sandbox Gitea registry requires authentication - an anonymous manifest GET
# returns 401 - so a pull secret is required in every namespace that pulls either
# image. Create it with a Gitea access token that has read:package scope:
#
#   kubectl -n <namespace> create secret docker-registry gitea-registry \
#     --docker-server=gitea.dataprovider01.sandbox-cat-dat.simpl-europe.eu \
#     --docker-username=<gitea-user> \
#     --docker-password=<gitea-token>
#
# The payload namespace needs it too: PipesK8sClient creates that Job, and a
# missing pull secret there leaves the op waiting on a pod that never starts,
# which surfaces as pod_wait_timeout rather than as an image error.
dagster:
  dagster-user-deployments:
    deployments:
      - name: distributed-execution
        image:
          repository: gitea.dataprovider01.sandbox-cat-dat.simpl-europe.eu/j.r/distributed-execution
          tag: 5122da4691f9
          pullPolicy: IfNotPresent
        imagePullSecrets:
          - name: gitea-registry
        env:
          - name: PIPES_PAYLOAD_IMAGE
            value: gitea.dataprovider01.sandbox-cat-dat.simpl-europe.eu/j.r/distributed-execution-payload:5122da4691f9
          - name: PIPES_PAYLOAD_NAMESPACE
            value: dagster
  runLauncher:
    config:
      k8sRunLauncher:
        imagePullSecrets:
          - name: gitea-registry
        runK8sConfig:
          containerConfig:
            env:
              - name: PIPES_PAYLOAD_IMAGE
                value: gitea.dataprovider01.sandbox-cat-dat.simpl-europe.eu/j.r/distributed-execution-payload:5122da4691f9
              - name: PIPES_PAYLOAD_NAMESPACE
                value: dagster
						
						
						
						
@@ -0,0 +1,34 @@
# Kubernetes RBAC required by the k8s_job_executor.
#
# The in_process and multiprocess executors need NONE of this - the run worker
# does all the work itself. Only tightly_coupled_k8s_job, which creates one
# Kubernetes Job per step, needs permission to manage Jobs and read their pods.
#
# Bind to the service account used by the Dagster run pods (dagster.serviceAccount.name).
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: distributed-execution-step-runner
  namespace: dagster
rules:
  - apiGroups: ["batch"]
    resources: ["jobs", "jobs/status"]
    verbs: ["create", "get", "list", "watch", "delete"]
  - apiGroups: [""]
    resources: ["pods", "pods/log", "pods/status"]
    verbs: ["get", "list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: distributed-execution-step-runner
  namespace: dagster
subjects:
  - kind: ServiceAccount
    name: dagster-dev
    namespace: dagster
roleRef:
  kind: Role
  name: distributed-execution-step-runner
  apiGroup: rbac.authorization.k8s.io
						
						
						
						
@@ -0,0 +1,54 @@
# Tightly coupled execution target - instance-level run launcher.
#
# Merge into the Dagster chart values. This is the half of the execution-target
# binding that lives OUTSIDE workflow code: it decides where the run worker for
# an entire run is placed, and what that pod can reach.
#
# Companion to src/distributed_execution/tightly_coupled/jobs.py, which decides
# how the steps INSIDE that run worker execute.
dagster:
  runLauncher:
    type: K8sRunLauncher
    config:
      k8sRunLauncher:
        # Namespace the run pods land in. Must be a namespace whose NetworkPolicy
        # permits egress to Postgres, object storage and Vault - see AC1
        # prerequisites in the user guide.
        jobNamespace: dagster
        # Surfaces step failures as pod failures so kubectl and Dagster agree.
        failPodOnRunFailure: true
        runK8sConfig:
          podTemplateSpecMetadata:
            annotations:
              vault.security.banzaicloud.io/vault-inject: "true"
              vault.security.banzaicloud.io/vault-addr: https://secrets.common01.dev.simpl-europe.eu
              vault.security.banzaicloud.io/vault-role: dev-role
              vault.security.banzaicloud.io/vault-skip-verify: "true"
              vault.security.banzaicloud.io/vault-path: kubernetes
          containerConfig:
            resources:
              requests:
                cpu: 100m
                memory: 128Mi
              limits:
                cpu: 500m
                memory: 512Mi
            # Everything below is what "tightly coupled" costs you: each run pod
            # needs credentials for, and network reachability to, the full
            # orchestration runtime. A loosely coupled target needs none of it.
            env:
              - name: DAGSTER_TELEMETRY_ENABLED
                value: "false"
              - name: TOKEN
                value: "vault:dev/data/dagster/dagster-workflow-vault-secret#VAULT_ACCESS_TOKEN"
              - name: S3_ENDPOINT_URL
                value: "https://s3.dev.simpl-europe.eu"
              - name: S3_ACCESS_KEY
                value: "vault:dev/data/dev-orchestration-platform#S3_ACCESS_KEY"
              - name: S3_SECRET_KEY
                value: "vault:dev/data/dev-orchestration-platform#S3_SECRET_KEY"
						
						
						
						
@@ -0,0 +1,56 @@
dagster:
  # ===========================================================================
  # CODE LOCATION - distributed-execution
  # ===========================================================================
  # Deploys this repository as a Dagster code location so the reference jobs
  # appear in the UI alongside production workflows.
  # MANAGED BY GITLAB CI/CD - DO NOT CHANGE MANUALLY
  dagster-user-deployments:
    deployments:
      - name: distributed-execution
        location_name: distributed-execution
        enabled: true
        enableSubchart: true
        codeServerArgs:
          - "--host"
          - "0.0.0.0"
          - "--port"
          - "4000"
          - "--python-file"
          - "src/distributed_execution/repository.py"
        envFrom:
          - secretRef:
              name: distributed-execution-dagster-db-secret
          - configMapRef:
              name: distributed-execution-dagster-db-config
        image:
          pullPolicy: IfNotPresent
          # $CI_REGISTRY_IMAGE for this project - the pipeline pushes here, so anything
          # else is a reference to an image nobody builds.
          repository: code.europa.eu:4567/simpl/simpl-open/data/supporting-data-services/distributed-execution/distributed-execution
          tag: 0.0.0 # DO NOT CHANGE - replaced automatically by the pipeline before releasing
        port: 4000
        resources:
          requests:
            cpu: 100m
            memory: 256Mi
          limits:
            cpu: 500m
            memory: 512Mi
        livenessProbe:
          enabled: true
          periodSeconds: 30
          timeoutSeconds: 5
          failureThreshold: 3
        readinessProbe:
          enabled: true
          periodSeconds: 10
          timeoutSeconds: 3
          failureThreshold: 3
		`@@ -0,0 +1,3 @@`
							`"""Reference implementations for Dagster distributed execution targets."""`

							`__version__ = "0.1.0"`
		`@@ -0,0 +1 @@`
							`"""Loosely coupled (dagster-pipes) execution-target reference implementations."""`
		`@@ -0,0 +1 @@`
							`"""Tightly coupled execution-target reference implementations."""`
		`@@ -0,0 +1 @@`
							`"""Tests for the distributed-execution reference implementations."""`