Dremio Reflection and Job Health Gate for Kestra

id: dremio-reflection-job-health-gate
namespace: company.team
description: |
  Query Dremio's sys.reflections and sys.jobs system tables on a schedule,
  flag any Reflection that failed to refresh and any query job stuck in
  RUNNING past a runtime threshold, classify the result as CLEAN,
  AT_RISK, or UNREACHABLE, alert Slack, and optionally cancel the stuck
  jobs through Dremio's REST API, then re-query to verify none remain
  RUNNING.

inputs:
  - id: dremio_jdbc_url
    type: STRING
    displayName: Dremio JDBC URL
    description: JDBC URL of a self-hosted Dremio OSS coordinator, format
      jdbc:dremio:direct=host:31010.
    defaults: "jdbc:dremio:direct=dremio:31010"

  - id: dremio_username
    type: STRING
    displayName: Dremio username
    defaults: "dremio"

  - id: dremio_rest_base_url
    type: STRING
    displayName: Dremio REST API base URL
    description: Base URL of the Dremio web/REST port, used only to cancel a stuck job.
    defaults: "http://dremio:9047"

  - id: reflection_failure_threshold
    type: INT
    displayName: Reflection failure threshold
    description: A Reflection with num_failures at or above this value is flagged,
      in addition to any Reflection already in a FAILED or
      CAN_ACCELERATE_WITH_FAILURES state.
    defaults: 1

  - id: stuck_job_after_minutes
    type: INT
    displayName: Stuck job threshold in minutes
    description: A job still in RUNNING status this many minutes after it was
      submitted is flagged as stuck.
    defaults: 30

  - id: auto_cancel_stuck_jobs
    type: BOOL
    displayName: Auto cancel stuck jobs
    description: Allow the REST cancel call to run. Failed Reflections are always
      report-only, they are never cancelled automatically.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_cancel_stuck_jobs, for a job
      cancel to apply.
    defaults: true

