From 2d44e008bd73b86ecd424852f6c74ceb067a51c3 Mon Sep 17 00:00:00 2001 From: Chris J Arges Date: Thu, 6 Aug 2026 17:49:44 -0500 Subject: [PATCH] s3: upload package files concurrently Upload package files in bounded concurrent batches while keeping metadata generation and conflicting paths ordered. Default to four uploads and allow per-endpoint configuration. https://github.com/aptly-dev/aptly/issues/1615 Signed-off-by: Chris J Arges --- context/context.go | 4 +- deb/publish.go | 133 +++++++++++++++++++----- deb/publish_test.go | 65 ++++++++++++ debian/aptly.conf | 5 +- s3/public.go | 72 ++++++++----- s3/public_test.go | 13 +++ system/t02_config/CreateConfigTest_gold | 4 + utils/config.go | 1 + utils/config_test.go | 2 + 9 files changed, 244 insertions(+), 55 deletions(-) diff --git a/context/context.go b/context/context.go index a1a522380..910b3dc76 100644 --- a/context/context.go +++ b/context/context.go @@ -436,11 +436,11 @@ func (context *AptlyContext) GetPublishedStorage(name string) (aptly.PublishedSt } var err error - publishedStorage, err = s3.NewPublishedStorage( + publishedStorage, err = s3.NewPublishedStorageWithUploadConcurrency( params.AccessKeyID, params.SecretAccessKey, params.SessionToken, params.Region, params.Endpoint, params.Bucket, params.ACL, params.Prefix, params.StorageClass, params.EncryptionMethod, params.PlusWorkaround, params.DisableMultiDel, - params.ForceSigV2, params.ForceVirtualHostedStyle, params.Debug) + params.ForceSigV2, params.ForceVirtualHostedStyle, params.UploadConcurrency, params.Debug) if err != nil { return nil, err } diff --git a/deb/publish.go b/deb/publish.go index 8ae71df94..cf76dc4dc 100644 --- a/deb/publish.go +++ b/deb/publish.go @@ -829,6 +829,32 @@ func (p *PublishedRepo) GetSkelFiles(skelDir string, component string) (map[stri return files, nil } +func (p *PublishedRepo) packagePublishPath(pkg *Package, component string) (string, bool, error) { + for _, arch := range p.Architectures { + if !pkg.MatchesArchitecture(arch) { + continue + } + + if pkg.IsInstaller { + if p.Distribution == aptly.DistributionFocal { + return filepath.Join("dists", p.Distribution, component, fmt.Sprintf("%s-%s", pkg.Name, arch), "current", "legacy-images"), true, nil + } + return filepath.Join("dists", p.Distribution, component, fmt.Sprintf("%s-%s", pkg.Name, arch), "current", "images"), true, nil + } + + poolDir, err := pkg.PoolDirectory() + if err != nil { + return "", false, err + } + if p.MultiDist { + return filepath.Join("pool", p.Distribution, component, poolDir), true, nil + } + return filepath.Join("pool", component, poolDir), true, nil + } + + return "", false, nil +} + // Publish publishes snapshot (repository) contents, links package files, generates Packages & Release files, signs them func (p *PublishedRepo) Publish(packagePool aptly.PackagePool, publishedStorageProvider aptly.PublishedStorageProvider, collectionFactory *CollectionFactory, signer pgp.Signer, progress aptly.Progress, forceOverwrite bool, skelDir string) error { @@ -919,6 +945,13 @@ func (p *PublishedRepo) Publish(packagePool aptly.PackagePool, publishedStorageP progress.InitBar(count, false, aptly.BarPublishGeneratePackageFiles) } + uploadConcurrency := 1 + if uploader, ok := publishedStorage.(interface{ UploadConcurrency() int }); ok { + if concurrency := uploader.UploadConcurrency(); concurrency > 1 { + uploadConcurrency = min(concurrency, 128) + } + } + for component, list := range lists { hadUdebs := false @@ -928,43 +961,24 @@ func (p *PublishedRepo) Publish(packagePool aptly.PackagePool, publishedStorageP } list.PrepareIndex() - contentIndexes := map[string]*ContentsIndex{} - err = list.ForEachIndexed(func(pkg *Package) error { + processPackage := func(pkg *Package) error { if progress != nil { progress.AddBar(1) } - for _, arch := range p.Architectures { - if pkg.MatchesArchitecture(arch) { - hadUdebs = hadUdebs || pkg.IsUdeb - - var relPath string - if !pkg.IsInstaller { - poolDir, err2 := pkg.PoolDirectory() - if err2 != nil { - return err2 - } - if p.MultiDist { - relPath = filepath.Join("pool", p.Distribution, component, poolDir) - } else { - relPath = filepath.Join("pool", component, poolDir) - } - - } else { - if p.Distribution == aptly.DistributionFocal { - relPath = filepath.Join("dists", p.Distribution, component, fmt.Sprintf("%s-%s", pkg.Name, arch), "current", "legacy-images") - } else { - relPath = filepath.Join("dists", p.Distribution, component, fmt.Sprintf("%s-%s", pkg.Name, arch), "current", "images") - } - } - + relPath, matches, err2 := p.packagePublishPath(pkg, component) + if err2 != nil { + return err2 + } + if matches { + hadUdebs = hadUdebs || pkg.IsUdeb + if uploadConcurrency == 1 { err = pkg.LinkFromPool(publishedStorage, packagePool, p.Prefix, relPath, forceOverwrite) if err != nil { return err } - break } } @@ -1017,7 +1031,70 @@ func (p *PublishedRepo) Publish(packagePool aptly.PackagePool, publishedStorageP pkg.contents = nil return batch.Write() - }) + } + + if uploadConcurrency == 1 { + err = list.ForEachIndexed(processPackage) + } else { + for offset := 0; offset < len(list.packagesIndex); offset += uploadConcurrency { + batch := list.packagesIndex[offset:min(offset+uploadConcurrency, len(list.packagesIndex))] + groups := map[string][]*Package{} + var paths []string + for _, pkg := range batch { + relPath, matches, err2 := p.packagePublishPath(pkg, component) + if err2 != nil { + return fmt.Errorf("unable to prepare package uploads: %s", err2) + } + if !matches { + continue + } + + files := pkg.Files() + for i := range files { + if _, err2 = files[i].GetPoolPath(packagePool); err2 != nil { + return fmt.Errorf("unable to prepare package uploads: %s", err2) + } + } + if pkg.IsSource { + pkg.Extra() + } + if _, exists := groups[relPath]; !exists { + paths = append(paths, relPath) + } + groups[relPath] = append(groups[relPath], pkg) + } + + done := make(chan struct{}, len(paths)) + uploadErrors := make([]error, len(paths)) + for i, relPath := range paths { + go func() { + defer func() { done <- struct{}{} }() + for _, pkg := range groups[relPath] { + if err := pkg.LinkFromPool(publishedStorage, packagePool, p.Prefix, relPath, forceOverwrite); err != nil { + uploadErrors[i] = err + return + } + } + }() + } + for range paths { + <-done + } + for _, uploadErr := range uploadErrors { + if uploadErr != nil { + return fmt.Errorf("unable to upload packages: %s", uploadErr) + } + } + for _, pkg := range batch { + if err = processPackage(pkg); err != nil { + break + } + } + if err != nil { + break + } + } + } if err != nil { return fmt.Errorf("unable to process packages: %s", err) diff --git a/deb/publish_test.go b/deb/publish_test.go index 20f78c4f9..fba08e6da 100644 --- a/deb/publish_test.go +++ b/deb/publish_test.go @@ -8,6 +8,8 @@ import ( "os" "path/filepath" "sort" + "sync" + "time" "github.com/aptly-dev/aptly/aptly" "github.com/aptly-dev/aptly/database" @@ -62,6 +64,49 @@ type FakeStorageProvider struct { storages map[string]aptly.PublishedStorage } +type concurrentTestStorage struct { + aptly.PublishedStorage + concurrency int + mu sync.Mutex + active int + maxActive int + release chan struct{} + releaseTriggered bool +} + +func (s *concurrentTestStorage) UploadConcurrency() int { + return s.concurrency +} + +func (s *concurrentTestStorage) LinkFromPool(publishedPrefix, publishedRelPath, fileName string, + sourcePool aptly.PackagePool, sourcePath string, sourceChecksums utils.ChecksumInfo, force bool) error { + if s.release == nil { + return s.PublishedStorage.LinkFromPool(publishedPrefix, publishedRelPath, fileName, sourcePool, sourcePath, sourceChecksums, force) + } + + s.mu.Lock() + s.active++ + s.maxActive = max(s.maxActive, s.active) + if s.active == s.concurrency && !s.releaseTriggered { + close(s.release) + s.releaseTriggered = true + } + s.mu.Unlock() + defer func() { + s.mu.Lock() + s.active-- + s.mu.Unlock() + }() + + select { + case <-s.release: + case <-time.After(time.Second): + return errors.New("package uploads did not overlap") + } + + return s.PublishedStorage.LinkFromPool(publishedPrefix, publishedRelPath, fileName, sourcePool, sourcePath, sourceChecksums, force) +} + func (p *FakeStorageProvider) GetPublishedStorage(name string) (aptly.PublishedStorage, error) { storage, ok := p.storages[name] if !ok { @@ -195,6 +240,8 @@ func (s *PublishedRepoSuite) TestNewPublishedRepo(c *C) { } func (s *PublishedRepoSuite) TestMultiDistPool(c *C) { + s.provider.storages[""] = &concurrentTestStorage{PublishedStorage: s.publishedStorage, concurrency: 3} + repo, err := NewPublishedRepo("", "ppa", "squeeze", nil, []string{"main"}, []interface{}{s.snapshot}, s.factory, true) c.Assert(err, IsNil) err = repo.Publish(s.packagePool, s.provider, s.factory, &NullSigner{}, nil, false, "") @@ -246,6 +293,24 @@ func (s *PublishedRepoSuite) TestMultiDistPool(c *C) { } +func (s *PublishedRepoSuite) TestPackageUploadsConcurrent(c *C) { + for _, pkg := range []*Package{s.p1, s.p2, s.p3} { + pkg.Source = "" + c.Assert(s.packageCollection.Update(pkg), IsNil) + } + + storage := &concurrentTestStorage{ + PublishedStorage: s.publishedStorage, + concurrency: 2, + release: make(chan struct{}), + } + s.provider.storages[""] = storage + + err := s.repo.Publish(s.packagePool, s.provider, s.factory, &NullSigner{}, nil, false, "") + c.Assert(err, IsNil) + c.Check(storage.maxActive, Equals, 2) +} + func (s *PublishedRepoSuite) TestPrefixNormalization(c *C) { for _, t := range []struct { diff --git a/debian/aptly.conf b/debian/aptly.conf index 043b46ebe..9875d5a58 100644 --- a/debian/aptly.conf +++ b/debian/aptly.conf @@ -281,6 +281,10 @@ s3_publish_endpoints: # # Disable path style visit, useful with non-AWS S3-compatible object stores # # which only support virtual hosted style # force_virtualhosted_style: false + # # Upload Concurrency (optional) + # # Maximum number of package files uploaded concurrently (up to 128), defaults to 4. + # # Set to 1 to upload package files sequentially. + # upload_concurrency: 4 # # Debug (optional) # # Enables detailed request/response dump for each S3 operation # debug: false @@ -424,4 +428,3 @@ packagepool_storage: # # See: Azure documentation https://docs.microsoft.com/en-us/azure/storage/common/storage-configure-connection-string # # defaults to "https://.blob.core.windows.net" # endpoint: "" - diff --git a/s3/public.go b/s3/public.go index c35de4250..56e0d8ebd 100644 --- a/s3/public.go +++ b/s3/public.go @@ -42,22 +42,25 @@ func (l *logger) Logf(classification logging.Classification, format string, v .. // PublishedStorage abstract file system with published files (actually hosted on S3) type PublishedStorage struct { - s3 *s3.Client - config *aws.Config - bucket string - acl types.ObjectCannedACL - prefix string - storageClass types.StorageClass - encryptionMethod types.ServerSideEncryption - plusWorkaround bool - disableMultiDel bool - pathCache map[string]string - pathCacheMutex sync.RWMutex + s3 *s3.Client + config *aws.Config + bucket string + acl types.ObjectCannedACL + prefix string + storageClass types.StorageClass + encryptionMethod types.ServerSideEncryption + plusWorkaround bool + disableMultiDel bool + uploadConcurrency int + pathCache map[string]string + pathCacheMutex sync.RWMutex // True if the bucket encrypts objects by default. encryptByDefault bool } +const defaultUploadConcurrency = 4 + // Check interface var ( _ aptly.PublishedStorage = (*PublishedStorage)(nil) @@ -93,14 +96,15 @@ func NewPublishedStorageRaw( o.HTTPSignerV4 = signer.NewSigner() o.BaseEndpoint = baseEndpoint }), - bucket: bucket, - config: config, - acl: acl, - prefix: prefix, - storageClass: types.StorageClass(storageClass), - encryptionMethod: types.ServerSideEncryption(encryptionMethod), - plusWorkaround: plusWorkaround, - disableMultiDel: disabledMultiDel, + bucket: bucket, + config: config, + acl: acl, + prefix: prefix, + storageClass: types.StorageClass(storageClass), + encryptionMethod: types.ServerSideEncryption(encryptionMethod), + plusWorkaround: plusWorkaround, + disableMultiDel: disabledMultiDel, + uploadConcurrency: defaultUploadConcurrency, } result.setKMSFlag() @@ -127,8 +131,8 @@ func (storage *PublishedStorage) setKMSFlag() { // keys, region and bucket name func NewPublishedStorage( accessKey, secretKey, sessionToken, region, endpoint, bucket, defaultACL, prefix, storageClass, encryptionMethod string, - plusWorkaround, disableMultiDel, _, forceVirtualHostedStyle, debug bool) (*PublishedStorage, error) { - + plusWorkaround, disableMultiDel, _, forceVirtualHostedStyle, debug bool, +) (*PublishedStorage, error) { opts := []func(*config.LoadOptions) error{config.WithRegion(region)} if accessKey != "" { opts = append(opts, config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(accessKey, secretKey, sessionToken))) @@ -149,6 +153,27 @@ func NewPublishedStorage( return result, err } +// NewPublishedStorageWithUploadConcurrency creates published storage with bounded concurrent package uploads. +func NewPublishedStorageWithUploadConcurrency( + accessKey, secretKey, sessionToken, region, endpoint, bucket, defaultACL, prefix, storageClass, encryptionMethod string, + plusWorkaround, disableMultiDel, forceSigV2, forceVirtualHostedStyle bool, uploadConcurrency int, debug bool, +) (*PublishedStorage, error) { + if uploadConcurrency < 0 || uploadConcurrency > 128 { + return nil, fmt.Errorf("upload concurrency must be between 0 and 128") + } + storage, err := NewPublishedStorage(accessKey, secretKey, sessionToken, region, endpoint, bucket, defaultACL, prefix, + storageClass, encryptionMethod, plusWorkaround, disableMultiDel, forceSigV2, forceVirtualHostedStyle, debug) + if err == nil && uploadConcurrency > 0 { + storage.uploadConcurrency = uploadConcurrency + } + return storage, err +} + +// UploadConcurrency returns the maximum number of concurrent package uploads. +func (storage *PublishedStorage) UploadConcurrency() int { + return storage.uploadConcurrency +} + // String returns the storage as string func (storage *PublishedStorage) String() string { return fmt.Sprintf("S3: %s:%s/%s", storage.config.Region, storage.bucket, storage.prefix) @@ -339,8 +364,8 @@ func (storage *PublishedStorage) RemoveDirs(path string, _ aptly.Progress) error // // LinkFromPool returns relative path for the published file to be included in package index func (storage *PublishedStorage) LinkFromPool(publishedPrefix, publishedRelPath, fileName string, sourcePool aptly.PackagePool, - sourcePath string, sourceChecksums utils.ChecksumInfo, force bool) error { - + sourcePath string, sourceChecksums utils.ChecksumInfo, force bool, +) error { publishedDirectory := filepath.Join(publishedPrefix, publishedRelPath) relPath := filepath.Join(publishedDirectory, fileName) poolPath := filepath.Join(storage.prefix, relPath) @@ -500,7 +525,6 @@ func (storage *PublishedStorage) RenameFile(oldName, newName string) error { // SymLink creates a copy of src file and adds link information as meta data func (storage *PublishedStorage) SymLink(src string, dst string) error { - params := &s3.CopyObjectInput{ Bucket: aws.String(storage.bucket), CopySource: aws.String(filepath.Join(storage.bucket, storage.prefix, src)), diff --git a/s3/public_test.go b/s3/public_test.go index d857e64dd..1ad5d3b8b 100644 --- a/s3/public_test.go +++ b/s3/public_test.go @@ -112,6 +112,19 @@ func (s *PublishedStorageSuite) TestPutFile(c *C) { c.Check(s.GetFile(c, "lala/a/b.txt"), DeepEquals, []byte("welcome to s3!")) } +func (s *PublishedStorageSuite) TestUploadConcurrency(c *C) { + c.Check(s.storage.UploadConcurrency(), Equals, defaultUploadConcurrency) + + storage, err := NewPublishedStorageWithUploadConcurrency("aa", "bb", "", "test-1", s.srv.URL(), "test", "", "", "", "", false, true, false, false, 7, false) + c.Assert(err, IsNil) + c.Check(storage.UploadConcurrency(), Equals, 7) + + _, err = NewPublishedStorageWithUploadConcurrency("aa", "bb", "", "test-1", s.srv.URL(), "test", "", "", "", "", false, true, false, false, -1, false) + c.Check(err, ErrorMatches, "upload concurrency must be between 0 and 128") + _, err = NewPublishedStorageWithUploadConcurrency("aa", "bb", "", "test-1", s.srv.URL(), "test", "", "", "", "", false, true, false, false, 129, false) + c.Check(err, ErrorMatches, "upload concurrency must be between 0 and 128") +} + func (s *PublishedStorageSuite) TestPutFilePlusWorkaround(c *C) { s.storage.plusWorkaround = true diff --git a/system/t02_config/CreateConfigTest_gold b/system/t02_config/CreateConfigTest_gold index d8e39c9dc..54bd77691 100644 --- a/system/t02_config/CreateConfigTest_gold +++ b/system/t02_config/CreateConfigTest_gold @@ -281,6 +281,10 @@ s3_publish_endpoints: # # Disable path style visit, useful with non-AWS S3-compatible object stores # # which only support virtual hosted style # force_virtualhosted_style: false + # # Upload Concurrency (optional) + # # Maximum number of package files uploaded concurrently (up to 128), defaults to 4. + # # Set to 1 to upload package files sequentially. + # upload_concurrency: 4 # # Debug (optional) # # Enables detailed request/response dump for each S3 operation # debug: false diff --git a/utils/config.go b/utils/config.go index 1c148e4f1..384505102 100644 --- a/utils/config.go +++ b/utils/config.go @@ -201,6 +201,7 @@ type S3PublishRoot struct { DisableMultiDel bool `json:"disableMultiDel" yaml:"disable_multidel"` ForceSigV2 bool `json:"forceSigV2" yaml:"force_sigv2"` ForceVirtualHostedStyle bool `json:"forceVirtualHostedStyle" yaml:"force_virtualhosted_style"` + UploadConcurrency int `json:"uploadConcurrency,omitempty" yaml:"upload_concurrency,omitempty"` Debug bool `json:"debug" yaml:"debug"` } diff --git a/utils/config_test.go b/utils/config_test.go index abc59e492..e975bd91a 100644 --- a/utils/config_test.go +++ b/utils/config_test.go @@ -211,6 +211,7 @@ func (s *ConfigSuite) TestLoadYAMLConfig(c *C) { c.Check(s.config.GetRootDir(), Equals, "/opt/aptly/") c.Check(s.config.DownloadConcurrency, Equals, 40) c.Check(s.config.DatabaseOpenAttempts, Equals, 10) + c.Check(s.config.S3PublishRoots["test"].UploadConcurrency, Equals, 8) } func (s *ConfigSuite) TestLoadYAMLErrorConfig(c *C) { @@ -388,6 +389,7 @@ s3_publish_endpoints: disable_multidel: true force_sigv2: true force_virtualhosted_style: true + upload_concurrency: 8 debug: true gcs_publish_endpoints: test: