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

*: add API concurrency metrics #7541

Merged
merged 28 commits into from
Dec 28, 2023
Merged

*: add API concurrency metrics #7541

Show file tree
Hide file tree
Changes from 16 commits
Commits
Show all changes
28 commits
Select commit Hold shift + click to select a range
b8be11c
refactor ratelimit pkg
CabinfeverB Oct 19, 2023
5c242b8
refactor ratelimit pkg
CabinfeverB Oct 19, 2023
b8cde6f
refactor test
CabinfeverB Oct 22, 2023
753c9d4
address comment
CabinfeverB Oct 23, 2023
9fb345f
Merge branch 'master' into rate_limit/refactor
CabinfeverB Oct 23, 2023
8c6a990
address comment
CabinfeverB Oct 23, 2023
12d9529
address comment
CabinfeverB Oct 23, 2023
a815282
address comment
CabinfeverB Nov 9, 2023
4748cf5
address comment
CabinfeverB Nov 9, 2023
5514e9b
concurrency metrics
CabinfeverB Dec 13, 2023
4d70445
fix conflict
CabinfeverB Dec 13, 2023
8da0b38
fix conflict
CabinfeverB Dec 13, 2023
ed2de59
address comment
CabinfeverB Dec 13, 2023
c620463
address comment
CabinfeverB Dec 14, 2023
31d416d
Merge branch 'master' into rate_limit/concurrency_limit
CabinfeverB Dec 18, 2023
12faadd
address comment
CabinfeverB Dec 18, 2023
d76914e
address comment
CabinfeverB Dec 21, 2023
ce33540
fix test
CabinfeverB Dec 21, 2023
b52c4f6
fix test
CabinfeverB Dec 21, 2023
e1ef4d2
Merge branch 'master' into rate_limit/concurrency_limit
CabinfeverB Dec 21, 2023
469005b
fix
CabinfeverB Dec 21, 2023
170acf1
merge master
CabinfeverB Dec 21, 2023
a1f8fd0
merge master
CabinfeverB Dec 26, 2023
bf14f9a
address comment
CabinfeverB Dec 27, 2023
62bc1d8
Merge branch 'master' into rate_limit/concurrency_limit
CabinfeverB Dec 28, 2023
b3311b7
fix
CabinfeverB Dec 28, 2023
d8cab6c
fix
CabinfeverB Dec 28, 2023
c34d2b1
address comment
CabinfeverB Dec 28, 2023
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
202 changes: 199 additions & 3 deletions metrics/grafana/pd.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8784,7 +8784,7 @@
"format": "time_series",
"intervalFactor": 2,
"legendFormat": "{{grpc_method}}",
"refId": "A",
"refId": "B",
"step": 4
}
],
Expand Down Expand Up @@ -8829,6 +8829,104 @@
"align": false,
"alignLevel": null
}
},
{
"aliasColors": {},
"bars": false,
"dashLength": 10,
"dashes": false,
"datasource": "${DS_TEST-CLUSTER}",
"description": "The concurrency number of each kind of gRPC commands",
"editable": true,
"error": false,
"fill": 0,
"grid": {},
"gridPos": {
"h": 8,
"w": 12,
"x": 0,
"y": 127
},
"id": 129,
"legend": {
"alignAsTable": true,
"avg": false,
"current": true,
"hideEmpty": true,
"hideZero": true,
"max": true,
"min": false,
"rightSide": true,
"show": true,
"sideWidth": 300,
"sortDesc": true,
"total": false,
"values": true
},
"lines": true,
"linewidth": 1,
"links": [],
"nullPointMode": "null as zero",
"paceLength": 10,
"percentage": false,
"pointradius": 5,
"points": false,
"renderer": "flot",
"seriesOverrides": [],
"spaceLength": 10,
"stack": false,
"steppedLine": false,
"targets": [
{
"expr": "pd_server_api_concurrency{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", kind=\"grpc\"}",
"format": "time_series",
"intervalFactor": 2,
"legendFormat": "{{api}}",
"refId": "E",
"step": 20
}
],
"thresholds": [],
"timeFrom": null,
"timeRegions": [],
"timeShift": null,
"title": "gRPC commands concurrency number",
"tooltip": {
"msResolution": false,
"shared": true,
"sort": 0,
"value_type": "cumulative"
},
"type": "graph",
"xaxis": {
"buckets": null,
"mode": "time",
"name": null,
"show": true,
"values": []
},
"yaxes": [
{
"format": "ops",
"label": null,
"logBase": 10,
"max": null,
"min": null,
"show": true
},
{
"format": "short",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
}
],
"yaxis": {
"align": false,
"alignLevel": null
}
}
],
"repeat": null,
Expand Down Expand Up @@ -8861,7 +8959,7 @@
"h": 8,
"w": 12,
"x": 0,
"y": 119
"y": 135
},
"id": 1461,
"legend": {
Expand Down Expand Up @@ -8957,7 +9055,7 @@
"h": 8,
"w": 12,
"x": 12,
"y": 119
"y": 135
},
"id": 1462,
"legend": {
Expand Down Expand Up @@ -9039,6 +9137,104 @@
"align": false,
"alignLevel": null
}
},
{
"aliasColors": {},
"bars": false,
"dashLength": 10,
"dashes": false,
"datasource": "${DS_TEST-CLUSTER}",
"description": "The concurrency number of each kind of HTTP commands",
"editable": true,
"error": false,
"fill": 0,
"grid": {},
"gridPos": {
"h": 8,
"w": 12,
"x": 0,
"y": 143
},
"id": 130,
"legend": {
"alignAsTable": true,
"avg": false,
"current": true,
"hideEmpty": true,
"hideZero": true,
"max": true,
"min": false,
"rightSide": true,
"show": true,
"sideWidth": 300,
"sortDesc": true,
"total": false,
"values": true
},
"lines": true,
"linewidth": 1,
"links": [],
"nullPointMode": "null as zero",
"paceLength": 10,
"percentage": false,
"pointradius": 5,
"points": false,
"renderer": "flot",
"seriesOverrides": [],
"spaceLength": 10,
"stack": false,
"steppedLine": false,
"targets": [
{
"expr": "pd_server_api_concurrency{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", kind=\"http\"}",
"format": "time_series",
"intervalFactor": 2,
"legendFormat": "{{api}}",
"refId": "E",
"step": 20
}
],
"thresholds": [],
"timeFrom": null,
"timeRegions": [],
"timeShift": null,
"title": "HTTP commands concurrency number",
"tooltip": {
"msResolution": false,
"shared": true,
"sort": 0,
"value_type": "cumulative"
},
"type": "graph",
"xaxis": {
"buckets": null,
"mode": "time",
"name": null,
"show": true,
"values": []
},
"yaxes": [
{
"format": "ops",
"label": null,
"logBase": 10,
"max": null,
"min": null,
"show": true
},
{
"format": "short",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
}
],
"yaxis": {
"align": false,
"alignLevel": null
}
}
],
"repeat": null,
Expand Down
20 changes: 19 additions & 1 deletion pkg/ratelimit/concurrency_limiter.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,18 +20,26 @@ type concurrencyLimiter struct {
mu syncutil.RWMutex
current uint64
limit uint64

// statistic
maxLimit uint64
}

