feat(clustermgr): add http dashboard and scope interface

. #1000911000 and #1000911005

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

blobstore/api/clustermgr/dashboard.go

@@ -0,0 +1,60 @@
+// Copyright 2026 The CubeFS Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License.
+
+package clustermgr
+
+import (
+	"context"
+
+	"github.com/cubefs/cubefs/blobstore/util"
+)
+
+type DashboardArgs struct {
+	// Force to fresh an immediate snapshot of dashboard
+	Force bool `json:"force,omitempty"`
+}
+
+type DashboardScore int
+
+const (
+	DashboardScoreOK       DashboardScore = 0
+	DashboardScoreNotice   DashboardScore = 1
+	DashboardScoreWarning  DashboardScore = 2
+	DashboardScoreMajor    DashboardScore = 3
+	DashboardScoreCritical DashboardScore = 4
+)
+
+type ClusterDashboard struct {
+	Score       DashboardScore `json:"score"`
+	GeneratedAt int64          `json:"generated_at"` // Unix nanoseconds
+
+	Scope ScopeStat `json:"scope"`
+}
+
+type ScopeStat struct {
+	Score  DashboardScore `json:"score"`
+	Scopes []ScopeUsage   `json:"scopes"`
+}
+
+// ScopeUsage records the current allocation progress of a single scope counter.
+type ScopeUsage struct {
+	Name     string `json:"name"`
+	Current  uint64 `json:"current"`
+	MaxValue uint64 `json:"max_value"`
+}
+
+func (c *Client) Dashboard(ctx context.Context, args *DashboardArgs) (ret ClusterDashboard, err error) {
+	err = c.GetWith(ctx, "/cluster/dashboard?force="+util.Any2String(args.Force), &ret)
+	return
+}

blobstore/clustermgr/dashboard.go

