dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-05 15:09:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Store client meta on participant join (#380)

Browse Source 
* Store client meta on participant join

capture region, time_to_connect, ip, node

Signed-off-by: shishir gowda <shishir@livekit.io>

* Update proto dep

Signed-off-by: shishir gowda <shishir@livekit.io>
This commit is contained in:
shishirng
2022-01-27 15:44:03 -05:00
committed by GitHub
parent 26eea78b54
commit 1e156025b4
7 changed files with 35 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
pkg/service/roommanager.go
+2 -1
View File
@@ -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)
pkg/telemetry/telemetryservice.go
+4 -3
View File
@@ -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)
}
}
pkg/telemetry/telemetryserviceinternalevents.go
+2 -1
View File
@@ -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,
})
}
pkg/telemetry/test/telemetry_service_events_test.go
+12 -1
View File
@@ -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) {
pkg/telemetry/test/telemetry_service_test.go
+12 -12
View File
@@ -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