バッチ処理の管理方法を考えたいので考えをまとめるためのメモスクラップ

やりたいこと

• バッチを管理したい
  • バッチのオンオフ、リトライを専用の管理画面からやりたい
  • 専用の管理画面は現状システムの管理画面があるためそこに項目として追加したい
• バッチ処理自体はすでにいくつかある
  • これらはTerraformなどコードベースで管理されている
  • オンオフの設定や、起動時間の設定はデプロイが必要

利用環境

• Google Cloud
  • Cloud Scheduler
  • Workflows
  • Cloud Functions 2nd gen
  • その他データ関係（CloudSQLやBigQuery、Cloud Storage
• Slack
  • 通知用

現状

1. バッチはPythonやらGoやらでCloud Functionsにデプロイするコードを記述
2. Workflowのyamlファイルを記述
3. Configファイルに上記の処理・ワークフローの連携関係や、起動条件のCron式などを記述
4. 上記の三つをTerraformに読み込ませ、いい感じにデプロイ

問題点

• バッチのオンオフにTerraformのデプロイが必要
  • コンソールから変更もできるが、過去それで設定ファイルをミスって巻き戻りが発生し事故った
• 定時バッチが週明けだったのが、週明け火曜日で実行したい、みたいな要望が上がった
  • そのほか「今欲しい」みたいなことがたまにあった
    • これは手動で出してた
• パラメータがいくつかあるが、半年に一回諸事情で書き換える必要がある
  • 手動で書き換えるのは負担
  • 前はStgとPrdで同じパラメータだったが、管理画面経由で設定できるようになったのでばらけた値が入るようになった（UUIDなので本来それが正しい
• 起動条件を変えるのが難しい（デプロイが必要でめんどい、が正しい
  • 週明けって言ってるけど週末も欲しい、とか

これらを解決したい

シンプルに問題点も分離して考える

• 汎用バッチの問題点
  • 起動時間の問題
    • 正直これは「毎日朝実行、バッチに起動条件として曜日や日付を設定して実行」でいい気がする
  • リトライ、手動実行の問題
    • これらも手動でworkflowをトリガーしてあげればいい気がする
  • パラメータの問題
    • 冷静に考えたらこれが必要なのは下の専門バッチと同等なので一旦パス
  • 起動条件変更
    • 起動時間の問題に合わせていい
• 専用バッチの問題点
  • 起動時間
    • 上に同じ
  • リトライ、手動実行
    • これも汎用と同じ
  • パラメータの問題
    • 専用バッチだけパラメータの設定項目を追加する？
  • 起動条件変更
    • こっちはこれの問題がそんなに出ない気がする

割とシンプルめでいいか？

どこまでコード管理されているべきで、どこから別ソース（実質データベース）管理されるべきか

汎用バッチは割と容易（というか、現状すでにデータベースからいくつかデータソースを取ってきているのでそこを伸ばせばいいはず
ただ手動実行のためのポイントを用意する必要があるのでそこに改修の手が入る

専用バッチの場合も汎用バッチと同様にしつつ専用バッチごとにパラメータと起動設定を持つか

ラバーダッキング的な感じで使ってたら整理できてきたのでこの辺で

何かしら別のポイントが出てきたらまた使うけど一旦クローズ

クローズしたけどもうちょい具体的なところを詰めたくなったので再開

まずCloud Schedulerでバッチ管理用の処理を立ち上げる（バッチのバッチ
現状朝9時起動と午前0時起動の二つがあったので、そこに合わせる……か、あるいは毎時起動させてバッチごとに何時に走るか判定し、起動時間の時間と同じものを走らせるようにする

各処理はバッチのトリガーを、上記のバッチのバッチが叩く、という形にする
このトリガーは手動で叩けば手動でのリトライや緊急でデータが欲しい時などに使いまわせる

変更していくならまずはバッチのトリガーをリスト化する状態を作るべきか

（しかしこの手の処理をやろうとするとローカルでテストしづらいのがつらい……stgのさらに前のdev環境があるからまだいいけど

ちょっと調べて修正する現状のバッチ起動プロセス

• Cloud Schedulerが特定時間にPubSubに送信
• PubSubの通知をEventarcが受信
• EventarcがWorkflow起動

こうだった

今回のフィルターを追加すると

• Cloud Schedulerが特定時間にPubSubに送信
• PubSub通知をEventarcが受信
• EventarcがWorkflow起動
• Workflowで途中のCloudFunctions（仮）が起動条件等を検索
• 起動条件に従って各種Workflowを起動

という感じか。今まで使っていたEventarc等を消す必要がある

まずトリガーがEventarcにならないように変更。Terraformを書き直して設定項目でCronの設定がないものはCloudSchedulerとEventarcを作らないように変更

