From 06789db33d0c9892d2fe3f4c99bb68a49ebbf50b Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Fri, 16 Oct 2026 08:15:33 -0700 Subject: [PATCH] feat: record accepted rows in every resource sample Each sample now carries ingest_rows_total, Fanout's cumulative accepted rows by signal from fanout_ingest_rows_total. Consumers can derive per-second ingest rates from consecutive samples, for example to replay the sustained phase second by second. The field is omitted when the metric is absent, and headline tables and gates are unchanged. Claude-Session: https://claude.ai/code/session_01XA52nWyQ3R5f5kK8kjpAQW --- README.md | 6 ++-- docs/metrics.md | 3 +- internal/bench/report.go | 2 +- internal/bench/runner.go | 1 + internal/bench/runner_test.go | 55 +++++++++++++++++++++++++++++++++++ internal/bench/types.go | 1 + 6 files changed, 63 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 2d27d2f..2a8509d 100644 --- a/README.md +++ b/README.md @@ -129,12 +129,12 @@ 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 and background-work samples +├── samples.jsonl per-second Fanout ingest, 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. +`samples.jsonl` includes cumulative `ingest_rows_total` by signal (`spans`, `logs`, `metrics`), `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. Derive per-second accepted-row rates from consecutive samples' `ingest_rows_total` deltas divided by their `at_seconds` difference. 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: @@ -186,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, including rollup durations, write-gate hold/wait durations, compaction duration/count/backlog, publish wait duration, and ingest backpressure duration; +- a raw time series in `samples.jsonl` for independent charting, including accepted rows by signal, 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 7073d81..ef0f8cf 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -38,6 +38,7 @@ Each `samples.jsonl` row also records the following `/metrics` values. The same | Sample field | Source | |---|---| +| `ingest_rows_total` | `fanout_ingest_rows_total`, keyed by `signal` (`spans`, `logs`, `metrics`) | | `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` | @@ -47,7 +48,7 @@ Each `samples.jsonl` row also records the following `/metrics` values. The same | `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. +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. For per-second accepted-row rates, divide each signal's `ingest_rows_total` delta by the difference in `at_seconds` between consecutive samples. `compaction_backlog_batches` is a current gauge rather than a cumulative counter. Series with additional labels are summed, retaining the indicated map key for ingest signals, 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 diff --git a/internal/bench/report.go b/internal/bench/report.go index 740784f..d3010c4 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, 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) + 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 `ingest_rows_total` by signal (`spans`, `logs`, `metrics`), `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 44bea16..a856632 100644 --- a/internal/bench/runner.go +++ b/internal/bench/runner.go @@ -405,6 +405,7 @@ func resourceSample(s *snapshot, at float64) ResourceSample { 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), + IngestRowsTotal: sampleTotalsByLabel(s, "fanout_ingest_rows_total", "signal"), 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"), diff --git a/internal/bench/runner_test.go b/internal/bench/runner_test.go index 6156769..992c3d4 100644 --- a/internal/bench/runner_test.go +++ b/internal/bench/runner_test.go @@ -15,6 +15,61 @@ import ( "time" ) +func TestResourceSampleIngestRowsTotal(t *testing.T) { + for _, tt := range []struct { + name string + fixture string + want map[string]float64 + }{ + { + name: "present", + fixture: `fanout_ingest_rows_total{signal="spans",transport="grpc"} 123 +fanout_ingest_rows_total{signal="spans",transport="http"} 7 +fanout_ingest_rows_total{signal="logs"} 45 +fanout_ingest_rows_total{signal="metrics"} 0 +process_cpu_seconds_total 12 +`, + want: map[string]float64{"spans": 130, "logs": 45, "metrics": 0}, + }, + { + name: "absent", + fixture: "process_cpu_seconds_total 12\n", + }, + } { + t.Run(tt.name, func(t *testing.T) { + snap, err := parsePrometheus(strings.NewReader(tt.fixture)) + if err != nil { + t.Fatal(err) + } + got := resourceSample(snap, 1) + if !reflect.DeepEqual(got.IngestRowsTotal, tt.want) { + t.Fatalf("ingest rows=%v; want %v", got.IngestRowsTotal, tt.want) + } + data, err := json.Marshal(got) + if err != nil { + t.Fatal(err) + } + var fields map[string]json.RawMessage + if err := json.Unmarshal(data, &fields); err != nil { + t.Fatal(err) + } + raw, present := fields["ingest_rows_total"] + if present != (tt.want != nil) { + t.Fatalf("JSON ingest_rows_total present=%t; want %t", present, tt.want != nil) + } + if present { + var decoded map[string]float64 + if err := json.Unmarshal(raw, &decoded); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(decoded, tt.want) { + t.Fatalf("JSON ingest rows=%v; want %v", decoded, tt.want) + } + } + }) + } +} + func TestResourceSampleBackgroundMetrics(t *testing.T) { const fixture = `process_cpu_seconds_total 12 process_resident_memory_bytes 1024 diff --git a/internal/bench/types.go b/internal/bench/types.go index 31a6f62..a27ff1e 100644 --- a/internal/bench/types.go +++ b/internal/bench/types.go @@ -144,6 +144,7 @@ type ResourceSample struct { ParquetBytes float64 `json:"parquet_bytes"` PublishWaiters float64 `json:"publish_waiters"` PublishTimeouts float64 `json:"publish_timeouts_total"` + IngestRowsTotal map[string]float64 `json:"ingest_rows_total,omitempty"` 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"`