fix: list across all drives to avoid stale disks

It was observed in VMware vsphere environment during a
pod replacement, `mc admin info` might report incorrect
offline nodes for the replaced drive. This issue eventually
goes away but requires quite a lot of time for all servers
to be in sync.

This PR fixes this behavior properly.

cmd/background-heal-ops.go

@@ -21,7 +21,6 @@ import (
 	"path"
 	"time"
 
-	"github.com/minio/minio/cmd/logger"
 	"github.com/minio/minio/pkg/madmin"
 )
 
@@ -129,24 +128,5 @@ func healDiskFormat(ctx context.Context, objAPI ObjectLayer, opts madmin.HealOpt
 		return madmin.HealResultItem{}, err
 	}
 
-	// Healing succeeded notify the peers to reload format and re-initialize disks.
-	// We will not notify peers if healing is not required.
-	if err == nil {
-		// Notify servers in background and retry if needed.
-		go func() {
-		retry:
-			for _, nerr := range globalNotificationSys.ReloadFormat(opts.DryRun) {
-				if nerr.Err != nil {
-					if nerr.Err.Error() == errServerNotInitialized.Error() {
-						time.Sleep(time.Second)
-						goto retry
-					}
-					logger.GetReqInfo(ctx).SetTags("peerAddress", nerr.Host.String())
-					logger.LogIf(ctx, nerr.Err)
-				}
-			}
-		}()
-	}
-
 	return res, nil
 }

cmd/background-newdisks-heal-ops.go

