DataHub Ingestion Health Gate with Kestra

id: datahub-ingestion-health-gate
namespace: company.team

description: |
  Run a DataHub metadata ingestion recipe on a schedule, classify the run as HEALTHY,
  DEGRADED, FAILED or PROBE_FAILED from the real exitCode and stdErrLineCount reported by
  the DataHub ingestion CLI, alert Slack when the catalog stops receiving fresh metadata,
  and log a final audit summary either way.

inputs:
  - id: source_type
    type: SELECT
    displayName: Metadata source connector type
    description: The DataHub ingestion recipe source type. Maps directly to the
      recipe's source.type field.
    values:
      - "postgres"
      - "mysql"
    defaults: "postgres"

  - id: source_host_port
    type: STRING
    displayName: Source host and port
    description: host:port of the database DataHub should ingest metadata from, for
      example postgres:5432 on the Docker network set in docker_network.
    defaults: "postgres:5432"

  - id: source_database
    type: STRING
    displayName: Source database name
    description: The database (catalog) to ingest metadata from. Maps to the
      recipe's source.config.database field.
    defaults: "datahub_source"

  - id: source_username
    type: STRING
    displayName: Source database username
    description: Username DataHub uses to connect to the source database. The
      password is read from the DATAHUB_SOURCE_PASSWORD secret, never from an
      input.
    defaults: "datahub"

  - id: datahub_gms_server
    type: STRING
    displayName: DataHub GMS server URL
    description: URL of the DataHub GMS endpoint used as the sink when dry_run is
      false. Reachable by container name on docker_network, for example
      http://datahub-gms:8080.
    defaults: "http://datahub-gms:8080"

  - id: docker_network
    type: STRING
    displayName: Docker network for the ingestion container
    description: Docker network the ingestion container joins so it can resolve the
      source database and the DataHub GMS server by container name. Run docker
      network ls to confirm the name.
    defaults: "datahub_network"

  - id: container_image
    type: STRING
    displayName: DataHub ingestion CLI image
    description: Container image that runs the DataHub ingestion CLI. Override to
      pin a specific DataHub CLI version instead of head.
    defaults: "acryldata/datahub-ingestion:head"

  - id: stderr_warning_threshold
    type: INT
    displayName: Stderr line count degraded threshold
    description: Number of stderr lines the ingestion run can produce before it is
      classified DEGRADED instead of HEALTHY, even when exitCode is 0.
    defaults: 5

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: When true, the recipe sink is console, metadata is printed only,
      nothing is written to the catalog. Set to false to ingest into the live
      DataHub GMS server named in datahub_gms_server.
    defaults: "true"

