GCP Workflows使って調査結果のcsvデータ作成＋調査結果のヒートマップhtmlのビルドを行う仕組みを作った話

弊社が提供しているタカノメサービスについて、GCP Workflowsを用いることで調査結果の出力フローを自動化することができました。本稿では、構築したワークフローの概要を紹介いたします。

<!-- more -->


## 開発背景
陸ごみ散布状況の調査サービスであるタカノメにおいて、調査のたびに調査結果を出力しています。これまではローカル上でPythonスクリプトを動かす必要があり、コードバージョンや秘密鍵の管理、環境構築面での課題がありました。そこで、社内の管理ページから調査結果を出力指示できるようにし、運用スタッフがコードを触らなくても良いように改善することとなりました。

## ワークフローの構築
下図の様に構築しました。大まかな流れは以下の通りです。

1. 変数定義: GCPプロジェクトID(projectId), ワークフローの実行ID(executionId), Cloud BuildのトリガーID(triggerId)を定義する
1. 変数マッピング: firestoreで配列型のフィールドを追加するため、配列データを型変換する
1. 管理データ作成: 調査結果の管理データを作成する。このデータで、ワークフローの処理成否も記録する
1. 通常処理
    1. Cloud Functionsにて、調査結果データ(csv, xlsx, json)を作成する
    1. Cloud Buildにて、調査結果のヒートマップhtmlを作成する
    1. 管理データにワークフローの処理OKを記録する
1. 例外処理: 調査結果を作成できなかった場合、管理データにワークフローの処理NGを記録する

![タカノメ 調査結果作成ワークフロー](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1967276/37858567-f9ae-4eb3-c36d-b0b1a1c349cd.png)

### 1. 変数定義
以下の通り、変数を定義しました。

```yaml
main:
    params: [input]  # ワークフロー呼び出し時に指定したパラメータがinputに入る
    steps:
    - setParams:
        assign:
            - projectId: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}  # 実行しているワークフローのプロジェクトID
            - executionId: ${sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}  # ワークフローの実行ID
            - triggerId: "実行したいCloud BuildのトリガーID"
            - targetIdsForFiretore: []   # 次のステップで変数マッピングする際に値を設定する
```

GCP Workflowsではいくつかの環境変数が事前に定義されていて、ワークフロー上で読み出すことができます[^1]。

