Repository: minio/minio (language: go)

Issue:
fatal error: concurrent map iteration and map write on github.com/minio/minio/cmd.cloneMSS crashing MinIO
## NOTE
If this case is urgent, please subscribe to [Subnet](https://min.io/pricing) so that our 24/7 support team may help you faster.

<!--- Provide a general summary of the issue in the Title above -->

## Expected Behavior
<!--- If you're describing a bug, tell us what should happen -->
<!--- If you're suggesting a change/improvement, tell us how it should work -->
Not to crash

## Current Behavior
<!--- If describing a bug, tell us what happens instead of the expected behavior -->
<!--- If suggesting a change/improvement, explain the difference from current behavior -->
https://gist.github.com/jamie-34254/a0246a63dd0dd11fe064386ef64239e5

## Steps to Reproduce (for bugs)
<!--- Provide a link to a live example, or an unambiguous set of steps to -->
<!--- reproduce this bug. Include code to reproduce, if relevant -->
<!--- and make sure you have followed https://github.com/minio/minio/tree/release/docs/debugging to capture relevant logs -->
Not 100% sure

## Context
<!--- How has this issue affected you? What are you trying to accomplish? -->
<!--- Providing context helps us come up with a solution that is most useful in the real world -->
Causes a MinIO instance to crash causing cluster stability issues

## Regression
<!-- Is this issue a regression? (Yes / No) -->
<!-- If Yes, optionally please include minio version or commit id or PR# that caused this regression, if you have these details. -->
It didn't happen a few months ago but does happen after we upgraded to a newer release, can't remember which version we were on but I believe it was from 2020 so quite an old release.

## Your Environment
<!--- Include as many relevant details about the environment you experienced the bug in -->
* Version used (`minio --version`): minio version RELEASE.2022-04-26T01-20-24Z
* Server setup and configuration: 4 minio servers in a cluster with caching enabled
* Operating System and version (`uname -a`): `Linux minio-04 4.19.0-12-amd64 #1 SMP Debian 4.19.152-1 (2020-10-18) x86_64 GNU/Linux`



Discussion / hints:
Do not use disk caching with distributed setup, this is not supported.
However let's fix the crash.. 

```
Apr 30 21:16:53 minio-04 minio[15080]: fatal error: concurrent map iteration and map write
Apr 30 21:16:53 minio-04 minio[15080]: goroutine 739511348 [running]:
Apr 30 21:16:53 minio-04 minio[15080]: runtime.throw({0x2617dd8, 0x110})
Apr 30 21:16:53 minio-04 minio[15080]:         runtime/panic.go:1198 +0x71 fp=0xc00b5f7740 sp=0xc00b5f7710 pc=0x4370f1
Apr 30 21:16:53 minio-04 minio[15080]: runtime.mapiternext(0x2254de0)
Apr 30 21:16:53 minio-04 minio[15080]:         runtime/map.go:858 +0x4eb fp=0xc00b5f77b0 sp=0xc00b5f7740 pc=0x40fbeb
Apr 30 21:16:53 minio-04 minio[15080]: github.com/minio/minio/cmd.cloneMSS(0xc276c0b500)
Apr 30 21:16:53 minio-04 minio[15080]:         github.com/minio/minio/cmd/utils.go:129 +0x98 fp=0xc00b5f7858 sp=0xc00b5f77b0 pc=0x1f20698
Apr 30 21:16:53 minio-04 minio[15080]: github.com/minio/minio/cmd.(*diskCache).put(_, {_, _}, {_, _}, {_, _}, {_, _}, 0x24, ...)
Apr 30 21:16:53 minio-04 minio[15080]:         github.com/minio/minio/cmd/disk-cache-backend.go:889 +0x3b6 fp=0xc00b5f7b78 sp=0xc00b5f7858 pc=0x1cd5436
Apr 30 21:16:53 minio-04 minio[15080]: 
```

Fix patch:
diff --git a/cmd/disk-cache-backend.go b/cmd/disk-cache-backend.go
index 8d38a3d6fbddf..f93f77c401877 100644
--- a/cmd/disk-cache-backend.go
+++ b/cmd/disk-cache-backend.go
@@ -884,6 +884,7 @@ func (c *diskCache) put(ctx context.Context, bucket, object string, data io.Read
 	}
 
 	if err := os.MkdirAll(cachePath, 0o777); err != nil {
+		removeAll(cachePath)
 		return oi, err
 	}
 	metadata := cloneMSS(opts.UserDefined)
@@ -892,6 +893,7 @@ func (c *diskCache) put(ctx context.Context, bucket, object string, data io.Read
 	if globalCacheKMS != nil {
 		reader, err = newCacheEncryptReader(data, bucket, object, metadata)
 		if err != nil {
+			removeAll(cachePath)
 			return oi, err
 		}
 		actualSize, _ = sio.EncryptedSize(uint64(size))
diff --git a/cmd/disk-cache.go b/cmd/disk-cache.go
index 12c8112ec3629..095b6d406a4a8 100644
--- a/cmd/disk-cache.go
+++ b/cmd/disk-cache.go
@@ -33,7 +33,6 @@ import (
 	"github.com/minio/minio/internal/config/cache"
 	"github.com/minio/minio/internal/disk"
 	"github.com/minio/minio/internal/hash"
-	xhttp "github.com/minio/minio/internal/http"
 	"github.com/minio/minio/internal/logger"
 	"github.com/minio/minio/internal/sync/errgroup"
 	"github.com/minio/pkg/wildcard"
@@ -739,35 +738,30 @@ func (c *cacheObjects) PutObject(ctx context.Context, bucket, object string, r *
 	// Initialize pipe to stream data to cache
 	rPipe, wPipe := io.Pipe()
 	infoCh := make(chan ObjectInfo)
-	errorCh := make(chan error)
 	go func() {
+		defer close(infoCh)
 		info, err := putObjectFn(ctx, bucket, object, NewPutObjReader(hashReader), opts)
-		if err != nil {
-			close(infoCh)
-			pipeReader.CloseWithError(err)
-			rPipe.CloseWithError(err)
-			errorCh <- err
-			return
+		pipeReader.CloseWithError(err)
+		rPipe.CloseWithError(err)
+		if err == nil {
+			infoCh <- info
 		}
-		close(errorCh)
-		infoCh <- info
 	}()
 
 	go func() {
 		_, err := dcache.put(lkctx.Context(), bucket, object, rPipe, r.Size(), nil, opts, false, false)
 		if err != nil {
-			rPipe.CloseWithError(err)
-			return
+			logger.LogIf(lkctx.Context(), err)
 		}
+		// We do not care about errors to cached backend.
+		rPipe.Close()
 	}()
 
 	mwriter := cacheMultiWriter(pipeWriter, wPipe)
 	_, err = io.Copy(mwriter, r)
 	pipeWriter.Close()
 	wPipe.Close()
-
 	if err != nil {
-		err = <-errorCh
 		return ObjectInfo{}, err
 	}
 	info := <-infoCh
@@ -803,8 +797,7 @@ func (c *cacheObjects) uploadObject(ctx context.Context, oi ObjectInfo) {
 		return
 	}
 	var opts ObjectOptions
-	opts.UserDefined = make(map[string]string)
-	opts.UserDefined[xhttp.ContentMD5] = oi.UserDefined["content-md5"]
+	opts.UserDefined = cloneMSS(oi.UserDefined)
 	objInfo, err := c.InnerPutObjectFn(ctx, oi.Bucket, oi.Name, NewPutObjReader(hashReader), opts)
 	wbCommitStatus := CommitComplete
 	size := objInfo.Size
diff --git a/cmd/erasure-multipart.go b/cmd/erasure-multipart.go
index 43ce216dbc1ad..e2b104180cb59 100644
--- a/cmd/erasure-multipart.go
+++ b/cmd/erasure-multipart.go
@@ -286,8 +286,10 @@ func (er erasureObjects) ListMultipartUploads(ctx context.Context, bucket, objec
 // disks. `uploads.json` carries metadata regarding on-going multipart
 // operation(s) on the object.
 func (er erasureObjects) newMultipartUpload(ctx context.Context, bucket string, object string, opts ObjectOptions) (string, error) {
+	userDefined := cloneMSS(opts.UserDefined)
+
 	onlineDisks := er.getDisks()
-	parityDrives := globalStorageClass.GetParityForSC(opts.UserDefined[xhttp.AmzStorageClass])
+	parityDrives := globalStorageClass.GetParityForSC(userDefined[xhttp.AmzStorageClass])
 	if parityDrives <= 0 {
 		parityDrives = er.defaultParityCount
 	}
@@ -308,7 +310,7 @@ func (er erasureObjects) newMultipartUpload(ctx context.Context, bucket string,
 		}
 	}
 	if parityOrig != parityDrives {
-		opts.UserDefined[minIOErasureUpgraded] = strconv.Itoa(parityOrig) + "->" + strconv.Itoa(parityDrives)
+		userDefined[minIOErasureUpgraded] = strconv.Itoa(parityOrig) + "->" + strconv.Itoa(parityDrives)
 	}
 
 	dataDrives := len(onlineDisks) - parityDrives
@@ -336,8 +338,8 @@ func (er erasureObjects) newMultipartUpload(ctx context.Context, bucket string,
 	}
 
 	// Guess content-type from the extension if possible.
-	if opts.UserDefined["content-type"] == "" {
-		opts.UserDefined["content-type"] = mimedb.TypeByExtension(path.Ext(object))
+	if userDefined["content-type"] == "" {
+		userDefined["content-type"] = mimedb.TypeByExtension(path.Ext(object))
 	}
 
 	modTime := opts.MTime
@@ -352,7 +354,7 @@ func (er erasureObjects) newMultipartUpload(ctx context.Context, bucket string,
 	for index := range partsMetadata {
 		partsMetadata[index].Fresh = true
 		partsMetadata[index].ModTime = modTime
-		partsMetadata[index].Metadata = opts.UserDefined
+		partsMetadata[index].Metadata = userDefined
 	}
 
 	uploadID := mustGetUUID()
@@ -375,10 +377,6 @@ func (er erasureObjects) newMultipartUpload(ctx context.Context, bucket string,
 func (er erasureObjects) NewMultipartUpload(ctx context.Context, bucket, object string, opts ObjectOptions) (string, error) {
 	auditObjectErasureSet(ctx, object, &er)
 
-	// No metadata is set, allocate a new one.
-	if opts.UserDefined == nil {
-		opts.UserDefined = make(map[string]string)
-	}
 	return er.newMultipartUpload(ctx, bucket, object, opts)
 }
 
diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go
index 77704dee7e68c..7b5f6b51d09d8 100644
--- a/cmd/erasure-object.go
+++ b/cmd/erasure-object.go
@@ -842,17 +842,14 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 
 	data := r.Reader
 
-	// No metadata is set, allocate a new one.
-	if opts.UserDefined == nil {
-		opts.UserDefined = make(map[string]string)
-	}
+	userDefined := cloneMSS(opts.UserDefined)
 
 	storageDisks := er.getDisks()
 
 	parityDrives := len(storageDisks) / 2
 	if !opts.MaxParity {
 		// Get parity and data drive count based on storage class metadata
-		parityDrives = globalStorageClass.GetParityForSC(opts.UserDefined[xhttp.AmzStorageClass])
+		parityDrives = globalStorageClass.GetParityForSC(userDefined[xhttp.AmzStorageClass])
 		if parityDrives <= 0 {
 			parityDrives = er.defaultParityCount
 		}
@@ -890,7 +887,7 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 			parityDrives = len(storageDisks) / 2
 		}
 		if parityOrig != parityDrives {
-			opts.UserDefined[minIOErasureUpgraded] = strconv.Itoa(parityOrig) + "->" + strconv.Itoa(parityDrives)
+			userDefined[minIOErasureUpgraded] = strconv.Itoa(parityOrig) + "->" + strconv.Itoa(parityDrives)
 		}
 	}
 	dataDrives := len(storageDisks) - parityDrives
@@ -1066,13 +1063,13 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 			Hash:       bitrotWriterSum(w),
 		})
 	}
-	if opts.UserDefined["etag"] == "" {
-		opts.UserDefined["etag"] = r.MD5CurrentHexString()
+	if userDefined["etag"] == "" {
+		userDefined["etag"] = r.MD5CurrentHexString()
 	}
 
 	// Guess content-type from the extension if possible.
-	if opts.UserDefined["content-type"] == "" {
-		opts.UserDefined["content-type"] = mimedb.TypeByExtension(path.Ext(object))
+	if userDefined["content-type"] == "" {
+		userDefined["content-type"] = mimedb.TypeByExtension(path.Ext(object))
 	}
 
 	modTime := opts.MTime
@@ -1083,7 +1080,7 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 	// Fill all the necessary metadata.
 	// Update `xl.meta` content on each disks.
 	for index := range partsMetadata {
-		partsMetadata[index].Metadata = opts.UserDefined
+		partsMetadata[index].Metadata = userDefined
 		partsMetadata[index].Size = n
 		partsMetadata[index].ModTime = modTime
 	}
diff --git a/cmd/gateway/s3/gateway-s3-sse.go b/cmd/gateway/s3/gateway-s3-sse.go
index 2b4dcba0415cb..8d530ca892b5a 100644
--- a/cmd/gateway/s3/gateway-s3-sse.go
+++ b/cmd/gateway/s3/gateway-s3-sse.go
@@ -364,6 +364,7 @@ func (l *s3EncObjects) GetObjectInfo(ctx context.Context, bucket string, object
 // CopyObject copies an object from source bucket to a destination bucket.
 func (l *s3EncObjects) CopyObject(ctx context.Context, srcBucket string, srcObject string, dstBucket string, dstObject string, srcInfo minio.ObjectInfo, s, d minio.ObjectOptions) (objInfo minio.ObjectInfo, err error) {
 	cpSrcDstSame := path.Join(srcBucket, srcObject) == path.Join(dstBucket, dstObject)
+	userDefined := minio.CloneMSS(srcInfo.UserDefined)
 	if cpSrcDstSame {
 		var gwMeta gwMetaV1
 		if s.ServerSideEncryption != nil && d.ServerSideEncryption != nil &&
@@ -378,16 +379,16 @@ func (l *s3EncObjects) CopyObject(ctx context.Context, srcBucket string, srcObje
 				d.ServerSideEncryption.Marshal(header)
 			}
 			for k, v := range header {
-				srcInfo.UserDefined[k] = v[0]
+				userDefined[k] = v[0]
 			}
-			gwMeta.Meta = srcInfo.UserDefined
+			gwMeta.Meta = userDefined
 			if err = l.writeGWMetadata(ctx, dstBucket, getDareMetaPath(dstObject), gwMeta, minio.ObjectOptions{}); err != nil {
 				return objInfo, minio.ErrorRespToObjectError(err)
 			}
 			return gwMeta.ToObjectInfo(dstBucket, dstObject), nil
 		}
 	}
-	dstOpts := minio.ObjectOptions{ServerSideEncryption: d.ServerSideEncryption, UserDefined: srcInfo.UserDefined}
+	dstOpts := minio.ObjectOptions{ServerSideEncryption: d.ServerSideEncryption, UserDefined: userDefined}
 	return l.PutObject(ctx, dstBucket, dstObject, srcInfo.PutObjReader, dstOpts)
 }
 
diff --git a/cmd/gateway/s3/gateway-s3.go b/cmd/gateway/s3/gateway-s3.go
index 7951b0be1c6ae..40720fd319f5e 100644
--- a/cmd/gateway/s3/gateway-s3.go
+++ b/cmd/gateway/s3/gateway-s3.go
@@ -502,17 +502,20 @@ func (l *s3Objects) GetObjectInfo(ctx context.Context, bucket string, object str
 // PutObject creates a new object with the incoming data,
 func (l *s3Objects) PutObject(ctx context.Context, bucket string, object string, r *minio.PutObjReader, opts minio.ObjectOptions) (objInfo minio.ObjectInfo, err error) {
 	data := r.Reader
+
+	userDefined := minio.CloneMSS(opts.UserDefined)
+
 	var tagMap map[string]string
-	if tagstr, ok := opts.UserDefined[xhttp.AmzObjectTagging]; ok && tagstr != "" {
+	if tagstr, ok := userDefined[xhttp.AmzObjectTagging]; ok && tagstr != "" {
 		tagObj, err := tags.ParseObjectTags(tagstr)
 		if err != nil {
 			return objInfo, minio.ErrorRespToObjectError(err, bucket, object)
 		}
 		tagMap = tagObj.ToMap()
-		delete(opts.UserDefined, xhttp.AmzObjectTagging)
+		delete(userDefined, xhttp.AmzObjectTagging)
 	}
 	putOpts := miniogo.PutObjectOptions{
-		UserMetadata:         opts.UserDefined,
+		UserMetadata:         userDefined,
 		ServerSideEncryption: opts.ServerSideEncryption,
 		UserTags:             tagMap,
 		// Content-Md5 is needed for buckets with object locking,
@@ -529,7 +532,7 @@ func (l *s3Objects) PutObject(ctx context.Context, bucket string, object string,
 		ETag:     ui.ETag,
 		Size:     ui.Size,
 		Key:      object,
-		Metadata: minio.ToMinioClientObjectInfoMetadata(opts.UserDefined),
+		Metadata: minio.ToMinioClientObjectInfoMetadata(userDefined),
 	}
 
 	return minio.FromMinioClientObjectInfo(bucket, oi), nil
@@ -544,8 +547,9 @@ func (l *s3Objects) CopyObject(ctx context.Context, srcBucket string, srcObject
 	// metadata input is already a trickled down value from interpreting x-amz-metadata-directive at
 	// handler layer. So what we have right now is supposed to be applied on the destination object anyways.
 	// So preserve it by adding "REPLACE" directive to save all the metadata set by CopyObject API.
-	srcInfo.UserDefined["x-amz-metadata-directive"] = "REPLACE"
-	srcInfo.UserDefined["x-amz-copy-source-if-match"] = srcInfo.ETag
+	userDefined := minio.CloneMSS(srcInfo.UserDefined)
+	userDefined["x-amz-metadata-directive"] = "REPLACE"
+	userDefined["x-amz-copy-source-if-match"] = srcInfo.ETag
 	header := make(http.Header)
 	if srcOpts.ServerSideEncryption != nil {
 		encrypt.SSECopy(srcOpts.ServerSideEncryption).Marshal(header)
@@ -556,10 +560,10 @@ func (l *s3Objects) CopyObject(ctx context.Context, srcBucket string, srcObject
 	}
 
 	for k, v := range header {
-		srcInfo.UserDefined[k] = v[0]
+		userDefined[k] = v[0]
 	}
 
-	if _, err = l.Client.CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo.UserDefined, miniogo.CopySrcOptions{}, miniogo.PutObjectOptions{}); err != nil {
+	if _, err = l.Client.CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, userDefined, miniogo.CopySrcOptions{}, miniogo.PutObjectOptions{}); err != nil {
 		return objInfo, minio.ErrorRespToObjectError(err, srcBucket, srcObject)
 	}
 	return l.GetObjectInfo(ctx, dstBucket, dstObject, dstOpts)
@@ -605,17 +609,18 @@ func (l *s3Objects) ListMultipartUploads(ctx context.Context, bucket string, pre
 // NewMultipartUpload upload object in multiple parts
 func (l *s3Objects) NewMultipartUpload(ctx context.Context, bucket string, object string, o minio.ObjectOptions) (uploadID string, err error) {
 	var tagMap map[string]string
-	if tagStr, ok := o.UserDefined[xhttp.AmzObjectTagging]; ok {
+	userDefined := minio.CloneMSS(o.UserDefined)
+	if tagStr, ok := userDefined[xhttp.AmzObjectTagging]; ok {
 		tagObj, err := tags.Parse(tagStr, true)
 		if err != nil {
 			return uploadID, minio.ErrorRespToObjectError(err, bucket, object)
 		}
 		tagMap = tagObj.ToMap()
-		delete(o.UserDefined, xhttp.AmzObjectTagging)
+		delete(userDefined, xhttp.AmzObjectTagging)
 	}
 	// Create PutObject options
 	opts := miniogo.PutObjectOptions{
-		UserMetadata:         o.UserDefined,
+		UserMetadata:         userDefined,
 		ServerSideEncryption: o.ServerSideEncryption,
 		UserTags:             tagMap,
 	}
@@ -645,9 +650,8 @@ func (l *s3Objects) CopyObjectPart(ctx context.Context, srcBucket, srcObject, de
 	if srcOpts.CheckPrecondFn != nil && srcOpts.CheckPrecondFn(srcInfo) {
 		return minio.PartInfo{}, minio.PreConditionFailed{}
 	}
-	srcInfo.UserDefined = map[string]string{
-		"x-amz-copy-source-if-match": srcInfo.ETag,
-	}
+	userDefined := minio.CloneMSS(srcInfo.UserDefined)
+	userDefined["x-amz-copy-source-if-match"] = srcInfo.ETag
 	header := make(http.Header)
 	if srcOpts.ServerSideEncryption != nil {
 		encrypt.SSECopy(srcOpts.ServerSideEncryption).Marshal(header)
@@ -657,11 +661,11 @@ func (l *s3Objects) CopyObjectPart(ctx context.Context, srcBucket, srcObject, de
 		dstOpts.ServerSideEncryption.Marshal(header)
 	}
 	for k, v := range header {
-		srcInfo.UserDefined[k] = v[0]
+		userDefined[k] = v[0]
 	}
 
 	completePart, err := l.Client.CopyObjectPart(ctx, srcBucket, srcObject, destBucket, destObject,
-		uploadID, partID, startOffset, length, srcInfo.UserDefined)
+		uploadID, partID, startOffset, length, userDefined)
 	if err != nil {
 		return p, minio.ErrorRespToObjectError(err, srcBucket, srcObject)
 	}
diff --git a/cmd/utils.go b/cmd/utils.go
index 02a1e3b468986..5be2e14cbf321 100644
--- a/cmd/utils.go
+++ b/cmd/utils.go
@@ -130,6 +130,9 @@ func getWriteQuorum(drive int) int {
 	return quorum
 }
 
+// CloneMSS is an exposed function of cloneMSS for gateway usage.
+var CloneMSS = cloneMSS
+
 // cloneMSS will clone a map[string]string.
 // If input is nil an empty map is returned, not nil.
 func cloneMSS(v map[string]string) map[string]string {


Test patch: