サーバレスなバッチを管理するためのKudaを作成しました
概要
こちらの記事でも書いたのですが、小さなバッチを運用するにはCloud Runのようなコンテナベースのサーバレス環境はとても向いています。
バッチサーバとしてインスタンスを常駐しておく必要が無いのでコスト面でも有利ですし、横にスケールさせやすいと言うのもあります。
ただ、バッチとして考えた場合にはジョブ管理ツールというかワークフローエンジンが無い。1ジョブで完結するものは良いですが、典型的なジョブは以下のように後続処理を持っています。
後続処理が並列で動いて集約の処理がそれを待つとか典型的な実装ですよね?
ただ、これをマイクロサービスでやるのはちょっと面倒。おそらくMSA的に正しく実装しようとすると全てをイベントドリブンで記述して「JOBaの後にJOBbが動く」と言ったことを連鎖的に書くのだと思います。
これは大規模な処理で並列度を高めるには良いと思うのですが、既存のマイグレーションや小規模な運用では少し面倒です。
なので、シンプルにそれを実現するためのツールとして「Kuda」を作りました。
Kudaを使うことで各マイクロサービスは後続のことを気にする事無く記述でき、ジョブのワークフローを外出しできます。
使い方
現状、アルファ版でとりあえず作ったレベルでバギー&GCPのみ対応ですが、一応動かす事は可能です。
まず、下記のようなjobs.yaml
というYAMLファイルでワークフローを記述します。
name: exec-batch tasks: - name: step-load url: https://xxx-app1.run.app - name: step-parallel1 dependencies: [step-load] url: https://xxx-app2.run.app - name: step-parallel2 dependencies: [step-load] url: https://xxx-app3.run.app - name: step-join dependencies: [step-parallel1, step-parallel2] url: https://xxx-app4.run.app
tasks
配下に対象となるジョブの一覧を書いていきます。
name
はそのまま名前で、url
がジョブとして実行したいマイクロサービスのURLとなります。dependencies
には依存(先行ジョブ)を書きます。
同じ依存を書けばそれらは並列で実行されますし、カンマ区切りで複数個書けば全てのジョブの完了を待ってから処理に入ります。
このファイルをGCS上に置いてkuda自体もCloud Runなどにデプロイすれば準備完了です。
ジョブのステータスを確認
JOBの実行状態を以下のようにURLを叩く事で確認できます。「READY -> RUNNING -> DONE」という状態遷移をします。サンプルは未稼働状態なので全部READY。
$ curl http://localhost:8080/jobq [(step-load, READY), (step-parallel1, READY), (step-parallel2, READY), (step-join, READY)]
実行
以下のようにexec
で実行を開始します。
$ curl http://localhost:8080/exec done
このURLは非同期なのでジョブの完了を待つ事なくすぐにレスポンスを返します。
ジョブの進捗を見る場合は先ほどのjobq
を実行するか、kuda側の標準出力を見る事で確認できます。
標準出力の結果は下記の通り。
07:09:29 INFO [cn.or.pa.ku.WorkflowManager]] (ForkJoinPool.commonPool-worker-1) START:JOB step-btc-collector 07:09:35 INFO [cn.or.pa.ku.WorkflowManager]] (ForkJoinPool.commonPool-worker-1) END:JOB step-btc-collector 07:09:35 INFO [cn.or.pa.ku.WorkflowManager]] (ForkJoinPool.commonPool-worker-3) START:JOB step-btc-predictor-dtree 07:09:35 INFO [cn.or.pa.ku.WorkflowManager]] (ForkJoinPool.commonPool-worker-5) START:JOB step-btc-predictor-knn 07:09:39 INFO [cn.or.pa.ku.WorkflowManager]] (ForkJoinPool.commonPool-worker-5) END:JOB step-btc-predictor-knn 07:09:39 INFO [cn.or.pa.ku.WorkflowManager]] (ForkJoinPool.commonPool-worker-3) END:JOB step-btc-predictor-dtree 07:09:39 INFO [cn.or.pa.ku.WorkflowManager]] (ForkJoinPool.commonPool-worker-1) START:JOB step-btc-scoring 07:09:45 INFO [cn.or.pa.ku.WorkflowManager]] (ForkJoinPool.commonPool-worker-1) END:JOB step-btc-scoring 07:09:45 INFO [cn.or.pa.ku.rs.WorkflowResource]] (executor-thread-3) END:WORKFLOW 07:09:45 INFO [cn.or.pa.ku.rs.WorkflowResource]] (executor-thread-3) CLEAR:JOBQ
仕組み
基本的な挙動としては
- execが実行されるとjobs.yamlを解析してDAGを作成
- DAGを元に実行状態を表すキューを作ってファイルに保存
- キューを読み込んで依存ジョブが全て
DONE
になってるREADY
のジョブがあれば全て実行 - 実行時にジョブのステータスを
RUNNNING
に変更してファイルに保存 - 実行時に実行完了時に自分自身(kuda)を呼ぶようにするために非同期処理で対象ジョブをキック
- 全てのジョブが
DONE
になるまで3-5を繰り返し
となります。
ポイントとしては一つ一つのリクエストは非同期処理で終わらせてるのでjobs.yamlに書いたジョブ全てが終わるまでkudaのリクエストが継続する訳ではない、ということです。
なので、小粒なジョブをたくさん書いてトータル実行時間が伸びたから親のリクエストであるKudaがタイムアウト、とかにはなら無いはず。まだ、ちゃんとテストして無いから肝心の部分がバグってるかもですがw
Kuda自体もCloud Runで動くサーバレス環境で状態はGCSなど外部に保存する想定なので並列度が上がれば勝手にスケールするはず。ただ、並列度が上がれば誤って同時実行しそうな気がするので排他ロックをDB使うなりで実現が必要な気はしてます。
今後の想定
とりあえず今後の想定としては以下の通り
- ジョブの実行状態とかが見えるWeb UIの作成
- Cloud Scheduler とかと連携してスケジューラーもWeb UIに統合
- quarkusで作ってあるのでnative-buildして高速化 # 現状実行時エラーになるのでjavaで稼働
- 排他処理周りの検討
- テストとリファクタリングをいっぱい
まとめ
hwrapと合わせて使うことで、Cloud Runなどを利用して簡単にサバーレスバッチを作るためのツールを作りました。
巨大なバッチならSparkとか使えばスケールアウト構成に出来るのですが、そうじゃない場合はk8sとかを使って上手くできないかなー、と思ってたのでようやく開発着手、というところですね。
Dockerというかk8sをベースにバッチ管理だとargoがありますが、あれよりも機能を減らしてでも手軽に運用出来るものに出来たら良いな、と思っています。HTTP対応してれば良いのでk8s/Cloud Runに本来的には縛られないし。
ちなみに名前の由来は「パイプライン」からの連想で「管」と、なんとなくリクエストを連鎖的に発生させて動かすイメージから「管狐」を連想したのが理由です。
それでは、Happy Hacking!