BigQuery High-Cost Query and Slot Contention Sentinel

id: bigquery-slot-contention-and-spill-sentinel
namespace: company.team
description: |
  Automated BigQuery query resource and cost governance sentinel. Queries INFORMATION_SCHEMA.JOBS_BY_PROJECT
  to identify queries with excessive slot consumption or massive on-demand terabyte scans, alerting FinOps on Slack.

concurrency:
  limit: 1

inputs:
  - id: region
    type: STRING
    displayName: GCP Region
    description: Google Cloud region where BigQuery query execution history resides.
    defaults: "us"

  - id: lookback_hours
    type: INT
    displayName: Audit Lookback Window (Hours)
    description: Historical time window in hours to inspect completed BigQuery query jobs.
    defaults: 1

  - id: max_billed_tb_threshold
    type: FLOAT
    displayName: Max Billed Terabytes Threshold
    description: Flag any single query that scanned and billed more than this volume
      in terabytes.
    defaults: 2.0

  - id: max_slot_minutes_threshold
    type: INT
    displayName: Max Slot Minutes Threshold
    description: Flag queries consuming more than this cumulative slot computation
      time in minutes.
    defaults: 30

  - id: cost_per_tb_usd
    type: FLOAT
    displayName: BigQuery Cost per TB ($)
    description: On-demand query pricing per terabyte (standard GCP rate is $6.25).
    defaults: 6.25

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

triggers:
  - id: hourly_finops_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Hourly automated BigQuery query cost and slot audit executed at minute 15.
    cron: "15 * * * *"

tasks:
  - id: audit_expensive_jobs
    type: io.kestra.plugin.gcp.bigquery.Query
    description: Query BigQuery INFORMATION_SCHEMA job execution history to detect
      expensive or resource-heavy queries.
    projectId: "{{ secret('GCP_PROJECT_ID') }}"
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT_JSON') }}"
    fetchType: FETCH
    sql: |
      SELECT
        job_id,
        user_email,
        creation_time,
        ROUND(TIMESTAMP_DIFF(end_time, start_time, SECOND), 1) AS duration_seconds,
        ROUND(total_bytes_billed / (1024 * 1024 * 1024 * 1024), 2) AS billed_terabytes,
        ROUND((total_bytes_billed / (1024 * 1024 * 1024 * 1024)) * {{ inputs.cost_per_tb_usd }}, 2) AS estimated_cost_usd,
        ROUND(total_slot_ms / (1000 * 60), 1) AS slot_minutes_consumed,
        SUBSTR(query, 1, 150) AS query_snippet
      FROM `{{ secret('GCP_PROJECT_ID') }}`.`region-{{ inputs.region }}`.INFORMATION_SCHEMA.JOBS_BY_PROJECT
      WHERE creation_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {{ inputs.lookback_hours }} HOUR)
        AND job_type = 'QUERY'
        AND statement_type != 'SCRIPT'
        AND (
          total_bytes_billed >= ({{ inputs.max_billed_tb_threshold }} * 1024 * 1024 * 1024 * 1024)
          OR (total_slot_ms / (1000 * 60)) >= {{ inputs.max_slot_minutes_threshold }}
        )
      ORDER BY total_bytes_billed DESC
      LIMIT 10;

  - id: evaluate_job_anomalies
    type: io.kestra.plugin.core.flow.If
    description: Branch execution when any query exceeds billed terabyte limits or
      slot consumption thresholds.
    condition: "{{ outputs.audit_expensive_jobs.rows | length > 0 }}"
    then:
      - id: notify_finops_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Broadcast critical BigQuery runaway query alert with cost estimates
          to Slack.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          *BigQuery High-Cost Query Alert*
          • *Job ID*: `{{ outputs.audit_expensive_jobs.rows[0].job_id }}`
          • *User*: `{{ outputs.audit_expensive_jobs.rows[0].user_email }}`
          • *Data Billed*: `{{ outputs.audit_expensive_jobs.rows[0].billed_terabytes }} TB` (Threshold: `{{ inputs.max_billed_tb_threshold }} TB`)
          • *Estimated Cost*: `${{ outputs.audit_expensive_jobs.rows[0].estimated_cost_usd }}`
          • *Slot Time Consumed*: `{{ outputs.audit_expensive_jobs.rows[0].slot_minutes_consumed }} min` (Threshold: `{{ inputs.max_slot_minutes_threshold }} min`)
          • *Duration*: `{{ outputs.audit_expensive_jobs.rows[0].duration_seconds }}s`
          • *Query Snippet*: `{{ outputs.audit_expensive_jobs.rows[0].query_snippet }}...`
          • *Remediation*: Review query partition filters, eliminate SELECT *, and configure maximum bytes billed limits in client sessions.

    else:
      - id: log_nominal_capacity
        type: io.kestra.plugin.core.log.Log
        description: Confirm that all BigQuery jobs executed within nominal compute and
          scan thresholds.
        message: "BigQuery job execution health is nominal. Zero queries exceeded {{
          inputs.max_billed_tb_threshold }} TB or {{
          inputs.max_slot_minutes_threshold }} slot minutes."

  - id: export_job_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Record BigQuery cost governance telemetry for historical FinOps analysis.
    format: |
      {
        "pipeline": "bigquery-slot-contention-and-spill-sentinel",
        "audited_at": "{{ execution.startDate }}",
        "region": "{{ inputs.region }}",
        "flagged_jobs_count": {{ outputs.audit_expensive_jobs.rows | length }},
        "status": "{{ outputs.audit_expensive_jobs.rows | length > 0 ? 'ANOMALY_DETECTED' : 'NOMINAL' }}"
      }