Skip to content

Commit

Permalink
This is an automated cherry-pick of #8702
Browse files Browse the repository at this point in the history 
close #8698

Signed-off-by: ti-chi-bot <[email protected]>
  • Loading branch information
@lhy1024 @ti-chi-bot
lhy1024 authored and ti-chi-bot committed Oct 30, 2024
1 parent 4fbf433 commit 0830b68
Show file tree
Hide file tree
Showing 12 changed files with 1,529 additions and 24 deletions.
708 changes: 708 additions & 0 deletions pkg/mcs/scheduling/server/cluster.go
View file

Large diffs are not rendered by default.

13 changes: 13 additions & 0 deletions pkg/mock/mockcluster/mockcluster.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@ type Cluster struct {

// NewCluster creates a new Cluster
func NewCluster(ctx context.Context, opts *config.PersistOptions) *Cluster {
<<<<<<< HEAD
clus := &Cluster{
BasicCluster: core.NewBasicCluster(),
IDAllocator: mockid.NewIDAllocator(),
Expand All @@ -71,6 +72,18 @@ func NewCluster(ctx context.Context, opts *config.PersistOptions) *Cluster {
suspectRegions: map[uint64]struct{}{},
StoreConfigManager: config.NewTestStoreConfigManager(nil),
ctx: ctx,
=======
bc := core.NewBasicCluster()
c := &Cluster{
ctx: ctx,
BasicCluster: bc,
IDAllocator: mockid.NewIDAllocator(),
HotStat: statistics.NewHotStat(ctx, bc),
HotBucketCache: buckets.NewBucketsCache(ctx),
PersistOptions: opts,
pendingProcessedRegions: map[uint64]struct{}{},
Storage: storage.NewStorageWithMemoryBackend(),
>>>>>>> 20087e290 (statistics: add gc in hot peer cache (#8702))
}
if clus.PersistOptions.GetReplicationConfig().EnablePlacementRules {
clus.initRuleManager()
Expand Down
28 changes: 28 additions & 0 deletions pkg/schedule/schedulers/hot_region_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1494,7 +1494,13 @@ func TestHotCacheUpdateCache(t *testing.T) {
re := require.New(t)
cancel, _, tc, _ := prepareSchedulersTest()
defer cancel()
<<<<<<< HEAD
tc.SetHotRegionCacheHitsThreshold(0)
=======
for i := range 3 {
tc.PutStore(core.NewStoreInfo(&metapb.Store{Id: uint64(i + 1)}))
}
>>>>>>> 20087e290 (statistics: add gc in hot peer cache (#8702))

// For read flow
addRegionInfo(tc, statistics.Read, []testRegionInfo{
Expand Down Expand Up @@ -1561,8 +1567,15 @@ func TestHotCacheKeyThresholds(t *testing.T) {
{ // only a few regions
cancel, _, tc, _ := prepareSchedulersTest()
defer cancel()
<<<<<<< HEAD
tc.SetHotRegionCacheHitsThreshold(0)
addRegionInfo(tc, statistics.Read, []testRegionInfo{
=======
for i := range 6 {
tc.PutStore(core.NewStoreInfo(&metapb.Store{Id: uint64(i + 1)}))
}
addRegionInfo(tc, utils.Read, []testRegionInfo{
>>>>>>> 20087e290 (statistics: add gc in hot peer cache (#8702))
{1, []uint64{1, 2, 3}, 0, 1, 0},
{2, []uint64{1, 2, 3}, 0, 1 * units.KiB, 0},
})
Expand All @@ -1580,6 +1593,9 @@ func TestHotCacheKeyThresholds(t *testing.T) {
{ // many regions
cancel, _, tc, _ := prepareSchedulersTest()
defer cancel()
for i := range 3 {
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 @@ -1633,7 +1649,13 @@ func TestHotCacheByteAndKey(t *testing.T) {
re := require.New(t)
cancel, _, tc, _ := prepareSchedulersTest()
defer cancel()
<<<<<<< HEAD
tc.SetHotRegionCacheHitsThreshold(0)
=======
for i := range 3 {
tc.PutStore(core.NewStoreInfo(&metapb.Store{Id: uint64(i + 1)}))
}
>>>>>>> 20087e290 (statistics: add gc in hot peer cache (#8702))
statistics.ThresholdsUpdateInterval = 0
defer func() {
statistics.ThresholdsUpdateInterval = 8 * time.Second
Expand Down Expand Up @@ -1760,6 +1782,9 @@ func TestHotCacheCheckRegionFlow(t *testing.T) {
func checkHotCacheCheckRegionFlow(re *require.Assertions, testCase testHotCacheCheckRegionFlowCase, enablePlacementRules bool) {
cancel, _, tc, oc := prepareSchedulersTest()
defer cancel()
for i := range 3 {
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 @@ -1835,6 +1860,9 @@ func TestHotCacheCheckRegionFlowWithDifferentThreshold(t *testing.T) {
func checkHotCacheCheckRegionFlowWithDifferentThreshold(re *require.Assertions, enablePlacementRules bool) {
cancel, _, tc, _ := prepareSchedulersTest()
defer cancel()
for i := range 3 {
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
7 changes: 6 additions & 1 deletion pkg/statistics/hot_cache.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -37,11 +37,16 @@ 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,
<<<<<<< HEAD
writeCache: NewHotPeerCache(ctx, Write),
readCache: NewHotPeerCache(ctx, Read),
=======
writeCache: NewHotPeerCache(ctx, cluster, utils.Write),
readCache: NewHotPeerCache(ctx, cluster, utils.Read),
>>>>>>> 20087e290 (statistics: add gc in hot peer cache (#8702))
}
go w.updateItems(w.readCache.taskQueue, w.runReadTask)
go w.updateItems(w.writeCache.taskQueue, w.runWriteTask)
Expand Down
70 changes: 70 additions & 0 deletions pkg/statistics/hot_peer_cache.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,7 @@ type thresholds struct {
metrics [DimLen + 1]prometheus.Gauge // 0 is for byte, 1 is for key, 2 is for query, 3 is for total length.
}

<<<<<<< HEAD
// hotPeerCache saves the hot peer's statistics.
type hotPeerCache struct {
kind RWType
Expand All @@ -91,6 +92,28 @@ func NewHotPeerCache(ctx context.Context, kind RWType) *hotPeerCache {
return &hotPeerCache{
kind: kind,
peersOfStore: make(map[uint64]*TopN),
=======
// 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[func(*HotPeerCache)]
thresholdsOfStore map[uint64]*thresholds // storeID -> thresholds
metrics map[uint64][utils.ActionTypeLen]prometheus.Gauge // storeID -> metrics
lastGCTime time.Time
}

// NewHotPeerCache creates a 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),
>>>>>>> 20087e290 (statistics: add gc in hot peer cache (#8702))
storesOfRegion: make(map[uint64]map[uint64]struct{}),
regionsOfStore: make(map[uint64]map[uint64]struct{}),
taskQueue: chanx.NewUnboundedChan[FlowItemTask](ctx, queueCap),
Expand Down Expand Up @@ -130,6 +153,7 @@ func (f *hotPeerCache) updateStat(item *HotPeerStat) {
return
}
f.incMetrics(item.actionType, item.StoreID)
f.gc()
}

func (f *hotPeerCache) incMetrics(action ActionType, storeID uint64) {
Expand Down Expand Up @@ -560,7 +584,53 @@ func (f *hotPeerCache) removeItem(item *HotPeerStat) {
}
}

<<<<<<< HEAD
func (f *hotPeerCache) coldItem(newItem, oldItem *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() {
for _, peers := range f.peersOfStore {
for _, peer := range peers.GetAll() {
item := peer.(*HotPeerStat)
item.actionType = utils.Remove
f.UpdateStat(item)
}
}
}

func coldItem(newItem, oldItem *HotPeerStat) {
>>>>>>> 20087e290 (statistics: add gc in hot peer cache (#8702))
newItem.HotDegree = oldItem.HotDegree - 1
newItem.AntiCount = oldItem.AntiCount - 1
if newItem.AntiCount <= 0 {
Expand Down
Loading

0 comments on commit 0830b68

Please sign in to comment.