fix: mitigate 429 errors in cloud function execution for validation reports (#883)

Author: davidgamez

functions-python/validation_report_processor/function_config.json

@@ -13,8 +13,8 @@
     }
   ],
   "ingress_settings": "ALLOW_INTERNAL_AND_GCLB",
-  "max_instance_request_concurrency": 1,
-  "max_instance_count": 5,
+  "max_instance_request_concurrency": 8,
+  "max_instance_count": 1,
   "min_instance_count": 0,
   "available_cpu": 1
 }

infra/functions-python/main.tf

@@ -23,6 +23,8 @@ terraform {
 #
 
 locals {
+  x_number_of_concurrent_instance = 4
+  deployment_timestamp = formatdate("YYYYMMDDhhmmss", timestamp())
   function_tokens_config = jsondecode(file("${path.module}../../../functions-python/tokens/function_config.json"))
   function_tokens_zip    = "${path.module}/../../functions-python/tokens/.dist/tokens.zip"
 
@@ -346,6 +348,34 @@ resource "google_cloudfunctions2_function" "extract_location_batch" {
 }
 
 # 3. functions/validation_report_processor cloud function
+# Create a queue for the cloud tasks
+# The 2X rate is defined as 4*2 concurrent dispatches and 1 dispatch per second
+# The name of the queue need to be dynamic due to GCP limitations
+# references:
+#   - https://cloud.google.com/tasks/docs/deleting-appengine-queues-and-tasks#deleting_queues
+#   - https://issuetracker.google.com/issues/263947953
+resource "google_cloud_tasks_queue" "cloud_tasks_2x_rate_queue" {
+  name     = "cloud-tasks-2x-rate-queue-${var.environment}-${local.deployment_timestamp}"
+  location = var.gcp_region
+
+  rate_limits {
+    max_concurrent_dispatches = local.x_number_of_concurrent_instance * 2
+    max_dispatches_per_second = 1
+  }
+
+  retry_config {
+    # This will make the cloud task retry for ~two hours
+    max_attempts  = 120
+    min_backoff   = "20s"
+    max_backoff   = "60s"
+    max_doublings = 2
+  }
+}
+
+output "processing_report_cloud_task_name" {
+  value = google_cloud_tasks_queue.cloud_tasks_2x_rate_queue.name
+}
+
 resource "google_cloudfunctions2_function" "process_validation_report" {
   name        = local.function_process_validation_report_config.name
   description = local.function_process_validation_report_config.description

infra/main.tf

@@ -116,6 +116,7 @@ module "workflows" {
   gcp_region         = var.gcp_region
   environment        = var.environment
   validator_endpoint = var.validator_endpoint
+  processing_report_cloud_task_name = module.functions-python.processing_report_cloud_task_name
 }
 
 module "feed-api-load-balancer" {

infra/workflows/main.tf

@@ -52,12 +52,32 @@ resource "google_project_iam_member" "workflows_invoker" {
   member  = "serviceAccount:${google_service_account.workflows_service_account.email}"
 }
 
+# This permission is added to allow the workflow to act as the service account and generate tokens.
+# This is required for the workflow to call a cloud tasks that generates a token to call the a cloud run service or function.
+resource "google_project_iam_member" "service_account_workflow_act_as_binding" {
+  project = var.project_id
+  role    = "roles/iam.serviceAccountUser" #iam.serviceAccounts.actAs
+  member  = "serviceAccount:${google_service_account.workflows_service_account.email}"
+}
+
 resource "google_project_iam_member" "cloud_run_invoker" {
   project = var.project_id
   role    = "roles/run.invoker"
   member  = "serviceAccount:${google_service_account.workflows_service_account.email}"
 }
 
+resource "google_project_iam_member" "service_account_workflow_cloudtasks_enqueuer" {
+  project = var.project_id
+  role    = "roles/cloudtasks.enqueuer"
+  member  = "serviceAccount:${google_service_account.workflows_service_account.email}"
+}
+
+resource "google_project_iam_member" "service_account_workflow_cloudtasks_viewer" {
+  project = var.project_id
+  role    = "roles/cloudtasks.viewer"
+  member  = "serviceAccount:${google_service_account.workflows_service_account.email}"
+}
+
 # Workflow to execute the GTFS Validator
 resource "google_workflows_workflow" "gtfs_validator_execution" {
   name                    = "gtfs_validator_execution"
@@ -70,6 +90,7 @@ resource "google_workflows_workflow" "gtfs_validator_execution" {
     reports_bucket_name   = lower(var.environment) == "prod" ? var.reports_bucket_name : "stg-${var.reports_bucket_name}"
     validator_endpoint    = var.validator_endpoint
     environment           = lower(var.environment)
+    processing_report_cloud_task_name       = var.processing_report_cloud_task_name
   }
   source_contents         = file("${path.module}../../../workflows/gtfs_validator_execution.yml")
 }

infra/workflows/vars.tf

@@ -51,3 +51,8 @@ variable "validator_endpoint" {
   type = string
   description = "URL of the validator endpoint"
 }
+
+variable "processing_report_cloud_task_name" {
+  type = string
+  description = "The cloud task name to call the process report task"
+}

workflows/gtfs_validator_execution.yml

@@ -19,6 +19,11 @@ main:
             - datasetID: ${text.split(data.resourceName, "/")[6]}
             - region: ${resource.labels.location}
             - projectID: ${resource.labels.project_id}
+            - environment: ${sys.get_env("environment")}
+            # The cloud tasks are sensitive to deletion.
+            # This is why we need to have this as a parameter set at deploying time.
+            - cloudTaskName: ${sys.get_env("processing_report_cloud_task_name")}
+            - serviceAccountEmail: ${"workflows-service-account@mobility-feeds-" + environment + ".iam.gserviceaccount.com"}
             - url: ${"https://storage.googleapis.com/" + datasetsBucket + "/" + feedID + "/" + datasetID + "/" + datasetID + ".zip"}
             - headers:
                   Content-Type: application/json
@@ -221,23 +226,73 @@ main:
             switch:
                 - condition: ${byPassDbUpdate}
                   next: successfulExecution
-      - updateDatabase:
-          call: http.post
-          args:
-            url: ${"https://" + region + "-" + projectID + ".cloudfunctions.net/process-validation-report"}
-            auth:
-              type: OIDC
-            body:
+      - createPayload:
+          assign:
+            - payloadTask:
                 feed_id: ${feedID}
                 dataset_id: ${datasetID}
                 validator_version: ${validatorVersion}
-            headers: ${headers}
-          result: updateDatabaseResponse
-      - logUpdateDatabaseResponse:
+            - payloadTaskBase64: '${base64.encode(json.encode(payloadTask))}'
+      - enqueueTask:
+          call: googleapis.cloudtasks.v2beta3.projects.locations.queues.tasks.create
+          args:
+            parent: ${"projects/" + projectID + "/locations/" + region + "/queues/" + cloudTaskName}
+            body:
+              task:
+                httpRequest:
+                  url: ${"https://" + region + "-" + projectID + ".cloudfunctions.net/process-validation-report"}
+                  httpMethod: POST
+                  oidcToken:
+                    serviceAccountEmail: ${serviceAccountEmail}
+                  body: ${payloadTaskBase64}
+                  headers:
+                    Content-Type: application/json
+          result: taskResponse
+      - logTaskCreation:
           call: sys.log
           args:
-              text: ${updateDatabaseResponse}
-              severity: INFO
+            text: ${"Created task for feed " + feedID + " Task ID:" + taskResponse.name}
+            severity: INFO
+      # The waitForTaskCompletion will wait for HTTP error status 404 from google API.
+      # The cloud tasks API doesn't return a status field and when the task success it HTTP returns 404(tasks not found).
+      - waitForTaskCompletion:
+          try:
+            steps:
+              - getTaskStatus:
+                  call: googleapis.cloudtasks.v2beta3.projects.locations.queues.tasks.get
+                  args:
+                    name: ${taskResponse.name}
+                  result: taskStatus
+              - printTaskStatus:
+                  call: sys.log
+                  args:
+                    text: ${taskStatus}
+                    severity: INFO
+                  next: retryTaskStatus
+          except:
+            as: error
+            steps:
+              - printExceptTaskStatus:
+                  call: sys.log
+                  args:
+                    text: ${"HTTP error raised while calling get tasks endpoint:" + error.message}
+                    severity: INFO
+              - handle404:
+                  switch:
+                    - condition: ${error.code == 404}
+                      next: logTaskCompletion
+                    - condition: ${error.code != 404}
+                      raise: ${error}
+      - retryTaskStatus:
+          call: sys.sleep
+          args:
+            seconds: 30
+          next: waitForTaskCompletion
+      - logTaskCompletion:
+          call: sys.log
+          args:
+            text: ${"Completed task for feed " + feedID + " Task ID:" + taskResponse.name}
+            severity: INFO
           next: successfulExecution
       - fileExistenceTimeout:
           steps: