Add subject_disk tmpfs mounts and disk_pressure_correctness cases#48
Add subject_disk tmpfs mounts and disk_pressure_correctness cases#48namles wants to merge 1 commit into
Conversation
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 <noreply@anthropic.com>
WalkthroughThe PR adds ChangesDisk pressure correctness
Sequenced generator buffers
Estimated code review effort: 4 (Complex) | ~45 minutes Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (2)
containers/generator/main.go (1)
1103-1109: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAvoid allocating the sequenced pool for JSON mode.
The JSON sequenced path uses
generateSequencedJSONLineand never readsseqBufs, so this allocates64 * (LineSize+1)bytes per connection worker unnecessarily. Gate the pool oncfg.Format != "json"or allocate it lazily.Proposed fix
- if cfg.Sequenced && len(sampleLines) == 0 { + if cfg.Sequenced && cfg.Format != "json" && len(sampleLines) == 0 {🤖 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 `@containers/generator/main.go` around lines 1103 - 1109, Update the seqBufs initialization in the cfg.Sequenced and empty sampleLines branch to skip allocation when cfg.Format is "json", since generateSequencedJSONLine does not use the pool; retain the existing pool creation and contents for non-JSON formats.internal/runner/runner.go (1)
1684-1687: 🚀 Performance & Scalability | 🔵 TrivialCollector is started but its metrics are never collected.
orch.UpServices("subject", "collector")starts the collector, but unlike the mainRun()flow, this function never callsorch.CopyMetricsCSV/orch.StopCollector, soAvgMemMB/DiskWriteBytes/etc. are never captured — a notable gap for a test whose entire premise is disk usage under pressure. Worth wiring in the same CSV copy + stop sequence used inRun(), or dropping the collector service if it's genuinely unused here.🤖 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/runner/runner.go` around lines 1684 - 1687, Update the phase-1 flow around UpServices("subject", "collector") to either collect the collector metrics using the same CopyMetricsCSV and StopCollector sequence as Run(), ensuring fields such as AvgMemMB and DiskWriteBytes are populated, or remove "collector" from the services started if this phase does not need collector metrics.
🤖 Prompt for all review comments with 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.
Inline comments:
In `@internal/config/case.go`:
- Around line 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.
- Around line 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.
In `@internal/runner/runner.go`:
- Around line 1859-1869: Update subjectDiskUsage to remove the sh -c invocation
and execute du directly, passing tc.SubjectDisk.Path as a separate argv element
so shell metacharacters cannot be interpreted. Preserve the existing error
handling and trim the returned du output before returning it.
- Around line 1732-1763: Update the drain loop around drainDeadline to clamp its
deadline against the existing runDeadline derived from r.opts.Timeout, while
retaining the three-minute drainTimeout as the local maximum. Ensure the phase
exits when either the configured run timeout or the drain timeout is reached,
consistent with the other drain loops in this file.
- Around line 1829-1832: Update runDiskPressureCorrectness so it skips the
saveResult persistence step when --drain is enabled, matching the drain behavior
in Run(). Preserve result generation and return handling, but prevent diagnostic
drain runs from overwriting the canonical results store.
---
Nitpick comments:
In `@containers/generator/main.go`:
- Around line 1103-1109: Update the seqBufs initialization in the cfg.Sequenced
and empty sampleLines branch to skip allocation when cfg.Format is "json", since
generateSequencedJSONLine does not use the pool; retain the existing pool
creation and contents for non-JSON formats.
In `@internal/runner/runner.go`:
- Around line 1684-1687: Update the phase-1 flow around UpServices("subject",
"collector") to either collect the collector metrics using the same
CopyMetricsCSV and StopCollector sequence as Run(), ensuring fields such as
AvgMemMB and DiskWriteBytes are populated, or remove "collector" from the
services started if this phase does not need collector metrics.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 1b2f9012-808c-4cd4-85d3-316be7588500
📒 Files selected for processing (6)
containers/generator/main.gointernal/config/case.gointernal/config/subject_disk_test.gointernal/orchestrator/docker.gointernal/orchestrator/subject_disk_test.gointernal/runner/runner.go
| // 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)?$`) |
There was a problem hiding this comment.
🎯 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.
| 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) | ||
| } | ||
| } |
There was a problem hiding this comment.
🎯 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.
| 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 | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Drain loop isn't bounded by r.opts.Timeout.
Every other drain loop in this file clamps its deadline against a runDeadline := startTime.Add(r.opts.Timeout) so "a run never overruns Options.Timeout" (see the comment at line ~412). Here, drainDeadline := time.Now().Add(drainTimeout) (a hardcoded 3 minutes) has no such clamp, so a disk_pressure_correctness run can exceed the configured timeout ceiling — a problem for CI/batch runs across many subjects with an overall time budget.
♻️ Proposed fix
startTime := time.Now()
+ runDeadline := startTime.Add(r.opts.Timeout)
...
drainTimeout := 3 * time.Minute
fmt.Printf(" phase 5: waiting for logs to drain (up to %s)…\n", drainTimeout)
...
drainDeadline := time.Now().Add(drainTimeout)
+ if drainDeadline.After(runDeadline) {
+ drainDeadline = runDeadline
+ }
for time.Now().Before(drainDeadline) {📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| 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 | |
| } | |
| startTime := time.Now() | |
| runDeadline := startTime.Add(r.opts.Timeout) | |
| 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) | |
| if drainDeadline.After(runDeadline) { | |
| drainDeadline = runDeadline | |
| } | |
| 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 | |
| } |
🤖 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/runner/runner.go` around lines 1732 - 1763, Update the drain loop
around drainDeadline to clamp its deadline against the existing runDeadline
derived from r.opts.Timeout, while retaining the three-minute drainTimeout as
the local maximum. Ensure the phase exits when either the configured run timeout
or the drain timeout is reached, consistent with the other drain loops in this
file.
| dir, err := r.saveResult(result, "") | ||
| if err != nil { | ||
| return result, fmt.Errorf("saving results: %w", err) | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map relevant symbols in the file first
ast-grep outline internal/runner/runner.go --view expanded || true
# Show line-numbered snippets around the cited area and Run()
nl -ba internal/runner/runner.go | sed -n '1750,1865p'
printf '\n----\n'
nl -ba internal/runner/runner.go | sed -n '1,260p' | sed -n '1,260p' >/dev/null
# Find Run() definition and nearby drain handling
rg -n "func \\(r \\*Runner\\) Run|Drain|saveResult\\(" internal/runner/runner.goRepository: VirtualMetric/PipeBench
Length of output: 10926
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
path = Path("internal/runner/runner.go")
lines = path.read_text().splitlines()
def show(start, end, title):
print(f"\n==== {title} ({start}-{end}) ====")
for i in range(start, min(end, len(lines)) + 1):
print(f"{i:5d}: {lines[i-1]}")
show(161, 260, "Run")
show(1605, 1850, "runDiskPressureCorrectness / saveResult call")
show(7267, 7315, "saveResult")
PYRepository: VirtualMetric/PipeBench
Length of output: 18007
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
import re
# Search for Drain references and results store methods
paths = [Path(p) for p in [
"internal/runner/runner.go",
"results",
]]
for p in paths:
if p.is_file():
print(f"\n=== {p} ===")
text = p.read_text().splitlines()
for i, line in enumerate(text, 1):
if re.search(r'\bDrain\b|saveResult\(|Store|Save\(', line):
print(f"{i:5d}: {line}")
elif p.is_dir():
print(f"\n=== {p}/ (grep) ===")
for file in p.rglob("*.go"):
text = file.read_text().splitlines()
for i, line in enumerate(text, 1):
if re.search(r'\bDrain\b|func \(.*Save\(|type Store|Save\(', line):
print(f"{file}:{i}: {line}")
PYRepository: VirtualMetric/PipeBench
Length of output: 1919
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
path = Path("internal/runner/runner.go")
lines = path.read_text().splitlines()
for start, end, title in [
(470, 505, "performance-type drain handling"),
(930, 955, "drain-mode save path"),
]:
print(f"\n==== {title} ({start}-{end}) ====")
for i in range(start, min(end, len(lines)) + 1):
print(f"{i:5d}: {lines[i-1]}")
PYRepository: VirtualMetric/PipeBench
Length of output: 3066
disk_pressure_correctness should honor --drain
Run() only suppresses persistence for performance cases in drain mode; runDiskPressureCorrectness() still calls saveResult() unconditionally, so a diagnostic --drain run here overwrites the canonical results store.
🤖 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/runner/runner.go` around lines 1829 - 1832, Update
runDiskPressureCorrectness so it skips the saveResult persistence step when
--drain is enabled, matching the drain behavior in Run(). Preserve result
generation and return handling, but prevent diagnostic drain runs from
overwriting the canonical results store.
| 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)) | ||
| } |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '--- subjectDiskUsage references ---\n'
rg -n "subjectDiskUsage|SubjectDisk|Validate\(" internal/runner internal -g '*.go'
printf '\n--- runner.go around subjectDiskUsage ---\n'
nl -ba internal/runner/runner.go | sed -n '1825,1895p'
printf '\n--- config definitions for SubjectDisk / validation ---\n'
rg -n "type .*SubjectDisk|SubjectDisk struct|Path string|func .*Validate" -g '*.go' internal config .
printf '\n--- relevant file list ---\n'
git ls-files | rg '(^|/)(runner|config).*\.(go)$'Repository: VirtualMetric/PipeBench
Length of output: 252
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- subjectDiskUsage references ---'
rg -n "subjectDiskUsage|SubjectDisk|Validate\(" internal/runner internal -g '*.go' || true
echo
echo '--- runner.go around subjectDiskUsage ---'
nl -ba internal/runner/runner.go | sed -n '1825,1895p'
echo
echo '--- config definitions for SubjectDisk / validation ---'
rg -n "type .*SubjectDisk|SubjectDisk struct|Path string|func .*Validate" -g '*.go' .
echo
echo '--- candidate files ---'
git ls-files | rg '(^|/)(runner|config).*\.(go)$'Repository: VirtualMetric/PipeBench
Length of output: 9677
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- internal/config/case.go:1048-1068 ---'
sed -n '1048,1068p' internal/config/case.go
echo
echo '--- internal/config/case.go:2438-2452 ---'
sed -n '2438,2452p' internal/config/case.go
echo
echo '--- internal/runner/runner.go:1856-1866 ---'
sed -n '1856,1866p' internal/runner/runner.goRepository: VirtualMetric/PipeBench
Length of output: 2450
Remove the shell from subjectDiskUsage. tc.SubjectDisk.Path is user-controlled config, and Validate() only enforces an absolute path, so a crafted value with shell metacharacters can execute arbitrary commands via sh -c. Call du directly with the path as an argv element and read the size from its output.
🧰 Tools
🪛 ast-grep (0.44.1)
[error] 1862-1863: An argument passed to exec.Command/exec.CommandContext is built by concatenating a string literal with dynamic input. If that input is attacker-controlled (and especially when the command is a shell such as sh -c/bash -c), this enables OS command injection. Pass untrusted data as separate, fixed arguments instead of interpolating it into a command string, avoid invoking a shell, and validate/escape the input where a shell is unavoidable.
Context: exec.Command("docker", "exec", container, "sh", "-c",
"du -sh "+tc.SubjectDisk.Path+" 2>/dev/null | cut -f1")
Note: [CWE-78] Improper Neutralization of Special Elements used in an OS Command ('OS Command Injection').
(command-injection-exec-concat-arg-go)
🤖 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/runner/runner.go` around lines 1859 - 1869, Update subjectDiskUsage
to remove the sh -c invocation and execute du directly, passing
tc.SubjectDisk.Path as a separate argv element so shell metacharacters cannot be
interpreted. Preserve the existing error handling and trim the returned du
output before returning it.
Source: Linters/SAST tools
1 participant
What
Three subject-agnostic additions that let cases exercise disk-exhaustion behavior — what a pipeline does when its buffer/storage volume fills up (crash? silent drop? backpressure?):
subject_diskcase block — mounts a size-limited tmpfs (mode01777so non-root images can write) at a case-chosen path inside the singular subject service, giving the subject a small dedicated "disk" without a specially built image:Size strings are validated at case load (
Validate) and pre-parsed to an integer byte count at compose render — the two are kept in lockstep (tested) so a case that loads can never fail later with a cryptic compose error.disk_pressure_correctnesscase type — receiver stays DOWN while a duration-bounded generator floods far more data than the volume holds. The subject must survive the pressure window (container-alive check), and once the receiver comes up, deliver everything it accepted (loss ≤expected_loss_pct). A subject that fills the volume until its durable layer hits ENOSPC fails on loss and/or the crash check. The runner logs the volume's actual usage after the flood for the report.Generator fidelity fix (sequenced mode) — rotate a pool of 64 distinct random paddings instead of one shared buffer. With identical padding on every line, subjects that compress their batches shrank the stream ~100:1 — a 1.2 GB flood landed only ~11 MB on the subject's storage volume, quietly defeating any disk-pressure scenario (and flattering compression-friendly workloads in general). Distinct paddings make wire bytes ≈ stored bytes at a cost of 64×
line_sizesetup memory per connection worker.bench-generatorimage needs a rebuild/push for this to take effect.Why
Verified end-to-end against a real subject: with a 64 MB
subject_disk, a build without disk admission control fills the volume to 100% and starts failing its store writes (insufficient resources), while a build with disk backpressure stays bounded and survives. Without the generator padding fix, neither outcome is reachable — compression absorbs the flood.Tests
TestParseByteSize— size-string table (units, case-insensitivity, whitespace, rejects64bb/1.5g/zero/negative)TestValidateSubjectDisk— case-load validation (absolute path, size shape, both-fields-required)TestComposeRendersSubjectDiskTmpfs— tmpfs block renders with pre-parsed byte size + mode, absent without the block, malformed size fails rendergo build ./... && go vet ./internal/... && go test ./internal/...all green; generator module builds standalone🤖 Generated with Claude Code
Summary by CodeRabbit