執行 Cloud Run 工作，處理儲存在 Cloud Storage 中的事件資料

您可以使用 Workflows 執行 Cloud Run 工作，以便在工作流程中進行更複雜的資料處理作業，或以系統化的方式自動調度管理現有工作。

本教學課程說明如何使用 Workflows 執行 Cloud Run 工作，處理儲存在 Cloud Storage bucket 中的事件資料。將事件酬載儲存在 Cloud Storage 值區中，可讓您使用客戶管理的加密金鑰加密資料，但如果將事件資料做為環境變數傳遞至 Cloud Run 作業，則無法加密資料。

下圖提供概要總覽：

建立 Cloud Run 工作

本教學課程使用 GitHub 上的範例程式碼。部署指令碼會建構容器映像檔，以建立 Cloud Run 工作。這個指令碼也會建立 Cloud Storage bucket。Cloud Run 工作會讀取儲存在 Cloud Storage bucket 中的任何事件資料，然後列印事件資料。

1. 如果您在 Cloud Shell 中執行部署指令碼，且Compute Engine 預設服務帳戶沒有編輯者角色，請將下列專案角色授予 Compute Engine 預設服務帳戶。(否則，您可以略過這個步驟，並在下一個步驟中繼續複製範例應用程式存放區)。

   注意： iam.automaticIamGrantsForDefaultServiceAccounts機構政策限制會禁止系統自動將「編輯者」角色授予預設服務帳戶。如果您是在 2024 年 5 月 3 日後建立機構，系統預設會強制執行這項限制。

   1. 授予Artifact Registry 寫入者角色 (roles/artifactregistry.writer)：

      gcloud projects add-iam-policy-binding PROJECT_ID \
          --member=serviceAccount:PROJECT_NUMBER-compute@ \
          --role=roles/artifactregistry.writer

      將 PROJECT_NUMBER 替換為專案編號。 Google Cloud您可以在 Google Cloud 控制台的「歡迎」頁面中找到專案編號，也可以執行下列指令：

      gcloud projects describe PROJECT_ID --format='value(projectNumber)'

   2. 授予 Storage 物件使用者角色 (roles/storage.objectUser)：

      gcloud projects add-iam-policy-binding PROJECT_ID \
          --member=serviceAccount:PROJECT_NUMBER-compute@ \
          --role=roles/storage.objectUser

   3. 授予記錄記錄寫入者角色 (roles/logging.logWriter)：

      gcloud projects add-iam-policy-binding PROJECT_ID \
          --member=serviceAccount:PROJECT_NUMBER-compute@ \
          --role=roles/logging.logWriter

2. 將範例應用程式存放區複製到本機電腦，即可取得程式碼範例：

   git clone https://github.com/GoogleCloudPlatform/workflows-demos.git

   或者，您也可以下載 ZIP 格式的範例檔案，

3. 變更為包含範例程式碼的目錄：

   cd workflows-demos/cloud-run-jobs-payload-gcs/message-payload-job

4. 執行部署指令碼，建立 Cloud Run 工作：

   ./deploy-job.sh

這個指令碼會建立名為 message-payload-PROJECT_ID 的 Cloud Storage bucket，其中 PROJECT_ID 是 Google Cloud 專案的 ID。系統也會建立名為 message-payload-job 的 Cloud Run 工作。

部署執行 Cloud Run 工作的工作流程

定義及部署工作流程，執行您剛建立的 Cloud Run 工作。工作流程定義是由一系列步驟組成，這些步驟會使用 Workflows 語法描述。

這項工作流程會接收事件、將事件資料儲存至 Cloud Storage 值區，然後執行 Cloud Run 工作來處理事件資料。

控制台

1. 前往 Google Cloud 控制台的「Workflows」頁面：

   前往「Workflows」頁面

2. 按一下「建立」。

3. 輸入新工作流程的名稱，例如 message-payload-workflow。

4. 選擇適當的區域，例如 us-central1。

5. 在「Service account」(服務帳戶) 欄位中，選取您先前建立的服務帳戶。

   服務帳戶是工作流程的身分。您應該已將下列角色授予服務帳戶：

   • Cloud Run 管理員：執行 Cloud Run 工作
   • 記錄寫入者：寫入記錄項目
   • Storage 物件建立者：在 Cloud Storage 中建立物件

6. 點選「下一步」。

7. 在工作流程編輯器中，輸入下列工作流程定義：

   main:
       params: [event]
       steps:
           - init:
               assign:
                   - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
                   - event_bucket: ${"message-payload-" + project_id}
                   - event_file: ${event.id + ".data.json"}
                   - job_name: message-payload-job
                   - job_location: us-central1
           - log_event:
               call: sys.log
               args:
                   data: ${event}
           - write_payload_to_gcs:
               call: http.post
               args:
                   url: ${"https://storage.googleapis.com/upload/storage/v1/b/" + event_bucket + "/o"}
                   auth:
                       type: OAuth2
                   query:
                       name: ${event_file}
                   body:
                       ${event.data}
           - run_job_to_process_payload:
               call: googleapis.run.v1.namespaces.jobs.run
               args:
                   name: ${"namespaces/" + project_id + "/jobs/" + job_name}
                   location: ${job_location}
                   body:
                       overrides:
                           containerOverrides:
                               env:
                                   - name: INPUT_BUCKET
                                     value: ${event_bucket}
                                   - name: INPUT_FILE
                                     value: ${event_file}
               result: job_execution
           - finish:
               return: ${job_execution}

