Persist in-memory replication stats to disk (#15594)

to avoid relying on scanner-calculated replication metrics.
This will improve the accuracy of the replication stats reported.

This PR also adds on to #15556 by handing replication
traffic that could not be queued by available workers to the 
MRF queue so that entries in `PENDING` status are healed faster.

cmd/bucket-replication.go

@@ -23,7 +23,6 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"math"
 	"net/http"
 	"path"
 	"reflect"
@@ -1369,6 +1368,7 @@ type ReplicationPool struct {
 	existingReplicaCh       chan ReplicateObjectInfo
 	existingReplicaDeleteCh chan DeletedObjectReplicationInfo
 	mrfSaveCh               chan MRFReplicateEntry
+	saveStateCh             chan struct{}
 
 	workerSize    int
 	mrfWorkerSize int
@@ -1391,6 +1391,7 @@ func NewReplicationPool(ctx context.Context, o ObjectLayer, opts replicationPool
 		existingReplicaDeleteCh: make(chan DeletedObjectReplicationInfo, 100000),
 		resyncState:             replicationResyncState{statusMap: make(map[string]BucketReplicationResyncStatus)},
 		mrfSaveCh:               make(chan MRFReplicateEntry, 100000),
+		saveStateCh:             make(chan struct{}, 1),
 		ctx:                     ctx,
 		objLayer:                o,
 	}
@@ -1401,6 +1402,7 @@ func NewReplicationPool(ctx context.Context, o ObjectLayer, opts replicationPool
 	go pool.updateResyncStatus(ctx, o)
 	go pool.processMRF()
 	go pool.persistMRF()
+	go pool.saveStatsToDisk()
 	return pool
 }
 
@@ -1526,10 +1528,12 @@ func (p *ReplicationPool) queueReplicaTask(ri ReplicateObjectInfo) {
 			close(p.replicaCh)
 			close(p.mrfReplicaCh)
 			close(p.existingReplicaCh)
+			close(p.saveStateCh)
 		})
 	case healCh <- ri:
 	case ch <- ri:
 	default:
+		globalReplicationPool.queueMRFSave(ri.ToMRFEntry())
 		logger.LogOnceIf(GlobalContext, fmt.Errorf("WARNING: Unable to keep up with incoming traffic - we recommend increasing number of replicate object workers with `mc admin config set api replication_workers=%d`", p.suggestedWorkers(false)), string(replicationSubsystem))
 	}
 }
@@ -1567,6 +1571,7 @@ func (p *ReplicationPool) queueReplicaDeleteTask(doi DeletedObjectReplicationInf
 		})
 	case ch <- doi:
 	default:
+		globalReplicationPool.queueMRFSave(doi.ToMRFEntry())
 		logger.LogOnceIf(GlobalContext, fmt.Errorf("WARNING: Unable to keep up with incoming deletes - we recommend increasing number of replicate workers with `mc admin config set api replication_workers=%d`", p.suggestedWorkers(false)), string(replicationSubsystem))
 	}
 }
@@ -1886,100 +1891,6 @@ func resyncTarget(oi ObjectInfo, arn string, resetID string, resetBeforeDate tim
 	return
 }
 