tasks:
  - id: probe_reflections
    type: io.kestra.plugin.jdbc.dremio.Query
    description: Read-only. Reflections already failed or failing to refresh.
    url: "{{ inputs.dremio_jdbc_url }}"
    username: "{{ inputs.dremio_username }}"
    password: "{{ secret('DREMIO_PASSWORD') }}"
    sql: >-
      SELECT reflection_id, reflection_name, dataset_name, status, num_failures,
      last_failure_message FROM sys.reflections WHERE status IN ('FAILED',
      'CAN_ACCELERATE_WITH_FAILURES')
         OR num_failures >= {{ inputs.reflection_failure_threshold }}
    fetchType: FETCH
    allowFailure: true

  - id: probe_stuck_jobs
    type: io.kestra.plugin.jdbc.dremio.Query
    description: Read-only. Every job currently in RUNNING status; the runtime
      threshold is applied afterward in jq, not in SQL.
    url: "{{ inputs.dremio_jdbc_url }}"
    username: "{{ inputs.dremio_username }}"
    password: "{{ secret('DREMIO_PASSWORD') }}"
    sql: >-
      SELECT job_id, user_name, queue_name, query, submitted_epoch_millis FROM
      sys.jobs WHERE status = 'RUNNING'
    fetchType: FETCH
    allowFailure: true

  - id: findings
    type: io.kestra.plugin.core.output.OutputValues
    description: Read-only. Combines failed Reflections with jobs that have been
      RUNNING past stuck_job_after_minutes into one findings array.
    values:
      items: >-
        {{ {'reflections': outputs.probe_reflections.rows ?? [], 'jobs':
        outputs.probe_stuck_jobs.rows ?? [], 'threshold':
        inputs.stuck_job_after_minutes} | jq('(.reflections | map({kind:
        "REFLECTION_FAILED", id: .reflection_id, name: .reflection_name,
        dataset: .dataset_name, status: .status, detail: (.last_failure_message
        // ("num_failures=" + (.num_failures | tostring)))})) as $r | .threshold
        as $t | (.jobs | map(. + {minutesRunning: (((now * 1000) -
        .submitted_epoch_millis) / 60000 | floor)}) | map(select(.minutesRunning
        > $t)) | map({kind: "STUCK_JOB", id: .job_id, name: .user_name, dataset:
        (.queue_name // "default"), status: "RUNNING", detail: (.minutesRunning
        | tostring) + " minutes running"})) as $j | $r + $j') | first | toJson
        }}

  - id: classify_status
    type: io.kestra.plugin.core.debug.Return
    description: UNREACHABLE when either probe has no rows field, AT_RISK when the
      findings array is non-empty, otherwise CLEAN.
    format: "{{ (outputs.probe_reflections.rows is defined and
      outputs.probe_reflections.rows is not null and
      outputs.probe_stuck_jobs.rows is defined and outputs.probe_stuck_jobs.rows
      is not null) ? (fromJson(outputs.findings.values.items) | length > 0 ?
      'AT_RISK' : 'CLEAN') : 'UNREACHABLE' }}"

  - id: route_by_status
    type: io.kestra.plugin.core.flow.Switch
    description: Log a quiet confirmation when clean, page Slack and optionally
      cancel stuck jobs when at risk, or page Slack when Dremio could not be
      queried.
    value: "{{ outputs.classify_status.value }}"
    cases:
      "CLEAN":
        - id: log_clean
          type: io.kestra.plugin.core.log.Log
          message: "No failed Reflection and no job RUNNING past {{
            inputs.stuck_job_after_minutes }} minutes."
      "AT_RISK":
        - id: alert_at_risk
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "text": "Dremio health gate AT_RISK: {{ fromJson(outputs.findings.values.items) | length }} finding(s): {{ fromJson(outputs.findings.values.items) | jq('map(\"\\(.kind) \\(.name // .id) (\\(.dataset)): \\(.detail)\") | join(\"; \")') | first }}. {{ inputs.auto_cancel_stuck_jobs and not inputs.dry_run ? 'Cancelling stuck jobs now.' : 'No job cancelled (auto_cancel_stuck_jobs=' ~ inputs.auto_cancel_stuck_jobs ~ ', dry_run=' ~ inputs.dry_run ~ ').' }} Failed Reflections are report-only. Execution {{ execution.id }}."
            }

        - id: cancel_gate
          type: io.kestra.plugin.core.flow.If
          condition: "{{ inputs.auto_cancel_stuck_jobs and not inputs.dry_run and
            fromJson(outputs.findings.values.items) | jq('map(select(.kind ==
            \"STUCK_JOB\")) | length') | first > 0 }}"
          then:
            - id: cancel_loop
              type: io.kestra.plugin.core.flow.Loop
              description: One REST cancel call per stuck job, one at a time.
              concurrencyLimit: 1
              values: "{{ fromJson(outputs.findings.values.items) | jq('map(select(.kind ==
                \"STUCK_JOB\"))') | first }}"
              tasks:
                - id: cancel_job
                  type: io.kestra.plugin.core.http.Request
                  description: "plugin-jdbc-dremio exposes only Query, Queries, and Trigger, there
                    is no SQL CANCEL JOB statement in Dremio, so this calls
                    Dremio's own job-cancel REST endpoint directly."
                  uri: "{{ inputs.dremio_rest_base_url }}/api/v3/job/{{ item.value.id }}/cancel"
                  method: POST
                  headers:
                    Authorization: "Bearer {{ secret('DREMIO_PAT') }}"
                  options:
                    allowFailed: true
                  allowFailure: true

            - id: reread_stuck_jobs
              type: io.kestra.plugin.jdbc.dremio.Query
              description: Read-only. Re-lists every job still RUNNING after the cancel
                requests were sent, to compare against the cancelled job IDs.
              url: "{{ inputs.dremio_jdbc_url }}"
              username: "{{ inputs.dremio_username }}"
              password: "{{ secret('DREMIO_PASSWORD') }}"
              sql: >-
                SELECT job_id, status FROM sys.jobs WHERE status = 'RUNNING'
              fetchType: FETCH
              allowFailure: true

            - id: verify_cancel
              type: io.kestra.plugin.core.flow.If
              description: Fails if any job ID that was sent a cancel request is still present
                among the still-RUNNING jobs.
              condition: "{{ {'cancelled_ids': (fromJson(outputs.findings.values.items) |
                jq('map(select(.kind == \"STUCK_JOB\") | .id)') | first),
                'still_running_ids': (outputs.reread_stuck_jobs.rows ?? [] |
                jq('map(.job_id)') | first)} | jq('(.cancelled_ids -
                (.cancelled_ids - .still_running_ids)) | length > 0') | first
                }}"
              then:
                - id: cancel_failed
                  type: io.kestra.plugin.core.execution.Fail
                  errorMessage: "Cancel request sent, but at least one job is still RUNNING."
              else:
                - id: cancelled
                  type: io.kestra.plugin.core.log.Log
                  message: "Sent a cancel request for {{ fromJson(outputs.findings.values.items) |
                    jq('map(select(.kind == \"STUCK_JOB\")) | length') | first
                    }} stuck job(s) and verified none are still RUNNING."
          else:
            - id: not_cancelled
              type: io.kestra.plugin.core.log.Log
              message: "At-risk findings recorded but no job cancelled
                (auto_cancel_stuck_jobs={{ inputs.auto_cancel_stuck_jobs }},
                dry_run={{ inputs.dry_run }})."
      "UNREACHABLE":
        - id: alert_unreachable
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "text": "Dremio health gate UNREACHABLE: querying sys.reflections or sys.jobs on {{ inputs.dremio_jdbc_url }} did not return rows (coordinator down, auth failure, or system tables not accessible to this user). Execution {{ execution.id }}."
            }
    defaults:
      - id: log_unclassified
        type: io.kestra.plugin.core.log.Log
        message: "Completed Dremio health audit with status: {{
          outputs.classify_status.value }}"

  - id: log_audit_status
    type: io.kestra.plugin.core.log.Log
    description: Post-audit status line recorded on every run regardless of branch,
      for a continuous Reflection and job health trend in the execution history.
    message: "Dremio health audit: url={{ inputs.dremio_jdbc_url }},
      reflection_failure_threshold={{ inputs.reflection_failure_threshold }},
      stuck_job_after_minutes={{ inputs.stuck_job_after_minutes }}, status={{
      outputs.classify_status.value }}, execution={{ execution.id }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack when the gate flow itself fails, so a broken audit
      never silently stops monitoring Reflection and job health.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Dremio health gate flow {{ flow.id }} FAILED (execution {{ execution.id }}). Reflection and job health status is now unknown; check the task logs and the Dremio coordinator manually."
      }

