Merge commit '6bd7fac875d9e9009915053d8a590abb372c5679' into feature/livekit-upgrade-1.13.1

livekit-server/pkg/telemetry/analyticsservice.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
+	"github.com/livekit/protocol/observability/roomobs"
 	"github.com/livekit/protocol/rpc"
 	"github.com/livekit/protocol/utils/guid"
 
@@ -34,8 +35,24 @@ type AnalyticsService interface {
 	SendStats(ctx context.Context, stats []*livekit.AnalyticsStat)
 	SendEvent(ctx context.Context, events *livekit.AnalyticsEvent)
 	SendNodeRoomStates(ctx context.Context, nodeRooms *livekit.AnalyticsNodeRooms)
+	RoomProjectReporter(ctx context.Context) roomobs.ProjectReporter
 }
 
+// ----------------------------
+
+var _ AnalyticsService = &NullAnalyticService{}
+
+type NullAnalyticService struct{}
+
+func (n NullAnalyticService) SendStats(_ context.Context, _ []*livekit.AnalyticsStat)             {}
+func (n NullAnalyticService) SendEvent(_ context.Context, _ *livekit.AnalyticsEvent)              {}
+func (n NullAnalyticService) SendNodeRoomStates(_ context.Context, _ *livekit.AnalyticsNodeRooms) {}
+func (n NullAnalyticService) RoomProjectReporter(_ctx context.Context) roomobs.ProjectReporter {
+	return nil
+}
+
+// ----------------------------
+
 type analyticsService struct {
 	analyticsKey   string
 	nodeID         string
@@ -95,3 +112,7 @@ func (a *analyticsService) SendNodeRoomStates(_ context.Context, nodeRooms *live
 		logger.Errorw("failed to send node room states", err)
 	}
 }
+
+func (a *analyticsService) RoomProjectReporter(_ context.Context) roomobs.ProjectReporter {
+	return roomobs.NewNoopProjectReporter()
+}

livekit-server/pkg/telemetry/events.go

@@ -20,15 +20,16 @@ import (
 
 	"google.golang.org/protobuf/types/known/timestamppb"
 
-	"github.com/livekit/livekit-server/pkg/sfu/mime"
 	"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
+	"github.com/livekit/protocol/codecs/mime"
+	"github.com/livekit/protocol/egress"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 	"github.com/livekit/protocol/utils/guid"
 	"github.com/livekit/protocol/webhook"
 )
 