-func getAllLatestReplicationStats(bucketsUsage map[string]BucketUsageInfo) (bucketsReplicationStats map[string]BucketReplicationStats) {
-	peerBucketStatsList := globalNotificationSys.GetClusterAllBucketStats(GlobalContext)
-	bucketsReplicationStats = make(map[string]BucketReplicationStats, len(bucketsUsage))
-
-	for bucket, u := range bucketsUsage {
-		bucketStats := make([]BucketStats, len(peerBucketStatsList))
-		for i, peerBucketStats := range peerBucketStatsList {
-			bucketStat, ok := peerBucketStats[bucket]
-			if !ok {
-				continue
-			}
-			bucketStats[i] = bucketStat
-		}
-		bucketsReplicationStats[bucket] = calculateBucketReplicationStats(bucket, u, bucketStats)
-	}
-
-	return bucketsReplicationStats
-}
-
-func calculateBucketReplicationStats(bucket string, u BucketUsageInfo, bucketStats []BucketStats) (s BucketReplicationStats) {
-	// accumulate cluster bucket stats
-	stats := make(map[string]*BucketReplicationStat)
-	var totReplicaSize int64
-	for _, bucketStat := range bucketStats {
-		totReplicaSize += bucketStat.ReplicationStats.ReplicaSize
-		for arn, stat := range bucketStat.ReplicationStats.Stats {
-			oldst := stats[arn]
-			if oldst == nil {
-				oldst = &BucketReplicationStat{}
-			}
-			stats[arn] = &BucketReplicationStat{
-				FailedCount:    stat.FailedCount + oldst.FailedCount,
-				FailedSize:     stat.FailedSize + oldst.FailedSize,
-				ReplicatedSize: stat.ReplicatedSize + oldst.ReplicatedSize,
-				Latency:        stat.Latency.merge(oldst.Latency),
-			}
-		}
-	}
-
-	// add initial usage stat to cluster stats
-	usageStat := globalReplicationStats.GetInitialUsage(bucket)
-	totReplicaSize += usageStat.ReplicaSize
-	for arn, stat := range usageStat.Stats {
-		st, ok := stats[arn]
-		if !ok {
-			st = &BucketReplicationStat{}
-			stats[arn] = st
-		}
-		st.ReplicatedSize += stat.ReplicatedSize
-		st.FailedSize += stat.FailedSize
-		st.FailedCount += stat.FailedCount
-	}
-	s = BucketReplicationStats{
-		Stats: make(map[string]*BucketReplicationStat, len(stats)),
-	}
-	var latestTotReplicatedSize int64
-	for _, st := range u.ReplicationInfo {
-		latestTotReplicatedSize += int64(st.ReplicatedSize)
-	}
-	// normalize computed real time stats with latest usage stat
-	for arn, tgtstat := range stats {
-		st := BucketReplicationStat{}
-		bu, ok := u.ReplicationInfo[arn]
-		if !ok {
-			bu = BucketTargetUsageInfo{}
-		}
-		// use in memory replication stats if it is ahead of usage info.
-		st.ReplicatedSize = int64(bu.ReplicatedSize)
-		if tgtstat.ReplicatedSize >= int64(bu.ReplicatedSize) {
-			st.ReplicatedSize = tgtstat.ReplicatedSize
-		}
-		s.ReplicatedSize += st.ReplicatedSize
-		// Reset FailedSize and FailedCount to 0 for negative overflows which can
-		// happen since data usage picture can lag behind actual usage state at the time of cluster start
-		st.FailedSize = int64(math.Max(float64(tgtstat.FailedSize), 0))
-		st.FailedCount = int64(math.Max(float64(tgtstat.FailedCount), 0))
-		st.Latency = tgtstat.Latency
-
-		s.Stats[arn] = &st
-		s.FailedSize += st.FailedSize
-		s.FailedCount += st.FailedCount
-	}
-	// normalize overall stats
-	s.ReplicaSize = int64(math.Max(float64(totReplicaSize), float64(u.ReplicaSize)))
-	s.ReplicatedSize = int64(math.Max(float64(s.ReplicatedSize), float64(latestTotReplicatedSize)))
-	return s
-}
-
-// get the most current of in-memory replication stats  and data usage info from crawler.
-func getLatestReplicationStats(bucket string, u BucketUsageInfo) (s BucketReplicationStats) {
-	bucketStats := globalNotificationSys.GetClusterBucketStats(GlobalContext, bucket)
-	return calculateBucketReplicationStats(bucket, u, bucketStats)
-}
-
 const resyncTimeInterval = time.Minute * 1
 
 // updateResyncStatus persists in-memory resync metadata stats to disk at periodic intervals
@@ -2520,7 +2431,6 @@ func (p *ReplicationPool) persistMRF() {
 			logger.LogOnceIf(p.ctx, fmt.Errorf("Unable to persist replication failures to disk:%w", err), string(replicationSubsystem))
 		}
 		entries = make(map[string]MRFReplicateEntry)
