dandri/livekit
1
0 Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-10-23 15:12:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

list-nodes command action (#87)

Browse Source 
* list-nodes command action

* refactor stats into a package

* store existing stats in redis, more refactoring

* dz review: atomic load
This commit is contained in:
Mathew Kamkar authored and GitHub committed 2021-08-24 10:14:39 -07:00
1 parent a8b0125e42
commit f5466c74e6
15 files changed
+428 -294
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

cmd/server/commands.go
+43
View File
@@ -3,10 +3,14 @@ package main
import (
"fmt"
"os"
"strconv"
"time"
"github.com/livekit/livekit-server/pkg/routing"
"github.com/livekit/livekit-server/pkg/service"
"github.com/livekit/protocol/auth"
"github.com/livekit/protocol/utils"
"github.com/olekukonko/tablewriter"
"github.com/urfave/cli/v2"
"gopkg.in/yaml.v3"
)
@@ -123,3 +127,42 @@ func createToken(c *cli.Context) error {
return nil
}
func listNodes(c *cli.Context) error {
conf, err := getConfig(c)
if err != nil {
return err
}
currentNode, err := routing.NewLocalNode(conf)
if err != nil {
return err
}
router, err := service.InitializeRouter(conf, currentNode)
if err != nil {
return err
}
nodes, err := router.ListNodes()
if err != nil {
return err
}
table := tablewriter.NewWriter(os.Stdout)
table.SetHeader([]string{"ID", "IP Address", "Num CPUs", "Num Clients", "Num Rooms", "Num Tracks In", "Num Tracks Out", "Started At", "Updated At"})
for _, node := range nodes {
cpus := strconv.Itoa(int(node.NumCpus))
clients := strconv.Itoa(int(node.Stats.NumClients))
rooms := strconv.Itoa(int(node.Stats.NumRooms))
tracksIn := strconv.Itoa(int(node.Stats.NumTracksIn))
tracksOut := strconv.Itoa(int(node.Stats.NumTracksOut))
startedAt := time.Unix(node.Stats.StartedAt, 0).String()
updatedAt := time.Unix(node.Stats.UpdatedAt, 0).String()
table.Append([]string{node.Id, node.Ip, cpus, clients, rooms, tracksIn, tracksOut, startedAt, updatedAt})
}
table.Render()
return nil
}
cmd/server/main.go
+5
View File
@@ -123,6 +123,11 @@ func main() {
},
},
},
{
Name: "list-nodes",
Usage: "list all nodes",
Action: listNodes,
},
},
Version: version.Version,
}
go.mod
+5 -2
View File
@@ -9,6 +9,7 @@ require (
github.com/go-logr/logr v1.0.0
github.com/go-logr/zapr v1.0.0
github.com/go-redis/redis/v8 v8.7.1
github.com/google/subcommands v1.2.0 // indirect
github.com/google/wire v0.5.0
github.com/gorilla/websocket v1.4.2
github.com/jxskiss/base62 v0.0.0-20191017122030-4f11678b909b
@@ -16,6 +17,7 @@ require (
github.com/magefile/mage v1.11.0
github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0
github.com/mitchellh/go-homedir v1.1.0
github.com/olekukonko/tablewriter v0.0.5
github.com/pion/ice/v2 v2.1.10
github.com/pion/interceptor v0.0.13
github.com/pion/ion-sfu v1.10.5
@@ -38,8 +40,9 @@ require (
github.com/urfave/negroni v1.0.0
go.uber.org/zap v1.18.1
golang.org/x/crypto v0.0.0-20210506145944-38f3c27a63bf // indirect
golang.org/x/sys v0.0.0-20210601080250-7ecdf8ef093b // indirect
golang.org/x/tools v0.1.2 // indirect
golang.org/x/mod v0.5.0 // indirect
golang.org/x/sys v0.0.0-20210816183151-1e6c022a8912 // indirect
golang.org/x/tools v0.1.5 // indirect
google.golang.org/protobuf v1.27.1
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b
)
go.sum
+12 -6
View File
@@ -158,8 +158,9 @@ github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXi
github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
github.com/google/pprof v0.0.0-20190515194954-54271f7e092f/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
github.com/google/subcommands v1.0.1 h1:/eqq+otEXm5vhfBrbREPCSVQbvofip6kIz+mX5TUH7k=
github.com/google/subcommands v1.0.1/go.mod h1:ZjhPrFU+Olkh9WazFPsl27BQ4UPiG37m3yTrtFlrHVk=
github.com/google/subcommands v1.2.0 h1:vWQspBTo2nEqTUFita5/KeEWlUL8kQObDFbub/EN9oE=
github.com/google/subcommands v1.2.0/go.mod h1:ZjhPrFU+Olkh9WazFPsl27BQ4UPiG37m3yTrtFlrHVk=
github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.2.0 h1:qJYtXnJRWmpe7m/3XlyhrsLrEURqHRM2kxzoxXqyUDs=
@@ -249,6 +250,8 @@ github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaO
github.com/mattn/go-isatty v0.0.3/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
github.com/mattn/go-isatty v0.0.4/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
github.com/mattn/go-runewidth v0.0.2/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU=
github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0=
github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0 h1:8E6DrFvII6QR4eJ3PkFvV+lc03P+2qwqTPLm1ax7694=
@@ -285,6 +288,8 @@ github.com/oklog/oklog v0.3.2/go.mod h1:FCV+B7mhrz4o+ueLpx+KqkyXRGMWOYEvfiXtdGtb
github.com/oklog/run v1.0.0/go.mod h1:dlhp/R75TPv97u0XWUtDeV/lRKWPKSdTuV0TZvrmrQA=
github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
github.com/olekukonko/tablewriter v0.0.0-20170122224234-a0225b3f23b5/go.mod h1:vsDQFd/mU46D+Z4whnwzcISnGGzXWMclvtLoiIKAKIo=
github.com/olekukonko/tablewriter v0.0.5 h1:P2Ga83D34wi1o9J6Wh1mRuqd4mF/x/lgBS7N7AbDhec=
github.com/olekukonko/tablewriter v0.0.5/go.mod h1:hPp6KlRPjbx+hW8ykQs1w3UBbZlj6HuIJcUGPhkA7kY=
github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk=
@@ -527,8 +532,9 @@ golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKG
golang.org/x/mod v0.1.0/go.mod h1:0QHyrYULN0/3qlju5TqG8bIK38QM8yzMo5ekMj3DlcY=
golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
golang.org/x/mod v0.4.2 h1:Gz96sIWK3OalVv/I/qNygP42zyoKp3xptRVCWRFEBvo=
golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
golang.org/x/mod v0.5.0 h1:UG21uOlmZabA4fW5i7ZX6bjw1xELEGg/ZLgZq9auk/Q=
golang.org/x/mod v0.5.0/go.mod h1:5OXOZSfqPIIbmVBIIKWRFfZjPR0E5r58TLhUjH0a2Ro=
golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -611,8 +617,8 @@ golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7w
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=
golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20210601080250-7ecdf8ef093b h1:qh4f65QIVFjq9eBURLEYWqaEXmOyqdUyiBSgaXWccWk=
golang.org/x/sys v0.0.0-20210601080250-7ecdf8ef093b/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20210816183151-1e6c022a8912 h1:uCLL3g5wH2xjxVREVuAbP9JM5PPKjRbXKRa6IBjkzmU=
golang.org/x/sys v0.0.0-20210816183151-1e6c022a8912/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -652,8 +658,8 @@ golang.org/x/tools v0.0.0-20200103221440-774c71fcf114/go.mod h1:TB2adYChydJhpapK
golang.org/x/tools v0.0.0-20201023174141-c8cfbd0f21e6/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0=
golang.org/x/tools v0.1.2 h1:kRBLX7v7Af8W7Gdbbc908OJcdgtK8bOz9Uaj8/F1ACA=
golang.org/x/tools v0.1.2/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
golang.org/x/tools v0.1.5 h1:ouewzE6p+/VEB31YYnTbEJdi8pFqKp4P4n85vwo3DHA=
golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
pkg/routing/redisrouter.go
+3 -3
View File
@@ -8,10 +8,10 @@ import (
"github.com/pkg/errors"
"google.golang.org/protobuf/proto"
"github.com/livekit/protocol/utils"
"github.com/livekit/livekit-server/pkg/logger"
"github.com/livekit/livekit-server/pkg/utils/stats"
livekit "github.com/livekit/livekit-server/proto"
"github.com/livekit/protocol/utils"
)
const (
@@ -309,7 +309,7 @@ func (r *RedisRouter) statsWorker() {
// update periodically seconds
select {
case <-time.After(statsUpdateInterval):
r.currentNode.Stats.UpdatedAt = time.Now().Unix()
stats.UpdateCurrentNodeStats(r.currentNode.Stats)
if err := r.RegisterNode(); err != nil {
logger.Errorw("could not update node", err, "nodeID", r.currentNode.Id)
}
pkg/rtc/mediatrack.go
+3 -2
View File
@@ -17,6 +17,7 @@ import (
"github.com/livekit/livekit-server/pkg/config"
"github.com/livekit/livekit-server/pkg/logger"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/pkg/utils/stats"
livekit "github.com/livekit/livekit-server/proto"
)
@@ -59,7 +60,7 @@ type MediaTrackParams struct {
BufferFactory *buffer.Factory
ReceiverConfig ReceiverConfig
AudioConfig config.AudioConfig
Stats *RoomStatsReporter
Stats *stats.RoomStatsReporter
Width uint32
Height uint32
}
@@ -244,7 +245,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
buff, rtcpReader := t.params.BufferFactory.GetBufferPair(uint32(track.SSRC()))
buff.OnFeedback(func(fb []rtcp.Packet) {
if t.params.Stats != nil {
t.params.Stats.incoming.HandleRTCP(fb)
t.params.Stats.Incoming.HandleRTCP(fb)
}
// feedback for the source RTCP
t.params.RTCPChan <- fb
pkg/rtc/participant.go
+2 -1
View File
@@ -20,6 +20,7 @@ import (
"github.com/livekit/livekit-server/pkg/logger"
"github.com/livekit/livekit-server/pkg/routing"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/pkg/utils/stats"
livekit "github.com/livekit/livekit-server/proto"
"github.com/livekit/livekit-server/version"
)
@@ -36,7 +37,7 @@ type ParticipantParams struct {
Sink routing.MessageSink
AudioConfig config.AudioConfig
ProtocolVersion types.ProtocolVersion
Stats *RoomStatsReporter
Stats *stats.RoomStatsReporter
ThrottleConfig config.PLIThrottleConfig
EnabledCodecs []*livekit.Codec
Hidden bool
pkg/rtc/room.go
+8 -7
View File
@@ -14,6 +14,7 @@ import (
"github.com/livekit/livekit-server/pkg/config"
"github.com/livekit/livekit-server/pkg/logger"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/livekit-server/pkg/utils/stats"
livekit "github.com/livekit/livekit-server/proto"
)
@@ -42,7 +43,7 @@ type Room struct {
// for active speaker updates
audioConfig *config.AudioConfig
statsReporter *RoomStatsReporter
statsReporter *stats.RoomStatsReporter
onParticipantChanged func(p types.Participant)
onClose func()
@@ -58,7 +59,7 @@ func NewRoom(room *livekit.Room, config WebRTCConfig, iceServers []*livekit.ICES
config: config,
iceServers: iceServers,
audioConfig: audioConfig,
statsReporter: NewRoomStatsReporter(room.Name),
statsReporter: stats.NewRoomStatsReporter(room.Name),
participants: make(map[string]types.Participant),
participantOpts: make(map[string]*ParticipantOptions),
bufferFactory: buffer.NewBufferFactory(config.Receiver.packetBufferSize, logger.GetLogger()),
@@ -111,7 +112,7 @@ func (r *Room) GetActiveSpeakers() []*livekit.SpeakerInfo {
return speakers
}
func (r *Room) GetStatsReporter() *RoomStatsReporter {
func (r *Room) GetStatsReporter() *stats.RoomStatsReporter {
return r.statsReporter
}
@@ -336,12 +337,12 @@ func (r *Room) Close() {
}
}
func (r *Room) GetIncomingStats() PacketStats {
return *r.statsReporter.incoming
func (r *Room) GetIncomingStats() stats.PacketStats {
return *r.statsReporter.Incoming
}
func (r *Room) GetOutgoingStats() PacketStats {
return *r.statsReporter.outgoing
func (r *Room) GetOutgoingStats() stats.PacketStats {
return *r.statsReporter.Outgoing
}
func (r *Room) OnClose(f func()) {
pkg/rtc/stats.go
-267
View File
@@ -1,267 +0,0 @@
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"
)
const livekitNamespace = "livekit"
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)
}
// 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) {
packetTotal.WithLabelValues(s.direction).Add(float64(count))
atomic.AddUint64(&s.PacketTotal, count)
}
func (s *PacketStats) IncrementNack(count uint64) {
nackTotal.WithLabelValues(s.direction).Add(float64(count))
atomic.AddUint64(&s.NackTotal, count)
}
func (s *PacketStats) IncrementPLI(count uint64) {
pliTotal.WithLabelValues(s.direction).Add(float64(count))
atomic.AddUint64(&s.PLITotal, count)
}
func (s *PacketStats) IncrementFIR(count uint64) {
firTotal.WithLabelValues(s.direction).Add(float64(count))
atomic.AddUint64(&s.FIRTotal, count)
}
func (s *PacketStats) HandleRTCP(pkts []rtcp.Packet) {
for _, rtcpPacket := range pkts {
switch rtcpPacket.(type) {
case *rtcp.TransportLayerNack:
s.IncrementNack(1)
case *rtcp.PictureLossIndication:
s.IncrementPLI(1)
case *rtcp.FullIntraRequest:
s.IncrementFIR(1)
}
}
}
func (s PacketStats) Copy() *PacketStats {
return &PacketStats{
roomName: s.roomName,
direction: s.direction,
PacketBytes: atomic.LoadUint64(&s.PacketBytes),
PacketTotal: atomic.LoadUint64(&s.PacketTotal),
NackTotal: atomic.LoadUint64(&s.NackTotal),
PLITotal: atomic.LoadUint64(&s.PLITotal),
FIRTotal: atomic.LoadUint64(&s.FIRTotal),
}
}
// StatsBufferWrapper wraps a buffer factory so we could get information on
// incoming packets
type StatsBufferWrapper struct {
createBufferFunc func(packetType packetio.BufferPacketType, ssrc uint32) io.ReadWriteCloser
stats *PacketStats
}
func (w *StatsBufferWrapper) CreateBuffer(packetType packetio.BufferPacketType, ssrc uint32) io.ReadWriteCloser {
writer := w.createBufferFunc(packetType, ssrc)
if packetType == packetio.RTPBufferPacket {
// wrap this in a counter class
return &rtpReporterWriter{
ReadWriteCloser: writer,
stats: w.stats,
}
}
return writer
}
type rtpReporterWriter struct {
io.ReadWriteCloser
stats *PacketStats
}
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)
}
// StatsInterceptor is created for each participant to keep of track of outgoing stats
// it adheres to Pion interceptor interface
type StatsInterceptor struct {
interceptor.NoOp
reporter *RoomStatsReporter
}
func NewStatsInterceptor(reporter *RoomStatsReporter) *StatsInterceptor {
return &StatsInterceptor{
reporter: reporter,
}
}
// BindRTCPWriter lets you modify any outgoing RTCP packets. It is called once per PeerConnection. The returned method
// will be called once per packet batch.
func (s *StatsInterceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interceptor.RTCPWriter {
return interceptor.RTCPWriterFunc(func(pkts []rtcp.Packet, attributes interceptor.Attributes) (int, error) {
s.reporter.outgoing.HandleRTCP(pkts)
return writer.Write(pkts, attributes)
})
}
// BindLocalStream lets you modify any outgoing RTP packets. It is called once for per LocalStream. The returned method
// will be called once per rtp packet.
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)
})
}
pkg/rtc/transport.go
+6 -5
View File
@@ -9,6 +9,7 @@ import (
"github.com/pion/webrtc/v3"
"github.com/livekit/livekit-server/pkg/logger"
"github.com/livekit/livekit-server/pkg/utils/stats"
livekit "github.com/livekit/livekit-server/proto"
)
@@ -41,7 +42,7 @@ type PCTransport struct {
type TransportParams struct {
Target livekit.SignalTarget
Config *WebRTCConfig
Stats *RoomStatsReporter
Stats *stats.RoomStatsReporter
EnabledCodecs []*livekit.Codec
}
@@ -59,9 +60,9 @@ func newPeerConnection(params TransportParams) (*webrtc.PeerConnection, *webrtc.
se := params.Config.SettingEngine
se.DisableMediaEngineCopy(true)
if params.Stats != nil && se.BufferFactory != nil {
wrapper := &StatsBufferWrapper{
createBufferFunc: se.BufferFactory,
stats: params.Stats.incoming,
wrapper := &stats.StatsBufferWrapper{
CreateBufferFunc: se.BufferFactory,
Stats: params.Stats.Incoming,
}
se.BufferFactory = wrapper.CreateBuffer
}
@@ -69,7 +70,7 @@ func newPeerConnection(params TransportParams) (*webrtc.PeerConnection, *webrtc.
ir := &interceptor.Registry{}
if params.Stats != nil && params.Target == livekit.SignalTarget_SUBSCRIBER {
// only capture subscriber for outbound streams
ir.Add(NewStatsInterceptor(params.Stats))
ir.Add(stats.NewStatsInterceptor(params.Stats))
}
api := webrtc.NewAPI(
webrtc.WithMediaEngine(me),
pkg/service/wire.go
+11
View File
@@ -17,3 +17,14 @@ func InitializeServer(conf *config.Config, keyProvider auth.KeyProvider, current
)
return &LivekitServer{}, nil
}
func InitializeRouter(conf *config.Config, currentNode routing.LocalNode) (routing.Router, error) {
wire.Build(
wire.NewSet(
createRedisClient,
createRouter,
),
)
return nil, nil
}
pkg/service/wire_gen.go
+11 -1
View File
@@ -1,7 +1,8 @@
// Code generated by Wire. DO NOT EDIT.
//go:generate go run github.com/google/wire/cmd/wire
//+build !wireinject
//go:build !wireinject
// +build !wireinject
package service
@@ -44,3 +45,12 @@ func InitializeServer(conf *config.Config, keyProvider auth.KeyProvider, current
}
return livekitServer, nil
}
func InitializeRouter(conf *config.Config, currentNode routing.LocalNode) (routing.Router, error) {
client, err := createRedisClient(conf)
if err != nil {
return nil, err
}
router := createRouter(client, currentNode)
return router, nil
}
pkg/utils/stats/packetstats.go
+112
View File
@@ -0,0 +1,112 @@
package stats
import (
"sync/atomic"
"github.com/pion/rtcp"
"github.com/prometheus/client_golang/prometheus"
)
var (
promPacketTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: livekitNamespace,
Subsystem: "packet",
Name: "total",
}, promLabels)
promPacketBytes = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: livekitNamespace,
Subsystem: "packet",
Name: "bytes",
}, promLabels)
promNackTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: livekitNamespace,
Subsystem: "nack",
Name: "total",
}, promLabels)
promPliTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: livekitNamespace,
Subsystem: "pli",
Name: "total",
}, promLabels)
promFirTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: livekitNamespace,
Subsystem: "fir",
Name: "total",
}, promLabels)
)
func initPacketStats() {
prometheus.MustRegister(promPacketTotal)
prometheus.MustRegister(promPacketBytes)
prometheus.MustRegister(promNackTotal)
prometheus.MustRegister(promPliTotal)
prometheus.MustRegister(promFirTotal)
}
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) {
promPacketBytes.WithLabelValues(s.direction).Add(float64(bytes))
atomic.AddUint64(&s.PacketBytes, bytes)
}
func (s *PacketStats) IncrementPackets(count uint64) {
promPacketTotal.WithLabelValues(s.direction).Add(float64(count))
atomic.AddUint64(&s.PacketTotal, count)
}
func (s *PacketStats) IncrementNack(count uint64) {
promNackTotal.WithLabelValues(s.direction).Add(float64(count))
atomic.AddUint64(&s.NackTotal, count)
}
func (s *PacketStats) IncrementPLI(count uint64) {
promPliTotal.WithLabelValues(s.direction).Add(float64(count))
atomic.AddUint64(&s.PLITotal, count)
}
func (s *PacketStats) IncrementFIR(count uint64) {
promFirTotal.WithLabelValues(s.direction).Add(float64(count))
atomic.AddUint64(&s.FIRTotal, count)
}
func (s *PacketStats) HandleRTCP(pkts []rtcp.Packet) {
for _, rtcpPacket := range pkts {
switch rtcpPacket.(type) {
case *rtcp.TransportLayerNack:
s.IncrementNack(1)
case *rtcp.PictureLossIndication:
s.IncrementPLI(1)
case *rtcp.FullIntraRequest:
s.IncrementFIR(1)
}
}
}
func (s PacketStats) Copy() *PacketStats {
return &PacketStats{
roomName: s.roomName,
direction: s.direction,
PacketBytes: atomic.LoadUint64(&s.PacketBytes),
PacketTotal: atomic.LoadUint64(&s.PacketTotal),
NackTotal: atomic.LoadUint64(&s.NackTotal),
PLITotal: atomic.LoadUint64(&s.PLITotal),
FIRTotal: atomic.LoadUint64(&s.FIRTotal),
}
}
pkg/utils/stats/roomstatsreporter.go
+112
View File
@@ -0,0 +1,112 @@
package stats
import (
"sync/atomic"
"time"
"github.com/prometheus/client_golang/prometheus"
)
var (
roomTotal int32
participantTotal int32
trackPublishedTotal int32
trackSubscribedTotal int32
promRoomTotal = prometheus.NewGauge(prometheus.GaugeOpts{
Namespace: livekitNamespace,
Subsystem: "room",
Name: "total",
})
promRoomDuration = 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,
},
})
promParticipantTotal = prometheus.NewGauge(prometheus.GaugeOpts{
Namespace: livekitNamespace,
Subsystem: "participant",
Name: "total",
})
promTrackPublishedTotal = prometheus.NewGaugeVec(prometheus.GaugeOpts{
Namespace: livekitNamespace,
Subsystem: "track",
Name: "published_total",
}, []string{"kind"})
promTrackSubscribedTotal = prometheus.NewGaugeVec(prometheus.GaugeOpts{
Namespace: livekitNamespace,
Subsystem: "track",
Name: "subscribed_total",
}, []string{"kind"})
)
func initRoomStatsReporter() {
prometheus.MustRegister(promRoomTotal)
prometheus.MustRegister(promRoomDuration)
prometheus.MustRegister(promParticipantTotal)
prometheus.MustRegister(promTrackPublishedTotal)
prometheus.MustRegister(promTrackSubscribedTotal)
}
// 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()
promRoomTotal.Add(1)
atomic.AddInt32(&roomTotal, 1)
}
func (r *RoomStatsReporter) RoomEnded() {
if !r.startedAt.IsZero() {
promRoomDuration.Observe(float64(time.Now().Sub(r.startedAt)) / float64(time.Second))
}
promRoomTotal.Sub(1)
atomic.AddInt32(&roomTotal, -1)
}
func (r *RoomStatsReporter) AddParticipant() {
promParticipantTotal.Add(1)
atomic.AddInt32(&participantTotal, 1)
}
func (r *RoomStatsReporter) SubParticipant() {
promParticipantTotal.Sub(1)
atomic.AddInt32(&participantTotal, -1)
}
func (r *RoomStatsReporter) AddPublishedTrack(kind string) {
promTrackPublishedTotal.WithLabelValues(kind).Add(1)
atomic.AddInt32(&trackPublishedTotal, 1)
}
func (r *RoomStatsReporter) SubPublishedTrack(kind string) {
promTrackPublishedTotal.WithLabelValues(kind).Sub(1)
atomic.AddInt32(&trackPublishedTotal, -1)
}
func (r *RoomStatsReporter) AddSubscribedTrack(kind string) {
promTrackSubscribedTotal.WithLabelValues(kind).Add(1)
atomic.AddInt32(&trackSubscribedTotal, 1)
}
func (r *RoomStatsReporter) SubSubscribedTrack(kind string) {
promTrackSubscribedTotal.WithLabelValues(kind).Sub(1)
atomic.AddInt32(&trackSubscribedTotal, -1)
}
pkg/utils/stats/stats.go
+95
View File
@@ -0,0 +1,95 @@
package stats
import (
"io"
"sync/atomic"
"time"
livekit "github.com/livekit/livekit-server/proto"
"github.com/pion/interceptor"
"github.com/pion/rtcp"
"github.com/pion/rtp"
"github.com/pion/transport/packetio"
)
const livekitNamespace = "livekit"
var (
promLabels = []string{"direction"}
)
func init() {
initPacketStats()
initRoomStatsReporter()
}
// StatsBufferWrapper wraps a buffer factory so we could get information on
// incoming packets
type StatsBufferWrapper struct {
CreateBufferFunc func(packetType packetio.BufferPacketType, ssrc uint32) io.ReadWriteCloser
Stats *PacketStats
}
func (w *StatsBufferWrapper) CreateBuffer(packetType packetio.BufferPacketType, ssrc uint32) io.ReadWriteCloser {
writer := w.CreateBufferFunc(packetType, ssrc)
if packetType == packetio.RTPBufferPacket {
// wrap this in a counter class
return &rtpReporterWriter{
ReadWriteCloser: writer,
stats: w.Stats,
}
}
return writer
}
type rtpReporterWriter struct {
io.ReadWriteCloser
stats *PacketStats
}
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)
}
// StatsInterceptor is created for each participant to keep of track of outgoing stats
// it adheres to Pion interceptor interface
type StatsInterceptor struct {
interceptor.NoOp
reporter *RoomStatsReporter
}
func NewStatsInterceptor(reporter *RoomStatsReporter) *StatsInterceptor {
return &StatsInterceptor{
reporter: reporter,
}
}
// BindRTCPWriter lets you modify any outgoing RTCP packets. It is called once per PeerConnection. The returned method
// will be called once per packet batch.
func (s *StatsInterceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interceptor.RTCPWriter {
return interceptor.RTCPWriterFunc(func(pkts []rtcp.Packet, attributes interceptor.Attributes) (int, error) {
s.reporter.Outgoing.HandleRTCP(pkts)
return writer.Write(pkts, attributes)
})
}
// BindLocalStream lets you modify any outgoing RTP packets. It is called once for per LocalStream. The returned method
// will be called once per rtp packet.
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)
})
}
func UpdateCurrentNodeStats(nodeStats *livekit.NodeStats) {
nodeStats.NumClients = uint32(atomic.LoadInt32(&participantTotal))
nodeStats.NumRooms = uint32(atomic.LoadInt32(&roomTotal))
nodeStats.NumTracksIn = uint32(atomic.LoadInt32(&trackPublishedTotal))
nodeStats.NumTracksOut = uint32(atomic.LoadInt32(&trackSubscribedTotal))
nodeStats.UpdatedAt = time.Now().Unix()
}