Imlement analytics per track statistics (#281)

New tests for telemetry + implementation of per track statistics

pkg/rtc/mediatrack.go

@@ -286,13 +286,13 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error {
 		go t.sendDownTrackBindingReports(sub)
 	})
 	downTrack.OnPacketSent(func(_ *sfu.DownTrack, size int) {
-		t.params.Telemetry.OnDownstreamPacket(subscriberID, size)
+		t.params.Telemetry.OnDownstreamPacket(subscriberID, t.ID(), size)
 	})
 	downTrack.OnPaddingSent(func(_ *sfu.DownTrack, size int) {
-		t.params.Telemetry.OnDownstreamPacket(subscriberID, size)
+		t.params.Telemetry.OnDownstreamPacket(subscriberID, t.ID(), size)
 	})
 	downTrack.OnRTCP(func(pkts []rtcp.Packet) {
-		t.params.Telemetry.HandleRTCP(livekit.StreamType_DOWNSTREAM, subscriberID, pkts)
+		t.params.Telemetry.HandleRTCP(livekit.StreamType_DOWNSTREAM, subscriberID, t.ID(), pkts)
 	})
 
 	downTrack.OnCloseHandler(func() {
@@ -437,7 +437,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
 	}
 
 	t.receiver.AddUpTrack(track, buff)
-	t.params.Telemetry.AddUpTrack(t.params.ParticipantID, buff)
+	t.params.Telemetry.AddUpTrack(t.params.ParticipantID, t.ID(), buff)
 
 	atomic.AddUint32(&t.numUpTracks, 1)
 	// LK-TODO: can remove this completely when VideoLayers protocol becomes the default as it has info from client or if we decide to use TrackInfo.Simulcast
@@ -631,6 +631,9 @@ func (t *MediaTrack) handlePublisherFeedback(packets []rtcp.Packet) {
 	var totalLost uint32
 	var maxSeqNum uint32
 
+	//forward to telemetry
+	t.params.Telemetry.HandleRTCP(livekit.StreamType_UPSTREAM, t.params.ParticipantID, t.ID(), packets)
+
 	for _, p := range packets {
 		switch pkt := p.(type) {
 		// sfu.Buffer generates ReceiverReports for the publisher

pkg/rtc/transport.go

@@ -7,7 +7,6 @@ import (
 	"github.com/bep/debounce"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
-	"github.com/pion/interceptor"
 	"github.com/pion/webrtc/v3"
 
 	"github.com/livekit/livekit-server/pkg/rtc/types"
@@ -70,16 +69,9 @@ func newPeerConnection(params TransportParams) (*webrtc.PeerConnection, *webrtc.
 	se := params.Config.SettingEngine
 	se.DisableMediaEngineCopy(true)
 
-	ir := &interceptor.Registry{}
-	// intercept pub -> SFU rtcp for analytics
-	if params.Telemetry != nil && params.Target == livekit.SignalTarget_PUBLISHER {
-		f := params.Telemetry.NewStatsInterceptorFactory(params.ParticipantID, params.ParticipantIdentity)
-		ir.Add(f)
-	}
 	api := webrtc.NewAPI(
 		webrtc.WithMediaEngine(me),
 		webrtc.WithSettingEngine(se),
-		webrtc.WithInterceptorRegistry(ir),
 	)
 	pc, err := api.NewPeerConnection(params.Config.Configuration)
 	return pc, me, err

pkg/sfu/buffer/buffer.go

@@ -618,6 +618,13 @@ func (b *Buffer) GetStats() (stats Stats) {
 	return
 }
 
+// Used only in tests
+func (b *Buffer) SetStatsTestOnly(stats Stats) {
+	b.Lock()
+	b.stats = stats
+	b.Unlock()
+}
+
 // GetLatestTimestamp returns the latest RTP timestamp factoring in potential RTP timestamp wrap-around
 func (b *Buffer) GetLatestTimestamp() (latestTimestamp uint32, latestTimestampTimeInNanosSinceEpoch int64) {
 	latestTimestamp = atomic.LoadUint32(&b.latestTimestamp)

pkg/telemetry/interceptor.go

@@ -1,46 +0,0 @@
-package telemetry
-
-import (
-	"github.com/livekit/protocol/livekit"
-	"github.com/pion/interceptor"
-	"github.com/pion/rtcp"
-)
-
-func (t *telemetryServiceInternal) NewStatsInterceptorFactory(participantID, identity string) *StatsInterceptorFactory {
-	return &StatsInterceptorFactory{
-		t:             t,
-		participantID: participantID,
-		identity:      identity,
-	}
-}
-
-type StatsInterceptorFactory struct {
-	t             TelemetryService
-	participantID string
-	identity      string
-}
-
-func (f *StatsInterceptorFactory) NewInterceptor(_ string) (interceptor.Interceptor, error) {
-	return &StatsInterceptor{
-		t:             f.t,
-		participantID: f.participantID,
-		identity:      f.identity,
-	}, nil
-}
-
-type StatsInterceptor struct {
-	interceptor.NoOp
-
-	t             TelemetryService
-	participantID string
-	identity      string
-}
-
-// 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.t.HandleRTCP(livekit.StreamType_UPSTREAM, s.participantID, pkts)
-		return writer.Write(pkts, attributes)
-	})
-}

pkg/telemetry/statsworker.go

@@ -19,15 +19,14 @@ type StatsWorker struct {
 	participantID string
 
 	sync.RWMutex
-	buffers map[uint32]*buffer.Buffer
-	drain   map[uint32]bool
+	upstreamBuffers      map[string][]*buffer.Buffer
+	drainUpstreamBuffers map[string]bool
 
-	incoming *Stats
-	outgoing *Stats
+	outgoingPerTrack map[string]*Stats
+	incomingPerTrack map[string]*Stats
 }
 
 type Stats struct {
-	sync.Mutex
 	next         *livekit.AnalyticsStat
 	totalPackets uint32
 	prevPackets  uint32
@@ -43,48 +42,64 @@ func newStatsWorker(ctx context.Context, t TelemetryReporter, roomID, roomName,
 		roomName:      roomName,
 		participantID: participantID,
 
-		buffers: make(map[uint32]*buffer.Buffer),
-		drain:   make(map[uint32]bool),
+		upstreamBuffers:      make(map[string][]*buffer.Buffer),
+		drainUpstreamBuffers: make(map[string]bool),
 
-		incoming: &Stats{next: &livekit.AnalyticsStat{
-			Kind:          livekit.StreamType_UPSTREAM,
-			RoomId:        roomID,
-			ParticipantId: participantID,
-			RoomName:      roomName,
-		}},
-		outgoing: &Stats{next: &livekit.AnalyticsStat{
-			Kind:          livekit.StreamType_DOWNSTREAM,
-			RoomId:        roomID,
-			ParticipantId: participantID,
-			RoomName:      roomName,
-		}},
+		outgoingPerTrack: make(map[string]*Stats),
+		incomingPerTrack: make(map[string]*Stats),
 	}
 	return s
 }
 
-func (s *StatsWorker) AddBuffer(buffer *buffer.Buffer) {
+func (s *StatsWorker) AddBuffer(trackID string, buffer *buffer.Buffer) {
 	s.Lock()
 	defer s.Unlock()
 
-	s.buffers[buffer.GetMediaSSRC()] = buffer
+	s.upstreamBuffers[trackID] = append(s.upstreamBuffers[trackID], buffer)
 }
 
-func (s *StatsWorker) OnDownstreamPacket(bytes int) {
-	s.outgoing.Lock()
-	defer s.outgoing.Unlock()
+func (s *StatsWorker) OnDownstreamPacket(trackID string, bytes int) {
+	s.Lock()
+	defer s.Unlock()
 
-	s.outgoing.totalPackets++
-	s.outgoing.totalBytes += uint64(bytes)
+	s.getOrCreateOutgoingStatsIfEmpty(trackID).totalBytes += uint64(bytes)
+	s.getOrCreateOutgoingStatsIfEmpty(trackID).totalPackets++
 }
 
-func (s *StatsWorker) OnRTCP(direction livekit.StreamType, stats *livekit.AnalyticsStat) {
-	ds := s.incoming
-	if direction == livekit.StreamType_DOWNSTREAM {
-		ds = s.outgoing
+func (s *StatsWorker) getOrCreateOutgoingStatsIfEmpty(trackID string) *Stats {
+	if s.outgoingPerTrack[trackID] == nil {
+		s.outgoingPerTrack[trackID] = &Stats{next: &livekit.AnalyticsStat{
+			Kind:          livekit.StreamType_DOWNSTREAM,
+			RoomId:        s.roomID,
+			ParticipantId: s.participantID,
+			RoomName:      s.roomName,
+		}}
 	}
+	return s.outgoingPerTrack[trackID]
+}
 
-	ds.Lock()
-	defer ds.Unlock()
+func (s *StatsWorker) getOrCreateIncomingStatsIfEmpty(trackID string) *Stats {
+	if s.incomingPerTrack[trackID] == nil {
+		s.incomingPerTrack[trackID] = &Stats{next: &livekit.AnalyticsStat{
+			Kind:          livekit.StreamType_UPSTREAM,
+			RoomId:        s.roomID,
+			ParticipantId: s.participantID,
+			RoomName:      s.roomName,
+		}}
+	}
+	return s.incomingPerTrack[trackID]
+}
+
+func (s *StatsWorker) OnRTCP(trackID string, direction livekit.StreamType, stats *livekit.AnalyticsStat) {
+	s.Lock()
+	defer s.Unlock()
+
+	var ds *Stats
+	if direction == livekit.StreamType_DOWNSTREAM {
+		ds = s.getOrCreateOutgoingStatsIfEmpty(trackID)
+	} else {
+		ds = s.getOrCreateIncomingStatsIfEmpty(trackID)
+	}
 
 	if stats.Delay > ds.next.Delay {
 		ds.next.Delay = stats.Delay
@@ -98,52 +113,70 @@ func (s *StatsWorker) OnRTCP(direction livekit.StreamType, stats *livekit.Analyt
 	ds.next.FirCount += stats.FirCount
 }
 
+func (s *StatsWorker) calculateTotalBytesPackets(allBuffers []*buffer.Buffer) (totalBytes uint64, totalPackets uint32) {
+	totalBytes = 0
+	totalPackets = 0
+
+	for _, buffer := range allBuffers {
+		totalBytes += buffer.GetStats().TotalByte
+		totalPackets += buffer.GetStats().PacketCount
+	}
+	return totalBytes, totalPackets
+}
+
 func (s *StatsWorker) Update() {
-	var packetsIn uint32
-	var bytesIn uint64
-
 	s.Lock()
+	defer s.Unlock()
+
 	ts := timestamppb.Now()
-	for _, buff := range s.buffers {
-		stats := buff.GetStats()
-		packetsIn += stats.PacketCount
-		bytesIn += stats.TotalByte
-	}
+	stats := make([]*livekit.AnalyticsStat, 0)
 
-	if len(s.drain) > 0 {
-		for ssrc := range s.drain {
-			delete(s.buffers, ssrc)
-		}
-		s.drain = make(map[uint32]bool)
-	}
-	s.Unlock()
-
-	s.incoming.Lock()
-	s.incoming.totalPackets = packetsIn
-	s.incoming.totalBytes = bytesIn
-	s.incoming.Unlock()
-
-	stats := make([]*livekit.AnalyticsStat, 0, 2)
-	upstream := s.update(s.incoming, ts)
-	if upstream != nil {
-		stats = append(stats, upstream)
-	}
-	downstream := s.update(s.outgoing, ts)
-	if downstream != nil {
-		stats = append(stats, downstream)
-	}
+	stats = s.collectUpstreamStats(ts, stats)
+	stats = s.collectDownstreamStats(ts, stats)
 
 	s.t.Report(s.ctx, stats)
 }
 
+func (s *StatsWorker) collectDownstreamStats(ts *timestamppb.Timestamp, stats []*livekit.AnalyticsStat) []*livekit.AnalyticsStat {
+	for trackID, trackDownStreamStats := range s.outgoingPerTrack {
+		analyticsStat := s.update(trackDownStreamStats, ts)
+		if analyticsStat != nil {
+			analyticsStat.TrackId = trackID
+			stats = append(stats, analyticsStat)
+		}
+	}
+	return stats
+}
+
+func (s *StatsWorker) collectUpstreamStats(ts *timestamppb.Timestamp, stats []*livekit.AnalyticsStat) []*livekit.AnalyticsStat {
+	for trackID, buffers := range s.upstreamBuffers {
+		totalBytes, totalPackets := s.calculateTotalBytesPackets(buffers)
+
+		s.getOrCreateIncomingStatsIfEmpty(trackID).totalBytes = totalBytes
+		s.getOrCreateIncomingStatsIfEmpty(trackID).totalPackets = totalPackets
+
+		analyticsStats := s.update(s.incomingPerTrack[trackID], ts)
+		if analyticsStats != nil {
+			analyticsStats.TrackId = trackID
+			stats = append(stats, analyticsStats)
+		}
+	}
+
+	if len(s.drainUpstreamBuffers) > 0 {
+		for trackID := range s.drainUpstreamBuffers {
+			delete(s.upstreamBuffers, trackID)
+			delete(s.incomingPerTrack, trackID)
+		}
+		s.drainUpstreamBuffers = make(map[string]bool)
+	}
+	return stats
+}
+
 func (s *StatsWorker) update(stats *Stats, ts *timestamppb.Timestamp) *livekit.AnalyticsStat {
 	if stats.totalBytes == 0 {
 		return nil
 	}
 
-	stats.Lock()
-	defer stats.Unlock()
-
 	next := stats.next
 	stats.next = &livekit.AnalyticsStat{
 		Kind:          next.Kind,
@@ -162,9 +195,9 @@ func (s *StatsWorker) update(stats *Stats, ts *timestamppb.Timestamp) *livekit.A
 	return next
 }
 
-func (s *StatsWorker) RemoveBuffer(ssrc uint32) {
+func (s *StatsWorker) RemoveBuffer(trackID string) {
 	s.Lock()
-	s.drain[ssrc] = true
+	s.drainUpstreamBuffers[trackID] = true
 	s.Unlock()
 }
 

pkg/telemetry/telemetryservice.go

@@ -15,10 +15,9 @@ const updateFrequency = time.Second * 10
 
 type TelemetryService interface {
 	// stats
-	NewStatsInterceptorFactory(participantID, identity string) *StatsInterceptorFactory
-	AddUpTrack(participantID string, buff *buffer.Buffer)
-	OnDownstreamPacket(participantID string, bytes int)
-	HandleRTCP(streamType livekit.StreamType, participantID string, pkts []rtcp.Packet)
+	AddUpTrack(participantID string, trackID string, buff *buffer.Buffer)
+	OnDownstreamPacket(participantID string, trackID string, bytes int)
+	HandleRTCP(streamType livekit.StreamType, participantID string, trackID string, pkts []rtcp.Packet)
 
 	// events
 	RoomStarted(ctx context.Context, room *livekit.Room)
@@ -56,16 +55,16 @@ func (t *telemetryService) run() {
 	}
 }
 
-func (t *telemetryService) AddUpTrack(participantID string, buff *buffer.Buffer) {
-	t.internalService.AddUpTrack(participantID, buff)
+func (t *telemetryService) AddUpTrack(participantID string, trackID string, buff *buffer.Buffer) {
+	t.internalService.AddUpTrack(participantID, trackID, buff)
 }
 
-func (t *telemetryService) OnDownstreamPacket(participantID string, bytes int) {
-	t.internalService.OnDownstreamPacket(participantID, bytes)
+func (t *telemetryService) OnDownstreamPacket(participantID string, trackID string, bytes int) {
+	t.internalService.OnDownstreamPacket(participantID, trackID, bytes)
 }
 
-func (t *telemetryService) HandleRTCP(streamType livekit.StreamType, participantID string, pkts []rtcp.Packet) {
-	t.internalService.HandleRTCP(streamType, participantID, pkts)
+func (t *telemetryService) HandleRTCP(streamType livekit.StreamType, participantID string, trackID string, pkts []rtcp.Packet) {
+	t.internalService.HandleRTCP(streamType, participantID, trackID, pkts)
 }
 
 func (t *telemetryService) RoomStarted(ctx context.Context, room *livekit.Room) {
@@ -107,7 +106,3 @@ func (t *telemetryService) RecordingStarted(ctx context.Context, ri *livekit.Rec
 func (t *telemetryService) RecordingEnded(ctx context.Context, ri *livekit.RecordingInfo) {
 	t.internalService.RecordingEnded(ctx, ri)
 }
-
-func (t *telemetryService) NewStatsInterceptorFactory(participantID, identity string) *StatsInterceptorFactory {
-	return t.internalService.NewStatsInterceptorFactory(participantID, identity)
-}

pkg/telemetry/telemetryserviceinternal.go

@@ -42,25 +42,25 @@ func NewTelemetryServiceInternal(notifier webhook.Notifier, analytics AnalyticsS
 	}
 }
 
-func (t *telemetryServiceInternal) AddUpTrack(participantID string, buff *buffer.Buffer) {
+func (t *telemetryServiceInternal) AddUpTrack(participantID string, trackID string, buff *buffer.Buffer) {
 	t.RLock()
 	w := t.workers[participantID]
 	t.RUnlock()
 	if w != nil {
-		w.AddBuffer(buff)
+		w.AddBuffer(trackID, buff)
 	}
 }
 
-func (t *telemetryServiceInternal) OnDownstreamPacket(participantID string, bytes int) {
+func (t *telemetryServiceInternal) OnDownstreamPacket(participantID string, trackID string, bytes int) {
 	t.RLock()
 	w := t.workers[participantID]
 	t.RUnlock()
 	if w != nil {
-		w.OnDownstreamPacket(bytes)
+		w.OnDownstreamPacket(trackID, bytes)
 	}
 }
 
-func (t *telemetryServiceInternal) HandleRTCP(streamType livekit.StreamType, participantID string, pkts []rtcp.Packet) {
+func (t *telemetryServiceInternal) HandleRTCP(streamType livekit.StreamType, participantID string, trackID string, pkts []rtcp.Packet) {
 	stats := &livekit.AnalyticsStat{}
 	for _, pkt := range pkts {
 		switch pkt := pkt.(type) {
@@ -94,7 +94,7 @@ func (t *telemetryServiceInternal) HandleRTCP(streamType livekit.StreamType, par
 	w := t.workers[participantID]
 	t.RUnlock()
 	if w != nil {
-		w.OnRTCP(streamType, stats)
+		w.OnRTCP(trackID, streamType, stats)
 	}
 }
 

pkg/telemetry/telemetryserviceinternalevents.go

@@ -59,12 +59,13 @@ func (t *telemetryServiceInternal) ParticipantJoined(ctx context.Context, room *
 	})
 
 	t.analytics.SendEvent(ctx, &livekit.AnalyticsEvent{
-		Type:        livekit.AnalyticsEventType_PARTICIPANT_JOINED,
-		Timestamp:   timestamppb.Now(),
-		RoomSid:     room.Sid,
-		Participant: participant,
-		Room:        room,
-		SdkType:     clientInfo.GetSdk(),
+		Type:          livekit.AnalyticsEventType_PARTICIPANT_JOINED,
+		Timestamp:     timestamppb.Now(),
+		RoomSid:       room.Sid,
+		ParticipantId: participant.Sid,
+		Participant:   participant,
+		Room:          room,
+		SdkType:       clientInfo.GetSdk(),
 	})
 }
 
@@ -115,7 +116,7 @@ func (t *telemetryServiceInternal) TrackUnpublished(ctx context.Context, partici
 	t.RUnlock()
 	if w != nil {
 		roomID = w.roomID
-		w.RemoveBuffer(ssrc)
+		w.RemoveBuffer(track.GetSid())
 		roomName = w.roomName
 	}
 

pkg/telemetry/test/telemetry_service_events_test.go

@@ -0,0 +1,52 @@
+package telemetrytest
+
+import (
+	"context"
+	"testing"
+
+	"github.com/livekit/protocol/livekit"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_OnParticipantJoin_EventIsSent(t *testing.T) {
+	fixture := createFixture()
+
+	//prepare
+	room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
+	partSID := "part1"
+	clientInfo := &livekit.ClientInfo{Sdk: 2}
+	participantInfo := &livekit.ParticipantInfo{Sid: partSID}
+
+	//do
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo)
+
+	//test
+	require.Equal(t, 1, fixture.analytics.SendEventCallCount())
+	_, event := fixture.analytics.SendEventArgsForCall(0)
+	require.Equal(t, livekit.AnalyticsEventType_PARTICIPANT_JOINED, event.Type)
+	require.Equal(t, partSID, event.ParticipantId)
+	require.Equal(t, participantInfo, event.Participant)
+	require.Equal(t, room.Sid, event.RoomSid)
+	require.Equal(t, room, event.Room)
+	require.Equal(t, clientInfo.Sdk, event.SdkType)
+}
+
+func Test_OnParticipantLeft_EventIsSent(t *testing.T) {
+	fixture := createFixture()
+
+	//prepare
+	room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
+	partSID := "part1"
+	participantInfo := &livekit.ParticipantInfo{Sid: partSID}
+
+	//do
+	fixture.sut.ParticipantLeft(context.Background(), room, participantInfo)
+
+	//test
+	require.Equal(t, 1, fixture.analytics.SendEventCallCount())
+	_, event := fixture.analytics.SendEventArgsForCall(0)
+	require.Equal(t, livekit.AnalyticsEventType_PARTICIPANT_LEFT, event.Type)
+	require.Equal(t, partSID, event.ParticipantId)
+	require.Equal(t, room.Sid, event.RoomSid)
+	require.Equal(t, room, event.Room)
+}

pkg/telemetry/test/telemetry_service_test.go

@@ -5,8 +5,10 @@ import (
 	"testing"
 
 	"github.com/livekit/protocol/livekit"
+	"github.com/pion/rtcp"
 	"github.com/stretchr/testify/require"
 
+	"github.com/livekit/livekit-server/pkg/sfu/buffer"
 	"github.com/livekit/livekit-server/pkg/telemetry"
 	"github.com/livekit/livekit-server/pkg/telemetry/telemetryfakes"
 )
@@ -23,7 +25,32 @@ func createFixture() *telemetryServiceFixture {
 	return fixture
 }
 
-func Test_OnDownstreamPacket(t *testing.T) {
+func Test_ParticipantAndRoomDataAreSentWithAnalytics(t *testing.T) {
+	fixture := createFixture()
+
+	//prepare
+	room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
+	partSID := "part1"
+	clientInfo := &livekit.ClientInfo{Sdk: 2}
+	participantInfo := &livekit.ParticipantInfo{Sid: partSID}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo)
+
+	//do
+	packet := 33
+	fixture.sut.OnDownstreamPacket(partSID, "", packet)
+	fixture.sut.SendAnalytics()
+
+	//test
+	require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
+	_, stats := fixture.analytics.SendStatsArgsForCall(0)
+	require.Equal(t, 1, len(stats))
+	require.Equal(t, livekit.StreamType_DOWNSTREAM, stats[0].Kind)
+	require.Equal(t, partSID, stats[0].ParticipantId)
+	require.Equal(t, room.Sid, stats[0].RoomId)
+	require.Equal(t, room.Name, stats[0].RoomName)
+}
+
+func Test_OnDownstreamPackets(t *testing.T) {
 	fixture := createFixture()
 
 	//prepare
@@ -37,8 +64,9 @@ func Test_OnDownstreamPacket(t *testing.T) {
 	packets := []int{33, 23}
 	totalBytes := packets[0] + packets[1]
 	totalPackets := len(packets)
+	trackID := "trackID"
 	for i := range packets {
-		fixture.sut.OnDownstreamPacket(partSID, packets[i])
+		fixture.sut.OnDownstreamPacket(partSID, trackID, packets[i])
 	}
 	fixture.sut.SendAnalytics()
 
@@ -49,6 +77,248 @@ func Test_OnDownstreamPacket(t *testing.T) {
 	require.Equal(t, livekit.StreamType_DOWNSTREAM, stats[0].Kind)
 	require.Equal(t, totalBytes, int(stats[0].TotalBytes))
 	require.Equal(t, totalPackets, int(stats[0].TotalPackets))
+	require.Equal(t, trackID, stats[0].TrackId)
+}
+
+func Test_OnDownstreamPackets_SeveralTracks(t *testing.T) {
+	fixture := createFixture()
+
+	//prepare
+	room := &livekit.Room{}
+	partSID := "part1"
+	clientInfo := &livekit.ClientInfo{Sdk: 2}
+	participantInfo := &livekit.ParticipantInfo{Sid: partSID}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo)
+
+	//do
+	packet1 := 33
+	trackID1 := "trackID1"
+	packet2 := 23
+	trackID2 := "trackID2"
+	fixture.sut.OnDownstreamPacket(partSID, trackID1, packet1)
+	fixture.sut.OnDownstreamPacket(partSID, trackID2, packet2)
+	fixture.sut.SendAnalytics()
+
+	//test
+	require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
+	_, stats := fixture.analytics.SendStatsArgsForCall(0)
+	require.Equal(t, 2, len(stats))
+
+	found1 := false
+	found2 := false
+	for _, sentStat := range stats {
+		if sentStat.TrackId == trackID1 {
+			found1 = true
+			require.Equal(t, packet1, int(sentStat.TotalBytes))
+			require.Equal(t, 1, int(sentStat.TotalPackets))
+		} else if sentStat.TrackId == trackID2 {
+			found2 = true
+			require.Equal(t, packet2, int(sentStat.TotalBytes))
+			require.Equal(t, 1, int(sentStat.TotalPackets))
+		}
+	}
+	require.True(t, found1)
+	require.True(t, found2)
+}
+
+func Test_OnDownStreamRTCP(t *testing.T) {
+	fixture := createFixture()
+
+	//prepare
+	room := &livekit.Room{}
+	partSID := "part1"
+	participantInfo := &livekit.ParticipantInfo{Sid: partSID}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil)
+
+	//do
+	pkts := []rtcp.Packet{
+		&rtcp.TransportLayerNack{},
+		&rtcp.PictureLossIndication{},
+		&rtcp.FullIntraRequest{},
+		&rtcp.ReceiverReport{
+			Reports: []rtcp.ReceptionReport{
+				{Delay: 1, Jitter: 5, TotalLost: 3},
+				{Delay: 4, Jitter: 2, TotalLost: 4},
+			},
+		},
+	}
+	trackID := "trackID1"
+	fixture.sut.OnDownstreamPacket(partSID, trackID, 1) // there should be bytes reported so that stats are sent
+	fixture.sut.HandleRTCP(livekit.StreamType_DOWNSTREAM, partSID, trackID, pkts)
+	fixture.sut.SendAnalytics()
+
+	//test
+	require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
+	_, stats := fixture.analytics.SendStatsArgsForCall(0)
+	require.Equal(t, 1, len(stats))
+	require.Equal(t, livekit.StreamType_DOWNSTREAM, stats[0].Kind)
+	require.Equal(t, 1, int(stats[0].NackCount))
+	require.Equal(t, 1, int(stats[0].PliCount))
+	require.Equal(t, 1, int(stats[0].FirCount))
+	require.Equal(t, 4, int(stats[0].Delay))      // max of delay, see list of rtcp.ReceptionReport above
+	require.Equal(t, 5, int(stats[0].Jitter))     // max of jitter, see list of rtcp.ReceptionReport above
+	require.Equal(t, 7, int(stats[0].PacketLost)) // sum of lost packets, see list of rtcp.ReceptionReport above
+	require.Equal(t, trackID, stats[0].TrackId)
+}
+
+func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) {
+	fixture := createFixture()
+
+	//prepare
+	room := &livekit.Room{}
+	partSID := "part1"
+	participantInfo := &livekit.ParticipantInfo{Sid: partSID}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil)
+
+	//do
+	pkts1 := []rtcp.Packet{
+		&rtcp.TransportLayerNack{},
+	}
+	pkts2 := []rtcp.Packet{
+		&rtcp.FullIntraRequest{},
+	}
+	trackID1 := "trackID1"
+	trackID2 := "trackID2"
+	fixture.sut.OnDownstreamPacket(partSID, trackID1, 1) // there should be bytes reported so that stats are sent
+	fixture.sut.HandleRTCP(livekit.StreamType_DOWNSTREAM, partSID, trackID1, pkts1)
+	fixture.sut.OnDownstreamPacket(partSID, trackID2, 1) // there should be bytes reported so that stats are sent
+	fixture.sut.HandleRTCP(livekit.StreamType_DOWNSTREAM, partSID, trackID2, pkts2)
+	fixture.sut.SendAnalytics()
+
+	//test
+	require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
+	_, stats := fixture.analytics.SendStatsArgsForCall(0)
+	require.Equal(t, 2, len(stats))
+
+	found1 := false
+	found2 := false
+	for _, sentStat := range stats {
+		if sentStat.TrackId == trackID1 {
+			found1 = true
+			require.Equal(t, livekit.StreamType_DOWNSTREAM, sentStat.Kind)
+			require.Equal(t, 1, int(sentStat.NackCount)) // see pkts1 above
+		} else if sentStat.TrackId == trackID2 {
+			found2 = true
+			require.Equal(t, livekit.StreamType_DOWNSTREAM, sentStat.Kind)
+			require.Equal(t, 1, int(sentStat.FirCount)) // see pkts2 above
+		}
+	}
+	require.True(t, found1)
+	require.True(t, found2)
+}
+func Test_OnUpstreamRTCP(t *testing.T) {
+	fixture := createFixture()
+
+	//prepare
+	room := &livekit.Room{}
+	partSID := "part1"
+	participantInfo := &livekit.ParticipantInfo{Sid: partSID}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil)
+
+	//do
+	pkts := []rtcp.Packet{
+		&rtcp.TransportLayerNack{},
+		&rtcp.PictureLossIndication{},
+		&rtcp.FullIntraRequest{},
+		&rtcp.ReceiverReport{
+			Reports: []rtcp.ReceptionReport{
+				{Delay: 1, Jitter: 5, TotalLost: 3},
+				{Delay: 4, Jitter: 2, TotalLost: 4},
+			},
+		},
+	}
+	// there should be bytes reported so that stats are sent
+	buf := &buffer.Buffer{}
+	buf.SetStatsTestOnly(buffer.Stats{
+		PacketCount: 1,
+		TotalByte:   1,
+	})
+	trackID := "trackID"
+	fixture.sut.AddUpTrack(partSID, trackID, buf)
+	fixture.sut.HandleRTCP(livekit.StreamType_UPSTREAM, partSID, trackID, pkts)
+	fixture.sut.SendAnalytics()
+
+	//test
+	require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
+	_, stats := fixture.analytics.SendStatsArgsForCall(0)
+	require.Equal(t, 1, len(stats))
+	require.Equal(t, livekit.StreamType_UPSTREAM, stats[0].Kind)
+	require.Equal(t, 1, int(stats[0].NackCount))
+	require.Equal(t, 1, int(stats[0].PliCount))
+	require.Equal(t, 1, int(stats[0].FirCount))
+	require.Equal(t, 4, int(stats[0].Delay))      // max of delay, see list of rtcp.ReceptionReport above
+	require.Equal(t, 5, int(stats[0].Jitter))     // max of jitter, see list of rtcp.ReceptionReport above
+	require.Equal(t, 7, int(stats[0].PacketLost)) // sum of lost packets, see list of rtcp.ReceptionReport above
+	require.Equal(t, trackID, stats[0].TrackId)
+}
+
+func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) {
+	fixture := createFixture()
+
+	//prepare
+	room := &livekit.Room{}
+	partSID := "part1"
+	participantInfo := &livekit.ParticipantInfo{Sid: partSID}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil)
+
+	// there should be bytes reported so that stats are sent
+	buf := &buffer.Buffer{}
+	totalBytes := 1
+	tolalPackets := 1
+	buf.SetStatsTestOnly(buffer.Stats{
+		PacketCount: uint32(tolalPackets),
+		TotalByte:   uint64(totalBytes),
+	})
+	trackID1 := "trackID1"
+	trackID2 := "trackID2"
+	fixture.sut.AddUpTrack(partSID, trackID1, buf)
+	fixture.sut.AddUpTrack(partSID, trackID2, buf) //using same buffer is not correct but for test it is fine
+	pkts1 := []rtcp.Packet{
+		&rtcp.TransportLayerNack{},
+	}
+	pkts2 := []rtcp.Packet{
+		&rtcp.FullIntraRequest{},
+	}
+	//do
+	fixture.sut.HandleRTCP(livekit.StreamType_UPSTREAM, partSID, trackID1, pkts1)
+	fixture.sut.HandleRTCP(livekit.StreamType_UPSTREAM, partSID, trackID2, pkts2)
+	fixture.sut.SendAnalytics()
+
+	//test
+	require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
+	_, stats := fixture.analytics.SendStatsArgsForCall(0)
+	require.Equal(t, 2, len(stats))
+
+	found1 := false
+	found2 := false
+	for _, sentStat := range stats {
+		if sentStat.TrackId == trackID1 {
+			found1 = true
+			require.Equal(t, livekit.StreamType_UPSTREAM, sentStat.Kind)
+			require.Equal(t, 1, int(sentStat.NackCount)) // see pkts1 above
+		} else if sentStat.TrackId == trackID2 {
+			found2 = true
+			require.Equal(t, livekit.StreamType_UPSTREAM, sentStat.Kind)
+			require.Equal(t, 1, int(sentStat.FirCount)) // see pkts2 above
+		}
+		require.Equal(t, totalBytes, int(sentStat.TotalBytes))
+		require.Equal(t, tolalPackets, int(sentStat.TotalPackets))
+	}
+	require.True(t, found1)
+	require.True(t, found2)
+
+	//remove 1 buffer
+	fixture.sut.TrackUnpublished(context.Background(), partSID, &livekit.TrackInfo{Sid: trackID2}, 0)
+	fixture.sut.SendAnalytics()
+	require.Equal(t, 2, fixture.analytics.SendStatsCallCount())
+	_, stats = fixture.analytics.SendStatsArgsForCall(1)
+	require.Equal(t, 2, len(stats)) // still 2 tracks, next call won't contain 1 track
+
+	//now only 1 track stats remaining
+	fixture.sut.SendAnalytics()
+	require.Equal(t, 3, fixture.analytics.SendStatsCallCount())
+	_, stats = fixture.analytics.SendStatsArgsForCall(2)
+	require.Equal(t, 1, len(stats)) // now only 1 track remaining
 }
 
 func Test_AnalyticsSentWhenParticipantLeaves(t *testing.T) {
@@ -66,3 +336,101 @@ func Test_AnalyticsSentWhenParticipantLeaves(t *testing.T) {
 	//test
 	require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
 }
+
+func Test_AddUpTrack(t *testing.T) {
+	fixture := createFixture()
+
+	//prepare
+	room := &livekit.Room{}
+	partSID := "part1"
+	participantInfo := &livekit.ParticipantInfo{Sid: partSID}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil)
+
+	//do
+	var totalBytes uint64 = 3
+	var totalPackets uint32 = 3
+	buf := &buffer.Buffer{}
+	bufferStats := buffer.Stats{
+		PacketCount: totalPackets,
+		TotalByte:   totalBytes,
+	}
+	buf.SetStatsTestOnly(bufferStats)
+	trackID := "trackID"
+	fixture.sut.AddUpTrack(partSID, trackID, buf)
+	fixture.sut.SendAnalytics()
+
+	//test
+	require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
+	_, stats := fixture.analytics.SendStatsArgsForCall(0)
+	require.Equal(t, 1, len(stats))
+	require.Equal(t, livekit.StreamType_UPSTREAM, stats[0].Kind)
+	require.Equal(t, totalBytes, stats[0].TotalBytes)
+	require.Equal(t, totalPackets, uint32(stats[0].TotalPackets))
+	require.Equal(t, trackID, stats[0].TrackId)
+}
+
+func Test_AddUpTrack_SeveralBuffers_Simulcast(t *testing.T) {
+	fixture := createFixture()
+
+	//prepare
+	room := &livekit.Room{}
+	partSID := "part1"
+	participantInfo := &livekit.ParticipantInfo{Sid: partSID}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil)
+	//do
+	trackID := "trackID"
+	//buffer 1
+	buf1 := &buffer.Buffer{}
+	buf1.SetStatsTestOnly(buffer.Stats{
+		PacketCount: 1,
+		TotalByte:   1,
+	})
+	fixture.sut.AddUpTrack(partSID, trackID, buf1)
+	//buffer 2
+	buf2 := &buffer.Buffer{}
+	buf2.SetStatsTestOnly(buffer.Stats{
+		PacketCount: 2,
+		TotalByte:   2,
+	})
+	fixture.sut.AddUpTrack(partSID, trackID, buf2)
+	fixture.sut.SendAnalytics()
+	//test
+	totalBytes := buf1.GetStats().TotalByte + buf2.GetStats().TotalByte
+	totalPackets := buf1.GetStats().PacketCount + buf2.GetStats().PacketCount
+	require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
+	_, stats := fixture.analytics.SendStatsArgsForCall(0)
+	require.Equal(t, 1, len(stats))
+	require.Equal(t, livekit.StreamType_UPSTREAM, stats[0].Kind)
+	require.Equal(t, totalBytes, stats[0].TotalBytes)
+	require.Equal(t, totalPackets, uint32(stats[0].TotalPackets))
+	require.Equal(t, trackID, stats[0].TrackId)
+}
+
+func Test_BothDownstreamAndUpstreamStatsAreSentTogether(t *testing.T) {
+	fixture := createFixture()
+
+	//prepare
+	room := &livekit.Room{}
+	partSID := "part1"
+	participantInfo := &livekit.ParticipantInfo{Sid: partSID}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil)
+
+	//do
+	//upstream bytes
+	buf := &buffer.Buffer{}
+	buf.SetStatsTestOnly(buffer.Stats{
+		PacketCount: 3,
+		TotalByte:   3,
+	})
+	fixture.sut.AddUpTrack(partSID, "trackID", buf)
+	//downstream bytes
+	fixture.sut.OnDownstreamPacket(partSID, "trackID1", 1)
+	fixture.sut.SendAnalytics()
+
+	//test
+	require.Equal(t, 1, fixture.analytics.SendStatsCallCount())
+	_, stats := fixture.analytics.SendStatsArgsForCall(0)
+	require.Equal(t, 2, len(stats))
+	require.Equal(t, livekit.StreamType_UPSTREAM, stats[0].Kind)
+	require.Equal(t, livekit.StreamType_DOWNSTREAM, stats[1].Kind)
+}