fix: make sure we list freeVersions like DEL marker with --versions (#19878)

freeVersions() was being incorrectly skipped; list it as
valid objects properly.

Co-authored-by: Krishnan Parthasarathi <Krishnan Parthasarathi>
This commit is contained in:
Harshavardhana 2024-06-07 15:18:44 -07:00 committed by GitHub
parent 2dd8faaedc
commit 29a25a538f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 63 additions and 12 deletions

View File

@ -300,7 +300,7 @@ func (e *metaCacheEntry) fileInfoVersions(bucket string) (FileInfoVersions, erro
}, nil
}
// Too small gains to reuse cache here.
return getFileInfoVersions(e.metadata, bucket, e.name, false)
return getFileInfoVersions(e.metadata, bucket, e.name, false, true)
}
// metaCacheEntries is a slice of metacache entries.

View File

@ -391,7 +391,7 @@ func (r *metacacheReader) filter(o listPathOptions) (entries metaCacheEntriesSor
if !o.InclDeleted && entry.isObject() && entry.isLatestDeletemarker() && !entry.isObjectDir() {
return true
}
if entry.isAllFreeVersions() {
if !o.InclDeleted && entry.isAllFreeVersions() {
return true
}
entries.o = append(entries.o, entry)

View File

@ -23,22 +23,40 @@ import (
"github.com/zeebo/xxh3"
)
func getFileInfoVersions(xlMetaBuf []byte, volume, path string, allParts bool) (FileInfoVersions, error) {
// getFileInfoVersions partitions this object's versions such that,
// - fivs.Versions has all the non-free versions
// - fivs.FreeVersions has all the free versions
//
// if inclFreeVersions is true all the versions are in fivs.Versions, free and non-free versions alike.
//
// Note: Only the scanner requires fivs.Versions to have exclusively non-free versions. This is used while enforcing NewerNoncurrentVersions lifecycle element.
func getFileInfoVersions(xlMetaBuf []byte, volume, path string, allParts, inclFreeVersions bool) (FileInfoVersions, error) {
fivs, err := getAllFileInfoVersions(xlMetaBuf, volume, path, allParts)
if err != nil {
return fivs, err
}
// If inclFreeVersions is false, partition the versions in fivs.Versions
// such that finally fivs.Versions has
// all the non-free versions and fivs.FreeVersions has all the free
// versions.
n := 0
for _, fi := range fivs.Versions {
// Filter our tier object delete marker
if !fi.TierFreeVersion() {
fivs.Versions[n] = fi
n++
// filter our tier object delete marker
if fi.TierFreeVersion() {
if !inclFreeVersions {
fivs.FreeVersions = append(fivs.FreeVersions, fi)
}
} else {
fivs.FreeVersions = append(fivs.FreeVersions, fi)
if !inclFreeVersions {
fivs.Versions[n] = fi
}
n++
}
}
fivs.Versions = fivs.Versions[:n]
if !inclFreeVersions {
fivs.Versions = fivs.Versions[:n]
}
// Update numversions
for i := range fivs.Versions {
fivs.Versions[i].NumVersions = n

View File

@ -18,6 +18,7 @@
package cmd
import (
"slices"
"sort"
"testing"
"time"
@ -145,7 +146,7 @@ func TestGetFileInfoVersions(t *testing.T) {
}
xl := xlMetaV2{}
var versions []FileInfo
var freeVersionIDs []string
var allVersionIDs, freeVersionIDs []string
for i := 0; i < 5; i++ {
fi := basefi
fi.VersionID = mustGetUUID()
@ -167,18 +168,31 @@ func TestGetFileInfoVersions(t *testing.T) {
// delete this version leading to a free version
xl.DeleteVersion(fi)
freeVersionIDs = append(freeVersionIDs, fi.TierFreeVersionID())
allVersionIDs = append(allVersionIDs, fi.TierFreeVersionID())
} else {
versions = append(versions, fi)
allVersionIDs = append(allVersionIDs, fi.VersionID)
}
}
buf, err := xl.AppendTo(nil)
if err != nil {
t.Fatalf("Failed to serialize xlmeta %v", err)
}
fivs, err := getFileInfoVersions(buf, basefi.Volume, basefi.Name, true)
fivs, err := getFileInfoVersions(buf, basefi.Volume, basefi.Name, true, false)
if err != nil {
t.Fatalf("getFileInfoVersions failed: %v", err)
}
chkNumVersions := func(fis []FileInfo) bool {
for i := 0; i < len(fis)-1; i++ {
if fis[i].NumVersions != fis[i+1].NumVersions {
return false
}
}
return true
}
if !chkNumVersions(fivs.Versions) {
t.Fatalf("Expected all versions to have the same NumVersions")
}
sort.Slice(versions, func(i, j int) bool {
if versions[i].IsLatest {
@ -194,6 +208,9 @@ func TestGetFileInfoVersions(t *testing.T) {
if fi.VersionID != versions[i].VersionID {
t.Fatalf("getFileInfoVersions: versions don't match at %d, version id expected %s but got %s", i, fi.VersionID, versions[i].VersionID)
}
if fi.NumVersions != len(fivs.Versions) {
t.Fatalf("getFileInfoVersions: version with %s version id expected to have %d as NumVersions but got %d", fi.VersionID, len(fivs.Versions), fi.NumVersions)
}
}
for i, free := range fivs.FreeVersions {
@ -201,4 +218,20 @@ func TestGetFileInfoVersions(t *testing.T) {
t.Fatalf("getFileInfoVersions: free versions don't match at %d, version id expected %s but got %s", i, free.VersionID, freeVersionIDs[i])
}
}
// versions are stored in xl-meta sorted in descending order of their ModTime
slices.Reverse(allVersionIDs)
fivs, err = getFileInfoVersions(buf, basefi.Volume, basefi.Name, true, true)
if err != nil {
t.Fatalf("getFileInfoVersions failed: %v", err)
}
if !chkNumVersions(fivs.Versions) {
t.Fatalf("Expected all versions to have the same NumVersions")
}
for i, fi := range fivs.Versions {
if fi.VersionID != allVersionIDs[i] {
t.Fatalf("getFileInfoVersions: all versions don't match at %d expected %s but got %s", i, allVersionIDs[i], fi.VersionID)
}
}
}

View File

@ -586,7 +586,7 @@ func (s *xlStorage) NSScanner(ctx context.Context, cache dataUsageCache, updates
// Remove filename which is the meta file.
item.transformMetaDir()
fivs, err := getFileInfoVersions(buf, item.bucket, item.objectPath(), false)
fivs, err := getFileInfoVersions(buf, item.bucket, item.objectPath(), false, false)
metaDataPoolPut(buf)
if err != nil {
res["err"] = err.Error()