mirror of
https://github.com/livekit/livekit.git
synced 2026-08-28 05:04:10 +00:00
Rework node stats a bit. (#3555)
* Rework node stats a bit. Related protocol PR - https://github.com/livekit/protocol/pull/1023 - Make a config for node stats measurements. Wanted to put the config in `routing` package, but a circular dependency forced me to put in config.go - Make rate calculations explicit, i. e. requested via config. Previously, it had some odd checks to decide when to calculate rate and it would have been calculating over different windows. - Report signal/data channel bytes every 5 seconds to stats collection module. Previously, it was doing it every 30 seconds and that meant some windows could have had a large spike NOTE: Still need to think about this for load calculations as a large number of participants leaving could flush in a small window and that could report a large spike in bytes/packets. Maybe need to ignore signal bytes for load calculation? * deps * use default node stats config if given config is nil * split out node stats into a struct for re-use * update config
This commit is contained in:
Raja Subramanian
+14
-5
@@ -27,6 +27,7 @@ import (
|
||||
"gopkg.in/yaml.v3"
|
||||
|
||||
"github.com/livekit/protocol/auth"
|
||||
"github.com/livekit/protocol/livekit"
|
||||
"github.com/livekit/protocol/utils"
|
||||
"github.com/livekit/protocol/utils/guid"
|
||||
|
||||
@@ -205,6 +206,10 @@ func listNodes(c *cli.Context) error {
|
||||
|
||||
for _, node := range nodes {
|
||||
stats := node.Stats
|
||||
rate := &livekit.NodeStatsRate{}
|
||||
if len(stats.Rates) > 0 {
|
||||
rate = stats.Rates[0]
|
||||
}
|
||||
|
||||
// Id and state
|
||||
idAndState := fmt.Sprintf("%s\n(%s)", node.Id, node.State.Enum().String())
|
||||
@@ -220,13 +225,17 @@ func listNodes(c *cli.Context) error {
|
||||
clientsAndTracks := fmt.Sprintf("%d\n%d / %d", stats.NumClients, stats.NumTracksIn, stats.NumTracksOut)
|
||||
|
||||
// Packet stats
|
||||
bytes := fmt.Sprintf("%sps / %sps\n%s / %s", humanize.Bytes(uint64(stats.BytesInPerSec)), humanize.Bytes(uint64(stats.BytesOutPerSec)),
|
||||
bytes := fmt.Sprintf("%sps / %sps\n%s / %s", humanize.Bytes(uint64(rate.BytesIn)), humanize.Bytes(uint64(rate.BytesOut)),
|
||||
humanize.Bytes(stats.BytesIn), humanize.Bytes(stats.BytesOut))
|
||||
packets := fmt.Sprintf("%s / %s\n%s / %s", humanize.Comma(int64(stats.PacketsInPerSec)), humanize.Comma(int64(stats.PacketsOutPerSec)),
|
||||
packets := fmt.Sprintf("%s / %s\n%s / %s", humanize.Comma(int64(rate.PacketsIn)), humanize.Comma(int64(rate.PacketsOut)),
|
||||
strings.TrimSpace(humanize.SIWithDigits(float64(stats.PacketsIn), 2, "")), strings.TrimSpace(humanize.SIWithDigits(float64(stats.PacketsOut), 2, "")))
|
||||
sysPackets := fmt.Sprintf("%.2f %%\n%v / %v", stats.SysPacketsDroppedPctPerSec*100, float64(stats.SysPacketsOutPerSec), float64(stats.SysPacketsDroppedPerSec))
|
||||
nacks := fmt.Sprintf("%.2f\n%s", stats.NackPerSec, strings.TrimSpace(humanize.SIWithDigits(float64(stats.NackTotal), 2, "")))
|
||||
retransmit := fmt.Sprintf("%.2f\n%s", stats.RetransmitPacketsOutPerSec, strings.TrimSpace(humanize.SIWithDigits(float64(stats.RetransmitPacketsOut), 2, "")))
|
||||
sysPacketsDroppedPct := float32(0)
|
||||
if rate.SysPacketsOut+rate.SysPacketsDropped > 0 {
|
||||
sysPacketsDroppedPct = float32(rate.SysPacketsDropped) / float32(rate.SysPacketsDropped+rate.SysPacketsOut)
|
||||
}
|
||||
sysPackets := fmt.Sprintf("%.2f %%\n%v / %v", sysPacketsDroppedPct*100, float64(rate.SysPacketsOut), float64(rate.SysPacketsDropped))
|
||||
nacks := fmt.Sprintf("%.2f\n%s", rate.NackTotal, strings.TrimSpace(humanize.SIWithDigits(float64(stats.NackTotal), 2, "")))
|
||||
retransmit := fmt.Sprintf("%.2f\n%s", rate.RetransmitPacketsOut, strings.TrimSpace(humanize.SIWithDigits(float64(stats.RetransmitPacketsOut), 2, "")))
|
||||
|
||||
// Date
|
||||
startedAndUpdated := fmt.Sprintf("%s\n%s", time.Unix(stats.StartedAt, 0).UTC().UTC().Format("2006-01-02 15:04:05"),
|
||||
|
||||
@@ -23,7 +23,7 @@ require (
|
||||
github.com/jxskiss/base62 v1.1.0
|
||||
github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1
|
||||
github.com/livekit/mediatransportutil v0.0.0-20250310153736-45596af895b6
|
||||
github.com/livekit/protocol v1.36.1
|
||||
github.com/livekit/protocol v1.36.2-0.20250326174620-fbbb1c3ae28a
|
||||
github.com/livekit/psrpc v0.6.1-0.20250205181828-a0beed2e4126
|
||||
github.com/mackerelio/go-osstat v0.2.5
|
||||
github.com/magefile/mage v1.15.0
|
||||
@@ -32,19 +32,19 @@ require (
|
||||
github.com/olekukonko/tablewriter v0.0.5
|
||||
github.com/ory/dockertest/v3 v3.11.0
|
||||
github.com/pion/datachannel v1.5.10
|
||||
github.com/pion/dtls/v3 v3.0.4
|
||||
github.com/pion/ice/v4 v4.0.7
|
||||
github.com/pion/dtls/v3 v3.0.5
|
||||
github.com/pion/ice/v4 v4.0.9
|
||||
github.com/pion/interceptor v0.1.37
|
||||
github.com/pion/rtcp v1.2.15
|
||||
github.com/pion/rtp v1.8.11
|
||||
github.com/pion/sctp v1.8.36
|
||||
github.com/pion/sdp/v3 v3.0.10
|
||||
github.com/pion/rtp v1.8.13
|
||||
github.com/pion/sctp v1.8.37
|
||||
github.com/pion/sdp/v3 v3.0.11
|
||||
github.com/pion/transport/v3 v3.0.7
|
||||
github.com/pion/turn/v4 v4.0.0
|
||||
github.com/pion/webrtc/v4 v4.0.11
|
||||
github.com/pion/webrtc/v4 v4.0.14
|
||||
github.com/pkg/errors v0.9.1
|
||||
github.com/prometheus/client_golang v1.20.5
|
||||
github.com/redis/go-redis/v9 v9.7.1
|
||||
github.com/redis/go-redis/v9 v9.7.3
|
||||
github.com/rs/cors v1.11.1
|
||||
github.com/stretchr/testify v1.10.0
|
||||
github.com/thoas/go-funk v0.9.3
|
||||
@@ -57,22 +57,22 @@ require (
|
||||
go.uber.org/zap v1.27.0
|
||||
golang.org/x/exp v0.0.0-20250305212735-054e65f0b394
|
||||
golang.org/x/sync v0.12.0
|
||||
google.golang.org/protobuf v1.36.5
|
||||
google.golang.org/protobuf v1.36.6
|
||||
gopkg.in/yaml.v3 v3.0.1
|
||||
)
|
||||
|
||||
require (
|
||||
buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.0-20241127180247-a33202765966.1 // indirect
|
||||
buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.6-20250307204501-0409229c3780.1 // indirect
|
||||
buf.build/go/protoyaml v0.3.1 // indirect
|
||||
cel.dev/expr v0.19.1 // indirect
|
||||
cel.dev/expr v0.22.1 // indirect
|
||||
dario.cat/mergo v1.0.0 // indirect
|
||||
github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 // indirect
|
||||
github.com/Microsoft/go-winio v0.6.2 // indirect
|
||||
github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 // indirect
|
||||
github.com/antlr4-go/antlr/v4 v4.13.0 // indirect
|
||||
github.com/antlr4-go/antlr/v4 v4.13.1 // indirect
|
||||
github.com/benbjohnson/clock v1.3.5 // indirect
|
||||
github.com/beorn7/perks v1.0.1 // indirect
|
||||
github.com/bufbuild/protovalidate-go v0.8.0 // indirect
|
||||
github.com/bufbuild/protovalidate-go v0.9.2 // indirect
|
||||
github.com/cenkalti/backoff/v4 v4.3.0 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/containerd/continuity v0.4.3 // indirect
|
||||
@@ -87,7 +87,7 @@ require (
|
||||
github.com/go-jose/go-jose/v3 v3.0.4 // indirect
|
||||
github.com/go-logr/logr v1.4.2 // indirect
|
||||
github.com/gogo/protobuf v1.3.2 // indirect
|
||||
github.com/google/cel-go v0.22.1 // indirect
|
||||
github.com/google/cel-go v0.24.1 // indirect
|
||||
github.com/google/go-cmp v0.6.0 // indirect
|
||||
github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
|
||||
github.com/google/subcommands v1.2.0 // indirect
|
||||
@@ -96,7 +96,7 @@ require (
|
||||
github.com/hashicorp/golang-lru v0.5.4 // indirect
|
||||
github.com/josharian/native v1.1.0 // indirect
|
||||
github.com/klauspost/compress v1.18.0 // indirect
|
||||
github.com/klauspost/cpuid/v2 v2.2.6 // indirect
|
||||
github.com/klauspost/cpuid/v2 v2.2.10 // indirect
|
||||
github.com/lithammer/shortuuid/v4 v4.2.0 // indirect
|
||||
github.com/mattn/go-runewidth v0.0.9 // indirect
|
||||
github.com/mdlayher/netlink v1.7.1 // indirect
|
||||
@@ -105,7 +105,7 @@ require (
|
||||
github.com/moby/docker-image-spec v1.3.1 // indirect
|
||||
github.com/moby/term v0.5.0 // indirect
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
|
||||
github.com/nats-io/nats.go v1.39.1 // indirect
|
||||
github.com/nats-io/nats.go v1.40.1 // indirect
|
||||
github.com/nats-io/nkeys v0.4.10 // indirect
|
||||
github.com/nats-io/nuid v1.0.1 // indirect
|
||||
github.com/opencontainers/go-digest v1.0.0 // indirect
|
||||
@@ -120,7 +120,7 @@ require (
|
||||
github.com/prometheus/client_model v0.6.1 // indirect
|
||||
github.com/prometheus/common v0.55.0 // indirect
|
||||
github.com/prometheus/procfs v0.15.1 // indirect
|
||||
github.com/puzpuzpuz/xsync/v3 v3.5.0 // indirect
|
||||
github.com/puzpuzpuz/xsync/v3 v3.5.1 // indirect
|
||||
github.com/russross/blackfriday/v2 v2.1.0 // indirect
|
||||
github.com/sirupsen/logrus v1.9.3 // indirect
|
||||
github.com/stoewer/go-strcase v1.3.0 // indirect
|
||||
@@ -137,8 +137,8 @@ require (
|
||||
golang.org/x/sys v0.31.0 // indirect
|
||||
golang.org/x/text v0.23.0 // indirect
|
||||
golang.org/x/tools v0.31.0 // indirect
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20250106144421-5f5ef82da422 // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20250303144028-a0af3efb3deb // indirect
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20250324211829-b45e905df463 // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20250324211829-b45e905df463 // indirect
|
||||
google.golang.org/grpc v1.71.0 // indirect
|
||||
gopkg.in/yaml.v2 v2.4.0 // indirect
|
||||
)
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.0-20241127180247-a33202765966.1 h1:ntAj16eF7AtUyzOOAFk5gvbAO52QmUKPKk7GmsIEORo=
|
||||
buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.0-20241127180247-a33202765966.1/go.mod h1:AxRT+qTj5PJCz2nyQzsR/qxAcveW5USRhJTt/edTO5w=
|
||||
buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.6-20250307204501-0409229c3780.1 h1:zgJPqo17m28+Lf5BW4xv3PvU20BnrmTcGYrog22lLIU=
|
||||
buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.6-20250307204501-0409229c3780.1/go.mod h1:avRlCjnFzl98VPaeCtJ24RrV/wwHFzB8sWXhj26+n/U=
|
||||
buf.build/go/protoyaml v0.3.1 h1:ucyzE7DRnjX+mQ6AH4JzN0Kg50ByHHu+yrSKbgQn2D4=
|
||||
buf.build/go/protoyaml v0.3.1/go.mod h1:0TzNpFQDXhwbkXb/ajLvxIijqbve+vMQvWY/b3/Dzxg=
|
||||
cel.dev/expr v0.19.1 h1:NciYrtDRIR0lNCnH1LFJegdjspNx9fI59O7TWcua/W4=
|
||||
cel.dev/expr v0.19.1/go.mod h1:MrpN08Q+lEBs+bGYdLxxHkZoUSsCp0nSKTs0nTymJgw=
|
||||
cel.dev/expr v0.22.1 h1:xoFEsNh972Yzey8N9TCPx2nDvMN7TMhQEzxLuj/iRrI=
|
||||
cel.dev/expr v0.22.1/go.mod h1:MrpN08Q+lEBs+bGYdLxxHkZoUSsCp0nSKTs0nTymJgw=
|
||||
dario.cat/mergo v1.0.0 h1:AGCNq9Evsj31mOgNPcLyXc+4PNABt905YmuqPYYpBWk=
|
||||
dario.cat/mergo v1.0.0/go.mod h1:uNxQE+84aUszobStD9th8a29P2fMDhsBdgRYvZOxGmk=
|
||||
filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
|
||||
@@ -14,8 +14,8 @@ github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERo
|
||||
github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU=
|
||||
github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 h1:TngWCqHvy9oXAN6lEVMRuU21PR1EtLVZJmdB18Gu3Rw=
|
||||
github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5/go.mod h1:lmUJ/7eu/Q8D7ML55dXQrVaamCz2vxCfdQBasLZfHKk=
|
||||
github.com/antlr4-go/antlr/v4 v4.13.0 h1:lxCg3LAv+EUK6t1i0y1V6/SLeUi0eKEKdhQAlS8TVTI=
|
||||
github.com/antlr4-go/antlr/v4 v4.13.0/go.mod h1:pfChB/xh/Unjila75QW7+VU4TSnWnnk9UTnmpPaOR2g=
|
||||
github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ=
|
||||
github.com/antlr4-go/antlr/v4 v4.13.1/go.mod h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw=
|
||||
github.com/benbjohnson/clock v1.3.5 h1:VvXlSJBzZpA/zum6Sj74hxwYI2DIxRWuNIoXAzHZz5o=
|
||||
github.com/benbjohnson/clock v1.3.5/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
|
||||
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
|
||||
@@ -26,8 +26,8 @@ github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
|
||||
github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
|
||||
github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
|
||||
github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
|
||||
github.com/bufbuild/protovalidate-go v0.8.0 h1:Xs3kCLCJ4tQiogJ0iOXm+ClKw/KviW3nLAryCGW2I3Y=
|
||||
github.com/bufbuild/protovalidate-go v0.8.0/go.mod h1:JPWZInGm2y2NBg3vKDKdDIkvDjyLv31J3hLH5GIFc/Q=
|
||||
github.com/bufbuild/protovalidate-go v0.9.2 h1:dUoPvFimovS74s3eeFNvHQOxFumRPsk390ifkzJCJ/4=
|
||||
github.com/bufbuild/protovalidate-go v0.9.2/go.mod h1:U9+WHAa6IOrLuqQEWPcxsyE4QEOTwm9fDpVbWXsR0zU=
|
||||
github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
|
||||
github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
|
||||
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
|
||||
@@ -92,8 +92,8 @@ github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
|
||||
github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
|
||||
github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
|
||||
github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
|
||||
github.com/google/cel-go v0.22.1 h1:AfVXx3chM2qwoSbM7Da8g8hX8OVSkBFwX+rz2+PcK40=
|
||||
github.com/google/cel-go v0.22.1/go.mod h1:BuznPXXfQDpXKWQ9sPW3TzlAJN5zzFe+i9tIs0yC4s8=
|
||||
github.com/google/cel-go v0.24.1 h1:jsBCtxG8mM5wiUJDSGUqU0K7Mtr3w7Eyv00rw4DiZxI=
|
||||
github.com/google/cel-go v0.24.1/go.mod h1:Hdf9TqOaTNSFQA1ybQaRqATVoK7m/zcf7IMhGXP5zI8=
|
||||
github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
|
||||
github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
|
||||
github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
||||
@@ -149,8 +149,8 @@ github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI
|
||||
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
|
||||
github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
|
||||
github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
|
||||
github.com/klauspost/cpuid/v2 v2.2.6 h1:ndNyv040zDGIDh8thGkXYjnFtiN02M1PVVF+JE/48xc=
|
||||
github.com/klauspost/cpuid/v2 v2.2.6/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws=
|
||||
github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE=
|
||||
github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
|
||||
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
|
||||
github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
|
||||
github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
|
||||
@@ -170,8 +170,8 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD
|
||||
github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
|
||||
github.com/livekit/mediatransportutil v0.0.0-20250310153736-45596af895b6 h1:6ZhtnY9I9knfm3ieIPpznQSEU2rDECO8yliW/ANLQ7U=
|
||||
github.com/livekit/mediatransportutil v0.0.0-20250310153736-45596af895b6/go.mod h1:36s+wwmU3O40IAhE+MjBWP3W71QRiEE9SfooSBvtBqY=
|
||||
github.com/livekit/protocol v1.36.1 h1:0cShRQxZQBQ/VquXrQRVD0qFwD84/7pffAN5dQM55Iw=
|
||||
github.com/livekit/protocol v1.36.1/go.mod h1:WrT/CYRxtMNOVUjnIPm5OjWtEkmreffTeE1PRZwlRg4=
|
||||
github.com/livekit/protocol v1.36.2-0.20250326174620-fbbb1c3ae28a h1:3oH/yRx6OTFc0JbUNkfhZXmW5zLZ61ZW02fqYFKjSrM=
|
||||
github.com/livekit/protocol v1.36.2-0.20250326174620-fbbb1c3ae28a/go.mod h1:WrT/CYRxtMNOVUjnIPm5OjWtEkmreffTeE1PRZwlRg4=
|
||||
github.com/livekit/psrpc v0.6.1-0.20250205181828-a0beed2e4126 h1:fzuYpAQbCid7ySPpQWWePfQOWUrs8x6dJ0T3Wl07n+Y=
|
||||
github.com/livekit/psrpc v0.6.1-0.20250205181828-a0beed2e4126/go.mod h1:X5WtEZ7OnEs72Fi5/J+i0on3964F1aynQpCalcgMqRo=
|
||||
github.com/mackerelio/go-osstat v0.2.5 h1:+MqTbZUhoIt4m8qzkVoXUJg1EuifwlAJSk4Yl2GXh+o=
|
||||
@@ -215,8 +215,8 @@ github.com/moby/term v0.5.0 h1:xt8Q1nalod/v7BqbG21f8mQPqH+xAaC9C3N3wfWbVP0=
|
||||
github.com/moby/term v0.5.0/go.mod h1:8FzsFHVUBGZdbDsJw/ot+X+d5HLUbvklYLJ9uGfcI3Y=
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
|
||||
github.com/nats-io/nats.go v1.39.1 h1:oTkfKBmz7W047vRxV762M67ZdXeOtUgvbBaNoQ+3PPk=
|
||||
github.com/nats-io/nats.go v1.39.1/go.mod h1:MgRb8oOdigA6cYpEPhXJuRVH6UE/V4jblJ2jQ27IXYM=
|
||||
github.com/nats-io/nats.go v1.40.1 h1:MLjDkdsbGUeCMKFyCFoLnNn/HDTqcgVa3EQm+pMNDPk=
|
||||
github.com/nats-io/nats.go v1.40.1/go.mod h1:wV73x0FSI/orHPSYoyMeJB+KajMDoWyXmFaRrrYaaTo=
|
||||
github.com/nats-io/nkeys v0.4.10 h1:glmRrpCmYLHByYcePvnTBEAwawwapjCPMjy2huw20wc=
|
||||
github.com/nats-io/nkeys v0.4.10/go.mod h1:OjRrnIKnWBFl+s4YK5ChQfvHP2fxqZexrKJoVVyWB3U=
|
||||
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
|
||||
@@ -235,10 +235,10 @@ github.com/ory/dockertest/v3 v3.11.0 h1:OiHcxKAvSDUwsEVh2BjxQQc/5EHz9n0va9awCtNG
|
||||
github.com/ory/dockertest/v3 v3.11.0/go.mod h1:VIPxS1gwT9NpPOrfD3rACs8Y9Z7yhzO4SB194iUDnUI=
|
||||
github.com/pion/datachannel v1.5.10 h1:ly0Q26K1i6ZkGf42W7D4hQYR90pZwzFOjTq5AuCKk4o=
|
||||
github.com/pion/datachannel v1.5.10/go.mod h1:p/jJfC9arb29W7WrxyKbepTU20CFgyx5oLo8Rs4Py/M=
|
||||
github.com/pion/dtls/v3 v3.0.4 h1:44CZekewMzfrn9pmGrj5BNnTMDCFwr+6sLH+cCuLM7U=
|
||||
github.com/pion/dtls/v3 v3.0.4/go.mod h1:R373CsjxWqNPf6MEkfdy3aSe9niZvL/JaKlGeFphtMg=
|
||||
github.com/pion/ice/v4 v4.0.7 h1:mnwuT3n3RE/9va41/9QJqN5+Bhc0H/x/ZyiVlWMw35M=
|
||||
github.com/pion/ice/v4 v4.0.7/go.mod h1:y3M18aPhIxLlcO/4dn9X8LzLLSma84cx6emMSu14FGw=
|
||||
github.com/pion/dtls/v3 v3.0.5 h1:OGWLu21/Wc5+H8R75F1BWvedH7H+nYUPFzJOew4k1iA=
|
||||
github.com/pion/dtls/v3 v3.0.5/go.mod h1:JVCnfmbgq45QoU07AaxFbdjF2iomKzYouVNy+W5kqmY=
|
||||
github.com/pion/ice/v4 v4.0.9 h1:VKgU4MwA2LUDVLq+WBkpEHTcAb8c5iCvFMECeuPOZNk=
|
||||
github.com/pion/ice/v4 v4.0.9/go.mod h1:y3M18aPhIxLlcO/4dn9X8LzLLSma84cx6emMSu14FGw=
|
||||
github.com/pion/interceptor v0.1.37 h1:aRA8Zpab/wE7/c0O3fh1PqY0AJI3fCSEM5lRWJVorwI=
|
||||
github.com/pion/interceptor v0.1.37/go.mod h1:JzxbJ4umVTlZAf+/utHzNesY8tmRkM2lVmkS82TTj8Y=
|
||||
github.com/pion/logging v0.2.3 h1:gHuf0zpoh1GW67Nr6Gj4cv5Z9ZscU7g/EaoC/Ke/igI=
|
||||
@@ -249,12 +249,12 @@ github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA=
|
||||
github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8=
|
||||
github.com/pion/rtcp v1.2.15 h1:LZQi2JbdipLOj4eBjK4wlVoQWfrZbh3Q6eHtWtJBZBo=
|
||||
github.com/pion/rtcp v1.2.15/go.mod h1:jlGuAjHMEXwMUHK78RgX0UmEJFV4zUKOFHR7OP+D3D0=
|
||||
github.com/pion/rtp v1.8.11 h1:17xjnY5WO5hgO6SD3/NTIUPvSFw/PbLsIJyz1r1yNIk=
|
||||
github.com/pion/rtp v1.8.11/go.mod h1:8uMBJj32Pa1wwx8Fuv/AsFhn8jsgw+3rUC2PfoBZ8p4=
|
||||
github.com/pion/sctp v1.8.36 h1:owNudmnz1xmhfYje5L/FCav3V9wpPRePHle3Zi+P+M0=
|
||||
github.com/pion/sctp v1.8.36/go.mod h1:cNiLdchXra8fHQwmIoqw0MbLLMs+f7uQ+dGMG2gWebE=
|
||||
github.com/pion/sdp/v3 v3.0.10 h1:6MChLE/1xYB+CjumMw+gZ9ufp2DPApuVSnDT8t5MIgA=
|
||||
github.com/pion/sdp/v3 v3.0.10/go.mod h1:88GMahN5xnScv1hIMTqLdu/cOcUkj6a9ytbncwMCq2E=
|
||||
github.com/pion/rtp v1.8.13 h1:8uSUPpjSL4OlwZI8Ygqu7+h2p9NPFB+yAZ461Xn5sNg=
|
||||
github.com/pion/rtp v1.8.13/go.mod h1:8uMBJj32Pa1wwx8Fuv/AsFhn8jsgw+3rUC2PfoBZ8p4=
|
||||
github.com/pion/sctp v1.8.37 h1:ZDmGPtRPX9mKCiVXtMbTWybFw3z/hVKAZgU81wcOrqs=
|
||||
github.com/pion/sctp v1.8.37/go.mod h1:cNiLdchXra8fHQwmIoqw0MbLLMs+f7uQ+dGMG2gWebE=
|
||||
github.com/pion/sdp/v3 v3.0.11 h1:VhgVSopdsBKwhCFoyyPmT1fKMeV9nLMrEKxNOdy3IVI=
|
||||
github.com/pion/sdp/v3 v3.0.11/go.mod h1:88GMahN5xnScv1hIMTqLdu/cOcUkj6a9ytbncwMCq2E=
|
||||
github.com/pion/srtp/v3 v3.0.4 h1:2Z6vDVxzrX3UHEgrUyIGM4rRouoC7v+NiF1IHtp9B5M=
|
||||
github.com/pion/srtp/v3 v3.0.4/go.mod h1:1Jx3FwDoxpRaTh1oRV8A/6G1BnFL+QI82eK4ms8EEJQ=
|
||||
github.com/pion/stun/v3 v3.0.0 h1:4h1gwhWLWuZWOJIJR9s2ferRO+W3zA/b6ijOI6mKzUw=
|
||||
@@ -263,8 +263,8 @@ github.com/pion/transport/v3 v3.0.7 h1:iRbMH05BzSNwhILHoBoAPxoB9xQgOaJk+591KC9P1
|
||||
github.com/pion/transport/v3 v3.0.7/go.mod h1:YleKiTZ4vqNxVwh77Z0zytYi7rXHl7j6uPLGhhz9rwo=
|
||||
github.com/pion/turn/v4 v4.0.0 h1:qxplo3Rxa9Yg1xXDxxH8xaqcyGUtbHYw4QSCvmFWvhM=
|
||||
github.com/pion/turn/v4 v4.0.0/go.mod h1:MuPDkm15nYSklKpN8vWJ9W2M0PlyQZqYt1McGuxG7mA=
|
||||
github.com/pion/webrtc/v4 v4.0.11 h1:0i7BNFH2n8LVp08q/dqM5iyZBXW4TITbD1+RwNqk/iY=
|
||||
github.com/pion/webrtc/v4 v4.0.11/go.mod h1:C+5JA7KiyLyoKyGh7hVFD/HCAon3IB/tfniocpZ9JoU=
|
||||
github.com/pion/webrtc/v4 v4.0.14 h1:nyds/sFRR+HvmWoBa6wrL46sSfpArE0qR883MBW96lg=
|
||||
github.com/pion/webrtc/v4 v4.0.14/go.mod h1:R3+qTnQTS03UzwDarYecgioNf7DYgTsldxnCXB821Kk=
|
||||
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
|
||||
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
@@ -277,10 +277,10 @@ github.com/prometheus/common v0.55.0 h1:KEi6DK7lXW/m7Ig5i47x0vRzuBsHuvJdi5ee6Y3G
|
||||
github.com/prometheus/common v0.55.0/go.mod h1:2SECS4xJG1kd8XF9IcM1gMX6510RAEL65zxzNImwdc8=
|
||||
github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc=
|
||||
github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
|
||||
github.com/puzpuzpuz/xsync/v3 v3.5.0 h1:i+cMcpEDY1BkNm7lPDkCtE4oElsYLn+EKF8kAu2vXT4=
|
||||
github.com/puzpuzpuz/xsync/v3 v3.5.0/go.mod h1:VjzYrABPabuM4KyBh1Ftq6u8nhwY5tBPKP9jpmh0nnA=
|
||||
github.com/redis/go-redis/v9 v9.7.1 h1:4LhKRCIduqXqtvCUlaq9c8bdHOkICjDMrr1+Zb3osAc=
|
||||
github.com/redis/go-redis/v9 v9.7.1/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw=
|
||||
github.com/puzpuzpuz/xsync/v3 v3.5.1 h1:GJYJZwO6IdxN/IKbneznS6yPkVC+c3zyY/j19c++5Fg=
|
||||
github.com/puzpuzpuz/xsync/v3 v3.5.1/go.mod h1:VjzYrABPabuM4KyBh1Ftq6u8nhwY5tBPKP9jpmh0nnA=
|
||||
github.com/redis/go-redis/v9 v9.7.3 h1:YpPyAayJV+XErNsatSElgRZZVCwXX9QzkKYNvO7x0wM=
|
||||
github.com/redis/go-redis/v9 v9.7.3/go.mod h1:bGUrSggJ9X9GUmZpZNEOQKaANxSGgOEBRltRTZHSvrA=
|
||||
github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
|
||||
github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
|
||||
github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=
|
||||
@@ -476,14 +476,14 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T
|
||||
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=
|
||||
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20250106144421-5f5ef82da422 h1:GVIKPyP/kLIyVOgOnTwFOrvQaQUzOzGMCxgFUOEmm24=
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20250106144421-5f5ef82da422/go.mod h1:b6h1vNKhxaSoEI+5jc3PJUCustfli/mRab7295pY7rw=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20250303144028-a0af3efb3deb h1:TLPQVbx1GJ8VKZxz52VAxl1EBgKXXbTiU9Fc5fZeLn4=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20250303144028-a0af3efb3deb/go.mod h1:LuRYeWDFV6WOn90g357N17oMCaxpgCnbi/44qJvDn2I=
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20250324211829-b45e905df463 h1:hE3bRWtU6uceqlh4fhrSnUyjKHMKB9KrTLLG+bc0ddM=
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20250324211829-b45e905df463/go.mod h1:U90ffi8eUL9MwPcrJylN5+Mk2v3vuPDptd5yyNUiRR8=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20250324211829-b45e905df463 h1:e0AIkUUhxyBKh6ssZNrAMeqhA7RKUj42346d1y02i2g=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20250324211829-b45e905df463/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A=
|
||||
google.golang.org/grpc v1.71.0 h1:kF77BGdPTQ4/JZWMlb9VpJ5pa25aqvVqogsxNHHdeBg=
|
||||
google.golang.org/grpc v1.71.0/go.mod h1:H0GRtasmQOh9LkFoCPDu3ZrwUtD1YGE+b2vYBYd/8Ec=
|
||||
google.golang.org/protobuf v1.36.5 h1:tPhr+woSbjfYvY6/GPufUoYizxw1cF/yFoxJ2fmpwlM=
|
||||
google.golang.org/protobuf v1.36.5/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
|
||||
google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY=
|
||||
google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
|
||||
|
||||
+19
-4
@@ -79,6 +79,8 @@ type Config struct {
|
||||
Development bool `yaml:"development,omitempty"`
|
||||
|
||||
Metric metric.MetricConfig `yaml:"metric,omitempty"`
|
||||
|
||||
NodeStats NodeStatsConfig `yaml:"node_stats,omitempty"`
|
||||
}
|
||||
|
||||
type RTCConfig struct {
|
||||
@@ -302,6 +304,18 @@ func DefaultAPIConfig() APIConfig {
|
||||
}
|
||||
}
|
||||
|
||||
type NodeStatsConfig struct {
|
||||
StatsUpdateInterval time.Duration `yaml:"stats_update_interval,omitempty"`
|
||||
StatsRateMeasurementIntervals []time.Duration `yaml:"stats_rate_measurement_intervals,omitempty"`
|
||||
StatsMaxDelay time.Duration `yaml:"stats_max_delay,omitempty"`
|
||||
}
|
||||
|
||||
var DefaultNodeStatsConfig = NodeStatsConfig{
|
||||
StatsUpdateInterval: 2 * time.Second,
|
||||
StatsRateMeasurementIntervals: []time.Duration{10 * time.Second},
|
||||
StatsMaxDelay: 30 * time.Second,
|
||||
}
|
||||
|
||||
var DefaultConfig = Config{
|
||||
Port: 7880,
|
||||
RTC: RTCConfig{
|
||||
@@ -377,10 +391,11 @@ var DefaultConfig = Config{
|
||||
StreamBufferSize: 1000,
|
||||
ConnectAttempts: 3,
|
||||
},
|
||||
PSRPC: rpc.DefaultPSRPCConfig,
|
||||
Keys: map[string]string{},
|
||||
Metric: metric.DefaultMetricConfig,
|
||||
WebHook: webhook.DefaultWebHookConfig,
|
||||
PSRPC: rpc.DefaultPSRPCConfig,
|
||||
Keys: map[string]string{},
|
||||
Metric: metric.DefaultMetricConfig,
|
||||
WebHook: webhook.DefaultWebHookConfig,
|
||||
NodeStats: DefaultNodeStatsConfig,
|
||||
}
|
||||
|
||||
func NewConfig(confString string, strictMode bool, c *cli.Context, baseFlags []cli.Flag) (*Config, error) {
|
||||
|
||||
@@ -23,6 +23,7 @@ import (
|
||||
"go.uber.org/zap/zapcore"
|
||||
"google.golang.org/protobuf/proto"
|
||||
|
||||
"github.com/livekit/livekit-server/pkg/config"
|
||||
"github.com/livekit/livekit-server/pkg/utils"
|
||||
"github.com/livekit/protocol/auth"
|
||||
"github.com/livekit/protocol/livekit"
|
||||
@@ -162,8 +163,9 @@ func CreateRouter(
|
||||
signalClient SignalClient,
|
||||
roomManagerClient RoomManagerClient,
|
||||
kps rpc.KeepalivePubSub,
|
||||
nodeStatsConfig config.NodeStatsConfig,
|
||||
) Router {
|
||||
lr := NewLocalRouter(node, signalClient, roomManagerClient)
|
||||
lr := NewLocalRouter(node, signalClient, roomManagerClient, nodeStatsConfig)
|
||||
|
||||
if rc != nil {
|
||||
return NewRedisRouter(lr, rc, kps)
|
||||
|
||||
@@ -21,6 +21,7 @@ import (
|
||||
|
||||
"go.uber.org/atomic"
|
||||
|
||||
"github.com/livekit/livekit-server/pkg/config"
|
||||
"github.com/livekit/protocol/livekit"
|
||||
"github.com/livekit/protocol/logger"
|
||||
)
|
||||
@@ -32,6 +33,7 @@ type LocalRouter struct {
|
||||
currentNode LocalNode
|
||||
signalClient SignalClient
|
||||
roomManagerClient RoomManagerClient
|
||||
nodeStatsConfig config.NodeStatsConfig
|
||||
|
||||
lock sync.RWMutex
|
||||
// channels for each participant
|
||||
@@ -44,11 +46,13 @@ func NewLocalRouter(
|
||||
currentNode LocalNode,
|
||||
signalClient SignalClient,
|
||||
roomManagerClient RoomManagerClient,
|
||||
nodeStatsConfig config.NodeStatsConfig,
|
||||
) *LocalRouter {
|
||||
return &LocalRouter{
|
||||
currentNode: currentNode,
|
||||
signalClient: signalClient,
|
||||
roomManagerClient: roomManagerClient,
|
||||
nodeStatsConfig: nodeStatsConfig,
|
||||
requestChannels: make(map[string]*MessageChannel),
|
||||
responseChannels: make(map[string]*MessageChannel),
|
||||
}
|
||||
@@ -146,8 +150,7 @@ func (r *LocalRouter) statsWorker() {
|
||||
if !r.isStarted.Load() {
|
||||
return
|
||||
}
|
||||
// update every 10 seconds
|
||||
<-time.After(statsUpdateInterval)
|
||||
<-time.After(r.nodeStatsConfig.StatsUpdateInterval)
|
||||
r.currentNode.UpdateNodeStats()
|
||||
}
|
||||
}
|
||||
|
||||
+12
-15
@@ -20,12 +20,10 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/livekit/protocol/livekit"
|
||||
"github.com/livekit/protocol/logger"
|
||||
"github.com/livekit/protocol/utils"
|
||||
"github.com/livekit/protocol/utils/guid"
|
||||
|
||||
"github.com/livekit/livekit-server/pkg/config"
|
||||
"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
|
||||
)
|
||||
|
||||
type LocalNode interface {
|
||||
@@ -45,8 +43,7 @@ type LocalNodeImpl struct {
|
||||
lock sync.RWMutex
|
||||
node *livekit.Node
|
||||
|
||||
// previous stats for computing averages
|
||||
prevStats *livekit.NodeStats
|
||||
nodeStats *NodeStats
|
||||
}
|
||||
|
||||
func NewLocalNode(conf *config.Config) (*LocalNodeImpl, error) {
|
||||
@@ -54,21 +51,27 @@ func NewLocalNode(conf *config.Config) (*LocalNodeImpl, error) {
|
||||
if conf != nil && conf.RTC.NodeIP == "" {
|
||||
return nil, ErrIPNotSet
|
||||
}
|
||||
nowUnix := time.Now().Unix()
|
||||
l := &LocalNodeImpl{
|
||||
node: &livekit.Node{
|
||||
Id: nodeID,
|
||||
NumCpus: uint32(runtime.NumCPU()),
|
||||
State: livekit.NodeState_SERVING,
|
||||
Stats: &livekit.NodeStats{
|
||||
StartedAt: time.Now().Unix(),
|
||||
UpdatedAt: time.Now().Unix(),
|
||||
StartedAt: nowUnix,
|
||||
UpdatedAt: nowUnix,
|
||||
},
|
||||
},
|
||||
}
|
||||
var nsc *config.NodeStatsConfig
|
||||
if conf != nil {
|
||||
l.node.Ip = conf.RTC.NodeIP
|
||||
l.node.Region = conf.Region
|
||||
|
||||
nsc = &conf.NodeStats
|
||||
}
|
||||
l.nodeStats = NewNodeStats(nsc, nowUnix)
|
||||
|
||||
return l, nil
|
||||
}
|
||||
|
||||
@@ -138,18 +141,12 @@ func (l *LocalNodeImpl) UpdateNodeStats() bool {
|
||||
l.lock.Lock()
|
||||
defer l.lock.Unlock()
|
||||
|
||||
if l.prevStats == nil {
|
||||
l.prevStats = l.node.Stats
|
||||
}
|
||||
updated, computedAvg, err := prometheus.GetUpdatedNodeStats(l.node.Stats, l.prevStats)
|
||||
stats, err := l.nodeStats.UpdateAndGetNodeStats()
|
||||
if err != nil {
|
||||
logger.Errorw("could not update node stats", err)
|
||||
return false
|
||||
}
|
||||
l.node.Stats = updated
|
||||
if computedAvg {
|
||||
l.prevStats = updated
|
||||
}
|
||||
|
||||
l.node.Stats = stats
|
||||
return true
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,82 @@
|
||||
// Copyright 2023 LiveKit, Inc.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package routing
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/livekit/protocol/livekit"
|
||||
"github.com/livekit/protocol/logger"
|
||||
|
||||
"github.com/livekit/livekit-server/pkg/config"
|
||||
"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
|
||||
)
|
||||
|
||||
type NodeStats struct {
|
||||
config config.NodeStatsConfig
|
||||
startedAt int64
|
||||
|
||||
lock sync.Mutex
|
||||
statsHistory []*livekit.NodeStats
|
||||
statsHistoryWritePtr int
|
||||
}
|
||||
|
||||
func NewNodeStats(conf *config.NodeStatsConfig, startedAt int64) *NodeStats {
|
||||
n := &NodeStats{
|
||||
startedAt: startedAt,
|
||||
}
|
||||
n.UpdateConfig(conf)
|
||||
return n
|
||||
}
|
||||
|
||||
func (n *NodeStats) UpdateConfig(conf *config.NodeStatsConfig) {
|
||||
n.lock.Lock()
|
||||
defer n.lock.Unlock()
|
||||
|
||||
if conf == nil {
|
||||
conf = &config.DefaultNodeStatsConfig
|
||||
}
|
||||
n.config = *conf
|
||||
|
||||
// set up stats history to be able to measure different rate windows
|
||||
var maxInterval time.Duration
|
||||
for _, rateInterval := range conf.StatsRateMeasurementIntervals {
|
||||
if rateInterval > maxInterval {
|
||||
maxInterval = rateInterval
|
||||
}
|
||||
}
|
||||
n.statsHistory = make([]*livekit.NodeStats, (maxInterval+conf.StatsUpdateInterval-1)/conf.StatsUpdateInterval)
|
||||
n.statsHistoryWritePtr = 0
|
||||
}
|
||||
|
||||
func (n *NodeStats) UpdateAndGetNodeStats() (*livekit.NodeStats, error) {
|
||||
n.lock.Lock()
|
||||
defer n.lock.Unlock()
|
||||
|
||||
stats, err := prometheus.GetNodeStats(
|
||||
n.startedAt,
|
||||
append(n.statsHistory[n.statsHistoryWritePtr:], n.statsHistory[0:n.statsHistoryWritePtr]...),
|
||||
n.config.StatsRateMeasurementIntervals,
|
||||
)
|
||||
if err != nil {
|
||||
logger.Errorw("could not update node stats", err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
n.statsHistory[n.statsHistoryWritePtr] = stats
|
||||
n.statsHistoryWritePtr = (n.statsHistoryWritePtr + 1) % len(n.statsHistory)
|
||||
return stats, nil
|
||||
}
|
||||
@@ -33,11 +33,6 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
// expire participant mappings after a day
|
||||
participantMappingTTL = 24 * time.Hour
|
||||
statsUpdateInterval = 2 * time.Second
|
||||
statsMaxDelaySeconds = float64(30)
|
||||
|
||||
// hash of node_id => Node proto
|
||||
NodesKey = "nodes"
|
||||
|
||||
@@ -209,11 +204,11 @@ func (r *RedisRouter) statsWorker() {
|
||||
for r.ctx.Err() == nil {
|
||||
// update periodically
|
||||
select {
|
||||
case <-time.After(statsUpdateInterval):
|
||||
case <-time.After(r.nodeStatsConfig.StatsUpdateInterval):
|
||||
r.kps.PublishPing(r.ctx, r.currentNode.NodeID(), &rpc.KeepalivePing{Timestamp: time.Now().Unix()})
|
||||
|
||||
delaySeconds := r.currentNode.SecondsSinceNodeStatsUpdate()
|
||||
if delaySeconds > statsMaxDelaySeconds {
|
||||
if delaySeconds > r.nodeStatsConfig.StatsMaxDelay.Seconds() {
|
||||
if !goroutineDumped {
|
||||
goroutineDumped = true
|
||||
buf := bytes.NewBuffer(nil)
|
||||
@@ -240,7 +235,7 @@ func (r *RedisRouter) keepaliveWorker(startedChan chan error) {
|
||||
close(startedChan)
|
||||
|
||||
for ping := range pings.Channel() {
|
||||
if time.Since(time.Unix(ping.Timestamp, 0)) > statsUpdateInterval {
|
||||
if time.Since(time.Unix(ping.Timestamp, 0)) > r.nodeStatsConfig.StatsUpdateInterval {
|
||||
logger.Infow("keep alive too old, skipping", "timestamp", ping.Timestamp)
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -33,12 +33,16 @@ var (
|
||||
NumCpus: 1,
|
||||
CpuLoad: 0.1,
|
||||
LoadAvgLast1Min: 0.0,
|
||||
NumRooms: 1,
|
||||
NumClients: 2,
|
||||
NumTracksIn: 4,
|
||||
NumTracksOut: 8,
|
||||
BytesInPerSec: 1000,
|
||||
BytesOutPerSec: 2000,
|
||||
NumRooms: 1,
|
||||
NumClients: 2,
|
||||
NumTracksIn: 4,
|
||||
NumTracksOut: 8,
|
||||
Rates: []*livekit.NodeStatsRate{
|
||||
{
|
||||
BytesIn: 1000,
|
||||
BytesOut: 2000,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
@@ -49,12 +53,16 @@ var (
|
||||
NumCpus: 1,
|
||||
CpuLoad: 0.5,
|
||||
LoadAvgLast1Min: 0.5,
|
||||
NumRooms: 5,
|
||||
NumClients: 10,
|
||||
NumTracksIn: 20,
|
||||
NumTracksOut: 200,
|
||||
BytesInPerSec: 5000,
|
||||
BytesOutPerSec: 10000,
|
||||
NumRooms: 5,
|
||||
NumClients: 10,
|
||||
NumTracksIn: 20,
|
||||
NumTracksOut: 200,
|
||||
Rates: []*livekit.NodeStatsRate{
|
||||
{
|
||||
BytesIn: 5000,
|
||||
BytesOut: 10000,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
@@ -65,12 +73,16 @@ var (
|
||||
NumCpus: 1,
|
||||
CpuLoad: 0.99,
|
||||
LoadAvgLast1Min: 2.0,
|
||||
NumRooms: 10,
|
||||
NumClients: 20,
|
||||
NumTracksIn: 40,
|
||||
NumTracksOut: 800,
|
||||
BytesInPerSec: 10000,
|
||||
BytesOutPerSec: 40000,
|
||||
NumRooms: 10,
|
||||
NumClients: 20,
|
||||
NumTracksIn: 40,
|
||||
NumTracksOut: 800,
|
||||
Rates: []*livekit.NodeStatsRate{
|
||||
{
|
||||
BytesIn: 10000,
|
||||
BytesOut: 40000,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
@@ -62,7 +62,12 @@ func LimitsReached(limitConfig config.LimitConfig, nodeStats *livekit.NodeStats)
|
||||
if limitConfig.NumTracks > 0 && limitConfig.NumTracks <= nodeStats.NumTracksIn+nodeStats.NumTracksOut {
|
||||
return true
|
||||
}
|
||||
if limitConfig.BytesPerSec > 0 && limitConfig.BytesPerSec <= nodeStats.BytesInPerSec+nodeStats.BytesOutPerSec {
|
||||
|
||||
rate := &livekit.NodeStatsRate{}
|
||||
if len(nodeStats.Rates) > 0 {
|
||||
rate = nodeStats.Rates[0]
|
||||
}
|
||||
if limitConfig.BytesPerSec > 0 && limitConfig.BytesPerSec <= rate.BytesIn+rate.BytesOut {
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -106,7 +111,16 @@ func SelectSortedNode(nodes []*livekit.Node, sortBy string) (*livekit.Node, erro
|
||||
return nodes[0], nil
|
||||
case "bytespersec":
|
||||
sort.Slice(nodes, func(i, j int) bool {
|
||||
return nodes[i].Stats.BytesInPerSec+nodes[i].Stats.BytesOutPerSec < nodes[j].Stats.BytesInPerSec+nodes[j].Stats.BytesOutPerSec
|
||||
ratei := &livekit.NodeStatsRate{}
|
||||
if len(nodes[i].Stats.Rates) > 0 {
|
||||
ratei = nodes[i].Stats.Rates[0]
|
||||
}
|
||||
|
||||
ratej := &livekit.NodeStatsRate{}
|
||||
if len(nodes[j].Stats.Rates) > 0 {
|
||||
ratej = nodes[j].Stats.Rates[0]
|
||||
}
|
||||
return ratei.BytesIn+ratei.BytesOut < ratej.BytesIn+ratej.BytesOut
|
||||
})
|
||||
return nodes[0], nil
|
||||
default:
|
||||
|
||||
@@ -486,7 +486,7 @@ func NewPCTransport(params TransportParams) (*PCTransport, error) {
|
||||
|
||||
if params.IsSendSide {
|
||||
if params.CongestionControlConfig.UseSendSideBWE {
|
||||
params.Logger.Infow("using send side BWE")
|
||||
params.Logger.Infow("using send side BWE", "pacerBehavior", params.CongestionControlConfig.SendSideBWEPacer)
|
||||
t.bwe = sendsidebwe.NewSendSideBWE(sendsidebwe.SendSideBWEParams{
|
||||
Config: params.CongestionControlConfig.SendSideBWE,
|
||||
Logger: params.Logger,
|
||||
|
||||
@@ -309,7 +309,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
// websocket established
|
||||
sigConn := NewWSSignalConnection(conn)
|
||||
pLogger.Debugw("sending join", "join", logger.Proto(initialResponse))
|
||||
pLogger.Debugw("sending initial response", "response", logger.Proto(initialResponse))
|
||||
count, err := sigConn.WriteResponse(initialResponse)
|
||||
if err != nil {
|
||||
pLogger.Warnw("could not write initial response", err)
|
||||
|
||||
@@ -53,6 +53,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
|
||||
createWebhookNotifier,
|
||||
createClientConfiguration,
|
||||
createForwardStats,
|
||||
getNodeStatsConfig,
|
||||
routing.CreateRouter,
|
||||
getLimitConf,
|
||||
config.DefaultAPIConfig,
|
||||
@@ -116,6 +117,7 @@ func InitializeRouter(conf *config.Config, currentNode routing.LocalNode) (routi
|
||||
getRoomConfig,
|
||||
routing.NewRoomManagerClient,
|
||||
rpc.NewKeepalivePubSub,
|
||||
getNodeStatsConfig,
|
||||
routing.CreateRouter,
|
||||
)
|
||||
|
||||
@@ -269,3 +271,7 @@ func createForwardStats(conf *config.Config) *sfu.ForwardStats {
|
||||
func newInProcessTurnServer(conf *config.Config, authHandler turn.AuthHandler) (*turn.Server, error) {
|
||||
return NewTurnServer(conf, authHandler, false)
|
||||
}
|
||||
|
||||
func getNodeStatsConfig(config *config.Config) config.NodeStatsConfig {
|
||||
return config.NodeStats
|
||||
}
|
||||
|
||||
+13
-7
@@ -60,7 +60,8 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
router := routing.CreateRouter(universalClient, currentNode, signalClient, roomManagerClient, keepalivePubSub)
|
||||
nodeStatsConfig := getNodeStatsConfig(conf)
|
||||
router := routing.CreateRouter(universalClient, currentNode, signalClient, roomManagerClient, keepalivePubSub, nodeStatsConfig)
|
||||
objectStore := createStore(universalClient)
|
||||
roomAllocator, err := NewRoomAllocator(conf, router, objectStore)
|
||||
if err != nil {
|
||||
@@ -89,23 +90,23 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
|
||||
}
|
||||
rtcEgressLauncher := NewEgressLauncher(egressClient, ioInfoService)
|
||||
topicFormatter := rpc.NewTopicFormatter()
|
||||
v, err := rpc.NewTypedRoomClient(clientParams)
|
||||
roomClient, err := rpc.NewTypedRoomClient(clientParams)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
v2, err := rpc.NewTypedParticipantClient(clientParams)
|
||||
participantClient, err := rpc.NewTypedParticipantClient(clientParams)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, v, v2)
|
||||
roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, roomClient, participantClient)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
v3, err := rpc.NewTypedAgentDispatchInternalClient(clientParams)
|
||||
agentDispatchInternalClient, err := rpc.NewTypedAgentDispatchInternalClient(clientParams)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
agentDispatchService := NewAgentDispatchService(v3, topicFormatter, roomAllocator, router)
|
||||
agentDispatchService := NewAgentDispatchService(agentDispatchInternalClient, topicFormatter, roomAllocator, router)
|
||||
egressService := NewEgressService(egressClient, rtcEgressLauncher, objectStore, ioInfoService, roomService)
|
||||
ingressConfig := getIngressConfig(conf)
|
||||
ingressClient, err := rpc.NewIngressClient(clientParams)
|
||||
@@ -176,7 +177,8 @@ func InitializeRouter(conf *config.Config, currentNode routing.LocalNode) (routi
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
router := routing.CreateRouter(universalClient, currentNode, signalClient, roomManagerClient, keepalivePubSub)
|
||||
nodeStatsConfig := getNodeStatsConfig(conf)
|
||||
router := routing.CreateRouter(universalClient, currentNode, signalClient, roomManagerClient, keepalivePubSub, nodeStatsConfig)
|
||||
return router, nil
|
||||
}
|
||||
|
||||
@@ -329,3 +331,7 @@ func createForwardStats(conf *config.Config) *sfu.ForwardStats {
|
||||
func newInProcessTurnServer(conf *config.Config, authHandler turn.AuthHandler) (*turn.Server, error) {
|
||||
return NewTurnServer(conf, authHandler, false)
|
||||
}
|
||||
|
||||
func getNodeStatsConfig(config2 *config.Config) config.NodeStatsConfig {
|
||||
return config2.NodeStats
|
||||
}
|
||||
|
||||
+104
-116
@@ -28,8 +28,6 @@ import (
|
||||
|
||||
const (
|
||||
livekitNamespace string = "livekit"
|
||||
|
||||
statsUpdateInterval = time.Second * 10
|
||||
)
|
||||
|
||||
var (
|
||||
@@ -104,22 +102,11 @@ func Init(nodeID string, nodeType livekit.NodeType) error {
|
||||
[]string{"type"},
|
||||
)
|
||||
|
||||
promSysDroppedPacketPctGauge = prometheus.NewGauge(
|
||||
prometheus.GaugeOpts{
|
||||
Namespace: livekitNamespace,
|
||||
Subsystem: "node",
|
||||
Name: "dropped_packets",
|
||||
ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()},
|
||||
Help: "System level dropped outgoing packet percentage.",
|
||||
},
|
||||
)
|
||||
|
||||
prometheus.MustRegister(MessageCounter)
|
||||
prometheus.MustRegister(MessageBytes)
|
||||
prometheus.MustRegister(ServiceOperationCounter)
|
||||
prometheus.MustRegister(TwirpRequestStatusCounter)
|
||||
prometheus.MustRegister(promSysPacketGauge)
|
||||
prometheus.MustRegister(promSysDroppedPacketPctGauge)
|
||||
|
||||
sysPacketsStart, sysDroppedPacketsStart, _ = getTCStats()
|
||||
|
||||
@@ -138,10 +125,10 @@ func Init(nodeID string, nodeType livekit.NodeType) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func GetUpdatedNodeStats(prev *livekit.NodeStats, prevAverage *livekit.NodeStats) (*livekit.NodeStats, bool, error) {
|
||||
func GetNodeStats(nodeStartedAt int64, prevStats []*livekit.NodeStats, rateIntervals []time.Duration) (*livekit.NodeStats, error) {
|
||||
loadAvg, err := getLoadAvg()
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var cpuLoad float64
|
||||
@@ -160,118 +147,119 @@ func GetUpdatedNodeStats(prev *livekit.NodeStats, prevAverage *livekit.NodeStats
|
||||
memUsed = memInfo.Used
|
||||
}
|
||||
|
||||
// do not error out, and use the information if it is available
|
||||
sysPackets, sysDroppedPackets, _ := getTCStats()
|
||||
promSysPacketGauge.WithLabelValues("out").Set(float64(sysPackets - sysPacketsStart))
|
||||
promSysPacketGauge.WithLabelValues("dropped").Set(float64(sysDroppedPackets - sysDroppedPacketsStart))
|
||||
|
||||
bytesInNow := bytesIn.Load()
|
||||
bytesOutNow := bytesOut.Load()
|
||||
packetsInNow := packetsIn.Load()
|
||||
packetsOutNow := packetsOut.Load()
|
||||
nackTotalNow := nackTotal.Load()
|
||||
retransmitBytesNow := retransmitBytes.Load()
|
||||
retransmitPacketsNow := retransmitPackets.Load()
|
||||
participantSignalConnectedNow := participantSignalConnected.Load()
|
||||
participantRTCInitNow := participantRTCInit.Load()
|
||||
participantRTConnectedCNow := participantRTCConnected.Load()
|
||||
trackPublishAttemptsNow := trackPublishAttempts.Load()
|
||||
trackPublishSuccessNow := trackPublishSuccess.Load()
|
||||
trackSubscribeAttemptsNow := trackSubscribeAttempts.Load()
|
||||
trackSubscribeSuccessNow := trackSubscribeSuccess.Load()
|
||||
forwardLatencyNow := forwardLatency.Load()
|
||||
forwardJitterNow := forwardJitter.Load()
|
||||
|
||||
updatedAt := time.Now().Unix()
|
||||
elapsed := updatedAt - prevAverage.UpdatedAt
|
||||
// include sufficient buffer to be sure a stats update had taken place
|
||||
computeAverage := elapsed > int64(statsUpdateInterval.Seconds()+2)
|
||||
if bytesInNow != prevAverage.BytesIn ||
|
||||
bytesOutNow != prevAverage.BytesOut ||
|
||||
packetsInNow != prevAverage.PacketsIn ||
|
||||
packetsOutNow != prevAverage.PacketsOut ||
|
||||
retransmitBytesNow != prevAverage.RetransmitBytesOut ||
|
||||
retransmitPacketsNow != prevAverage.RetransmitPacketsOut {
|
||||
computeAverage = true
|
||||
}
|
||||
|
||||
stats := &livekit.NodeStats{
|
||||
StartedAt: prev.StartedAt,
|
||||
UpdatedAt: updatedAt,
|
||||
NumRooms: roomCurrent.Load(),
|
||||
NumClients: participantCurrent.Load(),
|
||||
NumTracksIn: trackPublishedCurrent.Load(),
|
||||
NumTracksOut: trackSubscribedCurrent.Load(),
|
||||
NumTrackPublishAttempts: trackPublishAttemptsNow,
|
||||
NumTrackPublishSuccess: trackPublishSuccessNow,
|
||||
NumTrackSubscribeAttempts: trackSubscribeAttemptsNow,
|
||||
NumTrackSubscribeSuccess: trackSubscribeSuccessNow,
|
||||
BytesIn: bytesInNow,
|
||||
BytesOut: bytesOutNow,
|
||||
PacketsIn: packetsInNow,
|
||||
PacketsOut: packetsOutNow,
|
||||
RetransmitBytesOut: retransmitBytesNow,
|
||||
RetransmitPacketsOut: retransmitPacketsNow,
|
||||
NackTotal: nackTotalNow,
|
||||
ParticipantSignalConnected: participantSignalConnectedNow,
|
||||
ParticipantRtcInit: participantRTCInitNow,
|
||||
ParticipantRtcConnected: participantRTConnectedCNow,
|
||||
BytesInPerSec: prevAverage.BytesInPerSec,
|
||||
BytesOutPerSec: prevAverage.BytesOutPerSec,
|
||||
PacketsInPerSec: prevAverage.PacketsInPerSec,
|
||||
PacketsOutPerSec: prevAverage.PacketsOutPerSec,
|
||||
RetransmitBytesOutPerSec: prevAverage.RetransmitBytesOutPerSec,
|
||||
RetransmitPacketsOutPerSec: prevAverage.RetransmitPacketsOutPerSec,
|
||||
NackPerSec: prevAverage.NackPerSec,
|
||||
ForwardLatency: forwardLatencyNow,
|
||||
ForwardJitter: forwardJitterNow,
|
||||
ParticipantSignalConnectedPerSec: prevAverage.ParticipantSignalConnectedPerSec,
|
||||
ParticipantRtcInitPerSec: prevAverage.ParticipantRtcInitPerSec,
|
||||
ParticipantRtcConnectedPerSec: prevAverage.ParticipantRtcConnectedPerSec,
|
||||
NumCpus: uint32(cpuStats.NumCPU()), // this will round down to the nearest integer
|
||||
CpuLoad: float32(cpuLoad),
|
||||
MemoryTotal: memTotal,
|
||||
MemoryUsed: memUsed,
|
||||
LoadAvgLast1Min: float32(loadAvg.Loadavg1),
|
||||
LoadAvgLast5Min: float32(loadAvg.Loadavg5),
|
||||
LoadAvgLast15Min: float32(loadAvg.Loadavg15),
|
||||
SysPacketsOut: sysPackets,
|
||||
SysPacketsDropped: sysDroppedPackets,
|
||||
TrackPublishAttemptsPerSec: prevAverage.TrackPublishAttemptsPerSec,
|
||||
TrackPublishSuccessPerSec: prevAverage.TrackPublishSuccessPerSec,
|
||||
TrackSubscribeAttemptsPerSec: prevAverage.TrackSubscribeAttemptsPerSec,
|
||||
TrackSubscribeSuccessPerSec: prevAverage.TrackSubscribeSuccessPerSec,
|
||||
StartedAt: nodeStartedAt,
|
||||
UpdatedAt: time.Now().Unix(),
|
||||
NumRooms: roomCurrent.Load(),
|
||||
NumClients: participantCurrent.Load(),
|
||||
NumTracksIn: trackPublishedCurrent.Load(),
|
||||
NumTracksOut: trackSubscribedCurrent.Load(),
|
||||
NumTrackPublishAttempts: trackPublishAttempts.Load(),
|
||||
NumTrackPublishSuccess: trackPublishSuccess.Load(),
|
||||
NumTrackSubscribeAttempts: trackSubscribeAttempts.Load(),
|
||||
NumTrackSubscribeSuccess: trackSubscribeSuccess.Load(),
|
||||
BytesIn: bytesIn.Load(),
|
||||
BytesOut: bytesOut.Load(),
|
||||
PacketsIn: packetsIn.Load(),
|
||||
PacketsOut: packetsOut.Load(),
|
||||
RetransmitBytesOut: retransmitBytes.Load(),
|
||||
RetransmitPacketsOut: retransmitPackets.Load(),
|
||||
NackTotal: nackTotal.Load(),
|
||||
ParticipantSignalConnected: participantSignalConnected.Load(),
|
||||
ParticipantRtcInit: participantRTCInit.Load(),
|
||||
ParticipantRtcConnected: participantRTCConnected.Load(),
|
||||
ForwardLatency: forwardLatency.Load(),
|
||||
ForwardJitter: forwardJitter.Load(),
|
||||
NumCpus: uint32(cpuStats.NumCPU()), // this will round down to the nearest integer
|
||||
CpuLoad: float32(cpuLoad),
|
||||
MemoryTotal: memTotal,
|
||||
MemoryUsed: memUsed,
|
||||
LoadAvgLast1Min: float32(loadAvg.Loadavg1),
|
||||
LoadAvgLast5Min: float32(loadAvg.Loadavg5),
|
||||
LoadAvgLast15Min: float32(loadAvg.Loadavg15),
|
||||
SysPacketsOut: sysPackets,
|
||||
SysPacketsDropped: sysDroppedPackets,
|
||||
}
|
||||
|
||||
// update stats
|
||||
if computeAverage {
|
||||
stats.BytesInPerSec = perSec(prevAverage.BytesIn, bytesInNow, elapsed)
|
||||
stats.BytesOutPerSec = perSec(prevAverage.BytesOut, bytesOutNow, elapsed)
|
||||
stats.PacketsInPerSec = perSec(prevAverage.PacketsIn, packetsInNow, elapsed)
|
||||
stats.PacketsOutPerSec = perSec(prevAverage.PacketsOut, packetsOutNow, elapsed)
|
||||
stats.RetransmitBytesOutPerSec = perSec(prevAverage.RetransmitBytesOut, retransmitBytesNow, elapsed)
|
||||
stats.RetransmitPacketsOutPerSec = perSec(prevAverage.RetransmitPacketsOut, retransmitPacketsNow, elapsed)
|
||||
stats.NackPerSec = perSec(prevAverage.NackTotal, nackTotalNow, elapsed)
|
||||
stats.ParticipantSignalConnectedPerSec = perSec(prevAverage.ParticipantSignalConnected, participantSignalConnectedNow, elapsed)
|
||||
stats.ParticipantRtcInitPerSec = perSec(prevAverage.ParticipantRtcInit, participantRTCInitNow, elapsed)
|
||||
stats.ParticipantRtcConnectedPerSec = perSec(prevAverage.ParticipantRtcConnected, participantRTConnectedCNow, elapsed)
|
||||
stats.SysPacketsOutPerSec = perSec(uint64(prevAverage.SysPacketsOut), uint64(sysPackets), elapsed)
|
||||
stats.SysPacketsDroppedPerSec = perSec(uint64(prevAverage.SysPacketsDropped), uint64(sysDroppedPackets), elapsed)
|
||||
stats.TrackPublishAttemptsPerSec = perSec(uint64(prevAverage.NumTrackPublishAttempts), uint64(trackPublishAttemptsNow), elapsed)
|
||||
stats.TrackPublishSuccessPerSec = perSec(uint64(prevAverage.NumTrackPublishSuccess), uint64(trackPublishSuccessNow), elapsed)
|
||||
stats.TrackSubscribeAttemptsPerSec = perSec(uint64(prevAverage.NumTrackSubscribeAttempts), uint64(trackSubscribeAttemptsNow), elapsed)
|
||||
stats.TrackSubscribeSuccessPerSec = perSec(uint64(prevAverage.NumTrackSubscribeSuccess), uint64(trackSubscribeSuccessNow), elapsed)
|
||||
for _, rateInterval := range rateIntervals {
|
||||
for idx := len(prevStats) - 1; idx >= 0; idx-- {
|
||||
prev := prevStats[idx]
|
||||
if prev == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
packetTotal := stats.SysPacketsOutPerSec + stats.SysPacketsDroppedPerSec
|
||||
if packetTotal == 0 {
|
||||
stats.SysPacketsDroppedPctPerSec = 0
|
||||
} else {
|
||||
stats.SysPacketsDroppedPctPerSec = stats.SysPacketsDroppedPerSec / packetTotal
|
||||
if stats.UpdatedAt-prev.UpdatedAt >= int64(rateInterval.Seconds()) {
|
||||
if rate := getNodeStatsRate(append(prevStats[idx:], stats)); rate != nil {
|
||||
stats.Rates = append(stats.Rates, rate)
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
promSysDroppedPacketPctGauge.Set(float64(stats.SysPacketsDroppedPctPerSec))
|
||||
}
|
||||
|
||||
return stats, computeAverage, nil
|
||||
return stats, nil
|
||||
}
|
||||
|
||||
func getNodeStatsRate(statsHistory []*livekit.NodeStats) *livekit.NodeStatsRate {
|
||||
if len(statsHistory) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
elapsed := statsHistory[len(statsHistory)-1].UpdatedAt - statsHistory[0].UpdatedAt
|
||||
if elapsed <= 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
// time weighted averages
|
||||
var cpuLoad, memoryLoad float32
|
||||
for idx := len(statsHistory) - 1; idx > 0; idx-- {
|
||||
stats := statsHistory[idx]
|
||||
prevStats := statsHistory[idx-1]
|
||||
if stats == nil || prevStats == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
spanElapsed := stats.UpdatedAt - prevStats.UpdatedAt
|
||||
if spanElapsed <= 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
cpuLoad += stats.CpuLoad * float32(spanElapsed)
|
||||
if stats.MemoryTotal > 0 {
|
||||
memoryLoad += float32(stats.MemoryUsed) / float32(stats.MemoryTotal) * float32(spanElapsed)
|
||||
}
|
||||
}
|
||||
|
||||
earlier := statsHistory[0]
|
||||
later := statsHistory[len(statsHistory)-1]
|
||||
rate := &livekit.NodeStatsRate{
|
||||
StartedAt: earlier.UpdatedAt,
|
||||
EndedAt: later.UpdatedAt,
|
||||
Duration: elapsed,
|
||||
BytesIn: perSec(earlier.BytesIn, later.BytesIn, elapsed),
|
||||
BytesOut: perSec(earlier.BytesOut, later.BytesOut, elapsed),
|
||||
PacketsIn: perSec(earlier.PacketsIn, later.PacketsIn, elapsed),
|
||||
PacketsOut: perSec(earlier.PacketsOut, later.PacketsOut, elapsed),
|
||||
RetransmitBytesOut: perSec(earlier.RetransmitBytesOut, later.RetransmitBytesOut, elapsed),
|
||||
RetransmitPacketsOut: perSec(earlier.RetransmitPacketsOut, later.RetransmitPacketsOut, elapsed),
|
||||
NackTotal: perSec(earlier.NackTotal, later.NackTotal, elapsed),
|
||||
ParticipantSignalConnected: perSec(earlier.ParticipantSignalConnected, later.ParticipantSignalConnected, elapsed),
|
||||
ParticipantRtcInit: perSec(earlier.ParticipantRtcInit, later.ParticipantRtcInit, elapsed),
|
||||
ParticipantRtcConnected: perSec(earlier.ParticipantRtcConnected, later.ParticipantRtcConnected, elapsed),
|
||||
SysPacketsOut: perSec(uint64(earlier.SysPacketsOut), uint64(later.SysPacketsOut), elapsed),
|
||||
SysPacketsDropped: perSec(uint64(earlier.SysPacketsDropped), uint64(later.SysPacketsDropped), elapsed),
|
||||
TrackPublishAttempts: perSec(uint64(earlier.NumTrackPublishAttempts), uint64(later.NumTrackPublishAttempts), elapsed),
|
||||
TrackPublishSuccess: perSec(uint64(earlier.NumTrackPublishSuccess), uint64(later.NumTrackPublishSuccess), elapsed),
|
||||
TrackSubscribeAttempts: perSec(uint64(earlier.NumTrackSubscribeAttempts), uint64(later.NumTrackSubscribeAttempts), elapsed),
|
||||
TrackSubscribeSuccess: perSec(uint64(earlier.NumTrackSubscribeSuccess), uint64(later.NumTrackSubscribeSuccess), elapsed),
|
||||
CpuLoad: cpuLoad / float32(elapsed),
|
||||
MemoryLoad: memoryLoad / float32(elapsed),
|
||||
}
|
||||
return rate
|
||||
}
|
||||
|
||||
func perSec(prev, curr uint64, secs int64) float32 {
|
||||
|
||||
@@ -121,7 +121,7 @@ func (s *BytesTrackStats) report() {
|
||||
}
|
||||
|
||||
func (s *BytesTrackStats) reporter() {
|
||||
ticker := time.NewTicker(telemetryNonMediaStatsUpdateInterval)
|
||||
ticker := time.NewTicker(5 * time.Second)
|
||||
defer func() {
|
||||
ticker.Stop()
|
||||
s.report()
|
||||
|
||||
@@ -91,8 +91,7 @@ const (
|
||||
workerCleanupWait = 3 * time.Minute
|
||||
jobsQueueMinSize = 2048
|
||||
|
||||
telemetryStatsUpdateInterval = time.Second * 30
|
||||
telemetryNonMediaStatsUpdateInterval = time.Second * 30
|
||||
telemetryStatsUpdateInterval = time.Second * 30
|
||||
)
|
||||
|
||||
type telemetryService struct {
|
||||
|
||||
Reference in New Issue
Block a user