From 1b2f175c24296401aedf0c997b0921e8ea1a8577 Mon Sep 17 00:00:00 2001 From: Selman Uluc Date: Wed, 15 Jul 2026 09:59:18 +0300 Subject: [PATCH] Add subject_disk tmpfs mounts and disk_pressure_correctness cases Subjects buffering to local storage had no way to be tested against a full disk: the harness gave every subject the host's entire volume, so disk-exhaustion behavior (crash? silent drop? backpressure?) was never exercised. This adds the missing pieces, all subject-agnostic: - case `subject_disk: {path, size}` mounts a size-limited tmpfs (mode 01777, so non-root images can write) at the given path inside the singular subject service - a small dedicated "disk" for the subject's storage/buffer directory without a specially built image. Size accepts docker-style strings (64m, 1g, 512kb, plain bytes), validated at case load and pre-parsed to bytes at compose render so a bad value fails fast with a clear error. - case type `disk_pressure_correctness`: receiver stays DOWN while a duration-bounded generator floods far more data than the volume holds. The subject must survive the pressure window (container still running), then deliver everything it accepted once the receiver comes up. A subject that fills the volume until its durable layer hits ENOSPC fails on loss and/or the crash check. - generator: rotate 64 distinct random paddings in sequenced mode instead of one shared buffer. Identical padding let subjects that compress their batches shrink the stream ~100:1 (a 1.2 GB flood landed only ~11 MB on disk), which quietly defeated any disk-pressure scenario and flattered compression-friendly workloads generally. Distinct paddings make wire bytes ~ stored bytes. (bench-generator image needs a rebuild/push to take effect.) Unit tests cover size parsing, case validation (kept in lockstep so Validate acceptance can never fail at render time), and the compose tmpfs rendering. Co-Authored-By: Claude Fable 5 --- containers/generator/main.go | 34 ++- internal/config/case.go | 38 +++ internal/config/subject_disk_test.go | 61 +++++ internal/orchestrator/docker.go | 79 +++++- internal/orchestrator/subject_disk_test.go | 125 +++++++++ internal/runner/runner.go | 288 +++++++++++++++++++++ 6 files changed, 609 insertions(+), 16 deletions(-) create mode 100644 internal/config/subject_disk_test.go create mode 100644 internal/orchestrator/subject_disk_test.go diff --git a/containers/generator/main.go b/containers/generator/main.go index 244637c..208b053 100644 --- a/containers/generator/main.go +++ b/containers/generator/main.go @@ -1085,17 +1085,29 @@ func sendLinesConn(cfg config, connID int, clock *sendClock, write func([]byte) templateLine = generateLine(cfg.LineSize, cfg.Format) } - // Pre-allocate a reusable line buffer for sequenced mode so we don't - // regenerate random padding on every line (the old path ran rand.Intn - // LineSize times per line — cratered perf when validate_content was on). - // Padding is generated once and the prefix is rewritten in place. - // Last byte is reserved for '\n' so the hot-loop write callback can - // skip a per-line append-newline allocation. - var seqBuf []byte + // Pre-allocate a small POOL of line buffers for sequenced mode and + // rotate through them per line, rewriting only the CONN=/SEQ= prefix in + // place. Pre-generation keeps the hot loop free of rand calls (the old + // per-line rand.Intn path cratered perf when validate_content was on) — + // but a single shared buffer made every line's padding IDENTICAL, which + // let block compressors downstream (subjects that snappy/gzip their + // batches) shrink the stream ~100:1: a disk-pressure case that flooded + // 1.2 GB landed only ~11 MB on the subject's storage volume. Rotating + // 64 distinct random paddings makes lines inside a compressor block + // mutually incompressible — wire bytes ≈ stored bytes — while costing + // only 64×LineSize of setup memory per connection worker. + // Each buffer's last byte is reserved for '\n' so the hot-loop write + // callback can skip a per-line append-newline allocation. + const seqPadPool = 64 + var seqBufs [][]byte if cfg.Sequenced && len(sampleLines) == 0 { - seqBuf = make([]byte, cfg.LineSize+1) - copy(seqBuf, randString(cfg.LineSize)) - seqBuf[cfg.LineSize] = '\n' + seqBufs = make([][]byte, seqPadPool) + for i := range seqBufs { + buf := make([]byte, cfg.LineSize+1) + copy(buf, randString(cfg.LineSize)) + buf[cfg.LineSize] = '\n' + seqBufs[i] = buf + } } var linesSent, bytesSent int64 @@ -1175,7 +1187,7 @@ func sendLinesConn(cfg config, connID int, clock *sendClock, write func([]byte) if cfg.Format == "json" { line = generateSequencedJSONLine(connID, linesSent, cfg.LineSize) } else { - line = writeSequencedPrefix(seqBuf, connID, linesSent) + line = writeSequencedPrefix(seqBufs[linesSent%seqPadPool], connID, linesSent) } case linesSent%1000 == 0: // Sample every 1000th line with a timestamp for latency measurement diff --git a/internal/config/case.go b/internal/config/case.go index 7a2d1e8..ade1c26 100644 --- a/internal/config/case.go +++ b/internal/config/case.go @@ -105,6 +105,15 @@ type TestCase struct { // buckets. Mutually exclusive with AWS. See MinioConfig in cloud.go. Minio *MinioConfig `yaml:"minio"` + // SubjectDisk, when set, mounts a size-limited tmpfs at Path inside the + // subject container, giving the subject a small dedicated "disk" for its + // storage/buffer directory without a specially built image. Lets a case + // exercise disk-full and disk-backpressure behavior: fill the volume with + // more data than it can hold and observe whether the subject crashes, + // drops, or backpressures. Used by the disk_pressure_correctness type but + // honored on the singular subject service for any type. + SubjectDisk *SubjectDiskConfig `yaml:"subject_disk"` + // Requires lists subject capabilities every subject in this case must // declare (Subject.Capabilities); the runner fails fast on subjects // lacking one instead of starting a run that silently produces zero @@ -1032,6 +1041,13 @@ func (tc *TestCase) IsKafkaType() bool { return strings.HasPrefix(tc.Type, "kafka_") } +// subjectDiskSizeRe accepts the size strings the orchestrator's +// parseByteSize understands: plain bytes ("1048576"), plain bytes with a +// "b" suffix ("64b"), or a k/m/g unit with an optional trailing "b" +// ("64m", "64mb"), case-insensitive. Kept in lockstep with parseByteSize +// so a case that passes Validate can never fail at compose-render time. +var subjectDiskSizeRe = regexp.MustCompile(`(?i)^[0-9]+([kmg]b?|b)?$`) + // Validate runs structural checks that don't depend on runtime state. // Returns an error for cases where the singular and plural forms are both // set (ambiguous) or where required IDs on plural entries are missing. @@ -1039,6 +1055,17 @@ func (tc *TestCase) Validate() error { if len(tc.Generators) > 0 && (tc.Generator.Mode != "" || tc.Generator.Target != "") { return fmt.Errorf("case %q: both `generator:` and `generators:` are set — pick one", tc.Name) } + if tc.SubjectDisk != nil { + if tc.SubjectDisk.Path == "" || tc.SubjectDisk.Size == "" { + return fmt.Errorf("case %q: subject_disk requires both `path` and `size`", tc.Name) + } + if !strings.HasPrefix(tc.SubjectDisk.Path, "/") { + return fmt.Errorf("case %q: subject_disk.path must be absolute, got %q", tc.Name, tc.SubjectDisk.Path) + } + if !subjectDiskSizeRe.MatchString(tc.SubjectDisk.Size) { + return fmt.Errorf("case %q: subject_disk.size %q is not a valid tmpfs size (e.g. 64m, 1g, 1048576)", tc.Name, tc.SubjectDisk.Size) + } + } if len(tc.Receivers) > 0 && (tc.Receiver.Mode != "" || tc.Receiver.Listen != "") { return fmt.Errorf("case %q: both `receiver:` and `receivers:` are set — pick one", tc.Name) } @@ -2412,6 +2439,17 @@ func validateSampleFile(caseName, sampleFile string) error { return nil } +// SubjectDiskConfig mounts a size-limited tmpfs inside the subject container +// (see TestCase.SubjectDisk). Path is the in-container mount point — for +// vmetric that's /opt/vmetric/storage, the root of its NATS JetStream +// StoreDir and queue/WAL files. Size is a docker-compose tmpfs size string +// ("64m", "1g", or plain bytes). The mount is created mode 01777 so +// non-root subject images can write to it. +type SubjectDiskConfig struct { + Path string `yaml:"path"` + Size string `yaml:"size"` +} + type GeneratorConfig struct { // ID is only meaningful in the plural `generators:` form. It names the // docker-compose service (`generator-`) and, in sequenced/correctness diff --git a/internal/config/subject_disk_test.go b/internal/config/subject_disk_test.go new file mode 100644 index 0000000..bc7f957 --- /dev/null +++ b/internal/config/subject_disk_test.go @@ -0,0 +1,61 @@ +package config + +import ( + "strings" + "testing" +) + +// subjectDiskCase returns a minimal valid case with the given +// subject_disk block, so each test tweaks exactly one dimension. +func subjectDiskCase(disk *SubjectDiskConfig) *TestCase { + return &TestCase{ + Name: "disk-validate", + Type: "disk_pressure_correctness", + SubjectDisk: disk, + } +} + +func TestValidateSubjectDisk(t *testing.T) { + valid := []SubjectDiskConfig{ + {Path: "/opt/vmetric/storage", Size: "64m"}, + {Path: "/data", Size: "128MB"}, + {Path: "/data", Size: "1g"}, + {Path: "/data", Size: "512kb"}, + {Path: "/data", Size: "1048576"}, + {Path: "/data", Size: "64b"}, + } + for _, d := range valid { + if err := subjectDiskCase(&d).Validate(); err != nil { + t.Errorf("subject_disk %+v: unexpected error: %v", d, err) + } + } + + invalid := []struct { + disk SubjectDiskConfig + wantSub string + }{ + {SubjectDiskConfig{Path: "", Size: "64m"}, "requires both"}, + {SubjectDiskConfig{Path: "/data", Size: ""}, "requires both"}, + {SubjectDiskConfig{Path: "data", Size: "64m"}, "must be absolute"}, + {SubjectDiskConfig{Path: "/data", Size: "sixty-four"}, "not a valid tmpfs size"}, + {SubjectDiskConfig{Path: "/data", Size: "1.5g"}, "not a valid tmpfs size"}, + {SubjectDiskConfig{Path: "/data", Size: "64bb"}, "not a valid tmpfs size"}, + {SubjectDiskConfig{Path: "/data", Size: "-5m"}, "not a valid tmpfs size"}, + {SubjectDiskConfig{Path: "/data", Size: "64x"}, "not a valid tmpfs size"}, + } + for _, tt := range invalid { + err := subjectDiskCase(&tt.disk).Validate() + if err == nil { + t.Errorf("subject_disk %+v: expected error containing %q, got nil", tt.disk, tt.wantSub) + continue + } + if !strings.Contains(err.Error(), tt.wantSub) { + t.Errorf("subject_disk %+v: error %q does not contain %q", tt.disk, err, tt.wantSub) + } + } + + // No subject_disk block — nothing to validate. + if err := subjectDiskCase(nil).Validate(); err != nil { + t.Errorf("nil subject_disk: unexpected error: %v", err) + } +} diff --git a/internal/orchestrator/docker.go b/internal/orchestrator/docker.go index e3fa046..0e57cad 100644 --- a/internal/orchestrator/docker.go +++ b/internal/orchestrator/docker.go @@ -260,6 +260,13 @@ services: - "{{ .VaultTLSHost }}:/vault-tls:ro" {{- end }} {{- end }} +{{- if .SubjectDiskPath }} + - type: tmpfs + target: "{{ .SubjectDiskPath }}" + tmpfs: + size: {{ .SubjectDiskSizeBytes }} + mode: 01777 +{{- end }} {{- if .SubjectUser }} user: "{{ .SubjectUser }}" {{- end }} @@ -1734,6 +1741,14 @@ type composeVars struct { UseSharedData bool DeferReceiver bool + // SubjectDiskPath / SubjectDiskSizeBytes render a size-limited tmpfs + // mount inside the singular subject service (case subject_disk block) — + // a small dedicated "disk" for disk-full / disk-backpressure cases. + // Size is pre-parsed to bytes so the compose file carries an + // unambiguous integer. Empty path emits nothing. + SubjectDiskPath string + SubjectDiskSizeBytes int64 + // BenchSubnet, when set, pins the bench bridge network's IPAM subnet (instead // of letting Docker auto-assign one) so a cluster_ip_failover case can bind a // known virtual IP that won't collide with a container's assigned address. @@ -2171,6 +2186,25 @@ func writeCompose(path string, cfg RunConfig) error { subjectCmd = "" // no config-path override; subject finds the service config in its workdir } + // Resolve the subject_disk tmpfs mount up front so a malformed size + // string fails the run at compose-render time with a clear error, + // not as a cryptic docker-compose parse failure at up. + var ( + subjectDiskPath string + subjectDiskSizeBytes int64 + ) + if tc.SubjectDisk != nil { + if tc.SubjectDisk.Path == "" || tc.SubjectDisk.Size == "" { + return fmt.Errorf("subject_disk requires both path and size") + } + size, err := parseByteSize(tc.SubjectDisk.Size) + if err != nil { + return fmt.Errorf("subject_disk.size %q: %w", tc.SubjectDisk.Size, err) + } + subjectDiskPath = tc.SubjectDisk.Path + subjectDiskSizeBytes = size + } + vars := composeVars{ SubjectImage: s.ImageRef(), SubjectContainer: subjectContainer, @@ -2192,10 +2226,12 @@ func writeCompose(path string, cfg RunConfig) error { // unset one (the old defaultStr(.., "1"/"1g") behavior) silently // pinned --mem-limit-only runs to 1 CPU and --cpu-limit-only runs // to 1 GB, which throttled the subject into bogus results. - HasResourceLimits: cfg.CPULimit != "" || cfg.MemLimit != "", - CPULimit: cfg.CPULimit, - MemLimit: cfg.MemLimit, - UseSharedData: useSharedData, + HasResourceLimits: cfg.CPULimit != "" || cfg.MemLimit != "", + CPULimit: cfg.CPULimit, + MemLimit: cfg.MemLimit, + UseSharedData: useSharedData, + SubjectDiskPath: subjectDiskPath, + SubjectDiskSizeBytes: subjectDiskSizeBytes, // DeferReceiver drops the `generator.depends_on: receiver` link // so `UpServices("generator")` doesn't transitively start the // receiver. Needed for any test where the subject must buffer @@ -2204,7 +2240,8 @@ func writeCompose(path string, cfg RunConfig) error { // restart/crash phase has nothing left to recover. DeferReceiver: tc.Type == "persistence_correctness" || tc.Type == "persistence_restart_correctness" || - tc.Type == "persistence_crash_correctness", + tc.Type == "persistence_crash_correctness" || + tc.Type == "disk_pressure_correctness", GeneratorImage: cfg.GeneratorImage, ReceiverImage: cfg.ReceiverImage, CollectorImage: cfg.CollectorImage, @@ -2658,3 +2695,35 @@ func boolStr(b bool) string { } return "false" } + +// parseByteSize converts a human byte-size string ("64m", "1g", "512kb", +// or a plain integer byte count) to bytes. Suffixes are case-insensitive +// with an optional trailing "b"; units are binary (1k = 1024). +func parseByteSize(s string) (int64, error) { + v := strings.ToLower(strings.TrimSpace(s)) + mult := int64(1) + switch { + case strings.HasSuffix(v, "kb"), strings.HasSuffix(v, "k"): + mult = 1 << 10 + v = strings.TrimSuffix(strings.TrimSuffix(v, "b"), "k") + case strings.HasSuffix(v, "mb"), strings.HasSuffix(v, "m"): + mult = 1 << 20 + v = strings.TrimSuffix(strings.TrimSuffix(v, "b"), "m") + case strings.HasSuffix(v, "gb"), strings.HasSuffix(v, "g"): + mult = 1 << 30 + v = strings.TrimSuffix(strings.TrimSuffix(v, "b"), "g") + case strings.HasSuffix(v, "b"): + // Plain-bytes suffix ("64b"). Must come after the kb/mb/gb + // cases; keeps parsing consistent with the case-level + // subject_disk.size validation, which accepts it. + v = strings.TrimSuffix(v, "b") + } + n, err := strconv.ParseInt(strings.TrimSpace(v), 10, 64) + if err != nil { + return 0, fmt.Errorf("invalid byte size (want e.g. 64m, 1g, or bytes): %w", err) + } + if n <= 0 { + return 0, fmt.Errorf("byte size must be positive, got %d", n) + } + return n * mult, nil +} diff --git a/internal/orchestrator/subject_disk_test.go b/internal/orchestrator/subject_disk_test.go new file mode 100644 index 0000000..7138768 --- /dev/null +++ b/internal/orchestrator/subject_disk_test.go @@ -0,0 +1,125 @@ +package orchestrator + +import ( + "fmt" + "os" + "path/filepath" + "testing" + + "github.com/VirtualMetric/PipeBench/internal/config" +) + +// TestParseByteSize covers the size strings subject_disk accepts — the +// set must stay consistent with config's subject_disk.size validation so +// a case that passes Validate can never fail at compose-render time. +func TestParseByteSize(t *testing.T) { + good := map[string]int64{ + "64m": 64 << 20, + "64M": 64 << 20, + "64mb": 64 << 20, + "10MB": 10 << 20, + "1g": 1 << 30, + "2GB": 2 << 30, + "512k": 512 << 10, + "512kb": 512 << 10, + "1048576": 1 << 20, + "64b": 64, + " 64m ": 64 << 20, + } + for in, want := range good { + got, err := parseByteSize(in) + if err != nil { + t.Errorf("parseByteSize(%q): unexpected error: %v", in, err) + continue + } + if got != want { + t.Errorf("parseByteSize(%q) = %d, want %d", in, got, want) + } + } + + bad := []string{"", "abc", "-5m", "0", "m", "64x", "64bb", "1.5g"} + for _, in := range bad { + if got, err := parseByteSize(in); err == nil { + t.Errorf("parseByteSize(%q) = %d, want error", in, got) + } + } +} + +// TestComposeRendersSubjectDiskTmpfs verifies the subject_disk block +// renders a long-syntax tmpfs mount on the subject service with the size +// pre-parsed to bytes, and that cases without the block emit no tmpfs. +func TestComposeRendersSubjectDiskTmpfs(t *testing.T) { + render := func(t *testing.T, disk *config.SubjectDiskConfig) string { + t.Helper() + tc := &config.TestCase{ + Name: "disk-smoke", + Type: "correctness", + Duration: "10s", + Generator: config.GeneratorConfig{ + Mode: "tcp", + Target: "subject:9000", + }, + Receiver: config.ReceiverConfig{ + Mode: "tcp", + Listen: ":9001", + }, + SubjectDisk: disk, + } + subj := config.Subject{ + Name: "vmetric", + Image: "vmetric/director", + Version: "dev", + ConfigPath: "/config.yml", + } + tmp := t.TempDir() + composePath := filepath.Join(tmp, "compose.yaml") + cfg := RunConfig{ + TestCase: tc, + Subject: subj, + ConfigName: "default", + ConfigSrcPath: composePath, + TmpDir: tmp, + GeneratorImage: "img-gen", + ReceiverImage: "img-recv", + CollectorImage: "img-coll", + ReceiverHostPort: 19001, + } + if err := writeCompose(composePath, cfg); err != nil { + t.Fatalf("writeCompose: %v", err) + } + data, err := os.ReadFile(composePath) + if err != nil { + t.Fatal(err) + } + return string(data) + } + + out := render(t, &config.SubjectDiskConfig{Path: "/opt/vmetric/storage", Size: "64m"}) + mustContain(t, out, "- type: tmpfs") + mustContain(t, out, `target: "/opt/vmetric/storage"`) + mustContain(t, out, fmt.Sprintf("size: %d", int64(64<<20))) + mustContain(t, out, "mode: 01777") + + out = render(t, nil) + mustNotContain(t, out, "type: tmpfs") + + // A malformed size must fail the render with a clear error, not + // produce a compose file docker rejects later. + tc := &config.TestCase{ + Name: "disk-bad", Type: "correctness", + Generator: config.GeneratorConfig{Mode: "tcp", Target: "subject:9000"}, + Receiver: config.ReceiverConfig{Mode: "tcp", Listen: ":9001"}, + SubjectDisk: &config.SubjectDiskConfig{Path: "/data", Size: "sixty-four"}, + } + tmp := t.TempDir() + composePath := filepath.Join(tmp, "compose.yaml") + err := writeCompose(composePath, RunConfig{ + TestCase: tc, + Subject: config.Subject{Name: "vmetric", Image: "img", ConfigPath: "/config.yml"}, + TmpDir: tmp, ConfigSrcPath: composePath, + GeneratorImage: "g", ReceiverImage: "r", CollectorImage: "c", + }) + if err == nil { + t.Fatal("expected error for malformed subject_disk.size, got nil") + } +} diff --git a/internal/runner/runner.go b/internal/runner/runner.go index 7d51d32..ac7fd66 100644 --- a/internal/runner/runner.go +++ b/internal/runner/runner.go @@ -171,6 +171,14 @@ func (r *Runner) Run(tc *config.TestCase, subject config.Subject) (results.RunRe if tc.Type == "persistence_file_restart_correctness" { return r.runPersistenceFileRestartCorrectness(tc, subject) } + // Disk pressure: the subject gets a small size-limited storage volume + // (subject_disk tmpfs) and the receiver stays down while the generator + // pushes far more data than the volume holds. The subject must apply + // backpressure (slow ingestion) instead of filling the disk and crashing; + // once the receiver comes up, everything accepted must be delivered. + if tc.Type == "disk_pressure_correctness" { + return r.runDiskPressureCorrectness(tc, subject) + } // Kafka crash/restart correctness reuses the shutdown flow: produce to // the broker while the receiver is down (the subject consumes from Kafka // and buffers to its crash-resistant queue), kill/stop the subject, bring @@ -1580,6 +1588,286 @@ func (r *Runner) runPersistenceShutdownCorrectness(tc *config.TestCase, subject return result, nil } +// runDiskPressureCorrectness drives the disk-backpressure scenario. The +// subject gets a small dedicated storage volume (case subject_disk block — +// a size-limited tmpfs over its buffer/StoreDir root) and the receiver is +// deliberately DOWN while a duration-bounded generator pushes far more +// data than that volume can hold. A subject without disk admission control +// fills the volume until its durable layer (e.g. embedded NATS JetStream) +// hits ENOSPC — crashing outright or silently dropping records. A subject +// with disk backpressure must instead slow ingestion (TCP flow control — +// the generator simply sends fewer lines within its window; its per-write +// deadline guarantees it still exits at the duration bound) and stay alive. +// When the receiver comes up, every line the generator counted as sent +// must drain: expected loss per the case (typically 0), and the subject +// container must have been running continuously through the pressure +// window. +func (r *Runner) runDiskPressureCorrectness(tc *config.TestCase, subject config.Subject) (results.RunResult, error) { + configName := r.opts.ConfigName + subject = r.applySubjectOverrides(subject) + + fmt.Printf("→ test=%s subject=%s version=%s config=%s\n", + tc.Name, subject.Name, subject.Version, configName) + + configSrc, err := tc.ConfigFilePath(r.opts.CasesDir, configName, subject) + if err != nil { + return results.RunResult{}, err + } + configSrc, err = filepath.Abs(configSrc) + if err != nil { + return results.RunResult{}, fmt.Errorf("resolving config path: %w", err) + } + + tmpDir, err := os.MkdirTemp("", "bench-"+tc.Name+"-") + if err != nil { + return results.RunResult{}, err + } + if err := os.Chmod(tmpDir, 0o777); err != nil { + return results.RunResult{}, fmt.Errorf("chmod tmpdir: %w", err) + } + defer func() { + if !r.opts.NoCleanup { + os.RemoveAll(tmpDir) + } + }() + + extraEnv := map[string]string{} + if cfg, ok := tc.Configurations[configName]; ok { + maps.Copy(extraEnv, cfg.Env) + } + + caseDir, err := filepath.Abs(filepath.Join(r.opts.CasesDir, tc.Name)) + if err != nil { + return results.RunResult{}, fmt.Errorf("resolving case directory: %w", err) + } + + runCfg := orchestrator.RunConfig{ + TestCase: tc, + Subject: subject, + ConfigName: configName, + ConfigSrcPath: configSrc, + CaseDir: caseDir, + TmpDir: tmpDir, + GeneratorImage: r.opts.GeneratorImage, + ReceiverImage: r.opts.ReceiverImage, + CollectorImage: r.opts.CollectorImage, + ReceiverHostPort: r.opts.ReceiverHostPort, + ExtraSubjectEnv: extraEnv, + CPULimit: r.opts.CPULimit, + MemLimit: r.opts.MemLimit, + } + + cr, err := orchestrator.NewComposeRunner(r.ctx, runCfg) + if err != nil { + return results.RunResult{}, fmt.Errorf("compose setup: %w", err) + } + orch := cr + + subjectContainer := "bench-subject-" + subject.Name + for _, c := range []string{"bench-generator", "bench-receiver", "bench-collector", subjectContainer} { + _ = exec.Command("docker", "rm", "-f", c).Run() + } + _ = orch.Down() + + startTime := time.Now() + + cleanup := func() { + if !r.opts.NoCleanup { + fmt.Println(" tearing down…") + _ = orch.Down() + } + } + defer cleanup() + + // PHASE 1: subject + collector only — the receiver stays down so the + // subject cannot drain and its storage volume fills instead. + fmt.Println(" phase 1: starting subject on a size-limited storage volume (receiver is DOWN)…") + if err := orch.UpServices("subject", "collector"); err != nil { + return results.RunResult{}, fmt.Errorf("starting subject: %w", err) + } + + // PHASE 2: duration-bounded flood — the generator is configured to + // attempt far more volume than subject_disk.size holds. + fmt.Println(" phase 2: flooding subject (receiver still DOWN)…") + if err := orch.UpServices("generator"); err != nil { + return results.RunResult{}, fmt.Errorf("starting generator: %w", err) + } + + duration := tc.DurationOrDefault(60 * time.Second) + warmup := tc.WarmupOrDefault(5 * time.Second) + genTimeout := min(duration+warmup+2*time.Minute, r.opts.Timeout) + + fmt.Printf(" waiting for generator (up to %s)…\n", genTimeout) + if err := orch.WaitForGeneratorExit(genTimeout); err != nil { + return results.RunResult{}, fmt.Errorf("waiting for generator: %w", err) + } + + genStats := r.parseGeneratorStats(orch.GeneratorStdout()) + fmt.Printf(" generator sent %s lines (%s bytes) under disk pressure\n", + formatCount(genStats.LinesSent), formatCount(genStats.BytesSent)) + + // PHASE 3: pressure verdict — the subject must have survived the flood. + // Capture storage usage while the container is (hopefully) still up so + // the report shows how full the volume actually got. + subjectAlive := containerRunning(subjectContainer) + if usage := subjectDiskUsage(subjectContainer, tc); usage != "" { + fmt.Printf(" phase 3: subject storage usage after flood: %s (volume cap %s)\n", usage, tc.SubjectDisk.Size) + } + if subjectAlive { + fmt.Println(" phase 3: subject survived the disk-pressure window ✓") + } else { + fmt.Println(" phase 3: subject is NOT running after the disk-pressure window ✗") + fmt.Fprintf(os.Stderr, "\n --- subject (last 40 lines) ---\n%s", orch.Logs("subject", 40)) + } + + // PHASE 4: receiver up — whatever the subject accepted must now drain. + fmt.Println(" phase 4: starting receiver (drain)…") + if err := orch.UpServices("receiver"); err != nil { + return results.RunResult{}, fmt.Errorf("starting receiver: %w", err) + } + if err := sleepCtx(r.ctx, 3*time.Second); err != nil { + return results.RunResult{}, fmt.Errorf("interrupted: %w", err) + } + + drainTimeout := 3 * time.Minute + fmt.Printf(" phase 5: waiting for logs to drain (up to %s)…\n", drainTimeout) + + metricsPort, stopPortFwd, err := orch.ReceiverMetricsPort() + if err != nil { + return results.RunResult{}, fmt.Errorf("setting up receiver access: %w", err) + } + defer stopPortFwd() + + var lastCount int64 + stableRounds := 0 + drainDeadline := time.Now().Add(drainTimeout) + for time.Now().Before(drainDeadline) { + if err := sleepCtx(r.ctx, 5*time.Second); err != nil { + return results.RunResult{}, fmt.Errorf("interrupted: %w", err) + } + rm, err := r.queryReceiverMetrics(metricsPort, 10*time.Second) + if err != nil { + continue + } + fmt.Printf(" received: %s / %s lines\n", formatCount(rm.LinesReceived), formatCount(genStats.LinesSent)) + if rm.LinesReceived == lastCount && rm.LinesReceived > 0 { + stableRounds++ + if stableRounds >= 12 { + fmt.Println(" receiver stable — all logs drained") + break + } + } else { + stableRounds = 0 + } + lastCount = rm.LinesReceived + } + + recvMetrics, err := r.queryReceiverMetrics(metricsPort, 30*time.Second) + if err != nil { + return results.RunResult{}, fmt.Errorf("querying receiver metrics: %w", err) + } + + elapsed := time.Since(startTime).Seconds() + + lossPct := 0.0 + if genStats.LinesSent > 0 { + lossPct = 100.0 * (1.0 - float64(recvMetrics.LinesReceived)/float64(genStats.LinesSent)) + if lossPct < 0 { + lossPct = 0 + } + } + + passed := lossPct <= tc.Correctness.ExpectedLossPct + var errors []string + if !subjectAlive { + passed = false + errors = append(errors, "subject crashed/exited during disk pressure (container not running after the flood window)") + } + if lossPct > tc.Correctness.ExpectedLossPct { + errors = append(errors, fmt.Sprintf("expected loss <= %.2f%%, got %.2f%% (%s of %s lines lost)", + tc.Correctness.ExpectedLossPct, lossPct, + formatCount(genStats.LinesSent-recvMetrics.LinesReceived), formatCount(genStats.LinesSent))) + } + if recvMetrics.LinesReceived > genStats.LinesSent { + extra := recvMetrics.LinesReceived - genStats.LinesSent + if tc.Correctness.AllowOverDelivery { + fmt.Printf(" note: over-delivery of %s lines (at-least-once duplicates — not a failure)\n", formatCount(extra)) + } else { + passed = false + errors = append(errors, fmt.Sprintf("over-delivery: received %s lines but only %s were sent (%s extra/duplicate lines)", + formatCount(recvMetrics.LinesReceived), formatCount(genStats.LinesSent), formatCount(extra))) + } + } + if tc.Correctness.ValidateDedup && recvMetrics.Duplicates > 0 { + passed = false + errors = append(errors, fmt.Sprintf("expected 0 duplicates, got %s", formatCount(recvMetrics.Duplicates))) + } + + result := results.RunResult{ + TestName: tc.Name, + Config: configName, + Subject: subject.Name, + Version: subject.Version, + Hardware: hardwareID(), + Timestamp: startTime, + DurationSec: elapsed, + FirstSentNs: genStats.FirstSentNs, + LastSentNs: genStats.LastSentNs, + FirstReceivedNs: recvMetrics.FirstReceivedNs, + LastReceivedNs: recvMetrics.LastReceivedNs, + LinesIn: genStats.LinesSent, + LinesOut: recvMetrics.LinesReceived, + BytesIn: genStats.BytesSent, + BytesOut: recvMetrics.BytesReceived, + LossPercent: lossPct, + Passed: &passed, + } + if !passed { + result.FailReason = strings.Join(errors, "; ") + } + + dir, err := r.saveResult(result, "") + if err != nil { + return result, fmt.Errorf("saving results: %w", err) + } + + fmt.Printf(" done. results → %s\n", dir) + fmt.Printf(" lines sent: %s lines received: %s loss: %.2f%%\n", + formatCount(genStats.LinesSent), formatCount(recvMetrics.LinesReceived), lossPct) + if passed { + fmt.Println(" disk pressure correctness: PASSED ✓") + } else { + fmt.Println(" disk pressure correctness: FAILED ✗") + for _, e := range errors { + fmt.Printf(" - %s\n", e) + } + } + + if recvMetrics.LinesReceived == 0 { + fmt.Fprintln(os.Stderr, "\n WARNING: 0 lines received. Container logs:") + fmt.Fprintf(os.Stderr, "\n --- generator ---\n%s", orch.Logs("generator", 30)) + fmt.Fprintf(os.Stderr, "\n --- subject ---\n%s", orch.Logs("subject", 30)) + fmt.Fprintf(os.Stderr, "\n --- receiver ---\n%s", orch.Logs("receiver", 30)) + } + + return result, nil +} + +// subjectDiskUsage reports the subject_disk mount's on-disk usage inside the +// subject container ("du -sh" of the mount point), best-effort: empty when +// the case has no subject_disk block or the container is not running. +func subjectDiskUsage(container string, tc *config.TestCase) string { + if tc.SubjectDisk == nil { + return "" + } + out, err := exec.Command("docker", "exec", container, "sh", "-c", + "du -sh "+tc.SubjectDisk.Path+" 2>/dev/null | cut -f1").CombinedOutput() + if err != nil { + return "" + } + return strings.TrimSpace(string(out)) +} + // midDeliveryFlow parameterizes the shared kafka correctness driver // (runKafkaMidDeliveryAction): produce to the broker with the receiver live, // fire one disruptive action once the receiver has seen ~half of total_lines,