Workflows のチュートリアル（第 1 世代）

注: この内容は、Cloud Run functions（旧称 Cloud Functions）の第 1 世代を対象としています。

このドキュメントの Cloud Run 関数のバージョンについては、Workflows のチュートリアルをご覧ください。

このチュートリアルでは、Workflows を使用して一連のサービスをリンクする方法について説明します。2 つのパブリック HTTP サービス（Cloud Run functions を使用）、外部 REST API、プライベート Cloud Run サービスを接続すると、柔軟でサーバーレスなアプリケーションを作成できます。

最初の Cloud Run functions サービスをデプロイする

HTTP リクエストを受信すると、この HTTP 関数は 1～100 の間の乱数を生成し、その数値を JSON 形式で返します。

1. randomgen という名前のディレクトリを作成し、そのディレクトリに移動します。

   mkdir ~/randomgen
   cd ~/randomgen

2. 次の Python コードを含むテキスト ファイルを main.py というファイル名で作成します。

   import functions_framework
   import random
   from flask import jsonify
   
   
   @functions_framework.http
   def randomgen(request):
       randomNum = random.randint(1, 100)
       output = {"random": randomNum}
       return jsonify(output)

3. HTTP 処理用の Flask への依存関係をサポートするには、pip パッケージ管理システムのテキスト ファイルを作成します。ファイル名を requirements.txt にして、以下を追加します。

   flask>=1.0.2
   functions-framework==3.0.0

4. HTTP トリガーを使用して関数をデプロイし、未認証アクセスを許可します。

   gcloud functions deploy randomgen \
       --runtime python37 \
       --trigger-http \
       --allow-unauthenticated

関数のデプロイには数分かかることがあります。また、 Google Cloud コンソールで Cloud Run functions のインターフェースを使用して関数をデプロイすることもできます。

1. 関数がデプロイされたら、httpsTrigger.url プロパティを確認できます。

   gcloud functions describe randomgen

2. 次の curl コマンドを使用して、この関数を試すことができます。

   curl $(gcloud functions describe randomgen --format='value(httpsTrigger.url)')

   数値がランダムに生成され、返されます。

2 つ目の Cloud Run functions サービスをデプロイする

HTTP リクエストを受信すると、この HTTP 関数は JSON 本文から input を抽出し、2 を乗算して、結果を JSON 形式で返します。

1. multiply という名前のディレクトリを作成し、そのディレクトリに移動します。

   mkdir ~/multiply
   cd ~/multiply

2. 次の Python コードを含むテキスト ファイルを main.py というファイル名で作成します。

   import functions_framework
   from flask import jsonify
   
   
   @functions_framework.http
   def multiply(request):
       request_json = request.get_json()
       output = {"multiplied": 2 * request_json['input']}
       return jsonify(output)

3. HTTP 処理用の Flask への依存関係をサポートするには、pip パッケージ管理システムのテキスト ファイルを作成します。ファイル名を requirements.txt にして、以下を追加します。

   flask>=1.0.2
   functions-framework==3.0.0

4. HTTP トリガーを使用して関数をデプロイし、未認証アクセスを許可します。

   gcloud functions deploy multiply \
       --runtime python37 \
       --trigger-http \
       --allow-unauthenticated

関数のデプロイには数分かかる場合があります。また、 Google Cloud コンソールで Cloud Run functions のインターフェースを使用して関数をデプロイすることもできます。

1. 関数がデプロイされたら、httpsTrigger.url プロパティを確認できます。

   gcloud functions describe multiply

2. 次の curl コマンドを使用して、この関数を試すことができます。

   curl $(gcloud functions describe multiply --format='value(httpsTrigger.url)') \
       -X POST \
       -H "content-type: application/json" \
       -d '{"input": 5}'

   数値 10 が返されます。

ワークフローで 2 つの Cloud Run functions サービスを接続する

ワークフローは、ワークフロー構文で記述された一連のステップで構成され、YAML 形式または JSON 形式のいずれでも記述できます。これがワークフローの定義です。詳細な説明については、構文リファレンスのページをご覧ください。