outputs:
  - id: status
    type: STRING
    value: "{{ outputs.classify_status.value }}"

triggers:
  - id: dremio_health_gate_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Audit Reflections and running jobs every 15 minutes, always in safe mode.
    cron: "*/15 * * * *"
    disabled: true
    inputs:
      dremio_jdbc_url: "jdbc:dremio:direct=dremio:31010"
      dremio_username: "dremio"
      dremio_rest_base_url: "http://dremio:9047"
      reflection_failure_threshold: 1
      stuck_job_after_minutes: 30
      auto_cancel_stuck_jobs: false
      dry_run: true

A Dremio Reflection that stops refreshing does not fail loudly, queries that used to hit the accelerated Reflection silently fall back to a full scan of the raw dataset, and nobody notices until a dashboard that used to load in two seconds takes two minutes. Likewise, a query job stuck in RUNNING for hours quietly holds an engine slot that a real workload needs. This blueprint queries Dremio's own sys.reflections and sys.jobs system tables on a schedule, flags any Reflection that is FAILED, CAN_ACCELERATE_WITH_FAILURES, or has accumulated failures, flags any job still RUNNING past a runtime threshold, pages Slack the moment one is found or Dremio cannot be queried, and can cancel the stuck jobs through Dremio's REST API, then re-query to verify none remain RUNNING. plugin-jdbc-dremio ships only Query, Queries, and Trigger, there is no SQL statement to cancel a job in Dremio, so cancellation calls Dremio's own POST /api/v3/job/{id}/cancel REST endpoint directly, the same pattern Dremio's own REST client uses.

