Meltano ELT Pipeline State Drift and Freshness Gate

id: meltano-elt-pipeline-state-drift-and-freshness-gate
namespace: company.data

inputs:
  - id: git_repository_url
    type: STRING
    displayName: Meltano Project Repository URL
    defaults: "https://github.com/meltano/demo-project"
    description: HTTPS URL of the Git repository hosting the Meltano project
      configuration (`meltano.yml`). The default is Meltano's own real public
      demo project.

  - id: git_branch
    type: STRING
    displayName: Git Branch or Ref
    defaults: "main"
    description: Git branch, pull request ref, or commit SHA to clone and verify.

  - id: job_name
    type: STRING
    displayName: Meltano Job Name
    defaults: "tap-gitlab-to-target-jsonl"
    description: Identifier of the Meltano ELT job or pipeline whose state and
      schema are being audited. This label is informational only; the audit
      targets the extractors declared in meltano.yml and the state recorded
      under state_id.

  - id: state_id
    type: STRING
    displayName: Singer State ID
    defaults: "dev:tap-gitlab-to-target-jsonl"
    description: Replication state identifier used by Meltano state service to store
      incremental cursor bookmarks.

  - id: max_allowed_lag_hours
    type: INT
    displayName: Maximum Allowed Bookmark Lag (Hours)
    defaults: 24
    description: Maximum permissible hours between the latest incremental bookmark
      timestamp and execution time before flagging state stagnation.

  - id: check_schema_drift
    type: BOOL
    displayName: Enforce Schema Catalog Drift Check
    defaults: true
    description: When true, compares live tap discovery output against committed
      catalog selections to detect unmapped upstream columns.

  - id: slack_channel
    type: STRING
    displayName: Slack Notification Channel
    defaults: "#dataops-pipeline-alerts"
    description: Destination Slack channel for ELT freshness warnings and schema
      evolution alerts.

triggers:
  - id: elt_pre_flight_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Triggered immediately prior to scheduled warehouse loads to verify
      state integrity and prevent corrupted syncs.
    key: "{{ secret('MELTANO_WEBHOOK_KEY') }}"

  - id: hourly_state_freshness_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Scheduled hourly check auditing Singer incremental replication
      cursors across production taps.
    cron: "0 * * * *"

