셀프 스케줄링 반복 클라우드 작업 (Terraform + Python 코드와 함께)

Source: Dev.to

개요

완전 서버리스 클라우드 환경을 사용하면, 때때로 “이 함수를 정기적으로 실행하려면 어떻게 해야 할까?”라는 쉬운 질문도 답하기가 쉽지 않습니다. 전통적인 CRON 작업은 여기서 작동하지 않습니다.
대신, 예약된 Cloud Tasks가 스스로 다시 예약하도록 활용하면 정기 실행을 달성할 수 있습니다.

워크플로우

1. Cloud Tasks 큐를 설정합니다.
2. 다음 작업을 큐에 넣는 코드를 Cloud Run 함수에 추가합니다.
3. 체인을 시작할 초기 작업을 생성합니다.

Service Account

Cloud Function을 실행할 서비스 계정을 생성합니다.

resource "google_service_account" "" {
  account_id    = ""
  display_name = "My Service Account"
  description   = "Service account used for my function"
  project       = var.project_id
}

Cloud Tasks와 Cloud Run을 호출하기 위한 필요한 역할을 부여합니다.

resource "google_project_iam_member" "" {
  for_each = toset([
    "roles/run.invoker",
    "roles/cloudtasks.enqueuer"
  ])

  project = var.project_id
  role    = each.value
  member  = "serviceAccount:${google_service_account..email}"

  depends_on = [google_service_account.]
}

다음 작업을 큐에 넣을 때 서비스 계정이 자신을 가장하도록 허용합니다.

resource "google_service_account_iam_member" "" {
  service_account_id = google_service_account..name
  role               = "roles/iam.serviceAccountUser"
  member             = "serviceAccount:${google_service_account..email}"
}

Cloud Task Queue

예약된 HTTP 작업을 보관할 Cloud Tasks 큐를 생성합니다.

resource "google_cloud_tasks_queue" "" {
  name     = ""
  location = var.region

  http_target {
    http_method = "POST"
    oidc_token {
      service_account_email = 
    }
  }
}

함수에서 작업 큐에 넣기

다음 작업을 큐에 넣는 도우미 함수를 Cloud Run(또는 Cloud Functions) 코드에 추가하세요. days_increment와 예약 시간을 필요에 따라 조정하십시오.

import json
from datetime import datetime, timedelta, timezone
from google.cloud import tasks_v2
from google.protobuf import timestamp_pb2

def cloud_task_scheduler(days_increment: int = 1):
    try:
        client = tasks_v2.CloudTasksClient()
        parent = client.queue_path(GCP_PROJECT_ID, LOCATION, CLOUD_TASK_QUEUE_ID)
        FUNCTION_URL = ""

        # Payload for the next invocation
        payload = {"schedule_next_cloud_task": True}
        body_bytes = json.dumps(payload).encode("utf-8")

        # Schedule for the desired future time (e.g., 16:00 UTC)
        scheduled_date = (
            datetime.now(timezone.utc) + timedelta(days=days_increment)
        ).replace(hour=16, minute=0, second=0, microsecond=0)

        ts = timestamp_pb2.Timestamp()
        ts.FromDatetime(scheduled_date)

        task = {
            "http_request": {
                "http_method": tasks_v2.HttpMethod.POST,
                "url": FUNCTION_URL,
                "headers": {"Content-Type": "application/json"},
                "body": body_bytes,
            },
            "schedule_time": ts,
        }

        client.create_task(parent=parent, task=task)

    except Exception as e:
        raise Exception(f"Error scheduling next Cloud Task: {e}")

함수 끝에서(또는 적절한 위치에서) cloud_task_scheduler()를 호출하여 다음 실행이 큐에 들어가도록 하세요.

첫 번째 작업 예약

체인을 시작하려면 gcloud CLI를 사용해 초기 작업을 생성합니다.

gcloud tasks create-http-task my-first-task \
  --queue=my-queue \
  --url="my-function-url" \
  --method=POST \
  --body-content='{"schedule_next_cloud_task": true}' \
  --header="Content-Type":"application/json" \
  --oidc-service-account-email="my-sa-email" \
  --location="region" \
  --schedule-time="YYYY-MM-DDTHH:MM:SS+00:00"

명령을 실행하면 첫 번째 작업이 큐에 들어가 함수 호출을 기다리게 됩니다. 함수가 실행되면 다음 작업을 큐에 넣어 자체 유지 스케줄을 생성합니다.

왜 이렇게 작동하는가

• 비용 효율적 – Cloud Tasks는 관대한 무료 티어 내에서 사실상 무료입니다. 함수의 컴퓨팅 시간에 대해서만 비용을 지불합니다.
• 제로까지 확장 가능 – 다음 예약 실행을 기다리는 동안 리소스가 소비되지 않습니다.
• 유연함 – 간격, 시간대 등을 조정하거나 HTTP 본문을 통해 임의의 데이터를 전달하여 더 복잡한 워크플로우(예: 장기 작업 분할)를 처리할 수 있습니다.

이 패턴은 외부 크론 서비스를 관리하지 않고도 반복 작업을 실행할 수 있는 간단하고 서버리스한 방법을 제공합니다.