-func (t *telemetryService) NotifyEvent(ctx context.Context, event *livekit.WebhookEvent) {
+func (t *telemetryService) NotifyEvent(ctx context.Context, event *livekit.WebhookEvent, opts ...webhook.NotifyOption) {
 	if t.notifier == nil {
 		return
 	}
@@ -36,7 +37,7 @@ func (t *telemetryService) NotifyEvent(ctx context.Context, event *livekit.Webho
 	event.CreatedAt = time.Now().Unix()
 	event.Id = guid.New("EV_")
 
-	if err := t.notifier.QueueNotify(ctx, event); err != nil {
+	if err := t.notifier.QueueNotify(ctx, event, opts...); err != nil {
 		logger.Warnw("failed to notify webhook", err, "event", event.Event)
 	}
 }
@@ -79,6 +80,7 @@ func (t *telemetryService) ParticipantJoined(
 	clientInfo *livekit.ClientInfo,
 	clientMeta *livekit.AnalyticsClientMeta,
 	shouldSendEvent bool,
+	guard *ReferenceGuard,
 ) {
 	t.enqueue(func() {
 		_, found := t.getOrCreateWorker(
@@ -87,6 +89,7 @@ func (t *telemetryService) ParticipantJoined(
 			livekit.RoomName(room.Name),
 			livekit.ParticipantID(participant.Sid),
 			livekit.ParticipantIdentity(participant.Identity),
+			guard,
 		)
 		if !found {
 			prometheus.IncrementParticipantRtcConnected(1)
@@ -108,10 +111,11 @@ func (t *telemetryService) ParticipantActive(
 	participant *livekit.ParticipantInfo,
 	clientMeta *livekit.AnalyticsClientMeta,
 	isMigration bool,
+	guard *ReferenceGuard,
 ) {
 	t.enqueue(func() {
 		if !isMigration {
-			// consider participant joined only when they became active
+			// a participant is considered "joined" only when they become "active"
 			t.NotifyEvent(ctx, &livekit.WebhookEvent{
 				Event:       webhook.EventParticipantJoined,
 				Room:        room,
@@ -125,12 +129,13 @@ func (t *telemetryService) ParticipantActive(
 			livekit.RoomName(room.Name),
 			livekit.ParticipantID(participant.Sid),
 			livekit.ParticipantIdentity(participant.Identity),
+			guard,
 		)
 		if !found {
-			// need to also account for participant count
 			prometheus.AddParticipant()
 		}
 		worker.SetConnected()
+		prometheus.IncrementParticipantRtcActive(1)
 
 		ev := newParticipantEvent(livekit.AnalyticsEventType_PARTICIPANT_ACTIVE, room, participant)
 		ev.ClientMeta = clientMeta
@@ -161,6 +166,7 @@ func (t *telemetryService) ParticipantResumed(
 			livekit.RoomName(room.Name),
 			livekit.ParticipantID(participant.Sid),
 			livekit.ParticipantIdentity(participant.Identity),
+			nil,
 		)
 		if !found {
 			prometheus.AddParticipant()
@@ -179,37 +185,55 @@ func (t *telemetryService) ParticipantLeft(ctx context.Context,
 	room *livekit.Room,
 	participant *livekit.ParticipantInfo,
 	shouldSendEvent bool,
+	guard *ReferenceGuard,
 ) {
 	t.enqueue(func() {
 		isConnected := false
-		if worker, ok := t.getWorker(livekit.ParticipantID(participant.Sid)); ok {
+		if worker, ok := t.getWorker(livekit.RoomID(room.Sid), livekit.ParticipantID(participant.Sid)); ok {
 			isConnected = worker.IsConnected()
-			if worker.Close() {
+			if worker.Close(guard) {
 				prometheus.SubParticipant()
+			} else {
+				logger.Infow(
+					"stats worker active",
+					"room", room.Name,
+					"roomID", room.Sid,
+					"participant", participant.Identity,
+					"participantID", participant.Sid,
+					"worker", worker,
+				)
 			}
 		}
 
-		if isConnected && shouldSendEvent {
+		if shouldSendEvent {
+			webhookEvent := webhook.EventParticipantLeft
+			analyticsEvent := livekit.AnalyticsEventType_PARTICIPANT_LEFT
+			if !isConnected {
+				webhookEvent = webhook.EventParticipantConnectionAborted
+				analyticsEvent = livekit.AnalyticsEventType_PARTICIPANT_CONNECTION_ABORTED
+			}
 			t.NotifyEvent(ctx, &livekit.WebhookEvent{
-				Event:       webhook.EventParticipantLeft,
+				Event:       webhookEvent,
 				Room:        room,
 				Participant: participant,
 			})
 
-			t.SendEvent(ctx, newParticipantEvent(livekit.AnalyticsEventType_PARTICIPANT_LEFT, room, participant))
+			t.SendEvent(ctx, newParticipantEvent(analyticsEvent, room, participant))
 		}
 	})
 }
 
 func (t *telemetryService) TrackPublishRequested(
 	ctx context.Context,
+	roomID livekit.RoomID,
+	roomName livekit.RoomName,
 	participantID livekit.ParticipantID,
 	identity livekit.ParticipantIdentity,
 	track *livekit.TrackInfo,
 ) {
 	t.enqueue(func() {
-		prometheus.AddPublishAttempt(track.Type.String())
-		room := t.getRoomDetails(participantID)
+		prometheus.RecordTrackPublishAttempt(track.Type.String())
+		room := toMinimalRoomProto(roomID, roomName)
 		ev := newTrackEvent(livekit.AnalyticsEventType_TRACK_PUBLISH_REQUESTED, room, participantID, track)
 		if ev.Participant != nil {
 			ev.Participant.Identity = string(identity)
@@ -220,15 +244,21 @@ func (t *telemetryService) TrackPublishRequested(
 
 func (t *telemetryService) TrackPublished(
 	ctx context.Context,
+	roomID livekit.RoomID,
+	roomName livekit.RoomName,
 	participantID livekit.ParticipantID,
 	identity livekit.ParticipantIdentity,
 	track *livekit.TrackInfo,
+	shouldSendEvent bool,
 ) {
 	t.enqueue(func() {
 		prometheus.AddPublishedTrack(track.Type.String())
-		prometheus.AddPublishSuccess(track.Type.String())
+		prometheus.RecordTrackPublishSuccess(track.Type.String())
+		if !shouldSendEvent {
+			return
+		}
 
-		room := t.getRoomDetails(participantID)
+		room := toMinimalRoomProto(roomID, roomName)
 		participant := &livekit.ParticipantInfo{
 			Sid:      string(participantID),
 			Identity: string(identity),
@@ -246,22 +276,30 @@ func (t *telemetryService) TrackPublished(
 	})
 }
 
-func (t *telemetryService) TrackPublishedUpdate(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo) {
+func (t *telemetryService) TrackPublishedUpdate(
+	ctx context.Context,
+	roomID livekit.RoomID,
+	roomName livekit.RoomName,
+	participantID livekit.ParticipantID,
+	track *livekit.TrackInfo,
+) {
 	t.enqueue(func() {
-		room := t.getRoomDetails(participantID)
+		room := toMinimalRoomProto(roomID, roomName)
 		t.SendEvent(ctx, newTrackEvent(livekit.AnalyticsEventType_TRACK_PUBLISHED_UPDATE, room, participantID, track))
 	})
 }
 
 func (t *telemetryService) TrackMaxSubscribedVideoQuality(
 	ctx context.Context,
+	roomID livekit.RoomID,
+	roomName livekit.RoomName,
 	participantID livekit.ParticipantID,
 	track *livekit.TrackInfo,
 	mime mime.MimeType,
 	maxQuality livekit.VideoQuality,
 ) {
 	t.enqueue(func() {
-		room := t.getRoomDetails(participantID)
+		room := toMinimalRoomProto(roomID, roomName)
 		ev := newTrackEvent(livekit.AnalyticsEventType_TRACK_MAX_SUBSCRIBED_VIDEO_QUALITY, room, participantID, track)
 		ev.MaxSubscribedVideoQuality = maxQuality
 		ev.Mime = mime.String()
@@ -271,13 +309,15 @@ func (t *telemetryService) TrackMaxSubscribedVideoQuality(
 
 func (t *telemetryService) TrackSubscribeRequested(
 	ctx context.Context,
+	roomID livekit.RoomID,
+	roomName livekit.RoomName,
 	participantID livekit.ParticipantID,
 	track *livekit.TrackInfo,
 ) {
 	t.enqueue(func() {
 		prometheus.RecordTrackSubscribeAttempt()
 
-		room := t.getRoomDetails(participantID)
+		room := toMinimalRoomProto(roomID, roomName)
 		ev := newTrackEvent(livekit.AnalyticsEventType_TRACK_SUBSCRIBE_REQUESTED, room, participantID, track)
 		t.SendEvent(ctx, ev)
 	})
@@ -285,6 +325,8 @@ func (t *telemetryService) TrackSubscribeRequested(
 
 func (t *telemetryService) TrackSubscribed(
 	ctx context.Context,
+	roomID livekit.RoomID,
+	roomName livekit.RoomName,
 	participantID livekit.ParticipantID,
 	track *livekit.TrackInfo,
 	publisher *livekit.ParticipantInfo,
@@ -297,7 +339,7 @@ func (t *telemetryService) TrackSubscribed(
 			return
 		}
 
-		room := t.getRoomDetails(participantID)
+		room := toMinimalRoomProto(roomID, roomName)
 		ev := newTrackEvent(livekit.AnalyticsEventType_TRACK_SUBSCRIBED, room, participantID, track)
 		ev.Publisher = publisher
 		t.SendEvent(ctx, ev)
@@ -306,6 +348,8 @@ func (t *telemetryService) TrackSubscribed(
 
 func (t *telemetryService) TrackSubscribeFailed(
 	ctx context.Context,
+	roomID livekit.RoomID,
+	roomName livekit.RoomName,
 	participantID livekit.ParticipantID,
 	trackID livekit.TrackID,
 	err error,
@@ -314,7 +358,7 @@ func (t *telemetryService) TrackSubscribeFailed(
 	t.enqueue(func() {
 		prometheus.RecordTrackSubscribeFailure(err, isUserError)
 
-		room := t.getRoomDetails(participantID)
+		room := toMinimalRoomProto(roomID, roomName)
 		ev := newTrackEvent(livekit.AnalyticsEventType_TRACK_SUBSCRIBE_FAILED, room, participantID, &livekit.TrackInfo{
 			Sid: string(trackID),
 		})
@@ -325,6 +369,8 @@ func (t *telemetryService) TrackSubscribeFailed(
 
 func (t *telemetryService) TrackUnsubscribed(
 	ctx context.Context,
+	roomID livekit.RoomID,
+	roomName livekit.RoomName,
 	participantID livekit.ParticipantID,
 	track *livekit.TrackInfo,
 	shouldSendEvent bool,
@@ -333,7 +379,7 @@ func (t *telemetryService) TrackUnsubscribed(
 		prometheus.RecordTrackUnsubscribed(track.Type.String())
 
 		if shouldSendEvent {
-			room := t.getRoomDetails(participantID)
+			room := toMinimalRoomProto(roomID, roomName)
 			t.SendEvent(ctx, newTrackEvent(livekit.AnalyticsEventType_TRACK_UNSUBSCRIBED, room, participantID, track))
 		}
 	})
@@ -341,6 +387,8 @@ func (t *telemetryService) TrackUnsubscribed(
 
 func (t *telemetryService) TrackUnpublished(
 	ctx context.Context,
+	roomID livekit.RoomID,
+	roomName livekit.RoomName,
 	participantID livekit.ParticipantID,
 	identity livekit.ParticipantIdentity,
 	track *livekit.TrackInfo,
@@ -352,7 +400,7 @@ func (t *telemetryService) TrackUnpublished(
 			return
 		}
 
-		room := t.getRoomDetails(participantID)
+		room := toMinimalRoomProto(roomID, roomName)
 		participant := &livekit.ParticipantInfo{
 			Sid:      string(participantID),
 			Identity: string(identity),
@@ -370,28 +418,34 @@ func (t *telemetryService) TrackUnpublished(
 
 func (t *telemetryService) TrackMuted(
 	ctx context.Context,
+	roomID livekit.RoomID,
+	roomName livekit.RoomName,
 	participantID livekit.ParticipantID,
 	track *livekit.TrackInfo,
 ) {
 	t.enqueue(func() {
-		room := t.getRoomDetails(participantID)
+		room := toMinimalRoomProto(roomID, roomName)
 		t.SendEvent(ctx, newTrackEvent(livekit.AnalyticsEventType_TRACK_MUTED, room, participantID, track))
 	})
 }
 
 func (t *telemetryService) TrackUnmuted(
 	ctx context.Context,
+	roomID livekit.RoomID,
+	roomName livekit.RoomName,
 	participantID livekit.ParticipantID,
 	track *livekit.TrackInfo,
 ) {
 	t.enqueue(func() {
-		room := t.getRoomDetails(participantID)
+		room := toMinimalRoomProto(roomID, roomName)
 		t.SendEvent(ctx, newTrackEvent(livekit.AnalyticsEventType_TRACK_UNMUTED, room, participantID, track))
 	})
 }
 
 func (t *telemetryService) TrackPublishRTPStats(
 	ctx context.Context,
+	roomID livekit.RoomID,
+	roomName livekit.RoomName,
 	participantID livekit.ParticipantID,
 	trackID livekit.TrackID,
 	mimeType mime.MimeType,
@@ -399,7 +453,7 @@ func (t *telemetryService) TrackPublishRTPStats(
 	stats *livekit.RTPStats,
 ) {
 	t.enqueue(func() {
-		room := t.getRoomDetails(participantID)
+		room := toMinimalRoomProto(roomID, roomName)
 		ev := newRoomEvent(livekit.AnalyticsEventType_TRACK_PUBLISH_STATS, room)
 		ev.ParticipantId = string(participantID)
 		ev.TrackId = string(trackID)
@@ -412,13 +466,15 @@ func (t *telemetryService) TrackPublishRTPStats(
 
 func (t *telemetryService) TrackSubscribeRTPStats(
 	ctx context.Context,
+	roomID livekit.RoomID,
+	roomName livekit.RoomName,
 	participantID livekit.ParticipantID,
 	trackID livekit.TrackID,
 	mimeType mime.MimeType,
 	stats *livekit.RTPStats,
 ) {
 	t.enqueue(func() {
-		room := t.getRoomDetails(participantID)
+		room := toMinimalRoomProto(roomID, roomName)
 		ev := newRoomEvent(livekit.AnalyticsEventType_TRACK_SUBSCRIBE_STATS, room)
 		ev.ParticipantId = string(participantID)
 		ev.TrackId = string(trackID)
@@ -428,12 +484,19 @@ func (t *telemetryService) TrackSubscribeRTPStats(
 	})
 }
 
+func (t *telemetryService) NotifyEgressEvent(ctx context.Context, event string, info *livekit.EgressInfo) {
+	opts := egress.GetEgressNotifyOptions(info)
+
+	t.NotifyEvent(ctx, &livekit.WebhookEvent{
+		Event:      event,
+		EgressInfo: info,
+	}, opts...)
+}
+
 func (t *telemetryService) EgressStarted(ctx context.Context, info *livekit.EgressInfo) {
+
 	t.enqueue(func() {
-		t.NotifyEvent(ctx, &livekit.WebhookEvent{
-			Event:      webhook.EventEgressStarted,
-			EgressInfo: info,
-		})
+		t.NotifyEgressEvent(ctx, webhook.EventEgressStarted, info)
 
 		t.SendEvent(ctx, newEgressEvent(livekit.AnalyticsEventType_EGRESS_STARTED, info))
 	})
@@ -441,20 +504,15 @@ func (t *telemetryService) EgressStarted(ctx context.Context, info *livekit.Egre
 
 func (t *telemetryService) EgressUpdated(ctx context.Context, info *livekit.EgressInfo) {
 	t.enqueue(func() {
-		t.NotifyEvent(ctx, &livekit.WebhookEvent{
-			Event:      webhook.EventEgressUpdated,
-			EgressInfo: info,
-		})
+		t.NotifyEgressEvent(ctx, webhook.EventEgressUpdated, info)
+
 		t.SendEvent(ctx, newEgressEvent(livekit.AnalyticsEventType_EGRESS_UPDATED, info))
 	})
 }
 
 func (t *telemetryService) EgressEnded(ctx context.Context, info *livekit.EgressInfo) {
 	t.enqueue(func() {
-		t.NotifyEvent(ctx, &livekit.WebhookEvent{
-			Event:      webhook.EventEgressEnded,
-			EgressInfo: info,
-		})
+		t.NotifyEgressEvent(ctx, webhook.EventEgressEnded, info)
 
 		t.SendEvent(ctx, newEgressEvent(livekit.AnalyticsEventType_EGRESS_ENDED, info))
 	})
@@ -533,19 +591,6 @@ func (t *telemetryService) Webhook(ctx context.Context, webhookInfo *livekit.Web
 	})
 }
 
-// returns a livekit.Room with only name and sid filled out
-// returns nil if room is not found
-func (t *telemetryService) getRoomDetails(participantID livekit.ParticipantID) *livekit.Room {
-	if worker, ok := t.getWorker(participantID); ok {
-		return &livekit.Room{
-			Sid:  string(worker.roomID),
-			Name: string(worker.roomName),
-		}
-	}
-
-	return nil
-}
-
 func newRoomEvent(event livekit.AnalyticsEventType, room *livekit.Room) *livekit.AnalyticsEvent {
 	ev := &livekit.AnalyticsEvent{
 		Type:      event,
@@ -596,3 +641,10 @@ func newIngressEvent(event livekit.AnalyticsEventType, ingress *livekit.IngressI
 		Ingress:   ingress,
 	}
 }
+
+func toMinimalRoomProto(roomID livekit.RoomID, roomName livekit.RoomName) *livekit.Room {
+	return &livekit.Room{
+		Sid:  string(roomID),
+		Name: string(roomName),
+	}
+}

livekit-server/pkg/telemetry/events_test.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/livekit/livekit-server/pkg/telemetry"
 	"github.com/livekit/protocol/livekit"
 )
 
@@ -46,9 +47,10 @@ func Test_OnParticipantJoin_EventIsSent(t *testing.T) {
 		ClientConnectTime: 420,
 	}
 	participantInfo := &livekit.ParticipantInfo{Sid: partSID}
+	guard := &telemetry.ReferenceGuard{}
 
 	// do
-	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, clientMeta, true)
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, clientMeta, true, guard)
 	time.Sleep(time.Millisecond * 500)
 
 	// test
@@ -81,10 +83,11 @@ func Test_OnParticipantLeft_EventIsSent(t *testing.T) {
 	room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
 	partSID := "part1"
 	participantInfo := &livekit.ParticipantInfo{Sid: partSID}
+	guard := &telemetry.ReferenceGuard{}
 
 	// do
-	fixture.sut.ParticipantActive(context.Background(), room, participantInfo, &livekit.AnalyticsClientMeta{}, false)
-	fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true)
+	fixture.sut.ParticipantActive(context.Background(), room, participantInfo, &livekit.AnalyticsClientMeta{}, false, guard)
+	fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true, guard)
 	time.Sleep(time.Millisecond * 500)
 
 	// test
@@ -100,6 +103,8 @@ func Test_OnTrackUpdate_EventIsSent(t *testing.T) {
 	fixture := createFixture()
 
 	// prepare
+	roomID := "room1"
+	roomName := "RoomName"
 	partID := "part1"
 	trackID := "track1"
 	layer := &livekit.VideoLayer{
@@ -119,7 +124,7 @@ func Test_OnTrackUpdate_EventIsSent(t *testing.T) {
 	}
 
 	// do
-	fixture.sut.TrackPublishedUpdate(context.Background(), livekit.ParticipantID(partID), trackInfo)
+	fixture.sut.TrackPublishedUpdate(context.Background(), livekit.RoomID(roomID), livekit.RoomName(roomName), livekit.ParticipantID(partID), trackInfo)
 	time.Sleep(time.Millisecond * 500)
 
 	// test
@@ -158,9 +163,10 @@ func Test_OnParticipantActive_EventIsSent(t *testing.T) {
 		ClientAddr: "127.0.0.1",
 	}
 	participantInfo := &livekit.ParticipantInfo{Sid: partSID}
+	guard := &telemetry.ReferenceGuard{}
 
 	// do
-	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, clientMeta, true)
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, clientMeta, true, guard)
 	time.Sleep(time.Millisecond * 500)
 
 	// test
@@ -173,7 +179,7 @@ func Test_OnParticipantActive_EventIsSent(t *testing.T) {
 		ClientConnectTime: 420,
 	}
 
-	fixture.sut.ParticipantActive(context.Background(), room, participantInfo, clientMetaConnect, false)
+	fixture.sut.ParticipantActive(context.Background(), room, participantInfo, clientMetaConnect, false, guard)
 	time.Sleep(time.Millisecond * 500)
 
 	require.Equal(t, 2, fixture.analytics.SendEventCallCount())
@@ -210,9 +216,10 @@ func Test_OnTrackSubscribed_EventIsSent(t *testing.T) {
 		ClientAddr: "127.0.0.1",
 	}
 	participantInfo := &livekit.ParticipantInfo{Sid: partSID}
+	guard := &telemetry.ReferenceGuard{}
 
 	// do
-	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, clientMeta, true)
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, clientMeta, true, guard)
 	time.Sleep(time.Millisecond * 500)
 
 	// test
@@ -221,7 +228,7 @@ func Test_OnTrackSubscribed_EventIsSent(t *testing.T) {
 	require.Equal(t, room, event.Room)
 
 	// do
-	fixture.sut.TrackSubscribed(context.Background(), livekit.ParticipantID(partSID), trackInfo, publisherInfo, true)
+	fixture.sut.TrackSubscribed(context.Background(), livekit.RoomID(room.Sid), livekit.RoomName(room.Name), livekit.ParticipantID(partSID), trackInfo, publisherInfo, true)
 	time.Sleep(time.Millisecond * 500)
 
 	require.Eventually(t, func() bool {

livekit-server/pkg/telemetry/prometheus/debug.go

@@ -0,0 +1,40 @@
+// 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 prometheus
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/livekit/protocol/livekit"
+)
+
+var (
+	refCounts *prometheus.GaugeVec
+)
+
+func initDebugStats(nodeID string, nodeType livekit.NodeType) {
+	refCounts = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace:   livekitNamespace,
+		Subsystem:   "debug",
+		Name:        "ref_count",
+		ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()},
+	}, []string{"referrer"})
+
+	prometheus.MustRegister(refCounts)
+}
+
+func AddRef(referrer string, n int) {
+	refCounts.WithLabelValues(referrer).Add(float64(n))
+}

livekit-server/pkg/telemetry/prometheus/node.go

@@ -17,35 +17,34 @@ package prometheus
 import (
 	"time"
 
-	"github.com/mackerelio/go-osstat/memory"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/twitchtv/twirp"
 	"go.uber.org/atomic"
 
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/rpc"
 	"github.com/livekit/protocol/utils/hwstats"
+	"github.com/livekit/protocol/webhook"
 )
 
 const (
 	livekitNamespace string = "livekit"
-
-	statsUpdateInterval = time.Second * 10
 )
 
 var (
 	initialized atomic.Bool
 
-	MessageCounter            *prometheus.CounterVec
-	MessageBytes              *prometheus.CounterVec
-	ServiceOperationCounter   *prometheus.CounterVec
-	TwirpRequestStatusCounter *prometheus.CounterVec
+	promMessageCounter            *prometheus.CounterVec
+	promServiceOperationCounter   *prometheus.CounterVec
+	promTwirpRequestStatusCounter *prometheus.CounterVec
+	promTwirpRequestLatency       *prometheus.HistogramVec
 
-	sysPacketsStart              uint32
-	sysDroppedPacketsStart       uint32
-	promSysPacketGauge           *prometheus.GaugeVec
-	promSysDroppedPacketPctGauge prometheus.Gauge
+	sysPacketsStart        uint32
+	sysDroppedPacketsStart uint32
+	promSysPacketGauge     *prometheus.GaugeVec
 
-	cpuStats *hwstats.CPUStats
+	cpuStats    *hwstats.CPUStats
+	memoryStats *hwstats.MemoryStats
 )
 
 func Init(nodeID string, nodeType livekit.NodeType) error {
@@ -53,27 +52,17 @@ func Init(nodeID string, nodeType livekit.NodeType) error {
 		return nil
 	}
 
-	MessageCounter = prometheus.NewCounterVec(
+	promMessageCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace:   livekitNamespace,
 			Subsystem:   "node",
 			Name:        "messages",
 			ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()},
 		},
-		[]string{"type", "status"},
+		[]string{"type", "status", "direction"},
 	)
 
-	MessageBytes = prometheus.NewCounterVec(
-		prometheus.CounterOpts{
-			Namespace:   livekitNamespace,
-			Subsystem:   "node",
-			Name:        "message_bytes",
-			ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()},
-		},
-		[]string{"type", "message_type"},
-	)
-
-	ServiceOperationCounter = prometheus.NewCounterVec(
+	promServiceOperationCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace:   livekitNamespace,
 			Subsystem:   "node",
@@ -83,7 +72,7 @@ func Init(nodeID string, nodeType livekit.NodeType) error {
 		[]string{"type", "status", "error_type"},
 	)
 
-	TwirpRequestStatusCounter = prometheus.NewCounterVec(
+	promTwirpRequestStatusCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace:   livekitNamespace,
 			Subsystem:   "node",
@@ -93,6 +82,17 @@ func Init(nodeID string, nodeType livekit.NodeType) error {
 		[]string{"service", "method", "status", "code"},
 	)
 
+	promTwirpRequestLatency = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace:   livekitNamespace,
+			Subsystem:   "node",
+			Name:        "twirp_request_latency_ms",
+			ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()},
+			Buckets:     []float64{5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000, 30000, 60000},
+		},
+		[]string{"service", "method"},
+	)
+
 	promSysPacketGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace:   livekitNamespace,
@@ -104,30 +104,21 @@ 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(promMessageCounter)
+	prometheus.MustRegister(promServiceOperationCounter)
+	prometheus.MustRegister(promTwirpRequestStatusCounter)
+	prometheus.MustRegister(promTwirpRequestLatency)
 	prometheus.MustRegister(promSysPacketGauge)
-	prometheus.MustRegister(promSysDroppedPacketPctGauge)
 
 	sysPacketsStart, sysDroppedPacketsStart, _ = getTCStats()
 
 	initPacketStats(nodeID, nodeType)
 	initRoomStats(nodeID, nodeType)
 	rpc.InitPSRPCStats(prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()})
+	webhook.InitWebhookStats(prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()})
 	initQualityStats(nodeID, nodeType)
 	initDataPacketStats(nodeID, nodeType)
+	initDebugStats(nodeID, nodeType)
 
 	var err error
 	cpuStats, err = hwstats.NewCPUStats(nil)
@@ -135,145 +126,187 @@ func Init(nodeID string, nodeType livekit.NodeType) error {
 		return err
 	}
 
+	memoryStats, err = hwstats.NewMemoryStats()
+	if err != nil {
+		return err
+	}
+
 	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
-	}
-
-	var cpuLoad float64
-	cpuIdle := cpuStats.GetCPUIdle()
-	if cpuIdle > 0 {
-		cpuLoad = 1 - (cpuIdle / cpuStats.NumCPU())
+		return nil, err
 	}
 
 	// On MacOS, get "\"vm_stat\": executable file not found in $PATH" although it is in /usr/bin
 	// So, do not error out. Use the information if it is available.
-	memTotal := uint64(0)
-	memUsed := uint64(0)
-	memInfo, _ := memory.Get()
-	if memInfo != nil {
-		memTotal = memInfo.Total
-		memUsed = memInfo.Used
-	}
+	memUsed, memTotal, _ := memoryStats.GetMemory()
 
-	// 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(),
+		NumTrackPublishCancels:            trackPublishCancels.Load(),
+		NumTrackSubscribeAttempts:         trackSubscribeAttempts.Load(),
+		NumTrackSubscribeSuccess:          trackSubscribeSuccess.Load(),
+		NumTrackSubscribeCancels:          trackSubscribeCancels.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(),
+		ParticipantSignalFailed:           participantSignalFailed.Load(),
+		ParticipantSignalValidationFailed: participantSignalValidationFailed.Load(),
+		ParticipantRtcInit:                participantRTCInit.Load(),
+		ParticipantRtcConnected:           participantRTCConnected.Load(),
+		ParticipantRtcCanceled:            participantRTCCanceled.Load(),
+		ParticipantRtcActive:              participantRTCActive.Load(),
+		ForwardLatency:                    forwardLatency.Load(),
+		ForwardJitter:                     forwardJitter.Load(),
+		NumCpus:                           uint32(cpuStats.NumCPU()), // this will round down to the nearest integer
+		CpuLoad:                           float32(cpuStats.GetCPULoad()),
+		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, memoryUsed, memoryTotal, 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)
+		memoryUsed += float32(stats.MemoryUsed) * float32(spanElapsed)
+		memoryTotal += float32(stats.MemoryTotal) * 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),
+		ParticipantSignalFailed:           perSec(earlier.ParticipantSignalFailed, later.ParticipantSignalFailed, elapsed),
+		ParticipantSignalValidationFailed: perSec(earlier.ParticipantSignalValidationFailed, later.ParticipantSignalValidationFailed, elapsed),
+		ParticipantRtcInit:                perSec(earlier.ParticipantRtcInit, later.ParticipantRtcInit, elapsed),
+		ParticipantRtcConnected:           perSec(earlier.ParticipantRtcConnected, later.ParticipantRtcConnected, elapsed),
+		ParticipantRtcCanceled:            perSec(earlier.ParticipantRtcCanceled, later.ParticipantRtcCanceled, elapsed),
+		ParticipantRtcActive:              perSec(earlier.ParticipantRtcActive, later.ParticipantRtcActive, 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),
+		TrackPublishCancels:               perSec(uint64(earlier.NumTrackPublishCancels), uint64(later.NumTrackPublishCancels), elapsed),
+		TrackSubscribeAttempts:            perSec(uint64(earlier.NumTrackSubscribeAttempts), uint64(later.NumTrackSubscribeAttempts), elapsed),
+		TrackSubscribeSuccess:             perSec(uint64(earlier.NumTrackSubscribeSuccess), uint64(later.NumTrackSubscribeSuccess), elapsed),
+		TrackSubscribeCancels:             perSec(uint64(earlier.NumTrackSubscribeCancels), uint64(later.NumTrackSubscribeCancels), elapsed),
+		CpuLoad:                           cpuLoad / float32(elapsed),
+		MemoryLoad:                        memoryLoad / float32(elapsed),
+		MemoryUsed:                        memoryUsed / float32(elapsed),
+		MemoryTotal:                       memoryTotal / float32(elapsed),
+	}
+	return rate
 }
 
 func perSec(prev, curr uint64, secs int64) float32 {
 	return float32(curr-prev) / float32(secs)
 }
+
+func RecordSignalRequestSuccess() {
+	promMessageCounter.WithLabelValues("signal", "success", "request").Add(1)
+}
+
+func RecordSignalRequestFailure() {
+	promMessageCounter.WithLabelValues("signal", "failure", "request").Add(1)
+}
+
+func RecordSignalResponseSuccess() {
+	promMessageCounter.WithLabelValues("signal", "success", "response").Add(1)
+}
+
+func RecordSignalResponseFailure() {
+	promMessageCounter.WithLabelValues("signal", "failure", "response").Add(1)
+}
+
+func RecordServiceOperationSuccess(op string) {
+	promServiceOperationCounter.WithLabelValues(op, "success", "").Add(1)
+}
+
+func RecordServiceOperationError(op string, error string) {
+	promServiceOperationCounter.WithLabelValues(op, "error", error).Add(1)
+}
+
+func RecordTwirpRequestStatus(service string, method string, statusFamily string, code twirp.ErrorCode) {
+	promTwirpRequestStatusCounter.WithLabelValues(service, method, statusFamily, string(code)).Add(1)
+}
+
+func RecordTwirpRequestLatency(service, method string, duration time.Duration) {
+	promTwirpRequestLatency.WithLabelValues(service, method).Observe(float64(duration.Milliseconds()))
+}

livekit-server/pkg/telemetry/prometheus/packets.go

@@ -24,31 +24,40 @@ import (
 type Direction string
 
 const (
-	Incoming               Direction = "incoming"
-	Outgoing               Direction = "outgoing"
-	transmissionInitial              = "initial"
-	transmissionRetransmit           = "retransmit"
+	Incoming Direction = "incoming"
+	Outgoing Direction = "outgoing"
+)
+
+type TransmissionType string
+
+const (
+	TransmissionInitial    TransmissionType = "initial"
+	TransmissionRetransmit TransmissionType = "retransmit"
 )
 
 var (
-	bytesIn                    atomic.Uint64
-	bytesOut                   atomic.Uint64
-	packetsIn                  atomic.Uint64
-	packetsOut                 atomic.Uint64
-	nackTotal                  atomic.Uint64
-	retransmitBytes            atomic.Uint64
-	retransmitPackets          atomic.Uint64
-	participantSignalConnected atomic.Uint64
-	participantRTCConnected    atomic.Uint64
-	participantRTCInit         atomic.Uint64
-	forwardLatency             atomic.Uint32
-	forwardJitter              atomic.Uint32
+	bytesIn                           atomic.Uint64
+	bytesOut                          atomic.Uint64
+	packetsIn                         atomic.Uint64
+	packetsOut                        atomic.Uint64
+	nackTotal                         atomic.Uint64
+	retransmitBytes                   atomic.Uint64
+	retransmitPackets                 atomic.Uint64
+	participantSignalConnected        atomic.Uint64
+	participantSignalFailed           atomic.Uint64
+	participantSignalValidationFailed atomic.Uint64
+	participantRTCConnected           atomic.Uint64
+	participantRTCInit                atomic.Uint64
+	participantRTCCanceled            atomic.Uint64
+	participantRTCActive              atomic.Uint64
+	forwardLatency                    atomic.Uint32
+	forwardJitter                     atomic.Uint32
 
-	promPacketLabels          = []string{"direction", "transmission"}
+	promPacketLabels          = []string{"direction", "transmission", "country"}
 	promPacketTotal           *prometheus.CounterVec
 	promPacketBytes           *prometheus.CounterVec
-	promRTCPLabels            = []string{"direction"}
-	promStreamLabels          = []string{"direction", "source", "type"}
+	promRTCPLabels            = []string{"direction", "country"}
+	promStreamLabels          = []string{"direction", "source", "type", "country"}
 	promNackTotal             *prometheus.CounterVec
 	promPliTotal              *prometheus.CounterVec
 	promFirTotal              *prometheus.CounterVec
@@ -62,15 +71,7 @@ var (
 	promConnections           *prometheus.GaugeVec
 	promForwardLatency        prometheus.Gauge
 	promForwardJitter         prometheus.Gauge
-
-	promPacketTotalIncomingInitial    prometheus.Counter
-	promPacketTotalIncomingRetransmit prometheus.Counter
-	promPacketTotalOutgoingInitial    prometheus.Counter
-	promPacketTotalOutgoingRetransmit prometheus.Counter
-	promPacketBytesIncomingInitial    prometheus.Counter
-	promPacketBytesIncomingRetransmit prometheus.Counter
-	promPacketBytesOutgoingInitial    prometheus.Counter
-	promPacketBytesOutgoingRetransmit prometheus.Counter
+	promForwardLatencyHist    prometheus.Histogram
 )
 
 func initPacketStats(nodeID string, nodeType livekit.NodeType) {
@@ -135,7 +136,6 @@ func initPacketStats(nodeID string, nodeType livekit.NodeType) {
 		Subsystem:   "jitter",
 		Name:        "us",
 		ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()},
-
 		// 1ms, 10ms, 30ms, 50ms, 70ms, 100ms, 300ms, 600ms, 1s
 		Buckets: []float64{1000, 10000, 30000, 50000, 70000, 100000, 300000, 600000, 1000000},
 	}, promStreamLabels)
@@ -170,6 +170,25 @@ func initPacketStats(nodeID string, nodeType livekit.NodeType) {
 		Name:        "jitter",
 		ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()},
 	})
+	promForwardLatencyHist = prometheus.NewHistogram(prometheus.HistogramOpts{
+		Namespace:   livekitNamespace,
+		Subsystem:   "forward_latency",
+		Name:        "ns",
+		ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()},
+		// 50us, 100us, 250us, 500us, 1ms, 2ms, 3ms, 5ms, 10ms, 20ms
+		Buckets: []float64{
+			50 * 1000,
+			100 * 1000,
+			250 * 1000,
+			500 * 1000,
+			1 * 1000 * 1000,
+			2 * 1000 * 1000,
+			3 * 1000 * 1000,
+			5 * 1000 * 1000,
+			10 * 1000 * 1000,
+			20 * 1000 * 1000,
+		},
+	})
 
 	prometheus.MustRegister(promPacketTotal)
 	prometheus.MustRegister(promPacketBytes)
@@ -186,31 +205,17 @@ func initPacketStats(nodeID string, nodeType livekit.NodeType) {
 	prometheus.MustRegister(promConnections)
 	prometheus.MustRegister(promForwardLatency)
 	prometheus.MustRegister(promForwardJitter)
-
-	promPacketTotalIncomingInitial = promPacketTotal.WithLabelValues(string(Incoming), transmissionInitial)
-	promPacketTotalIncomingRetransmit = promPacketTotal.WithLabelValues(string(Incoming), transmissionRetransmit)
-	promPacketTotalOutgoingInitial = promPacketTotal.WithLabelValues(string(Outgoing), transmissionInitial)
-	promPacketTotalOutgoingRetransmit = promPacketTotal.WithLabelValues(string(Outgoing), transmissionRetransmit)
-	promPacketBytesIncomingInitial = promPacketBytes.WithLabelValues(string(Incoming), transmissionInitial)
-	promPacketBytesIncomingRetransmit = promPacketBytes.WithLabelValues(string(Incoming), transmissionRetransmit)
-	promPacketBytesOutgoingInitial = promPacketBytes.WithLabelValues(string(Outgoing), transmissionInitial)
-	promPacketBytesOutgoingRetransmit = promPacketBytes.WithLabelValues(string(Outgoing), transmissionRetransmit)
+	prometheus.MustRegister(promForwardLatencyHist)
 }
 
-func IncrementPackets(direction Direction, count uint64, retransmit bool) {
-	if direction == Incoming {
-		if retransmit {
-			promPacketTotalIncomingRetransmit.Add(float64(count))
-		} else {
-			promPacketTotalIncomingInitial.Add(float64(count))
-		}
+func IncrementPackets(country string, direction Direction, count uint64, retransmit bool) {
+	var transmission TransmissionType
+	if retransmit {
+		transmission = TransmissionRetransmit
 	} else {
-		if retransmit {
-			promPacketTotalOutgoingRetransmit.Add(float64(count))
-		} else {
-			promPacketTotalOutgoingInitial.Add(float64(count))
-		}
+		transmission = TransmissionInitial
 	}
+	promPacketTotal.WithLabelValues(string(direction), string(transmission), country).Add(float64(count))
 
 	if direction == Incoming {
 		packetsIn.Add(count)
@@ -222,20 +227,14 @@ func IncrementPackets(direction Direction, count uint64, retransmit bool) {
 	}
 }
 
-func IncrementBytes(direction Direction, count uint64, retransmit bool) {
-	if direction == Incoming {
-		if retransmit {
-			promPacketBytesIncomingRetransmit.Add(float64(count))
-		} else {
-			promPacketBytesIncomingInitial.Add(float64(count))
-		}
+func IncrementBytes(country string, direction Direction, count uint64, retransmit bool) {
+	var transmission TransmissionType
+	if retransmit {
+		transmission = TransmissionRetransmit
 	} else {
-		if retransmit {
-			promPacketBytesOutgoingRetransmit.Add(float64(count))
-		} else {
-			promPacketBytesOutgoingInitial.Add(float64(count))
-		}
+		transmission = TransmissionInitial
 	}
+	promPacketBytes.WithLabelValues(string(direction), string(transmission), country).Add(float64(count))
 
 	if direction == Incoming {
 		bytesIn.Add(count)
@@ -247,46 +246,53 @@ func IncrementBytes(direction Direction, count uint64, retransmit bool) {
 	}
 }
 
-func IncrementRTCP(direction Direction, nack, pli, fir uint32) {
+func IncrementRTCP(country string, direction Direction, nack, pli, fir uint32) {
 	if nack > 0 {
-		promNackTotal.WithLabelValues(string(direction)).Add(float64(nack))
+		promNackTotal.WithLabelValues(string(direction), country).Add(float64(nack))
 		nackTotal.Add(uint64(nack))
 	}
 	if pli > 0 {
-		promPliTotal.WithLabelValues(string(direction)).Add(float64(pli))
+		promPliTotal.WithLabelValues(string(direction), country).Add(float64(pli))
 	}
 	if fir > 0 {
-		promFirTotal.WithLabelValues(string(direction)).Add(float64(fir))
+		promFirTotal.WithLabelValues(string(direction), country).Add(float64(fir))
 	}
 }
 
-func RecordPacketLoss(direction Direction, trackSource livekit.TrackSource, trackType livekit.TrackType, lost, total uint32) {
+func RecordPacketLoss(
+	country string,
+	direction Direction,
+	trackSource livekit.TrackSource,
+	trackType livekit.TrackType,
+	lost uint32,
+	total uint32,
+) {
 	if total > 0 {
-		promPacketLoss.WithLabelValues(string(direction), trackSource.String(), trackType.String()).Observe(float64(lost) / float64(total) * 100)
+		promPacketLoss.WithLabelValues(string(direction), trackSource.String(), trackType.String(), country).Observe(float64(lost) / float64(total) * 100)
 	}
 	if lost > 0 {
-		promPacketLossTotal.WithLabelValues(string(direction), trackSource.String(), trackType.String()).Add(float64(lost))
+		promPacketLossTotal.WithLabelValues(string(direction), trackSource.String(), trackType.String(), country).Add(float64(lost))
 	}
 }
 
-func RecordPacketOutOfOrder(direction Direction, trackSource livekit.TrackSource, trackType livekit.TrackType, ooo, total uint32) {
+func RecordPacketOutOfOrder(country string, direction Direction, trackSource livekit.TrackSource, trackType livekit.TrackType, ooo, total uint32) {
 	if total > 0 {
-		promPacketOutOfOrder.WithLabelValues(string(direction), trackSource.String(), trackType.String()).Observe(float64(ooo) / float64(total) * 100)
+		promPacketOutOfOrder.WithLabelValues(string(direction), trackSource.String(), trackType.String(), country).Observe(float64(ooo) / float64(total) * 100)
 	}
 	if ooo > 0 {
-		promPacketOutOfOrderTotal.WithLabelValues(string(direction), trackSource.String(), trackType.String()).Add(float64(ooo))
+		promPacketOutOfOrderTotal.WithLabelValues(string(direction), trackSource.String(), trackType.String(), country).Add(float64(ooo))
 	}
 }
 
-func RecordJitter(direction Direction, trackSource livekit.TrackSource, trackType livekit.TrackType, jitter uint32) {
+func RecordJitter(country string, direction Direction, trackSource livekit.TrackSource, trackType livekit.TrackType, jitter uint32) {
 	if jitter > 0 {
-		promJitter.WithLabelValues(string(direction), trackSource.String(), trackType.String()).Observe(float64(jitter))
+		promJitter.WithLabelValues(string(direction), trackSource.String(), trackType.String(), country).Observe(float64(jitter))
 	}
 }
 
-func RecordRTT(direction Direction, trackSource livekit.TrackSource, trackType livekit.TrackType, rtt uint32) {
+func RecordRTT(country string, direction Direction, trackSource livekit.TrackSource, trackType livekit.TrackType, rtt uint32) {
 	if rtt > 0 {
-		promRTT.WithLabelValues(string(direction), trackSource.String(), trackType.String()).Observe(float64(rtt))
+		promRTT.WithLabelValues(string(direction), trackSource.String(), trackType.String(), country).Observe(float64(rtt))
 	}
 }
 
@@ -297,9 +303,17 @@ func IncrementParticipantJoin(join uint32) {
 	}
 }
 
-func IncrementParticipantJoinFail(join uint32) {
-	if join > 0 {
-		promParticipantJoin.WithLabelValues("signal_failed").Add(float64(join))
+func IncrementParticipantJoinFail(fail uint32) {
+	if fail > 0 {
+		participantSignalFailed.Add(uint64(fail))
+		promParticipantJoin.WithLabelValues("signal_failed").Add(float64(fail))
+	}
+}
+
+func IncrementParticipantJoinValidationFail(validationFail uint32) {
+	if validationFail > 0 {
+		participantSignalValidationFailed.Add(uint64(validationFail))
+		promParticipantJoin.WithLabelValues("signal_validation_failed").Add(float64(validationFail))
 	}
 }
 
@@ -317,6 +331,20 @@ func IncrementParticipantRtcConnected(join uint32) {
 	}
 }
 
+func IncrementParticipantRtcActive(active uint32) {
+	if active > 0 {
+		participantRTCActive.Add(uint64(active))
+		promParticipantJoin.WithLabelValues("rtc_active").Add(float64(active))
+	}
+}
+
+func IncrementParticipantRtcCanceled(numCancels uint64) {
+	if numCancels > 0 {
+		participantRTCCanceled.Add(numCancels)
+		promParticipantJoin.WithLabelValues("rtc_canceled").Add(float64(numCancels))
+	}
+}
+
 func AddConnection(direction Direction) {
 	promConnections.WithLabelValues(string(direction)).Add(1)
 }
@@ -325,12 +353,16 @@ func SubConnection(direction Direction) {
 	promConnections.WithLabelValues(string(direction)).Sub(1)
 }
 
-func RecordForwardLatency(_, latencyAvg uint32) {
-	forwardLatency.Store(latencyAvg)
-	promForwardLatency.Set(float64(latencyAvg))
+func RecordForwardLatencySample(forwardLatency int64) {
+	promForwardLatencyHist.Observe(float64(forwardLatency))
 }
 
-func RecordForwardJitter(_, jitterAvg uint32) {
-	forwardJitter.Store(jitterAvg)
-	promForwardJitter.Set(float64(jitterAvg))
+func RecordForwardLatency(longTermLatencyAvg uint32) {
+	forwardLatency.Store(longTermLatencyAvg)
+	promForwardLatency.Set(float64(longTermLatencyAvg))
+}
+
+func RecordForwardJitter(longTermJitterAvg uint32) {
+	forwardJitter.Store(longTermJitterAvg)
+	promForwardJitter.Set(float64(longTermJitterAvg))
 }

livekit-server/pkg/telemetry/prometheus/rooms.go

@@ -31,8 +31,10 @@ var (
 	trackSubscribedCurrent atomic.Int32
 	trackPublishAttempts   atomic.Int32
 	trackPublishSuccess    atomic.Int32
+	trackPublishCancels    atomic.Int32
 	trackSubscribeAttempts atomic.Int32
 	trackSubscribeSuccess  atomic.Int32
+	trackSubscribeCancels  atomic.Int32
 	// count the number of failures that are due to user error (permissions, track doesn't exist), so we could compute
 	// success rate by subtracting this from total attempts
 	trackSubscribeUserError atomic.Int32
@@ -162,30 +164,67 @@ func SubPublishedTrack(kind string) {
 	trackPublishedCurrent.Dec()
 }
 
-func AddPublishAttempt(kind string) {
+func RecordTrackPublishAttempt(kind string) {
 	trackPublishAttempts.Inc()
 	promTrackPublishCounter.WithLabelValues(kind, "attempt").Inc()
 }
 
-func AddPublishSuccess(kind string) {
+func RecordTrackPublishSuccess(kind string) {
 	trackPublishSuccess.Inc()
 	promTrackPublishCounter.WithLabelValues(kind, "success").Inc()
 }
 
-func RecordPublishTime(source livekit.TrackSource, trackType livekit.TrackType, d time.Duration, sdk livekit.ClientInfo_SDK, kind livekit.ParticipantInfo_Kind) {
-	recordPubSubTime(true, source, trackType, d, sdk, kind, 1)
+func RecordTrackPublishCancels(kind string, numCancels int32) {
+	trackPublishCancels.Add(numCancels)
+	promTrackPublishCounter.WithLabelValues(kind, "cancel").Add(float64(numCancels))
 }
 
-func RecordSubscribeTime(source livekit.TrackSource, trackType livekit.TrackType, d time.Duration, sdk livekit.ClientInfo_SDK, kind livekit.ParticipantInfo_Kind, count int) {
-	recordPubSubTime(false, source, trackType, d, sdk, kind, count)
+func RecordPublishTime(
+	country string,
+	source livekit.TrackSource,
+	trackType livekit.TrackType,
+	d time.Duration,
+	sdk livekit.ClientInfo_SDK,
+	kind livekit.ParticipantInfo_Kind,
+) {
+	recordPubSubTime(true, country, source, trackType, d, sdk, kind, 1)
 }
 
-func recordPubSubTime(isPublish bool, source livekit.TrackSource, trackType livekit.TrackType, d time.Duration, sdk livekit.ClientInfo_SDK, kind livekit.ParticipantInfo_Kind, count int) {
+func RecordSubscribeTime(
+	country string,
+	source livekit.TrackSource,
+	trackType livekit.TrackType,
+	d time.Duration,
+	sdk livekit.ClientInfo_SDK,
+	kind livekit.ParticipantInfo_Kind,
+	count int,
+) {
+	recordPubSubTime(false, country, source, trackType, d, sdk, kind, count)
+}
+
+func recordPubSubTime(
+	isPublish bool,
+	country string,
+	source livekit.TrackSource,
+	trackType livekit.TrackType,
+	d time.Duration,
+	sdk livekit.ClientInfo_SDK,
+	kind livekit.ParticipantInfo_Kind,
+	count int,
+) {
 	direction := "subscribe"
 	if isPublish {
 		direction = "publish"
 	}
-	promPubSubTime.WithLabelValues(direction, source.String(), trackType.String(), sdk.String(), kind.String(), strconv.Itoa(count)).Observe(float64(d.Milliseconds()))
+	promPubSubTime.WithLabelValues(
+		direction,
+		source.String(),
+		trackType.String(),
+		country,
+		sdk.String(),
+		kind.String(),
+		strconv.Itoa(count),
+	).Observe(float64(d.Milliseconds()))
 }
 
 func RecordTrackSubscribeSuccess(kind string) {
@@ -214,9 +253,15 @@ func RecordTrackSubscribeFailure(err error, isUserError bool) {
 
 	if isUserError {
 		trackSubscribeUserError.Inc()
+		trackSubscribeCancels.Inc()
 	}
 }
 
+func RecordTrackSubscribeCancels(numCancels int32) {
+	trackSubscribeCancels.Add(numCancels)
+	promTrackSubscribeCounter.WithLabelValues("cancel", "").Add(float64(numCancels))
+}
+
 func RecordSessionStartTime(protocolVersion int, d time.Duration) {
 	promSessionStartTime.WithLabelValues(strconv.Itoa(protocolVersion)).Observe(float64(d.Milliseconds()))
 }

livekit-server/pkg/telemetry/signalanddatastats.go

@@ -1,205 +0,0 @@
-// 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 telemetry
-
-import (
-	"context"
-	"fmt"
-	"sync"
-	"time"
-
-	"github.com/frostbyte73/core"
-	"go.uber.org/atomic"
-
-	"github.com/livekit/protocol/livekit"
-	"github.com/livekit/protocol/utils"
-)
-
-type BytesTrackType string
-
-const (
-	BytesTrackTypeData   BytesTrackType = "DT"
-	BytesTrackTypeSignal BytesTrackType = "SG"
-)
-
-// -------------------------------
-
-type TrafficTotals struct {
-	At           time.Time
-	SendBytes    uint64
-	SendMessages uint32
-	RecvBytes    uint64
-	RecvMessages uint32
-}
-
-// --------------------------------
-
-// stats for signal and data channel
-type BytesTrackStats struct {
-	trackID                              livekit.TrackID
-	pID                                  livekit.ParticipantID
-	send, recv                           atomic.Uint64
-	sendMessages, recvMessages           atomic.Uint32
-	totalSendBytes, totalRecvBytes       atomic.Uint64
-	totalSendMessages, totalRecvMessages atomic.Uint32
-	telemetry                            TelemetryService
-	done                                 core.Fuse
-}
-
-func NewBytesTrackStats(trackID livekit.TrackID, pID livekit.ParticipantID, telemetry TelemetryService) *BytesTrackStats {
-	s := &BytesTrackStats{
-		trackID:   trackID,
-		pID:       pID,
-		telemetry: telemetry,
-	}
-	go s.reporter()
-	return s
-}
-
-func (s *BytesTrackStats) AddBytes(bytes uint64, isSend bool) {
-	if isSend {
-		s.send.Add(bytes)
-		s.sendMessages.Inc()
-		s.totalSendBytes.Add(bytes)
-		s.totalSendMessages.Inc()
-	} else {
-		s.recv.Add(bytes)
-		s.recvMessages.Inc()
-		s.totalRecvBytes.Add(bytes)
-		s.totalRecvMessages.Inc()
-	}
-}
-
-func (s *BytesTrackStats) GetTrafficTotals() *TrafficTotals {
-	return &TrafficTotals{
-		At:           time.Now(),
-		SendBytes:    s.totalSendBytes.Load(),
-		SendMessages: s.totalSendMessages.Load(),
-		RecvBytes:    s.totalRecvBytes.Load(),
-		RecvMessages: s.totalRecvMessages.Load(),
-	}
-}
-
-func (s *BytesTrackStats) Stop() {
-	s.done.Break()
-}
-
-func (s *BytesTrackStats) report() {
-	if recv := s.recv.Swap(0); recv > 0 {
-		s.telemetry.TrackStats(StatsKeyForData(livekit.StreamType_UPSTREAM, s.pID, s.trackID), &livekit.AnalyticsStat{
-			Streams: []*livekit.AnalyticsStream{
-				{
-					PrimaryBytes:   recv,
-					PrimaryPackets: s.recvMessages.Swap(0),
-				},
-			},
-		})
-	}
-
-	if send := s.send.Swap(0); send > 0 {
-		s.telemetry.TrackStats(StatsKeyForData(livekit.StreamType_DOWNSTREAM, s.pID, s.trackID), &livekit.AnalyticsStat{
-			Streams: []*livekit.AnalyticsStream{
-				{
-					PrimaryBytes:   send,
-					PrimaryPackets: s.sendMessages.Swap(0),
-				},
-			},
-		})
-	}
-}
-
-func (s *BytesTrackStats) reporter() {
-	ticker := time.NewTicker(telemetryNonMediaStatsUpdateInterval)
-	defer func() {
-		ticker.Stop()
-		s.report()
-	}()
-
-	for {
-		select {
-		case <-s.done.Watch():
-			return
-		case <-ticker.C:
-			s.report()
-		}
-	}
-}
-
-// -----------------------------------------------------------------------
-
-type BytesSignalStats struct {
-	BytesTrackStats
-	ctx context.Context
-
-	mu sync.Mutex
-	ri *livekit.Room
-	pi *livekit.ParticipantInfo
-}
-
-func NewBytesSignalStats(ctx context.Context, telemetry TelemetryService) *BytesSignalStats {
-	return &BytesSignalStats{
-		BytesTrackStats: BytesTrackStats{
-			telemetry: telemetry,
-		},
-		ctx: ctx,
-	}
-}
-
-func (s *BytesSignalStats) ResolveRoom(ri *livekit.Room) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	if s.ri == nil && ri.GetSid() != "" {
-		s.ri = &livekit.Room{
-			Sid:  ri.Sid,
-			Name: ri.Name,
-		}
-		s.maybeStart()
-	}
-}
-
-func (s *BytesSignalStats) ResolveParticipant(pi *livekit.ParticipantInfo) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	if s.pi == nil && pi != nil {
-		s.pi = &livekit.ParticipantInfo{
-			Sid:      pi.Sid,
-			Identity: pi.Identity,
-		}
-		s.maybeStart()
-	}
-}
-
-func (s *BytesSignalStats) maybeStart() {
-	if s.ri == nil || s.pi == nil {
-		return
-	}
-
-	s.pID = livekit.ParticipantID(s.pi.Sid)
-	s.trackID = BytesTrackIDForParticipantID(BytesTrackTypeSignal, s.pID)
-
-	s.telemetry.ParticipantJoined(s.ctx, s.ri, s.pi, nil, nil, false)
-	go s.reporter()
-}
-
-func (s *BytesSignalStats) reporter() {
-	s.BytesTrackStats.reporter()
-	s.telemetry.ParticipantLeft(s.ctx, s.ri, s.pi, false)
-}
-
-// -----------------------------------------------------------------------
-
-func BytesTrackIDForParticipantID(typ BytesTrackType, participantID livekit.ParticipantID) livekit.TrackID {
-	return livekit.TrackID(fmt.Sprintf("%s_%s%s", utils.TrackPrefix, string(typ), participantID))
-}

livekit-server/pkg/telemetry/stats.go

@@ -20,6 +20,7 @@ import (
 )
 
 type StatsKey struct {
+	country       string
 	streamType    livekit.StreamType
 	participantID livekit.ParticipantID
 	trackID       livekit.TrackID
@@ -28,8 +29,16 @@ type StatsKey struct {
 	track         bool
 }
 
-func StatsKeyForTrack(streamType livekit.StreamType, participantID livekit.ParticipantID, trackID livekit.TrackID, trackSource livekit.TrackSource, trackType livekit.TrackType) StatsKey {
+func StatsKeyForTrack(
+	country string,
+	streamType livekit.StreamType,
+	participantID livekit.ParticipantID,
+	trackID livekit.TrackID,
+	trackSource livekit.TrackSource,
+	trackType livekit.TrackType,
+) StatsKey {
 	return StatsKey{
+		country:       country,
 		streamType:    streamType,
 		participantID: participantID,
 		trackID:       trackID,
@@ -39,15 +48,21 @@ func StatsKeyForTrack(streamType livekit.StreamType, participantID livekit.Parti
 	}
 }
 
-func StatsKeyForData(streamType livekit.StreamType, participantID livekit.ParticipantID, trackID livekit.TrackID) StatsKey {
+func StatsKeyForData(
+	country string,
+	streamType livekit.StreamType,
+	participantID livekit.ParticipantID,
+	trackID livekit.TrackID,
+) StatsKey {
 	return StatsKey{
+		country:       country,
 		streamType:    streamType,
 		participantID: participantID,
 		trackID:       trackID,
 	}
 }
 
-func (t *telemetryService) TrackStats(key StatsKey, stat *livekit.AnalyticsStat) {
+func (t *telemetryService) TrackStats(roomID livekit.RoomID, _roomName livekit.RoomName, key StatsKey, stat *livekit.AnalyticsStat) {
 	t.enqueue(func() {
 		direction := prometheus.Incoming
 		if key.streamType == livekit.StreamType_DOWNSTREAM {
@@ -76,23 +91,23 @@ func (t *telemetryService) TrackStats(key StatsKey, stat *livekit.AnalyticsStat)
 				bytes += stream.RetransmitBytes
 			}
 			if key.track {
-				prometheus.RecordPacketLoss(direction, key.trackSource, key.trackType, stream.PacketsLost, stream.PrimaryPackets+stream.PaddingPackets)
-				prometheus.RecordPacketOutOfOrder(direction, key.trackSource, key.trackType, stream.PacketsOutOfOrder, stream.PrimaryPackets+stream.PaddingPackets)
-				prometheus.RecordRTT(direction, key.trackSource, key.trackType, stream.Rtt)
-				prometheus.RecordJitter(direction, key.trackSource, key.trackType, stream.Jitter)
+				prometheus.RecordPacketLoss(key.country, direction, key.trackSource, key.trackType, stream.PacketsLost, stream.PrimaryPackets+stream.PaddingPackets)
+				prometheus.RecordPacketOutOfOrder(key.country, direction, key.trackSource, key.trackType, stream.PacketsOutOfOrder, stream.PrimaryPackets+stream.PaddingPackets)
+				prometheus.RecordRTT(key.country, direction, key.trackSource, key.trackType, stream.Rtt)
+				prometheus.RecordJitter(key.country, direction, key.trackSource, key.trackType, stream.Jitter)
 			}
 		}
-		prometheus.IncrementRTCP(direction, nacks, plis, firs)
-		prometheus.IncrementPackets(direction, uint64(packets), false)
-		prometheus.IncrementBytes(direction, bytes, false)
+		prometheus.IncrementRTCP(key.country, direction, nacks, plis, firs)
+		prometheus.IncrementPackets(key.country, direction, uint64(packets), false)
+		prometheus.IncrementBytes(key.country, direction, bytes, false)
 		if retransmitPackets != 0 {
-			prometheus.IncrementPackets(direction, uint64(retransmitPackets), true)
+			prometheus.IncrementPackets(key.country, direction, uint64(retransmitPackets), true)
 		}
 		if retransmitBytes != 0 {
-			prometheus.IncrementBytes(direction, retransmitBytes, true)
+			prometheus.IncrementBytes(key.country, direction, retransmitBytes, true)
 		}
 
-		if worker, ok := t.getWorker(key.participantID); ok {
+		if worker, ok := t.getWorker(roomID, key.participantID); ok {
 			worker.OnTrackStat(key.trackID, key.streamType, stat)
 		}
 	})

livekit-server/pkg/telemetry/stats_test.go

@@ -52,12 +52,13 @@ func Test_ParticipantAndRoomDataAreSentWithAnalytics(t *testing.T) {
 	partSID := livekit.ParticipantID("part1")
 	clientInfo := &livekit.ClientInfo{Sdk: 2}
 	participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
-	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, nil, true)
+	guard := &telemetry.ReferenceGuard{}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, nil, true, guard)
 
 	// do
 	packet := 33
 	stat := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{PrimaryBytes: uint64(packet)}}}
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_DOWNSTREAM, partSID, ""), stat)
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, ""), stat)
 
 	// flush
 	fixture.flush()
@@ -76,11 +77,12 @@ func Test_OnDownstreamPackets(t *testing.T) {
 	fixture := createFixture()
 
 	// prepare
-	room := &livekit.Room{}
+	room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
 	partSID := livekit.ParticipantID("part1")
 	clientInfo := &livekit.ClientInfo{Sdk: 2}
 	participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
-	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, nil, true)
+	guard := &telemetry.ReferenceGuard{}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, nil, true, guard)
 
 	// do
 	packets := []int{33, 23}
@@ -89,7 +91,7 @@ func Test_OnDownstreamPackets(t *testing.T) {
 	trackID := livekit.TrackID("trackID")
 	for i := range packets {
 		stat := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{PrimaryBytes: uint64(packets[i]), PrimaryPackets: uint32(1)}}}
-		fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_DOWNSTREAM, partSID, trackID), stat)
+		fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID), stat)
 	}
 
 	// flush
@@ -109,22 +111,23 @@ func Test_OnDownstreamPackets_SeveralTracks(t *testing.T) {
 	fixture := createFixture()
 
 	// prepare
-	room := &livekit.Room{}
+	room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
 	partSID := livekit.ParticipantID("part1")
 	clientInfo := &livekit.ClientInfo{Sdk: 2}
 	participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
-	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, nil, true)
+	guard := &telemetry.ReferenceGuard{}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, nil, true, guard)
 
 	// do
 	packet1 := 33
 	trackID1 := livekit.TrackID("trackID1")
 	stat1 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{PrimaryBytes: uint64(packet1), PrimaryPackets: 1}}}
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_DOWNSTREAM, partSID, trackID1), stat1)
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID1), stat1)
 
 	packet2 := 23
 	trackID2 := livekit.TrackID("trackID2")
 	stat2 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{PrimaryBytes: uint64(packet2), PrimaryPackets: 1}}}
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_DOWNSTREAM, partSID, trackID2), stat2)
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID2), stat2)
 
 	// flush
 	fixture.flush()
@@ -155,10 +158,11 @@ func Test_OnDownStreamStat(t *testing.T) {
 	fixture := createFixture()
 
 	// prepare
-	room := &livekit.Room{}
+	room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
 	partSID := livekit.ParticipantID("part1")
 	participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
-	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
+	guard := &telemetry.ReferenceGuard{}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
 
 	// do
 	stat1 := &livekit.AnalyticsStat{
@@ -175,7 +179,7 @@ func Test_OnDownStreamStat(t *testing.T) {
 		},
 	}
 	trackID := livekit.TrackID("trackID1")
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_DOWNSTREAM, partSID, trackID), stat1)
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID), stat1)
 
 	stat2 := &livekit.AnalyticsStat{
 		Streams: []*livekit.AnalyticsStream{
@@ -191,7 +195,7 @@ func Test_OnDownStreamStat(t *testing.T) {
 			},
 		},
 	}
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_DOWNSTREAM, partSID, trackID), stat2)
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID), stat2)
 
 	// flush
 	fixture.flush()
@@ -214,10 +218,11 @@ func Test_PacketLostDiffShouldBeSentToTelemetry(t *testing.T) {
 	fixture := createFixture()
 
 	// prepare
-	room := &livekit.Room{}
+	room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
 	partSID := livekit.ParticipantID("part1")
 	participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
-	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
+	guard := &telemetry.ReferenceGuard{}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
 
 	// do
 	trackID := livekit.TrackID("trackID1")
@@ -230,7 +235,7 @@ func Test_PacketLostDiffShouldBeSentToTelemetry(t *testing.T) {
 			},
 		},
 	}
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_DOWNSTREAM, partSID, trackID), stat1) // there should be bytes reported so that stats are sent
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID), stat1) // there should be bytes reported so that stats are sent
 
 	// flush
 	fixture.flush()
@@ -244,7 +249,7 @@ func Test_PacketLostDiffShouldBeSentToTelemetry(t *testing.T) {
 			},
 		},
 	}
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_DOWNSTREAM, partSID, trackID), stat2)
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID), stat2)
 
 	// flush
 	fixture.flush()
@@ -266,10 +271,11 @@ func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) {
 	fixture := createFixture()
 
 	// prepare
-	room := &livekit.Room{}
+	room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
 	partSID := livekit.ParticipantID("part1")
 	participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
-	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
+	guard := &telemetry.ReferenceGuard{}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
 
 	// do
 	trackID1 := livekit.TrackID("trackID1")
@@ -282,7 +288,7 @@ func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) {
 			},
 		},
 	}
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_DOWNSTREAM, partSID, trackID1), stat1) // there should be bytes reported so that stats are sent
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID1), stat1) // there should be bytes reported so that stats are sent
 
 	stat2 := &livekit.AnalyticsStat{
 		Streams: []*livekit.AnalyticsStream{
@@ -293,7 +299,7 @@ func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) {
 			},
 		},
 	}
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_DOWNSTREAM, partSID, trackID1), stat2)
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID1), stat2)
 
 	stat3 := &livekit.AnalyticsStat{
 		Streams: []*livekit.AnalyticsStream{
@@ -304,7 +310,7 @@ func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) {
 			},
 		},
 	}
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_DOWNSTREAM, partSID, trackID2), stat3)
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID2), stat3)
 
 	// flush
 	fixture.flush()