tasks:
  - id: prepare_working_directory
    type: io.kestra.plugin.core.flow.WorkingDirectory
    tasks:
      - id: clone_meltano_repo
        type: io.kestra.plugin.git.Clone
        description: Clones the target Meltano project repository to inspect project
          configuration and catalog manifests.
        url: "{{ inputs.git_repository_url }}"
        branch: "{{ inputs.git_branch }}"

      - id: audit_meltano_pipeline
        type: io.kestra.plugin.scripts.shell.Commands
        description: Runs containerized Meltano audit to inspect replication bookmarks,
          calculate state staleness, and detect schema drift.
        containerImage: python:3.11-slim
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        env:
          JOB_NAME: "{{ inputs.job_name }}"
          STATE_ID: "{{ inputs.state_id }}"
          MAX_LAG_HOURS: "{{ inputs.max_allowed_lag_hours }}"
          CHECK_SCHEMA_DRIFT: "{{ inputs.check_schema_drift }}"
          GITHUB_TOKEN: "{{ secret('GITHUB_TOKEN') }}"
        beforeCommands:
          - apt-get update -qq && apt-get install -y -qq git curl jq > /dev/null
            2>&1
          - pip install --quiet --no-cache-dir pyyaml meltano
        commands:
          - python3 audit_meltano.py
        inputFiles:
          audit_meltano.py: |
            import os
            import sys
            import json
            import subprocess
            import yaml
            from datetime import datetime, timezone

            job_name = os.environ.get("JOB_NAME", "tap-stripe-to-target-duckdb")
            state_id = os.environ.get("STATE_ID", "dev:tap-stripe-to-target-duckdb")
            max_lag_hours = int(os.environ.get("MAX_LAG_HOURS", "24"))
            check_schema_drift = os.environ.get("CHECK_SCHEMA_DRIFT", "true").lower() == "true"

            if not os.path.exists("meltano.yml"):
                print("No meltano.yml found in the cloned repository; nothing to audit.")
                output_payload = {
                    "compliance_status": "NO_MELTANO_PROJECT_FOUND",
                    "total_streams": 0,
                    "max_lag_hours": 0,
                    "stagnant_streams_count": 0,
                    "has_schema_drift": False,
                    "stagnant_stream_names": []
                }
                print(f'::{{"outputs": {json.dumps(output_payload)}}}::')
                sys.exit(0)

            install_cmd = subprocess.run(["meltano", "install"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
            if install_cmd.returncode != 0:
                print(f"meltano install failed: {install_cmd.stderr.strip()[-1000:]}")

            # Fetch the real Singer state from the project's configured state backend.
            state_cmd = subprocess.run(["meltano", "state", "get", state_id], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
            if state_cmd.returncode != 0:
                output_payload = {
                    "compliance_status": "STATE_FETCH_FAILED",
                    "total_streams": 0,
                    "max_lag_hours": 0,
                    "stagnant_streams_count": 0,
                    "has_schema_drift": False,
                    "stagnant_stream_names": []
                }
                print(f"meltano state get failed for state_id '{state_id}': {state_cmd.stderr.strip()[-1000:]}")
                print(f'::{{"outputs": {json.dumps(output_payload)}}}::')
                sys.exit(0)

            try:
                state_data = json.loads(state_cmd.stdout)
            except Exception as ex:
                state_data = {}

            # Real Singer state shape: {"bookmarks": {"<stream>": {"replication_key_value": ..., "partitions": [...]}}}
            bookmarks = state_data.get("bookmarks", {})
            now = datetime.now(timezone.utc)

            def bookmark_values(bm):
                if "partitions" in bm:
                    return [p.get("replication_key_value") for p in bm.get("partitions", []) if p.get("replication_key_value")]
                val = bm.get("replication_key_value")
                return [val] if val else []

            total_streams = len(bookmarks)
            stagnant_streams = []
            max_observed_lag = 0.0

            for stream_name, bm in bookmarks.items():
                values = bookmark_values(bm)
                if not values:
                    continue
                oldest_ts = None
                for rep_val in values:
                    try:
                        ts = datetime.fromisoformat(str(rep_val).replace("Z", "+00:00"))
                    except Exception:
                        continue
                    if oldest_ts is None or ts < oldest_ts:
                        oldest_ts = ts
                if oldest_ts is None:
                    continue
                lag_hours = (now - oldest_ts).total_seconds() / 3600.0
                if lag_hours > max_observed_lag:
                    max_observed_lag = lag_hours
                if lag_hours > max_lag_hours:
                    stagnant_streams.append({
                        "stream": stream_name,
                        "replication_key": bm.get("replication_key", "unknown"),
                        "bookmark_value": oldest_ts.isoformat(),
                        "lag_hours": round(lag_hours, 1)
                    })

            # Detect schema catalog drift: run live discovery against each configured extractor
            # and flag any discovered stream with no matching `select` rule in meltano.yml.
            drifted_streams = []
            if check_schema_drift:
                try:
                    with open("meltano.yml") as f:
                        project_config = yaml.safe_load(f) or {}
                except Exception:
                    project_config = {}

                extractors = (project_config.get("plugins") or {}).get("extractors") or []
                for extractor in extractors:
                    tap_name = extractor.get("name")
                    if not tap_name:
                        continue
                    select_rules = extractor.get("select") or ["*.*"]
                    selected_stream_names = set()
                    for rule in select_rules:
                        stream_part = rule.split(".")[0].lstrip("!")
                        selected_stream_names.add(stream_part)

                    discover_cmd = subprocess.run(
                        ["meltano", "invoke", tap_name, "--discover"],
                        stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
                    )
                    if discover_cmd.returncode != 0:
                        continue
                    try:
                        catalog = json.loads(discover_cmd.stdout)
                    except Exception:
                        continue

                    for stream in catalog.get("streams", []):
                        stream_name = stream.get("stream") or stream.get("tap_stream_id")
                        if not stream_name:
                            continue
                        if "*" not in selected_stream_names and stream_name not in selected_stream_names:
                            drifted_streams.append(f"{tap_name}:{stream_name}")

            if stagnant_streams:
                compliance_status = "STATE_BOOKMARK_STAGNANT"
            elif drifted_streams:
                compliance_status = "SCHEMA_DRIFT_DETECTED"
            else:
                compliance_status = "PIPELINE_SYNCHRONIZED_HEALTHY"

            report = [
                f"# Meltano ELT Pipeline Freshness & State Audit",
                f"- **Job Name**: `{job_name}`",
                f"- **State ID**: `{state_id}`",
                f"- **Compliance Status**: `{compliance_status}`",
                f"- **Total Streams Tracked**: {total_streams}",
                f"- **Max Observed Bookmark Lag**: {max_observed_lag:.1f} hours (SLA Limit: {max_lag_hours} hours)",
                f"- **Stagnant Streams**: {len(stagnant_streams)}",
                f"- **Schema Drift Detected**: {len(drifted_streams) > 0}",
                "",
                "## Incremental Stream Bookmarks"
            ]

            for s_name, bm in bookmarks.items():
                rep_k = bm.get("replication_key", "n/a")
                rep_v = bm.get("replication_key_value", "n/a")
                report.append(f"- `{s_name}`: cursor `{rep_k}` = `{rep_v}`")

            if stagnant_streams:
                report.append("\n## Stagnant Streams Exceeding SLA")
                for s in stagnant_streams:
                    report.append(f"- **{s['stream']}**: Lag of {s['lag_hours']}h exceeds {max_lag_hours}h threshold (Bookmark: `{s['bookmark_value']}`)")

            report_md = "\n".join(report)
            print(report_md)

            output_payload = {
                "compliance_status": compliance_status,
                "total_streams": total_streams,
                "max_lag_hours": round(max_observed_lag, 1),
                "stagnant_streams_count": len(stagnant_streams),
                "has_schema_drift": len(drifted_streams) > 0,
                "stagnant_stream_names": [s["stream"] for s in stagnant_streams],
                "drifted_stream_names": drifted_streams
            }
            print(f'::{{"outputs": {json.dumps(output_payload)}}}::')

  - id: route_governance_decision
    type: io.kestra.plugin.core.flow.Switch
    description: Evaluates Meltano state freshness and schema drift to gate
      downstream warehouse ingestion or alert Slack.
    value: "{{ outputs.audit_meltano_pipeline.vars.compliance_status }}"
    cases:
      "STATE_BOOKMARK_STAGNANT":
        - id: alert_stagnant_state_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Dispatches high-severity alert to DataOps Slack channel when Singer
            incremental bookmarks freeze past SLA.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": ":rotating_light: *Meltano Freshness Gate Tripped: Stagnant Replication State* :rotating_light:\n*Job:* `{{ inputs.job_name }}`\n*State ID:* `{{ inputs.state_id }}`\n*Observed Lag:* {{ outputs.audit_meltano_pipeline.vars.max_lag_hours }} hours (Limit: {{ inputs.max_allowed_lag_hours }}h)\n*Stagnant Streams:* {{ outputs.audit_meltano_pipeline.vars.stagnant_stream_names | join(', ') }}\n*Action:* Check upstream API rate limits and verify tap extraction cursor."
            }

        - id: halt_on_stagnant_state
          type: io.kestra.plugin.core.execution.Fail
          description: Halts workflow to prevent stale incremental syncs from triggering
            downstream data transformation pipelines.
          errorMessage: "FRESHNESS SLA BREACH: Meltano Singer bookmark lag exceeds maximum
            allowed threshold."

      "SCHEMA_DRIFT_DETECTED":
        - id: alert_schema_drift_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Alerts data engineering team when upstream API schema changes
            require warehouse target migration.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": ":warning: *Meltano Schema Drift Warning: Upstream Catalog Mismatch* :warning:\n*Job:* `{{ inputs.job_name }}`\n*State ID:* `{{ inputs.state_id }}`\n*Action:* Run `meltano select <tap> --list --all` to inspect unselected new stream properties."
            }

        - id: halt_on_schema_drift
          type: io.kestra.plugin.core.execution.Fail
          description: Halts execution when unmapped upstream schema changes threaten
            warehouse target stability.
          errorMessage: "SCHEMA DRIFT: Upstream Singer tap schema desynchronized from
            target data contract."

      "STATE_FETCH_FAILED":
        - id: alert_state_fetch_failed_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Alerts the team when the Meltano state backend could not be read,
            so freshness could not be verified.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": ":warning: *Meltano State Fetch Failed* :warning:\n*Job:* `{{ inputs.job_name }}`\n*State ID:* `{{ inputs.state_id }}`\n*Action:* `meltano state get` failed. Check that the state backend is reachable and the state ID exists; this is not the same as a verified-fresh audit."
            }

        - id: halt_on_state_fetch_failure
          type: io.kestra.plugin.core.execution.Fail
          description: Halts execution since freshness could not be verified; treating a
            failed state fetch as healthy would be unsafe.
          errorMessage: "STATE FETCH FAILED: Could not read Meltano state for the
            configured state_id. Freshness status unknown."

      "NO_MELTANO_PROJECT_FOUND":
        - id: log_no_project_found
          type: io.kestra.plugin.core.log.Log
          description: Records that the cloned repository had no meltano.yml at its root.
          message: "Meltano audit: no meltano.yml found in {{ inputs.git_repository_url }}
            (branch {{ inputs.git_branch }}). Nothing was audited."

    defaults:
      - id: log_clean_synchronization
        type: io.kestra.plugin.core.log.Log
        description: Logs compliance certification confirming Meltano Singer state
          cursors are active and within freshness SLAs.
        message: "Meltano state audit passed: All {{
          outputs.audit_meltano_pipeline.vars.total_streams }} streams in job
          '{{ inputs.job_name }}' are synchronized within SLA (Lag: {{
          outputs.audit_meltano_pipeline.vars.max_lag_hours }}h)."