Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

statistics: replace loads with []float64 in peer info #3729

Merged
merged 9 commits into from
Jun 2, 2021
Merged

statistics: replace loads with []float64 in peer info #3729

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
64dae82
statistics: replace loads with []float64 in peer info
lhy1024 Jun 1, 2021
5c3576f
remove unused code
lhy1024 Jun 1, 2021
939e506
address comment
lhy1024 Jun 2, 2021
c7069bd
address comment
lhy1024 Jun 2, 2021
9312d02
remove unused code
lhy1024 Jun 2, 2021
3d837f8
fix test
lhy1024 Jun 2, 2021
fa98344
address comment
lhy1024 Jun 2, 2021
2a0d92f
misc
lhy1024 Jun 2, 2021
dbcbfa8
Merge branch 'master' into reformat-peer
ti-chi-bot Jun 2, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 3 additions & 12 deletions pkg/mock/mockcluster/mockcluster.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -756,10 +756,7 @@ func (mc *Cluster) CheckRegionRead(region *core.RegionInfo) []*statistics.HotPee
interval := reportInterval.GetEndTimestamp() - reportInterval.GetStartTimestamp()
for _, peer := range region.GetPeers() {
peerInfo := core.NewPeerInfo(peer,
region.GetBytesWritten(),
region.GetKeysWritten(),
region.GetBytesRead(),
region.GetKeysRead(),
statistics.GetLoads(region),
interval)
item := mc.HotCache.CheckReadPeerSync(peerInfo, region)
if item != nil {
Expand All @@ -778,10 +775,7 @@ func (mc *Cluster) CheckRegionWrite(region *core.RegionInfo) []*statistics.HotPe
interval := reportInterval.GetEndTimestamp() - reportInterval.GetStartTimestamp()
for _, peer := range region.GetPeers() {
peerInfo := core.NewPeerInfo(peer,
region.GetBytesWritten(),
region.GetKeysWritten(),
region.GetBytesRead(),
region.GetKeysRead(),
statistics.GetLoads(region),
interval)
item := mc.HotCache.CheckWritePeerSync(peerInfo, region)
if item != nil {
Expand All @@ -800,10 +794,7 @@ func (mc *Cluster) CheckRegionLeaderRead(region *core.RegionInfo) []*statistics.
interval := reportInterval.GetEndTimestamp() - reportInterval.GetStartTimestamp()
peer := region.GetLeader()
peerInfo := core.NewPeerInfo(peer,
region.GetBytesWritten(),
region.GetKeysWritten(),
region.GetBytesRead(),
region.GetKeysRead(),
statistics.GetLoads(region),
interval)
item := mc.HotCache.CheckReadPeerSync(peerInfo, region)
if item != nil {
Expand Down
8 changes: 4 additions & 4 deletions server/cluster/cluster.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -555,8 +555,9 @@ func (c *RaftCluster) HandleStoreHeartbeat(stats *pdpb.StoreStats) error {
zap.Uint64("store-id", storeID))
continue
}
peerInfo := core.NewPeerInfo(peer, 0, 0,
peerStat.GetReadBytes(), peerStat.GetReadKeys(), interval)
peerInfo := core.NewPeerInfo(peer,
statistics.GetLoads(region),
interval)
item := statistics.NewPeerInfoItem(peerInfo, region)
c.hotStat.CheckReadAsync(item)
}
Expand Down Expand Up @@ -589,8 +590,7 @@ func (c *RaftCluster) processRegionHeartbeat(region *core.RegionInfo) error {
interval := reportInterval.GetEndTimestamp() - reportInterval.GetStartTimestamp()
for _, peer := range region.GetPeers() {
peerInfo := core.NewPeerInfo(peer,
region.GetBytesWritten(), region.GetKeysWritten(),
0, 0,
statistics.GetLoads(region),
interval)
item := statistics.NewPeerInfoItem(peerInfo, region)
c.hotStat.CheckWriteAsync(item)
Expand Down
8 changes: 0 additions & 8 deletions server/core/kind.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,11 +129,3 @@ func StringToKeyType(input string) KeyType {
panic("invalid key type: " + input)
}
}

// FlowStat indicates the stats of the flow
type FlowStat interface {
GetKeysWritten() uint64
GetBytesWritten() uint64
GetBytesRead() uint64
GetKeysRead() uint64
}
44 changes: 11 additions & 33 deletions server/core/peer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,52 +76,30 @@ func CountInJointState(peers ...*metapb.Peer) int {
// PeerInfo provides peer information
type PeerInfo struct {
*metapb.Peer
writtenBytes uint64
writtenKeys uint64
readBytes uint64
readKeys uint64
interval uint64
loads []float64
interval uint64
}

// NewPeerInfo creates PeerInfo
func NewPeerInfo(meta *metapb.Peer,
writtenBytes, writtenKeys, readBytes, readKeys uint64,
func NewPeerInfo(meta *metapb.Peer, loads []float64,
interval uint64) *PeerInfo {
return &PeerInfo{
Peer: meta,
writtenBytes: writtenBytes,
writtenKeys: writtenKeys,
readBytes: readBytes,
readKeys: readKeys,
interval: interval,
Peer: meta,
loads: loads,
interval: interval,
}
}

// GetKeysWritten provides peer written keys
func (p *PeerInfo) GetKeysWritten() uint64 {
return p.writtenKeys
}

// GetBytesWritten provides peer written bytes
func (p *PeerInfo) GetBytesWritten() uint64 {
return p.writtenBytes
}

// GetBytesRead provides peer read bytes
func (p *PeerInfo) GetBytesRead() uint64 {
return p.readBytes
}

// GetKeysRead provides read keys
func (p *PeerInfo) GetKeysRead() uint64 {
return p.readKeys
}

// GetStoreID provides located storeID
func (p *PeerInfo) GetStoreID() uint64 {
return p.GetStoreId()
}

// GetLoads provides loads
func (p *PeerInfo) GetLoads() []float64 {
return p.loads
}

// GetPeerID provides peer id
func (p *PeerInfo) GetPeerID() uint64 {
return p.GetId()
Expand Down
19 changes: 1 addition & 18 deletions server/statistics/hot_peer_cache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -162,7 +162,7 @@ func (f *hotPeerCache) CheckPeerFlow(peer *core.PeerInfo, region *core.RegionInf
return nil
}
storeID := peer.GetStoreID()
deltaLoads := getFlowDeltaLoads(peer)
deltaLoads := peer.GetLoads()
f.collectPeerMetrics(deltaLoads, interval)
loads := make([]float64, len(deltaLoads))
for i := range deltaLoads {
Expand Down Expand Up @@ -462,23 +462,6 @@ func (f *hotPeerCache) updateNewHotPeerStat(newItem *HotPeerStat, deltaLoads []f
return newItem
}

func getFlowDeltaLoads(stat core.FlowStat) []float64 {
ret := make([]float64, RegionStatCount)
for k := RegionStatKind(0); k < RegionStatCount; k++ {
switch k {
case RegionReadBytes:
ret[k] = float64(stat.GetBytesRead())
case RegionReadKeys:
ret[k] = float64(stat.GetKeysRead())
case RegionWriteBytes:
ret[k] = float64(stat.GetBytesWritten())
case RegionWriteKeys:
ret[k] = float64(stat.GetKeysWritten())
}
}
return ret
}

func (f *hotPeerCache) putInheritItem(item *HotPeerStat) {
f.inheritItem[item.RegionID] = item
}
Expand Down
10 changes: 2 additions & 8 deletions server/statistics/hot_peer_cache_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,10 +111,7 @@ func checkAndUpdate(c *C, cache *hotPeerCache, region *core.RegionInfo, expect i
res = append(res, cache.CollectExpiredItems(region)...)
for _, peer := range region.GetPeers() {
peerInfo := core.NewPeerInfo(peer,
region.GetBytesWritten(),
region.GetKeysWritten(),
region.GetBytesRead(),
region.GetKeysRead(),
statistics.GetLoads(region),
interval)
item := cache.CheckPeerFlow(peerInfo, region)
if item != nil {
Expand Down Expand Up @@ -346,10 +343,7 @@ func BenchmarkCheckRegionFlow(b *testing.B) {
peerInfos := make([]*core.PeerInfo, 0)
for _, peer := range newRegion.GetPeers() {
peerInfo := core.NewPeerInfo(peer,
region.GetBytesWritten(),
region.GetKeysWritten(),
region.GetBytesRead(),
region.GetKeysRead(),
GetLoads(region),
10)
peerInfos = append(peerInfos, peerInfo)
}
Expand Down
13 changes: 13 additions & 0 deletions server/statistics/util.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,9 @@ package statistics

import (
"fmt"

"github.com/tikv/pd/server/core"

lhy1024 marked this conversation as resolved.
Show resolved Hide resolved
)

const (
Expand All @@ -27,3 +30,13 @@ const (
func storeTag(id uint64) string {
return fmt.Sprintf("store-%d", id)
}

func GetLoads(r *core.RegionInfo) []float64 {
rleungx marked this conversation as resolved.
Show resolved Hide resolved
return []float64{
RegionWriteBytes: float64(r.GetBytesWritten()),
RegionWriteKeys: float64(r.GetKeysWritten()),
RegionReadBytes: float64(r.GetBytesRead()),
RegionReadKeys: float64(r.GetKeysRead()),
}
}