このページは Cloud Translation API によって翻訳されました。

コールバックを使用して待機する

コールバックを使用すると、ワークフローの実行では、別のサービスがコールバックエンドポイントにリクエストを行うことを待ち、そのリクエストによってワークフローの実行を再開できます。

コールバックを使用すると、ポーリングせずに指定したイベントを待機して、その発生をワークフローに知らせることが可能です。たとえば、商品の在庫が補充されたときや商品が発送されたときに通知を受け取るワークフローを作成できます。あるいは、注文の確認や翻訳の確認など、人間による操作ができるように待機することも可能です。

このページでは、コールバックエンドポイントをサポートし、外部プロセスからの HTTP リクエストがそのエンドポイントに到着するまで待機するワークフローを作成する方法について説明します。コールバックと Eventarc トリガーを使用してイベントを待機することもできます。

コールバックでは、次の 2 つの標準ライブラリ組み込み関数を使用する必要があります。

events.create_callback_endpoint - 指定された HTTP メソッドを想定するコールバックエンドポイントを作成します。
events.await_callback - コールバックが指定されたエンドポイントで受信されるのを待ちます。

コールバックリクエストを受信するエンドポイントを作成する

そのエンドポイントに到着する HTTP リクエストを受信できるコールバックエンドポイントを作成します。

手順に沿って新しいワークフローを作成するか、既存のワークフローを選択して更新しますが、まだデプロイはしないでください。
ワークフローの定義で、コールバックエンドポイントを作成するステップを追加します。
YAML
- create_callback: call: events.create_callback_endpoint args: http_callback_method: "METHOD" result: callback_details
JSON
[ { "create_callback": { "call": "events.create_callback_endpoint", "args": { "http_callback_method": "METHOD" }, "result": "callback_details" } } ]
METHOD は、予想される HTTP メソッド（GET、HEAD、POST、PUT、DELETE、OPTIONS、PATCH のいずれか）に置き換えます。デフォルト値は POST です。

結果のマップは callback_details で、作成したエンドポイントの URL を格納する url フィールドが含まれます。

これで、コールバックエンドポイントは、指定された HTTP メソッドを使用して受信リクエストを受信できるようになりました。作成したエンドポイントの URL を使用して、ワークフロー外部のプロセスからのコールバックをトリガーできます。たとえば、URL を Cloud Run 関数に渡します。

注: 同じワークフローには、複数のコールバックを含めることができます。
ワークフローの定義で、コールバックリクエストを待つステップを追加します。
YAML
- await_callback: call: events.await_callback args: callback: ${callback_details} timeout: TIMEOUT result: callback_request
JSON
[ { "await_callback": { "call": "events.await_callback", "args": { "callback": "${callback_details}", "timeout": TIMEOUT }, "result": "callback_request" } } ]
TIMEOUT は、ワークフローがリクエストを待機する最大秒数に置き換えます。デフォルトは 43200　(12 時間) です。リクエストを受信する前に時間が経過した場合は、TimeoutError が発生します。

最大実行時間があるので注意してください。詳細については、リクエストの上限をご覧ください。

引数には、以前の create_callback ステップの callback_details マップが渡されます。

注: await_callback が呼び出される前に複数のリクエストがコールバックエンドポイントに送信された場合は、最初のリクエストのみが受け入れられて返されます。後続のリクエストは、RESOURCE_EXHAUSTED の HTTP ステータスコードで拒否されます。

ワークフローをデプロイして、作成または更新を完了します。

リクエストを受信すると、リクエストのすべての詳細が callback_request マップに保存されます。そうすると、ヘッダー、本文、クエリパラメータの query マップなど、HTTP リクエスト全体にアクセスできます。次に例を示します。

YAML

    http_request:
      body:
      headers: {...}
      method: GET
      query: {}
      url: "/v1/projects/350446661175/locations/us-central1/workflows/workflow-1/executions/46804f42-dc83-46d6-87e4-93962866ed81/callbacks/49c80102-74d2-49cd-a70e-805a9fded94f_2de9b413-6332-412d-99c3-d7e9b6eeeda2"
    received_time: 2021-06-24 12:49:16.988072651 -0700 PDT m=+742581.005780667
    type: HTTP

JSON

    {
       "http_request":{
          "body":null,
          "headers":{
             ...
          },
          "method":"GET",
          "query":{
          },
          "url":"/v1/projects/350446661175/locations/us-central1/workflows/workflow-1/executions/46804f42-dc83-46d6-87e4-93962866ed81/callbacks/49c80102-74d2-49cd-a70e-805a9fded94f_2de9b413-6332-412d-99c3-d7e9b6eeeda2"
       },
       "received_time":"2021-06-24 12:49:16.988072651 -0700 PDT m=+742581.005780667",
       "type":"HTTP"
    }

