diff --git a/cmd/erasure-server-pool-consistency.go b/cmd/erasure-server-pool-consistency.go index 776e1e4cc..188ded74b 100644 --- a/cmd/erasure-server-pool-consistency.go +++ b/cmd/erasure-server-pool-consistency.go @@ -130,7 +130,12 @@ func (z *erasureServerPools) checkDataMovementCopies(ctx context.Context, bucket // removeMovedVersions retires only the source versions in the walker's // snapshot. A concurrent delete can create a new marker in the source pool; // deleting the whole prefix would remove that marker and expose moved data. -func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, object string, source int, versions []FileInfo) error { +// Versions in expired were not moved because ILM deletes them. They are +// removed from the source without a moved copy, as the queued expiry would +// remove them. Left behind, such a version can become the latest one in the +// source pool once the moved versions are gone, and the expiry, which +// re-evaluates the rule there, would no longer delete it. +func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, object string, source int, versions, expired []FileInfo) error { object = encodeDirObject(object) lk := z.NewNSLock(bucket, object) lkctx, err := lk.GetLock(ctx, globalOperationTimeout) @@ -168,7 +173,7 @@ func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, ob } // Successful movement is not enough if the destination changed again. // Keep the source unless another copy still preserves its read result. - if len(retained) == 0 || retained[0].ObjInfo.ModTime.Before(src.ModTime) { + if !isExpiredVersion(expired, version) && (len(retained) == 0 || retained[0].ObjInfo.ModTime.Before(src.ModTime)) { return PreConditionFailed{} } _, err = z.serverPools[source].DeleteObject(ctx, bucket, object, ObjectOptions{ @@ -182,6 +187,23 @@ func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, ob return nil } +// ilmDeletesVersionOnly reports whether the ILM action queued for a version +// that data movement skipped deletes that version and nothing else, rather +// than every version of the object or only the restored copy of a +// transitioned version. Such a version is passed to removeMovedVersions. +func ilmDeletesVersionOnly(action lifecycle.Action) bool { + return action == lifecycle.DeleteAction || action == lifecycle.DeleteVersionAction +} + +func isExpiredVersion(expired []FileInfo, version FileInfo) bool { + for _, fi := range expired { + if fi.VersionID == version.VersionID && fi.ModTime.Equal(version.ModTime) && fi.Deleted == version.Deleted { + return true + } + } + return false +} + // Healing also commits object metadata. Both queued healing and drive healing // must use the same namespace as pooled writes, rather than racing them under // a destination set's independent namespace. diff --git a/cmd/erasure-server-pool-decom.go b/cmd/erasure-server-pool-decom.go index e0064bfcd..d5a557451 100644 --- a/cmd/erasure-server-pool-decom.go +++ b/cmd/erasure-server-pool-decom.go @@ -796,9 +796,11 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool } for setIdx, set := range pool.sets { - filterLifecycle := func(bucket, object string, fi FileInfo) bool { + // filterLifecycle returns the action of an ILM deletion it queued + // for the version, which is then not moved, or NoneAction. + filterLifecycle := func(bucket, object string, fi FileInfo) lifecycle.Action { if lc == nil { - return false + return lifecycle.NoneAction } versioned := vc != nil && vc.Versioned(object) objInfo := fi.ToObjectInfo(bucket, object, versioned) @@ -807,12 +809,18 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool switch { case evt.Action.DeleteRestored(): // if restored copy has expired, delete it synchronously applyExpiryOnTransitionedObject(ctx, z, objInfo, evt, lcEventSrc_Decom) - return false + return lifecycle.NoneAction + case evt.Action == lifecycle.DeleteAction && versioned: + // Expiring the latest version of a versioned bucket only puts + // a delete marker on top of it. The version itself survives as + // a noncurrent version and has to be moved like any other. The + // scanner applies the expiry once it lives in another pool. + return lifecycle.NoneAction case evt.Action.Delete(): globalExpiryState.enqueueByDays(objInfo, evt, lcEventSrc_Decom) - return true + return evt.Action default: - return false + return lifecycle.NoneAction } } @@ -833,11 +841,18 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool versionsSorter(fivs.Versions).reverse() var decommissioned int + var expired []FileInfo for _, version := range fivs.Versions { stopFn := globalDecommissionMetrics.log(decomMetricDecommissionObject, idx, bi.Name, version.Name, version.VersionID) // Apply lifecycle rules on the objects that are expired. - if filterLifecycle(bi.Name, version.Name, version) { + if action := filterLifecycle(bi.Name, version.Name, version); action.Delete() { decommissioned++ + if ilmDeletesVersionOnly(action) { + // Its source copy goes with those of the moved + // versions. A version that takes the whole object + // with it stays until the expiry deletes them all. + expired = append(expired, version) + } stopFn(version.Size, errors.New("ILM expired object/version will be skipped")) continue } @@ -981,7 +996,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool // if all versions were decommissioned, then we can delete the object versions. if decommissioned == len(fivs.Versions) { stopFn := globalDecommissionMetrics.log(decomMetricDecommissionRemoveObject, idx, bi.Name, entry.name) - err := z.removeMovedVersions(ctx, bi.Name, entry.name, idx, fivs.Versions) + err := z.removeMovedVersions(ctx, bi.Name, entry.name, idx, fivs.Versions, expired) if err == nil { err = z.removeFreeVersions(ctx, bi.Name, entry.name, idx, fivs.FreeVersions) } @@ -1199,6 +1214,9 @@ func (z *erasureServerPools) checkAfterDecom(ctx context.Context, idx int) error case evt.Action.DeleteRestored(): // if restored copy has expired,delete it synchronously applyExpiryOnTransitionedObject(ctx, z, objInfo, evt, lcEventSrc_Decom) return false + case evt.Action == lifecycle.DeleteAction && versioned: + // The version survives as noncurrent; it must have moved. + return false case evt.Action.Delete(): globalExpiryState.enqueueByDays(objInfo, evt, lcEventSrc_Decom) return true diff --git a/cmd/erasure-server-pool-migration-delete_test.go b/cmd/erasure-server-pool-migration-delete_test.go index 904c87c0f..55bab37fc 100644 --- a/cmd/erasure-server-pool-migration-delete_test.go +++ b/cmd/erasure-server-pool-migration-delete_test.go @@ -132,7 +132,7 @@ func TestPoolsMigrationCleanupRechecksSource(t *testing.T) { _, err := z.serverPools[1].DeleteObject(t.Context(), bucket, oi.Name, ObjectOptions{VersionID: nullVersionID}) mustIAM(t, err) } - err := z.removeMovedVersions(t.Context(), bucket, oi.Name, 0, snapshot) + err := z.removeMovedVersions(t.Context(), bucket, oi.Name, 0, snapshot, nil) if change != "delete-uuid" && err == nil { t.Fatal("cleanup accepted changed or uncertain state") } diff --git a/cmd/erasure-server-pool-migration-ilm_test.go b/cmd/erasure-server-pool-migration-ilm_test.go new file mode 100644 index 000000000..8eb108750 --- /dev/null +++ b/cmd/erasure-server-pool-migration-ilm_test.go @@ -0,0 +1,456 @@ +// Copyright (c) 2026 Denis Babchuk +// +// This file is part of Silo 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 . + +package cmd + +import ( + "slices" + "strings" + "testing" + "time" + + "github.com/minio/minio/internal/amztime" + "github.com/minio/minio/internal/bucket/lifecycle" + objectlock "github.com/minio/minio/internal/bucket/object/lock" + "github.com/minio/minio/internal/bucket/versioning" + xhttp "github.com/minio/minio/internal/http" +) + +// Decommission and rebalance leave a version that ILM expires to the expiry +// workers. Expiring the current version of a versioned object only puts a +// delete marker on top of it: the version stays as a noncurrent version and +// has to be moved like any other. A version that ILM does delete is not moved, +// and the source copies of the versions that were moved must still be removed. + +// migrationILM is a bucket in pool 0 of two pools. The ILM expiry workers are +// replaced by a queue that nothing drains, so the expiries a decommission or +// rebalance queues run after it, as when the workers lag behind the walk. +type migrationILM struct { + t *testing.T + z *erasureServerPools + bucket string + queue chan expiryOp + objects []string + labels map[string]string // object and version ID -> label of a version + markers map[string]string // object and version ID -> label of a delete marker +} + +func newMigrationILM(t *testing.T, status versioning.State, lock bool, rules string) *migrationILM { + t.Helper() + z, _ := migrationTestPools(t) + ctx := t.Context() + m := &migrationILM{ + t: t, z: z, bucket: "migration-ilm", queue: make(chan expiryOp, 100), + labels: map[string]string{}, markers: map[string]string{}, + } + err := z.MakeBucket(ctx, m.bucket, MakeBucketOptions{LockEnabled: lock, VersioningEnabled: status == versioning.Enabled}) + if err != nil { + t.Fatal(err) + } + meta, err := globalBucketMetadataSys.Get(m.bucket) + if err != nil { + t.Fatal(err) + } + if status == versioning.Suspended { + meta.VersioningConfigXML = []byte("Suspended") + } + if rules != "" { + meta.LifecycleConfigXML = []byte("" + rules + "") + } + if err := meta.Save(ctx, z); err != nil { + t.Fatal(err) + } + globalBucketMetadataSys.Set(m.bucket, meta) + + es := &expiryState{ctx: ctx, objAPI: z} + workers := []chan expiryOp{m.queue} + es.workers.Store(&workers) + prev := globalExpiryState + globalExpiryState = es + t.Cleanup(func() { globalExpiryState = prev }) + return m +} + +func (m *migrationILM) key(object, versionID string) string { + if versionID == "" { + versionID = nullVersionID + } + return object + " " + versionID +} + +func (m *migrationILM) track(object string) { + if !slices.Contains(m.objects, object) { + m.objects = append(m.objects, object) + } +} + +// put writes a version of object to pool 0. +func (m *migrationILM) put(object, label string, opts ObjectOptions) ObjectInfo { + m.t.Helper() + oi := putConsistencyObject(m.t, m.z, m.bucket, object, 0, object+" "+label, opts) + m.labels[m.key(object, oi.VersionID)] = label + m.track(object) + return oi +} + +// marker writes a delete marker of object to pool 0. +func (m *migrationILM) marker(object, label string, mtime time.Time) { + m.t.Helper() + versionID := mustGetUUID() + _, err := m.z.serverPools[0].DeleteObject(m.t.Context(), m.bucket, object, + ObjectOptions{Versioned: true, VersionID: versionID, DeleteMarker: true, MTime: mtime}) + if err != nil { + m.t.Fatal(err) + } + m.markers[m.key(object, versionID)] = label + m.track(object) +} + +// tiered writes a version of object to pool 0 whose data is in a remote tier. +func (m *migrationILM) tiered(object, label string) ObjectInfo { + m.t.Helper() + ctx := m.t.Context() + oi := m.put(object, label, ObjectOptions{Versioned: true}) + infos, errs := readAllFileInfo(ctx, m.z.serverPools[0].getHashedSet(object).getDisks(), "", m.bucket, object, oi.VersionID, false, false) + if errs[0] != nil { + m.t.Fatal(errs[0]) + } + fi := infos[0] + fi.TransitionStatus, fi.TransitionTier, fi.TransitionedObjName = lifecycle.TransitionComplete, "WARM", "remote" + if err := m.z.serverPools[0].DecomTieredObject(ctx, m.bucket, object, fi, ObjectOptions{}); err != nil { + m.t.Fatal(err) + } + stored, err := m.z.serverPools[0].GetObjectInfo(ctx, m.bucket, object, ObjectOptions{VersionID: oi.VersionID}) + if err != nil || stored.TransitionedObject.Status != lifecycle.TransitionComplete { + m.t.Fatalf("fixture is not a transitioned version: %q, %v", stored.TransitionedObject.Status, err) + } + return oi +} + +func (m *migrationILM) label(object, versionID string, deleted bool) string { + key := m.key(object, versionID) + if l, ok := m.markers[key]; ok && deleted { + return l + } + if l, ok := m.labels[key]; ok && !deleted { + return l + } + if deleted { + return "DM" + } + return "?" +} + +// move moves the contents of pool 0 out with a decommission, as the admin API +// runs it, including the check for versions left behind, or with a rebalance +// pass, with the walker's accounting and the removal of the source copies. +// The rebalance goal is never reached, so the walk visits every object. The +// test fails unless a decommission completes. ILM has not caught up yet. +func (m *migrationILM) move(kind string) { + m.t.Helper() + ctx := m.t.Context() + if kind == "rebalance" { + migrationTestWalk(m.t, m.z, m.bucket, 0, kind) + m.z.rebalMu.Lock() + m.z.rebalMeta.PoolStats[0].Info.Status = rebalCompleted + m.z.rebalMu.Unlock() + return + } + if err := m.z.StartDecommission(ctx, 0); err != nil { + m.t.Fatal(err) + } + m.z.doDecommissionInRoutine(ctx, 0) + m.z.poolMetaMutex.RLock() + info := *m.z.poolMeta.Pools[0].Decommission + m.z.poolMetaMutex.RUnlock() + if !info.Complete || info.Failed || info.ItemsDecommissionFailed > 0 { + m.t.Fatalf("decommission did not complete: complete=%t failed=%t itemsFailed=%d; pool 0 holds %v", + info.Complete, info.Failed, info.ItemsDecommissionFailed, m.view(0)) + } +} + +// migrate moves the contents of pool 0 out and lets ILM catch up. +func (m *migrationILM) migrate(kind string) { + m.t.Helper() + m.move(kind) + m.catchUp() +} + +// catchUp lets ILM catch up with data movement out of pool 0: the queued +// expiries run as the expiry workers run them, and the rules are applied to +// the current version of every object as the scanner applies them in the pool +// that holds it now. A transitioned version is not expired: its remote tier is +// not configured here. +func (m *migrationILM) catchUp() { + m.t.Helper() + ctx := m.t.Context() + for len(m.queue) > 0 { + op := <-m.queue + task, ok := op.(expiryTask) + if !ok { + m.t.Fatalf("unexpected expiry operation %T", op) + } + oi := task.objInfo + m.t.Logf("%s queued %s of %s %s", task.src, task.event.Action, oi.Name, m.label(oi.Name, oi.VersionID, oi.DeleteMarker)) + if oi.TransitionedObject.Status == "" { + applyExpiryOnNonTransitionedObjects(ctx, m.z, oi, task.event, task.src) + } + } + + lc, err := globalLifecycleSys.Get(m.bucket) + if err != nil { + return // no rules + } + lr, err := globalBucketObjectLockSys.Get(m.bucket) + if err != nil { + m.t.Fatal(err) + } + for _, object := range m.objects { + oi, err := m.z.GetObjectInfo(ctx, m.bucket, object, ObjectOptions{}) + if isErrObjectNotFound(err) || isErrMethodNotAllowed(err) { + continue // deleted + } + if err != nil { + m.t.Fatal(err) + } + evt := evalActionFromLifecycle(ctx, *lc, lr, nil, oi) + if evt.Action.Delete() && oi.TransitionedObject.Status == "" { + applyExpiryOnNonTransitionedObjects(ctx, m.z, oi, evt, lcEventSrc_Scanner) + } + } +} + +// latest returns the label of the version a read of object returns. +func (m *migrationILM) latest(object string) string { + m.t.Helper() + oi, err := m.z.GetObjectInfo(m.t.Context(), m.bucket, object, ObjectOptions{}) + if err != nil && (!oi.DeleteMarker || !isErrObjectNotFound(err) && !isErrMethodNotAllowed(err)) { + m.t.Fatal(err) + } + return m.label(object, oi.VersionID, oi.DeleteMarker) +} + +// view returns the versions of each object in a pool, newest first. Delete +// markers and versions the test did not write are "DM" and "?". +func (m *migrationILM) view(pool int) map[string][]string { + m.t.Helper() + view := map[string][]string{} + for _, set := range m.z.serverPools[pool].sets { + err := set.listObjectsToDecommission(m.t.Context(), decomBucketInfo{Name: m.bucket}, func(entry metaCacheEntry) { + fivs, err := entry.fileInfoVersions(m.bucket) + if err != nil { + m.t.Errorf("%s: %v", entry.name, err) + return + } + for _, fi := range fivs.Versions { + view[entry.name] = append(view[entry.name], m.label(entry.name, fi.VersionID, fi.Deleted)) + } + }) + if err != nil { + m.t.Fatal(err) + } + } + return view +} + +// check fails the test unless pool 0 is empty and pool 1 holds the versions +// of want. +func (m *migrationILM) check(want map[string][]string) { + m.t.Helper() + if left := m.view(0); len(left) > 0 { + m.t.Errorf("pool 0 still holds %v", left) + } + got := m.view(1) + for _, object := range m.objects { + if !slices.Equal(got[object], want[object]) { + m.t.Errorf("%s: pool 1 holds %v, want %v", object, got[object], want[object]) + } + } +} + +// versions writes v1, v2 and v3 of each object. +func (m *migrationILM) versions(opts ObjectOptions, objects ...string) { + for _, object := range objects { + for _, label := range []string{"v1", "v2", "v3"} { + m.put(object, label, opts) + } + } +} + +// aged writes v1, v2 and v3 of each object, five, four and age days ago. +func (m *migrationILM) aged(age int, objects ...string) { + for _, object := range objects { + m.put(object, "v1", ObjectOptions{Versioned: true, MTime: UTCNow().AddDate(0, 0, -5)}) + m.put(object, "v2", ObjectOptions{Versioned: true, MTime: UTCNow().AddDate(0, 0, -4)}) + m.put(object, "v3", ObjectOptions{Versioned: true, MTime: UTCNow().AddDate(0, 0, -age)}) + } +} + +// migrationILMRule returns a rule for the objects under obj-. +func migrationILMRule(action string) string { + return "ruleEnabledobj-" + action + "" +} + +// migrationILMExpireCurrent expires the current versions at a past date (S3 +// expiration dates are midnights). +func migrationILMExpireCurrent() string { + date := UTCNow().Truncate(24*time.Hour).AddDate(0, 0, -2).Format(time.RFC3339) + return migrationILMRule("" + date + "") +} + +// testMigrationLifecycleExpiry runs the cases that decommission and rebalance +// share. +func testMigrationLifecycleExpiry(t *testing.T, kind string) { + versioned := ObjectOptions{Versioned: true} + + t.Run("versioned/no-lifecycle", func(t *testing.T) { + m := newMigrationILM(t, versioning.Enabled, false, "") + m.versions(versioned, "obj-01", "keep-01") + m.migrate(kind) + m.check(map[string][]string{"obj-01": {"v3", "v2", "v1"}, "keep-01": {"v3", "v2", "v1"}}) + }) + + // The expiry puts a delete marker on top of v3, which must be moved. + t.Run("versioned/expire-current", func(t *testing.T) { + m := newMigrationILM(t, versioning.Enabled, false, migrationILMExpireCurrent()) + m.versions(versioned, "obj-01", "keep-01") + m.migrate(kind) + m.check(map[string][]string{"obj-01": {"DM", "v3", "v2", "v1"}, "keep-01": {"v3", "v2", "v1"}}) + }) + + // Object lock does not hold back the delete marker, and v3 is under + // retention. + t.Run("versioned/expire-current-locked", func(t *testing.T) { + m := newMigrationILM(t, versioning.Enabled, true, migrationILMExpireCurrent()) + m.put("obj-01", "v1", versioned) + m.put("obj-01", "v2", versioned) + v3 := m.put("obj-01", "v3", ObjectOptions{Versioned: true, UserDefined: map[string]string{ + strings.ToLower(xhttp.AmzObjectLockMode): string(objectlock.RetCompliance), + strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): amztime.ISO8601Format(UTCNow().Add(24 * time.Hour)), + }}) + m.migrate(kind) + m.check(map[string][]string{"obj-01": {"DM", "v3", "v2", "v1"}}) + oi, err := m.z.serverPools[1].GetObjectInfo(t.Context(), m.bucket, "obj-01", ObjectOptions{VersionID: v3.VersionID}) + if mode := objectlock.GetObjectRetentionMeta(oi.UserDefined).Mode; err != nil || mode != objectlock.RetCompliance { + t.Errorf("v3 retention in pool 1: %q, %v", mode, err) + } + }) + + // In a suspended bucket the expiry puts a null delete marker on top: a + // versioned v3 stays, a null current version is replaced. + t.Run("suspended/expire-current", func(t *testing.T) { + m := newMigrationILM(t, versioning.Suspended, false, migrationILMExpireCurrent()) + m.versions(versioned, "obj-01") + m.put("obj-null", "v1", versioned) + m.put("obj-null", "null", ObjectOptions{VersionSuspended: true}) + m.migrate(kind) + m.check(map[string][]string{"obj-01": {"DM", "v3", "v2", "v1"}, "obj-null": {"DM", "v1"}}) + }) + + // v1 has been noncurrent for four days and is deleted by the queued + // expiry. The source copies of v2 and v3 must be removed although v1 has + // not been deleted yet. + t.Run("versioned/noncurrent-expiration", func(t *testing.T) { + m := newMigrationILM(t, versioning.Enabled, false, + migrationILMRule("1")) + m.aged(0, "obj-01", "keep-01") + m.migrate(kind) + m.check(map[string][]string{"obj-01": {"v3", "v2"}, "keep-01": {"v3", "v2", "v1"}}) + }) + + // Only a delete marker without noncurrent versions is expired. + t.Run("versioned/expired-delete-marker", func(t *testing.T) { + m := newMigrationILM(t, versioning.Enabled, false, + migrationILMRule("true")) + m.marker("obj-dm", "dm", UTCNow()) + m.put("obj-top", "v1", versioned) + m.marker("obj-top", "dm", UTCNow()) + m.marker("keep-dm", "dm", UTCNow()) + m.migrate(kind) + m.check(map[string][]string{"obj-top": {"dm", "v1"}, "keep-dm": {"dm"}}) + }) + + // ILM deletes every version of obj-01 with v3, and of obj-dm with its + // marker. Until the queued expiry does, the moved older versions must + // stay hidden. + for _, tc := range []struct{ name, rule, object, latest string }{ + {"expire-all-versions", "1true", "obj-01", "v3"}, + {"expire-marker-all-versions", "1", "obj-dm", "dm"}, + } { + t.Run("versioned/"+tc.name, func(t *testing.T) { + m := newMigrationILM(t, versioning.Enabled, false, migrationILMRule(tc.rule)) + m.aged(3, "obj-01", "keep-01") + m.put("obj-dm", "v1", ObjectOptions{Versioned: true, MTime: UTCNow().AddDate(0, 0, -5)}) + m.marker("obj-dm", "dm", UTCNow().AddDate(0, 0, -3)) + m.move(kind) + if got := m.latest(tc.object); got != tc.latest { + t.Errorf("%s reads as %s before ILM deletes it, want %s; pool 0 holds %v, pool 1 holds %v", + tc.object, got, tc.latest, m.view(0), m.view(1)) + } + m.catchUp() + want := map[string][]string{"obj-01": {"v3", "v2", "v1"}, "keep-01": {"v3", "v2", "v1"}, "obj-dm": {"dm", "v1"}} + delete(want, tc.object) + m.check(want) + }) + } + + t.Run("unversioned/expire-current", func(t *testing.T) { + m := newMigrationILM(t, "", false, migrationILMExpireCurrent()) + m.put("obj-01", "null", ObjectOptions{}) + m.put("keep-01", "null", ObjectOptions{}) + m.migrate(kind) + m.check(map[string][]string{"keep-01": {"null"}}) + }) +} + +func TestPoolsDecommissionLifecycleExpiry(t *testing.T) { + testMigrationLifecycleExpiry(t, "decommission") + + // A transitioned v3 is moved as metadata; its remote data stays. + // Rebalance does not move transitioned versions yet. + t.Run("versioned/expire-current-tiered", func(t *testing.T) { + m := newMigrationILM(t, versioning.Enabled, false, migrationILMExpireCurrent()) + m.put("obj-01", "v1", ObjectOptions{Versioned: true}) + m.put("obj-01", "v2", ObjectOptions{Versioned: true}) + v3 := m.tiered("obj-01", "v3") + m.migrate("decommission") + m.check(map[string][]string{"obj-01": {"v3", "v2", "v1"}}) + oi, err := m.z.serverPools[1].GetObjectInfo(t.Context(), m.bucket, "obj-01", ObjectOptions{VersionID: v3.VersionID}) + if err != nil || oi.TransitionedObject.Status != lifecycle.TransitionComplete { + t.Errorf("v3 in pool 1 is not transitioned: %q, %v", oi.TransitionedObject.Status, err) + } + }) +} + +// A current version that ILM expires in a versioned bucket stays as a +// noncurrent version. The check after decommission must count it when it is +// left in the pool. +func TestPoolsDecommissionCountsExpiredCurrentVersions(t *testing.T) { + m := newMigrationILM(t, versioning.Enabled, false, migrationILMExpireCurrent()) + m.migrate("decommission") + putConsistencyObject(t, m.z, m.bucket, "obj-01", 0, "v1", ObjectOptions{Versioned: true}) + if err := m.z.checkAfterDecom(t.Context(), 0); err == nil { + t.Fatal("decommission check passed with an expired current version left in the pool") + } +} + +// Rebalance removes the source copies of an object only when it has counted +// all of its versions. A version that ILM deletes must be counted, or the +// moved versions stay in pool 0 as duplicates. +func TestPoolsRebalanceLifecycleExpiry(t *testing.T) { + testMigrationLifecycleExpiry(t, "rebalance") +} diff --git a/cmd/erasure-server-pool-rebalance.go b/cmd/erasure-server-pool-rebalance.go index 6c64f9f28..0ff18b378 100644 --- a/cmd/erasure-server-pool-rebalance.go +++ b/cmd/erasure-server-pool-rebalance.go @@ -587,19 +587,28 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string, } for setIdx, set := range pool.sets { - filterLifecycle := func(bucket, object string, fi FileInfo) bool { + // filterLifecycle returns the action of an ILM deletion it queued + // for the version, which is then not moved, or NoneAction. + filterLifecycle := func(bucket, object string, fi FileInfo) lifecycle.Action { if lc == nil { - return false + return lifecycle.NoneAction } versioned := vc != nil && vc.Versioned(object) objInfo := fi.ToObjectInfo(bucket, object, versioned) evt := evalActionFromLifecycle(ctx, *lc, lr, rcfg, objInfo) - if evt.Action.Delete() { + switch { + case evt.Action == lifecycle.DeleteAction && versioned: + // Expiring the latest version of a versioned bucket only puts + // a delete marker on top of it. The version itself survives as + // a noncurrent version and has to be moved like any other. The + // scanner applies the expiry once it lives in another pool. + return lifecycle.NoneAction + case evt.Action.Delete(): globalExpiryState.enqueueByDays(objInfo, evt, lcEventSrc_Rebal) - return true + return evt.Action } - return false + return lifecycle.NoneAction } rebalanceEntry := func(entry metaCacheEntry) { @@ -624,6 +633,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string, versionsSorter(fivs.Versions).reverse() var rebalanced int + var expired []FileInfo for _, version := range fivs.Versions { stopFn := globalRebalanceMetrics.log(rebalanceMetricRebalanceObject, poolIdx, bucket, version.Name, version.VersionID) @@ -634,7 +644,15 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string, } // Apply lifecycle rules on the objects that are expired. - if filterLifecycle(bucket, version.Name, version) { + if action := filterLifecycle(bucket, version.Name, version); action.Delete() { + // Count the version, or the source copies of the moved + // ones stay behind as duplicates. Its own source copy + // goes with them when ILM deletes it alone; any other + // version stays until the expiry has run. + rebalanced++ + if ilmDeletesVersionOnly(action) { + expired = append(expired, version) + } stopFn(version.Size, errors.New("ILM expired object/version will be skipped")) continue } @@ -743,7 +761,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string, // if all versions were rebalanced, we can delete the object versions. if rebalanced == len(fivs.Versions) { stopFn := globalRebalanceMetrics.log(rebalanceMetricRebalanceRemoveObject, poolIdx, bucket, entry.name) - err := z.removeMovedVersions(ctx, bucket, entry.name, poolIdx, fivs.Versions) + err := z.removeMovedVersions(ctx, bucket, entry.name, poolIdx, fivs.Versions, expired) stopFn(0, err) auditLogRebalance(ctx, "Rebalance:DeleteObject", bucket, entry.name, "", err) if err != nil {