diff --git a/go.mod b/go.mod index 8c2534f6a..c9f0b50ce 100644 --- a/go.mod +++ b/go.mod @@ -86,8 +86,8 @@ require ( go.uber.org/multierr v1.6.0 // indirect golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d // indirect golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3 // indirect - golang.org/x/net v0.0.0-20220708220712-1185a9018129 // indirect - golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 // indirect + golang.org/x/net v0.0.0-20220728211354-c7608f3a8462 // indirect + golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10 // indirect golang.org/x/text v0.3.7 // indirect golang.org/x/tools v0.1.10 // indirect golang.org/x/xerrors v0.0.0-20220411194840-2f41105eb62f // indirect diff --git a/go.sum b/go.sum index ff73efc21..d3a7db3b3 100644 --- a/go.sum +++ b/go.sum @@ -532,8 +532,8 @@ golang.org/x/net v0.0.0-20220401154927-543a649e0bdd/go.mod h1:CfG3xpIq0wQ8r1q4Su golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk= golang.org/x/net v0.0.0-20220531201128-c960675eff93/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= golang.org/x/net v0.0.0-20220630215102-69896b714898/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= -golang.org/x/net v0.0.0-20220708220712-1185a9018129 h1:vucSRfWwTsoXro7P+3Cjlr6flUMtzCwzlvkxEQtHHB0= -golang.org/x/net v0.0.0-20220708220712-1185a9018129/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= +golang.org/x/net v0.0.0-20220728211354-c7608f3a8462 h1:UreQrH7DbFXSi9ZFox6FNT3WBooWmdANpU+IfkT1T4I= +golang.org/x/net v0.0.0-20220728211354-c7608f3a8462/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= @@ -622,8 +622,8 @@ golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220608164250-635b8c9b7f68/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220622161953-175b2fd9d664/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 h1:0A+M6Uqn+Eje4kHMK80dtF3JCXC4ykBgQG4Fe06QRhQ= -golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10 h1:WIoqL4EROvwiPdUtaip4VcDdpZ4kha7wBWZrbVKCIZg= +golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 42e2c2aad..8f68df662 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -238,6 +238,10 @@ func (p *ParticipantImpl) GetClientConfiguration() *livekit.ClientConfiguration return p.params.ClientConf } +func (p *ParticipantImpl) GetICEConnectionType() types.ICEConnectionType { + return p.TransportManager.GetICEConnectionType() +} + // SetMetadata attaches metadata to the participant func (p *ParticipantImpl) SetMetadata(metadata string) { p.lock.Lock() diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index cdad3a0d5..ea83f2ee8 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -251,7 +251,10 @@ func (r *Room) Join(participant types.LocalParticipant, opts *ParticipantOptions // start the workers once connectivity is established p.Start() - r.telemetry.ParticipantActive(context.Background(), r.ToProto(), p.ToProto(), &livekit.AnalyticsClientMeta{ClientConnectTime: uint32(time.Since(p.ConnectedAt()).Milliseconds())}) + r.telemetry.ParticipantActive(context.Background(), r.ToProto(), p.ToProto(), &livekit.AnalyticsClientMeta{ + ClientConnectTime: uint32(time.Since(p.ConnectedAt()).Milliseconds()), + ConnectionType: string(p.GetICEConnectionType()), + }) } else if state == livekit.ParticipantInfo_DISCONNECTED { // remove participant from room go r.RemoveParticipant(p.Identity(), types.ParticipantCloseReasonStateDisconnected) diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index d5bfbcc1e..ba8a67171 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -8,6 +8,7 @@ import ( "github.com/bep/debounce" "github.com/go-logr/logr" + "github.com/pion/ice/v2" "github.com/pion/interceptor" "github.com/pion/interceptor/pkg/cc" "github.com/pion/interceptor/pkg/gcc" @@ -1030,6 +1031,44 @@ func (t *PCTransport) RemoveTrackFromStreamAllocator(subTrack types.SubscribedTr t.streamAllocator.RemoveTrack(subTrack.DownTrack()) } +func (t *PCTransport) getICEConnectionType() types.ICEConnectionType { + unknown := types.ICEConnectionTypeUnknown + if t.pc == nil { + return unknown + } + p, err := t.getSelectedPair() + if err != nil { + return unknown + } + + if p.Remote.Typ == webrtc.ICECandidateTypeRelay { + return types.ICEConnectionTypeTURN + } else if p.Remote.Typ == webrtc.ICECandidateTypePrflx { + // if the remote relay candidate pings us *before* we get a relay candidate, + // Pion would have created a prflx candidate with the same address as the relay candidate. + // to report an accurate connection type, we'll compare to see if existing relay candidates match + t.lock.RLock() + allowedRemoteCandidates := t.allowedRemoteCandidates + t.lock.RUnlock() + + for _, ci := range allowedRemoteCandidates { + candidateValue := strings.TrimPrefix(ci, "candidate:") + candidate, err := ice.UnmarshalCandidate(candidateValue) + if err == nil && candidate.Type() == ice.CandidateTypeRelay { + if p.Remote.Address == candidate.Address() && + p.Remote.Port == uint16(candidate.Port()) && + p.Remote.Protocol.String() == candidate.NetworkType().NetworkShort() { + return types.ICEConnectionTypeTURN + } + } + } + } + if p.Remote.Protocol == webrtc.ICEProtocolTCP { + return types.ICEConnectionTypeTCP + } + return types.ICEConnectionTypeUDP +} + func (t *PCTransport) preparePC(previousAnswer webrtc.SessionDescription) error { // sticky data channel to first m-lines, if someday we don't send sdp without media streams to // client's subscribe pc after joining, should change this step @@ -1276,7 +1315,9 @@ func (t *PCTransport) clearLocalDescriptionSent() { t.cachedLocalCandidates = nil t.allowedLocalCandidates = nil + t.lock.Lock() t.allowedRemoteCandidates = nil + t.lock.Unlock() t.filteredLocalCandidates = nil t.filteredRemoteCandidates = nil } @@ -1325,13 +1366,15 @@ func (t *PCTransport) handleRemoteICECandidate(e *event) error { return nil } + t.lock.Lock() + t.allowedRemoteCandidates = append(t.allowedRemoteCandidates, c.Candidate) + t.lock.Unlock() + if t.pc.RemoteDescription() == nil { t.pendingRemoteCandidates = append(t.pendingRemoteCandidates, c) return nil } - t.allowedRemoteCandidates = append(t.allowedRemoteCandidates, c.Candidate) - t.params.Logger.Infow("add candidate ", "candidate", c.Candidate) if err := t.pc.AddICECandidate(*c); err != nil { return errors.Wrap(err, "add ice candidate failed") diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index b1fa37dc4..d29093af0 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -442,6 +442,10 @@ func (t *TransportManager) SubscriberAsPrimary() bool { return t.params.SubscriberAsPrimary } +func (t *TransportManager) GetICEConnectionType() types.ICEConnectionType { + return t.getTransport(true).getICEConnectionType() +} + func (t *TransportManager) getTransport(isPrimary bool) *PCTransport { pcTransport := t.publisher if (isPrimary && t.params.SubscriberAsPrimary) || (!isPrimary && !t.params.SubscriberAsPrimary) { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index b3716ad99..4db3a4cb0 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -217,6 +217,17 @@ type IceConfig struct { PreferPub PreferCandidateType } +// ------------------------------------------------------- + +type ICEConnectionType string + +const ( + ICEConnectionTypeUDP ICEConnectionType = "udp" + ICEConnectionTypeTCP ICEConnectionType = "tcp" + ICEConnectionTypeTURN ICEConnectionType = "turn" + ICEConnectionTypeUnknown ICEConnectionType = "unknown" +) + //counterfeiter:generate . LocalParticipant type LocalParticipant interface { Participant @@ -230,6 +241,7 @@ type LocalParticipant interface { IsReady() bool SubscriberAsPrimary() bool GetClientConfiguration() *livekit.ClientConfiguration + GetICEConnectionType() ICEConnectionType SetResponseSink(sink routing.MessageSink) CloseSignalConnection() diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 27a306494..37ec2a86e 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -236,6 +236,16 @@ type FakeLocalParticipant struct { getConnectionQualityReturnsOnCall map[int]struct { result1 *livekit.ConnectionQualityInfo } + GetICEConnectionTypeStub func() types.ICEConnectionType + getICEConnectionTypeMutex sync.RWMutex + getICEConnectionTypeArgsForCall []struct { + } + getICEConnectionTypeReturns struct { + result1 types.ICEConnectionType + } + getICEConnectionTypeReturnsOnCall map[int]struct { + result1 types.ICEConnectionType + } GetLoggerStub func() logger.Logger getLoggerMutex sync.RWMutex getLoggerArgsForCall []struct { @@ -1884,6 +1894,59 @@ func (fake *FakeLocalParticipant) GetConnectionQualityReturnsOnCall(i int, resul }{result1} } +func (fake *FakeLocalParticipant) GetICEConnectionType() types.ICEConnectionType { + fake.getICEConnectionTypeMutex.Lock() + ret, specificReturn := fake.getICEConnectionTypeReturnsOnCall[len(fake.getICEConnectionTypeArgsForCall)] + fake.getICEConnectionTypeArgsForCall = append(fake.getICEConnectionTypeArgsForCall, struct { + }{}) + stub := fake.GetICEConnectionTypeStub + fakeReturns := fake.getICEConnectionTypeReturns + fake.recordInvocation("GetICEConnectionType", []interface{}{}) + fake.getICEConnectionTypeMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) GetICEConnectionTypeCallCount() int { + fake.getICEConnectionTypeMutex.RLock() + defer fake.getICEConnectionTypeMutex.RUnlock() + return len(fake.getICEConnectionTypeArgsForCall) +} + +func (fake *FakeLocalParticipant) GetICEConnectionTypeCalls(stub func() types.ICEConnectionType) { + fake.getICEConnectionTypeMutex.Lock() + defer fake.getICEConnectionTypeMutex.Unlock() + fake.GetICEConnectionTypeStub = stub +} + +func (fake *FakeLocalParticipant) GetICEConnectionTypeReturns(result1 types.ICEConnectionType) { + fake.getICEConnectionTypeMutex.Lock() + defer fake.getICEConnectionTypeMutex.Unlock() + fake.GetICEConnectionTypeStub = nil + fake.getICEConnectionTypeReturns = struct { + result1 types.ICEConnectionType + }{result1} +} + +func (fake *FakeLocalParticipant) GetICEConnectionTypeReturnsOnCall(i int, result1 types.ICEConnectionType) { + fake.getICEConnectionTypeMutex.Lock() + defer fake.getICEConnectionTypeMutex.Unlock() + fake.GetICEConnectionTypeStub = nil + if fake.getICEConnectionTypeReturnsOnCall == nil { + fake.getICEConnectionTypeReturnsOnCall = make(map[int]struct { + result1 types.ICEConnectionType + }) + } + fake.getICEConnectionTypeReturnsOnCall[i] = struct { + result1 types.ICEConnectionType + }{result1} +} + func (fake *FakeLocalParticipant) GetLogger() logger.Logger { fake.getLoggerMutex.Lock() ret, specificReturn := fake.getLoggerReturnsOnCall[len(fake.getLoggerArgsForCall)] @@ -4721,6 +4784,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.getClientConfigurationMutex.RUnlock() fake.getConnectionQualityMutex.RLock() defer fake.getConnectionQualityMutex.RUnlock() + fake.getICEConnectionTypeMutex.RLock() + defer fake.getICEConnectionTypeMutex.RUnlock() fake.getLoggerMutex.RLock() defer fake.getLoggerMutex.RUnlock() fake.getPublishedTrackMutex.RLock()