GCP Dataflow Job Failure Sentinel

A Dataflow pipeline that fails does not necessarily page anyone - Dataflow's own console shows the job's state, but nothing watches it on a schedule unless a Cloud Monitoring alert was deliberately wired up for that specific job. This blueprint reads a Dataflow job's currentState through GetJob, Google's own documented 15-value JobState enum, flags JOB_STATE_FAILED or JOB_STATE_CANCELLED, and alerts Slack the moment that happens or the Dataflow API itself cannot be reached.

How it works

1. periodic_job_check (io.kestra.plugin.core.trigger.Schedule) runs every 30 minutes, shipped disabled so you can validate a manual run first.
2. probe_job_status (io.kestra.plugin.gcp.dataflow.GetJob, allowFailure: true) reads job_id's currentState.
3. evaluate_job_health (io.kestra.plugin.core.debug.Return) classifies UNREACHABLE (currentState undefined), AT_RISK (JOB_STATE_FAILED or JOB_STATE_CANCELLED), or HEALTHY (every other documented state, including in-progress ones).
4. route_by_health (io.kestra.plugin.core.flow.Switch) branches on that classification: HEALTHY logs the current state, AT_RISK pages Slack naming the failed state, UNREACHABLE pages Slack that the API could not be reached, and the defaults: block catches any unexpected classification value.
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 direct, Dataflow-native answer to "did this job actually fail", using Google's own documented JobState enum rather than an inferred signal from downstream data freshness.
• A three-way classification (HEALTHY/AT_RISK/UNREACHABLE) so a broken audit (bad credentials, an invalid job ID) is never mistaken for a healthy job.
• A final status log every run, so the execution history doubles as a job-reliability trend for job_id.

Who it's for

• Data engineering teams running critical Dataflow pipelines (streaming or batch) who want a Kestra-native alert the moment one fails or gets cancelled, independent of whatever Cloud Monitoring alerting policy may or may not be configured.
• Platform teams standardizing cross-cloud job governance in Kestra, auditing Dataflow the same way this repo's other blueprints audit AWS Glue or Apache Flink job runs.
• Anyone evaluating this plugin's Dataflow tasks who wants a worked example of GetJob used as a standing health check.

Why orchestrate this with Kestra

GetJob is a single read with no memory of the last check and no notification path of its own. Kestra supplies the schedule, the risk classification as a first-class branch, and an execution history that shows exactly when a job first failed and how long it took anyone to notice.

Prerequisites

• A GCP project with a Dataflow job to audit, and a service account with dataflow.jobs.get permission.
• A Slack incoming webhook for alerts.

Local testing: Google Cloud Dataflow is a fully managed, serverless service with no official self-hosted or local emulator - Dataflow jobs run on Google's own managed workers, not in a container you control. Validate this flow against a real GCP project and a real (even minimal, quick-to-finish) Dataflow job rather than attempting a fully local run - this has not been exercised in this environment.

Secrets

• GCP_PROJECT_ID / GCP_SERVICE_ACCOUNT: GCP credentials used by GetJob, matching this plugin's own documented connection pattern (see this repo's other GCP blueprints, for example gcs-object-encryption-compliance-gate.yaml).
• 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_GCP_SERVICE_ACCOUNT=$(echo -n '<your-service-account-json>' | 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_id (STRING, default "sample-dataflow-job-01"): Dataflow job audited.
• location (STRING, default us-central1): regional endpoint for the job.

Outputs

• outputs.probe_job_status.currentState / .createTime / .currentStateTime / .metrics: the audited job's own fields for every successful call.
• outputs.evaluate_job_health.value: HEALTHY, AT_RISK, or UNREACHABLE for every run.

Quick start

1. Launch a small, quick-to-finish Dataflow job (a Classic or Flex Template) in a test project and note its job_id and location.
2. Add the GCP_PROJECT_ID, GCP_SERVICE_ACCOUNT, and SLACK_WEBHOOK_URL secrets.
3. Run the flow manually while the job is still running or after it completes successfully and confirm log_healthy fires.
4. Point job_id at a job you know failed (an intentionally broken template is an easy way to produce one in a test project) and re-run; confirm alert_at_risk fires naming JOB_STATE_FAILED.
5. Enable periodic_job_check once you trust the check.

How to extend

• Loop over several job_id values with io.kestra.plugin.core.flow.Loop to audit every critical pipeline in a project in one pass.
• Compare currentStateTime against now() to also flag a job stuck in JOB_STATE_RUNNING far longer than expected, a distinct kind of risk from an outright failure - this blueprint deliberately keeps to terminal-state detection only, to avoid date-arithmetic logic that was not verified against a live Dataflow job in this environment.
• Chain this flow directly after a LaunchTemplate/LaunchFlexTemplate task (via a Subflow call passing through the new job's jobId) for same-execution failure detection on a critical pipeline launch, instead of waiting for the next scheduled sweep.
• If your use case needs active remediation on a still-running job, add io.kestra.plugin.gcp.dataflow.CancelJob (verified real, with a drain option for a graceful stop) behind the same double-gated auto_remediate/dry_run pattern this repo's other sentinels use - deliberately not attempted here, since JOB_STATE_FAILED/JOB_STATE_CANCELLED are both already-terminal states with nothing left to cancel.

Pitfalls

• This blueprint does not attempt remediation. JOB_STATE_FAILED and JOB_STATE_CANCELLED are both terminal per Google's own API reference - there is nothing left to cancel or drain on a job already in either state. CancelJob is a real, verified task in this plugin, but it is only meaningful against a job still in a non-terminal state, which is a different audit than the one this blueprint runs; see How to extend.
• JobState has 15 documented values, not just the four most people name. Verified against Google Cloud's own Dataflow REST API reference: JOB_STATE_UNKNOWN, JOB_STATE_STOPPED, JOB_STATE_RUNNING, JOB_STATE_DONE, JOB_STATE_FAILED, JOB_STATE_CANCELLED, JOB_STATE_UPDATED, JOB_STATE_DRAINING, JOB_STATE_DRAINED, JOB_STATE_PENDING, JOB_STATE_CANCELLING, JOB_STATE_QUEUED, JOB_STATE_RESOURCE_CLEANING_UP, JOB_STATE_PAUSING, JOB_STATE_PAUSED. This blueprint only treats JOB_STATE_FAILED/JOB_STATE_CANCELLED as a breach; every transitional and successful state is HEALTHY.
• 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, an invalid jobId or missing IAM permission would abort the flow before evaluate_job_health ever runs.
• Switch case keys are quoted strings matching the classifier's exact output. "HEALTHY", "AT_RISK", and "UNREACHABLE" must match evaluate_job_health.value verbatim; the defaults: block exists specifically to catch any value that does not match one of those three cases.
• Dataflow regional endpoints must match the job's execution region. Querying GetJob against the wrong location returns a 404/not-found from the Google Cloud Dataflow API, which this flow classifies as UNREACHABLE. Verify the job's exact regional location in the GCP Console before configuring the input.

Links

• plugin-gcp Dataflow documentation
• Google Cloud Dataflow JobState reference
• Kestra secrets (OSS vs Enterprise)