1. ホーム ディレクトリに戻ります。

   cd ~

2. 次の内容を含むテキスト ファイルを workflow.yaml という名前で作成します。

   - randomgen_function:
       call: http.get
       args:
           url: https://REGION-PROJECT_ID.cloudfunctions.net/randomgen
       result: randomgen_result
   - multiply_function:
       call: http.post
       args:
           url: https://REGION-PROJECT_ID.cloudfunctions.net/multiply
           body:
               input: ${randomgen_result.body.random}
       result: multiply_result
   - return_result:
       return: ${multiply_result}
   

   これにより、2 つの HTTP 関数がリンクされ、最終結果が返されます。

   注: 認証された呼び出しのみを受け入れる Cloud Run functions にリクエストを行うには、OIDC を使用して認証を行います。詳細については、ワークフローから認証済みリクエストを行うをご覧ください。

3. ワークフローを作成したら、デプロイして実行の準備ができます。

   gcloud workflows deploy WORKFLOW_NAME --source=workflow.yaml

   WORKFLOW_NAME は、ワークフローの名前に置き換えます。

4. ワークフローを実行します。

   gcloud workflows run WORKFLOW_NAME

   実行とは、ワークフローの定義に含まれるロジックを 1 回だけ実行することです。すべてのワークフローの実行は独立しており、Workflows の迅速なスケーリングにより、多数の同時実行が可能になります。

   ワークフローの実行後、出力は次のようになります。

   result: '{"body":{"multiplied":120},"code":200,"headers":{"Alt-Svc":"h3-29=\":443\";
   ...
   startTime: '2021-05-05T14:17:39.135251700Z'
   state: SUCCEEDED
   ...
   

ワークフローでパブリック REST サービスを接続する

既存のワークフローを更新し、数式を評価できるパブリック REST API（math.js）を接続します。例: curl https://api.mathjs.org/v4/?'expr=log(56)'

ワークフローはデプロイしてあるため、 Google Cloud コンソールの [ワークフロー] ページから編集することもできます。

1. ワークフローのソースファイルを編集し、次の内容で置き換えます。

   - randomgen_function:
       call: http.get
       args:
           url: https://REGION-PROJECT_ID.cloudfunctions.net/randomgen
       result: randomgen_result
   - multiply_function:
       call: http.post
       args:
           url: https://REGION-PROJECT_ID.cloudfunctions.net/multiply
           body:
               input: ${randomgen_result.body.random}
       result: multiply_result
   - log_function:
       call: http.get
       args:
           url: https://api.mathjs.org/v4/
           query:
               expr: ${"log(" + string(multiply_result.body.multiplied) + ")"}
       result: log_result
   - return_result:
       return: ${log_result}
   

   これにより、外部 REST サービスが Cloud Run functions サービスにリンクされ、最終結果が返されます。

2. 変更されたワークフローをデプロイします。

   gcloud workflows deploy WORKFLOW_NAME --source=workflow.yaml

Cloud Run サービスをデプロイする

HTTP リクエストを受信後に JSON 本文から input を抽出してその math.floor を計算し、結果を返す Cloud Run サービスをデプロイします。

1. floor という名前のディレクトリを作成し、そのディレクトリに移動します。

   mkdir ~/floor
   cd ~/floor

2. 次の Python コードを含むテキスト ファイルを app.py というファイル名で作成します。

   import json
   import logging
   import os
   import math
   
   from flask import Flask, request
   
   app = Flask(__name__)
   
   
   @app.route('/', methods=['POST'])
   def handle_post():
       content = json.loads(request.data)
       input = float(content['input'])
       return f"{math.floor(input)}", 200
   
   
   if __name__ != '__main__':
       # Redirect Flask logs to Gunicorn logs
       gunicorn_logger = logging.getLogger('gunicorn.error')
       app.logger.handlers = gunicorn_logger.handlers
       app.logger.setLevel(gunicorn_logger.level)
       app.logger.info('Service started...')
   else:
       app.run(debug=True, host='0.0.0.0', port=int(os.environ.get('PORT', 8080)))

