From d7d902f9aaa5e31449187d7bd2675aa79c028827 Mon Sep 17 00:00:00 2001 From: slasher Date: Tue, 28 Apr 2026 10:28:22 +0800 Subject: [PATCH] feat(clustermgr): add http dashboard and scope interface . #1000911000 and #1000911005 Signed-off-by: slasher --- blobstore/api/clustermgr/dashboard.go | 60 ++++ blobstore/clustermgr/dashboard.go | 184 +++++++++++ blobstore/clustermgr/dashboard_test.go | 289 ++++++++++++++++++ blobstore/clustermgr/handler.go | 2 + blobstore/clustermgr/scopemgr/scopemgr.go | 10 + .../clustermgr/scopemgr/scopemgr_test.go | 60 ++++ blobstore/clustermgr/svr.go | 6 + 7 files changed, 611 insertions(+) create mode 100644 blobstore/api/clustermgr/dashboard.go create mode 100644 blobstore/clustermgr/dashboard.go create mode 100644 blobstore/clustermgr/dashboard_test.go diff --git a/blobstore/api/clustermgr/dashboard.go b/blobstore/api/clustermgr/dashboard.go new file mode 100644 index 000000000..b0ea151f8 --- /dev/null +++ b/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 +} diff --git a/blobstore/clustermgr/dashboard.go b/blobstore/clustermgr/dashboard.go new file mode 100644 index 000000000..f12e1b153 --- /dev/null +++ b/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} +} diff --git a/blobstore/clustermgr/dashboard_test.go b/blobstore/clustermgr/dashboard_test.go new file mode 100644 index 000000000..07d511aea --- /dev/null +++ b/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) +} diff --git a/blobstore/clustermgr/handler.go b/blobstore/clustermgr/handler.go index ce6762b1f..81c671995 100644 --- a/blobstore/clustermgr/handler.go +++ b/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") diff --git a/blobstore/clustermgr/scopemgr/scopemgr.go b/blobstore/clustermgr/scopemgr/scopemgr.go index e4871eae1..dd2ac9aa7 100644 --- a/blobstore/clustermgr/scopemgr/scopemgr.go +++ b/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() diff --git a/blobstore/clustermgr/scopemgr/scopemgr_test.go b/blobstore/clustermgr/scopemgr/scopemgr_test.go index d463291af..7910ec6e9 100644 --- a/blobstore/clustermgr/scopemgr/scopemgr_test.go +++ b/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() + } +} diff --git a/blobstore/clustermgr/svr.go b/blobstore/clustermgr/svr.go index f4f0958c8..4a9531acd 100644 --- a/blobstore/clustermgr/svr.go +++ b/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()