### 2. 変数マッピング
ワークフローからfirestoreのフィールドを追加するとき、[Method: googleapis.firestore.v1.projects.databases.documents.createDocument](https://cloud.google.com/workflows/docs/reference/googleapis/firestore/v1/projects.databases.documents/createDocument)で定義されているフォーマットに従ってフィールドを指定する必要があります。プリミティブな型(string, timestampなど)であれば、以下の様に `フィールド名.型Value`の形式で指定すればOKです。

```yaml
            fields:
              hoge:
                stringValue: ${input.hoge}
              foo:
                timestampValue: ${time.format(sys.now())}
```

問題は配列型のデータの場合で、json形式での配列をそのまま`フィールド名.arrayValue`に渡してもエラーになります。arrayValueは

```yaml
            fields:
              bar:
                arrayValue:
                  values:
                    - stringValue: "aaa"
                    - stringValue: "bbb"
                    - stringValue: "ccc"
```

という形で、arrayValue.values[]下に stringValue: "xxx" や timestampValue: `YYYY-MM-DDTHH:mm:ss` の配列を渡す必要があります[^2]。

そこで、以下の様にfor文でlist型のtargetIdsForFiretoreに要素を足し込んでいくことで対応しています。

```yaml
    - mappingForFirestore:
        for:
            value: v
            in: ${input.targetIds}
            steps:
                - getStep:
                    assign:
                        - newValue:
                            stringValue: ${v}
                        - targetIdsForFiretore: ${list.concat(targetIdsForFiretore, newValue)}
```

### 3. 管理データ作成
前ステップで準備したtargetIdsForFiretoreおよびinputに渡した各パラメータを基に、firestoreに管理データdocumentを追加します。

callにて[googleapis.firestore.v1.projects.databases.documents.createDocument](https://cloud.google.com/workflows/docs/reference/googleapis/firestore/v1/projects.databases.documents/createDocument)コネクタを指定し、argsにcollection名やdocumentのIDを指定することでdocumentを作成できます。なお、timestampValueにはISO8601フォーマットの日時文字列を渡すことができます。

```yaml
    - createSurveySummaryFirestore:
        call: googleapis.firestore.v1.projects.databases.documents.createDocument
        args:
          parent: ${ "projects/" + projectId + "/databases/(default)/documents" }
          collectionId: surveysummary
          documentId: ${executionId}  # documentはワークフローごとに作成する
          body:
            fields:
              targetIds:
                arrayValue:
                  values: ${targetIdsForFiretore}
              name:
                stringValue: ${input.name}
              createdAt:
                timestampValue: ${time.format(sys.now())}
              status:
                stringValue: "prepare"
        result: resp
```

処理結果で403が返る場合、ワークフローを実行しているサービスアカウントのIAM権限でFirestoreの書き込み権限があるか確認してください。

### 4. 通常処理
調査結果データ(csv, xlsx)の作成と、ヒートマップhtmlの作成を順に行います。前者はCloud Functions、後者はCloud Buildで実施しますが、いずれも状況によっては失敗する可能性があります。

そこで、通常処理をtry文、例外処理をexcept文で対処します。

```yaml
    - createSurveyResultSteps:
        try:
            steps:
            - accumulateSurveyResult:
               # 通常時の処理1: Cloud Functionsを呼び出し、調査結果データを作成する
            - buildRenderer:
               # 通常時の処理2: Cloud Buildを呼び出し、ヒートマップhtmlを作成する
            - updateSurveySummaryFirestoreOnSuccess:
               # 通常時の処理3: 管理データについて、処理成功を記録する
        except:
            as: e
            steps:
            - logStep:
               # 例外時の処理1: エラーをロギングする
            - updateSurveySummaryFirestoreOnError:
               # 例外時の処理2: 管理データについて、処理失敗を記録する
```

#### 4.1. Cloud Functionsにて、調査結果データ(csv, xlsx, json)を作成する
事前に定義したFunctionのcreate_survey_resultを呼び出します。本Functionはus-east1に定義してあるので、下記の通り呼び出します。

```yaml
            - accumulateSurveyResult:
                call: http.post
                args:
                    url: ${ "https://us-east1-" + $projectId + ".cloudfunctions.net/create_survey_result" }
                    body:
                        request_id: ${executionId}
                        target_ids: ${input.targetIds}
                    auth:
                        type: OIDC
                result: response

```

このFunctionは要認証で定義しています。ワークフローにおいては、`auth.type = OIDC`を指定することでサービスアカウントのOpenId Connect トークンをAuthorizationヘッダのBearerに付与することができます。こうして、Functionを認証付きで実行することができます。

#### 4.2. Cloud Buildにて、調査結果のヒートマップhtmlを作成する
Cloud Buildを呼び出すには、[googleapis.cloudbuild.v1.projects.triggers.run](https://cloud.google.com/workflows/docs/reference/googleapis/cloudbuild/v1/projects.triggers/run?hl=ja) コネクタを使用します。呼び出しにあたり、projectIdおよびtriggerIdの指定が必要です。また、body.substitutions._変数名 に値を渡すことでビルド時の変数値を指定できます。

```yaml
            - buildRenderer:
                call: googleapis.cloudbuild.v1.projects.triggers.run
                args:
                  projectId: ${projectId}
                  triggerId: ${triggerId}
                  body:
                    substitutions:
                      _REQUEST_ID: ${executionId}  # ワークフローの実行IDを渡し、前ステップで作成した調査結果データを参照できるようにする
                result: runResult
```

#### 4.3. 管理データにワークフローの処理OKを記録する
一連の処理が成功したことを管理データに記録します。[googleapis.firestore.v1.projects.databases.documents.patch](https://cloud.google.com/workflows/docs/reference/googleapis/firestore/v1/projects.databases.documents/patch)によりdocumentをupsertできます。今回は処理結果のフィールド(status)だけを更新したいので、updateMaskでstatusを指定します。

```yaml
            - updateSurveySummaryFirestoreOnSuccess:
                call: googleapis.firestore.v1.projects.databases.documents.patch
                args:
                  name: ${ "projects/" + projectId + "/databases/(default)/documents/surveysummary/" + executionId}  # ここでcollectionIdとdocumentIdを指定する
                  updateMask:
                    fieldPaths: [ "status" ]  # 値を更新したいフィールドの一覧をここで指定する
                  body:
                    fields:
                      status:
                        stringValue: "success"
                result: resp
```

### 5. 例外処理
通常処理の途中で例外が発生し、調査結果を作成できないときに本フローを実行します。本フローではエラーが発生したことの記録が必要だったため、ロギングへのログ出力と管理データへの失敗記録を行っています。

```yaml
        except:
            as: e
            steps:
            - logStep:
                call: sys.log
                args:
                    text: ${json.encode_to_string(e)}
                    severity: ERROR
            - updateSurveySummaryFirestoreOnError:  # 記載省略。4.3節の内容からstringValueへの代入値を変えただけになるため
```

## ワークフローの呼び出し
フロントエンド側から直接呼び出すことはできないため、Workflows呼び出し用のCloud Functionsを定義しました。

Cloud Functions for Firebaseであれば、公式の[@google-cloud/workflows](https://www.npmjs.com/package/@google-cloud/workflows)ライブラリを利用できます。createExecutionを呼び出す際、execution.argumentにJSON文字列を渡します。すると、ワークフローに実行時引数(ワークフローのyaml中のinput変数)にデータを渡すことができます。

```typescript
import * as functions from "firebase-functions";
import { ExecutionsClient, WorkflowsClient } from "@google-cloud/workflows";

export const dispatchCreateSurveyResultWorkflow = functions.https.onCall(
  async (data: Record<string, unknown>) => {
    const config: Record<string, string> = JSON.parse(
      process.env.FIREBASE_CONFIG || "{}"
    );
    const projectId = config["projectId"] || "your-project-id";
    const region = "us-east1";
    if (!data["name"] || !data["targetIds"]) {
      return;
    }

    const client = new ExecutionsClient();
    const parent = new WorkflowsClient().workflowPath(
      projectId,  
      region,  // Workflowsのあるリージョン
      "create_survey_result_workflow"  // Workflows名
    );
    await client.createExecution({
      parent,
      execution: {
        argument: JSON.stringify(data),
      },
    });
    return {};
  }
);
```

あとはフロントエンドからhttpsCallableなどを使って本Functionを呼び出せば、ワークフローを走らせることができます。

[^1]: https://cloud.google.com/workflows/docs/reference/environment-variables
[^2]: https://cloud.google.com/workflows/docs/reference/googleapis/firestore/v1/Overview#arrayvalue