Snowflake Task Failure and Stream Staleness Sentinel

In Snowflake architectures, Tasks and Streams form the backbone of native Change Data Capture (CDC) and continuous ELT pipelines. Streams track row-level delta mutations (inserts, updates, deletes) on source tables using Snowflake Time Travel, while Tasks schedule SQL statements or stored procedures to consume those deltas downstream.

However, two failure modes silently plague production pipelines:

1. Task Execution Failures: When a task encounters a syntax error, warehouse exhaustion, permission issue, or deadlock, its state becomes FAILED in SNOWFLAKE.ACCOUNT_USAGE.TASK_HISTORY. After multiple consecutive failures, Snowflake automatically suspends dependent tasks in a task graph, stalling the pipeline.
2. Stream Staleness Disasters: If a stream is not consumed within the underlying table's Time Travel data retention period (defined by DATA_RETENTION_TIME_IN_DAYS, typically 1 to 90 days), the stream becomes stale (stale = true). Once a stream becomes stale, Snowflake permanently locks it from being read or advancing offsets, requiring manual intervention to drop and recreate the stream and potentially triggering costly full-table reconciliations.

This blueprint establishes an automated operational health gate for Snowflake Tasks and Streams. Executing on an hourly schedule or on-demand, it queries SNOWFLAKE.ACCOUNT_USAGE.TASK_HISTORY and SNOWFLAKE.ACCOUNT_USAGE.STREAMS to identify failed task executions and stale CDC streams. When an anomaly is detected, it dispatches an actionable incident triage card to Slack with root-cause error messages and remediation commands.

How it works

1. Scheduled Health Audit: The hourly_snowflake_health_audit trigger (io.kestra.plugin.core.trigger.Schedule) executes hourly. Shipped disabled: true by default so manual verification can occur first.
2. Task Failure Audit: audit_failed_tasks (io.kestra.plugin.jdbc.snowflake.Query) queries SNOWFLAKE.ACCOUNT_USAGE.TASK_HISTORY over the configured lookback window (default 24 hours) for tasks with state = 'FAILED'. It uses COUNT(*) OVER() to preserve the true account-wide incident count while limiting detail rows to 50. Aliases are double-quoted to guarantee exact row-key casing.
3. CDC Stream Staleness Audit: audit_stale_streams (io.kestra.plugin.jdbc.snowflake.Query) queries SNOWFLAKE.ACCOUNT_USAGE.STREAMS for active streams where stale = 'true' or stale_after < CURRENT_TIMESTAMP(). It similarly preserves the total count while retrieving detail records.
4. Health Evaluation Gate: evaluate_pipeline_health (io.kestra.plugin.core.flow.If) evaluates whether any failed tasks or stale CDC streams were returned.
5. Slack Incident Dispatch: When anomalies exist, notify_dataops_incident (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) delivers an operational card to Slack detailing affected task names, SQL error messages, and stale stream targets.
6. Compliant Logging: When all tasks succeeded and streams remain active, log_healthy_state records confirmed health in execution logs.
7. Audit Manifest: export_audit_manifest outputs a structured JSON telemetry summary for DataOps monitoring dashboards.
8. Sentinel Failure Handler: The errors block alerts Slack if the sentinel flow itself cannot connect to Snowflake.

Architecture diagram

flowchart TD
    A[Schedule: Hourly or On-Demand] --> B[audit_failed_tasks: Query TASK_HISTORY]
    B --> C[audit_stale_streams: Query STREAMS]
    C --> D{Failed Tasks or Stale Streams Detected?}
    D -- Yes --> E[notify_dataops_incident: Slack Webhook]
    D -- No --> F[log_healthy_state: Log Health Status]
    E --> G[export_audit_manifest: JSON Telemetry]
    F --> G

What you get

