diff --git a/src/vizier/services/adaptive_export/cmd/main.go b/src/vizier/services/adaptive_export/cmd/main.go index 97a3b907793..dffdf7dbfc8 100644 --- a/src/vizier/services/adaptive_export/cmd/main.go +++ b/src/vizier/services/adaptive_export/cmd/main.go @@ -578,7 +578,8 @@ func main() { // operator-side querier is disabled (no PushPixieTables), OrderQuery returns // an error and /query 502s — start/stop + dx_evidence_graph still work. ctrlSrv := control.New(activeSet, ctl) - ctrlSrv.SetGraphWriter(applier) // dx_evidence_graph ingest → ClickHouse + ctrlSrv.SetGraphWriter(applier) // dx_evidence_graph ingest → ClickHouse + ctrlSrv.SetManifestWriter(applier) // dx_evidence_manifest ingest → ClickHouse // Bearer-JWT auth on the control surface (CodeRabbit: protect control // endpoints). Same shared lib + signing key the broker/PEM use — dx // attaches the service JWT it already mints. Default-OFF so this can diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/apply.go b/src/vizier/services/adaptive_export/internal/clickhouse/apply.go index 1faeac12bda..0115795c79a 100644 --- a/src/vizier/services/adaptive_export/internal/clickhouse/apply.go +++ b/src/vizier/services/adaptive_export/internal/clickhouse/apply.go @@ -70,6 +70,10 @@ var OperatorOwnedTables = []string{ "dx_evidence_graph", // rule-ins-only VIEW over dx_evidence_graph; created AFTER it (depends on it). "dx_evidence_graph_malignant", + // dx §9 completeness manifest — one row per verdict naming the evidence rows + // dx consulted. Created on boot so POST /dx/evidence_manifest has a target. + // Independent of dx_evidence_graph. Not a pixie table → not in PixieTables(). + "dx_evidence_manifest", } // Applier applies operator-owned DDL to a ClickHouse cluster over the @@ -121,6 +125,20 @@ func (a *Applier) WriteEvidenceGraph(ctx context.Context, jsonEachRow []byte) er return err } +// WriteEvidenceManifest inserts one dx §9 completeness manifest (per verdict) +// into forensic_db.dx_evidence_manifest. jsonEachRow is a single JSONEachRow +// object whose keys are the column names; the control handler pre-renders the +// nested collections (case_window/findings/orders/seeds/chain) as JSON text so +// the insert is ClickHouse-version independent. No-op on empty input. +func (a *Applier) WriteEvidenceManifest(ctx context.Context, jsonEachRow []byte) error { + if len(jsonEachRow) == 0 { + return nil + } + _, err := a.c.Insert(ctx, "INSERT INTO forensic_db.dx_evidence_manifest FORMAT JSONEachRow", + jsonEachRow, chhttp.InsertOptions{}) + return err +} + // execute is the DDL primitive — used by Apply for CREATE statements. func (a *Applier) execute(ctx context.Context, sql string) error { _, err := a.c.Exec(ctx, sql) diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/apply_test.go b/src/vizier/services/adaptive_export/internal/clickhouse/apply_test.go index d9856d5bb01..3c834d11a80 100644 --- a/src/vizier/services/adaptive_export/internal/clickhouse/apply_test.go +++ b/src/vizier/services/adaptive_export/internal/clickhouse/apply_test.go @@ -228,7 +228,7 @@ func TestOperatorOwnedTables_DoesNotIncludeKubescape(t *testing.T) { // plugin can auto-DDL them with the wrong schema), then the operator's // own write targets in declared order. func TestOperatorOwnedTables_TrailingOperatorTables(t *testing.T) { - want := []string{"adaptive_attribution", "trigger_watermark", "ae_reconcile", "dx_evidence_graph", "dx_evidence_graph_malignant"} + want := []string{"adaptive_attribution", "trigger_watermark", "ae_reconcile", "dx_evidence_graph", "dx_evidence_graph_malignant", "dx_evidence_manifest"} got := OperatorOwnedTables[len(OperatorOwnedTables)-len(want):] for i, w := range want { if got[i] != w { diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go b/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go index 5b658ebcbae..c9513c75b88 100644 --- a/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go +++ b/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go @@ -71,6 +71,10 @@ var KnownTables = []string{ // dx_evidence_graph UI reads this by default so benign rows are filtered // in ClickHouse, not pulled. Must follow dx_evidence_graph (depends on it). "dx_evidence_graph_malignant", + // operator-owned dx §9 completeness manifest — one row per verdict naming + // the evidence rows dx consulted (POST /dx/evidence_manifest). NOT a pixie + // table. Independent of dx_evidence_graph. + "dx_evidence_manifest", } // ErrUnknownTable is returned by DDL / Columns when asked for a table diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql b/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql index 0f055847e49..92ae3310a76 100644 --- a/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql +++ b/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql @@ -530,3 +530,34 @@ CREATE TABLE IF NOT EXISTS forensic_db.dx_evidence_graph ( -- dx_evidence_graph UI reads by default so benign rows stay in ClickHouse. CREATE VIEW IF NOT EXISTS forensic_db.dx_evidence_graph_malignant AS SELECT * FROM forensic_db.dx_evidence_graph WHERE `condition` != ''; + +-- dx_evidence_manifest — the §9 completeness contract: one row per verdict +-- (ruled_in | metastasis), naming the evidence rows dx consulted so the +-- validator can join them against what AE persisted (write⊇read, checkable). +-- Operator-owned (dx emits the manifest via POST /dx/evidence_manifest, AE +-- persists it); NOT a pixie table. Column names are the manifest.Manifest +-- JSON tags (dx internal/manifest). Same event_time (unix NANOSECONDS) + +-- hostname read-path convention as dx_evidence_graph so it is px-readable. +-- The nested collections (case_window/findings/orders/seeds/chain) are stored +-- as JSON text in String columns; the control handler pre-renders them so the +-- JSONEachRow insert is ClickHouse-version independent. +CREATE TABLE IF NOT EXISTS forensic_db.dx_evidence_manifest ( + investigation_id String, + event_time UInt64, + hostname String, + `condition` String, + verdict String, + confidence Float64, + posterior Float64, + catalog_version String, + case_window String, + findings String, + orders String, + seeds String, + chain String, + evidence_hash String +) ENGINE = MergeTree() + ORDER BY (event_time, hostname) + PARTITION BY toYYYYMM(fromUnixTimestamp64Nano(event_time)) + TTL toDateTime(fromUnixTimestamp64Nano(event_time)) + INTERVAL 30 DAY DELETE + SETTINGS index_granularity = 8192; diff --git a/src/vizier/services/adaptive_export/internal/control/server.go b/src/vizier/services/adaptive_export/internal/control/server.go index 1ebce67fd64..20ae154de20 100644 --- a/src/vizier/services/adaptive_export/internal/control/server.go +++ b/src/vizier/services/adaptive_export/internal/control/server.go @@ -59,13 +59,20 @@ type graphWriter interface { WriteEvidenceGraph(ctx context.Context, jsonEachRow []byte) error } +// manifestWriter persists one dx §9 completeness manifest per verdict +// (JSONEachRow) to forensic_db.dx_evidence_manifest. nil → /dx/evidence_manifest 501s. +type manifestWriter interface { + WriteEvidenceManifest(ctx context.Context, jsonEachRow []byte) error +} + // Server is the control HTTP surface. type Server struct { - set exporter - runner queryRunner // may be nil; /query then returns 501 - graph graphWriter // may be nil; /dx/evidence_graph then returns 501 - mux *http.ServeMux - verify func(bearer string) error // nil → auth disabled; set via SetAuth + set exporter + runner queryRunner // may be nil; /query then returns 501 + graph graphWriter // may be nil; /dx/evidence_graph then returns 501 + manifest manifestWriter // may be nil; /dx/evidence_manifest then returns 501 + mux *http.ServeMux + verify func(bearer string) error // nil → auth disabled; set via SetAuth } // New builds the control server. runner may be nil for deployments that @@ -77,12 +84,16 @@ func New(set exporter, runner queryRunner) *Server { s.mux.HandleFunc("/export/stop", s.handleStop) s.mux.HandleFunc("/query", s.handleQuery) s.mux.HandleFunc("/dx/evidence_graph", s.handleDXEvidenceGraph) + s.mux.HandleFunc("/dx/evidence_manifest", s.handleDXEvidenceManifest) return s } // SetGraphWriter wires the dx_evidence_graph sink. func (s *Server) SetGraphWriter(g graphWriter) { s.graph = g } +// SetManifestWriter wires the dx_evidence_manifest sink. +func (s *Server) SetManifestWriter(m manifestWriter) { s.manifest = m } + // SetAuth turns on bearer-JWT auth for the control surface, verified with the // SAME shared lib + signing key the vizier broker/PEM use (px.dev/pixie/src/ // shared/services/utils). dx already mints a service JWT (GenerateJWTForService, @@ -147,6 +158,81 @@ func (s *Server) handleDXEvidenceGraph(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusAccepted) } +// dxManifest mirrors the wire shape of dx's manifest.Manifest (internal/manifest). +// Scalars map to typed forensic_db.dx_evidence_manifest columns; the nested +// collections are held as raw JSON and persisted as JSON text in String columns +// so the JSONEachRow insert is ClickHouse-version independent. +type dxManifest struct { + InvestigationID string `json:"investigation_id"` + EventTime int64 `json:"event_time"` + Hostname string `json:"hostname"` + Condition string `json:"condition"` + Verdict string `json:"verdict"` + Confidence float64 `json:"confidence"` + Posterior float64 `json:"posterior"` + CatalogVersion string `json:"catalog_version"` + CaseWindow json.RawMessage `json:"case_window"` + Findings json.RawMessage `json:"findings"` + Orders json.RawMessage `json:"orders"` + Seeds json.RawMessage `json:"seeds"` + Chain json.RawMessage `json:"chain"` + EvidenceHash string `json:"evidence_hash"` +} + +// handleDXEvidenceManifest ingests ONE dx completeness manifest (per verdict) +// and writes it to forensic_db.dx_evidence_manifest as a single JSONEachRow +// row, with the nested collections rendered as JSON text. +func (s *Server) handleDXEvidenceManifest(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + if s.manifest == nil { + w.WriteHeader(http.StatusNotImplemented) + return + } + var m dxManifest + if !decode(w, r, &m) { + w.WriteHeader(http.StatusBadRequest) + return + } + row := map[string]any{ + "investigation_id": m.InvestigationID, + "event_time": m.EventTime, + "hostname": m.Hostname, + "condition": m.Condition, + "verdict": m.Verdict, + "confidence": m.Confidence, + "posterior": m.Posterior, + "catalog_version": m.CatalogVersion, + "case_window": jsonText(m.CaseWindow), + "findings": jsonText(m.Findings), + "orders": jsonText(m.Orders), + "seeds": jsonText(m.Seeds), + "chain": jsonText(m.Chain), + "evidence_hash": m.EvidenceHash, + } + line, err := json.Marshal(row) + if err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + if err := s.manifest.WriteEvidenceManifest(r.Context(), append(line, '\n')); err != nil { + w.WriteHeader(http.StatusBadGateway) + return + } + w.WriteHeader(http.StatusAccepted) +} + +// jsonText renders a nested JSON value as compact text for a String column; +// nil/absent/null → "" so the column holds an empty string rather than "null". +func jsonText(raw json.RawMessage) string { + if len(raw) == 0 || string(raw) == "null" { + return "" + } + return string(raw) +} + // ── wire types ──────────────────────────────────────────────────────── type targetReq struct { Namespace string `json:"namespace"` diff --git a/src/vizier/services/adaptive_export/internal/control/server_test.go b/src/vizier/services/adaptive_export/internal/control/server_test.go index 630e8515f58..02ab9d285e3 100644 --- a/src/vizier/services/adaptive_export/internal/control/server_test.go +++ b/src/vizier/services/adaptive_export/internal/control/server_test.go @@ -17,6 +17,8 @@ package control import ( + "context" + "encoding/json" "net/http" "net/http/httptest" "strings" @@ -218,3 +220,60 @@ type fakeErr struct{} func (fakeErr) Error() string { return "boom" } var errFake = fakeErr{} + +type fakeManifest struct { + got string + err error +} + +func (f *fakeManifest) WriteEvidenceManifest(_ context.Context, jsonEachRow []byte) error { + f.got = string(jsonEachRow) + return f.err +} + +// TestEvidenceManifest: /dx/evidence_manifest is 501 without a writer; with one it +// persists ONE JSONEachRow row per verdict, scalars as typed columns and the nested +// collections (findings/case_window/...) rendered as JSON *text* so the insert is +// ClickHouse-version independent. +func TestEvidenceManifest(t *testing.T) { + srv := New(&fakeExporter{}, nil) + if r := do(t, srv, http.MethodPost, "/dx/evidence_manifest", `{"investigation_id":"i1"}`); r.StatusCode != http.StatusNotImplemented { + t.Fatalf("no writer: got %d, want 501", r.StatusCode) + } + + fm := &fakeManifest{} + srv.SetManifestWriter(fm) + body := `{"investigation_id":"i1","event_time":1730000000000000000,"hostname":"n1",` + + `"verdict":"ruled_in","confidence":0.9,"case_window":[1.0,2.0],` + + `"findings":[{"vector":"process"}],"evidence_hash":"h"}` + if r := do(t, srv, http.MethodPost, "/dx/evidence_manifest", body); r.StatusCode != http.StatusAccepted { + t.Fatalf("got %d, want 202", r.StatusCode) + } + if !strings.HasSuffix(fm.got, "\n") { + t.Fatalf("missing JSONEachRow newline terminator: %q", fm.got) + } + var row map[string]any + if err := json.Unmarshal([]byte(strings.TrimSpace(fm.got)), &row); err != nil { + t.Fatalf("row not valid JSON: %v (%s)", err, fm.got) + } + if row["investigation_id"] != "i1" || row["hostname"] != "n1" || row["verdict"] != "ruled_in" { + t.Fatalf("scalar columns wrong: %#v", row) + } + // event_time is a large nanos int; it must round-trip as an integer, not a float in sci notation. + if !strings.Contains(fm.got, `"event_time":1730000000000000000`) { + t.Fatalf("event_time not an integer literal: %s", fm.got) + } + // nested collections persisted as JSON text strings, not raw arrays. + if row["findings"] != `[{"vector":"process"}]` { + t.Fatalf("findings should be JSON text, got %#v", row["findings"]) + } + if row["case_window"] != `[1.0,2.0]` { + t.Fatalf("case_window should be JSON text, got %#v", row["case_window"]) + } + + // writer failure surfaces as 502. + fm.err = errFake + if r := do(t, srv, http.MethodPost, "/dx/evidence_manifest", body); r.StatusCode != http.StatusBadGateway { + t.Fatalf("writer error: got %d, want 502", r.StatusCode) + } +}