GCP で Cloud Run jobs が一般リリースとなったので、簡易的なバッチジョブ用途として Workflows との連携を個人開発のアプリに実装してみました。
前置き
個人の趣味プロで hp2fit というアプリを作っています。タニタの体組成計で計測しタニタの健康管理サービスである Health Planet に保存された体重と体脂肪を、毎時 Fitbit に登録するという連携アプリです。
毎時に特定の処理を実施する、という簡易バッチ的な処理をするためには、GCP の環境だと今までは以下のサービスを組み合わせることが多いかと思います。
- Cloud Scheduler
- Cloud Function もしくは Cloud Run
Cloud Scheduler で HTTP コールをトリガーに Cloud Function だったり Cloud Run でバッチ的な処理を行う方法で、比較的シンプルかつ処理が３０分以内に終わる場合には有効です。この３０分という時間制限は Cloud Scheduler の HTTP 呼び出しのタイムアウトが３０分であることに由来しています。
上述した hp2fit でも毎時の処理として batchjob という Cloud Run を Cloud Scheduler から呼び出すという構成を採用しています。
所詮は趣味で作った連携サービスで、サービスを無料で開放しているとはいえ、今のところ毎回３０秒程度の処理で終わっているため HTTP トリガーで問題がないのですが、せっかくなのでこの batchjob を Cloud Run Jobs に置き換えてバッチジョブとして呼び出す仕組みに変更してみたいと思います。
Cloud Run から Cloud Run jobs へ
まず、既存の batchjob は Cloud Run は HTTP のエンドポイントとして作成されていて、起動すると指定したポートで HTTP リクエストを待機する挙動となります。Cloud Run jobs では、実行されたら処理を行い、処理が完了したら終了する作りに変えておきます。
ここで、大量のデータを処理する場合は Cloud Run jobs は処理をタスクに分けて処理するようにします。どういうことかというと、同じ処理の繰り返しだけど１タスクを６０分以内に終わる分量にして、全体のデータが扱えるようなタスクを並列もしくは直列実行するようにします。このあたりについてはこちらの記事が参考になりますのでここでは割愛します。
Cloud Run jobs は既存の Cloud Run 同様、予め登録をしておき、実行時に execute で実行指示を出します。よって、Cloud Build とかで CI/CD のパイプラインを作っている場合は、jobs 用のコマンドライン引数に変更しておきます。
- name: 'gcr.io/cloud-builders/gcloud'
args: [
'run',
'jobs',
'deploy',
'batchjob',
'--image', '${_LOCATION}-docker.pkg.dev/${PROJECT_ID}/${_REPO}/${_IMAGE}',
'--region', 'asia-northeast1',
'--set-env-vars=KEEPDAYS=90']
substitutions:
_LOCATION: asia-northeast1
_REPO: hp2fit-repo
_IMAGE: batchjob
Cloud Run jobs の実行管理は Workflows を使う
Cloud Run jobs で60分を超える処理（タスクに文化させる）を実行管理させる場合、Cloud Run jobs の execution は Cloud Functions から直接呼び出すのではなく、Workflows をはさみます。
Workflows は他の Google Cloud プロダクトに簡単にアクセスできるようにコネクタが用意されていて、Cloud Run jobs のコネクタもあります。このコネクタを使うことで、最大１年間のタイムアウトが設定できます。つまり、Cloud Run jobs の実行完了を１年間待つことができるようになります。
動きとしてはこのような流れになります;
- Cloud Scheduler が登録されている Workflow を定期的に実行指示する。この際、実行指示として run ではなく execute で Workflow を起動する。execute だとワークフローが起動できたらすぐに戻るので、Cloud Scheduler の30分タイムアウトに引っかかりません。つまり、ワークフローを起動するだけでその後の処理の成否はここでは管理をしません。
- Workflow の中で、Cloud Run jobs の実行指示を出す。ここでは完了まで待ちます（引数として --async を指定しなければ jobs の完了まで待ちます）
- Cloud Run jobs の中でバッチ処理を行う
Cloud Run jobs を呼び出す Workflow の YAML ファイルはこちらになります。
main:
params: [event]
steps:
- init:
assign:
- project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
- job_name: batchjob
- job_location: asia-northeast1
- run_job:
call: googleapis.run.v1.namespaces.jobs.run
args:
name: ${"namespaces/" + project_id + "/jobs/" + job_name}
location: ${job_location}
result: job_execution
- finish:
return: ${job_execution}
Workflow を登録したら、ワークフローのトリガーのタブの中で Scheduler から呼び出す設定ができるので、そちらで登録すれば OK です。
ということで、HTTPトリガーでキックしていたバッチ処理を Workflow と Cloud Run jobs を使ったクラウドネイティブなバッチ処理アーキテクチャに変更できました。意外と簡単に。
今回の処理は単一の Cloud Run jobs を呼び出すだけなので、バッチ処理の運用管理としては Cloud Run jobs でも Workflows の実行結果を確認すれば良いです。もし、ある処理を行ったあとに次の処理を行うといったジョブのネットワークを構成する場合は、ジョブの全体管理として Workflow の実行結果を確認すれば良いだけで、なにかエラーが生じている場合は個別の Cloud Run jobs を確認すればよくなります。
Workflow のいいところは、ステップ数でコストがかかるので、この程度ならタダで使えます。そして Cloud Run jobs も実行している間だけのコストで、登録されている状態だけで実行待ちの状態はコストが掛からない点です。
待機している間のコストが掛からず、ある程度複雑な処理の組み合わせのワークフローを YAML で定義ができるという workflow は結構便利です。今後使われるシーンが増えそうです。