Skip to content

Add subject_disk tmpfs mounts and disk_pressure_correctness cases #48

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
namles wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Add subject_disk tmpfs mounts and disk_pressure_correctness cases #48

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 23 additions & 11 deletions containers/generator/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
38 changes: 38 additions & 0 deletions internal/config/case.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -1032,13 +1041,31 @@ 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)?$`)
Comment on lines +1044 to +1049

@coderabbitai coderabbitai Bot Jul 15, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

subjectDiskSizeRe accepts sizes that parseByteSize rejects — breaks the documented lockstep guarantee.

The regex ^[0-9]+([kmg]b?|b)?$ matches an all-zero numeral ("0", "0m", "0b", "0kb", …), but orchestrator.parseByteSize explicitly rejects non-positive sizes (if n <= 0 { return error }). A case with subject_disk: {path: /data, size: "0"} passes Validate() at load time but fails later in writeCompose, contradicting the comment directly above the regex: "a case that passes Validate can never fail at compose-render time."

🐛 Proposed fix — require at least one non-zero digit 
-var subjectDiskSizeRe = regexp.MustCompile(`(?i)^[0-9]+([kmg]b?|b)?$`)
+var subjectDiskSizeRe = regexp.MustCompile(`(?i)^[0-9]*[1-9][0-9]*([kmg]b?|b)?$`)

A more robust fix is to have this package expose (or directly reuse) the same byte-size parser writeCompose calls, so the two implementations can't drift again. Either way, please add "0"/"0m"/"0b" to TestValidateSubjectDisk's invalid cases once fixed — the current test suite doesn't cover this edge.

Also applies to: 1058-1068

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@internal/config/case.go` around lines 1044 - 1049, Update subjectDiskSizeRe
and the validation path around TestValidateSubjectDisk so zero-valued sizes such
as "0", "0m", and "0b" are rejected consistently with
orchestrator.parseByteSize; prefer reusing that parser if accessible, otherwise
require at least one non-zero digit in the regex. Add these zero-size forms to
TestValidateSubjectDisk’s invalid cases while preserving acceptance of valid
positive units.


// 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.
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)
}
}
Comment on lines +1058 to +1068

@coderabbitai coderabbitai Bot Jul 15, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

disk_pressure_correctness type doesn't require a subject_disk (or generator) block.

Unlike IsKafkaType(), which fails validation when its kafka: block or a kafka-mode generator is missing, nothing here enforces that a disk_pressure_correctness case actually declares subject_disk. If it's omitted, writeCompose simply renders no tmpfs limit, and runDiskPressureCorrectness runs to completion against an effectively unlimited volume — the run will almost certainly report PASS without ever exercising the backpressure path it's named for. Similarly, if generator:/generators: is empty, orch.UpServices("generator") in the runner will fail against a service the compose template never emitted.

Consider adding, alongside the existing per-type checks (e.g. the IsKafkaType() block):

if tc.Type == "disk_pressure_correctness" {
    if tc.SubjectDisk == nil {
        return fmt.Errorf("case %q: type %q requires a `subject_disk:` block", tc.Name, tc.Type)
    }
    if len(tc.AllGenerators()) == 0 {
        return fmt.Errorf("case %q: type %q requires a generator", tc.Name, tc.Type)
    }
}

Also worth guarding: subject_disk combined with cluster: silently no-ops (the tmpfs mount only renders on the singular subject branch) — may be worth an explicit rejection too, per the doc comment's own "singular subject service" caveat.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@internal/config/case.go` around lines 1058 - 1068, The case validation must
require disk_pressure_correctness configurations to declare a subject disk and
at least one generator, preventing runs without backpressure or a generated
service. Add a per-type check alongside IsKafkaType() that rejects nil
tc.SubjectDisk and empty tc.AllGenerators() with case-specific errors; also
reject subject_disk combined with cluster configuration if the existing
documentation requires a singular subject service.

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)
}
Expand Down Expand Up @@ -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-<id>`) and, in sequenced/correctness
Expand Down
61 changes: 61 additions & 0 deletions internal/config/subject_disk_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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)
}
}
79 changes: 74 additions & 5 deletions internal/orchestrator/docker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 }}
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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,
Expand All @@ -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
Expand All @@ -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,
Expand Down Expand Up @@ -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
}
Loading
Loading