diff --git a/cmd/server/commands.go b/cmd/server/commands.go index d71c3860a..b7732a468 100644 --- a/cmd/server/commands.go +++ b/cmd/server/commands.go @@ -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 +} diff --git a/cmd/server/main.go b/cmd/server/main.go index b3a6a1f96..8d4a2556f 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -123,6 +123,11 @@ func main() { }, }, }, + { + Name: "list-nodes", + Usage: "list all nodes", + Action: listNodes, + }, }, Version: version.Version, } diff --git a/go.mod b/go.mod index 908eb0f98..55db9ea7b 100644 --- a/go.mod +++ b/go.mod @@ -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 ) diff --git a/go.sum b/go.sum index 039c530e4..b2e2eb113 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/pkg/routing/redisrouter.go b/pkg/routing/redisrouter.go index 9f6f63bc7..cd4b8da5d 100644 --- a/pkg/routing/redisrouter.go +++ b/pkg/routing/redisrouter.go @@ -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) } diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 2eeff2aa4..b75eb6d50 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -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 diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 431e0da4d..f2268187d 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -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 diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index ef053c994..6a520405d 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -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()) { diff --git a/pkg/rtc/stats.go b/pkg/rtc/stats.go deleted file mode 100644 index 52bf7e940..000000000 --- a/pkg/rtc/stats.go +++ /dev/null @@ -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) - }) -} diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 644174751..e7988e7a3 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -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), diff --git a/pkg/service/wire.go b/pkg/service/wire.go index 0175f0489..c679438cc 100644 --- a/pkg/service/wire.go +++ b/pkg/service/wire.go @@ -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 +} diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index 73b4d70a2..1cd5676e5 100644 --- a/pkg/service/wire_gen.go +++ b/pkg/service/wire_gen.go @@ -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 +} diff --git a/pkg/utils/stats/packetstats.go b/pkg/utils/stats/packetstats.go new file mode 100644 index 000000000..6a2f9d105 --- /dev/null +++ b/pkg/utils/stats/packetstats.go @@ -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), + } +} diff --git a/pkg/utils/stats/roomstatsreporter.go b/pkg/utils/stats/roomstatsreporter.go new file mode 100644 index 000000000..aba8c53b9 --- /dev/null +++ b/pkg/utils/stats/roomstatsreporter.go @@ -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) +} diff --git a/pkg/utils/stats/stats.go b/pkg/utils/stats/stats.go new file mode 100644 index 000000000..fc779b09e --- /dev/null +++ b/pkg/utils/stats/stats.go @@ -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() +}