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

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 {

internal/orchestrator/awsinit.go

@@ -56,6 +56,25 @@ func writeAWSInit(path string, aws *config.AWSConfig) error {
 			"awslocal sns subscribe --topic-arn '%s' --protocol sqs --notification-endpoint '%s' --attributes RawMessageDelivery=true\n",
 			aws.TopicARN(s.Topic), aws.QueueARN(s.Queue))
 	}
+	for _, so := range aws.SeedObjects {
+		prefix := so.Prefix
+		if prefix == "" {
+			prefix = "seed/"
+		}
+		marker := so.Marker
+		if marker == "" {
+			marker = "SEED"
+		}
+		// Build each object body with awk (busybox ships it), then upload —
+		// so the objects exist before LocalStack reports init "completed" and
+		// the subject's depends_on gate releases. All values are int-formatted
+		// or charset-validated, so single-quoting is defense in depth.
+		fmt.Fprintf(&b, "i=0; while [ \"$i\" -lt %d ]; do\n", so.Objects)
+		fmt.Fprintf(&b, "  awk -v o=\"$i\" 'BEGIN{for(l=0;l<%d;l++) printf \"%s-OBJ%%d-LINE%%d\\n\", o, l}' > /tmp/pb-seed-obj\n", so.Lines, marker)
+		fmt.Fprintf(&b, "  awslocal s3 cp /tmp/pb-seed-obj 's3://%s/%sobj-'\"$i\"'.log'\n", so.Bucket, prefix)
+		b.WriteString("  i=$((i+1))\n")
+		b.WriteString("done\n")
+	}
 
 	b.WriteString("echo 'pipebench aws init complete'\n")
 	return os.WriteFile(path, []byte(b.String()), 0o755)

internal/orchestrator/compose_render_test.go

@@ -632,6 +632,53 @@ func TestComposeRendersAWS(t *testing.T) {
 	mustContain(t, string(script), `"QueueArn":"arn:aws:sqs:us-east-1:000000000000:bench-events"`)
 }
 
+// TestComposeRendersAWSSeedObjects verifies a `seed_objects:` entry renders a
+// pre-upload loop into the LocalStack init script (objects exist before the
+// subject starts, for list-mode backlog cases).
+func TestComposeRendersAWSSeedObjects(t *testing.T) {
+	tc := &config.TestCase{
+		Name:     "aws-seed",
+		Type:     "correctness",
+		Duration: "10s",
+		AWS: &config.AWSConfig{
+			Buckets: []string{"bench-in"},
+			SeedObjects: []config.AWSSeedObjects{
+				{Bucket: "bench-in", Prefix: "seed/", Objects: 10, Lines: 100, Marker: "SEED"},
+			},
+		},
+		Receiver:    config.ReceiverConfig{Mode: "tcp", Listen: ":9001"},
+		Correctness: config.CorrectnessConfig{MinReceived: 1000},
+	}
+	if err := tc.Validate(); err != nil {
+		t.Fatalf("validate: %v", err)
+	}
+	subj := config.Subject{Name: "vmetric", Image: "vmetric/director", Version: "2.0.3", ConfigPath: "/config.yml"}
+	tmp, err := os.MkdirTemp("", "compose-seed-")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(tmp)
+	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)
+	}
+	script, err := os.ReadFile(filepath.Join(tmp, "aws-init.sh"))
+	if err != nil {
+		t.Fatalf("aws-init.sh not written: %v", err)
+	}
+	mustContain(t, string(script), "awslocal s3 mb 's3://bench-in'")
+	mustContain(t, string(script), `while [ "$i" -lt 10 ]`)
+	mustContain(t, string(script), "for(l=0;l<100;l++)")
+	mustContain(t, string(script), "SEED-OBJ%d-LINE%d")
+	mustContain(t, string(script), "'s3://bench-in/seed/obj-'\"$i\"'.log'")
+}
+
 // TestComposeRendersAzure verifies an `azure:` case renders the Azurite
 // service plus the one-shot azure-init (receiver image), gates the subject on
 // init completion, and injects the connection string where needed.

internal/results/compare.go

@@ -279,6 +279,20 @@ func AggregateAllMetricsFromCSVWindow(csvPath string, startNs, endNs int64) (Agg
 			}
 		}
 
