worm: when enabled, avoid renaming the existing object in tmp directory (#6560)

In XL PutObject & CompleteMultipartUpload, the existing object is renamed
to the temporary directory before checking if worm is enabled or not.

Most of the times, this doesn't cause an issue unless two uploads to the
same location occurs at the same time. Since there is no locking in object
handlers, both uploads will reach XL layer. The second client acquiring
write lock in put object or complete upload in XL will rename the object
to the temporary directory before doing the check and returning the error (wrong!).

This commit fixes then the behavior: no rename to temporary directory if
worm is enabled.
master
Anis Elleuch 6 years ago committed by Harshavardhana
parent 307765591d
commit ea9408ccbb
  1. 12
      cmd/xl-v1-multipart.go
  2. 12
      cmd/xl-v1-object.go

@ -739,6 +739,11 @@ func (xl xlObjects) CompleteMultipartUpload(ctx context.Context, bucket string,
}
if xl.isObject(bucket, object) {
// Deny if WORM is enabled
if globalWORMEnabled {
return ObjectInfo{}, ObjectAlreadyExists{Bucket: bucket, Object: object}
}
// Rename if an object already exists to temporary location.
newUniqueID := mustGetUUID()
@ -767,13 +772,6 @@ func (xl xlObjects) CompleteMultipartUpload(ctx context.Context, bucket string,
}
}
// Deny if WORM is enabled
if globalWORMEnabled {
if xl.isObject(bucket, object) {
return ObjectInfo{}, ObjectAlreadyExists{Bucket: bucket, Object: object}
}
}
// Rename the multipart object to final location.
if _, err = renameObject(ctx, onlineDisks, minioMetaMultipartBucket, uploadIDPath, bucket, object, writeQuorum); err != nil {
return oi, toObjectErr(err, bucket, object)

@ -788,6 +788,11 @@ func (xl xlObjects) putObject(ctx context.Context, bucket string, object string,
}
if xl.isObject(bucket, object) {
// Deny if WORM is enabled
if globalWORMEnabled {
return ObjectInfo{}, ObjectAlreadyExists{Bucket: bucket, Object: object}
}
// Rename if an object already exists to temporary location.
newUniqueID := mustGetUUID()
@ -816,13 +821,6 @@ func (xl xlObjects) putObject(ctx context.Context, bucket string, object string,
return ObjectInfo{}, toObjectErr(err, bucket, object)
}
// Deny if WORM is enabled
if globalWORMEnabled {
if xl.isObject(bucket, object) {
return ObjectInfo{}, ObjectAlreadyExists{Bucket: bucket, Object: object}
}
}
// Rename the successfully written temporary object to final location.
if _, err = renameObject(ctx, onlineDisks, minioMetaTmpBucket, tempObj, bucket, object, writeQuorum); err != nil {
return ObjectInfo{}, toObjectErr(err, bucket, object)

Loading…
Cancel
Save