@@ -106,23 +106,26 @@ func monitorLocalDisksAndHeal(ctx context.Context, z *erasureZones, bgSeq *healS
 		case <-time.After(defaultMonitorNewDiskInterval):
 			waitForLowHTTPReq(int32(globalEndpoints.NEndpoints()), time.Second)
 
-			var erasureSetInZoneEndpointToHeal = make([]map[int]Endpoint, len(z.zones))
-			for i := range z.zones {
-				erasureSetInZoneEndpointToHeal[i] = map[int]Endpoint{}
-			}
-
+			var erasureSetInZoneEndpointToHeal []map[int]Endpoints
 			healDisks := globalBackgroundHealState.getHealLocalDisks()
-			// heal only if new disks found.
-			for _, endpoint := range healDisks {
-				logger.Info(fmt.Sprintf("Found drives to heal %d, proceeding to heal content...",
-					len(healDisks)))
-
+			if len(healDisks) > 0 {
 				// Reformat disks
 				bgSeq.sourceCh <- healSource{bucket: SlashSeparator}
 
 				// Ensure that reformatting disks is finished
 				bgSeq.sourceCh <- healSource{bucket: nopHeal}
 
+				logger.Info(fmt.Sprintf("Found drives to heal %d, proceeding to heal content...",
+					len(healDisks)))
+
+				erasureSetInZoneEndpointToHeal = make([]map[int]Endpoints, len(z.zones))
+				for i := range z.zones {
+					erasureSetInZoneEndpointToHeal[i] = map[int]Endpoints{}
+				}
+			}
+
+			// heal only if new disks found.
+			for _, endpoint := range healDisks {
 				// Load the new format of this passed endpoint
 				_, format, err := connectEndpoint(endpoint)
 				if err != nil {
@@ -142,20 +145,22 @@ func monitorLocalDisksAndHeal(ctx context.Context, z *erasureZones, bgSeq *healS
 					continue
 				}
 
-				erasureSetInZoneEndpointToHeal[zoneIdx][setIndex] = endpoint
+				erasureSetInZoneEndpointToHeal[zoneIdx][setIndex] = append(erasureSetInZoneEndpointToHeal[zoneIdx][setIndex], endpoint)
 			}
 
 			for i, setMap := range erasureSetInZoneEndpointToHeal {
-				for setIndex, endpoint := range setMap {
-					logger.Info("Healing disk '%s' on %s zone", endpoint, humanize.Ordinal(i+1))
+				for setIndex, endpoints := range setMap {
+					for _, ep := range endpoints {
+						logger.Info("Healing disk '%s' on %s zone", ep, humanize.Ordinal(i+1))
 
-					if err := healErasureSet(ctx, setIndex, z.zones[i].sets[setIndex], z.zones[i].setDriveCount); err != nil {
-						logger.LogIf(ctx, err)
-						continue
+						if err := healErasureSet(ctx, setIndex, z.zones[i].sets[setIndex], z.zones[i].setDriveCount); err != nil {
+							logger.LogIf(ctx, err)
+							continue
+						}
+
+						// Only upon success pop the healed disk.
+						globalBackgroundHealState.popHealLocalDisks(ep)
 					}
-
-					// Only upon success pop the healed disk.
-					globalBackgroundHealState.popHealLocalDisks(endpoint)
 				}
 			}
 		}

cmd/bucket-replication.go

@@ -170,24 +170,29 @@ func putReplicationOpts(dest replication.Destination, objInfo ObjectInfo) (putOp
 
 // replicateObject replicates the specified version of the object to destination bucket
 // The source object is then updated to reflect the replication status.
-func replicateObject(ctx context.Context, bucket, object, versionID string, objectAPI ObjectLayer, eventArg *eventArgs, healPending bool) {
+func replicateObject(ctx context.Context, objInfo ObjectInfo, objectAPI ObjectLayer) {
+	bucket := objInfo.Bucket
+	object := objInfo.Name
+
 	cfg, err := getReplicationConfig(ctx, bucket)
 	if err != nil {
 		logger.LogIf(ctx, err)
 		return
 	}
+
 	tgt := globalBucketTargetSys.GetReplicationTargetClient(ctx, cfg.RoleArn)
 	if tgt == nil {
 		return
 	}
+
 	gr, err := objectAPI.GetObjectNInfo(ctx, bucket, object, nil, http.Header{}, readLock, ObjectOptions{
-		VersionID: versionID,
+		VersionID: objInfo.VersionID,
 	})
 	if err != nil {
 		return
 	}
 
-	objInfo := gr.ObjInfo
+	objInfo = gr.ObjInfo
 	size, err := objInfo.GetActualSize()
 	if err != nil {
 		logger.LogIf(ctx, err)
@@ -200,6 +205,11 @@ func replicateObject(ctx context.Context, bucket, object, versionID string, obje
 		gr.Close()
 		return
 	}
+
+	// if heal encounters a pending replication status, either replication
+	// has failed due to server shutdown or crawler and PutObject replication are in contention.
+	healPending := objInfo.ReplicationStatus == replication.Pending
+
 	// In the rare event that replication is in pending state either due to
 	// server shut down/crash before replication completed or healing and PutObject
 	// race - do an additional stat to see if the version ID exists
@@ -219,22 +229,25 @@ func replicateObject(ctx context.Context, bucket, object, versionID string, obje
 	gr.Close()
 	if err != nil {
 		replicationStatus = replication.Failed
-		// Notify replication failure  event.
-		if eventArg == nil {
-			eventArg = &eventArgs{
-				BucketName: bucket,
-				Object:     objInfo,
-				Host:       "Internal: [Replication]",
-			}
-		}
-		eventArg.EventName = event.OperationReplicationFailed
-		eventArg.Object.UserDefined[xhttp.AmzBucketReplicationStatus] = replicationStatus.String()
-		sendEvent(*eventArg)
 	}
 	objInfo.UserDefined[xhttp.AmzBucketReplicationStatus] = replicationStatus.String()
 	if objInfo.UserTags != "" {
 		objInfo.UserDefined[xhttp.AmzObjectTagging] = objInfo.UserTags
 	}
+
+	// FIXME: add support for missing replication events
+	// - event.ObjectReplicationNotTracked
+	// - event.ObjectReplicationMissedThreshold
+	// - event.ObjectReplicationReplicatedAfterThreshold
+	if replicationStatus == replication.Failed {
+		sendEvent(eventArgs{
+			EventName:  event.ObjectReplicationFailed,
+			BucketName: bucket,
+			Object:     objInfo,
+			Host:       "Internal: [Replication]",
+		})
+	}
+
 	objInfo.metadataOnly = true // Perform only metadata updates.
 	if _, err = objectAPI.CopyObject(ctx, bucket, object, bucket, object, objInfo, ObjectOptions{
 		VersionID: objInfo.VersionID,
@@ -267,3 +280,42 @@ func filterReplicationStatusMetadata(metadata map[string]string) map[string]stri
 	delKey(xhttp.AmzBucketReplicationStatus)
 	return dst
 }
+
+type replicationState struct {
+	// add future metrics here
+	replicaCh chan ObjectInfo
+}
+
+func (r *replicationState) queueReplicaTask(oi ObjectInfo) {
+	select {
+	case r.replicaCh <- oi:
+	default:
+	}
+}
+
+var globalReplicationState *replicationState
+
+func newReplicationState() *replicationState {
+	return &replicationState{
+		// TODO: currently keeping it conservative
+		// but eventually can be tuned in future
+		replicaCh: make(chan ObjectInfo, 100),
+	}
+}
+
+func initBackgroundReplication(ctx context.Context, objectAPI ObjectLayer) {
+	if globalReplicationState == nil {
+		return
+	}
+	go func() {
+		defer close(globalReplicationState.replicaCh)
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case oi := <-globalReplicationState.replicaCh:
+				replicateObject(ctx, oi, objectAPI)
+			}
+		}
+	}()
+}

cmd/data-crawler.go

@@ -766,9 +766,6 @@ func sleepDuration(d time.Duration, x float64) {
 func (i *crawlItem) healReplication(ctx context.Context, o ObjectLayer, meta actionMeta) {
 	if meta.oi.ReplicationStatus == replication.Pending ||
 		meta.oi.ReplicationStatus == replication.Failed {
-		// if heal encounters a pending replication status, either replication
-		// has failed due to server shutdown or crawler and PutObject replication are in contention.
-		healPending := meta.oi.ReplicationStatus == replication.Pending
-		replicateObject(ctx, meta.oi.Bucket, meta.oi.Name, meta.oi.VersionID, o, nil, healPending)
+		globalReplicationState.queueReplicaTask(meta.oi)
 	}
 }

cmd/erasure-metadata.go

@@ -103,11 +103,17 @@ func (fi FileInfo) ToObjectInfo(bucket, object string) ObjectInfo {
 			IsDir:  true,
 		}
 	}
+
+	versionID := fi.VersionID
+	if globalBucketVersioningSys.Enabled(bucket) && versionID == "" {
+		versionID = nullVersionID
+	}
+
 	objInfo := ObjectInfo{
 		IsDir:           false,
 		Bucket:          bucket,
 		Name:            object,
-		VersionID:       fi.VersionID,
+		VersionID:       versionID,
 		IsLatest:        fi.IsLatest,
 		DeleteMarker:    fi.Deleted,
 		Size:            fi.Size,

cmd/erasure-sets.go

@@ -80,9 +80,6 @@ type erasureSets struct {
 
 	disksConnectEvent chan diskConnectInfo
 
-	// Done channel to control monitoring loop.
-	disksConnectDoneCh chan struct{}
-
 	// Distribution algorithm of choice.
 	distributionAlgo string
 	deploymentID     [16]byte
@@ -115,7 +112,7 @@ func (s *erasureSets) getDiskMap() map[string]StorageAPI {
 	for i := 0; i < s.setCount; i++ {
 		for j := 0; j < s.setDriveCount; j++ {
 			disk := s.erasureDisks[i][j]
-			if disk == nil {
+			if disk == OfflineDisk {
 				continue
 			}
 			if !disk.IsOnline() {
@@ -211,14 +208,16 @@ func (s *erasureSets) connectDisks() {
 			disk, format, err := connectEndpoint(endpoint)
 			if err != nil {
 				if endpoint.IsLocal && errors.Is(err, errUnformattedDisk) {
-					logger.Info(fmt.Sprintf("Found unformatted drive %s, attempting to heal...", endpoint))
 					globalBackgroundHealState.pushHealLocalDisks(endpoint)
+					logger.Info(fmt.Sprintf("Found unformatted drive %s, attempting to heal...", endpoint))
 				} else {
 					printEndpointError(endpoint, err, true)
 				}
 				return
 			}
+			s.erasureDisksMu.RLock()
 			setIndex, diskIndex, err := findDiskIndex(s.format, format)
+			s.erasureDisksMu.RUnlock()
 			if err != nil {
 				if endpoint.IsLocal {
 					globalBackgroundHealState.pushHealLocalDisks(endpoint)
@@ -256,8 +255,6 @@ func (s *erasureSets) monitorAndConnectEndpoints(ctx context.Context, monitorInt
 		select {
 		case <-ctx.Done():
 			return
-		case <-s.disksConnectDoneCh:
-			return
 		case <-time.After(monitorInterval):
 			s.connectDisks()
 		}
@@ -318,7 +315,6 @@ func newErasureSets(ctx context.Context, endpoints Endpoints, storageDisks []Sto
 		listTolerancePerSet: setDriveCount / 2,
 		format:              format,
 		disksConnectEvent:   make(chan diskConnectInfo),
-		disksConnectDoneCh:  make(chan struct{}),
 		distributionAlgo:    format.Erasure.DistributionAlgo,
 		deploymentID:        uuid.MustParse(format.ID),
 		pool:                NewMergeWalkPool(globalMergeLookupTimeout),
@@ -1191,16 +1187,12 @@ func (s *erasureSets) ReloadFormat(ctx context.Context, dryRun bool) (err error)
 		return err
 	}
 
-	// kill the monitoring loop such that we stop writing
-	// to indicate that we will re-initialize everything
-	// with new format.
-	s.disksConnectDoneCh <- struct{}{}
+	s.erasureDisksMu.Lock()
 
 	// Replace with new reference format.
 	s.format = refFormat
 
 	// Close all existing disks and reconnect all the disks.
-	s.erasureDisksMu.Lock()
 	for _, disk := range storageDisks {
 		if disk == nil {
 			continue
@@ -1223,10 +1215,8 @@ func (s *erasureSets) ReloadFormat(ctx context.Context, dryRun bool) (err error)
 		s.endpointStrings[m*s.setDriveCount+n] = disk.String()
 		s.erasureDisks[m][n] = disk
 	}
-	s.erasureDisksMu.Unlock()
 
-	// Restart monitoring loop to monitor reformatted disks again.
-	go s.monitorAndConnectEndpoints(GlobalContext, defaultMonitorConnectEndpointInterval)
+	s.erasureDisksMu.Unlock()
 
 	return nil
 }
@@ -1400,16 +1390,12 @@ func (s *erasureSets) HealFormat(ctx context.Context, dryRun bool) (res madmin.H
 			return madmin.HealResultItem{}, err
 		}
 
-		// kill the monitoring loop such that we stop writing
-		// to indicate that we will re-initialize everything
-		// with new format.
-		s.disksConnectDoneCh <- struct{}{}
+		s.erasureDisksMu.Lock()
 
 		// Replace with new reference format.
 		s.format = refFormat
 
 		// Disconnect/relinquish all existing disks, lockers and reconnect the disks, lockers.
-		s.erasureDisksMu.Lock()
 		for _, disk := range storageDisks {
 			if disk == nil {
 				continue
@@ -1432,10 +1418,8 @@ func (s *erasureSets) HealFormat(ctx context.Context, dryRun bool) (res madmin.H
 			s.endpointStrings[m*s.setDriveCount+n] = disk.String()
 			s.erasureDisks[m][n] = disk
 		}
-		s.erasureDisksMu.Unlock()
 
-		// Restart our monitoring loop to start monitoring newly formatted disks.
-		go s.monitorAndConnectEndpoints(GlobalContext, defaultMonitorConnectEndpointInterval)
+		s.erasureDisksMu.Unlock()
 	}
 
 	return res, nil

cmd/erasure-zones.go

@@ -18,6 +18,7 @@ package cmd
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"math/rand"
@@ -645,7 +646,7 @@ func (z *erasureZones) listObjectsNonSlash(ctx context.Context, bucket, prefix,
 
 	for _, zone := range z.zones {
 		zonesEntryChs = append(zonesEntryChs,
-			zone.startMergeWalksN(ctx, bucket, prefix, "", true, endWalkCh, zone.listTolerancePerSet, false))
+			zone.startMergeWalksN(ctx, bucket, prefix, "", true, endWalkCh, zone.setDriveCount, false))
 		zonesListTolerancePerSet = append(zonesListTolerancePerSet, zone.listTolerancePerSet)
 	}
 
@@ -764,7 +765,7 @@ func (z *erasureZones) listObjectsSplunk(ctx context.Context, bucket, prefix, ma
 		entryChs, endWalkCh := zone.poolSplunk.Release(listParams{bucket, recursive, marker, prefix})
 		if entryChs == nil {
 			endWalkCh = make(chan struct{})
-			entryChs = zone.startMergeWalksN(ctx, bucket, prefix, marker, recursive, endWalkCh, zone.listTolerancePerSet, true)
+			entryChs = zone.startMergeWalksN(ctx, bucket, prefix, marker, recursive, endWalkCh, zone.setDriveCount, true)
 		}
 		zonesEntryChs = append(zonesEntryChs, entryChs)
 		zonesEndWalkCh = append(zonesEndWalkCh, endWalkCh)
@@ -856,7 +857,7 @@ func (z *erasureZones) listObjects(ctx context.Context, bucket, prefix, marker,
 		entryChs, endWalkCh := zone.pool.Release(listParams{bucket, recursive, marker, prefix})
 		if entryChs == nil {
 			endWalkCh = make(chan struct{})
-			entryChs = zone.startMergeWalksN(ctx, bucket, prefix, marker, recursive, endWalkCh, zone.listTolerancePerSet+1, false)
+			entryChs = zone.startMergeWalksN(ctx, bucket, prefix, marker, recursive, endWalkCh, zone.setDriveCount, false)
 		}
 		zonesEntryChs = append(zonesEntryChs, entryChs)
 		zonesEndWalkCh = append(zonesEndWalkCh, endWalkCh)
@@ -1225,7 +1226,7 @@ func (z *erasureZones) listObjectVersions(ctx context.Context, bucket, prefix, m
 		entryChs, endWalkCh := zone.poolVersions.Release(listParams{bucket, recursive, marker, prefix})
 		if entryChs == nil {
 			endWalkCh = make(chan struct{})
-			entryChs = zone.startMergeWalksVersionsN(ctx, bucket, prefix, marker, recursive, endWalkCh, zone.listTolerancePerSet+1)
+			entryChs = zone.startMergeWalksVersionsN(ctx, bucket, prefix, marker, recursive, endWalkCh, zone.setDriveCount)
 		}
 		zonesEntryChs = append(zonesEntryChs, entryChs)
 		zonesEndWalkCh = append(zonesEndWalkCh, endWalkCh)
@@ -1608,13 +1609,8 @@ func (z *erasureZones) ListBuckets(ctx context.Context) (buckets []BucketInfo, e
 }
 
 func (z *erasureZones) ReloadFormat(ctx context.Context, dryRun bool) error {
-	// Acquire lock on format.json
-	formatLock := z.NewNSLock(ctx, minioMetaBucket, formatConfigFile)
-	if err := formatLock.GetRLock(globalOperationTimeout); err != nil {
-		return err
-	}
-	defer formatLock.RUnlock()
-
+	// No locks needed since reload happens in HealFormat under
+	// write lock across all nodes.
 	for _, zone := range z.zones {
 		if err := zone.ReloadFormat(ctx, dryRun); err != nil {
 			return err
@@ -1639,13 +1635,13 @@ func (z *erasureZones) HealFormat(ctx context.Context, dryRun bool) (madmin.Heal
 	var countNoHeal int
 	for _, zone := range z.zones {
 		result, err := zone.HealFormat(ctx, dryRun)
-		if err != nil && err != errNoHealRequired {
+		if err != nil && !errors.Is(err, errNoHealRequired) {
 			logger.LogIf(ctx, err)
 			continue
 		}
 		// Count errNoHealRequired across all zones,
 		// to return appropriate error to the caller
-		if err == errNoHealRequired {
+		if errors.Is(err, errNoHealRequired) {
 			countNoHeal++
 		}
 		r.DiskCount += result.DiskCount
@@ -1653,10 +1649,21 @@ func (z *erasureZones) HealFormat(ctx context.Context, dryRun bool) (madmin.Heal
 		r.Before.Drives = append(r.Before.Drives, result.Before.Drives...)
 		r.After.Drives = append(r.After.Drives, result.After.Drives...)
 	}
+
+	// Healing succeeded notify the peers to reload format and re-initialize disks.
+	// We will not notify peers if healing is not required.
+	for _, nerr := range globalNotificationSys.ReloadFormat(dryRun) {
+		if nerr.Err != nil {
+			logger.GetReqInfo(ctx).SetTags("peerAddress", nerr.Host.String())
+			logger.LogIf(ctx, nerr.Err)
+		}
+	}
+
 	// No heal returned by all zones, return errNoHealRequired
 	if countNoHeal == len(z.zones) {
 		return r, errNoHealRequired
 	}
+
 	return r, nil
 }
 

cmd/erasure.go

@@ -140,6 +140,7 @@ func getDisksInfo(disks []StorageAPI, endpoints []string) (disksInfo []madmin.Di
 		index := index
 		g.Go(func() error {
 			if disks[index] == OfflineDisk {
+				logger.LogIf(GlobalContext, fmt.Errorf("%s: %s", errDiskNotFound, endpoints[index]))
 				disksInfo[index] = madmin.Disk{
 					State:    diskErrToDriveState(errDiskNotFound),
 					Endpoint: endpoints[index],
@@ -149,11 +150,9 @@ func getDisksInfo(disks []StorageAPI, endpoints []string) (disksInfo []madmin.Di
 			}
 			info, err := disks[index].DiskInfo(context.TODO())
 			if err != nil {
-				if !IsErr(err, baseErrs...) {
-					reqInfo := (&logger.ReqInfo{}).AppendTags("disk", disks[index].String())
-					ctx := logger.SetReqInfo(GlobalContext, reqInfo)
-					logger.LogIf(ctx, err)
-				}
+				reqInfo := (&logger.ReqInfo{}).AppendTags("disk", disks[index].String())
+				ctx := logger.SetReqInfo(GlobalContext, reqInfo)
+				logger.LogIf(ctx, err)
 				disksInfo[index] = madmin.Disk{
 					State:    diskErrToDriveState(err),
 					Endpoint: endpoints[index],

cmd/object-handlers.go

@@ -1251,16 +1251,9 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 	response := generateCopyObjectResponse(objInfo.ETag, objInfo.ModTime)
 	encodedSuccessResponse := encodeResponse(response)
 	if mustReplicate(ctx, r, dstBucket, dstObject, objInfo.UserDefined, objInfo.ReplicationStatus.String()) {
-		go replicateObject(GlobalContext, dstBucket, dstObject, objInfo.VersionID, objectAPI, &eventArgs{
-			EventName:    event.ObjectCreatedCopy,
-			BucketName:   dstBucket,
-			Object:       objInfo,
-			ReqParams:    extractReqParams(r),
-			RespElements: extractRespElements(w),
-			UserAgent:    r.UserAgent(),
-			Host:         handlers.GetSourceIP(r),
-		}, false)
+		globalReplicationState.queueReplicaTask(objInfo)
 	}
+
 	setPutObjHeaders(w, objInfo, false)
 	// We must not use the http.Header().Set method here because some (broken)
 	// clients expect the x-amz-copy-source-version-id header key to be literally
@@ -1504,7 +1497,7 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
 		return
 	}
 	if mustReplicate(ctx, r, bucket, object, metadata, "") {
-		metadata[xhttp.AmzBucketReplicationStatus] = string(replication.Pending)
+		metadata[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
 	}
 	if r.Header.Get(xhttp.AmzBucketReplicationStatus) == replication.Replica.String() {
 		if s3Err = isPutActionAllowed(getRequestAuthType(r), bucket, object, r, iampolicy.ReplicateObjectAction); s3Err != ErrNone {
@@ -1567,15 +1560,7 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
 		}
 	}
 	if mustReplicate(ctx, r, bucket, object, metadata, "") {
-		go replicateObject(GlobalContext, bucket, object, objInfo.VersionID, objectAPI, &eventArgs{
-			EventName:    event.ObjectCreatedPut,
-			BucketName:   bucket,
-			Object:       objInfo,
-			ReqParams:    extractReqParams(r),
-			RespElements: extractRespElements(w),
-			UserAgent:    r.UserAgent(),
-			Host:         handlers.GetSourceIP(r),
-		}, false)
+		globalReplicationState.queueReplicaTask(objInfo)
 	}
 	setPutObjHeaders(w, objInfo, false)
 
@@ -1692,7 +1677,7 @@ func (api objectAPIHandlers) NewMultipartUploadHandler(w http.ResponseWriter, r
 		return
 	}
 	if mustReplicate(ctx, r, bucket, object, metadata, "") {
-		metadata[xhttp.AmzBucketReplicationStatus] = string(replication.Pending)
+		metadata[xhttp.AmzBucketReplicationStatus] = replication.Pending.String()
 	}
 	// We need to preserve the encryption headers set in EncryptRequest,
 	// so we do not want to override them, copy them instead.
@@ -2645,16 +2630,9 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite
 
 	setPutObjHeaders(w, objInfo, false)
 	if mustReplicate(ctx, r, bucket, object, objInfo.UserDefined, objInfo.ReplicationStatus.String()) {
-		go replicateObject(GlobalContext, bucket, object, objInfo.VersionID, objectAPI, &eventArgs{
-			EventName:    event.ObjectCreatedCompleteMultipartUpload,
-			BucketName:   bucket,
-			Object:       objInfo,
-			ReqParams:    extractReqParams(r),
-			RespElements: extractRespElements(w),
-			UserAgent:    r.UserAgent(),
-			Host:         handlers.GetSourceIP(r),
-		}, false)
+		globalReplicationState.queueReplicaTask(objInfo)
 	}
+
 	// Write success response.
 	writeSuccessResponseXML(w, encodedSuccessResponse)
 

cmd/peer-rest-client.go

@@ -481,11 +481,7 @@ func (client *peerRESTClient) DeleteBucketMetadata(bucket string) error {
 // ReloadFormat - reload format on the peer node.
 func (client *peerRESTClient) ReloadFormat(dryRun bool) error {
 	values := make(url.Values)
-	if dryRun {
-		values.Set(peerRESTDryRun, "true")
-	} else {
-		values.Set(peerRESTDryRun, "false")
-	}
+	values.Set(peerRESTDryRun, strconv.FormatBool(dryRun))
 
 	respBody, err := client.call(peerRESTMethodReloadFormat, values, nil, -1)
 	if err != nil {

cmd/rest/client.go

@@ -103,9 +103,6 @@ func (c *Client) Call(ctx context.Context, method string, values url.Values, bod
 	}
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url.String()+method+querySep+values.Encode(), body)
 	if err != nil {
-		if xnet.IsNetworkOrHostDown(err) {
-			c.MarkOffline()
-		}
 		return nil, &NetworkError{err}
 	}
 	req.Header.Set("Authorization", "Bearer "+c.newAuthToken(req.URL.Query().Encode()))
@@ -173,7 +170,6 @@ func NewClient(url *url.URL, newCustomTransport func() *http.Transport, newAuthT
 		url:                 url,
 		newAuthToken:        newAuthToken,
 		connected:           online,
-
 		MaxErrResponseSize:  4096,
 		HealthCheckInterval: 200 * time.Millisecond,
 		HealthCheckTimeout:  time.Second,
@@ -191,21 +187,18 @@ func (c *Client) MarkOffline() {
 	// Start goroutine that will attempt to reconnect.
 	// If server is already trying to reconnect this will have no effect.
 	if c.HealthCheckFn != nil && atomic.CompareAndSwapInt32(&c.connected, online, offline) {
-		if c.httpIdleConnsCloser != nil {
-			c.httpIdleConnsCloser()
-		}
-		go func() {
+		go func(healthFunc func() bool) {
 			ticker := time.NewTicker(c.HealthCheckInterval)
 			defer ticker.Stop()
 			for range ticker.C {
-				if status := atomic.LoadInt32(&c.connected); status == closed {
+				if atomic.LoadInt32(&c.connected) == closed {
 					return
 				}
-				if c.HealthCheckFn() {
+				if healthFunc() {
 					atomic.CompareAndSwapInt32(&c.connected, offline, online)
 					return
 				}
 			}
-		}()
+		}(c.HealthCheckFn)
 	}
 }

cmd/server-main.go

@@ -224,9 +224,10 @@ func initSafeMode(ctx context.Context, newObject ObjectLayer) (err error) {
 		}
 	}(txnLk)
 
-	// Enable healing to heal drives if possible
+	// Enable background operations for erasure coding
 	if globalIsErasure {
 		initAutoHeal(ctx, newObject)
+		initBackgroundReplication(ctx, newObject)
 	}
 
 	// allocate dynamic timeout once before the loop
@@ -444,6 +445,7 @@ func serverMain(ctx *cli.Context) {
 		// New global heal state
 		globalAllHealState = newHealState()
 		globalBackgroundHealState = newHealState()
+		globalReplicationState = newReplicationState()
 	}
 
 	// Initialize all sub-systems

cmd/storage-rest-server.go

@@ -106,6 +106,7 @@ func (s *storageRESTServer) IsValid(w http.ResponseWriter, r *http.Request) bool
 		s.writeErrorResponse(w, err)
 		return false
 	}
+
 	diskID := r.URL.Query().Get(storageRESTDiskID)
 	if diskID == "" {
 		// Request sent empty disk-id, we allow the request
@@ -113,6 +114,7 @@ func (s *storageRESTServer) IsValid(w http.ResponseWriter, r *http.Request) bool
 		// or create format.json
 		return true
 	}
+
 	storedDiskID, err := s.storage.GetDiskID()
 	if err != nil {
 		s.writeErrorResponse(w, err)

cmd/web-handlers.go

@@ -1166,16 +1166,9 @@ func (web *webAPIHandlers) Upload(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 	if mustReplicate {
-		go replicateObject(GlobalContext, bucket, object, objInfo.VersionID, objectAPI, &eventArgs{
-			EventName:    event.ObjectCreatedPut,
-			BucketName:   bucket,
-			Object:       objInfo,
-			ReqParams:    extractReqParams(r),
-			RespElements: extractRespElements(w),
-			UserAgent:    r.UserAgent(),
-			Host:         handlers.GetSourceIP(r),
-		}, false)
+		globalReplicationState.queueReplicaTask(objInfo)
 	}
+
 	// Notify object created event.
 	sendEvent(eventArgs{
 		EventName:    event.ObjectCreatedPut,

docs/orchestration/docker-compose/docker-compose.yaml

@@ -5,7 +5,7 @@ version: '3.7'
 # it through port 9000.
 services:
   minio1:
-    image: minio/minio:RELEASE.2020-09-10T22-02-45Z
+    image: minio/minio:RELEASE.2020-09-16T04-22-35Z
     volumes:
       - data1-1:/data1
       - data1-2:/data2
@@ -22,7 +22,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2020-09-10T22-02-45Z
+    image: minio/minio:RELEASE.2020-09-16T04-22-35Z
     volumes:
       - data2-1:/data1
       - data2-2:/data2
@@ -39,7 +39,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2020-09-10T22-02-45Z
+    image: minio/minio:RELEASE.2020-09-16T04-22-35Z
     volumes:
       - data3-1:/data1
       - data3-2:/data2
@@ -56,7 +56,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2020-09-10T22-02-45Z
+    image: minio/minio:RELEASE.2020-09-16T04-22-35Z
     volumes:
       - data4-1:/data1
       - data4-2:/data2

docs/orchestration/docker-swarm/docker-compose-secrets.yaml

@@ -2,7 +2,7 @@ version: '3.7'
 
 services:
   minio1:
-    image: minio/minio:RELEASE.2020-09-10T22-02-45Z
+    image: minio/minio:RELEASE.2020-09-16T04-22-35Z
     hostname: minio1
     volumes:
       - minio1-data:/export
@@ -29,7 +29,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2020-09-10T22-02-45Z
+    image: minio/minio:RELEASE.2020-09-16T04-22-35Z
     hostname: minio2
     volumes:
       - minio2-data:/export
@@ -56,7 +56,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2020-09-10T22-02-45Z
+    image: minio/minio:RELEASE.2020-09-16T04-22-35Z
     hostname: minio3
     volumes:
       - minio3-data:/export
@@ -83,7 +83,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2020-09-10T22-02-45Z
+    image: minio/minio:RELEASE.2020-09-16T04-22-35Z
     hostname: minio4
     volumes:
       - minio4-data:/export

docs/orchestration/docker-swarm/docker-compose.yaml

@@ -2,7 +2,7 @@ version: '3.7'
 
 services:
   minio1:
-    image: minio/minio:RELEASE.2020-09-10T22-02-45Z
+    image: minio/minio:RELEASE.2020-09-16T04-22-35Z
     hostname: minio1
     volumes:
       - minio1-data:/export
@@ -33,7 +33,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2020-09-10T22-02-45Z
+    image: minio/minio:RELEASE.2020-09-16T04-22-35Z
     hostname: minio2
     volumes:
       - minio2-data:/export
@@ -64,7 +64,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2020-09-10T22-02-45Z
+    image: minio/minio:RELEASE.2020-09-16T04-22-35Z
     hostname: minio3
     volumes:
       - minio3-data:/export
@@ -95,7 +95,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2020-09-10T22-02-45Z
+    image: minio/minio:RELEASE.2020-09-16T04-22-35Z
     hostname: minio4
     volumes:
       - minio4-data:/export

pkg/event/name.go

@@ -45,7 +45,11 @@ const (
 	ObjectRemovedDeleteMarkerCreated
 	BucketCreated
 	BucketRemoved
-	OperationReplicationFailed
+	ObjectReplicationAll
+	ObjectReplicationFailed
+	ObjectReplicationMissedThreshold
+	ObjectReplicationReplicatedAfterThreshold
+	ObjectReplicationNotTracked
 )
 
 // Expand - returns expanded values of abbreviated event type.
@@ -71,6 +75,13 @@ func (name Name) Expand() []Name {
 			ObjectRemovedDelete,
 			ObjectRemovedDeleteMarkerCreated,
 		}
+	case ObjectReplicationAll:
+		return []Name{
+			ObjectReplicationFailed,
+			ObjectReplicationNotTracked,
+			ObjectReplicationMissedThreshold,
+			ObjectReplicationReplicatedAfterThreshold,
+		}
 	default:
 		return []Name{name}
 	}
@@ -113,8 +124,14 @@ func (name Name) String() string {
 		return "s3:ObjectRemoved:Delete"
 	case ObjectRemovedDeleteMarkerCreated:
 		return "s3:ObjectRemoved:DeleteMarkerCreated"
-	case OperationReplicationFailed:
+	case ObjectReplicationFailed:
 		return "s3:Replication:OperationFailedReplication"
+	case ObjectReplicationNotTracked:
+		return "s3:Replication:OperationNotTracked"
+	case ObjectReplicationMissedThreshold:
+		return "s3:Replication:OperationMissedThreshold"
+	case ObjectReplicationReplicatedAfterThreshold:
+		return "s3:Replication:OperationReplicatedAfterThreshold"
 	}
 
 	return ""
@@ -199,8 +216,16 @@ func ParseName(s string) (Name, error) {
 		return ObjectRemovedDelete, nil
 	case "s3:ObjectRemoved:DeleteMarkerCreated":
 		return ObjectRemovedDeleteMarkerCreated, nil
+	case "s3:Replication:*":
+		return ObjectReplicationAll, nil
 	case "s3:Replication:OperationFailedReplication":
-		return OperationReplicationFailed, nil
+		return ObjectReplicationFailed, nil
+	case "s3:Replication:OperationMissedThreshold":
+		return ObjectReplicationMissedThreshold, nil
+	case "s3:Replication:OperationReplicatedAfterThreshold":
+		return ObjectReplicationReplicatedAfterThreshold, nil
+	case "s3:Replication:OperationNotTracked":
+		return ObjectReplicationNotTracked, nil
 	default:
 		return 0, &ErrInvalidEventName{s}
 	}