diff --git a/src/vizier/services/adaptive_export/cmd/main.go b/src/vizier/services/adaptive_export/cmd/main.go index 30db7220f77..97a3b907793 100644 --- a/src/vizier/services/adaptive_export/cmd/main.go +++ b/src/vizier/services/adaptive_export/cmd/main.go @@ -576,9 +576,9 @@ func main() { // Wire the controller as the /query runner: dx OrderQuery → one-shot pixie // capture written to forensic_db (write⊇read; entlein/dx#93). When the // operator-side querier is disabled (no PushPixieTables), OrderQuery returns - // an error and /query 502s — start/stop + dx_attack_graph still work. + // an error and /query 502s — start/stop + dx_evidence_graph still work. ctrlSrv := control.New(activeSet, ctl) - ctrlSrv.SetGraphWriter(applier) // dx_attack_graph ingest → ClickHouse + ctrlSrv.SetGraphWriter(applier) // dx_evidence_graph 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 84b3afbbcb0..1faeac12bda 100644 --- a/src/vizier/services/adaptive_export/internal/clickhouse/apply.go +++ b/src/vizier/services/adaptive_export/internal/clickhouse/apply.go @@ -67,9 +67,9 @@ var OperatorOwnedTables = []string{ // dx_evidence_graph UI (px.DataFrame clickhouse_dsn) has a real, // globally-registered table to read. dx emits edges, AE persists. // Not a pixie socket_tracer table → not in PixieTables(). - "dx_attack_graph", - // rule-ins-only VIEW over dx_attack_graph; created AFTER it (depends on it). - "dx_attack_graph_malicious", + "dx_evidence_graph", + // rule-ins-only VIEW over dx_evidence_graph; created AFTER it (depends on it). + "dx_evidence_graph_malignant", } // Applier applies operator-owned DDL to a ClickHouse cluster over the @@ -108,15 +108,15 @@ func (a *Applier) Apply(ctx context.Context) error { return nil } -// WriteAttackGraph inserts dx evidence-graph edges into -// forensic_db.dx_attack_graph. jsonEachRow is newline-delimited JSON objects +// WriteEvidenceGraph inserts dx evidence-graph edges into +// forensic_db.dx_evidence_graph. jsonEachRow is newline-delimited JSON objects // whose keys are the column names (JSONEachRow; unknown keys are skipped, // missing columns default). No-op on empty input. -func (a *Applier) WriteAttackGraph(ctx context.Context, jsonEachRow []byte) error { +func (a *Applier) WriteEvidenceGraph(ctx context.Context, jsonEachRow []byte) error { if len(jsonEachRow) == 0 { return nil } - _, err := a.c.Insert(ctx, "INSERT INTO forensic_db.dx_attack_graph FORMAT JSONEachRow", + _, err := a.c.Insert(ctx, "INSERT INTO forensic_db.dx_evidence_graph FORMAT JSONEachRow", jsonEachRow, chhttp.InsertOptions{}) return err } 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 e108e05540c..d9856d5bb01 100644 --- a/src/vizier/services/adaptive_export/internal/clickhouse/apply_test.go +++ b/src/vizier/services/adaptive_export/internal/clickhouse/apply_test.go @@ -59,7 +59,7 @@ func TestApply_ExecutesEveryOperatorOwnedTable(t *testing.T) { } // Spot-check that the SECOND call is for the first OperatorOwnedTables entry, // and that the LAST call is for the last OperatorOwnedTables entry (robust to - // new operator-owned tables being appended, e.g. dx_attack_graph). + // new operator-owned tables being appended, e.g. dx_evidence_graph). if !strings.Contains(bodies[1], "forensic_db."+OperatorOwnedTables[0]) { t.Fatalf("second DDL not for %s; got: %s", OperatorOwnedTables[0], bodies[1]) } @@ -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_attack_graph", "dx_attack_graph_malicious"} + want := []string{"adaptive_attribution", "trigger_watermark", "ae_reconcile", "dx_evidence_graph", "dx_evidence_graph_malignant"} 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 e4503bb340c..5b658ebcbae 100644 --- a/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go +++ b/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go @@ -66,11 +66,11 @@ var KnownTables = []string{ "ae_reconcile", // operator-owned dx evidence-graph edge list (read by the Pixie // dx_evidence_graph UI via clickhouse_dsn). NOT a pixie table. - "dx_attack_graph", - // rule-ins-only VIEW over dx_attack_graph (condition != ''); the + "dx_evidence_graph", + // rule-ins-only VIEW over dx_evidence_graph (condition != ''); the // dx_evidence_graph UI reads this by default so benign rows are filtered - // in ClickHouse, not pulled. Must follow dx_attack_graph (depends on it). - "dx_attack_graph_malicious", + // in ClickHouse, not pulled. Must follow dx_evidence_graph (depends on it). + "dx_evidence_graph_malignant", } // 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 494285b3d12..0f055847e49 100644 --- a/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql +++ b/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql @@ -487,7 +487,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.ae_reconcile ( -- unbounded storage (CodeRabbit). 30d matches the pixie observation tables. TTL toDateTime(ts) + INTERVAL 30 DAY DELETE; --- dx_attack_graph — dx evidence-graph edge list: one row per directed hop of an +-- dx_evidence_graph — dx evidence-graph edge list: one row per directed hop of an -- investigation (delivery/egress/execution/exfil/pivot), read by the Pixie -- dx_evidence_graph UI via px.DataFrame(clickhouse_dsn=...). Operator-owned -- (dx emits the edges, AE persists them); NOT a pixie socket_tracer table. @@ -498,7 +498,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.ae_reconcile ( -- event_time". Same convention as kubescape_logs. event_time is nanos, so the -- partition/TTL use fromUnixTimestamp64Nano (toDateTime would read ns as seconds -- → year ~58e9 → broken partitions; see the soc#225 fix). -CREATE TABLE IF NOT EXISTS forensic_db.dx_attack_graph ( +CREATE TABLE IF NOT EXISTS forensic_db.dx_evidence_graph ( investigation_id String, event_time UInt64, hostname String, @@ -526,7 +526,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.dx_attack_graph ( TTL toDateTime(fromUnixTimestamp64Nano(event_time)) + INTERVAL 30 DAY DELETE SETTINGS index_granularity = 8192; --- dx_attack_graph_malicious — rule-ins-only view (condition != '') the +-- dx_evidence_graph_malignant — rule-ins-only view (condition != '') the -- dx_evidence_graph UI reads by default so benign rows stay in ClickHouse. -CREATE VIEW IF NOT EXISTS forensic_db.dx_attack_graph_malicious AS - SELECT * FROM forensic_db.dx_attack_graph WHERE `condition` != ''; +CREATE VIEW IF NOT EXISTS forensic_db.dx_evidence_graph_malignant AS + SELECT * FROM forensic_db.dx_evidence_graph WHERE `condition` != ''; diff --git a/src/vizier/services/adaptive_export/internal/control/server.go b/src/vizier/services/adaptive_export/internal/control/server.go index 88e67d76369..1ebce67fd64 100644 --- a/src/vizier/services/adaptive_export/internal/control/server.go +++ b/src/vizier/services/adaptive_export/internal/control/server.go @@ -54,16 +54,16 @@ type queryRunner interface { } // graphWriter persists dx evidence-graph edges (newline-delimited JSON, -// JSONEachRow) to forensic_db.dx_attack_graph. nil → /dx/attack_graph 501s. +// JSONEachRow) to forensic_db.dx_evidence_graph. nil → /dx/evidence_graph 501s. type graphWriter interface { - WriteAttackGraph(ctx context.Context, jsonEachRow []byte) error + WriteEvidenceGraph(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/attack_graph 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 } @@ -76,11 +76,11 @@ func New(set exporter, runner queryRunner) *Server { s.mux.HandleFunc("/export/start", s.handleStart) s.mux.HandleFunc("/export/stop", s.handleStop) s.mux.HandleFunc("/query", s.handleQuery) - s.mux.HandleFunc("/dx/attack_graph", s.handleDXAttackGraph) + s.mux.HandleFunc("/dx/evidence_graph", s.handleDXEvidenceGraph) return s } -// SetGraphWriter wires the dx_attack_graph sink. +// SetGraphWriter wires the dx_evidence_graph sink. func (s *Server) SetGraphWriter(g graphWriter) { s.graph = g } // SetAuth turns on bearer-JWT auth for the control surface, verified with the @@ -115,9 +115,9 @@ func (s *Server) Handler() http.Handler { }) } -// handleDXAttackGraph ingests a JSON array of dx evidence-graph edges and writes -// them to forensic_db.dx_attack_graph (as JSONEachRow). -func (s *Server) handleDXAttackGraph(w http.ResponseWriter, r *http.Request) { +// handleDXEvidenceGraph ingests a JSON array of dx evidence-graph edges and writes +// them to forensic_db.dx_evidence_graph (as JSONEachRow). +func (s *Server) handleDXEvidenceGraph(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { w.WriteHeader(http.StatusMethodNotAllowed) return @@ -140,7 +140,7 @@ func (s *Server) handleDXAttackGraph(w http.ResponseWriter, r *http.Request) { buf.Write(e) buf.WriteByte('\n') } - if err := s.graph.WriteAttackGraph(r.Context(), buf.Bytes()); err != nil { + if err := s.graph.WriteEvidenceGraph(r.Context(), buf.Bytes()); err != nil { w.WriteHeader(http.StatusBadGateway) return } @@ -175,7 +175,7 @@ func (t targetReq) target() anomaly.Target { } // maxControlBodyBytes caps a single control-surface request body. The -// largest legitimate payload we accept is /dx/attack_graph which is a +// largest legitimate payload we accept is /dx/evidence_graph which is a // JSON array of pre-marshalled JSONEachRow lines — measured live the // hottest dx rule-in pass fits in ~256 KiB. 4 MiB is well above that // and below the per-pod memory headroom an oversized POST could diff --git a/src/vizier/services/adaptive_export/internal/controller/BUILD.bazel b/src/vizier/services/adaptive_export/internal/controller/BUILD.bazel index 5e19fbeaf1e..6024b7ce5c7 100644 --- a/src/vizier/services/adaptive_export/internal/controller/BUILD.bazel +++ b/src/vizier/services/adaptive_export/internal/controller/BUILD.bazel @@ -34,7 +34,10 @@ go_library( pl_go_test( name = "controller_test", - srcs = ["controller_test.go"], + srcs = [ + "controller_test.go", + "order_query_test.go", + ], embed = [":controller"], deps = [ "//src/vizier/services/adaptive_export/internal/anomaly", diff --git a/src/vizier/services/adaptive_export/internal/controller/order_query_test.go b/src/vizier/services/adaptive_export/internal/controller/order_query_test.go index 8ba079c8f13..5519e9d4d25 100644 --- a/src/vizier/services/adaptive_export/internal/controller/order_query_test.go +++ b/src/vizier/services/adaptive_export/internal/controller/order_query_test.go @@ -46,6 +46,7 @@ func (s *recordingSink) Write(context.Context, []sink.AttributionRow) error { re func (s *recordingSink) QueryActive(context.Context, string) ([]sink.AttributionRow, error) { return nil, nil } + func (s *recordingSink) WritePixieRows(_ context.Context, table string, rows []map[string]any) error { if s.werr != nil { return s.werr @@ -55,6 +56,7 @@ func (s *recordingSink) WritePixieRows(_ context.Context, table string, rows []m s.written[table] += len(rows) return nil } + func (s *recordingSink) count(table string) int { s.mu.Lock() defer s.mu.Unlock() @@ -104,7 +106,7 @@ func TestOrderQueryWritesRows(t *testing.T) { } // With the operator-side querier disabled, OrderQuery errors (so /query 502s) — -// start/stop + dx_attack_graph remain usable. +// start/stop + dx_evidence_graph remain usable. func TestOrderQueryNoQuerierErrors(t *testing.T) { snk := newRecordingSink() lo, hi := oqWindow()