diff --git a/cmd/batch-authz-review6_test.go b/cmd/batch-authz-review6_test.go index 7fefe6f74..c11a84c4d 100644 --- a/cmd/batch-authz-review6_test.go +++ b/cmd/batch-authz-review6_test.go @@ -330,7 +330,7 @@ func TestBatchReviewRotationKeepsMigrationCleanup(t *testing.T) { t.Errorf("rotation changed a draining source after its version was copied (rotation error: %v)", rotationErr) } snapshot := []FileInfo{{VersionID: before.VersionID, ModTime: beforeInfo.ModTime}} - if err := z.removeMovedVersions(t.Context(), bucket, object, 0, snapshot, nil); err != nil { + if _, err := z.removeMovedVersions(t.Context(), bucket, object, 0, snapshot, nil); err != nil { t.Fatalf("rotation prevented migration cleanup: %v", err) } if _, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: before.VersionID}); !isErrObjectNotFound(err) && !isErrVersionNotFound(err) { diff --git a/cmd/erasure-server-pool-consistency.go b/cmd/erasure-server-pool-consistency.go index f44e2ce1f..79ba0f01c 100644 --- a/cmd/erasure-server-pool-consistency.go +++ b/cmd/erasure-server-pool-consistency.go @@ -271,16 +271,17 @@ func (z *erasureServerPools) applyMigrationExpiry(ctx context.Context, bucket, o // in the source until the ordinary expiry path has deleted them: that path // owns replication state and tier cleanup, and it re-evaluates the current // lifecycle rule. A migration cleanup must never turn an asynchronous ILM -// decision into an unconditional physical purge. -func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, object string, source int, versions []FileInfo, expired []migrationExpiredVersion) error { +// decision into an unconditional physical purge. It returns how many source +// versions it deleted. +func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, object string, source int, versions []FileInfo, expired []migrationExpiredVersion) (removed int, err error) { if err := z.applyMigrationExpiry(ctx, bucket, object, source, expired); err != nil { - return err + return 0, err } object = encodeDirObject(object) lk := z.NewNSLock(bucket, object) lkctx, err := lk.GetLock(ctx, globalOperationTimeout) if err != nil { - return err + return 0, err } ctx = lkctx.Context() defer lk.Unlock(lkctx) @@ -294,7 +295,7 @@ func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, ob continue } if err != nil { - return err + return removed, err } var src *ObjectInfo var retained []PoolObjInfo @@ -309,7 +310,7 @@ func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, ob continue } if !src.ModTime.Equal(version.ModTime) || src.DeleteMarker != version.Deleted { - return PreConditionFailed{} + return removed, PreConditionFailed{} } if isExpiredVersion(expired, version) { // The logical expiry path owns this version. Keeping the source is @@ -319,17 +320,20 @@ 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) { - return PreConditionFailed{} + return removed, PreConditionFailed{} } _, err = z.serverPools[source].DeleteObject(ctx, bucket, object, ObjectOptions{ VersionID: vid, NoLock: true, NoAuditLog: true, SkipFreeVersion: true, // preserve the old prefix cleanup's tier-GC behavior }) if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) { - return err + return removed, err + } + if err == nil { + removed++ } } - return nil + return removed, nil } // ilmDeletesVersionOnly reports whether the ILM action queued for a version diff --git a/cmd/erasure-server-pool-decom.go b/cmd/erasure-server-pool-decom.go index 16ea90b4d..4f1614380 100644 --- a/cmd/erasure-server-pool-decom.go +++ b/cmd/erasure-server-pool-decom.go @@ -28,6 +28,7 @@ import ( "slices" "sort" "strings" + "sync/atomic" "time" "github.com/dustin/go-humanize" @@ -752,7 +753,25 @@ func (set *erasureObjects) listObjectsToDecommission(ctx context.Context, bi dec return err } +// decommissionPool moves bucket bi out of pool idx. A key below another key +// ("k/b" under "k") is a directory inside the directory of that key, and the +// listing does not descend into the directory of an object: a walk sees "k/b" +// only once "k" has left the pool. The walk is repeated while it removes +// source versions, so each pass moves the keys that the previous one +// uncovered. Each repeat needs a version removed from the pool, which takes +// no new writes, so the repeats end. func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool *erasureSets, bi decomBucketInfo) error { + for { + removed, err := z.decommissionPoolWalk(ctx, idx, pool, bi) + if err != nil || removed == 0 || contextCanceled(ctx) { + return err + } + } +} + +// decommissionPoolWalk walks bucket bi of pool idx once and returns the number +// of keys whose source versions it removed. +func (z *erasureServerPools) decommissionPoolWalk(ctx context.Context, idx int, pool *erasureSets, bi decomBucketInfo) (int64, error) { ctx = logger.SetReqInfo(ctx, &logger.ReqInfo{}) const envDecomWorkers = "_MINIO_DECOMMISSION_WORKERS" @@ -766,9 +785,11 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool // add that many extra workers. workerSize += len(pool.sets) + var removed atomic.Int64 + wk, err := workers.New(workerSize) if err != nil { - return err + return 0, err } var vc *versioning.Versioning @@ -779,24 +800,24 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool if bi.Name != minioMetaBucket { vc, err = globalBucketVersioningSys.Get(bi.Name) if err != nil { - return err + return 0, err } // Check if the current bucket has a configured lifecycle policy lc, lifecycleUpdatedAt, err = globalBucketMetadataSys.GetLifecycleConfig(bi.Name) if err != nil && !errors.Is(err, BucketLifecycleNotFound{Bucket: bi.Name}) { - return err + return 0, err } // Check if bucket is object locked. lr, err = globalBucketObjectLockSys.Get(bi.Name) if err != nil { - return err + return 0, err } rcfg, err = getReplicationConfig(ctx, bi.Name) if err != nil { - return err + return 0, err } } @@ -1019,7 +1040,10 @@ 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, expired) + n, err := z.removeMovedVersions(ctx, bi.Name, entry.name, idx, fivs.Versions, expired) + if n > 0 { + removed.Add(1) + } if err == nil { err = z.removeFreeVersions(ctx, bi.Name, entry.name, idx, fivs.FreeVersions) } @@ -1062,7 +1086,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool }(setIdx) } wk.Wait() - return nil + return removed.Load(), nil } //msgp:ignore decomMetrics diff --git a/cmd/erasure-server-pool-lifecycle-authority_test.go b/cmd/erasure-server-pool-lifecycle-authority_test.go index e3bd7aaa5..82a13dde2 100644 --- a/cmd/erasure-server-pool-lifecycle-authority_test.go +++ b/cmd/erasure-server-pool-lifecycle-authority_test.go @@ -1666,7 +1666,7 @@ func TestPoolsMigrationLifecycleRefusedExpiryKeepsSource(t *testing.T) { t.Fatal(err) } version := FileInfo{Name: key, VersionID: id, ModTime: source.ModTime} - err = s.z.removeMovedVersions(ctx, s.bucket, key, 0, []FileInfo{version}, []migrationExpiredVersion{{ + _, err = s.z.removeMovedVersions(ctx, s.bucket, key, 0, []FileInfo{version}, []migrationExpiredVersion{{ version: version, event: lifecycle.Event{Action: lifecycle.DeleteAction}, lifecycleUpdatedAt: updatedAt, diff --git a/cmd/erasure-server-pool-migration-delete_test.go b/cmd/erasure-server-pool-migration-delete_test.go index 55bab37fc..557cc2efa 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, nil) + _, 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-rebalance.go b/cmd/erasure-server-pool-rebalance.go index 81160fb0e..fb9eb996d 100644 --- a/cmd/erasure-server-pool-rebalance.go +++ b/cmd/erasure-server-pool-rebalance.go @@ -779,7 +779,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, expired) + _, 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 {