feat(clustermgr): fresh all volume every n-tick

. #1000995387

Signed-off-by: slasher <shenjie1@oppo.com>

blobstore/clustermgr/dashboard.go

@@ -32,6 +32,7 @@ import (
 	"github.com/cubefs/cubefs/blobstore/common/proto"
 	"github.com/cubefs/cubefs/blobstore/common/rpc"
 	"github.com/cubefs/cubefs/blobstore/common/trace"
+	"github.com/cubefs/cubefs/blobstore/util"
 	"github.com/cubefs/cubefs/blobstore/util/defaulter"
 )
 
@@ -102,9 +103,12 @@ type dashboardMgr struct {
 	snapshot atomic.Value // stores *dashboardSnapshot
 
 	// volumes is a persistent cache of VolumeBasic keyed by Vid.
-	// RangeUpdateVolume updates entries in-place on each fresh()
-	volumes map[proto.Vid]*clustermgr.VolumeBasic
-	volMu   sync.RWMutex
+	// RangeUpdateVolume updates entries in-place
+	// only called DashboardFreshVolumeTick reached
+	volumes    map[proto.Vid]*clustermgr.VolumeBasic
+	volMu      sync.RWMutex
+	freshTick  uint64
+	lastVolume clustermgr.VolumeStat
 
 	freshCh chan struct{} // buffered(1); loop listens for force-fresh signals
 
@@ -227,22 +231,24 @@ func (d *dashboardMgr) Simulate(nodes []string) clustermgr.ClusterDashboard {
 
 func (d *dashboardMgr) loopFresh() {
 	defaulter.LessOrEqual(&d.service.DashboardFreshIntervalS, 60)
+	defaulter.IntegerLessOrEqual(&d.service.DashboardFreshVolumeTick, 5)
 
 	interval := time.Duration(d.service.DashboardFreshIntervalS) * time.Second
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
 
 	for {
+		_, ctx := trace.StartSpanFromContext(context.Background(), "")
 		select {
 		case <-ticker.C:
-			d.fresh()
+			d.fresh(ctx, false)
 			select {
 			case <-d.freshCh:
 			default:
 			}
 
 		case <-d.freshCh:
-			d.fresh()
+			d.fresh(ctx, true)
 			ticker.Reset(interval)
 
 		case <-d.service.closeCh:
@@ -251,13 +257,13 @@ func (d *dashboardMgr) loopFresh() {
 	}
 }
 
-func (d *dashboardMgr) fresh() {
+func (d *dashboardMgr) fresh(ctx context.Context, force bool) {
 	now := time.Now()
 
 	scope := buildScope(d.service.ScopeMgr.Stat())
 	services, _ := d.service.ServiceMgr.ListServiceInfo()
 	allDisks, expiredDisks := d.blobnode.DisksSnapshot()
-	allNodes := d.blobnode.ListNodes(context.Background(), proto.NodeStatusInvalid)
+	allNodes := d.blobnode.ListNodes(ctx, proto.NodeStatusInvalid)
 	droppedNodes := make(map[proto.NodeID]struct{})
 	nodeHost := make(map[proto.NodeID]string, len(allNodes))
 	for _, n := range allNodes {
@@ -282,11 +288,15 @@ func (d *dashboardMgr) fresh() {
 	}
 
 	d.volMu.Lock()
-	d.service.VolumeMgr.RangeUpdateVolume(context.Background(), d.volumes)
-	volume := buildVolume(d.volumes,
-		d.service.VolumeMgr.AllocatableSize,
-		d.service.VolumeMgr.RetainThreshold,
-		d.service.VolumeMgr.AllocatableDiskLoadThreshold)
+	if force || d.freshTick%uint64(util.Max(1, d.service.DashboardFreshVolumeTick)) == 0 {
+		d.service.VolumeMgr.RangeUpdateVolume(ctx, d.volumes)
+		d.lastVolume = buildVolume(d.volumes,
+			d.service.VolumeMgr.AllocatableSize,
+			d.service.VolumeMgr.RetainThreshold,
+			d.service.VolumeMgr.AllocatableDiskLoadThreshold)
+	}
+	d.freshTick++
+	volume := d.lastVolume
 	safety := buildVolumeSafety(d.volumes, unsafeDiskSet)
 	d.volMu.Unlock()
 

blobstore/clustermgr/dashboard_test.go

@@ -132,7 +132,7 @@ func TestDashboardFresh_UpdatesSnapshotAndNotifiesWaiters(t *testing.T) {
 	}
 
 	t1 := time.Now()
-	d.fresh()
+	d.fresh(context.Background(), true)
 
 	select {
 	case <-ch:
@@ -157,7 +157,7 @@ func TestDashboardFresh_NoPanicWhenNoWaiter(t *testing.T) {
 	svc, cleanup := initTestService(t)
 	defer cleanup()
 	require.NotPanics(t, func() {
-		svc.dashboardMgr.fresh()
+		svc.dashboardMgr.fresh(context.Background(), false)
 	})
 }
 

blobstore/clustermgr/svr.go

@@ -131,6 +131,7 @@ type Config struct {
 	MetricReportIntervalM    int                       `json:"metric_report_interval_m"`
 	ConsistentCheckIntervalM int                       `json:"consistent_check_interval_m"`
 	DashboardFreshIntervalS  int                       `json:"dashboard_fresh_interval_s"`
+	DashboardFreshVolumeTick int                       `json:"dashboard_fresh_volume_tick"`
 
 	BrokenVolumeUnitReportNum int `json:"broken_volume_unit_report_num"`
 	BrokenShardUnitReportNum  int `json:"broken_shard_unit_report_num"`

blobstore/clustermgr/volumemgr/volumemgr_test.go

@@ -22,6 +22,7 @@ import (
 	"os"
 	"path"
 	"strconv"
+	"sync"
 	"testing"
 	"time"
 
@@ -1479,3 +1480,49 @@ func TestVolumeStat(t *testing.T) {
 	stat.addSize(vid3, proto.VolumeStatusActive, freeSize1)
 	require.Equal(t, spaceBeforeAdd, stat.getWriteSpace())
 }
+
+func makeBenchVolumeMgr(n int) *VolumeMgr {
+	num := uint32(testConfig.VolumeSliceMapNum)
+	shards := &shardedVolumes{
+		num:   num,
+		m:     make(map[uint32]map[proto.Vid]*volume, num),
+		locks: make(map[uint32]*sync.RWMutex, num),
+	}
+	for i := uint32(0); i < num; i++ {
+		shards.m[i] = make(map[proto.Vid]*volume)
+		shards.locks[i] = &sync.RWMutex{}
+	}
+	v := &VolumeMgr{all: shards}
+	for _, vol := range generateVolume(codemode.EC12P9, n, 1) {
+		v.all.putVol(vol) //nolint:errcheck
+	}
+	return v
+}
+
+func BenchmarkRangeUpdateVolume(b *testing.B) {
+	cases := []struct {
+		name  string
+		count int
+	}{
+		{"1w", 10_000},
+		{"10w", 100_000},
+		{"100w", 1_000_000},
+	}
+
+	for _, tc := range cases {
+		tc := tc
+		b.Run(tc.name, func(b *testing.B) {
+			mgr := makeBenchVolumeMgr(tc.count)
+			ctx := context.Background()
+
+			cache := make(map[proto.Vid]*clustermgr.VolumeBasic, tc.count)
+			mgr.RangeUpdateVolume(ctx, cache)
+
+			b.ResetTimer()
+			b.ReportAllocs()
+			for i := 0; i < b.N; i++ {
+				mgr.RangeUpdateVolume(ctx, cache)
+			}
+		})
+	}
+}