mirror of
https://github.com/minio/minio.git
synced 2024-12-24 06:05:55 -05:00
parent
b8f5a7db33
commit
a366143c5b
@ -40,7 +40,6 @@ import (
|
|||||||
"github.com/minio/minio/internal/hash"
|
"github.com/minio/minio/internal/hash"
|
||||||
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"
|
||||||
iampolicy "github.com/minio/pkg/iam/policy"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
const throttleDeadline = 1 * time.Hour
|
const throttleDeadline = 1 * time.Hour
|
||||||
@ -134,16 +133,7 @@ func getMustReplicateOptions(o ObjectInfo, op replication.Type) mustReplicateOpt
|
|||||||
|
|
||||||
// mustReplicate returns 2 booleans - true if object meets replication criteria and true if replication is to be done in
|
// mustReplicate returns 2 booleans - true if object meets replication criteria and true if replication is to be done in
|
||||||
// a synchronous manner.
|
// a synchronous manner.
|
||||||
func mustReplicate(ctx context.Context, r *http.Request, bucket, object string, opts mustReplicateOptions) (replicate bool, sync bool) {
|
func mustReplicate(ctx context.Context, bucket, object string, mopts mustReplicateOptions) (replicate bool, sync bool) {
|
||||||
if s3Err := isPutActionAllowed(ctx, getRequestAuthType(r), bucket, "", r, iampolicy.GetReplicationConfigurationAction); s3Err != ErrNone {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
return mustReplicater(ctx, bucket, object, opts)
|
|
||||||
}
|
|
||||||
|
|
||||||
// mustReplicater returns 2 booleans - true if object meets replication criteria and true if replication is to be done in
|
|
||||||
// a synchronous manner.
|
|
||||||
func mustReplicater(ctx context.Context, bucket, object string, mopts mustReplicateOptions) (replicate bool, sync bool) {
|
|
||||||
if globalIsGateway {
|
if globalIsGateway {
|
||||||
return replicate, sync
|
return replicate, sync
|
||||||
}
|
}
|
||||||
@ -1440,7 +1430,7 @@ func (c replicationConfig) Resync(ctx context.Context, oi ObjectInfo) bool {
|
|||||||
// Ignore previous replication status when deciding if object can be re-replicated
|
// Ignore previous replication status when deciding if object can be re-replicated
|
||||||
objInfo := oi.Clone()
|
objInfo := oi.Clone()
|
||||||
objInfo.ReplicationStatus = replication.StatusType("")
|
objInfo.ReplicationStatus = replication.StatusType("")
|
||||||
replicate, _ = mustReplicater(ctx, oi.Bucket, oi.Name, getMustReplicateOptions(objInfo, replication.ExistingObjectReplicationType))
|
replicate, _ = mustReplicate(ctx, oi.Bucket, oi.Name, getMustReplicateOptions(objInfo, replication.ExistingObjectReplicationType))
|
||||||
}
|
}
|
||||||
return c.resync(oi, replicate)
|
return c.resync(oi, replicate)
|
||||||
}
|
}
|
||||||
|
@ -1314,7 +1314,7 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
|
|||||||
if rs := r.Header.Get(xhttp.AmzBucketReplicationStatus); rs != "" {
|
if rs := r.Header.Get(xhttp.AmzBucketReplicationStatus); rs != "" {
|
||||||
srcInfo.UserDefined[xhttp.AmzBucketReplicationStatus] = rs
|
srcInfo.UserDefined[xhttp.AmzBucketReplicationStatus] = rs
|
||||||
}
|
}
|
||||||
if ok, _ := mustReplicate(ctx, r, dstBucket, dstObject, getMustReplicateOptions(srcInfo, replication.UnsetReplicationType)); ok {
|
if ok, _ := mustReplicate(ctx, dstBucket, dstObject, getMustReplicateOptions(srcInfo, replication.UnsetReplicationType)); ok {
|
||||||
srcInfo.UserDefined[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
|
srcInfo.UserDefined[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
|
||||||
}
|
}
|
||||||
// Store the preserved compression metadata.
|
// Store the preserved compression metadata.
|
||||||
@ -1422,7 +1422,7 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
|
|||||||
response := generateCopyObjectResponse(objInfo.ETag, objInfo.ModTime)
|
response := generateCopyObjectResponse(objInfo.ETag, objInfo.ModTime)
|
||||||
encodedSuccessResponse := encodeResponse(response)
|
encodedSuccessResponse := encodeResponse(response)
|
||||||
|
|
||||||
if replicate, sync := mustReplicate(ctx, r, dstBucket, dstObject, getMustReplicateOptions(objInfo, replication.UnsetReplicationType)); replicate {
|
if replicate, sync := mustReplicate(ctx, dstBucket, dstObject, getMustReplicateOptions(objInfo, replication.UnsetReplicationType)); replicate {
|
||||||
scheduleReplication(ctx, objInfo.Clone(), objectAPI, sync, replication.ObjectReplicationType)
|
scheduleReplication(ctx, objInfo.Clone(), objectAPI, sync, replication.ObjectReplicationType)
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -1657,7 +1657,7 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
|
|||||||
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
|
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if ok, _ := mustReplicate(ctx, r, bucket, object, getMustReplicateOptions(ObjectInfo{
|
if ok, _ := mustReplicate(ctx, bucket, object, getMustReplicateOptions(ObjectInfo{
|
||||||
UserDefined: metadata,
|
UserDefined: metadata,
|
||||||
}, replication.ObjectReplicationType)); ok {
|
}, replication.ObjectReplicationType)); ok {
|
||||||
metadata[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
|
metadata[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
|
||||||
@ -1752,7 +1752,7 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if replicate, sync := mustReplicate(ctx, r, bucket, object, getMustReplicateOptions(ObjectInfo{
|
if replicate, sync := mustReplicate(ctx, bucket, object, getMustReplicateOptions(ObjectInfo{
|
||||||
UserDefined: metadata,
|
UserDefined: metadata,
|
||||||
}, replication.ObjectReplicationType)); replicate {
|
}, replication.ObjectReplicationType)); replicate {
|
||||||
scheduleReplication(ctx, objInfo.Clone(), objectAPI, sync, replication.ObjectReplicationType)
|
scheduleReplication(ctx, objInfo.Clone(), objectAPI, sync, replication.ObjectReplicationType)
|
||||||
@ -1983,7 +1983,7 @@ func (api objectAPIHandlers) PutObjectExtractHandler(w http.ResponseWriter, r *h
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
if ok, _ := mustReplicate(ctx, r, bucket, object, getMustReplicateOptions(ObjectInfo{
|
if ok, _ := mustReplicate(ctx, bucket, object, getMustReplicateOptions(ObjectInfo{
|
||||||
UserDefined: metadata,
|
UserDefined: metadata,
|
||||||
}, replication.ObjectReplicationType)); ok {
|
}, replication.ObjectReplicationType)); ok {
|
||||||
metadata[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
|
metadata[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
|
||||||
@ -2041,7 +2041,7 @@ func (api objectAPIHandlers) PutObjectExtractHandler(w http.ResponseWriter, r *h
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
if replicate, sync := mustReplicate(ctx, r, bucket, object, getMustReplicateOptions(ObjectInfo{
|
if replicate, sync := mustReplicate(ctx, bucket, object, getMustReplicateOptions(ObjectInfo{
|
||||||
UserDefined: metadata,
|
UserDefined: metadata,
|
||||||
}, replication.ObjectReplicationType)); replicate {
|
}, replication.ObjectReplicationType)); replicate {
|
||||||
scheduleReplication(ctx, objInfo.Clone(), objectAPI, sync, replication.ObjectReplicationType)
|
scheduleReplication(ctx, objInfo.Clone(), objectAPI, sync, replication.ObjectReplicationType)
|
||||||
@ -2146,7 +2146,7 @@ func (api objectAPIHandlers) NewMultipartUploadHandler(w http.ResponseWriter, r
|
|||||||
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
|
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if ok, _ := mustReplicate(ctx, r, bucket, object, getMustReplicateOptions(ObjectInfo{
|
if ok, _ := mustReplicate(ctx, bucket, object, getMustReplicateOptions(ObjectInfo{
|
||||||
UserDefined: metadata,
|
UserDefined: metadata,
|
||||||
}, replication.ObjectReplicationType)); ok {
|
}, replication.ObjectReplicationType)); ok {
|
||||||
metadata[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
|
metadata[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
|
||||||
@ -3143,7 +3143,7 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite
|
|||||||
}
|
}
|
||||||
|
|
||||||
setPutObjHeaders(w, objInfo, false)
|
setPutObjHeaders(w, objInfo, false)
|
||||||
if replicate, sync := mustReplicate(ctx, r, bucket, object, getMustReplicateOptions(objInfo, replication.ObjectReplicationType)); replicate {
|
if replicate, sync := mustReplicate(ctx, bucket, object, getMustReplicateOptions(objInfo, replication.ObjectReplicationType)); replicate {
|
||||||
scheduleReplication(ctx, objInfo.Clone(), objectAPI, sync, replication.ObjectReplicationType)
|
scheduleReplication(ctx, objInfo.Clone(), objectAPI, sync, replication.ObjectReplicationType)
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -3410,7 +3410,7 @@ func (api objectAPIHandlers) PutObjectLegalHoldHandler(w http.ResponseWriter, r
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
objInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = strings.ToUpper(string(legalHold.Status))
|
objInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = strings.ToUpper(string(legalHold.Status))
|
||||||
replicate, sync := mustReplicate(ctx, r, bucket, object, getMustReplicateOptions(objInfo, replication.MetadataReplicationType))
|
replicate, sync := mustReplicate(ctx, bucket, object, getMustReplicateOptions(objInfo, replication.MetadataReplicationType))
|
||||||
if replicate {
|
if replicate {
|
||||||
objInfo.UserDefined[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
|
objInfo.UserDefined[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
|
||||||
}
|
}
|
||||||
@ -3589,7 +3589,7 @@ func (api objectAPIHandlers) PutObjectRetentionHandler(w http.ResponseWriter, r
|
|||||||
objInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] = ""
|
objInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] = ""
|
||||||
objInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = ""
|
objInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = ""
|
||||||
}
|
}
|
||||||
replicate, sync := mustReplicate(ctx, r, bucket, object, getMustReplicateOptions(objInfo, replication.MetadataReplicationType))
|
replicate, sync := mustReplicate(ctx, bucket, object, getMustReplicateOptions(objInfo, replication.MetadataReplicationType))
|
||||||
if replicate {
|
if replicate {
|
||||||
objInfo.UserDefined[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
|
objInfo.UserDefined[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
|
||||||
}
|
}
|
||||||
@ -3790,7 +3790,7 @@ func (api objectAPIHandlers) PutObjectTaggingHandler(w http.ResponseWriter, r *h
|
|||||||
|
|
||||||
oi := objInfo.Clone()
|
oi := objInfo.Clone()
|
||||||
oi.UserTags = tagsStr
|
oi.UserTags = tagsStr
|
||||||
replicate, sync := mustReplicate(ctx, r, bucket, object, getMustReplicateOptions(oi, replication.MetadataReplicationType))
|
replicate, sync := mustReplicate(ctx, bucket, object, getMustReplicateOptions(oi, replication.MetadataReplicationType))
|
||||||
if replicate {
|
if replicate {
|
||||||
opts.UserDefined = make(map[string]string)
|
opts.UserDefined = make(map[string]string)
|
||||||
opts.UserDefined[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
|
opts.UserDefined[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
|
||||||
@ -3865,7 +3865,7 @@ func (api objectAPIHandlers) DeleteObjectTaggingHandler(w http.ResponseWriter, r
|
|||||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
|
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
replicate, sync := mustReplicate(ctx, r, bucket, object, getMustReplicateOptions(oi, replication.MetadataReplicationType))
|
replicate, sync := mustReplicate(ctx, bucket, object, getMustReplicateOptions(oi, replication.MetadataReplicationType))
|
||||||
if replicate {
|
if replicate {
|
||||||
opts.UserDefined = make(map[string]string)
|
opts.UserDefined = make(map[string]string)
|
||||||
opts.UserDefined[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
|
opts.UserDefined[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
|
||||||
|
Loading…
Reference in New Issue
Block a user