diff --git a/pkg/telemetry/events.go b/pkg/telemetry/events.go index 2231b6a2b..fa6257173 100644 --- a/pkg/telemetry/events.go +++ b/pkg/telemetry/events.go @@ -184,6 +184,18 @@ func (t *telemetryService) ParticipantResumed( }) } +// RoomIDChanged re-keys the room's stats workers. +// +// NOTE: this shares the queue with the stats and participant events it races with, so +// ops raised before the id changed (carrying `prevRoomID`) are applied before the +// re-key and ops raised after it (carrying the new id) are applied after. Callers +// should raise this as soon as the room starts reporting the new id. +func (t *telemetryService) RoomIDChanged(ctx context.Context, prevRoomID livekit.RoomID, room *livekit.Room) { + t.enqueue(func() { + t.reKeyRoom(prevRoomID, livekit.RoomID(room.Sid), livekit.RoomName(room.Name)) + }) +} + func (t *telemetryService) ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, diff --git a/pkg/telemetry/stats_test.go b/pkg/telemetry/stats_test.go index 4729d99dd..f4f9d3dbe 100644 --- a/pkg/telemetry/stats_test.go +++ b/pkg/telemetry/stats_test.go @@ -626,6 +626,133 @@ func Test_BothDownstreamAndUpstreamStatsAreSentTogether(t *testing.T) { require.Equal(t, livekit.StreamType_DOWNSTREAM, stats[1].Kind) } +func Test_RoomIDChangeReKeysStatsWorkers(t *testing.T) { + fixture := createFixture() + + // prepare + room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"} + partSID := livekit.ParticipantID("part1") + participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)} + trackID := livekit.TrackID("trackID") + guard := &telemetry.ReferenceGuard{} + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard) + + stat1 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{PrimaryBytes: 33}}} + fixture.sut.TrackStats(livekit.RoomID(room.Sid), livekit.RoomName(room.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID), stat1) + + // do - the room restarts and gets a new id + restartedRoom := &livekit.Room{Sid: "RestartedSid", Name: "RoomName"} + fixture.sut.RoomIDChanged(context.Background(), livekit.RoomID(room.Sid), restartedRoom) + + // stats reported with the new id reach the same worker + stat2 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{PrimaryBytes: 44}}} + fixture.sut.TrackStats(livekit.RoomID(restartedRoom.Sid), livekit.RoomName(restartedRoom.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID), stat2) + + fixture.flush() + + // one worker, one flush, but two stats - each attributed to the session it was collected in + require.Equal(t, 1, fixture.analytics.SendStatsCallCount()) + _, stats := fixture.analytics.SendStatsArgsForCall(0) + require.Equal(t, 2, len(stats)) + + byRoom := map[string]*livekit.AnalyticsStat{} + for _, stat := range stats { + require.Equal(t, string(partSID), stat.ParticipantId) + byRoom[stat.RoomId] = stat + } + require.Len(t, byRoom, 2) + require.Equal(t, uint64(33), byRoom[room.Sid].Streams[0].PrimaryBytes) + require.Equal(t, uint64(44), byRoom[restartedRoom.Sid].Streams[0].PrimaryBytes) + + // the worker moved rather than being duplicated, so closing it out drains everything + fixture.sut.ParticipantLeft(context.Background(), restartedRoom, participantInfo, true, guard) + fixture.flush() + require.Equal(t, 1, fixture.analytics.SendStatsCallCount()) +} + +// a forwarded participant is in more than one room at a time under the same participant +// id, so re-keying one of those rooms must leave the other alone +func Test_RoomIDChangeLeavesForwardedParticipantAlone(t *testing.T) { + fixture := createFixture() + + // prepare - the same participant id in a source room and a forwarding destination room + sourceRoom := &livekit.Room{Sid: "SourceSid", Name: "SourceRoom"} + destRoom := &livekit.Room{Sid: "DestSid", Name: "DestRoom"} + partSID := livekit.ParticipantID("part1") + participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)} + trackID := livekit.TrackID("trackID") + fixture.sut.ParticipantJoined(context.Background(), sourceRoom, participantInfo, nil, nil, true, &telemetry.ReferenceGuard{}) + fixture.sut.ParticipantJoined(context.Background(), destRoom, participantInfo, nil, nil, true, &telemetry.ReferenceGuard{}) + + // do - only the destination room restarts + restartedDest := &livekit.Room{Sid: "RestartedDestSid", Name: "DestRoom"} + fixture.sut.RoomIDChanged(context.Background(), livekit.RoomID(destRoom.Sid), restartedDest) + + stat1 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{PrimaryBytes: 33}}} + fixture.sut.TrackStats(livekit.RoomID(sourceRoom.Sid), livekit.RoomName(sourceRoom.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID), stat1) + stat2 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{PrimaryBytes: 44}}} + fixture.sut.TrackStats(livekit.RoomID(restartedDest.Sid), livekit.RoomName(restartedDest.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID), stat2) + + fixture.flush() + + // the source room's worker is untouched, the destination room's worker moved + byRoom := map[string]*livekit.AnalyticsStat{} + for i := 0; i < fixture.analytics.SendStatsCallCount(); i++ { + _, stats := fixture.analytics.SendStatsArgsForCall(i) + for _, stat := range stats { + byRoom[stat.RoomId] = stat + } + } + require.Len(t, byRoom, 2) + require.Equal(t, uint64(33), byRoom[sourceRoom.Sid].Streams[0].PrimaryBytes) + require.Equal(t, sourceRoom.Name, byRoom[sourceRoom.Sid].RoomName) + require.Equal(t, uint64(44), byRoom[restartedDest.Sid].Streams[0].PrimaryBytes) + require.Equal(t, restartedDest.Name, byRoom[restartedDest.Sid].RoomName) +} + +// a re-key should never land on a room that already has workers, but if it does only +// one worker can be keyed at (room, participant) and the superseded one must not be +// left unreachable in the flush list +func Test_RoomIDChangeParticipantCollision(t *testing.T) { + fixture := createFixture() + + // prepare - the same participant id in the room being re-keyed and in its destination + prevRoom := &livekit.Room{Sid: "PrevSid", Name: "PrevRoom"} + destRoom := &livekit.Room{Sid: "DestSid", Name: "DestRoom"} + partSID := livekit.ParticipantID("part1") + participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)} + trackID := livekit.TrackID("trackID") + fixture.sut.ParticipantJoined(context.Background(), prevRoom, participantInfo, nil, nil, true, &telemetry.ReferenceGuard{}) + fixture.sut.ParticipantJoined(context.Background(), destRoom, participantInfo, nil, nil, true, &telemetry.ReferenceGuard{}) + + stat1 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{PrimaryBytes: 33}}} + fixture.sut.TrackStats(livekit.RoomID(prevRoom.Sid), livekit.RoomName(prevRoom.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID), stat1) + + // do + fixture.sut.RoomIDChanged(context.Background(), livekit.RoomID(prevRoom.Sid), destRoom) + + // the superseded worker drains what it collected under the room it was in + fixture.flush() + byRoom := map[string]*livekit.AnalyticsStat{} + for i := 0; i < fixture.analytics.SendStatsCallCount(); i++ { + _, stats := fixture.analytics.SendStatsArgsForCall(i) + for _, stat := range stats { + byRoom[stat.RoomId] = stat + } + } + require.Equal(t, uint64(33), byRoom[prevRoom.Sid].Streams[0].PrimaryBytes) + + // the worker already keyed at the destination wins and keeps receiving stats + stat2 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{PrimaryBytes: 44}}} + fixture.sut.TrackStats(livekit.RoomID(destRoom.Sid), livekit.RoomName(destRoom.Name), telemetry.StatsKeyForData("test", livekit.StreamType_DOWNSTREAM, partSID, trackID), stat2) + + fixture.flush() + _, stats := fixture.analytics.SendStatsArgsForCall(fixture.analytics.SendStatsCallCount() - 1) + require.Equal(t, 1, len(stats)) + require.Equal(t, destRoom.Sid, stats[0].RoomId) + require.Equal(t, uint64(44), stats[0].Streams[0].PrimaryBytes) +} + func (f *telemetryServiceFixture) flush() { time.Sleep(time.Millisecond * 500) f.sut.FlushStats() diff --git a/pkg/telemetry/statsworker.go b/pkg/telemetry/statsworker.go index 59429a0d9..2271c3e9f 100644 --- a/pkg/telemetry/statsworker.go +++ b/pkg/telemetry/statsworker.go @@ -52,6 +52,18 @@ func (s *ReferenceCount) Release(guard *ReferenceGuard) bool { return s.count == 0 } +// Take hands over every reference held, leaving none behind. +func (s *ReferenceCount) Take() int { + count := s.count + s.count = 0 + return count +} + +// Absorb takes on references handed over from elsewhere. +func (s *ReferenceCount) Absorb(count int) { + s.count += count +} + func (s ReferenceCount) MarshalLogObject(e zapcore.ObjectEncoder) error { e.AddInt("count", s.count) return nil @@ -59,19 +71,39 @@ func (s ReferenceCount) MarshalLogObject(e zapcore.ObjectEncoder) error { // ---------------------------------------- +// statsBatch is stats collected while the worker was in one room +type statsBatch struct { + roomID livekit.RoomID + roomName livekit.RoomName + incomingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat + outgoingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat +} + +func (b statsBatch) isEmpty() bool { + return len(b.incomingPerTrack) == 0 && len(b.outgoingPerTrack) == 0 +} + +// ---------------------------------------- + // StatsWorker handles participant stats type StatsWorker struct { next *StatsWorker ctx context.Context t TelemetryService - roomID livekit.RoomID - roomName livekit.RoomName participantID livekit.ParticipantID participantIdentity livekit.ParticipantIdentity - isConnected bool - lock sync.RWMutex + lock sync.RWMutex + // the room a worker belongs to can change mid-session, so it is mutable state + // guarded by `lock`. it is kept in sync with the key the worker is filed under in + // telemetryService.workers, see telemetryService.reKeyRoom. + roomID livekit.RoomID + roomName livekit.RoomName + // batches sealed off by a room change, they carry the room they were collected + // in and go out on the next flush + sealed []statsBatch + isConnected bool outgoingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat incomingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat refCount ReferenceCount @@ -115,6 +147,51 @@ func (s *StatsWorker) ParticipantID() livekit.ParticipantID { return s.participantID } +func (s *StatsWorker) RoomID() livekit.RoomID { + s.lock.RLock() + defer s.lock.RUnlock() + + return s.roomID +} + +// SetRoom re-points the worker at a room. +// +// Stats collected so far are sealed off rather than re-stamped - a room id changes +// because the previous session ended, so what was collected under it belongs to it. +// Sealing keeps the re-key free of any sending, the sealed stats go out on the next +// flush like every other stat. +func (s *StatsWorker) SetRoom(roomID livekit.RoomID, roomName livekit.RoomName) { + s.lock.Lock() + defer s.lock.Unlock() + + if s.roomID == roomID && s.roomName == roomName { + return + } + + if batch := s.sealStatsLocked(); !batch.isEmpty() { + s.sealed = append(s.sealed, batch) + } + + s.roomID = roomID + s.roomName = roomName +} + +// sealStatsLocked hands over everything collected since the last seal, stamped +// with the room it was collected in +func (s *StatsWorker) sealStatsLocked() statsBatch { + batch := statsBatch{ + roomID: s.roomID, + roomName: s.roomName, + incomingPerTrack: s.incomingPerTrack, + outgoingPerTrack: s.outgoingPerTrack, + } + + s.incomingPerTrack = make(map[livekit.TrackID][]*livekit.AnalyticsStat) + s.outgoingPerTrack = make(map[livekit.TrackID][]*livekit.AnalyticsStat) + + return batch +} + func (s *StatsWorker) SetConnected() { s.lock.Lock() s.isConnected = true @@ -132,19 +209,24 @@ func (s *StatsWorker) Flush(now time.Time, closeWait time.Duration) bool { ts := timestamppb.New(now) s.lock.Lock() - stats := make([]*livekit.AnalyticsStat, 0, len(s.incomingPerTrack)+len(s.outgoingPerTrack)) - - incomingPerTrack := s.incomingPerTrack - s.incomingPerTrack = make(map[livekit.TrackID][]*livekit.AnalyticsStat) - - outgoingPerTrack := s.outgoingPerTrack - s.outgoingPerTrack = make(map[livekit.TrackID][]*livekit.AnalyticsStat) + // anything sealed off by a room change goes out along with the current batch, + // each stamped with the room it was collected in + batches := append(s.sealed, s.sealStatsLocked()) + s.sealed = nil closed := !s.closedAt.IsZero() && now.Sub(s.closedAt) > closeWait s.lock.Unlock() - stats = s.collectStats(ts, livekit.StreamType_UPSTREAM, incomingPerTrack, stats) - stats = s.collectStats(ts, livekit.StreamType_DOWNSTREAM, outgoingPerTrack, stats) + numTracks := 0 + for _, batch := range batches { + numTracks += len(batch.incomingPerTrack) + len(batch.outgoingPerTrack) + } + + stats := make([]*livekit.AnalyticsStat, 0, numTracks) + for _, batch := range batches { + stats = s.collectStats(ts, batch, livekit.StreamType_UPSTREAM, stats) + stats = s.collectStats(ts, batch, livekit.StreamType_DOWNSTREAM, stats) + } if len(stats) > 0 { s.t.SendStats(s.ctx, stats) } @@ -167,6 +249,34 @@ func (s *StatsWorker) Close(guard *ReferenceGuard) bool { return ok } +// ForceClose closes the worker irrespective of outstanding references. Used when a +// worker can no longer be reached through the worker map, so that it drains and is +// reaped instead of lingering in the flush list forever. +// +// Its references are handed over to `successor`, the worker that can be reached in its +// place, so that whoever holds one still has a live worker to close. A ReferenceGuard +// records that it activated some worker, not which one, so leaving them behind would +// strand the successor with references it can never see released. +func (s *StatsWorker) ForceClose(successor *StatsWorker) bool { + s.lock.Lock() + if !s.closedAt.IsZero() { + s.lock.Unlock() + return false + } + + s.closedAt = time.Now() + count := s.refCount.Take() + s.lock.Unlock() + + if successor != nil && count != 0 { + successor.lock.Lock() + successor.refCount.Absorb(count) + successor.lock.Unlock() + } + + return true +} + func (s *StatsWorker) Closed(guard *ReferenceGuard) bool { s.lock.Lock() defer s.lock.Unlock() @@ -179,10 +289,15 @@ func (s *StatsWorker) Closed(guard *ReferenceGuard) bool { func (s *StatsWorker) collectStats( ts *timestamppb.Timestamp, + batch statsBatch, streamType livekit.StreamType, - perTrack map[livekit.TrackID][]*livekit.AnalyticsStat, stats []*livekit.AnalyticsStat, ) []*livekit.AnalyticsStat { + perTrack := batch.incomingPerTrack + if streamType == livekit.StreamType_DOWNSTREAM { + perTrack = batch.outgoingPerTrack + } + for trackID, analyticsStats := range perTrack { coalesced := coalesce(analyticsStats) if coalesced == nil { @@ -192,9 +307,9 @@ func (s *StatsWorker) collectStats( coalesced.TimeStamp = ts coalesced.TrackId = string(trackID) coalesced.Kind = streamType - coalesced.RoomId = string(s.roomID) + coalesced.RoomId = string(batch.roomID) coalesced.ParticipantId = string(s.participantID) - coalesced.RoomName = string(s.roomName) + coalesced.RoomName = string(batch.roomName) stats = append(stats, coalesced) } return stats diff --git a/pkg/telemetry/statsworker_test.go b/pkg/telemetry/statsworker_test.go index f7ae48816..f89992c94 100644 --- a/pkg/telemetry/statsworker_test.go +++ b/pkg/telemetry/statsworker_test.go @@ -16,4 +16,48 @@ func TestStatsWorker(t *testing.T) { require.True(t, w.Close(&g1)) require.True(t, w.Closed(&g1)) }) + + // a ReferenceGuard records that it activated some worker, not which one, so a + // superseded worker has to hand its references to the one reachable in its place + t.Run("force close hands references to the successor", func(t *testing.T) { + t.Run("a guard shared by both workers", func(t *testing.T) { + // the second worker never got a reference, the guard was already activated + var g ReferenceGuard + superseded := newStatsWorker(t.Context(), nil, "", "", "", "", &g) + survivor := newStatsWorker(t.Context(), nil, "", "", "", "", &g) + require.Equal(t, 1, superseded.refCount.count) + require.Equal(t, 0, survivor.refCount.count) + + require.True(t, superseded.ForceClose(survivor)) + require.Equal(t, 0, superseded.refCount.count) + require.Equal(t, 1, survivor.refCount.count) + + // without the hand over this would leave the survivor at -1 and never closed + require.True(t, survivor.Close(&g)) + require.True(t, survivor.Closed(&g)) + }) + + t.Run("a guard per worker", func(t *testing.T) { + var gSuperseded, gSurvivor ReferenceGuard + superseded := newStatsWorker(t.Context(), nil, "", "", "", "", &gSuperseded) + survivor := newStatsWorker(t.Context(), nil, "", "", "", "", &gSurvivor) + + require.True(t, superseded.ForceClose(survivor)) + require.Equal(t, 2, survivor.refCount.count) + + // the superseded worker's owner departs, it must not close the survivor early + require.False(t, survivor.Close(&gSuperseded)) + require.True(t, survivor.Close(&gSurvivor)) + }) + + t.Run("closing an already closed worker holds on to its references", func(t *testing.T) { + var g ReferenceGuard + superseded := newStatsWorker(t.Context(), nil, "", "", "", "", &g) + survivor := newStatsWorker(t.Context(), nil, "", "", "", "", nil) + + require.True(t, superseded.ForceClose(nil)) + require.False(t, superseded.ForceClose(survivor)) + require.Equal(t, 0, survivor.refCount.count) + }) + }) } diff --git a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go index aed5e7cf3..12d542127 100644 --- a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go +++ b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go @@ -136,6 +136,13 @@ type FakeTelemetryService struct { arg2 *livekit.Room arg3 livekit.RoomEndReason } + RoomIDChangedStub func(context.Context, livekit.RoomID, *livekit.Room) + roomIDChangedMutex sync.RWMutex + roomIDChangedArgsForCall []struct { + arg1 context.Context + arg2 livekit.RoomID + arg3 *livekit.Room + } RoomProjectReporterStub func(context.Context) roomobs.ProjectReporter roomProjectReporterMutex sync.RWMutex roomProjectReporterArgsForCall []struct { @@ -915,6 +922,40 @@ func (fake *FakeTelemetryService) RoomEndedArgsForCall(i int) (context.Context, return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 } +func (fake *FakeTelemetryService) RoomIDChanged(arg1 context.Context, arg2 livekit.RoomID, arg3 *livekit.Room) { + fake.roomIDChangedMutex.Lock() + fake.roomIDChangedArgsForCall = append(fake.roomIDChangedArgsForCall, struct { + arg1 context.Context + arg2 livekit.RoomID + arg3 *livekit.Room + }{arg1, arg2, arg3}) + stub := fake.RoomIDChangedStub + fake.recordInvocation("RoomIDChanged", []interface{}{arg1, arg2, arg3}) + fake.roomIDChangedMutex.Unlock() + if stub != nil { + fake.RoomIDChangedStub(arg1, arg2, arg3) + } +} + +func (fake *FakeTelemetryService) RoomIDChangedCallCount() int { + fake.roomIDChangedMutex.RLock() + defer fake.roomIDChangedMutex.RUnlock() + return len(fake.roomIDChangedArgsForCall) +} + +func (fake *FakeTelemetryService) RoomIDChangedCalls(stub func(context.Context, livekit.RoomID, *livekit.Room)) { + fake.roomIDChangedMutex.Lock() + defer fake.roomIDChangedMutex.Unlock() + fake.RoomIDChangedStub = stub +} + +func (fake *FakeTelemetryService) RoomIDChangedArgsForCall(i int) (context.Context, livekit.RoomID, *livekit.Room) { + fake.roomIDChangedMutex.RLock() + defer fake.roomIDChangedMutex.RUnlock() + argsForCall := fake.roomIDChangedArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 +} + func (fake *FakeTelemetryService) RoomProjectReporter(arg1 context.Context) roomobs.ProjectReporter { fake.roomProjectReporterMutex.Lock() ret, specificReturn := fake.roomProjectReporterReturnsOnCall[len(fake.roomProjectReporterArgsForCall)] diff --git a/pkg/telemetry/telemetryservice.go b/pkg/telemetry/telemetryservice.go index b3fd92b8d..c8e31efd0 100644 --- a/pkg/telemetry/telemetryservice.go +++ b/pkg/telemetry/telemetryservice.go @@ -19,6 +19,7 @@ import ( "sync" "time" + "github.com/livekit/livekit-server/pkg/telemetry/prometheus" "github.com/livekit/livekit-server/pkg/utils" "github.com/livekit/protocol/codecs/mime" "github.com/livekit/protocol/livekit" @@ -45,6 +46,9 @@ type TelemetryService interface { 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, guard *ReferenceGuard) + // RoomIDChanged - the room kept its session, but got a different id (a provisional room id + // replaced by the resolved one), re-keys the stats workers of every participant in the room + RoomIDChanged(ctx context.Context, prevRoomID livekit.RoomID, room *livekit.Room) // TrackPublishRequested - a publication attempt has been received TrackPublishRequested(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) // TrackPublished - a publication attempt has been successful @@ -115,6 +119,8 @@ func (n NullTelemetryService) ParticipantResumed(ctx context.Context, room *live } func (n NullTelemetryService) ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool, guard *ReferenceGuard) { } +func (n NullTelemetryService) RoomIDChanged(ctx context.Context, prevRoomID livekit.RoomID, room *livekit.Room) { +} func (n NullTelemetryService) TrackPublishRequested(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) { } func (n NullTelemetryService) TrackPublished(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) { @@ -166,11 +172,6 @@ const ( telemetryStatsUpdateInterval = time.Second * 30 ) -type statsWorkerKey struct { - roomID livekit.RoomID - participantID livekit.ParticipantID -} - type telemetryService struct { AnalyticsService @@ -178,7 +179,7 @@ type telemetryService struct { jobsQueue *utils.OpsQueue workersMu sync.RWMutex - workers map[statsWorkerKey]*StatsWorker + workers map[livekit.RoomID]map[livekit.ParticipantID]*StatsWorker workerList *StatsWorker flushMu sync.Mutex @@ -194,7 +195,7 @@ func NewTelemetryService(notifier webhook.QueuedNotifier, analytics AnalyticsSer FlushOnStop: true, Logger: logger.GetLogger(), }), - workers: make(map[statsWorkerKey]*StatsWorker), + workers: make(map[livekit.RoomID]map[livekit.ParticipantID]*StatsWorker), } t.jobsQueue.Start() @@ -243,9 +244,12 @@ func (t *telemetryService) FlushStats() { if reap != nil { t.workersMu.Lock() for reap != nil { - key := statsWorkerKey{reap.roomID, reap.participantID} - if reap == t.workers[key] { - delete(t.workers, key) + roomID := reap.RoomID() + if roomWorkers := t.workers[roomID]; reap == roomWorkers[reap.participantID] { + delete(roomWorkers, reap.participantID) + if len(roomWorkers) == 0 { + delete(t.workers, roomID) + } } reap = reap.next } @@ -267,7 +271,7 @@ func (t *telemetryService) getWorker(roomID livekit.RoomID, participantID liveki t.workersMu.RLock() defer t.workersMu.RUnlock() - worker, ok = t.workers[statsWorkerKey{roomID, participantID}] + worker, ok = t.workers[roomID][participantID] return } @@ -282,8 +286,8 @@ func (t *telemetryService) getOrCreateWorker( t.workersMu.Lock() defer t.workersMu.Unlock() - key := statsWorkerKey{roomID, participantID} - worker, ok := t.workers[key] + roomWorkers := t.workers[roomID] + worker, ok := roomWorkers[participantID] if ok && !worker.Closed(guard) { return worker, true } @@ -306,7 +310,11 @@ func (t *telemetryService) getOrCreateWorker( worker.SetConnected() } - t.workers[key] = worker + if roomWorkers == nil { + roomWorkers = make(map[livekit.ParticipantID]*StatsWorker) + t.workers[roomID] = roomWorkers + } + roomWorkers[participantID] = worker worker.next = t.workerList t.workerList = worker @@ -314,6 +322,70 @@ func (t *telemetryService) getOrCreateWorker( return worker, false } +// reKeyRoom files every one of a room's stats workers under `roomID` instead of +// `prevRoomID`. +// +// A room can be restarted while participants are connected and reporting stats, which +// gives it a new id. As every worker of the room moves at once, the move is a single map +// splice - the workers themselves are untouched and keep their place in the flush list. +// Each worker then seals off what it collected under `prevRoomID` so those stats stay +// attributed to the session that ended (see StatsWorker.SetRoom). +func (t *telemetryService) reKeyRoom(prevRoomID livekit.RoomID, roomID livekit.RoomID, roomName livekit.RoomName) { + if prevRoomID == roomID { + return + } + + t.workersMu.Lock() + defer t.workersMu.Unlock() + + roomWorkers := t.workers[prevRoomID] + if len(roomWorkers) == 0 { + delete(t.workers, prevRoomID) + return + } + delete(t.workers, prevRoomID) + + existing := t.workers[roomID] + if existing == nil { + t.workers[roomID] = roomWorkers + } else { + // should not happen as a room id is only ever replaced by a freshly minted one + logger.Warnw( + "telemetry re-keying room into an existing entry", nil, + "prevRoomID", prevRoomID, + "room", roomName, + "roomID", roomID, + "numWorkers", len(roomWorkers), + "numExistingWorkers", len(existing), + ) + } + + for participantID, worker := range roomWorkers { + if existing != nil { + if survivor, ok := existing[participantID]; ok { + // only one worker can be keyed at (room, participant) and the one already + // filed there wins, close the superseded one so that it drains and is + // reaped instead of lingering in the flush list unreachable + if worker.ForceClose(survivor) { + prometheus.SubParticipant() + } + continue + } + existing[participantID] = worker + } + + worker.SetRoom(roomID, roomName) + } + + logger.Infow( + "telemetry re-keyed room", + "prevRoomID", prevRoomID, + "room", roomName, + "roomID", roomID, + "numWorkers", len(roomWorkers), + ) +} + func (t *telemetryService) LocalRoomState(ctx context.Context, info *livekit.AnalyticsNodeRooms) { t.enqueue(func() { t.SendNodeRoomStates(ctx, info)