feat(orchestrator): add AWS S3 seed objects to LocalStack init

containers/collector/main.go

@@ -100,36 +100,75 @@ func main() {
 }
 
 func runDockerMode(ctx context.Context, ticker *time.Ticker, emit func(MetricsRow)) {
-	container := mustEnv("COLLECTOR_TARGET_CONTAINER")
+	// COLLECTOR_TARGET_CONTAINER is a comma-separated container list. Single-
+	// subject cases pass one name; director-cluster cases pass every node so
+	// each row captures the whole cluster's footprint (per-tick sum across
+	// nodes). Deltas (net/blkio) are computed per container against its own
+	// previous sample, then summed.
+	var targets []string
+	for _, t := range strings.Split(mustEnv("COLLECTOR_TARGET_CONTAINER"), ",") {
+		if t = strings.TrimSpace(t); t != "" {
+			targets = append(targets, t)
+		}
+	}
+	if len(targets) == 0 {
+		fmt.Fprintln(os.Stderr, "collector: COLLECTOR_TARGET_CONTAINER has no container names")
+		os.Exit(1)
+	}
 	dockerHost := getEnv("DOCKER_HOST", "unix:///var/run/docker.sock")
 
 	httpClient := buildClient(dockerHost)
-	var prev *dockerStats
+	prev := make([]*dockerStats, len(targets))
 	var errCount int
 
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			stats, err := fetchDockerStats(httpClient, dockerHost, container)
-			if err != nil {
-				errCount++
-				// Log the first error immediately, then every 30 ticks,
-				// so a silent permission/DNS/container-missing failure is
-				// diagnosable but we don't spam the run.
-				if errCount == 1 || errCount%30 == 0 {
-					fmt.Fprintf(os.Stderr, "collector: fetch stats (err #%d): %v\n", errCount, err)
+			var row MetricsRow
+			sampled := false
+			for i, container := range targets {
+				stats, err := fetchDockerStats(httpClient, dockerHost, container)
+				if err != nil {
+					errCount++
+					// Log the first error immediately, then every 30 ticks,
+					// so a silent permission/DNS/container-missing failure is
+					// diagnosable but we don't spam the run.
+					if errCount == 1 || errCount%30 == 0 {
+						fmt.Fprintf(os.Stderr, "collector: fetch stats %s (err #%d): %v\n", container, errCount, err)
+					}
+					continue
 				}
+				addRow(&row, dockerStatsToRow(stats, prev[i]))
+				prev[i] = stats
+				sampled = true
+			}
+			if !sampled {
 				continue
 			}
-			row := dockerStatsToRow(stats, prev)
+			row.Epoch = time.Now().Unix()
+			// Summed usage regularly exceeds 100 (multi-core containers,
+			// multi-node clusters); idle is a percentage, so floor it at 0.
+			row.CpuIdl = max(100.0-row.CpuUsr, 0)
 			emit(row)
-			prev = stats
 		}
 	}
 }
 
+// addRow accumulates one container's sample into the tick's combined row.
+// Epoch and CpuIdl are derived once per tick by the caller, not summed.
+func addRow(dst *MetricsRow, s MetricsRow) {
+	dst.CpuUsr += s.CpuUsr
+	dst.MemUsed += s.MemUsed
+	dst.MemCach += s.MemCach
+	dst.MemFree += s.MemFree
+	dst.NetRecv += s.NetRecv
+	dst.NetSend += s.NetSend
+	dst.DskRead += s.DskRead
+	dst.DskWrit += s.DskWrit
+}
+
 // --- Docker Stats API ---
 
 type dockerStats struct {
@@ -198,9 +237,10 @@ func fetchDockerStats(client *http.Client, dockerHost, container string) (*docke
 	return &s, nil
 }
 
+// dockerStatsToRow converts one container's stats snapshot into a partial
+// row: per-tick fields (Epoch, CpuIdl) are derived by the caller after the
+// per-target rows are summed, so they stay zero here.
 func dockerStatsToRow(cur *dockerStats, prev *dockerStats) MetricsRow {
-	now := time.Now().Unix()
-
 	var cpuPct float64
 	cpuDelta := float64(cur.CPUStats.CPUUsage.TotalUsage) - float64(cur.PreCPUStats.CPUUsage.TotalUsage)
 	sysDelta := float64(cur.CPUStats.SystemUsage) - float64(cur.PreCPUStats.SystemUsage)
@@ -269,9 +309,7 @@ func dockerStatsToRow(cur *dockerStats, prev *dockerStats) MetricsRow {
 	}
 
 	return MetricsRow{
-		Epoch:   now,
 		CpuUsr:  cpuPct,
-		CpuIdl:  100.0 - cpuPct,
 		MemUsed: memUsed,
 		MemCach: cache,
 		MemFree: memFree,

containers/collector/main_test.go

@@ -100,3 +100,44 @@ func TestMemUsage(t *testing.T) {
 		})
 	}
 }
