diff --git a/README.md b/README.md index 6fca979..6a60a01 100644 --- a/README.md +++ b/README.md @@ -159,7 +159,7 @@ emits benchstat output) can be pointed at any `addr` from `systems.yml` — see in `queries/native/`; they measure the same retrieval work, not identical semantics. Queries with no equivalent report `n/a`. - **Datasets.** Metrics are live `node_exporter` series multiplied to 10 hosts by - vmagent during a prewarm (`PREWARM` env, default 90s); some engines have PromQL + vmagent during a prewarm (`PREWARM` env, default 30s); some engines have PromQL dialect gaps (e.g. GreptimeDB rejects `topk` and bare `{__name__=~...}`), reported as `ERR`/empty. Logs match the loghub suite when `LOGHUB_DIR` is set, else synthetic. Traces are diff --git a/cmd/benchctl/main.go b/cmd/benchctl/main.go index 8a85629..37bbb1c 100644 --- a/cmd/benchctl/main.go +++ b/cmd/benchctl/main.go @@ -54,6 +54,8 @@ func run(args []string) error { return e.Ingest(must(rest, 0, "ingest ")) case "query": return e.Query(must(rest, 0, "query [runs]"), runs(rest, 1, e.Runs), 3) + case "check": + return e.Check(must(rest, 0, "check ")) case "collect": return e.Collect() case "report": @@ -92,6 +94,9 @@ func benchLane(e *bench.Env, lane string, n int) error { if err := e.Query(lane, n, 3); err != nil { return err } + if err := e.Check(lane); err != nil { + return err + } return e.Collect() } @@ -101,6 +106,7 @@ func usage() { up [metrics|logs|traces|all] build + start a lane (default all) ingest push canonical dataset (fan-out) query [runs] replay suite against every engine + check verify query results (count vs reference) collect capture storage + memory footprint report render results/REPORT.md bench [runs] up→ingest→settle→query→collect diff --git a/compose/base.yml b/compose/base.yml index ca32b66..5c46e8f 100644 --- a/compose/base.yml +++ b/compose/base.yml @@ -51,7 +51,7 @@ services: entrypoint: ["/bin/sh", "-c"] command: - | - for b in mimir loki tempo greptimedb; do + for b in mimir loki tempo greptimedb oteldb-s3; do aws --endpoint-url http://fs:9000 s3 mb "s3://$$b" 2>/dev/null || true done echo "buckets ready" diff --git a/compose/metrics.yml b/compose/metrics.yml index bd4e856..5f32991 100644 --- a/compose/metrics.yml +++ b/compose/metrics.yml @@ -28,6 +28,7 @@ services: - -promscrape.config=/etc/vmagent/scrape.yml - -remoteWrite.url=http://oteldb:19291/api/v1/write - -remoteWrite.url=http://oteldb-ch:19291/api/v1/write + - -remoteWrite.url=http://oteldb-s3:19291/api/v1/write - -remoteWrite.url=http://prometheus:9090/api/v1/write - -remoteWrite.url=http://victoriametrics:8428/api/v1/write - -remoteWrite.url=http://mimir:9009/api/v1/push @@ -77,9 +78,11 @@ services: ports: ["127.0.0.1:9009:9009"] # GreptimeDB standalone, object storage on go-faster/fs S3 (bucket: greptimedb). + # In [logs]/[traces] too: it ingests the OTLP log/trace streams into otel_logs / + # opentelemetry_traces and is queried over SQL (systems.yml, queries/native/greptimedb.yml). greptimedb: - image: greptime/greptimedb:v0.11.0 - profiles: [metrics] + image: greptime/greptimedb:v1.1.1 + profiles: [metrics, logs, traces] networks: [bench] depends_on: fs-init: { condition: service_completed_successfully } diff --git a/compose/shared.yml b/compose/shared.yml index 2b5d48c..1cc6446 100644 --- a/compose/shared.yml +++ b/compose/shared.yml @@ -7,6 +7,7 @@ volumes: oteldb-data: + oteldb-s3-data: clickhouse-data: clickhouse-oteldb-data: @@ -38,6 +39,32 @@ services: - "127.0.0.1:9010:9010" # pprof - "127.0.0.1:19291:19291" # prometheus remote-write ingest + # oteldb, S3 variant: same image and --embedded, but the embedded storage + # engine's object-store backend is the shared go-faster/fs S3 (bucket oteldb-s3), + # putting oteldb in the apples-to-apples object-store group. Only the WAL is local. + oteldb-s3: + image: ${OTELDB_IMAGE:-ghcr.io/oteldb/oteldb:v0.41.0} + command: ["--config=/etc/otel/cfg.yml", "--embedded"] + networks: [bench] + depends_on: + fs-init: { condition: service_completed_successfully } + volumes: + - ../config/oteldb/oteldb-s3.yml:/etc/otel/cfg.yml:ro + - oteldb-s3-data:/data + environment: + - OTEL_LOG_LEVEL=info + - OTEL_METRICS_EXPORTER=none + - OTEL_LOGS_EXPORTER=none + - OTEL_TRACES_EXPORTER=none + - PPROF_ADDR=0.0.0.0:9010 + - GOMAXPROCS=${BENCH_CPUS:-} + - GOMEMLIMIT=${BENCH_MEMLIMIT:-} + ports: + - "127.0.0.1:9093:9090" # promql + - "127.0.0.1:3103:3100" # logql + - "127.0.0.1:3203:3200" # traceql + - "127.0.0.1:9012:9010" # pprof + # oteldb, ClickHouse-backed variant: same image WITHOUT --embedded, serving all # three signals from oteldb's chstorage on a dedicated ClickHouse (so its disk # footprint is isolated from gigapipe's). Distinct host ports from the embedded diff --git a/config/otelcol/logs.yml b/config/otelcol/logs.yml index 36d2fae..b4ff871 100644 --- a/config/otelcol/logs.yml +++ b/config/otelcol/logs.yml @@ -17,9 +17,19 @@ exporters: otlp/oteldb-ch: endpoint: oteldb-ch:4317 tls: { insecure: true } + otlp/oteldb-s3: + endpoint: oteldb-s3:4317 + tls: { insecure: true } otlphttp/victorialogs: logs_endpoint: http://victorialogs:9428/insert/opentelemetry/v1/logs tls: { insecure: true } + otlphttp/greptimedb: + # GreptimeDB OTLP logs: default (no pipeline) -> structured table with a + # fulltext-indexed `body`, `severity_text`/`severity_number`, and JSON + # log/resource attributes. Table selected by the x-greptime-log-table-name header. + logs_endpoint: http://greptimedb:4000/v1/otlp/v1/logs + headers: { x-greptime-log-table-name: otel_logs } + tls: { insecure: true } loki/loki: endpoint: http://loki:3100/loki/api/v1/push tls: { insecure: true } @@ -35,6 +45,8 @@ service: exporters: - otlp/oteldb - otlp/oteldb-ch + - otlp/oteldb-s3 - otlphttp/victorialogs + - otlphttp/greptimedb - loki/loki - loki/gigapipe diff --git a/config/otelcol/traces.yml b/config/otelcol/traces.yml index 22c2133..935a381 100644 --- a/config/otelcol/traces.yml +++ b/config/otelcol/traces.yml @@ -15,12 +15,24 @@ exporters: otlp/oteldb-ch: endpoint: oteldb-ch:4317 tls: { insecure: true } + otlp/oteldb-s3: + endpoint: oteldb-s3:4317 + tls: { insecure: true } otlp/tempo: endpoint: tempo:4317 tls: { insecure: true } otlphttp/victoriatraces: traces_endpoint: http://victoriatraces:10428/insert/opentelemetry/v1/traces tls: { insecure: true } + otlphttp/greptimedb: + # GreptimeDB OTLP traces -> table opentelemetry_traces (columns span_name, + # duration_nano, parent_span_id, span_attributes/resource_attributes JSON), + # queried via SQL. v1.1.x requires the built-in trace pipeline header. + traces_endpoint: http://greptimedb:4000/v1/otlp/v1/traces + headers: + x-greptime-trace-table-name: opentelemetry_traces + x-greptime-pipeline-name: greptime_trace_v1 + tls: { insecure: true } otlp/gigapipe: endpoint: gigapipe:4317 tls: { insecure: true } @@ -33,6 +45,8 @@ service: exporters: - otlp/oteldb - otlp/oteldb-ch + - otlp/oteldb-s3 - otlp/tempo - otlphttp/victoriatraces + - otlphttp/greptimedb - otlp/gigapipe diff --git a/config/oteldb/oteldb-s3.yml b/config/oteldb/oteldb-s3.yml new file mode 100644 index 0000000..b289323 --- /dev/null +++ b/config/oteldb/oteldb-s3.yml @@ -0,0 +1,58 @@ +# oteldb under test, S3 variant. Same embedded go-faster/storage engine as +# config/oteldb/oteldb.yml, but its object-store backend is the shared +# go-faster/fs S3 server (bucket: oteldb-s3) instead of a local `file` volume — +# the apples-to-apples object-store substrate Mimir/Loki/Tempo/GreptimeDB use. +# Only the WAL is local (S3 is stateless); see storage.s3 below. +storage: + backend: s3 + # Local write-ahead log (the only on-disk state for the s3 backend). + wal_dir: /data/wal + s3: + bucket: oteldb-s3 + endpoint: http://fs:9000 + region: us-east-1 + force_path_style: true # go-faster/fs serves path-style (endpoint/bucket/key) + access_key_id: bench + secret_access_key: benchbench + # Parallelize LogQL result materialization across workers, like the file variant. + log_query_parallelism: 8 + # Flush head data to S3 on this interval so a short bench actually exercises (and + # measures the footprint of) the object store rather than serving only from the + # in-memory head. Default (0) is size-triggered only — a one-shot ingest would + # never reach S3. + flush_interval: 10s + +prometheus: # PromQL HTTP API (Prometheus-compatible) + bind: 0.0.0.0:9090 + max_samples: 1_000_000 + timeout: 1m + enable_negative_offset: true +loki: # LogQL HTTP API (Loki-compatible) + bind: 0.0.0.0:3100 +tempo: # TraceQL HTTP API (Tempo-compatible) + bind: 0.0.0.0:3200 +health_check: + bind: 0.0.0.0:13133 + +# Ingest receivers: Prometheus remote-write for metrics, OTLP for logs + traces. +# Under --embedded the oteldbexporter writes to the embedded storage engine, so +# its DSN is ignored — but every pipeline must still name an exporter. +otelcol: + receivers: + prometheusremotewrite: + endpoint: 0.0.0.0:19291 + time_threshold: 100000h + otlp: + protocols: + grpc: + endpoint: 0.0.0.0:4317 + http: + endpoint: 0.0.0.0:4318 + exporters: + oteldbexporter: + dsn: clickhouse://default:default@clickhouse:9000 # ignored under --embedded + service: + pipelines: + metrics: { receivers: [prometheusremotewrite], exporters: [oteldbexporter] } + logs: { receivers: [otlp], exporters: [oteldbexporter] } + traces: { receivers: [otlp], exporters: [oteldbexporter] } diff --git a/internal/bench/check.go b/internal/bench/check.go new file mode 100644 index 0000000..abc01a1 --- /dev/null +++ b/internal/bench/check.go @@ -0,0 +1,269 @@ +package bench + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "path/filepath" + "strconv" + "strings" + "time" +) + +// Correctness statuses for a single (engine, query) probe. +const ( + statusOK = "OK" // 2xx and a non-empty result + statusEmpty = "EMPTY" // 2xx but zero results + statusErr = "ERR" // transport error, non-2xx, or unparseable body + statusNA = "N/A" // native-dialect engine has no equivalent for this query +) + +// probeResult is one engine's outcome for one query. +type probeResult struct { + count int + status string + apples bool +} + +// Check validates query *results* (not just latency): it issues each suite query +// once per engine, parses the response into a result count, and compares every +// apples engine against a reference (the first apples engine that returns data, +// i.e. oteldb). It catches the failures a latency run hides — a 200-but-empty +// response from a dialect gap, a broken filter, or missing data. Writes +// results//correctness.csv. +func (e *Env) Check(signal string) error { + suite, err := LoadSuite(e.Dir, signal) + if err != nil { + return err + } + systems, err := LoadSystems(filepath.Join(e.Dir, "systems.yml")) + if err != nil { + return err + } + sys := e.selected(systems.For(signal)) + client := &http.Client{Timeout: 120 * time.Second} + + // probes[queryID][engine] = result + probes := map[string]map[string]probeResult{} + for _, q := range suite.Queries { + probes[q.ID] = map[string]probeResult{} + } + + for _, s := range sys { + native, err := LoadNative(e.Dir, s.Name) + if err != nil { + return err + } + // Apples engines use the reference language; a native override only applies + // to a system's non-apples signals (see Query). + if s.Apples { + native = nil + } + fmt.Printf(">> [%s] checking %s (%s)\n", signal, s.Name, s.Lang) + for _, q := range suite.Queries { + useLang, text := s.Lang, q.Q + if native != nil { + useLang = native.API + t, ok := native.query(q.ID) + if !ok { + probes[q.ID][s.Name] = probeResult{0, statusNA, s.Apples} + continue + } + text = t + } + count, status := e.probeQuery(client, s.Addr, useLang, q.Type, text, suite) + probes[q.ID][s.Name] = probeResult{count, status, s.Apples} + } + } + + // Resolve a reference per query (first apples engine that returned data) and + // score every engine against it. + rows := [][]string{{"query_id", "engine", "count", "status", "verdict"}} + pass, fail := 0, 0 + for _, q := range suite.Queries { + byEngine := probes[q.ID] + ref, refCount := "", 0 + for _, s := range sys { + if p := byEngine[s.Name]; s.Apples && p.status == statusOK { + ref, refCount = s.Name, p.count + break + } + } + for _, s := range sys { + p := byEngine[s.Name] + verdict := verdictFor(s.Name, ref, p.count, refCount, p.status, p.apples) + switch verdict { + case "empty", "err": + fail++ + case "n/a", "none": + default: + pass++ + } + rows = append(rows, []string{q.ID, s.Name, strconv.Itoa(p.count), p.status, verdict}) + fmt.Printf(" %-26s %-12s %-6s %s\n", q.ID, s.Name, glyphFor(verdict), countStr(p.count, p.status)) + } + } + + out := filepath.Join(e.Dir, "results", signal, "correctness.csv") + if err := writeCSV(out, rows); err != nil { + return err + } + fmt.Printf(">> wrote %s (%d ok, %d failed)\n", out, pass, fail) + return nil +} + +// verdictFor scores one engine's result against the reference. Non-apples engines +// (native dialects) are judged on non-emptiness only — never on count agreement, +// since they answer a semantically different query over the same data. +func verdictFor(engine, ref string, count, refCount int, status string, apples bool) string { + switch status { + case statusNA: + return "n/a" + case statusErr: + return "err" + case statusEmpty: + if ref == "" { + // No apples engine returned data for this query — likely a selector + // that matches nothing or data outside the window, not an engine fault. + return "none" + } + return "empty" + } + // status OK + switch { + case engine == ref: + return "ref" + case !apples || ref == "": + return "ok" + case count == refCount: + return "match" + default: + return "diff" + } +} + +func glyphFor(v string) string { + switch v { + case "ref": + return "★ref" + case "match", "ok": + return "✓" + case "diff": + return "≠" + case "empty": + return "✗∅" + case "err": + return "✗ERR" + case "none": + return "∅?" + default: + return "-" + } +} + +func countStr(n int, status string) string { + switch status { + case statusOK: + return strconv.Itoa(n) + case statusEmpty: + return "∅" + case statusErr: + return "ERR" + default: + return "n/a" + } +} + +// probeQuery issues one query and returns its result count and status. It retries +// once on an empty result to ride out a transient flush gap. +func (e *Env) probeQuery(c *http.Client, addr, lang, typ, q string, s *Suite) (int, string) { + for attempt := 0; attempt < 2; attempt++ { + start, end := s.Window(time.Now().Unix()) + u := buildURL(addr, lang, typ, q, start, end, s.Range.Step) + if u == "" { + return 0, statusErr + } + resp, err := c.Get(u) + if err != nil { + return 0, statusErr + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return 0, statusErr + } + count, ok := resultCount(lang, body) + if !ok { + return 0, statusErr + } + if count > 0 { + return count, statusOK + } + if attempt == 0 { + time.Sleep(500 * time.Millisecond) + } + } + return 0, statusEmpty +} + +// resultCount extracts the number of results from a response body for a query +// language. ok=false when the body cannot be parsed as that API's success shape. +func resultCount(lang string, body []byte) (int, bool) { + switch lang { + case "promql", "logql": + // Prometheus/Loki envelope: {"status":"success","data":{"result":[...]}} + var r struct { + Status string `json:"status"` + Data struct { + Result []json.RawMessage `json:"result"` + } `json:"data"` + } + if err := json.Unmarshal(body, &r); err != nil { + return 0, false + } + if r.Status != "" && r.Status != "success" { + return 0, false + } + return len(r.Data.Result), true + case "traceql": + var r struct { + Traces []json.RawMessage `json:"traces"` + } + if err := json.Unmarshal(body, &r); err != nil { + return 0, false + } + return len(r.Traces), true + case "jaeger": + var r struct { + Data []json.RawMessage `json:"data"` + } + if err := json.Unmarshal(body, &r); err != nil { + return 0, false + } + return len(r.Data), true + case "logsql": + // VictoriaLogs streams newline-delimited JSON objects, one per log line. + n := 0 + for _, line := range strings.Split(string(body), "\n") { + if strings.TrimSpace(line) != "" { + n++ + } + } + return n, true + case "greptimesql": + // GreptimeDB v1 SQL response: {"output":[{"records":{"rows":[[...],...]}}]}. + var r struct { + Output []struct { + Records struct { + Rows []json.RawMessage `json:"rows"` + } `json:"records"` + } `json:"output"` + } + if err := json.Unmarshal(body, &r); err != nil || len(r.Output) == 0 { + return 0, false + } + return len(r.Output[0].Records.Rows), true + } + return 0, false +} diff --git a/internal/bench/compose.go b/internal/bench/compose.go index 65518d4..acba642 100644 --- a/internal/bench/compose.go +++ b/internal/bench/compose.go @@ -58,6 +58,7 @@ var laneDrivers = map[string][]string{ // engineDeps are non-engine services a given engine needs beyond the lane // driver. oteldb's file backend is self-contained, so it has none. var engineDeps = map[string][]string{ + "oteldb-s3": {"fs", "fs-init"}, "oteldb-ch": {"clickhouse-oteldb"}, "gigapipe": {"clickhouse"}, "mimir": {"fs", "fs-init"}, diff --git a/internal/bench/ingest.go b/internal/bench/ingest.go index 865a897..4e6aa40 100644 --- a/internal/bench/ingest.go +++ b/internal/bench/ingest.go @@ -39,9 +39,9 @@ func (e *Env) otelbench(args ...string) error { // node_exporter targets and remote-writes the live stream to every engine from // the moment the lane is up. Here we just let data accumulate (current // timestamps, in order) so the query window has samples. Duration: PREWARM env -// (default 90s). +// (default 30s). func (e *Env) ingestMetrics() error { - prewarm := 90 * time.Second + prewarm := 30 * time.Second if v := os.Getenv("PREWARM"); v != "" { if d, err := time.ParseDuration(v); err == nil { prewarm = d diff --git a/internal/bench/query.go b/internal/bench/query.go index 2b488de..a7f9b8f 100644 --- a/internal/bench/query.go +++ b/internal/bench/query.go @@ -37,6 +37,12 @@ func (e *Env) Query(signal string, count, warmup int) error { if err != nil { return err } + // Apples engines speak the signal's reference language directly; ignore any + // native override (a system like greptimedb has one only for the signals + // where it is non-apples, e.g. SQL over logs/traces — not for PromQL metrics). + if sys.Apples { + native = nil + } fmt.Printf(">> [%s] %s (%s) -> %s\n", signal, sys.Name, sys.Lang, sys.Addr) rows := [][]string{{"query_id", "lang", "p50_ms", "p90_ms", "p99_ms", "samples", "errors"}} @@ -137,6 +143,12 @@ func buildURL(addr, lang, typ, q string, start, end int64, step int) string { } // native dialects key on lang only (type is ignored) switch lang { + case "greptimesql": + // q is a full SQL statement that self-bounds its time window; GreptimeDB's + // SQL HTTP API accepts it as a GET query param. + v.Set("sql", q) + v.Set("db", "public") + return addr + "/v1/sql?" + v.Encode() case "logsql": v.Set("query", q) v.Set("start", strconv.FormatInt(start*1000, 10)) diff --git a/internal/bench/report.go b/internal/bench/report.go index 1804db2..b46c4a9 100644 --- a/internal/bench/report.go +++ b/internal/bench/report.go @@ -75,6 +75,8 @@ func (e *Env) Report() error { b.WriteString("|\n") } b.WriteString("\n") + + e.writeCorrectness(&b, signal, suite, systems) } if rows, ok := readCSV(filepath.Join(e.Dir, "results", "footprint.csv")); ok { @@ -95,6 +97,114 @@ func (e *Env) Report() error { return nil } +// writeCorrectness renders the per-signal correctness matrix (rows = query, +// columns = engine, cell = result count with a verdict glyph) from +// results//correctness.csv, plus a one-line pass/fail summary. No-op when +// the file is absent (a latency-only run). +func (e *Env) writeCorrectness(b *strings.Builder, signal string, suite *Suite, systems *Systems) { + rows, ok := readCSV(filepath.Join(e.Dir, "results", signal, "correctness.csv")) + if !ok { + return + } + // cell[query][engine] = rendered cell; track verdict counts for the summary. + cell := map[string]map[string]string{} + fails, none := 0, 0 + for _, r := range rows[1:] { + if len(r) < 5 { + continue + } + id, engine, count, status, verdict := r[0], r[1], r[2], r[3], r[4] + if cell[id] == nil { + cell[id] = map[string]string{} + } + cell[id][engine] = correctnessCell(count, status, verdict) + switch verdict { + case "empty", "err": + fails++ + case "none": + none++ + } + } + + var cols []System + for _, sys := range e.selected(systems.For(signal)) { + for _, r := range rows[1:] { + if r[1] == sys.Name { + cols = append(cols, sys) + break + } + } + } + if len(cols) == 0 { + return + } + + b.WriteString("**Correctness** — result count per query; `★`=reference, `≠`=count differs from reference, `∅`=missing data the reference returned, `∅?`=no engine had data (query matched nothing / out of window), `ERR`=query failed.\n\n") + b.WriteString("| query ") + for _, c := range cols { + mark := "" + if !c.Apples { + mark = "*" + } + fmt.Fprintf(b, "| %s%s ", c.Name, mark) + } + b.WriteString("|\n|---") + for range cols { + b.WriteString("|---") + } + b.WriteString("|\n") + for _, q := range suite.Queries { + fmt.Fprintf(b, "| `%s` ", q.ID) + for _, c := range cols { + v := cell[q.ID][c.Name] + if v == "" { + v = "–" + } + fmt.Fprintf(b, "| %s ", v) + } + b.WriteString("|\n") + } + switch { + case fails == 0 && none == 0: + b.WriteString("\n_All engines agree: every query returned data on every engine._\n\n") + case fails == 0: + fmt.Fprintf(b, "\n_No divergence; %d quer(y/ies) returned no data on any engine (`∅?` — matched nothing or out of window)._\n\n", none) + default: + fmt.Fprintf(b, "\n_%d correctness failure(s): an engine returned no data (`∅`/`ERR`) for a query the reference answered.%s_\n\n", + fails, noneSuffix(none)) + } +} + +func noneSuffix(none int) string { + if none == 0 { + return "" + } + return fmt.Sprintf(" (%d more matched nothing on any engine.)", none) +} + +func correctnessCell(count, status, verdict string) string { + switch verdict { + case "ref": + return count + " ★" + case "diff": + return count + " ≠" + case "match", "ok": + return count + case "empty": + return "∅" + case "none": + return "∅?" + case "err": + return "ERR" + case "n/a": + return "n/a" + } + if status == statusOK { + return count + } + return "–" +} + // readP90 returns id -> p90_ms from a results CSV. func readP90(path string) (map[string]string, bool) { rows, ok := readCSV(path) diff --git a/queries/native/greptimedb.yml b/queries/native/greptimedb.yml new file mode 100644 index 0000000..98b2a00 --- /dev/null +++ b/queries/native/greptimedb.yml @@ -0,0 +1,55 @@ +# GreptimeDB native SQL equivalents of the LogQL (logs) and TraceQL (traces) +# suites, matched by id. apples=false: same dataset/ingest, not LogQL/TraceQL +# semantics. One file serves both signals — each suite only looks up its own ids. +# GreptimeDB ingests the OTLP streams into `otel_logs` and `opentelemetry_traces` +# and is queried over its SQL HTTP API (GET /v1/sql?sql=...&db=public). Each query +# self-bounds its window with `now() - INTERVAL` (the suite lookback). +# +# Schema (see storage/_ref/greptimedb/src/servers/src/otlp/logs.rs): +# timestamp (ns) · severity_text (OTLP SeverityText, title-case "Error"/"Warn") +# · severity_number · body (raw log line, FULLTEXT-indexed) · log_attributes +# (JSON: http.method, http.status_code) · resource_attributes (JSON). +# The synthetic stream sets no service.name, so the "service" selector is all rows. +# Each query self-bounds the window with `now() - INTERVAL '5 minutes'` (the suite +# lookback), so no start/end is threaded in. +system: greptimedb +api: greptimesql +queries: + - id: select_service + q: "SELECT body FROM otel_logs WHERE timestamp >= now() - INTERVAL '5 minutes' LIMIT 1000" + - id: level_error + q: "SELECT body FROM otel_logs WHERE severity_text = 'Error' AND timestamp >= now() - INTERVAL '5 minutes' LIMIT 1000" + - id: level_warn + q: "SELECT body FROM otel_logs WHERE severity_text = 'Warn' AND timestamp >= now() - INTERVAL '5 minutes' LIMIT 1000" + - id: error_token + q: "SELECT body FROM otel_logs WHERE severity_text = 'Error' AND body LIKE '%method%' AND timestamp >= now() - INTERVAL '5 minutes' LIMIT 1000" + - id: body_get + q: "SELECT body FROM otel_logs WHERE body LIKE '%GET%' AND timestamp >= now() - INTERVAL '5 minutes' LIMIT 1000" + - id: status_regex + q: "SELECT body FROM otel_logs WHERE body LIKE '%\"status\":4%' AND timestamp >= now() - INTERVAL '5 minutes' LIMIT 1000" + - id: json_status + q: "SELECT body FROM otel_logs WHERE (body LIKE '%\"status\":4%' OR body LIKE '%\"status\":5%') AND timestamp >= now() - INTERVAL '5 minutes' LIMIT 1000" + - id: lines_by_level + q: "SELECT severity_text, count(*) AS lines FROM otel_logs WHERE timestamp >= now() - INTERVAL '5 minutes' GROUP BY severity_text" + - id: rate_by_level + q: "SELECT severity_text, count(*) AS lines FROM otel_logs WHERE timestamp >= now() - INTERVAL '5 minutes' GROUP BY severity_text" + - id: error_rate + q: "SELECT count(*) AS lines FROM otel_logs WHERE severity_text = 'Error' AND timestamp >= now() - INTERVAL '5 minutes'" + + # traces — over opentelemetry_traces, ingested with the built-in greptime_trace_v1 + # pipeline (GreptimeDB v1.1.x). That flattens spans to columns: service_name, + # span_name, duration_nano, parent_span_id, and one column per span attribute + # (e.g. "span_attributes.net.peer.ip"). Returns DISTINCT trace_id (LIMIT 20) to + # mirror Tempo's trace-level results. + - id: by_service + q: "SELECT DISTINCT trace_id FROM opentelemetry_traces WHERE service_name = 'frontend' AND timestamp >= now() - INTERVAL '60 minutes' LIMIT 20" + - id: any_span + q: "SELECT DISTINCT trace_id FROM opentelemetry_traces WHERE timestamp >= now() - INTERVAL '60 minutes' LIMIT 20" + - id: by_duration + q: "SELECT DISTINCT trace_id FROM opentelemetry_traces WHERE duration_nano > 1000000 AND timestamp >= now() - INTERVAL '60 minutes' LIMIT 20" + - id: by_span_name + q: "SELECT DISTINCT trace_id FROM opentelemetry_traces WHERE span_name = 'okey-dokey-0' AND timestamp >= now() - INTERVAL '60 minutes' LIMIT 20" + - id: by_attr + q: "SELECT DISTINCT trace_id FROM opentelemetry_traces WHERE \"span_attributes.net.peer.ip\" <> '' AND timestamp >= now() - INTERVAL '60 minutes' LIMIT 20" + - id: structural + q: "SELECT DISTINCT t.trace_id FROM opentelemetry_traces t JOIN opentelemetry_traces p ON t.trace_id = p.trace_id AND t.parent_span_id = p.span_id WHERE p.service_name = 'frontend' AND t.timestamp >= now() - INTERVAL '60 minutes' LIMIT 20" diff --git a/queries/native/victorialogs.yml b/queries/native/victorialogs.yml index 75401ff..b6ad9ed 100644 --- a/queries/native/victorialogs.yml +++ b/queries/native/victorialogs.yml @@ -1,27 +1,32 @@ # VictoriaLogs native LogsQL equivalents of logs.logql.yml (matched by id). # apples=false: same dataset/ingest, not LogQL semantics. Queried via -# GET /select/logsql/query. Field names follow VictoriaLogs' OTLP mapping -# (service.name -> service.name, severity -> level via the log record). +# GET /select/logsql/query. +# +# Field names follow VictoriaLogs' OTLP mapping (verified against the live schema): +# _msg = log body · severity = OTLP SeverityText (title-case "Error"/"Warn") +# · http.method · http.status_code · otelbench.resource. +# Note there is NO service.name (the synthetic stream sets none), so the "service" +# selector matches every line (`*`), and the level field is `severity`, not `level`. system: victorialogs api: logsql queries: - id: select_service - q: 'service.name:="unknown_service"' + q: '*' - id: level_error - q: 'level:=ERROR' + q: 'severity:=Error' - id: level_warn - q: 'level:=WARN' + q: 'severity:=Warn' - id: error_token - q: 'level:=ERROR AND "method"' + q: 'severity:=Error AND "method"' - id: body_get q: '"GET"' - id: status_regex q: '_msg:~"\"status\":4[0-9][0-9]"' - id: json_status - q: '_msg:~"\"status\":4[0-9][0-9]"' + q: '_msg:~"\"status\":[45][0-9][0-9]"' - id: lines_by_level - q: '* | stats by (level) count() lines' + q: '* | stats by (severity) count() lines' - id: rate_by_level - q: '* | stats by (level) count() lines' + q: '* | stats by (severity) count() lines' - id: error_rate - q: 'level:=ERROR | stats count() lines' + q: 'severity:=Error | stats count() lines' diff --git a/results/pprof/FINDINGS.md b/results/pprof/FINDINGS.md new file mode 100644 index 0000000..21b6a5c --- /dev/null +++ b/results/pprof/FINDINGS.md @@ -0,0 +1,112 @@ +# oteldb query-path profiling — findings + +Captured by `scripts/profile.sh` under each signal's query load against oteldb +(`ghcr.io/oteldb/oteldb:0245c0a4`, embedded `file` backend, unconstrained CPU). +Profiles in this dir: `.{cpu,heap,allocs}.pb.gz`, binary `oteldb.bin`. +CPU = 20 s under load; `heap` = in-use snapshot taken hot; `allocs` = +alloc_space **baseline-diffed** over the load window (attributable, not +cumulative-since-start). + +Explore: +``` +go tool pprof -http=: results/pprof/oteldb.bin results/pprof/logs.allocs.pb.gz +go tool pprof -list 'logColumns.record' results/pprof/oteldb.bin results/pprof/logs.allocs.pb.gz +``` + +## The one-paragraph summary + +Same two-front split as before. **Resident storage (in-use heap)** is identical +across signals — the live in-memory state: metrics head `sampleBuf.appendSample` +**~33%** and the record engine `recordCols.appendClone`+`cloneBytes` **~37%** +(clone-on-append copies every column value into a fresh slab). Those own ~70% of +RSS; shrink them to cut footprint. **Query latency** is allocation/GC bound and +the hot spot keeps moving release-to-release: on **metrics** the new #1 is +`engine.sortedWindow` (37% of allocs, 22% CPU — it sorts+copies each window); +**logs** de-concentrated from the 97%-`newRecordCols` spike of `3cc3f30e` into a +spread (`logColumns.record` 43% cum, `signal.DecodeAttributes` 19%, +`logqlabels.LabelSet.Set` 17.5%, `newRecordCols` 10%) and now saturates cores +(450% CPU util); **traces** are unchanged — a `SelectSpansets → scanSpans → +materializeSpans` chain at ~45% GC. Highest-leverage fix is still **pool/stop +cloning `recordCols`** plus killing the per-row attribute re-materialization +(`DecodeAttributes` + `pcommon.Map` rebuild) that is now visible on both logs and +traces. + +## Per signal + +### metrics — now sortedWindow-bound +- **CPU (~42% util):** `engine.sortedWindow` **22% cum** (with + `slices.partialInsertionSort` 4.7% on the `{ts,val,sf}` tuples), GC moderate + (`scanObject` 14.6% cum, `tryDefer` 8.9%, `spanClass` 5%, `memclr` 2.85%), + `memmove` 3.9%. +- **Allocs:** `engine.sortedWindow` **37.4% cum** (1.97 GB flat) — sorts and + copies each fetch window; `promql.floatSamples` 17%, `ringbuffer.Push` 12%, + `engine.windowCopy` 11%, `ensureCap[int64|float64]` ~8.5% (decode buffers grown + not pooled), prometheus `ScratchBuilder.Add` 3.5%. +- **Levers:** sort window indices in place / reuse the `{ts,val,sf}` scratch; + pool `windowCopy`/`floatSamples` and pre-size `ensureCap` to the known point + count; reuse ringbuffer backing arrays across steps. + +### logs — de-concentrated, now core-saturating +- **CPU (~450% util — heavily parallel):** GC ~40% (`scanObject` 16% cum, + `tryDefer` 10.6%, `spanClass` 4.5%) but real work now dominates: `memmove` + 7.5%, `signal.DecodeAttributes` 11.4% cum, `aeshashbody` 2.7% (map hashing for + label sets). +- **Allocs:** `storagebackend.logColumns.record` **43.4% cum** (16.3 GB flat); + `signal.DecodeAttributes` 19.4%; `logqlabels.LabelSet.Set` 17.5% + + `NewLabelSet` 3% (label-set maps rebuilt per line); `newRecordCols` 10.2%; + `pcommon.Map.PutEmpty` 7.8% + `NewAnyValueStringValue` 3% (pdata rebuild); + `bytealg.MakeNoZero` 6.25%. +- **Levers:** decode attributes into reused buffers (kills ~19%); reuse + `LabelSet` maps instead of `Set`/`NewLabelSet` per line (~20%); pool + `recordCols`; avoid the `pcommon.Map` round-trip in `logColumns.record`. +- **Note:** the v`3cc3f30e` pathology (`newRecordCols` 97% of logs allocs) is + **gone** — work is now spread, so logs runs much wider (33→450% CPU util) but + the per-line attribute/label-set rebuild is the new dominant cost. + +### traces — unchanged SelectSpansets → scanSpans → materializeSpans chain +- **CPU (193% util; ~45% GC):** `scanObject` 33% cum, `tryDefer` 18%, `spanClass` + 7.8%, `memclr` 6.5%; work: `recordCols.appendRow` 4%, `DecodeAttributes`. +- **Allocs:** `SelectSpansets` 98% cum → `scanSpans` 71% cum, `materializeSpans` + 34% cum (8.4 GB flat), `newRecordCols` 12.3%, `permute[[]uint8]` 9.5% + + `permute[int64]` 2.8% (byte copies during sort), `DecodeAttributes` 8%, + `pcommon.Map.PutEmpty` 6.3% + `NewMap`/`NewAnyValueStringValue` ~3% + (`otelAttrs` 12.7% cum rebuilding pdata). +- **Levers:** pool `recordCols`; sort by index-permute not `permute` byte copies; + decode attributes into reused buffers; avoid rebuilding `pcommon.Map` per span. + +## Cross-cutting (do these first — ordered by leverage) +1. **`recordengine.recordCols` — pool on read, stop cloning on resident.** 37% of + resident heap (`appendClone`/`cloneBytes`) and a top-5 query allocator on every + signal. Pool the column set on the fetch path; store resident values as offsets + into one shared byte slab instead of `cloneBytes` per value. (`storage/recordengine`) +2. **`signal.DecodeAttributes` + `pcommon.Map` rebuild — reuse buffers, skip the + round-trip.** Now ~19% of logs allocs and ~14% of traces (DecodeAttributes + + PutEmpty/NewMap). Decode into reused buffers; don't rebuild `pcommon.Map` per + row/span. (`storage/signal`, `internal/storagebackend`) +3. **`engine.sortedWindow` — sort in place / reuse scratch.** New metrics #1 (37% + allocs, 22% CPU). (`storage/engine`) +4. **`logqlabels.LabelSet` — reuse maps.** ~20% of logs allocs via `Set`/`NewLabelSet`. + (`internal/logql/logqlengine/logqlabels`) +5. **`sampleBuf.appendSample` — 33% of resident heap.** Metrics head in-memory + samples; chunk/flush sooner or pack denser to cut RSS. (`storage/engine`) +6. GC tuning (`GOGC`/soft memlimit) is a band-aid; the allocation fixes are the cure. + +## What changed vs 3cc3f30e (for trend tracking) +- **Metrics:** `windowCopy` (was 41% of allocs) dropped to 11%; new top is + `sortedWindow` 37% + `ensureCap` reappearing. CPU util 33→42%. +- **Logs: de-concentrated.** `newRecordCols` 97% → 10%; cost spread across + `logColumns.record`/`DecodeAttributes`/`LabelSet.Set`. CPU util 58→450% (now + parallelizes instead of serializing on one allocator) — wall-clock p90s + basically flat (see REPORT.md), so this is a throughput/GC-pressure shift. +- **Traces:** essentially identical shape and magnitudes. +- **Resident heap:** unchanged shape — `sampleBuf` + `recordCols` clone still + dominate; storage-footprint levers are the same. + +## Methodology caveats +- Metrics captured with **vmagent paused** (clean read, no live-ingest cache + churn); logs/traces queried over their ingested dataset with collectors idle. +- **In-use heap is a whole-process snapshot** — includes the resident ingestion + head (`ConsumeMetrics`/`ConsumeLogs`), which is exactly the in-RAM storage to + shrink, so it's reported as-is. +- Single-node `file` backend; the S3 backend would add object-fetch latency/allocs + not seen here. diff --git a/results/pprof/LOGS-FINDINGS.md b/results/pprof/LOGS-FINDINGS.md new file mode 100644 index 0000000..6c68046 --- /dev/null +++ b/results/pprof/LOGS-FINDINGS.md @@ -0,0 +1,80 @@ +# oteldb LogQL query performance — investigation (EXPLAIN ANALYZE + pprof) + +Image `ghcr.io/oteldb/oteldb:logql-fix` (embedded `file` backend), ~75k log +records in a 300s window. Profiles: `logs-cpu.pb.gz`, `logs-heap.pb.gz` (+ binary +`oteldb.bin`). Driven by the LogQL suite (`queries/logs.logql.yml`). + +## The one-line finding + +**Storage is not the bottleneck — the LogQL engine is.** EXPLAIN ANALYZE +(`X-Oteldb-Profile: 1`) shows the fetch returns ~75k rows in **3–8 ms**, but the +query spends **80–253 ms of self-time above the fetch**, materializing and +processing every one of those rows. + +``` +lines_by_level query 257.8ms (self 253.2ms) └─ recordengine.fetch 4.5ms rows=74700 series_matched=10 +select_service query 92.5ms (self 84.8ms) └─ recordengine.fetch 7.7ms rows=74700 +status_regex query 84.0ms (self 80.7ms) └─ recordengine.fetch 3.4ms rows=74700 +json_status query 86.1ms (self 82.4ms) └─ recordengine.fetch 3.7ms rows=74700 +``` + +Every query fetches and materializes **all 74700 rows** — even `select_service` +with `limit=1000` (the limit is applied *after* materializing everything). + +## Where the engine self-time goes (CPU, 20s under load) + +- **Per-row materialization** — `storagebackend.logColumns.recordInto` 15% cum, + `signal.AppendAttributes` 4% (decoding each row's attributes back into a + `pcommon.Map`). +- **GC** ~40% cum (`tryDeferToSpanScan` 15%, `scanObjectsSmall`/`scanObject` ~20%, + `mallocgcSmallScanNoHeader` 10%) — driven by the per-row allocations below. +- **Map-based label sets** ~12% — `aeshashbody` 5% + `mapaccess`/`getWithoutKey` + ~5% + `maps.Clone` 2% (the label set is a `map[Label]Value`, rebuilt per row). +- **Metric grouping** — `aggregatedLabels.forEach` 8% (the `sum by (level)` path). +- **Sort via reflection** ~6% — `sort.SliceStable` → `reflectlite.Swapper` + + `typedmemmove` sorting 74700 entries with a reflection-based swapper. + +## What allocates (heap alloc_space) + +`recordInto` 22% · `LabelSet.Set`+`buildSet`+`NewLabelSet`+`AggregatedLabelsFromSet`+ +`aggregatedLabels.By` ≈ **35%** (label sets) · `newRecordCols` 7% · `slices.Grow` +(attr KeyValue) 7% · `pcommon.Map.PutEmpty`+`NewAnyValue*` ≈ 8% (pdata round-trip) · +`rangeAggIterator.Next`+`fillWindow` ≈ **29%** (the metric-window aggregation that +makes `lines_by_level`/`rate_by_level` the slowest at ~240 ms). + +## Fixes, by leverage + +1. **Limit pushdown / lazy materialization.** `EvalPipeline` fetches every record + and `materialize` builds a full label set per row, then truncates to + `params.Limit` (logs.go:96–103). For the filter queries (`select_service`, + `status_regex`, `json_status`, ~85 ms) only `limit` entries are kept — push the + limit into the fetch/materialize so it stops early instead of building 74700 + and discarding 73700. Biggest win for non-metric queries. +2. **Slice-backed / pooled `LabelSet`** instead of `map[Label]Value`. The map per + row (hash + clone) is ~12% CPU and ~35% of allocs across every query. Reuse the + set (it's already scratch in `materializeRange`) and avoid `maps.Clone` in the + aggregated-labels path. +3. **`slices.SortStableFunc` instead of `sort.SliceStable`** in `sortEntries` + (logs.go:294) — removes the reflection swapper (~6% CPU), a trivial change. +4. **Reuse window buffers in `logqlmetric.rangeAggIterator`** (`Next`/`fillWindow`, + ~29% of allocs) — the dominant cost of `lines_by_level`/`rate_by_level`; avoid + re-allocating per step/window and re-scanning overlapping windows. +5. **Skip the `pcommon.Map` round-trip in `recordInto`/`SetFromRecord`** — decode + stored attributes into a reused buffer rather than rebuilding pdata Values + (`PutEmpty`/`NewAnyValueStringValue`, ~8% allocs + `AppendAttributes`). + +GreptimeDB answers these same queries in 4–7 ms (columnar SQL with indexed +columns + bounded scans); oteldb's gap is almost entirely (1) materializing more +rows than needed and (2) per-row map/pdata allocation, not the fetch. + +## Reproduce + +``` +# per-query plan (server logs the tree): +curl -s -H 'X-Oteldb-Profile: 1' -G localhost:3100/loki/api/v1/query_range \ + --data-urlencode 'query=sum by (level) (count_over_time({service_name="unknown_service"}[1m]))' \ + --data start= --data end= --data step=30 +docker logs oteldb-bench-oteldb-1 | grep 'EXPLAIN ANALYZE' + +go tool pprof -http=: results/pprof/oteldb.bin results/pprof/logs-cpu.pb.gz +``` diff --git a/results/pprof/findings/logs-investigation-evidence.md b/results/pprof/findings/logs-investigation-evidence.md new file mode 100644 index 0000000..6c68046 --- /dev/null +++ b/results/pprof/findings/logs-investigation-evidence.md @@ -0,0 +1,80 @@ +# oteldb LogQL query performance — investigation (EXPLAIN ANALYZE + pprof) + +Image `ghcr.io/oteldb/oteldb:logql-fix` (embedded `file` backend), ~75k log +records in a 300s window. Profiles: `logs-cpu.pb.gz`, `logs-heap.pb.gz` (+ binary +`oteldb.bin`). Driven by the LogQL suite (`queries/logs.logql.yml`). + +## The one-line finding + +**Storage is not the bottleneck — the LogQL engine is.** EXPLAIN ANALYZE +(`X-Oteldb-Profile: 1`) shows the fetch returns ~75k rows in **3–8 ms**, but the +query spends **80–253 ms of self-time above the fetch**, materializing and +processing every one of those rows. + +``` +lines_by_level query 257.8ms (self 253.2ms) └─ recordengine.fetch 4.5ms rows=74700 series_matched=10 +select_service query 92.5ms (self 84.8ms) └─ recordengine.fetch 7.7ms rows=74700 +status_regex query 84.0ms (self 80.7ms) └─ recordengine.fetch 3.4ms rows=74700 +json_status query 86.1ms (self 82.4ms) └─ recordengine.fetch 3.7ms rows=74700 +``` + +Every query fetches and materializes **all 74700 rows** — even `select_service` +with `limit=1000` (the limit is applied *after* materializing everything). + +## Where the engine self-time goes (CPU, 20s under load) + +- **Per-row materialization** — `storagebackend.logColumns.recordInto` 15% cum, + `signal.AppendAttributes` 4% (decoding each row's attributes back into a + `pcommon.Map`). +- **GC** ~40% cum (`tryDeferToSpanScan` 15%, `scanObjectsSmall`/`scanObject` ~20%, + `mallocgcSmallScanNoHeader` 10%) — driven by the per-row allocations below. +- **Map-based label sets** ~12% — `aeshashbody` 5% + `mapaccess`/`getWithoutKey` + ~5% + `maps.Clone` 2% (the label set is a `map[Label]Value`, rebuilt per row). +- **Metric grouping** — `aggregatedLabels.forEach` 8% (the `sum by (level)` path). +- **Sort via reflection** ~6% — `sort.SliceStable` → `reflectlite.Swapper` + + `typedmemmove` sorting 74700 entries with a reflection-based swapper. + +## What allocates (heap alloc_space) + +`recordInto` 22% · `LabelSet.Set`+`buildSet`+`NewLabelSet`+`AggregatedLabelsFromSet`+ +`aggregatedLabels.By` ≈ **35%** (label sets) · `newRecordCols` 7% · `slices.Grow` +(attr KeyValue) 7% · `pcommon.Map.PutEmpty`+`NewAnyValue*` ≈ 8% (pdata round-trip) · +`rangeAggIterator.Next`+`fillWindow` ≈ **29%** (the metric-window aggregation that +makes `lines_by_level`/`rate_by_level` the slowest at ~240 ms). + +## Fixes, by leverage + +1. **Limit pushdown / lazy materialization.** `EvalPipeline` fetches every record + and `materialize` builds a full label set per row, then truncates to + `params.Limit` (logs.go:96–103). For the filter queries (`select_service`, + `status_regex`, `json_status`, ~85 ms) only `limit` entries are kept — push the + limit into the fetch/materialize so it stops early instead of building 74700 + and discarding 73700. Biggest win for non-metric queries. +2. **Slice-backed / pooled `LabelSet`** instead of `map[Label]Value`. The map per + row (hash + clone) is ~12% CPU and ~35% of allocs across every query. Reuse the + set (it's already scratch in `materializeRange`) and avoid `maps.Clone` in the + aggregated-labels path. +3. **`slices.SortStableFunc` instead of `sort.SliceStable`** in `sortEntries` + (logs.go:294) — removes the reflection swapper (~6% CPU), a trivial change. +4. **Reuse window buffers in `logqlmetric.rangeAggIterator`** (`Next`/`fillWindow`, + ~29% of allocs) — the dominant cost of `lines_by_level`/`rate_by_level`; avoid + re-allocating per step/window and re-scanning overlapping windows. +5. **Skip the `pcommon.Map` round-trip in `recordInto`/`SetFromRecord`** — decode + stored attributes into a reused buffer rather than rebuilding pdata Values + (`PutEmpty`/`NewAnyValueStringValue`, ~8% allocs + `AppendAttributes`). + +GreptimeDB answers these same queries in 4–7 ms (columnar SQL with indexed +columns + bounded scans); oteldb's gap is almost entirely (1) materializing more +rows than needed and (2) per-row map/pdata allocation, not the fetch. + +## Reproduce + +``` +# per-query plan (server logs the tree): +curl -s -H 'X-Oteldb-Profile: 1' -G localhost:3100/loki/api/v1/query_range \ + --data-urlencode 'query=sum by (level) (count_over_time({service_name="unknown_service"}[1m]))' \ + --data start= --data end= --data step=30 +docker logs oteldb-bench-oteldb-1 | grep 'EXPLAIN ANALYZE' + +go tool pprof -http=: results/pprof/oteldb.bin results/pprof/logs-cpu.pb.gz +``` diff --git a/results/pprof/findings/storage-pushdown-proposals.md b/results/pprof/findings/storage-pushdown-proposals.md new file mode 100644 index 0000000..5b852ec --- /dev/null +++ b/results/pprof/findings/storage-pushdown-proposals.md @@ -0,0 +1,163 @@ +# LogQL query performance — storage-level pushdown proposals + +Audience: someone working in `/src/oteldb/storage` (+ the embedded +`internal/storagebackend` adapter in `/src/oteldb/oteldb`). Companion to +`results/pprof/LOGS-FINDINGS.md` (the raw EXPLAIN ANALYZE + pprof evidence). + +## TL;DR + +The LogQL slowness is **engine-side materialization, not the fetch**. EXPLAIN +ANALYZE: the storage fetch returns ~75k rows in **3–8 ms**, but the query spends +**80–253 ms above it** building a label set per row and (for metric queries) +bucketing all samples in Go. Two storage-level pushdowns remove most of that, and +**both extension points already exist in the codebase** — one is implemented for +ClickHouse but not for the embedded engine. + +| Bottleneck (measured) | Pushdown | Status in tree | +|---|---|---| +| Metric queries `lines_by_level`/`rate_by_level` ~240 ms | range-agg step bucketing (`EvalBucketedSample`) | **interface exists; chstorage implements it, embedded does NOT** | +| Filter queries `select_service`/`status_regex` ~85 ms | `Limit` + ordered top-N in the fetch | `fetch.Request` has no `Limit`/order field yet | + +--- + +## Proposal A — push range-aggregation bucketing into the record engine (biggest win) + +**Problem.** `sum by (level) (count_over_time({…}[1m]))` currently streams every +matching record to the engine, which builds a `LabelSet` per row and buckets +samples in Go. Profile: `logqlmetric.rangeAggIterator.Next`/`fillWindow` ≈ **29% +of allocations**; storage fetch is 4.5 ms of the 257 ms. + +**The hook already exists.** `internal/logql/logqlengine/engine_plan.go`: + +```go +// BucketedSampleNode is an optional capability of [SampleNode] implementations +// that can push range-aggregation step-bucketing (e.g. +// sum by(...) (count_over_time(...))) down into the storage layer, instead of +// streaming raw samples for [RangeAggregation] to bucket in Go. +type BucketedSampleNode interface { + EvalBucketedSample(ctx context.Context, params EvalParams, window time.Duration) (StepIterator, error) +} +``` + +- **chstorage already implements it** → `internal/chstorage/querier_logs_node.go` + (oteldb-ch pushes the bucketing into ClickHouse; that's why oteldb-ch logs + metric queries are fast and the embedded engine is ~30× slower). +- **The embedded `logStreamNode` does NOT** (`internal/storagebackend/logs.go`), + so the engine falls back to the raw `SampleNode` path + `rangeAggIterator`. + +**What to build.** Implement `EvalBucketedSample` on the embedded record-engine +path: scan the grouping column(s) (e.g. `level`/`severity`) + timestamps +columnarly, fold into step-aligned trailing windows, and emit one `Step` per +(group, output step) — never materializing a per-row `LabelSet`. Precedent to +mirror is the **metrics vertical**: `storage/engine/aggregate.go` +`AggregateRange` (`aggPushdownSafe`, `bucketSeries`), which already does +step-bucketed aggregation and even short-circuits from the **stats sidecar** when +the plan is pushdown-safe. The record engine wants the analogous +"grouped count/rate over time buckets" fetch mode. + +**Scope of support (start small, expand):** `count_over_time` and `rate` +(rate = count/window) grouped by stored label columns are the common, fully +columnar cases (cover the suite's `lines_by_level`/`rate_by_level`/`error_rate`). +`bytes_over_time`, parser-derived group keys (`| json | …`), and unwrap can stay +on the Go fallback initially. + +**Expected impact:** `lines_by_level`/`rate_by_level` from ~240 ms toward the +single-digit ms the columnar scan implies (GreptimeDB does the equivalent +`GROUP BY severity_text` in ~7 ms). + +--- + +## Proposal B — `Limit` + ordered top-N in the fetch (filter queries) + +**Problem.** `EvalPipeline` (`internal/storagebackend/logs.go:66–106`) fetches +**every** record in the window, `materialize` builds a full `LabelSet` per row, +then it sorts (`sortEntries`, `sort.SliceStable` → reflection swapper) and +truncates to `params.Limit` (lines 102–103). For `select_service` with +`limit=1000` it builds 74 700 label sets and throws away 73 700. + +**Today `fetch.Request` cannot express it.** It has `Matchers`, `Conditions`, +`SecondPass`, `Projection`, `Recycle` — filtering and column projection push +down, but there is **no `Limit` and no scan direction**. Records are stored +timestamp-ordered within a part (timestamp is the part sort key, +`recordengine/schema.go`), so ordered top-N is natural. + +**What to build.** +1. Add `Limit int` + `Reverse bool` (or a `Direction`) to `fetch.Request`. +2. Record engine: scan parts in timestamp order for the requested direction, + apply `Matchers`/`Conditions`/`SecondPass` during the scan, and **stop after + `Limit` surviving rows** (merge across overlapping parts to keep global order). +3. `storagebackend` sets `req.Limit`/direction from `EvalParams.Limit`/`Direction` + when the LogQL pipeline is limit-safe (selector + already-offloaded line + filters; no downstream parser/label-format that changes the result set). + When safe, this also lets the engine **drop the sort** entirely (storage + returns ordered rows). + +**Limit-safety gate.** Only push the limit when every pipeline stage between the +selector and the limit is already offloaded to the fetch (the optimizer knows +this — it sets `n.conditions`/`pipelineLabels`). `{…}` and `{…} |~ "…"` qualify; +`| json | status>=400` does not (the parsed field is consumed after the limit) — +unless that predicate is itself pushed (see Proposal C). + +**Expected impact:** `select_service`/`status_regex` from ~85 ms toward the +~5–8 ms fetch floor (materialize 1 000 instead of 74 700; no reflection sort). + +--- + +## Proposal C (optional) — push the json-status predicate as a SecondPass + +`{…} | json | status>=400` currently parses + filters in the engine, so neither +filtering nor limit can be pushed. `fetch.Request.SecondPass func(*Batch) bool` +already exists for "predicates not expressible as a single-column Match" (it sees +the candidate row's materialized batch). A `| json | ` whose +parsed field is used **only** for filtering could be lowered to a `SecondPass` +over the body column. That drops the rows at the fetch layer and then re-enables +Proposal B's limit pushdown for these queries too. Bigger optimizer change; +lower priority than A and B. + +--- + +## What pushdown does NOT solve (engine-side, complementary) + +For queries that must still return many rows through a non-pushable pipeline, the +per-row cost remains. These are independent of storage and worth doing alongside: + +- **`LabelSet` is a `map[Label]Value`** (`logqlabels/label_set.go`), rebuilt per + row → ~35% of allocs (`Set`/`buildSet`/`AggregatedLabelsFromSet`/`maps.Clone`, + `aeshashbody`). A slice-backed / pooled set helps every non-aggregated query. +- **`recordInto` rebuilds a `pcommon.Map` per row** (`PutEmpty`/`NewAnyValue*`, + `AppendAttributes`, `slices.Grow`) → ~22% allocs. Decode into a reused buffer. +- **`sortEntries` uses `sort.SliceStable`** (reflection swapper, ~6% CPU) → + `slices.SortStableFunc`. Trivial, and moot once Proposal B returns ordered rows. + +Pushdown reduces how often (and over how many rows) these run; the engine fixes +cap the worst case. + +--- + +## Suggested order + +1. **Proposal A** (`EvalBucketedSample` on the embedded path) — largest win, the + interface + a working ClickHouse reference + the metrics `AggregateRange` + precedent already exist. +2. **Proposal B** (`fetch.Request.Limit` + ordered scan) — smaller, self-contained, + also kills the engine sort. +3. Engine-side `LabelSet`/`recordInto`/sort cleanups as a complement. +4. Proposal C if `| json | …`-style filter queries matter. + +## Reproduce / verify + +``` +# per-query plan (set X-Oteldb-Profile: 1, server logs the operator tree): +curl -s -H 'X-Oteldb-Profile: 1' -G localhost:3100/loki/api/v1/query_range \ + --data-urlencode 'query=sum by (level) (count_over_time({service_name="unknown_service"}[1m]))' \ + --data start= --data end= --data step=30 +docker logs oteldb-bench-oteldb-1 | grep 'EXPLAIN ANALYZE' + +go tool pprof -http=: results/pprof/oteldb.bin results/pprof/logs-cpu.pb.gz +# bench just these two engines after a change: +./benchctl --only oteldb,oteldb-ch bench logs && ./benchctl --only oteldb,oteldb-ch report +``` +Key references: `engine_plan.go` (`BucketedSampleNode`), `internal/chstorage/ +querier_logs_node.go` (reference impl), `storage/engine/aggregate.go` +(`AggregateRange` precedent), `internal/storagebackend/logs.go` (`EvalPipeline`, +`materialize`, `sortEntries`), `storage/query/fetch` (`Request`). diff --git a/systems.yml b/systems.yml index ae0cb26..b02505b 100644 --- a/systems.yml +++ b/systems.yml @@ -18,6 +18,7 @@ metrics: # reference language: PromQL (Prometheus HTTP API /api/v1) oteldb: { addr: http://127.0.0.1:9090, lang: promql, apples: true, storage: local } oteldb-ch: { addr: http://127.0.0.1:9092, lang: promql, apples: true, storage: clickhouse, volume: clickhouse-oteldb-data } + oteldb-s3: { addr: http://127.0.0.1:9093, lang: promql, apples: true, storage: fs } prometheus: { addr: http://127.0.0.1:9091, lang: promql, apples: true, storage: local } victoriametrics: { addr: http://127.0.0.1:8428/prometheus, lang: promql, apples: true, storage: local } mimir: { addr: http://127.0.0.1:9009/prometheus, lang: promql, apples: true, storage: fs } @@ -27,13 +28,17 @@ metrics: # reference language: PromQL (Prometheus HTTP A logs: # reference language: LogQL (Loki HTTP API /loki/api/v1) oteldb: { addr: http://127.0.0.1:3100, lang: logql, apples: true, storage: local } oteldb-ch: { addr: http://127.0.0.1:3102, lang: logql, apples: true, storage: clickhouse, volume: clickhouse-oteldb-data } + oteldb-s3: { addr: http://127.0.0.1:3103, lang: logql, apples: true, storage: fs } loki: { addr: http://127.0.0.1:3101, lang: logql, apples: true, storage: fs } + greptimedb: { addr: http://127.0.0.1:4000, lang: sql, apples: false, storage: fs } # native SQL over otel_logs victorialogs: { addr: http://127.0.0.1:9428, lang: logsql, apples: false, storage: local } # native LogsQL gigapipe: { addr: http://127.0.0.1:3110, lang: logql, apples: true, storage: clickhouse } traces: # reference language: TraceQL (Tempo HTTP API /api/search) oteldb: { addr: http://127.0.0.1:3200, lang: traceql, apples: true, storage: local } oteldb-ch: { addr: http://127.0.0.1:3202, lang: traceql, apples: true, storage: clickhouse, volume: clickhouse-oteldb-data } + oteldb-s3: { addr: http://127.0.0.1:3203, lang: traceql, apples: true, storage: fs } tempo: { addr: http://127.0.0.1:3201, lang: traceql, apples: true, storage: fs } + greptimedb: { addr: http://127.0.0.1:4000, lang: sql, apples: false, storage: fs } # native SQL over opentelemetry_traces victoriatraces: { addr: http://127.0.0.1:10428, lang: native, apples: false, storage: local } # Jaeger query API gigapipe: { addr: http://127.0.0.1:3110, lang: traceql, apples: true, storage: clickhouse }