Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[RFC] Split reflists to share their contents across snapshots #1282

Open
wants to merge 7 commits into
base: master
Choose a base branch
from
Open

[RFC] Split reflists to share their contents across snapshots #1282

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
4be09fd
Use github.com/saracen/walker for file walk operations
refi64 Sep 5, 2023
19a705f
Split reflists to share their contents across snapshots
refi64 Nov 10, 2023
1a3346a
fix golangci-lint error
neolynx Apr 21, 2024
174bdc2
fix golangci-lint errors
neolynx Apr 21, 2024
2a07494
fix unit tests
neolynx Jun 17, 2024
483dcf1
make compatible with go 1.19
neolynx Jun 17, 2024
3e99e37
handle packageRefs == nil
neolynx Jun 20, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion api/api.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -229,7 +229,7 @@ func maybeRunTaskInBackground(c *gin.Context, name string, resources []string, p

// Common piece of code to show list of packages,
// with searching & details if requested
func showPackages(c *gin.Context, reflist *deb.PackageRefList, collectionFactory *deb.CollectionFactory) {
func showPackages(c *gin.Context, reflist deb.AnyRefList, collectionFactory *deb.CollectionFactory) {
result := []*deb.Package{}

list, err := deb.NewPackageListFromRefList(reflist, collectionFactory.PackageCollection(), nil)
Expand Down
81 changes: 61 additions & 20 deletions api/db.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import (
"sort"

"github.com/aptly-dev/aptly/aptly"
"github.com/aptly-dev/aptly/database"
"github.com/aptly-dev/aptly/deb"
"github.com/aptly-dev/aptly/task"
"github.com/aptly-dev/aptly/utils"
Expand All @@ -28,18 +29,22 @@ func apiDbCleanup(c *gin.Context) {

collectionFactory := context.NewCollectionFactory()

// collect information about referenced packages...
existingPackageRefs := deb.NewPackageRefList()
// collect information about referenced packages and their reflist buckets...
existingPackageRefs := deb.NewSplitRefList()
existingBuckets := deb.NewRefListDigestSet()

reflistMigration := collectionFactory.RefListCollection().NewMigration()

out.Printf("Loading mirrors, local repos, snapshots and published repos...")
err = collectionFactory.RemoteRepoCollection().ForEach(func(repo *deb.RemoteRepo) error {
e := collectionFactory.RemoteRepoCollection().LoadComplete(repo)
if e != nil {
sl := deb.NewSplitRefList()
e := collectionFactory.RefListCollection().LoadCompleteAndMigrate(sl, repo.RefKey(), reflistMigration)
if e != nil && e != database.ErrNotFound {
return e
}
if repo.RefList() != nil {
existingPackageRefs = existingPackageRefs.Merge(repo.RefList(), false, true)
}

existingPackageRefs = existingPackageRefs.Merge(sl, false, true)
existingBuckets.AddAllInRefList(sl)

return nil
})
Expand All @@ -48,14 +53,14 @@ func apiDbCleanup(c *gin.Context) {
}

err = collectionFactory.LocalRepoCollection().ForEach(func(repo *deb.LocalRepo) error {
e := collectionFactory.LocalRepoCollection().LoadComplete(repo)
if e != nil {
sl := deb.NewSplitRefList()
e := collectionFactory.RefListCollection().LoadCompleteAndMigrate(sl, repo.RefKey(), reflistMigration)
if e != nil && e != database.ErrNotFound {
return e
}

if repo.RefList() != nil {
existingPackageRefs = existingPackageRefs.Merge(repo.RefList(), false, true)
}
existingPackageRefs = existingPackageRefs.Merge(sl, false, true)
existingBuckets.AddAllInRefList(sl)

return nil
})
Expand All @@ -64,12 +69,14 @@ func apiDbCleanup(c *gin.Context) {
}

err = collectionFactory.SnapshotCollection().ForEach(func(snapshot *deb.Snapshot) error {
e := collectionFactory.SnapshotCollection().LoadComplete(snapshot)
sl := deb.NewSplitRefList()
e := collectionFactory.RefListCollection().LoadCompleteAndMigrate(sl, snapshot.RefKey(), reflistMigration)
if e != nil {
return e
}

existingPackageRefs = existingPackageRefs.Merge(snapshot.RefList(), false, true)
existingPackageRefs = existingPackageRefs.Merge(sl, false, true)
existingBuckets.AddAllInRefList(sl)

return nil
})
Expand All @@ -81,25 +88,37 @@ func apiDbCleanup(c *gin.Context) {
if published.SourceKind != deb.SourceLocalRepo {
return nil
}
e := collectionFactory.PublishedRepoCollection().LoadComplete(published, collectionFactory)
if e != nil {
return e
}

for _, component := range published.Components() {
existingPackageRefs = existingPackageRefs.Merge(published.RefList(component), false, true)
sl := deb.NewSplitRefList()
e := collectionFactory.RefListCollection().LoadCompleteAndMigrate(sl, published.RefKey(component), reflistMigration)
if e != nil {
return e
}

existingPackageRefs = existingPackageRefs.Merge(sl, false, true)
existingBuckets.AddAllInRefList(sl)
}
return nil
})
if err != nil {
return nil, err
}

err = reflistMigration.Flush()
if err != nil {
return nil, err
}
if stats := reflistMigration.Stats(); stats.Reflists > 0 {
out.Printf("Split %d reflist(s) into %d bucket(s) (%d segment(s))",
stats.Reflists, stats.Buckets, stats.Segments)
}

// ... and compare it to the list of all packages
out.Printf("Loading list of all packages...")
allPackageRefs := collectionFactory.PackageCollection().AllPackageRefs()

toDelete := allPackageRefs.Subtract(existingPackageRefs)
toDelete := allPackageRefs.Subtract(existingPackageRefs.Flatten())

// delete packages that are no longer referenced
out.Printf("Deleting unreferenced packages (%d)...", toDelete.Len())
Expand All @@ -120,6 +139,28 @@ func apiDbCleanup(c *gin.Context) {
}
}

bucketsToDelete, err := collectionFactory.RefListCollection().AllBucketDigests()
if err != nil {
return nil, err
}

bucketsToDelete.RemoveAll(existingBuckets)

out.Printf("Deleting unreferenced reflist buckets (%d)...", bucketsToDelete.Len())
if bucketsToDelete.Len() > 0 {
batch := db.CreateBatch()
err := bucketsToDelete.ForEach(func(digest []byte) error {
return collectionFactory.RefListCollection().UnsafeDropBucket(digest, batch)
})
if err != nil {
return nil, err
}

if err := batch.Write(); err != nil {
return nil, err
}
}

// now, build a list of files that should be present in Repository (package pool)
out.Printf("Building list of files referenced by packages...")
referencedFiles := make([]string, 0, existingPackageRefs.Len())
Expand Down
6 changes: 1 addition & 5 deletions api/files.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,11 +170,7 @@ func apiFilesListFiles(c *gin.Context) {
listLock := &sync.Mutex{}
root := filepath.Join(context.UploadPath(), utils.SanitizePath(c.Params.ByName("dir")))

err := filepath.Walk(root, func(path string, _ os.FileInfo, err error) error {
if err != nil {
return err
}

err := walker.Walk(root, func(path string, _ os.FileInfo) error {
if path == root {
return nil
}
Expand Down
2 changes: 1 addition & 1 deletion api/metrics.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@ func countPackagesByRepos() {

components := repo.Components()
for _, c := range components {
count := float64(len(repo.RefList(c).Refs))
count := float64(repo.RefList(c).Len())
apiReposPackageCountGauge.WithLabelValues(fmt.Sprintf("%s", (repo.SourceNames())), repo.Distribution, c).Set(count)
}

Expand Down
12 changes: 6 additions & 6 deletions api/mirror.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,7 @@ func apiMirrorsCreate(c *gin.Context) {
return
}

err = collection.Add(repo)
err = collection.Add(repo, collectionFactory.RefListCollection())
if err != nil {
AbortWithJSONError(c, 500, fmt.Errorf("unable to add mirror: %s", err))
return
Expand Down Expand Up @@ -229,7 +229,7 @@ func apiMirrorsShow(c *gin.Context) {
return
}

err = collection.LoadComplete(repo)
err = collection.LoadComplete(repo, collectionFactory.RefListCollection())
if err != nil {
AbortWithJSONError(c, 500, fmt.Errorf("unable to show: %s", err))
}
Expand Down Expand Up @@ -260,7 +260,7 @@ func apiMirrorsPackages(c *gin.Context) {
return
}

err = collection.LoadComplete(repo)
err = collection.LoadComplete(repo, collectionFactory.RefListCollection())
if err != nil {
AbortWithJSONError(c, 500, fmt.Errorf("unable to show: %s", err))
}
Expand Down Expand Up @@ -491,12 +491,12 @@ func apiMirrorsUpdate(c *gin.Context) {
e := context.ReOpenDatabase()
if e == nil {
remote.MarkAsIdle()
collection.Update(remote)
collection.Update(remote, collectionFactory.RefListCollection())
}
}()

remote.MarkAsUpdating()
err = collection.Update(remote)
err = collection.Update(remote, collectionFactory.RefListCollection())
if err != nil {
return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
}
Expand Down Expand Up @@ -654,7 +654,7 @@ func apiMirrorsUpdate(c *gin.Context) {

log.Info().Msgf("%s: Finalizing download...", b.Name)
remote.FinalizeDownload(collectionFactory, out)
err = collectionFactory.RemoteRepoCollection().Update(remote)
err = collectionFactory.RemoteRepoCollection().Update(remote, collectionFactory.RefListCollection())
if err != nil {
return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
}
Expand Down
10 changes: 5 additions & 5 deletions api/publish.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -297,10 +297,10 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
switch s := source.(type) {
case *deb.Snapshot:
snapshotCollection := collectionFactory.SnapshotCollection()
err = snapshotCollection.LoadComplete(s)
err = snapshotCollection.LoadComplete(s, collectionFactory.RefListCollection())
case *deb.LocalRepo:
localCollection := collectionFactory.LocalRepoCollection()
err = localCollection.LoadComplete(s)
err = localCollection.LoadComplete(s, collectionFactory.RefListCollection())
default:
err = fmt.Errorf("unexpected type for source: %T", source)
}
Expand Down Expand Up @@ -352,7 +352,7 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to publish: %s", err)
}

err = collection.Add(published)
err = collection.Add(published, collectionFactory.RefListCollection())
if err != nil {
return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to save to DB: %s", err)
}
Expand Down Expand Up @@ -469,7 +469,7 @@ func apiPublishUpdateSwitch(c *gin.Context) {
resources := []string{string(published.Key())}
taskName := fmt.Sprintf("Update published %s repository %s/%s", published.SourceKind, published.StoragePrefix(), published.Distribution)
maybeRunTaskInBackground(c, taskName, resources, func(out aptly.Progress, _ *task.Detail) (*task.ProcessReturnValue, error) {
err = collection.LoadComplete(published, collectionFactory)
err = collection.LoadComplete(published, collectionFactory, collectionFactory.RefListCollection())
if err != nil {
return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("Unable to update: %s", err)
}
Expand All @@ -495,7 +495,7 @@ func apiPublishUpdateSwitch(c *gin.Context) {
return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("Unable to update: %s", err)
}

err = collection.Update(published)
err = collection.Update(published, collectionFactory.RefListCollection())
if err != nil {
return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to save to DB: %s", err)
}
Expand Down
25 changes: 16 additions & 9 deletions api/repos.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,6 +155,13 @@ func apiReposCreate(c *gin.Context) {
return
}

collection := collectionFactory.LocalRepoCollection()
err := collection.Add(repo, collectionFactory.RefListCollection())
if err != nil {
AbortWithJSONError(c, 400, err)
return
}

err := localRepoCollection.Add(repo)
if err != nil {
AbortWithJSONError(c, http.StatusInternalServerError, err)
Expand Down Expand Up @@ -218,7 +225,7 @@ func apiReposEdit(c *gin.Context) {
repo.DefaultComponent = *b.DefaultComponent
}

err = collection.Update(repo)
err = collection.Update(repo, collectionFactory.RefListCollection())
if err != nil {
AbortWithJSONError(c, 500, err)
return
Expand Down Expand Up @@ -326,7 +333,7 @@ func apiReposPackagesShow(c *gin.Context) {
return
}

err = collection.LoadComplete(repo)
err = collection.LoadComplete(repo, collectionFactory.RefListCollection())
if err != nil {
AbortWithJSONError(c, 500, err)
return
Expand Down Expand Up @@ -360,7 +367,7 @@ func apiReposPackagesAddDelete(c *gin.Context, taskNamePrefix string, cb func(li
resources := []string{string(repo.Key())}

maybeRunTaskInBackground(c, taskNamePrefix+repo.Name, resources, func(out aptly.Progress, _ *task.Detail) (*task.ProcessReturnValue, error) {
err = collection.LoadComplete(repo)
err = collection.LoadComplete(repo, collectionFactory.RefListCollection())
if err != nil {
return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, err
}
Expand Down Expand Up @@ -389,9 +396,9 @@ func apiReposPackagesAddDelete(c *gin.Context, taskNamePrefix string, cb func(li
}
}

repo.UpdateRefList(deb.NewPackageRefListFromPackageList(list))
repo.UpdateRefList(deb.NewSplitRefListFromPackageList(list))

err = collectionFactory.LocalRepoCollection().Update(repo)
err = collectionFactory.LocalRepoCollection().Update(repo, collectionFactory.RefListCollection())
if err != nil {
return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to save: %s", err)
}
Expand Down Expand Up @@ -517,7 +524,7 @@ func apiReposPackageFromDir(c *gin.Context) {
resources := []string{string(repo.Key())}
resources = append(resources, sources...)
maybeRunTaskInBackground(c, taskName, resources, func(out aptly.Progress, _ *task.Detail) (*task.ProcessReturnValue, error) {
err = collection.LoadComplete(repo)
err = collection.LoadComplete(repo, collectionFactory.RefListCollection())
if err != nil {
return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, err
}
Expand Down Expand Up @@ -552,9 +559,9 @@ func apiReposPackageFromDir(c *gin.Context) {
return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to import package files: %s", err)
}

repo.UpdateRefList(deb.NewPackageRefListFromPackageList(list))
repo.UpdateRefList(deb.NewSplitRefListFromPackageList(list))

err = collectionFactory.LocalRepoCollection().Update(repo)
err = collectionFactory.LocalRepoCollection().Update(repo, collectionFactory.RefListCollection())
if err != nil {
return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to save: %s", err)
}
Expand Down Expand Up @@ -867,7 +874,7 @@ func apiReposIncludePackageFromDir(c *gin.Context) {
_, failedFiles2, err = deb.ImportChangesFiles(
changesFiles, reporter, acceptUnsigned, ignoreSignature, forceReplace, noRemoveFiles, verifier,
repoTemplate, context.Progress(), collectionFactory.LocalRepoCollection(), collectionFactory.PackageCollection(),
context.PackagePool(), collectionFactory.ChecksumCollection, nil, query.Parse)
collectionFactory.RefListCollection(), context.PackagePool(), collectionFactory.ChecksumCollection, nil, query.Parse)
failedFiles = append(failedFiles, failedFiles2...)

if err != nil {
Expand Down
Loading
Loading