Retain current and upto NewerNoncurrentVersions versions (#17909)

applyNewerNoncurrentVersionLimit method should pass along versions
unaffected by NewerNoncurrentVersions rule for further ILM evaluation.
This commit is contained in:
Krishnan Parthasarathi 2023-08-24 09:26:29 -07:00 committed by GitHub
parent 4a6af93c83
commit 87cb0081ec
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 141 additions and 1 deletions

View File

@ -997,7 +997,6 @@ func (i *scannerItem) applyNewerNoncurrentVersionLimit(ctx context.Context, _ Ob
versioned := vcfg != nil && vcfg.Versioned(i.objectPath())
// current version + most recent lim noncurrent versions
objectInfos := make([]ObjectInfo, 0, len(fivs))
if i.lifeCycle == nil {
@ -1017,6 +1016,10 @@ func (i *scannerItem) applyNewerNoncurrentVersionLimit(ctx context.Context, _ Ob
}
overflowVersions := fivs[lim+1:]
// Retain the current version + most recent lim noncurrent versions
for _, fi := range fivs[:lim+1] {
objectInfos = append(objectInfos, fi.ToObjectInfo(i.bucket, i.objectPath(), versioned))
}
toDel := make([]ObjectToDelete, 0, len(overflowVersions))
for _, fi := range overflowVersions {

137
cmd/data-scanner_test.go Normal file
View File

@ -0,0 +1,137 @@
// Copyright (c) 2015-2023 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package cmd
import (
"context"
"encoding/xml"
"sync"
"testing"
"time"
"github.com/google/uuid"
"github.com/minio/minio/internal/bucket/lifecycle"
"github.com/minio/minio/internal/bucket/versioning"
)
func TestApplyNewerNoncurrentVersionsLimit(t *testing.T) {
objAPI, disks, err := prepareErasure(context.Background(), 8)
if err != nil {
t.Fatalf("Failed to initialize object layer: %v", err)
}
defer removeRoots(disks)
setObjectLayer(objAPI)
globalBucketMetadataSys = NewBucketMetadataSys()
globalBucketObjectLockSys = &BucketObjectLockSys{}
globalBucketVersioningSys = &BucketVersioningSys{}
globalExpiryState = newExpiryState()
var wg sync.WaitGroup
wg.Add(1)
expired := make([]ObjectToDelete, 0, 5)
go func() {
defer wg.Done()
for t := range globalExpiryState.byNewerNoncurrentCh {
expired = append(expired, t.versions...)
}
}()
lc := lifecycle.Lifecycle{
Rules: []lifecycle.Rule{
{
ID: "max-versions",
Status: "Enabled",
NoncurrentVersionExpiration: lifecycle.NoncurrentVersionExpiration{
NewerNoncurrentVersions: 1,
},
},
},
}
lcXML, err := xml.Marshal(lc)
if err != nil {
t.Fatalf("Failed to marshal lifecycle config: %v", err)
}
vcfg := versioning.Versioning{
Status: "Enabled",
}
vcfgXML, err := xml.Marshal(vcfg)
if err != nil {
t.Fatalf("Failed to marshal versioning config: %v", err)
}
bucket := "bucket"
obj := "obj-1"
now := time.Now()
meta := BucketMetadata{
Name: bucket,
Created: now,
LifecycleConfigXML: lcXML,
VersioningConfigXML: vcfgXML,
VersioningConfigUpdatedAt: now,
LifecycleConfigUpdatedAt: now,
lifecycleConfig: &lc,
versioningConfig: &vcfg,
}
globalBucketMetadataSys.Set(bucket, meta)
item := scannerItem{
Path: obj,
bucket: bucket,
prefix: "",
objectName: obj,
lifeCycle: &lc,
}
modTime := time.Now()
uuids := make([]uuid.UUID, 5)
for i := range uuids {
uuids[i] = uuid.UUID([16]byte{15: uint8(i + 1)})
}
fivs := make([]FileInfo, 5)
for i := 0; i < 5; i++ {
fivs[i] = FileInfo{
Volume: bucket,
Name: obj,
VersionID: uuids[i].String(),
IsLatest: i == 0,
ModTime: modTime.Add(-1 * time.Duration(i) * time.Minute),
Size: 1 << 10,
NumVersions: 5,
}
}
versioned := vcfg.Status == "Enabled"
wants := make([]ObjectInfo, 2)
for i, fi := range fivs[:2] {
wants[i] = fi.ToObjectInfo(bucket, obj, versioned)
}
gots, err := item.applyNewerNoncurrentVersionLimit(context.TODO(), objAPI, fivs)
if err != nil {
t.Fatalf("Failed with err: %v", err)
}
if len(gots) != len(wants) {
t.Fatalf("Expected %d objects but got %d", len(wants), len(gots))
}
// Close expiry state's channel to inspect object versions enqueued for expiration
close(globalExpiryState.byNewerNoncurrentCh)
wg.Wait()
for _, obj := range expired {
switch obj.ObjectV.VersionID {
case uuids[2].String(), uuids[3].String(), uuids[4].String():
default:
t.Errorf("Unexpected versionID being expired: %#v\n", obj)
}
}
}