Snowflake Cost Anomaly Detector and FinOps Alerting

id: snowflake-cost-anomaly-detector
namespace: company.team
description: |
  Automated FinOps monitoring pipeline for Snowflake data warehouses. Tracks daily credit
  consumption across all compute warehouses, computes rolling 7-day baselines, detects anomalous
  cost spikes, and triggers Slack alerts to prevent unexpected cloud bills.

concurrency:
  limit: 1

inputs:
  - id: alert_threshold_percent
    type: INT
    displayName: Cost Spike Alert Threshold (%)
    description: Percentage increase over rolling baseline that flags an anomaly.
    defaults: 25

  - id: lookback_days
    type: INT
    displayName: Baseline Lookback Window (Days)
    description: Number of days to include when calculating normal historical consumption.
    defaults: 7

  - id: estimated_credit_price_usd
    type: FLOAT
    displayName: Estimated Price per Credit (USD)
    description: Average dollar cost per Snowflake credit for financial impact estimation.
    defaults: 3.00

  - id: slack_channel
    type: STRING
    displayName: FinOps Slack Channel
    description: Destination channel for cost spike alerts.
    defaults: "#finops-alerts"

triggers:
  - id: daily_finops_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Daily FinOps spend audit executed at 07:00 UTC. Can also be
      triggered manually.
    cron: "0 7 * * *"

tasks:
  - id: calculate_baseline_and_spikes
    type: io.kestra.plugin.jdbc.snowflake.Query
    description: Query Snowflake Account Usage metering history to evaluate
      warehouse spend trends.
    url: "{{ secret('SNOWFLAKE_JDBC_URL') }}"
    username: "{{ secret('SNOWFLAKE_USER') }}"
    password: "{{ secret('SNOWFLAKE_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      WITH daily_warehouse_spend AS (
        SELECT
          warehouse_name,
          DATE_TRUNC('day', start_time) AS usage_date,
          ROUND(SUM(credits_used), 2) AS credits_consumed
        FROM snowflake.account_usage.warehouse_metering_history
        WHERE start_time >= DATEADD(day, -{{ inputs.lookback_days }}, CURRENT_DATE())
        GROUP BY warehouse_name, usage_date
      ),
      baseline_metrics AS (
        SELECT
          warehouse_name,
          AVG(CASE WHEN usage_date < CURRENT_DATE() - 1 THEN credits_consumed END) AS avg_historical_credits,
          MAX(CASE WHEN usage_date = CURRENT_DATE() - 1 THEN credits_consumed ELSE 0 END) AS yesterday_credits
        FROM daily_warehouse_spend
        GROUP BY warehouse_name
      )
      SELECT
        warehouse_name,
        ROUND(yesterday_credits, 2) AS credits_yesterday,
        ROUND(COALESCE(avg_historical_credits, 0), 2) AS baseline_credits,
        ROUND(yesterday_credits * {{ inputs.estimated_credit_price_usd }}, 2) AS estimated_cost_usd,
        CASE
          WHEN COALESCE(avg_historical_credits, 0) > 0 THEN
            ROUND(((yesterday_credits - avg_historical_credits) / avg_historical_credits) * 100, 1)
          ELSE 0
        END AS spike_percentage
      FROM baseline_metrics
      WHERE yesterday_credits > 0
      ORDER BY spike_percentage DESC;

  - id: evaluate_spend_anomalies
    type: io.kestra.plugin.core.flow.If
    description: Branch execution when any warehouse exceeds the allowed spike threshold.
    condition: "{{ outputs.calculate_baseline_and_spikes.rows[0].spike_percentage >=
      inputs.alert_threshold_percent }}"
    then:
      - id: notify_finops_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Broadcast critical cost anomaly alert with impact figures to Slack.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          *Snowflake Cost Anomaly Detected* 🚨
          • *Offending Warehouse*: `{{ outputs.calculate_baseline_and_spikes.rows[0].warehouse_name }}`
          • *Credits Consumed (Yesterday)*: `{{ outputs.calculate_baseline_and_spikes.rows[0].credits_yesterday }}` credits
          • *7-Day Baseline*: `{{ outputs.calculate_baseline_and_spikes.rows[0].baseline_credits }}` credits
          • *Spike*: `+{{ outputs.calculate_baseline_and_spikes.rows[0].spike_percentage }}%` above baseline
          • *Estimated Spend*: `${{ outputs.calculate_baseline_and_spikes.rows[0].estimated_cost_usd }}`
          • *Action Required*: Inspect Snowflake Query History for unoptimized Cartesian joins or runaway clustering tasks.

    else:
      - id: log_healthy_spend
        type: io.kestra.plugin.core.log.Log
        description: Confirm warehouse consumption is healthy and within budget limits.
        message: "Snowflake spend within normal bounds. Top warehouse: {{
          outputs.calculate_baseline_and_spikes.rows[0].warehouse_name }} (+{{
          outputs.calculate_baseline_and_spikes.rows[0].spike_percentage }}% vs
          baseline)."

  - id: export_finops_audit
    type: io.kestra.plugin.core.debug.Return
    description: Output structured execution audit for historical tracking and
      compliance records.
    format: |
      {
        "pipeline": "snowflake-cost-anomaly-detector",
        "audit_timestamp": "{{ execution.startDate }}",
        "top_warehouse": "{{ outputs.calculate_baseline_and_spikes.rows[0].warehouse_name }}",
        "spike_percentage": {{ outputs.calculate_baseline_and_spikes.rows[0].spike_percentage }},
        "anomaly_flagged": {{ outputs.calculate_baseline_and_spikes.rows[0].spike_percentage >= inputs.alert_threshold_percent }}
      }