💬

【GCP】Workflowsを触ってみた

2021/03/19に公開

Google Cloud

Workflows

tech

はじめに

GCPのWorkflowsを触ってみたので備忘録です。

この記事に書いていることは下記になります。

ワークフローを動かしかた
メインとなるワークフロー構文を確認
cloudfunctionsと繋いで確認

GCP Workflowsとは

GCPのWorkflowsはAPI呼び出しやCloud Functions, Cloud Runなどのサービスを組み合わせてワークフローを構築することができます。
AWSのStep Functionsと似たようなサービスです。(触ってみてはStep Functionsのほうができることが多いと思いました。)

マイクロサービスで設計されているとき、複数のサービスに跨って処理を行うことがあると思います。
その場合はイベント処理等を使用すると思いますが、使用するのリソースが多くなればなるほど全体のフローを把握することは大変になります。

例えば次のようにCloudFunctionsからpubsubを経由してイベント実行している場合、ドキュメントが存在しないと実装からワークフローを把握することが大変です。
また、アプリケーションの都合でCloudFunctionsでは本来の処理に関係ない、pubsubのメッセージ作成もする必要が発生します。

こういった時にWorkflowsを使用すると処理の遷移が把握できて、各処理はメインの処理のみに集中できます。

ワークフローを試す

まずはGCPコンソールからワークフローを作って実行してみます。

GCPコンソールからワークフローを選択し、作成をクリック

ワークフローなどを入力。2021/3/19時点では3リージョンからの選択でした。

サンプルのワークフローが作られるので、そのままデプロイします。

ワークフローの実行をクリックする

実行結果が表示されます。

ワークフローの定義を行うとUIで実行順が確認できるので処理の流れが分かりやすいです。

ワークフローの作成

ワークフローはYAMLもしくはJSONのワークフロー構文で作成します。
実行時にパラメータを受け取ったり、条件分岐、ループなども行うことが可能です。

Steps

ワークフローには最低１つのStepが必要になります。
そして、stepには下記がサポートされています。

HTTPエンドポイントの呼び出し
変数の割り当て
スリープ
条件付き分岐
値を返す

HTTPエンドポイントの呼び出し

HTTPエンドポイントは下記のように定義できます。
公式リファレンス

- STEP_NAME:
  # 必須項目でhttp.get、http.post、http.requestから選択する
  call: {http.get|http.post|http.request}
  args:
    # 必須項目でリクエストをするURLを記載する
    url: URL_VALUE
    # callをhttp.requestにした場合、必須項目となる。GET,POST,PATCH,DELETEのようなリクエストメソッドを記載
    method: REQUEST_METHOD
    # httpリクエストのヘッダを記載
    headers:
        KEY:VALUE
    # httpリクエストのbodyを記載
    body:
        KEY:VALUE
    # httpリクエストのqueryを記載
    query:
        KEY:VALUE
    # APIに認証が必要な場合に使用する
    auth:
        type:{OIDC|OAuth2}
    # タイムアウトの秒数を指定できる
    timeout: VALUE_IN_SECONDS
  # レスポンスの結果を格納する
  result: RESPONSE_VALUE

例えば下記のようなワークフローを定義すると

- 日付を取得:
  call: http.get
  args:
    # ワークフローのサンプルAPIが用意されている
    url: https://us-central1-workflowsample.cloudfunctions.net/datetime
  result: currentTime
- 結果確認:
  return: ${currentTime}

UIではこのように表示されます。

実行結果は下記になります。

{
  "body": {
    "dayOfTheWeek": "Thursday"
  },
  "code": 200,
  "headers": {
    "Alt-Svc": "h3-29=\":443\"; ma=2592000,h3-T051=\":443\"; ma=2592000,h3-Q050=\":443\"; ma=2592000,h3-Q046=\":443\"; ma=2592000,h3-Q043=\":443\"; ma=2592000,quic=\":443\"; ma=2592000; v=\"46,43\"",
    "Cache-Control": "private",
    "Content-Length": "28",
    "Content-Type": "application/json",
    "Date": "Thu, 18 Mar 2021 15:45:28 GMT",
    "Function-Execution-Id": "65g9tfc9ngh5",
    "Server": "Google Frontend",
    "X-Cloud-Trace-Context": "e632a6e77f1d1dc1c827d597a7b02fe8;o=1"
  }
}

変数の割り当て

変数の初期化、更新ができます。変数は別のstepで使用することもできます。

- STEP_NAME:
  assign:
    - VARIABLE_NAME: VALUE

下記のワークフローを定義して実行すると、6が出力されます。

- 変数初期化:
  assign:
    - number: 5
    - number_plus_one: ${number+1}
    - other_number: 10
    - string: "hello"
- 結果確認:
   return: ${number_plus_one}

スリープ

ワークフローの実行を一時停止することができます。
APIの反映待ちやバッチで一時停止するときなどに使用できます。

- STEP_NAME:
  call: sys.sleep
  args:
    # スリープする秒数
    seconds: SLEEP_IN_SECONDS

