使用Terraform部署GCP Dataflow作业时如何自动排空旧作业？

当然可以实现这个流程，核心思路是通过Terraform的本地执行功能或自定义资源，先调用GCP工具/API完成旧作业排空，再触发新作业的部署。以下是两种可行方案：

方法一：使用null_resource + gcloud命令 #

依赖GCP CLI工具gcloud，需确保运行Terraform的环境已配置GCP认证并安装gcloud。

1. （可选）通过数据源获取旧作业状态，用于确认作业存在

data "google_dataflow_job" "old_job" {
  project  = var.gcp_project
  job_id   = var.old_dataflow_job_id
  location = var.dataflow_region
}

2. 创建null_resource执行排空操作，配置触发条件

resource "null_resource" "drain_old_dataflow_job" {
  triggers = {
    # 新作业配置变更时触发排空
    new_job_config = jsonencode(var.new_dataflow_job_config)
  }

  provisioner "local-exec" {
    command = <<EOT
      # 检查旧作业是否处于运行状态，是则执行排空
      if gcloud dataflow jobs describe ${data.google_dataflow_job.old_job.job_id} --region=${var.dataflow_region} --format="value(state)" | grep -q "RUNNING"; then
        gcloud dataflow jobs drain ${data.google_dataflow_job.old_job.job_id} --region=${var.dataflow_region}
        # 轮询等待排空完成，可根据作业规模调整间隔
        until ! gcloud dataflow jobs describe ${data.google_dataflow_job.old_job.job_id} --region=${var.dataflow_region} --format="value(state)" | grep -q "DRAINING"; do
          sleep 30
        done
      fi
    EOT
  }
}

3. 让新Dataflow作业依赖排空操作，确保顺序执行

resource "google_dataflow_job" "new_job" {
  depends_on = [null_resource.drain_old_dataflow_job]

  project           = var.gcp_project
  name              = var.new_dataflow_job_name
  template_gcs_path = var.dataflow_template_path
  location          = var.dataflow_region
  # 其他作业配置参数...
}

方法二：直接调用Dataflow REST API（无需gcloud） #

若环境无法安装gcloud，可通过HTTP请求调用API完成排空：

resource "null_resource" "drain_old_dataflow_job" {
  triggers = {
    new_job_config = jsonencode(var.new_dataflow_job_config)
  }

  provisioner "local-exec" {
    command = <<EOT
      # 获取GCP认证令牌
      ACCESS_TOKEN=$(gcloud auth print-access-token)
      # 查询旧作业当前状态
      JOB_STATE=$(curl -s -H "Authorization: Bearer $ACCESS_TOKEN" "https://dataflow.googleapis.com/v1b3/projects/${var.gcp_project}/locations/${var.dataflow_region}/jobs/${var.old_dataflow_job_id}" | jq -r ".currentState")
      
      if [ "$JOB_STATE" = "RUNNING" ]; then
        # 发送排空请求
        curl -X POST -H "Authorization: Bearer $ACCESS_TOKEN" -H "Content-Type: application/json" "https://dataflow.googleapis.com/v1b3/projects/${var.gcp_project}/locations/${var.dataflow_region}/jobs/${var.old_dataflow_job_id}:drain"
        # 轮询等待排空完成
        while true; do
          JOB_STATE=$(curl -s -H "Authorization: Bearer $ACCESS_TOKEN" "https://dataflow.googleapis.com/v1b3/projects/${var.gcp_project}/locations/${var.dataflow_region}/jobs/${var.old_dataflow_job_id}" | jq -r ".currentState")
          if [ "$JOB_STATE" != "DRAINING" ]; then
            break
          fi
          sleep 30
        done
      fi
    EOT
  }
}

# 新作业依赖排空操作
resource "google_dataflow_job" "new_job" {
  depends_on = [null_resource.drain_old_dataflow_job]
  # 作业配置参数...
}

注意事项 #

• 确保Terraform运行环境拥有dataflow.jobs.get和dataflow.jobs.update权限
• 排空作业耗时取决于作业规模，可按需调整轮询间隔和超时逻辑
• 若存在多个旧作业，可扩展逻辑实现批量排空

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