Executor種別
現在実装されている executor は shell と python_module の2種類です。モデルの型定義には http_post / sql / python_eval / container も宣言されていますが未実装で、これらの workload は worker が claim せずスキップします。
shell
任意のシェルコマンドを 1タスク = 1プロセス として実行します。
{
"command": ["bash", "-c", "echo task={task.pk}"],
"cwd": null,
"env": {"MY_VAR": "value"},
"timeout_secs": 60
}テンプレート変数 {task.pk} {task.extra.KEY} が展開されます。成否は success_criteria (既定 {"type":"exit_code","expected":0}) の通り、プロセスの exit code で判定されます。
| executor_config の鍵 | 意味 |
|---|---|
command | 実行するコマンドの配列 (argv)。テンプレート変数を展開。 |
cwd | 作業ディレクトリ (任意、null で既定)。 |
env | 追加の環境変数 (任意)。 |
timeout_secs | プロセスのタイムアウト秒。 |
python_module
プラグインを呼びます (プラグインSDK)。instance は worker プロセスでキャッシュされるため、重い setup() は1回だけ実行されます。python_module では exit code が無いため、成否は process() の戻り値 / 例外で決まります。
{
"source_path": "plugins/my_plugin",
"module": "main",
"callable": "process",
"init_kwargs": {
"api_url": "https://api.example.com",
"timeout_secs": 10
}
}| executor_config の鍵 | 意味 |
|---|---|
source_path | プラグインのディレクトリ (例 plugins/my_plugin / /opt/pipeline/plugins/my_plugin)。 |
module | ディレクトリ内の .py ファイル名 (拡張子なし) or サブパッケージ名。 |
callable | 呼ぶ関数名。既定 process。 |
init_kwargs | setup(**init_kwargs) に渡す dict。フォームで組み立てられます。 |
max_duration_ms | (任意) process() がこれを超えたら失敗扱い。0/未指定で無効。 |
callable は既定で process です。モジュール解決の内部 (slug 込みキーによる同名衝突回避) は プラグインSDK を参照してください。
success_criteria / on_success / on_failure
success_criteria は type: exit_code のときだけ実ロジックを持ち、それ以外は process() の戻り値 (成功フラグ) がそのまま使われます。on_success / on_failure フィールドはモデル・DB に存在しますが、現状ランタイムでは消費されません (予約項目)。チェーン実行が要る場合は plugin 側か self_loop で組んでください。