3. 同じディレクトリに、次の内容の Dockerfile を作成します。

   # Use an official lightweight Python image.
   # https://hub.docker.com/_/python
   FROM python:3.7-slim
   
   # Install production dependencies.
   RUN pip install Flask gunicorn
   
   # Copy local code to the container image.
   WORKDIR /app
   COPY . .
   
   # Run the web service on container startup. Here we use the gunicorn
   # webserver, with one worker process and 8 threads.
   # For environments with multiple CPU cores, increase the number of workers
   # to be equal to the cores available.
   CMD exec gunicorn --bind 0.0.0.0:8080 --workers 1 --threads 8 app:app

4. Docker コンテナ イメージを保存できる Artifact Registry 標準リポジトリを作成します。

   gcloud artifacts repositories create REPOSITORY \
       --repository-format=docker \
       --location=${REGION}

   REPOSITORY は、リポジトリの一意の名前に置き換えます。

5. コンテナ イメージをビルドします。

   export SERVICE_NAME=floor
   gcloud builds submit --tag ${REGION}-docker.pkg.dev/PROJECT_ID/REPOSITORY/${SERVICE_NAME}

6. コンテナ イメージを Cloud Run にデプロイし、認証された呼び出しのみが受け入れられるようにします。

   gcloud run deploy ${SERVICE_NAME} \
       --image ${REGION}-docker.pkg.dev/PROJECT_ID/REPOSITORY/${SERVICE_NAME}:latest \
       --platform managed \
       --no-allow-unauthenticated

サービス URL が表示されたら、デプロイは完了しています。ワークフロー定義を更新するときに、この URL を指定する必要があります。

ワークフローで Cloud Run サービスを接続する

既存のワークフローを更新して、Cloud Run サービスの URL を指定します。

1. ワークフローのソースファイルを編集し、次の内容で置き換えます。

   - randomgen_function:
       call: http.get
       args:
           url: https://REGION-PROJECT_ID.cloudfunctions.net/randomgen
       result: randomgen_result
   - multiply_function:
       call: http.post
       args:
           url: https://REGION-PROJECT_ID.cloudfunctions.net/multiply
           body:
               input: ${randomgen_result.body.random}
       result: multiply_result
   - log_function:
       call: http.get
       args:
           url: https://api.mathjs.org/v4/
           query:
               expr: ${"log(" + string(multiply_result.body.multiplied) + ")"}
       result: log_result
   - floor_function:
       call: http.post
       args:
           url: CLOUD_RUN_SERVICE_URL
           auth:
               type: OIDC
           body:
               input: ${log_result.body}
       result: floor_result
   - create_output_map:
       assign:
         - outputMap:
             randomResult: ${randomgen_result}
             multiplyResult: ${multiply_result}
             logResult: ${log_result}
             floorResult: ${floor_result}
   - return_output:
       return: ${outputMap}
   

   CLOUD_RUN_SERVICE_URL は、Cloud Run サービス URL に置き換えます。

   これにより、ワークフローで Cloud Run サービスが接続されます。auth キーを使用すると、Cloud Run サービスの呼び出しで認証トークンが渡されます。詳細については、ワークフローから認証済みリクエストを行うをご覧ください。

2. 変更されたワークフローをデプロイします。

   gcloud workflows deploy WORKFLOW_NAME \
       --source=workflow.yaml

3. 最終ワークフローを実行します。

   gcloud workflows run WORKFLOW_NAME

   出力は次のようになります。

   result: '{"Floor":{"body":"4","code":200
     ...
     "Log":{"body":"4.02535169073515","code":200
     ...
     "Multiply":{"body":{"multiplied":56},"code":200
     ...
     "Random":{"body":{"random":28},"code":200
     ...
   startTime: '2023-11-13T21:22:56.782669001Z'
   state: SUCCEEDED
   

これで完了です。一連のサービスを接続するワークフローのデプロイと実行が完了しました。

式、条件付きジャンプ、Base64 エンコードまたはデコード、サブワークフローなどを使用して複雑なワークフローを作成するには、Workflows 構文リファレンスと標準ライブラリの概要をご覧ください。