8. 按一下 [Deploy] (部署)。

gcloud

1. 為工作流程建立原始碼檔案：

   touch message-payload-workflow.yaml

2. 將下列工作流程定義複製到 message-payload-workflow.yaml：

   main:
       params: [event]
       steps:
           - init:
               assign:
                   - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
                   - event_bucket: ${"message-payload-" + project_id}
                   - event_file: ${event.id + ".data.json"}
                   - job_name: message-payload-job
                   - job_location: us-central1
           - log_event:
               call: sys.log
               args:
                   data: ${event}
           - write_payload_to_gcs:
               call: http.post
               args:
                   url: ${"https://storage.googleapis.com/upload/storage/v1/b/" + event_bucket + "/o"}
                   auth:
                       type: OAuth2
                   query:
                       name: ${event_file}
                   body:
                       ${event.data}
           - run_job_to_process_payload:
               call: googleapis.run.v1.namespaces.jobs.run
               args:
                   name: ${"namespaces/" + project_id + "/jobs/" + job_name}
                   location: ${job_location}
                   body:
                       overrides:
                           containerOverrides:
                               env:
                                   - name: INPUT_BUCKET
                                     value: ${event_bucket}
                                   - name: INPUT_FILE
                                     value: ${event_file}
               result: job_execution
           - finish:
               return: ${job_execution}

3. 輸入下列指令來部署工作流程：

   gcloud workflows deploy message-payload-workflow \
       --location=us-central1 \
       --source=message-payload-workflow.yaml \
       --service-account=SERVICE_ACCOUNT_NAME@PROJECT_ID.

   更改下列內容：

   • SERVICE_ACCOUNT_NAME：您先前建立的服務帳戶名稱
   • PROJECT_ID：您的Google Cloud 專案 ID

   服務帳戶是工作流程的身分。您應該已將下列角色授予服務帳戶：

   • roles/logging.logWriter：用於寫入記錄項目
   • roles/run.admin：執行 Cloud Run 工作
   • roles/storage.objectCreator：在 Cloud Storage 中建立物件

工作流程會執行下列作業：

1. init 步驟：接受事件做為引數，並設定必要變數。

2. log_event 步驟：使用函式 sys.log 在 Cloud Logging 中建立記錄項目。

3. write_payload_to_gcs 步驟：發出 HTTP POST 要求，並將事件酬載資料寫入 Cloud Storage bucket 檔案。

4. run_job_to_process_payload 步驟：使用 Cloud Run Admin API 連接器方法 googleapis.run.v1.namespaces.jobs.run 執行作業。Cloud Storage bucket 和資料檔案名稱會從工作流程以覆寫變數的形式傳遞至工作。

5. finish 步驟：傳回工作執行資訊，做為工作流程的結果。

建立 Pub/Sub 主題

建立 Pub/Sub 主題，以便將訊息發布至該主題。本範例使用 Pub/Sub 事件，說明如何使用 Workflows 轉送事件，並將事件儲存至 Cloud Storage，以便 Cloud Run 工作處理事件資料。

控制台

1. 前往 Google Cloud 控制台的「主題」頁面。

   前往「主題」

2. 按一下 「建立主題」。

3. 在「Topic ID」(主題 ID) 欄位中，輸入主題的 ID，例如 message-payload-topic。

4. 保留「新增預設訂閱項目」選項。

5. 請勿選取其他選項。

6. 點選「建立」。

gcloud

如要使用 ID message-payload-topic 建立主題，請執行 gcloud pubsub topics create 指令：

gcloud pubsub topics create message-payload-topic

建立 Eventarc 觸發條件，將事件轉送至工作流程

如要自動執行工作流程，進而執行 Cloud Run 工作，請建立 Eventarc 觸發條件，回應 Pub/Sub 事件，並將事件傳送至工作流程。每當訊息寫入 Pub/Sub 主題，事件就會觸發工作流程的執行作業。

控制台

1. 前往 Google Cloud 控制台的「Workflows」頁面：

   前往「Workflows」頁面

2. 按一下工作流程名稱，例如 message-payload-workflow。

3. 在「工作流程詳細資料」頁面中，按一下「編輯」。

4. 在「Edit workflow」(編輯工作流程) 頁面的「Triggers」(觸發條件) 區段中，按一下「Add new trigger」(新增觸發條件) >「Eventarc」。

   「Eventarc trigger」(Eventarc 觸發條件) 窗格隨即開啟。