tasks:
  - id: route_ingestion_mode
    type: io.kestra.plugin.core.flow.If
    description: Picks the recipe sink. console prints metadata without writing it,
      datahub-rest writes to the live GMS server.
    condition: "{{ inputs.dry_run }}"
    then:
      - id: probe_console
        type: io.kestra.plugin.datahub.Ingestion
        description: Dry run probe. Ingests metadata and prints it to stdout without
          writing to the catalog.
        allowFailure: true
        containerImage: "{{ inputs.container_image }}"
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          networkMode: "{{ inputs.docker_network }}"
        recipe:
          source:
            type: "{{ inputs.source_type }}"
            config:
              host_port: "{{ inputs.source_host_port }}"
              database: "{{ inputs.source_database }}"
              username: "{{ inputs.source_username }}"
              password: "{{ secret('DATAHUB_SOURCE_PASSWORD') }}"
          sink:
            type: "console"
    else:
      - id: probe_live
        type: io.kestra.plugin.datahub.Ingestion
        description: Live probe. Ingests metadata and writes it to the DataHub GMS server.
        allowFailure: true
        containerImage: "{{ inputs.container_image }}"
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          networkMode: "{{ inputs.docker_network }}"
        recipe:
          source:
            type: "{{ inputs.source_type }}"
            config:
              host_port: "{{ inputs.source_host_port }}"
              database: "{{ inputs.source_database }}"
              username: "{{ inputs.source_username }}"
              password: "{{ secret('DATAHUB_SOURCE_PASSWORD') }}"
          sink:
            type: "datahub-rest"
            config:
              server: "{{ inputs.datahub_gms_server }}"

  - id: classify_ingestion_status
    type: io.kestra.plugin.core.debug.Return
    description: Reads exitCode and stdErrLineCount from whichever probe ran and
      turns them into one status string.
    format: "{{ (outputs.probe_live is not defined and outputs.probe_console is not
      defined) ? 'PROBE_FAILED' : ((outputs.probe_live is defined ?
      outputs.probe_live.exitCode : outputs.probe_console.exitCode) != 0 ?
      'FAILED' : ((outputs.probe_live is defined ?
      outputs.probe_live.stdErrLineCount :
      outputs.probe_console.stdErrLineCount) > inputs.stderr_warning_threshold ?
      'DEGRADED' : 'HEALTHY')) }}"

  - id: route_by_ingestion_status
    type: io.kestra.plugin.core.flow.Switch
    description: Branch A logs a healthy run. Branch B alerts Slack when the run
      failed to start, exited non-zero, or produced too many stderr lines.
    value: "{{ outputs.classify_ingestion_status.value }}"
    cases:
      "PROBE_FAILED":
        - id: alert_probe_failed
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "DataHub ingestion probe for source " ~ inputs.source_type ~ " did not return a result. The ingestion container likely failed to start. Check that " ~ inputs.docker_network ~ " can reach " ~ inputs.source_host_port ~ " and " ~ inputs.datahub_gms_server ~ "."} | toJson }}
      "FAILED":
        - id: alert_failed
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "DataHub ingestion FAILED for source " ~ inputs.source_type ~ " (dry_run=" ~ inputs.dry_run ~ "). exitCode " ~ (outputs.probe_live is defined ? outputs.probe_live.exitCode : outputs.probe_console.exitCode) ~ ". The metadata catalog did not receive this run."} | toJson }}
      "DEGRADED":
        - id: alert_degraded
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "DataHub ingestion DEGRADED for source " ~ inputs.source_type ~ ". stderr line count " ~ (outputs.probe_live is defined ? outputs.probe_live.stdErrLineCount : outputs.probe_console.stdErrLineCount) ~ " exceeded the threshold of " ~ inputs.stderr_warning_threshold ~ ". Metadata may be incomplete."} | toJson }}
      "HEALTHY":
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          message: "DataHub ingestion healthy for source {{ inputs.source_type }}.
            exitCode 0, stderr within the {{ inputs.stderr_warning_threshold }}
            line threshold."

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: Final audit log confirming the gate ran, regardless of the branch
      taken above.
    message: "DataHub ingestion health gate completed. status={{
      outputs.classify_ingestion_status.value }} source={{ inputs.source_type }}
      dry_run={{ inputs.dry_run }} gms={{ inputs.datahub_gms_server }}"

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {{ {"text": "DataHub ingestion health gate flow failed (execution " ~ execution.id ~ "). Check the execution logs."} | toJson }}

triggers:
  - id: six_hourly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled, and always in dry_run mode by default.
    cron: "0 */6 * * *"
    disabled: true
    inputs:
      source_type: "postgres"
      source_host_port: "postgres:5432"
      source_database: "datahub_source"
      source_username: "datahub"
      datahub_gms_server: "http://datahub-gms:8080"
      docker_network: "datahub_network"
      container_image: "acryldata/datahub-ingestion:head"
      stderr_warning_threshold: 5
      dry_run: "true"