@@ -335,10 +341,11 @@ func Test_OnUpstreamStat(t *testing.T) {
 	fixture := createFixture()
 
 	// prepare
-	room := &livekit.Room{}
+	room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
 	partSID := livekit.ParticipantID("part1")
 	participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
-	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
+	guard := &telemetry.ReferenceGuard{}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
 
 	// do
 	stat1 := &livekit.AnalyticsStat{
@@ -357,7 +364,7 @@ func Test_OnUpstreamStat(t *testing.T) {
 	}
 	trackID := livekit.TrackID("trackID")
 
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_UPSTREAM, partSID, trackID), stat1)
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_UPSTREAM, partSID, trackID), stat1)
 
 	stat2 := &livekit.AnalyticsStat{
 		Streams: []*livekit.AnalyticsStream{
@@ -373,7 +380,7 @@ func Test_OnUpstreamStat(t *testing.T) {
 			},
 		},
 	}
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_UPSTREAM, partSID, trackID), stat2)
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_UPSTREAM, partSID, trackID), stat2)
 
 	// flush
 	fixture.flush()
@@ -396,11 +403,12 @@ func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) {
 	fixture := createFixture()
 
 	// prepare
-	room := &livekit.Room{}
+	room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
 	partSID := livekit.ParticipantID("part1")
 	identity := livekit.ParticipantIdentity("part1Identity")
 	participantInfo := &livekit.ParticipantInfo{Sid: string(partSID), Identity: string(identity)}
-	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
+	guard := &telemetry.ReferenceGuard{}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
 
 	// there should be bytes reported so that stats are sent
 	totalBytes := 1
@@ -416,8 +424,8 @@ func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) {
 			},
 		},
 	}
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_UPSTREAM, partSID, trackID1), stat1)
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_UPSTREAM, partSID, trackID2), stat1) // using same buffer is not correct but for test it is fine
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_UPSTREAM, partSID, trackID1), stat1)
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_UPSTREAM, partSID, trackID2), stat1) // using same buffer is not correct but for test it is fine
 
 	// do
 	totalBytes++
