support for exposing stats via prometheus

This commit is contained in:
David Zhao committed 2021-05-11 22:04:59 -07:00
1 parent a8eba51c33
commit 681010072c
12 files changed
+217 -66

No files matched your search

+3
View File
@@ -12,6 +12,9 @@ redis:
# username: myuser
# password: mypassword
# when enabled, LiveKit will expose prometheus metrics on :6789/metrics
#prometheus_port: 6789
# WebRTC configuration
rtc:
# the main UDP port to transport WebRTC data
+1 -1
View File
@@ -27,8 +27,8 @@ require (
github.com/pion/turn/v2 v2.0.5
github.com/pion/webrtc/v3 v3.0.29
github.com/pkg/errors v0.9.1
github.com/prometheus/client_golang v1.10.0
github.com/prometheus/common v0.19.0 // indirect
github.com/prometheus/procfs v0.6.0 // indirect
github.com/russross/blackfriday/v2 v2.1.0 // indirect
github.com/stretchr/testify v1.7.0
github.com/thoas/go-funk v0.8.0
+4 -1
View File
@@ -363,8 +363,9 @@ github.com/prometheus/client_golang v0.9.3/go.mod h1:/TN21ttK/J9q6uSwhBd54HahCDf
github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
github.com/prometheus/client_golang v1.3.0/go.mod h1:hJaj2vgQTGQmVCsAACORcieXFeDPbaTKGT+JTgUa3og=
github.com/prometheus/client_golang v1.7.1/go.mod h1:PY5Wy2awLA44sXw4AOSfFBetzPP4j5+D6mVACh+pe2M=
github.com/prometheus/client_golang v1.9.0 h1:Rrch9mh17XcxvEu9D9DEpb4isxjGBtcevQjKvxPRQIU=
github.com/prometheus/client_golang v1.9.0/go.mod h1:FqZLKOZnGdFAhOK4nqGHa7D66IdsO+O441Eve7ptJDU=
github.com/prometheus/client_golang v1.10.0 h1:/o0BDeWzLWXNZ+4q5gXltUvaMpJqckTa+jTNoB+z4cg=
github.com/prometheus/client_golang v1.10.0/go.mod h1:WJM3cc3yu7XKBKa/I8WeZm+V3eltZnBwfENSU7mdogU=
github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
github.com/prometheus/client_model v0.0.0-20190115171406-56726106282f/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
@@ -379,6 +380,7 @@ github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y8
github.com/prometheus/common v0.7.0/go.mod h1:DjGbpBbp5NYNiECxcL/VnbXCCaQpKd3tt26CguLLsqA=
github.com/prometheus/common v0.10.0/go.mod h1:Tlit/dnDKsSWFlCLTWaA1cyBgKHSMdTB80sz/V91rCo=
github.com/prometheus/common v0.15.0/go.mod h1:U+gB1OBLb1lF3O42bTCL+FK18tX9Oar16Clt/msog/s=
github.com/prometheus/common v0.18.0/go.mod h1:U+gB1OBLb1lF3O42bTCL+FK18tX9Oar16Clt/msog/s=
github.com/prometheus/common v0.19.0 h1:Itb4+NjG9wRdkAWgVucbM/adyIXxEhbw0866e0uZE6A=
github.com/prometheus/common v0.19.0/go.mod h1:U+gB1OBLb1lF3O42bTCL+FK18tX9Oar16Clt/msog/s=
github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
@@ -594,6 +596,7 @@ golang.org/x/sys v0.0.0-20201214210602-f9fddec55a1e/go.mod h1:h1NjWce9XRLGQEsW7w
golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210119212857-b64e53b001e4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210309074719-68d13333faf2/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210420072515-93ed5bcd2bfe/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+9 -8
View File
@@ -9,14 +9,15 @@ import (
)
type Config struct {
Port uint32 `yaml:"port"`
RTC RTCConfig `yaml:"rtc"`
Redis RedisConfig `yaml:"redis"`
Audio AudioConfig `yaml:"audio"`
TURN TURNConfig `yaml:"turn"`
KeyFile string `yaml:"key_file"`
Keys map[string]string `yaml:"keys"`
LogLevel string `yaml:"log_level"`
Port uint32 `yaml:"port"`
PrometheusPort uint32 `yaml:"prometheus_port"`
RTC RTCConfig `yaml:"rtc"`
Redis RedisConfig `yaml:"redis"`
Audio AudioConfig `yaml:"audio"`
TURN TURNConfig `yaml:"turn"`
KeyFile string `yaml:"key_file"`
Keys map[string]string `yaml:"keys"`
LogLevel string `yaml:"log_level"`
Development bool `yaml:"development"`
}
+6 -3
View File
@@ -56,7 +56,7 @@ type MediaTrackParams struct {
BufferFactory *buffer.Factory
ReceiverConfig ReceiverConfig
AudioConfig config.AudioConfig
Stats *StatsReporter
Stats *RoomStatsReporter
}
func NewMediaTrack(track *webrtc.TrackRemote, params MediaTrackParams) *MediaTrack {
@@ -195,6 +195,7 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error {
sub.RemoveSubscribedTrack(t.params.ParticipantID, subTrack)
sub.Negotiate()
t.params.Stats.SubSubscribedTrack(t.kind.String())
})
t.subscribedTracks[sub.ID()] = subTrack
@@ -203,15 +204,15 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error {
sub.AddSubscribedTrack(t.params.ParticipantID, subTrack)
sub.Negotiate()
t.params.Stats.AddSubscribedTrack(t.kind.String())
return nil
}
// AddReceiver adds a new RTP receiver to the track, returns true if this is a new track
// AddReceiver adds a new RTP receiver to the track
func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.TrackRemote, twcc *twcc.Responder) {
t.lock.Lock()
defer t.lock.Unlock()
//rid := track.RID()
buff, rtcpReader := t.params.BufferFactory.GetBufferPair(uint32(track.SSRC()))
buff.OnFeedback(func(fb []rtcp.Packet) {
if t.params.Stats != nil {
@@ -260,10 +261,12 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
onclose := t.onClose
t.lock.Unlock()
t.RemoveAllSubscribers()
t.params.Stats.SubPublishedTrack(t.kind.String())
if onclose != nil {
onclose()
}
})
t.params.Stats.AddPublishedTrack(t.kind.String())
}
t.receiver.AddUpTrack(track, buff, true)
+1 -1
View File
@@ -35,7 +35,7 @@ type ParticipantParams struct {
Sink routing.MessageSink
AudioConfig config.AudioConfig
ProtocolVersion types.ProtocolVersion
Stats *StatsReporter
Stats *RoomStatsReporter
}
type ParticipantImpl struct {
+13 -11
View File
@@ -34,9 +34,7 @@ type Room struct {
audioUpdateInterval uint32
lastActiveSpeakers []*livekit.SpeakerInfo
// aggregate stats
incomingStats PacketStats
outgoingStats PacketStats
statsReporter *RoomStatsReporter
onParticipantChanged func(p types.Participant)
onClose func()
@@ -49,6 +47,7 @@ func NewRoom(room *livekit.Room, config WebRTCConfig, iceServers []*livekit.ICES
iceServers: iceServers,
audioUpdateInterval: audioUpdateInterval,
lock: sync.RWMutex{},
statsReporter: NewRoomStatsReporter(room.Name),
participants: make(map[string]types.Participant),
}
if r.EmptyTimeout == 0 {
@@ -57,6 +56,7 @@ func NewRoom(room *livekit.Room, config WebRTCConfig, iceServers []*livekit.ICES
if r.CreationTime == 0 {
r.CreationTime = time.Now().Unix()
}
r.statsReporter.RoomStarted()
go r.audioUpdateWorker()
return r
}
@@ -98,6 +98,10 @@ func (r *Room) GetActiveSpeakers() []*livekit.SpeakerInfo {
return speakers
}
func (r *Room) GetStatsReporter() *RoomStatsReporter {
return r.statsReporter
}
func (r *Room) FirstJoinedAt() int64 {
j := r.joinedAt.Load()
if t, ok := j.(int64); ok {
@@ -134,6 +138,8 @@ func (r *Room) Join(participant types.Participant) error {
r.joinedAt.Store(time.Now().Unix())
}
r.statsReporter.AddParticipant()
// it's important to set this before connection, we don't want to miss out on any publishedTracks
participant.OnTrackPublished(r.onTrackAdded)
participant.OnStateChange(func(p types.Participant, oldState livekit.ParticipantInfo_State) {
@@ -191,6 +197,7 @@ func (r *Room) RemoveParticipant(identity string) {
if !ok {
return
}
r.statsReporter.SubParticipant()
// send broadcast only if it's not already closed
sendUpdates := p.State() != livekit.ParticipantInfo_DISCONNECTED
@@ -204,12 +211,6 @@ func (r *Room) RemoveParticipant(identity string) {
// close participant as well
_ = p.Close()
// add participant stats
if pi, ok := p.(*ParticipantImpl); ok && pi.params.Stats != nil {
r.incomingStats.addFrom(pi.params.Stats.incoming)
r.outgoingStats.addFrom(pi.params.Stats.outgoing)
}
if len(r.participants) == 0 {
r.leftAt.Store(time.Now().Unix())
}
@@ -286,17 +287,18 @@ func (r *Room) Close() {
}
logger.Infow("closing room", "room", r.Sid, "name", r.Name)
r.statsReporter.RoomEnded()
if r.onClose != nil {
r.onClose()
}
}
func (r *Room) GetIncomingStats() PacketStats {
return r.incomingStats
return *r.statsReporter.incoming
}
func (r *Room) GetOutgoingStats() PacketStats {
return r.outgoingStats
return *r.statsReporter.outgoing
}
func (r *Room) OnClose(f func()) {
+155 -38
View File
@@ -3,46 +3,180 @@ package rtc
import (
"io"
"sync/atomic"
"time"
"github.com/pion/interceptor"
"github.com/pion/rtcp"
"github.com/pion/rtp"
"github.com/pion/transport/packetio"
"github.com/prometheus/client_golang/prometheus"
)
type PacketStats struct {
roomName string
identity string
kind string // incoming or outgoing
const livekitNamespace = "livekit"
PacketCount uint64 `json:"packetCount"`
NackCount uint64 `json:"nackCount"`
PLICount uint64 `json:"pliCount"`
FIRCount uint64 `json:"firCount"`
var (
promLabels = []string{"direction"}
packetTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: livekitNamespace,
Subsystem: "packet",
Name: "total",
}, promLabels)
packetBytes = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: livekitNamespace,
Subsystem: "packet",
Name: "bytes",
}, promLabels)
nackTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: livekitNamespace,
Subsystem: "nack",
Name: "total",
}, promLabels)
pliTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: livekitNamespace,
Subsystem: "pli",
Name: "total",
}, promLabels)
firTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: livekitNamespace,
Subsystem: "fir",
Name: "total",
}, promLabels)
roomTotal = prometheus.NewGauge(prometheus.GaugeOpts{
Namespace: livekitNamespace,
Subsystem: "room",
Name: "total",
})
roomDuration = prometheus.NewHistogram(prometheus.HistogramOpts{
Namespace: livekitNamespace,
Subsystem: "room",
Name: "duration_seconds",
Buckets: []float64{
5, 10, 60, 5 * 60, 10 * 60, 30 * 60, 60 * 60, 2 * 60 * 60, 5 * 60 * 60, 10 * 60 * 60,
},
})
participantTotal = prometheus.NewGauge(prometheus.GaugeOpts{
Namespace: livekitNamespace,
Subsystem: "participant",
Name: "total",
})
trackPublishedTotal = prometheus.NewGaugeVec(prometheus.GaugeOpts{
Namespace: livekitNamespace,
Subsystem: "track",
Name: "published_total",
}, []string{"kind"})
trackSubscribedTotal = prometheus.NewGaugeVec(prometheus.GaugeOpts{
Namespace: livekitNamespace,
Subsystem: "track",
Name: "subscribed_total",
}, []string{"kind"})
)
func init() {
prometheus.MustRegister(packetTotal)
prometheus.MustRegister(packetBytes)
prometheus.MustRegister(nackTotal)
prometheus.MustRegister(pliTotal)
prometheus.MustRegister(firTotal)
prometheus.MustRegister(roomTotal)
prometheus.MustRegister(roomDuration)
prometheus.MustRegister(participantTotal)
prometheus.MustRegister(trackPublishedTotal)
prometheus.MustRegister(trackSubscribedTotal)
}
func newPacketStats(room, identity, kind string) *PacketStats {
return &PacketStats{
roomName: room,
identity: identity,
kind: kind,
// RoomStatsReporter is created for each room
type RoomStatsReporter struct {
roomName string
startedAt time.Time
incoming *PacketStats
outgoing *PacketStats
}
func NewRoomStatsReporter(roomName string) *RoomStatsReporter {
return &RoomStatsReporter{
roomName: roomName,
incoming: newPacketStats(roomName, "incoming"),
outgoing: newPacketStats(roomName, "outgoing"),
}
}
func (r *RoomStatsReporter) RoomStarted() {
r.startedAt = time.Now()
roomTotal.Add(1)
}
func (r *RoomStatsReporter) RoomEnded() {
if !r.startedAt.IsZero() {
roomDuration.Observe(float64(time.Now().Sub(r.startedAt)) / float64(time.Second))
}
roomTotal.Sub(1)
}
func (r *RoomStatsReporter) AddParticipant() {
participantTotal.Add(1)
}
func (r *RoomStatsReporter) SubParticipant() {
participantTotal.Sub(1)
}
func (r *RoomStatsReporter) AddPublishedTrack(kind string) {
trackPublishedTotal.WithLabelValues(kind).Add(1)
}
func (r *RoomStatsReporter) SubPublishedTrack(kind string) {
trackPublishedTotal.WithLabelValues(kind).Sub(1)
}
func (r *RoomStatsReporter) AddSubscribedTrack(kind string) {
trackSubscribedTotal.WithLabelValues(kind).Add(1)
}
func (r *RoomStatsReporter) SubSubscribedTrack(kind string) {
trackSubscribedTotal.WithLabelValues(kind).Sub(1)
}
type PacketStats struct {
roomName string
direction string // incoming or outgoing
PacketBytes uint64 `json:"packetBytes"`
PacketTotal uint64 `json:"packetTotal"`
NackTotal uint64 `json:"nackTotal"`
PLITotal uint64 `json:"pliTotal"`
FIRTotal uint64 `json:"firTotal"`
}
func newPacketStats(room, direction string) *PacketStats {
return &PacketStats{
roomName: room,
direction: direction,
}
}
func (s *PacketStats) IncrementBytes(bytes uint64) {
packetBytes.WithLabelValues(s.direction).Add(float64(bytes))
atomic.AddUint64(&s.PacketBytes, bytes)
}
func (s *PacketStats) IncrementPackets(count uint64) {
atomic.AddUint64(&s.PacketCount, count)
packetTotal.WithLabelValues(s.direction).Add(float64(count))
atomic.AddUint64(&s.PacketTotal, count)
}
func (s *PacketStats) IncrementNack(count uint64) {
atomic.AddUint64(&s.NackCount, count)
nackTotal.WithLabelValues(s.direction).Add(float64(count))
atomic.AddUint64(&s.NackTotal, count)
}
func (s *PacketStats) IncrementPLI(count uint64) {
atomic.AddUint64(&s.PLICount, count)
pliTotal.WithLabelValues(s.direction).Add(float64(count))
atomic.AddUint64(&s.PLITotal, count)
}
func (s *PacketStats) IncrementFIR(count uint64) {
atomic.AddUint64(&s.FIRCount, count)
firTotal.WithLabelValues(s.direction).Add(float64(count))
atomic.AddUint64(&s.FIRTotal, count)
}
func (s *PacketStats) HandleRTCP(pkts []rtcp.Packet) {
@@ -58,25 +192,6 @@ func (s *PacketStats) HandleRTCP(pkts []rtcp.Packet) {
}
}
func (s *PacketStats) addFrom(o *PacketStats) {
s.PacketCount += atomic.LoadUint64(&o.PacketCount)
s.NackCount += atomic.LoadUint64(&o.NackCount)
s.PLICount += atomic.LoadUint64(&o.PLICount)
s.FIRCount += atomic.LoadUint64(&o.FIRCount)
}
type StatsReporter struct {
incoming *PacketStats
outgoing *PacketStats
}
func NewStatsReporter(roomName, identity string) *StatsReporter {
return &StatsReporter{
incoming: newPacketStats(roomName, identity, "incoming"),
outgoing: newPacketStats(roomName, identity, "outgoing"),
}
}
// StatsBufferWrapper wraps a buffer factory so we could get information on
// incoming packets
type StatsBufferWrapper struct {
@@ -103,6 +218,7 @@ type rtpReporterWriter struct {
func (w *rtpReporterWriter) Write(p []byte) (n int, err error) {
w.stats.IncrementPackets(1)
w.stats.IncrementBytes(uint64(len(p)))
return w.ReadWriteCloser.Write(p)
}
@@ -110,10 +226,10 @@ func (w *rtpReporterWriter) Write(p []byte) (n int, err error) {
// it adheres to Pion interceptor interface
type StatsInterceptor struct {
interceptor.NoOp
reporter *StatsReporter
reporter *RoomStatsReporter
}
func NewStatsInterceptor(reporter *StatsReporter) *StatsInterceptor {
func NewStatsInterceptor(reporter *RoomStatsReporter) *StatsInterceptor {
return &StatsInterceptor{
reporter: reporter,
}
@@ -133,6 +249,7 @@ func (s *StatsInterceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interce
func (s *StatsInterceptor) BindLocalStream(_ *interceptor.StreamInfo, writer interceptor.RTPWriter) interceptor.RTPWriter {
return interceptor.RTPWriterFunc(func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) {
s.reporter.outgoing.IncrementPackets(1)
s.reporter.outgoing.IncrementBytes(uint64(len(payload)))
return writer.Write(header, payload, attributes)
})
}
+1 -1
View File
@@ -41,7 +41,7 @@ type PCTransport struct {
type TransportParams struct {
Target livekit.SignalTarget
Config *WebRTCConfig
Stats *StatsReporter
Stats *RoomStatsReporter
}
func newPeerConnection(params TransportParams) (*webrtc.PeerConnection, *webrtc.MediaEngine, error) {
+1 -1
View File
@@ -109,7 +109,7 @@ func ToProtoTrackKind(kind webrtc.RTPCodecType) livekit.TrackType {
case webrtc.RTPCodecTypeAudio:
return livekit.TrackType_AUDIO
}
panic("unsupported track kind")
panic("unsupported track direction")
}
func IsEOF(err error) bool {
+1 -1
View File
@@ -239,7 +239,7 @@ func (r *RoomManager) StartSession(roomName string, pi routing.ParticipantInit,
Sink: responseSink,
AudioConfig: r.config.Audio,
ProtocolVersion: pv,
Stats: rtc.NewStatsReporter(roomName, pi.Identity),
Stats: room.GetStatsReporter(),
})
if err != nil {
logger.Errorw("could not create participant", "error", err)
+22
View File
@@ -9,6 +9,7 @@ import (
"time"
"github.com/pion/turn/v2"
"github.com/prometheus/client_golang/prometheus/promhttp"
"github.com/urfave/negroni"
"github.com/livekit/livekit-server/pkg/config"
@@ -25,6 +26,7 @@ type LivekitServer struct {
roomServer livekit.TwirpServer
rtcService *RTCService
httpServer *http.Server
promServer *http.Server
router routing.Router
roomManager *RoomManager
turnServer *turn.Server
@@ -71,6 +73,13 @@ func NewLivekitServer(conf *config.Config,
Handler: configureMiddlewares(mux, middlewares...),
}
if conf.PrometheusPort > 0 {
s.promServer = &http.Server{
Addr: fmt.Sprintf(":%d", conf.PrometheusPort),
Handler: promhttp.Handler(),
}
}
// hook up router to the RoomManager
router.OnNewParticipantRTC(roomManager.StartSession)
router.OnRTCMessage(roomManager.handleRTCMessage)
@@ -120,6 +129,16 @@ func (s *LivekitServer) Start() error {
return err
}
if s.promServer != nil {
promLn, err := net.Listen("tcp", s.promServer.Addr)
if err != nil {
return err
}
go func() {
_ = s.promServer.Serve(promLn)
}()
}
go func() {
values := []interface{}{
"address", s.httpServer.Addr,
@@ -137,6 +156,9 @@ func (s *LivekitServer) Start() error {
"rtc.port_range_end", s.config.RTC.ICEPortRangeEnd,
)
}
if s.config.PrometheusPort != 0 {
values = append(values, "prometheus_port", s.config.PrometheusPort)
}
logger.Infow("starting LiveKit server", values...)
if err := s.httpServer.Serve(ln); err != http.ErrServerClosed {
logger.Errorw("could not start server", "error", err)