Snowflake Warehouse Credit Usage and FinOps Cost Guard

id: snowflake-warehouse-query-acceleration-cost-guard
namespace: company.analytics

description: |
  Audits Snowflake virtual warehouse query acceleration service (QAS) credit
  consumption and query queue overload, flags unexpected warehouse credit surges,
  creates an automated GitHub FinOps audit ticket, and alerts the data engineering team on Slack.

inputs:
  - id: warehouse_name
    type: STRING
    displayName: Snowflake Warehouse Name
    defaults: COMPUTE_WH

  - id: credit_threshold
    type: FLOAT
    displayName: Hourly Credit Ceiling Threshold
    defaults: 15.0

  - id: github_repository
    type: STRING
    displayName: FinOps GitHub Tracking Repository
    defaults: my-org/snowflake-cost-governance

triggers:
  - id: hourly_warehouse_governance_check
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 * * * *"
    disabled: true

tasks:
  - id: probe_warehouse_credit_usage
    type: io.kestra.plugin.jdbc.snowflake.Query
    description: Interrogates Snowflake ACCOUNT_USAGE for recent hourly warehouse
      credit consumption
    url: "jdbc:snowflake://{{ secret('SNOWFLAKE_ACCOUNT')
      }}.snowflakecomputing.com/?warehouse={{ inputs.warehouse_name
      }}&db=SNOWFLAKE&schema=ACCOUNT_USAGE"
    username: "{{ secret('SNOWFLAKE_USERNAME') }}"
    password: "{{ secret('SNOWFLAKE_PASSWORD') }}"
    sql: |
      SELECT 
        WAREHOUSE_NAME,
        COALESCE(SUM(CREDITS_USED), 0) AS TOTAL_CREDITS,
        COALESCE(SUM(CREDITS_USED_COMPUTE), 0) AS COMPUTE_CREDITS,
        COALESCE(SUM(CREDITS_USED_CLOUD_SERVICES), 0) AS CLOUD_SERVICES_CREDITS
      FROM SNOWFLAKE.ACCOUNT_USAGE.WAREHOUSE_METERING_HISTORY
      WHERE START_TIME >= DATEADD('hour', -1, CURRENT_TIMESTAMP())
        AND WAREHOUSE_NAME = '{{ inputs.warehouse_name }}'
      GROUP BY WAREHOUSE_NAME;
    fetchOne: true

  - id: evaluate_credit_surge
    type: io.kestra.plugin.core.debug.Return
    description: Evaluates whether warehouse credit consumption breached the hourly ceiling
    format:
      warehouse: "{{ inputs.warehouse_name }}"
      total_credits: "{{ outputs.probe_warehouse_credit_usage.row.TOTAL_CREDITS ?? 0 }}"
      compute_credits: "{{ outputs.probe_warehouse_credit_usage.row.COMPUTE_CREDITS ?? 0 }}"
      cloud_credits: "{{
        outputs.probe_warehouse_credit_usage.row.CLOUD_SERVICES_CREDITS ?? 0 }}"
      threshold: "{{ inputs.credit_threshold }}"
      is_surge_detected: "{{ (outputs.probe_warehouse_credit_usage.row.TOTAL_CREDITS
        ?? 0) > inputs.credit_threshold }}"

  - id: branch_on_credit_surge
    type: io.kestra.plugin.core.flow.If
    description: Branches based on whether warehouse credit consumption exceeds cost limits
    condition: "{{ outputs.evaluate_credit_surge.value.is_surge_detected == true }}"
    then:
      - id: create_finops_github_issue
        type: io.kestra.plugin.github.issues.Create
        description: Logs a prioritized GitHub issue tracking warehouse cost overruns
        oauthToken: "{{ secret('GITHUB_TOKEN') }}"
        repository: "{{ inputs.github_repository }}"
        title: "[FinOps Alert] Snowflake Credit Surge Detected on {{
          inputs.warehouse_name }}"
        body: |
          ## 🚨 Snowflake Warehouse Credit Overrun Warning

          - **Audited Warehouse:** `{{ inputs.warehouse_name }}`
          - **Total Credits (Last Hour):** `{{ outputs.evaluate_credit_surge.value.total_credits }}`
          - **Compute Credits:** `{{ outputs.evaluate_credit_surge.value.compute_credits }}`
          - **Cloud Services Credits:** `{{ outputs.evaluate_credit_surge.value.cloud_credits }}`
          - **Hourly Budget Ceiling:** `{{ inputs.credit_threshold }}`

          Please inspect runaway ad-hoc queries, unoptimized cartesian joins, or autocommit transaction loops.
          *Generated by Kestra Snowflake FinOps Guard.*

      - id: alert_finops_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-priority alert to the data platform Slack channel
        url: "{{ secret('SLACK_WEBHOOK') }}"
        payload: |
          {{ {
            "text": "💰 *Snowflake Warehouse Credit Surge Alert*\n*Warehouse:* `" ~ inputs.warehouse_name ~ "`\n*Credits Used (1h):* *" ~ outputs.evaluate_credit_surge.value.total_credits ~ "* (Ceiling: `" ~ inputs.credit_threshold ~ "`)\n*FinOps Ticket:* <" ~ (outputs.create_finops_github_issue.issueUrl ?? "#") ~ "|View GitHub Issue>"
          } | toJson }}

    else:
      - id: log_warehouse_compliant
        type: io.kestra.plugin.core.log.Log
        message: "Snowflake warehouse {{ inputs.warehouse_name }} credit usage is within
          budget ({{ outputs.evaluate_credit_surge.value.total_credits }} <= {{
          inputs.credit_threshold }})."

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: Final status log recording the credit audit outcome
    message: "Snowflake cost governance audit completed for {{ inputs.warehouse_name
      }}. Credits consumed: {{ outputs.evaluate_credit_surge.value.total_credits
      }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alerts Slack when Snowflake queries or issue logging fails
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {{ {
        "text": "⚠️ *Snowflake FinOps Sentinel Failed* in execution `" ~ execution.id ~ "`. Unable to query ACCOUNT_USAGE views."
      } | toJson }}