5. 在「觸發條件名稱」欄位中，輸入觸發條件的名稱，例如 message-payload-trigger。

6. 在「Event provider」(事件提供者) 清單中，選取「Cloud Pub/Sub」。

7. 在「Event」(事件) 清單中，選取「google.cloud.pubsub.topic.v1.messagePublished」。

8. 從「Select a Cloud Pub/Sub topic」(選取 Cloud Pub/Sub 主題) 清單中，選取您先前建立的 Pub/Sub 主題。

9. 在「Service account」(服務帳戶) 欄位中，選取您先前建立的服務帳戶。

   服務帳戶會做為觸發程序的 ID。您應該已將下列角色授予服務帳戶：

   • Eventarc 事件接收者：接收事件
   • 工作流程叫用者：執行工作流程

10. 按一下「儲存觸發條件」。

    現在「編輯工作流程」頁面的「觸發條件」部分會顯示 Eventarc 觸發條件。

11. 點選「下一步」。

12. 按一下 [Deploy] (部署)。

gcloud

執行下列指令，建立 Eventarc 觸發條件：

gcloud eventarc triggers create message-payload-trigger \
    --location=us-central1 \
    --destination-workflow=message-payload-workflow \
    --destination-workflow-location=us-central1 \
    --event-filters="type=google.cloud.pubsub.topic.v1.messagePublished" \
    --transport-topic=projects/PROJECT_ID/topics/message-payload-topic \
    --service-account=SERVICE_ACCOUNT_NAME@PROJECT_ID.

更改下列內容：

• PROJECT_ID：您的 Google Cloud 專案 ID
• SERVICE_ACCOUNT_NAME：先前建立的服務帳戶名稱。

服務帳戶會做為觸發程序的 ID。您應該已將下列角色授予服務帳戶：

• roles/eventarc.eventReceiver：接收事件
• roles/workflows.invoker：執行工作流程

觸發工作流程

將訊息發布至 Pub/Sub 主題並產生事件，測試端對端系統。詳情請參閱使用事件或 Pub/Sub 訊息觸發工作流程。

1. 將訊息發布至 Pub/Sub 主題，產生事件：

   gcloud pubsub topics publish message-payload-topic --message="Hello World"
   

   系統會將事件傳送至工作流程，該工作流程會記錄事件訊息、將事件資料儲存至 Cloud Storage bucket，並執行 Cloud Run 工作來處理儲存在 Cloud Storage 中的資料。這可能需要一分鐘的時間。

2. 查看工作執行作業，確認 Cloud Run 工作是否如預期執行：

   gcloud run jobs executions list --job=message-payload-job

   輸出內容中應會顯示新的工作執行作業。

3. 如要查看觸發工作流程所建立的事件相關記錄項目，請執行下列指令：

   gcloud logging read "resource.type=cloud_run_job AND textPayload:Payload"
   

4. 尋找類似下列內容的記錄項目：

   textPayload: "Payload: {'message': {'data': 'SGVsbG8gV29ybGQ=', 'messageId': '8834108235224238',\
   \ 'publishTime': '2023-09-20T17:07:52.921Z'}, 'subscription': 'projects/MY_PROJECT/subscriptions/eventarc-us-central1-message-payload-trigger-sub-741'}"
   ...
   resource:
   labels:
     job_name: message-payload-job
     location: us-central1
     project_id: MY_PROJECT
   type: cloud_run_job
   textPayload: Processing message payload gs://message-payload-MY_PROJECT/8254002311197919.data.json
   

5. 您可以查看 Cloud Storage bucket 物件中的事件資料，確認結果是否符合預期。

   1. 擷取 bucket 名稱：

      gcloud storage ls

      輸出結果會與下列內容相似：

      gs://message-payload-PROJECT_ID/

   2. 列出值區中的物件：

      gcloud storage ls gs://message-payload-PROJECT_ID/** --recursive

      畫面會顯示如下的輸出內容：

      gs://message-payload-PROJECT_ID/OBJECT_ID.data.json

      記下 OBJECT_ID，以便在下一個步驟中使用。

   3. 將 bucket 中的物件下載為檔案：

      gcloud storage cp gs://message-payload-PROJECT_ID/OBJECT_ID.data.json message-event.txt

      將 OBJECT_ID 替換為上一個步驟中傳回的 ID。

   4. 在文字編輯器中開啟 message-event.txt 檔案，寫入檔案的事件主體應類似下列內容：

      {
        "message": {
          "data": "SGVsbG8gV29ybGQ=",
          "messageId": "8254002311197919",
          "publishTime": "2023-09-20T16:54:29.312Z"
        },
        "subscription": "projects/MY_PROJECT/subscriptions/eventarc-us-central1-message-payload-trigger-sub-741"
      }
      

      請注意，如果從 Base64 格式解碼 SGVsbG8gV29ybGQ= 的資料值，系統會傳回「Hello World」。