Snowflake Query Tagging and FinOps Cost Allocation Governance Guard

id: snowflake-query-tag-and-cost-allocation-governance-guard
namespace: company.team
description: |
  Automated Snowflake FinOps governance guard for query tagging and cost allocation compliance.
  Audits SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY to identify untagged or non-compliant queries,
  aggregating unallocated warehouse compute spend and alerting FinOps and Data Platform teams.

concurrency:
  limit: 1

inputs:
  - id: lookback_hours
    type: INT
    displayName: Lookback Period (Hours)
    description: Number of hours of historical Snowflake query execution activity to
      inspect.
    defaults: 24

  - id: max_untagged_percentage_threshold
    type: FLOAT
    displayName: Max Untagged Query Threshold (%)
    description: Maximum acceptable percentage of untagged queries before triggering
      an alert.
    defaults: 5.0

  - id: credit_price_usd
    type: FLOAT
    displayName: Estimated Credit Price ($)
    description: Estimated dollar cost per Snowflake compute credit (standard
      Enterprise rate is $3.00).
    defaults: 3.00

  - id: max_offending_roles_to_list
    type: INT
    displayName: Top Offending Roles Limit
    description: Maximum number of offending roles and users to report in the alert
      breakdown.
    defaults: 5

  - id: slack_channel
    type: STRING
    displayName: Alert Channel
    description: Slack channel destination for FinOps query tag governance alerts.
    defaults: "#finops-governance"

triggers:
  - id: daily_tagging_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Scheduled daily query tagging compliance audit executed at 06:00 UTC.
    cron: "0 6 * * *"