下記のワークフローを定義して実行すると

- 変数初期化:
    assign:
        - number: 5
        - number_plus_one: ${number+1}
        - other_number: 10
        - string: "hello"
- スリープ:
    call: sys.sleep
    args:
        seconds: 10
- 結果確認:
   return: ${number_plus_one}

実行時間が10秒になっていることが分かります。

分岐

switch構文で条件分岐を表現できます。

- STEP_NAME_A:
    # 最大10個のconditionを追加できる
    switch:
        - condition: ${EXPRESSION_A}
          # nextのstepに移行する
          next: STEP_NAME_B
        - condition: ${EXPRESSION_B}
          next: STEP_NAME_C
    # どの条件にも引っかからない場合
    next: STEP_NAME_D

下記のワークフローを定義すると

- 日付取得:
    call: http.get
    args:
      url: https://us-central1-workflowsample.cloudfunctions.net/datetime
    result: currentTime
- 曜日チェック:
    switch:
      - condition: ${currentTime.body.dayOfTheWeek == "Friday"}
        next: 金曜日
      - condition: ${currentTime.body.dayOfTheWeek == "Saturday" OR currentTime.body.dayOfTheWeek == "Sunday"}
        next: 週末
    next: 仕事の曜日
- 金曜日:
    return: "金曜日!"
- 週末:
    return: "週末！！"
- 仕事の曜日:
    return: "仕事の曜日"

こういった表示になります。

ワークフローの終了

例外が投げられた時もしくは、end、returnが存在しているときにワークフローは終了します。

- STEP_NAME:
    ...
    next: end

- STEP_NAME:
    ...
    return: ${VARIABLE}

ループの表現

ループはnextで前のstepを指定することで可能です。

このワークフローを定義すると

- define:
    assign:
      - array: ["foo", "ba", "r"]
      - result: ""
      - i: 0
- check_condition:
    switch:
      - condition: ${len(array) > i}
        next: iterate
    next: exit_loop
- iterate:
    assign:
      - result: ${result + array[i]}
      - i: ${i+1}
    next: check_condition
- exit_loop:
    return:
        concat_result: ${result}

こういった表示になります。

実行時のパラメータの受け取り

実行時のパラメータを受け取ることが可能です。

main:
  # 受け取るパラメータを定義
  params: [args]
  steps:
  - step1:
      assign:
        - outputVar: ${"Hello " + args.firstName + " " + args.lastName}
  - step2:
      return: ${outputVar}

実行時に下記オブジェクトを渡すことが可能です。
{"firstName":"Sherlock","lastName":"Holmes"}

ログの出力

ワークフロー実行のログは自動生成されないため、明示的に作成する必要があります。
sys.logを使うことで出力ができます。

- log:
    call: sys.log
    args:
      text: "ログ出力"
      severity: "INFO"

cloudfunctionとの連携

cloudfunctionとの疎通を確認します。(とりあえず繋げて動かすだけです)

cloudfunctionの関数

index.js

exports.helloWorld = (req, res) => {
  res.status(200).send({
    status: req.body.status,
    count: req.body.count + 1
  });
};

ワークフロー定義

- 変数初期化:
    assign:
      - status: "start"
      - count: 0
- ファンクション実行:
    call: http.post
    args:
      url: https://asia-northeast1-${PROJECT_NAME}.cloudfunctions.net/workflow-sample
      auth:
        # これを指定することで認証が可能になる
        type: OIDC
      body:
        status: ${status}
        count: ${count}
    result: returnValue
- ステータス確認:
    switch:
      - condition: ${returnValue.body.status == "start"}
        next: ステータスをupdateに更新
      - condition: ${returnValue.body.status == "create"}
        next: ステータスをcreateに更新
    next: 終了
- ステータスをupdateに更新:
    assign:
      - status: "update"
      - count: ${returnValue.body.count}
    next: スリープ
- スリープ:
    call: sys.sleep
    args:
      # スリープする秒数
      seconds: 10
    next: ファンクション実行
- ステータスをcreateに更新:
    assign:
      - status: "create"
      - count: ${returnValue.body.count}
    next: ファンクション実行
- 終了:
    return: ${returnValue}

実行すると全ての分岐を通って完了します。

おわりに

この他にも例外処理やリトライ処理もワークフローで定義可能のため、業務でも使える印象です。
しかし、現在実行中のタスクをUIで確認できない。並列でタスクを処理できないなどもありStep Functionsを比べると機能的に劣っている印象です。今後の機能追加に期待したいです！！

また、ワークフローの詳細の実行数に何も表示されなかったのですが、よく分かりませんでした...

参考資料

ワークフローのサンプルがたくさんあります。
https://cloud.google.com/workflows/docs/sample-workflows?hl=ja
ワークフローの構文リファレンスです
https://cloud.google.com/workflows/docs/reference/syntax?hl=ja
GCP Workflows試してみた
https://zenn.dev/ogataka50/articles/fde9633be02080