Dagster Job Run Compliance Gate

This repo's own dagster-trigger-job.yaml blueprint shows how to launch a Dagster job and block until it finishes - and its own "How to extend" section names the two things it deliberately leaves out: a schedule, and branching on the run's status to alert on FAILURE. This blueprint is that extension. It triggers job_name, waits for a terminal state, classifies the result into a three-state risk level, alerts Slack on anything but a clean SUCCESS, and can launch one fresh retry run - leaving the original failed or canceled run exactly as Dagster recorded it.

How it works

1. periodic_job_compliance_check (io.kestra.plugin.core.trigger.Schedule) runs every 6 hours, always forcing auto_remediate: "false" and dry_run: "true" through its own inputs: override regardless of this flow's defaults. Shipped disabled so you can validate a manual run first.
2. trigger_and_wait_job_run (io.kestra.plugin.dagster.TriggerRun, wait: true, allowFailure: true) launches job_name and blocks until it reaches a terminal state, returning status (QUEUED, STARTING, STARTED, SUCCESS, FAILURE, or CANCELED, per the plugin's own documentation).
3. classify_run_outcome (io.kestra.plugin.core.debug.Return) classifies the result as UNREACHABLE (status undefined - the GraphQL call itself failed), AT_RISK (status is anything other than SUCCESS), or HEALTHY.
4. route_by_outcome (io.kestra.plugin.core.flow.Switch) branches on that classification:
   • HEALTHY: log_healthy records the successful run.
   • AT_RISK: a nested route_remediation (Switch on auto_remediate) either considers remediation ("true", gated again by dry_run through check_dry_run) or only logs that remediation is disabled ("false"). When both gates allow it, retrigger_job_run (io.kestra.plugin.dagster.TriggerRun, wait: false) launches a brand new run. alert_at_risk always fires to Slack regardless of which path ran.
   • UNREACHABLE: alert_unreachable pages Slack directly - a broken GraphQL call must never read as "job succeeded".
5. log_audit_complete always runs last, printing the classification regardless of which branch fired.
6. The errors block alerts Slack separately if the flow itself fails outside these handled branches.

What you get

• A scheduled compliance check on a critical Dagster job, not just a one-shot manual trigger.
• A three-way classification (HEALTHY/AT_RISK/UNREACHABLE) so a broken GraphQL connection is never mistaken for a successful run.
• Remediation that only ever launches a brand new run; the original run's own record, logs, and status in Dagster are never altered, cancelled, or deleted.
• A final status log every run, so the execution history doubles as a job-reliability trend for job_name.

Who it's for

• Data engineering teams running Dagster who want a Kestra-native alert the moment a critical job comes back FAILURE or CANCELED, instead of relying on Dagster's own notification setup.
• Platform teams standardizing cross-tool job governance in Kestra, where Dagster is one orchestrator among several being audited the same way.
• Anyone extending this repo's own dagster-trigger-job.yaml - that blueprint's own "How to extend" section names exactly this: a schedule, and a branch on status that alerts on failure.

Why orchestrate this with Kestra

Dagster's own scheduler can re-run a job on a cron, but a single trigger-and-wait call makes no decision about what counts as a compliance breach, does not notify anyone by itself, and has no independent authorization gate before launching a retry. Kestra supplies the schedule, the three-way risk classification as a first-class branch, two independent authorization gates before a new run is launched, and an execution history that shows exactly when a job first came back non-SUCCESS and whether a retry was launched.

Prerequisites

• A running Dagster instance with its GraphQL API reachable from Kestra, and job_name's repository/location known.
• A Slack incoming webhook for alerts.

Local testing: docker run -d --name dagster-compliance-gate -p 3001:3000 dagster/dagster-k8s:latest dagster-webserver -h 0.0.0.0 -p 3000 docker network connect YOUR_KESTRA_NETWORK dagster-compliance-gate (only needed if the Kestra Worker runs in a separate Docker network than this container; set DAGSTER_BASE_URL to http://dagster-compliance-gate:3000/graphql from inside that network, or http://localhost:3001/graphql from the host). A bare webserver container has no code location loaded; deploy a sample job and code location per Dagster's own open-source quickstart before exercising AT_RISK/HEALTHY end to end - this has not been done in this environment.

Secrets

• DAGSTER_BASE_URL: Dagster GraphQL endpoint, matching this plugin's own documented pattern (see dagster-trigger-job.yaml).
• DAGSTER_TOKEN: bearer token used in the Authorization header for Dagster API authentication.
• SLACK_WEBHOOK_URL: Slack incoming webhook used by alert_at_risk, alert_unreachable, and the errors block.
• In Kestra OSS (no Enterprise secrets backend), secrets are supplied as environment variables prefixed SECRET_, base64-encoded, and read back in flows with {{ secret('NAME') }} - for example SECRET_DAGSTER_TOKEN=$(echo -n '<your-dagster-token>' | base64). This keeps credentials out of the flow YAML but, per Kestra's own documentation, offers no encryption at rest or access control beyond the host environment; use the Enterprise secrets backend for stronger guarantees.

Inputs

• job_name (STRING, default example_job): job triggered and audited.
• repository (STRING, no default): repository containing job_name.
• location (STRING, no default): code location containing repository.
• max_wait_duration (STRING, default PT30M): ceiling this flow waits for a terminal state.
• auto_remediate (SELECT: "false", "true"; default "false"): must be "true" for a retry run to even be considered.
• dry_run (SELECT: "true", "false"; default "true"): must be explicitly "false", together with auto_remediate: "true", for a fresh run to actually launch.

Outputs

• outputs.trigger_and_wait_job_run.runId / .status / .startTime / .endTime: the audited run's own result for every attempted run.
• outputs.classify_run_outcome.value: HEALTHY, AT_RISK, or UNREACHABLE for every run.
• outputs.retrigger_job_run.runId: present only when remediation actually launched a new run.

Quick start

1. Deploy a sample Dagster job and code location against the container above (or point this flow at an existing Dagster instance).
2. Add the DAGSTER_BASE_URL, DAGSTER_TOKEN, and SLACK_WEBHOOK_URL secrets.
3. Run the flow manually against a job that succeeds and confirm log_healthy fires.
4. Point job_name at a job that deliberately raises an exception and re-run; confirm alert_at_risk fires with status: FAILURE, and that log_remediation_disabled logs without launching a retry.
5. Run once with auto_remediate: "true" and dry_run: "true" against the same failing job and confirm log_dry_run_remediation describes the retry without launching it.
6. Run once with auto_remediate: "true" and dry_run: "false" and confirm retrigger_job_run launches, visible as a new run in the Dagster UI.
7. Enable periodic_job_compliance_check once you trust the check; it always runs in the safe auto_remediate: "false" / dry_run: "true" mode regardless of what you leave the flow's own defaults set to.

How to extend

• Loop over several job_name/repository/location triples with io.kestra.plugin.core.flow.Loop to audit every critical job in a code location in one run.
• Cap the number of automatic retries by tracking a counter in vars or an external store, so a job that fails repeatedly escalates instead of retrying forever on every scheduled audit.
• Route the remediation branch through a Pause approval task instead of (or alongside) the static auto_remediate input, for a human-reviewed retry on a production job.
• Pass body.tags on both TriggerRun calls (per this plugin's own documented pattern in dagster-trigger-job.yaml) to stamp each audited and retried run with the originating Kestra execution.id, for full bidirectional lineage.

Pitfalls

• This plugin has only one task, TriggerRun. Verified from the plugin's own documentation: there is no separate "get run status" task to audit an already-running or already-finished run without launching a new one - every audit in this blueprint is itself a real trigger, not a read-only check.
• wait: true blocks for up to maxDuration. A job that genuinely needs longer than the default PT30M to finish will be classified UNREACHABLE-adjacent behavior at the Kestra level (the task itself may time out) well before it actually fails in Dagster; raise max_wait_duration to match the job's real expected runtime.
• The retry launches a brand new run; it does not resume or retry the original run in place. Dagster's own run identity means retrigger_job_run.runId is always different from trigger_and_wait_job_run.runId - this blueprint does not attempt to use any Dagster-side "retry from failure" mechanism, since no such task is documented on this plugin.
• allowFailure: true is what lets UNREACHABLE be classified instead of failing the execution outright. A base Task property, not specific to this plugin; without it, a GraphQL connection failure would abort the flow before classify_run_outcome ever runs.
• auto_remediate and dry_run are independent gates, not a single boolean. Both must be "true"/"false" respectively for retrigger_job_run to run; setting only one leaves the other still blocking.
• Switch case keys are quoted strings matching the classifier's exact output. "HEALTHY", "AT_RISK", and "UNREACHABLE" must match classify_run_outcome.value verbatim; the defaults: block exists specifically to catch any value that does not match one of those three cases.

Links

• plugin-dagster documentation
• plugin-dagster TriggerRun task reference
• Kestra secrets (OSS vs Enterprise)