Skip to content

Commit

Permalink
statistics: add gc in hot peer cache (#8702) (#8897)
Browse files Browse the repository at this point in the history 
close #8698

Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>
Signed-off-by: lhy1024 <admin@liudos.us>

Co-authored-by: lhy1024 <liuhanyang@pingcap.com>
Co-authored-by: lhy1024 <admin@liudos.us>
Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com>
  • Loading branch information
@ti-chi-bot @lhy1024 @ti-chi-bot
4 people authored Dec 12, 2024
1 parent e2bda4b commit 57fdd79
Show file tree
Hide file tree
Showing 11 changed files with 183 additions and 75 deletions.
2 changes: 1 addition & 1 deletion pkg/mcs/scheduling/server/cluster.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,7 @@ func NewCluster(parentCtx context.Context, persistConfig *config.PersistConfig,
ruleManager: ruleManager,
labelerManager: labelerManager,
persistConfig: persistConfig,
hotStat: statistics.NewHotStat(ctx),
hotStat: statistics.NewHotStat(ctx, basicCluster),
labelStats: statistics.NewLabelStatistics(),
regionStats: statistics.NewRegionStatistics(basicCluster, persistConfig, ruleManager),
storage: storage,
Expand Down
5 changes: 3 additions & 2 deletions pkg/mock/mockcluster/mockcluster.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -63,11 +63,12 @@ type Cluster struct {

// NewCluster creates a new Cluster
func NewCluster(ctx context.Context, opts *config.PersistOptions) *Cluster {
bc := core.NewBasicCluster()
c := &Cluster{
ctx: ctx,
BasicCluster: core.NewBasicCluster(),
BasicCluster: bc,
IDAllocator: mockid.NewIDAllocator(),
HotStat: statistics.NewHotStat(ctx),
HotStat: statistics.NewHotStat(ctx, bc),
HotBucketCache: buckets.NewBucketsCache(ctx),
PersistOptions: opts,
suspectRegions: map[uint64]struct{}{},
Expand Down
18 changes: 18 additions & 0 deletions pkg/schedule/schedulers/hot_region_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1658,6 +1658,9 @@ func TestHotCacheUpdateCache(t *testing.T) {
cancel, _, tc, _ := prepareSchedulersTest()
defer cancel()
tc.SetHotRegionCacheHitsThreshold(0)
for i := 0; i < 3; i++ {
tc.PutStore(core.NewStoreInfo(&metapb.Store{Id: uint64(i + 1)}))
}

// For read flow
addRegionInfo(tc, utils.Read, []testRegionInfo{
Expand Down Expand Up @@ -1725,6 +1728,9 @@ func TestHotCacheKeyThresholds(t *testing.T) {
cancel, _, tc, _ := prepareSchedulersTest()
defer cancel()
tc.SetHotRegionCacheHitsThreshold(0)
for i := 0; i < 6; i++ {
tc.PutStore(core.NewStoreInfo(&metapb.Store{Id: uint64(i + 1)}))
}
addRegionInfo(tc, utils.Read, []testRegionInfo{
{1, []uint64{1, 2, 3}, 0, 1, 0},
{2, []uint64{1, 2, 3}, 0, 1 * units.KiB, 0},
Expand All @@ -1743,6 +1749,9 @@ func TestHotCacheKeyThresholds(t *testing.T) {
{ // many regions
cancel, _, tc, _ := prepareSchedulersTest()
defer cancel()
for i := 0; i < 3; i++ {
tc.PutStore(core.NewStoreInfo(&metapb.Store{Id: uint64(i + 1)}))
}
regions := []testRegionInfo{}
for i := 1; i <= 1000; i += 2 {
regions = append(regions,
Expand Down Expand Up @@ -1797,6 +1806,9 @@ func TestHotCacheByteAndKey(t *testing.T) {
cancel, _, tc, _ := prepareSchedulersTest()
defer cancel()
tc.SetHotRegionCacheHitsThreshold(0)
for i := 0; i < 3; i++ {
tc.PutStore(core.NewStoreInfo(&metapb.Store{Id: uint64(i + 1)}))
}
statistics.ThresholdsUpdateInterval = 0
defer func() {
statistics.ThresholdsUpdateInterval = 8 * time.Second
Expand Down Expand Up @@ -1923,6 +1935,9 @@ func TestHotCacheCheckRegionFlow(t *testing.T) {
func checkHotCacheCheckRegionFlow(re *require.Assertions, testCase testHotCacheCheckRegionFlowCase, enablePlacementRules bool) {
cancel, _, tc, oc := prepareSchedulersTest()
defer cancel()
for i := 0; i < 3; i++ {
tc.PutStore(core.NewStoreInfo(&metapb.Store{Id: uint64(i + 1)}))
}
tc.SetClusterVersion(versioninfo.MinSupportedVersion(versioninfo.Version4_0))
tc.SetEnablePlacementRules(enablePlacementRules)
labels := []string{"zone", "host"}
Expand Down Expand Up @@ -1998,6 +2013,9 @@ func TestHotCacheCheckRegionFlowWithDifferentThreshold(t *testing.T) {
func checkHotCacheCheckRegionFlowWithDifferentThreshold(re *require.Assertions, enablePlacementRules bool) {
cancel, _, tc, _ := prepareSchedulersTest()
defer cancel()
for i := 0; i < 3; i++ {
tc.PutStore(core.NewStoreInfo(&metapb.Store{Id: uint64(i + 1)}))
}
tc.SetClusterVersion(versioninfo.MinSupportedVersion(versioninfo.Version4_0))
tc.SetEnablePlacementRules(enablePlacementRules)
labels := []string{"zone", "host"}
Expand Down
6 changes: 3 additions & 3 deletions pkg/statistics/hot_cache.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,11 +38,11 @@ type HotCache struct {
}

// NewHotCache creates a new hot spot cache.
func NewHotCache(ctx context.Context) *HotCache {
func NewHotCache(ctx context.Context, cluster *core.BasicCluster) *HotCache {
w := &HotCache{
ctx: ctx,
writeCache: NewHotPeerCache(ctx, utils.Write),
readCache: NewHotPeerCache(ctx, utils.Read),
writeCache: NewHotPeerCache(ctx, cluster, utils.Write),
readCache: NewHotPeerCache(ctx, cluster, utils.Read),
}
go w.updateItems(w.readCache.taskQueue, w.runReadTask)
go w.updateItems(w.writeCache.taskQueue, w.runWriteTask)
Expand Down
37 changes: 35 additions & 2 deletions pkg/statistics/hot_peer_cache.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -60,20 +60,22 @@ type thresholds struct {
// hotPeerCache saves the hot peer's statistics.
type hotPeerCache struct {
kind utils.RWType
cluster *core.BasicCluster
peersOfStore map[uint64]*utils.TopN // storeID -> hot peers
storesOfRegion map[uint64]map[uint64]struct{} // regionID -> storeIDs
regionsOfStore map[uint64]map[uint64]struct{} // storeID -> regionIDs
topNTTL time.Duration
taskQueue *chanx.UnboundedChan[FlowItemTask]
thresholdsOfStore map[uint64]*thresholds // storeID -> thresholds
metrics map[uint64][utils.ActionTypeLen]prometheus.Gauge // storeID -> metrics
// TODO: consider to remove store info when store is offline.
lastGCTime time.Time
}

// NewHotPeerCache creates a hotPeerCache
func NewHotPeerCache(ctx context.Context, kind utils.RWType) *hotPeerCache {
func NewHotPeerCache(ctx context.Context, cluster *core.BasicCluster, kind utils.RWType) *hotPeerCache {
return &hotPeerCache{
kind: kind,
cluster: cluster,
peersOfStore: make(map[uint64]*utils.TopN),
storesOfRegion: make(map[uint64]map[uint64]struct{}),
regionsOfStore: make(map[uint64]map[uint64]struct{}),
Expand Down Expand Up @@ -114,6 +116,7 @@ func (f *hotPeerCache) updateStat(item *HotPeerStat) {
return
}
f.incMetrics(item.actionType, item.StoreID)
f.gc()
}

func (f *hotPeerCache) incMetrics(action utils.ActionType, storeID uint64) {
Expand Down Expand Up @@ -544,6 +547,36 @@ func (f *hotPeerCache) removeItem(item *HotPeerStat) {
}
}

func (f *hotPeerCache) gc() {
if time.Since(f.lastGCTime) < f.topNTTL {
return
}
f.lastGCTime = time.Now()
// remove tombstone stores
stores := make(map[uint64]struct{})
for _, storeID := range f.cluster.GetStores() {
stores[storeID.GetID()] = struct{}{}
}
for storeID := range f.peersOfStore {
if _, ok := stores[storeID]; !ok {
delete(f.peersOfStore, storeID)
delete(f.regionsOfStore, storeID)
delete(f.thresholdsOfStore, storeID)
delete(f.metrics, storeID)
}
}
// remove expired items
for _, peers := range f.peersOfStore {
regions := peers.RemoveExpired()
for _, regionID := range regions {
delete(f.storesOfRegion, regionID)
for storeID := range f.regionsOfStore {
delete(f.regionsOfStore[storeID], regionID)
}
}
}
}

// removeAllItem removes all items of the cache.
// It is used for test.
func (f *hotPeerCache) removeAllItem() {
Expand Down
Loading

0 comments on commit 57fdd79

Please sign in to comment.