Various improvements in replication (#11949)

- collect real time replication metrics for prometheus.
- add pending_count, failed_count metric for total pending/failed replication operations.

- add API to get replication metrics

- add MRF worker to handle spill-over replication operations

- multiple issues found with replication
- fixes an issue when client sends a bucket
 name with `/` at the end from SetRemoteTarget
 API call make sure to trim the bucket name to 
 avoid any extra `/`.

- hold write locks in GetObjectNInfo during replication
  to ensure that object version stack is not overwritten
  while reading the content.

- add additional protection during WriteMetadata() to
  ensure that we always write a valid FileInfo{} and avoid
  ever writing empty FileInfo{} to the lowest layers.

Co-authored-by: Poorna Krishnamoorthy <poorna@minio.io>
Co-authored-by: Harshavardhana <harsha@minio.io>

cmd/bucket-replication.go

@@ -291,6 +291,13 @@ func replicateDelete(ctx context.Context, dobj DeletedObjectVersionInfo, objectA
 			versionPurgeStatus = Complete
 		}
 	}
+	prevStatus := dobj.DeleteMarkerReplicationStatus
+	currStatus := replicationStatus
+	if dobj.VersionID != "" {
+		prevStatus = string(dobj.VersionPurgeStatus)
+		currStatus = string(versionPurgeStatus)
+	}
+	globalReplicationStats.Update(ctx, dobj.Bucket, 0, replication.StatusType(currStatus), replication.StatusType(prevStatus), replication.DeleteReplicationType) // to decrement pending count
 
 	var eventName = event.ObjectReplicationComplete
 	if replicationStatus == string(replication.Failed) || versionPurgeStatus == Failed {
@@ -594,7 +601,7 @@ func replicateObject(ctx context.Context, objInfo ObjectInfo, objectAPI ObjectLa
 		})
 		return
 	}
