diff --git a/cmd/sourcemaps/upload.go b/cmd/sourcemaps/upload.go index 6dac8c315..2600e3de7 100644 --- a/cmd/sourcemaps/upload.go +++ b/cmd/sourcemaps/upload.go @@ -2,6 +2,7 @@ package sourcemaps import ( "bytes" + "context" "encoding/json" "fmt" "io" @@ -15,6 +16,7 @@ import ( "github.com/spf13/cobra" "github.com/spf13/viper" + "golang.org/x/sync/errgroup" cmdAnalytics "github.com/launchdarkly/ldcli/cmd/analytics" "github.com/launchdarkly/ldcli/cmd/cliflags" @@ -34,6 +36,8 @@ const ( defaultPath = "." defaultBackendUrl = "https://pri.observability.app.launchdarkly.com" + uploadConcurrency = 8 + getSourceMapUrlsQuery = ` query GetSourceMapUploadUrls($api_key: String!, $project_id: String!, $paths: [String!]!) { get_source_map_upload_urls_ld( @@ -157,10 +161,8 @@ func runE(client resources.Client) func(cmd *cobra.Command, args []string) error return fmt.Errorf("failed to get upload URLs: %w", err) } - for i, file := range files { - if err := uploadFile(file.Path, uploadUrls[i], file.Name); err != nil { - return fmt.Errorf("failed to upload file %s: %w", file.Path, err) - } + if err := uploadFiles(files, uploadUrls); err != nil { + return err } fmt.Println("Successfully uploaded all sourcemaps") @@ -309,6 +311,29 @@ func getSourceMapUploadUrls(apiKey, projectID string, paths []string, backendUrl return urlsResp.Data.GetSourceMapUploadUrls, nil } +// uploadFiles uploads each file to its presigned URL, running up to +// uploadConcurrency uploads at once. After the first failure it starts no new +// uploads and returns that failure. +func uploadFiles(files []SourceMapFile, uploadUrls []string) error { + g, ctx := errgroup.WithContext(context.Background()) + g.SetLimit(uploadConcurrency) + + for i, file := range files { + g.Go(func() error { + if ctx.Err() != nil { + return nil + } + + if err := uploadFile(file.Path, uploadUrls[i], file.Name); err != nil { + return fmt.Errorf("failed to upload file %s: %w", file.Path, err) + } + return nil + }) + } + + return g.Wait() +} + func uploadFile(filePath, uploadUrl, name string) error { fileContent, err := os.ReadFile(filePath) if err != nil { diff --git a/cmd/sourcemaps/upload_test.go b/cmd/sourcemaps/upload_test.go index 3f78d0cd6..36a6da9e4 100644 --- a/cmd/sourcemaps/upload_test.go +++ b/cmd/sourcemaps/upload_test.go @@ -9,7 +9,9 @@ import ( "net/url" "os" "path/filepath" + "sync/atomic" "testing" + "time" "github.com/stretchr/testify/assert" @@ -128,6 +130,68 @@ func TestUploadFile(t *testing.T) { assert.Contains(t, err.Error(), "upload failed with status code: 500") } +func writeUploadTestFiles(t *testing.T, n int) []SourceMapFile { + tempDir := t.TempDir() + files := make([]SourceMapFile, 0, n) + for i := 0; i < n; i++ { + name := fmt.Sprintf("chunk%d.js.map", i) + path := filepath.Join(tempDir, name) + assert.NoError(t, os.WriteFile(path, []byte("{}"), 0644)) + files = append(files, SourceMapFile{Path: path, Name: name}) + } + return files +} + +func TestUploadFilesRunsConcurrently(t *testing.T) { + files := writeUploadTestFiles(t, uploadConcurrency*4) + + var inFlight, maxInFlight, uploaded atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + n := inFlight.Add(1) + defer inFlight.Add(-1) + for { + m := maxInFlight.Load() + if n <= m || maxInFlight.CompareAndSwap(m, n) { + break + } + } + time.Sleep(20 * time.Millisecond) + uploaded.Add(1) + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + urls := make([]string, len(files)) + for i := range urls { + urls[i] = fmt.Sprintf("%s/%d", server.URL, i) + } + + assert.NoError(t, uploadFiles(files, urls)) + assert.Equal(t, int32(len(files)), uploaded.Load()) + assert.Greater(t, maxInFlight.Load(), int32(1)) + assert.LessOrEqual(t, maxInFlight.Load(), int32(uploadConcurrency)) +} + +func TestUploadFilesStopsAfterFirstError(t *testing.T) { + files := writeUploadTestFiles(t, uploadConcurrency*10) + + var requests atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + w.WriteHeader(http.StatusForbidden) + })) + defer server.Close() + + urls := make([]string, len(files)) + for i := range urls { + urls[i] = server.URL + } + + err := uploadFiles(files, urls) + assert.ErrorContains(t, err, "upload failed with status code: 403") + assert.Less(t, requests.Load(), int32(len(files))) +} + func TestNewUploadCmd(t *testing.T) { client := resources.NewClient("") cmd := NewUploadCmd(client, func(accessToken, baseURI string, analyticsOptOut bool) analytics.Tracker { diff --git a/go.mod b/go.mod index e97c4f63a..984d3993a 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ require ( github.com/atotto/clipboard v0.1.4 github.com/aws/aws-sdk-go-v2 v1.47.3 github.com/aws/aws-sdk-go-v2/config v1.33.5 + github.com/aws/aws-sdk-go-v2/credentials v1.20.5 github.com/aws/aws-sdk-go-v2/service/devopsagent v1.17.0 github.com/aws/aws-sdk-go-v2/service/iam v1.64.0 github.com/aws/aws-sdk-go-v2/service/sts v1.51.4 @@ -48,6 +49,7 @@ require ( github.com/stretchr/testify v1.11.1 go.uber.org/mock v0.6.0 golang.org/x/exp v0.0.0-20240808152545-0cdaa3abc0fa + golang.org/x/sync v0.19.0 golang.org/x/term v0.40.0 gopkg.in/yaml.v3 v3.0.1 ) @@ -57,7 +59,6 @@ require ( github.com/alecthomas/chroma/v2 v2.14.0 // indirect github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.20 // indirect - github.com/aws/aws-sdk-go-v2/credentials v1.20.5 // indirect github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.20.0 // indirect github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.6 // indirect github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.6 // indirect @@ -127,7 +128,6 @@ require ( go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/mod v0.33.0 // indirect golang.org/x/net v0.50.0 // indirect - golang.org/x/sync v0.19.0 // indirect golang.org/x/sys v0.45.0 // indirect golang.org/x/text v0.34.0 // indirect golang.org/x/tools v0.42.0 // indirect