用語とアーキテクチャ
Pipeline を構成する要素と、タスクが流れる仕組みを押さえます。
用語
| 用語 | 説明 |
|---|---|
| 制御プレーン | FastAPI + SQLite で動く単一プロセス。Web UI 配信 + REST API + worker heartbeat 受付を担います。既定 listen は 0.0.0.0:8000。 |
| ワーカー | 各ホストで動くデーモン。制御プレーンに約5秒ごとに heartbeat を送り、キューからタスクを claim して処理します。HTTP だけで会話する pull 型です。 |
| エージェント | (任意) 1ホスト = 1エージェントで、そのホストの worker 子プロセスを desired-state に従って spawn/監督します。VRAM 実測を報告し実効 desired を受け取る新方式です。 |
| スーパーバイザー | (任意) 需要 + 余力を見て workload の優先度 / worker 配分を自動調整する単一権威オーケストレーター。プラグインとして動きます。LLM advisor を足せば AI 提案も取り込めます。 |
| ワークロード | 「何をどう実行するか」の定義。slug で一意、Web UI から登録する1単位です。 |
| キュー | workload 毎に1つ作られるタスクの待ち行列。<slug>_queue テーブルとして保存されます。格納先は SQLite (既定) / MariaDB を選べます。 |
| タスク | キューに1行入る処理単位。pk (主キー) + extra (任意 JSON) を持ちます。 |
| Run | タスクの実行履歴1件。成否 / 経過時間 / stdout / stderr / output_json が保存されます。 |
| Executor | タスクを実際に処理する方式。実装済みは shell と python_module の2種類です。 |
| プラグイン | plugins/<name>/ に置いた Python モジュール。plugin.yaml + setup/process/cleanup を export します。 |
| フロー / タンク | Flow図上の可視化単位。workload を「装置」、キューを「タンク (液面 = 滞留量)」、外部システムを「外部ノード」として配管でつないだプラント図で表示します。 |
アーキテクチャ全体像
Pipeline は「1つの制御プレーン + N台のワーカー」で構成されます。ワーカーは pull 型なので、制御プレーンは配りに行きません。ワーカー側が自発的に heartbeat / claim / complete を HTTP で叩きます。
┌─────────────────────────────────────────────┐
│ 制御プレーン (pipeline run) │
│ FastAPI + SQLite (+ 任意で MariaDB secondary) │
│ ・Web UI 配信 │
│ ・REST API (/api/v1/...) │
│ ・キュー (各 <slug>_queue テーブル) │
│ ・self_loop watchdog / flow rate 集約 │
└───────────────▲──────────────▲────────────────┘
heartbeat/ │ claim/ │ service-log/
config poll │ complete │ metrics push
┌─────────────┴───┐ ┌───────┴───────┐
│ Worker (host A) │ │ Worker (host B) │ ...
│ plugin process │ │ plugin process │
└──────────────────┘ └─────────────────┘- 単一プロセス mode —
pipeline run --devは制御プレーン + in-process worker + Web UI を1プロセスで立ち上げます。SQLite 1ファイルで完結します。 - スケールアウト — 各ホストで
pipeline worker --control-url ...を起動。追加はデーモン起動だけ (制御プレーン側の設定変更は不要)。 - デプロイ — プラグインやコードの配信は Web UI の Deploy 画面から SSH/rsync で一括配信できます。
タスクのライフサイクル
- enqueue — Web UI または POST API でキューに pk を1件追加 (state
pending)。同じ pk はINSERT IGNOREされ重複しません。 - claim — worker がその workload を対象に
batch_size件まで claim (stateclaimed、lease_secs付きリース)。 - process — worker がプラグインの
process(task, ctx, state)を呼びます。 - complete / fail — 成功ならキュー行を削除し runs に記録、失敗なら attempt 増加 + state
failedに戻します。 - retry — failed は
max_attempts未満なら再び claim 対象に。上限到達で dead-letter として残ります (claim されなくなる)。 - lease 失効 — claim 後
lease_secs内に complete/fail が来なければ、他 worker が再 claim できます (worker 突然死からの自動回収)。
self_loop ワークロード: キューを消費する代わりに、自分で次の tick を自己 enqueue して回り続けるループ型プラグインもあります (dispatcher / supervisor / 投入系)。詳細は
プラグインSDK を参照してください。
タスクはどうやってキューに入るか
「ワーカーがキューから取り出す」側は分かりました。ではそのキューには誰がタスクを入れるのでしょうか。Pipeline には入り口が3経路あります。用途に応じて選びます。
| 経路 | 誰が | 典型用途 |
|---|---|---|
| 手動投入 | オペレータが Web UI の 📨、または POST /workloads/{slug}/tasks | 試験・再実行・アドホックな1件処理。 |
| 外部システムから POST | CI / cron / 別サービスが API を叩く | 「アップロードされたら処理」等のイベント連携。バルクは tasks/batch。 |
| dispatcher プラグイン | self_loop 型プラグインが定期的に上流を見て自動 enqueue | 常時流し続けるパイプライン本体。最も一般的。 |
3つ目が Pipeline らしい形です。dispatcher (self_loop) が上流ソースを見て新規対象を見つけ、下流ワークロードのキューへ tasks/batch で流し込み、consumer がそれを1件ずつ処理します。「発見」と「重い処理」を別ワークロードに分けることで、後者だけを worker 追加で水平スケールできます。
┌──────────┐ fetch ┌─────────────┐ tasks/batch ┌──────────────┐
│ 上流ソース │ ─────────▶ │ dispatcher │ ─────────────▶ │ consumer │ → process()
│ (feed/DB…) │ │ (self_loop) │ 下流キューへ │ (per-task) │
└──────────┘ └─────────────┘ └──────────────┘手を動かして体験するには チュートリアルのステップ6 を、実装の詳細は プラグインSDK を参照してください。
「入力ソース」機能について: README には DB / SQL / file glob / HTTP poll を入力ソースとして選ぶ構想が書かれていますが、これは将来の実装予定 (未実装) です。現状タスクを供給する正規の方法は上記3経路、とりわけ dispatcher プラグインです。
なぜ pull 型なのか
制御プレーンがワーカーに仕事を「押し込む (push)」のではなく、ワーカーが「取りに行く (pull)」設計には理由があります。
- スケールが自明 — ワーカーは起動して heartbeat を送るだけで参加します。制御プレーン側にワーカー一覧の設定を持たなくてよい。
- 障害に強い — ワーカーが落ちても、claim した lease が失効すれば他のワーカーが同じタスクを再 claim します。押し込み型のように「配ったのに相手が死んだ」状態を追跡する必要がありません。
- 背圧が自然にかかる — 忙しいワーカーは次を取りに来ないだけ。過負荷ワーカーへ仕事が積み上がりません。
- NAT/FW を越えやすい — ワーカー → 制御プレーンの一方向 HTTP だけで動きます。