@@ -0,0 +1,184 @@
+// Copyright 2026 The CubeFS Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License.
+
+package clustermgr
+
+import (
+	"math"
+	"sort"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/cubefs/cubefs/blobstore/api/clustermgr"
+	"github.com/cubefs/cubefs/blobstore/common/rpc"
+	"github.com/cubefs/cubefs/blobstore/common/trace"
+	"github.com/cubefs/cubefs/blobstore/util/defaulter"
+)
+
+const rebuildCooldown = 5 * time.Second
+
+func (s *Service) Dashboard(c *rpc.Context) {
+	ctx := c.Request.Context()
+	span := trace.SpanFromContextSafe(ctx)
+
+	args := new(clustermgr.DashboardArgs)
+	if err := c.ParseArgs(args); err != nil {
+		c.RespondError(err)
+		return
+	}
+	span.Infof("accept dashboard request, force=%v", args.Force)
+
+	if args.Force {
+		select {
+		case <-s.dashboardMgr.Refresh():
+		case <-ctx.Done():
+			c.RespondError(ctx.Err())
+			return
+		}
+	}
+
+	snap := s.dashboardMgr.GetSnapshot()
+	c.RespondJSON(snap.dashboard)
+}
+
+type dashboardSnapshot struct {
+	dashboard clustermgr.ClusterDashboard
+}
+
+type dashboardMgr struct {
+	service  *Service
+	snapshot atomic.Value // stores *dashboardSnapshot
+
+	freshCh chan struct{} // buffered(1); loop listens for force-fresh signals
+
+	processLock   sync.Mutex
+	processCh     chan struct{} // non-nil while a fresh is pending; closed on completion
+	cooldownUntil time.Time     // Refresh() returns a pre-closed channel before this time
+}
+
+func newDashboardMgr(service *Service) *dashboardMgr {
+	d := &dashboardMgr{
+		service: service,
+		freshCh: make(chan struct{}, 1),
+	}
+	d.snapshot.Store(&dashboardSnapshot{})
+	return d
+}
+
+func (d *dashboardMgr) Refresh() <-chan struct{} {
+	d.processLock.Lock()
+	defer d.processLock.Unlock()
+
+	if d.processCh != nil {
+		return d.processCh
+	}
+
+	// within cooldown window: snapshot is fresh enough, return a pre-closed
+	if time.Now().Before(d.cooldownUntil) {
+		ch := make(chan struct{})
+		close(ch)
+		return ch
+	}
+
+	ch := make(chan struct{})
+	d.processCh = ch
+	select {
+	case d.freshCh <- struct{}{}:
+	default:
+	}
+	return ch
+}
+
+func (d *dashboardMgr) GetSnapshot() *dashboardSnapshot {
+	return d.snapshot.Load().(*dashboardSnapshot)
+}
+
+func (d *dashboardMgr) loopFresh() {
+	defaulter.LessOrEqual(&d.service.DashboardFreshIntervalS, 60)
+
+	interval := time.Duration(d.service.DashboardFreshIntervalS) * time.Second
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			d.fresh()
+			select {
+			case <-d.freshCh:
+			default:
+			}
+
+		case <-d.freshCh:
+			d.fresh()
+			ticker.Reset(interval)
+
+		case <-d.service.closeCh:
+			return
+		}
+	}
+}
+
+func (d *dashboardMgr) fresh() {
+	d.snapshot.Store(&dashboardSnapshot{
+		dashboard: clustermgr.ClusterDashboard{
+			Scope:       d.buildScope(),
+			GeneratedAt: time.Now().UnixNano(),
+		},
+	})
+
+	d.processLock.Lock()
+	ch := d.processCh
+	d.processCh = nil
+	d.cooldownUntil = time.Now().Add(rebuildCooldown)
+	d.processLock.Unlock()
+
+	if ch != nil {
+		close(ch) // wake all force waiters
+	}
+}
+
+func (d *dashboardMgr) buildScope() clustermgr.ScopeStat {
+	return buildScopeFromMap(d.service.ScopeMgr.Stat())
+}
+
+func buildScopeFromMap(rawScopes map[string]uint64) clustermgr.ScopeStat {
+	scopeMaxValue := func(name string) uint64 {
+		switch name {
+		case BidScopeName, "space_id":
+			return math.MaxUint64
+		default:
+			return math.MaxUint32
+		}
+	}
+
+	score := clustermgr.DashboardScoreOK
+	scopes := make([]clustermgr.ScopeUsage, 0, len(rawScopes))
+	for name, cur := range rawScopes {
+		maxVal := scopeMaxValue(name)
+		if cur > maxVal/2 {
+			score = clustermgr.DashboardScoreNotice
+		}
+		scopes = append(scopes, clustermgr.ScopeUsage{
+			Name:     name,
+			Current:  cur,
+			MaxValue: maxVal,
+		})
+	}
+	sort.Slice(scopes, func(i, j int) bool {
+		return scopes[i].Name < scopes[j].Name
+	})
+	return clustermgr.ScopeStat{Score: score, Scopes: scopes}
+}

blobstore/clustermgr/dashboard_test.go