A data catalog is only as trustworthy as its last successful ingestion. When a DataHub ingestion recipe silently starts failing, swallowing a bad connector password, a schema change, or a network change, nobody notices until an analyst finds a dataset with no owner, no lineage, and no fresh schema. This blueprint turns a DataHub ingestion run into a Day 2 governance gate: it runs the recipe, reads the exact exitCode and stdErrLineCount the DataHub ingestion CLI reports, classifies the run, and alerts Slack before stale metadata becomes everyone else's problem.

How it works

route_ingestion_mode (io.kestra.plugin.core.flow.If) checks the dry_run input. When true it runs probe_console, when false it runs probe_live. Both are io.kestra.plugin.datahub.Ingestion tasks with allowFailure: true so a failed probe does not stop the flow before it can be classified and alerted on.
- probe_console uses recipe sink type: console, metadata is printed to stdout only, nothing is written to the catalog.
- probe_live uses recipe sink type: datahub-rest with config.server pointed at datahub_gms_server, metadata is written to the live catalog. Both use io.kestra.plugin.scripts.runner.docker.Docker as the taskRunner with networkMode set to docker_network, so the ingestion container can resolve the source database and the GMS server by container name.
classify_ingestion_status (io.kestra.plugin.core.debug.Return) reads exitCode and stdErrLineCount from whichever probe ran (guarded with is defined, since a probe that never started has no output at all) and returns one status string: PROBE_FAILED (no output at all), FAILED (non-zero exitCode), DEGRADED (zero exitCode but stdErrLineCount over stderr_warning_threshold), or HEALTHY.
route_by_ingestion_status (io.kestra.plugin.core.flow.Switch) branches on that status:
- HEALTHY: log_healthy (io.kestra.plugin.core.log.Log) logs it, nothing else happens.
- DEGRADED, FAILED, PROBE_FAILED: a dedicated io.kestra.plugin.slack.notifications.SlackIncomingWebhook task posts a message built only from the probe's own exitCode or stdErrLineCount, never an invented number.
log_audit_summary (io.kestra.plugin.core.log.Log) always runs last and logs the final status, the source type, dry_run, and the GMS server, closing out the audit cycle.
The errors block posts to Slack if the flow itself fails outside the gate logic above.

What you get

A real signal on whether your metadata catalog is still receiving ingestion runs, not just whether the flow executed.
A four-way classification (HEALTHY, DEGRADED, FAILED, PROBE_FAILED) built only from fields the DataHub ingestion CLI actually returns: exitCode and stdErrLineCount.
A safe-by-default dry run mode (console sink) so you can validate connectivity and recipe correctness before ever writing to the catalog.
A Slack alert that always carries the real number that triggered it.

Who it's for

Data platform and data governance teams responsible for catalog freshness and trust.
Teams running DataHub ingestion on a schedule who currently find out about a broken connector from a confused analyst instead of an alert.
Teams evaluating the DataHub plugin who want a working example of io.kestra.plugin.datahub.Ingestion with both a console and a datahub-rest sink.

Why orchestrate this with Kestra

The DataHub ingestion CLI runs a recipe and exits, it has no concept of a schedule, a threshold, or an alert. Kestra adds the schedule, the dry run and live modes behind one input, the classification logic that turns a raw exit code into an operational signal, the conditional Slack alerting, and a full execution history of every ingestion run next to every other pipeline you orchestrate. The whole gate is one declarative YAML file you can version and review like code.

Prerequisites

A local DataHub instance. The standard way to run one locally is the DataHub CLI quickstart, which manages a Docker Compose stack of DataHub's own containers (pip install acryl-datahub first):
```
DATAHUB_MAPPED_GMS_PORT=8081 datahub docker quickstart
```
This maps datahub-gms to host port 8081 instead of its default 8080, which otherwise collides with Kestra's own default UI port. The containers join a Docker network named datahub_network.
Connect your Kestra container to that network so the ingestion task's Docker task runner can resolve datahub-gms and your source database by name. Run docker network ls first to confirm the exact network name on your machine, then:
```
docker network connect datahub_network YOUR_KESTRA_NETWORK
```

