From dbe06aa8d1855d827132f2ea5918f4b304fd96c9 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Fri, 14 Aug 2026 16:26:08 +0800 Subject: [PATCH] Experimental WARP (#4649) * Experimental WARP * fix panic * go dep * stats --- go.mod | 6 ++ go.sum | 12 ++-- pkg/config/config.go | 5 ++ pkg/rtc/participant.go | 14 +++-- pkg/rtc/room.go | 1 + pkg/rtc/transport.go | 6 ++ pkg/rtc/transportmanager.go | 3 + pkg/rtc/types/interfaces.go | 1 + .../typesfakes/fake_local_participant.go | 63 +++++++++++++++++++ pkg/service/roommanager.go | 20 +++--- pkg/telemetry/events.go | 3 +- pkg/telemetry/events_test.go | 4 +- pkg/telemetry/prometheus/packets.go | 35 ++++++----- pkg/telemetry/prometheus/rooms.go | 6 +- .../telemetryfakes/fake_telemetry_service.go | 22 ++++--- pkg/telemetry/telemetryservice.go | 4 +- 16 files changed, 152 insertions(+), 53 deletions(-) diff --git a/go.mod b/go.mod index 093625e43..ead98871c 100644 --- a/go.mod +++ b/go.mod @@ -153,3 +153,9 @@ require ( google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d // indirect google.golang.org/grpc v1.83.0 // indirect ) + +replace github.com/pion/webrtc/v4 => github.com/livekit/webrtc-pion/v4 v4.2.18-warp.1 + +replace github.com/pion/dtls/v3 => github.com/livekit/dtls/v3 v3.1.5-warp.1 + +replace github.com/pion/ice/v4 => github.com/livekit/ice/v4 v4.4.0-warp.1 diff --git a/go.sum b/go.sum index 19ddb1945..19c9c30d1 100644 --- a/go.sum +++ b/go.sum @@ -156,6 +156,10 @@ github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/lithammer/shortuuid/v4 v4.2.0 h1:LMFOzVB3996a7b8aBuEXxqOBflbfPQAiVzkIcHO0h8c= github.com/lithammer/shortuuid/v4 v4.2.0/go.mod h1:D5noHZ2oFw/YaKCfGy0YxyE7M0wMbezmMjPdhyEFe6Y= +github.com/livekit/dtls/v3 v3.1.5-warp.1 h1:5jo2qQreDnUrgw6Al1F+9xVyXHvWQ2IaQkjF53FHEyU= +github.com/livekit/dtls/v3 v3.1.5-warp.1/go.mod h1:gz1K4jg6c+fq86oQMH4pilpCEOEPwmEr2jY+VcF/mkU= +github.com/livekit/ice/v4 v4.4.0-warp.1 h1:P9Y1f02VVx6kkRJxY3jzoS7HnblFbYkgMwiu6iOqCOU= +github.com/livekit/ice/v4 v4.4.0-warp.1/go.mod h1:obAyD+J+Hzs7QA7Y8YXHp5uIn6gb7z87pKedXZkrcFU= github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5ATTo469PQPkqzdoU7be46ryiCDO3boc= github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0 h1:XHNNzebIKZRkLimla/hFGrAIX5EMWHctrgt3hLw7s+I= @@ -164,6 +168,8 @@ github.com/livekit/protocol v1.50.5-0.20260811022948-6ad5660c9f8f h1:1Tod/QfZbX2 github.com/livekit/protocol v1.50.5-0.20260811022948-6ad5660c9f8f/go.mod h1:edX/q09IZsPPR6SRK1xqA2lcdskpXIZfVvZAPDVV4yw= github.com/livekit/psrpc v0.7.3 h1:bekuZt/ZQzg8+/M8G6G5jq7bvV9fAKdPHSOZeTwrIIc= github.com/livekit/psrpc v0.7.3/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw= +github.com/livekit/webrtc-pion/v4 v4.2.18-warp.1 h1:fH+v4W+NFp9FfPzON6FaUFNmazGcctaAhb2P+Ksf+1s= +github.com/livekit/webrtc-pion/v4 v4.2.18-warp.1/go.mod h1:rbKGHo2OpNUImWTvRIV776/3xjjq/t47H3IZiTtwluc= github.com/mackerelio/go-osstat v0.2.8 h1:I2duicTaCGWoM53XwAwA9OIe1inu0xnVs8/pqOWWVr4= github.com/mackerelio/go-osstat v0.2.8/go.mod h1:SyS3XxKdoSKJnTGTkN5Yrh6VUQVuAURACfE6y+2DN4k= github.com/magefile/mage v1.17.2 h1:fyXVu1eadI8Ap1HCCNgEhJ5McIWiYhLR8uol64ZZc40= @@ -231,10 +237,6 @@ github.com/ory/dockertest/v4 v4.0.0 h1:i19aFsO/VXE0VrMk4ifnKW4G/KIJ93PCjLOslxXoP github.com/ory/dockertest/v4 v4.0.0/go.mod h1:b5Ofu8VIxWNhXFvQcLu17pRNQdoUBKtXBW74G4Ygzx8= github.com/pion/datachannel v1.6.2 h1:7EXQ8TH3vTouBUdRWYbcX2edSx9Yj6k5zl5P+qyxEPc= github.com/pion/datachannel v1.6.2/go.mod h1:pzbdAZvyGtXbcHM1hBbsFaOTf40lZizU/dNlvVOak6E= -github.com/pion/dtls/v3 v3.1.5 h1:9xJtVsHwMYeSjPp5Hh1FTis4DchnQWtnOa5o+6ygqfc= -github.com/pion/dtls/v3 v3.1.5/go.mod h1:gz1K4jg6c+fq86oQMH4pilpCEOEPwmEr2jY+VcF/mkU= -github.com/pion/ice/v4 v4.4.0 h1:wvHDDqimaC38Y7MVpD46Y63p246ChvXd87VKoLYS5b4= -github.com/pion/ice/v4 v4.4.0/go.mod h1:obAyD+J+Hzs7QA7Y8YXHp5uIn6gb7z87pKedXZkrcFU= github.com/pion/interceptor v0.1.47 h1:yw8t5pJ2f8t78NgU+8EmxhaqYLXS7uFCC/tAGOaSDBo= github.com/pion/interceptor v0.1.47/go.mod h1:7yoRBzaIDETPC6cIN8Zj9EyGqHv1ImOpcTFPha6MuOM= github.com/pion/logging v0.2.4 h1:tTew+7cmQ+Mc1pTBLKH2puKsOvhm32dROumOZ655zB8= @@ -261,8 +263,6 @@ github.com/pion/transport/v4 v4.1.0 h1:8S+nF2reM2cJuqC6g78OVy2BBgmbdns+acx3jA97B github.com/pion/transport/v4 v4.1.0/go.mod h1:06hFI+jCFcok2X2MekVufNZ/uzNZXivGBPfviSVcjgM= github.com/pion/turn/v5 v5.0.12 h1:6+b69ivQQXSlyfkp2AKripqD2k3W32qXK8QzCzpJWPI= github.com/pion/turn/v5 v5.0.12/go.mod h1:CQACsRDJtjQ+6RSrGHrS2PCIerLwbW3uqXRqOvtjAFg= -github.com/pion/webrtc/v4 v4.2.18 h1:smA/3g6Gy4RohM0VIZ5KKY/12TQbxv3XFgpUMyb2EUI= -github.com/pion/webrtc/v4 v4.2.18/go.mod h1:vmzi6s+rvhoIuT94DPqivB+0xJXs9rG4QRD+4MgBtlY= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= diff --git a/pkg/config/config.go b/pkg/config/config.go index f4490f35c..2ef6a3efe 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -122,6 +122,11 @@ type RTCConfig struct { TURNServers []TURNServer `yaml:"turn_servers,omitempty"` + // EnableWarp turns on WARP = SPED (DTLS-in-STUN, saves DTLS round-trip) + + // SNAP (SCTP INIT in SDP, saves SCTP/data-channel init round-trip). + // Off by default; experimental. + EnableWarp bool `yaml:"enable_warp,omitempty"` + // Deprecated StrictACKs bool `yaml:"strict_acks,omitempty"` diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index e7f33097a..f35228f33 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -233,6 +233,7 @@ type ParticipantParams struct { EnableStartAtDesiredQuality bool MigrationWaitDuration time.Duration ExcludeIPv6LocalCandidates bool + EnableWarp bool } type ParticipantImpl struct { @@ -1453,9 +1454,9 @@ func (p *ParticipantImpl) recordRTCState(closeReason types.ParticipantCloseReaso } if p.IsConnectionCanceled(closeReason) { - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, p.params.EnableWarp) } else { - prometheus.IncrementParticipantRtcFailure(1) + prometheus.IncrementParticipantRtcFailure(1, p.params.EnableWarp) } } @@ -2100,6 +2101,7 @@ func (p *ParticipantImpl) setupTransportManager() error { FireOnTrackBySdp: p.params.FireOnTrackBySdp, EnableDataTracks: p.params.EnableDataTracks, ExcludeIPv6LocalCandidates: p.params.ExcludeIPv6LocalCandidates, + EnableWarp: p.params.EnableWarp, } if p.params.SyncStreams && p.params.PlayoutDelay.GetEnabled() && p.params.ClientInfo.isFirefox() { // we will disable playout delay for Firefox if the user is expecting @@ -2683,8 +2685,8 @@ func (p *ParticipantImpl) onPrimaryTransportInitialConnected() { } if !p.sessionStartRecorded.Swap(true) { - prometheus.RecordSessionStartTime(int(p.ProtocolVersion()), time.Since(p.params.SessionStartTime)) - prometheus.IncrementParticipantRtcSuccess(1) + prometheus.RecordSessionStartTime(int(p.ProtocolVersion()), p.params.EnableWarp, time.Since(p.params.SessionStartTime)) + prometheus.IncrementParticipantRtcSuccess(1, p.params.EnableWarp) } p.updateState(livekit.ParticipantInfo_ACTIVE) } @@ -4442,3 +4444,7 @@ func (p *ParticipantImpl) PerformRpc(req *livekit.PerformRpcRequest, resultCh ch p.rpcLock.Unlock() }() } + +func (p *ParticipantImpl) IsWarpEnabled() bool { + return p.params.EnableWarp +} diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index b6e97a8b0..74f7f3059 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -1265,6 +1265,7 @@ func (r *Room) onStateChange(p types.LocalParticipant) { p.ToProto(), meta, false, + p.IsWarpEnabled(), p.TelemetryGuard(), ) diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index c1b5c0c01..cdd3eb6e1 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -330,6 +330,7 @@ type TransportParams struct { DatachannelMaxReceiverBufferSize int EnableDataTracks bool + EnableWarp bool } func newPeerConnection( @@ -382,6 +383,11 @@ func newPeerConnection( se.EnableSCTPZeroChecksum(true) } + if params.EnableWarp { + se.EnableSped(true) + se.EnableSctpSnap(true) + } + // // Disable SRTP replay protection (https://datatracker.ietf.org/doc/html/rfc3711#page-15). // Needed due to lack of RTX stream support in Pion. diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 151af1bf1..9e8cf51b1 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -100,6 +100,7 @@ type TransportManagerParams struct { FireOnTrackBySdp bool EnableDataTracks bool ExcludeIPv6LocalCandidates bool + EnableWarp bool } type TransportManager struct { @@ -172,6 +173,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro FireOnTrackBySdp: params.FireOnTrackBySdp, EnableDataTracks: params.EnableDataTracks, ExcludeIPv6LocalCandidates: params.ExcludeIPv6LocalCandidates, + EnableWarp: params.EnableWarp, }) if err != nil { return nil, err @@ -200,6 +202,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro FireOnTrackBySdp: params.FireOnTrackBySdp, EnableDataTracks: params.EnableDataTracks, ExcludeIPv6LocalCandidates: params.ExcludeIPv6LocalCandidates, + EnableWarp: params.EnableWarp, }) if err != nil { return nil, err diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 225cd9861..ddc462e1f 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -430,6 +430,7 @@ type LocalParticipant interface { GetPublisherICESessionUfrag() (string, error) SupportsMoving() error GetLastReliableSequence(migrateOut bool) uint32 + IsWarpEnabled() bool SwapResponseSink(sink routing.MessageSink, reason SignallingCloseReason) GetResponseSink() routing.MessageSink diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 6996a81e3..d4fcfd21b 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -965,6 +965,16 @@ type FakeLocalParticipant struct { isUsingSinglePeerConnectionReturnsOnCall map[int]struct { result1 bool } + IsWarpEnabledStub func() bool + isWarpEnabledMutex sync.RWMutex + isWarpEnabledArgsForCall []struct { + } + isWarpEnabledReturns struct { + result1 bool + } + isWarpEnabledReturnsOnCall map[int]struct { + result1 bool + } IssueFullReconnectStub func(types.ParticipantCloseReason) issueFullReconnectMutex sync.RWMutex issueFullReconnectArgsForCall []struct { @@ -6628,6 +6638,59 @@ func (fake *FakeLocalParticipant) IsUsingSinglePeerConnectionReturnsOnCall(i int }{result1} } +func (fake *FakeLocalParticipant) IsWarpEnabled() bool { + fake.isWarpEnabledMutex.Lock() + ret, specificReturn := fake.isWarpEnabledReturnsOnCall[len(fake.isWarpEnabledArgsForCall)] + fake.isWarpEnabledArgsForCall = append(fake.isWarpEnabledArgsForCall, struct { + }{}) + stub := fake.IsWarpEnabledStub + fakeReturns := fake.isWarpEnabledReturns + fake.recordInvocation("IsWarpEnabled", []interface{}{}) + fake.isWarpEnabledMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) IsWarpEnabledCallCount() int { + fake.isWarpEnabledMutex.RLock() + defer fake.isWarpEnabledMutex.RUnlock() + return len(fake.isWarpEnabledArgsForCall) +} + +func (fake *FakeLocalParticipant) IsWarpEnabledCalls(stub func() bool) { + fake.isWarpEnabledMutex.Lock() + defer fake.isWarpEnabledMutex.Unlock() + fake.IsWarpEnabledStub = stub +} + +func (fake *FakeLocalParticipant) IsWarpEnabledReturns(result1 bool) { + fake.isWarpEnabledMutex.Lock() + defer fake.isWarpEnabledMutex.Unlock() + fake.IsWarpEnabledStub = nil + fake.isWarpEnabledReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalParticipant) IsWarpEnabledReturnsOnCall(i int, result1 bool) { + fake.isWarpEnabledMutex.Lock() + defer fake.isWarpEnabledMutex.Unlock() + fake.IsWarpEnabledStub = nil + if fake.isWarpEnabledReturnsOnCall == nil { + fake.isWarpEnabledReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.isWarpEnabledReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeLocalParticipant) IssueFullReconnect(arg1 types.ParticipantCloseReason) { fake.issueFullReconnectMutex.Lock() fake.issueFullReconnectArgsForCall = append(fake.issueFullReconnectArgsForCall, struct { diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 0447f14e9..57282b3c3 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -306,7 +306,7 @@ func (r *RoomManager) StartSession( room, err := r.getOrCreateRoom(ctx, createRoom) if err != nil { if pi.Identity != "" { - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, false) } return err } @@ -359,7 +359,7 @@ func (r *RoomManager) StartSession( Leave: leave, }, }) - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, participant.IsWarpEnabled()) return errors.New("could not restart closed participant") } @@ -383,11 +383,11 @@ func (r *RoomManager) StartSession( pi.ReconnectReason, ); err != nil { participant.GetLogger().Warnw("could not resume participant", err) - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, participant.IsWarpEnabled()) return err } r.telemetry.ParticipantResumed(ctx, room.ToProto(), participant.ToProto(), r.currentNode.NodeID(), pi.ReconnectReason) - prometheus.IncrementParticipantRtcActive(1) + prometheus.IncrementParticipantRtcActive(1, participant.IsWarpEnabled()) go room.HandleSyncState(participant, pi.SyncState) @@ -419,7 +419,7 @@ func (r *RoomManager) StartSession( Leave: leave, }, }) - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, false) return errors.New("could not restart participant") } @@ -484,6 +484,7 @@ func (r *RoomManager) StartSession( enabledCodecs = append(enabledCodecs, &livekit.Codec{Mime: mime.MimeTypeRTX.String()}) } + enableWarp := !useOneShotSignallingMode && r.config.RTC.EnableWarp participant, err = rtc.NewParticipant(rtc.ParticipantParams{ Identity: pi.Identity, Name: pi.Name, @@ -541,9 +542,10 @@ func (r *RoomManager) StartSession( EnableDataTracks: r.config.EnableDataTracks, EnableParticipantDataBlob: r.config.EnableParticipantDataBlob, EnableRTPStreamRestartDetection: r.config.RTC.EnableRTPStreamRestartDetection, + EnableWarp: enableWarp, }) if err != nil { - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, enableWarp) return err } iceConfig := r.setIceConfig(room.Name(), participant) @@ -559,7 +561,7 @@ func (r *RoomManager) StartSession( if err = room.Join(participant, requestSource, &opts, iceServers); err != nil { pLogger.Errorw("could not join room", err) _ = participant.Close(true, types.ParticipantCloseReasonJoinFailed, false) - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, enableWarp) return err } @@ -571,7 +573,7 @@ func (r *RoomManager) StartSession( participantServerClosers.Close() pLogger.Errorw("could not join register participant topic", err) _ = participant.Close(true, types.ParticipantCloseReasonMessageBusFailed, false) - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, enableWarp) return err } @@ -582,7 +584,7 @@ func (r *RoomManager) StartSession( participantServerClosers.Close() pLogger.Errorw("could not join register participant topic for rtc rest participant server", err) _ = participant.Close(true, types.ParticipantCloseReasonMessageBusFailed, false) - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, enableWarp) return err } } diff --git a/pkg/telemetry/events.go b/pkg/telemetry/events.go index bf684156e..989dd870e 100644 --- a/pkg/telemetry/events.go +++ b/pkg/telemetry/events.go @@ -111,6 +111,7 @@ func (t *telemetryService) ParticipantActive( participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool, + isWarp bool, guard *ReferenceGuard, ) { t.enqueue(func() { @@ -135,7 +136,7 @@ func (t *telemetryService) ParticipantActive( prometheus.AddParticipant() } worker.SetConnected() - prometheus.IncrementParticipantRtcActive(1) + prometheus.IncrementParticipantRtcActive(1, isWarp) ev := newParticipantEvent(livekit.AnalyticsEventType_PARTICIPANT_ACTIVE, room, participant) ev.ClientMeta = clientMeta diff --git a/pkg/telemetry/events_test.go b/pkg/telemetry/events_test.go index 3677791b4..03fca0e4a 100644 --- a/pkg/telemetry/events_test.go +++ b/pkg/telemetry/events_test.go @@ -86,7 +86,7 @@ func Test_OnParticipantLeft_EventIsSent(t *testing.T) { guard := &telemetry.ReferenceGuard{} // do - fixture.sut.ParticipantActive(context.Background(), room, participantInfo, &livekit.AnalyticsClientMeta{}, false, guard) + fixture.sut.ParticipantActive(context.Background(), room, participantInfo, &livekit.AnalyticsClientMeta{}, false, false, guard) fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true, guard) time.Sleep(time.Millisecond * 500) @@ -179,7 +179,7 @@ func Test_OnParticipantActive_EventIsSent(t *testing.T) { ClientConnectTime: 420, } - fixture.sut.ParticipantActive(context.Background(), room, participantInfo, clientMetaConnect, false, guard) + fixture.sut.ParticipantActive(context.Background(), room, participantInfo, clientMetaConnect, false, false, guard) time.Sleep(time.Millisecond * 500) require.Equal(t, 2, fixture.analytics.SendEventCallCount()) diff --git a/pkg/telemetry/prometheus/packets.go b/pkg/telemetry/prometheus/packets.go index 6ca9ce436..37e67e895 100644 --- a/pkg/telemetry/prometheus/packets.go +++ b/pkg/telemetry/prometheus/packets.go @@ -15,6 +15,8 @@ package prometheus import ( + "strconv" + "github.com/prometheus/client_golang/prometheus" "go.uber.org/atomic" @@ -149,7 +151,7 @@ func initPacketStats(nodeID string, nodeType livekit.NodeType) { Subsystem: "participant_join", Name: "total", ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()}, - }, []string{"state"}) + }, []string{"state", "warp"}) promConnections = prometheus.NewGaugeVec(prometheus.GaugeOpts{ Namespace: livekitNamespace, Subsystem: "connection", @@ -297,71 +299,72 @@ func RecordRTT(country string, direction Direction, trackSource livekit.TrackSou func IncrementParticipantJoin(join uint32) { if join > 0 { participantSignalConnected.Add(uint64(join)) - promParticipantJoin.WithLabelValues("signal_connected").Add(float64(join)) + promParticipantJoin.WithLabelValues("signal_connected", "").Add(float64(join)) } } func IncrementParticipantJoinFail(fail uint32) { if fail > 0 { - promParticipantJoin.WithLabelValues("signal_failed").Add(float64(fail)) + promParticipantJoin.WithLabelValues("signal_failed", "").Add(float64(fail)) } } func IncrementParticipantJoinValidationFail(validationFail uint32) { if validationFail > 0 { - promParticipantJoin.WithLabelValues("signal_validation_failed").Add(float64(validationFail)) + promParticipantJoin.WithLabelValues("signal_validation_failed", "").Add(float64(validationFail)) } } func IncrementParticipantJoinUpgradeFail(upgradeFail uint32) { if upgradeFail > 0 { - promParticipantJoin.WithLabelValues("signal_upgrade_failed").Add(float64(upgradeFail)) + promParticipantJoin.WithLabelValues("signal_upgrade_failed", "").Add(float64(upgradeFail)) } } func IncrementParticipantJoinWriteInitialResponseFail(writeInitialResponseFail uint32) { if writeInitialResponseFail > 0 { - promParticipantJoin.WithLabelValues("signal_write_initial_response_failed").Add(float64(writeInitialResponseFail)) + promParticipantJoin.WithLabelValues("signal_write_initial_response_failed", "").Add(float64(writeInitialResponseFail)) } } func IncrementParticipantRtcInit(init uint32) { if init > 0 { participantRTCInit.Add(uint64(init)) - promParticipantJoin.WithLabelValues("rtc_init").Add(float64(init)) + promParticipantJoin.WithLabelValues("rtc_init", "").Add(float64(init)) } } func IncrementParticipantRtcConnected(connected uint32) { if connected > 0 { participantRTCConnected.Add(uint64(connected)) - promParticipantJoin.WithLabelValues("rtc_connected").Add(float64(connected)) + promParticipantJoin.WithLabelValues("rtc_connected", "").Add(float64(connected)) } } -func IncrementParticipantRtcActive(active uint32) { +func IncrementParticipantRtcActive(active uint32, warp bool) { if active > 0 { participantRTCActive.Add(uint64(active)) - promParticipantJoin.WithLabelValues("rtc_active").Add(float64(active)) + promParticipantJoin.WithLabelValues("rtc_active", strconv.FormatBool(warp)).Add(float64(active)) } } -func IncrementParticipantRtcCanceled(canceled uint64) { +func IncrementParticipantRtcCanceled(canceled uint64, warp bool) { if canceled > 0 { participantRTCCanceled.Add(canceled) - promParticipantJoin.WithLabelValues("rtc_canceled").Add(float64(canceled)) + promParticipantJoin.WithLabelValues("rtc_canceled", strconv.FormatBool(warp)).Add(float64(canceled)) } } -func IncrementParticipantRtcSuccess(success uint64) { +// todo: check if need to record warp to rtcSucc/Failure +func IncrementParticipantRtcSuccess(success uint64, warp bool) { if success > 0 { - promParticipantJoin.WithLabelValues("rtc_success").Add(float64(success)) + promParticipantJoin.WithLabelValues("rtc_success", strconv.FormatBool(warp)).Add(float64(success)) } } -func IncrementParticipantRtcFailure(failure uint64) { +func IncrementParticipantRtcFailure(failure uint64, warp bool) { if failure > 0 { - promParticipantJoin.WithLabelValues("rtc_failure").Add(float64(failure)) + promParticipantJoin.WithLabelValues("rtc_failure", strconv.FormatBool(warp)).Add(float64(failure)) } } diff --git a/pkg/telemetry/prometheus/rooms.go b/pkg/telemetry/prometheus/rooms.go index 0733413a3..9a7a9c35b 100644 --- a/pkg/telemetry/prometheus/rooms.go +++ b/pkg/telemetry/prometheus/rooms.go @@ -113,7 +113,7 @@ func initRoomStats(nodeID string, nodeType livekit.NodeType) { Name: "start_time_ms", ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()}, Buckets: prometheus.ExponentialBucketsRange(100, 10000, 15), - }, []string{"protocol_version"}) + }, []string{"protocol_version", "warp"}) promSessionDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ Namespace: livekitNamespace, Subsystem: "session", @@ -291,8 +291,8 @@ func RecordSessionJoinLatency(protocolVersion int, d time.Duration) { promSessionJoinLatency.WithLabelValues(strconv.Itoa(protocolVersion)).Observe(float64(d.Milliseconds())) } -func RecordSessionStartTime(protocolVersion int, d time.Duration) { - promSessionStartTime.WithLabelValues(strconv.Itoa(protocolVersion)).Observe(float64(d.Milliseconds())) +func RecordSessionStartTime(protocolVersion int, warp bool, d time.Duration) { + promSessionStartTime.WithLabelValues(strconv.Itoa(protocolVersion), strconv.FormatBool(warp)).Observe(float64(d.Milliseconds())) } func RecordSessionDuration(protocolVersion int, d time.Duration) { diff --git a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go index 5c0f28881..25a241bef 100644 --- a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go +++ b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go @@ -83,7 +83,7 @@ type FakeTelemetryService struct { arg2 string arg3 *livekit.EgressInfo } - ParticipantActiveStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, *telemetry.ReferenceGuard) + ParticipantActiveStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, bool, *telemetry.ReferenceGuard) participantActiveMutex sync.RWMutex participantActiveArgsForCall []struct { arg1 context.Context @@ -91,7 +91,8 @@ type FakeTelemetryService struct { arg3 *livekit.ParticipantInfo arg4 *livekit.AnalyticsClientMeta arg5 bool - arg6 *telemetry.ReferenceGuard + arg6 bool + arg7 *telemetry.ReferenceGuard } ParticipantJoinedStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.ClientInfo, *livekit.AnalyticsClientMeta, bool, *telemetry.ReferenceGuard) participantJoinedMutex sync.RWMutex @@ -698,7 +699,7 @@ func (fake *FakeTelemetryService) NotifyEgressEventArgsForCall(i int) (context.C return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 } -func (fake *FakeTelemetryService) ParticipantActive(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 *livekit.AnalyticsClientMeta, arg5 bool, arg6 *telemetry.ReferenceGuard) { +func (fake *FakeTelemetryService) ParticipantActive(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 *livekit.AnalyticsClientMeta, arg5 bool, arg6 bool, arg7 *telemetry.ReferenceGuard) { fake.participantActiveMutex.Lock() fake.participantActiveArgsForCall = append(fake.participantActiveArgsForCall, struct { arg1 context.Context @@ -706,13 +707,14 @@ func (fake *FakeTelemetryService) ParticipantActive(arg1 context.Context, arg2 * arg3 *livekit.ParticipantInfo arg4 *livekit.AnalyticsClientMeta arg5 bool - arg6 *telemetry.ReferenceGuard - }{arg1, arg2, arg3, arg4, arg5, arg6}) + arg6 bool + arg7 *telemetry.ReferenceGuard + }{arg1, arg2, arg3, arg4, arg5, arg6, arg7}) stub := fake.ParticipantActiveStub - fake.recordInvocation("ParticipantActive", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6}) + fake.recordInvocation("ParticipantActive", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6, arg7}) fake.participantActiveMutex.Unlock() if stub != nil { - fake.ParticipantActiveStub(arg1, arg2, arg3, arg4, arg5, arg6) + fake.ParticipantActiveStub(arg1, arg2, arg3, arg4, arg5, arg6, arg7) } } @@ -722,17 +724,17 @@ func (fake *FakeTelemetryService) ParticipantActiveCallCount() int { return len(fake.participantActiveArgsForCall) } -func (fake *FakeTelemetryService) ParticipantActiveCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, *telemetry.ReferenceGuard)) { +func (fake *FakeTelemetryService) ParticipantActiveCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, bool, *telemetry.ReferenceGuard)) { fake.participantActiveMutex.Lock() defer fake.participantActiveMutex.Unlock() fake.ParticipantActiveStub = stub } -func (fake *FakeTelemetryService) ParticipantActiveArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, *telemetry.ReferenceGuard) { +func (fake *FakeTelemetryService) ParticipantActiveArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, bool, *telemetry.ReferenceGuard) { fake.participantActiveMutex.RLock() defer fake.participantActiveMutex.RUnlock() argsForCall := fake.participantActiveArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6, argsForCall.arg7 } func (fake *FakeTelemetryService) ParticipantJoined(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 *livekit.ClientInfo, arg5 *livekit.AnalyticsClientMeta, arg6 bool, arg7 *telemetry.ReferenceGuard) { diff --git a/pkg/telemetry/telemetryservice.go b/pkg/telemetry/telemetryservice.go index 28c90400b..bdf9c820c 100644 --- a/pkg/telemetry/telemetryservice.go +++ b/pkg/telemetry/telemetryservice.go @@ -40,7 +40,7 @@ type TelemetryService interface { // 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, guard *ReferenceGuard) // ParticipantActive - a participant establishes media connection - ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool, guard *ReferenceGuard) + ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool, isWarp 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 @@ -108,7 +108,7 @@ func (n NullTelemetryService) RoomStarted(ctx context.Context, room *livekit.Roo 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) ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool, isWarp bool, guard *ReferenceGuard) { } func (n NullTelemetryService) ParticipantResumed(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, nodeID livekit.NodeID, reason livekit.ReconnectReason) { }