Improve frequency of stats update (#673)

* Improve frequency of stats update

Prometheus stats are updated as the data becomes available, instead of
aggregated along with telemetry batches. Node availability decisions can
now react much faster to these stats.

* use the same intervals for connection quality updates

pkg/rtc/room.go

@@ -9,6 +9,7 @@ import (
 	"go.uber.org/atomic"
 	"google.golang.org/protobuf/proto"
 
+	"github.com/livekit/livekit-server/pkg/sfu/connectionquality"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 
@@ -799,11 +800,12 @@ func (r *Room) audioUpdateWorker() {
 }
 
 func (r *Room) connectionQualityWorker() {
+	ticker := time.NewTicker(connectionquality.UpdateInterval)
+	defer ticker.Stop()
+
 	// send updates to only users that are subscribed to each other
-	for {
-		if r.IsClosed() {
-			return
-		}
+	for !r.IsClosed() {
+		<-ticker.C
 
 		participants := r.GetParticipants()
 		connectionInfos := make(map[livekit.ParticipantID]*livekit.ConnectionQualityInfo, len(participants))
@@ -838,8 +840,6 @@ func (r *Room) connectionQualityWorker() {
 					"participant", op.Identity())
 			}
 		}
-
-		time.Sleep(time.Second * 5)
 	}
 }
 

pkg/sfu/connectionquality/connectionstats.go

@@ -14,7 +14,7 @@ import (
 )
 
 const (
-	connectionQualityUpdateInterval = 5 * time.Second
+	UpdateInterval = 2 * time.Second
 )
 
 type ConnectionStatsParams struct {
@@ -47,7 +47,7 @@ func NewConnectionStats(params ConnectionStatsParams) *ConnectionStats {
 }
 
 func (cs *ConnectionStats) Start() {
-	go cs.updateStats()
+	go cs.updateStatsWorker()
 }
 
 func (cs *ConnectionStats) Close() {
@@ -127,10 +127,10 @@ func (cs *ConnectionStats) getStat() *livekit.AnalyticsStat {
 	}
 }
 
-func (cs *ConnectionStats) updateStats() {
+func (cs *ConnectionStats) updateStatsWorker() {
 	interval := cs.params.UpdateInterval
 	if interval == 0 {
-		interval = connectionQualityUpdateInterval
+		interval = UpdateInterval
 	}
 
 	tk := time.NewTicker(interval)

pkg/telemetry/telemetryserviceinternal.go

@@ -53,12 +53,18 @@ func (t *telemetryServiceInternal) TrackStats(streamType livekit.StreamType, par
 	nacks := uint32(0)
 	plis := uint32(0)
 	firs := uint32(0)
+	packets := uint32(0)
+	bytes := uint64(0)
 	for _, stream := range stat.Streams {
 		nacks += stream.Nacks
 		plis += stream.Plis
 		firs += stream.Firs
+		packets += stream.PrimaryPackets + stream.RetransmitPackets + stream.PaddingPackets
+		bytes += stream.PrimaryBytes + stream.RetransmitBytes + stream.PaddingBytes
 	}
 	prometheus.IncrementRTCP(direction, nacks, plis, firs)
+	prometheus.IncrementPackets(direction, uint64(packets))
+	prometheus.IncrementBytes(direction, bytes)
 
 	if w := t.getStatsWorker(participantID); w != nil {
 		w.OnTrackStat(trackID, streamType, stat)
@@ -66,26 +72,6 @@ func (t *telemetryServiceInternal) TrackStats(streamType livekit.StreamType, par
 }
 
 func (t *telemetryServiceInternal) Report(ctx context.Context, stats []*livekit.AnalyticsStat) {
-	for _, stat := range stats {
-		if len(stat.Streams) == 0 {
-			continue
-		}
-
-		direction := prometheus.Incoming
-		if stat.Kind == livekit.StreamType_DOWNSTREAM {
-			direction = prometheus.Outgoing
-		}
-
-		packets := uint32(0)
-		bytes := uint64(0)
-		for _, stream := range stat.Streams {
-			packets += stream.PrimaryPackets + stream.RetransmitPackets + stream.PaddingPackets
-			bytes += stream.PrimaryBytes + stream.RetransmitBytes + stream.PaddingBytes
-		}
-		prometheus.IncrementPackets(direction, uint64(packets))
-		prometheus.IncrementBytes(direction, bytes)
-	}
-
 	t.analytics.SendStats(ctx, stats)
 }