mirror of
https://github.com/pgsty/minio.git
synced 2026-11-02 13:52:08 +03:00
Compare commits
12 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 0fc5a0b527 | |||
| a47bacbc0f | |||
| 0a546c7a76 | |||
| 8dac7f7c14 | |||
| 4cc3bc1de6 | |||
| 0cfca29cb2 | |||
| a7310d59fe | |||
| a995921e57 | |||
| 19910230aa | |||
| 7ce7fc8d1e | |||
| fb1a619ee1 | |||
| 3f13c26eed |
@@ -102,7 +102,7 @@ jobs:
|
|||||||
run: go test -race ./cmd -run '^Test(Multipart|PaginateMultipartUploads|ListMultipartUploads|CleanupStaleUploads|ExtractModTime)' -count=1 -timeout=5m
|
run: go test -race ./cmd -run '^Test(Multipart|PaginateMultipartUploads|ListMultipartUploads|CleanupStaleUploads|ExtractModTime)' -count=1 -timeout=5m
|
||||||
|
|
||||||
- name: Run listing metadata recovery, pagination and cancellation tests under race detector
|
- name: Run listing metadata recovery, pagination and cancellation tests under race detector
|
||||||
run: go test -race ./cmd -run '^Test(ListFallback|ListingCache|ListingMerge|ListingRecovery|ListingPending|ListingEntry|ListingSecondLock|StreamMetadataParts|LocalLockerListing|DistributedListing|MergeXLV2LatestNullCrossParity|CountXLV2MixedECGroups|ListCrossParity)' -count=1 -timeout=5m
|
run: go test -race ./cmd -run '^Test(ListFallback|ListingCache|ListingMerge|ListingRecovery|ListingPending|ListingEntry|ListingSecondLock|StreamMetadataParts|LocalLockerListing|DistributedListing|MergeXLV2LatestNullCrossParity|CountXLV2MixedECGroups|ListCrossParity|ListObjectVersionsCrossParityNull|MigrationCrossParity)' -count=1 -timeout=5m
|
||||||
|
|
||||||
- name: Run lock cancellation and late-grant cleanup tests under race detector
|
- name: Run lock cancellation and late-grant cleanup tests under race detector
|
||||||
run: go test -race ./internal/dsync ./internal/lsync -run '^Test(DRWMutex|LRWMutex)' -count=1 -timeout=5m
|
run: go test -race ./internal/dsync ./internal/lsync -run '^Test(DRWMutex|LRWMutex)' -count=1 -timeout=5m
|
||||||
|
|||||||
+50
-17
@@ -37,6 +37,7 @@ import (
|
|||||||
"github.com/minio/minio/internal/amztime"
|
"github.com/minio/minio/internal/amztime"
|
||||||
sse "github.com/minio/minio/internal/bucket/encryption"
|
sse "github.com/minio/minio/internal/bucket/encryption"
|
||||||
"github.com/minio/minio/internal/bucket/lifecycle"
|
"github.com/minio/minio/internal/bucket/lifecycle"
|
||||||
|
"github.com/minio/minio/internal/bucket/replication"
|
||||||
"github.com/minio/minio/internal/event"
|
"github.com/minio/minio/internal/event"
|
||||||
xhttp "github.com/minio/minio/internal/http"
|
xhttp "github.com/minio/minio/internal/http"
|
||||||
"github.com/minio/minio/internal/logger"
|
"github.com/minio/minio/internal/logger"
|
||||||
@@ -631,31 +632,63 @@ func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *Ob
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
// Delete remote object from warm-tier
|
// Recheck the lifecycle decision and create the hot-tier tombstone before
|
||||||
//
|
// touching the remote tier. If the lifecycle configuration changed after
|
||||||
// Unless the expiry only puts a delete marker on top of the version, as
|
// the scanner queued this task, DeleteObject rejects the stale action while
|
||||||
// for an object that is not transitioned: with versioning enabled, or
|
// the remote object is still protected.
|
||||||
// suspended and the version is not the null version, the version stays
|
|
||||||
// as a noncurrent version and must remain readable. Its remote object is
|
|
||||||
// removed later, through the free version added when the version itself
|
|
||||||
// is deleted.
|
|
||||||
opts.VersionSuspended = globalBucketVersioningSys.PrefixSuspended(oi.Bucket, oi.Name)
|
opts.VersionSuspended = globalBucketVersioningSys.PrefixSuspended(oi.Bucket, oi.Name)
|
||||||
keepRemote := opts.VersionID == "" &&
|
keepRemote := opts.VersionID == "" &&
|
||||||
(opts.Versioned || (opts.VersionSuspended && oi.VersionID != "" && oi.VersionID != nullVersionID))
|
(opts.Versioned || (opts.VersionSuspended && oi.VersionID != "" && oi.VersionID != nullVersionID))
|
||||||
|
if lcEvent.Action == lifecycle.DeleteAction || lcEvent.Action == lifecycle.DeleteVersionAction {
|
||||||
|
opts.SetEvalMetadataFn(func(goi *ObjectInfo, gerr error) (dsc ReplicateDecision, err error) {
|
||||||
|
return checkReplicateDelete(ctx, oi.Bucket, ObjectToDelete{
|
||||||
|
ObjectV: ObjectV{
|
||||||
|
ObjectName: goi.Name,
|
||||||
|
VersionID: opts.VersionID,
|
||||||
|
},
|
||||||
|
}, *goi, opts, gerr), nil
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// Delete the hot-tier namespace first. For a transitioned version this
|
||||||
|
// leaves a durable free-version record until the remote delete succeeds;
|
||||||
|
// that record is the retry point if the process stops between the two
|
||||||
|
// operations.
|
||||||
|
delInfo, err := objectAPI.DeleteObject(ctx, oi.Bucket, oi.Name, opts)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
if !keepRemote {
|
if !keepRemote {
|
||||||
err := deleteObjectFromRemoteTier(ctx, oi.TransitionedObject.Name, oi.TransitionedObject.VersionID, oi.TransitionedObject.Tier)
|
if err := deleteObjectFromRemoteTier(ctx, oi.TransitionedObject.Name, oi.TransitionedObject.VersionID, oi.TransitionedObject.Tier); err != nil {
|
||||||
if err == nil {
|
|
||||||
// Skip adding free version since we successfully deleted the
|
|
||||||
// remote object
|
|
||||||
opts.SkipFreeVersion = true
|
|
||||||
} else {
|
|
||||||
transitionLogIf(ctx, err)
|
transitionLogIf(ctx, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Now, delete object from hot-tier namespace
|
// Queue replication of the ILM-created delete marker or version purge,
|
||||||
if _, err := objectAPI.DeleteObject(ctx, oi.Bucket, oi.Name, opts); err != nil {
|
// mirroring the DELETE handler's post-delete scheduling.
|
||||||
return err
|
if delInfo.ReplicationStatus == replication.Pending || delInfo.VersionPurgeStatus == replication.VersionPurgePending {
|
||||||
|
dmVersionID, versionID := "", ""
|
||||||
|
if delInfo.DeleteMarker && delInfo.VersionPurgeStatus.Empty() {
|
||||||
|
dmVersionID = delInfo.VersionID
|
||||||
|
} else {
|
||||||
|
versionID = delInfo.VersionID
|
||||||
|
}
|
||||||
|
if delInfo.Name == "" {
|
||||||
|
delInfo.Name = oi.Name
|
||||||
|
}
|
||||||
|
scheduleReplicationDelete(ctx, DeletedObjectReplicationInfo{
|
||||||
|
DeletedObject: DeletedObject{
|
||||||
|
ObjectName: delInfo.Name,
|
||||||
|
VersionID: versionID,
|
||||||
|
DeleteMarkerVersionID: dmVersionID,
|
||||||
|
DeleteMarkerMTime: DeleteMarkerMTime{delInfo.ModTime},
|
||||||
|
DeleteMarker: delInfo.DeleteMarker,
|
||||||
|
ReplicationState: delInfo.ReplicationState(),
|
||||||
|
},
|
||||||
|
Bucket: oi.Bucket,
|
||||||
|
EventType: ReplicateILMDelete,
|
||||||
|
}, objectAPI)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Send audit for the lifecycle delete operation
|
// Send audit for the lifecycle delete operation
|
||||||
|
|||||||
@@ -2062,6 +2062,8 @@ const (
|
|||||||
ReplicateIncoming = "replicate:incoming"
|
ReplicateIncoming = "replicate:incoming"
|
||||||
// ReplicateIncomingDelete - audit trail of inline replication of deletes.
|
// ReplicateIncomingDelete - audit trail of inline replication of deletes.
|
||||||
ReplicateIncomingDelete = "replicate:incoming:delete"
|
ReplicateIncomingDelete = "replicate:incoming:delete"
|
||||||
|
// ReplicateILMDelete - audit trail of delete replication triggered by ILM expiry.
|
||||||
|
ReplicateILMDelete = "replicate:ilm:delete"
|
||||||
|
|
||||||
// ReplicateHeal - audit trail for healing of failed/pending replications
|
// ReplicateHeal - audit trail for healing of failed/pending replications
|
||||||
ReplicateHeal = "replicate:heal"
|
ReplicateHeal = "replicate:heal"
|
||||||
|
|||||||
+67
-8
@@ -776,7 +776,7 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
|
|||||||
}
|
}
|
||||||
|
|
||||||
// We got an entry which we should be able to heal.
|
// We got an entry which we should be able to heal.
|
||||||
fiv, err := entry.fileInfoVersions(bucket)
|
fiv, err := entry.allFileInfoVersions(bucket)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
err := bgSeq.queueHealTask(healSource{
|
err := bgSeq.queueHealTask(healSource{
|
||||||
bucket: bucket,
|
bucket: bucket,
|
||||||
@@ -1130,15 +1130,9 @@ eventLoop:
|
|||||||
applyExpiryRule(event, lcEventSrc_Scanner, oi)
|
applyExpiryRule(event, lcEventSrc_Scanner, oi)
|
||||||
|
|
||||||
case lifecycle.DeleteVersionAction: // noncurrent versions expiration
|
case lifecycle.DeleteVersionAction: // noncurrent versions expiration
|
||||||
opts := objOpts[idx]
|
|
||||||
remainingVersions--
|
remainingVersions--
|
||||||
size = 0
|
size = 0
|
||||||
toDel = append(toDel, ObjectToDelete{
|
toDel = append(toDel, ilmPurgeToDelete(ctx, i.bucket, oi, i.replication.Config != nil))
|
||||||
ObjectV: ObjectV{
|
|
||||||
ObjectName: opts.Name,
|
|
||||||
VersionID: opts.VersionID,
|
|
||||||
},
|
|
||||||
})
|
|
||||||
noncurrentEvents = append(noncurrentEvents, event)
|
noncurrentEvents = append(noncurrentEvents, event)
|
||||||
|
|
||||||
case lifecycle.TransitionAction, lifecycle.TransitionVersionAction:
|
case lifecycle.TransitionAction, lifecycle.TransitionVersionAction:
|
||||||
@@ -1240,6 +1234,21 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay
|
|||||||
// use prefix delete on exact object (this is an optimization to avoid fan-out calls)
|
// use prefix delete on exact object (this is an optimization to avoid fan-out calls)
|
||||||
opts.DeletePrefixObject = true
|
opts.DeletePrefixObject = true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ILM-created delete markers and version purges must carry the same
|
||||||
|
// replication decision as API-created deletes. DeleteAll is intentionally
|
||||||
|
// excluded: it is an object-local purge and has no single version to
|
||||||
|
// replicate.
|
||||||
|
if lcEvent.Action == lifecycle.DeleteAction || lcEvent.Action == lifecycle.DeleteVersionAction {
|
||||||
|
opts.SetEvalMetadataFn(func(oi *ObjectInfo, gerr error) (dsc ReplicateDecision, err error) {
|
||||||
|
return checkReplicateDelete(ctx, obj.Bucket, ObjectToDelete{
|
||||||
|
ObjectV: ObjectV{
|
||||||
|
ObjectName: oi.Name,
|
||||||
|
VersionID: opts.VersionID,
|
||||||
|
},
|
||||||
|
}, *oi, opts, gerr), nil
|
||||||
|
})
|
||||||
|
}
|
||||||
var (
|
var (
|
||||||
dobj ObjectInfo
|
dobj ObjectInfo
|
||||||
err error
|
err error
|
||||||
@@ -1276,6 +1285,29 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay
|
|||||||
dobj = obj
|
dobj = obj
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Queue replication of the ILM-created delete marker or version purge,
|
||||||
|
// mirroring the DELETE handler's post-delete scheduling.
|
||||||
|
if dobj.ReplicationStatus == replication.Pending || dobj.VersionPurgeStatus == replication.VersionPurgePending {
|
||||||
|
dmVersionID, versionID := "", ""
|
||||||
|
if dobj.DeleteMarker && dobj.VersionPurgeStatus.Empty() {
|
||||||
|
dmVersionID = dobj.VersionID
|
||||||
|
} else {
|
||||||
|
versionID = dobj.VersionID
|
||||||
|
}
|
||||||
|
scheduleReplicationDelete(ctx, DeletedObjectReplicationInfo{
|
||||||
|
DeletedObject: DeletedObject{
|
||||||
|
ObjectName: dobj.Name,
|
||||||
|
VersionID: versionID,
|
||||||
|
DeleteMarkerVersionID: dmVersionID,
|
||||||
|
DeleteMarkerMTime: DeleteMarkerMTime{dobj.ModTime},
|
||||||
|
DeleteMarker: dobj.DeleteMarker,
|
||||||
|
ReplicationState: dobj.ReplicationState(),
|
||||||
|
},
|
||||||
|
Bucket: obj.Bucket,
|
||||||
|
EventType: ReplicateILMDelete,
|
||||||
|
}, objLayer)
|
||||||
|
}
|
||||||
|
|
||||||
tags := newLifecycleAuditEvent(src, lcEvent).Tags()
|
tags := newLifecycleAuditEvent(src, lcEvent).Tags()
|
||||||
tags["version-id"] = dobj.VersionID
|
tags["version-id"] = dobj.VersionID
|
||||||
|
|
||||||
@@ -1309,6 +1341,33 @@ func applyExpiryRule(event lifecycle.Event, src lcEventSrc, obj ObjectInfo) {
|
|||||||
globalExpiryState.enqueueByDays(obj, event, src)
|
globalExpiryState.enqueueByDays(obj, event, src)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ilmPurgeToDelete builds the bulk delete entry for an ILM-initiated removal
|
||||||
|
// of a noncurrent version. The entry carries a pending version-purge state so
|
||||||
|
// the purge is replicated and retried like an API version purge.
|
||||||
|
func ilmPurgeToDelete(ctx context.Context, bucket string, oi ObjectInfo, replicate bool) ObjectToDelete {
|
||||||
|
otd := ObjectToDelete{
|
||||||
|
ObjectV: ObjectV{
|
||||||
|
ObjectName: oi.Name,
|
||||||
|
VersionID: oi.VersionID,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
if !replicate {
|
||||||
|
return otd
|
||||||
|
}
|
||||||
|
delOpts := ObjectOptions{
|
||||||
|
Versioned: globalBucketVersioningSys.PrefixEnabled(bucket, oi.Name),
|
||||||
|
VersionSuspended: globalBucketVersioningSys.PrefixSuspended(bucket, oi.Name),
|
||||||
|
}
|
||||||
|
dsc := checkReplicateDelete(ctx, bucket, otd, oi, delOpts, nil)
|
||||||
|
if !dsc.ReplicateAny() {
|
||||||
|
return otd
|
||||||
|
}
|
||||||
|
otd.VersionPurgeStatus = replication.VersionPurgePending
|
||||||
|
otd.VersionPurgeStatuses = dsc.PendingStatus()
|
||||||
|
otd.ReplicateDecisionStr = dsc.String()
|
||||||
|
return otd
|
||||||
|
}
|
||||||
|
|
||||||
// objectPath returns the prefix and object name.
|
// objectPath returns the prefix and object name.
|
||||||
func (i *scannerItem) objectPath() string {
|
func (i *scannerItem) objectPath() string {
|
||||||
return pathJoin(i.prefix, i.objectName)
|
return pathJoin(i.prefix, i.objectName)
|
||||||
|
|||||||
@@ -127,10 +127,119 @@ func (z *erasureServerPools) checkDataMovementCopies(ctx context.Context, bucket
|
|||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type migrationExpiredVersion struct {
|
||||||
|
version FileInfo
|
||||||
|
event lifecycle.Event
|
||||||
|
lifecycleUpdatedAt time.Time
|
||||||
|
source lcEventSrc
|
||||||
|
}
|
||||||
|
|
||||||
|
// applyMigrationExpiry runs the logical expiry path before migration retires
|
||||||
|
// any remaining source copies. This keeps replication and tier cleanup under
|
||||||
|
// their normal ownership and makes lifecycle configuration changes fail safe.
|
||||||
|
func (z *erasureServerPools) applyMigrationExpiry(ctx context.Context, bucket, object string, source int, expired []migrationExpiredVersion) error {
|
||||||
|
if len(expired) == 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
lc, lifecycleUpdatedAt, err := globalBucketMetadataSys.GetLifecycleConfig(bucket)
|
||||||
|
if err != nil {
|
||||||
|
return PreConditionFailed{}
|
||||||
|
}
|
||||||
|
for _, pending := range expired {
|
||||||
|
if !pending.lifecycleUpdatedAt.Equal(lifecycleUpdatedAt) {
|
||||||
|
return PreConditionFailed{}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
lr, err := globalBucketObjectLockSys.Get(bucket)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
rcfg, err := getReplicationConfig(ctx, bucket)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
encodedObject := encodeDirObject(object)
|
||||||
|
seenDeleteAll := false
|
||||||
|
eventSource := lcEventSrc_Decom
|
||||||
|
for _, pending := range expired {
|
||||||
|
if pending.source != lcEventSrc_None {
|
||||||
|
eventSource = pending.source
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, pending := range expired {
|
||||||
|
if pending.event.Action.DeleteAll() {
|
||||||
|
if seenDeleteAll {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
seenDeleteAll = true
|
||||||
|
}
|
||||||
|
|
||||||
|
versionID := pending.version.VersionID
|
||||||
|
if versionID == "" {
|
||||||
|
versionID = nullVersionID
|
||||||
|
}
|
||||||
|
copies, err := z.objectPoolInfos(ctx, bucket, encodedObject, ObjectOptions{
|
||||||
|
VersionID: versionID,
|
||||||
|
NoAuditLog: true,
|
||||||
|
})
|
||||||
|
if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
current := copies[0].ObjInfo
|
||||||
|
for _, copy := range copies {
|
||||||
|
if copy.Index == source {
|
||||||
|
current = copy.ObjInfo
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
current.Bucket = bucket
|
||||||
|
current.Name = object
|
||||||
|
currentEvent := evalActionFromLifecycle(ctx, *lc, lr, rcfg, current)
|
||||||
|
if currentEvent.Action != pending.event.Action {
|
||||||
|
return PreConditionFailed{}
|
||||||
|
}
|
||||||
|
|
||||||
|
var ok bool
|
||||||
|
if current.TransitionedObject.Status != "" && !pending.event.Action.DeleteAll() {
|
||||||
|
ok = applyExpiryOnTransitionedObject(ctx, z, current, pending.event, eventSource)
|
||||||
|
} else {
|
||||||
|
ok = applyExpiryOnNonTransitionedObjects(ctx, z, current, pending.event, eventSource)
|
||||||
|
}
|
||||||
|
if ok {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
// A concurrent expiry may have removed the addressed version after
|
||||||
|
// the first lookup. Anything still present means the logical delete
|
||||||
|
// was not committed and migration must stop before purging its source.
|
||||||
|
if _, err := z.objectPoolInfos(ctx, bucket, encodedObject, ObjectOptions{VersionID: versionID, NoAuditLog: true}); err == nil {
|
||||||
|
return PreConditionFailed{}
|
||||||
|
} else if !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// removeMovedVersions retires only the source versions in the walker's
|
// removeMovedVersions retires only the source versions in the walker's
|
||||||
// snapshot. A concurrent delete can create a new marker in the source pool;
|
// 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.
|
// 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 must stay
|
||||||
|
// 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 {
|
||||||
|
if err := z.applyMigrationExpiry(ctx, bucket, object, source, expired); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
object = encodeDirObject(object)
|
object = encodeDirObject(object)
|
||||||
lk := z.NewNSLock(bucket, object)
|
lk := z.NewNSLock(bucket, object)
|
||||||
lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
|
lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
|
||||||
@@ -166,6 +275,11 @@ func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, ob
|
|||||||
if !src.ModTime.Equal(version.ModTime) || src.DeleteMarker != version.Deleted {
|
if !src.ModTime.Equal(version.ModTime) || src.DeleteMarker != version.Deleted {
|
||||||
return PreConditionFailed{}
|
return PreConditionFailed{}
|
||||||
}
|
}
|
||||||
|
if isExpiredVersion(expired, version) {
|
||||||
|
// The logical expiry path owns this version. Keeping the source is
|
||||||
|
// also the retry point if that path failed or the pool check races it.
|
||||||
|
continue
|
||||||
|
}
|
||||||
// Successful movement is not enough if the destination changed again.
|
// Successful movement is not enough if the destination changed again.
|
||||||
// Keep the source unless another copy still preserves its read result.
|
// Keep the source unless another copy still preserves its read result.
|
||||||
if len(retained) == 0 || retained[0].ObjInfo.ModTime.Before(src.ModTime) {
|
if len(retained) == 0 || retained[0].ObjInfo.ModTime.Before(src.ModTime) {
|
||||||
@@ -182,6 +296,24 @@ func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, ob
|
|||||||
return nil
|
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 []migrationExpiredVersion, version FileInfo) bool {
|
||||||
|
for _, pending := range expired {
|
||||||
|
fi := pending.version
|
||||||
|
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
|
// Healing also commits object metadata. Both queued healing and drive healing
|
||||||
// must use the same namespace as pooled writes, rather than racing them under
|
// must use the same namespace as pooled writes, rather than racing them under
|
||||||
// a destination set's independent namespace.
|
// a destination set's independent namespace.
|
||||||
|
|||||||
@@ -769,6 +769,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
|
|||||||
|
|
||||||
var vc *versioning.Versioning
|
var vc *versioning.Versioning
|
||||||
var lc *lifecycle.Lifecycle
|
var lc *lifecycle.Lifecycle
|
||||||
|
var lifecycleUpdatedAt time.Time
|
||||||
var lr objectlock.Retention
|
var lr objectlock.Retention
|
||||||
var rcfg *replication.Config
|
var rcfg *replication.Config
|
||||||
if bi.Name != minioMetaBucket {
|
if bi.Name != minioMetaBucket {
|
||||||
@@ -778,7 +779,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Check if the current bucket has a configured lifecycle policy
|
// Check if the current bucket has a configured lifecycle policy
|
||||||
lc, err = globalLifecycleSys.Get(bi.Name)
|
lc, lifecycleUpdatedAt, err = globalBucketMetadataSys.GetLifecycleConfig(bi.Name)
|
||||||
if err != nil && !errors.Is(err, BucketLifecycleNotFound{Bucket: bi.Name}) {
|
if err != nil && !errors.Is(err, BucketLifecycleNotFound{Bucket: bi.Name}) {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -796,9 +797,11 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
|
|||||||
}
|
}
|
||||||
|
|
||||||
for setIdx, set := range pool.sets {
|
for setIdx, set := range pool.sets {
|
||||||
filterLifecycle := func(bucket, object string, fi FileInfo) bool {
|
// filterLifecycle returns an ILM deletion that migration must handle
|
||||||
|
// before retiring the source copy, or NoneAction.
|
||||||
|
filterLifecycle := func(bucket, object string, fi FileInfo) lifecycle.Action {
|
||||||
if lc == nil {
|
if lc == nil {
|
||||||
return false
|
return lifecycle.NoneAction
|
||||||
}
|
}
|
||||||
versioned := vc != nil && vc.Versioned(object)
|
versioned := vc != nil && vc.Versioned(object)
|
||||||
objInfo := fi.ToObjectInfo(bucket, object, versioned)
|
objInfo := fi.ToObjectInfo(bucket, object, versioned)
|
||||||
@@ -806,13 +809,20 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
|
|||||||
evt := evalActionFromLifecycle(ctx, *lc, lr, rcfg, objInfo)
|
evt := evalActionFromLifecycle(ctx, *lc, lr, rcfg, objInfo)
|
||||||
switch {
|
switch {
|
||||||
case evt.Action.DeleteRestored(): // if restored copy has expired, delete it synchronously
|
case evt.Action.DeleteRestored(): // if restored copy has expired, delete it synchronously
|
||||||
applyExpiryOnTransitionedObject(ctx, z, objInfo, evt, lcEventSrc_Decom)
|
if !applyExpiryOnTransitionedObject(ctx, z, objInfo, evt, lcEventSrc_Decom) {
|
||||||
return false
|
return evt.Action
|
||||||
|
}
|
||||||
|
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():
|
case evt.Action.Delete():
|
||||||
globalExpiryState.enqueueByDays(objInfo, evt, lcEventSrc_Decom)
|
return evt.Action
|
||||||
return true
|
|
||||||
default:
|
default:
|
||||||
return false
|
return lifecycle.NoneAction
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -833,11 +843,24 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
|
|||||||
versionsSorter(fivs.Versions).reverse()
|
versionsSorter(fivs.Versions).reverse()
|
||||||
|
|
||||||
var decommissioned int
|
var decommissioned int
|
||||||
|
var expired []migrationExpiredVersion
|
||||||
for _, version := range fivs.Versions {
|
for _, version := range fivs.Versions {
|
||||||
stopFn := globalDecommissionMetrics.log(decomMetricDecommissionObject, idx, bi.Name, version.Name, version.VersionID)
|
stopFn := globalDecommissionMetrics.log(decomMetricDecommissionObject, idx, bi.Name, version.Name, version.VersionID)
|
||||||
// Apply lifecycle rules on the objects that are expired.
|
// 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++
|
decommissioned++
|
||||||
|
if ilmDeletesVersionOnly(action) || (action.DeleteAll() && !version.IsRemote()) {
|
||||||
|
// Run the logical expiry before retiring the source.
|
||||||
|
// A full-object expiry is safe here only for hot-tier
|
||||||
|
// metadata; transitioned objects retain their tier cleanup
|
||||||
|
// ownership in the ordinary expiry worker.
|
||||||
|
expired = append(expired, migrationExpiredVersion{
|
||||||
|
version: version,
|
||||||
|
event: lifecycle.Event{Action: action},
|
||||||
|
lifecycleUpdatedAt: lifecycleUpdatedAt,
|
||||||
|
source: lcEventSrc_Decom,
|
||||||
|
})
|
||||||
|
}
|
||||||
stopFn(version.Size, errors.New("ILM expired object/version will be skipped"))
|
stopFn(version.Size, errors.New("ILM expired object/version will be skipped"))
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
@@ -981,7 +1004,10 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
|
|||||||
// if all versions were decommissioned, then we can delete the object versions.
|
// if all versions were decommissioned, then we can delete the object versions.
|
||||||
if decommissioned == len(fivs.Versions) {
|
if decommissioned == len(fivs.Versions) {
|
||||||
stopFn := globalDecommissionMetrics.log(decomMetricDecommissionRemoveObject, idx, bi.Name, entry.name)
|
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)
|
||||||
|
}
|
||||||
stopFn(0, err)
|
stopFn(0, err)
|
||||||
auditLogDecom(ctx, "DecomDeleteObject", bi.Name, entry.name, "", err)
|
auditLogDecom(ctx, "DecomDeleteObject", bi.Name, entry.name, "", err)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -1097,6 +1123,15 @@ func (z *erasureServerPools) decommissionInBackground(ctx context.Context, idx i
|
|||||||
stopFn(0, err)
|
stopFn(0, err)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
// Do not retire the bucket from the resumable queue until every
|
||||||
|
// source version and free version is gone. In particular, an ILM
|
||||||
|
// action queued during the walk may still be pending; marking the
|
||||||
|
// bucket complete here would allow the source pool to be removed
|
||||||
|
// before that action performs its logical and tier cleanup.
|
||||||
|
if err := z.checkAfterDecomBucket(ctx, idx, bucket); err != nil {
|
||||||
|
stopFn(0, err)
|
||||||
|
return err
|
||||||
|
}
|
||||||
stopFn(0, nil)
|
stopFn(0, nil)
|
||||||
|
|
||||||
z.poolMetaMutex.Lock()
|
z.poolMetaMutex.Lock()
|
||||||
@@ -1108,101 +1143,85 @@ func (z *erasureServerPools) decommissionInBackground(ctx context.Context, idx i
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// removeFreeVersions removes the free versions fvs of object from the pool
|
||||||
|
// idx being decommissioned, as the scanner removes free versions: first the
|
||||||
|
// remote object of each, then the free version. A free version is not moved.
|
||||||
|
// In another pool it would be a delete marker on top of the object, and once
|
||||||
|
// this pool is gone nothing would remove its remote object. A remote object
|
||||||
|
// that is already gone is not an error.
|
||||||
|
func (z *erasureServerPools) removeFreeVersions(ctx context.Context, bucket, object string, idx int, fvs []FileInfo) error {
|
||||||
|
if len(fvs) == 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
for _, fv := range fvs {
|
||||||
|
err := deleteObjectFromRemoteTier(ctx, fv.TransitionedObjName, fv.TransitionVersionID, fv.TransitionTier)
|
||||||
|
if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
|
||||||
|
return fmt.Errorf("unable to remove the remote object %s of a free version of %s/%s from tier %s: %w",
|
||||||
|
fv.TransitionedObjName, bucket, object, fv.TransitionTier, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
object = encodeDirObject(object)
|
||||||
|
lk := z.NewNSLock(bucket, object)
|
||||||
|
lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
ctx = lkctx.Context()
|
||||||
|
defer lk.Unlock(lkctx)
|
||||||
|
for _, fv := range fvs {
|
||||||
|
_, err := z.serverPools[idx].DeleteObject(ctx, bucket, object, ObjectOptions{
|
||||||
|
VersionID: fv.VersionID,
|
||||||
|
InclFreeVersions: true,
|
||||||
|
NoLock: true,
|
||||||
|
NoAuditLog: true,
|
||||||
|
})
|
||||||
|
if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// checkAfterDecomBucket verifies that a source bucket is physically empty
|
||||||
|
// before it is removed from the resumable decommission queue. ILM decisions
|
||||||
|
// are deliberately not treated as an exemption: an expired version remains
|
||||||
|
// until the expiry path has performed its logical delete, replication, and
|
||||||
|
// tier cleanup.
|
||||||
|
func (z *erasureServerPools) checkAfterDecomBucket(ctx context.Context, idx int, bi decomBucketInfo) error {
|
||||||
|
versionsFound := 0
|
||||||
|
for _, set := range z.serverPools[idx].sets {
|
||||||
|
if err := set.listObjectsToDecommission(ctx, bi, func(entry metaCacheEntry) {
|
||||||
|
if !entry.isObject() {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if bi.Name == minioMetaBucket && strings.Contains(entry.name, dataUsageCacheName) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
fivs, err := entry.fileInfoVersions(bi.Name)
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
versionsFound += len(fivs.Versions) + len(fivs.FreeVersions)
|
||||||
|
}); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if versionsFound > 0 {
|
||||||
|
return fmt.Errorf("at least %d object(s)/version(s) were found in bucket `%s` after decommissioning", versionsFound, bi.Name)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
func (z *erasureServerPools) checkAfterDecom(ctx context.Context, idx int) error {
|
func (z *erasureServerPools) checkAfterDecom(ctx context.Context, idx int) error {
|
||||||
buckets, err := z.getBucketsToDecommission(ctx)
|
buckets, err := z.getBucketsToDecommission(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
pool := z.serverPools[idx]
|
|
||||||
for _, set := range pool.sets {
|
|
||||||
for _, bi := range buckets {
|
for _, bi := range buckets {
|
||||||
var vc *versioning.Versioning
|
if err := z.checkAfterDecomBucket(ctx, idx, bi); err != nil {
|
||||||
var lc *lifecycle.Lifecycle
|
|
||||||
var lr objectlock.Retention
|
|
||||||
var rcfg *replication.Config
|
|
||||||
if bi.Name != minioMetaBucket {
|
|
||||||
vc, err = globalBucketVersioningSys.Get(bi.Name)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Check if the current bucket has a configured lifecycle policy
|
|
||||||
lc, err = globalLifecycleSys.Get(bi.Name)
|
|
||||||
if err != nil && !errors.Is(err, BucketLifecycleNotFound{Bucket: bi.Name}) {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Check if bucket is object locked.
|
|
||||||
lr, err = globalBucketObjectLockSys.Get(bi.Name)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
rcfg, err = getReplicationConfig(ctx, bi.Name)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
filterLifecycle := func(bucket, object string, fi FileInfo) bool {
|
|
||||||
if lc == nil {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
versioned := vc != nil && vc.Versioned(object)
|
|
||||||
objInfo := fi.ToObjectInfo(bucket, object, versioned)
|
|
||||||
|
|
||||||
evt := evalActionFromLifecycle(ctx, *lc, lr, rcfg, objInfo)
|
|
||||||
switch {
|
|
||||||
case evt.Action.DeleteRestored(): // if restored copy has expired,delete it synchronously
|
|
||||||
applyExpiryOnTransitionedObject(ctx, z, objInfo, evt, lcEventSrc_Decom)
|
|
||||||
return false
|
|
||||||
case evt.Action.Delete():
|
|
||||||
globalExpiryState.enqueueByDays(objInfo, evt, lcEventSrc_Decom)
|
|
||||||
return true
|
|
||||||
default:
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
var versionsFound int
|
|
||||||
if err = set.listObjectsToDecommission(ctx, bi, func(entry metaCacheEntry) {
|
|
||||||
if !entry.isObject() {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
// `.usage-cache.bin` still exists, must be not readable ignore it.
|
|
||||||
if bi.Name == minioMetaBucket && strings.Contains(entry.name, dataUsageCacheName) {
|
|
||||||
// skipping bucket usage cache name, as its autogenerated.
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
fivs, err := entry.fileInfoVersions(bi.Name)
|
|
||||||
if err != nil {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
var ignored int
|
|
||||||
|
|
||||||
for _, version := range fivs.Versions {
|
|
||||||
// Apply lifecycle rules on the objects that are expired.
|
|
||||||
if filterLifecycle(bi.Name, version.Name, version) {
|
|
||||||
ignored++
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
versionsFound += len(fivs.Versions) - ignored
|
|
||||||
}); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
if versionsFound > 0 {
|
|
||||||
return fmt.Errorf("at least %d object(s)/version(s) were found in bucket `%s` after decommissioning", versionsFound, bi.Name)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,526 @@
|
|||||||
|
// 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 <http://www.gnu.org/licenses/>.
|
||||||
|
|
||||||
|
package cmd
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"slices"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
madmin "github.com/minio/madmin-go/v3"
|
||||||
|
"github.com/minio/minio/internal/bucket/lifecycle"
|
||||||
|
objectlock "github.com/minio/minio/internal/bucket/object/lock"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Deleting a version whose data is in a remote tier leaves a free version in
|
||||||
|
// its place: a record of the remote object, kept until the scanner removes
|
||||||
|
// both. A free version is not a version of the object. There is no delete
|
||||||
|
// marker, and the version below it is the latest one; reads already skip it.
|
||||||
|
|
||||||
|
// freeVersionNoncurrent expires versions one day after they become noncurrent.
|
||||||
|
const freeVersionNoncurrent = "<Rule><ID>noncurrent</ID><Status>Enabled</Status><Filter><Prefix></Prefix></Filter>" +
|
||||||
|
"<NoncurrentVersionExpiration><NoncurrentDays>1</NoncurrentDays></NoncurrentVersionExpiration></Rule>"
|
||||||
|
|
||||||
|
// freeVersionTier is a remote tier held in memory whose removals can fail.
|
||||||
|
type freeVersionTier struct {
|
||||||
|
*tierExpiryTestBackend
|
||||||
|
removeErr error
|
||||||
|
}
|
||||||
|
|
||||||
|
func (b *freeVersionTier) Remove(ctx context.Context, object string, rv remoteVersionID) error {
|
||||||
|
if b.removeErr != nil {
|
||||||
|
return b.removeErr
|
||||||
|
}
|
||||||
|
return b.tierExpiryTestBackend.Remove(ctx, object, rv)
|
||||||
|
}
|
||||||
|
|
||||||
|
type freeVersionTest struct {
|
||||||
|
t *testing.T
|
||||||
|
z *erasureServerPools
|
||||||
|
bucket string
|
||||||
|
tier *freeVersionTier
|
||||||
|
labels map[string]string // version ID -> label
|
||||||
|
}
|
||||||
|
|
||||||
|
// newFreeVersionTest returns two pools with a versioned bucket under rules and
|
||||||
|
// a remote tier WARM. In pool 0, obj-01 has v1 and v2 and a tiered v3, written
|
||||||
|
// five, four and three days ago; the client deleted v3. obj-02 had only a
|
||||||
|
// tiered v1, deleted as well.
|
||||||
|
func newFreeVersionTest(t *testing.T, rules string) *freeVersionTest {
|
||||||
|
t.Helper()
|
||||||
|
z, _ := migrationTestPools(t)
|
||||||
|
ctx := t.Context()
|
||||||
|
f := &freeVersionTest{t: t, z: z, bucket: "free-version", labels: map[string]string{}}
|
||||||
|
if err := z.MakeBucket(ctx, f.bucket, MakeBucketOptions{VersioningEnabled: true}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if rules != "" {
|
||||||
|
meta, err := globalBucketMetadataSys.Get(f.bucket)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
meta.LifecycleConfigXML = []byte("<LifecycleConfiguration>" + rules + "</LifecycleConfiguration>")
|
||||||
|
if err := meta.Save(ctx, z); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
globalBucketMetadataSys.Set(f.bucket, meta)
|
||||||
|
}
|
||||||
|
f.tier = &freeVersionTier{tierExpiryTestBackend: &tierExpiryTestBackend{objects: map[string][]byte{}}}
|
||||||
|
prevTier := globalTierConfigMgr
|
||||||
|
globalTierConfigMgr = NewTierConfigMgr()
|
||||||
|
globalTierConfigMgr.drivercache["WARM"] = f.tier
|
||||||
|
t.Cleanup(func() { globalTierConfigMgr = prevTier })
|
||||||
|
|
||||||
|
now := UTCNow()
|
||||||
|
f.put("obj-01", "v1", now.AddDate(0, 0, -5))
|
||||||
|
f.put("obj-01", "v2", now.AddDate(0, 0, -4))
|
||||||
|
f.deleteVersion(f.tiered("obj-01", "v3", now.AddDate(0, 0, -3)))
|
||||||
|
f.deleteVersion(f.tiered("obj-02", "v1", now.AddDate(0, 0, -3)))
|
||||||
|
return f
|
||||||
|
}
|
||||||
|
|
||||||
|
// put writes a version of object to pool 0.
|
||||||
|
func (f *freeVersionTest) put(object, label string, mtime time.Time) ObjectInfo {
|
||||||
|
f.t.Helper()
|
||||||
|
oi := putConsistencyObject(f.t, f.z, f.bucket, object, 0, object+" "+label, ObjectOptions{Versioned: true, MTime: mtime})
|
||||||
|
f.labels[oi.VersionID] = label
|
||||||
|
return oi
|
||||||
|
}
|
||||||
|
|
||||||
|
// tiered writes a version of object to pool 0 whose data is in the tier.
|
||||||
|
func (f *freeVersionTest) tiered(object, label string, mtime time.Time) ObjectInfo {
|
||||||
|
f.t.Helper()
|
||||||
|
oi := f.put(object, label, mtime)
|
||||||
|
infos, errs := readAllFileInfo(f.t.Context(), f.z.serverPools[0].getHashedSet(object).getDisks(), "", f.bucket, object, oi.VersionID, false, false)
|
||||||
|
if errs[0] != nil {
|
||||||
|
f.t.Fatal(errs[0])
|
||||||
|
}
|
||||||
|
fi := infos[0]
|
||||||
|
fi.TransitionStatus, fi.TransitionTier, fi.TransitionedObjName = lifecycle.TransitionComplete, "WARM", f.remote(object, label)
|
||||||
|
f.tier.objects[fi.TransitionedObjName] = []byte(object + " " + label)
|
||||||
|
if err := f.z.serverPools[0].DecomTieredObject(f.t.Context(), f.bucket, object, fi, ObjectOptions{}); err != nil {
|
||||||
|
f.t.Fatal(err)
|
||||||
|
}
|
||||||
|
return oi
|
||||||
|
}
|
||||||
|
|
||||||
|
// remote is the name of the remote object of a tiered version.
|
||||||
|
func (f *freeVersionTest) remote(object, label string) string {
|
||||||
|
return f.bucket + "/" + object + "/" + label
|
||||||
|
}
|
||||||
|
|
||||||
|
// deleteVersion deletes a tiered version by its ID, as a client does, which
|
||||||
|
// leaves a free version in its place.
|
||||||
|
func (f *freeVersionTest) deleteVersion(oi ObjectInfo) {
|
||||||
|
f.t.Helper()
|
||||||
|
if _, err := f.z.DeleteObject(f.t.Context(), f.bucket, oi.Name, ObjectOptions{Versioned: true, VersionID: oi.VersionID}); err != nil {
|
||||||
|
f.t.Fatal(err)
|
||||||
|
}
|
||||||
|
if got := f.stored(0, oi.Name); len(got) == 0 || got[0] != "free" {
|
||||||
|
f.t.Fatalf("fixture: %s holds %v, want a free version on top", oi.Name, got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (f *freeVersionTest) label(fi FileInfo) string {
|
||||||
|
switch {
|
||||||
|
case fi.TierFreeVersion():
|
||||||
|
return "free"
|
||||||
|
case fi.Deleted:
|
||||||
|
return "DM"
|
||||||
|
case f.labels[fi.VersionID] != "":
|
||||||
|
return f.labels[fi.VersionID]
|
||||||
|
}
|
||||||
|
return "?"
|
||||||
|
}
|
||||||
|
|
||||||
|
// xlMeta reads the metadata of object in pool from its first drive, nil if
|
||||||
|
// the pool has none.
|
||||||
|
func (f *freeVersionTest) xlMeta(pool int, object string) []byte {
|
||||||
|
f.t.Helper()
|
||||||
|
disk := f.z.serverPools[pool].getHashedSet(object).getDisks()[0]
|
||||||
|
buf, err := disk.ReadAll(f.t.Context(), f.bucket, pathJoin(object, xlStorageFormatFile))
|
||||||
|
if errors.Is(err, errFileNotFound) {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
f.t.Fatal(err)
|
||||||
|
}
|
||||||
|
return buf
|
||||||
|
}
|
||||||
|
|
||||||
|
// stored returns what pool holds of object, newest first, free versions
|
||||||
|
// included.
|
||||||
|
func (f *freeVersionTest) stored(pool int, object string) []string {
|
||||||
|
f.t.Helper()
|
||||||
|
buf := f.xlMeta(pool, object)
|
||||||
|
if buf == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
fivs, err := getFileInfoVersions(buf, f.bucket, object, true)
|
||||||
|
if err != nil {
|
||||||
|
f.t.Fatal(err)
|
||||||
|
}
|
||||||
|
var got []string
|
||||||
|
for _, fi := range fivs.Versions {
|
||||||
|
got = append(got, f.label(fi))
|
||||||
|
}
|
||||||
|
return got
|
||||||
|
}
|
||||||
|
|
||||||
|
// versions labels the versions of a listing as "<object> <label>[ latest]".
|
||||||
|
func (f *freeVersionTest) versions(objects []ObjectInfo) []string {
|
||||||
|
var got []string
|
||||||
|
for _, oi := range objects {
|
||||||
|
label := oi.Name + " " + f.label(FileInfo{VersionID: oi.VersionID, Deleted: oi.DeleteMarker})
|
||||||
|
if oi.IsLatest {
|
||||||
|
label += " latest"
|
||||||
|
}
|
||||||
|
got = append(got, label)
|
||||||
|
}
|
||||||
|
return got
|
||||||
|
}
|
||||||
|
|
||||||
|
// listVersions returns the versions ListObjectVersions lists in one page.
|
||||||
|
func (f *freeVersionTest) listVersions() []string {
|
||||||
|
f.t.Helper()
|
||||||
|
lv, err := f.z.ListObjectVersions(f.t.Context(), f.bucket, "", "", "", "", 100)
|
||||||
|
if err != nil {
|
||||||
|
f.t.Fatal(err)
|
||||||
|
}
|
||||||
|
return f.versions(lv.Objects)
|
||||||
|
}
|
||||||
|
|
||||||
|
// expiryQueue replaces the expiry workers with a queue for the test.
|
||||||
|
func (f *freeVersionTest) expiryQueue() chan expiryOp {
|
||||||
|
queue := make(chan expiryOp, 100)
|
||||||
|
es := &expiryState{ctx: f.t.Context(), objAPI: f.z}
|
||||||
|
workers := []chan expiryOp{queue}
|
||||||
|
es.workers.Store(&workers)
|
||||||
|
prev := globalExpiryState
|
||||||
|
globalExpiryState = es
|
||||||
|
f.t.Cleanup(func() { globalExpiryState = prev })
|
||||||
|
return queue
|
||||||
|
}
|
||||||
|
|
||||||
|
// expired returns the labels of the versions queued for expiry.
|
||||||
|
func (f *freeVersionTest) expired(queue chan expiryOp) []string {
|
||||||
|
var got []string
|
||||||
|
for len(queue) > 0 {
|
||||||
|
switch task := (<-queue).(type) {
|
||||||
|
case expiryTask:
|
||||||
|
got = append(got, f.labels[task.objInfo.VersionID])
|
||||||
|
case noncurrentVersionsTask:
|
||||||
|
for _, v := range task.versions {
|
||||||
|
got = append(got, f.labels[v.VersionID])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return got
|
||||||
|
}
|
||||||
|
|
||||||
|
// The scanner reads the versions of obj-01 without the free version. v2 is the
|
||||||
|
// latest version, and a noncurrent expiry must not count it as noncurrent
|
||||||
|
// since the deleted v3 was written. Otherwise the scanner deletes it.
|
||||||
|
func TestFreeVersionScannerLatestVersion(t *testing.T) {
|
||||||
|
f := newFreeVersionTest(t, freeVersionNoncurrent)
|
||||||
|
fivs, err := getFileInfoVersions(f.xlMeta(0, "obj-01"), f.bucket, "obj-01", false)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
var got []string
|
||||||
|
for _, fi := range fivs.Versions {
|
||||||
|
got = append(got, f.label(fi))
|
||||||
|
}
|
||||||
|
if want := []string{"v2", "v1"}; !slices.Equal(got, want) || len(fivs.FreeVersions) != 1 {
|
||||||
|
t.Fatalf("scanner versions %v with %d free versions, want %v with 1", got, len(fivs.FreeVersions), want)
|
||||||
|
}
|
||||||
|
v2, v1 := fivs.Versions[0], fivs.Versions[1]
|
||||||
|
if !v2.IsLatest || !v2.SuccessorModTime.IsZero() {
|
||||||
|
t.Errorf("v2: latest %v, successor mod time %v; want latest without a successor", v2.IsLatest, v2.SuccessorModTime)
|
||||||
|
}
|
||||||
|
if v1.IsLatest || !v1.SuccessorModTime.Equal(v2.ModTime) {
|
||||||
|
t.Errorf("v1: latest %v, successor mod time %v; want noncurrent since v2 at %v", v1.IsLatest, v1.SuccessorModTime, v2.ModTime)
|
||||||
|
}
|
||||||
|
if !fivs.LatestModTime.Equal(v2.ModTime) {
|
||||||
|
t.Errorf("latest mod time %v, want that of v2 %v", fivs.LatestModTime, v2.ModTime)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Apply the rules to these versions as the scanner does.
|
||||||
|
lc, err := globalLifecycleSys.Get(f.bucket)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
objInfos := make([]ObjectInfo, len(fivs.Versions))
|
||||||
|
for i, fi := range fivs.Versions {
|
||||||
|
objInfos[i] = fi.ToObjectInfo(f.bucket, "obj-01", true)
|
||||||
|
}
|
||||||
|
queue := f.expiryQueue()
|
||||||
|
item := scannerItem{Path: "obj-01", bucket: f.bucket, objectName: "obj-01", lifeCycle: lc}
|
||||||
|
item.applyActions(t.Context(), f.z, objInfos, objectlock.Retention{}, &sizeSummary{}, nil)
|
||||||
|
if expired, want := f.expired(queue), []string{"v1"}; !slices.Equal(expired, want) {
|
||||||
|
t.Errorf("scanner queued %v for expiry, want %v", expired, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ListObjectsV2 lists obj-01: its latest version is v2, not a delete marker.
|
||||||
|
// obj-02 has no versions left.
|
||||||
|
func TestFreeVersionListObjects(t *testing.T) {
|
||||||
|
f := newFreeVersionTest(t, "")
|
||||||
|
lo, err := f.z.ListObjectsV2(t.Context(), f.bucket, "", "", "", 100, false, "")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
var got []string
|
||||||
|
for _, oi := range lo.Objects {
|
||||||
|
got = append(got, oi.Name)
|
||||||
|
}
|
||||||
|
if want := []string{"obj-01"}; !slices.Equal(got, want) {
|
||||||
|
t.Errorf("ListObjectsV2: %v, want %v", got, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The listing checks of the latest version skip the free version: obj-01 is
|
||||||
|
// shown, obj-02 has nothing to show. Both the encoded metadata and the
|
||||||
|
// decoded form, which the listing caches, are checked.
|
||||||
|
func TestFreeVersionLatestDeleteMarker(t *testing.T) {
|
||||||
|
f := newFreeVersionTest(t, "")
|
||||||
|
for _, tc := range []struct {
|
||||||
|
object string
|
||||||
|
hidden bool
|
||||||
|
}{{"obj-01", false}, {"obj-02", true}} {
|
||||||
|
entry := metaCacheEntry{name: tc.object, metadata: f.xlMeta(0, tc.object)}
|
||||||
|
meta, _, err := isIndexedMetaV2(entry.metadata)
|
||||||
|
if err != nil || meta == nil {
|
||||||
|
t.Fatalf("%s: metadata is not indexed: %v", tc.object, err)
|
||||||
|
}
|
||||||
|
if got := meta.IsLatestDeleteMarker(); got != tc.hidden {
|
||||||
|
t.Errorf("%s: IsLatestDeleteMarker %v, want %v", tc.object, got, tc.hidden)
|
||||||
|
}
|
||||||
|
if got := meta.AllHidden(true); got != tc.hidden {
|
||||||
|
t.Errorf("%s: AllHidden(true) %v, want %v", tc.object, got, tc.hidden)
|
||||||
|
}
|
||||||
|
if got := meta.AllHidden(false); got != tc.hidden {
|
||||||
|
t.Errorf("%s: AllHidden(false) %v, want %v", tc.object, got, tc.hidden)
|
||||||
|
}
|
||||||
|
if got := entry.isLatestDeletemarker(); got != tc.hidden {
|
||||||
|
t.Errorf("%s: isLatestDeletemarker %v, want %v", tc.object, got, tc.hidden)
|
||||||
|
}
|
||||||
|
if _, err := entry.xlmeta(); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if got := entry.isLatestDeletemarker(); got != tc.hidden {
|
||||||
|
t.Errorf("%s: isLatestDeletemarker of the decoded metadata %v, want %v", tc.object, got, tc.hidden)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ListObjectVersions lists v2 as the latest version of obj-01 and no delete
|
||||||
|
// markers. Keys with nothing but a free version have no versions to list and
|
||||||
|
// do not take up a page. Walk returns the same versions.
|
||||||
|
func TestFreeVersionListObjectVersions(t *testing.T) {
|
||||||
|
f := newFreeVersionTest(t, "")
|
||||||
|
f.deleteVersion(f.tiered("obj-00", "v1", UTCNow().AddDate(0, 0, -3)))
|
||||||
|
want := []string{"obj-01 v2 latest", "obj-01 v1"}
|
||||||
|
if got := f.listVersions(); !slices.Equal(got, want) {
|
||||||
|
t.Errorf("ListObjectVersions: %v, want %v", got, want)
|
||||||
|
}
|
||||||
|
|
||||||
|
results := make(chan itemOrErr[ObjectInfo], 100)
|
||||||
|
if err := f.z.Walk(t.Context(), f.bucket, "", results, WalkOptions{}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
var walked []ObjectInfo
|
||||||
|
for r := range results {
|
||||||
|
if r.Err != nil {
|
||||||
|
t.Fatal(r.Err)
|
||||||
|
}
|
||||||
|
walked = append(walked, r.Item)
|
||||||
|
}
|
||||||
|
if got, want := f.versions(walked), []string{"obj-01 v1", "obj-01 v2 latest"}; !slices.Equal(got, want) {
|
||||||
|
t.Errorf("Walk: %v, want %v", got, want)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Page by page, one version a page.
|
||||||
|
for _, delimiter := range []string{"", "/"} {
|
||||||
|
var got []string
|
||||||
|
marker, versionMarker := "", ""
|
||||||
|
for range 10 {
|
||||||
|
lv, err := f.z.ListObjectVersions(t.Context(), f.bucket, "", marker, versionMarker, delimiter, 1)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
got = append(got, f.versions(lv.Objects)...)
|
||||||
|
if !lv.IsTruncated {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
marker, versionMarker = lv.NextMarker, lv.NextVersionIDMarker
|
||||||
|
}
|
||||||
|
if !slices.Equal(got, want) {
|
||||||
|
t.Errorf("ListObjectVersions page by page, delimiter %q: %v, want %v", delimiter, got, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Healing still visits the free versions, so a free version lost on some
|
||||||
|
// drives is healed and the scanner still finds it.
|
||||||
|
func TestFreeVersionHealObjects(t *testing.T) {
|
||||||
|
f := newFreeVersionTest(t, "")
|
||||||
|
var got []string
|
||||||
|
err := f.z.HealObjects(t.Context(), f.bucket, "", madmin.HealOpts{Recursive: true}, func(_, object, versionID string, _ madmin.HealScanMode) error {
|
||||||
|
label := f.labels[versionID]
|
||||||
|
if label == "" {
|
||||||
|
label = "free"
|
||||||
|
}
|
||||||
|
got = append(got, object+" "+label)
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
slices.Sort(got)
|
||||||
|
if want := []string{"obj-01 free", "obj-01 v1", "obj-01 v2", "obj-02 free"}; !slices.Equal(got, want) {
|
||||||
|
t.Errorf("healed versions %v, want %v", got, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Decommission and rebalance do not move a free version: in another pool it
|
||||||
|
// would be a delete marker that hides v2. Decommission removes the remote
|
||||||
|
// object and the free version, as the scanner would before the pool is
|
||||||
|
// removed. Rebalance leaves both to the scanner of the source pool.
|
||||||
|
func TestFreeVersionDataMovement(t *testing.T) {
|
||||||
|
for _, kind := range []string{"decommission", "rebalance"} {
|
||||||
|
for _, rules := range []string{"", freeVersionNoncurrent} {
|
||||||
|
name := kind
|
||||||
|
if rules != "" {
|
||||||
|
name += "/noncurrent-expiry"
|
||||||
|
}
|
||||||
|
t.Run(name, func(t *testing.T) {
|
||||||
|
f := newFreeVersionTest(t, rules)
|
||||||
|
queue := f.expiryQueue()
|
||||||
|
if failed := migrationTestWalk(t, f.z, f.bucket, 0, kind); failed != 0 {
|
||||||
|
t.Fatalf("%d versions failed to move", failed)
|
||||||
|
}
|
||||||
|
if expired := f.expired(queue); slices.Contains(expired, "v2") {
|
||||||
|
t.Errorf("%s queued the latest version v2 for expiry: %v", kind, expired)
|
||||||
|
}
|
||||||
|
|
||||||
|
// v1 has been noncurrent for days; whether a move skips an
|
||||||
|
// expired version is out of scope here.
|
||||||
|
if got := f.stored(1, "obj-01"); len(got) == 0 || got[0] != "v2" || slices.Contains(got, "DM") {
|
||||||
|
t.Errorf("pool 1 holds %v of obj-01, want v2 on top and no delete marker", got)
|
||||||
|
}
|
||||||
|
if got := f.stored(1, "obj-02"); len(got) != 0 {
|
||||||
|
t.Errorf("pool 1 holds %v of obj-02, want nothing", got)
|
||||||
|
}
|
||||||
|
if body := migrationTestBody(t, f.z.GetObjectNInfo, f.bucket, "obj-01", ""); body != "obj-01 v2" {
|
||||||
|
t.Errorf("obj-01 reads %q, want %q", body, "obj-01 v2")
|
||||||
|
}
|
||||||
|
if got := f.listVersions(); len(got) == 0 || got[0] != "obj-01 v2 latest" || slices.Contains(got, "obj-02 DM latest") {
|
||||||
|
t.Errorf("ListObjectVersions after %s: %v", kind, got)
|
||||||
|
}
|
||||||
|
|
||||||
|
if rules != "" {
|
||||||
|
// The expired v1 is not moved, so its source is kept.
|
||||||
|
return
|
||||||
|
}
|
||||||
|
var wantSource []string // decommission
|
||||||
|
removed := true
|
||||||
|
if kind == "rebalance" {
|
||||||
|
wantSource, removed = []string{"free"}, false
|
||||||
|
}
|
||||||
|
for _, object := range []string{"obj-01", "obj-02"} {
|
||||||
|
if got := f.stored(0, object); !slices.Equal(got, wantSource) {
|
||||||
|
t.Errorf("pool 0 holds %v of %s, want %v", got, object, wantSource)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for _, remote := range []string{f.remote("obj-01", "v3"), f.remote("obj-02", "v1")} {
|
||||||
|
if slices.Contains(f.tier.removed, remote) != removed {
|
||||||
|
t.Errorf("remote object %s of a deleted version removed: %v, want %v", remote, !removed, removed)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// decommission decommissions pool 0 as the admin API does, including the check
|
||||||
|
// for versions left behind, and returns its state.
|
||||||
|
func (f *freeVersionTest) decommission() PoolDecommissionInfo {
|
||||||
|
f.t.Helper()
|
||||||
|
ctx := f.t.Context()
|
||||||
|
if err := f.z.StartDecommission(ctx, 0); err != nil {
|
||||||
|
f.t.Fatal(err)
|
||||||
|
}
|
||||||
|
f.z.doDecommissionInRoutine(ctx, 0)
|
||||||
|
f.z.poolMetaMutex.RLock()
|
||||||
|
defer f.z.poolMetaMutex.RUnlock()
|
||||||
|
return *f.z.poolMeta.Pools[0].Decommission
|
||||||
|
}
|
||||||
|
|
||||||
|
// While the tier is unavailable, the decommission moves the versions but keeps
|
||||||
|
// the free versions, and the check for versions left behind fails it. A retry
|
||||||
|
// with the tier back completes it.
|
||||||
|
func TestFreeVersionDecommissionTierUnavailable(t *testing.T) {
|
||||||
|
f := newFreeVersionTest(t, "")
|
||||||
|
f.tier.removeErr = errors.New("tier unavailable")
|
||||||
|
if info := f.decommission(); info.Complete || !info.Failed {
|
||||||
|
t.Fatalf("decommission with the tier unavailable: complete %v, failed %v; want failed", info.Complete, info.Failed)
|
||||||
|
}
|
||||||
|
if got, want := f.stored(1, "obj-01"), []string{"v2", "v1"}; !slices.Equal(got, want) {
|
||||||
|
t.Errorf("pool 1 holds %v of obj-01, want %v", got, want)
|
||||||
|
}
|
||||||
|
for _, object := range []string{"obj-01", "obj-02"} {
|
||||||
|
if got, want := f.stored(0, object), []string{"free"}; !slices.Equal(got, want) {
|
||||||
|
t.Errorf("pool 0 holds %v of %s, want %v", got, object, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if body := migrationTestBody(t, f.z.GetObjectNInfo, f.bucket, "obj-01", ""); body != "obj-01 v2" {
|
||||||
|
t.Errorf("obj-01 reads %q, want %q", body, "obj-01 v2")
|
||||||
|
}
|
||||||
|
if len(f.tier.removed) != 0 {
|
||||||
|
t.Errorf("remote objects removed: %v", f.tier.removed)
|
||||||
|
}
|
||||||
|
|
||||||
|
// The failed decommission moved the bucket metadata, so pool 0 has no
|
||||||
|
// .minio.sys/buckets left. On a server the scanner recreates it when it
|
||||||
|
// saves its usage cache there; it does not run here.
|
||||||
|
for _, set := range f.z.serverPools[0].sets {
|
||||||
|
for _, disk := range set.getDisks() {
|
||||||
|
if err := disk.MakeVol(t.Context(), pathJoin(minioMetaBucket, bucketMetaPrefix)); err != nil && !errors.Is(err, errVolumeExists) {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
f.tier.removeErr = nil
|
||||||
|
if info := f.decommission(); !info.Complete || info.Failed {
|
||||||
|
t.Fatalf("retry: complete %v, failed %v; want complete", info.Complete, info.Failed)
|
||||||
|
}
|
||||||
|
if got, want := f.stored(1, "obj-01"), []string{"v2", "v1"}; !slices.Equal(got, want) {
|
||||||
|
t.Errorf("pool 1 holds %v of obj-01 after the retry, want %v", got, want)
|
||||||
|
}
|
||||||
|
for _, object := range []string{"obj-01", "obj-02"} {
|
||||||
|
if got := f.stored(0, object); len(got) != 0 {
|
||||||
|
t.Errorf("pool 0 holds %v of %s after the retry", got, object)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if want := []string{f.remote("obj-01", "v3"), f.remote("obj-02", "v1")}; !slices.Equal(slices.Sorted(slices.Values(f.tier.removed)), want) {
|
||||||
|
t.Errorf("remote objects removed: %v, want %v", f.tier.removed, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -132,7 +132,7 @@ func TestPoolsMigrationCleanupRechecksSource(t *testing.T) {
|
|||||||
_, err := z.serverPools[1].DeleteObject(t.Context(), bucket, oi.Name, ObjectOptions{VersionID: nullVersionID})
|
_, err := z.serverPools[1].DeleteObject(t.Context(), bucket, oi.Name, ObjectOptions{VersionID: nullVersionID})
|
||||||
mustIAM(t, err)
|
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 {
|
if change != "delete-uuid" && err == nil {
|
||||||
t.Fatal("cleanup accepted changed or uncertain state")
|
t.Fatal("cleanup accepted changed or uncertain state")
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,588 @@
|
|||||||
|
// 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 <http://www.gnu.org/licenses/>.
|
||||||
|
|
||||||
|
package cmd
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"net/http"
|
||||||
|
"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/replication"
|
||||||
|
"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("<VersioningConfiguration><Status>Suspended</Status></VersioningConfiguration>")
|
||||||
|
}
|
||||||
|
if rules != "" {
|
||||||
|
meta.LifecycleConfigXML = []byte("<LifecycleConfiguration>" + rules + "</LifecycleConfiguration>")
|
||||||
|
}
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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 "<Rule><ID>rule</ID><Status>Enabled</Status><Filter><Prefix>obj-</Prefix></Filter>" + action + "</Rule>"
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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("<Expiration><Date>" + date + "</Date></Expiration>")
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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("<NoncurrentVersionExpiration><NoncurrentDays>1</NoncurrentDays></NoncurrentVersionExpiration>"))
|
||||||
|
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("<Expiration><ExpiredObjectDeleteMarker>true</ExpiredObjectDeleteMarker></Expiration>"))
|
||||||
|
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"}})
|
||||||
|
})
|
||||||
|
|
||||||
|
// Each rule deletes every version of one object. Migration must execute the
|
||||||
|
// logical DeleteAll action before it retires the source pool, so the object
|
||||||
|
// selected by the rule is gone from both pools before any later scan.
|
||||||
|
for _, tc := range []struct {
|
||||||
|
name, rule, deleted string
|
||||||
|
}{
|
||||||
|
{"expire-all-versions", "<Expiration><Days>1</Days><ExpiredObjectAllVersions>true</ExpiredObjectAllVersions></Expiration>", "obj-01"},
|
||||||
|
{"expire-marker-all-versions", "<DelMarkerExpiration><Days>1</Days></DelMarkerExpiration>", "obj-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 len(m.queue) != 0 {
|
||||||
|
t.Fatalf("migration left stale expiry tasks queued: %d", len(m.queue))
|
||||||
|
}
|
||||||
|
want := map[string][]string{
|
||||||
|
"obj-01": {"v3", "v2", "v1"},
|
||||||
|
"keep-01": {"v3", "v2", "v1"},
|
||||||
|
"obj-dm": {"dm", "v1"},
|
||||||
|
}
|
||||||
|
delete(want, tc.deleted)
|
||||||
|
m.check(want)
|
||||||
|
m.catchUp()
|
||||||
|
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"}})
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// A lifecycle action found during the migration walk must not be applied
|
||||||
|
// after the bucket's lifecycle configuration changes. The source copy stays
|
||||||
|
// available for a fresh scan under the new policy.
|
||||||
|
func TestPoolsMigrationLifecycleChangePreservesSource(t *testing.T) {
|
||||||
|
m := newMigrationILM(t, versioning.Enabled, false, migrationILMExpireCurrent())
|
||||||
|
oi := m.put("obj-01", "v1", ObjectOptions{Versioned: true})
|
||||||
|
_, updatedAt, err := globalBucketMetadataSys.GetLifecycleConfig(m.bucket)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
meta, err := globalBucketMetadataSys.Get(m.bucket)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
meta.LifecycleConfigXML = []byte(`<LifecycleConfiguration><Rule><ID>changed</ID><Status>Enabled</Status><Filter><Prefix>other-</Prefix></Filter><Expiration><Days>1</Days></Expiration></Rule></LifecycleConfiguration>`)
|
||||||
|
meta.LifecycleConfigUpdatedAt = updatedAt.Add(time.Second)
|
||||||
|
if err := meta.Save(t.Context(), m.z); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
globalBucketMetadataSys.Set(m.bucket, meta)
|
||||||
|
|
||||||
|
err = m.z.applyMigrationExpiry(t.Context(), m.bucket, oi.Name, 0, []migrationExpiredVersion{{
|
||||||
|
version: FileInfo{
|
||||||
|
Name: oi.Name,
|
||||||
|
VersionID: oi.VersionID,
|
||||||
|
ModTime: oi.ModTime,
|
||||||
|
},
|
||||||
|
event: lifecycle.Event{Action: lifecycle.DeleteAction},
|
||||||
|
lifecycleUpdatedAt: updatedAt,
|
||||||
|
source: lcEventSrc_Decom,
|
||||||
|
}})
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("migration expiry succeeded after lifecycle configuration changed")
|
||||||
|
}
|
||||||
|
if _, err := m.z.serverPools[0].GetObjectInfo(t.Context(), m.bucket, oi.Name, ObjectOptions{VersionID: oi.VersionID}); err != nil {
|
||||||
|
t.Fatalf("source version was removed after lifecycle configuration changed: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// A migration must route an ILM version purge through the normal delete
|
||||||
|
// replication path before it can retire the source copy. Otherwise the
|
||||||
|
// source pool can disappear while the replica still retains the version.
|
||||||
|
func TestPoolsMigrationLifecycleExpiryReplication(t *testing.T) {
|
||||||
|
m := newMigrationILM(t, versioning.Enabled, false,
|
||||||
|
migrationILMRule("<NoncurrentVersionExpiration><NoncurrentDays>1</NoncurrentDays></NoncurrentVersionExpiration>"))
|
||||||
|
m.aged(0, "obj-01")
|
||||||
|
|
||||||
|
ctx, cancel := context.WithCancel(t.Context())
|
||||||
|
defer cancel()
|
||||||
|
stats := NewReplicationStats(ctx, nil)
|
||||||
|
oldStats := globalReplicationStats.Swap(stats)
|
||||||
|
defer globalReplicationStats.Store(oldStats)
|
||||||
|
|
||||||
|
target := newILMTestTarget(t)
|
||||||
|
registerILMTestTarget(t, m.z, m.bucket, target, func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
opts := ObjectOptions{VersionID: r.URL.Query().Get("versionId"), Versioned: true}
|
||||||
|
switch r.Method {
|
||||||
|
case http.MethodHead:
|
||||||
|
oi, err := m.z.GetObjectInfo(r.Context(), target.bucket, "obj-01", opts)
|
||||||
|
if err != nil {
|
||||||
|
writeErrorResponseHeadersOnly(w, toAPIError(r.Context(), err))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if oi.DeleteMarker {
|
||||||
|
w.Header().Set(xhttp.AmzDeleteMarker, "true")
|
||||||
|
w.Header().Set(xhttp.AmzVersionID, oi.VersionID)
|
||||||
|
}
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
case http.MethodDelete:
|
||||||
|
target.deletes.Add(1)
|
||||||
|
opts.DeleteMarker = r.Header.Get(xhttp.MinIOSourceDeleteMarker) == "true"
|
||||||
|
opts.ReplicationRequest = true
|
||||||
|
opts.SetReplicaStatus(replication.Replica)
|
||||||
|
if _, err := m.z.DeleteObject(r.Context(), target.bucket, "obj-01", opts); err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
|
||||||
|
writeErrorResponse(r.Context(), w, toAPIError(r.Context(), err), r.URL)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
w.WriteHeader(http.StatusNoContent)
|
||||||
|
default:
|
||||||
|
t.Errorf("unexpected remote method %s", r.Method)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
setILMTestReplicationConfig(t, m.bucket, target.arn, target.bucket)
|
||||||
|
restorePool, receive := newILMTestReplicationPool(ctx, m.z, stats)
|
||||||
|
defer restorePool()
|
||||||
|
|
||||||
|
if err := m.z.StartDecommission(ctx, 0); err != nil {
|
||||||
|
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 {
|
||||||
|
t.Fatalf("decommission completed before ILM purge replication: complete=%t failed=%t", info.Complete, info.Failed)
|
||||||
|
}
|
||||||
|
|
||||||
|
deletion := receive()
|
||||||
|
if deletion.ObjectName != "obj-01" || deletion.VersionID == "" || deletion.VersionPurgeStatus() != replication.VersionPurgePending {
|
||||||
|
t.Fatalf("unexpected migration purge: %+v", deletion)
|
||||||
|
}
|
||||||
|
if result := replicateDelete(ctx, deletion, m.z); result.VersionPurgeStatus() != replication.VersionPurgeComplete {
|
||||||
|
t.Fatalf("migration purge replication result: %+v", result)
|
||||||
|
}
|
||||||
|
if target.deletes.Load() != 1 {
|
||||||
|
t.Fatalf("replication target saw %d deletes, want 1", target.deletes.Load())
|
||||||
|
}
|
||||||
|
|
||||||
|
// A failed decommission may already have moved the source pool's bucket
|
||||||
|
// metadata. Recreate the metadata directory before the resumable retry,
|
||||||
|
// as the server's scanner would do in production.
|
||||||
|
for _, set := range m.z.serverPools[0].sets {
|
||||||
|
for _, disk := range set.getDisks() {
|
||||||
|
if err := disk.MakeVol(ctx, pathJoin(minioMetaBucket, bucketMetaPrefix)); err != nil && !errors.Is(err, errVolumeExists) {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := m.z.StartDecommission(ctx, 0); err != nil {
|
||||||
|
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 {
|
||||||
|
t.Fatalf("decommission did not complete after ILM purge replication: complete=%t failed=%t", info.Complete, info.Failed)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
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")
|
||||||
|
}
|
||||||
@@ -25,6 +25,7 @@ import (
|
|||||||
"io"
|
"io"
|
||||||
"maps"
|
"maps"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
"slices"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
@@ -906,6 +907,88 @@ func TestPoolsMigrationRetiresSupersededCopy(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestPoolsMigrationAfterCompleteRemoval(t *testing.T) {
|
||||||
|
ctx, cancel := context.WithCancel(t.Context())
|
||||||
|
defer cancel()
|
||||||
|
dirs, err := getRandomDisks(32)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
previousEndpoints := globalEndpoints
|
||||||
|
endpoints := mustGetPoolEndpoints(0, dirs[:16]...)
|
||||||
|
endpoints = append(endpoints, mustGetPoolEndpoints(1, dirs[16:]...)...)
|
||||||
|
globalEndpoints = endpoints
|
||||||
|
obj, _, err := initObjectLayer(ctx, endpoints)
|
||||||
|
if err != nil {
|
||||||
|
globalEndpoints = previousEndpoints
|
||||||
|
removeRoots(dirs)
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
z := obj.(*erasureServerPools)
|
||||||
|
previous := newObjectLayerFn()
|
||||||
|
setObjectLayer(z)
|
||||||
|
var reopened *erasureServerPools
|
||||||
|
oldShutdown := false
|
||||||
|
t.Cleanup(func() {
|
||||||
|
if reopened != nil {
|
||||||
|
_ = reopened.Shutdown(context.Background())
|
||||||
|
}
|
||||||
|
if !oldShutdown {
|
||||||
|
_ = z.Shutdown(context.Background())
|
||||||
|
}
|
||||||
|
removeRoots(dirs)
|
||||||
|
globalEndpoints = previousEndpoints
|
||||||
|
setObjectLayer(previous)
|
||||||
|
})
|
||||||
|
|
||||||
|
bucket, _, err := initAPIHandlerTest(ctx, z, nil, MakeBucketOptions{})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
const object = "object"
|
||||||
|
mtime := time.Now().UTC().Add(-time.Hour)
|
||||||
|
oldBody := bytes.Repeat([]byte{'a'}, 64<<10)
|
||||||
|
newBody := bytes.Repeat([]byte{'b'}, 256<<10)
|
||||||
|
nullQuorumCrossParityOverwrite(t, z, 0, bucket, object, oldBody, newBody, mtime, ObjectOptions{})
|
||||||
|
|
||||||
|
z.poolMetaMutex.Lock()
|
||||||
|
z.poolMeta.Pools[0].Decommission = &PoolDecommissionInfo{}
|
||||||
|
z.poolMetaMutex.Unlock()
|
||||||
|
if err := z.decommissionPool(ctx, 0, z.serverPools[0], decomBucketInfo{Name: bucket}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := z.CompleteDecommission(ctx, 0); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := z.Shutdown(context.Background()); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
oldShutdown = true
|
||||||
|
|
||||||
|
globalEndpoints = mustGetPoolEndpoints(0, dirs[16:]...)
|
||||||
|
reopenedObj, _, err := initObjectLayer(t.Context(), globalEndpoints)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
reopened = reopenedObj.(*erasureServerPools)
|
||||||
|
setObjectLayer(reopened)
|
||||||
|
_, router, err := initAPIHandlerTest(t.Context(), reopened, nil, MakeBucketOptions{})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
get := nullQuorumRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object))
|
||||||
|
if get.Code != http.StatusOK || !bytes.Equal(get.Body.Bytes(), newBody) {
|
||||||
|
t.Fatalf("GET after pool removal: status=%d bytes=%d", get.Code, get.Body.Len())
|
||||||
|
}
|
||||||
|
if code, keys := nullQuorumListV2Keys(t, router, bucket); code != http.StatusOK || !slices.Equal(keys, []string{object}) {
|
||||||
|
t.Fatalf("ListObjectsV2 after pool removal: status=%d keys=%v", code, keys)
|
||||||
|
}
|
||||||
|
if code, versions := nullQuorumListVersions(t, router, bucket); code != http.StatusOK || len(versions) != 1 || versions[0].Key != object || versions[0].VersionID != nullVersionID || versions[0].Size != int64(len(newBody)) {
|
||||||
|
t.Fatalf("ListObjectVersions after pool removal: status=%d versions=%+v", code, versions)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// A tiered move that fails, here because another pool cannot be read, must keep
|
// A tiered move that fails, here because another pool cannot be read, must keep
|
||||||
// its source like any other failed version.
|
// its source like any other failed version.
|
||||||
func TestPoolsMigrationTieredFailureKeepsSource(t *testing.T) {
|
func TestPoolsMigrationTieredFailureKeepsSource(t *testing.T) {
|
||||||
|
|||||||
@@ -542,6 +542,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
|
|||||||
|
|
||||||
var vc *versioning.Versioning
|
var vc *versioning.Versioning
|
||||||
var lc *lifecycle.Lifecycle
|
var lc *lifecycle.Lifecycle
|
||||||
|
var lifecycleUpdatedAt time.Time
|
||||||
var lr objectlock.Retention
|
var lr objectlock.Retention
|
||||||
var rcfg *replication.Config
|
var rcfg *replication.Config
|
||||||
if bucket != minioMetaBucket {
|
if bucket != minioMetaBucket {
|
||||||
@@ -551,7 +552,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Check if the current bucket has a configured lifecycle policy
|
// Check if the current bucket has a configured lifecycle policy
|
||||||
lc, err = globalLifecycleSys.Get(bucket)
|
lc, lifecycleUpdatedAt, err = globalBucketMetadataSys.GetLifecycleConfig(bucket)
|
||||||
if err != nil && !errors.Is(err, BucketLifecycleNotFound{Bucket: bucket}) {
|
if err != nil && !errors.Is(err, BucketLifecycleNotFound{Bucket: bucket}) {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -587,19 +588,27 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
|
|||||||
}
|
}
|
||||||
|
|
||||||
for setIdx, set := range pool.sets {
|
for setIdx, set := range pool.sets {
|
||||||
filterLifecycle := func(bucket, object string, fi FileInfo) bool {
|
// filterLifecycle returns an ILM deletion that migration must handle
|
||||||
|
// before retiring the source copy, or NoneAction.
|
||||||
|
filterLifecycle := func(bucket, object string, fi FileInfo) lifecycle.Action {
|
||||||
if lc == nil {
|
if lc == nil {
|
||||||
return false
|
return lifecycle.NoneAction
|
||||||
}
|
}
|
||||||
versioned := vc != nil && vc.Versioned(object)
|
versioned := vc != nil && vc.Versioned(object)
|
||||||
objInfo := fi.ToObjectInfo(bucket, object, versioned)
|
objInfo := fi.ToObjectInfo(bucket, object, versioned)
|
||||||
|
|
||||||
evt := evalActionFromLifecycle(ctx, *lc, lr, rcfg, objInfo)
|
evt := evalActionFromLifecycle(ctx, *lc, lr, rcfg, objInfo)
|
||||||
if evt.Action.Delete() {
|
switch {
|
||||||
globalExpiryState.enqueueByDays(objInfo, evt, lcEventSrc_Rebal)
|
case evt.Action == lifecycle.DeleteAction && versioned:
|
||||||
return true
|
// 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():
|
||||||
|
return evt.Action
|
||||||
}
|
}
|
||||||
return false
|
return lifecycle.NoneAction
|
||||||
}
|
}
|
||||||
|
|
||||||
rebalanceEntry := func(entry metaCacheEntry) {
|
rebalanceEntry := func(entry metaCacheEntry) {
|
||||||
@@ -624,6 +633,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
|
|||||||
versionsSorter(fivs.Versions).reverse()
|
versionsSorter(fivs.Versions).reverse()
|
||||||
|
|
||||||
var rebalanced int
|
var rebalanced int
|
||||||
|
var expired []migrationExpiredVersion
|
||||||
for _, version := range fivs.Versions {
|
for _, version := range fivs.Versions {
|
||||||
stopFn := globalRebalanceMetrics.log(rebalanceMetricRebalanceObject, poolIdx, bucket, version.Name, version.VersionID)
|
stopFn := globalRebalanceMetrics.log(rebalanceMetricRebalanceObject, poolIdx, bucket, version.Name, version.VersionID)
|
||||||
|
|
||||||
@@ -634,7 +644,20 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Apply lifecycle rules on the objects that are expired.
|
// 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) || action.DeleteAll() {
|
||||||
|
expired = append(expired, migrationExpiredVersion{
|
||||||
|
version: version,
|
||||||
|
event: lifecycle.Event{Action: action},
|
||||||
|
lifecycleUpdatedAt: lifecycleUpdatedAt,
|
||||||
|
source: lcEventSrc_Rebal,
|
||||||
|
})
|
||||||
|
}
|
||||||
stopFn(version.Size, errors.New("ILM expired object/version will be skipped"))
|
stopFn(version.Size, errors.New("ILM expired object/version will be skipped"))
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
@@ -743,7 +766,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
|
|||||||
// if all versions were rebalanced, we can delete the object versions.
|
// if all versions were rebalanced, we can delete the object versions.
|
||||||
if rebalanced == len(fivs.Versions) {
|
if rebalanced == len(fivs.Versions) {
|
||||||
stopFn := globalRebalanceMetrics.log(rebalanceMetricRebalanceRemoveObject, poolIdx, bucket, entry.name)
|
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)
|
stopFn(0, err)
|
||||||
auditLogRebalance(ctx, "Rebalance:DeleteObject", bucket, entry.name, "", err)
|
auditLogRebalance(ctx, "Rebalance:DeleteObject", bucket, entry.name, "", err)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -2868,7 +2868,7 @@ func (z *erasureServerPools) HealObjects(ctx context.Context, bucket, prefix str
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
fivs, err := entry.fileInfoVersions(bucket)
|
fivs, err := entry.allFileInfoVersions(bucket)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return healObjectFn(bucket, entry.name, "", scanMode)
|
return healObjectFn(bucket, entry.name, "", scanMode)
|
||||||
}
|
}
|
||||||
|
|||||||
+1
-1
@@ -423,7 +423,7 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
|
|||||||
encodedEntryName := encodeDirObject(entry.name)
|
encodedEntryName := encodeDirObject(entry.name)
|
||||||
|
|
||||||
var result healEntryResult
|
var result healEntryResult
|
||||||
fivs, err := entry.fileInfoVersions(bucket)
|
fivs, err := entry.allFileInfoVersions(bucket)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
res, err := healInPool(ctx, bucket, encodedEntryName, "",
|
res, err := healInPool(ctx, bucket, encodedEntryName, "",
|
||||||
madmin.HealOpts{
|
madmin.HealOpts{
|
||||||
|
|||||||
@@ -177,14 +177,12 @@ func (e metaCacheEntry) isInDir(dir, separator string) bool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// isLatestDeletemarker returns whether the latest version is a delete marker.
|
// isLatestDeletemarker returns whether the latest version is a delete marker.
|
||||||
|
// Free versions are not versions of the object and are skipped.
|
||||||
// If metadata is NOT versioned false will always be returned.
|
// If metadata is NOT versioned false will always be returned.
|
||||||
// If v2 and UNABLE to load metadata true will be returned.
|
// If v2 and UNABLE to load metadata true will be returned.
|
||||||
func (e *metaCacheEntry) isLatestDeletemarker() bool {
|
func (e *metaCacheEntry) isLatestDeletemarker() bool {
|
||||||
if e.cached != nil {
|
if e.cached != nil {
|
||||||
if len(e.cached.versions) == 0 {
|
return e.cached.isLatestDeleteMarker()
|
||||||
return true
|
|
||||||
}
|
|
||||||
return e.cached.versions[0].header.Type == DeleteType
|
|
||||||
}
|
}
|
||||||
if !isXL2V1Format(e.metadata) {
|
if !isXL2V1Format(e.metadata) {
|
||||||
return false
|
return false
|
||||||
@@ -196,10 +194,10 @@ func (e *metaCacheEntry) isLatestDeletemarker() bool {
|
|||||||
}
|
}
|
||||||
// Fall back...
|
// Fall back...
|
||||||
xlMeta, err := e.xlmeta()
|
xlMeta, err := e.xlmeta()
|
||||||
if err != nil || len(xlMeta.versions) == 0 {
|
if err != nil {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
return xlMeta.versions[0].header.Type == DeleteType
|
return xlMeta.isLatestDeleteMarker()
|
||||||
}
|
}
|
||||||
|
|
||||||
// isAllFreeVersions returns if all objects are free versions.
|
// isAllFreeVersions returns if all objects are free versions.
|
||||||
@@ -287,9 +285,21 @@ func (e *metaCacheEntry) xlmeta() (*xlMetaV2, error) {
|
|||||||
return e.cached, nil
|
return e.cached, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// fileInfoVersions returns the metadata as FileInfoVersions.
|
// fileInfoVersions returns the metadata as FileInfoVersions. The free
|
||||||
|
// versions of deleted tiered versions are not versions of the object; they
|
||||||
|
// are in FreeVersions.
|
||||||
// If entry is a directory it is returned as that.
|
// If entry is a directory it is returned as that.
|
||||||
func (e *metaCacheEntry) fileInfoVersions(bucket string) (FileInfoVersions, error) {
|
func (e *metaCacheEntry) fileInfoVersions(bucket string) (FileInfoVersions, error) {
|
||||||
|
return e.getFileInfoVersions(bucket, false)
|
||||||
|
}
|
||||||
|
|
||||||
|
// allFileInfoVersions returns the metadata as FileInfoVersions with the free
|
||||||
|
// versions among the versions, for healing.
|
||||||
|
func (e *metaCacheEntry) allFileInfoVersions(bucket string) (FileInfoVersions, error) {
|
||||||
|
return e.getFileInfoVersions(bucket, true)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (e *metaCacheEntry) getFileInfoVersions(bucket string, inclFreeVersions bool) (FileInfoVersions, error) {
|
||||||
if e.isDir() {
|
if e.isDir() {
|
||||||
return FileInfoVersions{
|
return FileInfoVersions{
|
||||||
Volume: bucket,
|
Volume: bucket,
|
||||||
@@ -304,7 +314,7 @@ func (e *metaCacheEntry) fileInfoVersions(bucket string) (FileInfoVersions, erro
|
|||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
// Too small gains to reuse cache here.
|
// Too small gains to reuse cache here.
|
||||||
return getFileInfoVersions(e.metadata, bucket, e.name, true)
|
return getFileInfoVersions(e.metadata, bucket, e.name, inclFreeVersions)
|
||||||
}
|
}
|
||||||
|
|
||||||
// metaCacheEntries is a slice of metacache entries.
|
// metaCacheEntries is a slice of metacache entries.
|
||||||
|
|||||||
@@ -117,6 +117,10 @@ func (d *nullQuorumWalkDisk) WalkDir(ctx context.Context, opts WalkDirOptions, o
|
|||||||
}
|
}
|
||||||
|
|
||||||
func nullQuorumBackend(t *testing.T) (*erasureServerPools, string, http.Handler) {
|
func nullQuorumBackend(t *testing.T) (*erasureServerPools, string, http.Handler) {
|
||||||
|
return nullQuorumBackendWithVersioning(t, false)
|
||||||
|
}
|
||||||
|
|
||||||
|
func nullQuorumBackendWithVersioning(t *testing.T, suspended bool) (*erasureServerPools, string, http.Handler) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
ctx, cancel := context.WithCancel(t.Context())
|
ctx, cancel := context.WithCancel(t.Context())
|
||||||
obj, dirs, err := prepareErasure16(ctx)
|
obj, dirs, err := prepareErasure16(ctx)
|
||||||
@@ -133,10 +137,16 @@ func nullQuorumBackend(t *testing.T) (*erasureServerPools, string, http.Handler)
|
|||||||
removeRoots(dirs)
|
removeRoots(dirs)
|
||||||
setObjectLayer(previous)
|
setObjectLayer(previous)
|
||||||
})
|
})
|
||||||
bucket, router, err := initAPIHandlerTest(ctx, z, nil, MakeBucketOptions{})
|
bucket, router, err := initAPIHandlerTest(ctx, z, nil, MakeBucketOptions{VersioningEnabled: suspended})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
if suspended {
|
||||||
|
if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketVersioningConfig,
|
||||||
|
[]byte(`<VersioningConfiguration><Status>Suspended</Status></VersioningConfiguration>`)); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
return z, bucket, router
|
return z, bucket, router
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -158,6 +168,36 @@ func nullQuorumPut(t *testing.T, z *erasureServerPools, bucket, object string, b
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func nullQuorumCrossParityOverwrite(t *testing.T, z *erasureServerPools, pool int, bucket, object string, oldBody, newBody []byte, oldTime time.Time, opts ObjectOptions) time.Time {
|
||||||
|
t.Helper()
|
||||||
|
set := z.serverPools[pool].getHashedSet(object)
|
||||||
|
disks := set.getDisks()
|
||||||
|
put := func(body []byte, opts ObjectOptions) {
|
||||||
|
if _, err := z.serverPools[pool].PutObject(t.Context(), bucket, object,
|
||||||
|
mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), opts); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
oldOpts := opts
|
||||||
|
oldOpts.MTime = oldTime
|
||||||
|
oldOpts.MaxParity = false
|
||||||
|
put(oldBody, oldOpts)
|
||||||
|
old := make([][]byte, len(disks))
|
||||||
|
for i, disk := range disks {
|
||||||
|
old[i] = mustReadNullQuorumMeta(t, disk, bucket, object)
|
||||||
|
}
|
||||||
|
newTime := oldTime.Add(time.Minute)
|
||||||
|
opts.MTime = newTime
|
||||||
|
opts.MaxParity = true
|
||||||
|
put(newBody, opts)
|
||||||
|
for i := 12; i < len(disks); i++ {
|
||||||
|
if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, old[i]); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return newTime
|
||||||
|
}
|
||||||
|
|
||||||
func nullQuorumListV2Keys(t *testing.T, router http.Handler, bucket string) (int, []string) {
|
func nullQuorumListV2Keys(t *testing.T, router http.Handler, bucket string) (int, []string) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", ""))
|
rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", ""))
|
||||||
@@ -175,6 +215,29 @@ func nullQuorumListV2Keys(t *testing.T, router http.Handler, bucket string) (int
|
|||||||
return rec.Code, keys
|
return rec.Code, keys
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type nullQuorumListedVersion struct {
|
||||||
|
Key string
|
||||||
|
VersionID string `xml:"VersionId"`
|
||||||
|
LastModified string
|
||||||
|
Size int64
|
||||||
|
IsLatest bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func nullQuorumListVersions(t *testing.T, router http.Handler, bucket string) (int, []nullQuorumListedVersion) {
|
||||||
|
t.Helper()
|
||||||
|
rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectVersionsURL("", bucket, "", "1000", ""))
|
||||||
|
if rec.Code != http.StatusOK {
|
||||||
|
return rec.Code, nil
|
||||||
|
}
|
||||||
|
var list struct {
|
||||||
|
Versions []nullQuorumListedVersion `xml:"Version"`
|
||||||
|
}
|
||||||
|
if err := xml.Unmarshal(rec.Body.Bytes(), &list); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
return rec.Code, list.Versions
|
||||||
|
}
|
||||||
|
|
||||||
func TestListObjectsSingleNullQuorumHTTP(t *testing.T) {
|
func TestListObjectsSingleNullQuorumHTTP(t *testing.T) {
|
||||||
z, bucket, router := nullQuorumBackend(t)
|
z, bucket, router := nullQuorumBackend(t)
|
||||||
oldBody := bytes.Repeat([]byte{'a'}, 8192)
|
oldBody := bytes.Repeat([]byte{'a'}, 8192)
|
||||||
|
|||||||
@@ -193,6 +193,11 @@ func (o *listPathOptions) shouldSkip(ctx context.Context, entry metaCacheEntry)
|
|||||||
if !o.InclDeleted && entry.isObject() && entry.isLatestDeletemarker() && !entry.isObjectDir() {
|
if !o.InclDeleted && entry.isObject() && entry.isLatestDeletemarker() && !entry.isObjectDir() {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
// Free versions are not versions of the object. Without other versions
|
||||||
|
// there is nothing to list, and the entry must not count toward the limit.
|
||||||
|
if o.InclDeleted && entry.isObject() && entry.isAllFreeVersions() {
|
||||||
|
return true
|
||||||
|
}
|
||||||
if o.Lifecycle != nil || o.Replication.Config != nil {
|
if o.Lifecycle != nil || o.Replication.Config != nil {
|
||||||
return triggerExpiryAndRepl(ctx, *o, entry)
|
return triggerExpiryAndRepl(ctx, *o, entry)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -28,6 +28,7 @@ import (
|
|||||||
|
|
||||||
"github.com/minio/minio/internal/amztime"
|
"github.com/minio/minio/internal/amztime"
|
||||||
"github.com/minio/minio/internal/bucket/lifecycle"
|
"github.com/minio/minio/internal/bucket/lifecycle"
|
||||||
|
"github.com/minio/minio/internal/bucket/replication"
|
||||||
"github.com/minio/minio/internal/crypto"
|
"github.com/minio/minio/internal/crypto"
|
||||||
"github.com/minio/minio/internal/event"
|
"github.com/minio/minio/internal/event"
|
||||||
"github.com/minio/minio/internal/hash"
|
"github.com/minio/minio/internal/hash"
|
||||||
@@ -499,6 +500,20 @@ func deleteObjectVersions(ctx context.Context, o ObjectLayer, bucket string, toD
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
sendEvent(evArgs)
|
sendEvent(evArgs)
|
||||||
|
|
||||||
|
// ILM purges carry a delete replication decision. Queue the purge
|
||||||
|
// after the local delete, just as the multi-object DELETE handler
|
||||||
|
// does for API-originated deletes.
|
||||||
|
if errs[i] == nil || isErrObjectNotFound(errs[i]) || isErrVersionNotFound(errs[i]) {
|
||||||
|
if dobj.VersionPurgeStatus() == replication.VersionPurgePending ||
|
||||||
|
dobj.DeleteMarkerReplicationStatus() == replication.Pending {
|
||||||
|
scheduleReplicationDelete(ctx, DeletedObjectReplicationInfo{
|
||||||
|
DeletedObject: dobj,
|
||||||
|
Bucket: bucket,
|
||||||
|
EventType: ReplicateILMDelete,
|
||||||
|
}, o)
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,333 @@
|
|||||||
|
// Copyright (c) 2015-2026 MinIO, Inc.
|
||||||
|
// Copyright (c) 2026 PGSTY
|
||||||
|
//
|
||||||
|
// 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"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"strings"
|
||||||
|
"sync/atomic"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/minio/madmin-go/v3"
|
||||||
|
"github.com/minio/minio-go/v7"
|
||||||
|
"github.com/minio/minio/internal/auth"
|
||||||
|
"github.com/minio/minio/internal/bucket/lifecycle"
|
||||||
|
"github.com/minio/minio/internal/bucket/replication"
|
||||||
|
xhttp "github.com/minio/minio/internal/http"
|
||||||
|
"github.com/minio/minio/internal/once"
|
||||||
|
)
|
||||||
|
|
||||||
|
// ilmDeleteTarget models one replication destination for ILM expiry tests.
|
||||||
|
type ilmDeleteTarget struct {
|
||||||
|
arn, bucket string
|
||||||
|
deletes atomic.Int32
|
||||||
|
}
|
||||||
|
|
||||||
|
func newILMTestTarget(t *testing.T) *ilmDeleteTarget {
|
||||||
|
return &ilmDeleteTarget{arn: "arn:minio:replication::" + mustGetUUID() + ":bucket", bucket: getRandomBucketName()}
|
||||||
|
}
|
||||||
|
|
||||||
|
func registerILMTestTarget(t *testing.T, obj ObjectLayer, bucket string, target *ilmDeleteTarget, handler http.HandlerFunc) {
|
||||||
|
t.Helper()
|
||||||
|
if err := obj.MakeBucket(t.Context(), target.bucket, MakeBucketOptions{}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
remote := httptest.NewServer(handler)
|
||||||
|
t.Cleanup(remote.Close)
|
||||||
|
client, err := minio.New(strings.TrimPrefix(remote.URL, "http://"), &minio.Options{Region: "us-east-1", MaxRetries: 1})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
globalBucketTargetSys.Lock()
|
||||||
|
globalBucketTargetSys.arnRemotesMap[target.arn] = arnTarget{Client: &TargetClient{Client: client, ARN: target.arn, Bucket: target.bucket}, lastRefresh: UTCNow()}
|
||||||
|
globalBucketTargetSys.targetsMap[bucket] = append(globalBucketTargetSys.targetsMap[bucket], madmin.BucketTarget{Arn: target.arn, TargetBucket: target.bucket})
|
||||||
|
globalBucketTargetSys.Unlock()
|
||||||
|
globalBucketTargetSys.hMutex.Lock()
|
||||||
|
globalBucketTargetSys.hc[client.EndpointURL().Host] = epHealth{Online: true}
|
||||||
|
globalBucketTargetSys.hMutex.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
|
func setILMTestReplicationConfig(t *testing.T, bucket, arn, targetBucket string) {
|
||||||
|
t.Helper()
|
||||||
|
cfg := replication.Config{}
|
||||||
|
rule := configs[0].Rules[0]
|
||||||
|
rule.Priority = 1
|
||||||
|
rule.Destination = replication.Destination{ARN: arn, Bucket: targetBucket}
|
||||||
|
cfg.Rules = append(cfg.Rules, rule)
|
||||||
|
cfg.RoleArn = arn
|
||||||
|
meta, err := globalBucketMetadataSys.Get(bucket)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
meta.replicationConfig = &cfg
|
||||||
|
globalBucketMetadataSys.Set(bucket, meta)
|
||||||
|
}
|
||||||
|
|
||||||
|
func newILMTestReplicationPool(ctx context.Context, obj ObjectLayer, stats *ReplicationStats) (restore func(), receive func() DeletedObjectReplicationInfo) {
|
||||||
|
oldPool := globalReplicationPool
|
||||||
|
p := &ReplicationPool{ctx: ctx, objLayer: obj, workers: []chan ReplicationWorkerOperation{make(chan ReplicationWorkerOperation, 8)}, stats: stats, mrfSaveCh: make(chan MRFReplicateEntry, 8)}
|
||||||
|
globalReplicationPool = once.NewSingleton[ReplicationPool]()
|
||||||
|
globalReplicationPool.Set(p)
|
||||||
|
return func() { globalReplicationPool = oldPool }, func() DeletedObjectReplicationInfo {
|
||||||
|
select {
|
||||||
|
case op := <-p.workers[0]:
|
||||||
|
d, ok := op.(DeletedObjectReplicationInfo)
|
||||||
|
if !ok {
|
||||||
|
return DeletedObjectReplicationInfo{}
|
||||||
|
}
|
||||||
|
return d
|
||||||
|
case <-time.After(3 * time.Second):
|
||||||
|
return DeletedObjectReplicationInfo{}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestReplicateILMExpiryDeleteMarker verifies that a delete marker created by
|
||||||
|
// ILM expiry carries delete replication state and is queued for replication,
|
||||||
|
// exactly like a marker created through the S3 DELETE API.
|
||||||
|
func TestReplicateILMExpiryDeleteMarker(t *testing.T) {
|
||||||
|
defer DetectTestLeak(t)()
|
||||||
|
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
|
||||||
|
t: t,
|
||||||
|
objAPITest: testReplicateILMExpiryDeleteMarker,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
func testReplicateILMExpiryDeleteMarker(obj ObjectLayer, instanceType, bucket string, _ http.Handler, creds auth.Credentials, t *testing.T) {
|
||||||
|
ctx, cancel := context.WithCancel(t.Context())
|
||||||
|
defer cancel()
|
||||||
|
stats := NewReplicationStats(ctx, nil)
|
||||||
|
oldStats := globalReplicationStats.Swap(stats)
|
||||||
|
defer globalReplicationStats.Store(oldStats)
|
||||||
|
|
||||||
|
const name = "ilm-object"
|
||||||
|
if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketVersioningConfig, enabledBucketVersioningConfig); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
// DeleteObject refuses ILM expiry (Expiration.Expire) without lifecycle
|
||||||
|
// rules; a date in the past makes every version eligible immediately.
|
||||||
|
ilmXML := []byte(`<LifecycleConfiguration><Rule><ID>ilm-expire</ID><Status>Enabled</Status><Filter/><Expiration><Date>2020-01-01T00:00:00Z</Date></Expiration></Rule></LifecycleConfiguration>`)
|
||||||
|
if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketLifecycleConfig, ilmXML); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
oi, err := obj.PutObject(ctx, bucket, name, mustGetPutObjReader(t, strings.NewReader("data"), 4, "", ""), ObjectOptions{Versioned: true})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
target := newILMTestTarget(t)
|
||||||
|
registerILMTestTarget(t, obj, bucket, target, func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
topts := ObjectOptions{VersionID: r.URL.Query().Get("versionId"), Versioned: true}
|
||||||
|
switch r.Method {
|
||||||
|
case http.MethodHead:
|
||||||
|
roi, rerr := obj.GetObjectInfo(r.Context(), target.bucket, name, topts)
|
||||||
|
if roi.DeleteMarker {
|
||||||
|
w.Header().Set(xhttp.AmzDeleteMarker, "true")
|
||||||
|
w.Header().Set(xhttp.AmzVersionID, roi.VersionID)
|
||||||
|
}
|
||||||
|
if rerr != nil {
|
||||||
|
writeErrorResponseHeadersOnly(w, toAPIError(r.Context(), rerr))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
case http.MethodDelete:
|
||||||
|
target.deletes.Add(1)
|
||||||
|
if topts.VersionID == "" {
|
||||||
|
t.Error("missing remote versionId")
|
||||||
|
}
|
||||||
|
topts.DeleteMarker = r.Header.Get(xhttp.MinIOSourceDeleteMarker) == "true"
|
||||||
|
topts.ReplicationRequest = true
|
||||||
|
topts.SetReplicaStatus(replication.Replica)
|
||||||
|
if _, err := obj.DeleteObject(r.Context(), target.bucket, name, topts); err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
|
||||||
|
writeErrorResponse(r.Context(), w, toAPIError(r.Context(), err), r.URL)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
w.WriteHeader(http.StatusNoContent)
|
||||||
|
default:
|
||||||
|
t.Errorf("unexpected remote method %s", r.Method)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
setILMTestReplicationConfig(t, bucket, target.arn, target.bucket)
|
||||||
|
restorePool, receive := newILMTestReplicationPool(ctx, obj, stats)
|
||||||
|
defer restorePool()
|
||||||
|
|
||||||
|
if ok := applyExpiryOnNonTransitionedObjects(ctx, obj, oi, lifecycle.Event{Action: lifecycle.DeleteAction}, lcEventSrc_Scanner); !ok {
|
||||||
|
t.Fatalf("%s: ILM expiry did not apply", instanceType)
|
||||||
|
}
|
||||||
|
|
||||||
|
deletion := receive()
|
||||||
|
if deletion.ObjectName == "" {
|
||||||
|
t.Fatalf("%s: ILM delete marker was not queued for replication", instanceType)
|
||||||
|
}
|
||||||
|
if !deletion.DeleteMarker || deletion.VersionID != "" || deletion.DeleteMarkerVersionID == "" {
|
||||||
|
t.Fatalf("%s: unexpected queued delete: %+v", instanceType, deletion)
|
||||||
|
}
|
||||||
|
if deletion.EventType != ReplicateILMDelete {
|
||||||
|
t.Fatalf("%s: queued event type = %q, want %q", instanceType, deletion.EventType, ReplicateILMDelete)
|
||||||
|
}
|
||||||
|
|
||||||
|
// The ILM-created marker must carry pending replication state.
|
||||||
|
markerVID := deletion.DeleteMarkerVersionID
|
||||||
|
moi, merr := obj.GetObjectInfo(ctx, bucket, name, ObjectOptions{VersionID: markerVID, Versioned: true})
|
||||||
|
if !isErrMethodNotAllowed(merr) || !moi.DeleteMarker {
|
||||||
|
t.Fatalf("%s: ILM marker missing: %v %v", instanceType, moi, merr)
|
||||||
|
}
|
||||||
|
if moi.ReplicationStatus != replication.Pending {
|
||||||
|
t.Fatalf("%s: ILM marker replication status = %q, want PENDING", instanceType, moi.ReplicationStatus)
|
||||||
|
}
|
||||||
|
|
||||||
|
result := replicateDelete(ctx, deletion, obj)
|
||||||
|
if result.ReplicationStatus() != replication.Completed {
|
||||||
|
t.Fatalf("%s: marker replication result: %+v", instanceType, result)
|
||||||
|
}
|
||||||
|
if target.deletes.Load() != 1 {
|
||||||
|
t.Fatalf("%s: target saw %d deletes, want 1", instanceType, target.deletes.Load())
|
||||||
|
}
|
||||||
|
// The marker must exist on the target.
|
||||||
|
roi, rerr := obj.GetObjectInfo(ctx, target.bucket, name, ObjectOptions{VersionID: markerVID, Versioned: true})
|
||||||
|
if !isErrMethodNotAllowed(rerr) || !roi.DeleteMarker {
|
||||||
|
t.Fatalf("%s: target marker missing: %v %v", instanceType, roi, rerr)
|
||||||
|
}
|
||||||
|
// The source marker records the completed replication per target. The
|
||||||
|
// composite ReplicationStatus field is unreliable on read-back for
|
||||||
|
// markers carrying a replication timestamp, so assert the raw state.
|
||||||
|
moi, merr = obj.GetObjectInfo(ctx, bucket, name, ObjectOptions{VersionID: markerVID})
|
||||||
|
if !isErrMethodNotAllowed(merr) || !strings.Contains(moi.ReplicationStatusInternal, "=COMPLETED;") {
|
||||||
|
t.Fatalf("%s: source marker state = %q (err %v), want COMPLETED", instanceType, moi.ReplicationStatusInternal, merr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestReplicateILMExpiryNoncurrentPurge verifies that a noncurrent version
|
||||||
|
// removed by ILM carries a pending version purge state, is queued for delete
|
||||||
|
// replication and converges on the target.
|
||||||
|
func TestReplicateILMExpiryNoncurrentPurge(t *testing.T) {
|
||||||
|
defer DetectTestLeak(t)()
|
||||||
|
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
|
||||||
|
t: t,
|
||||||
|
objAPITest: testReplicateILMExpiryNoncurrentPurge,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
func testReplicateILMExpiryNoncurrentPurge(obj ObjectLayer, instanceType, bucket string, _ http.Handler, creds auth.Credentials, t *testing.T) {
|
||||||
|
ctx, cancel := context.WithCancel(t.Context())
|
||||||
|
defer cancel()
|
||||||
|
stats := NewReplicationStats(ctx, nil)
|
||||||
|
oldStats := globalReplicationStats.Swap(stats)
|
||||||
|
defer globalReplicationStats.Store(oldStats)
|
||||||
|
|
||||||
|
const name = "ilm-object"
|
||||||
|
if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketVersioningConfig, enabledBucketVersioningConfig); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if _, err := obj.PutObject(ctx, bucket, name, mustGetPutObjReader(t, strings.NewReader("v1"), 2, "", ""), ObjectOptions{Versioned: true}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
v1Info, err := obj.PutObject(ctx, bucket, name, mustGetPutObjReader(t, strings.NewReader("v2"), 2, "", ""), ObjectOptions{Versioned: true})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
v1 := v1Info.VersionID
|
||||||
|
if v1 == "" {
|
||||||
|
t.Fatalf("%s: missing version ID on second version", instanceType)
|
||||||
|
}
|
||||||
|
|
||||||
|
target := newILMTestTarget(t)
|
||||||
|
registerILMTestTarget(t, obj, bucket, target, func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
topts := ObjectOptions{VersionID: r.URL.Query().Get("versionId"), Versioned: true}
|
||||||
|
switch r.Method {
|
||||||
|
case http.MethodDelete:
|
||||||
|
target.deletes.Add(1)
|
||||||
|
if topts.VersionID == "" {
|
||||||
|
t.Error("missing remote versionId")
|
||||||
|
}
|
||||||
|
topts.DeleteMarker = r.Header.Get(xhttp.MinIOSourceDeleteMarker) == "true"
|
||||||
|
topts.ReplicationRequest = true
|
||||||
|
topts.SetReplicaStatus(replication.Replica)
|
||||||
|
if _, err := obj.DeleteObject(r.Context(), target.bucket, name, topts); err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
|
||||||
|
writeErrorResponse(r.Context(), w, toAPIError(r.Context(), err), r.URL)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
w.WriteHeader(http.StatusNoContent)
|
||||||
|
default:
|
||||||
|
t.Errorf("unexpected remote method %s", r.Method)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
// Pre-create a marker version on the target with the same version ID as
|
||||||
|
// the noncurrent version being purged, so the purge can be observed.
|
||||||
|
if _, err := obj.DeleteObject(ctx, target.bucket, name, ObjectOptions{VersionID: v1, Versioned: true, DeleteMarker: true, ReplicationRequest: true}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
setILMTestReplicationConfig(t, bucket, target.arn, target.bucket)
|
||||||
|
restorePool, receive := newILMTestReplicationPool(ctx, obj, stats)
|
||||||
|
defer restorePool()
|
||||||
|
|
||||||
|
// The scanner evaluates noncurrent version expiry per version; build the
|
||||||
|
// delete entry exactly the way the scanner does for the oldest version.
|
||||||
|
vi, verr := obj.GetObjectInfo(ctx, bucket, name, ObjectOptions{VersionID: v1, Versioned: true})
|
||||||
|
if verr != nil {
|
||||||
|
t.Fatalf("%s: missing noncurrent version: %v", instanceType, verr)
|
||||||
|
}
|
||||||
|
otd := ilmPurgeToDelete(ctx, bucket, vi, true)
|
||||||
|
if otd.VersionPurgeStatus != replication.VersionPurgePending || otd.VersionPurgeStatuses == "" || otd.ReplicateDecisionStr == "" {
|
||||||
|
t.Fatalf("%s: ILM purge entry carries no replication state: %+v", instanceType, otd)
|
||||||
|
}
|
||||||
|
if plain := ilmPurgeToDelete(ctx, bucket, vi, false); plain.VersionPurgeStatus != "" || plain.ReplicateDecisionStr != "" {
|
||||||
|
t.Fatalf("%s: non-replicating purge entry carries state: %+v", instanceType, plain)
|
||||||
|
}
|
||||||
|
|
||||||
|
deleteObjectVersions(ctx, obj, bucket, []ObjectToDelete{otd}, []lifecycle.Event{{Action: lifecycle.DeleteVersionAction}})
|
||||||
|
|
||||||
|
// The version must remain locally as a tombstone with pending purge
|
||||||
|
// state until replication completes.
|
||||||
|
pi, perr := obj.GetObjectInfo(ctx, bucket, name, ObjectOptions{VersionID: v1, Versioned: true})
|
||||||
|
if !isErrMethodNotAllowed(perr) {
|
||||||
|
t.Fatalf("%s: purged version metadata lost: %v", instanceType, perr)
|
||||||
|
}
|
||||||
|
if pi.VersionPurgeStatus != replication.VersionPurgePending {
|
||||||
|
t.Fatalf("%s: purged version state = %q, want PENDING", instanceType, pi.VersionPurgeStatus)
|
||||||
|
}
|
||||||
|
|
||||||
|
deletion := receive()
|
||||||
|
if deletion.ObjectName == "" {
|
||||||
|
t.Fatalf("%s: ILM noncurrent purge was not queued for replication", instanceType)
|
||||||
|
}
|
||||||
|
if deletion.VersionID != v1 || deletion.DeleteMarkerVersionID != "" || deletion.DeleteMarker {
|
||||||
|
t.Fatalf("%s: unexpected queued purge: %+v", instanceType, deletion)
|
||||||
|
}
|
||||||
|
|
||||||
|
result := replicateDelete(ctx, deletion, obj)
|
||||||
|
if result.VersionPurgeStatus() != replication.VersionPurgeComplete {
|
||||||
|
t.Fatalf("%s: purge replication result: %+v", instanceType, result)
|
||||||
|
}
|
||||||
|
if target.deletes.Load() != 1 {
|
||||||
|
t.Fatalf("%s: target saw %d deletes, want 1", instanceType, target.deletes.Load())
|
||||||
|
}
|
||||||
|
// The purge must have converged on the target.
|
||||||
|
if _, terr := obj.GetObjectInfo(ctx, target.bucket, name, ObjectOptions{VersionID: v1, Versioned: true}); !isErrVersionNotFound(terr) && !isErrObjectNotFound(terr) {
|
||||||
|
t.Fatalf("%s: target version still present: %v", instanceType, terr)
|
||||||
|
}
|
||||||
|
// Once the purge completes the local tombstone is freed and the version
|
||||||
|
// is gone on both sides.
|
||||||
|
if _, terr := obj.GetObjectInfo(ctx, bucket, name, ObjectOptions{VersionID: v1}); !isErrVersionNotFound(terr) && !isErrObjectNotFound(terr) {
|
||||||
|
t.Fatalf("%s: purged version still present locally: %v", instanceType, terr)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -19,6 +19,7 @@ package cmd
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"errors"
|
"errors"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/zeebo/xxh3"
|
"github.com/zeebo/xxh3"
|
||||||
)
|
)
|
||||||
@@ -28,8 +29,12 @@ import (
|
|||||||
// - fivs.FreeVersions has all the 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.
|
// if inclFreeVersions is true all the versions are in fivs.Versions, free and non-free versions alike.
|
||||||
|
// Otherwise IsLatest, SuccessorModTime and LatestModTime are those of the
|
||||||
|
// non-free versions.
|
||||||
//
|
//
|
||||||
// Note: Only the scanner requires fivs.Versions to have exclusively non-free versions. This is used while enforcing NewerNoncurrentVersions lifecycle element.
|
// Only healing needs the free versions among the versions. Free versions are
|
||||||
|
// not versions of the object, and lifecycle evaluation, listings and data
|
||||||
|
// movement must not count them.
|
||||||
func getFileInfoVersions(xlMetaBuf []byte, volume, path string, inclFreeVersions bool) (FileInfoVersions, error) {
|
func getFileInfoVersions(xlMetaBuf []byte, volume, path string, inclFreeVersions bool) (FileInfoVersions, error) {
|
||||||
fivs, err := getAllFileInfoVersions(xlMetaBuf, volume, path, true)
|
fivs, err := getAllFileInfoVersions(xlMetaBuf, volume, path, true)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -56,6 +61,19 @@ func getFileInfoVersions(xlMetaBuf []byte, volume, path string, inclFreeVersions
|
|||||||
}
|
}
|
||||||
if !inclFreeVersions {
|
if !inclFreeVersions {
|
||||||
fivs.Versions = fivs.Versions[:n]
|
fivs.Versions = fivs.Versions[:n]
|
||||||
|
// A free version is not a version of the object. The latest
|
||||||
|
// version, and the successor of each other version, is among
|
||||||
|
// the rest.
|
||||||
|
for i := range fivs.Versions {
|
||||||
|
fivs.Versions[i].IsLatest = i == 0
|
||||||
|
fivs.Versions[i].SuccessorModTime = time.Time{}
|
||||||
|
if i > 0 {
|
||||||
|
fivs.Versions[i].SuccessorModTime = fivs.Versions[i-1].ModTime
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if n > 0 {
|
||||||
|
fivs.LatestModTime = fivs.Versions[0].ModTime
|
||||||
|
}
|
||||||
}
|
}
|
||||||
// Update numversions
|
// Update numversions
|
||||||
for i := range fivs.Versions {
|
for i := range fivs.Versions {
|
||||||
|
|||||||
@@ -20,6 +20,7 @@ package cmd
|
|||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
"slices"
|
"slices"
|
||||||
"testing"
|
"testing"
|
||||||
@@ -73,21 +74,25 @@ func TestMergeXLV2LatestNullCrossParity(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
input = append(input, make([][]xlMetaV2ShallowVersion, tt.empty)...)
|
input = append(input, make([][]xlMetaV2ShallowVersion, tt.empty)...)
|
||||||
|
// Latest-only listing and listing all versions (decommission,
|
||||||
|
// rebalance) must select the same version.
|
||||||
|
for _, requested := range []int{1, 0} {
|
||||||
for order, streams := range quorumVersionOrders(input) {
|
for order, streams := range quorumVersionOrders(input) {
|
||||||
got := mergeXLV2Versions(tt.quorum, false, 1, streams...)
|
got := mergeXLV2Versions(tt.quorum, false, requested, streams...)
|
||||||
if tt.want < 0 {
|
if tt.want < 0 {
|
||||||
if len(got) != 0 {
|
if len(got) != 0 {
|
||||||
t.Fatalf("order %d: subquorum groups were combined: %v", order, got)
|
t.Fatalf("requested %d, order %d: subquorum groups were combined: %v", requested, order, got)
|
||||||
}
|
}
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if len(got) != 1 || got[0].header != versions[tt.want].header {
|
if len(got) != 1 || got[0].header != versions[tt.want].header {
|
||||||
t.Fatalf("order %d: got %v, want group %d", order, got, tt.want)
|
t.Fatalf("requested %d, order %d: got %v, want group %d", requested, order, got, tt.want)
|
||||||
}
|
}
|
||||||
xl := xlMetaV2{versions: got}
|
xl := xlMetaV2{versions: got}
|
||||||
fi, err := xl.ToFileInfo("bucket", "object", "", false, true)
|
fi, err := xl.ToFileInfo("bucket", "object", "", false, true)
|
||||||
if err != nil || !fi.IsValid() || fi.Erasure.DataBlocks != int(versions[tt.want].header.EcM) {
|
if err != nil || !fi.IsValid() || fi.Erasure.DataBlocks != int(versions[tt.want].header.EcM) {
|
||||||
t.Fatalf("order %d: selected metadata is invalid: %+v, %v", order, fi, err)
|
t.Fatalf("requested %d, order %d: selected metadata is invalid: %+v, %v", requested, order, fi, err)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
@@ -97,15 +102,13 @@ func TestMergeXLV2LatestNullCrossParity(t *testing.T) {
|
|||||||
func TestMergeXLV2LatestNullCrossParityScope(t *testing.T) {
|
func TestMergeXLV2LatestNullCrossParityScope(t *testing.T) {
|
||||||
old := crossParityNullVersion(t, 100, 12, 4)
|
old := crossParityNullVersion(t, 100, 12, 4)
|
||||||
newer := crossParityNullVersion(t, 200, 8, 8)
|
newer := crossParityNullVersion(t, 200, 8, 8)
|
||||||
for _, kind := range []string{"strict", "all-versions", "uuid", "delete", "free", "history", "legacy-modern-ec", "partial-ec"} {
|
for _, kind := range []string{"strict", "uuid", "delete", "free", "history", "legacy-modern-ec", "partial-ec"} {
|
||||||
t.Run(kind, func(t *testing.T) {
|
t.Run(kind, func(t *testing.T) {
|
||||||
a, b := old, newer
|
a, b := old, newer
|
||||||
strict, requested := false, 1
|
strict := false
|
||||||
switch kind {
|
switch kind {
|
||||||
case "strict":
|
case "strict":
|
||||||
strict = true
|
strict = true
|
||||||
case "all-versions":
|
|
||||||
requested = 0
|
|
||||||
case "uuid":
|
case "uuid":
|
||||||
a.header.VersionID, b.header.VersionID = [16]byte{1}, [16]byte{1}
|
a.header.VersionID, b.header.VersionID = [16]byte{1}, [16]byte{1}
|
||||||
case "delete":
|
case "delete":
|
||||||
@@ -140,8 +143,10 @@ func TestMergeXLV2LatestNullCrossParityScope(t *testing.T) {
|
|||||||
history.header.VersionID = [16]byte{1}
|
history.header.VersionID = [16]byte{1}
|
||||||
input[0] = append(input[0], history)
|
input[0] = append(input[0], history)
|
||||||
}
|
}
|
||||||
|
for _, requested := range []int{1, 0} {
|
||||||
if got := mergeXLV2Versions(8, strict, requested, input...); len(got) != 0 {
|
if got := mergeXLV2Versions(8, strict, requested, input...); len(got) != 0 {
|
||||||
t.Fatalf("excluded input changed: %v", got)
|
t.Fatalf("requested %d: excluded input changed: %v", requested, got)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
@@ -189,25 +194,14 @@ func TestListCrossParityReadableMajority(t *testing.T) {
|
|||||||
t.Run(fmt.Sprintf("bytes-%d", size), func(t *testing.T) {
|
t.Run(fmt.Sprintf("bytes-%d", size), func(t *testing.T) {
|
||||||
z, bucket, router := nullQuorumBackend(t)
|
z, bucket, router := nullQuorumBackend(t)
|
||||||
set := z.serverPools[0].sets[0]
|
set := z.serverPools[0].sets[0]
|
||||||
disks := set.getDisks()
|
|
||||||
const object = "object"
|
const object = "object"
|
||||||
mtime := time.Now().UTC().Add(-time.Hour)
|
mtime := time.Now().UTC().Add(-time.Hour)
|
||||||
oldBody := bytes.Repeat([]byte{'a'}, size)
|
oldBody := bytes.Repeat([]byte{'a'}, size)
|
||||||
newBody := bytes.Repeat([]byte{'b'}, size)
|
newBody := bytes.Repeat([]byte{'b'}, size)
|
||||||
nullQuorumPut(t, z, bucket, object, oldBody, ObjectOptions{MTime: mtime})
|
newTime := nullQuorumCrossParityOverwrite(t, z, 0, bucket, object, oldBody, newBody, mtime, ObjectOptions{})
|
||||||
old := make([][]byte, len(disks))
|
|
||||||
for i, disk := range disks {
|
|
||||||
old[i] = mustReadNullQuorumMeta(t, disk, bucket, object)
|
|
||||||
}
|
|
||||||
nullQuorumPut(t, z, bucket, object, newBody, ObjectOptions{MTime: mtime.Add(time.Minute), MaxParity: true})
|
|
||||||
for i := 12; i < len(disks); i++ {
|
|
||||||
if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, old[i]); err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
for _, readData := range []bool{false, true} {
|
for _, readData := range []bool{false, true} {
|
||||||
fi, _, _, err := set.getObjectFileInfo(t.Context(), bucket, object, ObjectOptions{}, readData)
|
fi, _, _, err := set.getObjectFileInfo(t.Context(), bucket, object, ObjectOptions{}, readData)
|
||||||
if err != nil || !fi.ModTime.Equal(mtime.Add(time.Minute)) || fi.Erasure.DataBlocks != 8 {
|
if err != nil || !fi.ModTime.Equal(newTime) || fi.Erasure.DataBlocks != 8 {
|
||||||
t.Fatalf("readData=%v: full metadata read lost readable majority: %+v, %v", readData, fi, err)
|
t.Fatalf("readData=%v: full metadata read lost readable majority: %+v, %v", readData, fi, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -220,3 +214,76 @@ func TestListCrossParityReadableMajority(t *testing.T) {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestListObjectVersionsCrossParityNull(t *testing.T) {
|
||||||
|
for _, tc := range []struct {
|
||||||
|
name string
|
||||||
|
suspended bool
|
||||||
|
}{
|
||||||
|
{name: "unversioned"},
|
||||||
|
{name: "suspended", suspended: true},
|
||||||
|
} {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
z, bucket, router := nullQuorumBackendWithVersioning(t, tc.suspended)
|
||||||
|
const object = "object"
|
||||||
|
mtime := time.Now().UTC().Add(-time.Hour)
|
||||||
|
oldBody := bytes.Repeat([]byte{'a'}, 64<<10)
|
||||||
|
newBody := bytes.Repeat([]byte{'b'}, 256<<10)
|
||||||
|
nullQuorumCrossParityOverwrite(t, z, 0, bucket, object, oldBody, newBody, mtime,
|
||||||
|
ObjectOptions{VersionSuspended: tc.suspended})
|
||||||
|
|
||||||
|
code, versions := nullQuorumListVersions(t, router, bucket)
|
||||||
|
if code != http.StatusOK || len(versions) != 1 {
|
||||||
|
t.Fatalf("ListObjectVersions: status=%d versions=%+v", code, versions)
|
||||||
|
}
|
||||||
|
got := versions[0]
|
||||||
|
if got.Key != object || got.VersionID != nullVersionID || !got.IsLatest || got.Size != int64(len(newBody)) || got.LastModified == "" {
|
||||||
|
t.Fatalf("ListObjectVersions selected the wrong null version: %+v", got)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Decommission and rebalance list every version. A readable degraded overwrite
|
||||||
|
// must not be skipped there, or removing the pool loses it.
|
||||||
|
func TestMigrationCrossParityReadableMajority(t *testing.T) {
|
||||||
|
z, bucket, _ := nullQuorumBackend(t)
|
||||||
|
set := z.serverPools[0].sets[0]
|
||||||
|
const object = "object"
|
||||||
|
mtime := time.Now().UTC().Add(-time.Hour)
|
||||||
|
oldBody := bytes.Repeat([]byte{'a'}, 256<<10)
|
||||||
|
newBody := bytes.Repeat([]byte{'b'}, 256<<10)
|
||||||
|
nullQuorumCrossParityOverwrite(t, z, 0, bucket, object, oldBody, newBody, mtime, ObjectOptions{})
|
||||||
|
for _, kind := range []string{"rebalance", "decommission"} {
|
||||||
|
t.Run(kind, func(t *testing.T) {
|
||||||
|
var names []string
|
||||||
|
consume := func(entry metaCacheEntry) {
|
||||||
|
versions, err := entry.fileInfoVersions(bucket)
|
||||||
|
if err != nil || len(versions.Versions) != 1 || !versions.Versions[0].ModTime.Equal(mtime.Add(time.Minute)) {
|
||||||
|
t.Errorf("invalid migration metadata: %+v, %v", versions, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
reader, err := set.GetObjectNInfo(t.Context(), bucket, entry.name, nil, nil, ObjectOptions{VersionID: nullVersionID, NoLock: true, NoDecryption: true})
|
||||||
|
if err != nil {
|
||||||
|
t.Error(err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
body, err := io.ReadAll(reader)
|
||||||
|
reader.Close()
|
||||||
|
if err != nil || !bytes.Equal(body, newBody) {
|
||||||
|
t.Errorf("migration could not read selected object data: %v", err)
|
||||||
|
}
|
||||||
|
names = append(names, entry.name)
|
||||||
|
}
|
||||||
|
var err error
|
||||||
|
if kind == "rebalance" {
|
||||||
|
err = set.listObjectsToRebalance(t.Context(), bucket, consume)
|
||||||
|
} else {
|
||||||
|
err = set.listObjectsToDecommission(t.Context(), decomBucketInfo{Name: bucket}, consume)
|
||||||
|
}
|
||||||
|
if err != nil || !slices.Equal(names, []string{object}) {
|
||||||
|
t.Fatalf("migration listing skipped a readable object: %v, %v", names, err)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
+27
-13
@@ -1138,6 +1138,17 @@ func (x *xlMetaV2) latestModtime() time.Time {
|
|||||||
return time.Unix(0, x.versions[0].header.ModTime)
|
return time.Unix(0, x.versions[0].header.ModTime)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// isLatestDeleteMarker returns true if the latest version that is not a free
|
||||||
|
// version is a delete marker, or if there is none.
|
||||||
|
func (x *xlMetaV2) isLatestDeleteMarker() bool {
|
||||||
|
for _, ver := range x.versions {
|
||||||
|
if !ver.header.FreeVersion() {
|
||||||
|
return ver.header.Type == DeleteType
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
func (x *xlMetaV2) addVersion(ver xlMetaV2Version) error {
|
func (x *xlMetaV2) addVersion(ver xlMetaV2Version) error {
|
||||||
modTime := ver.getModTime().UnixNano()
|
modTime := ver.getModTime().UnixNano()
|
||||||
if !ver.Valid() {
|
if !ver.Valid() {
|
||||||
@@ -2027,9 +2038,11 @@ func mergeXLV2Versions(quorum int, strict bool, requestedVersions int, versions
|
|||||||
if !checkedSingleNull {
|
if !checkedSingleNull {
|
||||||
singleNull = singleNullVersionStreams(originalVersions, false)
|
singleNull = singleNullVersionStreams(originalVersions, false)
|
||||||
// Availability-optimized writes can change EC while a drive
|
// Availability-optimized writes can change EC while a drive
|
||||||
// is offline. Only latest-only non-strict reads may recount
|
// is offline. Only non-strict reads may recount these
|
||||||
// these ordinary null versions across separate modern EC groups.
|
// ordinary null versions across separate modern EC groups.
|
||||||
mixedECNull = !singleNull && !strict && requestedVersions == 1 &&
|
// Each stream holds just this version, so listing all
|
||||||
|
// versions (decommission, rebalance) must recount too.
|
||||||
|
mixedECNull = !singleNull && !strict &&
|
||||||
singleNullVersionStreams(originalVersions, true)
|
singleNullVersionStreams(originalVersions, true)
|
||||||
checkedSingleNull = true
|
checkedSingleNull = true
|
||||||
}
|
}
|
||||||
@@ -2289,6 +2302,7 @@ func (x xlMetaBuf) ListVersions(volume, path string, allParts bool) ([]FileInfo,
|
|||||||
}
|
}
|
||||||
|
|
||||||
// IsLatestDeleteMarker returns true if latest version is a deletemarker or there are no versions.
|
// IsLatestDeleteMarker returns true if latest version is a deletemarker or there are no versions.
|
||||||
|
// Free versions are not versions of the object and are skipped.
|
||||||
// If any error occurs false is returned.
|
// If any error occurs false is returned.
|
||||||
func (x xlMetaBuf) IsLatestDeleteMarker() bool {
|
func (x xlMetaBuf) IsLatestDeleteMarker() bool {
|
||||||
vers, headerV, _, buf, err := decodeXLHeaders(x)
|
vers, headerV, _, buf, err := decodeXLHeaders(x)
|
||||||
@@ -2298,13 +2312,17 @@ func (x xlMetaBuf) IsLatestDeleteMarker() bool {
|
|||||||
if vers == 0 {
|
if vers == 0 {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
isDeleteMarker := false
|
isDeleteMarker := true // only free versions
|
||||||
|
|
||||||
_ = decodeVersions(buf, vers, func(idx int, hdr, _ []byte) error {
|
_ = decodeVersions(buf, vers, func(idx int, hdr, _ []byte) error {
|
||||||
var xl xlMetaV2VersionHeader
|
var xl xlMetaV2VersionHeader
|
||||||
if _, err := xl.unmarshalV(headerV, hdr); err != nil {
|
if _, err := xl.unmarshalV(headerV, hdr); err != nil {
|
||||||
|
isDeleteMarker = false
|
||||||
return errDoneForNow
|
return errDoneForNow
|
||||||
}
|
}
|
||||||
|
if xl.FreeVersion() {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
isDeleteMarker = xl.Type == DeleteType
|
isDeleteMarker = xl.Type == DeleteType
|
||||||
return errDoneForNow
|
return errDoneForNow
|
||||||
})
|
})
|
||||||
@@ -2312,7 +2330,7 @@ func (x xlMetaBuf) IsLatestDeleteMarker() bool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// AllHidden returns true are no versions that would show up in a listing (ie all free markers)
|
// AllHidden returns true are no versions that would show up in a listing (ie all free markers)
|
||||||
// Optionally also return early if top is a delete marker.
|
// Optionally also return early if the top version that is not free is a delete marker.
|
||||||
func (x xlMetaBuf) AllHidden(topDeleteMarker bool) bool {
|
func (x xlMetaBuf) AllHidden(topDeleteMarker bool) bool {
|
||||||
vers, headerV, _, buf, err := decodeXLHeaders(x)
|
vers, headerV, _, buf, err := decodeXLHeaders(x)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -2328,16 +2346,12 @@ func (x xlMetaBuf) AllHidden(topDeleteMarker bool) bool {
|
|||||||
if _, err := xl.unmarshalV(headerV, hdr); err != nil {
|
if _, err := xl.unmarshalV(headerV, hdr); err != nil {
|
||||||
return errDoneForNow
|
return errDoneForNow
|
||||||
}
|
}
|
||||||
if topDeleteMarker && idx == 0 && xl.Type == DeleteType {
|
if xl.FreeVersion() {
|
||||||
hidden = true
|
|
||||||
return errDoneForNow
|
|
||||||
}
|
|
||||||
if !xl.FreeVersion() {
|
|
||||||
hidden = false
|
|
||||||
return errDoneForNow
|
|
||||||
}
|
|
||||||
// Check next version
|
// Check next version
|
||||||
return nil
|
return nil
|
||||||
|
}
|
||||||
|
hidden = topDeleteMarker && xl.Type == DeleteType
|
||||||
|
return errDoneForNow
|
||||||
})
|
})
|
||||||
return hidden
|
return hidden
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user