Add a method to get forward stats via API method. (#4664)

Use by cloud simulated tracks.
Introduces a lock, but it is used only in the background worker (where
it will be almost always uncontested) and when API GetStats is accessed.
Does not touch the per-packet Update path.

pkg/sfu/forwardstats.go

@@ -2,6 +2,7 @@ package sfu
 
 import (
 	"math"
+	"sync"
 	"time"
 
 	"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
@@ -163,7 +164,9 @@ type ForwardStats struct {
 	samples forwardSampleBuffer
 
 	// ring of per-summary-interval summaries covering the report window.
-	// only touched by the background worker, so it needs no locking.
+	// written by the background worker (flush) and read both by the worker
+	// (report) and by external callers (GetStats), so it is guarded by lock.
+	lock     sync.Mutex
 	ring     []forwardSummary
 	ringHead int
 	ringLen  int
@@ -236,18 +239,52 @@ func (s *ForwardStats) flush() {
 		summ = summ.addSample(transitNs)
 	})
 
+	s.lock.Lock()
 	s.ring[s.ringHead] = summ
 	s.ringHead = (s.ringHead + 1) % len(s.ring)
 	if s.ringLen < len(s.ring) {
 		s.ringLen++
 	}
+	s.lock.Unlock()
+}
+
+// summarize merges the ring summaries covering the most recent window. A
+// window <= 0 (or >= the report window) covers the entire ring.
+func (s *ForwardStats) summarize(window time.Duration) forwardSummary {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	n := s.ringLen
+	if window > 0 && s.summaryInterval > 0 {
+		want := int((window + s.summaryInterval - 1) / s.summaryInterval)
+		if want < 1 {
+			want = 1
+		}
+		if want < n {
+			n = want
+		}
+	}
+
+	// walk backwards from the most recent entry (ringHead-1) over n entries.
+	var w forwardSummary
+	for i := 0; i < n; i++ {
+		idx := (s.ringHead - 1 - i + len(s.ring)) % len(s.ring)
+		w = w.merge(s.ring[idx])
+	}
+	return w
+}
+
+// GetStats returns the mean latency and jitter (std dev) of the forwarding
+// transit over the most recent duration. The duration is rounded up to a whole
+// number of summary intervals (the smallest bucket span that covers it). A
+// duration <= 0, or one that meets/exceeds the report window, covers the full
+// window.
+func (s *ForwardStats) GetStats(duration time.Duration) (time.Duration, time.Duration) {
+	return s.summarize(duration).meanStdDev()
 }
 
 func (s *ForwardStats) report() {
-	var w forwardSummary
-	for i := 0; i < s.ringLen; i++ {
-		w = w.merge(s.ring[i])
-	}
+	w := s.summarize(0)
 
 	latency, jitter := w.meanStdDev()
 	if dropped := s.samples.takeDropped(); dropped > 0 {

pkg/sfu/forwardstats_test.go

@@ -278,6 +278,37 @@ func TestForwardStats_ReportWindow(t *testing.T) {
 	require.NotPanics(t, s.report)
 }
 
+func TestForwardStats_GetStats(t *testing.T) {
+	initPrometheus(t)
+
+	// 5 buckets, each covering one 100ms summary interval.
+	s := &ForwardStats{ring: make([]forwardSummary, 5), summaryInterval: 100 * time.Millisecond}
+
+	// fold five 100ms buckets, one sample each: 1ms, 2ms, 3ms, 4ms, 5ms.
+	for i := 1; i <= 5; i++ {
+		s.Update(0, int64(i)*int64(time.Millisecond))
+		s.flush()
+	}
+	require.Equal(t, 5, s.ringLen)
+
+	// a duration <= 0 covers the whole window: mean of 1..5ms == 3ms.
+	latency, jitter := s.GetStats(0)
+	require.InDelta(t, float64(3*time.Millisecond), float64(latency), float64(50*time.Microsecond))
+	require.Greater(t, jitter, time.Duration(0))
+
+	// a duration meeting/exceeding the window also covers it.
+	fullLatency, _ := s.GetStats(time.Second)
+	require.InDelta(t, float64(3*time.Millisecond), float64(fullLatency), float64(50*time.Microsecond))
+
+	// ~200ms rounds up to the two most recent buckets (4ms, 5ms): mean == 4.5ms.
+	shortLatency, _ := s.GetStats(200 * time.Millisecond)
+	require.InDelta(t, float64(4500*time.Microsecond), float64(shortLatency), float64(50*time.Microsecond))
+
+	// a sub-interval duration still yields at least the most recent bucket (5ms).
+	lastLatency, _ := s.GetStats(time.Nanosecond)
+	require.InDelta(t, float64(5*time.Millisecond), float64(lastLatency), float64(50*time.Microsecond))
+}
+
 func TestForwardStats_Lifecycle(t *testing.T) {
 	initPrometheus(t)