adaptive_export: restore dark-vector tracepoint table export onto main (8 dx_* tables)

.gitignore

@@ -79,3 +79,6 @@ Vagrantfile
 # Ignore symlink to bazel_output_base/external
 external
 !bazel/external
+
+# Stray root go build output (accidental commit of the adaptive_export binary)
+/cmd

mypy.ini

@@ -3,6 +3,13 @@ python_version = 3.8
 show_column_numbers = True
 show_error_context = False
 
+# arc lint invokes mypy once per file, in parallel. Those processes share
+# the on-disk incremental cache in the repo root, and concurrent writers
+# corrupt it -- yielding nondeterministic `INTERNAL ERROR` crashes (the file
+# that crashes varies run to run). Disable the cache so each invocation is
+# self-contained; per-file runs get no incremental benefit anyway.
+cache_dir = /dev/null
+
 # suppress errors about unsatisfied imports
 ignore_missing_imports = True
 

src/e2e_test/adaptive_export_loadtest/CONTRACTS.md

@@ -47,7 +47,9 @@ flowchart TD
 flowchart LR
     ENV["ENV (all non-empty or FATAL):<br/>PIXIE_CLUSTER_ID · CLUSTER_NAME<br/>PIXIE_API_KEY · CLICKHOUSE_DSN"] --> BOOT
     CM["cm/pl-cloud-config<br/>PL_CLOUD_ADDR=…:443"] -->|"C11 🔴 missing :443 → crashloop"| BOOT