+
+func TestAddRow(t *testing.T) {
+	var dst MetricsRow
+	addRow(&dst, MetricsRow{CpuUsr: 40, MemUsed: 100, MemCach: 10, MemFree: 50, NetRecv: 1, NetSend: 2, DskRead: 3, DskWrit: 4})
+	addRow(&dst, MetricsRow{CpuUsr: 15, MemUsed: 200, MemCach: 20, MemFree: 70, NetRecv: 10, NetSend: 20, DskRead: 30, DskWrit: 40})
+
+	if dst.CpuUsr != 55 {
+		t.Errorf("CpuUsr = %v, want 55", dst.CpuUsr)
+	}
+	if dst.MemUsed != 300 || dst.MemCach != 30 || dst.MemFree != 120 {
+		t.Errorf("mem = %d/%d/%d, want 300/30/120", dst.MemUsed, dst.MemCach, dst.MemFree)
+	}
+	if dst.NetRecv != 11 || dst.NetSend != 22 || dst.DskRead != 33 || dst.DskWrit != 44 {
+		t.Errorf("io = %d/%d/%d/%d, want 11/22/33/44", dst.NetRecv, dst.NetSend, dst.DskRead, dst.DskWrit)
+	}
+	// Epoch and CpuIdl are derived per tick by the caller, never summed.
+	if dst.Epoch != 0 || dst.CpuIdl != 0 {
+		t.Errorf("Epoch/CpuIdl = %d/%v, want 0/0", dst.Epoch, dst.CpuIdl)
+	}
+}
+
+func TestDockerStatsToRowLeavesPerTickFieldsZero(t *testing.T) {
+	// Epoch and CpuIdl are derived once per tick by runDockerMode after the
+	// per-target rows are summed (CpuIdl floored at 0 there, since summed
+	// usage can exceed 100). A per-container row must leave them zero, or a
+	// stale per-container idle would leak into the combined row's contract.
+	var s dockerStats
+	s.CPUStats.CPUUsage.TotalUsage = 2_000_000
+	s.CPUStats.SystemUsage = 10_000_000
+	s.CPUStats.OnlineCPUs = 4
+	s.PreCPUStats.CPUUsage.TotalUsage = 1_000_000
+	s.PreCPUStats.SystemUsage = 9_000_000
+
+	row := dockerStatsToRow(&s, nil)
+	if row.CpuUsr <= 0 {
+		t.Fatalf("CpuUsr = %v, want > 0 (sanity: stats deltas present)", row.CpuUsr)
+	}
+	if row.Epoch != 0 || row.CpuIdl != 0 {
+		t.Fatalf("Epoch/CpuIdl = %d/%v, want 0/0 (caller-derived per tick)", row.Epoch, row.CpuIdl)
+	}
+}

containers/receiver/main.go

@@ -642,10 +642,34 @@ func receiveTCP(cfg config, cnt *counters, val *validator) error {
 	}
 }
 
+// stampingReader refreshes the shard's receive-window timestamps on every
+// successful read from the connection. recordLine samples time.Now() only
+// every 1024 lines (and finish() only runs at connection close), so a burst
+// smaller than 1024 lines on a connection the sender keeps open would leave
+// lastNs == firstNs and the runner's receive-window rate at 0. Stamping per
+// read costs one time.Now() per socket read — already syscall-scale — and
+// bounds the window by actual socket activity.
+type stampingReader struct {
+	r     io.Reader
+	shard *connStats
+}
+
+func (sr *stampingReader) Read(p []byte) (int, error) {
+	n, err := sr.r.Read(p)
+	if n > 0 {
+		now := time.Now().UnixNano()
+		sr.shard.firstNs.CompareAndSwap(0, now)
+		sr.shard.lastNs.Store(now)
+	}
+	return n, err
+}
+
 func handleConn(conn net.Conn, shard *connStats, val *validator, cfg config) {
 	defer conn.Close()
-	defer shard.finish()
-	scanner := bufio.NewScanner(conn)
+	// No shard.finish() here: the stampingReader already recorded the exact
+	// time of the last data read. finish() would overwrite it with the close
+	// time, inflating the receive window when the sender idles before closing.
+	scanner := bufio.NewScanner(&stampingReader{r: conn, shard: shard})
 	scanner.Buffer(make([]byte, 1024*1024), 1024*1024)
 	needsValidation := cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.RequiredSubstring != ""
 	// Pass the scanner's internal slice directly. shard.recordLine + validator
@@ -660,6 +684,12 @@ func handleConn(conn net.Conn, shard *connStats, val *validator, cfg config) {
 			val.recordLine(b, cfg)
 		}
 	}
