feat(caretaker): add triage Cloud Run job workflow (#28468)

2026-08-09 08:27:00 -07:00 · 2026-08-07 14:02:14 -05:00
parent d9b600b1c9
commit 8cb94fe645
1 changed files with 102 additions and 0 deletions
@@ -0,0 +1,102 @@
+# Google Cloud Workflow invoked by the Ingestion Layer to run a Cloud Run Job,
+# writing to Firestore and publishing to a Pub/Sub DLQ on failure.
+
+main:
+  params: ['event']
+  steps:
+    - init:
+        assign:
+          - project_id: '${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}'
+          - database_id: '${sys.get_env("FIRESTORE_DATABASE")}'
+          - collection_name: '${sys.get_env("FIRESTORE_COLLECTION")}'
+          - job_name: 'triage-worker'
+          - job_location: 'us-west1'
+          - base64_data: '${event.data.message.data}'
+          - workflow_exec_id: '${sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}'
+          - dlq_topic: '${"projects/" + project_id + "/topics/incoming-issues-dlq"}'
+          - payload: '${json.decode(text.decode(base64.decode(base64_data)))}'
+          - owner: '${text.split(payload.repository, "/")[0]}'
+          - repo: '${text.split(payload.repository, "/")[1]}'
+          - issue_number: '${payload.issue_number}'
+          - doc_id: '${"github_" + owner + "_" + repo + "_" + string(issue_number)}'
+    - run_processing_job:
+        try:
+          call: 'googleapis.run.v1.namespaces.jobs.run'
+          args:
+            name: '${"namespaces/" + project_id + "/jobs/" + job_name}'
+            location: '${job_location}'
+            body:
+              overrides:
+                containerOverrides:
+                  env:
+                    - name: 'ISSUE_DETAILS'
+                      value: '${base64_data}'
+                    - name: 'WORKFLOW_EXECUTION_ID'
+                      value: '${workflow_exec_id}'
+          result: 'job_execution'
+        retry:
+          predicate: '${retry_predicate}'
+          max_retries: 1
+          backoff:
+            # wait 5 seconds before the retry
+            initial_delay: 5
+            max_delay: 60
+            multiplier: 2
+        except:
+          as: 'error'
+          steps:
+            - update_firestore_needs_human:
+                call: 'googleapis.firestore.v1.projects.databases.documents.patch'
+                args:
+                  name: '${"projects/" + project_id + "/databases/" + database_id + "/documents/" + collection_name + "/" + doc_id}'
+                  updateMask:
+                    fieldPaths:
+                      - 'status'
+                      - 'error'
+                      - 'lock.holder'
+                      - 'lock.expires_at'
+                      - 'updated_at'
+                  body:
+                    fields:
+                      status:
+                        stringValue: 'NEEDS_HUMAN'
+                      error:
+                        stringValue: '${"Job cancelled or crashed terminally: " + error.message}'
+                      lock:
+                        mapValue:
+                          fields:
+                            holder:
+                              nullValue: 'NULL_VALUE'
+                            expires_at:
+                              nullValue: 'NULL_VALUE'
+                      updated_at:
+                        timestampValue: '${sys.now()}'
+            - publish_to_dlq:
+                call: 'googleapis.pubsub.v1.projects.topics.publish'
+                args:
+                  topic: '${dlq_topic}'
+                  body:
+                    messages:
+                      - data: '${base64_data}'
+                        attributes:
+                          error: '${error.message}'
+                          workflow_id: '${workflow_exec_id}'
+                          origin: 'workflow_failure'
+            - workflow_failed:
+                raise: '${"Terminal failure. DLQ message sent and Firestore updated. Error is " + error.message}'
+
+    - success_log:
+        return:
+          status: 'SUCCESS'
+          job_details:
+            name: '${job_name}'
+            execution_id: '${job_execution.metadata.name}'
+            region: '${job_location}'
+          log_view_url: '${"https://console.cloud.google.com/run/jobs/executions/details/" + job_location + "/" + job_execution.metadata.name + "?project=" + project_id}'
+
+# retry on any error returned from the Cloud Run Job
+retry_predicate:
+  params: ['e']
+  steps:
+    - check_retry:
+        return: true