Workflows で実現するサーバーレスなワークフロー

Published in

google-cloud-jp

9 min readDec 11, 2020

この記事は Google Cloud Japan Customer Engineer Advent Calendar 2020 の 12日目の記事です。本記事ではサーバーレスワークフローを実現するWorkflows というプロダクトを紹介していきます。

TL;DR

Workflows で信頼性の高いワークフローを実行できる
ステップ間で変数の引き渡しも可能
エラーハンドリングもビルトインでサポートされている

はじめに

Google Cloud Next OnAir 2020 で Google Cloud Workflows（12/12 時点でPreview）という新しいプロダクトが紹介されました。Cloud Workflows を利用することで、Google Cloud と HTTP ベースのワークフローのオーケストレーションが可能になります。

この記事では、プロダクトのことを Workflows、ワークフローそのものや、Workflows にデプロイしたリソースはカタカナでワークフローと記述します。

Workflows の概要

Workflows はワークフロータスクをリンクさせることのできる、サーバーレスでマネージドなワークフローツールです。Workflows はサーバーレスなのでインフラを意識することなくスケーリングし、また利用がないときはゼロにスケールイン可能です。スケールが非常に早いので、レイテンシーが重要となるときにも向いていそうですね。

Cloud Composer というワークフロー管理のプロダクトも Google Cloud にはありますが、こちらはデフォルトでGKE環境が立ち上がり、Apache Airflow の記述に沿ってワークフローを記述していきます。

Workflows はサーバーレスで動作し、YAML でワークフローを定義するためシンプルで軽量なワークフローを簡単に実現することが可能です。プログラミングをしない方でも簡単に使い始めることができるでしょう。

ユースケース

Workflows のユースケースとしては、例えば以下のようなものをセッション動画で紹介しています。

請求書の作成

請求書を作成
請求書を PDF を変換
作成した PDF をユーザーに送信する

支払期限を過ぎている顧客にリマインダーを送信

顧客リストを取得
顧客ごとに支払期限を過ぎているか確認
支払期限を過ぎている場合にはリマインダーを送信する

IT 自動化ワークフロー

VM インスタンスを起動
60 秒待機
アプリケーションの起動を確認
起動していない場合はログに記述し、メンバーに通知する

他にもデータパイプラインやビルドパイプラインにも利用出来るのではないでしょうか。例えば Dataflow で Job を実行すると、Job Id が取得できますので、定期的に Job の実行結果を監視し、正常終了したら次の Dataflow Job を実行するといった形です。Cloud Scheduler と組み合わせることで、定時実行する Job をワークフローに組み込むことが可能です。

Workflows の使い方

1. 定義を作成する

まず Workflows は、ワークフローの定義を YAML 形式で記述します。構文はこちらを参照してください。

2. デプロイする

作成したワークフロー定義ファイルを、Workflows にデプロイします。デプロイが完了するとすぐに実行が可能です。

3. 実行する

実行は定義に含まれたロジックの 1 回の処理を指し、実行されたステップに対して課金されます。すべての実行は独立して処理されます。

Workflows の定義

ステップ

Workflows の定義は、複数のステップを組み合わせて作成していきます。ステップは一つのワークフローを構成する処理の最小単位です。HTTP のエンドポイントにリクエストを投げたり、レスポンスを分解して変数への代入する、処理が完了するまでスリープする、といった単位で分割することになると思います。

- step1:
    ...
- step2:
    ...

定義の中で、変数の宣言が可能です。

- assign_vars:
    assign:
        - number: 5
        - number_plus_one: ${number+1}
        - other_number: 10
        - string: "hello"

スリープ処理は以下のような構文で作成します

- sleep:
    call: sys.sleep
    args:
        seconds: 1

ジャンプ

エンドポイントからのレスポンスに応じて処理を分けたい場合、ジャンプが有効です。Switch 構文が用意されているので、分岐先のステップを指定することで処理をジャンプしたり、繰り返すことが出来ます。

- conditionalSwitch:
    switch:
      - condition: ${currentTime.body.dayOfTheWeek == "Friday"}
        next: friday
      - condition: ${currentTime.body.dayOfTheWeek == "Saturday" OR currentTime.body.dayOfTheWeek == "Sunday"}
        next: weekend
    next: workWeek

ワークフローの終了

ワークフローの実行のみで完結する場合には、next: end を定義すると、処理が終了します。何かしらの値を返したい場合は、最後に実行されるステップで return: value することで値を返すことが可能です。

エラーハンドリング

Workflows は、エラーが発生したときの処理もビルトインでサポートしています。定義に含めることでエラー時には通知するといったワークフローも作成可能です。

- step_a:
    raise:
        code: 55
        message: "Something went wrong."

上記の構文サンプルはリファレンス（https://cloud.google.com/workflows/docs/reference/syntax）を参照しています。他にも多くの構文がありますので、是非リファレンスを参照ください！

実際に使ってみよう!

さて、実際に Workflows を使ってみましょう。サンプルが用意されているので、それを利用します。チュートリアルは、エンドポイント（https://us-central1-workflowsample.cloudfunctions.net/datetime）から、今日の曜日を取得し、Wikipedia の API で曜日に関連するウィキペディアの記事のリストを取得しています。

事前準備
課金が有効になっているプロジェクトで、Workflows の API を有効化します。

ワークフローの構成

1. 作成をクリックし、作成するワークフローの構成を入力していきます。現時点ではリージョンは us-central1 が利用可能です。入力し、次へをクリックします。

2. ワークフローの定義は、こちらのエディタで行います。このままデプロイをクリックします。

チュートリアルのコードは、以下のように 3 ステップで構成されています。ステップ：getCurrentTime で取得した曜日を CurrentDateTime に格納しています。この CurrentDateTime は値を取り出し、次のステップのクエリパラメータとして利用しています。

- getCurrentTime:
    call: http.get
    args:
        url: https://us-central1-workflowsample.cloudfunctions.net/datetime
    result: CurrentDateTime
- readWikipedia:
    call: http.get
    args:
        url: https://en.wikipedia.org/w/api.php
        query:
            action: opensearch
            search: ${CurrentDateTime.body.dayOfTheWeek}
    result: WikiResult
- returnOutput:
    return: ${WikiResult.body[1]}