diff --git a/.gitignore b/.gitignore index 0b5d53bd0b9..756ce9e3ac2 100644 --- a/.gitignore +++ b/.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 diff --git a/mypy.ini b/mypy.ini index 67f8a7e5215..68e7b4df66b 100644 --- a/mypy.ini +++ b/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 diff --git a/src/e2e_test/adaptive_export_loadtest/CONTRACTS.md b/src/e2e_test/adaptive_export_loadtest/CONTRACTS.md index f7cd131d130..55d4785ec2a 100644 --- a/src/e2e_test/adaptive_export_loadtest/CONTRACTS.md +++ b/src/e2e_test/adaptive_export_loadtest/CONTRACTS.md @@ -47,7 +47,9 @@ flowchart TD flowchart LR ENV["ENV (all non-empty or FATAL):
PIXIE_CLUSTER_ID Β· CLUSTER_NAME
PIXIE_API_KEY Β· CLICKHOUSE_DSN"] --> BOOT CM["cm/pl-cloud-config
PL_CLOUD_ADDR=…:443"] -->|"C11 πŸ”΄ missing :443 β†’ crashloop"| BOOT - BOOT["AE boot"] --> DDL["C12 self-applies forensic_db DDL
(ADAPTIVE_SKIP_APPLY=false)"] + BOOT["AE boot"] --> DDL["C12a self-applies forensic_db DDL
(schemata+tables, ADAPTIVE_SKIP_APPLY=false)"] + BOOT --> TRACE["C12b/C17 deploys dark-vector bpftraces
(dc_snoop, creds_change) via UpsertTracepoint
mutation (INSTALL_PRESET_SCRIPTS=true)"] + BOOT --> PRESETS["C12c registers ch-<table> export presets
+ native-DSN plugin (C16)"] BOOT --> CTRLPLANE["control plane: CH only"] BOOT --> DATAPLANE["data plane: needs query-broker
(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-`) 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 `_deploy.pxl` (`import pxtrace` + `UpsertTracepoint`, TTL 876000h β‰ˆ permanent), run as a mutation via `deployDesiredTracepoints` with retry. The matching `ch-` 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) diff --git a/src/e2e_test/adaptive_export_loadtest/suite/creds_change_calibration_test.go b/src/e2e_test/adaptive_export_loadtest/suite/creds_change_calibration_test.go new file mode 100644 index 00000000000..d27384347d0 --- /dev/null +++ b/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") +} diff --git a/src/vizier/services/adaptive_export/cmd/BUILD.bazel b/src/vizier/services/adaptive_export/cmd/BUILD.bazel index 1ebaf3c27cd..a1bbeb52e1c 100644 --- a/src/vizier/services/adaptive_export/cmd/BUILD.bazel +++ b/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"], +) diff --git a/src/vizier/services/adaptive_export/cmd/leader_test.go b/src/vizier/services/adaptive_export/cmd/leader_test.go new file mode 100644 index 00000000000..7f64cc1d4ba --- /dev/null +++ b/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) + } +} diff --git a/src/vizier/services/adaptive_export/cmd/main.go b/src/vizier/services/adaptive_export/cmd/main.go index dffdf7dbfc8..359a3fe0408 100644 --- a/src/vizier/services/adaptive_export/cmd/main.go +++ b/src/vizier/services/adaptive_export/cmd/main.go @@ -51,6 +51,9 @@ import ( "time" log "github.com/sirupsen/logrus" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" "px.dev/pixie/src/api/go/pxapi" "px.dev/pixie/src/shared/services" @@ -82,10 +85,17 @@ const ( // envWindowBeforeSec / envWindowAfterSec / envTriggerPollMS / // envPruneIntervalSec are programmatic overrides per the spec. - envWindowBeforeSec = "ADAPTIVE_WINDOW_BEFORE_SEC" - envWindowAfterSec = "ADAPTIVE_WINDOW_AFTER_SEC" - envTriggerPollMS = "ADAPTIVE_TRIGGER_POLL_MS" - envPruneIntervalSec = "ADAPTIVE_PRUNE_INTERVAL_SEC" + envWindowBeforeSec = "ADAPTIVE_WINDOW_BEFORE_SEC" + envWindowAfterSec = "ADAPTIVE_WINDOW_AFTER_SEC" + // envQueryLagSec trails the per-table fan-out watermark this far behind + // wall-clock so sparse late-flushed rows (dns_events, dc_snoop) stay + // queryable. Default 30s (see controller.Config.QueryLag). + envQueryLagSec = "ADAPTIVE_QUERY_LAG_SEC" + // envExportAllFloorSec bounds how often the dx-steered full capture + // (OrderExportAll) re-runs for the same target. Default 30s. + envExportAllFloorSec = "ADAPTIVE_EXPORT_ALL_FLOOR_SEC" + envTriggerPollMS = "ADAPTIVE_TRIGGER_POLL_MS" + envPruneIntervalSec = "ADAPTIVE_PRUNE_INTERVAL_SEC" // envPushRefreshSec overrides controller.PushRefreshInterval. Unset β†’ // 30s default. A NEGATIVE value selects single-shot mode (one pull per @@ -120,6 +130,14 @@ const ( // users author scripts in the Pixie UI. envInstallPresets = "INSTALL_PRESET_SCRIPTS" + // envDeployTracepoints controls the AE's permanent bpftrace deploy + // (dc_snoop, creds_change). Decoupled from the retention firehose: + // the tracepoints stay upserted (876000h TTL) so they collect + // indefinitely, while INSTALL_PRESET_SCRIPTS governs only whether the + // cluster-wide cron *export* runs. Defaults to true (unset = deploy); + // set to "false" to skip the deploy. + envDeployTracepoints = "DEPLOY_TRACEPOINTS" + // === Throughput-protection knobs for the pushPixieRows fan-out. // All default to 0 (= legacy unbounded behavior preserved). envMaxParallelQueriesPerHash = "ADAPTIVE_MAX_PARALLEL_QUERIES_PER_HASH" @@ -210,6 +228,15 @@ func main() { } log.WithField("hostname", hostname).Info("operator pod is node-local") + // The AE runs as a DaemonSet (one pod per node), but the retention plugin, its + // cron scripts, and the bpftrace tracepoints are CLUSTER-scoped. If every pod + // registered them, each preset would get one duplicate cron script per node and + // every dark table would be exported N times (observed 2x β†’ 35% duplicate rows + // on a 2-node rig). Elect a single deterministic leader so exactly one pod does + // the cluster-scoped boot setup; the node-local trigger/data-plane still runs on + // every pod. + clusterSetupLeader := isClusterSetupLeader(ctx, hostname) + chEndpoint := chHTTPEndpoint(cfg.ClickHouse().Host(), os.Getenv(envCHHTTPEndpoint)) log.WithField("endpoint", chEndpoint).Info("clickhouse HTTP endpoint resolved") @@ -253,14 +280,31 @@ func main() { pluginClient = nil } if pluginClient != nil { - chDSN := cfg.ClickHouse().DSN() - exportURL, err := pluginClient.EnsureClickHousePluginEnabled(chDSN) + // The retention plugin's export sink is the query engine's native + // ClickHouseExportSink (clickhouse-cpp over TCP :9000), NOT the AE's own + // HTTP write path (:8123). It requires the native DSN format + // clickhouse://user:pass@host:9000/db; an HTTP DSN crashes the sink on + // connect and takes the vizier Unhealthy. See config.NativeDSN(). + chDSN := cfg.ClickHouse().NativeDSN() + // Boot-race: the vizier's plugin service can 404 GetClickHousePlugin for + // the first few seconds after the AE comes up. Retry a bounded number of + // times so a cold start doesn't permanently skip plugin enablement. + var exportURL string + var err error + for attempt := 1; attempt <= 5; attempt++ { + exportURL, err = pluginClient.EnsureClickHousePluginEnabled(chDSN) + if err == nil { + break + } + log.WithError(err).WithField("attempt", attempt).Warn("ensure ClickHouse plugin enabled failed β€” retrying (vizier plugin service may still be starting)") + time.Sleep(6 * time.Second) + } if err != nil { // non-fatal β€” the operator's own write path doesn't depend on // the plugin; analyst joins against pixie-table rows do, but a // missing plugin is a deployment misconfiguration the user // surfaces via UI. - log.WithError(err).Warn("could not ensure ClickHouse plugin is enabled β€” pixie tables will not be populated until you turn it on in the Pixie UI") + log.WithError(err).Warn("could not ensure ClickHouse plugin is enabled after retries β€” pixie tables will not be populated until you turn it on in the Pixie UI") } else { log.WithField("export_url", exportURL).Info("clickhouse retention plugin is enabled") } @@ -268,13 +312,24 @@ func main() { // 3b. (optional) install Pixie's preset retention scripts so the // pixie observation tables actually receive rows. Without this, // the plugin is enabled but does nothing. - if strings.EqualFold(os.Getenv(envInstallPresets), "true") { + if strings.EqualFold(os.Getenv(envInstallPresets), "true") && clusterSetupLeader { installed, err := installPresetScripts(pluginClient, cfg.Pixie().ClusterID(), cfg.Worker().ClusterName()) if err != nil { log.WithError(err).Warn("INSTALL_PRESET_SCRIPTS=true but install failed β€” pixie tables will stay empty") } else { log.WithField("installed", installed).Info("preset retention scripts installed on cluster") } + } else if clusterSetupLeader { + // Firehose off: purge any stale operator-managed cron scripts so the + // cluster-wide export stops. Evidence export is then driven per + // kubescape-alert by dx (OrderQuery β†’ /query) over the deployed + // tracepoints, deduped β€” not a continuous whole-cluster firehose. + purged, err := purgePresetScripts(pluginClient, cfg.Pixie().ClusterID(), cfg.Worker().ClusterName()) + if err != nil { + log.WithError(err).Warn("could not purge retention scripts β€” firehose may still be running") + } else if purged > 0 { + log.WithField("purged", purged).Info("retention firehose disabled β€” purged cluster-wide cron scripts (dx steers per-anomaly export)") + } } } @@ -292,7 +347,8 @@ func main() { wmStore, err := trigger.NewClickHouseWatermarkStore( chEndpoint, cfg.ClickHouse().Database(), cfg.ClickHouse().User(), cfg.ClickHouse().Password(), - httpTimeout) + httpTimeout, + ) if err != nil { log.WithError(err).Fatal("failed to create persistent watermark store") } @@ -355,10 +411,15 @@ func main() { }) ctlCfg := controller.Config{ - Hostname: hostname, - Rec: rec, - Before: durEnv(envWindowBeforeSec, 5*time.Minute, time.Second), - After: durEnv(envWindowAfterSec, 5*time.Minute, time.Second), + Hostname: hostname, + Rec: rec, + Before: durEnv(envWindowBeforeSec, 5*time.Minute, time.Second), + After: durEnv(envWindowAfterSec, 5*time.Minute, time.Second), + QueryLag: durEnv(envQueryLagSec, 30*time.Second, time.Second), + ExportAllFloor: durEnv(envExportAllFloorSec, 30*time.Second, time.Second), + // EXPORT_MODE=never β†’ the kubescape trigger stops self-steering; only a + // control client (dx) drives exports via /export/start + /query. + DisableSelfSteer: strings.EqualFold(strings.TrimSpace(os.Getenv("EXPORT_MODE")), "never"), MaxParallelQueriesPerHash: intEnvOrZero(envMaxParallelQueriesPerHash), MaxInflightQueriesGlobal: intEnvOrZero(envMaxInflightQueriesGlobal), EmptyResultSkipAfterN: intEnvOrZero(envEmptyResultSkipAfterN), @@ -409,15 +470,35 @@ func main() { // loop. All three need a live pxapi client; constructing once avoids // holding two parallel grpc streams for the same vizier. passthroughEnabled := strings.EqualFold(os.Getenv(envPassthrough), "true") + // The AE owns bpftrace deployment. Cron/retention export scripts cannot + // deploy a tracepoint (their pxtrace mutation is dropped by the cron + // executor), so the AE deploys the desired bpftraces itself via a mutation + // ExecuteScript over the pixie adapter. Decoupled from the retention + // firehose: default-on (unset = deploy) so the traces stay permanent even + // when INSTALL_PRESET_SCRIPTS is off and dx steers per-anomaly export. + deployTracepoints := !strings.EqualFold(os.Getenv(envDeployTracepoints), "false") && len(script.DesiredTracepoints()) > 0 && clusterSetupLeader var pixieAdapterInst *pixieapi.Adapter - if len(ctlCfg.PushPixieTables) > 0 || streamingMode || passthroughEnabled { + if len(ctlCfg.PushPixieTables) > 0 || streamingMode || passthroughEnabled || deployTracepoints { var adapter *pixieapi.Adapter - if direct := os.Getenv("ADAPTIVE_VIZIER_DIRECT_ADDR"); direct != "" { - // Direct mode β€” bypass the cloud's passthrough proxy and - // connect to the in-cluster vizier-query-broker. Use this - // on self-hosted clouds where pxapi.WithAPIKey isn't - // authorized for the cluster (e.g. a freshly-deployed - // vizier whose ID isn't yet linked to the API key's owner). + // DEFAULT to pem-direct: query THIS node's own vizier-pem at HOST_IP:50305 + // directly. It is node-local (matches the node-scoped AE), desync-immune + // (no kelvin/broker aggregation β€” the recurring "Agent ids not the same + // size" desync silently drops passthrough/broker queries) and fast. The + // default kicks in when the deploy provides HOST_IP (downward-API + // status.hostIP) + PL_JWT_SIGNING_KEY (the direct-query JWT); otherwise it + // falls back to cloud passthrough. An explicit ADAPTIVE_VIZIER_DIRECT_ADDR + // still wins (e.g. broker :50300 for a cluster-wide query). + direct := os.Getenv("ADAPTIVE_VIZIER_DIRECT_ADDR") + if direct == "" { + if hip := strings.TrimSpace(os.Getenv("HOST_IP")); hip != "" && os.Getenv("PL_JWT_SIGNING_KEY") != "" { + direct = hip + ":50305" + _ = os.Setenv("ADAPTIVE_VIZIER_DIRECT_ADDR", direct) // NewDirectFromEnv reads it + log.WithField("addr", direct).Info("pixieapi: defaulting to pem-direct (node-local PEM)") + } + } + if direct != "" { + // Direct mode β€” bypass the cloud passthrough proxy and connect to the + // in-cluster vizier-pem (pem-direct, default) or query-broker. a, err := pixieapi.NewDirectFromEnv(cfg.Pixie().ClusterID()) if err != nil { log.WithError(err).Fatal("ADAPTIVE_VIZIER_DIRECT_ADDR set but direct-mode adapter init failed") @@ -437,6 +518,31 @@ func main() { if len(ctlCfg.PushPixieTables) > 0 { ctl = ctl.WithPixieQuerier(&pixieAdapter{a: adapter}) } + if deployTracepoints { + // pem-direct (:50305) serves fast node-local QUERIES but refuses + // MUTATIONS ("direct-query: mutations out of scope #29"), so the + // bpftrace deploy (an UpsertTracepoint mutation) must go through a + // mutation-capable path. When the query adapter is pem-direct, deploy + // the tracepoints via the in-cluster broker (:50300) with the same + // service JWT, then keep querying via pem-direct. + mutAdapter := adapter + if strings.HasSuffix(direct, ":50305") && os.Getenv("PL_JWT_SIGNING_KEY") != "" { + brokerAddr := os.Getenv("ADAPTIVE_MUTATION_ADDR") + if brokerAddr == "" { + brokerAddr = "vizier-query-broker-svc.pl.svc.cluster.local:50300" + } + if m, err := pixieapi.NewDirect(cfg.Pixie().ClusterID(), pixieapi.DirectOptions{ + VizierAddr: brokerAddr, + SigningKey: os.Getenv("PL_JWT_SIGNING_KEY"), + }); err != nil { + log.WithError(err).Warn("could not build broker-direct mutation adapter β€” tracepoint deploy may fail on pem-direct") + } else { + mutAdapter = m + log.WithField("addr", brokerAddr).Info("tracepoint deploy via broker-direct (pem-direct can't mutate)") + } + } + deployDesiredTracepoints(ctx, mutAdapter) + } } // 5. Rehydrate active state across crashes. @@ -796,44 +902,139 @@ func (p *pixieAdapter) Query(ctx context.Context, src string) ([]map[string]any, return out, nil } -// installPresetScripts purges any stale ClickHouse-plugin retention -// scripts on the cluster, then installs the operator's built-in PxL -// scripts targeting the 13 socket_tracer tables we DDL'd. Cloud-side -// "presets" are deliberately ignored: in this fork the legacy -// "conn_stats export" / "dc snoop export" / "stack_traces export" -// preset names predate the rev-2 schema and would silently fail to -// write. conn_stats is now in the rev-2 schema, but it -// ships as "ch-conn_stats" (operator-managed naming) β€” the legacy -// "conn_stats export" preset name is still purged below so a stale -// one doesn't double-write. -func installPresetScripts(client *pixie.Client, clusterID, clusterName string) (int, error) { - current, err := client.GetClusterScripts(clusterID, clusterName) +// isClusterSetupLeader reports whether THIS AE pod should perform the +// CLUSTER-SCOPED boot setup β€” registering the retention cron scripts and +// deploying the bpftrace tracepoints. The AE is a DaemonSet (one pod per node), +// but those are cluster-wide: if every pod did them, each preset would get one +// duplicate cron script per node and every dark table would be exported once per +// node (observed 2x β†’ ~35% duplicate rows on a 2-node rig). We elect a single +// deterministic leader: the AE pod on the lexicographically-smallest node name. +// Every pod computes the same winner from the same pod list, so no coordination +// or lease is needed; pod-list RBAC is already held. Fail-open (return true) on +// any error β€” a transient duplicate is safer than skipping setup entirely. +func isClusterSetupLeader(ctx context.Context, myNode string) bool { + ns := os.Getenv("PL_NAMESPACE") + if ns == "" { + ns = "pl" + } + fallback := func(reason string, err error) bool { + log.WithError(err).Warnf("cluster-setup leader check (%s) β€” proceeding as leader", reason) + return true + } + cfg, err := rest.InClusterConfig() if err != nil { - return 0, fmt.Errorf("get cluster scripts: %w", err) + return fallback("in-cluster config", err) } - currentNames := make([]string, 0, len(current)) - for _, s := range current { - currentNames = append(currentNames, s.Name) + cs, err := kubernetes.NewForConfig(cfg) + if err != nil { + return fallback("clientset", err) } - log.WithFields(log.Fields{ - "already_on_cluster": len(current), - "cluster_script_names": currentNames, - }).Info("preset script install β€” purging managed + installing built-ins") + pods, err := cs.CoreV1().Pods(ns).List(ctx, metav1.ListOptions{LabelSelector: "name=adaptive-export"}) + if err != nil { + return fallback("pod list", err) + } + nodes := make([]string, 0, len(pods.Items)) + for _, p := range pods.Items { + nodes = append(nodes, p.Spec.NodeName) + } + minNode := leaderNode(nodes) + if minNode == "" { + return fallback("no scheduled AE pods found", nil) + } + leader := myNode == minNode + log.WithFields(log.Fields{"my_node": myNode, "leader_node": minNode, "is_leader": leader}). + Info("cluster-setup leader election β€” only the leader registers retention scripts + deploys tracepoints") + return leader +} + +// leaderNode picks the deterministic cluster-setup leader: the lexicographically +// smallest non-empty node name. Every AE pod computes the same winner from the +// same DaemonSet pod list, so no lease/coordination is needed. Empty input (or +// all-empty node names) yields "" β€” the caller then fails open. +func leaderNode(nodes []string) string { + smallest := "" + for _, n := range nodes { + if n != "" && (smallest == "" || n < smallest) { + smallest = n + } + } + return smallest +} - // Purge ONLY scripts we recognise as operator-managed or as legacy - // presets we know are broken in the rev-2 schema. User-authored - // retention scripts are left alone. +// deployDesiredTracepoints deploys the AE-owned bpftraces (script.DesiredTracepoints) +// via a mutation ExecuteScript over the pixie adapter. The retention/cron export +// path cannot deploy a tracepoint β€” the cron executor drops the pxtrace mutation, +// so the dark-vector output tables (dc_snoop, creds_change, …) never get created +// that way. The AE therefore owns tracepoint deployment here. UpsertTracepoint is +// idempotent (create-if-absent / no-op), so re-running on every boot is safe; each +// deploy is retried because deployment can transiently fail while PEMs (re)register. +func deployDesiredTracepoints(ctx context.Context, adapter *pixieapi.Adapter) { + for _, tp := range script.DesiredTracepoints() { + // Fire the deploy mutation. pxapi's result collector cannot decode the + // mutation-info response the vizier returns for a pxtrace deploy + // ("stream: unimplemented type"), so a Query error here is NOT a + // deployment failure β€” the UpsertTracepoint applies server-side + // regardless. Success is confirmed below by the tracepoint's output + // table becoming queryable (PENDING_STATE -> RUNNING_STATE). + if _, err := adapter.Query(ctx, tp.Script); err != nil { + log.WithError(err).WithField("tracepoint", tp.Name). + Debug("deploy mutation returned a stream error (expected for pxtrace mutations) β€” confirming via table") + } + // Confirm the tracepoint reached RUNNING by polling its output table. A + // plain DataFrame query on a not-yet-deployed table fails PxL compilation + // ("Table '' not found"); once the tracepoint is RUNNING the query + // compiles and returns (0 rows is fine β€” RUNNING, just no captures yet). + // Re-fire the deploy every few attempts in case the first didn't take. + verify := "import px\npx.display(px.DataFrame(table='" + tp.Table + "', start_time='-5s').head(1))\n" + running := false + for attempt := 1; attempt <= 12; attempt++ { + if _, err := adapter.Query(ctx, verify); err == nil { + running = true + break + } + if attempt%4 == 0 { + _, _ = adapter.Query(ctx, tp.Script) + } + time.Sleep(5 * time.Second) + } + if running { + log.WithFields(log.Fields{"tracepoint": tp.Name, "table": tp.Table}). + Info("bpftrace tracepoint deployed + RUNNING (permanent, idempotent upsert)") + } else { + log.WithField("tracepoint", tp.Name). + Warn("bpftrace tracepoint not confirmed RUNNING after deploy β€” its dark table stays empty until it deploys") + } + } +} + +// purgePresetScripts deletes the operator-managed (ch-*) + legacy retention +// cron scripts from the cluster, leaving user-authored scripts alone. Used both +// as the first step of installPresetScripts (reconcile) and standalone when +// INSTALL_PRESET_SCRIPTS is off, to stop the cluster-wide export firehose so +// that dx steers per-anomaly export instead. Returns the number purged. +func purgePresetScripts(client *pixie.Client, clusterID, clusterName string) (int, error) { + current, err := client.GetClusterScripts(clusterID, clusterName) + if err != nil { + return 0, fmt.Errorf("get cluster scripts: %w", err) + } + purged := 0 for _, s := range current { if !isOperatorManagedScript(s.Name) { - log.WithField("script", s.Name). - Debug("preset install β€” leaving user-authored script alone") continue } if err := client.DeleteDataRetentionScript(s.ScriptID); err != nil { log.WithError(err).WithField("script", s.Name).Warn("failed to delete stale script") continue } - log.WithField("script", s.Name).Info("purged stale retention script") + purged++ + log.WithField("script", s.Name).Info("purged retention script") + } + return purged, nil +} + +func installPresetScripts(client *pixie.Client, clusterID, clusterName string) (int, error) { + if _, err := purgePresetScripts(client, clusterID, clusterName); err != nil { + return 0, err } // Install built-ins. @@ -907,7 +1108,18 @@ func builtinPresetScripts() []*script.ScriptDefinition { "df = px.DataFrame(table='" + t + "', start_time='-15s')\n" + "df.namespace = px.upid_to_namespace(df.upid)\n" + "df.pod = px.upid_to_pod_name(df.upid)\n" + - "px.display(df, '" + t + "')\n" + // Provide event_time (nanoseconds) so the ClickHouse export sink emits + // it as DateTime64(9) via its normal type map, instead of auto-appending + // a DateTime64(3) millisecond column that mismatches the table's + // DateTime64(9) event_time and crashes the sink. + "df.event_time = df.time_\n" + + // Export via the native OTel ClickHouse sink (like DarkVectorPresets), + // NOT px.display. px.display relies on the retention plugin routing the + // display output to ClickHouse, which does not write on this stack + // (verified on a clean rig: dns_events/conn_stats/http_events = 0 parts + // ever, while the px.export dark tables dc_snoop/stack_trace populate). + // px.export writes directly through the same sink the dark presets use. + "px.export(df, px.otel.ClickHouseRows(table='" + t + "'))\n" out = append(out, &script.ScriptDefinition{ Name: "ch-" + t, Description: "adaptive_export builtin preset for " + t, @@ -916,5 +1128,9 @@ func builtinPresetScripts() []*script.ScriptDefinition { IsPreset: false, }) } - return out + // Dark-vector + profiler tracepoint/profiler export scripts (dc_snoop, + // stack_trace, creds_change) β€” permanent tracepoints (876000h) + OTelβ†’ + // ClickHouse export; registered if-not-present at boot alongside the + // protocol presets. + return append(out, script.DarkVectorPresets()...) } diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/BUILD.bazel b/src/vizier/services/adaptive_export/internal/clickhouse/BUILD.bazel index b83bc98cad7..4826adf14e4 100644 --- a/src/vizier/services/adaptive_export/internal/clickhouse/BUILD.bazel +++ b/src/vizier/services/adaptive_export/internal/clickhouse/BUILD.bazel @@ -39,6 +39,8 @@ pl_go_test( "columns_test.go", "ddl_test.go", "insert_test.go", + "metadata_invariants_test.go", + "timestamp_invariants_test.go", ], embed = [":clickhouse"], ) diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/apply.go b/src/vizier/services/adaptive_export/internal/clickhouse/apply.go index 0115795c79a..411b79f443d 100644 --- a/src/vizier/services/adaptive_export/internal/clickhouse/apply.go +++ b/src/vizier/services/adaptive_export/internal/clickhouse/apply.go @@ -56,6 +56,15 @@ var OperatorOwnedTables = []string{ // with "conn_stats schema drift, missing columns". Locked down by // TestOperatorOwnedTables_CoversAllPixieTables in apply_test.go. "conn_stats", + "dc_snoop", + "creds_change", + "stack_trace", + "dx_vfs_events", + "dx_unlink", + "dx_dlookup", + "dx_mprotect", + "dx_bpf", + "dx_ptrace", // operator's write targets. "adaptive_attribution", "trigger_watermark", diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go b/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go index c9513c75b88..0eb315882e9 100644 --- a/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go +++ b/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go @@ -57,6 +57,15 @@ var KnownTables = []string{ // conn_stats β€” re-added to rev-2 schema; counts per // (remote_addr, remote_port, protocol) on each retention-script pull. "conn_stats", + "dc_snoop", + "creds_change", + "stack_trace", + "dx_vfs_events", + "dx_unlink", + "dx_dlookup", + "dx_mprotect", + "dx_bpf", + "dx_ptrace", // operator-owned attribution table "adaptive_attribution", // operator-owned persistent trigger cursor @@ -128,6 +137,15 @@ func PixieTables() []string { "mux_events", "tls_events", "conn_stats", + "dc_snoop", + "creds_change", + "stack_trace", + "dx_vfs_events", + "dx_unlink", + "dx_dlookup", + "dx_mprotect", + "dx_bpf", + "dx_ptrace", } } diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/metadata_invariants_test.go b/src/vizier/services/adaptive_export/internal/clickhouse/metadata_invariants_test.go new file mode 100644 index 00000000000..33ba3d5f106 --- /dev/null +++ b/src/vizier/services/adaptive_export/internal/clickhouse/metadata_invariants_test.go @@ -0,0 +1,95 @@ +// 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 clickhouse + +import "testing" + +// darkVectorTables is the set of bpftrace + native-profiler tables. Every entry +// is a dark-vector observation table whose rows are attributed to a workload via +// the process_stats pid-merge (or, for stack_trace, via upid). Kept in lockstep +// with pxl.darkVectorTables (+ stack_trace, the profiler) β€” TestDarkVectorSetMatchesPixieTables +// guards that this list stays a subset of the operator-owned pixie tables. +var darkVectorTables = []string{ + "dc_snoop", "creds_change", "stack_trace", + "dx_vfs_events", "dx_unlink", "dx_dlookup", "dx_mprotect", "dx_bpf", "dx_ptrace", +} + +// TestDarkVectorTablesHaveFullMetadata enforces attribution CONSISTENCY: every +// dark-vector / profiler table carries the identical full k8s metadata set +// (namespace, pod, container, hostname), so any dark-vector event is attributable +// to its workload uniformly. A table missing one of these is an inconsistency +// that dx projection + forensic joins would silently drop β€” this was the concrete +// gap in dx_vfs_events/dx_unlink (no comm) and all six dx_ tables (no container) +// before they were reconciled to the canonical dc_snoop/creds_change shape. +func TestDarkVectorTablesHaveFullMetadata(t *testing.T) { + required := []string{"namespace", "pod", "container", "hostname"} + for _, tbl := range darkVectorTables { + cols, err := Columns(tbl) + if err != nil { + t.Fatalf("Columns(%q): %v", tbl, err) + } + have := make(map[string]bool, len(cols)) + for _, c := range cols { + have[c] = true + } + for _, req := range required { + if !have[req] { + t.Errorf("dark-vector table %s is missing metadata column %q (cols=%v) β€” every dark table must carry namespace/pod/container/hostname for uniform workload attribution", tbl, req, cols) + } + } + } +} + +// TestDarkVectorTablesCarryProcessIdentity β€” a dark-vector row must name the +// process it came from: comm (tracepoint tables) or upid (the profiler). Without +// it the pid/comm-filterable evidence base has nothing to filter on. +func TestDarkVectorTablesCarryProcessIdentity(t *testing.T) { + for _, tbl := range darkVectorTables { + cols, err := Columns(tbl) + if err != nil { + t.Fatalf("Columns(%q): %v", tbl, err) + } + var comm, upid bool + for _, c := range cols { + switch c { + case "comm": + comm = true + case "upid": + upid = true + } + } + if !comm && !upid { + t.Errorf("dark-vector table %s carries no process identity (need comm or upid) cols=%v", tbl, cols) + } + } +} + +// TestDarkVectorSetMatchesPixieTables keeps darkVectorTables honest: every dark +// table is an operator-owned pixie observation table (so the metadata + nanosecond +// guards above actually cover it). Catches a dark table being renamed/removed in +// PixieTables() without updating this set. +func TestDarkVectorSetMatchesPixieTables(t *testing.T) { + pixie := make(map[string]bool) + for _, p := range PixieTables() { + pixie[p] = true + } + for _, d := range darkVectorTables { + if !pixie[d] { + t.Errorf("dark-vector table %q is not in PixieTables() β€” it would escape the nanosecond + metadata guards", d) + } + } +} diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql b/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql index 61a582d8335..6fa8ed7f002 100644 --- a/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql +++ b/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql @@ -108,9 +108,9 @@ CREATE TABLE IF NOT EXISTS forensic_db.http_events ( latency Int64, hostname String, event_time DateTime64(9, 'UTC') DEFAULT toDateTime64(time_, 9) -) ENGINE = MergeTree() +) ENGINE = ReplacingMergeTree() PARTITION BY toYYYYMM(event_time) - ORDER BY (hostname, event_time); + ORDER BY (hostname, event_time, time_, upid, trace_role, remote_port, local_port, latency, req_method, req_path); -- http2_messages.beta β€” http2_messages_table.h CREATE TABLE IF NOT EXISTS forensic_db.`http2_messages.beta` ( @@ -153,9 +153,9 @@ CREATE TABLE IF NOT EXISTS forensic_db.dns_events ( latency Int64, hostname String, event_time DateTime64(9, 'UTC') DEFAULT toDateTime64(time_, 9) -) ENGINE = MergeTree() +) ENGINE = ReplacingMergeTree() PARTITION BY toYYYYMM(event_time) - ORDER BY (hostname, event_time); + ORDER BY (hostname, event_time, time_, upid, trace_role, remote_port, local_port, latency, req_body); -- redis_events β€” redis_table.h CREATE TABLE IF NOT EXISTS forensic_db.redis_events ( @@ -175,9 +175,9 @@ CREATE TABLE IF NOT EXISTS forensic_db.redis_events ( latency Int64, hostname String, event_time DateTime64(9, 'UTC') DEFAULT toDateTime64(time_, 9) -) ENGINE = MergeTree() +) ENGINE = ReplacingMergeTree() PARTITION BY toYYYYMM(event_time) - ORDER BY (hostname, event_time); + ORDER BY (hostname, event_time, time_, upid, trace_role, remote_port, local_port, latency, req_cmd); -- mysql_events β€” mysql_table.h CREATE TABLE IF NOT EXISTS forensic_db.mysql_events ( @@ -384,9 +384,9 @@ CREATE TABLE IF NOT EXISTS forensic_db.conn_stats ( bytes_recv Int64, hostname String, event_time DateTime64(9, 'UTC') DEFAULT toDateTime64(time_, 9) -) ENGINE = MergeTree() +) ENGINE = ReplacingMergeTree() PARTITION BY toYYYYMM(event_time) - ORDER BY (hostname, event_time); + ORDER BY (hostname, event_time, time_, upid, remote_addr, remote_port, trace_role); -- ============================================================================ -- adaptive_attribution β€” operator's only write target in ClickHouse. @@ -566,3 +566,127 @@ CREATE TABLE IF NOT EXISTS forensic_db.dx_evidence_manifest ( PARTITION BY toYYYYMM(fromUnixTimestamp64Nano(event_time)) TTL toDateTime(fromUnixTimestamp64Nano(event_time)) + INTERVAL 30 DAY DELETE SETTINGS index_granularity = 8192; + +-- ── dx dark-vector tracepoint tables (entlein/dx#126) ──────────────────────── +-- Fed by AE-owned bpftrace UpsertTracepoint probes (constantly enabled, no TTL). +-- Emit raw kernel pid+comm (NOT upid); namespace/pod enriched at pull time via a +-- process_stats join on pid. One column per line (schema-verify parser is line-oriented). +-- (dx_dcsnoop superseded by forensic_db.dc_snoop β€” canonical DateTime64(9)/Int64 +-- schema with full k8s metadata; see the dark-vector section above.) +CREATE TABLE IF NOT EXISTS forensic_db.dx_vfs_events ( + time_ DateTime64(9, 'UTC'), + pid Int64, + comm String, + op String, + file String, + namespace String, + pod String, + container String, + hostname String, + event_time DateTime64(9, 'UTC') +) ENGINE = MergeTree ORDER BY (event_time, pod); + +CREATE TABLE IF NOT EXISTS forensic_db.dx_unlink ( + time_ DateTime64(9, 'UTC'), + pid Int64, + comm String, + op String, + file String, + namespace String, + pod String, + container String, + hostname String, + event_time DateTime64(9, 'UTC') +) ENGINE = MergeTree ORDER BY (event_time, pod); + +CREATE TABLE IF NOT EXISTS forensic_db.dx_dlookup ( + time_ DateTime64(9, 'UTC'), + pid Int64, + comm String, + file String, + namespace String, + pod String, + container String, + hostname String, + event_time DateTime64(9, 'UTC') +) ENGINE = MergeTree ORDER BY (event_time, pod); + +CREATE TABLE IF NOT EXISTS forensic_db.dx_mprotect ( + time_ DateTime64(9, 'UTC'), + pid Int64, + comm String, + prot UInt64, + namespace String, + pod String, + container String, + hostname String, + event_time DateTime64(9, 'UTC') +) ENGINE = MergeTree ORDER BY (event_time, pod); + +-- (dx_creds superseded by forensic_db.creds_change β€” canonical schema with +-- old_uid/new_uid + full k8s metadata.) +CREATE TABLE IF NOT EXISTS forensic_db.dx_bpf ( + time_ DateTime64(9, 'UTC'), + pid Int64, + comm String, + namespace String, + pod String, + container String, + hostname String, + event_time DateTime64(9, 'UTC') +) ENGINE = MergeTree ORDER BY (event_time, pod); + +CREATE TABLE IF NOT EXISTS forensic_db.dx_ptrace ( + time_ DateTime64(9, 'UTC'), + pid Int64, + comm String, + namespace String, + pod String, + container String, + hostname String, + event_time DateTime64(9, 'UTC') +) ENGINE = MergeTree ORDER BY (event_time, pod); + +-- dc_snoop (dentry cache, V1/V2 process+file) β€” exported via the OTel/ClickHouse +-- retention plugin (px.export). pid-keyed; t = R (reference) / M (miss). +-- One column per line (schema-verify parser is line-oriented). +CREATE TABLE IF NOT EXISTS forensic_db.dc_snoop ( + time_ DateTime64(9, 'UTC'), + pid Int64, + comm String, + t String, + file String, + namespace String, + pod String, + container String, + hostname String, + event_time DateTime64(9, 'UTC') +) ENGINE = ReplacingMergeTree ORDER BY (time_, pid, comm, t, file, pod); + +-- stack_trace (native continuous profiler stack_traces.beta, V9) β€” OTel export. +CREATE TABLE IF NOT EXISTS forensic_db.stack_trace ( + time_ DateTime64(9, 'UTC'), + upid String, + namespace String, + pod String, + container String, + hostname String, + stack_trace_id Int64, + stack_trace String, + count Int64, + event_time DateTime64(9, 'UTC') +) ENGINE = ReplacingMergeTree ORDER BY (time_, upid, stack_trace_id, pod); + +-- creds_change (commit_creds privilege-escalation to root, V7) β€” OTel export. +CREATE TABLE IF NOT EXISTS forensic_db.creds_change ( + time_ DateTime64(9, 'UTC'), + pid Int64, + comm String, + old_uid Int64, + new_uid Int64, + namespace String, + pod String, + container String, + hostname String, + event_time DateTime64(9, 'UTC') +) ENGINE = ReplacingMergeTree ORDER BY (time_, pid, comm, old_uid, new_uid, pod); diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/timestamp_invariants_test.go b/src/vizier/services/adaptive_export/internal/clickhouse/timestamp_invariants_test.go new file mode 100644 index 00000000000..1491b8f7b39 --- /dev/null +++ b/src/vizier/services/adaptive_export/internal/clickhouse/timestamp_invariants_test.go @@ -0,0 +1,72 @@ +// 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 clickhouse + +import ( + "regexp" + "testing" +) + +// dt64Millis matches a DateTime64(3) COLUMN TYPE (millisecond scale). It does NOT +// match the toDateTime64(x, 3) function form (that reads "DateTime64(time_…"), so +// only real millisecond column declarations trip it. +var dt64Millis = regexp.MustCompile(`DateTime64\(3[,)]`) + +// dt64Nanos matches a DateTime64(9) column type (nanosecond scale). +var dt64Nanos = regexp.MustCompile(`DateTime64\(9[,)]`) + +// TestPixieTablesUseNanosecondTimestamps is the ONE-unit guardrail (contracts +// C1 + C16): every pixie observation table stores its timestamps as +// DateTime64(9) β€” nanoseconds β€” and NEVER DateTime64(3) milliseconds. A single +// millisecond column silently corrupts the invariant shared by three writers of +// these tables: the AE HTTP write path, the retention-plugin native ClickHouse +// export sink (which emits event_time = time_ as DateTime64(9) precisely to avoid +// its own millisecond auto-append), and dx/soc joins that read event_time as +// nanoseconds. kubescape_logs (unix-ns UInt64 INPUT) and alerts (kubescape's own +// millisecond alert stream) are not pixie observation tables and are excluded +// from PixieTables() by construction β€” this test locks that boundary. +func TestPixieTablesUseNanosecondTimestamps(t *testing.T) { + for _, tbl := range PixieTables() { + ddl, err := DDL(tbl) + if err != nil { + t.Fatalf("DDL(%q): %v", tbl, err) + } + if dt64Millis.MatchString(ddl) { + t.Errorf("%s declares a DateTime64(3) millisecond column β€” pixie observation tables MUST use DateTime64(9) nanoseconds (contracts C1/C16). No millisecond timestamps.", tbl) + } + if !dt64Nanos.MatchString(ddl) { + t.Errorf("%s has no DateTime64(9) timestamp column β€” expected nanosecond time_ and event_time.", tbl) + } + } +} + +// TestNoMillisecondTimestampReintroducedAnywhere is a coarser backstop: outside +// the two documented millisecond consumers (kubescape alerts + the kubescape_logs +// magnitude-normalizer), no forensic_db table the operator owns as a pixie +// observation table may carry a DateTime64(3) column. Guards against a future +// table being added with the wrong scale. +func TestNoMillisecondTimestampReintroducedAnywhere(t *testing.T) { + for _, tbl := range PixieTables() { + ddl, err := DDL(tbl) + if err != nil { + t.Fatalf("DDL(%q): %v", tbl, err) + } + if dt64Millis.MatchString(ddl) { + t.Fatalf("pixie table %s reintroduced a millisecond timestamp β€” the one-unit (nanosecond) invariant is broken", tbl) + } + } +} diff --git a/src/vizier/services/adaptive_export/internal/config/config.go b/src/vizier/services/adaptive_export/internal/config/config.go index 7c518513d9a..86e60250e34 100644 --- a/src/vizier/services/adaptive_export/internal/config/config.go +++ b/src/vizier/services/adaptive_export/internal/config/config.go @@ -163,11 +163,15 @@ func setUpConfig() error { log.SetLevel(log.InfoLevel) // Try to read configuration from environment variables first - clickhouseDSN := os.Getenv(envClickHouseDSN) - pixieClusterID := os.Getenv(envPixieClusterID) - pixieAPIKey := os.Getenv(envPixieAPIKey) - clusterName := os.Getenv(envClusterName) - pixieHost := getEnvWithDefault(envPixieEndpoint, defPixieHostname) + clickhouseDSN := strings.TrimSpace(os.Getenv(envClickHouseDSN)) + pixieClusterID := strings.TrimSpace(os.Getenv(envPixieClusterID)) + // TrimSpace: a secret sourced via `kubectl --from-file` keeps the file's + // trailing newline. In the pixie-api-key gRPC metadata header that newline is + // an HTTP/2 protocol violation β†’ the cloud PluginService replies RST_STREAM + // PROTOCOL_ERROR (looks like an auth failure but isn't). Trim it defensively. + pixieAPIKey := strings.TrimSpace(os.Getenv(envPixieAPIKey)) + clusterName := strings.TrimSpace(os.Getenv(envClusterName)) + pixieHost := strings.TrimSpace(getEnvWithDefault(envPixieEndpoint, defPixieHostname)) enableDebug := os.Getenv(envVerbose) if strings.EqualFold(enableDebug, boolTrue) { @@ -422,6 +426,7 @@ func (s *settings) BuildDate() string { type ClickHouse interface { DSN() string + NativeDSN() string Host() string Port() string User() string @@ -454,7 +459,17 @@ func (c *clickhouse) validate() error { return nil } -func (c *clickhouse) DSN() string { return c.dsn } +func (c *clickhouse) DSN() string { return c.dsn } + +// NativeDSN builds the ClickHouse retention-plugin export DSN in the format the +// query engine's ClickHouseExportSink requires: [clickhouse://]user:pass@host:port/db, +// where port is the NATIVE TCP port (9000), NOT the HTTP port (8123). The sink uses +// the clickhouse-cpp native client β€” an HTTP DSN (http:// scheme / :8123) makes it +// parse "http" as the username and crash on connect, taking the whole vizier +// Unhealthy. This is distinct from DSN() (the AE's own HTTP write endpoint). +func (c *clickhouse) NativeDSN() string { + return fmt.Sprintf("clickhouse://%s:%s@%s:%s/%s", c.user, c.password, c.host, c.port, c.database) +} func (c *clickhouse) Host() string { return c.host } func (c *clickhouse) Port() string { return c.port } func (c *clickhouse) User() string { return c.user } diff --git a/src/vizier/services/adaptive_export/internal/control/server.go b/src/vizier/services/adaptive_export/internal/control/server.go index 20ae154de20..96292715fcb 100644 --- a/src/vizier/services/adaptive_export/internal/control/server.go +++ b/src/vizier/services/adaptive_export/internal/control/server.go @@ -53,6 +53,27 @@ type queryRunner interface { OrderQuery(target anomaly.Target, table string, start, end time.Time, queryID string) error } +// exportAller is the optional "steer-all" capability behind /export/start: given +// a target, capture the COMPLETE evidence set (every configured pixie table) for +// its pod. The controller implements it. Optional (type-asserted) so start/stop- +// only deployments and test mocks that only implement queryRunner still compile. +type exportAller interface { + OrderExportAll(target anomaly.Target, start, end time.Time) +} + +// controlExportLookback is how far back /export/start reaches when a client +// (dx) steers a full capture β€” it sends only t_end, so AE captures +// [t_end-lookback, t_end]. 600s mirrors dx's Β±300s referral window so the +// anomaly is comfortably inside the pulled slice. +const controlExportLookback = 600 * time.Second + +// The control API carries timestamps in the pipeline's ONE unit: unix +// NANOSECONDS β€” the same unit as forensic_db.*.event_time and dx's referral +// windows. Read them with time.Unix(0, ns). (This spot previously did +// time.Unix(ns, 0), reading nanos AS seconds β†’ a year-56-billion window that +// overlaps no data, so every dx-steered export β€” all dark tables included β€” +// silently returned zero rows.) + // graphWriter persists dx evidence-graph edges (newline-delimited JSON, // JSONEachRow) to forensic_db.dx_evidence_graph. nil β†’ /dx/evidence_graph 501s. type graphWriter interface { @@ -242,13 +263,13 @@ type targetReq struct { type startReq struct { targetReq - TEnd int64 `json:"t_end"` // unix seconds + TEnd int64 `json:"t_end"` // unix NANOSECONDS (pipeline-wide unit) } type queryReq struct { targetReq Table string `json:"table"` - Window [2]int64 `json:"window"` // [start,end] unix seconds + Window [2]int64 `json:"window"` // [start,end] unix NANOSECONDS (pipeline-wide unit) QueryID string `json:"query_id"` } @@ -289,7 +310,17 @@ func (s *Server) handleStart(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusBadRequest) return } - s.set.Upsert(req.key(), time.Unix(req.TEnd, 0)) + s.set.Upsert(req.key(), time.Unix(0, req.TEnd).UTC()) + // Steer-all: when a querier is wired (pull mode), a StartExport is dx telling + // AE "grab the complete evidence set for this pod." The activeSet.Upsert above + // only feeds streaming mode, so in pull mode drive the full-table capture here. + // Async: the fan-out is slow (one query per table); dx must get its 202 back + // immediately and not block on the export. + if ea, ok := s.runner.(exportAller); ok { + hi := time.Unix(0, req.TEnd).UTC() + lo := hi.Add(-controlExportLookback) + go ea.OrderExportAll(req.target(), lo, hi) + } w.WriteHeader(http.StatusAccepted) } @@ -323,7 +354,7 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { return } err := s.runner.OrderQuery(req.target(), req.Table, - time.Unix(req.Window[0], 0), time.Unix(req.Window[1], 0), req.QueryID) + time.Unix(0, req.Window[0]).UTC(), time.Unix(0, req.Window[1]).UTC(), req.QueryID) if err != nil { w.WriteHeader(http.StatusBadGateway) 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 dfb98722002..15b3122ad0e 100644 --- a/src/vizier/services/adaptive_export/internal/control/server_test.go +++ b/src/vizier/services/adaptive_export/internal/control/server_test.go @@ -54,6 +54,39 @@ func (f *fakeRunner) OrderQuery(t anomaly.Target, table string, start, end time. return f.err } +// fakeExportAller implements BOTH queryRunner and exportAller β€” the controller's +// real shape. It sends the OrderExportAll target on a channel so the test can +// assert /export/start drove the steer-all full-evidence capture. +type fakeExportAller struct { + fakeRunner + exported chan anomaly.Target +} + +func (f *fakeExportAller) OrderExportAll(t anomaly.Target, start, end time.Time) { + f.exported <- t +} + +// TestStartExportDrivesSteerAll pins the steer-all contract: a POST /export/start +// (what dx sends default-on per referral) triggers OrderExportAll for the pod β€” +// i.e. dx steers AE to grab the complete evidence set, no per-table decision. +func TestStartExportDrivesSteerAll(t *testing.T) { + rn := &fakeExportAller{exported: make(chan anomaly.Target, 1)} + srv := New(&fakeExporter{}, rn) + r := do(t, srv, http.MethodPost, "/export/start", + `{"namespace":"redis","pod":"redis-1","t_end":1785000000}`) + if r.StatusCode != http.StatusAccepted { + t.Fatalf("want 202, got %d", r.StatusCode) + } + select { + case got := <-rn.exported: + if got.Pod != "redis-1" || got.Namespace != "redis" { + t.Fatalf("steer-all target wrong: %+v", got) + } + case <-time.After(2 * time.Second): + t.Fatal("OrderExportAll not called by /export/start within 2s") + } +} + func do(t *testing.T, srv *Server, method, path, body string) *http.Response { t.Helper() req := httptest.NewRequest(method, path, strings.NewReader(body)) @@ -104,8 +137,9 @@ func TestControlAuth(t *testing.T) { func TestStartExportUpserts(t *testing.T) { ex := &fakeExporter{} srv := New(ex, nil) + // t_end is unix NANOSECONDS (the pipeline-wide unit) β€” 1717200600s expressed in ns. resp := do(t, srv, http.MethodPost, "/export/start", - `{"namespace":"svc-poc","pod":"chain-backend-abc","comm":"sh","t_end":1717200600}`) + `{"namespace":"svc-poc","pod":"chain-backend-abc","comm":"sh","t_end":1717200600000000000}`) if resp.StatusCode != http.StatusAccepted { t.Fatalf("status = %d, want 202", resp.StatusCode) } @@ -113,8 +147,8 @@ func TestStartExportUpserts(t *testing.T) { ex.upserts[0].Namespace != "svc-poc" { t.Fatalf("upsert = %+v, want one for svc-poc/chain-backend-abc", ex.upserts) } - if ex.lastEnd != time.Unix(1717200600, 0) { - t.Fatalf("tEnd = %v, want 1717200600", ex.lastEnd) + if !ex.lastEnd.Equal(time.Unix(0, 1717200600000000000)) { + t.Fatalf("tEnd = %v, want %v", ex.lastEnd, time.Unix(0, 1717200600000000000).UTC()) } } diff --git a/src/vizier/services/adaptive_export/internal/controller/controller.go b/src/vizier/services/adaptive_export/internal/controller/controller.go index 9045b843fec..5a4b44ec1c4 100644 --- a/src/vizier/services/adaptive_export/internal/controller/controller.go +++ b/src/vizier/services/adaptive_export/internal/controller/controller.go @@ -32,6 +32,7 @@ package controller import ( "context" "errors" + "fmt" "sync" "time" @@ -110,6 +111,30 @@ type Config struct { // to be unambiguous. PushRefreshInterval time.Duration + // QueryLag holds the per-table watermark this far behind wall-clock: + // each pass queries up to now-QueryLag, not now. Sparse tables (dns_events, + // dc_snoop) emit events that socket_tracer/stirling flush a few seconds + // after they occur; without a lag the watermark advances past an event's + // time_ before it is queryable, so it is skipped forever. Continuous tables + // (conn_stats) always have fresh post-watermark rows so they never notice β€” + // which is why sparse tables lost ALL rows while continuous ones exported + // fully. Defaulted to 30s in defaulted(); 0 keeps the legacy (lossy) behavior + // only if set negative is not used β€” env ADAPTIVE_QUERY_LAG_SEC overrides. + QueryLag time.Duration + + // DisableSelfSteer, when true, stops the kubescape trigger from spawning its own + // pushPixieRows fan-out β€” the AE then exports ONLY what a control client (dx) + // orders via /export/start (OrderExportAll) or /query (OrderQuery). Set by + // EXPORT_MODE=never in main.go. Inverted bool so the zero value preserves the + // legacy self-steering behavior (and existing tests that build Config directly). + DisableSelfSteer bool + + // ExportAllFloor bounds how often the control-surface steer-all (OrderExportAll) + // re-captures the SAME target. dx fires StartExport per referral (~1s floor), so + // without this a sustained attack floods the broker with redundant full-table + // captures over overlapping windows. Defaulted to 30s in defaulted(). + ExportAllFloor time.Duration + // === Throughput-protection knobs === // // At high anomaly rates (many concurrent active hashes), the default @@ -176,6 +201,12 @@ func (c *Config) defaulted() Config { if out.PushRefreshInterval == 0 { out.PushRefreshInterval = 30 * time.Second } + if out.QueryLag == 0 { + out.QueryLag = 30 * time.Second + } + if out.ExportAllFloor == 0 { + out.ExportAllFloor = 30 * time.Second + } return out } @@ -209,6 +240,9 @@ type Controller struct { emptyCacheMu sync.Mutex emptyStreak map[string]int // consecutive 0-row returns emptySkipUntil map[string]time.Time // skip this (ns,pod,table) until this time + + exportAllMu sync.Mutex + exportAllAt map[string]time.Time // per-target floor for OrderExportAll (steer-all) } // New wires a Controller. nil clock falls through to RealClock. @@ -232,6 +266,7 @@ func New(trig Trigger, snk Sink, cfg Config, clk Clock) *Controller { inFlight: map[anomaly.AnomalyHash]bool{}, emptyStreak: map[string]int{}, emptySkipUntil: map[string]time.Time{}, + exportAllAt: map[string]time.Time{}, } if defaulted.MaxInflightQueriesGlobal > 0 { c.globalSem = make(chan struct{}, defaulted.MaxInflightQueriesGlobal) @@ -306,6 +341,53 @@ func (c *Controller) OrderQuery(target anomaly.Target, table string, start, end return nil } +// OrderExportAll runs a one-shot OrderQuery for EVERY configured pixie table for +// the target/window β€” the control-surface "steer-all" path. A control client +// (dx) asks AE to capture the COMPLETE evidence set for an anomaly's pod, with NO +// per-table relevance decision: dx filters only to (namespace, pod), AE grabs +// everything that could be relevant. Tables run concurrently (each OrderQuery +// takes the globalSem itself, so MaxInflightQueriesGlobal still bounds broker +// load), best-effort β€” a per-table error is logged and skipped so one slow/empty +// table can't block the rest. The deterministic query_id makes overlapping +// anomalies on the same pod idempotent (same target+table+window β†’ same id). +func (c *Controller) OrderExportAll(target anomaly.Target, start, end time.Time) { + if c.querier == nil || len(c.cfg.PushPixieTables) == 0 { + return + } + // Per-target floor: dx sends StartExport on EVERY referral (its own floor is + // ~1s), so a sustained attack fires OrderExportAll many times per second for + // the SAME pod. Each call is a full 20-table capture over a rolling ~600s + // window that already overlaps the previous one, so re-running them just floods + // the broker (globalSem saturates, nothing completes). Collapse the burst: one + // full capture per target per ExportAllFloor β€” the rolling window still covers + // every event. + tk := target.Namespace + "/" + target.Pod + c.exportAllMu.Lock() + if last, ok := c.exportAllAt[tk]; ok && c.clock.Now().Sub(last) < c.cfg.ExportAllFloor { + c.exportAllMu.Unlock() + return + } + c.exportAllAt[tk] = c.clock.Now() + c.exportAllMu.Unlock() + log.WithFields(log.Fields{ + "pod": target.Pod, "namespace": target.Namespace, "tables": len(c.cfg.PushPixieTables), + }).Info("OrderExportAll: dx-steered full-evidence capture for anomaly pod") + var wg sync.WaitGroup + for _, table := range c.cfg.PushPixieTables { + wg.Add(1) + go func(table string) { + defer wg.Done() + qid := fmt.Sprintf("steerall:%s/%s:%s:%d-%d", + target.Namespace, target.Pod, table, start.Unix(), end.Unix()) + if err := c.OrderQuery(target, table, start, end, qid); err != nil { + log.WithError(err).WithFields(log.Fields{"table": table, "pod": target.Pod}). + Warn("OrderExportAll: table export failed (skipped)") + } + }(table) + } + wg.Wait() +} + // Rehydrate populates the in-memory active set from ClickHouse so a // restarted operator picks up where it left off. Idempotent. Call // once at boot before Run. @@ -323,7 +405,7 @@ func (c *Controller) Rehydrate(ctx context.Context) error { // 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] { + if !c.cfg.DisableSelfSteer && c.querier != nil && len(c.cfg.PushPixieTables) > 0 && !c.inFlight[row.AnomalyHash] { c.inFlight[row.AnomalyHash] = true resume = append(resume, row) } @@ -416,7 +498,7 @@ func (c *Controller) handle(ctx context.Context, ev kubescape.Event) { snapshot := *row // Decide AND mark inFlight under the same mutex acquisition so two // rapid events for the same hash can't both decide to spawn. - spawn := c.querier != nil && len(c.cfg.PushPixieTables) > 0 && !c.inFlight[hash] + spawn := !c.cfg.DisableSelfSteer && c.querier != nil && len(c.cfg.PushPixieTables) > 0 && !c.inFlight[hash] if spawn { c.inFlight[hash] = true } @@ -564,7 +646,11 @@ func (c *Controller) pushPixieRows(ctx context.Context, initial sink.Attribution continue } sliceStart := lastUpper[table] - sliceEnd := now + // Trail the watermark by QueryLag so sparse late-flushed rows are + // still queryable when this slice runs. QueryFor's `now` (for its + // relative start_time pad) stays real-now so the DataFrame window + // still covers the slice. + sliceEnd := now.Add(-c.cfg.QueryLag) if !sliceEnd.After(sliceStart) { continue // tiny / inverted slice β€” skip } diff --git a/src/vizier/services/adaptive_export/internal/passthrough/passthrough.go b/src/vizier/services/adaptive_export/internal/passthrough/passthrough.go index d556c131c1c..14a5418de8c 100644 --- a/src/vizier/services/adaptive_export/internal/passthrough/passthrough.go +++ b/src/vizier/services/adaptive_export/internal/passthrough/passthrough.go @@ -70,7 +70,14 @@ type sink interface { type Config struct { Window time.Duration Refresh time.Duration - Tables []string + // QueryTimeout bounds a single table's pixie query (entlein/dx#7). The + // firehose pull used to bound query+write by Refresh, which is far too tight + // for a heavy protocol: pgsql_events carries full SQL text and its + // socket_tracer parse is expensive, so the ExecuteScript deadline-exceeded and + // pgsql_events landed 0 rows in forensic_db. Decoupled from Refresh and + // defaulted generous (matches the OrderQuery path's 180s budget). + QueryTimeout time.Duration + Tables []string // Rec records per-pull read/wrote counts (ADAPTIVE_RECONCILE). nil β†’ // defaulted to reconcile.Nop{} in New (instrument off). Rec reconcile.Recorder @@ -106,6 +113,9 @@ func New(q querier, s sink, cfg Config) *Loop { if cfg.Refresh <= 0 { cfg.Refresh = 30 * time.Second } + if cfg.QueryTimeout <= 0 { + cfg.QueryTimeout = 150 * time.Second // #7: heavy pgsql pull needs headroom + } if len(cfg.Tables) == 0 { cfg.Tables = clickhouse.PixieTables() } @@ -259,7 +269,7 @@ func (l *Loop) pull(ctx context.Context, table, src string, sliceStart, sliceEnd // 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) + ctx, cancel := context.WithTimeout(ctx, l.cfg.QueryTimeout) defer cancel() rows, err := l.q.Query(ctx, src) if err != nil { diff --git a/src/vizier/services/adaptive_export/internal/pixie/BUILD.bazel b/src/vizier/services/adaptive_export/internal/pixie/BUILD.bazel index 29f239170a0..eca74633f49 100644 --- a/src/vizier/services/adaptive_export/internal/pixie/BUILD.bazel +++ b/src/vizier/services/adaptive_export/internal/pixie/BUILD.bazel @@ -15,6 +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_test") go_library( name = "pixie", @@ -32,3 +33,12 @@ go_library( "@org_golang_google_grpc//metadata", ], ) + +pl_go_test( + name = "pixie_test", + srcs = ["auth_invariants_test.go"], + embed = [":pixie"], + deps = [ + "@org_golang_google_grpc//metadata", + ], +) diff --git a/src/vizier/services/adaptive_export/internal/pixie/auth_invariants_test.go b/src/vizier/services/adaptive_export/internal/pixie/auth_invariants_test.go new file mode 100644 index 00000000000..966c1182e83 --- /dev/null +++ b/src/vizier/services/adaptive_export/internal/pixie/auth_invariants_test.go @@ -0,0 +1,137 @@ +// 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 pixie + +import ( + "context" + "os" + "path/filepath" + "runtime" + "strings" + "testing" + + "google.golang.org/grpc/metadata" +) + +// TestCloudClientAuthIsPixieAPIKeyHeader pins the ONE cloud-auth mechanism: the +// plugin client authenticates via the canonical "pixie-api-key" gRPC metadata +// header β€” what the pixie cloud api-server trusts for external clients β€” and +// never a bearer/JWT or a hand-rolled scheme. (Cluster service JWTs are rejected +// by the cloud api for this surface; see NewClient's doc comment.) This must not +// drift into a second mechanism. +func TestCloudClientAuthIsPixieAPIKeyHeader(t *testing.T) { + c, err := NewClient(context.Background(), "test-key-123", "cloud.example.org:443") + if err != nil { + t.Fatalf("NewClient: %v", err) + } + md, ok := metadata.FromOutgoingContext(c.ctx) + if !ok { + t.Fatal("NewClient attached no outgoing gRPC metadata β€” no auth header set") + } + if got := md.Get("pixie-api-key"); len(got) != 1 || got[0] != "test-key-123" { + t.Errorf(`cloud auth header pixie-api-key = %v, want ["test-key-123"]`, got) + } + // The cloud client must NOT use the in-cluster bearer/JWT header β€” that path + // is a different, non-interchangeable mechanism (jwtutils service JWT). + if got := md.Get("authorization"); len(got) != 0 { + t.Errorf("cloud client must not set an authorization/bearer header (that is the in-cluster JWT path), got %v", got) + } +} + +// TestCloudClientRejectsEmptyKey β€” no silent unauthenticated fallback: an empty +// key is a hard error, never a proceed-anyway. +func TestCloudClientRejectsEmptyKey(t *testing.T) { + if _, err := NewClient(context.Background(), "", "cloud.example.org:443"); err == nil { + t.Fatal("NewClient(empty key) must error, not proceed unauthenticated") + } +} + +// TestNoAuthReinvention walks the whole adaptive_export source tree and enforces +// "one auth method per context, no wheels reinvented": +// +// - Every JWT is minted/verified through the SHARED pixie lib +// px.dev/pixie/src/shared/services/utils (jwtutils) β€” never a hand-rolled +// golang-jwt/dgrijalva/jwt.New/SignedString/jwt.Parse. +// - The "pixie-api-key" cloud header lives ONLY in internal/pixie (the single +// cloud plugin client), never sprinkled across packages. +// +// This is the guardrail behind the recurring "use the API-based auth, don't +// reinvent it" rule: the AE has exactly two auth surfaces β€” cloud=pixie-api-key, +// in-cluster=jwtutils service JWT β€” and both go through canonical code. +func TestNoAuthReinvention(t *testing.T) { + root := aeSourceRoot(t) + + // Hand-rolled JWT crypto is banned β€” jwtutils is the only sanctioned path. + bannedJWT := []string{"golang-jwt", "dgrijalva/jwt", "jwt.New(", ".SignedString(", "jwt.Parse("} + + apiKeyHeaderIn := map[string]bool{} + err := filepath.WalkDir(root, func(path string, d os.DirEntry, err error) error { + if err != nil { + return err + } + if d.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") { + return nil + } + b, err := os.ReadFile(path) + if err != nil { + return err + } + src := string(b) + rel, _ := filepath.Rel(root, path) + for _, banned := range bannedJWT { + if strings.Contains(src, banned) { + t.Errorf("%s uses %q β€” JWTs must go through the shared jwtutils lib (GenerateJWTForService / SignJWTClaims / ParseToken), no hand-rolled crypto", rel, banned) + } + } + // Match the header string literal only (not doc-comment prose). + if strings.Contains(src, `"pixie-api-key"`) { + apiKeyHeaderIn[rel] = true + } + return nil + }) + if err != nil { + t.Fatalf("walk AE source: %v", err) + } + + for f := range apiKeyHeaderIn { + if filepath.Dir(f) != filepath.Join("internal", "pixie") { + t.Errorf(`the "pixie-api-key" cloud auth header appears in %s β€” it must be centralized in internal/pixie (one cloud auth surface)`, f) + } + } +} + +// aeSourceRoot returns the adaptive_export package root (the dir containing this +// test's package), resolved from the compiled test's own file path. +func aeSourceRoot(t *testing.T) string { + t.Helper() + _, file, _, ok := runtime.Caller(0) + if !ok { + t.Skip("runtime.Caller unavailable β€” cannot locate source tree") + } + // .../adaptive_export/internal/pixie/auth_invariants_test.go -> .../adaptive_export + root := filepath.Dir(filepath.Dir(filepath.Dir(file))) + // In a sandboxed build (bazel) the source tree is not laid out on disk at this + // path β€” the source-walk guard is a `go test` check; skip it there rather than + // fail. The behavioral auth tests above still run everywhere. + if filepath.Base(root) != "adaptive_export" { + t.Skipf("source tree not at %q (sandboxed build) β€” skipping source-walk guard", root) + } + if _, err := os.Stat(filepath.Join(root, "cmd", "main.go")); err != nil { + t.Skipf("AE source tree not readable at %q (sandboxed build) β€” skipping source-walk guard", root) + } + return root +} diff --git a/src/vizier/services/adaptive_export/internal/pxl/compile.go b/src/vizier/services/adaptive_export/internal/pxl/compile.go index de3d16d0aad..cdd21c5313c 100644 --- a/src/vizier/services/adaptive_export/internal/pxl/compile.go +++ b/src/vizier/services/adaptive_export/internal/pxl/compile.go @@ -61,12 +61,61 @@ func CompilePassthrough(table string, window time.Duration) (string, error) { b.WriteString("df = px.DataFrame(table='" + table + "', start_time='" + relStart + "')\n") b.WriteString("df = df[df.time_ >= px.int64_to_time(%d)]\n") b.WriteString("df = df[df.time_ < px.int64_to_time(%d)]\n") - b.WriteString("df.namespace = px.upid_to_namespace(df.upid)\n") - b.WriteString("df.pod = px.upid_to_pod_name(df.upid)\n") + b.WriteString(PodEnrichPxL(table)) b.WriteString("px.display(df, '" + table + "')\n") return b.String(), nil } +// darkVectorTables are the pid-keyed dx tracepoint tables (entlein/dx#126). They +// emit raw kernel pid + comm (no upid), so pod is resolved differently β€” see +// PodEnrichPxL. +var darkVectorTables = map[string]bool{ + "dc_snoop": true, "creds_change": true, + "dx_vfs_events": true, "dx_unlink": true, "dx_dlookup": true, + "dx_mprotect": true, "dx_bpf": true, "dx_ptrace": true, +} + +// IsDarkVector reports whether table is a pid-keyed dx tracepoint table. +func IsDarkVector(table string) bool { return darkVectorTables[table] } + +// darkProcStatsWindow bounds the process_stats scan used to resolve pod/namespace +// for the dark-vector pid-merge. Kept short on purpose: a wide window is the +// dominant cost of the dark query and starved / timed it out under load. Long- +// lived workload pids are sampled continuously so a 2-minute window resolves +// them; transient attack pids never enter process_stats regardless. +const darkProcStatsWindow = "-2m" + +// PodEnrichPxL returns the PxL that populates df.namespace + df.pod for a table. +// +// Native socket_tracer tables carry upid β†’ direct px.upid_to_* resolution (df.pod +// is the namespaced "/" key). The dx dark-vector tracepoint tables emit +// raw kernel pid with NO upid (px.upid_to_* / ctx['pod'] fail outright), so pod is +// resolved by merging process_stats on pid ONLY β€” the validated join (dx#126, +// join-pod.pxl). NOT pid+asid: on a dynamic tracepoint px.asid() is the +// aggregator/kelvin asid, not the per-PEM asid of the data, so pid+asid never +// matches. df.pod here is the BARE pod name (proc.ctx['pod']). Best-effort: blank +// for host/transient pids (correct β€” no pod). +func PodEnrichPxL(table string) string { + if darkVectorTables[table] { + // process_stats is the COST of the dark-table merge: a busy node samples + // every live pid every ~10-30s, so a wide window is a huge scan that + // competes with the fast native-table queries for the shared query-slot + // budget β€” the heavy pull that starved / timed out the dark capture. A + // short window still resolves the pods that matter: long-lived workload + // pids (redis-server) appear in every sample, so darkProcStatsWindow is + // plenty; genuinely transient attack pids (whoami/cat children) never land + // in process_stats at all and resolve blank either way (see comment above). + return "proc = px.DataFrame(table='process_stats', start_time='" + darkProcStatsWindow + "')\n" + + "proc.pod = proc.ctx['pod']\n" + + "proc.namespace = proc.ctx['namespace']\n" + + "proc.pid = px.upid_to_pid(proc.upid)\n" + + "proc = proc.groupby(['pod', 'namespace', 'pid']).agg()\n" + + "df = df.merge(proc, how='left', left_on=['pid'], right_on=['pid'], suffixes=['', '_x'])\n" + } + return "df.namespace = px.upid_to_namespace(df.upid)\n" + + "df.pod = px.upid_to_pod_name(df.upid)\n" +} + // Render fills a CompilePassthrough template with the precise [sliceStart, // sliceEnd) bounds for one tick. func Render(tmpl string, sliceStart, sliceEnd time.Time) string { diff --git a/src/vizier/services/adaptive_export/internal/pxl/compile_test.go b/src/vizier/services/adaptive_export/internal/pxl/compile_test.go index 724e12e827c..e5a16cc3e56 100644 --- a/src/vizier/services/adaptive_export/internal/pxl/compile_test.go +++ b/src/vizier/services/adaptive_export/internal/pxl/compile_test.go @@ -86,3 +86,25 @@ func TestCompilePassthrough_UnknownTable(t *testing.T) { t.Fatalf("err=%v want ErrUnknownTable", err) } } + +// TestPodEnrichPxL_DarkVsNative locks the #126 split: native tables resolve pod +// via upid; the pid-keyed dark-vector tracepoint tables use the validated +// process_stats pid-merge (pid ONLY, not pid+asid) and never reference df.upid. +func TestPodEnrichPxL_DarkVsNative(t *testing.T) { + native := PodEnrichPxL("http_events") + if !strings.Contains(native, "px.upid_to_pod_name(df.upid)") { + t.Errorf("native table must resolve pod via upid: %q", native) + } + for _, tbl := range []string{"dc_snoop", "creds_change", "dx_bpf", "dx_ptrace"} { + q := PodEnrichPxL(tbl) + if strings.Contains(q, "df.upid") { + t.Errorf("%s (pid-keyed) must NOT reference df.upid: %q", tbl, q) + } + if !strings.Contains(q, "left_on=['pid'], right_on=['pid']") { + t.Errorf("%s must merge process_stats on pid only: %q", tbl, q) + } + if !IsDarkVector(tbl) { + t.Errorf("%s should be IsDarkVector", tbl) + } + } +} diff --git a/src/vizier/services/adaptive_export/internal/pxl/queryfor.go b/src/vizier/services/adaptive_export/internal/pxl/queryfor.go index 96db2507f5f..4f9d8d6d37c 100644 --- a/src/vizier/services/adaptive_export/internal/pxl/queryfor.go +++ b/src/vizier/services/adaptive_export/internal/pxl/queryfor.go @@ -19,6 +19,7 @@ package pxl import ( "errors" "fmt" + "os" "regexp" "strconv" "strings" @@ -62,33 +63,132 @@ func QueryFor(table string, t anomaly.Target, sliceStart, sliceEnd, now time.Tim var b strings.Builder b.WriteString(pxSetMaxRows) b.WriteString("import px\n") - b.WriteString("df = px.DataFrame(table='" + table + "', start_time='" + relStart + "')\n") + b.WriteString("df = px.DataFrame(table='" + pixieSourceFor(table) + "', start_time='" + relStart + "')\n") b.WriteString("df = df[df.time_ >= px.int64_to_time(" + strconv.FormatInt(sliceStart.UnixNano(), 10) + ")]\n") b.WriteString("df = df[df.time_ < px.int64_to_time(" + strconv.FormatInt(sliceEnd.UnixNano(), 10) + ")]\n") - b.WriteString("df.namespace = px.upid_to_namespace(df.upid)\n") - // px.upid_to_pod_name returns "/" (carnot: + // Native tables: px.upid_to_pod_name returns "/" (carnot: // metadata_ops.h UPIDToPodNameUDF::Exec β†’ absl::Substitute("$0/$1", ns, name)), - // not the bare pod name. Filtering against bare t.Pod would always - // miss; build the namespaced key when we have both fields. - b.WriteString("df.pod = px.upid_to_pod_name(df.upid)\n") - if t.Namespace != "" { - b.WriteString("df = df[df.namespace == '" + escapePxL(t.Namespace) + "']\n") - } - if t.Pod != "" { + // not the bare pod name. Dark-vector tracepoint tables (pid-keyed) resolve pod + // via a process_stats pid-merge instead and yield a BARE pod name (dx#126). + if table == "stack_trace" { + // stack_trace is the CANONICAL native continuous profiler (stack_traces.beta, + // upid-keyed β€” NOT a pid tracepoint, so NOT a dark-vector pid-merge). Resolve + // pod/namespace/container/hostname exactly like the export preset + // (script/presets/stack_trace.pxl) and stamp event_time = time_ so the CH + // stack_trace row is complete. df.ctx['pod'] is the NAMESPACED "/" + // key (verified live), so the pod filter is namespaced β€” same as the native + // upid_to_pod_name path below. + b.WriteString("df.namespace = df.ctx['namespace']\n") + b.WriteString("df.pod = df.ctx['pod']\n") + b.WriteString("df.container = df.ctx['container']\n") + b.WriteString("df.hostname = px.upid_to_node_name(df.upid)\n") + b.WriteString("df.event_time = df.time_\n") if t.Namespace != "" { - // Both fields present β€” use exact equality on the namespaced key. - b.WriteString("df = df[df.pod == '" + escapePxL(t.Namespace+"/"+t.Pod) + "']\n") - } else { - // Pod-only fallback: df.pod is "/", so a bare-pod - // equality always misses. Regex-anchor "/" via - // px.regex_match so the defensive path stays functional. - b.WriteString("df = df[px.regex_match('^[^/]+/" + escapePxL(regexp.QuoteMeta(t.Pod)) + "$', df.pod)]\n") + b.WriteString("df = df[df.namespace == '" + escapePxL(t.Namespace) + "']\n") + } + if t.Pod != "" { + if t.Namespace != "" { + b.WriteString("df = df[df.pod == '" + escapePxL(t.Namespace+"/"+t.Pod) + "']\n") + } else { + b.WriteString("df = df[px.regex_match('^[^/]+/" + escapePxL(regexp.QuoteMeta(t.Pod)) + "$', df.pod)]\n") + } + } + } else if IsDarkVector(table) { + // Dark-vector tracepoints emit a RAW kernel pid. The malignant transient + // pids an incident actually produces β€” an attack's whoami/cat/getent + // children β€” are too short-lived to land in process_stats, so their + // pod/namespace resolves BLANK; a pod (or even namespace) filter drops + // exactly the evidence, which is why the dark tables came back empty. + // The AE is node-local (pem-direct β†’ the node's own PEM), so the query is + // already scoped to the alert's node. + // + // ORDER MATTERS: drop the infra/self comms FIRST (env-driven, no recompile), + // THEN do the process_stats pid-merge. The node's dark stream is huge + // (Formatter/vector/runc/... thousands of rows per window); merging every + // one against process_stats is the query that timed out and silently + // dropped dc_snoop. Filtering comm first shrinks the merge to the handful + // of workload rows (bash/redis/whoami/cat), so the dark capture completes. + b.WriteString(darkCommExclusion(table)) + b.WriteString(PodEnrichPxL(table)) + } else { + b.WriteString(PodEnrichPxL(table)) + if t.Namespace != "" { + b.WriteString("df = df[df.namespace == '" + escapePxL(t.Namespace) + "']\n") + } + if t.Pod != "" { + if t.Namespace != "" { + // upid_to_pod_name is "/" β€” exact equality on the namespaced key. + b.WriteString("df = df[df.pod == '" + escapePxL(t.Namespace+"/"+t.Pod) + "']\n") + } else { + // Pod-only fallback: df.pod is "/", so a bare-pod + // equality always misses. Regex-anchor "/". + b.WriteString("df = df[px.regex_match('^[^/]+/" + escapePxL(regexp.QuoteMeta(t.Pod)) + "$', df.pod)]\n") + } } } b.WriteString("px.display(df, '" + table + "')\n") return b.String(), nil } +// pixieSourceFor returns the Pixie table a builtin is sourced FROM when it +// differs from the ClickHouse table it is written TO. stack_trace is written to +// CH as 'stack_trace' but sourced from the CANONICAL native continuous profiler +// 'stack_traces.beta' β€” the always-on Pixie profiler, NOT an AE-invented table. +// (Dotted-name DataFrames compile fine in a direct query; verified live.) +func pixieSourceFor(table string) string { + if table == "stack_trace" { + return "stack_traces.beta" + } + return table +} + +// darkVectorHasComm lists the dark-vector tables that carry a `comm` column, so +// the infra-comm exclusion only emits for those (stack_trace is upid-only). +var darkVectorHasComm = map[string]bool{ + "dc_snoop": true, "creds_change": true, "dx_vfs_events": true, + "dx_unlink": true, "dx_dlookup": true, "dx_mprotect": true, + "dx_bpf": true, "dx_ptrace": true, +} + +// darkExcludeCommsDefault is the node's own infra/self comms dropped from the +// node-scoped dark capture so the workload's activity stands out. Overridable at +// runtime via DC_SNOOP_EXCLUDE_COMMS (csv) β€” a process can be added without a +// recompile. Kept in sync with script.presets defaultExcludeComms. +var darkExcludeCommsDefault = []string{ + "pem", "kelvin", "containerd", "containerd-shim", "runc", "node-agent", + "runc:[2:INIT]", "runc:[1:CHILD]", + "vizier-query-broker", "vizier-metadata", "nats-server", "k3s-server", + "k3s-agent", "systemd", "systemd-journal", "SystemLogFlush", "kubelet", + "AsyncInsertQ", "BgSchPool", "Collector", "AsyncMetrics", "MergeMutate", + "MergeTreeIndex", "CgrpMemUsgObsr", "coredns", "metadata", "storage", + "operator", "iptables", "iptables-save", "iptables-restor", "ip6tables", + "ConfigReloader", "clickhouse-oper", "Formatter", "(setup.sh)", "cmd", + "vector-worker", "metrics-server", "local-path-prov", "portmap", + "(udev-worker)", "systemd-resolve", "systemd-timesyn", +} + +// darkCommExclusion builds the infra-comm drop filter for a dark-vector table +// that has a comm column. Returns "" for comm-less tables (stack_trace). +func darkCommExclusion(table string) string { + if !darkVectorHasComm[table] { + return "" + } + comms := darkExcludeCommsDefault + if v := strings.TrimSpace(os.Getenv("DC_SNOOP_EXCLUDE_COMMS")); v != "" { + comms = nil + for _, s := range strings.Split(v, ",") { + if s = strings.TrimSpace(s); s != "" { + comms = append(comms, s) + } + } + } + var b strings.Builder + for _, c := range comms { + b.WriteString("df = df[df.comm != '" + escapePxL(c) + "']\n") + } + return b.String() +} + // pxlEscaper turns raw bytes that could break out of a PxL single-quoted // string into their Python-style escape sequences. The backslash MUST be // mapped FIRST so its own substitution doesn't get double-escaped when diff --git a/src/vizier/services/adaptive_export/internal/pxl/tables.go b/src/vizier/services/adaptive_export/internal/pxl/tables.go index c29284ad58a..d04f107fa2f 100644 --- a/src/vizier/services/adaptive_export/internal/pxl/tables.go +++ b/src/vizier/services/adaptive_export/internal/pxl/tables.go @@ -72,6 +72,21 @@ var builtinTables = []TableSpec{ {Name: "mux_events", Protocol: "Mux (Twitter Finagle)"}, {Name: "tls_events", Protocol: "TLS handshake"}, {Name: "conn_stats", Protocol: "Connection-level statistics"}, + // Dark-vector tracepoint + profiler tables the AE deploys + exports itself. + // dc_snoop (V1/V2), creds_change (V7), stack_trace (V9) are the active set β€” + // canonical schemas (DateTime64(9) time_/event_time, Int64 ids, full k8s + // metadata via the process_stats pid-merge), matching the export presets and + // dx#129's darkTables keys. The remaining dx_* entries are reserved for the + // bpftraces still to be written (V2 write/unlink, V6 mprotect, V8 bpf/ptrace). + {Name: "dc_snoop", Protocol: "tracepoint (dentry lookup, V1/V2)"}, + {Name: "creds_change", Protocol: "tracepoint (commit_creds priv-esc, V7)"}, + {Name: "stack_trace", Protocol: "profiler (stack_traces.beta, V9)"}, + {Name: "dx_vfs_events", Protocol: "tracepoint"}, + {Name: "dx_unlink", Protocol: "tracepoint"}, + {Name: "dx_dlookup", Protocol: "tracepoint"}, + {Name: "dx_mprotect", Protocol: "tracepoint"}, + {Name: "dx_bpf", Protocol: "tracepoint"}, + {Name: "dx_ptrace", Protocol: "tracepoint"}, } // Registry is the extension surface for users to register their own diff --git a/src/vizier/services/adaptive_export/internal/pxl/tables_test.go b/src/vizier/services/adaptive_export/internal/pxl/tables_test.go index 273c0f625ee..d0d8a38ed54 100644 --- a/src/vizier/services/adaptive_export/internal/pxl/tables_test.go +++ b/src/vizier/services/adaptive_export/internal/pxl/tables_test.go @@ -26,8 +26,12 @@ import ( // mysql_events, pgsql_events, cql_events, mongodb_events, // kafka_events.beta, amqp_events, mux_events, tls_events, conn_stats). // Update this guard if the spec adds / removes a table. +// 13 socket_tracer tables + 9 dark-vector tables: dc_snoop (V1/V2), creds_change +// (V7), stack_trace (V9) β€” canonical schemas, active β€” plus dx_vfs_events, +// dx_unlink, dx_dlookup, dx_mprotect, dx_bpf, dx_ptrace reserved for the bpftraces +// still to be written (entlein/dx#126). Update this guard if the spec changes. func TestBuiltinTables_Count(t *testing.T) { - const want = 13 + const want = 22 if got := len(builtinTables); got != want { t.Fatalf("builtinTables = %d entries, want %d", got, want) } diff --git a/src/vizier/services/adaptive_export/internal/script/BUILD.bazel b/src/vizier/services/adaptive_export/internal/script/BUILD.bazel index 28d764063a4..71ad44e2e9c 100644 --- a/src/vizier/services/adaptive_export/internal/script/BUILD.bazel +++ b/src/vizier/services/adaptive_export/internal/script/BUILD.bazel @@ -15,10 +15,27 @@ # 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_test") go_library( name = "script", - srcs = ["script.go"], + srcs = [ + "presets.go", + "script.go", + ], + embedsrcs = [ + "presets/creds_change.pxl", + "presets/creds_change_deploy.pxl", + "presets/dc_snoop.pxl", + "presets/dc_snoop_deploy.pxl", + "presets/stack_trace.pxl", + ], importpath = "px.dev/pixie/src/vizier/services/adaptive_export/internal/script", visibility = ["//src/vizier/services/adaptive_export:__subpackages__"], ) + +pl_go_test( + name = "script_test", + srcs = ["presets_test.go"], + embed = [":script"], +) diff --git a/src/vizier/services/adaptive_export/internal/script/presets.go b/src/vizier/services/adaptive_export/internal/script/presets.go new file mode 100644 index 00000000000..212f869b50b --- /dev/null +++ b/src/vizier/services/adaptive_export/internal/script/presets.go @@ -0,0 +1,107 @@ +// Copyright 2018- The Pixie Authors. +// SPDX-License-Identifier: Apache-2.0 + +package script + +import ( + _ "embed" + "fmt" + "os" + "strings" +) + +var defaultExcludeNamespaces = []string{ + "pl", "honey", "px-operator", "olm", "clickhouse", + "kube-system", "kube-public", "kube-node-lease", "local-path-storage", +} + +var defaultExcludeComms = []string{ + "k3s-server", "k3s-agent", "containerd", "containerd-shim", + "runc", "runc:[2:INIT]", "runc:[1:CHILD]", "node-agent", "kelvin", + "vizier-pem", "vizier-query-broker", "vizier-metadata", + "systemd", "systemd-journal", "iptables", "ip6tables", "kubelet", + "operator", "storage", +} + +func csvEnv(key string, def []string) []string { + v := os.Getenv(key) + if v == "" { + return def + } + out := []string{} + for _, s := range strings.Split(v, ",") { + if s = strings.TrimSpace(s); s != "" { + out = append(out, s) + } + } + return out +} + +// dcSnoopExclusion builds the dc_snoop noise filter (namespace + comm drops) from +// DC_SNOOP_EXCLUDE_NAMESPACES / DC_SNOOP_EXCLUDE_COMMS, substituted into +// dc_snoop.pxl at # __DC_SNOOP_EXCLUSION__ so a process can be added without a +// recompile. Kept in sync with dx benchlive.writeSelfExclusion. +func dcSnoopExclusion() string { + var b strings.Builder + for _, ns := range csvEnv("DC_SNOOP_EXCLUDE_NAMESPACES", defaultExcludeNamespaces) { + fmt.Fprintf(&b, "df = df[df.namespace != '%s']\n", ns) + } + for _, c := range csvEnv("DC_SNOOP_EXCLUDE_COMMS", defaultExcludeComms) { + fmt.Fprintf(&b, "df = df[df.comm != '%s']\n", c) + } + return strings.TrimRight(b.String(), "\n") +} + +// Dark-vector + profiler retention/export scripts, embedded so the operator can +// register them (if not already present) at boot via CreateRetentionScript. +// Each keeps its tracepoint permanently upserted ("876000h" β‰ˆ 100y, effectively +// permanent β€” no reliance on the 24h re-run) and exports its table to ClickHouse +// via the OTel plugin (px.export + px.otel.ClickHouseRows). stack_trace needs no +// tracepoint β€” it exports the native continuous profiler (stack_traces.beta). + +//go:embed presets/dc_snoop.pxl +var dcSnoopScript string + +//go:embed presets/stack_trace.pxl +var stackTraceScript string + +//go:embed presets/creds_change.pxl +var credsChangeScript string + +//go:embed presets/dc_snoop_deploy.pxl +var dcSnoopDeployScript string + +//go:embed presets/creds_change_deploy.pxl +var credsChangeDeployScript string + +// TracepointDef is a bpftrace tracepoint the AE deploys itself at boot. The +// retention/cron export path cannot deploy tracepoints (its pxtrace mutation is +// dropped), so the AE owns deployment via a mutation ExecuteScript (pixieapi). +// Script is an `import pxtrace` + UpsertTracepoint program (idempotent upsert, +// permanent TTL); Table is the tracepoint output table its export preset reads. +type TracepointDef struct { + Name string + Table string + Script string +} + +// DesiredTracepoints is the source of truth for the bpftraces the AE deploys at +// boot. stack_traces.beta (V9) is the native continuous profiler and needs no +// tracepoint, so it is absent here (its export preset works with no deploy). +// Extend this list as new bpftraces (V6 mprotect, V8 bpf/ptrace, …) are added. +func DesiredTracepoints() []TracepointDef { + return []TracepointDef{ + {Name: "dc_snoop", Table: "dc_snoop", Script: dcSnoopDeployScript}, + {Name: "creds_change", Table: "creds_change", Script: credsChangeDeployScript}, + } +} + +// DarkVectorPresets are the tracepoint/profiler export scripts the operator +// registers if-not-present. Names are operator-managed (reconciled on boot). +func DarkVectorPresets() []*ScriptDefinition { + return []*ScriptDefinition{ + {Name: "ch-dc_snoop", Description: "dc_snoop (dentry cache: process+file, V1/V2) β†’ ClickHouse", FrequencyS: 10, Script: strings.Replace(dcSnoopScript, "# __DC_SNOOP_EXCLUSION__", dcSnoopExclusion(), 1)}, + {Name: "ch-stack_trace", Description: "stack_traces.beta (continuous profiler, V9) β†’ ClickHouse", FrequencyS: 10, Script: stackTraceScript}, + {Name: "ch-creds_change", Description: "commit_creds privilege-escalation to root (V7) β†’ ClickHouse", FrequencyS: 10, Script: credsChangeScript}, + } +} diff --git a/src/vizier/services/adaptive_export/internal/script/presets/creds_change.pxl b/src/vizier/services/adaptive_export/internal/script/presets/creds_change.pxl new file mode 100644 index 00000000000..2eebda41e31 --- /dev/null +++ b/src/vizier/services/adaptive_export/internal/script/presets/creds_change.pxl @@ -0,0 +1,41 @@ +# 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 + +# creds_change EXPORT (retention cron) - query the creds_change tracepoint table +# and export to ClickHouse. The tracepoint is deployed by the AE at boot +# (creds_change_deploy.pxl); a cron script cannot deploy it. Read + export only. + +import px + +table_name = 'creds_change' +df = px.DataFrame(table=table_name, start_time=px.plugin.start_time, end_time=px.plugin.end_time) + +# pid -> pod/namespace attribution via a process_stats merge on pid (the tracepoint +# emits a raw kernel pid, no upid). Validated join; blank pod for host/transient +# pids. See dc_snoop.pxl for why it is pid-only (not pid+asid). +proc = px.DataFrame(table='process_stats', start_time='-5m') +proc.namespace = proc.ctx['namespace'] +proc.pod = proc.ctx['pod'] +proc.container = proc.ctx['container'] +proc.hostname = px.upid_to_node_name(proc.upid) +proc.pid = px.upid_to_pid(proc.upid) +proc = proc.groupby(['namespace', 'pod', 'container', 'hostname', 'pid']).agg() +df = df.merge(proc, how='left', left_on=['pid'], right_on=['pid'], suffixes=['', '_x']) +df = df[['time_', 'pid', 'comm', 'old_uid', 'new_uid', 'namespace', 'pod', 'container', 'hostname']] + +# event_time as nanoseconds -> sink emits DateTime64(9) (see dc_snoop.pxl). +df.event_time = df.time_ +px.export(df, px.otel.ClickHouseRows(table=table_name)) diff --git a/src/vizier/services/adaptive_export/internal/script/presets/creds_change_deploy.pxl b/src/vizier/services/adaptive_export/internal/script/presets/creds_change_deploy.pxl new file mode 100644 index 00000000000..206ba05df13 --- /dev/null +++ b/src/vizier/services/adaptive_export/internal/script/presets/creds_change_deploy.pxl @@ -0,0 +1,43 @@ +# 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 + +# creds_change tracepoint DEPLOYMENT (mutation) - run once at AE boot via the +# pixieapi mutation path. Privilege-escalation to root: a process whose creds +# are committed with new uid==0 while its previous (real) uid was >0. +# kprobe:commit_creds - arg0 is the NEW cred; the OLD uid is the current task's +# real_cred (still in place at commit time). V7 credential vector. Idempotent +# upsert; TTL 876000h (~permanent). See dc_snoop_deploy.pxl for why the AE owns +# tracepoint deployment separately from the cron export path. + +import pxtrace + +program = """ +#include +#include + +kprobe:commit_creds +{ + $new = (struct cred *)arg0; + $newuid = $new->uid.val; + $olduid = ((struct task_struct *)curtask)->real_cred->uid.val; + if ($newuid == 0 && $olduid > 0) { + printf("time_:%llu pid:%d comm:%s old_uid:%u new_uid:%u", + nsecs, pid, comm, $olduid, $newuid); + } +} +""" + +pxtrace.UpsertTracepoint('creds_change', 'creds_change', program, pxtrace.kprobe(), "876000h") diff --git a/src/vizier/services/adaptive_export/internal/script/presets/dc_snoop.pxl b/src/vizier/services/adaptive_export/internal/script/presets/dc_snoop.pxl new file mode 100644 index 00000000000..b930f3907ef --- /dev/null +++ b/src/vizier/services/adaptive_export/internal/script/presets/dc_snoop.pxl @@ -0,0 +1,56 @@ +# 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 + +# dc_snoop EXPORT (retention cron) - query the dc_snoop tracepoint table and +# export to ClickHouse. The tracepoint itself is deployed by the AE at boot +# (dc_snoop_deploy.pxl) - a cron script CANNOT deploy a tracepoint (its mutation +# is dropped), so this script only reads + exports the already-deployed table. + +import px + +table_name = 'dc_snoop' +df = px.DataFrame(table=table_name, start_time=px.plugin.start_time, end_time=px.plugin.end_time) + +# pid -> pod/namespace attribution. Dark-vector tracepoints emit a raw kernel pid +# with NO upid, so px.upid_to_* / ctx['pod'] fail outright; resolve pod by merging +# process_stats on pid ONLY - the validated join (PodEnrichPxL, dx#126). NOT +# pid+asid: on a dynamic tracepoint px.asid() is the aggregator/kelvin asid, not +# the per-PEM asid of the data, so pid+asid never matches. Best-effort left join: +# pod/namespace stay blank for host/transient pids (correct - they have no pod). +proc = px.DataFrame(table='process_stats', start_time='-5m') +proc.namespace = proc.ctx['namespace'] +proc.pod = proc.ctx['pod'] +proc.container = proc.ctx['container'] +proc.hostname = px.upid_to_node_name(proc.upid) +proc.pid = px.upid_to_pid(proc.upid) +proc = proc.groupby(['namespace', 'pod', 'container', 'hostname', 'pid']).agg() +df = df.merge(proc, how='left', left_on=['pid'], right_on=['pid'], suffixes=['', '_x']) +# Keep exactly the forensic_db.dc_snoop columns (drop the merge's pid_x etc.), +# else the export sink sends an unknown column and the INSERT fails. +df = df[['time_', 'pid', 'comm', 't', 'file', 'namespace', 'pod', 'container', 'hostname']] + +# Drop known infrastructure namespaces + process comms (blank-namespace workload +# rows are kept). The filter is injected here from env by presets.go +# (DC_SNOOP_EXCLUDE_NAMESPACES / DC_SNOOP_EXCLUDE_COMMS) so a process can be added +# without recompiling. Kept in sync with dx benchlive.writeSelfExclusion. +# __DC_SNOOP_EXCLUSION__ + +# Provide event_time (nanoseconds) explicitly so the ClickHouse export sink +# emits it as DateTime64(9) via its normal type map, instead of auto-appending +# a DateTime64(3) millisecond column (which would break the nanosecond-consistent +# event_time contract shared with the AE HTTP path + dx/soc reads). +df.event_time = df.time_ +px.export(df, px.otel.ClickHouseRows(table=table_name)) diff --git a/src/vizier/services/adaptive_export/internal/script/presets/dc_snoop_deploy.pxl b/src/vizier/services/adaptive_export/internal/script/presets/dc_snoop_deploy.pxl new file mode 100644 index 00000000000..bf8333f0a4b --- /dev/null +++ b/src/vizier/services/adaptive_export/internal/script/presets/dc_snoop_deploy.pxl @@ -0,0 +1,60 @@ +# 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 + +# dc_snoop tracepoint DEPLOYMENT (mutation) - run once at AE boot via the +# pixieapi mutation path (pxapi sets Mutation:true for `import pxtrace`). +# The retention/cron export path CANNOT deploy tracepoints (the mutation is +# dropped), so the AE owns this deploy separately. UpsertTracepoint is +# idempotent, so re-running at every boot is safe (create-if-absent / no-op). +# TTL 876000h (~100y) = effectively permanent; no reliance on the 24h re-run. + +import pxtrace + +program = """ +#include + +// from fs/namei.c: +struct nameidata { + struct path path; + struct qstr last; + // [...] +}; + +// comment out this block to avoid showing hits: +kprobe:lookup_fast, +kprobe:lookup_fast.constprop.* +{ + $nd = (struct nameidata *)arg0; + printf("time_:%llu pid:%d comm:%s t:%s file:%s", + nsecs, pid, comm, "R", str($nd->last.name)); +} + +kprobe:d_lookup +{ + $name = (struct qstr *)arg1; + @fname[tid] = $name->name; +} + +kretprobe:d_lookup +/@fname[tid]/ +{ + printf("time_:%llu pid:%d comm:%s t:%s file:%s", + nsecs, pid, comm, "M", str(@fname[tid])); + delete(@fname[tid]); +} +""" + +pxtrace.UpsertTracepoint('dc_snoop', 'dc_snoop', program, pxtrace.kprobe(), "876000h") diff --git a/src/vizier/services/adaptive_export/internal/script/presets/stack_trace.pxl b/src/vizier/services/adaptive_export/internal/script/presets/stack_trace.pxl new file mode 100644 index 00000000000..6fc29452bf0 --- /dev/null +++ b/src/vizier/services/adaptive_export/internal/script/presets/stack_trace.pxl @@ -0,0 +1,40 @@ +# 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 + +import px + +# Native continuous profiler (stack_traces.beta) - always-on, no tracepoint. +# Exports folded user+kernel stacks per pod -> ClickHouse via the OTel plugin. +# This is the OTel "profiles" signal for the SBoB stack-trace vector (V9). +df = px.DataFrame(table='stack_traces.beta', start_time=px.plugin.start_time, end_time=px.plugin.end_time) +df.namespace = df.ctx['namespace'] +df.pod = df.ctx['pod'] +df.container = df.ctx['container'] +df.hostname = px.upid_to_node_name(df.upid) +df = df[['time_', 'upid', 'namespace', 'pod', 'container', 'hostname', 'stack_trace_id', 'stack_trace', 'count']] + +# EXCLUDE our own monitoring/infra pods (see dc_snoop.pxl) - the profiler otherwise +# fills stack_trace with samples of Pixie's own pem/kelvin/vizier + the AE + CH, +# which is self-observation noise, not workload control-flow evidence. +df = df[df.namespace != 'pl'] +df = df[df.namespace != 'px-operator'] +df = df[df.namespace != 'olm'] +df = df[df.namespace != 'clickhouse'] +df = df[df.namespace != 'kube-system'] + +# event_time as nanoseconds -> sink emits DateTime64(9) (see dc_snoop.pxl). +df.event_time = df.time_ +px.export(df, px.otel.ClickHouseRows(table='stack_trace')) diff --git a/src/vizier/services/adaptive_export/internal/script/presets_test.go b/src/vizier/services/adaptive_export/internal/script/presets_test.go new file mode 100644 index 00000000000..2ea0f01ab22 --- /dev/null +++ b/src/vizier/services/adaptive_export/internal/script/presets_test.go @@ -0,0 +1,50 @@ +// Copyright 2018- The Pixie Authors. +// SPDX-License-Identifier: Apache-2.0 + +package script + +import ( + "strings" + "testing" +) + +func chDcSnoop(t *testing.T) string { + t.Helper() + for _, p := range DarkVectorPresets() { + if p.Name == "ch-dc_snoop" { + return p.Script + } + } + t.Fatal("ch-dc_snoop preset not found") + return "" +} + +func TestDcSnoopExclusionDefault(t *testing.T) { + s := chDcSnoop(t) + if strings.Contains(s, "#__DC_SNOOP_EXCLUSION__") { + t.Fatal("exclusion placeholder was not substituted") + } + for _, want := range []string{ + "df = df[df.comm != 'k3s-server']", + "df = df[df.comm != 'runc:[2:INIT]']", + "df = df[df.namespace != 'honey']", + } { + if !strings.Contains(s, want) { + t.Errorf("default filter missing: %s", want) + } + } + if strings.Contains(s, "df = df[df.namespace != '']") { + t.Error("must NOT drop blank-namespace rows") + } +} + +func TestDcSnoopExclusionConfigurable(t *testing.T) { + t.Setenv("DC_SNOOP_EXCLUDE_COMMS", "foo, bar") + s := chDcSnoop(t) + if !strings.Contains(s, "df = df[df.comm != 'foo']") || !strings.Contains(s, "df = df[df.comm != 'bar']") { + t.Error("DC_SNOOP_EXCLUDE_COMMS override not applied") + } + if strings.Contains(s, "k3s-server") { + t.Error("env override should replace, not append to, the default comm list") + } +} diff --git a/src/vizier/services/adaptive_export/internal/sink/clickhouse.go b/src/vizier/services/adaptive_export/internal/sink/clickhouse.go index 1f30fb8a187..11b0aed7202 100644 --- a/src/vizier/services/adaptive_export/internal/sink/clickhouse.go +++ b/src/vizier/services/adaptive_export/internal/sink/clickhouse.go @@ -181,8 +181,14 @@ func (s *ClickHouseHTTP) WritePixieRows(ctx context.Context, table string, rows if strings.Contains(table, ".") { identifier = "`" + table + "`" } + // SETTINGS async_insert=0: write synchronously so the rows land (and are + // counted in the response) immediately. Fresh ClickHouse deployments default + // async_insert=1, which buffers the INSERT and returns written_rows=0 β€” the + // AE then looks like it "wrote 0" while the evidence trickles in minutes + // later (or is lost on a flush failure). This makes the evidence write a + // self-contained per-PG-fix-free contract. res, err := s.c.Insert(ctx, - fmt.Sprintf("INSERT INTO %s.%s FORMAT JSONEachRow", s.cfg.Database, identifier), + fmt.Sprintf("INSERT INTO %s.%s SETTINGS async_insert=0 FORMAT JSONEachRow", s.cfg.Database, identifier), buf.Bytes(), chhttp.InsertOptions{FailLoud: true}) if err != nil { return fmt.Errorf("sink: pixie POST %s: %w", table, err)