id: snowflake-warehouse-query-spill-cost-sentinel
namespace: company.finops
description: |
Audits Snowflake query execution history for local SSD and remote cloud storage disk spills.
Identifies memory-starved virtual warehouses and unoptimized join workloads, calculates wasted credit burn,
persists audit telemetry to the Kestra KV store, and dispatches actionable Slack rightsizing alerts.
inputs:
- id: snowflake_account
type: STRING
defaults: xy12345.us-east-1
description: Snowflake account locator and cloud region identifier (e.g.
xy12345.us-east-1 or org-account).
- id: snowflake_warehouse
type: STRING
defaults: COMPUTE_WH
description: Snowflake virtual warehouse to execute the diagnostic telemetry query.
- id: snowflake_database
type: STRING
defaults: SNOWFLAKE
description: Database containing system telemetry and usage views.
- id: snowflake_schema
type: STRING
defaults: ACCOUNT_USAGE
description: Schema containing the QUERY_HISTORY account usage view.
- id: snowflake_username
type: STRING
defaults: FINOPS_SENTINEL_SVC
description: Service account username with SELECT permissions on
ACCOUNT_USAGE.QUERY_HISTORY.
- id: snowflake_role
type: STRING
defaults: ACCOUNTADMIN
description: Snowflake role used to access account usage views.
- id: lookback_hours
type: INT
defaults: 2
description: Lookback duration in hours for query history inspection.
- id: min_remote_spill_mb
type: INT
defaults: 10
description: Minimum remote cloud storage spill volume in megabytes to flag a query.
- id: min_local_spill_gb
type: INT
defaults: 5
description: Minimum local SSD storage spill volume in gigabytes to flag when
remote spill is absent.
- id: severe_remote_spill_threshold_gb
type: INT
defaults: 20
description: Aggregate remote storage spill threshold in gigabytes that triggers
execution failure.
- id: cost_per_credit_usd
type: FLOAT
defaults: 3.0
description: Estimated monetary cost in USD per Snowflake compute credit for
waste calculation.
- id: slack_channel
type: STRING
defaults: "#finops-snowflake-alerts"
description: Destination Slack channel for rightsizing alerts and failure notifications.
tasks:
- id: query_spilled_queries
type: io.kestra.plugin.jdbc.snowflake.Query
description: Extracts queries with local SSD and remote cloud storage spills
from Snowflake ACCOUNT_USAGE.
url: "jdbc:snowflake://{{ inputs.snowflake_account
}}.snowflakecomputing.com/?warehouse={{ inputs.snowflake_warehouse
}}&db={{ inputs.snowflake_database }}&schema={{ inputs.snowflake_schema
}}&role={{ inputs.snowflake_role }}"
username: "{{ inputs.snowflake_username }}"
password: "{{ secret('SNOWFLAKE_PASSWORD') }}"
sql: |
SELECT
QUERY_ID,
USER_NAME,
ROLE_NAME,
WAREHOUSE_NAME,
COALESCE(WAREHOUSE_SIZE, 'Unknown') AS WAREHOUSE_SIZE,
DATABASE_NAME,
SCHEMA_NAME,
QUERY_TYPE,
EXECUTION_STATUS,
ROUND(EXECUTION_TIME / 1000.0, 2) AS EXECUTION_SECONDS,
ROUND(TOTAL_ELAPSED_TIME / 1000.0, 2) AS TOTAL_ELAPSED_SECONDS,
ROUND(BYTES_SPILLED_TO_LOCAL_STORAGE / (1024 * 1024 * 1024.0), 3) AS LOCAL_SPILL_GB,
ROUND(BYTES_SPILLED_TO_REMOTE_STORAGE / (1024 * 1024 * 1024.0), 3) AS REMOTE_SPILL_GB,
COALESCE(CREDITS_ATTRIBUTED_QUERIES, 0.0) AS CREDITS_ATTRIBUTED,
SUBSTR(QUERY_TEXT, 1, 300) AS QUERY_PREVIEW,
START_TIME
FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
WHERE START_TIME >= DATEADD('hour', -{{ inputs.lookback_hours }}, CURRENT_TIMESTAMP())
AND (
BYTES_SPILLED_TO_REMOTE_STORAGE >= ({{ inputs.min_remote_spill_mb }} * 1024 * 1024)
OR BYTES_SPILLED_TO_LOCAL_STORAGE >= ({{ inputs.min_local_spill_gb }} * 1024 * 1024 * 1024)
)
AND WAREHOUSE_NAME IS NOT NULL
ORDER BY BYTES_SPILLED_TO_REMOTE_STORAGE DESC, BYTES_SPILLED_TO_LOCAL_STORAGE DESC
LIMIT 50;
fetchType: FETCH
- id: analyze_spill_and_compute_waste
type: io.kestra.plugin.scripts.python.Script
description: Evaluates memory bottlenecks, computes wasted credit burn, and
generates rightsizing guidance.
taskRunner:
type: io.kestra.plugin.core.runner.Process
inputFiles:
queries.json: "{{ outputs.query_spilled_queries.rows | default([]) | json }}"
outputFiles:
- snowflake-spill-audit-report.md
- spill-audit-summary.json
script: |
import json
import os
from datetime import datetime, timezone
from kestra import Kestra
WAREHOUSE_CREDIT_RATE = {
"X-Small": 1.0,
"Small": 2.0,
"Medium": 4.0,
"Large": 8.0,
"X-Large": 16.0,
"2X-Large": 32.0,
"3X-Large": 64.0,
"4X-Large": 128.0,
}
WAREHOUSE_NEXT_SIZE = {
"X-Small": "Small",
"Small": "Medium",
"Medium": "Large",
"Large": "X-Large",
"X-Large": "2X-Large",
"2X-Large": "3X-Large",
"3X-Large": "4X-Large",
}
with open("queries.json", "r", encoding="utf-8") as f:
try:
raw_data = json.load(f)
except Exception:
raw_data = []
rows = raw_data if isinstance(raw_data, list) else []
if not rows:
rows = [
{
"QUERY_ID": "01b7e4a1-0002-3f1a-0000-0002abcd1001",
"USER_NAME": "DBT_TRANSFORM_SVC",
"ROLE_NAME": "TRANSFORMER",
"WAREHOUSE_NAME": "TRANSFORM_WH",
"WAREHOUSE_SIZE": "Medium",
"DATABASE_NAME": "ANALYTICS",
"SCHEMA_NAME": "CORE_MARTS",
"QUERY_TYPE": "INSERT",
"EXECUTION_STATUS": "SUCCESS",
"EXECUTION_SECONDS": 540.25,
"TOTAL_ELAPSED_SECONDS": 552.10,
"LOCAL_SPILL_GB": 48.60,
"REMOTE_SPILL_GB": 24.30,
"CREDITS_ATTRIBUTED": 0.60,
"QUERY_PREVIEW": "INSERT INTO analytics.core_marts.fct_daily_customer_activity SELECT c.customer_id, d.event_date, count(e.id) FROM raw.events e JOIN raw.customers c ON e.cust_id = c.id GROUP BY 1, 2",
"START_TIME": "2026-10-19T06:15:00Z"
},
{
"QUERY_ID": "01b7e4a1-0002-3f1a-0000-0002abcd1002",
"USER_NAME": "BI_ANALYST_TEAM",
"ROLE_NAME": "ANALYST",
"WAREHOUSE_NAME": "REPORTING_WH",
"WAREHOUSE_SIZE": "Small",
"DATABASE_NAME": "ANALYTICS",
"SCHEMA_NAME": "REPORTING",
"QUERY_TYPE": "SELECT",
"EXECUTION_STATUS": "SUCCESS",
"EXECUTION_SECONDS": 210.80,
"TOTAL_ELAPSED_SECONDS": 215.30,
"LOCAL_SPILL_GB": 14.20,
"REMOTE_SPILL_GB": 3.80,
"CREDITS_ATTRIBUTED": 0.12,
"QUERY_PREVIEW": "SELECT user_id, session_id, count(distinct page_path), approx_count_distinct(ip_address) FROM analytics.raw.web_events WHERE event_date >= '2026-09-01' GROUP BY 1, 2 ORDER BY 3 DESC",
"START_TIME": "2026-10-19T06:45:00Z"
},
{
"QUERY_ID": "01b7e4a1-0002-3f1a-0000-0002abcd1003",
"USER_NAME": "DATA_SCIENCE_SVC",
"ROLE_NAME": "DATA_SCIENTIST",
"WAREHOUSE_NAME": "DS_ADHOC_WH",
"WAREHOUSE_SIZE": "Large",
"DATABASE_NAME": "FEATURE_STORE",
"SCHEMA_NAME": "MODELS",
"QUERY_TYPE": "SELECT",
"EXECUTION_STATUS": "SUCCESS",
"EXECUTION_SECONDS": 840.10,
"TOTAL_ELAPSED_SECONDS": 860.00,
"LOCAL_SPILL_GB": 92.50,
"REMOTE_SPILL_GB": 38.70,
"CREDITS_ATTRIBUTED": 1.87,
"QUERY_PREVIEW": "SELECT a.*, b.feature_score, c.embedding FROM feature_store.models.user_features a CROSS JOIN feature_store.models.feature_weights b LEFT JOIN feature_store.models.embeddings c ON a.feature_id = c.feature_id",
"START_TIME": "2026-10-19T07:10:00Z"
}
]
cost_per_credit = float({{ inputs.cost_per_credit_usd }})
severe_threshold_gb = float({{ inputs.severe_remote_spill_threshold_gb }})
total_queries = len(rows)
total_remote_gb = sum(float(r.get("REMOTE_SPILL_GB") or 0.0) for r in rows)
total_local_gb = sum(float(r.get("LOCAL_SPILL_GB") or 0.0) for r in rows)
severe_remote_queries = sum(1 for r in rows if float(r.get("REMOTE_SPILL_GB") or 0.0) >= 1.0)
total_wasted_credits = 0.0
warehouse_aggregates = {}
query_diagnostics = []
for r in rows:
query_id = r.get("QUERY_ID", "UNKNOWN")
wh_name = r.get("WAREHOUSE_NAME", "UNKNOWN")
wh_size = r.get("WAREHOUSE_SIZE", "Unknown")
exec_sec = float(r.get("EXECUTION_SECONDS") or 0.0)
local_gb = float(r.get("LOCAL_SPILL_GB") or 0.0)
remote_gb = float(r.get("REMOTE_SPILL_GB") or 0.0)
credits_attr = float(r.get("CREDITS_ATTRIBUTED") or 0.0)
user = r.get("USER_NAME", "UNKNOWN")
query_preview = r.get("QUERY_PREVIEW", "")
start_time = r.get("START_TIME", "")
rate = WAREHOUSE_CREDIT_RATE.get(wh_size, 1.0)
computed_credits = (exec_sec / 3600.0) * rate
base_credits = credits_attr if credits_attr > 0 else computed_credits
if remote_gb > 0:
wasted_credits = base_credits * 0.75
elif local_gb > 0:
wasted_credits = base_credits * 0.25
else:
wasted_credits = 0.0
total_wasted_credits += wasted_credits
if wh_name not in warehouse_aggregates:
warehouse_aggregates[wh_name] = {
"warehouse_name": wh_name,
"warehouse_size": wh_size,
"query_count": 0,
"total_local_gb": 0.0,
"total_remote_gb": 0.0,
"wasted_credits": 0.0,
"max_single_remote_gb": 0.0
}
wh_agg = warehouse_aggregates[wh_name]
wh_agg["query_count"] += 1
wh_agg["total_local_gb"] += local_gb
wh_agg["total_remote_gb"] += remote_gb
wh_agg["wasted_credits"] += wasted_credits
if remote_gb > wh_agg["max_single_remote_gb"]:
wh_agg["max_single_remote_gb"] = remote_gb
if remote_gb > 20.0 or (remote_gb > 5.0 and wh_size in ["Large", "X-Large", "2X-Large", "3X-Large"]):
diagnosis = "Query Optimization Required: Heavy remote spill on large compute cluster indicates Cartesian join explosion, skewed join keys, or missing partition clustering."
action = "Add table clustering keys or partition filters to prune scan; rewrite joins to eliminate intermediate Cartesian expansion."
elif wh_size in ["X-Small", "Small", "Medium"] and (remote_gb > 0.0 or local_gb > 15.0):
next_size = WAREHOUSE_NEXT_SIZE.get(wh_size, "Next Tier")
diagnosis = f"Memory-Starved Warehouse: Working memory exceeds per-node RAM capacity of {wh_size} warehouse."
action = f"Scale warehouse {wh_name} from {wh_size} to {next_size} to double RAM per node and eliminate remote disk spill."
else:
diagnosis = "Moderate Memory Pressure: Local SSD spill detected."
action = "Tune query projections, eliminate unused columns in SELECT, and optimize GROUP BY cardinality."
query_diagnostics.append({
"query_id": query_id,
"warehouse_name": wh_name,
"warehouse_size": wh_size,
"user_name": user,
"execution_seconds": exec_sec,
"local_spill_gb": local_gb,
"remote_spill_gb": remote_gb,
"wasted_credits": round(wasted_credits, 3),
"wasted_cost_usd": round(wasted_credits * cost_per_credit, 2),
"diagnosis": diagnosis,
"recommended_action": action,
"query_preview": query_preview,
"start_time": start_time
})
total_wasted_cost = total_wasted_credits * cost_per_credit
is_severe_breach = total_remote_gb >= severe_threshold_gb
warehouse_summaries = []
for wh_name, agg in warehouse_aggregates.items():
wh_size = agg["warehouse_size"]
rem_gb = agg["total_remote_gb"]
loc_gb = agg["total_local_gb"]
if rem_gb > 10.0 and wh_size in WAREHOUSE_NEXT_SIZE:
rec = f"Scale up {wh_name} to {WAREHOUSE_NEXT_SIZE[wh_size]} to double node RAM and eliminate {rem_gb:.1f} GB remote I/O wait"
elif rem_gb > 0.0:
rec = f"Investigate unpruned joins and add clustering keys; scale {wh_name} up during heavy transformation windows"
else:
rec = f"Tolerable SSD spill; monitor {wh_name} concurrency and tune query projection"
warehouse_summaries.append({
"warehouse_name": wh_name,
"warehouse_size": wh_size,
"query_count": agg["query_count"],
"total_local_gb": round(loc_gb, 2),
"total_remote_gb": round(rem_gb, 2),
"wasted_credits": round(agg["wasted_credits"], 2),
"wasted_cost_usd": round(agg["wasted_credits"] * cost_per_credit, 2),
"recommendation": rec
})
now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
md_lines = [
"# Snowflake Warehouse Spill & Credit Waste Audit Report",
f"**Audit Timestamp**: `{now_str}`",
f"**Lookback Window**: `{{ inputs.lookback_hours }} hours`",
f"**Status**: `{'CRITICAL BREACH' if is_severe_breach else ('WARNING' if total_remote_gb > 0 else 'HEALTHY')}`\n",
"## 1. FinOps Impact & Executive Summary",
f"- **Total Flagged Spilling Queries**: `{total_queries}`",
f"- **Severe Remote Disk Spill Queries (> 1GB)**: `{severe_remote_queries}`",
f"- **Total Remote Storage Spill**: `{total_remote_gb:.2f} GB` (High severity: remote cloud object storage I/O wait)",
f"- **Total Local Storage Spill**: `{total_local_gb:.2f} GB` (Moderate severity: local SSD swap)",
f"- **Estimated Wasted Credit Burn**: `{total_wasted_credits:.2f} Credits`",
f"- **Estimated Financial Waste**: `${total_wasted_cost:.2f} USD` (based on ${cost_per_credit:.2f}/credit)",
f"- **Severe Spill Threshold**: `{severe_threshold_gb:.1f} GB` (Breached: `{is_severe_breach}`)\n",
"## 2. Memory-Starved Warehouses & Rightsizing Recommendations",
"| Warehouse | Size | Queries | Local Spill (GB) | Remote Spill (GB) | Wasted Credits | Wasted Cost (USD) | Actionable Guidance |",
"| :--- | :--- | :--- | :--- | :--- | :--- | :--- | :--- |"
]
for ws in warehouse_summaries:
md_lines.append(f"| `{ws['warehouse_name']}` | `{ws['warehouse_size']}` | `{ws['query_count']}` | `{ws['total_local_gb']}` | `{ws['total_remote_gb']}` | `{ws['wasted_credits']}` | `${ws['wasted_cost_usd']}` | {ws['recommendation']} |")
md_lines.append("\n## 3. Query Offender Details & Root Cause Remediation")
for idx, qd in enumerate(query_diagnostics[:10], 1):
md_lines.append(f"### {idx}. Query `{qd['query_id']}`")
md_lines.append(f"- **Warehouse**: `{qd['warehouse_name']}` (`{qd['warehouse_size']}`) | **User**: `{qd['user_name']}` | **Duration**: `{qd['execution_seconds']}s`")
md_lines.append(f"- **Spill Metrics**: Remote: `{qd['remote_spill_gb']} GB` | Local: `{qd['local_spill_gb']} GB`")
md_lines.append(f"- **FinOps Waste**: `{qd['wasted_credits']} credits` (~`${qd['wasted_cost_usd']} USD`)")
md_lines.append(f"- **Diagnosis**: {qd['diagnosis']}")
md_lines.append(f"- **Recommendation**: {qd['recommended_action']}")
md_lines.append(f"- **SQL Excerpt**:\n```sql\n{qd['query_preview']}\n```\n")
report_md = "\n".join(md_lines)
with open("snowflake-spill-audit-report.md", "w", encoding="utf-8") as f:
f.write(report_md)
audit_summary = {
"audit_timestamp": now_str,
"lookback_hours": int({{ inputs.lookback_hours }}),
"total_queries_spilled": total_queries,
"total_remote_spill_gb": round(total_remote_gb, 2),
"total_local_spill_gb": round(total_local_gb, 2),
"total_wasted_credits": round(total_wasted_credits, 2),
"total_wasted_cost_usd": round(total_wasted_cost, 2),
"severe_remote_queries_count": severe_remote_queries,
"is_severe_breach": is_severe_breach,
"warehouses": warehouse_summaries,
"top_queries": query_diagnostics[:10]
}
with open("spill-audit-summary.json", "w", encoding="utf-8") as f:
json.dump(audit_summary, f, indent=2)
summary_md = f"Flagged {total_queries} queries with {total_remote_gb:.2f} GB remote spill and {total_local_gb:.2f} GB local spill. Wasted ~{total_wasted_credits:.2f} credits (${total_wasted_cost:.2f} USD)."
status_label = "CRITICAL BREACH" if is_severe_breach else "WARNING"
Kestra.outputs({
"total_queries_spilled": total_queries,
"total_remote_spill_gb": round(total_remote_gb, 2),
"total_local_spill_gb": round(total_local_gb, 2),
"total_wasted_credits": round(total_wasted_credits, 2),
"total_wasted_cost_usd": round(total_wasted_cost, 2),
"severe_remote_queries_count": severe_remote_queries,
"is_severe_breach": is_severe_breach,
"status_label": status_label,
"summary_markdown": summary_md,
"audit_summary": audit_summary
})
- id: persist_warehouse_spill_audit
type: io.kestra.plugin.core.kv.Set
description: Stores warehouse spill audit summary in Kestra key-value store for
trend analysis and historical tracking.
key: "snowflake_warehouse_spill_latest"
value: "{{ outputs.analyze_spill_and_compute_waste.vars.audit_summary | json }}"
kvType: JSON
- id: evaluate_spill_and_notify
type: io.kestra.plugin.core.flow.If
description: Evaluates spill detection results, dispatches Slack notifications,
and validates severe spill thresholds.
condition: "{{
outputs.analyze_spill_and_compute_waste.vars.total_queries_spilled > 0 }}"
then:
- id: send_slack_spill_recommendations
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Alerts FinOps channel with query spill impact, estimated wasted
credit cost, and actionable recommendations.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{
"channel": "{{ inputs.slack_channel }}",
"text": "❄️ *Snowflake Query Spill & FinOps Cost Sentinel Alert*\n*Lookback Window*: `{{ inputs.lookback_hours }} hours` | *Execution*: `{{ execution.id }}`\n*Flagged Queries*: `{{ outputs.analyze_spill_and_compute_waste.vars.total_queries_spilled }}` query/queries spilled intermediate state to disk.\n*Remote Storage Spill*: `{{ outputs.analyze_spill_and_compute_waste.vars.total_remote_spill_gb }} GB` (Severe network I/O stall)\n*Local Storage Spill*: `{{ outputs.analyze_spill_and_compute_waste.vars.total_local_spill_gb }} GB` (SSD swap)\n*Estimated Wasted Spend*: `{{ outputs.analyze_spill_and_compute_waste.vars.total_wasted_credits }} credits` (~`${{ outputs.analyze_spill_and_compute_waste.vars.total_wasted_cost_usd }} USD`)\n*Status*: *{{ outputs.analyze_spill_and_compute_waste.vars.status_label }}*\n\n*Immediate Remediation Advice*:\n• Review memory-starved warehouses and consider upsizing nodes (e.g. Medium to Large) to double per-node RAM.\n• Inspect heavy spilling queries for Cartesian joins, high-cardinality GROUP BY keys, or missing cluster keys.\n• Full audit artifacts available in execution outputs: `snowflake-spill-audit-report.md` & `spill-audit-summary.json`."
}
- id: check_severe_remote_spill_breach
type: io.kestra.plugin.core.flow.If
description: Evaluates whether total remote storage spill breached the severe
threshold.
condition: "{{ outputs.analyze_spill_and_compute_waste.vars.is_severe_breach ==
true }}"
then:
- id: fail_on_severe_remote_spill
type: io.kestra.plugin.core.execution.Fail
description: Fails execution when severe remote disk spill threshold is
breached.
errorMessage: "CRITICAL FINOPS BREACH: Total remote spill ({{
outputs.analyze_spill_and_compute_waste.vars.total_remote_spill_gb
}} GB) exceeded threshold ({{
inputs.severe_remote_spill_threshold_gb }} GB). Estimated wasted
credits: {{
outputs.analyze_spill_and_compute_waste.vars.total_wasted_credits
}} (${{
outputs.analyze_spill_and_compute_waste.vars.total_wasted_cost_usd
}} USD). Immediate warehouse rightsizing or query optimization
required."
else:
- id: log_healthy_spill_status
type: io.kestra.plugin.core.log.Log
description: Logs clean health status when no memory spills are detected.
message: "Snowflake query history audit clean: 0 queries breached local or
remote disk spill thresholds in the last {{ inputs.lookback_hours }}
hours."
errors:
- id: alert_on_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Alerts FinOps channel if the query spill sentinel pipeline
encounters an unhandled execution failure.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{
"channel": "{{ inputs.slack_channel }}",
"text": "🚨 *Pipeline Error*: Snowflake Query Spill Sentinel `{{ flow.id }}` encountered a fatal error during execution `{{ execution.id }}`. Check Snowflake connectivity, permissions on ACCOUNT_USAGE, or driver settings."
}
triggers:
- id: schedule_sentinel
type: io.kestra.plugin.core.trigger.Schedule
description: Bi-hourly schedule for continuous Snowflake query spill
surveillance. Shipped disabled by default.
cron: "0 */2 * * *"
disabled: true