How it works

probe_reflections (io.kestra.plugin.jdbc.dremio.Query) selects Reflections that are already FAILED, CAN_ACCELERATE_WITH_FAILURES, or at or above reflection_failure_threshold failures, read-only, with allowFailure: true.
probe_stuck_jobs selects every job currently RUNNING; the runtime threshold is applied afterward, not in SQL, also read-only with allowFailure: true.
findings (io.kestra.plugin.core.output.OutputValues) combines the failed Reflections with the jobs whose computed runtime exceeds stuck_job_after_minutes into one findings array, each entry tagged REFLECTION_FAILED or STUCK_JOB.
classify_status (io.kestra.plugin.core.debug.Return) returns UNREACHABLE when either probe's rows field is not defined, AT_RISK when the findings array is non-empty, otherwise CLEAN.
route_by_status (io.kestra.plugin.core.flow.Switch) branches on that string:
- CLEAN: logs a quiet confirmation.
- AT_RISK: alert_at_risk posts every finding to Slack. cancel_gate runs only when auto_cancel_stuck_jobs is true, dry_run is false, and at least one STUCK_JOB finding exists: cancel_loop (concurrencyLimit: 1) sends one REST cancel request per stuck job, reread_stuck_jobs re-queries sys.jobs for those job IDs, and verify_cancel fails the flow if any of them is still RUNNING. Failed Reflections are always report-only, there is no automated fix for a broken Reflection from this flow.
- UNREACHABLE: alert_unreachable posts a separate Slack message.
log_audit_status records a status line on every run regardless of branch.
The errors block alerts Slack separately if the flow itself fails outright.
The dremio_health_gate_schedule trigger polls every 15 minutes with explicit input overrides, shipped disabled: true and always in safe mode (auto_cancel_stuck_jobs: false, dry_run: true).

What you get

A scheduled, system-table-native audit of Dremio's acceleration layer and job queue with zero custom client code.
A three-way status output (CLEAN / AT_RISK / UNREACHABLE) you can chart or gate other flows on.
A Slack page naming every failed Reflection and every stuck job with its runtime, with cancellation applied and verified behind auto_cancel_stuck_jobs and dry_run.
A standing audit log line per run and a separate failure channel for the gate flow itself.

Who it's for

Data platform teams running self-hosted Dremio as a semantic and acceleration layer over a lakehouse, who need to know when a Reflection silently stops accelerating queries.
Teams who have had a runaway or stuck query hold an engine slot for hours with nobody noticing until concurrency ran out.
Teams evaluating the Dremio plugin who want an example of sys.reflections and sys.jobs used together for a Day 2 operations check, beyond a basic Query and transform.

Why orchestrate this with Kestra

A one-off look at the Dremio UI's Jobs or Reflections tab proves the state for one moment. This flow keeps every audit in the execution history, turns the result into a typed output other flows can depend on, and gives you a branch to grow into real remediation, cancelling a stuck job, rebuilding a broken Reflection's dataset, or opening a ticket, without rewriting the audit itself.

Prerequisites

A self-hosted Dremio OSS coordinator, not Dremio Cloud, reachable from the Kestra Worker on its JDBC port (31010 by default) and its REST/web port (9047 by default). At least one Reflection configured so sys.reflections has rows.
For local testing: docker run -d --name dremio -p 9047:9047 -p 31010:31010 -p 32010:32010 dremio/dremio-oss:latest, then connect it to your Kestra worker's Docker network so dremio_jdbc_url and dremio_rest_base_url resolve: docker network connect YOUR_KESTRA_NETWORK dremio (run docker network ls to find the real network name). The flow reaches Dremio on its container-internal ports over that network; the published host ports are only for browsing the Dremio UI from your own machine, after completing Dremio's first-run admin setup at http://localhost:9047.
A Dremio Personal Access Token (generate it from Account Settings in the Dremio UI) for the REST cancel call.
A Slack incoming webhook for the at-risk and unreachable alerts.

Secrets