+		// A sample with zero CPU AND zero memory is a Docker stats snapshot
+		// of a stopped container (the crash/restart drivers stop and restart
+		// the subject mid-run, and Docker returns zeroed stats for an exited
+		// container). A running process never has 0 RSS, so real idle samples
+		// (cpu 0, mem > 0) are kept; counting stopped-window rows would
+		// dilute the averages with time the subject wasn't running.
+		if cpuIdx >= 0 && cpuIdx < len(record) && memIdx >= 0 && memIdx < len(record) {
+			cpuV, _ := strconv.ParseFloat(record[cpuIdx], 64)
+			memV, _ := strconv.ParseFloat(record[memIdx], 64)
+			if cpuV == 0 && memV == 0 {
+				continue
+			}
+		}
+
 		if cpuIdx >= 0 && cpuIdx < len(record) {
 			v, _ := strconv.ParseFloat(record[cpuIdx], 64)
 			cpuSum += v

internal/results/compare_test.go

@@ -0,0 +1,103 @@
+package results
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func writeMetricsCSV(t *testing.T, rows string) string {
+	t.Helper()
+	path := filepath.Join(t.TempDir(), "metrics.csv")
+	header := "epoch,cpu_usr,mem_used,net_recv,net_send,dsk_read,dsk_writ,load_avg1,load_avg5,load_avg15\n"
+	if err := os.WriteFile(path, []byte(header+rows), 0o644); err != nil {
+		t.Fatalf("writing csv: %v", err)
+	}
+	return path
+}
+
+func TestAggregateAllMetricsFromCSV(t *testing.T) {
+	t.Parallel()
+
+	// Stopped-container samples (cpu 0 AND mem 0) are dropped so the
+	// crash/restart down window doesn't dilute the averages.
+	t.Run("all-zero rows skipped", func(t *testing.T) {
+		t.Parallel()
+		csv := writeMetricsCSV(t,
+			"100,0,0,0,0,0,0,0,0,0\n"+
+				"101,10,104857600,0,0,0,0,0,0,0\n"+
+				"102,0,0,0,0,0,0,0,0,0\n"+
+				"103,20,209715200,0,0,0,0,0,0,0\n")
+		m, err := AggregateAllMetricsFromCSV(csv)
+		if err != nil {
+			t.Fatalf("aggregate: %v", err)
+		}
+		if m.Samples != 2 {
+			t.Fatalf("Samples = %d, want 2", m.Samples)
+		}
+		if m.CPUAvg != 15 || m.CPUMax != 20 {
+			t.Fatalf("cpu avg/max = %v/%v, want 15/20", m.CPUAvg, m.CPUMax)
+		}
+		if m.MemAvgMB != 150 || m.MemMaxMB != 200 {
+			t.Fatalf("mem avg/max = %v/%v, want 150/200", m.MemAvgMB, m.MemMaxMB)
+		}
+	})
+
+	// A running-but-idle sample (cpu 0, mem > 0) is a real sample and stays.
+	t.Run("idle row kept", func(t *testing.T) {
+		t.Parallel()
+		csv := writeMetricsCSV(t,
+			"100,0,104857600,0,0,0,0,0,0,0\n"+
+				"101,10,104857600,0,0,0,0,0,0,0\n")
+		m, err := AggregateAllMetricsFromCSV(csv)
+		if err != nil {
+			t.Fatalf("aggregate: %v", err)
+		}
+		if m.Samples != 2 {
+			t.Fatalf("Samples = %d, want 2", m.Samples)
+		}
+		if m.CPUAvg != 5 {
+			t.Fatalf("CPUAvg = %v, want 5", m.CPUAvg)
+		}
+	})
+
+	// Net/disk totals accumulate only over surviving rows.
+	t.Run("io totals over kept rows", func(t *testing.T) {
+		t.Parallel()
+		csv := writeMetricsCSV(t,
+			"100,0,0,999,999,999,999,0,0,0\n"+
+				"101,10,104857600,100,200,300,400,0,0,0\n")
+		m, err := AggregateAllMetricsFromCSV(csv)
+		if err != nil {
+			t.Fatalf("aggregate: %v", err)
+		}
+		if m.NetRecv != 100 || m.NetSend != 200 || m.DiskRead != 300 || m.DiskWrite != 400 {
+			t.Fatalf("io = net %d/%d disk %d/%d, want 100/200 300/400",
+				m.NetRecv, m.NetSend, m.DiskRead, m.DiskWrite)
+		}
+	})
+}
+
+func TestAggregateAllMetricsFromCSVWindow(t *testing.T) {
+	t.Parallel()
+
+	// The epoch window filter still applies alongside the zero-row skip.
+	csv := writeMetricsCSV(t,
+		"100,10,104857600,0,0,0,0,0,0,0\n"+
+			"200,0,0,0,0,0,0,0,0,0\n"+
+			"201,30,314572800,0,0,0,0,0,0,0\n"+
+			"300,50,524288000,0,0,0,0,0,0,0\n")
+	startNs := int64(200) * int64(time.Second)
+	endNs := int64(250) * int64(time.Second)
+	m, err := AggregateAllMetricsFromCSVWindow(csv, startNs, endNs)
+	if err != nil {
+		t.Fatalf("aggregate: %v", err)
+	}
+	if m.Samples != 1 {
+		t.Fatalf("Samples = %d, want 1 (window keeps 200-201, zero row dropped)", m.Samples)
+	}
+	if m.CPUAvg != 30 || m.MemMaxMB != 300 {
+		t.Fatalf("cpu/mem = %v/%v, want 30/300", m.CPUAvg, m.MemMaxMB)
+	}
+}