-		return
 	}
 	for {
 		select {
@@ -2531,6 +2441,9 @@ func (p *ReplicationPool) persistMRF() {
 			close(p.mrfSaveCh)
 			saveMRFToDisk(true)
 			return
+		case <-p.saveStateCh:
+			saveMRFToDisk(true)
+			return
 		case e, ok := <-p.mrfSaveCh:
 			if !ok {
 				return
@@ -2683,3 +2596,91 @@ func (p *ReplicationPool) queueMRFHeal(file string) error {
 	}
 	return nil
 }
+
+// load replication stats from disk
+func (p *ReplicationPool) loadStatsFromDisk() (rs map[string]BucketReplicationStats, e error) {
+	data, err := readConfig(p.ctx, p.objLayer, getReplicationStatsPath(globalLocalNodeName))
+	if err != nil {
+		if !errors.Is(err, errConfigNotFound) {
+			return rs, nil
+		}
+		return rs, err
+	}
+
+	if len(data) <= 4 {
+		logger.LogIf(p.ctx, fmt.Errorf("replication stats: no data"))
+		return map[string]BucketReplicationStats{}, nil
+	}
+	// Read repl stats meta header
+	switch binary.LittleEndian.Uint16(data[0:2]) {
+	case replStatsMetaFormat:
+	default:
+		return rs, fmt.Errorf("replication stats: unknown format: %d", binary.LittleEndian.Uint16(data[0:2]))
+	}
+	switch binary.LittleEndian.Uint16(data[2:4]) {
+	case replStatsVersion:
+	default:
+		return rs, fmt.Errorf("replication stats: unknown version: %d", binary.LittleEndian.Uint16(data[2:4]))
+	}
+	ss := BucketStatsMap{}
+	if _, err = ss.UnmarshalMsg(data[4:]); err != nil {
+		return rs, err
+	}
+	rs = make(map[string]BucketReplicationStats, len(ss.Stats))
+	for bucket, st := range ss.Stats {
+		rs[bucket] = st.ReplicationStats
+	}
+
+	return rs, nil
+}
+
+func (p *ReplicationPool) saveStatsToDisk() {
+	if p == nil || p.objLayer == nil {
+		return
+	}
+	sTimer := time.NewTimer(replStatsSaveInterval)
+	defer sTimer.Stop()
+	for {
+		select {
+		case <-sTimer.C:
+			dui, err := loadDataUsageFromBackend(GlobalContext, newObjectLayerFn())
+			if err == nil && !dui.LastUpdate.IsZero() {
+				globalReplicationStats.getAllLatest(dui.BucketsUsage)
+			}
+			p.saveStats(p.ctx)
+			sTimer.Reset(replStatsSaveInterval)
+		case <-p.ctx.Done():
+			return
+		}
+	}
+}
+
+// save replication stats to .minio.sys/buckets/replication/node-name.stats
+func (p *ReplicationPool) saveStats(ctx context.Context) error {
+	bsm := globalReplicationStats.getAllCachedLatest()
+	if len(bsm.Stats) == 0 {
+		return nil
+	}
+	data := make([]byte, 4, 4+bsm.Msgsize())
+	// Add the replication stats meta header.
+	binary.LittleEndian.PutUint16(data[0:2], replStatsMetaFormat)
+	binary.LittleEndian.PutUint16(data[2:4], replStatsVersion)
+	// Add data
+	data, err := bsm.MarshalMsg(data)
+	if err != nil {
+		return err
+	}
+	return saveConfig(ctx, p.objLayer, getReplicationStatsPath(globalLocalNodeName), data)
+}
+
+// SaveState saves replication stats and mrf data before server restart
+func (p *ReplicationPool) SaveState(ctx context.Context) error {
+	go func() {
+		select {
+		case p.saveStateCh <- struct{}{}:
+		case <-p.ctx.Done():
+			return
+		}
+	}()
+	return p.saveStats(ctx)
+}