diff --git a/go.mod b/go.mod index c3de2a855..5c9436da2 100644 --- a/go.mod +++ b/go.mod @@ -14,7 +14,7 @@ require ( github.com/google/wire v0.5.0 github.com/gorilla/websocket v1.4.2 github.com/hashicorp/golang-lru v0.5.4 - github.com/livekit/protocol v0.11.12-0.20220124192741-b94955852f2a + github.com/livekit/protocol v0.11.12-0.20220127201730-77f0aed23c8a github.com/magefile/mage v1.11.0 github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0 github.com/mitchellh/go-homedir v1.1.0 diff --git a/go.sum b/go.sum index ea2da9e58..8d297d9cc 100644 --- a/go.sum +++ b/go.sum @@ -132,8 +132,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY/mKrcvA= github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts= -github.com/livekit/protocol v0.11.12-0.20220124192741-b94955852f2a h1:iYvpBQ5a7IgVWZfl5A+VggN2H86MAwYOoAwwvq6Qeq4= -github.com/livekit/protocol v0.11.12-0.20220124192741-b94955852f2a/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg= +github.com/livekit/protocol v0.11.12-0.20220127201730-77f0aed23c8a h1:e16IjJzW23GUXypGbSxn6Esa3xWujJo1QyGsz/68f/M= +github.com/livekit/protocol v0.11.12-0.20220127201730-77f0aed23c8a/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg= github.com/magefile/mage v1.11.0 h1:C/55Ywp9BpgVVclD3lRnSYCwXTYxmSppIgLeDYlNuls= github.com/magefile/mage v1.11.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A= github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0= diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index fedefd594..4f08219a3 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -285,7 +285,8 @@ func (r *RoomManager) StartSession(ctx context.Context, roomName livekit.RoomNam // update room store with new numParticipants updateParticipantCount() - r.telemetry.ParticipantJoined(ctx, room.Room, participant.ToProto(), pi.Client) + clientMeta := &livekit.AnalyticsClientMeta{Region: r.currentNode.Region, Node: r.currentNode.Id} + r.telemetry.ParticipantJoined(ctx, room.Room, participant.ToProto(), pi.Client, clientMeta) participant.OnClose(func(p types.LocalParticipant, disallowedSubscriptions map[livekit.TrackID]livekit.ParticipantID) { if err := r.roomStore.DeleteParticipant(ctx, roomName, p.Identity()); err != nil { pLogger.Errorw("could not delete participant", err) diff --git a/pkg/telemetry/telemetryservice.go b/pkg/telemetry/telemetryservice.go index 42c7bce15..78f4519ff 100644 --- a/pkg/telemetry/telemetryservice.go +++ b/pkg/telemetry/telemetryservice.go @@ -17,7 +17,7 @@ type TelemetryService interface { // events RoomStarted(ctx context.Context, room *livekit.Room) RoomEnded(ctx context.Context, room *livekit.Room) - ParticipantJoined(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientInfo *livekit.ClientInfo) + ParticipantJoined(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientInfo *livekit.ClientInfo, clientMeta *livekit.AnalyticsClientMeta) ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo) TrackPublished(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo) TrackUnpublished(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo, ssrc uint32) @@ -81,9 +81,10 @@ func (t *telemetryService) RoomEnded(ctx context.Context, room *livekit.Room) { } } -func (t *telemetryService) ParticipantJoined(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientInfo *livekit.ClientInfo) { +func (t *telemetryService) ParticipantJoined(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, + clientInfo *livekit.ClientInfo, clientMeta *livekit.AnalyticsClientMeta) { t.jobQueue <- func() { - t.internalService.ParticipantJoined(ctx, room, participant, clientInfo) + t.internalService.ParticipantJoined(ctx, room, participant, clientInfo, clientMeta) } } diff --git a/pkg/telemetry/telemetryserviceinternalevents.go b/pkg/telemetry/telemetryserviceinternalevents.go index c60cb93a7..3da6339f0 100644 --- a/pkg/telemetry/telemetryserviceinternalevents.go +++ b/pkg/telemetry/telemetryserviceinternalevents.go @@ -45,7 +45,7 @@ func (t *telemetryServiceInternal) RoomEnded(ctx context.Context, room *livekit. } func (t *telemetryServiceInternal) ParticipantJoined(ctx context.Context, room *livekit.Room, - participant *livekit.ParticipantInfo, clientInfo *livekit.ClientInfo) { + participant *livekit.ParticipantInfo, clientInfo *livekit.ClientInfo, clientMeta *livekit.AnalyticsClientMeta) { t.workers[livekit.ParticipantID(participant.Sid)] = newStatsWorker(ctx, t, livekit.RoomID(room.Sid), livekit.RoomName(room.Name), livekit.ParticipantID(participant.Sid)) prometheus.AddParticipant() @@ -64,6 +64,7 @@ func (t *telemetryServiceInternal) ParticipantJoined(ctx context.Context, room * Participant: participant, Room: room, ClientInfo: clientInfo, + ClientMeta: clientMeta, }) } diff --git a/pkg/telemetry/test/telemetry_service_events_test.go b/pkg/telemetry/test/telemetry_service_events_test.go index 9fcad999c..5cff258dc 100644 --- a/pkg/telemetry/test/telemetry_service_events_test.go +++ b/pkg/telemetry/test/telemetry_service_events_test.go @@ -23,10 +23,16 @@ func Test_OnParticipantJoin_EventIsSent(t *testing.T) { Browser: "chrome", BrowserVersion: "97.0.1", } + clientMeta := &livekit.AnalyticsClientMeta{ + Region: "dark-side", + Node: "moon", + ClientAddr: "127.0.0.1", + ClientConnectTime: 420, + } participantInfo := &livekit.ParticipantInfo{Sid: partSID} // do - fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo) + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, clientMeta) // test require.Equal(t, 1, fixture.analytics.SendEventCallCount()) @@ -44,6 +50,11 @@ func Test_OnParticipantJoin_EventIsSent(t *testing.T) { require.Equal(t, clientInfo.DeviceModel, event.ClientInfo.DeviceModel) require.Equal(t, clientInfo.Browser, event.ClientInfo.Browser) require.Equal(t, clientInfo.BrowserVersion, event.ClientInfo.BrowserVersion) + + require.Equal(t, clientMeta.Region, event.ClientMeta.Region) + require.Equal(t, clientMeta.Node, event.ClientMeta.Node) + require.Equal(t, clientMeta.ClientAddr, event.ClientMeta.ClientAddr) + require.Equal(t, clientMeta.ClientConnectTime, event.ClientMeta.ClientConnectTime) } func Test_OnParticipantLeft_EventIsSent(t *testing.T) { diff --git a/pkg/telemetry/test/telemetry_service_test.go b/pkg/telemetry/test/telemetry_service_test.go index def671263..ee045d9b5 100644 --- a/pkg/telemetry/test/telemetry_service_test.go +++ b/pkg/telemetry/test/telemetry_service_test.go @@ -31,7 +31,7 @@ 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) + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, nil) // do packet := 33 @@ -57,7 +57,7 @@ func Test_OnDownstreamPackets(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) + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, nil) // do packets := []int{33, 23} @@ -90,7 +90,7 @@ func Test_OnDownstreamPackets_SeveralTracks(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) + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, nil) // do packet1 := 33 @@ -133,7 +133,7 @@ func Test_OnDownStreamRTCP(t *testing.T) { room := &livekit.Room{} partSID := livekit.ParticipantID("part1") participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)} - fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil) + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil) // do stat1 := &livekit.AnalyticsStat{NackCount: 1, PliCount: 1, Jitter: 3, PacketLost: 3, TotalBytes: 1, TotalPackets: 1} @@ -165,7 +165,7 @@ func Test_PacketLostDiffShouldBeSentToTelemetry(t *testing.T) { room := &livekit.Room{} partSID := livekit.ParticipantID("part1") participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)} - fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil) + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil) // do trackID := livekit.TrackID("trackID1") @@ -196,7 +196,7 @@ func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) { room := &livekit.Room{} partSID := livekit.ParticipantID("part1") participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)} - fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil) + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil) // do @@ -240,7 +240,7 @@ func Test_OnUpstreamRTCP(t *testing.T) { room := &livekit.Room{} partSID := livekit.ParticipantID("part1") participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)} - fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil) + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil) // do @@ -274,7 +274,7 @@ func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) { room := &livekit.Room{} partSID := livekit.ParticipantID("part1") participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)} - fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil) + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil) // there should be bytes reported so that stats are sent totalBytes := 1 @@ -339,7 +339,7 @@ func Test_AnalyticsSentWhenParticipantLeaves(t *testing.T) { room := &livekit.Room{} partSID := "part1" participantInfo := &livekit.ParticipantInfo{Sid: partSID} - fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil) + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil) // do fixture.sut.ParticipantLeft(context.Background(), room, participantInfo) @@ -355,7 +355,7 @@ func Test_AddUpTrack(t *testing.T) { room := &livekit.Room{} partSID := livekit.ParticipantID("part1") participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)} - fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil) + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil) // do var totalBytes uint64 = 3 @@ -383,7 +383,7 @@ func Test_AddUpTrack_SeveralBuffers_Simulcast(t *testing.T) { room := &livekit.Room{} partSID := livekit.ParticipantID("part1") participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)} - fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil) + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil) // do trackID := livekit.TrackID("trackID") @@ -410,7 +410,7 @@ func Test_BothDownstreamAndUpstreamStatsAreSentTogether(t *testing.T) { room := &livekit.Room{} partSID := livekit.ParticipantID("part1") participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)} - fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil) + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil) // do // upstream bytes