diff --git a/README.md b/README.md index 8eeb72b..3b27154 100644 --- a/README.md +++ b/README.md @@ -87,9 +87,9 @@ Nothing is installed system-wide. Existing Fanout services and data directories |---|---|---:|---:|---| | `quick` | 1 topology per signal and mixed | 5s | 10s | installation smoke test | | `standard` | 1, 2, 4, 8 processes per signal | 15s | 60s | engineering comparison | -| `publish` | 1, 2, 4, 8, 16 processes per signal | 30s | 5m | repeatable public evidence | +| `publish` | 1, 2, 4, 8, 16, 32, 64 processes per signal | 30s | 5m | repeatable public evidence | -The default topology is 8 workers per process at 2,500 items/s per worker with 1,000-item OTLP batches. A mixed step launches the same topology independently for traces, logs, and metrics. Without read load, the sustained proof reruns the mixed topology with the highest **accepted** throughput. With `--query-rate`, it chooses the highest-throughput topology that also completed at least 95% of the requested reads with no errors or scheduler shedding. If no sweep point qualifies, it repeats the best read-completion point and fails the sustained-read quality gate rather than publishing a write-only peak as mixed capacity. +The default topology is 8 workers per process at 2,500 items/s per worker with 1,000-item OTLP batches. A mixed step launches the same topology independently for traces, logs, and metrics. Without read load, the sustained proof reruns the mixed topology with the highest **accepted** throughput. With `--query-rate`, it chooses the highest-throughput topology that also completed at least 95% of the requested reads with no errors or scheduler shedding. If no sweep point qualifies, it repeats the best read-completion point and fails the sustained-read quality gate rather than publishing a write-only peak as mixed capacity. Both reports record the selected sustained process count per signal. The wider publish sweep adds independent exporters to get past telemetrygen's per-process limits; see [the source analysis](docs/telemetrygen.md). For publication runs, dedicate the default ccx33 to Fanout and use the separate ccx43 for generators, with three repetitions. Run ingest-only and read-under-write as two groups sharing the same fresh pair of VMs; see [the publication commands](docs/publishing.md#publication-runs). For example: @@ -165,7 +165,7 @@ The terminal output is optimized for decisions: ╰─ PASS — report is internally valid ────────────────────────────────╯ ``` -“Valid” means the evidence is internally trustworthy. It does not mean every offered target was sustained. Each phase is separately classified as `target-sustained`, `near-saturation`, `saturated-or-generator-limited`, or `invalid`. +“Valid” means the evidence is internally trustworthy. It does not mean every offered target was sustained. Each phase is separately classified as `target-sustained`, `generator-limited`, `saturated`, `near-saturation`, `saturated-or-generator-limited`, or `invalid`. Split runs record average generator-host CPU cores and logical CPUs for every trial, from `/proc/stat` reads before and after the phase. After validity and `target-sustained` checks, generator CPU at or above 90% means `generator-limited`; below 90%, an accepted-throughput gain under 10% over the preceding process count for the same signal and repetition means `saturated`. Same-host runs retain the acceptance-ratio fallback. See [metric definitions](docs/metrics.md) for the full rule. Regenerate Markdown without rerunning load: diff --git a/docs/metrics.md b/docs/metrics.md index 6216462..0ba66bc 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -32,6 +32,22 @@ 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. +## 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`. + +Average cores = logical CPUs × (Δtotal − Δidle − Δiowait) / Δtotal. Total is the sum of user, nice, system, idle, iowait, irq, softirq, and steal ticks; guest and guest_nice are already included in user/nice and are excluded from the sum. Logical CPUs are counted from the per-CPU lines. This is whole-generator-host utilisation, including other processes, rather than telemetrygen-only CPU. Failed reads, unusable counter deltas, or changed logical CPU counts stop the run with a partial report; they cannot masquerade as idle CPU. + +## Limit classification + +Invalid trials (generator failures, dropped rows, restarts, or no accepted data) take precedence. Accepting at least 95% of the offered count remains `target-sustained`. For other split trials: + +- Generator CPU averaging **at least 90%** of its logical CPUs means `generator-limited`. +- With generator CPU **below 90%**, a sweep whose accepted items/s rose **less than 10%** over the preceding process count is `saturated`, including a decline. Compare only the same signal and repetition; an invalid or zero-throughput previous point cannot establish saturation. Warmup and sustained phases do not establish sweep plateaus. +- Otherwise retain the acceptance-ratio fallback: at least 80% is `near-saturation`; below 80% is `saturated-or-generator-limited`. The latter can still occur while split throughput is scaling or no valid prior point is available. Low CPU alone cannot rule out acknowledgement latency or other generator limits; see [telemetrygen's export paths](telemetrygen.md). + +Same-host runs keep the existing fallback: at least 95% is `target-sustained`, at least 80% is `near-saturation`, otherwise `saturated-or-generator-limited`. Saturation describes the measured plateau under this workload; it is not proof of a CPU-only server bottleneck. + ## Read workload Authenticated reads rotate equally across user operations `dashboard-refresh`, `drill-traces` (System overview service-map selection), and `trace-detail`, at 15m, 1h, 6h, 12h, and 24h. A refresh follows `page.tsx`, `use-panel-results.ts`, and `refresh.ts`: one sorted batch of all initially visible panel IDs with the full dashboard spec, range and refresh setting, service All, comparison per spec, and layout widths rounded to 100 pixels for a fixed 1664-pixel viewport. It then sends `POST /api/annotations` with the union of the returned time-panel bounds, when annotations are enabled and those bounds are present. diff --git a/docs/publishing.md b/docs/publishing.md index 9ea8a5a..642f0b0 100644 --- a/docs/publishing.md +++ b/docs/publishing.md @@ -9,6 +9,7 @@ A performance number is useful only when another engineer can explain and reprod - [ ] exact telemetrygen version; - [ ] 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; - [ ] three publish-preset repetitions with Fanout alone on a fresh default `ccx33` VM and generators on a separate `ccx43`; - [ ] signal-specific peak results and a sustained mixed result; @@ -42,6 +43,8 @@ The second sentence discards the machine, duration, signal mix, generator placem ## Publication runs +The publish sweep uses 1, 2, 4, 8, 16, 32, and 64 processes per signal, each with 8 workers at 2,500 items/s and 1,000-item batches. Sweep steps last 30 seconds after a 15-second warmup; the selected mixed topology then runs for five minutes. Reports state the sustained process count per signal. Selection still uses highest accepted mixed throughput meeting the read target, with the best read-completion fallback if none qualifies. Generator CPU and sweep scaling determine [limit attribution](metrics.md#limit-classification). + Every publication run uses one invocation of `scripts/hetzner.sh` with Fanout alone on the default Hetzner Cloud `ccx33` (8 dedicated vCPUs, 32 GB RAM) and generators on a separate `ccx43` (16 dedicated vCPUs, 64 GB RAM), both in `fsn1` on `ubuntu-24.04`. The two VMs share a private network (`10.77.0.0/16`, cloud subnet `10.77.0.0/24`) in the location’s network zone. Public IPs serve provisioning; generators send OTLP to Fanout over private IPs. `FANOUT_BENCH_HCLOUD_TYPE` and `FANOUT_BENCH_HCLOUD_GENERATOR_TYPE` override the Fanout and generator types respectively; location, image, and SSH-key settings apply to both. Use the publish preset with three repetitions in each group. Separate the ingest-only and read-under-write groups with a standalone `--` so both groups share the fresh pair of VMs and network while each gets fresh stages and a Fanout data directory: ```sh @@ -58,7 +61,7 @@ scripts/hetzner.sh \ The script uses the active `hcloud` context or `HCLOUD_TOKEN`, records both VM types, location, and image alongside the detected Fanout hardware, rebuilds the harness before provisioning, and waits for SSH on both VMs. Groups must be nonempty and use matching SSH port and identity settings; they share a temporary known-hosts file and both host descriptions. Every group runs even if an earlier one fails, downloads its own evidence, and receives a one-line summary at the end; any failed group makes the invocation exit nonzero. The script attempts to delete both servers and then the network after the last group and on early exits, including Ctrl-C. Failed deletion after three attempts prints each resource name for manual cleanup and exits nonzero. Hetzner bills every started hour on both VMs. The hourly VM cost is **ccx33 hourly rate + ccx43 hourly rate** for the chosen location, plus applicable public IPv4 charges and tax; a run crossing an hour boundary incurs another started hour on each VM. Confirm current rates before provisioning; grouping the runs shares the billing time of both machines. -The local provisioner detects both Linux platforms, builds the harness for Fanout and telemetrygen for the generator, and uploads only the binaries needed by each host. It generates an ephemeral ed25519 key, appends the public key to the generator’s root authorized keys with a unique marker, and puts the private key (`0600`) in the Fanout stage (`0700`). `managed` on Fanout owns setup, lifecycle, sampling, reads, storage verification, and reporting. Only telemetrygen launches move to the generator, through private-network SSH with a bounded remote timeout. Cleanup removes the marked key and both stages. Reports record the generator description in the host section and `generator_on_target_host: false`; reproduction commands redact both public targets and omit the private addresses. +The local provisioner detects both Linux platforms, builds the harness for Fanout and telemetrygen for the generator, and uploads only the binaries needed by each host. It generates an ephemeral ed25519 key, appends the public key to the generator’s root authorized keys with a unique marker, and puts the private key (`0600`) in the Fanout stage (`0700`). `managed` on Fanout owns setup, lifecycle, sampling, reads, storage verification, and reporting. Telemetrygen runs on the generator through private-network SSH with a bounded remote timeout; generator CPU reads reuse that SSH setup before and after each phase. Cleanup removes the marked key and both stages. Reports record the generator description in the host section and `generator_on_target_host: false`; reproduction commands redact both public targets and omit the private addresses. A plain `fanout-bench run --host root@` without a generator host remains a same-host run on any clean Linux host. It is useful for diagnostics, but generator CPU contention limits what it says about Fanout’s own ingest ceiling. diff --git a/docs/telemetrygen.md b/docs/telemetrygen.md new file mode 100644 index 0000000..e497557 --- /dev/null +++ b/docs/telemetrygen.md @@ -0,0 +1,49 @@ +# telemetrygen v0.157.0 throughput limits + +The configured 20,000 items/s per process is an approximate generation target, not an achieved send rate. There is no hard-coded 13,000 items/s ceiling. The source has per-item timer pacing, a single export lane for traces, and synchronous export pauses for log and metric workers. Adding processes adds independent pacing loops and exporters, which explains why the split run can scale while Fanout still has spare CPU. + +This analysis uses the supplied local module-cache source, including its pinned dependencies: OpenTelemetry Go trace SDK and trace/metric exporters v1.44.0, log exporter v0.20.0, and `golang.org/x/time` v0.15.0 (`telemetrygen/go.mod:12–28`). References below are relative to those module roots. + +## Rate and batching + +`--workers` creates generator goroutines. Each gets the full `--rate`, so eight workers at 2,500 request 20,000 items/s. Each worker creates a token limiter with **burst 1** and waits for each item. A trace with one child consumes two tokens: one for the parent and one for the child; the rate already counts spans. `--batch-size 1000` batches delivery but does not batch limiter reservations. + +At 2,500/s, token spacing is 400 microseconds. `rate.Wait` uses a Go timer when it must wait, and accumulated tokens are capped at one. Timer wakeup and scheduler delays can therefore reduce generation below the requested rate even on a host with spare CPU; a late worker cannot catch up with a large burst. The source establishes this mechanism, but does not quantify how much of the observed shortfall came from pacing versus export latency. + +| Source | Evidence | +|---|---| +| telemetrygen `internal/config/config.go:264–265,306–307,334` | Per-worker rate, worker count, batch flag/size; batching defaults on | +| telemetrygen `pkg/traces/traces.go:148–167` | One trace goroutine per worker, each given the configured rate | +| telemetrygen `pkg/traces/worker.go:100,107,142` | Burst-one limiter and separate parent/child waits | +| telemetrygen `pkg/logs/worker.go:49,93` | Burst-one limiter and one wait per record | +| telemetrygen `pkg/metrics/worker.go:100,219` | Burst-one limiter and one wait per metric | +| x/time `rate/rate.go:241–249,274–289,393–399` | Timer-based waits and burst cap | + +## Export concurrency and drops + +**Traces:** all workers share one `BatchSpanProcessor` and one OTLP exporter. The processor collects up to 1,000 spans or waits up to one second, then calls the exporter synchronously from its single queue-processing goroutine. The gRPC call waits for the receiver's response before returning. Eight workers do not create eight concurrent trace exports. + +The trace SDK queue defaults to **2,048 spans** (unless overridden through the SDK environment). telemetrygen sets the batch size and timeout but never enables `WithBlocking`. When the queue fills, `enqueueDrop` increments its dropped counter and discards spans instead of blocking generation. These losses happen before Fanout and do not increment Fanout's dropped-row counter. A successful telemetrygen exit therefore does not prove every generated span reached Fanout. + +**Logs and metrics:** each worker constructs its own exporter and buffer. Reaching the batch size calls `Export` inline; generation in that worker pauses until export completes. Exporters serialize calls within each exporter, but exporters belong to different workers, so a process can have up to eight concurrent log or metric exports. These paths call exporters directly: no SDK batch-processor queue is involved and there is no corresponding SDK queue-full drop. Remaining buffered items are flushed at the end. + +| Source | Evidence | +|---|---| +| telemetrygen `pkg/traces/traces.go:58,71–98` | One exporter and shared processor; batch timeout and size; no blocking option | +| trace SDK `trace/batch_span_processor.go:22,98–112,126–129` | Default queue size and one processor goroutine | +| trace SDK `trace/batch_span_processor.go:285–309,321–355` | Synchronous batch exports, then queue processing resumes | +| trace SDK `trace/batch_span_processor.go:392–398,417–431` | Default nonblocking queue-full drops | +| trace exporter `exporter.go:30–40`; trace gRPC exporter `client.go:232–249` | Conversion, upload, synchronous gRPC response and retry handling | +| telemetrygen `pkg/logs/logs.go:72–106`; `pkg/metrics/metrics.go:70–107` | Exporter created separately for each worker | +| telemetrygen `pkg/logs/worker.go:115–152`; `pkg/metrics/worker.go:241–280` | Synchronous batch export, buffer reset and final flush | +| log gRPC exporter `exporter.go:61–73`, `client.go:171–187` | Exporter lock and synchronous RPC/retry | +| metric gRPC exporter `exporter.go:92–106`, `client.go:129–145` | Exporter lock and synchronous RPC/retry | + +## Acknowledgement latency + +Let `B` be batch size, `W` workers, `r` requested rate per worker, and `L` time spent exporting a full batch, including encoding, transport, server acknowledgement and any retries. + +- **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. diff --git a/internal/bench/generator_cpu.go b/internal/bench/generator_cpu.go new file mode 100644 index 0000000..c2d1b0e --- /dev/null +++ b/internal/bench/generator_cpu.go @@ -0,0 +1,85 @@ +package bench + +import ( + "context" + "fmt" + "strconv" + "strings" + "time" +) + +type cpuStat struct { + total uint64 + idle uint64 + logicalCPUs int +} + +func (r *Runner) generatorCPUStat(ctx context.Context) (cpuStat, error) { + if r.readGeneratorCPU != nil { + return r.readGeneratorCPU(ctx) + } + ctx, cancel := context.WithTimeout(ctx, 15*time.Second) + defer cancel() + cmd := generatorSSHCommand(ctx, r.Config, "cat /proc/stat") + output, err := cmd.Output() + if err != nil { + return cpuStat{}, err + } + return parseCPUStat(string(output)) +} + +func parseCPUStat(data string) (cpuStat, error) { + var stat cpuStat + found := false + for _, line := range strings.Split(data, "\n") { + fields := strings.Fields(line) + if len(fields) == 0 { + continue + } + if fields[0] != "cpu" { + if suffix, ok := strings.CutPrefix(fields[0], "cpu"); ok { + if n, err := strconv.Atoi(suffix); err == nil && n >= 0 { + stat.logicalCPUs++ + } + } + continue + } + if len(fields) < 5 { + return cpuStat{}, fmt.Errorf("/proc/stat CPU line has fewer than four counters") + } + found = true + // user, nice, system, idle, iowait, irq, softirq, steal. + // guest and guest_nice are already included in user and nice. + for i := 1; i < len(fields) && i <= 8; i++ { + value, err := strconv.ParseUint(fields[i], 10, 64) + if err != nil { + return cpuStat{}, fmt.Errorf("/proc/stat CPU counter: %w", err) + } + stat.total += value + if i == 4 || i == 5 { + stat.idle += value + } + } + } + if !found || stat.logicalCPUs == 0 { + return cpuStat{}, fmt.Errorf("/proc/stat is missing aggregate CPU counters or logical CPUs") + } + return stat, nil +} + +func generatorCPUUsage(before, after cpuStat) (*GeneratorCPU, error) { + if before.logicalCPUs <= 0 || before.logicalCPUs != after.logicalCPUs { + return nil, fmt.Errorf("generator logical CPU count missing or changed during trial") + } + if after.total <= before.total || after.idle < before.idle { + return nil, fmt.Errorf("generator CPU counters did not advance or moved backwards") + } + total, idle := after.total-before.total, after.idle-before.idle + if idle > total { + return nil, fmt.Errorf("generator idle CPU delta exceeds total CPU delta") + } + return &GeneratorCPU{ + AverageCPUCores: float64(before.logicalCPUs) * float64(total-idle) / float64(total), + LogicalCPUs: before.logicalCPUs, + }, nil +} diff --git a/internal/bench/generator_cpu_test.go b/internal/bench/generator_cpu_test.go new file mode 100644 index 0000000..cef7d3a --- /dev/null +++ b/internal/bench/generator_cpu_test.go @@ -0,0 +1,67 @@ +package bench + +import ( + "context" + "math" + "slices" + "testing" +) + +func TestGeneratorCPUUsageFromProcStat(t *testing.T) { + before, err := parseCPUStat("cpu 100 20 30 400 50 6 7 8 90 10\ncpu0 0\ncpu1 0\nintr 123\n") + if err != nil { + t.Fatal(err) + } + // Guest fields must not contribute to either total; iowait is idle. + if before.total != 621 || before.idle != 450 || before.logicalCPUs != 2 { + t.Fatalf("baseline=%+v", before) + } + after, err := parseCPUStat("cpu 160 20 50 410 55 8 9 9 150 50\ncpu0 0\ncpu1 0\n") + if err != nil { + t.Fatal(err) + } + usage, err := generatorCPUUsage(before, after) + if err != nil { + t.Fatal(err) + } + if usage.LogicalCPUs != 2 || math.Abs(usage.AverageCPUCores-1.7) > 1e-12 { + t.Fatalf("usage=%+v want 1.7 / 2 cores", usage) + } + for _, idle := range []uint64{0, 100} { + usage, err := generatorCPUUsage(cpuStat{logicalCPUs: 16}, cpuStat{total: 100, idle: idle, logicalCPUs: 16}) + if err != nil || usage.AverageCPUCores != 16*(1-float64(idle)/100) { + t.Fatalf("idle=%d usage=%+v error=%v", idle, usage, err) + } + } +} + +func TestGeneratorCPURejectsUnusableSamples(t *testing.T) { + for _, data := range []string{"", "cpu 1 2 3\ncpu0 0", "cpu 1 bad 3 4\ncpu0 0", "cpu 1 2 3 4", "cpu0 1 2 3 4", "cpu -1 2 3 4\ncpu0 0"} { + if _, err := parseCPUStat(data); err == nil { + t.Errorf("accepted malformed stat %q", data) + } + } + before := cpuStat{total: 100, idle: 50, logicalCPUs: 2} + for _, after := range []cpuStat{ + before, + {total: 99, idle: 50, logicalCPUs: 2}, + {total: 110, idle: 49, logicalCPUs: 2}, + {total: 110, idle: 61, logicalCPUs: 2}, + {total: 110, idle: 50, logicalCPUs: 3}, + {total: 110, idle: 50}, + } { + if _, err := generatorCPUUsage(before, after); err == nil { + t.Errorf("accepted unusable delta: %+v -> %+v", before, after) + } + } +} + +func TestGeneratorCPUCommandUsesGeneratorSSHSetup(t *testing.T) { + cfg := Config{GeneratorSSH: "root@10.77.0.3", GeneratorSSHPort: 2222, GeneratorSSHIdentity: "/stage/key", GeneratorSSHKnownHosts: "/stage/known_hosts"} + load, cancel := generatorCommand(context.Background(), cfg, 0, "--version") + defer cancel() + stat := generatorSSHCommand(context.Background(), cfg, "cat /proc/stat") + if !slices.Equal(load.Args[:len(load.Args)-1], stat.Args[:len(stat.Args)-1]) || stat.Args[len(stat.Args)-1] != "cat /proc/stat" { + t.Fatalf("CPU command does not reuse generator connection setup: %q", stat.Args) + } +} diff --git a/internal/bench/preset.go b/internal/bench/preset.go index 032da90..6e26c11 100644 --- a/internal/bench/preset.go +++ b/internal/bench/preset.go @@ -27,7 +27,7 @@ func Plan(cfg Config) ([]PhaseSpec, error) { case "standard": s = shape{[]int{1, 2, 4, 8}, 8, 2_500, 15 * time.Second, 60 * time.Second, 10 * time.Second} case "publish": - s = shape{[]int{1, 2, 4, 8, 16}, 8, 2_500, 30 * time.Second, 5 * time.Minute, 15 * time.Second} + s = shape{[]int{1, 2, 4, 8, 16, 32, 64}, 8, 2_500, 30 * time.Second, 5 * time.Minute, 15 * time.Second} default: return nil, fmt.Errorf("unknown preset %q (want quick, standard, or publish)", cfg.Preset) } diff --git a/internal/bench/preset_test.go b/internal/bench/preset_test.go index b461d6f..160dadb 100644 --- a/internal/bench/preset_test.go +++ b/internal/bench/preset_test.go @@ -1,10 +1,47 @@ package bench import ( + "slices" "testing" "time" ) +func TestPublishPlanExtendsEverySweepAndKeepsShape(t *testing.T) { + plan, err := Plan(Config{Preset: "publish", Repeat: 2, BatchSize: 1000, QueryRate: 20}) + if err != nil { + t.Fatal(err) + } + if len(plan) != 58 { + t.Fatalf("phases=%d want 58", len(plan)) + } + for repetition := 1; repetition <= 2; repetition++ { + for _, signal := range []Signal{Traces, Logs, Metrics, Mixed} { + var processes []int + for _, phase := range plan { + if phase.Kind != "sweep" || phase.Repetition != repetition || phase.Signal != signal { + continue + } + processes = append(processes, phase.ProcessesPerSignal) + if phase.WorkersPerProcess != 8 || phase.RatePerWorker != 2500 || phase.BatchSize != 1000 || phase.Duration != 30*time.Second || phase.DurationSeconds != 30 { + t.Fatalf("publish shape changed: %+v", phase) + } + if (signal == Mixed && phase.QueryRate != 20) || (signal != Mixed && phase.QueryRate != 0) { + t.Fatalf("query load changed: %+v", phase) + } + } + if !slices.Equal(processes, []int{1, 2, 4, 8, 16, 32, 64}) { + t.Errorf("%s repetition %d: processes=%v", signal, repetition, processes) + } + } + } + if p := plan[0]; p.Kind != "warmup" || p.Duration != 15*time.Second || p.ProcessesPerSignal != 1 { + t.Fatalf("warmup changed: %+v", p) + } + if p := plan[len(plan)-1]; p.Kind != "sustained" || p.Signal != Mixed || p.Duration != 5*time.Minute || p.ProcessesPerSignal != 0 || p.QueryRate != 20 { + t.Fatalf("sustained selection or shape changed: %+v", p) + } +} + func TestStandardPlanIsDeterministicAndBalanced(t *testing.T) { cfg := Config{Preset: "standard", Repeat: 1, BatchSize: 1000} got, err := Plan(cfg) diff --git a/internal/bench/report.go b/internal/bench/report.go index 6dc6d11..8dcad81 100644 --- a/internal/bench/report.go +++ b/internal/bench/report.go @@ -145,6 +145,11 @@ func Markdown(r Report) string { fmt.Fprintf(&b, "| %s | %s | %s | %.2f cores | %s |\n", row.name, formatNumber(row.peak.AcceptedPerSecond), formatNumber(row.peak.OfferedPerSecond), row.peak.AverageCPUCores, formatBytes(row.peak.PeakRSSBytes)) } fmt.Fprintf(&b, "\n**Meaning:** accepted throughput is the delta of `fanout_ingest_rows_total` divided by the interval between the two counter scrapes that bracket the phase. It is not telemetrygen's requested or locally generated count.\n\n") + for _, t := range r.Trials { + if t.Spec.Kind == "sustained" { + fmt.Fprintf(&b, "**Sustained mixed topology:** %d processes per signal, %d workers per process.\n\n", t.Spec.ProcessesPerSignal, t.Spec.WorkersPerProcess) + } + } fmt.Fprintf(&b, "## Test identity\n\n| Setting | Value |\n|---|---|\n") machine := display(r.Host.CPUModel) if r.Host.Description != "" { @@ -154,9 +159,13 @@ func Markdown(r Report) string { for _, row := range settings { fmt.Fprintf(&b, "| %s | %s |\n", row[0], row[1]) } - fmt.Fprintf(&b, "\n## Complete phase breakdown\n\n| Phase | Signal | Topology | Duration | Offered/s | Accepted/s | Ratio | CPU | Peak RSS | Queue peak | Result |\n|---|---|---:|---:|---:|---:|---:|---:|---:|---:|---|\n") + fmt.Fprintf(&b, "\n## Complete phase breakdown\n\n| Phase | Signal | Topology (processes per signal × workers) | Duration | Offered/s | Accepted/s | Ratio | Fanout CPU cores | Generator CPU cores / logical CPUs | Peak RSS | Queue peak | Result |\n|---|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---|\n") for _, t := range r.Trials { - fmt.Fprintf(&b, "| %s | %s | %d×%d workers | %.1fs | %s | %s | %.1f%% | %.2f | %s | %.0f | %s |\n", t.Spec.Name, t.Spec.Signal, t.Spec.ProcessesPerSignal, t.Spec.WorkersPerProcess, t.ElapsedSeconds, formatNumber(t.OfferedPerSecond), formatNumber(t.AcceptedPerSecond), t.AcceptanceRatio*100, t.Resources.AverageCPUCores, formatBytes(t.Resources.PeakRSSBytes), t.Resources.PeakQueueDepth, t.Classification) + generatorCPU := "n/a" + if t.GeneratorCPU != nil { + generatorCPU = fmt.Sprintf("%.2f / %d", t.GeneratorCPU.AverageCPUCores, t.GeneratorCPU.LogicalCPUs) + } + fmt.Fprintf(&b, "| %s | %s | %d×%d workers | %.1fs | %s | %s | %.1f%% | %.2f | %s | %s | %.0f | %s |\n", t.Spec.Name, t.Spec.Signal, t.Spec.ProcessesPerSignal, t.Spec.WorkersPerProcess, t.ElapsedSeconds, formatNumber(t.OfferedPerSecond), formatNumber(t.AcceptedPerSecond), t.AcceptanceRatio*100, t.Resources.AverageCPUCores, generatorCPU, formatBytes(t.Resources.PeakRSSBytes), t.Resources.PeakQueueDepth, t.Classification) } fmt.Fprintf(&b, "\n## Quality gates\n\n") for _, gate := range r.Summary.Gates { diff --git a/internal/bench/report_test.go b/internal/bench/report_test.go index 22da3fa..ccd8d83 100644 --- a/internal/bench/report_test.go +++ b/internal/bench/report_test.go @@ -14,6 +14,47 @@ func fixtureReport() Report { return Report{SchemaVersion: SchemaVersion, RunID: "test", StartedAt: time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC), Build: Build{HarnessVersion: "v1", HarnessCommit: "abc", TelemetrygenVersion: "v0.157.0", FanoutVersion: "v2026.10.1-alpha"}, Host: Host{OS: "linux", Architecture: "amd64", CPUModel: "AMD EPYC", LogicalCPUs: 8, MemoryBytes: 32 << 30, GeneratorLocal: true}, Settings: Settings{Preset: "quick", Transport: "plaintext gRPC", BatchSize: 1000, SampleSeconds: 1, Methodology: "accepted"}, WorkloadSHA256: strings.Repeat("a", 64), Trials: []Trial{{Spec: PhaseSpec{Name: "traces", Signal: Traces, Kind: "sweep", ProcessesPerSignal: 1, WorkersPerProcess: 2}, ElapsedSeconds: 5, OfferedPerSecond: 2000, AcceptedPerSecond: 1900, AcceptanceRatio: .95, Accepted: Counts{Spans: 9500}, Resources: Resources{AverageCPUCores: 1.2, PeakRSSBytes: 100 << 20}, Classification: "target-sustained"}}, Summary: Summary{PeakTraces: Peak{"traces", 2000, 1900, 1.2, 100 << 20}, Passed: true, Gates: []Gate{{"counters", true, "ok"}}, Warnings: []string{"synthetic"}}, ReproductionCommand: "FANOUT_BENCH_INGEST_TOKEN=… fanout-bench run"} } +func TestGeneratorCPUAndSustainedTopologyArtifacts(t *testing.T) { + report := fixtureReport() + report.Host.GeneratorLocal = false + report.Trials[0].GeneratorCPU = &GeneratorCPU{AverageCPUCores: 14.4, LogicalCPUs: 16} + sustained := report.Trials[0] + sustained.Spec = PhaseSpec{Name: "mixed-sustained", Kind: "sustained", Signal: Mixed, ProcessesPerSignal: 32, WorkersPerProcess: 8} + report.Trials = append(report.Trials, sustained) + dir := t.TempDir() + if err := WriteArtifacts(dir, report); err != nil { + t.Fatal(err) + } + loaded, err := LoadReport(filepath.Join(dir, "report.json")) + if err != nil { + t.Fatal(err) + } + if cpu := loaded.Trials[0].GeneratorCPU; cpu == nil || cpu.AverageCPUCores != 14.4 || cpu.LogicalCPUs != 16 { + t.Fatalf("CPU measurement lost: %+v", cpu) + } + if loaded.Trials[1].Spec.ProcessesPerSignal != 32 { + t.Fatal("sustained process count missing from JSON") + } + data, err := os.ReadFile(filepath.Join(dir, "report.md")) + if err != nil { + t.Fatal(err) + } + for _, want := range []string{"Generator CPU cores / logical CPUs", "14.40 / 16", "**Sustained mixed topology:** 32 processes per signal, 8 workers per process."} { + if !strings.Contains(string(data), want) { + t.Errorf("Markdown missing %q", want) + } + } + // Older and same-host reports have no generator CPU measurement. + legacy := fixtureReport() + data, err = json.Marshal(legacy) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(data), "generator_cpu") || !strings.Contains(Markdown(legacy), "| n/a |") { + t.Fatal("missing generator CPU was presented as measured") + } +} + func TestArtifactsAreCompleteAndContainNoSecrets(t *testing.T) { dir := t.TempDir() report := fixtureReport() diff --git a/internal/bench/runner.go b/internal/bench/runner.go index b02b532..22e8a8c 100644 --- a/internal/bench/runner.go +++ b/internal/bench/runner.go @@ -22,13 +22,14 @@ import ( ) type Runner struct { - Config Config - Version string - Out io.Writer - DeferSummary bool - client *http.Client - queryTargets []queryTarget - readSummary string + Config Config + Version string + Out io.Writer + DeferSummary bool + client *http.Client + queryTargets []queryTarget + readSummary string + readGeneratorCPU func(context.Context) (cpuStat, error) } func (r *Runner) Run(ctx context.Context) (*Report, error) { @@ -86,7 +87,7 @@ func (r *Runner) Run(ctx context.Context) (*Report, error) { plan[i] = spec } fmt.Fprintf(r.Out, "\n %s %-18s offered %s items/s for %s\n", phaseIcon(spec.Kind), spec.Name, formatNumber(spec.OfferedPerSecond()), spec.Duration) - trial, err := r.runPhase(ctx, spec) + trial, err := r.runPhase(ctx, spec, report.Trials) if err != nil { return r.abandon(report, fmt.Errorf("phase %s: %w", spec.Name, err)) } @@ -166,7 +167,15 @@ func (r *Runner) validate() error { return nil } -func (r *Runner) runPhase(ctx context.Context, spec PhaseSpec) (Trial, error) { +func (r *Runner) runPhase(ctx context.Context, spec PhaseSpec, previous []Trial) (Trial, error) { + var generatorBefore cpuStat + if r.Config.GeneratorSSH != "" { + var err error + generatorBefore, err = r.generatorCPUStat(ctx) + if err != nil { + return Trial{}, fmt.Errorf("baseline generator CPU: %w", err) + } + } before, err := r.scrape(ctx) if err != nil { return Trial{}, fmt.Errorf("baseline metrics: %w", err) @@ -202,11 +211,23 @@ func (r *Runner) runPhase(ctx context.Context, spec PhaseSpec) (Trial, error) { // duration and stays the denominator for completed queries. window := time.Since(started).Seconds() cancel() - samples := <-samplesDone queries.finish() + samples := <-samplesDone + var generatorCPU *GeneratorCPU + var cpuErr error + if r.Config.GeneratorSSH != "" { + var generatorAfter cpuStat + generatorAfter, cpuErr = r.generatorCPUStat(ctx) + if cpuErr == nil { + generatorCPU, cpuErr = generatorCPUUsage(generatorBefore, generatorAfter) + } + } if err != nil { return Trial{}, fmt.Errorf("final metrics: %w", err) } + if cpuErr != nil { + return Trial{}, fmt.Errorf("final generator CPU: %w", cpuErr) + } afterDisk, _ := directorySize(r.Config.DataDir) accepted := subtractCounts(acceptedCounts(after), acceptedCounts(before)) if accepted.Spans < 0 || accepted.LogRecords < 0 || accepted.MetricPoints < 0 { @@ -228,13 +249,14 @@ func (r *Runner) runPhase(ctx context.Context, spec PhaseSpec) (Trial, error) { OfferedPerSecond: spec.OfferedPerSecond(), Accepted: accepted, AcceptedPerSecond: perSecond(accepted.Total(), window), AcceptanceRatio: ratio, RowsDropped: dropped, GeneratorProcesses: processCount, GeneratorFailures: failures, + GeneratorCPU: generatorCPU, ProcessRestarted: restarted, Resources: resources, QueryLatency: queries.hist.snapshot(), QueryAttempts: queries.attempts, QueryFailures: queries.failures, QueryKindCompleted: queries.completedByKind(), QueryShed: queries.shed, QueryCanceled: queries.canceled, Samples: samples, } - trial.Classification, trial.ClassificationNotes = classify(trial) + trial.Classification, trial.ClassificationNotes = classify(trial, previous) r.readSummary = queries.operationSummary(spec.Name) if trial.Spec.QueryRate > 0 && !queryTargetMet(trial) { trial.ClassificationNotes = append(trial.ClassificationNotes, "the requested query rate was not sustained") @@ -304,6 +326,10 @@ func generatorCommand(ctx context.Context, cfg Config, duration time.Duration, a for i := range remote { remote[i] = shellQuote(remote[i]) } + return generatorSSHCommand(ctx, cfg, strings.Join(remote, " ")), cancel +} + +func generatorSSHCommand(ctx context.Context, cfg Config, command string) *exec.Cmd { port := cfg.GeneratorSSHPort if port == 0 { port = 22 @@ -311,8 +337,8 @@ func generatorCommand(ctx context.Context, cfg Config, duration time.Duration, a ssh := []string{"-p", strconv.Itoa(port), "-i", cfg.GeneratorSSHIdentity, "-o", "ConnectTimeout=10", "-o", "ServerAliveInterval=15", "-o", "ServerAliveCountMax=3", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=accept-new", - "-o", "UserKnownHostsFile=" + cfg.GeneratorSSHKnownHosts, cfg.GeneratorSSH, "--", strings.Join(remote, " ")} - return exec.CommandContext(ctx, "ssh", ssh...), cancel + "-o", "UserKnownHostsFile=" + cfg.GeneratorSSHKnownHosts, cfg.GeneratorSSH, "--", command} + return exec.CommandContext(ctx, "ssh", ssh...) } func generatorArgs(cfg Config, signal Signal, spec PhaseSpec) []string { @@ -401,7 +427,7 @@ func resourceDelta(before, after *snapshot, samples []ResourceSample, elapsed fl return result } -func classify(t Trial) (string, []string) { +func classify(t Trial, previous []Trial) (string, []string) { var notes []string if t.GeneratorFailures > 0 { notes = append(notes, "one or more telemetrygen processes failed") @@ -415,18 +441,46 @@ func classify(t Trial) (string, []string) { if len(notes) > 0 { return "invalid", notes } - switch { - case t.Accepted.Total() == 0: + if t.Accepted.Total() == 0 { return "invalid", []string{"Fanout accepted no telemetry"} - case t.AcceptanceRatio >= .95: + } + if t.AcceptanceRatio >= .95 { return "target-sustained", nil + } + if cpu := t.GeneratorCPU; cpu != nil && cpu.LogicalCPUs > 0 { + if cpu.AverageCPUCores >= .9*float64(cpu.LogicalCPUs) { + return "generator-limited", []string{"generator host averaged at least 90% CPU utilisation"} + } + if prev := previousSweepTrial(t.Spec, previous); prev != nil && prev.Classification != "invalid" && prev.AcceptedPerSecond > 0 && t.AcceptedPerSecond-prev.AcceptedPerSecond < .1*prev.AcceptedPerSecond { + return "saturated", []string{fmt.Sprintf("accepted throughput rose less than 10%% over %s (%d processes per signal) while generator CPU was below 90%%", prev.Spec.Name, prev.Spec.ProcessesPerSignal)} + } + } + switch { case t.AcceptanceRatio >= .80: return "near-saturation", []string{"accepted less than 95% of the requested workload"} default: + if t.GeneratorCPU != nil { + return "saturated-or-generator-limited", []string{"the target was not sustained; throughput is still scaling or no valid previous sweep point is available"} + } return "saturated-or-generator-limited", []string{"the target was not sustained; separate load-generator hosts are needed to attribute the limit"} } } +// Compare only the preceding process count in this signal's current repetition. +// Warmup and sustained phases do not establish sweep plateaus. +func previousSweepTrial(spec PhaseSpec, trials []Trial) *Trial { + if spec.Kind != "sweep" { + return nil + } + var previous *Trial + for i := range trials { + p := &trials[i] + if p.Spec.Kind == "sweep" && p.Spec.Signal == spec.Signal && p.Spec.Repetition == spec.Repetition && p.Spec.ProcessesPerSignal < spec.ProcessesPerSignal && (previous == nil || p.Spec.ProcessesPerSignal > previous.Spec.ProcessesPerSignal) { + previous = p + } + } + return previous +} func bestMixedTopology(trials []Trial) int { bestRate, processes := -1.0, 0 fallbackCompletion, fallbackRate, fallbackProcesses := -1.0, -1.0, 0 @@ -565,7 +619,13 @@ func settingsFrom(c Config) Settings { SampleInterval: c.SampleInterval, SampleSeconds: c.SampleInterval.Seconds(), GeneratorTimeout: c.GeneratorTimeout, TimeoutSeconds: c.GeneratorTimeout.Seconds(), Cooldown: c.Cooldown, CooldownSeconds: c.Cooldown.Seconds(), DataDir: c.DataDir, QueryURL: c.QueryURL, QueryRate: c.QueryRate, BatchSize: c.BatchSize, TraceChildSpans: 1, - Methodology: "Fanout server-side accepted-row counter deltas divided by measured wall time; telemetrygen emits one parent and one child span per trace tick."} + Methodology: "Fanout server-side accepted-row counter deltas divided by measured wall time; telemetrygen emits one parent and one child span per trace tick. " + + "Split runs bracket each phase with generator-host /proc/stat reads over the generator SSH setup, including startup and export draining. " + + "Generator CPU cores = logical CPUs × (total tick delta − idle and iowait tick delta) / total tick delta; guest ticks are not counted twice. " + + "Invalid trials take precedence; accepting at least 95% of the target remains target-sustained. Otherwise generator CPU at or above 90% is generator-limited; " + + "below 90%, a sweep gain under 10% over the preceding process count for the same signal and repetition is saturated. " + + "Without a valid previous sweep point or a plateau, the existing acceptance-ratio classification applies; same-host runs retain that fallback. " + + "The sustained phase uses the mixed sweep topology with highest accepted throughput meeting the read target, or the best read-completion fallback if none qualifies."} } func workloadHash(settings Settings, plan []PhaseSpec) string { diff --git a/internal/bench/runner_test.go b/internal/bench/runner_test.go index d38281f..bd5591f 100644 --- a/internal/bench/runner_test.go +++ b/internal/bench/runner_test.go @@ -3,13 +3,84 @@ package bench import ( "context" "errors" + "io" + "net/http" "os" "slices" "strings" + "sync/atomic" "testing" "time" ) +func TestRunPhaseStopsQueriesBeforeFinalGeneratorCPURead(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + var dispatched atomic.Int64 + firstQuery := make(chan struct{}, 1) + scrapes := 0 + r := &Runner{ + Config: Config{GeneratorSSH: "unused", MetricsURL: "http://bench/metrics", QueryURL: "http://bench", SampleInterval: time.Hour}, + queryTargets: newQueryRun().targets, + client: &http.Client{Transport: releaseTransport(func(req *http.Request) (*http.Response, error) { + body := `{"results":[{"status":"empty"}]}` + if req.URL.Path == "/metrics" { + scrapes++ + if scrapes == 2 { + // Keep the phase open until a query is in flight. + select { + case <-firstQuery: + case <-ctx.Done(): + return nil, ctx.Err() + } + } + body = "fanout_rows_accepted_total{signal=\"logs\"} 1\n" + } else { + dispatched.Add(1) + select { + case firstQuery <- struct{}{}: + default: + } + select { + case <-time.After(20 * time.Millisecond): + case <-req.Context().Done(): + return nil, req.Context().Err() + } + } + return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(body)), Header: make(http.Header)}, nil + })}, + } + cpuReads := 0 + var beforeCPURead, afterCPURead int64 + r.readGeneratorCPU = func(ctx context.Context) (cpuStat, error) { + cpuReads++ + if cpuReads == 2 { + beforeCPURead = dispatched.Load() + select { + case <-time.After(100 * time.Millisecond): + case <-ctx.Done(): + return cpuStat{}, ctx.Err() + } + afterCPURead = dispatched.Load() + } + return cpuStat{total: uint64(cpuReads * 100), idle: uint64(cpuReads * 50), logicalCPUs: 1}, nil + } + // No generator processes are needed to exercise the phase boundary. + trial, err := r.runPhase(ctx, PhaseSpec{Signal: Logs, QueryRate: 100}, nil) + if err != nil { + t.Fatal(err) + } + if cpuReads != 2 || beforeCPURead == 0 { + t.Fatalf("CPU reads=%d queries before final read=%d; workload was not exercised", cpuReads, beforeCPURead) + } + if afterCPURead != beforeCPURead { + t.Fatalf("queries dispatched during final CPU read: %d", afterCPURead-beforeCPURead) + } + if trial.QueryLatency.Count != dispatched.Load() || trial.QueryFailures != 0 || trial.QueryCanceled != 0 { + t.Fatalf("in-flight queries did not drain: dispatched=%d completed=%d failures=%d canceled=%d", dispatched.Load(), trial.QueryLatency.Count, trial.QueryFailures, trial.QueryCanceled) + } +} + func TestGeneratorCommandWrapsSSHAndQuotesEveryArgument(t *testing.T) { cfg := Config{Telemetrygen: "/stage/generator's binary", GeneratorSSH: "root@10.77.0.3", GeneratorSSHPort: 2222, GeneratorSSHIdentity: "/stage/key with spaces", GeneratorSSHKnownHosts: "/stage/known hosts", GeneratorTimeout: 30 * time.Second} args := []string{"logs", "--otlp-header", "Authorization=\"Bearer token's $(touch /bad); value\""} @@ -107,13 +178,62 @@ func TestClassifySeparatesValidityFromSaturation(t *testing.T) { {Trial{Accepted: Counts{Spans: 100}, AcceptanceRatio: 1, RowsDropped: 1}, "invalid"}, } for _, tt := range tests { - got, _ := classify(tt.trial) + got, _ := classify(tt.trial, nil) if got != tt.want { t.Errorf("classify(%+v)=%q want %q", tt.trial, got, tt.want) } } } +func TestClassifySplitGeneratorCPUAndSweepPlateau(t *testing.T) { + baseline := Trial{Spec: PhaseSpec{Name: "logs-p16-r2", Kind: "sweep", Signal: Logs, Repetition: 2, ProcessesPerSignal: 16}, AcceptedPerSecond: 100, Classification: "target-sustained"} + for _, tc := range []struct { + name string + cores float64 + rate float64 + ratio float64 + change func(*Trial, *Trial) + want string + }{ + {name: "90 percent boundary", cores: 14.4, rate: 200, ratio: .5, want: "generator-limited"}, + {name: "CPU limit precedes plateau", cores: 16, rate: 100, ratio: .5, want: "generator-limited"}, + {name: "plateau below CPU threshold", cores: 14.39, rate: 109.99, ratio: .5, want: "saturated"}, + {name: "declining throughput", cores: 8, rate: 80, ratio: .5, want: "saturated"}, + {name: "exactly ten percent gain", cores: 8, rate: 110, ratio: .5, want: "saturated-or-generator-limited"}, + {name: "still scaling", cores: 8, rate: 190, ratio: .5, want: "saturated-or-generator-limited"}, + {name: "target unchanged despite CPU", cores: 16, rate: 100, ratio: .95, want: "target-sustained"}, + {name: "target unchanged despite plateau", cores: 8, rate: 100, ratio: .95, want: "target-sustained"}, + {name: "CPU before near saturation", cores: 16, rate: 200, ratio: .9, want: "generator-limited"}, + {name: "plateau before near saturation", cores: 8, rate: 100, ratio: .9, want: "saturated"}, + {name: "same host fallback", rate: 100, ratio: .5, change: func(t, _ *Trial) { t.GeneratorCPU = nil }, want: "saturated-or-generator-limited"}, + {name: "first process count", cores: 8, rate: 100, ratio: .5, change: func(t, _ *Trial) { t.Spec.ProcessesPerSignal = 1 }, want: "saturated-or-generator-limited"}, + {name: "other signal", cores: 8, rate: 100, ratio: .5, change: func(_, p *Trial) { p.Spec.Signal = Traces }, want: "saturated-or-generator-limited"}, + {name: "other repetition", cores: 8, rate: 100, ratio: .5, change: func(_, p *Trial) { p.Spec.Repetition = 1 }, want: "saturated-or-generator-limited"}, + {name: "warmup is not baseline", cores: 8, rate: 100, ratio: .5, change: func(_, p *Trial) { p.Spec.Kind = "warmup" }, want: "saturated-or-generator-limited"}, + {name: "sustained has no plateau comparison", cores: 8, rate: 100, ratio: .5, change: func(t, _ *Trial) { t.Spec.Kind = "sustained" }, want: "saturated-or-generator-limited"}, + {name: "invalid baseline", cores: 8, rate: 100, ratio: .5, change: func(_, p *Trial) { p.Classification = "invalid" }, want: "saturated-or-generator-limited"}, + {name: "zero baseline", cores: 8, rate: 100, ratio: .5, change: func(_, p *Trial) { p.AcceptedPerSecond = 0 }, want: "saturated-or-generator-limited"}, + {name: "invalid overrides CPU", cores: 16, rate: 100, ratio: 1, change: func(t, _ *Trial) { t.GeneratorFailures = 1 }, want: "invalid"}, + {name: "no accepted data", cores: 16, ratio: .5, change: func(t, _ *Trial) { t.Accepted = Counts{} }, want: "invalid"}, + } { + t.Run(tc.name, func(t *testing.T) { + previous := baseline + trial := Trial{Spec: PhaseSpec{Kind: "sweep", Signal: Logs, Repetition: 2, ProcessesPerSignal: 32}, Accepted: Counts{LogRecords: 1}, AcceptedPerSecond: tc.rate, AcceptanceRatio: tc.ratio, GeneratorCPU: &GeneratorCPU{AverageCPUCores: tc.cores, LogicalCPUs: 16}} + if tc.change != nil { + tc.change(&trial, &previous) + } + // Older points must not replace the immediately preceding process count. + older := baseline + older.Spec.ProcessesPerSignal = 8 + older.AcceptedPerSecond = 1 + got, _ := classify(trial, []Trial{previous, older}) + if got != tc.want { + t.Fatalf("classification=%q want %q", got, tc.want) + } + }) + } +} + func TestSecretRedactionAndBoundedOutput(t *testing.T) { b := &boundedBuffer{limit: 8} _, _ = b.Write([]byte("012345secret")) diff --git a/internal/bench/types.go b/internal/bench/types.go index c081594..0c08dd8 100644 --- a/internal/bench/types.go +++ b/internal/bench/types.go @@ -175,6 +175,11 @@ type Gate struct { Detail string `json:"detail"` } +type GeneratorCPU struct { + AverageCPUCores float64 `json:"average_cpu_cores"` + LogicalCPUs int `json:"logical_cpus"` +} + type Trial struct { Spec PhaseSpec `json:"settings"` StartedAt time.Time `json:"started_at"` @@ -186,6 +191,7 @@ type Trial struct { RowsDropped float64 `json:"rows_dropped"` GeneratorProcesses int `json:"generator_processes"` GeneratorFailures int `json:"generator_failures"` + GeneratorCPU *GeneratorCPU `json:"generator_cpu,omitempty"` ProcessRestarted bool `json:"fanout_restarted"` Resources Resources `json:"resources"` QueryLatency Latency `json:"query_latency_ms,omitempty"`