+	// A scan error (oversized line, connection reset) silently truncates the
+	// count for this connection — surface it so a lossy-looking run is
+	// diagnosable from the receiver log. EOF is not reported by Err().
+	if err := scanner.Err(); err != nil {
+		fmt.Fprintf(os.Stderr, "receiver: conn %s: %v\n", conn.RemoteAddr(), err)
+	}
 }
 
 func receiveFile(cfg config, cnt *counters, val *validator) error {

containers/receiver/main_test.go

@@ -0,0 +1,85 @@
+package main
+
+import (
+	"bufio"
+	"io"
+	"strings"
+	"testing"
+	"time"
+)
+
+// chunkedReader yields one fixed chunk per Read call with a small delay
+// between chunks, so consecutive reads land on distinct timestamps.
+type chunkedReader struct {
+	chunks []string
+	delay  time.Duration
+}
+
+func (c *chunkedReader) Read(p []byte) (int, error) {
+	if len(c.chunks) == 0 {
+		return 0, io.EOF
+	}
+	if c.delay > 0 {
+		time.Sleep(c.delay)
+	}
+	n := copy(p, c.chunks[0])
+	if n == len(c.chunks[0]) {
+		c.chunks = c.chunks[1:]
+	} else {
+		c.chunks[0] = c.chunks[0][n:]
+	}
+	return n, nil
+}
+
+func TestStampingReader(t *testing.T) {
+	// A burst far below recordLine's 1024-line sampling threshold must still
+	// produce a non-empty receive window (lastNs > firstNs) when the data
+	// arrives across multiple reads — the regression behind EPS 0 on
+	// 1,000-line crash/restart cases.
+	shard := &connStats{}
+	src := &chunkedReader{
+		chunks: []string{"a\nb\n", "c\nd\n"},
+		delay:  5 * time.Millisecond,
+	}
+	scanner := bufio.NewScanner(&stampingReader{r: src, shard: shard})
+	lines := 0
+	for scanner.Scan() {
+		shard.recordLine(int64(len(scanner.Bytes())) + 1)
+		lines++
+	}
+	if err := scanner.Err(); err != nil {
+		t.Fatalf("scan: %v", err)
+	}
+	if lines != 4 {
+		t.Fatalf("lines = %d, want 4", lines)
+	}
+	first, last := shard.firstNs.Load(), shard.lastNs.Load()
+	if first == 0 {
+		t.Fatal("firstNs not stamped")
+	}
+	if last <= first {
+		t.Fatalf("lastNs (%d) <= firstNs (%d): receive window empty", last, first)
+	}
+}
+
+func TestStampingReaderFirstNsStable(t *testing.T) {
+	// firstNs is stamped once on the first read and never moves.
+	shard := &connStats{}
+	sr := &stampingReader{r: strings.NewReader("x\n"), shard: shard}
+	buf := make([]byte, 16)
+	if _, err := sr.Read(buf); err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	first := shard.firstNs.Load()
+	time.Sleep(2 * time.Millisecond)
+	sr.r = strings.NewReader("y\n")
+	if _, err := sr.Read(buf); err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	if got := shard.firstNs.Load(); got != first {
+		t.Fatalf("firstNs moved: %d → %d", first, got)
+	}
+	if shard.lastNs.Load() <= first {
+		t.Fatal("lastNs not refreshed on second read")
+	}
+}

internal/config/cloud.go

@@ -77,6 +77,30 @@ type AWSConfig struct {
 	// Subscriptions wires SNS → SQS (RawMessageDelivery=true) so an
 	// SNS-target case is observable by the `sqs` receiver mode.
 	Subscriptions []AWSSubscription `yaml:"subscriptions"`
+
+	// SeedObjects pre-uploads synthetic objects into a declared bucket at
+	// init, before the subject starts — for list-mode / first-run backlog
+	// source cases where the subject must find objects already present rather
+	// than a generator streaming them in during the run.
+	SeedObjects []AWSSeedObjects `yaml:"seed_objects"`
+}
+
+// AWSSeedObjects pre-uploads Objects synthetic objects (Lines lines each,
+// each line prefixed with Marker) under Prefix in Bucket during LocalStack
+// init.
+type AWSSeedObjects struct {
+	// Bucket must be one of the declared Buckets.
+	Bucket string `yaml:"bucket"`
+	// Prefix is the key prefix for seeded objects (default "seed/").
+	Prefix string `yaml:"prefix"`
+	// Objects is the number of objects to create (> 0).
+	Objects int `yaml:"objects"`
+	// Lines is the number of lines per object (> 0).
+	Lines int `yaml:"lines"`
+	// Marker is the per-line content prefix (default "SEED"); validated
+	// against the cloud-name charset so it cannot inject into the init shell
+	// script.
+	Marker string `yaml:"marker"`
 }
 
 // AWSStream declares a Kinesis stream created at init.
@@ -420,6 +444,27 @@ func (tc *TestCase) validateAWS() error {
 			return err
 		}
 	}