@@ -431,7 +439,7 @@ func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) {
 			},
 		},
 	}
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_UPSTREAM, partSID, trackID1), stat2)
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_UPSTREAM, partSID, trackID1), stat2)
 
 	stat3 := &livekit.AnalyticsStat{
 		Streams: []*livekit.AnalyticsStream{
@@ -442,7 +450,7 @@ func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) {
 			},
 		},
 	}
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_UPSTREAM, partSID, trackID2), stat3)
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_UPSTREAM, partSID, trackID2), stat3)
 
 	// flush
 	fixture.flush()
@@ -471,7 +479,7 @@ func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) {
 	require.True(t, found2)
 
 	// remove 1 track - track stats were flushed above, so no more calls to SendStats
-	fixture.sut.TrackUnpublished(context.Background(), partSID, identity, &livekit.TrackInfo{Sid: string(trackID2)}, true)
+	fixture.sut.TrackUnpublished(context.Background(), livekit.RoomID(room.Sid), livekit.RoomName(room.Name), partSID, identity, &livekit.TrackInfo{Sid: string(trackID2)}, true)
 
 	// flush
 	fixture.flush()
@@ -486,10 +494,11 @@ func Test_AnalyticsSentWhenParticipantLeaves(t *testing.T) {
 	room := &livekit.Room{}
 	partSID := "part1"
 	participantInfo := &livekit.ParticipantInfo{Sid: partSID}
-	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
+	guard := &telemetry.ReferenceGuard{}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
 
 	// do
-	fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true)
+	fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true, guard)
 
 	// should not be called if there are no track stats
 	time.Sleep(time.Millisecond * 500)