HTTP 本文がテキストまたは JSON の場合、ワークフローは本文のデコードを試みます。それ以外の場合は、生のバイトが返されます。

コールバックエンドポイントへのリクエストを承認する

リクエストをコールバックエンドポイントに送信するには、Cloud Run、Cloud Run 関数などの Google Cloud サービス、サードパーティのサービスに対して、適切な Identity and Access Management（IAM）権限（具体的には Workflows 起動元ロールに含まれている workflows.callbacks.send）を設定して、処理を行うことを許可する必要があります。

直接リクエストを行う

有効期間が短いサービスアカウントの認証情報を作成する最も簡単な方法は、直接リクエストを作成することです。このフローには、呼び出し元と認証情報が作成されるサービスアカウントの 2 つの ID が含まれます。このページの基本的なワークフローの呼び出しは、直接リクエストの例です。詳細については、IAM を使用してアクセスを制御すると直接リクエスト権限をご覧ください。

OAuth 2.0 アクセストークンを生成する

アプリケーションがコールバックエンドポイントを呼び出すことを承認するには、ワークフローに関連付けられたサービスアカウントの OAuth 2.0 アクセストークンを生成します。必要な権限（Workflows Editor や Workflows Admin と、Service Account Token Creator ロール）が付与されている場合は、generateAccessToken メソッドを実行して自分でトークンを生成することもできます。

generateAccessToken リクエストが成功すると、返されたレスポンス本文に OAuth 2.0 アクセストークンと有効期限が含まれます（デフォルトでは、OAuth 2.0 アクセストークンは最大 1 時間有効です）。次に例を示します。

  {
  "accessToken": "eyJ0eXAi...NiJ9",
  "expireTime": "2020-04-07T15:01:23.045123456Z"
  }

次の例のように、accessToken コードは、コールバックエンドポイント URL への curl 呼び出しで使用できます。

  curl -X GET -H "Authorization: Bearer ACCESS_TOKEN_STRING" CALLBACK_URL

  curl -X POST -H "Content-Type: application/json" -H "Authorization: Bearer ACCESS_TOKEN_STRING" -d '{"foo" : "bar"}' CALLBACK_URL

Cloud Run 関数の OAuth トークンを生成する

ワークフローと同じプロジェクト内で同じサービスアカウントを使用して Cloud Run 関数からコールバックを呼び出す場合は、関数自体で OAuth アクセストークンを生成できます。次に例を示します。

const {GoogleAuth} = require('google-auth-library');
const auth = new GoogleAuth();
const token = await auth.getAccessToken();
console.log("Token", token);

