diff --git a/pkg/upload/s3.go b/pkg/upload/s3.go
index d413ea0..3a77a3f 100644
--- a/pkg/upload/s3.go
+++ b/pkg/upload/s3.go
@@ -113,6 +113,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 +150,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 +304,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 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
if prefix == "" {
@@ -316,7 +320,77 @@ 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.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.
+ 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 {
+ // 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
+
+ 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"
+
+// listSizes returns the size of every object under prefix, keyed by full key.
+func (u *s3Uploader) listSizes(
+ ctx context.Context, prefix string,
+) (map[string]int64, error) {
+ sizes := make(map[string]int64)
+
+ 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
+ }
+
+ sizes[*obj.Key] = *obj.Size
+ }
+ }
+
+ 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 cb22c15..ae65b59 100644
--- a/pkg/upload/s3_test.go
+++ b/pkg/upload/s3_test.go
@@ -6,6 +6,8 @@ import (
"net/http/httptest"
"os"
"path/filepath"
+ "strconv"
+ "strings"
"sync"
"testing"
@@ -96,6 +98,21 @@ type fakeS3 struct {
singlePuts []string
multiparts []string
parts int
+ // stored mimics bucket contents so ListObjectsV2 can report them back.
+ stored map[string]int64
+ partBytes map[string]int64
+}
+
+func newFakeS3() *fakeS3 {
+ return &fakeS3{
+ 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) {
+ f.stored[strings.TrimPrefix(path, "/b/")] = size
}
func (f *fakeS3) handler() http.Handler {
@@ -107,6 +124,19 @@ 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, size := range f.stored {
+ if strings.HasPrefix(k, q.Get("prefix")) {
+ body += `` + k +
+ `` + strconv.FormatInt(size, 10) + ``
+ }
+ }
+
+ _, _ = 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 +144,22 @@ 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")
+
+ f.store(key, f.partBytes[key])
+
_, _ = w.Write([]byte(`` +
`b` + key + `"etag"` +
``))
case r.Method == http.MethodPut:
- _, _ = io.Copy(io.Discard, r.Body)
+ n, _ := io.Copy(io.Discard, r.Body)
f.singlePuts = append(f.singlePuts, key)
+ f.store(key, n)
w.Header().Set("ETag", `"etag"`)
default:
w.WriteHeader(http.StatusNotImplemented)
@@ -136,7 +171,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 +208,54 @@ 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 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("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)
+}