-    BOOT["AE boot"] --> DDL["C12 self-applies forensic_db DDL<br/>(ADAPTIVE_SKIP_APPLY=false)"]
+    BOOT["AE boot"] --> DDL["C12a self-applies forensic_db DDL<br/>(schemata+tables, ADAPTIVE_SKIP_APPLY=false)"]
+    BOOT --> TRACE["C12b/C17 deploys dark-vector bpftraces<br/>(dc_snoop, creds_change) via UpsertTracepoint<br/>mutation (INSTALL_PRESET_SCRIPTS=true)"]
+    BOOT --> PRESETS["C12c registers ch-&lt;table&gt; export presets<br/>+ native-DSN plugin (C16)"]
     BOOT --> CTRLPLANE["control plane: CH only"]
     BOOT --> DATAPLANE["data plane: needs query-broker<br/>(C7) + ADAPTIVE_PUSH_PIXIE_ROWS"]
 ```
@@ -67,10 +69,13 @@ flowchart LR
 | C9 | a protocol table row is written only if Pixie returned ≥1 row | ✅ `WritePixieRows len==0 → nil` | ok (empty workload → 0 rows, by design) |
 | C10 | join key: `events.pod` = `"ns/pod"` (upid_to_pod_name) vs `adaptive_attribution.pod` = **bare** pod | ❌ asymmetric | ⚠️ consumers must `concat(namespace,'/',pod)` to join (burned the volume tool) |
 | C11 | `PL_CLOUD_ADDR` carries `:443` | ❌ | 🔴 missing → AE crashloops / 0 writes (per-PG fix) |
-| C12 | AE owns + self-applies the `forensic_db` DDL | ✅ when `ADAPTIVE_SKIP_APPLY=false` | ok; but DDL TTL/PARTITION assume seconds (C1) |
+| C12 | **AE owns the schemata, the table deployments, AND the trace deployments.** (a) self-applies the `forensic_db` DDL (schemata + tables) via `apply.go`; (b) deploys + keeps the dark-vector **bpftrace tracepoints** (`script.DesiredTracepoints()` → `dc_snoop`, `creds_change`, …) at boot via a **mutation** `ExecuteScript` (`import pxtrace` + `UpsertTracepoint`, permanent TTL, idempotent upsert); (c) registers the retention **export presets** (`ch-<table>`) that read those tables + native profiler and export to CH. | ✅ (a) when `ADAPTIVE_SKIP_APPLY=false`; ✅ (b)(c) when `INSTALL_PRESET_SCRIPTS=true` | The retention/cron export path **cannot** deploy tracepoints (its `pxtrace` mutation is dropped) — hence the AE owns deployment separately (C17). DDL TTL/PARTITION assume seconds (C1). |
 | C13 | `adaptive_attribution` / protocol writes are durable | ❌ best-effort: logged, non-fatal, **not retried** | 🔴 silent loss under CH hiccup; AE-4 retry+count |
 | C14 | **DX⊇AE invariant**: AE write-set ⊇ DX read-set (AE persists everything dx queries) | ❌ by convention | ⚠️ validated per-table in the load-test, not enforced in code |
 | C15 | **Write-duration (the one DX steers on):** once an anomaly opens a pod's window, AE **keeps re-pulling + writing that pod's forensic data continuously** until `t_end` expires OR DX explicitly stops it. `t_end = now + After`, extended by each new anomaly for the hash. | ❌ partial | 🔴 **last week's "wrote then stopped" bug.** Premature stop modes under investigation (E8-data RCA): (a) F8 — extension anomalies dropped → `t_end` not extended → expires early; (b) EmptyResultSkip negative cache skips a (pod,table) mid-window after N empty pulls; (c) prune/in-flight race; (d) my `PUSH_REFRESH=-1` single-shot is a TEST affordance that *violates* this contract (writes once) — production must re-pull. |
+| C16 | **Retention-plugin export uses the NATIVE ClickHouse DSN + nanosecond `event_time`.** The plugin sink is the query engine's native `ClickHouseExportSink` (clickhouse-cpp, **TCP :9000**), NOT the AE's own HTTP write path (:8123). AE must pass `config.NativeDSN()` = `clickhouse://user:pass@host:9000/db` (an HTTP DSN makes the sink parse "http" as the username → segfault → vizier Unhealthy). Every export preset sets `df.event_time = df.time_` so the sink emits `event_time` as `DateTime64(9)` nanos via its normal type map, instead of auto-appending a `DateTime64(3)` millis column (which mismatches the DDL + breaks C1's nanos-everywhere). Table column types must match the sink map exactly (`upid`→String, all ints→Int64, `time_`→DateTime64(9)) or the INSERT throws and the client segfaults. | ✅ `NativeDSN()` + boot-race retry (aeprod36); ✅ `df.event_time` in every preset (aeprod37) | Deploy sets `CLICKHOUSE_PORT=9000` (native); AE's own HTTP writes still target :8123 via `chHTTPEndpoint` (never uses `Port()`). |
+| C17 | **AE deploys the desired bpftraces at boot; the cron export path never does.** `script.DesiredTracepoints()` is the source of truth (currently `dc_snoop`, `creds_change`; `stack_traces.beta`/V9 needs none — native profiler). Each is a `<name>_deploy.pxl` (`import pxtrace` + `UpsertTracepoint`, TTL 876000h ≈ permanent), run as a mutation via `deployDesiredTracepoints` with retry. The matching `ch-<name>` export preset is query-only. | ✅ when `INSTALL_PRESET_SCRIPTS=true` (needs the pixie adapter — direct-mode `ADAPTIVE_VIZIER_DIRECT_ADDR` or cloud) | Extend `DesiredTracepoints()` as new bpftraces (V6 mprotect, V8 bpf/ptrace, …) land. Splitting deploy vs export was required because the cron executor drops the tracepoint mutation. |
+| C18 | **Dark-vector rows carry full k8s metadata attribution** (`namespace`, `pod`, `container`, `hostname`=node). Tracepoint tables emit a raw kernel pid with no upid, so the `dc_snoop`/`creds_change` export presets resolve the metadata by a **process_stats merge on pid** (`px.upid_to_pid` + `ctx['namespace'/'pod'/'container']` + `px.upid_to_node_name`; the validated PodEnrichPxL join, pid-only not pid+asid — see compile.go). Best-effort **left** join: blank for host/transient pids (correct — they have no pod), so a short-lived process must live long enough to be sampled by process_stats. `stack_trace` needs no merge — `stack_traces.beta` carries upid and resolves via `ctx`. | ✅ in the presets + DDL (`container` column added) | pid collisions across nodes are a known best-effort limitation of the pid-only join. The creds_change calibration verifies namespace/pod/container/node resolve to the firing workload. |
 
 ## DX steering contract (what DX can rely on / control)
 

src/e2e_test/adaptive_export_loadtest/suite/creds_change_calibration_test.go

