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"`