tasks:
  - id: audit_query_tag_compliance
    type: io.kestra.plugin.jdbc.snowflake.Query
    description: Query Snowflake account usage query history and attribution history
      to compute query tag compliance and unallocated credit cost.
    url: "{{ secret('SNOWFLAKE_JDBC_URL') }}"
    username: "{{ secret('SNOWFLAKE_USER') }}"
    password: "{{ secret('SNOWFLAKE_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: |
      SELECT
        COUNT(*) AS total_warehouse_queries,
        COUNT(CASE WHEN qh.query_tag IS NULL OR TRIM(qh.query_tag) = '' THEN 1 END) AS untagged_queries,
        COUNT(CASE WHEN qh.query_tag IS NOT NULL AND TRIM(qh.query_tag) != '' THEN 1 END) AS tagged_queries,
        ROUND(
          (COUNT(CASE WHEN qh.query_tag IS NULL OR TRIM(qh.query_tag) = '' THEN 1 END) * 100.0) /
          NULLIF(COUNT(*), 0),
          2
        ) AS untagged_percentage,
        ROUND(
          SUM(CASE WHEN qh.query_tag IS NULL OR TRIM(qh.query_tag) = '' THEN qh.total_elapsed_time ELSE 0 END) / 3600000.0,
          2
        ) AS untagged_execution_hours,
        ROUND(
          COALESCE(SUM(CASE WHEN qh.query_tag IS NULL OR TRIM(qh.query_tag) = '' THEN qa.credits_attributed_compute ELSE 0 END), 0.0),
          2
        ) AS untagged_compute_credits,
        ROUND(
          COALESCE(SUM(CASE WHEN qh.query_tag IS NULL OR TRIM(qh.query_tag) = '' THEN qa.credits_attributed_compute ELSE 0 END), 0.0) * {{ inputs.credit_price_usd }},
          2
        ) AS estimated_untagged_cost_usd,
        COUNT(DISTINCT CASE WHEN qh.query_tag IS NULL OR TRIM(qh.query_tag) = '' THEN qh.user_name END) AS offending_users_count,
        COUNT(DISTINCT CASE WHEN qh.query_tag IS NULL OR TRIM(qh.query_tag) = '' THEN qh.role_name END) AS offending_roles_count
      FROM snowflake.account_usage.query_history qh
      LEFT JOIN snowflake.account_usage.query_attribution_history qa
        ON qh.query_id = qa.query_id
      WHERE qh.start_time >= DATEADD('hour', -{{ inputs.lookback_hours }}, CURRENT_TIMESTAMP())
        AND qh.warehouse_name IS NOT NULL
        AND qh.execution_status = 'SUCCESS';

  - id: fetch_offending_roles_summary
    type: io.kestra.plugin.jdbc.snowflake.Query
    description: Identify the top users, roles, and warehouses responsible for
      untagged queries.
    url: "{{ secret('SNOWFLAKE_JDBC_URL') }}"
    username: "{{ secret('SNOWFLAKE_USER') }}"
    password: "{{ secret('SNOWFLAKE_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT
        role_name,
        user_name,
        warehouse_name,
        COUNT(*) AS untagged_query_count,
        ROUND(SUM(total_elapsed_time) / 1000.0, 1) AS total_elapsed_seconds
      FROM snowflake.account_usage.query_history
      WHERE start_time >= DATEADD('hour', -{{ inputs.lookback_hours }}, CURRENT_TIMESTAMP())
        AND warehouse_name IS NOT NULL
        AND execution_status = 'SUCCESS'
        AND (query_tag IS NULL OR TRIM(query_tag) = '')
      GROUP BY role_name, user_name, warehouse_name
      ORDER BY untagged_query_count DESC
      LIMIT {{ inputs.max_offending_roles_to_list }};

  - id: evaluate_tagging_compliance
    type: io.kestra.plugin.core.flow.If
    description: Branch execution when untagged query rate exceeds the
      organizational threshold.
    condition: "{{ (outputs.audit_query_tag_compliance.row.UNTAGGED_PERCENTAGE |
      default(0.0)) > inputs.max_untagged_percentage_threshold }}"
    then:
      - id: notify_finops_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Broadcast critical query tagging compliance alert to the FinOps
          Slack channel.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "*Snowflake FinOps Governance Alert: Query Tag Compliance Breached*\n• *Untagged Query Rate*: `{{ outputs.audit_query_tag_compliance.row.UNTAGGED_PERCENTAGE }}%` (Threshold: `{{ inputs.max_untagged_percentage_threshold }}%`)\n• *Untagged Queries*: `{{ outputs.audit_query_tag_compliance.row.UNTAGGED_QUERIES }}` out of `{{ outputs.audit_query_tag_compliance.row.TOTAL_WAREHOUSE_QUERIES }}` total queries in past `{{ inputs.lookback_hours }}h`\n• *Untagged Compute Credits*: `{{ outputs.audit_query_tag_compliance.row.UNTAGGED_COMPUTE_CREDITS }}` credits (~${{ outputs.audit_query_tag_compliance.row.ESTIMATED_UNTAGGED_COST_USD }} USD at ${{ inputs.credit_price_usd }}/credit)\n• *Untagged Execution Time*: `{{ outputs.audit_query_tag_compliance.row.UNTAGGED_EXECUTION_HOURS }} hours`\n• *Affected Entities*: `{{ outputs.audit_query_tag_compliance.row.OFFENDING_USERS_COUNT }} users` across `{{ outputs.audit_query_tag_compliance.row.OFFENDING_ROLES_COUNT }} roles`\n• *Top Offending Role*: `{{ outputs.fetch_offending_roles_summary.rows[0].ROLE_NAME | default('N/A') }}` (User: `{{ outputs.fetch_offending_roles_summary.rows[0].USER_NAME | default('N/A') }}`, Warehouse: `{{ outputs.fetch_offending_roles_summary.rows[0].WAREHOUSE_NAME | default('N/A') }}`, Queries: `{{ outputs.fetch_offending_roles_summary.rows[0].UNTAGGED_QUERY_COUNT | default(0) }}`)\n• *Impact*: Untagged workloads prevent accurate FinOps cost attribution and chargeback across business units and engineering domains.\n• *Remediation*: Require service accounts, dbt profiles, and BI tools to execute: `ALTER SESSION SET QUERY_TAG = '{\"team\": \"<team_name>\", \"cost_center\": \"<id>\", \"env\": \"prod\"}';` before executing queries."
          }

    else:
      - id: log_nominal_tagging
        type: io.kestra.plugin.core.log.Log
        description: Log confirmation that query tagging compliance is within acceptable
          limits.
        message: "Snowflake query tagging compliance is nominal: {{
          outputs.audit_query_tag_compliance.row.UNTAGGED_PERCENTAGE }}%
          untagged queries (below {{ inputs.max_untagged_percentage_threshold
          }}% threshold)."

  - id: export_governance_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Record governance telemetry manifest for FinOps reporting and
      compliance audits.
    format: |
      {
        "pipeline": "snowflake-query-tag-and-cost-allocation-governance-guard",
        "audited_at": "{{ execution.startDate }}",
        "lookback_hours": {{ inputs.lookback_hours }},
        "total_queries": {{ outputs.audit_query_tag_compliance.row.TOTAL_WAREHOUSE_QUERIES | default(0) }},
        "untagged_queries": {{ outputs.audit_query_tag_compliance.row.UNTAGGED_QUERIES | default(0) }},
        "untagged_percentage": {{ outputs.audit_query_tag_compliance.row.UNTAGGED_PERCENTAGE | default(0.0) }},
        "untagged_compute_credits": {{ outputs.audit_query_tag_compliance.row.UNTAGGED_COMPUTE_CREDITS | default(0.0) }},
        "estimated_untagged_cost_usd": {{ outputs.audit_query_tag_compliance.row.ESTIMATED_UNTAGGED_COST_USD | default(0.0) }},
        "threshold_percentage": {{ inputs.max_untagged_percentage_threshold }},
        "status": "{{ (outputs.audit_query_tag_compliance.row.UNTAGGED_PERCENTAGE | default(0.0)) > inputs.max_untagged_percentage_threshold ? 'NON_COMPLIANT' : 'COMPLIANT' }}"
      }

