From 5bf609dc83a79d52d1fe92075217083d2f8d37ab Mon Sep 17 00:00:00 2001 From: Stefan Date: Tue, 11 Aug 2026 09:44:22 +0200 Subject: [PATCH 1/2] perf(upload): skip suite objects the bucket already holds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A suite directory is content-addressed: the hash is a digest of its pre-run and test step file contents, so a given key under suites// always holds the same bytes. UploadSuiteDir re-sent all of them on every completed run. For the jochemnet bloatnet suite that is ~12.4 GB a run — a 9.4 GiB pre-run bundle plus 2.96 GB of fixtures — and that suite alone completes 12.8 runs a day, so roughly 158 GB/day of PUTs that overwrite identical objects. It only went unnoticed because the pre-run bundle was failing with EntityTooLarge before it could be sent twice. List the destination prefix once (3 requests for 2733 keys, versus a HEAD per file) and skip anything already there at the same size. Where the ETag is a plain MD5 — objects small enough to have gone up in a single part — verify that too, so the decision is content-based rather than size-based for all but the one file whose identity the suite hash already pins. A multipart ETag is a digest of part digests, so checking it would mean re-reading the whole bundle, the exact cost this skip exists to avoid. summary.json is never skipped: metadata labels can change between runs without affecting the suite hash, which is why CreateSuiteOutput rewrites it every time. --- pkg/upload/s3.go | 141 +++++++++++++++++++++++++++++++++++++++++- pkg/upload/s3_test.go | 101 +++++++++++++++++++++++++++++- 2 files changed, 237 insertions(+), 5 deletions(-) diff --git a/pkg/upload/s3.go b/pkg/upload/s3.go index d413ea0..2c233a5 100644 --- a/pkg/upload/s3.go +++ b/pkg/upload/s3.go @@ -2,7 +2,10 @@ package upload import ( "context" + "crypto/md5" //nolint:gosec // S3 defines the ETag as MD5; not a security check + "encoding/hex" "fmt" + "io" "mime" "os" "path/filepath" @@ -113,6 +116,7 @@ func (u *s3Uploader) Preflight(ctx context.Context) error { type uploadJob struct { localPath string key string + size int64 } // Upload walks localDir and uploads all files to S3 under the configured prefix. @@ -149,6 +153,7 @@ func (u *s3Uploader) collectJobs(localDir, keyPrefix string) ([]uploadJob, error jobs = append(jobs, uploadJob{ localPath: path, key: keyPrefix + "/" + filepath.ToSlash(relPath), + size: info.Size(), }) return nil @@ -302,7 +307,9 @@ func (u *s3Uploader) resolvePrefix(baseName string) string { } // UploadSuiteDir uploads all files in a suite directory to S3 under -// prefix + "/suites/" + dirname. +// prefix + "/suites/" + dirname. Objects already present at the same size are +// skipped: a suite hash is derived from its step file contents, so a given key +// under it always holds the same bytes and re-sending them is pure waste. func (u *s3Uploader) UploadSuiteDir(ctx context.Context, localSuiteDir string) error { prefix := u.cfg.Prefix if prefix == "" { @@ -316,7 +323,137 @@ func (u *s3Uploader) UploadSuiteDir(ctx context.Context, localSuiteDir string) e return fmt.Errorf("walking suite directory %s: %w", localSuiteDir, err) } - return u.uploadJobs(ctx, jobs, keyPrefix) + remote, err := u.listObjects(ctx, keyPrefix) + if err != nil { + // A listing failure only costs us the optimisation, so fall back to + // uploading everything rather than failing the suite. + u.log.WithError(err).WithField("prefix", keyPrefix). + Warn("Failed to list existing suite objects; uploading all files") + + return u.uploadJobs(ctx, jobs, keyPrefix) + } + + pending := make([]uploadJob, 0, len(jobs)) + + var skipped, skippedBytes int64 + + for _, job := range jobs { + if u.suiteObjectUnchanged(job, remote[job.key]) { + skipped++ + skippedBytes += job.size + + continue + } + + pending = append(pending, job) + } + + if skipped > 0 { + u.log.WithFields(logrus.Fields{ + "skipped": skipped, + "skipped_bytes": skippedBytes, + "pending": len(pending), + "prefix": keyPrefix, + }).Info("Skipping suite objects already present at the same size") + } + + return u.uploadJobs(ctx, pending, keyPrefix) +} + +// suiteSummaryFile is the one file in a suite directory that is rewritten +// rather than content-addressed. +const suiteSummaryFile = "summary.json" + +// remoteObject is what a listing tells us about an object already in the bucket. +type remoteObject struct { + size int64 + etag string +} + +// suiteObjectUnchanged reports whether the bucket already holds this file. +// +// Size is the baseline test: a suite hash is derived from its step file +// contents, so a given key under it always holds the same bytes. Where the +// ETag is a plain MD5 — objects small enough to have gone up in a single part +// — we verify that too, which costs a local read but no network. Multipart +// ETags are a digest of part digests, so checking one means re-reading the +// whole file, the exact cost this skip exists to avoid; those fall back to +// size, backed by the content-addressing. +func (u *s3Uploader) suiteObjectUnchanged(job uploadJob, remote remoteObject) bool { + // summary.json is rewritten on every run — metadata labels can change + // without affecting the suite hash — so it never gets skipped. + if filepath.Base(job.key) == suiteSummaryFile { + return false + } + + if remote.size != job.size || remote.size == 0 { + return false + } + + if job.size >= uploadPartSize { + return true + } + + sum, err := fileMD5(job.localPath) + if err != nil { + u.log.WithError(err).WithField("path", job.localPath). + Debug("Failed to hash local file; re-uploading") + + return false + } + + return strings.Trim(remote.etag, `"`) == sum +} + +// fileMD5 returns the hex MD5 of a file, matching how S3 reports the ETag of a +// single-part object. Not used as a security primitive. +func fileMD5(path string) (string, error) { + f, err := os.Open(path) + if err != nil { + return "", fmt.Errorf("opening file: %w", err) + } + + defer func() { _ = f.Close() }() + + h := md5.New() //nolint:gosec // S3 defines the ETag as MD5; not a security check + + if _, err := io.Copy(h, f); err != nil { + return "", fmt.Errorf("hashing file: %w", err) + } + + return hex.EncodeToString(h.Sum(nil)), nil +} + +// listObjects returns every object under prefix, keyed by full key. +func (u *s3Uploader) listObjects( + ctx context.Context, prefix string, +) (map[string]remoteObject, error) { + objects := make(map[string]remoteObject) + + paginator := s3.NewListObjectsV2Paginator(u.client, &s3.ListObjectsV2Input{ + Bucket: aws.String(u.cfg.Bucket), + Prefix: aws.String(prefix + "/"), + }) + + for paginator.HasMorePages() { + page, err := paginator.NextPage(ctx) + if err != nil { + return nil, fmt.Errorf("listing objects under %q: %w", prefix, err) + } + + for _, obj := range page.Contents { + if obj.Key == nil || obj.Size == nil { + continue + } + + objects[*obj.Key] = remoteObject{ + size: *obj.Size, + etag: aws.ToString(obj.ETag), + } + } + } + + return objects, nil } // detectContentType returns a MIME type based on file extension. diff --git a/pkg/upload/s3_test.go b/pkg/upload/s3_test.go index cb22c15..960dcbb 100644 --- a/pkg/upload/s3_test.go +++ b/pkg/upload/s3_test.go @@ -1,11 +1,15 @@ package upload import ( + "crypto/md5" //nolint:gosec // mirrors how S3 computes a single-part ETag + "encoding/hex" "io" "net/http" "net/http/httptest" "os" "path/filepath" + "strconv" + "strings" "sync" "testing" @@ -96,6 +100,26 @@ type fakeS3 struct { singlePuts []string multiparts []string parts int + // stored mimics bucket contents so ListObjectsV2 can report them back. + stored map[string]storedObject + partBytes map[string]int64 +} + +type storedObject struct { + size int64 + etag string +} + +func newFakeS3() *fakeS3 { + return &fakeS3{ + stored: make(map[string]storedObject), + partBytes: make(map[string]int64), + } +} + +// store records an object under its bucket-relative key. Callers hold f.mu. +func (f *fakeS3) store(path string, size int64, etag string) { + f.stored[strings.TrimPrefix(path, "/b/")] = storedObject{size: size, etag: etag} } func (f *fakeS3) handler() http.Handler { @@ -107,6 +131,20 @@ func (f *fakeS3) handler() http.Handler { defer f.mu.Unlock() switch { + case r.Method == http.MethodGet && q.Get("list-type") == "2": + w.Header().Set("Content-Type", "application/xml") + + body := `false` + + for k, obj := range f.stored { + if strings.HasPrefix(k, q.Get("prefix")) { + body += `` + k + + `` + strconv.FormatInt(obj.size, 10) + + `"` + obj.etag + `"` + } + } + + _, _ = w.Write([]byte(body + ``)) case r.Method == http.MethodPost && q.Has("uploads"): w.Header().Set("Content-Type", "application/xml") f.multiparts = append(f.multiparts, key) @@ -114,17 +152,25 @@ func (f *fakeS3) handler() http.Handler { `b` + key + `up-1` + ``)) case r.Method == http.MethodPut && q.Has("partNumber"): - _, _ = io.Copy(io.Discard, r.Body) + n, _ := io.Copy(io.Discard, r.Body) f.parts++ + f.partBytes[key] += n w.Header().Set("ETag", `"etag"`) case r.Method == http.MethodPost && q.Has("uploadId"): w.Header().Set("Content-Type", "application/xml") + + // A real multipart ETag is a digest of the part digests with a + // "-N" suffix, never a plain MD5 of the object. + f.store(key, f.partBytes[key], "composite-2") + _, _ = w.Write([]byte(`` + `b` + key + `"etag"` + ``)) case r.Method == http.MethodPut: - _, _ = io.Copy(io.Discard, r.Body) + h := md5.New() //nolint:gosec // mirrors how S3 computes a single-part ETag + n, _ := io.Copy(h, r.Body) f.singlePuts = append(f.singlePuts, key) + f.store(key, n, hex.EncodeToString(h.Sum(nil))) w.Header().Set("ETag", `"etag"`) default: w.WriteHeader(http.StatusNotImplemented) @@ -136,7 +182,7 @@ func (f *fakeS3) handler() http.Handler { // PutObject caps at 5 GiB and fails with EntityTooLarge on the multi-GB // pre-run bundles a stateful suite carries. func TestUploadFileSplitsLargeFiles(t *testing.T) { - fake := &fakeS3{} + fake := newFakeS3() srv := httptest.NewServer(fake.handler()) defer srv.Close() @@ -173,3 +219,52 @@ func TestUploadFileSplitsLargeFiles(t *testing.T) { assert.Equal(t, 2, fake.parts) assert.Equal(t, []string{"/b/suites/h/small.bin"}, fake.singlePuts) } + +// A suite directory is content-addressed by its hash, so re-uploading it on +// every run re-sends bytes the bucket already holds — for a stateful suite +// that is ~12 GB a run, most of it one pre-run bundle. +func TestUploadSuiteDirSkipsUnchangedObjects(t *testing.T) { + fake := newFakeS3() + srv := httptest.NewServer(fake.handler()) + + defer srv.Close() + + uploader, err := NewS3Uploader(logrus.New(), &config.S3UploadConfig{ + Bucket: "b", + EndpointURL: srv.URL, + Region: "us-east-1", + AccessKeyID: "id", + SecretAccessKey: "secret", + ForcePathStyle: true, + ParallelUploads: 4, + Prefix: "results", + }) + require.NoError(t, err) + + dir := filepath.Join(t.TempDir(), "0d93b5bf3b970403") + require.NoError(t, os.MkdirAll(filepath.Join(dir, "benchmark", "t1"), 0o750)) + require.NoError(t, os.MkdirAll(filepath.Join(dir, ".eest-meta"), 0o750)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "benchmark", "t1", "test.request"), []byte("payload"), 0o600)) + require.NoError(t, os.WriteFile(filepath.Join(dir, ".eest-meta", "fixtures.ini"), []byte("meta"), 0o600)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "summary.json"), []byte(`{"hash":"x"}`), 0o600)) + + require.NoError(t, uploader.UploadSuiteDir(t.Context(), dir)) + assert.Len(t, fake.singlePuts, 3, "first upload sends every file") + + fake.singlePuts = nil + + require.NoError(t, uploader.UploadSuiteDir(t.Context(), dir)) + + // summary.json is rewritten every run — labels change without changing the + // suite hash — so it alone is re-sent. + assert.Equal(t, []string{"/b/results/suites/0d93b5bf3b970403/summary.json"}, fake.singlePuts) + + // A changed file is re-sent even though its key already exists. + fake.singlePuts = nil + require.NoError(t, os.WriteFile(filepath.Join(dir, "benchmark", "t1", "test.request"), []byte("CHANGED"), 0o600)) + require.NoError(t, uploader.UploadSuiteDir(t.Context(), dir)) + assert.ElementsMatch(t, []string{ + "/b/results/suites/0d93b5bf3b970403/summary.json", + "/b/results/suites/0d93b5bf3b970403/benchmark/t1/test.request", + }, fake.singlePuts, "same size, different bytes: the ETag catches it") +} From 405330b44fbded8e134f0c3e20d0c3dafa7dedc3 Mon Sep 17 00:00:00 2001 From: Stefan Date: Tue, 11 Aug 2026 10:24:51 +0200 Subject: [PATCH 2/2] refactor(upload): compare size alone when skipping suite objects MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The ETag check was there to catch same-size-different-bytes. With the pre-run cap in #304 every remaining suite file is under the part size, so that branch covers all of them — meaning a full read and MD5 of the whole suite, 1-3 GB, on every run, to avoid re-sending the same 1-3 GB. It was insurance against something the design already rules out: a suite hash is a digest of exactly these step files, and only benchmarkoor writes under that prefix, so a key cannot hold different bytes at the same size without the hash changing too. Drop it and compare size, which the listing already gives us for free. summary.json stays exempt. It is the one file that is rewritten rather than content-addressed, so a label edit that happens to preserve its length is plausible in a way it is not for the rest. Net 63 lines and one crypto import gone. --- pkg/upload/s3.go | 89 +++++++------------------------------------ pkg/upload/s3_test.go | 37 +++++++----------- 2 files changed, 27 insertions(+), 99 deletions(-) diff --git a/pkg/upload/s3.go b/pkg/upload/s3.go index 2c233a5..3a77a3f 100644 --- a/pkg/upload/s3.go +++ b/pkg/upload/s3.go @@ -2,10 +2,7 @@ package upload import ( "context" - "crypto/md5" //nolint:gosec // S3 defines the ETag as MD5; not a security check - "encoding/hex" "fmt" - "io" "mime" "os" "path/filepath" @@ -307,8 +304,8 @@ func (u *s3Uploader) resolvePrefix(baseName string) string { } // UploadSuiteDir uploads all files in a suite directory to S3 under -// prefix + "/suites/" + dirname. Objects already present at the same size are -// skipped: a suite hash is derived from its step file contents, so a given key +// prefix + "/suites/" + dirname. Objects already there at the same size are +// skipped: a suite hash is a digest of its step file contents, so a given key // under it always holds the same bytes and re-sending them is pure waste. func (u *s3Uploader) UploadSuiteDir(ctx context.Context, localSuiteDir string) error { prefix := u.cfg.Prefix @@ -323,7 +320,7 @@ func (u *s3Uploader) UploadSuiteDir(ctx context.Context, localSuiteDir string) e return fmt.Errorf("walking suite directory %s: %w", localSuiteDir, err) } - remote, err := u.listObjects(ctx, keyPrefix) + remote, err := u.listSizes(ctx, keyPrefix) if err != nil { // A listing failure only costs us the optimisation, so fall back to // uploading everything rather than failing the suite. @@ -338,7 +335,10 @@ func (u *s3Uploader) UploadSuiteDir(ctx context.Context, localSuiteDir string) e var skipped, skippedBytes int64 for _, job := range jobs { - if u.suiteObjectUnchanged(job, remote[job.key]) { + // summary.json is rewritten every run — metadata labels change without + // affecting the suite hash — so it is the one file never skipped. + if size, ok := remote[job.key]; ok && size == job.size && + filepath.Base(job.key) != suiteSummaryFile { skipped++ skippedBytes += job.size @@ -364,71 +364,11 @@ func (u *s3Uploader) UploadSuiteDir(ctx context.Context, localSuiteDir string) e // rather than content-addressed. const suiteSummaryFile = "summary.json" -// remoteObject is what a listing tells us about an object already in the bucket. -type remoteObject struct { - size int64 - etag string -} - -// suiteObjectUnchanged reports whether the bucket already holds this file. -// -// Size is the baseline test: a suite hash is derived from its step file -// contents, so a given key under it always holds the same bytes. Where the -// ETag is a plain MD5 — objects small enough to have gone up in a single part -// — we verify that too, which costs a local read but no network. Multipart -// ETags are a digest of part digests, so checking one means re-reading the -// whole file, the exact cost this skip exists to avoid; those fall back to -// size, backed by the content-addressing. -func (u *s3Uploader) suiteObjectUnchanged(job uploadJob, remote remoteObject) bool { - // summary.json is rewritten on every run — metadata labels can change - // without affecting the suite hash — so it never gets skipped. - if filepath.Base(job.key) == suiteSummaryFile { - return false - } - - if remote.size != job.size || remote.size == 0 { - return false - } - - if job.size >= uploadPartSize { - return true - } - - sum, err := fileMD5(job.localPath) - if err != nil { - u.log.WithError(err).WithField("path", job.localPath). - Debug("Failed to hash local file; re-uploading") - - return false - } - - return strings.Trim(remote.etag, `"`) == sum -} - -// fileMD5 returns the hex MD5 of a file, matching how S3 reports the ETag of a -// single-part object. Not used as a security primitive. -func fileMD5(path string) (string, error) { - f, err := os.Open(path) - if err != nil { - return "", fmt.Errorf("opening file: %w", err) - } - - defer func() { _ = f.Close() }() - - h := md5.New() //nolint:gosec // S3 defines the ETag as MD5; not a security check - - if _, err := io.Copy(h, f); err != nil { - return "", fmt.Errorf("hashing file: %w", err) - } - - return hex.EncodeToString(h.Sum(nil)), nil -} - -// listObjects returns every object under prefix, keyed by full key. -func (u *s3Uploader) listObjects( +// listSizes returns the size of every object under prefix, keyed by full key. +func (u *s3Uploader) listSizes( ctx context.Context, prefix string, -) (map[string]remoteObject, error) { - objects := make(map[string]remoteObject) +) (map[string]int64, error) { + sizes := make(map[string]int64) paginator := s3.NewListObjectsV2Paginator(u.client, &s3.ListObjectsV2Input{ Bucket: aws.String(u.cfg.Bucket), @@ -446,14 +386,11 @@ func (u *s3Uploader) listObjects( continue } - objects[*obj.Key] = remoteObject{ - size: *obj.Size, - etag: aws.ToString(obj.ETag), - } + sizes[*obj.Key] = *obj.Size } } - return objects, nil + return sizes, nil } // detectContentType returns a MIME type based on file extension. diff --git a/pkg/upload/s3_test.go b/pkg/upload/s3_test.go index 960dcbb..ae65b59 100644 --- a/pkg/upload/s3_test.go +++ b/pkg/upload/s3_test.go @@ -1,8 +1,6 @@ package upload import ( - "crypto/md5" //nolint:gosec // mirrors how S3 computes a single-part ETag - "encoding/hex" "io" "net/http" "net/http/httptest" @@ -101,25 +99,20 @@ type fakeS3 struct { multiparts []string parts int // stored mimics bucket contents so ListObjectsV2 can report them back. - stored map[string]storedObject + stored map[string]int64 partBytes map[string]int64 } -type storedObject struct { - size int64 - etag string -} - func newFakeS3() *fakeS3 { return &fakeS3{ - stored: make(map[string]storedObject), + stored: make(map[string]int64), partBytes: make(map[string]int64), } } // store records an object under its bucket-relative key. Callers hold f.mu. -func (f *fakeS3) store(path string, size int64, etag string) { - f.stored[strings.TrimPrefix(path, "/b/")] = storedObject{size: size, etag: etag} +func (f *fakeS3) store(path string, size int64) { + f.stored[strings.TrimPrefix(path, "/b/")] = size } func (f *fakeS3) handler() http.Handler { @@ -136,11 +129,10 @@ func (f *fakeS3) handler() http.Handler { body := `false` - for k, obj := range f.stored { + for k, size := range f.stored { if strings.HasPrefix(k, q.Get("prefix")) { body += `` + k + - `` + strconv.FormatInt(obj.size, 10) + - `"` + obj.etag + `"` + `` + strconv.FormatInt(size, 10) + `` } } @@ -159,18 +151,15 @@ func (f *fakeS3) handler() http.Handler { case r.Method == http.MethodPost && q.Has("uploadId"): w.Header().Set("Content-Type", "application/xml") - // A real multipart ETag is a digest of the part digests with a - // "-N" suffix, never a plain MD5 of the object. - f.store(key, f.partBytes[key], "composite-2") + f.store(key, f.partBytes[key]) _, _ = w.Write([]byte(`` + `b` + key + `"etag"` + ``)) case r.Method == http.MethodPut: - h := md5.New() //nolint:gosec // mirrors how S3 computes a single-part ETag - n, _ := io.Copy(h, r.Body) + n, _ := io.Copy(io.Discard, r.Body) f.singlePuts = append(f.singlePuts, key) - f.store(key, n, hex.EncodeToString(h.Sum(nil))) + f.store(key, n) w.Header().Set("ETag", `"etag"`) default: w.WriteHeader(http.StatusNotImplemented) @@ -259,12 +248,14 @@ func TestUploadSuiteDirSkipsUnchangedObjects(t *testing.T) { // suite hash — so it alone is re-sent. assert.Equal(t, []string{"/b/results/suites/0d93b5bf3b970403/summary.json"}, fake.singlePuts) - // A changed file is re-sent even though its key already exists. + // A file whose size no longer matches is re-sent. Same size with different + // bytes is deliberately not detected: a suite key is content-addressed, so + // that cannot happen without the hash changing too. fake.singlePuts = nil - require.NoError(t, os.WriteFile(filepath.Join(dir, "benchmark", "t1", "test.request"), []byte("CHANGED"), 0o600)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "benchmark", "t1", "test.request"), []byte("longer payload"), 0o600)) require.NoError(t, uploader.UploadSuiteDir(t.Context(), dir)) assert.ElementsMatch(t, []string{ "/b/results/suites/0d93b5bf3b970403/summary.json", "/b/results/suites/0d93b5bf3b970403/benchmark/t1/test.request", - }, fake.singlePuts, "same size, different bytes: the ETag catches it") + }, fake.singlePuts) }