char
/
dendrite
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Roomserver input backpressure metric 

Squashed commit of the following:

commit 56e934ac0aeedcfb2c072010959ba49734d4e0cb
Author: Neil Alexander <neilalexander@users.noreply.github.com>
Date:   Fri Jul 2 09:39:30 2021 +0100

    Fix metric

commit 3911f3a0c17b164b012e881c085ceca30f5de408
Author: Neil Alexander <neilalexander@users.noreply.github.com>
Date:   Fri Jul 2 09:36:29 2021 +0100

    Register correct metric

commit a9ddbfaed421538a701151801e9451198a8be4f3
Author: Neil Alexander <neilalexander@users.noreply.github.com>
Date:   Fri Jul 2 09:33:33 2021 +0100

    Try to capture RS input backpressure metric
main
Neil Alexander 2021-07-02 09:48:55 +01:00
parent 2647f6e9c5
commit 192a7a7923
No known key found for this signature in database
GPG Key ID: A02A2019A2BB0944
1 changed files with 21 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
roomserver/internal/input/input.go
View File

@ -28,6 +28,7 @@ import (
"github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/storage" "github.com/matrix-org/dendrite/roomserver/storage"
"github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/gomatrixserverlib"
"github.com/prometheus/client_golang/prometheus"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"go.uber.org/atomic" "go.uber.org/atomic"
) )
@ -64,6 +65,9 @@ func (w *inputWorker) start() {
if !ok { if !ok {
continue continue
} }
roomserverInputBackpressure.With(prometheus.Labels{
"room_id": task.event.Event.RoomID(),
}).Dec()
hooks.Run(hooks.KindNewEventReceived, task.event.Event) hooks.Run(hooks.KindNewEventReceived, task.event.Event)
_, task.err = w.r.processRoomEvent(task.ctx, task.event) _, task.err = w.r.processRoomEvent(task.ctx, task.event)
if task.err == nil { if task.err == nil {
@ -120,6 +124,20 @@ func (r *Inputer) WriteOutputEvents(roomID string, updates []api.OutputEvent) er
return errs return errs
} }
func init() {
prometheus.MustRegister(roomserverInputBackpressure)
}
var roomserverInputBackpressure = prometheus.NewGaugeVec(
prometheus.GaugeOpts{
Namespace: "dendrite",
Subsystem: "roomserver",
Name: "input_backpressure",
Help: "How many events are queued for input for a given room",
},
[]string{"room_id"},
)
// InputRoomEvents implements api.RoomserverInternalAPI // InputRoomEvents implements api.RoomserverInternalAPI
func (r *Inputer) InputRoomEvents( func (r *Inputer) InputRoomEvents(
_ context.Context, _ context.Context,
@ -164,6 +182,9 @@ func (r *Inputer) InputRoomEvents(
go worker.start() go worker.start()
} }
worker.input.push(tasks[i]) worker.input.push(tasks[i])
roomserverInputBackpressure.With(prometheus.Labels{
"room_id": roomID,
}).Inc()
} }
// Wait for all of the workers to return results about our tasks. // Wait for all of the workers to return results about our tasks.