@@ -500,10 +509,11 @@ func Test_AddUpTrack(t *testing.T) {
 	fixture := createFixture()
 
 	// prepare
-	room := &livekit.Room{}
+	room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
 	partSID := livekit.ParticipantID("part1")
 	participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
-	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
+	guard := &telemetry.ReferenceGuard{}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
 
 	// do
 	var totalBytes uint64 = 3
@@ -518,7 +528,7 @@ func Test_AddUpTrack(t *testing.T) {
 		},
 	}
 	trackID := livekit.TrackID("trackID")
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_UPSTREAM, partSID, trackID), stat)
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_UPSTREAM, partSID, trackID), stat)
 
 	// flush
 	fixture.flush()
@@ -537,10 +547,11 @@ func Test_AddUpTrack_SeveralBuffers_Simulcast(t *testing.T) {
 	fixture := createFixture()
 
 	// prepare
-	room := &livekit.Room{}
+	room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
 	partSID := livekit.ParticipantID("part1")
 	participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
-	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
+	guard := &telemetry.ReferenceGuard{}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
 
 	// do
 	trackID := livekit.TrackID("trackID")
@@ -556,7 +567,7 @@ func Test_AddUpTrack_SeveralBuffers_Simulcast(t *testing.T) {
 			},
 		},
 	}
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_UPSTREAM, partSID, trackID), stat1)
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_UPSTREAM, partSID, trackID), stat1)
 
 	// flush
 	fixture.flush()