DREMIO_PASSWORD: password used by every Query task against Dremio's JDBC port.
DREMIO_PAT: Dremio Personal Access Token, sent as a Bearer token to the job-cancel REST endpoint.
SLACK_WEBHOOK_URL: Slack incoming webhook used for the at-risk alert, the unreachable alert, and the flow-failure alert.
OSS note: if your Kestra instance reads secrets from environment variables instead of a secrets backend, set them as base64, for example echo -n "<your-password>" | base64 exported as SECRET_DREMIO_PASSWORD, and likewise for SECRET_DREMIO_PAT and SECRET_SLACK_WEBHOOK_URL.

Inputs

Input	Default	Purpose
`dremio_jdbc_url`	`jdbc:dremio:direct=dremio:31010`	JDBC URL of the coordinator
`dremio_username`	`dremio`	JDBC username
`dremio_rest_base_url`	`http://dremio:9047`	REST base URL, used only to cancel a job
`reflection_failure_threshold`	`1`	`num_failures` at or above which a Reflection is flagged
`stuck_job_after_minutes`	`30`	Runtime past which a `RUNNING` job is flagged
`auto_cancel_stuck_jobs`	`false`	Allow the REST cancel call to run
`dry_run`	`true`	Must be `false` for a cancel to apply

Quick start

Add the DREMIO_PASSWORD, DREMIO_PAT, and SLACK_WEBHOOK_URL secrets.
Point dremio_jdbc_url and dremio_rest_base_url at your coordinator.
Run the flow once and read status.
Run it with auto_cancel_stuck_jobs: true and dry_run: false while a long-running job is active to confirm it gets cancelled and verified.
Enable the dremio_health_gate_schedule trigger, which always runs in safe mode.

Outputs

status (STRING): the classified outcome, {{ outputs.classify_status.value }}, also exposed as the flow-level output {{ outputs.status }}.
{{ outputs.findings.values.items }}: a JSON array of every finding, each with kind, id, name, dataset, status, and detail.

Pitfalls

This blueprint has not been executed against a live Dremio coordinator; the exact set of sys.jobs status values and the availability of sys.reflections columns can vary by Dremio version. Verify both against your own deployment before enabling the schedule.
Querying sys.reflections and sys.jobs requires a user with sufficient privileges; a restricted user may see an empty or partial result that reads as falsely CLEAN.
A coordinator with no Reflections configured at all audits zero entries and always reads CLEAN for that half of the check; configure at least one Reflection to exercise the AT_RISK branch.
stuck_job_after_minutes is evaluated against submitted_epoch_millis, the time the job entered the system, not execution start; a job queued for a long time behind a busy engine before it starts running is not counted as stuck by this definition.
Cancelling a job through the REST API stops it but does not roll back any side effects of a CREATE TABLE AS or similar DDL/DML job that had already written partial results.

How to extend

Loop over a list of coordinators with a Loop task to audit a multi-coordinator Dremio deployment from one flow.
Parse considered_count, matched_count, and accelerated_count from sys.reflections to alert on a Reflection nobody's queries actually use, a cost-cleanup candidate, not just an outright failure.
Chain a downstream task after cancel_failed to escalate to the dataset owner instead of only failing the flow.
Add an OAuth2 or API-key secret type instead of a long-lived PAT once your Dremio deployment supports it.

Dremio Reflection & Job Health Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

Outputs

Pitfalls

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Dremio Reflection & Job Health Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

Outputs

Pitfalls

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

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

AES-256 Argon2id Cryptographic Landing and Compliance Gate

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

AI Data Quality Anomaly Advisor and SQL Remediation Generator

Adanos Asset Sentiment Drift Sentinel

Actian Vector Table Fragmentation and Storage Sentinel

More Infrastructure Blueprints

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

ADLS World-Writable ACL Sentinel

AI Incident Post-Mortem and Root Cause Generator

Aerospike Stop-Writes Headroom Sentinel

Audit Every Agent Tool Call Against Policy Before It Goes Unnoticed

Automatically Review GitHub Pull Requests with OpenAI GPT-4o and Post Comments

New to Kestra?