mirror of
https://github.com/livekit/livekit.git
synced 2026-08-26 07:09:51 +00:00
Experimental WARP (#4649)
* Experimental WARP * fix panic * go dep * stats
This commit is contained in:
cnderrauber
@@ -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
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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"`
|
||||
|
||||
|
||||
+10
-4
@@ -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
|
||||
}
|
||||
|
||||
@@ -1265,6 +1265,7 @@ func (r *Room) onStateChange(p types.LocalParticipant) {
|
||||
p.ToProto(),
|
||||
meta,
|
||||
false,
|
||||
p.IsWarpEnabled(),
|
||||
p.TelemetryGuard(),
|
||||
)
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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())
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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) {
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user