From 2fa35def2c9a2c0a3620d9b46693b0ca49598631 Mon Sep 17 00:00:00 2001 From: Krishnan Parthasarathi Date: Wed, 21 Dec 2022 16:24:07 -0800 Subject: [PATCH] Fix DeleteObject when only free versions remain (#16289) --- cmd/data-scanner.go | 3 +- cmd/erasure-object.go | 14 +++---- cmd/metacache-entries.go | 2 +- cmd/object-api-interface.go | 2 + cmd/xl-storage-format-utils.go | 2 +- cmd/xl-storage-format-v2.go | 24 +++++++++++- cmd/xl-storage-format_test.go | 2 +- cmd/xl-storage-free-version_test.go | 57 +++++++++++++++++++++++++++++ cmd/xl-storage.go | 2 +- 9 files changed, 95 insertions(+), 13 deletions(-) diff --git a/cmd/data-scanner.go b/cmd/data-scanner.go index 5095a431f..04448b564 100644 --- a/cmd/data-scanner.go +++ b/cmd/data-scanner.go @@ -997,7 +997,8 @@ func (i *scannerItem) applyTierObjSweep(ctx context.Context, o ObjectLayer, oi O // Remove this free version _, err = o.DeleteObject(ctx, oi.Bucket, oi.Name, ObjectOptions{ - VersionID: oi.VersionID, + VersionID: oi.VersionID, + InclFreeVersions: true, }) if err == nil { auditLogLifecycle(ctx, oi, ILMFreeVersionDelete) diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index b09f3ad72..5e5ae62fd 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -93,7 +93,7 @@ func (er erasureObjects) CopyObject(ctx context.Context, srcBucket, srcObject, d if srcOpts.VersionID != "" { metaArr, errs = readAllFileInfo(ctx, storageDisks, srcBucket, srcObject, srcOpts.VersionID, true) } else { - metaArr, errs = readAllXL(ctx, storageDisks, srcBucket, srcObject, true) + metaArr, errs = readAllXL(ctx, storageDisks, srcBucket, srcObject, true, false) } readQuorum, writeQuorum, err := objectQuorumFromMeta(ctx, metaArr, errs, er.defaultParityCount) @@ -501,7 +501,7 @@ func (er erasureObjects) deleteIfDangling(ctx context.Context, bucket, object st return m, err } -func readAllXL(ctx context.Context, disks []StorageAPI, bucket, object string, readData bool) ([]FileInfo, []error) { +func readAllXL(ctx context.Context, disks []StorageAPI, bucket, object string, readData, inclFreeVers bool) ([]FileInfo, []error) { metadataArray := make([]*xlMetaV2, len(disks)) metaFileInfos := make([]FileInfo, len(metadataArray)) metadataShallowVersions := make([][]xlMetaV2ShallowVersion, len(disks)) @@ -566,7 +566,7 @@ func readAllXL(ctx context.Context, disks []StorageAPI, bucket, object string, r readQuorum := (len(disks) + 1) / 2 meta := &xlMetaV2{versions: mergeXLV2Versions(readQuorum, false, 1, metadataShallowVersions...)} - lfi, err := meta.ToFileInfo(bucket, object, "") + lfi, err := meta.ToFileInfo(bucket, object, "", inclFreeVers) if err != nil { for i := range errs { if errs[i] == nil { @@ -596,7 +596,7 @@ func readAllXL(ctx context.Context, disks []StorageAPI, bucket, object string, r // make sure to preserve this for diskmtime based healing bugfix. diskMTime := metaFileInfos[index].DiskMTime - metaFileInfos[index], errs[index] = metadataArray[index].ToFileInfo(bucket, object, versionID) + metaFileInfos[index], errs[index] = metadataArray[index].ToFileInfo(bucket, object, versionID, inclFreeVers) if errs[index] != nil { continue } @@ -620,7 +620,7 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s if opts.VersionID != "" { metaArr, errs = readAllFileInfo(ctx, disks, bucket, object, opts.VersionID, readData) } else { - metaArr, errs = readAllXL(ctx, disks, bucket, object, readData) + metaArr, errs = readAllXL(ctx, disks, bucket, object, readData, opts.InclFreeVersions) } readQuorum, _, err := objectQuorumFromMeta(ctx, metaArr, errs, er.defaultParityCount) @@ -1714,7 +1714,7 @@ func (er erasureObjects) PutObjectMetadata(ctx context.Context, bucket, object s if opts.VersionID != "" { metaArr, errs = readAllFileInfo(ctx, disks, bucket, object, opts.VersionID, false) } else { - metaArr, errs = readAllXL(ctx, disks, bucket, object, false) + metaArr, errs = readAllXL(ctx, disks, bucket, object, false, false) } readQuorum, _, err := objectQuorumFromMeta(ctx, metaArr, errs, er.defaultParityCount) @@ -1787,7 +1787,7 @@ func (er erasureObjects) PutObjectTags(ctx context.Context, bucket, object strin if opts.VersionID != "" { metaArr, errs = readAllFileInfo(ctx, disks, bucket, object, opts.VersionID, false) } else { - metaArr, errs = readAllXL(ctx, disks, bucket, object, false) + metaArr, errs = readAllXL(ctx, disks, bucket, object, false, false) } readQuorum, _, err := objectQuorumFromMeta(ctx, metaArr, errs, er.defaultParityCount) diff --git a/cmd/metacache-entries.go b/cmd/metacache-entries.go index 6d941af23..5443c990f 100644 --- a/cmd/metacache-entries.go +++ b/cmd/metacache-entries.go @@ -211,7 +211,7 @@ func (e *metaCacheEntry) fileInfo(bucket string) (FileInfo, error) { ModTime: timeSentinel1970, }, nil } - return e.cached.ToFileInfo(bucket, e.name, "") + return e.cached.ToFileInfo(bucket, e.name, "", false) } return getFileInfo(e.metadata, bucket, e.name, "", false) } diff --git a/cmd/object-api-interface.go b/cmd/object-api-interface.go index 6bd544251..7edfa6e6f 100644 --- a/cmd/object-api-interface.go +++ b/cmd/object-api-interface.go @@ -96,6 +96,8 @@ type ObjectOptions struct { // IndexCB will return any index created but the compression. // Object must have been read at this point. IndexCB func() []byte + + InclFreeVersions bool } // ExpirationOptions represents object options for object expiration at objectLayer. diff --git a/cmd/xl-storage-format-utils.go b/cmd/xl-storage-format-utils.go index 3d9ea8864..11844a1f0 100644 --- a/cmd/xl-storage-format-utils.go +++ b/cmd/xl-storage-format-utils.go @@ -122,7 +122,7 @@ func getFileInfo(xlMetaBuf []byte, volume, path, versionID string, data bool) (F }, nil } inData = xlMeta.data - fi, err = xlMeta.ToFileInfo(volume, path, versionID) + fi, err = xlMeta.ToFileInfo(volume, path, versionID, false) } if !data || err != nil { return fi, err diff --git a/cmd/xl-storage-format-v2.go b/cmd/xl-storage-format-v2.go index ac04eafe0..81df38b35 100644 --- a/cmd/xl-storage-format-v2.go +++ b/cmd/xl-storage-format-v2.go @@ -1700,7 +1700,7 @@ func (x *xlMetaV2) AddLegacy(m *xlMetaV1Object) error { // ToFileInfo converts xlMetaV2 into a common FileInfo datastructure // for consumption across callers. -func (x xlMetaV2) ToFileInfo(volume, path, versionID string) (fi FileInfo, err error) { +func (x xlMetaV2) ToFileInfo(volume, path, versionID string, inclFreeVers bool) (fi FileInfo, err error) { var uv uuid.UUID if versionID != "" && versionID != nullVersionID { uv, err = uuid.Parse(versionID) @@ -1712,12 +1712,29 @@ func (x xlMetaV2) ToFileInfo(volume, path, versionID string) (fi FileInfo, err e var succModTime int64 isLatest := true nonFreeVersions := len(x.versions) + + var ( + freeFi FileInfo + freeFound bool + ) found := false for _, ver := range x.versions { header := &ver.header // skip listing free-version unless explicitly requested via versionID if header.FreeVersion() { nonFreeVersions-- + // remember the latest free version; will return this FileInfo if no non-free version remain + var freeVersion xlMetaV2Version + if inclFreeVers && !freeFound { + // ignore unmarshalling errors, will return errFileNotFound in that case + if _, err := freeVersion.unmarshalV(x.metaV, ver.meta); err == nil { + if freeFi, err = freeVersion.ToFileInfo(volume, path); err == nil { + freeFi.IsLatest = true // when this is returned, it would be the latest free version remaining. + freeFound = true + } + } + } + if header.VersionID != uv { continue } @@ -1749,6 +1766,11 @@ func (x xlMetaV2) ToFileInfo(volume, path, versionID string) (fi FileInfo, err e } if !found { if versionID == "" { + if inclFreeVers && nonFreeVersions == 0 { + if freeFound { + return freeFi, nil + } + } return FileInfo{}, errFileNotFound } diff --git a/cmd/xl-storage-format_test.go b/cmd/xl-storage-format_test.go index ffdc1beb2..05ed2a9f7 100644 --- a/cmd/xl-storage-format_test.go +++ b/cmd/xl-storage-format_test.go @@ -485,7 +485,7 @@ func BenchmarkXlMetaV2Shallow(b *testing.B) { b.Fatal(err) } // List... - _, err = xl.ToFileInfo("volume", "path", ids[rng.Intn(size)]) + _, err = xl.ToFileInfo("volume", "path", ids[rng.Intn(size)], false) if err != nil { b.Fatal(err) } diff --git a/cmd/xl-storage-free-version_test.go b/cmd/xl-storage-free-version_test.go index b2f67378b..3fb927110 100644 --- a/cmd/xl-storage-free-version_test.go +++ b/cmd/xl-storage-free-version_test.go @@ -18,6 +18,7 @@ package cmd import ( + "errors" "testing" "time" @@ -128,6 +129,11 @@ func TestFreeVersion(t *testing.T) { report() fatalErr(err) + // At this point the version stack must look as below, + // v3 --> free version 00000000-0000-0000-0000-0000000000f2 (from removal of null version) + // v2 --> free version 00000000-0000-0000-0000-0000000000f1 (from overwriting of null version ) + // v1 --> non-free version 00000000-0000-0000-0000-000000000001 + // Check number of free-versions freeVersions, err := xl.listFreeVersions(newtierfi.Volume, newtierfi.Name) if err != nil { @@ -137,6 +143,57 @@ func TestFreeVersion(t *testing.T) { t.Fatalf("Expected two free versions but got %d", len(freeVersions)) } + freeVersionsTests := []struct { + vol string + name string + inclFreeVers bool + afterFn func(fi FileInfo) (string, error) + expectedFree bool + expectedErr error + }{ + // ToFileInfo with 'inclFreeVers = true' should return the latest + // non-free version if one is present + { + vol: newtierfi.Volume, + name: newtierfi.Name, + inclFreeVers: true, + afterFn: xl.DeleteVersion, + expectedFree: false, + }, + // ToFileInfo with 'inclFreeVers = true' must return the latest free + // version when no non-free versions are present. + { + vol: newtierfi.Volume, + name: newtierfi.Name, + inclFreeVers: true, + expectedFree: true, + }, + // ToFileInfo with 'inclFreeVers = false' must return errFileNotFound + // when no non-free version exist. + { + vol: newtierfi.Volume, + name: newtierfi.Name, + inclFreeVers: false, + expectedErr: errFileNotFound, + }, + } + + for _, ft := range freeVersionsTests { + fi, err := xl.ToFileInfo(ft.vol, ft.name, "", ft.inclFreeVers) + if err != nil && !errors.Is(err, ft.expectedErr) { + t.Fatalf("ToFileInfo failed due to %v", err) + } + if got := fi.TierFreeVersion(); got != ft.expectedFree { + t.Fatalf("Expected free-version=%v but got free-version=%v", ft.expectedFree, got) + } + if ft.afterFn != nil { + _, err = ft.afterFn(fi) + if err != nil { + t.Fatalf("ft.afterFn failed with err %v", err) + } + } + } + // Simulate scanner removing free-version freefi := newtierfi for _, fvID := range fvIDs { diff --git a/cmd/xl-storage.go b/cmd/xl-storage.go index d18f0210d..6d141f4a8 100644 --- a/cmd/xl-storage.go +++ b/cmd/xl-storage.go @@ -2335,7 +2335,7 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f } // Replace the data of null version or any other existing version-id - ofi, err := xlMeta.ToFileInfo(dstVolume, dstPath, reqVID) + ofi, err := xlMeta.ToFileInfo(dstVolume, dstPath, reqVID, false) if err == nil && !ofi.Deleted { if xlMeta.SharedDataDirCountStr(reqVID, ofi.DataDir) == 0 { // Purge the destination path as we are not preserving anything