@@ -576,10 +587,11 @@ func Test_BothDownstreamAndUpstreamStatsAreSentTogether(t *testing.T) {
 	fixture := createFixture()
 
 	// prepare
-	room := &livekit.Room{}
+	room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
 	partSID := livekit.ParticipantID("part1")
 	participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
-	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
+	guard := &telemetry.ReferenceGuard{}
+	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
 
 	// do
 	// upstream bytes
@@ -591,7 +603,7 @@ func Test_BothDownstreamAndUpstreamStatsAreSentTogether(t *testing.T) {
 			},
 		},
 	}
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_UPSTREAM, partSID, "trackID"), stat1)
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_UPSTREAM, partSID, "trackID"), stat1)
 	// downstream bytes
 	stat2 := &livekit.AnalyticsStat{
 		Streams: []*livekit.AnalyticsStream{
@@ -601,7 +613,7 @@ func Test_BothDownstreamAndUpstreamStatsAreSentTogether(t *testing.T) {
 			},
 		},
 	}
-	fixture.sut.TrackStats(telemetry.StatsKeyForData(livekit.StreamType_DOWNSTREAM, partSID, "trackID1"), stat2)
+	fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, "trackID1"), stat2)
 
 	// flush
 	fixture.flush()

livekit-server/pkg/telemetry/statsconn.go

@@ -17,7 +17,7 @@ package telemetry
 import (
 	"net"
 
-	"github.com/pion/turn/v4"
+	"github.com/pion/turn/v5"
 
 	"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
 )