Automated Snowflake FinOps watchdog auditing SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY and QUERY_ATTRIBUTION_HISTORY for query tag compliance. Identifies untagged warehouse workloads, attributes unallocated compute credits and estimated dollar cost, and alerts FinOps teams on Slack when non-compliance exceeds organizational thresholds.

What this blueprint does

In enterprise cloud data platforms, finance and engineering leadership rely on Snowflake Query Tags (QUERY_TAG) to map warehouse compute costs to business units, cost centers, microservices, and dbt models. When queries run without tags, compute charges fall into an unallocated "black hole," breaking FinOps chargeback models.

This blueprint automates continuous query tagging governance:

Inspects all warehouse queries executed across your Snowflake account over a configurable lookback window (default 24 hours).
Measures total queries versus untagged queries (QUERY_TAG IS NULL OR TRIM(QUERY_TAG) = '').
Calculates the untagged query percentage and total unallocated compute hours.
Identifies the top offending roles, users, and warehouses responsible for untagged activity.
Evaluates whether the untagged rate exceeds the organization's compliance SLA threshold (default 5%).
Dispatches an actionable alert to Slack with offending roles and recommended remediation scripts if breached.
Emits a structured FinOps governance manifest for auditing and tracking compliance trends over time.

How it works

Scheduled or On-Demand Trigger: Runs automatically every morning at 06:00 UTC via daily_tagging_audit or manually with custom lookback hours.
Compliance Aggregation: The audit_query_tag_compliance task executes an optimized SQL aggregation against snowflake.account_usage.query_history, filtering for successful warehouse queries and calculating the untagged percentage.
Offender Breakdown: The fetch_offending_roles_summary task isolates the top roles, users, and virtual warehouses driving the highest volume of untagged queries.
SLA Evaluation: The evaluate_tagging_compliance conditional task compares the computed untagged percentage against inputs.max_untagged_percentage_threshold.
Slack Alerting: If compliance is breached, notify_finops_slack sends a detailed Slack notification highlighting the offending role, user, warehouse, and remediation commands.
Nominal Logging: If within threshold, log_nominal_tagging logs a confirmation message.
Audit Manifest: The export_governance_manifest task records a JSON summary of the audit for ingestion into cloud governance dashboards.

What you get

Comprehensive visibility into untagged Snowflake compute spend.
Automated detection of teams, service accounts, or BI tools failing to set query tags.
Early warning before monthly billing cycles to enforce cost allocation discipline.
Zero human overhead for continuous enterprise FinOps policy enforcement.

Who it is for

FinOps Practitioners & Cloud Economists: Enforcing cloud financial governance and chargeback across Snowflake accounts.
Data Platform & Infrastructure Engineers: Maintaining observability and tagging standards across pipelines and BI connections.
Analytics Engineering Managers: Ensuring all dbt jobs, Airflow DAGs, and Kestra flows tag their sessions properly.