@@ -0,0 +1,289 @@
+// Copyright 2026 The CubeFS Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License.
+
+package clustermgr
+
+import (
+	"math"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/cubefs/cubefs/blobstore/api/clustermgr"
+)
+
+func newTestDashboardMgr() *dashboardMgr {
+	d := &dashboardMgr{
+		freshCh: make(chan struct{}, 1),
+	}
+	d.snapshot.Store(&dashboardSnapshot{})
+	return d
+}
+
+func TestDashboardRefresh_IdleTriggersSignal(t *testing.T) {
+	d := newTestDashboardMgr()
+
+	ch := d.Refresh()
+	require.NotNil(t, ch)
+
+	select {
+	case <-d.freshCh:
+	default:
+		t.Fatal("expected freshCh to have a pending signal")
+	}
+
+	d.processLock.Lock()
+	require.NotNil(t, d.processCh) // processCh is set
+	d.processLock.Unlock()
+}
+
+func TestDashboardRefresh_InFlightSharesChannel(t *testing.T) {
+	d := newTestDashboardMgr()
+
+	ch1 := d.Refresh()
+	ch2 := d.Refresh()
+	require.Equal(t, ch1, ch2)
+
+	// freshCh must have exactly one signal (not two)
+	count := 0
+	for {
+		select {
+		case <-d.freshCh:
+			count++
+		default:
+			goto done
+		}
+	}
+done:
+	require.Equal(t, 1, count)
+}
+
+func TestDashboardRefresh_CooldownReturnsClosedChannel(t *testing.T) {
+	d := newTestDashboardMgr()
+
+	// Manually set a future cooldown and nil processCh
+	d.processLock.Lock()
+	d.cooldownUntil = time.Now().Add(10 * time.Second)
+	d.processLock.Unlock()
+
+	ch := d.Refresh()
+
+	select {
+	case <-ch:
+	default:
+		t.Fatal("expected a pre-closed channel during cooldown")
+	}
+
+	select {
+	case <-d.freshCh:
+		t.Fatal("freshCh should not be signaled during cooldown")
+	default:
+	}
+}
+
+func TestDashboardRefresh_AfterCooldownExpiry(t *testing.T) {
+	d := newTestDashboardMgr()
+
+	d.processLock.Lock()
+	d.cooldownUntil = time.Now().Add(-1 * time.Second) // Expired cooldown
+	d.processLock.Unlock()
+
+	ch := d.Refresh()
+
+	select {
+	case <-d.freshCh:
+	default:
+		t.Fatal("expected freshCh signal after cooldown expired")
+	}
+
+	select {
+	case <-ch:
+		t.Fatal("channel should not be closed yet")
+	default:
+	}
+}
+
+func TestDashboardFresh_UpdatesSnapshotAndNotifiesWaiters(t *testing.T) {
+	svc, cleanup := initTestService(t)
+	defer cleanup()
+
+	d := svc.dashboardMgr
+
+	ch := d.Refresh()
+	select {
+	case <-d.freshCh:
+	default:
+	}
+
+	t1 := time.Now()
+	d.fresh()
+
+	select {
+	case <-ch:
+	case <-time.After(time.Second):
+		t.Fatal("waiter not unblocked after fresh()")
+	}
+
+	// Snapshot GeneratedAt must be after t1
+	snap := d.GetSnapshot()
+	require.Greater(t, snap.dashboard.GeneratedAt, t1.UnixNano())
+
+	d.processLock.Lock()
+	require.Nil(t, d.processCh)
+	d.processLock.Unlock()
+
+	d.processLock.Lock()
+	require.True(t, time.Now().Before(d.cooldownUntil))
+	d.processLock.Unlock()
+}
+
+func TestDashboardFresh_NoPanicWhenNoWaiter(t *testing.T) {
+	svc, cleanup := initTestService(t)
+	defer cleanup()
+	require.NotPanics(t, func() {
+		svc.dashboardMgr.fresh()
+	})
+}
+
+func TestDashboardRefresh_ConcurrentCallersShareResult(t *testing.T) {
+	d := newTestDashboardMgr()
+
+	const n = 50
+	var wg sync.WaitGroup
+	channels := make([](<-chan struct{}), n)
+
+	wg.Add(n)
+	for i := 0; i < n; i++ {
+		i := i
+		go func() {
+			defer wg.Done()
+			channels[i] = d.Refresh()
+		}()
+	}
+	wg.Wait()
+
+	// All goroutines must have received a non-nil channel.
+	for i, ch := range channels {
+		require.NotNil(t, ch, "channel[%d] is nil", i)
+	}
+
+	// After fresh() closes processCh all channels must become readable.
+	d.processLock.Lock()
+	ch := d.processCh
+	d.processCh = nil
+	d.processLock.Unlock()
+	if ch != nil {
+		close(ch)
+	}
+
+	for i, ch := range channels {
+		select {
+		case <-ch:
+		case <-time.After(time.Second):
+			t.Fatalf("channel[%d] not readable after close", i)
+		}
+	}
+}
+
+// buildScope()
+
+func TestDashboardBuildScope_EmptyScopes(t *testing.T) {
+	svc, cleanup := initTestService(t)
+	defer cleanup()
+	result := svc.dashboardMgr.buildScope()
+	require.Equal(t, clustermgr.DashboardScoreOK, result.Score)
+	require.Empty(t, result.Scopes)
+}
+
+func TestDashboardBuildScope_SortedAndCorrectMaxValue(t *testing.T) {
+	fakeScopes := map[string]uint64{
+		"vid":        10,
+		"diskid":     5,
+		BidScopeName: 100,
+		"space_id":   200,
+		"nodeid":     3,
+	}
+	result := buildScopeFromMap(fakeScopes)
+
+	for i := 1; i < len(result.Scopes); i++ {
+		require.Less(t, result.Scopes[i-1].Name, result.Scopes[i].Name,
+			"scopes not sorted at index %d", i)
+	}
+
+	for _, s := range result.Scopes {
+		if s.Name == BidScopeName || s.Name == "space_id" {
+			require.Equal(t, uint64(math.MaxUint64), s.MaxValue, "scope %s", s.Name)
+		} else {
+			require.Equal(t, uint64(math.MaxUint32), s.MaxValue, "scope %s", s.Name)
+		}
+	}
+}
+
+func TestDashboardBuildScope_ScoreOKWhenBelowHalf(t *testing.T) {
+	fakeScopes := map[string]uint64{
+		"vid":    100, // MaxUint32/2 = 2147483647; 100 is far below half
+		"diskid": 1000,
+	}
+	result := buildScopeFromMap(fakeScopes)
+	require.Equal(t, clustermgr.DashboardScoreOK, result.Score)
+}
+
+func TestDashboardBuildScope_ScoreNoticeWhenAboveHalf(t *testing.T) {
+	fakeScopes := map[string]uint64{
+		"vid":    math.MaxUint32/2 + 1, // just above half → Notice
+		"diskid": 100,
+	}
+	result := buildScopeFromMap(fakeScopes)
+	require.Equal(t, clustermgr.DashboardScoreNotice, result.Score)
+}
+
+func TestDashboardBuildScope_BidScopeNoOverflow(t *testing.T) {
+	fakeScopes := map[string]uint64{
+		BidScopeName: math.MaxUint64 / 2,
+	}
+	result := buildScopeFromMap(fakeScopes)
+	require.Equal(t, clustermgr.DashboardScoreOK, result.Score)
+
+	// current = MaxUint64/2+1 should be Notice
+	fakeScopes[BidScopeName] = math.MaxUint64/2 + 1
+	result = buildScopeFromMap(fakeScopes)
+	require.Equal(t, clustermgr.DashboardScoreNotice, result.Score)
+}
+
+
+func TestDashboardGetSnapshot_NeverNil(t *testing.T) {
+	d := newTestDashboardMgr()
+	require.NotNil(t, d.GetSnapshot())
+}
+
+func TestDashboardLoopFresh_ForceRefresh(t *testing.T) {
+	svc, cleanup := initTestService(t)
+	defer cleanup()
+
+	d := svc.dashboardMgr
+
+	origAt := d.snapshot.Load().(*dashboardSnapshot).dashboard.GeneratedAt
+
+	ch := d.Refresh()
+	select {
+	case <-ch:
+	case <-time.After(5 * time.Second):
+		t.Fatal("force refresh timed out")
+	}
+
+	snap := d.GetSnapshot()
+	require.Greater(t, snap.dashboard.GeneratedAt, origAt)
+}