-	gr, err := objectAPI.GetObjectNInfo(ctx, bucket, object, nil, http.Header{}, readLock, ObjectOptions{
+	gr, err := objectAPI.GetObjectNInfo(ctx, bucket, object, nil, http.Header{}, writeLock, ObjectOptions{
 		VersionID: objInfo.VersionID,
 	})
 	if err != nil {
@@ -604,10 +611,10 @@ func replicateObject(ctx context.Context, objInfo ObjectInfo, objectAPI ObjectLa
 			Object:     objInfo,
 			Host:       "Internal: [Replication]",
 		})
-		logger.LogIf(ctx, err)
+		logger.LogIf(ctx, fmt.Errorf("Unable to update replicate for %s/%s(%s): %w", bucket, object, objInfo.VersionID, err))
 		return
 	}
-	defer gr.Close() // hold read lock for entire transaction
+	defer gr.Close() // hold write lock for entire transaction
 
 	objInfo = gr.ObjInfo
 	size, err := objInfo.GetActualSize()
@@ -644,7 +651,7 @@ func replicateObject(ctx context.Context, objInfo ObjectInfo, objectAPI ObjectLa
 		rtype = getReplicationAction(objInfo, oi)
 		if rtype == replicateNone {
 			// object with same VersionID already exists, replication kicked off by
-			// PutObject might have completed.
+			// PutObject might have completed
 			return
 		}
 	}
@@ -656,7 +663,8 @@ func replicateObject(ctx context.Context, objInfo ObjectInfo, objectAPI ObjectLa
 		srcOpts := miniogo.CopySrcOptions{
 			Bucket:    dest.Bucket,
 			Object:    object,
-			VersionID: objInfo.VersionID}
+			VersionID: objInfo.VersionID,
+		}
 		dstOpts := miniogo.PutObjectOptions{
 			Internal: miniogo.AdvancedPutOptions{
 				SourceVersionID:    objInfo.VersionID,
@@ -712,6 +720,7 @@ func replicateObject(ctx context.Context, objInfo ObjectInfo, objectAPI ObjectLa
 		defer r.Close()
 	}
 
+	prevReplStatus := objInfo.ReplicationStatus
 	objInfo.UserDefined[xhttp.AmzBucketReplicationStatus] = replicationStatus.String()
 	if objInfo.UserTags != "" {
 		objInfo.UserDefined[xhttp.AmzObjectTagging] = objInfo.UserTags
@@ -736,6 +745,11 @@ func replicateObject(ctx context.Context, objInfo ObjectInfo, objectAPI ObjectLa
 			logger.LogIf(ctx, fmt.Errorf("Unable to update replication metadata for %s/%s(%s): %w", bucket, objInfo.Name, objInfo.VersionID, err))
 		}
 	}
+	opType := replication.MetadataReplicationType
+	if rtype == replicateAll {
+		opType = replication.ObjectReplicationType
+	}
+	globalReplicationStats.Update(ctx, bucket, size, replicationStatus, prevReplStatus, opType)
 	sendEvent(eventArgs{
 		EventName:  eventName,
 		BucketName: bucket,
@@ -774,38 +788,61 @@ type DeletedObjectVersionInfo struct {
 }
 
 var (
-	globalReplicationPool *ReplicationPool
+	globalReplicationPool  *ReplicationPool
+	globalReplicationStats *ReplicationStats
 )
 
 // ReplicationPool describes replication pool
 type ReplicationPool struct {
-	mu              sync.Mutex
-	size            int
-	replicaCh       chan ObjectInfo
-	replicaDeleteCh chan DeletedObjectVersionInfo
-	killCh          chan struct{}
-	wg              sync.WaitGroup
-	ctx             context.Context
-	objLayer        ObjectLayer
+	mu                 sync.Mutex
+	size               int
+	replicaCh          chan ObjectInfo
+	replicaDeleteCh    chan DeletedObjectVersionInfo
+	mrfReplicaCh       chan ObjectInfo
+	mrfReplicaDeleteCh chan DeletedObjectVersionInfo
+	killCh             chan struct{}
+	wg                 sync.WaitGroup
+	ctx                context.Context
+	objLayer           ObjectLayer
 }
 
 // NewReplicationPool creates a pool of replication workers of specified size
 func NewReplicationPool(ctx context.Context, o ObjectLayer, sz int) *ReplicationPool {
 	pool := &ReplicationPool{
-		replicaCh:       make(chan ObjectInfo, 10000),
-		replicaDeleteCh: make(chan DeletedObjectVersionInfo, 10000),
-		ctx:             ctx,
-		objLayer:        o,
+		replicaCh:          make(chan ObjectInfo, 1000),
+		replicaDeleteCh:    make(chan DeletedObjectVersionInfo, 1000),
+		mrfReplicaCh:       make(chan ObjectInfo, 100000),
+		mrfReplicaDeleteCh: make(chan DeletedObjectVersionInfo, 100000),
+		ctx:                ctx,
+		objLayer:           o,
 	}
-	go func() {
-		<-ctx.Done()
-		close(pool.replicaCh)
-		close(pool.replicaDeleteCh)
-	}()
 	pool.Resize(sz)
+	// add long running worker for handling most recent failures/pending replications
+	go pool.AddMRFWorker()
 	return pool
 }
 
+// AddMRFWorker adds a pending/failed replication worker to handle requests that could not be queued
+// to the other workers
+func (p *ReplicationPool) AddMRFWorker() {
+	for {
+		select {
+		case <-p.ctx.Done():
+			return
+		case oi, ok := <-p.mrfReplicaCh:
+			if !ok {
+				return
+			}
+			replicateObject(p.ctx, oi, p.objLayer)
+		case doi, ok := <-p.mrfReplicaDeleteCh:
+			if !ok {
+				return
+			}
+			replicateDelete(p.ctx, doi, p.objLayer)
+		}
+	}
+}
+
 // AddWorker adds a replication worker to the pool
 func (p *ReplicationPool) AddWorker() {
 	defer p.wg.Done()
@@ -846,28 +883,39 @@ func (p *ReplicationPool) Resize(n int) {
 	}
 }
 
-func (p *ReplicationPool) queueReplicaTask(oi ObjectInfo) {
+func (p *ReplicationPool) queueReplicaTask(ctx context.Context, oi ObjectInfo) {
 	if p == nil {
 		return
 	}
 	select {
+	case <-ctx.Done():
+		close(p.replicaCh)
+		close(p.mrfReplicaCh)
 	case p.replicaCh <- oi:
+	case p.mrfReplicaCh <- oi:
+		// queue all overflows into the mrfReplicaCh to handle incoming pending/failed operations
 	default:
 	}
 }
 
-func (p *ReplicationPool) queueReplicaDeleteTask(doi DeletedObjectVersionInfo) {
+func (p *ReplicationPool) queueReplicaDeleteTask(ctx context.Context, doi DeletedObjectVersionInfo) {
 	if p == nil {
 		return
 	}
 	select {
+	case <-ctx.Done():
+		close(p.replicaDeleteCh)
+		close(p.mrfReplicaDeleteCh)
 	case p.replicaDeleteCh <- doi:
+	case p.mrfReplicaDeleteCh <- doi:
+		// queue all overflows into the mrfReplicaDeleteCh to handle incoming pending/failed operations
 	default:
 	}
 }
 
 func initBackgroundReplication(ctx context.Context, objectAPI ObjectLayer) {
 	globalReplicationPool = NewReplicationPool(ctx, objectAPI, globalAPIConfig.getReplicationWorkers())
+	globalReplicationStats = NewReplicationStats(ctx, objectAPI)
 }
 
 // get Reader from replication target if active-active replication is in place and
@@ -1003,11 +1051,14 @@ func proxyHeadToReplicationTarget(ctx context.Context, bucket, object string, op
 	return oi, proxy, err
 }
 
-func scheduleReplication(ctx context.Context, objInfo ObjectInfo, o ObjectLayer, sync bool) {
+func scheduleReplication(ctx context.Context, objInfo ObjectInfo, o ObjectLayer, sync bool, opType replication.Type) {
 	if sync {
 		replicateObject(ctx, objInfo, o)
 	} else {
-		globalReplicationPool.queueReplicaTask(objInfo)
+		globalReplicationPool.queueReplicaTask(GlobalContext, objInfo)
+	}
+	if sz, err := objInfo.GetActualSize(); err == nil {
+		globalReplicationStats.Update(ctx, objInfo.Bucket, sz, objInfo.ReplicationStatus, replication.StatusType(""), opType)
 	}
 }
 
@@ -1015,6 +1066,7 @@ func scheduleReplicationDelete(ctx context.Context, dv DeletedObjectVersionInfo,
 	if sync {
 		replicateDelete(ctx, dv, o)
 	} else {
-		globalReplicationPool.queueReplicaDeleteTask(dv)
+		globalReplicationPool.queueReplicaDeleteTask(GlobalContext, dv)
 	}
+	globalReplicationStats.Update(ctx, dv.Bucket, 0, replication.Pending, replication.StatusType(""), replication.DeleteReplicationType)
 }