@@ -52,7 +52,8 @@ func NewConn(c net.Conn, direction prometheus.Direction) *Conn {
 func (c *Conn) Read(b []byte) (n int, err error) {
 	n, err = c.Conn.Read(b)
 	if n > 0 {
-		prometheus.IncrementBytes(prometheus.Incoming, uint64(n), false)
+		prometheus.IncrementBytes("", prometheus.Incoming, uint64(n), false)
+		prometheus.IncrementPackets("", prometheus.Incoming, 1, false)
 	}
 	return
 }
@@ -60,7 +61,8 @@ func (c *Conn) Read(b []byte) (n int, err error) {
 func (c *Conn) Write(b []byte) (n int, err error) {
 	n, err = c.Conn.Write(b)
 	if n > 0 {
-		prometheus.IncrementBytes(prometheus.Outgoing, uint64(n), false)
+		prometheus.IncrementBytes("", prometheus.Outgoing, uint64(n), false)
+		prometheus.IncrementPackets("", prometheus.Outgoing, 1, false)
 	}
 	return
 }
@@ -83,8 +85,8 @@ func NewPacketConn(c net.PacketConn, direction prometheus.Direction) *PacketConn
 func (c *PacketConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
 	n, addr, err = c.PacketConn.ReadFrom(p)
 	if n > 0 {
-		prometheus.IncrementBytes(prometheus.Incoming, uint64(n), false)
-		prometheus.IncrementPackets(prometheus.Incoming, 1, false)
+		prometheus.IncrementBytes("", prometheus.Incoming, uint64(n), false)
+		prometheus.IncrementPackets("", prometheus.Incoming, 1, false)
 	}
 	return
 }
@@ -92,8 +94,8 @@ func (c *PacketConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
 func (c *PacketConn) WriteTo(p []byte, addr net.Addr) (n int, err error) {
 	n, err = c.PacketConn.WriteTo(p, addr)
 	if n > 0 {
-		prometheus.IncrementBytes(prometheus.Outgoing, uint64(n), false)
-		prometheus.IncrementPackets(prometheus.Outgoing, 1, false)
+		prometheus.IncrementBytes("", prometheus.Outgoing, uint64(n), false)
+		prometheus.IncrementPackets("", prometheus.Outgoing, 1, false)
 	}
 	return
 }
@@ -111,8 +113,8 @@ func NewRelayAddressGenerator(g turn.RelayAddressGenerator) *RelayAddressGenerat
 	return &RelayAddressGenerator{RelayAddressGenerator: g}
 }
 
-func (g *RelayAddressGenerator) AllocatePacketConn(network string, requestedPort int) (net.PacketConn, net.Addr, error) {
-	conn, addr, err := g.RelayAddressGenerator.AllocatePacketConn(network, requestedPort)
+func (g *RelayAddressGenerator) AllocatePacketConn(c turn.AllocateListenerConfig) (net.PacketConn, net.Addr, error) {
+	conn, addr, err := g.RelayAddressGenerator.AllocatePacketConn(c)
 	if err != nil {
 		return nil, addr, err
 	}
@@ -120,11 +122,20 @@ func (g *RelayAddressGenerator) AllocatePacketConn(network string, requestedPort
 	return NewPacketConn(conn, prometheus.Outgoing), addr, err
 }
 
-func (g *RelayAddressGenerator) AllocateConn(network string, requestedPort int) (net.Conn, net.Addr, error) {
-	conn, addr, err := g.RelayAddressGenerator.AllocateConn(network, requestedPort)
+func (g *RelayAddressGenerator) AllocateConn(c turn.AllocateConnConfig) (net.Conn, error) {
+	conn, err := g.RelayAddressGenerator.AllocateConn(c)
+	if err != nil {
+		return nil, err
+	}
+
+	return NewConn(conn, prometheus.Outgoing), err
+}
+
+func (g *RelayAddressGenerator) AllocateListener(c turn.AllocateListenerConfig) (net.Listener, net.Addr, error) {
+	l, addr, err := g.RelayAddressGenerator.AllocateListener(c)
 	if err != nil {
 		return nil, addr, err
 	}
 
-	return NewConn(conn, prometheus.Outgoing), addr, err
+	return NewListener(l), addr, err
 }

livekit-server/pkg/telemetry/statsworker.go

@@ -19,6 +19,7 @@ import (
 	"sync"
 	"time"
 
+	"go.uber.org/zap/zapcore"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/livekit/livekit-server/pkg/utils"
@@ -27,6 +28,37 @@ import (
 	protoutils "github.com/livekit/protocol/utils"
 )
 
+type ReferenceGuard struct {
+	activated, released bool
+}
+
+type ReferenceCount struct {
+	count int
+}
+
+func (s *ReferenceCount) Activate(guard *ReferenceGuard) {
+	if guard != nil && !guard.activated {
+		guard.activated = true
+		s.count++
+	}
+}
+
+func (s *ReferenceCount) Release(guard *ReferenceGuard) bool {
+	if guard == nil || !guard.activated || guard.released {
+		return false
+	}
+	guard.released = true
+	s.count--
+	return s.count == 0
+}
+
+func (s ReferenceCount) MarshalLogObject(e zapcore.ObjectEncoder) error {
+	e.AddInt("count", s.count)
+	return nil
+}
+
+// ----------------------------------------
+
 // StatsWorker handles participant stats
 type StatsWorker struct {
 	next *StatsWorker
@@ -42,6 +74,7 @@ type StatsWorker struct {
 	lock             sync.RWMutex
 	outgoingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat
 	incomingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat
+	refCount         ReferenceCount
 	closedAt         time.Time
 }
 
@@ -52,6 +85,7 @@ func newStatsWorker(
 	roomName livekit.RoomName,
 	participantID livekit.ParticipantID,
 	identity livekit.ParticipantIdentity,
+	guard *ReferenceGuard,
 ) *StatsWorker {
 	s := &StatsWorker{
 		ctx:                 ctx,
@@ -63,6 +97,7 @@ func newStatsWorker(
 		outgoingPerTrack:    make(map[livekit.TrackID][]*livekit.AnalyticsStat),
 		incomingPerTrack:    make(map[livekit.TrackID][]*livekit.AnalyticsStat),
 	}
+	s.refCount.Activate(guard)
 	return s
 }
 
@@ -93,7 +128,7 @@ func (s *StatsWorker) IsConnected() bool {
 	return s.isConnected
 }
 
-func (s *StatsWorker) Flush(now time.Time) bool {
+func (s *StatsWorker) Flush(now time.Time, closeWait time.Duration) bool {
 	ts := timestamppb.New(now)
 
 	s.lock.Lock()
@@ -105,7 +140,7 @@ func (s *StatsWorker) Flush(now time.Time) bool {
 	outgoingPerTrack := s.outgoingPerTrack
 	s.outgoingPerTrack = make(map[livekit.TrackID][]*livekit.AnalyticsStat)
 
-	closed := !s.closedAt.IsZero() && now.Sub(s.closedAt) > workerCleanupWait
+	closed := !s.closedAt.IsZero() && now.Sub(s.closedAt) > closeWait
 	s.lock.Unlock()
 
 	stats = s.collectStats(ts, livekit.StreamType_UPSTREAM, incomingPerTrack, stats)
@@ -117,10 +152,14 @@ func (s *StatsWorker) Flush(now time.Time) bool {
 	return closed
 }
 
-func (s *StatsWorker) Close() bool {
+func (s *StatsWorker) Close(guard *ReferenceGuard) bool {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
+	if !s.refCount.Release(guard) {
+		return false
+	}
+
 	ok := s.closedAt.IsZero()
 	if ok {
 		s.closedAt = time.Now()
@@ -128,10 +167,14 @@ func (s *StatsWorker) Close() bool {
 	return ok
 }
 
-func (s *StatsWorker) Closed() bool {
+func (s *StatsWorker) Closed(guard *ReferenceGuard) bool {
 	s.lock.Lock()
 	defer s.lock.Unlock()
-	return !s.closedAt.IsZero()
+	if s.closedAt.IsZero() {
+		s.refCount.Activate(guard)
+		return false
+	}
+	return true
 }
 
 func (s *StatsWorker) collectStats(
@@ -157,6 +200,20 @@ func (s *StatsWorker) collectStats(
 	return stats
 }
 
+func (s *StatsWorker) MarshalLogObject(e zapcore.ObjectEncoder) error {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
+	e.AddString("room", string(s.roomName))
+	e.AddString("roomID", string(s.roomID))
+	e.AddString("participant", string(s.participantIdentity))
+	e.AddString("participantID", string(s.participantID))
+	e.AddBool("isConnected", s.isConnected)
+	e.AddTime("closedAt", s.closedAt)
+	e.AddObject("refCount", s.refCount)
+	return nil
+}
+
 // -------------------------------------------------------------------------
 
 // create a single stream and single video layer post aggregation
@@ -253,8 +310,9 @@ func coalesce(stats []*livekit.AnalyticsStat) *livekit.AnalyticsStat {
 
 	stat := &livekit.AnalyticsStat{
 		MinScore:    minScore,
-		MedianScore: utils.MedianFloat32(scores),
+		MedianScore: utils.Median(scores),
 		Streams:     []*livekit.AnalyticsStream{coalescedStream},
+		Mime:        stats[len(stats)-1].Mime, // use the latest Mime
 	}
 	numScores := len(scores)
 	if numScores > 0 {
@@ -263,6 +321,41 @@ func coalesce(stats []*livekit.AnalyticsStat) *livekit.AnalyticsStat {
 	return stat
 }
 
+type CondensedStat struct {
+	StartTime   time.Time
+	EndTime     time.Time
+	Bytes       uint64
+	Packets     uint32
+	PacketsLost uint32
+	Frames      uint32
+}
+
+func CondenseStat(stat *livekit.AnalyticsStat) (ps CondensedStat, ok bool) {
+	if ok = isValid(stat); !ok {
+		return
+	}
+
+	for _, stream := range stat.Streams {
+		startTime := stream.StartTime.AsTime()
+		endTime := stream.EndTime.AsTime()
+		if ps.StartTime.IsZero() || startTime.Before(ps.StartTime) {
+			ps.StartTime = startTime
+		}
+		if endTime.After(ps.EndTime) {
+			ps.EndTime = endTime
+		}
+
+		ps.Bytes += stream.PrimaryBytes
+		ps.Packets += stream.PrimaryPackets
+		ps.PacketsLost += stream.PacketsLost
+		if stream.Frames > ps.Frames {
+			ps.Frames = stream.Frames
+		}
+	}
+
+	return
+}
+
 func isValid(stat *livekit.AnalyticsStat) bool {
 	for _, analyticsStream := range stat.Streams {
 		if int32(analyticsStream.PrimaryPackets) < 0 ||

livekit-server/pkg/telemetry/statsworker_test.go

@@ -0,0 +1,19 @@
+package telemetry
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestStatsWorker(t *testing.T) {
+	t.Run("reference counted close works", func(t *testing.T) {
+		var g0, g1 ReferenceGuard
+		w := newStatsWorker(t.Context(), nil, "", "", "", "", &g0)
+		require.False(t, w.Closed(&g1))
+		require.False(t, w.Close(&g0))
+		require.False(t, w.Closed(&g1))
+		require.True(t, w.Close(&g1))
+		require.True(t, w.Closed(&g1))
+	})
+}

livekit-server/pkg/telemetry/telemetryfakes/fake_analytics_service.go

@@ -7,9 +7,21 @@ import (
 
 	"github.com/livekit/livekit-server/pkg/telemetry"
 	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/observability/roomobs"
 )
 
 type FakeAnalyticsService struct {
+	RoomProjectReporterStub        func(context.Context) roomobs.ProjectReporter
+	roomProjectReporterMutex       sync.RWMutex
+	roomProjectReporterArgsForCall []struct {
+		arg1 context.Context
+	}
+	roomProjectReporterReturns struct {
+		result1 roomobs.ProjectReporter
+	}
+	roomProjectReporterReturnsOnCall map[int]struct {
+		result1 roomobs.ProjectReporter
+	}
 	SendEventStub        func(context.Context, *livekit.AnalyticsEvent)
 	sendEventMutex       sync.RWMutex
 	sendEventArgsForCall []struct {
@@ -32,6 +44,67 @@ type FakeAnalyticsService struct {
 	invocationsMutex sync.RWMutex
 }
 
+func (fake *FakeAnalyticsService) RoomProjectReporter(arg1 context.Context) roomobs.ProjectReporter {
+	fake.roomProjectReporterMutex.Lock()
+	ret, specificReturn := fake.roomProjectReporterReturnsOnCall[len(fake.roomProjectReporterArgsForCall)]
+	fake.roomProjectReporterArgsForCall = append(fake.roomProjectReporterArgsForCall, struct {
+		arg1 context.Context
+	}{arg1})
+	stub := fake.RoomProjectReporterStub
+	fakeReturns := fake.roomProjectReporterReturns
+	fake.recordInvocation("RoomProjectReporter", []interface{}{arg1})
+	fake.roomProjectReporterMutex.Unlock()
+	if stub != nil {
+		return stub(arg1)
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeAnalyticsService) RoomProjectReporterCallCount() int {
+	fake.roomProjectReporterMutex.RLock()
+	defer fake.roomProjectReporterMutex.RUnlock()
+	return len(fake.roomProjectReporterArgsForCall)
+}
+
+func (fake *FakeAnalyticsService) RoomProjectReporterCalls(stub func(context.Context) roomobs.ProjectReporter) {
+	fake.roomProjectReporterMutex.Lock()
+	defer fake.roomProjectReporterMutex.Unlock()
+	fake.RoomProjectReporterStub = stub
+}
+
+func (fake *FakeAnalyticsService) RoomProjectReporterArgsForCall(i int) context.Context {
+	fake.roomProjectReporterMutex.RLock()
+	defer fake.roomProjectReporterMutex.RUnlock()
+	argsForCall := fake.roomProjectReporterArgsForCall[i]
+	return argsForCall.arg1
+}
+
+func (fake *FakeAnalyticsService) RoomProjectReporterReturns(result1 roomobs.ProjectReporter) {
+	fake.roomProjectReporterMutex.Lock()
+	defer fake.roomProjectReporterMutex.Unlock()
+	fake.RoomProjectReporterStub = nil
+	fake.roomProjectReporterReturns = struct {
+		result1 roomobs.ProjectReporter
+	}{result1}
+}
+
+func (fake *FakeAnalyticsService) RoomProjectReporterReturnsOnCall(i int, result1 roomobs.ProjectReporter) {
+	fake.roomProjectReporterMutex.Lock()
+	defer fake.roomProjectReporterMutex.Unlock()
+	fake.RoomProjectReporterStub = nil
+	if fake.roomProjectReporterReturnsOnCall == nil {
+		fake.roomProjectReporterReturnsOnCall = make(map[int]struct {
+			result1 roomobs.ProjectReporter
+		})
+	}
+	fake.roomProjectReporterReturnsOnCall[i] = struct {
+		result1 roomobs.ProjectReporter
+	}{result1}
+}
+
 func (fake *FakeAnalyticsService) SendEvent(arg1 context.Context, arg2 *livekit.AnalyticsEvent) {
 	fake.sendEventMutex.Lock()
 	fake.sendEventArgsForCall = append(fake.sendEventArgsForCall, struct {
@@ -139,12 +212,6 @@ func (fake *FakeAnalyticsService) SendStatsArgsForCall(i int) (context.Context,
 func (fake *FakeAnalyticsService) Invocations() map[string][][]interface{} {
 	fake.invocationsMutex.RLock()
 	defer fake.invocationsMutex.RUnlock()
-	fake.sendEventMutex.RLock()
-	defer fake.sendEventMutex.RUnlock()
-	fake.sendNodeRoomStatesMutex.RLock()
-	defer fake.sendNodeRoomStatesMutex.RUnlock()
-	fake.sendStatsMutex.RLock()
-	defer fake.sendStatsMutex.RUnlock()
 	copiedInvocations := map[string][][]interface{}{}
 	for key, value := range fake.invocations {
 		copiedInvocations[key] = value

livekit-server/pkg/telemetry/telemetryservice.go

@@ -19,8 +19,8 @@ import (
 	"sync"
 	"time"
 
-	"github.com/livekit/livekit-server/pkg/sfu/mime"
 	"github.com/livekit/livekit-server/pkg/utils"
+	"github.com/livekit/protocol/codecs/mime"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 	"github.com/livekit/protocol/webhook"
@@ -31,70 +31,145 @@ import (
 //counterfeiter:generate . TelemetryService
 type TelemetryService interface {
 	// TrackStats is called periodically for each track in both directions (published/subscribed)
-	TrackStats(key StatsKey, stat *livekit.AnalyticsStat)
+	TrackStats(roomID livekit.RoomID, roomName livekit.RoomName, key StatsKey, stat *livekit.AnalyticsStat)
 
 	// events
 	RoomStarted(ctx context.Context, room *livekit.Room)
 	RoomEnded(ctx context.Context, room *livekit.Room)
+
 	// ParticipantJoined - a participant establishes signal connection to a room
-	ParticipantJoined(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientInfo *livekit.ClientInfo, clientMeta *livekit.AnalyticsClientMeta, shouldSendEvent bool)
+	ParticipantJoined(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientInfo *livekit.ClientInfo, clientMeta *livekit.AnalyticsClientMeta, shouldSendEvent bool, guard *ReferenceGuard)
 	// ParticipantActive - a participant establishes media connection
-	ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool)
+	ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool, guard *ReferenceGuard)
 	// ParticipantResumed - there has been an ICE restart or connection resume attempt, and we've received their signal connection
 	ParticipantResumed(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, nodeID livekit.NodeID, reason livekit.ReconnectReason)
 	// ParticipantLeft - the participant leaves the room, only sent if ParticipantActive has been called before
-	ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool)
+	ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool, guard *ReferenceGuard)
 	// TrackPublishRequested - a publication attempt has been received
-	TrackPublishRequested(ctx context.Context, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo)
+	TrackPublishRequested(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo)
 	// TrackPublished - a publication attempt has been successful
-	TrackPublished(ctx context.Context, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo)
+	TrackPublished(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool)
 	// TrackUnpublished - a participant unpublished a track
-	TrackUnpublished(ctx context.Context, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool)
+	TrackUnpublished(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool)
 	// TrackSubscribeRequested - a participant requested to subscribe to a track
-	TrackSubscribeRequested(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo)
+	TrackSubscribeRequested(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, track *livekit.TrackInfo)
 	// TrackSubscribed - a participant subscribed to a track successfully
-	TrackSubscribed(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo, publisher *livekit.ParticipantInfo, shouldSendEvent bool)
+	TrackSubscribed(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, track *livekit.TrackInfo, publisher *livekit.ParticipantInfo, shouldSendEvent bool)
 	// TrackUnsubscribed - a participant unsubscribed from a track successfully
-	TrackUnsubscribed(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo, shouldSendEvent bool)
+	TrackUnsubscribed(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, track *livekit.TrackInfo, shouldSendEvent bool)
 	// TrackSubscribeFailed - failure to subscribe to a track
-	TrackSubscribeFailed(ctx context.Context, participantID livekit.ParticipantID, trackID livekit.TrackID, err error, isUserError bool)
+	TrackSubscribeFailed(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, trackID livekit.TrackID, err error, isUserError bool)
 	// TrackMuted - the publisher has muted the Track
-	TrackMuted(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo)
+	TrackMuted(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, track *livekit.TrackInfo)
 	// TrackUnmuted - the publisher has muted the Track
-	TrackUnmuted(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo)
+	TrackUnmuted(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, track *livekit.TrackInfo)
 	// TrackPublishedUpdate - track metadata has been updated
-	TrackPublishedUpdate(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo)
+	TrackPublishedUpdate(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, track *livekit.TrackInfo)
 	// TrackMaxSubscribedVideoQuality - publisher is notified of the max quality subscribers desire
-	TrackMaxSubscribedVideoQuality(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo, mime mime.MimeType, maxQuality livekit.VideoQuality)
-	TrackPublishRTPStats(ctx context.Context, participantID livekit.ParticipantID, trackID livekit.TrackID, mimeType mime.MimeType, layer int, stats *livekit.RTPStats)
-	TrackSubscribeRTPStats(ctx context.Context, participantID livekit.ParticipantID, trackID livekit.TrackID, mimeType mime.MimeType, stats *livekit.RTPStats)
+	TrackMaxSubscribedVideoQuality(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, track *livekit.TrackInfo, mime mime.MimeType, maxQuality livekit.VideoQuality)
+	TrackPublishRTPStats(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, trackID livekit.TrackID, mimeType mime.MimeType, layer int, stats *livekit.RTPStats)
+	TrackSubscribeRTPStats(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, trackID livekit.TrackID, mimeType mime.MimeType, stats *livekit.RTPStats)
+
 	EgressStarted(ctx context.Context, info *livekit.EgressInfo)
 	EgressUpdated(ctx context.Context, info *livekit.EgressInfo)
 	EgressEnded(ctx context.Context, info *livekit.EgressInfo)
+
 	IngressCreated(ctx context.Context, info *livekit.IngressInfo)
 	IngressDeleted(ctx context.Context, info *livekit.IngressInfo)
 	IngressStarted(ctx context.Context, info *livekit.IngressInfo)
 	IngressUpdated(ctx context.Context, info *livekit.IngressInfo)
 	IngressEnded(ctx context.Context, info *livekit.IngressInfo)
+
 	LocalRoomState(ctx context.Context, info *livekit.AnalyticsNodeRooms)
+
 	Report(ctx context.Context, reportInfo *livekit.ReportInfo)
+
 	APICall(ctx context.Context, apiCallInfo *livekit.APICallInfo)
+
 	Webhook(ctx context.Context, webhookInfo *livekit.WebhookInfo)
 
 	// helpers
 	AnalyticsService
-	NotifyEvent(ctx context.Context, event *livekit.WebhookEvent)
+	NotifyEgressEvent(ctx context.Context, event string, info *livekit.EgressInfo)
 	FlushStats()
 }
 
+// -----------------------------
+
+var _ TelemetryService = (*NullTelemetryService)(nil)
+
+type NullTelemetryService struct {
+	NullAnalyticService
+}
+
+func (n NullTelemetryService) TrackStats(roomID livekit.RoomID, roomName livekit.RoomName, key StatsKey, stat *livekit.AnalyticsStat) {
+}
+func (n NullTelemetryService) RoomStarted(ctx context.Context, room *livekit.Room) {}
+func (n NullTelemetryService) RoomEnded(ctx context.Context, room *livekit.Room)   {}
+func (n NullTelemetryService) ParticipantJoined(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientInfo *livekit.ClientInfo, clientMeta *livekit.AnalyticsClientMeta, shouldSendEvent bool, guard *ReferenceGuard) {
+}
+func (n NullTelemetryService) ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool, guard *ReferenceGuard) {
+}
+func (n NullTelemetryService) ParticipantResumed(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, nodeID livekit.NodeID, reason livekit.ReconnectReason) {
+}
+func (n NullTelemetryService) ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool, guard *ReferenceGuard) {
+}
+func (n NullTelemetryService) TrackPublishRequested(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo) {
+}
+func (n NullTelemetryService) TrackPublished(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) {
+}
+func (n NullTelemetryService) TrackUnpublished(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) {
+}
+func (n NullTelemetryService) TrackSubscribeRequested(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, track *livekit.TrackInfo) {
+}
+func (n NullTelemetryService) TrackSubscribed(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, track *livekit.TrackInfo, publisher *livekit.ParticipantInfo, shouldSendEvent bool) {
+}
+func (n NullTelemetryService) TrackUnsubscribed(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, track *livekit.TrackInfo, shouldSendEvent bool) {
+}
+func (n NullTelemetryService) TrackSubscribeFailed(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, trackID livekit.TrackID, err error, isUserError bool) {
+}
+func (n NullTelemetryService) TrackMuted(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, track *livekit.TrackInfo) {
+}
+func (n NullTelemetryService) TrackUnmuted(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, track *livekit.TrackInfo) {
+}
+func (n NullTelemetryService) TrackPublishedUpdate(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, track *livekit.TrackInfo) {
+}
+func (n NullTelemetryService) TrackMaxSubscribedVideoQuality(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, track *livekit.TrackInfo, mime mime.MimeType, maxQuality livekit.VideoQuality) {
+}
+func (n NullTelemetryService) TrackPublishRTPStats(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, trackID livekit.TrackID, mimeType mime.MimeType, layer int, stats *livekit.RTPStats) {
+}
+func (n NullTelemetryService) TrackSubscribeRTPStats(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, trackID livekit.TrackID, mimeType mime.MimeType, stats *livekit.RTPStats) {
+}
+func (n NullTelemetryService) EgressStarted(ctx context.Context, info *livekit.EgressInfo)          {}
+func (n NullTelemetryService) EgressUpdated(ctx context.Context, info *livekit.EgressInfo)          {}
+func (n NullTelemetryService) EgressEnded(ctx context.Context, info *livekit.EgressInfo)            {}
+func (n NullTelemetryService) IngressCreated(ctx context.Context, info *livekit.IngressInfo)        {}
+func (n NullTelemetryService) IngressDeleted(ctx context.Context, info *livekit.IngressInfo)        {}
+func (n NullTelemetryService) IngressStarted(ctx context.Context, info *livekit.IngressInfo)        {}
+func (n NullTelemetryService) IngressUpdated(ctx context.Context, info *livekit.IngressInfo)        {}
+func (n NullTelemetryService) IngressEnded(ctx context.Context, info *livekit.IngressInfo)          {}
+func (n NullTelemetryService) LocalRoomState(ctx context.Context, info *livekit.AnalyticsNodeRooms) {}
+func (n NullTelemetryService) Report(ctx context.Context, reportInfo *livekit.ReportInfo)           {}
+func (n NullTelemetryService) APICall(ctx context.Context, apiCallInfo *livekit.APICallInfo)        {}
+func (n NullTelemetryService) Webhook(ctx context.Context, webhookInfo *livekit.WebhookInfo)        {}
+func (n NullTelemetryService) NotifyEgressEvent(ctx context.Context, event string, info *livekit.EgressInfo) {
+}
+func (n NullTelemetryService) FlushStats() {}
+
+// -----------------------------
+
 const (
 	workerCleanupWait = 3 * time.Minute
 	jobsQueueMinSize  = 2048
 
-	telemetryStatsUpdateInterval         = time.Second * 30
-	telemetryNonMediaStatsUpdateInterval = time.Second * 30
+	telemetryStatsUpdateInterval = time.Second * 30
 )
 
+type statsWorkerKey struct {
+	roomID        livekit.RoomID
+	participantID livekit.ParticipantID
+}
+
 type telemetryService struct {
 	AnalyticsService
 
@@ -102,7 +177,7 @@ type telemetryService struct {
 	jobsQueue *utils.OpsQueue
 
 	workersMu  sync.RWMutex
-	workers    map[livekit.ParticipantID]*StatsWorker
+	workers    map[statsWorkerKey]*StatsWorker
 	workerList *StatsWorker
 
 	flushMu sync.Mutex
@@ -118,12 +193,7 @@ func NewTelemetryService(notifier webhook.QueuedNotifier, analytics AnalyticsSer
 			FlushOnStop: true,
 			Logger:      logger.GetLogger(),
 		}),
-		workers: make(map[livekit.ParticipantID]*StatsWorker),
-	}
-	if t.notifier != nil {
-		t.notifier.RegisterProcessedHook(func(ctx context.Context, whi *livekit.WebhookInfo) {
-			t.Webhook(ctx, whi)
-		})
+		workers: make(map[statsWorkerKey]*StatsWorker),
 	}
 
 	t.jobsQueue.Start()
@@ -144,7 +214,7 @@ func (t *telemetryService) FlushStats() {
 	var prev, reap *StatsWorker
 	for worker != nil {
 		next := worker.next
-		if closed := worker.Flush(now); closed {
+		if closed := worker.Flush(now, workerCleanupWait); closed {
 			if prev == nil {
 				// this worker was at the head of the list
 				t.workersMu.Lock()
@@ -172,8 +242,9 @@ func (t *telemetryService) FlushStats() {
 	if reap != nil {
 		t.workersMu.Lock()
 		for reap != nil {
-			if reap == t.workers[reap.participantID] {
-				delete(t.workers, reap.participantID)
+			key := statsWorkerKey{reap.roomID, reap.participantID}
+			if reap == t.workers[key] {
+				delete(t.workers, key)
 			}
 			reap = reap.next
 		}
@@ -191,11 +262,11 @@ func (t *telemetryService) enqueue(op func()) {
 	t.jobsQueue.Enqueue(op)
 }
 
-func (t *telemetryService) getWorker(participantID livekit.ParticipantID) (worker *StatsWorker, ok bool) {
+func (t *telemetryService) getWorker(roomID livekit.RoomID, participantID livekit.ParticipantID) (worker *StatsWorker, ok bool) {
 	t.workersMu.RLock()
 	defer t.workersMu.RUnlock()
 
-	worker, ok = t.workers[participantID]
+	worker, ok = t.workers[statsWorkerKey{roomID, participantID}]
 	return
 }
 
@@ -205,12 +276,14 @@ func (t *telemetryService) getOrCreateWorker(
 	roomName livekit.RoomName,
 	participantID livekit.ParticipantID,
 	participantIdentity livekit.ParticipantIdentity,
+	guard *ReferenceGuard,
 ) (*StatsWorker, bool) {
 	t.workersMu.Lock()
 	defer t.workersMu.Unlock()
 
-	worker, ok := t.workers[participantID]
-	if ok && !worker.Closed() {
+	key := statsWorkerKey{roomID, participantID}
+	worker, ok := t.workers[key]
+	if ok && !worker.Closed(guard) {
 		return worker, true
 	}
 
@@ -226,12 +299,13 @@ func (t *telemetryService) getOrCreateWorker(
 		roomName,
 		participantID,
 		participantIdentity,
+		guard,
 	)
 	if existingIsConnected {
 		worker.SetConnected()
 	}
 
-	t.workers[participantID] = worker
+	t.workers[key] = worker
 
 	worker.next = t.workerList
 	t.workerList = worker