一旦Terraform planしてもらおう

次、スーパーバッチトリガー（仮）を作る

基本的には従来のバッチ処理の延長線上で作ればいいのでとりあえず作る

起動はWorkflowの内部でループさせて送信させることにする。WorkflowからPOSTさせることで起動させられるのでそれで

パラメータはどうするか悩み中。基本的にスーパートリガーが起動させるバッチの種類と対象のID的なものを持ってくるタイミングで入れるのは確定だが、別途テーブルを持つべきか、DBのバッチ・IDの結びつけテーブルに持たせるべきか

パラメータはほぼ決まり切ったものが多いのでもっと汎用的に設定してもいい気がする
たとえば

• ID
• 実行日時

あたりはほぼほぼどれでも使うので、汎用的に使うものはもう一括で管理してしまうか

https://cloud.google.com/workflows/docs/reference/googleapis/workflowexecutions/v1beta/projects.locations.workflows.executions/run

ワークフローからワークフローを起動させるにはこのAPIがいいらしい
ちなみに run と create の二つがあるが、run の方が簡単らしい（ドキュメント曰く。リンクは run
まあ用途に合わせて使えば良さそう

これを使う場合、バッチのトリガーURLを入れるのではなくIDを入れる必要がある
あと標準だと30分以内に返ってこないとダメらしい。ただこれは設定で変えられる（最大1年らしい　長すぎんか

とりあえず適当に当てはめてみる

バッチ管理テーブルに起動週のカラムを入れていたがこれだと毎日実行が管理できないことがわかった

trigger_week ENUM('sun', 'mon', 'tue', 'wed', 'thu', 'fri', 'sat') NOT NULL, -- 何曜日に実行するか

こういう設定だとダメ。配列型で設定できると嬉しいがMySQLで配列を使うならJSON型でやるしかなくなる

他の解決法を出すなら、ビット型を使いビット配列で管理するか、Cron式で管理するかになる。Cron式はパースがめんどいのでやりたくないがビット配列もまためんどいな……

ついでに上記の曜日管理と同様に、

trigger_hour INT NOT NULL, -- 何時に実行するか。0-23

みたいな項目もあり同じ問題を抱えているが、1日に何度も起動させるようなバッチは現状ないので保留

概ねできた。テストバッチの起動も成功

あとなんか考えとくところあるだろうか。渡ったパラメータの処理だけはのちの自分のためにもまとめておこう

まず呼び出し側のワークフロー

簡略化しているのでもしかしたらこのままコピペっても上手くいかないかもしれないけどとりあえず

- get-caller-config:
    call: http.post
    args:
      url: ${caller-config-function-trigger-url}
      body:
        foo: bar
    result: ${caller}
- trigger-workflow:
    call: googleapis.workflowexecutions.v1.projects.locations.workflows.executions.run
    args:
      workflow_id: ${caller.workflow_id}
      argument: ${caller.argument}

こんな感じ。
簡略化しているというのは、実際は get-caller-config は複数の起動条件を持ってきてループ＆パラレルにしている（プラスして成否の通知処理でエラーハンドリングもある）が、今回は単一の起動条件を持ってきている

続いて関数。上記の get-caller-config のステップで実行される関数

def main(request):
    req_json = request.get_json()
    foo = req_json["foo"] # bar が入っている

    # なんやかんやでこんな感じの結果を返すとする
    return {
        "workflow_id" : "call-workflow-name",
        "argument" : {
            "hoge" : "fuga"
        }
    }

こんな感じとする。この場合、ワークフロー側の ${caller} 変数はMap形式で workflow_id と argument のキーを持つ状態になる。つまり、

- trigger-workflow:
    call: googleapis.workflowexecutions.v1.projects.locations.workflows.executions.run
    args:
      workflow_id: "call_workflow_name"
      argument: { "hoge" : "fuga" }

実際はこう呼び出されることになる。

続いて呼ばれる側

main:
  params: [args]
  steps:
    - log:
        call: sys.log
        args:
            text: ${args}
            severity: INFO

とすると、

{"hoge":"fuga"}

とマップがそのまま出力される。つまり、呼び出す側の argument に入れたMapはそのまま渡されると見ていいらしい。なのでこんなふうにして変数にassignできる

# ↑の続き
    - assign-args:
        assign:
          - hoge: ${args.hoge}

これでワークフロー内部の変数として使える（assignしなくてもいいにはいいが、やっておいた方が多分後で楽になるタイプ

ひとまずやりたいことはできた

パラメータやらの持ち方や各ワークフローの設計にもう一押しできそうではあるがバッチを管理するバッチとそのバッチ管理バッチから各種バッチを起動させるという点においては既存のワークフローのある程度の使い回し等ができそうな土台ができたのでOKとする

というわけでクローズ