如何用Terraform结合BigQuery定时查询实现空表Slack告警？

整体流程 #

由于BigQuery定时查询仅支持邮箱或Pub/Sub通知，我们通过以下链路实现需求：

• BigQuery每日执行空表检测查询，当表为空时，ASSERT语句触发查询失败
• 失败事件推送到指定Pub/Sub主题
• Cloud Function订阅该Pub/Sub主题，收到事件后调用Slack Webhook发送告警到alerts频道

Terraform配置实现 #

1. 创建Pub/Sub主题与订阅 #

用于接收BigQuery定时查询的失败通知，并对接Cloud Function：

resource "google_pubsub_topic" "bq_empty_table_alert" {
  name = "bq-empty-table-alert-topic"
}

resource "google_pubsub_subscription" "bq_empty_table_alert" {
  name  = "bq-empty-table-alert-sub"
  topic = google_pubsub_topic.bq_empty_table_alert.name

  push_config {
    push_endpoint = google_cloudfunctions_function.bq_slack_alert.https_trigger_url
  }
}

2. 配置BigQuery定时查询 #

使用你提供的ASSERT检测语句，设置每日调度规则，指定失败时推送到上述Pub/Sub主题：

resource "google_bigquery_data_transfer_config" "daily_empty_table_check" {
  display_name   = "Daily Empty Table Check"
  data_source_id = "scheduled_query"
  project_id     = var.project_id

  params = {
    query = <<-EOF
      ASSERT NOT EXISTS(SELECT COUNT(*) FROM ${var.project_id}.${local.dataset_name}.${local.table_name} where DATE(UPDATED_AT) < DATE(CURRENT_TIMESTAMP())) AS "Table has no records"
    EOF
    destination_table_name_template = "" # 仅执行断言，无需输出表
    write_disposition               = "WRITE_TRUNCATE"
    schedule                        = "0 8 * * *" # 每日8点执行，可按需调整
  }

  notification_pubsub_topic = google_pubsub_topic.bq_empty_table_alert.id
  disabled                  = false
}

3. 创建Cloud Function（Python示例） #

编写代码实现Pub/Sub消息解析与Slack告警推送，再用Terraform部署：

本地代码文件main.py

import os
import requests
import base64

def send_slack_alert(event, context):
    # 解析Pub/Sub消息内容
    pubsub_message = base64.b64decode(event['data']).decode('utf-8')
    
    # 从环境变量获取Slack Webhook地址
    slack_webhook_url = os.environ.get('SLACK_WEBHOOK_URL')
    
    # 构造Slack告警内容
    slack_payload = {
        "channel": "#alerts",
        "text": f"⚠️ BigQuery空表告警：\n查询失败详情：{pubsub_message}"
    }
    
    # 发送告警到Slack
    response = requests.post(
        slack_webhook_url,
        json=slack_payload,
        headers={"Content-Type": "application/json"}
    )
    response.raise_for_status()

Terraform部署配置

# 存储Cloud Function代码的GCS Bucket
resource "google_storage_bucket" "function_bucket" {
  name     = "${var.project_id}-function-code-bucket"
  location = var.region
}

# 上传打包后的代码包（需提前将main.py打包为zip）
resource "google_storage_bucket_object" "function_archive" {
  name   = "bq-slack-alert-function.zip"
  bucket = google_storage_bucket.function_bucket.name
  source = "./bq-slack-alert-function.zip"
}

# 创建Cloud Function
resource "google_cloudfunctions_function" "bq_slack_alert" {
  name        = "bq-slack-alert-function"
  runtime     = "python311"
  project     = var.project_id
  region      = var.region

  source_archive_bucket = google_storage_bucket.function_bucket.name
  source_archive_object = google_storage_bucket_object.function_archive.name

  entry_point = "send_slack_alert"

  environment_variables = {
    SLACK_WEBHOOK_URL = var.slack_webhook_url
  }

  event_trigger {
    event_type = "google.pubsub.topic.publish"
    resource   = google_pubsub_topic.bq_empty_table_alert.id
    failure_policy {
      retry = false # 避免重复推送告警
    }
  }

  service_account_email = google_service_account.function_sa.email
}

# 创建Function专用服务账号
resource "google_service_account" "function_sa" {
  account_id   = "bq-slack-alert-sa"
  display_name = "BigQuery to Slack Alert Service Account"
}

# 授予服务账号Pub/Sub订阅权限
resource "google_project_iam_member" "function_pubsub_permission" {
  project = var.project_id
  role    = "roles/pubsub.subscriber"
  member  = "serviceAccount:${google_service_account.function_sa.email}"
}

4. 变量定义（variables.tf） #

variable "project_id" {
  type        = string
  description = "GCP项目ID"
}

variable "region" {
  type        = string
  description = "GCP部署区域"
  default     = "us-central1"
}

variable "dataset_name" {
  type        = string
  description = "BigQuery数据集名称"
}

variable "table_name" {
  type        = string
  description = "待检测的BigQuery表名称"
}

variable "slack_webhook_url" {
  type        = string
  description = "Slack alerts频道的Incoming Webhook地址"
}

关键说明 #

• ASSERT语句会在表为空时抛出错误，触发BigQuery定时查询的失败通知机制
• Cloud Function仅在收到Pub/Sub事件时执行，无额外资源消耗
• 需提前在Slack中创建Incoming Webhook，获取对应URL并传入Terraform变量

内容的提问来源于stack exchange，提问作者Andrii Chertok