Audit Databricks Clusters Left Running Outside Work Hours

id: databricks-job-cluster-auto-termination-audit
namespace: company.team
description: |
  Databricks interactive clusters left running outside work hours burn compute
  when auto termination is disabled or misconfigured. A governance job POSTs
  the cluster inventory to a keyed webhook, Python flags every running cluster
  with termination disabled or idle time past policy, and a Loop walks the
  offenders alerting Slack on each one. A final If escalates the summary with
  the wasted compute hour estimate and can fail the run when violations
  exist. The offender list, the counts, the wasted hours estimate and the
  audit verdict are published as outputs.

triggers:
  - id: on_cluster_inventory
    type: io.kestra.plugin.core.trigger.Webhook
    description: A governance or usage export job POSTs the cluster inventory here
      after each collection window so the audit runs against fresh activity
      data.
    key: "{{ secret('DATABRICKS_AUDIT_WEBHOOK_KEY') }}"
  - id: nightly_idle_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Daily 22:00 UTC fallback net that rescores the staged cluster
      inventory, shipped disabled until an operator switches it on.
    cron: "0 22 * * *"
    disabled: true

inputs:
  - id: workspace_name
    type: STRING
    defaults: prod-analytics
    description: Databricks workspace the inventory came from, echoed in every alert
      so multi workspace estates stay distinguishable.
  - id: idle_policy_minutes
    type: INT
    defaults: 120
    description: Minutes a running cluster may sit idle before the audit flags it,
      applied only to clusters that have auto termination enabled.
  - id: fail_on_violations
    type: BOOL
    defaults: true
    description: Whether any violation turns the execution red after the Slack
      alerts, flip to false for report only runs.

