More fixes for delete marker replication (#11504)
continuation of PR#11491 for multiple server pools and bi-directional replication. Moving proxying for GET/HEAD to handler level rather than server pool layer as this was also causing incorrect proxying of HEAD. Also fixing metadata update on CopyObject - minio-go was not passing source version ID in X-Amz-Copy-Source header
This commit is contained in:
parent
466e95bb59
commit
e6b4ea7618
|
@ -184,6 +184,8 @@ func checkReplicateDelete(ctx context.Context, bucket string, dobj ObjectToDelet
|
||||||
if gerr != nil {
|
if gerr != nil {
|
||||||
validReplStatus := false
|
validReplStatus := false
|
||||||
switch oi.ReplicationStatus {
|
switch oi.ReplicationStatus {
|
||||||
|
// Point to note: even if two way replication with Deletemarker or Delete sync is enabled,
|
||||||
|
// deletion of a REPLICA object/deletemarker will not trigger a replication to the other cluster.
|
||||||
case replication.Pending, replication.Completed, replication.Failed:
|
case replication.Pending, replication.Completed, replication.Failed:
|
||||||
validReplStatus = true
|
validReplStatus = true
|
||||||
}
|
}
|
||||||
|
@ -648,9 +650,14 @@ func replicateObject(ctx context.Context, objInfo ObjectInfo, objectAPI ObjectLa
|
||||||
replicationStatus := replication.Completed
|
replicationStatus := replication.Completed
|
||||||
if rtype != replicateAll {
|
if rtype != replicateAll {
|
||||||
// replicate metadata for object tagging/copy with metadata replacement
|
// replicate metadata for object tagging/copy with metadata replacement
|
||||||
|
srcOpts := miniogo.CopySrcOptions{
|
||||||
|
Bucket: dest.Bucket,
|
||||||
|
Object: object,
|
||||||
|
VersionID: objInfo.VersionID}
|
||||||
dstOpts := miniogo.PutObjectOptions{Internal: miniogo.AdvancedPutOptions{SourceVersionID: objInfo.VersionID}}
|
dstOpts := miniogo.PutObjectOptions{Internal: miniogo.AdvancedPutOptions{SourceVersionID: objInfo.VersionID}}
|
||||||
c := &miniogo.Core{Client: tgt.Client}
|
c := &miniogo.Core{Client: tgt.Client}
|
||||||
if _, err = c.CopyObject(ctx, dest.Bucket, object, dest.Bucket, object, getCopyObjMetadata(objInfo, dest), dstOpts); err != nil {
|
|
||||||
|
if _, err = c.CopyObject(ctx, dest.Bucket, object, dest.Bucket, object, getCopyObjMetadata(objInfo, dest), srcOpts, dstOpts); err != nil {
|
||||||
replicationStatus = replication.Failed
|
replicationStatus = replication.Failed
|
||||||
logger.LogIf(ctx, fmt.Errorf("Unable to replicate metadata for object %s/%s(%s): %s", bucket, objInfo.Name, objInfo.VersionID, err))
|
logger.LogIf(ctx, fmt.Errorf("Unable to replicate metadata for object %s/%s(%s): %s", bucket, objInfo.Name, objInfo.VersionID, err))
|
||||||
}
|
}
|
||||||
|
@ -876,6 +883,7 @@ func proxyGetToReplicationTarget(ctx context.Context, bucket, object string, rs
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, false
|
return nil, false
|
||||||
}
|
}
|
||||||
|
reader.ObjInfo = oi.Clone()
|
||||||
return reader, true
|
return reader, true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -521,13 +521,6 @@ func (z *erasureServerPools) GetObjectNInfo(ctx context.Context, bucket, object
|
||||||
}
|
}
|
||||||
return gr, nil
|
return gr, nil
|
||||||
}
|
}
|
||||||
if isProxyable(ctx, bucket) {
|
|
||||||
// proxy to replication target if active-active replication is in place.
|
|
||||||
reader, proxy := proxyGetToReplicationTarget(ctx, bucket, object, rs, h, opts)
|
|
||||||
if reader != nil && proxy {
|
|
||||||
return reader, nil
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if opts.VersionID != "" {
|
if opts.VersionID != "" {
|
||||||
return gr, VersionNotFound{Bucket: bucket, Object: object, VersionID: opts.VersionID}
|
return gr, VersionNotFound{Bucket: bucket, Object: object, VersionID: opts.VersionID}
|
||||||
}
|
}
|
||||||
|
@ -573,13 +566,6 @@ func (z *erasureServerPools) GetObjectInfo(ctx context.Context, bucket, object s
|
||||||
return objInfo, nil
|
return objInfo, nil
|
||||||
}
|
}
|
||||||
object = decodeDirObject(object)
|
object = decodeDirObject(object)
|
||||||
// proxy HEAD to replication target if active-active replication configured on bucket
|
|
||||||
if isProxyable(ctx, bucket) {
|
|
||||||
oi, proxy, err := proxyHeadToReplicationTarget(ctx, bucket, object, opts)
|
|
||||||
if proxy {
|
|
||||||
return oi, err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if opts.VersionID != "" {
|
if opts.VersionID != "" {
|
||||||
return objInfo, VersionNotFound{Bucket: bucket, Object: object, VersionID: opts.VersionID}
|
return objInfo, VersionNotFound{Bucket: bucket, Object: object, VersionID: opts.VersionID}
|
||||||
}
|
}
|
||||||
|
|
|
@ -515,7 +515,7 @@ func (l *s3Objects) CopyObject(ctx context.Context, srcBucket string, srcObject
|
||||||
srcInfo.UserDefined[k] = v[0]
|
srcInfo.UserDefined[k] = v[0]
|
||||||
}
|
}
|
||||||
|
|
||||||
if _, err = l.Client.CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo.UserDefined, miniogo.PutObjectOptions{}); err != nil {
|
if _, err = l.Client.CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo.UserDefined, miniogo.CopySrcOptions{}, miniogo.PutObjectOptions{}); err != nil {
|
||||||
return objInfo, minio.ErrorRespToObjectError(err, srcBucket, srcObject)
|
return objInfo, minio.ErrorRespToObjectError(err, srcBucket, srcObject)
|
||||||
}
|
}
|
||||||
return l.GetObjectInfo(ctx, dstBucket, dstObject, dstOpts)
|
return l.GetObjectInfo(ctx, dstBucket, dstObject, dstOpts)
|
||||||
|
|
|
@ -411,19 +411,40 @@ func (api objectAPIHandlers) GetObjectHandler(w http.ResponseWriter, r *http.Req
|
||||||
|
|
||||||
gr, err := getObjectNInfo(ctx, bucket, object, rs, r.Header, readLock, opts)
|
gr, err := getObjectNInfo(ctx, bucket, object, rs, r.Header, readLock, opts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if isErrPreconditionFailed(err) {
|
var (
|
||||||
return
|
reader *GetObjectReader
|
||||||
}
|
proxy bool
|
||||||
if globalBucketVersioningSys.Enabled(bucket) && gr != nil {
|
)
|
||||||
// Versioning enabled quite possibly object is deleted might be delete-marker
|
if isProxyable(ctx, bucket) {
|
||||||
// if present set the headers, no idea why AWS S3 sets these headers.
|
// proxy to replication target if active-active replication is in place.
|
||||||
if gr.ObjInfo.VersionID != "" && gr.ObjInfo.DeleteMarker {
|
reader, proxy = proxyGetToReplicationTarget(ctx, bucket, object, rs, r.Header, opts)
|
||||||
w.Header()[xhttp.AmzVersionID] = []string{gr.ObjInfo.VersionID}
|
if reader != nil && proxy {
|
||||||
w.Header()[xhttp.AmzDeleteMarker] = []string{strconv.FormatBool(gr.ObjInfo.DeleteMarker)}
|
gr = reader
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
if reader == nil || !proxy {
|
||||||
return
|
if isErrPreconditionFailed(err) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if globalBucketVersioningSys.Enabled(bucket) && gr != nil {
|
||||||
|
if !gr.ObjInfo.VersionPurgeStatus.Empty() {
|
||||||
|
// Shows the replication status of a permanent delete of a version
|
||||||
|
w.Header()[xhttp.MinIODeleteReplicationStatus] = []string{string(gr.ObjInfo.VersionPurgeStatus)}
|
||||||
|
}
|
||||||
|
if !gr.ObjInfo.ReplicationStatus.Empty() && gr.ObjInfo.DeleteMarker {
|
||||||
|
w.Header()[xhttp.MinIODeleteMarkerReplicationStatus] = []string{string(gr.ObjInfo.ReplicationStatus)}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Versioning enabled quite possibly object is deleted might be delete-marker
|
||||||
|
// if present set the headers, no idea why AWS S3 sets these headers.
|
||||||
|
if gr.ObjInfo.VersionID != "" && gr.ObjInfo.DeleteMarker {
|
||||||
|
w.Header()[xhttp.AmzVersionID] = []string{gr.ObjInfo.VersionID}
|
||||||
|
w.Header()[xhttp.AmzDeleteMarker] = []string{strconv.FormatBool(gr.ObjInfo.DeleteMarker)}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
||||||
|
return
|
||||||
|
}
|
||||||
}
|
}
|
||||||
defer gr.Close()
|
defer gr.Close()
|
||||||
|
|
||||||
|
@ -577,23 +598,37 @@ func (api objectAPIHandlers) HeadObjectHandler(w http.ResponseWriter, r *http.Re
|
||||||
|
|
||||||
objInfo, err := getObjectInfo(ctx, bucket, object, opts)
|
objInfo, err := getObjectInfo(ctx, bucket, object, opts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if globalBucketVersioningSys.Enabled(bucket) {
|
var (
|
||||||
if !objInfo.VersionPurgeStatus.Empty() {
|
proxy bool
|
||||||
// Shows the replication status of a permanent delete of a version
|
perr error
|
||||||
w.Header()[xhttp.MinIODeleteReplicationStatus] = []string{string(objInfo.VersionPurgeStatus)}
|
oi ObjectInfo
|
||||||
}
|
)
|
||||||
if !objInfo.ReplicationStatus.Empty() && objInfo.DeleteMarker {
|
// proxy HEAD to replication target if active-active replication configured on bucket
|
||||||
w.Header()[xhttp.MinIODeleteMarkerReplicationStatus] = []string{string(objInfo.ReplicationStatus)}
|
if isProxyable(ctx, bucket) {
|
||||||
}
|
oi, proxy, perr = proxyHeadToReplicationTarget(ctx, bucket, object, opts)
|
||||||
// Versioning enabled quite possibly object is deleted might be delete-marker
|
if proxy && perr == nil {
|
||||||
// if present set the headers, no idea why AWS S3 sets these headers.
|
objInfo = oi
|
||||||
if objInfo.VersionID != "" && objInfo.DeleteMarker {
|
|
||||||
w.Header()[xhttp.AmzVersionID] = []string{objInfo.VersionID}
|
|
||||||
w.Header()[xhttp.AmzDeleteMarker] = []string{strconv.FormatBool(objInfo.DeleteMarker)}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
writeErrorResponseHeadersOnly(w, toAPIError(ctx, err))
|
if !proxy || perr != nil {
|
||||||
return
|
if globalBucketVersioningSys.Enabled(bucket) {
|
||||||
|
if !objInfo.VersionPurgeStatus.Empty() {
|
||||||
|
// Shows the replication status of a permanent delete of a version
|
||||||
|
w.Header()[xhttp.MinIODeleteReplicationStatus] = []string{string(objInfo.VersionPurgeStatus)}
|
||||||
|
}
|
||||||
|
if !objInfo.ReplicationStatus.Empty() && objInfo.DeleteMarker {
|
||||||
|
w.Header()[xhttp.MinIODeleteMarkerReplicationStatus] = []string{string(objInfo.ReplicationStatus)}
|
||||||
|
}
|
||||||
|
// Versioning enabled quite possibly object is deleted might be delete-marker
|
||||||
|
// if present set the headers, no idea why AWS S3 sets these headers.
|
||||||
|
if objInfo.VersionID != "" && objInfo.DeleteMarker {
|
||||||
|
w.Header()[xhttp.AmzVersionID] = []string{objInfo.VersionID}
|
||||||
|
w.Header()[xhttp.AmzDeleteMarker] = []string{strconv.FormatBool(objInfo.DeleteMarker)}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
writeErrorResponseHeadersOnly(w, toAPIError(ctx, err))
|
||||||
|
return
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Automatically remove the object/version is an expiry lifecycle rule can be applied
|
// Automatically remove the object/version is an expiry lifecycle rule can be applied
|
||||||
|
|
2
go.mod
2
go.mod
|
@ -49,7 +49,7 @@ require (
|
||||||
github.com/minio/cli v1.22.0
|
github.com/minio/cli v1.22.0
|
||||||
github.com/minio/highwayhash v1.0.1
|
github.com/minio/highwayhash v1.0.1
|
||||||
github.com/minio/md5-simd v1.1.1 // indirect
|
github.com/minio/md5-simd v1.1.1 // indirect
|
||||||
github.com/minio/minio-go/v7 v7.0.8
|
github.com/minio/minio-go/v7 v7.0.9-0.20210210235136-83423dddb072
|
||||||
github.com/minio/selfupdate v0.3.1
|
github.com/minio/selfupdate v0.3.1
|
||||||
github.com/minio/sha256-simd v0.1.1
|
github.com/minio/sha256-simd v0.1.1
|
||||||
github.com/minio/simdjson-go v0.2.1
|
github.com/minio/simdjson-go v0.2.1
|
||||||
|
|
2
go.sum
2
go.sum
|
@ -419,6 +419,8 @@ github.com/minio/md5-simd v1.1.1 h1:9ojcLbuZ4gXbB2sX53MKn8JUZ0sB/2wfwsEcRw+I08U=
|
||||||
github.com/minio/md5-simd v1.1.1/go.mod h1:XpBqgZULrMYD3R+M28PcmP0CkI7PEMzB3U77ZrKZ0Gw=
|
github.com/minio/md5-simd v1.1.1/go.mod h1:XpBqgZULrMYD3R+M28PcmP0CkI7PEMzB3U77ZrKZ0Gw=
|
||||||
github.com/minio/minio-go/v7 v7.0.8 h1:snnHtYkHz3TKrQJY1jTQGOZqnue79pbYTukuwqz/QvM=
|
github.com/minio/minio-go/v7 v7.0.8 h1:snnHtYkHz3TKrQJY1jTQGOZqnue79pbYTukuwqz/QvM=
|
||||||
github.com/minio/minio-go/v7 v7.0.8/go.mod h1:pEZBUa+L2m9oECoIA6IcSK8bv/qggtQVLovjeKK5jYc=
|
github.com/minio/minio-go/v7 v7.0.8/go.mod h1:pEZBUa+L2m9oECoIA6IcSK8bv/qggtQVLovjeKK5jYc=
|
||||||
|
github.com/minio/minio-go/v7 v7.0.9-0.20210210235136-83423dddb072 h1:zlheLAzZ66jYLUsa81R8gwPtSgKRI5FMJyAKuaJpkHE=
|
||||||
|
github.com/minio/minio-go/v7 v7.0.9-0.20210210235136-83423dddb072/go.mod h1:pEZBUa+L2m9oECoIA6IcSK8bv/qggtQVLovjeKK5jYc=
|
||||||
github.com/minio/selfupdate v0.3.1 h1:BWEFSNnrZVMUWXbXIgLDNDjbejkmpAmZvy/nCz1HlEs=
|
github.com/minio/selfupdate v0.3.1 h1:BWEFSNnrZVMUWXbXIgLDNDjbejkmpAmZvy/nCz1HlEs=
|
||||||
github.com/minio/selfupdate v0.3.1/go.mod h1:b8ThJzzH7u2MkF6PcIra7KaXO9Khf6alWPvMSyTDCFM=
|
github.com/minio/selfupdate v0.3.1/go.mod h1:b8ThJzzH7u2MkF6PcIra7KaXO9Khf6alWPvMSyTDCFM=
|
||||||
github.com/minio/sha256-simd v0.1.1 h1:5QHSlgo3nt5yKOJrC7W8w7X+NFl8cMPZm96iu8kKUJU=
|
github.com/minio/sha256-simd v0.1.1 h1:5QHSlgo3nt5yKOJrC7W8w7X+NFl8cMPZm96iu8kKUJU=
|
||||||
|
|
Loading…
Reference in a new issue