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) +}