Skip to content

Commit 7d05306

Browse files
PothulapatiPothulapati
authored
feat(tools/replay): Add pipeline latency distribution data (#4990)
feat(replay): add latency distributions * also add avg latency * also include pipeline range * display both at the end
1 parent 680722d commit 7d05306

File tree

4 files changed

+104
-1
lines changed

4 files changed

+104
-1
lines changed

‎tools/replay/go.mod‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ require (
1313
github.com/davecgh/go-spew v1.1.1 // indirect
1414
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
1515
github.com/gookit/color v1.4.2 // indirect
16+
github.com/influxdata/tdigest v0.0.1 // indirect
1617
github.com/mattn/go-runewidth v0.0.13 // indirect
1718
github.com/pmezard/go-difflib v1.0.0 // indirect
1819
github.com/rivo/uniseg v0.2.0 // indirect

‎tools/replay/go.sum‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,11 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
99
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
1010
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
1111
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
12+
github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
1213
github.com/gookit/color v1.4.2 h1:tXy44JFSFkKnELV6WaMo/lLfu/meqITX3iAV52do7lk=
1314
github.com/gookit/color v1.4.2/go.mod h1:fqRyamkC1W8uxl+lxCQxOT09l/vYfZ+QeiX3rKQHCoQ=
15+
github.com/influxdata/tdigest v0.0.1 h1:XpFptwYmnEKUqmkcDjrzffswZ3nvNeevbUSLPP/ZzIY=
16+
github.com/influxdata/tdigest v0.0.1/go.mod h1:Z0kXnxzbTC2qrx4NaIzYkE1k66+6oEDQTvL95hQFh5Y=
1417
github.com/mattn/go-runewidth v0.0.13 h1:lTGmDsbAYt5DmK6OnoV7EuIF1wEIFAcxld6ypU4OSgU=
1518
github.com/mattn/go-runewidth v0.0.13/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
1619
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -27,11 +30,15 @@ github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5Cc
2730
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
2831
github.com/xo/terminfo v0.0.0-20210125001918-ca9a967f8778 h1:QldyIu/L63oPpyvQmHgvgickp1Yw510KJOqX7H24mg8=
2932
github.com/xo/terminfo v0.0.0-20210125001918-ca9a967f8778/go.mod h1:2MuV+tbUrU1zIOPMxZ5EncGwgmMJsa+9ucAQZXxsObs=
33+
golang.org/x/exp v0.0.0-20180321215751-8460e604b9de/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
3034
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
3135
golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44 h1:Bli41pIlzTzf3KEY06n+xnzK/BESIg2ze4Pgfh/aI8c=
3236
golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
3337
golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d h1:SZxvLBoTP5yHO3Frd4z4vrF+DBX9vMVanchswa69toE=
3438
golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
39+
golang.org/x/tools v0.0.0-20180525024113-a5b4c53f6e8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
40+
gonum.org/v1/gonum v0.0.0-20181121035319-3f7ecaa7e8ca/go.mod h1:Y+Yx5eoAFn32cQvJDxZx5Dpnq+c3wtXuadVZAcxbbBo=
41+
gonum.org/v1/netlib v0.0.0-20181029234149-ec6d1f5cefe6/go.mod h1:wa6Ws7BG/ESfp6dHfk7C6KdzKA7wR7u/rKwOGE66zvw=
3542
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
3643
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
3744
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

‎tools/replay/main.go‎

Lines changed: 45 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,20 +19,62 @@ var fPace = flag.Bool("pace", true, "whether to pace the traffic according to th
1919
var fSkip = flag.Uint("skip", 0, "skip N records")
2020

2121
func RenderTable(area *pterm.AreaPrinter, files []string, workers []FileWorker) {
22-
tableData := pterm.TableData{{"file", "parsed", "processed", "delayed", "clients"}}
22+
tableData := pterm.TableData{{"file", "parsed", "processed", "delayed", "clients", "avg(us)", "p50(us)", "p75(us)", "p90(us)", "p99(us)"}}
2323
for i := range workers {
24+
workers[i].latencyMu.Lock()
25+
avg := 0.0
26+
if workers[i].latencyCount > 0 {
27+
avg = workers[i].latencySum / float64(workers[i].latencyCount)
28+
}
29+
p50 := workers[i].latencyDigest.Quantile(0.5)
30+
p75 := workers[i].latencyDigest.Quantile(0.75)
31+
p90 := workers[i].latencyDigest.Quantile(0.9)
32+
p99 := workers[i].latencyDigest.Quantile(0.99)
33+
workers[i].latencyMu.Unlock()
2434
tableData = append(tableData, []string{
2535
files[i],
2636
fmt.Sprint(atomic.LoadUint64(&workers[i].parsed)),
2737
fmt.Sprint(atomic.LoadUint64(&workers[i].processed)),
2838
fmt.Sprint(atomic.LoadUint64(&workers[i].delayed)),
2939
fmt.Sprint(atomic.LoadUint64(&workers[i].clients)),
40+
fmt.Sprintf("%.0f", avg),
41+
fmt.Sprintf("%.0f", p50),
42+
fmt.Sprintf("%.0f", p75),
43+
fmt.Sprintf("%.0f", p90),
44+
fmt.Sprintf("%.0f", p99),
3045
})
3146
}
3247
content, _ := pterm.DefaultTable.WithHasHeader().WithBoxed().WithData(tableData).Srender()
3348
area.Update(content)
3449
}
3550

51+
// RenderPipelineRangesTable renders the latency digests for each pipeline range
52+
func RenderPipelineRangesTable(area *pterm.AreaPrinter, files []string, workers []FileWorker) {
53+
tableData := pterm.TableData{{"file", "Pipeline Range", "p50(us)", "p75(us)", "p90(us)", "p99(us)"}}
54+
for i := range workers {
55+
workers[i].latencyMu.Lock()
56+
for _, rng := range pipelineRanges {
57+
if digest, ok := workers[i].perRange[rng.label]; ok {
58+
p50 := digest.Quantile(0.5)
59+
p75 := digest.Quantile(0.75)
60+
p90 := digest.Quantile(0.9)
61+
p99 := digest.Quantile(0.99)
62+
tableData = append(tableData, []string{
63+
files[i],
64+
rng.label,
65+
fmt.Sprintf("%.0f", p50),
66+
fmt.Sprintf("%.0f", p75),
67+
fmt.Sprintf("%.0f", p90),
68+
fmt.Sprintf("%.0f", p99),
69+
})
70+
}
71+
}
72+
workers[i].latencyMu.Unlock()
73+
}
74+
content, _ := pterm.DefaultTable.WithHasHeader().WithBoxed().WithData(tableData).Srender()
75+
area.Update(content)
76+
}
77+
3678
func Run(files []string) {
3779
timeOffset := time.Now().Add(500 * time.Millisecond).Sub(DetermineBaseTime(files))
3880
fmt.Println("Offset -> ", timeOffset)
@@ -64,6 +106,8 @@ func Run(files []string) {
64106
}
65107

66108
RenderTable(area, files, workers) // to show last stats
109+
areaPipelineRanges, _ := pterm.DefaultArea.WithCenter().Start()
110+
RenderPipelineRangesTable(areaPipelineRanges, files, workers) // to render per pipeline-range latency digests
67111
}
68112

69113
func Print(files []string) {

‎tools/replay/workers.go‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import (
1010
"sync/atomic"
1111
"time"
1212

13+
"github.com/influxdata/tdigest"
1314
"github.com/redis/go-redis/v9"
1415
)
1516

@@ -47,6 +48,18 @@ type ClientWorker struct {
4748
pipe redis.Pipeliner
4849
}
4950

51+
// Pipeline length ranges for summary
52+
var pipelineRanges = []struct {
53+
label string
54+
min int
55+
max int // inclusive, except last
56+
}{
57+
{"0-29", 0, 29},
58+
{"30-79", 30, 79},
59+
{"80-199", 80, 199},
60+
{"200+", 200, 1 << 30},
61+
}
62+
5063
// Handles a single file and distributes messages to clients
5164
type FileWorker struct {
5265
clientGroup sync.WaitGroup
@@ -56,6 +69,33 @@ type FileWorker struct {
5669
delayed uint64
5770
parsed uint64
5871
clients uint64
72+
73+
latencyDigest *tdigest.TDigest
74+
latencyMu sync.Mutex
75+
76+
latencySum float64 // sum of all batch latencies (microseconds)
77+
latencyCount uint64 // number of batches
78+
79+
// per-pipeline-range latency digests
80+
perRange map[string]*tdigest.TDigest
81+
}
82+
83+
// Helper function to track latency and update digests
84+
func trackLatency(worker *FileWorker, batchLatency float64, size int) {
85+
worker.latencyMu.Lock()
86+
defer worker.latencyMu.Unlock()
87+
worker.latencyDigest.Add(batchLatency, 1)
88+
worker.latencySum += batchLatency
89+
worker.latencyCount++
90+
// Add to per-range digest
91+
if worker.perRange != nil {
92+
for _, rng := range pipelineRanges {
93+
if size >= rng.min && size <= rng.max {
94+
worker.perRange[rng.label].Add(batchLatency, 1)
95+
break
96+
}
97+
}
98+
}
5999
}
60100

61101
func (c *ClientWorker) Run(pace bool, worker *FileWorker) {
@@ -79,13 +119,19 @@ func (c *ClientWorker) Run(pace bool, worker *FileWorker) {
79119

80120
if msg.HasMore == 0 {
81121
size := c.pipe.Len()
122+
start := time.Now()
82123
c.pipe.Exec(context.Background())
124+
batchLatency := float64(time.Since(start).Microseconds())
125+
trackLatency(worker, batchLatency, size)
83126
c.processed += uint(size)
84127
}
85128
}
86129

87130
if size := c.pipe.Len(); size >= 0 {
131+
start := time.Now()
88132
c.pipe.Exec(context.Background())
133+
batchLatency := float64(time.Since(start).Microseconds())
134+
trackLatency(worker, batchLatency, size)
89135
c.processed += uint(size)
90136
}
91137

@@ -106,6 +152,11 @@ func NewClient(w *FileWorker, pace bool) *ClientWorker {
106152
}
107153

108154
func (w *FileWorker) Run(file string, wg *sync.WaitGroup) {
155+
w.latencyDigest = tdigest.NewWithCompression(1000)
156+
w.perRange = make(map[string]*tdigest.TDigest)
157+
for _, rng := range pipelineRanges {
158+
w.perRange[rng.label] = tdigest.NewWithCompression(500)
159+
}
109160
clients := make(map[uint32]*ClientWorker, 0)
110161
recordId := uint64(0)
111162
err := parseRecords(file, func(r Record) bool {

0 commit comments

Comments
 (0)