From bd7e3beda41207a8895a9ce55cbb05ef5a782305 Mon Sep 17 00:00:00 2001
From: David Zhao <dz@livekit.io>
Date: Mon, 9 May 2022 08:55:06 -0700
Subject: [PATCH] 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                              | 12 ++++-----
 pkg/sfu/connectionquality/connectionstats.go |  8 +++---
 pkg/telemetry/telemetryserviceinternal.go    | 26 +++++---------------
 3 files changed, 16 insertions(+), 30 deletions(-)

diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go
index 580c2b1a6..9059eee32 100644
--- a/pkg/rtc/room.go
+++ b/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)
 	}
 }
 
diff --git a/pkg/sfu/connectionquality/connectionstats.go b/pkg/sfu/connectionquality/connectionstats.go
index 0683598fe..8ea44af9b 100644
--- a/pkg/sfu/connectionquality/connectionstats.go
+++ b/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)
diff --git a/pkg/telemetry/telemetryserviceinternal.go b/pkg/telemetry/telemetryserviceinternal.go
index 90515ef6a..8a189b3dd 100644
--- a/pkg/telemetry/telemetryserviceinternal.go
+++ b/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)
 }