Why orchestrate this with Kestra

Snowflake resource monitors manage warehouse quotas but cannot enforce tagging policies or inspect session metadata. Kestra orchestrates the entire governance lifecycle declaratively: connecting securely through JDBC, inspecting account-wide usage views, branching on FinOps compliance thresholds, and notifying engineering stakeholders directly where they work.

Inputs

Name	Type	Default	Description
`lookback_hours`	INT	`24`	Number of hours of historical query activity to audit.
`max_untagged_percentage_threshold`	FLOAT	`5.0`	Maximum acceptable percentage of untagged queries before triggering an alert.
`credit_price_usd`	FLOAT	`3.00`	Estimated dollar cost per Snowflake compute credit.
`max_offending_roles_to_list`	INT	`5`	Maximum number of offending roles and users to list in the breakdown.
`slack_channel`	STRING	`#finops-governance`	Slack channel destination for FinOps query tag governance alerts.

Expected outputs

{{ outputs.audit_query_tag_compliance.row }}: Aggregated metrics including TOTAL_WAREHOUSE_QUERIES, UNTAGGED_QUERIES, UNTAGGED_PERCENTAGE, UNTAGGED_COMPUTE_CREDITS, ESTIMATED_UNTAGGED_COST_USD, and UNTAGGED_EXECUTION_HOURS.
{{ outputs.fetch_offending_roles_summary.rows }}: Array of top offending roles, users, and warehouses with individual UNTAGGED_QUERY_COUNT values.
{{ outputs.evaluate_tagging_compliance }}: Status of the conditional SLA evaluation.
{{ outputs.export_governance_manifest.value }}: Structured JSON telemetry manifest recording compliance state and timestamp.

Prerequisites

A Snowflake account with access to the SNOWFLAKE.ACCOUNT_USAGE schema.
A Snowflake service account with a role granted IMPORTED PRIVILEGES on the SNOWFLAKE database.
A Slack Incoming Webhook URL targeting your FinOps notification 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: Username of the service account authorized to query account usage.
SNOWFLAKE_PASSWORD: Password for the Snowflake service account.
SLACK_WEBHOOK_URL: Slack Incoming Webhook endpoint URL used for sending alert notifications.

Quick start

Configure SNOWFLAKE_JDBC_URL, SNOWFLAKE_USER, SNOWFLAKE_PASSWORD, and SLACK_WEBHOOK_URL in your Kestra namespace secrets.
Import this flow YAML into your Kestra instance.
Click Execute in the UI to run an immediate audit across the last 24 hours of Snowflake activity.
Review execution outputs to inspect your account's query tagging compliance rate.

Common pitfalls and troubleshooting

Attribution and Usage Data Latency: While SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY updates within 45 minutes, QUERY_ATTRIBUTION_HISTORY lags by up to a few hours before compute credits are allocated. Running audits with a 24-hour lookback window ensures consistent and reconciled metrics rather than zero attribution for very recent queries.
Excluded Queries: Queries running without a virtual warehouse (such as metadata-only commands or failed syntax checks) are filtered out to focus exclusively on billable compute spend.
Structured JSON Tags: If your organization enforces specific JSON keys (e.g., cost_center), extend the WHERE clause to check TRY_PARSE_JSON(query_tag):cost_center IS NOT NULL.

How to extend

Add automated PagerDuty or Jira ticket creation for persistent repeat-offender roles.
Filter for specific production warehouses (e.g., ANALYTICS_WH, ETL_WH) to apply stricter tagging thresholds.
Enforce mandatory tagging at the warehouse level by integrating with automated user permission reviews.

Snowflake Query Tagging and FinOps Cost Allocation Governance Guard

What this blueprint does

How it works

What you get

Who it is for

Why orchestrate this with Kestra

Inputs

Expected outputs

Prerequisites

Secrets

Quick start

Common pitfalls and troubleshooting

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Snowflake Query Tagging and FinOps Cost Allocation Governance Guard

What this blueprint does

How it works

What you get

Who it is for

Why orchestrate this with Kestra

Inputs

Expected outputs

Prerequisites

Secrets

Quick start

Common pitfalls and troubleshooting

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

More Business Blueprints

New to Kestra?