tasks:
  # --- 1. inventory workspace ---
  - id: inventory_workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: One shared directory so the staged cluster inventory and the
      auditor read the same file for the whole run.
    tasks:
      # --- 2. inventory fixture ---
      - id: stage_cluster_inventory
        type: io.kestra.plugin.scripts.shell.Commands
        description: Write a Databricks cluster inventory with a heredoc, five clusters
          covering disabled auto termination, idle time past policy, a healthy
          interactive cluster and a terminated one, so trial runs always have an
          inventory to audit.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          entryPoint: []
        containerImage: alpine:3.20
        commands:
          - |
            set -u
            cat > cluster_inventory.json <<'JSON'
            {
              "workspace": "prod-analytics",
              "clusters": [
                {"id": "0113-482917-adept910", "name": "etl-dbx-prod", "state": "RUNNING", "auto_termination_minutes": 0, "idle_minutes": 145},
                {"id": "0113-556201-calm2234", "name": "analyst-adhoc", "state": "RUNNING", "auto_termination_minutes": 120, "idle_minutes": 40},
                {"id": "0113-667312-brisk4455", "name": "streaming-cdc", "state": "TERMINATED", "auto_termination_minutes": 60, "idle_minutes": 0},
                {"id": "0113-778423-eager5566", "name": "ml-feature-dev", "state": "RUNNING", "auto_termination_minutes": null, "idle_minutes": 95},
                {"id": "0113-889534-fizzy7788", "name": "bi-cache-refresh", "state": "RUNNING", "auto_termination_minutes": 60, "idle_minutes": 200}
              ]
            }
            JSON
            echo "staged $(wc -c < cluster_inventory.json) bytes of cluster inventory"

      # --- 3. inventory audit ---
      - id: audit_clusters
        type: io.kestra.plugin.scripts.python.Script
        description: Flag every running cluster with auto termination disabled or idle
          time past the policy, estimate the wasted compute hours per offender,
          and emit the verdict and the offender rows through the outputs
          protocol.
        containerImage: python:3.12-slim
        env:
          CLUSTERS_JSON: "{{ trigger.body ?? '' }}"
          IDLE_POLICY_MINUTES: "{{ inputs.idle_policy_minutes }}"
        script: |
          import json
          import os

          DEFAULT_INVENTORY = """
          {
            "workspace": "prod-analytics",
            "clusters": [
              {"id": "0113-482917-adept910", "name": "etl-dbx-prod", "state": "RUNNING", "auto_termination_minutes": 0, "idle_minutes": 145},
              {"id": "0113-556201-calm2234", "name": "analyst-adhoc", "state": "RUNNING", "auto_termination_minutes": 120, "idle_minutes": 40},
              {"id": "0113-667312-brisk4455", "name": "streaming-cdc", "state": "TERMINATED", "auto_termination_minutes": 60, "idle_minutes": 0},
              {"id": "0113-778423-eager5566", "name": "ml-feature-dev", "state": "RUNNING", "auto_termination_minutes": null, "idle_minutes": 95},
              {"id": "0113-889534-fizzy7788", "name": "bi-cache-refresh", "state": "RUNNING", "auto_termination_minutes": 60, "idle_minutes": 200}
            ]
          }
          """


          def as_int(value, fallback=0):
              try:
                  return int(value)
              except (TypeError, ValueError):
                  try:
                      return int(float(value))
                  except (TypeError, ValueError):
                      return fallback


          raw = (os.environ.get("CLUSTERS_JSON") or "").strip() or DEFAULT_INVENTORY.strip()
          policy = max(0, as_int(os.environ.get("IDLE_POLICY_MINUTES"), 120))

          outputs = {
              "audit_ok": False,
              "workspace": "unknown",
              "cluster_count": 0,
              "offender_count": 0,
              "wasted_hours": 0.0,
              "offenders": [],
              "clusters": [],
          }

          try:
              document = json.loads(raw)
          except ValueError:
              document = None

          if isinstance(document, dict):
              items = document.get("clusters")
              if not isinstance(items, list):
                  items = []
              rows = []
              offenders = []
              wasted = 0.0
              for cluster in items:
                  if not isinstance(cluster, dict):
                      continue
                  state = str(cluster.get("state") or "").strip().upper()
                  auto_raw = cluster.get("auto_termination_minutes")
                  if isinstance(auto_raw, bool) or not isinstance(auto_raw, (int, float)):
                      auto_term = 0
                  else:
                      auto_term = int(auto_raw)
                  idle = max(0, as_int(cluster.get("idle_minutes")))
                  flagged = False
                  reason = ""
                  if state == "RUNNING":
                      if auto_term <= 0:
                          flagged = True
                          reason = "termination_disabled"
                      elif idle > policy:
                          flagged = True
                          reason = "idle_beyond_policy"
                  row = {
                      "id": str(cluster.get("id") or "unknown"),
                      "name": str(cluster.get("name") or "unknown"),
                      "state": state or "UNKNOWN",
                      "auto_termination_minutes": auto_term,
                      "idle_minutes": idle,
                      "flagged": flagged,
                      "reason": reason,
                      "wasted_hours": round(idle / 60.0, 2) if flagged else 0.0,
                  }
                  rows.append(row)
                  if flagged:
                      offenders.append(row)
                      wasted += row["wasted_hours"]
              outputs.update({
                  "audit_ok": True,
                  "workspace": str(document.get("workspace") or "unknown"),
                  "cluster_count": len(rows),
                  "offender_count": len(offenders),
                  "wasted_hours": round(wasted, 2),
                  "offenders": offenders,
                  "clusters": rows,
              })

          print("audit_ok=%s clusters=%d offenders=%d wasted_hours=%.2f policy=%d" % (
              outputs["audit_ok"], outputs["cluster_count"], outputs["offender_count"],
              outputs["wasted_hours"], policy))
          print("::" + json.dumps({"outputs": outputs}) + "::")

  # --- 4. offender sweep ---
  - id: remediate_offenders
    type: io.kestra.plugin.core.flow.Loop
    description: Walk the flagged clusters one at a time so each violation gets its
      own alert with the exact idle counters instead of one blended message.
    concurrencyLimit: 1
    values: "{{ outputs.audit_clusters.vars.offenders }}"
    tasks:
      - id: route_offender
        type: io.kestra.plugin.core.flow.If
        description: Branch on the violation reason the auditor already computed, so a
          disabled auto termination setting and an idle overrun get different
          remediation advice.
        condition: "{{ item.value.reason == 'termination_disabled' }}"
        then:
          - id: alert_disabled_termination
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Post the cluster name, its id, the idle minutes burned and the
              wasted compute hours so the owner enables auto termination on the
              cluster.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "Databricks cluster {{ item.value.name }} ({{ item.value.id }}) in {{ inputs.workspace_name }} runs with auto termination DISABLED and has been idle {{ item.value.idle_minutes }} minutes - {{ item.value.wasted_hours }} compute hours already burned. Set an auto termination timeout on the cluster or decommission it. Execution {{ execution.id }}."
              }
        else:
          - id: alert_idle_overrun
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Post the cluster name, its timeout, the idle minutes past policy
              and the wasted compute hours so the owner tightens the timeout or
              stops the cluster.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "Databricks cluster {{ item.value.name }} ({{ item.value.id }}) in {{ inputs.workspace_name }} has a {{ item.value.auto_termination_minutes }} minute timeout but sat idle {{ item.value.idle_minutes }} minutes, past the {{ inputs.idle_policy_minutes }} minute policy - {{ item.value.wasted_hours }} compute hours burned. Tighten the timeout or stop the cluster. Execution {{ execution.id }}."
              }

  # --- 5. audit verdict ---
  - id: audit_verdict
    type: io.kestra.plugin.core.flow.If
    description: Branch on the audit verdict so a workspace with violations
      escalates the summary and can fail the run, while a clean workspace is
      logged.
    condition: "{{ outputs.audit_clusters.vars.audit_ok and
      outputs.audit_clusters.vars.offender_count > 0 }}"
    then:
      # --- 6. audit escalation ---
      - id: alert_audit_summary
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post the workspace, the offender count against the cluster total
          and the wasted compute hour estimate so finops sees the whole blast
          radius in one message.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "Databricks auto termination audit for {{ inputs.workspace_name }} - {{ outputs.audit_clusters.vars.offender_count }} of {{ outputs.audit_clusters.vars.cluster_count }} cluster(s) violate policy, an estimated {{ outputs.audit_clusters.vars.wasted_hours }} compute hours idle outside work hours. Offenders {{ outputs.audit_clusters.vars.offenders | map(attribute='name') | join(', ') }}. Execution {{ execution.id }}."
          }
      - id: enforce_audit
        type: io.kestra.plugin.core.flow.If
        description: Apply the fail_on_violations policy, turning the run red for
          governance pipelines and staying report only when the input is flipped
          off.
        condition: "{{ inputs.fail_on_violations }}"
        then:
          - id: fail_audit
            type: io.kestra.plugin.core.execution.Fail
            description: Fail the execution so the governance pipeline stops instead of
              accepting another night of idle clusters.
            errorMessage: "Databricks audit found {{
              outputs.audit_clusters.vars.offender_count }} cluster(s) violating
              auto termination policy in {{ inputs.workspace_name }}, an
              estimated {{ outputs.audit_clusters.vars.wasted_hours }} compute
              hours wasted."
        else:
          - id: log_audit_softened
            type: io.kestra.plugin.core.log.Log
            description: Record that violations exist but the fail policy is off, keeping
              the run green for report only estates.
            message: "Databricks audit violations in {{ inputs.workspace_name }} not
              enforced - fail_on_violations is false, {{
              outputs.audit_clusters.vars.offender_count }} offender(s) alerted
              only."
    else:
      - id: log_audit_clean
        type: io.kestra.plugin.core.log.Log
        description: Record the clean or unreadable outcome so a compliant workspace
          still leaves evidence the audit ran.
        message: "Databricks auto termination audit for {{ inputs.workspace_name }}
          clean - {{ outputs.audit_clusters.vars.cluster_count }} cluster(s)
          checked, {{ outputs.audit_clusters.vars.offender_count }}
          violation(s), policy {{ inputs.idle_policy_minutes }} minutes."

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalate any failure to the ops channel with the execution url.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED: {{ flow.id }} {{ execution.id }}"}

outputs:
  - id: idle_clusters
    type: ARRAY
    description: Rows for every violating cluster with id, name, state, timeout,
      idle minutes, reason and wasted compute hours.
    value: "{{ outputs.audit_clusters.vars.offenders }}"
  - id: offender_count
    type: INT
    description: Number of running clusters with auto termination disabled or idle
      time past the policy, for example 3 on the bundled inventory.
    value: "{{ outputs.audit_clusters.vars.offender_count }}"
  - id: wasted_hours
    type: FLOAT
    description: Estimated idle compute hours across every violating cluster in the
      window, for example 7.33 on the bundled inventory.
    value: "{{ outputs.audit_clusters.vars.wasted_hours }}"
  - id: cluster_count
    type: INT
    description: Clusters present in the scored inventory regardless of state, the
      denominator behind the violation share.
    value: "{{ outputs.audit_clusters.vars.cluster_count }}"
  - id: audit_ok
    type: BOOL
    description: True when the inventory parsed and was scored, false when the
      posted body could not be read.
    value: "{{ outputs.audit_clusters.vars.audit_ok }}"