From 4315f934217d30d503c953c8a2fc16dc95cfbaf0 Mon Sep 17 00:00:00 2001 From: Harshavardhana Date: Sun, 17 Jan 2021 01:11:48 -0800 Subject: [PATCH] fix: make sure parentDirIsObject is used at set level (#11280) parentDirIsObject is not using set level understanding to check for parent objects, without this it can lead to objects that can actually reside on a separate set as objects and would conflict. --- cmd/erasure-multipart.go | 2 +- cmd/erasure-object.go | 2 +- cmd/erasure-sets.go | 6 ++++++ cmd/metacache-set.go | 5 +++-- cmd/object-api-interface.go | 20 ++++++++++---------- 5 files changed, 21 insertions(+), 14 deletions(-) diff --git a/cmd/erasure-multipart.go b/cmd/erasure-multipart.go index d4330663a..30fe8f3e6 100644 --- a/cmd/erasure-multipart.go +++ b/cmd/erasure-multipart.go @@ -690,7 +690,7 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str // Check if an object is present as one of the parent dir. // -- FIXME. (needs a new kind of lock). - if er.parentDirIsObject(ctx, bucket, path.Dir(object)) { + if opts.ParentIsObject != nil && opts.ParentIsObject(ctx, bucket, path.Dir(object)) { return oi, toObjectErr(errFileParentIsFile, bucket, object) } diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index 45127199f..6755c82cc 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -621,7 +621,7 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st // Check if an object is present as one of the parent dir. // -- FIXME. (needs a new kind of lock). // -- FIXME (this also causes performance issue when disks are down). - if er.parentDirIsObject(ctx, bucket, path.Dir(object)) { + if opts.ParentIsObject != nil && opts.ParentIsObject(ctx, bucket, path.Dir(object)) { return ObjectInfo{}, toObjectErr(errFileParentIsFile, bucket, object) } diff --git a/cmd/erasure-sets.go b/cmd/erasure-sets.go index 27ad739fa..3a3d2fb00 100644 --- a/cmd/erasure-sets.go +++ b/cmd/erasure-sets.go @@ -773,8 +773,13 @@ func (s *erasureSets) GetObject(ctx context.Context, bucket, object string, star return s.getHashedSet(object).GetObject(ctx, bucket, object, startOffset, length, writer, etag, opts) } +func (s *erasureSets) parentDirIsObject(ctx context.Context, bucket, parent string) bool { + return s.getHashedSet(parent).parentDirIsObject(ctx, bucket, parent) +} + // PutObject - writes an object to hashedSet based on the object name. func (s *erasureSets) PutObject(ctx context.Context, bucket string, object string, data *PutObjReader, opts ObjectOptions) (objInfo ObjectInfo, err error) { + opts.ParentIsObject = s.parentDirIsObject return s.getHashedSet(object).PutObject(ctx, bucket, object, data, opts) } @@ -1064,6 +1069,7 @@ func (s *erasureSets) AbortMultipartUpload(ctx context.Context, bucket, object, // CompleteMultipartUpload - completes a pending multipart transaction, on hashedSet based on object name. func (s *erasureSets) CompleteMultipartUpload(ctx context.Context, bucket, object, uploadID string, uploadedParts []CompletePart, opts ObjectOptions) (objInfo ObjectInfo, err error) { + opts.ParentIsObject = s.parentDirIsObject return s.getHashedSet(object).CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, opts) } diff --git a/cmd/metacache-set.go b/cmd/metacache-set.go index b75e7f13d..b64661c13 100644 --- a/cmd/metacache-set.go +++ b/cmd/metacache-set.go @@ -680,8 +680,9 @@ func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions) (entr logger.LogIf(ctx, err) custom := b.headerKV() _, err = er.putObject(ctx, minioMetaBucket, o.objectPath(b.n), NewPutObjReader(r, nil, nil), ObjectOptions{ - UserDefined: custom, - NoLock: true, // No need to hold namespace lock, each prefix caches uniquely. + UserDefined: custom, + NoLock: true, // No need to hold namespace lock, each prefix caches uniquely. + ParentIsObject: nil, }) if err != nil { metaMu.Lock() diff --git a/cmd/object-api-interface.go b/cmd/object-api-interface.go index 5709159e7..a7c0143fe 100644 --- a/cmd/object-api-interface.go +++ b/cmd/object-api-interface.go @@ -44,16 +44,16 @@ type ObjectOptions struct { MTime time.Time // Is only set in POST/PUT operations Expires time.Time // Is only used in POST/PUT operations - DeleteMarker bool // Is only set in DELETE operations for delete marker replication - UserDefined map[string]string // only set in case of POST/PUT operations - PartNumber int // only useful in case of GetObject/HeadObject - CheckPrecondFn CheckPreconditionFn // only set during GetObject/HeadObject/CopyObjectPart preconditional valuation - DeleteMarkerReplicationStatus string // Is only set in DELETE operations - VersionPurgeStatus VersionPurgeStatusType // Is only set in DELETE operations for delete marker version to be permanently deleted. - TransitionStatus string // status of the transition - NoLock bool // indicates to lower layers if the caller is expecting to hold locks. - ProxyRequest bool // only set for GET/HEAD in active-active replication scenario - + DeleteMarker bool // Is only set in DELETE operations for delete marker replication + UserDefined map[string]string // only set in case of POST/PUT operations + PartNumber int // only useful in case of GetObject/HeadObject + CheckPrecondFn CheckPreconditionFn // only set during GetObject/HeadObject/CopyObjectPart preconditional valuation + DeleteMarkerReplicationStatus string // Is only set in DELETE operations + VersionPurgeStatus VersionPurgeStatusType // Is only set in DELETE operations for delete marker version to be permanently deleted. + TransitionStatus string // status of the transition + NoLock bool // indicates to lower layers if the caller is expecting to hold locks. + ProxyRequest bool // only set for GET/HEAD in active-active replication scenario + ParentIsObject func(ctx context.Context, bucket, parent string) bool // Used to verify if parent is an object. } // BucketOptions represents bucket options for ObjectLayer bucket operations