Reference Airflow and Dagster pipelines, tested against a server (INT-10, #1715)

.github/workflows/ci.yml

@@ -897,6 +897,78 @@ jobs:
             ${{ runner.temp }}/gpml-conformance/suite/results/summary.json
           if-no-files-found: warn
 
+  # INT-10 asks for an orchestration integration that is documented AND tested:
+  # one reference pipeline per tool, run against a server. This runs both --
+  # the Airflow DAG through `dag.test()` and the Dagster assets through
+  # `materialize` -- against the release binary `test` built, each test on its
+  # own server with a fresh data directory (examples/orchestration/tests/).
+  #
+  # A reference pipeline that is not run rots exactly like documentation does,
+  # only later: it keeps importing cleanly long after the HTTP API it calls has
+  # moved, and the first person to find out is the one who copied it.
+  #
+  # Not in `ci-ok`'s `needs`, for the reason `gpml` is not: making a new suite
+  # block every merge -- here, one that installs Airflow from PyPI -- is a policy
+  # decision, separate from adding the suite. tests/ci_gate_enumeration.rs
+  # records the exemption so it cannot be forgotten.
+  orchestration:
+    name: orchestration (Airflow and Dagster reference pipelines)
+    needs: [changes, test]
+    if: needs.changes.outputs.code == 'true'
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with: {python-version: "3.12"}
+      - uses: dtolnay/rust-toolchain@stable
+      - uses: Swatinem/rust-cache@v2
+
+      # The binary `test` already built. A missing artifact is not a failure:
+      # the next step builds the engine instead, as the gpml job does.
+      - name: Discard any cached release executable
+        run: rm -f target/release/samyama
+
+      - name: Take the release binary from the test job
+        id: orchestration_release_binary
+        continue-on-error: true
+        uses: actions/download-artifact@v4
+        with:
+          name: samyama-release
+          path: target/release
+
+      - name: Use it if it arrived, otherwise build it
+        run: |
+          set -euo pipefail
+          if [ "${{ steps.orchestration_release_binary.outcome }}" = "success" ] && [ -f target/release/samyama ]; then
+            # An artifact download does not preserve the executable bit.
+            chmod +x target/release/samyama
+            echo "using the binary built by the test job"
+          else
+            echo "::notice::no release artifact; building the engine"
+            rm -f target/release/samyama
+            cargo build --release --bin samyama --locked
+          fi
+
+      - name: Install Airflow and Dagster at the tested versions
+        working-directory: examples/orchestration
+        # Airflow's constraints file is per Airflow version and per Python
+        # version; both are read rather than restated, so bumping the pin in
+        # requirements.txt cannot leave this step installing against the old
+        # constraints.
+        run: |
+          set -euo pipefail
+          airflow=$(sed -n 's/^apache-airflow==//p' requirements.txt)
+          py=$(python -c 'import sys; print(f"{sys.version_info[0]}.{sys.version_info[1]}")')
+          pip install -r requirements.txt \
+            -c "https://raw.githubusercontent.com/apache/airflow/constraints-${airflow}/constraints-${py}.txt"
+
+      - name: Run both pipelines against the engine
+        working-directory: examples/orchestration
+        env:
+          SAMYAMA_BIN: ${{ github.workspace }}/target/release/samyama
+        run: python -m pytest -v tests/
+
   # API-09: the SDKs users already have keep working against this build.
   #
   # docs/VERSIONING.md promises that the previous minor's SDKs work against the

docs/ORCHESTRATION-AIRFLOW.md

@@ -0,0 +1,128 @@
+# Loading Samyama from Apache Airflow
+
+How to run the reference DAG,
+[`examples/orchestration/samyama_airflow_dag.py`](../examples/orchestration/samyama_airflow_dag.py),
+against your server and your files. What the pipeline does, the guarantees it
+makes and what it does not do are in [ORCHESTRATION.md](./ORCHESTRATION.md).
+
+Tested with Airflow 3.3.2 on Python 3.12. Airflow 2 is not supported: the DAG
+uses Airflow 3's `airflow.sdk`.
+
+---
+
+## 1. A credential for the pipeline
+
+Skip this if your server runs without `--auth-file`.
+
+```bash
+samyama auth-token pipeline
+```
+
+prints a line for the credential file and, once, the token. Add the line to
+the file the server reads with `:roles=write` on the end -- a pipeline that
+loads data has no business managing tenants or restoring snapshots:
+
+```text
+pipeline:<sha256 of the token>:roles=write
+```
+
+Keep the token for step 3. The server stores only its digest.
+
+## 2. Install, and put the files in the DAGs folder
+
+```bash
+pip install "apache-airflow==3.3.2" \
+  -c https://raw.githubusercontent.com/apache/airflow/constraints-3.3.2/constraints-3.12.txt
+
+cp examples/orchestration/samyama_airflow_dag.py \
+   examples/orchestration/samyama_pipeline.py \
+   "$AIRFLOW_HOME/dags/"
+cp -r examples/orchestration/sample "$AIRFLOW_HOME/dags/"
+```
+
+Both `.py` files: the DAG imports `samyama_pipeline`, and Airflow puts the DAGs
+folder on `sys.path`. `sample/` is the default input, found next to
+`samyama_pipeline.py`; leave it out if every run will name its own files, and a
+run that does not fails at once with `no such file`. Nothing else is needed --
+`samyama_pipeline` is standard library only, and there is no Samyama provider
+package to install.
+
+## 3. A Connection to the server
+
+The DAG reads the server from an Airflow Connection, `samyama_default` unless a
+run says otherwise. `schema` is the protocol, the convention Airflow's own HTTP
+hook uses; the token, if any, goes in `password`:
+
+```bash
+airflow connections add samyama_default \
+  --conn-type http \
+  --conn-host graph.internal \
+  --conn-port 8080 \
+  --conn-schema https \
+  --conn-password "$SAMYAMA_TOKEN"
+```
+
+Any secrets backend Airflow supports works the same way, because the DAG asks
+Airflow for the Connection rather than reading a variable of its own. As an
+environment variable:
+
+```bash
+export AIRFLOW_CONN_SAMYAMA_DEFAULT='{"conn_type": "http", "host": "graph.internal", "port": 8080, "schema": "https", "password": "..."}'
+```
+
+Optional: `{"timeout_seconds": 60}` in the Connection's extras sets how long one
+statement may take. The default is 30.
+
+## 4. Run it
+
+Once, without the scheduler, to see it work -- with the bundled sample:
+
+```bash
+airflow dags test samyama_accounts_transfers
+```
+
+With your own files:
+
+```bash
+airflow dags test samyama_accounts_transfers \
+  -c '{"accounts_csv": "/data/accounts.csv", "transfers_csv": "/data/transfers.csv"}'
+```
+
+The same JSON is what you pass to `airflow dags trigger --conf`, or type into
+the *Trigger DAG* form in the UI.
+
+| Param | Default | |
+|---|---|---|
+| `samyama_conn_id` | `samyama_default` | the Connection to use |
+| `accounts_csv` | the bundled sample | columns `id, holder, country, opened` |
+| `transfers_csv` | the bundled sample | columns `id, src, dst, amount, at` |
+| `batch_size` | `500` | rows per statement |
+
+The DAG has `schedule=None`: it runs when triggered. Give it a schedule, or an
+asset to wait on, when you deploy it -- that choice belongs to the deployment,
+not to a reference.
+
+## 5. When a run fails
+
+The failing task's log names the cause. Two kinds:
+
+- **Retried, with backoff, up to three times**: the server could not be
+  reached, timed out, or answered 5xx.
+- **Failed at once, no retry**: the server refused the statement or the token,
+  the input is wrong (the message names the file and line), or validation found
+  the graph disagrees with the files (the message names each check, what it
+  expected and what it found).
+
+A failed run can leave part of a load in the graph. Fix the cause and run it
+again: every write is an upsert keyed on the source id, so the second run
+completes the first rather than duplicating it.
+
+## How this page is tested
+
+`examples/orchestration/tests/test_airflow.py` runs the DAG through
+`dag.test()` against a real server: a full load, a second run that must leave
+the graph unchanged, a file naming an unknown account (one try, nothing
+written), a token taken from the Connection, a missing token (one try), and a
+Connection named by the `samyama_conn_id` param. The commands above were run as
+written against a local server before this page was published; they are not
+re-run by CI.

docs/ORCHESTRATION-DAGSTER.md

@@ -0,0 +1,114 @@
+# Loading Samyama from Dagster
+
+How to run the reference assets,
+[`examples/orchestration/samyama_dagster.py`](../examples/orchestration/samyama_dagster.py),
+against your server and your files. What the pipeline does, the guarantees it
+makes and what it does not do are in [ORCHESTRATION.md](./ORCHESTRATION.md).
+
+Tested with Dagster 1.13.26 on Python 3.12.
+
+---
+
+## 1. A credential for the pipeline
+
+Skip this if your server runs without `--auth-file`.
+
+```bash
+samyama auth-token pipeline
+```
+
+prints a line for the credential file and, once, the token. Add the line to
+the file the server reads with `:roles=write` on the end -- a pipeline that
+loads data has no business managing tenants or restoring snapshots:
+
+```text
+pipeline:<sha256 of the token>:roles=write
+```
+
+## 2. Install, and put the files in your code location
+
+```bash
+pip install "dagster==1.13.26" "dagster-webserver==1.13.26"
+cp examples/orchestration/samyama_dagster.py examples/orchestration/samyama_pipeline.py my_project/
+cp -r examples/orchestration/sample my_project/
+```
+
+Both `.py` files: `samyama_dagster` imports `samyama_pipeline`, which is
+standard library only. `sample/` is the default input, found next to
+`samyama_pipeline.py`; leave it out if every run will name its own files. There
+is no Samyama integration package to install.
+
+## 3. Point it at the server
+
+| Variable | |
+|---|---|
+| `SAMYAMA_URL` | required -- the server's HTTP endpoint, e.g. `https://graph.internal:8080` |
+| `SAMYAMA_TOKEN` | the token from step 1; leave unset for a server without an auth file |
+
+The URL uses `dg.EnvVar`; the optional token is read when the resource creates
+its client on the execution worker, so worker-only secrets are available. The
+token value is not stored with the code location or shown in the UI.
+
+## 4. Run it
+
+```bash
+export SAMYAMA_URL=http://localhost:8080
+cd my_project
+dagster dev -f samyama_dagster.py
+```
+
+and materialize the `samyama` group from the UI, or without the UI:
+
+```bash
+dagster asset materialize -f samyama_dagster.py --select '*'
+```
+
+The source files and batch size are the `source_files` resource. They default
+to the bundled sample; point a run at your own in the launchpad's config:
+
+```yaml
+resources:
+  source_files:
+    config:
+      accounts_csv: /data/accounts.csv
+      transfers_csv: /data/transfers.csv
+      batch_size: 500
+```
+
+| Asset | Depends on | Check |
+|---|---|---|
+| `samyama_schema` | | |
+| `accounts` | `samyama_schema` | `accounts_match_source` |
+| `transfers` | `accounts` | `transfers_match_source` |
+
+The checks are **blocking**: if `accounts_match_source` fails, `transfers` is
+not loaded on top of accounts that do not match their file. Each check's
+metadata records, per condition, how many it expected and how many it found.
+
+No schedule or sensor is defined; add one when you deploy it -- that choice
+belongs to the deployment, not to a reference.
+
+## 5. When a run fails
+
+- **Retried, with exponential backoff, up to three times**: the server could
+  not be reached, timed out, or answered 5xx.
+- **Failed at once, no retry** (`Failure(allow_retries=False)`): the server
+  refused the statement or the token, or the input is wrong -- the message
+  names the file and line.
+- **A check failed**: the graph disagrees with the files. The check's
+  description names each condition that failed.
+
+A failed run can leave part of a load in the graph. Fix the cause and
+materialize again: every write is an upsert keyed on the source id, so the
+second run completes the first rather than duplicating it.
+
+## How this page is tested
+
+`examples/orchestration/tests/test_dagster.py` materializes the assets and
+checks against a real server: a full load with both checks passing, a second
+run that must leave the graph unchanged, a file naming an unknown account (the
+step fails, is not retried, writes nothing), a failing accounts check that must
+stop `transfers` from loading, a token from the resource, a missing token, and
+the shipped definitions reading `SAMYAMA_URL` and `SAMYAMA_TOKEN` from the
+environment. The commands above were run as written against a local server
+before this page was published; they are not re-run by CI.