From e0800970eedbe87c37cc1e8005a3dd4be698d55f Mon Sep 17 00:00:00 2001 From: jinyuchen724 <627388936@qq.com> Date: Mon, 29 Jun 2026 17:27:11 +0800 Subject: [PATCH 1/5] feat: add multiple endpoints support for load balancing Add --endpoint-urls flag to distribute requests across multiple S3 endpoints using round-robin load balancing. --- README.md | 14 +++ command/app.go | 26 +++++ e2e/app_test.go | 56 ++++++++++ storage/s3.go | 260 ++++++++++++++++++++++++++++++++++++++------- storage/s3_test.go | 99 +++++++++++++++++ storage/storage.go | 4 +- 6 files changed, 418 insertions(+), 41 deletions(-) diff --git a/README.md b/README.md index b6854e0d2..7f1c3f79d 100644 --- a/README.md +++ b/README.md @@ -566,6 +566,20 @@ all variants will return your GCS buckets. acceleration and GCS. If a custom endpoint is provided, it'll fallback to path-style. +### Multiple endpoints (load balancing) + +For high-throughput workloads, `s5cmd` can distribute requests across multiple S3-compatible endpoints using round-robin load balancing. + + s5cmd --endpoint-urls "http://10.26.16.129:17410,http://10.26.16.130:17410" cp 's3://bucket/*' . + +or using the environment variable: + + S3_ENDPOINT_URLS="http://10.26.16.129:17410,http://10.26.16.130:17410" s5cmd cp 's3://bucket/*' . + +Each request is dispatched to the next endpoint in round-robin order across all workers. + +> **Note:** `--endpoint-url` and `--endpoint-urls` cannot be used together. + ### Retry logic `s5cmd` uses an exponential backoff retry mechanism for transient or potential diff --git a/command/app.go b/command/app.go index f1699e43d..47de51119 100644 --- a/command/app.go +++ b/command/app.go @@ -46,6 +46,11 @@ var app = &cli.App{ Usage: "override default S3 host for custom services", EnvVars: []string{"S3_ENDPOINT_URL"}, }, + &cli.StringSliceFlag{ + Name: "endpoint-urls", + Usage: "multiple S3 endpoints for load balancing (comma-separated or multiple flags)", + EnvVars: []string{"S3_ENDPOINT_URLS"}, + }, &cli.BoolFlag{ Name: "no-verify-ssl", Usage: "disable SSL certificate verification", @@ -98,6 +103,7 @@ var app = &cli.App{ logLevel := c.String("log") isStat := c.Bool("stat") endpointURL := c.String("endpoint-url") + endpointURLs := c.StringSlice("endpoint-urls") log.Init(logLevel, printJSON) parallel.Init(workerCount) @@ -118,10 +124,18 @@ var app = &cli.App{ return err } + // Validate endpoint-url and endpoint-urls are not used together + if endpointURL != "" && len(endpointURLs) > 0 { + err := fmt.Errorf(`"endpoint-url" and "endpoint-urls" flags cannot be used together`) + printError(commandFromContext(c), c.Command.Name, err) + return err + } + if isStat { stat.InitStat() } + // Validate single endpoint URL if endpointURL != "" { if !strings.HasPrefix(endpointURL, "http") { err := fmt.Errorf(`bad value for --endpoint-url %v: scheme is missing. Must be of the form http:/// or https:///`, endpointURL) @@ -130,6 +144,17 @@ var app = &cli.App{ } } + // Validate multiple endpoint URLs + if len(endpointURLs) > 0 { + for _, ep := range endpointURLs { + if !strings.HasPrefix(ep, "http") { + err := fmt.Errorf(`bad value for --endpoint-urls %v: scheme is missing. Must be of the form http:/// or https:///`, ep) + printError(commandFromContext(c), c.Command.Name, err) + return err + } + } + } + return nil }, CommandNotFound: func(c *cli.Context, command string) { @@ -181,6 +206,7 @@ func NewStorageOpts(c *cli.Context) storage.Options { return storage.Options{ DryRun: c.Bool("dry-run"), Endpoint: c.String("endpoint-url"), + Endpoints: c.StringSlice("endpoint-urls"), MaxRetries: c.Int("retry-count"), NoSignRequest: c.Bool("no-sign-request"), NoVerifySSL: c.Bool("no-verify-ssl"), diff --git a/e2e/app_test.go b/e2e/app_test.go index 80c3d5720..56c5319b0 100644 --- a/e2e/app_test.go +++ b/e2e/app_test.go @@ -312,3 +312,59 @@ func TestAppEndpointShouldHaveScheme(t *testing.T) { }) } } + +func TestAppEndpointURLsValidation(t *testing.T) { + t.Parallel() + + testcases := []struct { + name string + flags []string + expectedError error + expectedExitCode int + }{ + { + name: "valid_endpoint_urls", + flags: []string{"--endpoint-urls", "http://s3-1:9000,http://s3-2:9000"}, + expectedExitCode: 0, + }, + { + name: "endpoint_urls_env_var_valid", + flags: []string{"--endpoint-urls", "https://minio.example.com"}, + expectedExitCode: 0, + }, + { + name: "endpoint_urls_missing_scheme", + flags: []string{"--endpoint-urls", "s3-1:9000"}, + expectedError: fmt.Errorf(`ERROR bad value for --endpoint-urls s3-1:9000: scheme is missing. Must be of the form http:/// or https:///`), + expectedExitCode: 1, + }, + { + name: "endpoint_url_and_endpoint_urls_conflict", + flags: []string{"--endpoint-url", "http://s3-1:9000", "--endpoint-urls", "http://s3-2:9000"}, + expectedError: fmt.Errorf(`ERROR "endpoint-url" and "endpoint-urls" flags cannot be used together`), + expectedExitCode: 1, + }, + } + + for _, tc := range testcases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + _, s5cmd := setup(t) + + cmd := s5cmd(tc.flags...) + result := icmd.RunCmd(cmd) + + result.Assert(t, icmd.Expected{ExitCode: tc.expectedExitCode}) + + if tc.expectedError == nil && result.Stderr() == "" { + return + } + + assertLines(t, result.Stderr(), map[int]compareFunc{ + 0: equals("%v", tc.expectedError), + }) + }) + } +} diff --git a/storage/s3.go b/storage/s3.go index 3313be66e..77fc51abf 100644 --- a/storage/s3.go +++ b/storage/s3.go @@ -17,6 +17,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/aws/aws-sdk-go/aws" @@ -54,9 +55,12 @@ const ( // Re-used AWS sessions dramatically improve performance. var globalSessionCache = &SessionCache{ - sessions: map[Options]*session.Session{}, + sessions: map[sessionCacheKey]*session.Session{}, } +// Global counter for round-robin load balancing across all S3 instances +var globalPoolCounter uint64 + // S3 is a storage type which interacts with S3API, DownloaderAPI and // UploaderAPI. type S3 struct { @@ -68,6 +72,11 @@ type S3 struct { useListObjectsV1 bool noSuchUploadRetryCount int requestPayer string + + // Multi-endpoint support: pool of clients for load balancing + apiPool []s3iface.S3API + downloaderPool []s3manageriface.DownloaderAPI + uploaderPool []s3manageriface.UploaderAPI } func (s *S3) RequestPayer() *string { @@ -77,6 +86,21 @@ func (s *S3) RequestPayer() *string { return &s.requestPayer } +// getNextClient returns the next S3 client from the pool using round-robin +// If no pool exists (single endpoint), returns the default client +func (s *S3) getNextClient() (s3iface.S3API, s3manageriface.DownloaderAPI, s3manageriface.UploaderAPI) { + if len(s.apiPool) == 0 { + // Single endpoint mode + return s.api, s.downloader, s.uploader + } + + // Multi-endpoint mode: round-robin selection using global counter + idx := atomic.AddUint64(&globalPoolCounter, 1) - 1 + poolIdx := idx % uint64(len(s.apiPool)) + + return s.apiPool[poolIdx], s.downloaderPool[poolIdx], s.uploaderPool[poolIdx] +} + func parseEndpoint(endpoint string) (urlpkg.URL, error) { if endpoint == "" { return sentinelURL, nil @@ -97,21 +121,75 @@ func newS3Storage(ctx context.Context, opts Options) (*S3, error) { return nil, err } + // Create the primary session (for single endpoint or first endpoint in multi-endpoint) awsSession, err := globalSessionCache.newSession(ctx, opts) if err != nil { return nil, err } - return &S3{ - api: s3.New(awsSession), - downloader: s3manager.NewDownloader(awsSession), - uploader: s3manager.NewUploader(awsSession), + s3Storage := &S3{ endpointURL: endpointURL, dryRun: opts.DryRun, useListObjectsV1: opts.UseListObjectsV1, requestPayer: opts.RequestPayer, noSuchUploadRetryCount: opts.NoSuchUploadRetryCount, - }, nil + } + + // If multiple endpoints are configured, create a pool of clients + if len(opts.Endpoints) > 0 { + // Determine region once for all endpoints + region := opts.region + if region == "" { + // Use default region for custom endpoints + region = endpoints.UsEast1RegionID + } + + // Create a separate session for each endpoint + for _, endpoint := range opts.Endpoints { + // Create options for this specific endpoint + endpointOpts := Options{ + MaxRetries: opts.MaxRetries, + NoSuchUploadRetryCount: opts.NoSuchUploadRetryCount, + Endpoint: endpoint, // Use single endpoint + Endpoints: nil, // Clear multi-endpoint to avoid recursion + NoVerifySSL: opts.NoVerifySSL, + DryRun: opts.DryRun, + NoSignRequest: opts.NoSignRequest, + UseListObjectsV1: opts.UseListObjectsV1, + RequestPayer: opts.RequestPayer, + Profile: opts.Profile, + CredentialFile: opts.CredentialFile, + LogLevel: opts.LogLevel, + } + endpointOpts.bucket = opts.bucket + endpointOpts.region = region // Use determined region + + // Create session for this endpoint + sess, err := globalSessionCache.newSession(ctx, endpointOpts) + if err != nil { + return nil, fmt.Errorf("failed to create session for endpoint %s: %v", endpoint, err) + } + + // Create S3 clients directly from session + // The session already has the correct endpoint and S3ForcePathStyle settings + s3Client := s3.New(sess) + s3Storage.apiPool = append(s3Storage.apiPool, s3Client) + s3Storage.downloaderPool = append(s3Storage.downloaderPool, s3manager.NewDownloader(sess)) + s3Storage.uploaderPool = append(s3Storage.uploaderPool, s3manager.NewUploader(sess)) + } + + // Set primary clients to first in pool + s3Storage.api = s3Storage.apiPool[0] + s3Storage.downloader = s3Storage.downloaderPool[0] + s3Storage.uploader = s3Storage.uploaderPool[0] + } else { + // Single endpoint: use default session + s3Storage.api = s3.New(awsSession) + s3Storage.downloader = s3manager.NewDownloader(awsSession) + s3Storage.uploader = s3manager.NewUploader(awsSession) + } + + return s3Storage, nil } // Stat retrieves metadata from S3 object without returning the object itself. @@ -125,7 +203,9 @@ func (s *S3) Stat(ctx context.Context, url *url.URL) (*Object, error) { input.SetVersionId(url.VersionID) } - output, err := s.api.HeadObjectWithContext(ctx, input) + // Use round-robin client selection for load balancing + api, _, _ := s.getNextClient() + output, err := api.HeadObjectWithContext(ctx, input) if err != nil { if errHasCode(err, "NotFound") { return nil, &ErrGivenObjectNotFound{ObjectAbsPath: url.Absolute()} @@ -184,7 +264,8 @@ func (s *S3) listObjectVersions(ctx context.Context, url *url.URL) <-chan *Objec var now time.Time - err := s.api.ListObjectVersionsPagesWithContext(ctx, &listInput, + api, _, _ := s.getNextClient() + err := api.ListObjectVersionsPagesWithContext(ctx, &listInput, func(p *s3.ListObjectVersionsOutput, lastPage bool) bool { for _, c := range p.CommonPrefixes { prefix := aws.StringValue(c.Prefix) @@ -314,7 +395,9 @@ func (s *S3) listObjectsV2(ctx context.Context, url *url.URL) <-chan *Object { var now time.Time - err := s.api.ListObjectsV2PagesWithContext(ctx, &listInput, func(p *s3.ListObjectsV2Output, lastPage bool) bool { + // Use round-robin client selection for load balancing + api, _, _ := s.getNextClient() + err := api.ListObjectsV2PagesWithContext(ctx, &listInput, func(p *s3.ListObjectsV2Output, lastPage bool) bool { for _, c := range p.CommonPrefixes { prefix := aws.StringValue(c.Prefix) if !url.Match(prefix) { @@ -405,7 +488,8 @@ func (s *S3) listObjects(ctx context.Context, url *url.URL) <-chan *Object { var now time.Time - err := s.api.ListObjectsPagesWithContext(ctx, &listInput, func(p *s3.ListObjectsOutput, lastPage bool) bool { + api, _, _ := s.getNextClient() + err := api.ListObjectsPagesWithContext(ctx, &listInput, func(p *s3.ListObjectsOutput, lastPage bool) bool { for _, c := range p.CommonPrefixes { prefix := aws.StringValue(c.Prefix) if !url.Match(prefix) { @@ -564,7 +648,8 @@ func (s *S3) Copy(ctx context.Context, from, to *url.URL, metadata Metadata) err input.Metadata = m } - _, err := s.api.CopyObject(input) + api, _, _ := s.getNextClient() + _, err := api.CopyObject(input) return err } @@ -579,7 +664,8 @@ func (s *S3) Read(ctx context.Context, src *url.URL) (io.ReadCloser, error) { input.SetVersionId(src.VersionID) } - resp, err := s.api.GetObjectWithContext(ctx, input) + api, _, _ := s.getNextClient() + resp, err := api.GetObjectWithContext(ctx, input) if err != nil { return nil, err } @@ -593,7 +679,8 @@ func (s *S3) Presign(ctx context.Context, from *url.URL, expire time.Duration) ( RequestPayer: s.RequestPayer(), } - req, _ := s.api.GetObjectRequest(input) + api, _, _ := s.getNextClient() + req, _ := api.GetObjectRequest(input) return req.Presign(expire) } @@ -621,7 +708,9 @@ func (s *S3) Get( input.VersionId = aws.String(from.VersionID) } - return s.downloader.DownloadWithContext(ctx, to, input, func(u *s3manager.Downloader) { + // Use round-robin client selection for load balancing + _, downloader, _ := s.getNextClient() + return downloader.DownloadWithContext(ctx, to, input, func(u *s3manager.Downloader) { u.PartSize = partSize u.Concurrency = concurrency }) @@ -748,7 +837,8 @@ func (s *S3) Select(ctx context.Context, url *url.URL, query *SelectQuery, resul OutputSerialization: outputFormat, } - resp, err := s.api.SelectObjectContentWithContext(ctx, input) + api, _, _ := s.getNextClient() + resp, err := api.SelectObjectContentWithContext(ctx, input) if err != nil { return err } @@ -877,7 +967,10 @@ func (s *S3) Put( u.PartSize = partSize u.Concurrency = concurrency } - _, err := s.uploader.UploadWithContext(ctx, input, uploaderOptsFn) + + // Use round-robin client selection for load balancing + _, _, uploader := s.getNextClient() + _, err := uploader.UploadWithContext(ctx, input, uploaderOptsFn) if errHasCode(err, s3.ErrCodeNoSuchUpload) && s.noSuchUploadRetryCount > 0 { return s.retryOnNoSuchUpload(ctx, to, input, err, uploaderOptsFn) @@ -907,7 +1000,8 @@ func (s *S3) retryOnNoSuchUpload(ctx aws.Context, to *url.URL, input *s3manager. msg := log.DebugMessage{Err: fmt.Sprintf("Retrying to upload %v upon error: %q", to, err.Error())} log.Debug(msg) - _, err = s.uploader.UploadWithContext(ctx, input, uploaderOpts...) + _, _, uploader := s.getNextClient() + _, err = uploader.UploadWithContext(ctx, input, uploaderOpts...) } if errHasCode(err, s3.ErrCodeNoSuchUpload) && s.noSuchUploadRetryCount > 0 { @@ -1007,8 +1101,9 @@ func (s *S3) doDelete(ctx context.Context, chunk chunk, resultch chan *Object) { // GCS does not support multi delete. if IsGoogleEndpoint(s.endpointURL) { + api, _, _ := s.getNextClient() for _, k := range chunk.Keys { - _, err := s.api.DeleteObjectWithContext(ctx, &s3.DeleteObjectInput{ + _, err := api.DeleteObjectWithContext(ctx, &s3.DeleteObjectInput{ Bucket: aws.String(chunk.Bucket), Key: k.Key, RequestPayer: s.RequestPayer(), @@ -1025,7 +1120,8 @@ func (s *S3) doDelete(ctx context.Context, chunk chunk, resultch chan *Object) { } bucket := chunk.Bucket - o, err := s.api.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{ + api, _, _ := s.getNextClient() + o, err := api.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{ Bucket: aws.String(bucket), Delete: &s3.Delete{Objects: chunk.Keys}, RequestPayer: s.RequestPayer(), @@ -1092,7 +1188,8 @@ func (s *S3) MultiDelete(ctx context.Context, urlch <-chan *url.URL) <-chan *Obj // ListBuckets is a blocking list-operation which gets bucket list and returns // the buckets that match with given prefix. func (s *S3) ListBuckets(ctx context.Context, prefix string) ([]Bucket, error) { - o, err := s.api.ListBucketsWithContext(ctx, &s3.ListBucketsInput{}) + api, _, _ := s.getNextClient() + o, err := api.ListBucketsWithContext(ctx, &s3.ListBucketsInput{}) if err != nil { return nil, err } @@ -1116,7 +1213,8 @@ func (s *S3) MakeBucket(ctx context.Context, name string) error { return nil } - _, err := s.api.CreateBucketWithContext(ctx, &s3.CreateBucketInput{ + api, _, _ := s.getNextClient() + _, err := api.CreateBucketWithContext(ctx, &s3.CreateBucketInput{ Bucket: aws.String(name), }) return err @@ -1128,7 +1226,8 @@ func (s *S3) RemoveBucket(ctx context.Context, name string) error { return nil } - _, err := s.api.DeleteBucketWithContext(ctx, &s3.DeleteBucketInput{ + api, _, _ := s.getNextClient() + _, err := api.DeleteBucketWithContext(ctx, &s3.DeleteBucketInput{ Bucket: aws.String(name), }) return err @@ -1140,7 +1239,8 @@ func (s *S3) SetBucketVersioning(ctx context.Context, versioningStatus, bucket s return nil } - _, err := s.api.PutBucketVersioningWithContext(ctx, &s3.PutBucketVersioningInput{ + api, _, _ := s.getNextClient() + _, err := api.PutBucketVersioningWithContext(ctx, &s3.PutBucketVersioningInput{ Bucket: aws.String(bucket), VersioningConfiguration: &s3.VersioningConfiguration{ Status: aws.String(versioningStatus), @@ -1151,7 +1251,8 @@ func (s *S3) SetBucketVersioning(ctx context.Context, versioningStatus, bucket s // GetBucketVersioning returnsversioning property of the bucket func (s *S3) GetBucketVersioning(ctx context.Context, bucket string) (string, error) { - output, err := s.api.GetBucketVersioningWithContext(ctx, &s3.GetBucketVersioningInput{ + api, _, _ := s.getNextClient() + output, err := api.GetBucketVersioningWithContext(ctx, &s3.GetBucketVersioningInput{ Bucket: aws.String(bucket), }) if err != nil || output.Status == nil { @@ -1162,7 +1263,8 @@ func (s *S3) GetBucketVersioning(ctx context.Context, bucket string) (string, er } func (s *S3) HeadBucket(ctx context.Context, url *url.URL) error { - _, err := s.api.HeadBucketWithContext(ctx, &s3.HeadBucketInput{ + api, _, _ := s.getNextClient() + _, err := api.HeadBucketWithContext(ctx, &s3.HeadBucketInput{ Bucket: aws.String(url.Bucket), }) return err @@ -1179,7 +1281,8 @@ func (s *S3) HeadObject(ctx context.Context, url *url.URL) (*Object, *Metadata, input.SetVersionId(url.VersionID) } - output, err := s.api.HeadObjectWithContext(ctx, input) + api, _, _ := s.getNextClient() + output, err := api.HeadObjectWithContext(ctx, input) if err != nil { if errHasCode(err, "NotFound") { return nil, nil, &ErrGivenObjectNotFound{ObjectAbsPath: url.Absolute()} @@ -1220,11 +1323,53 @@ func (l sdkLogger) Log(args ...interface{}) { log.Trace(msg) } +// sessionCacheKey is a comparable key for caching sessions +type sessionCacheKey struct { + MaxRetries int + NoSuchUploadRetryCount int + Endpoint string + EndpointsKey string // Concatenated endpoints for comparison + NoVerifySSL bool + DryRun bool + NoSignRequest bool + UseListObjectsV1 bool + LogLevel log.LogLevel + RequestPayer string + Profile string + CredentialFile string + bucket string + region string +} + +// makeSessionCacheKey creates a comparable cache key from Options +func makeSessionCacheKey(opts Options) sessionCacheKey { + endpointsKey := "" + if len(opts.Endpoints) > 0 { + endpointsKey = strings.Join(opts.Endpoints, "|") + } + return sessionCacheKey{ + MaxRetries: opts.MaxRetries, + NoSuchUploadRetryCount: opts.NoSuchUploadRetryCount, + Endpoint: opts.Endpoint, + EndpointsKey: endpointsKey, + NoVerifySSL: opts.NoVerifySSL, + DryRun: opts.DryRun, + NoSignRequest: opts.NoSignRequest, + UseListObjectsV1: opts.UseListObjectsV1, + LogLevel: opts.LogLevel, + RequestPayer: opts.RequestPayer, + Profile: opts.Profile, + CredentialFile: opts.CredentialFile, + bucket: opts.bucket, + region: opts.region, + } +} + // SessionCache holds session.Session according to s3Opts and it synchronizes // access/modification. type SessionCache struct { sync.Mutex - sessions map[Options]*session.Session + sessions map[sessionCacheKey]*session.Session } // newSession initializes a new AWS session with region fallback and custom @@ -1233,7 +1378,8 @@ func (sc *SessionCache) newSession(ctx context.Context, opts Options) (*session. sc.Lock() defer sc.Unlock() - if sess, ok := sc.sessions[opts]; ok { + cacheKey := makeSessionCacheKey(opts) + if sess, ok := sc.sessions[cacheKey]; ok { return sess, nil } @@ -1248,9 +1394,34 @@ func (sc *SessionCache) newSession(ctx context.Context, opts Options) (*session. ) } - endpointURL, err := parseEndpoint(opts.Endpoint) - if err != nil { - return nil, err + // Check if multiple endpoints are provided for load balancing + var endpointURL urlpkg.URL + var httpClient *http.Client + + if len(opts.Endpoints) > 0 { + // Multiple endpoints: Parse first endpoint for session initialization + // We'll apply RoundRobinTransport later to avoid issues with region detection + firstEndpoint, err := parseEndpoint(opts.Endpoints[0]) + if err != nil { + return nil, err + } + endpointURL = firstEndpoint + + // Use standard client for session initialization + if opts.NoVerifySSL { + httpClient = insecureHTTPClient + } + } else { + // Single endpoint: use standard behavior + var err error + endpointURL, err = parseEndpoint(opts.Endpoint) + if err != nil { + return nil, err + } + + if opts.NoVerifySSL { + httpClient = insecureHTTPClient + } } // use virtual-host-style if the endpoint is known to support it, @@ -1265,12 +1436,13 @@ func (sc *SessionCache) newSession(ctx context.Context, opts Options) (*session. endpointURL = sentinelURL } - var httpClient *http.Client - if opts.NoVerifySSL { - httpClient = insecureHTTPClient - } + // For multiple endpoints, we still need to set the first endpoint + // for region detection and initial setup, but the RoundRobinTransport + // will handle actual request routing + endpointStr := endpointURL.String() + awsCfg = awsCfg. - WithEndpoint(endpointURL.String()). + WithEndpoint(endpointStr). WithS3ForcePathStyle(!isVirtualHostStyle). WithS3UseAccelerate(useAccelerate). WithHTTPClient(httpClient). @@ -1317,12 +1489,19 @@ func (sc *SessionCache) newSession(ctx context.Context, opts Options) (*session. if opts.region != "" { sess.Config.Region = aws.String(opts.region) } else { - if err := setSessionRegion(ctx, sess, opts.bucket); err != nil { - return nil, err + // For multiple endpoints, skip automatic region detection to avoid + // GetBucketRegion calls being distributed across endpoints + if len(opts.Endpoints) > 0 { + // Use default region for custom endpoints + sess.Config.Region = aws.String(endpoints.UsEast1RegionID) + } else { + if err := setSessionRegion(ctx, sess, opts.bucket); err != nil { + return nil, err + } } } - sc.sessions[opts] = sess + sc.sessions[cacheKey] = sess return sess, nil } @@ -1330,7 +1509,8 @@ func (sc *SessionCache) newSession(ctx context.Context, opts Options) (*session. func (sc *SessionCache) clear() { sc.Lock() defer sc.Unlock() - sc.sessions = map[Options]*session.Session{} + sc.sessions = map[sessionCacheKey]*session.Session{} + atomic.StoreUint64(&globalPoolCounter, 0) } func setSessionRegion(ctx context.Context, sess *session.Session, bucket string) error { diff --git a/storage/s3_test.go b/storage/s3_test.go index 11b52ebc3..c30ba3a55 100644 --- a/storage/s3_test.go +++ b/storage/s3_test.go @@ -25,7 +25,9 @@ import ( "github.com/aws/aws-sdk-go/aws/session" "github.com/aws/aws-sdk-go/awstesting/unit" "github.com/aws/aws-sdk-go/service/s3" + "github.com/aws/aws-sdk-go/service/s3/s3iface" "github.com/aws/aws-sdk-go/service/s3/s3manager" + "github.com/aws/aws-sdk-go/service/s3/s3manager/s3manageriface" "github.com/google/go-cmp/cmp" "gotest.tools/v3/assert" @@ -1341,3 +1343,100 @@ func (e tempError) Error() string { return e.err.Error() } func (e tempError) Temporary() bool { return e.temp } func (e *tempError) Unwrap() error { return e.err } + +func TestGetNextClientRoundRobin(t *testing.T) { + // Reset global counter to make test deterministic + atomic.StoreUint64(&globalPoolCounter, 0) + + api0 := s3.New(unit.Session) + api1 := s3.New(unit.Session) + api2 := s3.New(unit.Session) + + s3s := &S3{ + apiPool: []s3iface.S3API{api0, api1, api2}, + downloaderPool: []s3manageriface.DownloaderAPI{nil, nil, nil}, + uploaderPool: []s3manageriface.UploaderAPI{nil, nil, nil}, + } + + // Should cycle through pool indices 0, 1, 2, 0, 1, 2 ... + expected := []s3iface.S3API{api0, api1, api2, api0, api1, api2} + for i, want := range expected { + got, _, _ := s3s.getNextClient() + if got != want { + t.Fatalf("call %d: expected api at index %d, got different api", i, i%3) + } + } +} + +func TestGetNextClientSingleEndpointFallback(t *testing.T) { + defaultAPI := s3.New(unit.Session) + s3s := &S3{ + api: defaultAPI, + apiPool: nil, // no pool + } + + got, _, _ := s3s.getNextClient() + if got != defaultAPI { + t.Fatal("expected default api when no pool configured") + } +} + +func TestMakeSessionCacheKeyEndpoints(t *testing.T) { + base := Options{Endpoint: "http://s3.example.com"} + + multi := Options{Endpoints: []string{"http://s3-1:9000", "http://s3-2:9000"}} + + keyBase := makeSessionCacheKey(base) + keyMulti := makeSessionCacheKey(multi) + if keyBase == keyMulti { + t.Fatal("expected different cache keys for single vs multi-endpoint options") + } + + // Same endpoints in same order → same key + multiSame := Options{Endpoints: []string{"http://s3-1:9000", "http://s3-2:9000"}} + if makeSessionCacheKey(multi) != makeSessionCacheKey(multiSame) { + t.Fatal("expected same cache key for identical endpoint lists") + } + + // Different endpoint order → different key + multiReverse := Options{Endpoints: []string{"http://s3-2:9000", "http://s3-1:9000"}} + if makeSessionCacheKey(multi) == makeSessionCacheKey(multiReverse) { + t.Fatal("expected different cache keys for different endpoint orderings") + } +} + +func TestNewS3StorageMultiEndpointPoolSize(t *testing.T) { + // Start fake S3-compatible servers (just need to respond to region detection) + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + srv1 := httptest.NewServer(handler) + defer srv1.Close() + srv2 := httptest.NewServer(handler) + defer srv2.Close() + srv3 := httptest.NewServer(handler) + defer srv3.Close() + + globalSessionCache.clear() + + opts := Options{ + Endpoints: []string{srv1.URL, srv2.URL, srv3.URL}, + NoSignRequest: true, + } + + s3s, err := newS3Storage(context.Background(), opts) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if got := len(s3s.apiPool); got != 3 { + t.Fatalf("expected pool size 3, got %d", got) + } + if len(s3s.downloaderPool) != 3 || len(s3s.uploaderPool) != 3 { + t.Fatal("downloader/uploader pool size mismatch") + } + // Primary client should be set to first pool entry + if s3s.api != s3s.apiPool[0] { + t.Fatal("expected s3s.api to point to apiPool[0]") + } +} diff --git a/storage/storage.go b/storage/storage.go index 1a6171263..b5e62080f 100644 --- a/storage/storage.go +++ b/storage/storage.go @@ -61,6 +61,7 @@ func NewRemoteClient(ctx context.Context, url *url.URL, opts Options) (*S3, erro MaxRetries: opts.MaxRetries, NoSuchUploadRetryCount: opts.NoSuchUploadRetryCount, Endpoint: opts.Endpoint, + Endpoints: opts.Endpoints, // CRITICAL: Must copy Endpoints for multi-endpoint support NoVerifySSL: opts.NoVerifySSL, DryRun: opts.DryRun, NoSignRequest: opts.NoSignRequest, @@ -86,7 +87,8 @@ func NewClient(ctx context.Context, url *url.URL, opts Options) (Storage, error) type Options struct { MaxRetries int NoSuchUploadRetryCount int - Endpoint string + Endpoint string // Single endpoint (deprecated, use Endpoints for multiple) + Endpoints []string // Multiple endpoints for load balancing NoVerifySSL bool DryRun bool NoSignRequest bool From a731c42509004e23d470668949afff200e966469 Mon Sep 17 00:00:00 2001 From: jinyuchen724 <627388936@qq.com> Date: Mon, 29 Jun 2026 19:26:21 +0800 Subject: [PATCH 2/5] ci: update MinIO image to latest tag The bitnami/minio:2023.7.18 tag no longer exists on Docker Hub. Switch to official minio/minio:latest image and adjust health check command to work with the official image. --- .github/workflows/ci.yml | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index eceb7f72f..8586aa94f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -45,17 +45,18 @@ jobs: runs-on: ${{ matrix.os }}-latest services: minio: - image: ${{ (matrix.os == 'ubuntu') && 'bitnami/minio:2023.7.18' || ''}} + image: ${{ (matrix.os == 'ubuntu') && 'minio/minio:latest' || ''}} ports: - 45677:9000 options: >- - --health-cmd "curl -I http://localhost:9000/minio/health/live -s" + --health-cmd "curl -f http://localhost:9000/minio/health/live || exit 1" --health-interval 10s --health-timeout 5s --health-retries 5 env: MINIO_ROOT_USER: minioadmin MINIO_ROOT_PASSWORD: minioadmin + MINIO_VOLUMES: /data steps: - uses: actions/checkout@v2 - uses: actions/setup-go@v2 From 6df0538ee1a0424919dc469d4d8cfa68ac5a7f46 Mon Sep 17 00:00:00 2001 From: jinyuchen724 <627388936@qq.com> Date: Tue, 30 Jun 2026 17:37:33 +0800 Subject: [PATCH 3/5] ci: use bitnamilegacy/minio image --- .github/workflows/ci.yml | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8586aa94f..614797f58 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -45,18 +45,17 @@ jobs: runs-on: ${{ matrix.os }}-latest services: minio: - image: ${{ (matrix.os == 'ubuntu') && 'minio/minio:latest' || ''}} + image: ${{ (matrix.os == 'ubuntu') && 'bitnamilegacy/minio:2023.7.18' || ''}} ports: - 45677:9000 options: >- - --health-cmd "curl -f http://localhost:9000/minio/health/live || exit 1" + --health-cmd "curl -I http://localhost:9000/minio/health/live -s" --health-interval 10s --health-timeout 5s --health-retries 5 env: MINIO_ROOT_USER: minioadmin MINIO_ROOT_PASSWORD: minioadmin - MINIO_VOLUMES: /data steps: - uses: actions/checkout@v2 - uses: actions/setup-go@v2 From 2605cf70d5449ffa081cc4b05ac72ff01b0f1b7d Mon Sep 17 00:00:00 2001 From: jinyuchen724 <627388936@qq.com> Date: Tue, 30 Jun 2026 18:20:59 +0800 Subject: [PATCH 4/5] test: fix endpoint-urls e2e test colliding with injected endpoint-url --- e2e/app_test.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/e2e/app_test.go b/e2e/app_test.go index 56c5319b0..ca6c0c78f 100644 --- a/e2e/app_test.go +++ b/e2e/app_test.go @@ -351,9 +351,10 @@ func TestAppEndpointURLsValidation(t *testing.T) { t.Run(tc.name, func(t *testing.T) { t.Parallel() - _, s5cmd := setup(t) - - cmd := s5cmd(tc.flags...) + // Invoke the binary directly instead of the setup() helper, + // which injects "--endpoint-url" and would collide with the + // "--endpoint-urls" flag under test. + cmd := icmd.Command(s5cmdPath, tc.flags...) result := icmd.RunCmd(cmd) result.Assert(t, icmd.Expected{ExitCode: tc.expectedExitCode}) From 824fb7fccdcb35267f4a2e1fb0b520403392e4b8 Mon Sep 17 00:00:00 2001 From: jinyuchen724 <627388936@qq.com> Date: Tue, 30 Jun 2026 18:52:56 +0800 Subject: [PATCH 5/5] ci: pin macOS runner to macos-15 (macos-latest moved to macos-26) --- .github/workflows/ci.yml | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 614797f58..5888bbfe4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -20,7 +20,10 @@ jobs: - windows name: build (${{ matrix.os }}/go-${{ matrix.go-version }}) - runs-on: ${{ matrix.os }}-latest + # Pin macOS to macos-15: macos-latest moved to macos-26 in June 2026, + # whose linker/dyld rejects the older Go toolchains' test binaries + # (missing LC_UUID load command). macos-15 is the last image these pass on. + runs-on: ${{ matrix.os == 'macos' && 'macos-15' || format('{0}-latest', matrix.os) }} steps: - uses: actions/checkout@v2 - uses: actions/setup-go@v2 @@ -42,7 +45,8 @@ jobs: - windows name: test (${{ matrix.os }}/go-${{ matrix.go-version }}) - runs-on: ${{ matrix.os }}-latest + # See build job: pin macOS to macos-15 (macos-latest is macos-26). + runs-on: ${{ matrix.os == 'macos' && 'macos-15' || format('{0}-latest', matrix.os) }} services: minio: image: ${{ (matrix.os == 'ubuntu') && 'bitnamilegacy/minio:2023.7.18' || ''}}