func newConcurrencyLimiter(limit uint64) *concurrencyLimiter {
return &concurrencyLimiter{limit: limit}
}

const unlimit = uint64(0)

func (l *concurrencyLimiter) allow() bool {
l.mu.Lock()
defer l.mu.Unlock()

if l.current+1 <= l.limit {
if l.limit == unlimit || l.current+1 <= l.limit {
rleungx marked this conversation as resolved.
Show resolved Hide resolved
l.current++
if l.current > l.maxLimit {
l.maxLimit = l.current
}
return true
}
return false
Expand Down Expand Up @@ -66,3 +74,13 @@ func (l *concurrencyLimiter) getCurrent() uint64 {

return l.current
}

func (l *concurrencyLimiter) getMaxConcurrency() uint64 {
l.mu.RLock()
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should be Lock?

defer func() {
l.maxLimit = 0
l.mu.RUnlock()
}()

return l.maxLimit
}
9 changes: 9 additions & 0 deletions pkg/ratelimit/concurrency_limiter_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,9 +31,18 @@ func TestConcurrencyLimiter(t *testing.T) {
cl.release()
re.True(cl.allow())
re.Equal(uint64(10), cl.getLimit())
re.Equal(uint64(10), cl.getMaxConcurrency())
re.Equal(uint64(0), cl.getMaxConcurrency())
cl.setLimit(5)
re.Equal(uint64(5), cl.getLimit())
re.Equal(uint64(10), cl.getCurrent())
cl.release()
re.Equal(uint64(9), cl.getCurrent())
for i := 0; i < 9; i++ {
cl.release()
}
for i := 0; i < 5; i++ {
re.True(cl.allow())
}
re.Equal(uint64(5), cl.getMaxConcurrency())
}
48 changes: 45 additions & 3 deletions pkg/ratelimit/controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,11 @@
package ratelimit

import (
"context"
"sync"
"time"

"github.com/prometheus/client_golang/prometheus"
"golang.org/x/time/rate"
)

Expand All @@ -27,12 +30,51 @@ type Controller struct {
limiters sync.Map
// the label which is in labelAllowList won't be limited, and only inited by hard code.
labelAllowList map[string]struct{}

ctx context.Context
cancel context.CancelFunc
apiType string
concurrencyGauge *prometheus.GaugeVec
}

// NewController returns a global limiter which can be updated in the later.
func NewController() *Controller {
return &Controller{
labelAllowList: make(map[string]struct{}),
func NewController(ctx context.Context, typ string, concurrencyGauge *prometheus.GaugeVec) *Controller {
ctx, cancel := context.WithCancel(ctx)
l := &Controller{
ctx: ctx,
cancel: cancel,
labelAllowList: make(map[string]struct{}),
apiType: typ,
concurrencyGauge: concurrencyGauge,
}
if concurrencyGauge != nil {
go l.collectMetrics()
}
return l
}

func (l *Controller) Close() {
l.cancel()
}

func (l *Controller) collectMetrics() {
tricker := time.NewTicker(time.Second * 5)
defer tricker.Stop()
for {
select {
case <-l.ctx.Done():
return
case <-tricker.C:
l.limiters.Range(func(key, value any) bool {
limiter := value.(*limiter)
label := key.(string)
// Due to not in hot path, no need to save sub Gauge.
if con := limiter.getConcurrencyLimiter(); con != nil {
l.concurrencyGauge.WithLabelValues(l.apiType, label).Set(float64(con.getMaxConcurrency()))
}
return true
})
}
}
}

Expand Down
Loading
Loading