クラスメソッド株式会社データアナリティクス事業本部所属のニューシロです。
今回はGoogle CloudのサービスであるCloud Schedulerの再試行について調べてみました。
前提
Cloud Schedulerとは
Cloud Schedulerは、フルマネージドのcronジョブサービスです。
Cloud Scheduler では、作業単位のスケジュールを設定して、定義した回数または一定の間隔で実行できます。これらの作業単位は、一般的に cron ジョブと呼ばれています。代表的な使い方としては、レポートメールを毎日送信する、10 分間隔でキャッシュ データを更新する、1 時間に 1 回要約情報を更新する、などがあります。
Cloud Schedulerはオプションとして、ジョブが正常に完了しなかった場合、指数バックオフを使用してジョブを再試行する機能を持ちます。 今回はこの再試行の挙動について確認していきたいと思います。
本題
今回想定するCloud Schedulerの実行ケース
今回は、Cloud SchedulerでHTTPトリガーのCloud Functionsを実行するケースを考えます。
Cloud SchedulerでCloud FunctionsのHTTPエンドポイントにリクエストを送信し、Cloud Functionsを起動させてみます。その際、意図的にCloud Functionsで処理を失敗させることによって、Cloud Schedulerの再試行時の挙動を検証していきます。
Cloud Functions実装
設定
今回利用したCloud Functionsの設定です。箇条書きにした項目以外はデフォルトを使用しています。
- 第2世代
- HTTPトリガー
- Python3.12
- 認証が必要
コード
Cloud Functionsに実装したコードです。
JSONデータを受け取り、messageをキーに持つ値を出力します。
main.py
import functions_framework
@functions_framework.http
def cloud_scheduler_test(request):
print("Function execution started.")
request_json = request.get_json()
print(f'message: {request_json["message"]}')
return "OK"requirements.txt
functions-framework==3.*こちらをデプロイします。細かい手順については本題ではないため割愛させていただきます。
Cloud Scheduler実装
Cloud Schedulerを実装します。今回の設定は以下の通りです。
コンソールで作成します。Cloud Schedulerの画面からジョブを作成をクリックして進めていきます。
URLには、先ほどデプロイしたCloud FunctionsのHTTPSエンドポイントを記載します。
メッセージは{"message": "Hello!"}にし、また今回メッセージはJSONのためヘッダーのContent-Typeはapplication/jsonに設定しています。
Cloud Functionsを認証が必要に設定しているため、OIDCトークンを追加を選択しサービスアカウントを設定します。今回はデフォルトのサービスアカウントを指定しました。
まずは再試行回数は0回で設定します。こちらは後ほど変更します。
準備ができたため、Cloud Schedulerを実行していきます。4パターンに分けて、Cloud Schedulerの挙動を確認します。
①ジョブの成功
まずはこのままCloud Schedulerを強制実行します。
Cloud Functionsが成功して、Cloud Functionsのログにmessage: Hello!が出力されています。
Cloud Schedulerのコンソール上の表示も成功になっています。
②ジョブの失敗（再試行無し）
次はジョブを失敗させてみます。
Cloud Schedulerで送信するメッセージのキーをmessageからhogeに変更します。
これにより、Cloud Functionsコード内のrequest_json["message"]の箇所でエラーが発生する想定です。Cloud Schedulerを強制実行します。
Cloud Functionsのログを確認すると、失敗していることがわかります。エラーログにはKeyError: 'message'の表示がありました。
Cloud Schedulerのログも見てみましょう。
ジョブの開始と終了のログがそれぞれ1つずつ出力され、また終了ログはエラーログとして出力されています。
Cloud Schedulerのコンソール上の表示も失敗になっています。
③ジョブの失敗（再試行有り）
いよいよ本題です。Cloud Schedulerの再試行回数を0回から5回に変更してみます。最大試行時間、バックオフ時間、最大倍増回数はデフォルトのままです。
これにより、ジョブ失敗時にジョブが再試行されるはずです。
メッセージ本文は前回と同一のため何回実行しても成功することはありませんが、これで再試行時の挙動を確認できるはずです。改めて、Cloud Schedulerを強制実行します。
Cloud Schedulerのログのみ表示します。
1回の実行につき、ジョブの開始と終了のログがそれぞれ1つずつ出力されます。
最初の実行1回（緑枠）に5回の再試行を加えた、計6回のジョブの実行が行われたことがわかります。
Cloud Schedulerのログのみ表示していますが、ジョブを受け取ったCloud Functionsも同じく6回処理が行われておりました。
また、それぞれの時間間隔も5秒→10秒→20秒→40秒→80秒と、指数関数的に増加しています。
想定通りに再試行が実行されたことがわかりました。
④ジョブ試行の期限超過
また、この再試行は試行期限内に応答が無い場合も発生します。
必要に応じて、HTTP ターゲットと App Engine HTTP ターゲットの場合は、ジョブ試行の期限を構成します。この期限までにリクエスト ハンドラが応答しない場合、リクエストはキャンセルされ、試行は失敗とみなされます。
こちらも試してみます。試行期限を15秒、再試行回数は1回に設定します。
また処理自体は成功するように、メッセージは{"message": "Hello!"}に戻しておきます。
使用したCloud Functionsのコードにimport timeとtime.sleep(30)を追加し、処理に約30秒かかるように設定することで、期限超過の再試行を発生させてみます。
main.py
import time
import functions_framework
@functions_framework.http
def cloud_scheduler_test(request):
print("Function execution started.")
time.sleep(30)
request_json = request.get_json()
print(f'message: {request_json["message"]}')
return "OK"Cloud Schedulerを強制実行します。
Cloud Schedulerのログです。前述の通り、Cloud Functionsの実行終了までに約30秒かかるようにしているため、15秒以内に応答を返すことはできません。
ログから、試行期限である15秒後にCloud Schedulerでエラーログ（赤枠）が出力され、その後最小バックオフ時間でデフォルト値である5秒後にCloud Schedulerが再試行されている（青枠）ことがわかります。
指定した期限内に応答しない場合も、Cloud Schedulerは再試行することが確認できました。
Cloud Schedulerで長い時間の処理を実行する場合、試行期限は長い時間に設定する必要がありそうです。こちらは15秒〜30分の範囲で指定できます。
最後に感想
Cloud Schedulerの再試行は、Pub/Subの再試行と似た挙動だと感じました。
ただ、最大試行時間が最大30分という点はPub/Subより長くて使いやすそうだなと感じました（Pub/Subは再試行時間が最大10分）。うまく使い分けていこうと思います。
9
