dlt Dataset Freshness Monitor

id: dlt-load-freshness-monitor
namespace: company.team
description: |
  Monitor the freshness of every dlt dataset in the warehouse. The flow
  queries dlt's own _dlt_loads audit table, computes the age of the last
  successful load per dataset, and alerts when any dataset is staler than
  the threshold, catching pipelines that stopped without failing loudly.

inputs:
  - id: warehouse_path
    type: STRING
    defaults: /data/warehouse.duckdb
    description: Persistent path for the DuckDB destination file. Point at a volume
      that survives task runs; incremental state and merge history live here.

  - id: max_age_hours
    type: INT
    defaults: 26
    description: Alert when the newest completed load in a dataset is older than
      this many hours.

tasks:
  - id: check_freshness
    type: io.kestra.plugin.dlt.CLI
    description: Query the _dlt_loads audit table that dlt maintains in every
      dataset, compute the hours since the last completed load, and emit stale
      datasets as outputs.
    inputFiles:
      check_freshness.py: |
        import duckdb
        import json
        import os
        from datetime import datetime, timezone

        MAX_AGE_HOURS = float(os.environ["MAX_AGE_HOURS"])
        DB_PATH = os.environ["WAREHOUSE_DB_PATH"]

        con = duckdb.connect(DB_PATH, read_only=True)
        schemas = [
            r[0]
            for r in con.execute(
                "SELECT schema_name FROM information_schema.schemata"
            ).fetchall()
        ]

        report, stale = {}, {}
        now = datetime.now(timezone.utc)
        for schema in schemas:
            tables = [
                r[0]
                for r in con.execute(
                    f"SELECT table_name FROM information_schema.tables WHERE table_schema = '{schema}'"
                ).fetchall()
            ]
            if "_dlt_loads" not in tables:
                continue
            row = con.execute(
                f'SELECT max(inserted_at) FROM "{schema}"."_dlt_loads" WHERE status = 0'
            ).fetchone()
            if row and row[0]:
                last = row[0]
                if last.tzinfo is None:
                    last = last.replace(tzinfo=timezone.utc)
                age_hours = round((now - last).total_seconds() / 3600, 1)
                report[schema] = age_hours
                if age_hours > MAX_AGE_HOURS:
                    stale[schema] = age_hours

        fmt = lambda d: ", ".join(f"{k}: {v}h" for k, v in d.items()) or "none"
        print("::" + json.dumps({"outputs": {"ages_hours": report, "stale": stale, "stale_count": len(stale), "ages_summary": fmt(report), "stale_summary": fmt(stale)}}) + "::")
    env:
      MAX_AGE_HOURS: "{{ inputs.max_age_hours }}"
      WAREHOUSE_DB_PATH: "{{ inputs.warehouse_path }}"
    beforeCommands:
      - pip install duckdb --quiet
    commands:
      - python check_freshness.py

  - id: route_result
    type: io.kestra.plugin.core.flow.If
    description: Alert only when at least one dataset breached the freshness threshold.
    condition: "{{ outputs.check_freshness.vars.stale_count > 0 }}"
    then:
      - id: alert_stale
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "STALE dlt datasets detected: {{ outputs.check_freshness.vars.stale_summary }} exceeded {{ inputs.max_age_hours }}h since their last completed load. A pipeline probably stopped without failing (execution {{ execution.id }})."
          }
    else:
      - id: log_fresh
        type: io.kestra.plugin.core.log.Log
        message: "All dlt datasets fresh: {{ outputs.check_freshness.vars.ages_summary
          }} since last completed load, threshold {{ inputs.max_age_hours }}h."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: A broken monitor is worse than a stale dataset; alert when the
      check itself cannot run.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "dlt freshness monitor FAILED to run (execution {{ execution.id }}). Datasets are currently unmonitored; check warehouse connectivity."
      }

triggers:
  - id: morning_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check freshness every morning after the nightly loads should have finished.
    cron: "0 8 * * *"
    disabled: true