MyFavMirrors/minio
1
0
Fork 0
mirror of https://github.com/minio/minio.git synced 2024-12-24 06:05:55 -05:00
Code Issues Packages Projects Releases Wiki Activity

fix: resync of replication of delete markers (#12932)

Browse Source 
Fixes #12919
This commit is contained in:
Poorna Krishnamoorthy 2021-08-23 17:48:22 -04:00 committed by GitHub
parent db35bcf2ce
commit 674c6f7a7b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 27 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
cmd/bucket-replication.go
View File

@ -337,6 +337,17 @@ func replicateDelete(ctx context.Context, dobj DeletedObjectReplicationInfo, obj
} }
ctx = lkctx.Context() ctx = lkctx.Context()
defer lk.Unlock(lkctx.Cancel) defer lk.Unlock(lkctx.Cancel)
// early return if already replicated delete marker for existing object replication
if dobj.DeleteMarkerVersionID != "" && dobj.OpType == replication.ExistingObjectReplicationType {
_, err := tgt.StatObject(ctx, rcfg.GetDestination().Bucket, dobj.ObjectName, miniogo.StatObjectOptions{
VersionID: versionID,
Internal: miniogo.AdvancedGetOptions{
ReplicationProxyRequest: "false",
}})
if isErrMethodNotAllowed(ErrorRespToObjectError(err, dobj.Bucket, dobj.ObjectName)) {
return
}
}
rmErr := tgt.RemoveObject(ctx, rcfg.GetDestination().Bucket, dobj.ObjectName, miniogo.RemoveObjectOptions{ rmErr := tgt.RemoveObject(ctx, rcfg.GetDestination().Bucket, dobj.ObjectName, miniogo.RemoveObjectOptions{
VersionID: versionID, VersionID: versionID,
@ -766,7 +777,7 @@ func replicateObject(ctx context.Context, ri ReplicateObjectInfo, objectAPI Obje
if rtype == replicateNone { if rtype == replicateNone {
// object with same VersionID already exists, replication kicked off by // object with same VersionID already exists, replication kicked off by
// PutObject might have completed // PutObject might have completed
if objInfo.ReplicationStatus == replication.Pending || objInfo.ReplicationStatus == replication.Failed { if objInfo.ReplicationStatus == replication.Pending || objInfo.ReplicationStatus == replication.Failed || ri.OpType == replication.ExistingObjectReplicationType {
// if metadata is not updated for some reason after replication, such as // if metadata is not updated for some reason after replication, such as
// 503 encountered while updating metadata - make sure to set ReplicationStatus // 503 encountered while updating metadata - make sure to set ReplicationStatus
// as Completed. // as Completed.
@ -782,6 +793,9 @@ func replicateObject(ctx context.Context, ri ReplicateObjectInfo, objectAPI Obje
for k, v := range objInfo.UserDefined { for k, v := range objInfo.UserDefined {
popts.UserDefined[k] = v popts.UserDefined[k] = v
} }
if ri.OpType == replication.ExistingObjectReplicationType {
popts.UserDefined[xhttp.MinIOReplicationResetStatus] = fmt.Sprintf("%s;%s", UTCNow().Format(http.TimeFormat), ri.ResetID)
}
popts.UserDefined[xhttp.AmzBucketReplicationStatus] = replication.Completed.String() popts.UserDefined[xhttp.AmzBucketReplicationStatus] = replication.Completed.String()
if objInfo.UserTags != "" { if objInfo.UserTags != "" {
popts.UserDefined[xhttp.AmzObjectTagging] = objInfo.UserTags popts.UserDefined[xhttp.AmzObjectTagging] = objInfo.UserTags

2
cmd/data-scanner.go
View File

@ -1173,7 +1173,7 @@ func (i *scannerItem) healReplication(ctx context.Context, o ObjectLayer, oi Obj
return return
} }
// if replication status is Complete on DeleteMarker and existing object resync required // if replication status is Complete on DeleteMarker and existing object resync required
if existingObjResync && oi.ReplicationStatus == replication.Completed { if existingObjResync && (oi.ReplicationStatus == replication.Completed) {
i.healReplicationDeletes(ctx, o, oi, existingObjResync) i.healReplicationDeletes(ctx, o, oi, existingObjResync)
return return
} }

2
cmd/erasure-object.go
View File

@ -441,7 +441,7 @@ func (er erasureObjects) getObjectInfo(ctx context.Context, bucket, object strin
} }
objInfo = fi.ToObjectInfo(bucket, object) objInfo = fi.ToObjectInfo(bucket, object)
if !fi.VersionPurgeStatus.Empty() { if !fi.VersionPurgeStatus.Empty() && opts.VersionID != "" {
// Make sure to return object info to provide extra information. // Make sure to return object info to provide extra information.
return objInfo, toObjectErr(errMethodNotAllowed, bucket, object) return objInfo, toObjectErr(errMethodNotAllowed, bucket, object)
} }

4
cmd/gateway-common.go
View File

@ -333,6 +333,10 @@ func ErrorRespToObjectError(err error, params ...string) error {
err = PartTooSmall{} err = PartTooSmall{}
} }
switch minioErr.StatusCode {
case http.StatusMethodNotAllowed:
err = toObjectErr(errMethodNotAllowed, bucket, object)
}
return err return err
} }

6
cmd/object-api-errors.go
View File

@ -686,3 +686,9 @@ func isErrPreconditionFailed(err error) bool {
_, ok := err.(PreConditionFailed) _, ok := err.(PreConditionFailed)
return ok return ok
} }
// isErrMethodNotAllowed - Check if error type is MethodNotAllowed.
func isErrMethodNotAllowed(err error) bool {
var methodNotAllowed MethodNotAllowed
return errors.As(err, &methodNotAllowed)
}