blobstore/clustermgr/handler.go

@@ -194,6 +194,7 @@ func NewHandler(service *Service) *rpc.Router {
 
 	//==================manage==========================
 	rpc.RegisterArgsParser(&clustermgr.SetClusterReadonlyArgs{}, "json")
+	rpc.RegisterArgsParser(&clustermgr.DashboardArgs{}, "json")
 
 	rpc.POST("/member/add", service.MemberAdd, rpc.OptArgsBody())
 
@@ -206,6 +207,7 @@ func NewHandler(service *Service) *rpc.Router {
 	rpc.GET("/snapshot/dump", service.SnapshotDump)
 
 	rpc.POST("/cluster/set", service.ClusterSetReadonly, rpc.OptArgsQuery())
+	rpc.GET("/cluster/dashboard", service.Dashboard, rpc.OptArgsQuery())
 
 	//==================kv==========================
 	rpc.RegisterArgsParser(&clustermgr.ListKvOpts{}, "json")

blobstore/clustermgr/scopemgr/scopemgr.go

@@ -98,6 +98,16 @@ func (s *ScopeMgr) GetCurrent(name string) uint64 {
 	return s.scopeItems[name]
 }
 
+func (s *ScopeMgr) Stat() map[string]uint64 {
+	s.lock.RLock()
+	items := make(map[string]uint64, len(s.scopeItems))
+	for name, cur := range s.scopeItems {
+		items[name] = cur
+	}
+	s.lock.RUnlock()
+	return items
+}
+
 func (s *ScopeMgr) applyCommit(ctx context.Context, args *allocCtx) error {
 	s.lock.Lock()
 	defer s.lock.Unlock()

blobstore/clustermgr/scopemgr/scopemgr_test.go

@@ -20,6 +20,7 @@ import (
 	"math/rand"
 	"os"
 	"strconv"
+	"sync"
 	"testing"
 
 	base_ "github.com/cubefs/cubefs/blobstore/clustermgr/base"
@@ -122,3 +123,62 @@ func TestScopeMgr(t *testing.T) {
 		scopeMgr.Flush(ctx)
 	}
 }
+
+func TestScopeMgrStat(t *testing.T) {
+	tmpDBPath := "/tmp/tmpnormaldb" + strconv.Itoa(rand.Intn(10000000000))
+	defer os.RemoveAll(tmpDBPath)
+
+	db, err := normaldb.OpenNormalDB(tmpDBPath)
+	require.NoError(t, err)
+	defer db.Close()
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockRaftServer := mocks.NewMockRaftServer(ctrl)
+	_, ctx := trace.StartSpanFromContext(context.Background(), "")
+
+	scopeMgr, err := NewScopeMgr(db)
+	require.NoError(t, err)
+	scopeMgr.SetRaftServer(mockRaftServer)
+
+	{
+		got := scopeMgr.Stat()
+		require.Empty(t, got)
+	}
+	{
+		mockRaftServer.EXPECT().Propose(gomock.Any(), gomock.Any()).Return(nil)
+
+		_, _, err := scopeMgr.Alloc(ctx, "vid", 100)
+		require.NoError(t, err)
+
+		got := scopeMgr.Stat()
+		require.Equal(t, 1, len(got))
+		require.Equal(t, scopeMgr.GetCurrent("vid"), got["vid"])
+	}
+	{
+		mockRaftServer.EXPECT().Propose(gomock.Any(), gomock.Any()).Return(nil).Times(2)
+
+		_, _, err := scopeMgr.Alloc(ctx, "diskid", 50)
+		require.NoError(t, err)
+		_, _, err = scopeMgr.Alloc(ctx, "bid", 200)
+		require.NoError(t, err)
+
+		got := scopeMgr.Stat()
+		require.Equal(t, 3, len(got))
+		require.Equal(t, scopeMgr.GetCurrent("vid"), got["vid"])
+		require.Equal(t, scopeMgr.GetCurrent("diskid"), got["diskid"])
+		require.Equal(t, scopeMgr.GetCurrent("bid"), got["bid"])
+	}
+	{
+		var wg sync.WaitGroup
+		for range [20]struct{}{} {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				require.Equal(t, 3, len(scopeMgr.Stat()))
+			}()
+		}
+		wg.Wait()
+	}
+}

blobstore/clustermgr/svr.go

@@ -130,6 +130,7 @@ type Config struct {
 	ChunkSize                uint64                    `json:"chunk_size"`
 	MetricReportIntervalM    int                       `json:"metric_report_interval_m"`
 	ConsistentCheckIntervalM int                       `json:"consistent_check_interval_m"`
+	DashboardFreshIntervalS  int                       `json:"dashboard_fresh_interval_s"`
 
 	BrokenVolumeUnitReportNum int `json:"broken_volume_unit_report_num"`
 	BrokenShardUnitReportNum  int `json:"broken_shard_unit_report_num"`
@@ -166,6 +167,8 @@ type Service struct {
 	closeCh                chan interface{}
 	consulClient           *api.Client
 	*Config
+
+	dashboardMgr *dashboardMgr
 }
 
 func init() {
@@ -363,6 +366,9 @@ func New(cfg *Config) (*Service, error) {
 	// start raft node background progress
 	go raftNode.Start()
 
+	service.dashboardMgr = newDashboardMgr(service)
+	go service.dashboardMgr.loopFresh()
+
 	// start service background loop
 	go service.loop()