try {
  const resp = await fetch(url, {
      method: 'POST',
      headers: {
          'accept': 'application/json',
          'content-type': 'application/json',
          'authorization': `Bearer ${token}`
      },
      body: JSON.stringify({ approved })
  });
  console.log("Response = ", JSON.stringify(resp));

  const result = await resp.json();
  console.log("Outcome = ", JSON.stringify(result));

さらに詳しい内容については、コールバックを使用した人間参加型ワークフローの作成に関するチュートリアルをご覧ください。

オフラインアクセスをリクエストする

アクセストークンは定期的に期限が切れ、関連する API リクエストでは無効な認証情報になります。トークンに関連付けられたスコープへのオフラインアクセスをリクエストした場合は、ユーザーに権限を求めることなくアクセストークンを更新できます。ユーザーがいないときに Google API にアクセスする必要があるアプリケーションには、オフラインアクセスのリクエストが必須です。詳細については、アクセストークンの更新（オフラインアクセス）をご覧ください。

コールバックを使用してワークフローをただ 1 回だけ呼び出す

コールバックは完全にべき等です。つまり、コールバックが失敗した場合、予期しない結果または副作用を発生させることなくそれを再試行できます。

コールバックエンドポイントを作成すると、URL は受信リクエストを受信する準備が整い、通常は対応する呼び出しを await_callback に行う前に、呼び出し元に返されます。ただし、await_callback ステップの実行時にコールバック URL がまだ受信されていない場合、エンドポイントが受信される（またはタイムアウトが発生する）まで、ワークフローの実行はブロックされます。受信すると、ワークフローの実行が再開され、コールバックが処理されます。

create_callback_endpoint ステップを実行してコールバックエンドポイントを作成すると、ワークフローに単一のコールバックスロットを使用できるようになります。コールバックリクエストを受信すると、コールバックが処理できるようになるまで、このスロットにはコールバックペイロードが入ります。await_callback ステップが実行されると、コールバックが処理され、スロットが空になり、別のコールバックで使用できるようになります。コールバックエンドポイントを再利用して、await_callback を再度呼び出すことができます。

await_callback が 1 回だけ呼び出され、2 番目のコールバックを受信した場合、次のいずれかのシナリオが発生し、適切な HTTP ステータスコードが返されます。

HTTP 429: Too Many Requests は、最初のコールバックは正常に受信されたものの、処理待ちのままであることを示します。2 番目のコールバックはワークフローによって拒否されます。
HTTP 200: Success は、最初のコールバックが正常に受信され、レスポンスが返されたことを示します。2 番目のコールバックは保存され、await_callback が 2 回呼び出されない限り処理されない場合があります。それが発生する前にワークフローが終了すると、2 番目のコールバックリクエストは決して処理されず、破棄されます。
HTTP 404: Page Not Found は、ワークフローがもう実行されていないことを示します。最初のコールバックが処理されてワークフローが完了したか、ワークフローが失敗しました。これを判断するには、ワークフローの実行状態をクエリする必要があります。

並列コールバック

ステップが並列で実行され、親スレッドによってコールバックが作成され、子ステップで待機される場合、前述と同じパターンが適用されます。

次の例では、create_callback_endpoint ステップが実行されると、1 つのコールバックスロットが作成されます。await_callback に対する後続の呼び出しごとに、新しいコールバックスロットが開きます。すべてのスレッドが実行されており、コールバックリクエストが実行される前に待機している場合は、10 件のコールバックを同時に実行できます。追加のコールバックを行うこともできますが、保存され、決して処理されません。

YAML

  - createCallbackInParent:
    call: events.create_callback_endpoint
    args:
      http_callback_method: "POST"
    result: callback_details
  - parallelStep:
    parallel:
        for:
            range: [1, 10]
            value: loopValue
            steps:
              - waitForCallbackInChild:
                  call: events.await_callback
                  args:
                      callback: ${callback_details}

JSON

  [
    {
      "createCallbackInParent": {
        "call": "events.create_callback_endpoint",
        "args": {
          "http_callback_method": "POST"
        },
        "result": "callback_details"
      }
    },
    {
      "parallelStep": {
        "parallel": {
          "for": {
            "range": [
              1,
              10
            ],
            "value": "loopValue",
            "steps": [
              {
                "waitForCallbackInChild": {
                  "call": "events.await_callback",
                  "args": {
                    "callback": "${callback_details}"
                  }
                }
              }
            ]
          }
        }
      }
    }
  ]

コールバックは、await_callback へのブランチによって各呼び出しが行われるのと同じ順序で処理されます。ただし、ブランチの実行順序は決まってはおらず、さまざまなパスを使用して結果に到達する可能性があります。詳細については、並列ステップをご覧ください。

基本的なコールバックワークフローを試す

基本的なワークフローを作成し、そのワークフローのコールバックエンドポイントの呼び出しを curl を使用してテストできます。ワークフローが存在するプロジェクトに対して、Workflows Editor または Workflows Admin の権限を持っている必要があります。

次のワークフローを作成してデプロイした後、それを実行します。

YAML

    - create_callback:
        call: events.create_callback_endpoint
        args:
            http_callback_method: "GET"
        result: callback_details
    - print_callback_details:
        call: sys.log
        args:
            severity: "INFO"
            text: ${"Listening for callbacks on " + callback_details.url}
    - await_callback:
        call: events.await_callback
        args:
            callback: ${callback_details}
            timeout: 3600
        result: callback_request
    - print_callback_request:
        call: sys.log
        args:
            severity: "INFO"
            text: ${"Received " + json.encode_to_string(callback_request.http_request)}
    - return_callback_result:
        return: ${callback_request.http_request}

JSON

    [
      {
        "create_callback": {
          "call": "events.create_callback_endpoint",
          "args": {
            "http_callback_method": "GET"
          },
          "result": "callback_details"
        }
      },
      {
        "print_callback_details": {
          "call": "sys.log",
          "args": {
            "severity": "INFO",
            "text": "${\"Listening for callbacks on \" + callback_details.url}"
          }
        }
      },
      {
        "await_callback": {
          "call": "events.await_callback",
          "args": {
            "callback": "${callback_details}",
            "timeout": 3600
          },
          "result": "callback_request"
        }
      },
      {
        "print_callback_request": {
          "call": "sys.log",
          "args": {
            "severity": "INFO",
            "text": "${\\"Received \\" + json.encode_to_string(callback_request.http_request)}"
          }
        }
      },
      {
        "return_callback_result": {
          "return": "${callback_request.http_request}"
        }
      }
    ]

ワークフローの実行後、コールバックリクエストを受信するかタイムアウトが経過するまで、ワークフローの実行状態は ACTIVE になります。

実行状態を確認して、コールバック URL を取得します。
Console
1. Google Cloud コンソールの [ワークフロー] ページに移動します。
  [ワークフロー] に移動
2. 実行したワークフローの名前をクリックします。
  
  ワークフローの実行状態が表示されます。
3. [Logs] タブをクリックします。
4. 次のようなログエントリを探します。
  Listening for callbacks on https://workflowexecutions.googleapis.com/v1/projects/...
5. 次のコマンドで使用するコールバック URL をコピーします。
gcloud
1. まず、実行 ID を取得します。
  gcloud logging read "Listening for callbacks" --freshness=DURATION
  DURATION は、返されるログエントリを制限する適切な時間に置き換えます（ワークフローを複数回実行している場合）。
  たとえば、--freshness=t10m は、10 分以内のログエントリを返します。詳細については、gcloud topic datetimes を参照してください:
  
  実行 ID が返されます。コールバック URL は textPayload フィールドでも返されます。次の手順で使用するために両方の値をコピーします。
2. 次のコマンドを実行します。
  gcloud workflows executions describe WORKFLOW_EXECUTION_ID --workflow=WORKFLOW_NAME
  ワークフロー実行の状態が返されます。
これで、curl コマンドを使用してコールバックエンドポイントを呼び出せます。
```
curl -X GET -H "Authorization: Bearer $(gcloud auth print-access-token)" CALLBACK_URL
```
なお、POST エンドポイントには、Content-Type 表現ヘッダーを使用する必要があります。次に例を示します。
```
curl -X POST -H "Content-Type: application/json" -H "Authorization: Bearer $(gcloud auth print-access-token)" -d '{"foo" : "bar"}' CALLBACK_URL
```
CALLBACK_URL を前のステップでコピーした URL に置き換えます。
Google Cloud Console または Google Cloud CLI を使用して、ワークフローの実行の状態が SUCCEEDED になったことを確認します。
次のような textPayload が返されているログエントリを探します。
```
Received {"body":null,"headers":...
```

サンプル

ここでは、構文の例を示します。

タイムアウトエラーをキャッチする

このサンプルでは、タイムアウトエラーをキャッチしてシステムログにエラーを書き込むコードを、前のサンプルに追加します。

YAML

    main:
      steps:
        - create_callback:
            call: events.create_callback_endpoint
            args:
                http_callback_method: "GET"
            result: callback_details
        - print_callback_details:
            call: sys.log
            args:
                severity: "INFO"
                text: ${"Listening for callbacks on " + callback_details.url}
        - await_callback:
            try:
                call: events.await_callback
                args:
                    callback: ${callback_details}
                    timeout: 3600
                result: callback_request
            except:
                as: e
                steps:
                    - log_error:
                        call: sys.log
                        args:
                            severity: "ERROR"
                            text: ${"Received error " + e.message}
                        next: end
        - print_callback_result:
            call: sys.log
            args:
                severity: "INFO"
                text: ${"Received " + json.encode_to_string(callback_request.http_request)}

JSON

    {
      "main": {
        "steps": [
          {
            "create_callback": {
              "call": "events.create_callback_endpoint",
              "args": {
                "http_callback_method": "GET"
              },
              "result": "callback_details"
            }
          },
          {
            "print_callback_details": {
              "call": "sys.log",
              "args": {
                "severity": "INFO",
                "text": "${\"Listening for callbacks on \" + callback_details.url}"
              }
            }
          },
          {
            "await_callback": {
              "try": {
                "call": "events.await_callback",
                "args": {
                  "callback": "${callback_details}",
                  "timeout": 3600
                },
                "result": "callback_request"
              },
              "except": {
                "as": "e",
                "steps": [
                  {
                    "log_error": {
                      "call": "sys.log",
                      "args": {
                        "severity": "ERROR",
                        "text": "${\"Received error \" + e.message}"
                      },
                      "next": "end"
                    }
                  }
                ]
              }
            }
          },
          {
            "print_callback_result": {
              "call": "sys.log",
              "args": {
                "severity": "INFO",
                "text": "${\"Received \" + json.encode_to_string(callback_request.http_request)}"
              }
            }
          }
        ]
      }
    }

再試行ループでコールバックを待機する

このサンプルは、再試行ステップを実装して前のサンプルを変更します。ワークフローは、カスタム再試行述語を使用して、タイムアウトが発生したときに警告をログに記録し、コールバックエンドポイントで待機を再試行します（最大 5 回）。コールバックが受信される前に再試行の割り当てが使い果たされると、最終的なタイムアウトエラーによりワークフローが失敗します。

YAML

    main:
      steps:
        - create_callback:
            call: events.create_callback_endpoint
            args:
                http_callback_method: "GET"
            result: callback_details
        - print_callback_details:
            call: sys.log
            args:
                severity: "INFO"
                text: ${"Listening for callbacks on " + callback_details.url}
        - await_callback:
            try:
                call: events.await_callback
                args:
                    callback: ${callback_details}
                    timeout: 60.0
                result: callback_request
            retry:
                predicate: ${log_timeout}
                max_retries: 5
                backoff:
                    initial_delay: 1
                    max_delay: 10
                    multiplier: 2
        - print_callback_result:
            call: sys.log
            args:
                severity: "INFO"
                text: ${"Received " + json.encode_to_string(callback_request.http_request)}
    log_timeout:
        params: [e]
        steps:
          - when_to_repeat:
              switch:
                - condition: ${"TimeoutError" in e.tags}
                  steps:
                      - log_error_and_retry:
                          call: sys.log
                          args:
                              severity: "WARNING"
                              text: "Timed out waiting for callback, retrying"
                      - exit_predicate:
                          return: true
          - otherwise:
              return: false

JSON

    {
      "main": {
        "steps": [
          {
            "create_callback": {
              "call": "events.create_callback_endpoint",
              "args": {
                "http_callback_method": "GET"
              },
              "result": "callback_details"
            }
          },
          {
            "print_callback_details": {
              "call": "sys.log",
              "args": {
                "severity": "INFO",
                "text": "${\"Listening for callbacks on \" + callback_details.url}"
              }
            }
          },
          {
            "await_callback": {
              "try": {
                "call": "events.await_callback",
                "args": {
                  "callback": "${callback_details}",
                  "timeout": 60
                },
                "result": "callback_request"
              },
              "retry": {
                "predicate": "${log_timeout}",
                "max_retries": 5,
                "backoff": {
                  "initial_delay": 1,
                  "max_delay": 10,
                  "multiplier": 2
                }
              }
            }
          },
          {
            "print_callback_result": {
              "call": "sys.log",
              "args": {
                "severity": "INFO",
                "text": "${\"Received \" + json.encode_to_string(callback_request.http_request)}"
              }
            }
          }
        ]
      },
      "log_timeout": {
        "params": [
          "e"
        ],
        "steps": [
          {
            "when_to_repeat": {
              "switch": [
                {
                  "condition": "${\"TimeoutError\" in e.tags}",
                  "steps": [
                    {
                      "log_error_and_retry": {
                        "call": "sys.log",
                        "args": {
                          "severity": "WARNING",
                          "text": "Timed out waiting for callback, retrying"
                        }
                      }
                    },
                    {
                      "exit_predicate": {
                        "return": true
                      }
                    }
                  ]
                }
              ]
            }
          },
          {
            "otherwise": {
              "return": false
            }
          }
        ]
      }
    }

コールバックを使用して待機する

コールバック リクエストを受信するエンドポイントを作成する

YAML

JSON

YAML

JSON

YAML

JSON

コールバック エンドポイントへのリクエストを承認する

直接リクエストを行う

OAuth 2.0 アクセス トークンを生成する

Cloud Run 関数の OAuth トークンを生成する

オフライン アクセスをリクエストする

コールバックを使用してワークフローをただ 1 回だけ呼び出す

並列コールバック

YAML

JSON

基本的なコールバック ワークフローを試す

YAML

JSON

Console

gcloud

サンプル

タイムアウト エラーをキャッチする

YAML

JSON

再試行ループでコールバックを待機する

YAML

JSON

次のステップ

コールバックリクエストを受信するエンドポイントを作成する

コールバックエンドポイントへのリクエストを承認する

OAuth 2.0 アクセストークンを生成する

オフラインアクセスをリクエストする

基本的なコールバックワークフローを試す

タイムアウトエラーをキャッチする