Send connection type to telemetry (#964)

* Send connection type to telemetry

When connected, determine how the participant's primary connection is
connected and report it in ParticipantActive event.

* address feedback

* fixed case where prflx is reported instead of relay

* incorporate comments

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

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=

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()

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)

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")

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) {

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()

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()