A source database reachable on the same network, for example:

docker run -d --name postgres --network datahub_network -e POSTGRES_PASSWORD=<password> -e POSTGRES_DB=datahub_source postgres:16

A Slack incoming webhook for alerts.

Secrets

DATAHUB_SOURCE_PASSWORD: password for the source database the ingestion recipe reads from. In Kestra OSS, set it as the base64-encoded environment variable SECRET_DATAHUB_SOURCE_PASSWORD.
SLACK_WEBHOOK: Slack incoming webhook URL used by the alert tasks and the errors block. In Kestra OSS, set it as the base64-encoded environment variable SECRET_SLACK_WEBHOOK.

Inputs

Input	Type	Default	Allowed values
`source_type`	SELECT	`postgres`	`postgres`, `mysql`
`source_host_port`	STRING	`postgres:5432`	any `host:port`
`source_database`	STRING	`datahub_source`	any database name
`source_username`	STRING	`datahub`	any username
`datahub_gms_server`	STRING	`http://datahub-gms:8080`	any GMS URL
`docker_network`	STRING	`datahub_network`	any Docker network name
`container_image`	STRING	`acryldata/datahub-ingestion:head`	any DataHub ingestion image
`stderr_warning_threshold`	INT	`5`	any non-negative integer
`dry_run`	BOOL	`true`	`true`, `false`

Quick start

Start DataHub locally with DATAHUB_MAPPED_GMS_PORT=8081 datahub docker quickstart and a source database container on the same datahub_network.
Connect your Kestra container to datahub_network.
Add the DATAHUB_SOURCE_PASSWORD and SLACK_WEBHOOK secrets.
Run this flow manually with the default dry_run: true and read log_audit_summary for the classified status.
Switch dry_run to false once the dry run reports HEALTHY, then enable the six_hourly trigger.

How to extend

Read outputs.probe_live.vars or outputs.probe_console.vars for any structured counts your recipe's source.config is set up to emit, and fold them into the classification.
Add a source_type value for another DataHub source and branch the recipe config block per type if the fields differ.
Replace the stderr_warning_threshold check with a comparison against the previous run's stdErrLineCount stored in a Kestra KV entry, to catch a slow drift instead of a fixed threshold.
Chain a downstream task after HEALTHY that queries the DataHub GraphQL API to assert a specific dataset's lastIngested timestamp moved forward.

Pitfalls

DataHub's datahub-gms container defaults to host port 8080, the same default port Kestra's own UI uses. Remap it with DATAHUB_MAPPED_GMS_PORT before starting Kestra on the same host.
The ingestion container and the source database must be reachable from each other and from the DataHub GMS server on the same Docker network, or the probe returns PROBE_FAILED instead of a classified exit code.
The console sink accepts no config block at all, adding one is a silent no-op at best and a recipe validation error at worst. Only datahub-rest needs config.server.
A private app token or service account scoped only to the source database is enough, the ingestion recipe never needs catalog-admin credentials for the source side.

DataHub Ingestion Health Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

DataHub Ingestion Health Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Aerospike Stop-Writes Headroom Sentinel

Actian Vector Table Fragmentation and Storage Sentinel

Read out an A/B test with a sample ratio check and a guardrail metric

Apache ActiveMQ Artemis Dead Letter Address (DLA) Message Accumulation Sentinel

Automate Customer Feedback Triage, Sentiment Analysis, and Escalation with Google Gemini and Slack

AI Data Quality Anomaly Advisor and SQL Remediation Generator

More Infrastructure Blueprints

ADLS World-Writable ACL Sentinel

Renew TLS certificates through ACME, verify them, deploy them and roll back on failure

AI Incident Post-Mortem and Root Cause Generator

Environmental Disaster Early Warning System

ACM Certificate Expiry Compliance Sentinel

Autonomous Database Incident Investigation Agent with DuckDB and AI

New to Kestra?