From 7f4b78dbd98e07cf847282510813861a900f81d2 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Thu, 15 Oct 2026 08:40:14 -0700 Subject: [PATCH] feat: record Fanout's background work in resource samples A benchmark run could show Fanout slowing down without showing why: samples recorded CPU, memory, queue depth and Parquet files, but nothing about rollups, write gates, backpressure or compaction. Each sample now also carries cumulative rollup durations by component, write-gate hold and wait time by operation, compaction time and count, Parquet publish wait, ingest backpressure time and the compaction backlog gauge. Fields are omitted when a Fanout release does not export them, and the report's headline tables and gates are unchanged. Claude-Session: https://claude.ai/code/session_01XA52nWyQ3R5f5kK8kjpAQW --- README.md | 6 +- docs/metrics.md | 17 ++++++ docs/publishing.md | 2 +- docs/telemetrygen.md | 2 +- internal/bench/report.go | 2 +- internal/bench/runner.go | 28 ++++++++- internal/bench/runner_test.go | 112 ++++++++++++++++++++++++++++++++++ internal/bench/types.go | 26 +++++--- 8 files changed, 180 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index a074c59..2d27d2f 100644 --- a/README.md +++ b/README.md @@ -129,11 +129,13 @@ Every run produces one self-contained directory: results/20261014T190000Z/ ├── report.json authoritative schema-versioned result ├── report.md readable, publication-ready report -├── samples.jsonl per-second Fanout resource samples +├── samples.jsonl per-second Fanout resource and background-work samples ├── fanout-repair.txt post-run authoritative-storage verification └── fanout.log setup-token-scrubbed server log ``` +`samples.jsonl` includes cumulative `rollup_seconds` by rollup (`service`, `edge`, `version`, `read_cache`), `write_gate_hold_seconds` and `write_gate_wait_seconds` by operation, `compaction_seconds_total`, `compactions_total`, `publish_wait_seconds_total`, and `ingest_backpressure_seconds_total`, plus the current `compaction_backlog_batches` gauge. Missing metrics from older Fanout releases leave these fields absent or zero. See [sample definitions](docs/metrics.md#background-work-samples) for their sources and interpretation. + The terminal output is optimized for decisions: ```text @@ -184,7 +186,7 @@ For every phase the report includes: - generator process count/failures and Fanout dropped-row/restart checks; - average Fanout CPU cores, peak/end RSS, peak/end Go heap, allocations, GC pause; - peak ingest queue depth, Parquet file count/bytes, and optional local data-directory growth; -- a raw time series in `samples.jsonl` for independent charting; +- a raw time series in `samples.jsonl` for independent charting, including rollup durations, write-gate hold/wait durations, compaction duration/count/backlog, publish wait duration, and ingest backpressure duration; - attempted/completed read rate, latency, failures, scheduler shedding, and phase-end cancellations when query load is enabled. See [docs/metrics.md](docs/metrics.md) for definitions and [docs/publishing.md](docs/publishing.md) for the public-results checklist. diff --git a/docs/metrics.md b/docs/metrics.md index 0ba66bc..7073d81 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -32,6 +32,23 @@ CPU is the Fanout process only. On a same-host run it does not include telemetry The sampled series above, plus `fanout_rows_dropped_total`, `process_start_time_seconds`, `fanout_parquet_publish_waiters`, and `fanout_parquet_publish_timeouts_total`, come from Fanout v2026.10.1-alpha and its registered Go/process collectors. +## Background work samples + +Each `samples.jsonl` row also records the following `/metrics` values. The same fields appear in `report.json` under `trials[].samples`. + +| Sample field | Source | +|---|---| +| `rollup_seconds` | `fanout_rollup_component_duration_seconds_sum`, keyed by `rollup` (`service`, `edge`, `version`, `read_cache`) | +| `write_gate_hold_seconds` | `fanout_write_gate_hold_seconds_sum`, keyed by `operation` | +| `write_gate_wait_seconds` | `fanout_write_gate_wait_seconds_sum`, keyed by `operation` | +| `compaction_seconds_total` | `fanout_telemetry_operation_duration_seconds_sum` for `operation="compaction"` | +| `compactions_total` | `fanout_telemetry_operation_duration_seconds_count` for `operation="compaction"` | +| `publish_wait_seconds_total` | `fanout_parquet_publish_wait_seconds_sum` | +| `ingest_backpressure_seconds_total` | `fanout_ingest_backpressure_seconds_total` | +| `compaction_backlog_batches` | `fanout_compaction_backlog_batches` | + +Durations and counts are cumulative values at scrape time, not phase deltas. Subtract adjacent samples to attribute work during an interval, accounting for process restarts. `compaction_backlog_batches` is a current gauge rather than a cumulative counter. Series with additional labels are summed, retaining the indicated map key for rollups and write gates. Older Fanout releases may omit these metrics; absent maps and zero scalar values are omitted from JSON, and missing metrics do not fail a run. These samples provide attribution evidence without changing headline tables or quality gates. + ## Generator resources Split trials read `/proc/stat` on the generator before and after each phase using the same SSH target, port, identity, known-hosts file, and connection options as telemetrygen. The interval includes startup and final exports. `report.json` records `trials[].generator_cpu.average_cpu_cores` and `logical_cpus`; `report.md` shows both in each phase row. Same-host and older reports omit this measurement and render it as `n/a`. diff --git a/docs/publishing.md b/docs/publishing.md index 2512dfb..cee491d 100644 --- a/docs/publishing.md +++ b/docs/publishing.md @@ -10,7 +10,7 @@ A performance number is useful only when another engineer can explain and reprod - [ ] Fanout and generator Hetzner host descriptions, Fanout CPU model, architecture, logical CPU count, memory, OS, and generator placement; - [ ] transport, authentication path, batch size, topology, rate, duration, and repetitions; - [ ] average generator-host CPU cores and logical CPUs for each split trial, plus the selected sustained process count per signal; -- [ ] raw `report.json`, `report.md`, and `samples.jsonl` from the same run; +- [ ] raw `report.json`, `report.md`, and `samples.jsonl` from the same run, retaining the [background-work fields](metrics.md#background-work-samples) for rollups, write gates, compaction, publish waits, and ingest backpressure when available; - [ ] three publish-preset repetitions with Fanout alone on a fresh default `ccx33` VM and generators on a separate `ccx33`; - [ ] signal-specific peak results and a sustained mixed result; - [ ] a separate read-under-write result when claiming interactive performance; diff --git a/docs/telemetrygen.md b/docs/telemetrygen.md index e497557..ded5836 100644 --- a/docs/telemetrygen.md +++ b/docs/telemetrygen.md @@ -46,4 +46,4 @@ Let `B` be batch size, `W` workers, `r` requested rate per worker, and `L` time - **Traces:** generation and export overlap through the queue. Long-run throughput is bounded approximately by `min(actual generation rate, B / L)`. Larger `W` can fill the queue faster without increasing export concurrency. Longer acknowledgements lower throughput and can cause queue drops. At 1,000 spans/batch, 13,000 spans/s would correspond to roughly 77 ms per serial export **if export were the active bottleneck**. - **Logs/metrics:** each worker alternates filling a batch and exporting it. Ignoring startup/final partial batches, per-process throughput is approximately `W × B / (actual batch-fill time + L)`. Even with ideal pacing, batch fill is about `B / r`, so the idealized rate is `W × B / (B / r + L)`, rather than a latency-independent `W × r`. Burst-one pacing prevents recovering most tokens missed during export. -The supplied split report's single-process trials accepted approximately 13.5k spans/s, 12.5k records/s and 12.5k points/s. Similar rates across these different export paths cannot be attributed solely to the serialized trace processor. Neither that report nor `samples.jsonl` measures generator pacing or RPC acknowledgement latency, so the exact contribution of each mechanism is unresolved. The wider process sweep and generator CPU measurements provide the next attribution evidence without changing the workload shape. +The supplied split report's single-process trials accepted approximately 13.5k spans/s, 12.5k records/s and 12.5k points/s. Similar rates across these different export paths cannot be attributed solely to the serialized trace processor. Neither that report nor `samples.jsonl` measures generator pacing or RPC acknowledgement latency, so the exact contribution of each mechanism is unresolved. New samples include Fanout's [rollup, write-gate, compaction, publish-wait, and ingest-backpressure metrics](metrics.md#background-work-samples) to help attribute server work. The wider process sweep and generator CPU measurements provide the next attribution evidence without changing the workload shape. diff --git a/internal/bench/report.go b/internal/bench/report.go index 8dcad81..740784f 100644 --- a/internal/bench/report.go +++ b/internal/bench/report.go @@ -186,7 +186,7 @@ func Markdown(r Report) string { for _, warning := range r.Summary.Warnings { fmt.Fprintf(&b, "- %s\n", warning) } - fmt.Fprintf(&b, "\n## Reproduce\n\n```sh\n%s\n```\n\nRaw evidence: `report.json` (authoritative structured report) and `samples.jsonl` (time-series resource samples).\n", r.ReproductionCommand) + fmt.Fprintf(&b, "\n## Reproduce\n\n```sh\n%s\n```\n\nRaw evidence: `report.json` (authoritative structured report) and `samples.jsonl` (time-series resource samples, including cumulative `rollup_seconds`, `write_gate_hold_seconds`, `write_gate_wait_seconds`, `compaction_seconds_total`, `compactions_total`, `publish_wait_seconds_total`, and `ingest_backpressure_seconds_total`, plus the current `compaction_backlog_batches` gauge when available).\n", r.ReproductionCommand) return b.String() } diff --git a/internal/bench/runner.go b/internal/bench/runner.go index 22e8a8c..44bea16 100644 --- a/internal/bench/runner.go +++ b/internal/bench/runner.go @@ -404,10 +404,36 @@ func resourceSample(s *snapshot, at float64) ResourceSample { RSSBytes: s.total("process_resident_memory_bytes", nil), HeapBytes: s.total("go_memstats_heap_alloc_bytes", nil), QueueDepth: s.total("fanout_ingest_queue_depth", nil), ParquetFiles: s.total("fanout_parquet_files", nil), ParquetBytes: s.total("fanout_parquet_size_bytes", nil), PublishWaiters: s.total("fanout_parquet_publish_waiters", nil), - PublishTimeouts: s.total("fanout_parquet_publish_timeouts_total", nil), + PublishTimeouts: s.total("fanout_parquet_publish_timeouts_total", nil), + RollupSeconds: sampleTotalsByLabel(s, "fanout_rollup_component_duration_seconds_sum", "rollup"), + WriteGateHoldSeconds: sampleTotalsByLabel(s, "fanout_write_gate_hold_seconds_sum", "operation"), + WriteGateWaitSeconds: sampleTotalsByLabel(s, "fanout_write_gate_wait_seconds_sum", "operation"), + CompactionSecondsTotal: s.total("fanout_telemetry_operation_duration_seconds_sum", map[string]string{"operation": "compaction"}), + CompactionsTotal: s.total("fanout_telemetry_operation_duration_seconds_count", map[string]string{"operation": "compaction"}), + PublishWaitSecondsTotal: s.total("fanout_parquet_publish_wait_seconds_sum", nil), + IngestBackpressureSecondsTotal: s.total("fanout_ingest_backpressure_seconds_total", nil), + CompactionBacklogBatches: s.total("fanout_compaction_backlog_batches", nil), } } +func sampleTotalsByLabel(s *snapshot, name, label string) map[string]float64 { + var totals map[string]float64 + if s == nil { + return nil + } + for _, sample := range s.samples { + key := sample.labels[label] + if sample.name != name || key == "" { + continue + } + if totals == nil { + totals = make(map[string]float64) + } + totals[key] += sample.value + } + return totals +} + func resourceDelta(before, after *snapshot, samples []ResourceSample, elapsed float64) Resources { result := Resources{ EndRSSBytes: after.total("process_resident_memory_bytes", nil), EndHeapBytes: after.total("go_memstats_heap_alloc_bytes", nil), diff --git a/internal/bench/runner_test.go b/internal/bench/runner_test.go index bd5591f..6156769 100644 --- a/internal/bench/runner_test.go +++ b/internal/bench/runner_test.go @@ -2,10 +2,12 @@ package bench import ( "context" + "encoding/json" "errors" "io" "net/http" "os" + "reflect" "slices" "strings" "sync/atomic" @@ -13,6 +15,116 @@ import ( "time" ) +func TestResourceSampleBackgroundMetrics(t *testing.T) { + const fixture = `process_cpu_seconds_total 12 +process_resident_memory_bytes 1024 +go_memstats_heap_alloc_bytes 512 +fanout_ingest_queue_depth 3 +fanout_parquet_files 4 +fanout_parquet_size_bytes 2048 +fanout_parquet_publish_waiters 2 +fanout_parquet_publish_timeouts_total 1 +fanout_rollup_component_duration_seconds_sum{rollup="service",signal="spans"} 1.25 +fanout_rollup_component_duration_seconds_sum{rollup="service",signal="logs"} 0.75 +fanout_rollup_component_duration_seconds_sum{rollup="edge"} 2.5 +fanout_rollup_component_duration_seconds_sum{rollup="version"} 3.5 +fanout_rollup_component_duration_seconds_sum{rollup="read_cache"} 4.5 +fanout_rollup_component_duration_seconds_count{rollup="service"} 100 +fanout_write_gate_hold_seconds_sum{operation="ingest",signal="spans"} 5 +fanout_write_gate_hold_seconds_sum{operation="ingest",signal="logs"} 0.5 +fanout_write_gate_hold_seconds_sum{operation="compaction"} 6.5 +fanout_write_gate_wait_seconds_sum{operation="ingest",signal="spans"} 7 +fanout_write_gate_wait_seconds_sum{operation="ingest",signal="logs"} 0.5 +fanout_write_gate_wait_seconds_sum{operation="compaction"} 8.5 +fanout_write_gate_wait_seconds_count{operation="ingest"} 100 +fanout_telemetry_operation_duration_seconds_sum{operation="compaction",signal="spans"} 9 +fanout_telemetry_operation_duration_seconds_sum{operation="compaction",signal="logs"} 0.5 +fanout_telemetry_operation_duration_seconds_sum{operation="ingest"} 100 +fanout_telemetry_operation_duration_seconds_count{operation="compaction",signal="spans"} 10 +fanout_telemetry_operation_duration_seconds_count{operation="compaction",signal="logs"} 2 +fanout_telemetry_operation_duration_seconds_count{operation="ingest"} 100 +fanout_parquet_publish_wait_seconds_sum{signal="spans"} 11 +fanout_parquet_publish_wait_seconds_sum{signal="logs"} 0.5 +fanout_ingest_backpressure_seconds_total 12.5 +fanout_compaction_backlog_batches 13 +` + snap, err := parsePrometheus(strings.NewReader(fixture)) + if err != nil { + t.Fatal(err) + } + got := resourceSample(snap, 1.23456) + want := ResourceSample{ + AtSeconds: 1.235, CPUSeconds: 12, RSSBytes: 1024, HeapBytes: 512, + QueueDepth: 3, ParquetFiles: 4, ParquetBytes: 2048, PublishWaiters: 2, PublishTimeouts: 1, + RollupSeconds: map[string]float64{"service": 2, "edge": 2.5, "version": 3.5, "read_cache": 4.5}, + WriteGateHoldSeconds: map[string]float64{"ingest": 5.5, "compaction": 6.5}, + WriteGateWaitSeconds: map[string]float64{"ingest": 7.5, "compaction": 8.5}, + CompactionSecondsTotal: 9.5, + CompactionsTotal: 12, + PublishWaitSecondsTotal: 11.5, + IngestBackpressureSecondsTotal: 12.5, + CompactionBacklogBatches: 13, + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("sample=%+v; want %+v", got, want) + } + // Check the public JSON names as well as the in-memory sample. + data, err := json.Marshal(got) + if err != nil { + t.Fatal(err) + } + var fields map[string]any + if err := json.Unmarshal(data, &fields); err != nil { + t.Fatal(err) + } + for key, want := range map[string]any{ + "rollup_seconds": map[string]any{"service": float64(2), "edge": 2.5, "version": 3.5, "read_cache": 4.5}, + "write_gate_hold_seconds": map[string]any{"ingest": 5.5, "compaction": 6.5}, + "write_gate_wait_seconds": map[string]any{"ingest": 7.5, "compaction": 8.5}, + "compaction_seconds_total": 9.5, + "compactions_total": float64(12), + "publish_wait_seconds_total": 11.5, + "ingest_backpressure_seconds_total": 12.5, + "compaction_backlog_batches": float64(13), + } { + if !reflect.DeepEqual(fields[key], want) { + t.Errorf("JSON %s=%v; want %v", key, fields[key], want) + } + } +} + +func TestResourceSampleWithoutBackgroundMetrics(t *testing.T) { + const fixture = `process_cpu_seconds_total 12 +process_resident_memory_bytes 1024 +go_memstats_heap_alloc_bytes 512 +fanout_ingest_queue_depth 3 +fanout_parquet_files 4 +fanout_parquet_size_bytes 2048 +fanout_parquet_publish_waiters 2 +fanout_parquet_publish_timeouts_total 1 +` + snap, err := parsePrometheus(strings.NewReader(fixture)) + if err != nil { + t.Fatal(err) + } + got := resourceSample(snap, 1.23456) + want := ResourceSample{ + AtSeconds: 1.235, CPUSeconds: 12, RSSBytes: 1024, HeapBytes: 512, + QueueDepth: 3, ParquetFiles: 4, ParquetBytes: 2048, PublishWaiters: 2, PublishTimeouts: 1, + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("legacy sample=%+v; want %+v", got, want) + } + data, err := json.Marshal(got) + if err != nil { + t.Fatal(err) + } + const wantJSON = `{"at_seconds":1.235,"cpu_seconds":12,"rss_bytes":1024,"heap_bytes":512,"queue_depth":3,"parquet_files":4,"parquet_bytes":2048,"publish_waiters":2,"publish_timeouts_total":1}` + if string(data) != wantJSON { + t.Fatalf("legacy JSON=%s; want %s", data, wantJSON) + } +} + func TestRunPhaseStopsQueriesBeforeFinalGeneratorCPURead(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() diff --git a/internal/bench/types.go b/internal/bench/types.go index 0c08dd8..31a6f62 100644 --- a/internal/bench/types.go +++ b/internal/bench/types.go @@ -135,15 +135,23 @@ type Counts struct { func (c Counts) Total() float64 { return c.Spans + c.LogRecords + c.MetricPoints } type ResourceSample struct { - AtSeconds float64 `json:"at_seconds"` - CPUSeconds float64 `json:"cpu_seconds"` - RSSBytes float64 `json:"rss_bytes"` - HeapBytes float64 `json:"heap_bytes"` - QueueDepth float64 `json:"queue_depth"` - ParquetFiles float64 `json:"parquet_files"` - ParquetBytes float64 `json:"parquet_bytes"` - PublishWaiters float64 `json:"publish_waiters"` - PublishTimeouts float64 `json:"publish_timeouts_total"` + AtSeconds float64 `json:"at_seconds"` + CPUSeconds float64 `json:"cpu_seconds"` + RSSBytes float64 `json:"rss_bytes"` + HeapBytes float64 `json:"heap_bytes"` + QueueDepth float64 `json:"queue_depth"` + ParquetFiles float64 `json:"parquet_files"` + ParquetBytes float64 `json:"parquet_bytes"` + PublishWaiters float64 `json:"publish_waiters"` + PublishTimeouts float64 `json:"publish_timeouts_total"` + RollupSeconds map[string]float64 `json:"rollup_seconds,omitempty"` + WriteGateHoldSeconds map[string]float64 `json:"write_gate_hold_seconds,omitempty"` + WriteGateWaitSeconds map[string]float64 `json:"write_gate_wait_seconds,omitempty"` + CompactionSecondsTotal float64 `json:"compaction_seconds_total,omitempty"` + CompactionsTotal float64 `json:"compactions_total,omitempty"` + PublishWaitSecondsTotal float64 `json:"publish_wait_seconds_total,omitempty"` + IngestBackpressureSecondsTotal float64 `json:"ingest_backpressure_seconds_total,omitempty"` + CompactionBacklogBatches float64 `json:"compaction_backlog_batches,omitempty"` } type Latency struct {