資料流

本頁說明 Events 如何實際進入 Osprey、Results 如何輸出，以及平台架構與預設 docker-compose 設定不同時，應調整哪些擴充點，而不必 fork Osprey。

架構概覽

Osprey 的核心是一條 pipeline。Events 傳入後由 SML Rules 評估，再輸出 Verdicts 與 Effects。

   Kafka topic(s) / PubSub / gRPC
                │
                ▼
   Osprey Coordinator (Rust, optional)
                │
                ▼
       Osprey Worker (Python)
     evaluates rules against the
      event, produces an
      ExecutionResult
                │
                ▼
   Output sink(s) + execution result store
   (stdout, Kafka, Postgres, GCS, MinIO, BigTable, or your own plugin)
                │
                ▼
        Druid + UI API → Osprey UI

Worker 評估 Rules 的架構，也就是單一 event 如何依 Rules 計算結果，請見英文官方文件撰寫規則開頭的圖。Coordinator 的 priority queues、bidirectional streaming 與 synchronous API 等內部設計，請見 Coordinator README，其中包含更完整的 component diagram 與可執行範例。

輸入資料

有三種方式可以將 event 傳入 Osprey。

1. Kafka，預設方式

Worker 預設從 Kafka topic osprey.actions_input 消費資料。這個值是 docker-compose.yaml 中的 OSPREY_KAFKA_INPUT_STREAM_TOPIC，實作位於 osprey_worker/src/osprey/worker/sinks/sink/input_stream.py 的 KafkaInputStream。每則 message 是下列結構的 JSON blob。

{
  "send_time": "<Go-formatted timestamp>",
  "data": {
    "action_id": 123,
    "action_name": "user_login",
    "data": { "...": "your event's actual fields" }
  }
}

Rules 會比對 action_name。內層 data object 則包含 event 的實際欄位，並成為 UDFs 與 Feature extractors 讀取的值。

2. Google Cloud PubSub

另一個 input source 是 InputStreamSource.PUBSUB，實作位於 osprey_worker/src/osprey/worker/sinks/input_stream_chooser.py。設定時使用 PUBSUB_OSPREY_PROJECT_ID 與 PUBSUB_OSPREY_RULES_SINK_SUBSCRIPTION。

3. Coordinator 的 synchronous gRPC API

若執行選用的 Rust coordinator osprey_coordinator/，外部服務可以直接提交單一 action 並立即取得回覆，完全不經過 Kafka。當呼叫端需要 synchronous Verdict，而非只將 event 傳入 Queue 時，可以採用這個方式。Sync Action API 使用 port 19951；workers 使用的 bidirectional streaming 是另一個 port 19950。可執行的 grpcurl 範例與完整設定請見 Coordinator README。

不 fork Osprey，改用自己的平台資料

若 Events 不是來自 Kafka、PubSub 或 coordinator，或沒有使用上述 JSON envelope，不需要 fork Osprey。下列兩個 Plugin hooks 專門處理這類情況。

• register_action_proto_deserializer　將自有 protobuf Action message 轉換為引擎預期的 JSON dict 結構
• register_input_stream　換成完全自訂的 input source，例如不同的 Queue system 或 database poller

兩者都是 pluggy hooks，註冊方式與 UDFs 及 output sinks 相同。Plugin 機制請見英文官方文件 Integrations & Plugins，可執行範例位於 example_plugins/src/register_plugins.py。

輸出資料

Rule 完成評估後，Result 會透過兩套大致獨立的機制輸出。

Output sinks

每個註冊的 BaseOutputSink 都會收到一份 ExecutionResult，註冊 hook 為 register_output_sinks。Stdlib 內建下列項目。

• StdoutOutputSink　將結果印到 stdout。若沒有其他設定，這是預設值，適合本機開發
• KafkaOutputSink　預設寫入 osprey.execution_results，對應 OSPREY_KAFKA_OUTPUT_TOPIC
• StoredExecutionResultOutputSink　透過目前設定的 ExecutionResultStore 保存資料

可以使用相同 hook 註冊自己的 output sink。這個擴充點可將 Results 傳入 review Queue、webhook 或其他外部系統。

Execution result storage

ExecutionResultStore backends 會另外保存完整 Execution Results，供後續查詢。使用 OSPREY_EXECUTION_RESULT_STORAGE_BACKEND 選擇啟用的 backend，選擇器位於 osprey_worker/src/osprey/worker/_stdlibplugin/execution_result_store_chooser.py。

Results 如何進入介面供查詢

Druid 會消費 Worker 的 Kafka output，並支援 UI API 的即時查詢。調查查詢介面、Rules與 Features Registries 都使用這些資料。若執行環境沒有啟用 Kafka output，Druid 就沒有資料可以建立索引。此時即使 Rules 正常評估，查詢介面仍會是空白。

資料治理提醒

輸入 Events、完整 Execution Results 與 Druid 索引可能保存不同範圍及期間的資料。正式導入前，應逐一確認欄位、保存位置、存取者、保留時間與刪除方式，並避免把 production secrets 寫入設定檔、Events 或 logs。