@@ -0,0 +1,216 @@
+// 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"
+)
+
+// credsSentinel is a distinctive real-uid used only by this calibration, so the
+// resulting creds_change row is unambiguous (no collision with a system daemon
+// that happens to change credentials to root).
+const credsSentinel = 12345
+
+// TestCredsChangeCalibration is the end-to-end calibration for the creds_change
+// dark-vector tracepoint (V7 credential vector). It guarantees, on every run,
+// the two properties the trace exists to provide:
+//
+//	a) the TRACE WORKS — the AE-deployed commit_creds bpftrace captures a real
+//	   privilege escalation (a process whose REAL uid transitions >0 -> 0), and
+//	b) ATTRIBUTION reaches ClickHouse — the event flows Pixie -> AE retention
+//	   export -> forensic_db.creds_change carrying its pid + comm identity (the
+//	   filterable base dx projects as V7 IPC/credential evidence).
+//
+// The escalation is fired deterministically with a stock image and no custom
+// binary: a root container drops its REAL uid to the sentinel while KEEPING
+// effective uid 0 (so it stays privileged), then setuid(0) pulls the real uid
+// back to 0 — exactly the commit_creds(new_uid==0 && old_uid>0) the tracepoint
+// filters for. Reading the effective/saved trick wrong is the classic pitfall:
+// setuid(0) from a non-privileged euid changes only euid, leaving the real uid
+// untouched (no match), so we must retain euid=0 across the drop.
+//
+// Live + e2e gated: AELOAD_LIVE=1 AELOAD_E2E=1. Requires kubectl + a deployed AE
+// with INSTALL_PRESET_SCRIPTS=true (so the AE has deployed the creds_change
+// tracepoint at boot and registered its export preset).
+func TestCredsChangeCalibration(t *testing.T) {
+	e := RequireLiveEnv(t)
+	if os.Getenv("AELOAD_E2E") != "1" {
+		t.Skip("AELOAD_E2E != 1 — creds_change calibration (fires a real privilege escalation) skipped")
+	}
+	requireRunning(t, e, e.AENS, e.AEDaemon)
+
+	const ns, job = "creds-calib", "creds-calib"
+	// Everything on/after this instant is "post-fire". -2s absorbs minor clock
+	// skew between the test host and the ClickHouse/PEM nodes.
+	fireStart := time.Now().Add(-2 * time.Second).UnixNano()
+
+	base := e.QueryInt(t, fmt.Sprintf(
+		"SELECT count() FROM forensic_db.creds_change WHERE old_uid=%d AND new_uid=0", credsSentinel))
+	t.Logf("baseline creds_change(old_uid=%d,new_uid=0) = %d", credsSentinel, base)
+
+	// --- fire the escalation on the AE's node ---
+	kubeTry("create", "namespace", ns)
+	t.Cleanup(func() { kubeTry("delete", "namespace", ns, "--wait=false") })
+	kubeApplyStdin(t, credsCalibJob(ns, job, e.Node, credsSentinel))
+	waitCredsJobRan(t, ns)
+
+	// --- assert (a) trace works + (b) attribution in CH ---
+	// Export is the retention-plugin cron (10s) + native-sink lag; poll to 3m.
+	var got credsRow
+	deadline := time.Now().Add(3 * time.Minute)
+	for time.Now().Before(deadline) {
+		if got = e.queryCredsRow(t, credsSentinel, fireStart); got.count > 0 {
+			break
+		}
+		time.Sleep(5 * time.Second)
+	}
+	require.Positivef(t, got.count,
+		"no creds_change row (old_uid=%d,new_uid=0) reached forensic_db after firing the escalation — "+
+			"the commit_creds tracepoint is not deployed/capturing OR the AE export is not flowing to ClickHouse",
+		credsSentinel)
+	t.Logf("(a) TRACE OK: creds_change captured count=%d pid=%d comm=%q old_uid=%d new_uid=0",
+		got.count, got.pid, got.comm, credsSentinel)
+
+	// (b) attribution base: the pid + comm identity dx filters/projects on.
+	require.Positivef(t, got.pid, "creds_change row carries no pid — attribution incomplete")
+	require.NotEmptyf(t, got.comm, "creds_change row carries no comm — attribution incomplete")
+	t.Logf("(b) ATTRIBUTION OK: pid=%d comm=%q reached ClickHouse", got.pid, got.comm)
+
+	// (c) pod/namespace attribution via the process_stats pid-merge in the export
+	// preset. The escalation pod sleeps so process_stats captures its pid. This is
+	// a best-effort left join, so assert-or-log: when attribution lands we verify
+	// it is the calibration's own namespace (correctness); a merge miss is logged,
+	// not a hard flake. Harden to require once proven stable on a live rig.
+	if got.pod != "" || got.namespace != "" {
+		t.Logf("(c) METADATA ATTRIBUTION OK: namespace=%q pod=%q container=%q node=%q reached ClickHouse",
+			got.namespace, got.pod, got.container, got.hostname)
+		require.Containsf(t, got.namespace, ns,
+			"creds_change namespace=%q did not resolve to the calibration namespace %q", got.namespace, ns)
+	} else {
+		t.Logf("NOTE: creds_change pod/namespace empty — process_stats pid-merge did not attribute pid=%d "+
+			"(best-effort join; check the enrichment / process_stats coverage)", got.pid)
+	}
+}
+
+// credsRow is the single freshest sentinel escalation row read back from CH.
+type credsRow struct {
+	count     int
+	pid       int
+	comm      string
+	namespace string
+	pod       string
+	container string
+	hostname  string
+}
+
+// queryCredsRow reads the creds_change row(s) for the sentinel escalation fired
+// on/after sinceNanos. count>0 proves the trace + export worked; pid/comm/pod
+// carry the attribution.
+func (e Env) queryCredsRow(t *testing.T, oldUID int, sinceNanos int64) credsRow {
+	where := fmt.Sprintf(
+		"old_uid=%d AND new_uid=0 AND toUnixTimestamp64Nano(event_time) >= %d", oldUID, sinceNanos)
+	r := credsRow{count: e.QueryInt(t, "SELECT count() FROM forensic_db.creds_change WHERE "+where)}
+	if r.count == 0 {
+		return r
+	}
+	// anyIf(x, x!='') prefers an attributed row if any export landed pod/namespace,
+	// so a later enriched write wins over an earlier bare one for the same event.
+	r.pid = e.QueryInt(t, "SELECT any(pid) FROM forensic_db.creds_change WHERE "+where)
+	r.comm = strings.TrimSpace(e.Query(t, "SELECT any(comm) FROM forensic_db.creds_change WHERE "+where))
+	r.namespace = strings.TrimSpace(e.Query(t, "SELECT anyIf(namespace, namespace!='') FROM forensic_db.creds_change WHERE "+where))
+	r.pod = strings.TrimSpace(e.Query(t, "SELECT anyIf(pod, pod!='') FROM forensic_db.creds_change WHERE "+where))
+	r.container = strings.TrimSpace(e.Query(t, "SELECT anyIf(container, container!='') FROM forensic_db.creds_change WHERE "+where))
+	r.hostname = strings.TrimSpace(e.Query(t, "SELECT anyIf(hostname, hostname!='') FROM forensic_db.creds_change WHERE "+where))
+	return r
+}
+
+// credsCalibJob renders a one-shot Job that fires exactly one
+// commit_creds(new_uid==0 && old_uid>0). setresuid(sentinel,0,0) drops the REAL
+// uid to the sentinel while keeping effective uid 0 (privileged); setuid(0) then
+// pulls the real uid back to 0 — the escalation the tracepoint filters for.
+// python is present in python:3-slim; no custom image or setuid binary needed.
+func credsCalibJob(ns, name, node string, oldUID int) string {
+	// After firing the escalation the process sleeps ~20s so it is alive long
+	// enough for Pixie's process_stats to capture its pid — the dc_snoop/
+	// creds_change export presets resolve pod/namespace by merging process_stats
+	// on pid, and a sub-second process would never be sampled (empty attribution).
+	py := fmt.Sprintf(
+		"import os,time; os.setresuid(%d,0,0); os.setuid(0); print('credcalib escalated', os.getresuid()); time.sleep(20)",
+		oldUID)
+	nodeLine := ""
+	if node != "" {
+		nodeLine = "\n      nodeName: " + node
+	}
+	return fmt.Sprintf(`apiVersion: batch/v1
+kind: Job
+metadata:
+  name: %s
+  namespace: %s
+spec:
+  backoffLimit: 1
+  ttlSecondsAfterFinished: 120
+  template:
+    metadata:
+      labels: { app: creds-calib }
+    spec:
+      restartPolicy: Never%s
+      containers:
+      - name: escalate
+        image: python:3-slim
+        command: ["python3","-c","%s"]
+        securityContext:
+          runAsUser: 0
+          allowPrivilegeEscalation: true
+          capabilities:
+            add: ["SETUID","SETGID"]
+`, name, ns, nodeLine, py)
+}
+
+// kubeApplyStdin applies a manifest piped over stdin (kubectl apply -f -).
+func kubeApplyStdin(t *testing.T, manifest string) {
+	t.Helper()
+	cmd := exec.Command("kubectl", "apply", "-f", "-")
+	cmd.Stdin = strings.NewReader(manifest)
+	out, err := cmd.CombinedOutput()
+	require.NoErrorf(t, err, "kubectl apply creds-calib job:\n%s\n%s", manifest, string(out))
+}
+
+// waitCredsJobRan blocks until the escalation pod reached a terminal phase. The
+// commit_creds event fires the instant setuid(0) runs, so either Succeeded or
+// Failed means the trace has already had its chance to capture.
+func waitCredsJobRan(t *testing.T, ns string) {
+	t.Helper()
+	deadline := time.Now().Add(90 * time.Second)
+	for time.Now().Before(deadline) {
+		out, _ := exec.Command("kubectl", "-n", ns, "get", "pods", "-l", "app=creds-calib",
+			"-o", "jsonpath={.items[*].status.phase}").CombinedOutput()
+		phase := strings.TrimSpace(string(out))
+		if strings.Contains(phase, "Succeeded") || strings.Contains(phase, "Failed") {
+			t.Logf("creds-calib pod phase: %s", phase)
+			return
+		}
+		time.Sleep(3 * time.Second)
+	}
+	t.Log("creds-calib pod did not reach a terminal phase in 90s — polling ClickHouse anyway")
+}