• Early detection of failed Snowflake tasks before downstream dimensional marts and dashboards report missing data.
• Continuous monitoring of CDC stream expiration dates (stale_after) preventing catastrophic stream staleness and data loss.
• Automated Slack incident notifications with root-cause SQL error messages and object identifiers.
• Structured JSON audit telemetry on every run for DataOps observability.

Who it's for

• Analytics Engineers maintaining Snowflake Tasks and Streams for dbt or continuous ELT pipelines.
• Data Platform Engineers responsible for data freshness SLAs and warehouse reliability.
• DataOps Teams managing multi-database Snowflake accounts and change data capture infrastructure.

Why orchestrate this with Kestra

Snowflake provides no out-of-the-box email or webhook alerting for task failures or stream staleness without writing complex custom stored procedures and external network access rules. Kestra delivers a declarative, serverless alternative: it connects over standard Snowflake JDBC, evaluates task history and stream metadata in a single pipeline, and coordinates alerts with your incident management channels.

Inputs

Name	Type	Default	Description
lookback_hours	INT	24	Number of hours of historical Snowflake task execution history to inspect.

Expected outputs

• {{ outputs.audit_failed_tasks.rows }}: Array of failed task records containing database_name, schema_name, task_name, state, error_code, error_message, scheduled_time, and total_failed_tasks.
• {{ outputs.audit_stale_streams.rows }}: Array of stale stream records containing database_name, schema_name, stream_name, table_name, stream_mode, stale_after, and total_stale_streams.
• {{ outputs.export_audit_manifest.value }}: Structured JSON telemetry manifest recording execution timestamp and incident status.

Prerequisites

• A Snowflake account with access to the SNOWFLAKE.ACCOUNT_USAGE shared schema.
• A Snowflake service account with a role granted IMPORTED PRIVILEGES on the SNOWFLAKE database.
• A Slack Incoming Webhook configured for your target DataOps alert channel.

Secrets

• SNOWFLAKE_JDBC_URL: Snowflake JDBC connection string (e.g. jdbc:snowflake://<account_locator>.snowflakecomputing.com?warehouse=COMPUTE_WH&db=SNOWFLAKE&schema=ACCOUNT_USAGE).
• SNOWFLAKE_USER: Service account username.
• SNOWFLAKE_PASSWORD: Service account password.
• SLACK_WEBHOOK_URL: Slack Incoming Webhook URL pointing to your designated DataOps channel.

Quick start

1. Configure SNOWFLAKE_JDBC_URL, SNOWFLAKE_USER, SNOWFLAKE_PASSWORD, and SLACK_WEBHOOK_URL in your Kestra namespace secrets.
2. Import this flow YAML into your Kestra instance.
3. Click Execute in the UI to run an initial audit of task runs and stream staleness.
4. Verify outputs and enable hourly_snowflake_health_audit for continuous monitoring.

Common pitfalls and troubleshooting

• Account Usage Latency: Views in SNOWFLAKE.ACCOUNT_USAGE (such as TASK_HISTORY and STREAMS) can have up to 45 to 120 minutes of latency. For real-time sub-minute task status within a single database, you can point queries at TABLE(INFORMATION_SCHEMA.TASK_HISTORY(...)) instead.
• Recreating Stale Streams: When a stream becomes stale, querying it returns Stream has become stale. To recover, execute CREATE OR REPLACE STREAM <stream_name> ON TABLE <table_name>; and execute an initial reconciliation query against the source table.
• Resuming Suspended Tasks: When a task fails repeatedly, Snowflake suspends it. After addressing the underlying issue, re-enable it via ALTER TASK <name> RESUME;.

How to extend

• Add an automated auto-remediation task running ALTER TASK <task_name> RESUME; for tasks suspended by transient warehouse pauses.
• Escalate critical stream staleness alerts to PagerDuty or Opsgenie using Kestra notification plugins.
• Route task alerts by schema or database to specific team channels.

Links

• Kestra Snowflake Plugin Documentation
• Kestra Slack Plugin Documentation
• Snowflake Task History Documentation
• Snowflake Streams Documentation