+	for _, so := range tc.AWS.SeedObjects {
+		if _, ok := buckets[so.Bucket]; !ok {
+			return fmt.Errorf("case %q: seed_objects references undeclared bucket %q", tc.Name, so.Bucket)
+		}
+		if so.Prefix != "" {
+			if err := validateCloudName(tc.Name, "aws seed prefix", so.Prefix); err != nil {
+				return err
+			}
+		}
+		if so.Marker != "" {
+			if err := validateCloudName(tc.Name, "aws seed marker", so.Marker); err != nil {
+				return err
+			}
+		}
+		if so.Objects <= 0 {
+			return fmt.Errorf("case %q: seed_objects for bucket %q requires objects > 0, got %d", tc.Name, so.Bucket, so.Objects)
+		}
+		if so.Lines <= 0 {
+			return fmt.Errorf("case %q: seed_objects for bucket %q requires lines > 0, got %d", tc.Name, so.Bucket, so.Lines)
+		}
+	}
 	return nil
 }
 

internal/config/cloud_test.go

@@ -219,6 +219,43 @@ func TestAWSConfigDefaults(t *testing.T) {
 	}
 }
 
+func TestValidateAWSSeedObjects(t *testing.T) {
+	base := func(so AWSSeedObjects) *TestCase {
+		return &TestCase{
+			Name:     "seed-case",
+			Type:     "correctness",
+			Duration: "10s",
+			AWS: &AWSConfig{
+				Buckets:     []string{"bench-in"},
+				SeedObjects: []AWSSeedObjects{so},
+			},
+			Receiver:    ReceiverConfig{Mode: "tcp", Listen: ":9001"},
+			Correctness: CorrectnessConfig{},
+		}
+	}
+
+	tests := []struct {
+		name    string
+		so      AWSSeedObjects
+		wantErr bool
+	}{
+		{name: "valid", so: AWSSeedObjects{Bucket: "bench-in", Objects: 10, Lines: 100}},
+		{name: "valid with prefix and marker", so: AWSSeedObjects{Bucket: "bench-in", Prefix: "seed/", Objects: 1, Lines: 1, Marker: "SEED"}},
+		{name: "undeclared bucket", so: AWSSeedObjects{Bucket: "nope", Objects: 1, Lines: 1}, wantErr: true},
+		{name: "zero objects", so: AWSSeedObjects{Bucket: "bench-in", Objects: 0, Lines: 1}, wantErr: true},
+		{name: "zero lines", so: AWSSeedObjects{Bucket: "bench-in", Objects: 1, Lines: 0}, wantErr: true},
+		{name: "injecting marker", so: AWSSeedObjects{Bucket: "bench-in", Objects: 1, Lines: 1, Marker: "a'; rm -rf /"}, wantErr: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := base(tt.so).validateAWS()
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("validateAWS() err = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
 func TestMinioConfigDefaults(t *testing.T) {
 	m := &MinioConfig{Buckets: []string{"bench-out"}}
 	if got, want := m.ImageOrDefault(), "minio/minio:RELEASE.2025-04-22T22-12-26Z"; got != want {