src/vizier/services/adaptive_export/cmd/BUILD.bazel

@@ -15,7 +15,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 load("@io_bazel_rules_go//go:def.bzl", "go_library")
-load("@px//bazel:pl_build_system.bzl", "pl_go_binary")
+load("@px//bazel:pl_build_system.bzl", "pl_go_binary", "pl_go_test")
 
 go_library(
     name = "cmd_lib",
@@ -40,6 +40,9 @@ go_library(
         "//src/vizier/services/adaptive_export/internal/streaming",
         "//src/vizier/services/adaptive_export/internal/trigger",
         "@com_github_sirupsen_logrus//:logrus",
+        "@io_k8s_apimachinery//pkg/apis/meta/v1:meta",
+        "@io_k8s_client_go//kubernetes",
+        "@io_k8s_client_go//rest",
     ],
 )
 
@@ -48,3 +51,9 @@ pl_go_binary(
     embed = [":cmd_lib"],
     visibility = ["//visibility:public"],
 )
+
+pl_go_test(
+    name = "cmd_test",
+    srcs = ["leader_test.go"],
+    embed = [":cmd_lib"],
+)

src/vizier/services/adaptive_export/cmd/leader_test.go

@@ -0,0 +1,63 @@
+// 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 main
+
+import "testing"
+
+// TestLeaderNodeIsDeterministic pins the cluster-setup leader election: the
+// lexicographically smallest node name, computed identically by every AE pod, so
+// exactly one pod registers the cluster-scoped retention scripts + tracepoints.
+// This is the guard against the DaemonSet duplicate-registration bug (N pods → N
+// duplicate cron scripts → N-times-duplicated dark-table exports).
+func TestLeaderNodeIsDeterministic(t *testing.T) {
+	cases := []struct {
+		name  string
+		nodes []string
+		want  string
+	}{
+		{"two nodes — smallest wins", []string{"node-01", "cplane-01"}, "cplane-01"},
+		{"order independent", []string{"cplane-01", "node-01"}, "cplane-01"},
+		{"skips empty node names", []string{"node-b", "", "node-a"}, "node-a"},
+		{"single pod is its own leader", []string{"only-node"}, "only-node"},
+		{"no scheduled pods", []string{}, ""},
+		{"all empty", []string{"", ""}, ""},
+		{"duplicates collapse", []string{"n2", "n1", "n1", "n2"}, "n1"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			if got := leaderNode(c.nodes); got != c.want {
+				t.Errorf("leaderNode(%v) = %q, want %q", c.nodes, got, c.want)
+			}
+		})
+	}
+}
+
+// TestLeaderNodeElectsExactlyOne — across every pod's view of the SAME node set,
+// exactly one node is the leader (the invariant that prevents duplicate setup).
+func TestLeaderNodeElectsExactlyOne(t *testing.T) {
+	nodes := []string{"node-03", "node-01", "node-02"}
+	leader := leaderNode(nodes)
+	winners := 0
+	for _, myNode := range nodes {
+		if myNode == leader {
+			winners++
+		}
+	}
+	if winners != 1 {
+		t.Fatalf("expected exactly one leader among %v, got %d (leader=%q)", nodes, winners, leader)
+	}
+}