dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-07-28 16:29:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Log ICE reconnected. (#2999)

Browse Source 
To increase visibility of ICE reconnect, logging reconnected at Infow
level. Otherwise, it is hard to see if an ICE restart finished
successfully.

Also, cleaning up ICEConnectionDetails a bit. Just separate out
read-only fields into its own struct and use it for read-only export.
This commit is contained in:
Raja Subramanian
2024-09-12 13:16:41 +05:30
committed by GitHub
parent ce663f9550
commit 1b5bb4dddc
7 changed files with 120 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/room.go
+14 -14
View File
@@ -455,10 +455,10 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me
meta := &livekit.AnalyticsClientMeta{
ClientConnectTime: uint32(time.Since(p.ConnectedAt()).Milliseconds()),
}
cds := p.GetICEConnectionDetails()
for _, cd := range cds {
if cd.Type != types.ICEConnectionTypeUnknown {
meta.ConnectionType = string(cd.Type)
infos := p.GetICEConnectionInfo()
for _, info := range infos {
if info.Type != types.ICEConnectionTypeUnknown {
meta.ConnectionType = string(info.Type)
break
}
}
@@ -469,7 +469,7 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me
false,
)
p.GetLogger().Infow("participant active", connectionDetailsFields(cds)...)
p.GetLogger().Infow("participant active", connectionDetailsFields(infos)...)
} else if state == livekit.ParticipantInfo_DISCONNECTED {
// remove participant from room
// participant should already be closed and have a close reason, so NONE is fine here
@@ -695,7 +695,7 @@ func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity, pID livek
r.protoProxy.MarkDirty(immediateChange)
if !p.HasConnected() {
fields := append(connectionDetailsFields(p.GetICEConnectionDetails()),
fields := append(connectionDetailsFields(p.GetICEConnectionInfo()),
"reason", reason.String(),
)
p.GetLogger().Infow("removing participant without connection", fields...)
@@ -1804,12 +1804,12 @@ func IsCloseNotifySkippable(closeReason types.ParticipantCloseReason) bool {
return closeReason == types.ParticipantCloseReasonDuplicateIdentity
}
func connectionDetailsFields(cds []*types.ICEConnectionDetails) []interface{} {
func connectionDetailsFields(infos []*types.ICEConnectionInfo) []interface{} {
var fields []interface{}
connectionType := types.ICEConnectionTypeUnknown
for _, cd := range cds {
candidates := make([]string, 0, len(cd.Remote)+len(cd.Local))
for _, c := range cd.Local {
for _, info := range infos {
candidates := make([]string, 0, len(info.Remote)+len(info.Local))
for _, c := range info.Local {
cStr := "[local]"
if c.Selected {
cStr += "[selected]"
@@ -1822,7 +1822,7 @@ func connectionDetailsFields(cds []*types.ICEConnectionDetails) []interface{} {
cStr += " " + c.Local.String()
candidates = append(candidates, cStr)
}
for _, c := range cd.Remote {
for _, c := range info.Remote {
cStr := "[remote]"
if c.Selected {
cStr += "[selected]"
@@ -1836,10 +1836,10 @@ func connectionDetailsFields(cds []*types.ICEConnectionDetails) []interface{} {
candidates = append(candidates, cStr)
}
if len(candidates) > 0 {
fields = append(fields, fmt.Sprintf("%sCandidates", strings.ToLower(cd.Transport.String())), candidates)
fields = append(fields, fmt.Sprintf("%sCandidates", strings.ToLower(info.Transport.String())), candidates)
}
if cd.Type != types.ICEConnectionTypeUnknown {
connectionType = cd.Type
if info.Type != types.ICEConnectionTypeUnknown {
connectionType = info.Type
}
}
fields = append(fields, "connectionType", connectionType)
pkg/rtc/transport.go
+31 -3
View File
@@ -35,6 +35,7 @@ import (
"github.com/pion/webrtc/v3"
"github.com/pkg/errors"
"go.uber.org/atomic"
"go.uber.org/zap/zapcore"
lkinterceptor "github.com/livekit/mediatransportutil/pkg/interceptor"
lktwcc "github.com/livekit/mediatransportutil/pkg/twcc"
@@ -131,6 +132,30 @@ func (e event) String() string {
// -------------------------------------------------------
type wrappedICECandidatePairLogger struct {
pair *webrtc.ICECandidatePair
}
func (w wrappedICECandidatePairLogger) MarshalLogObject(e zapcore.ObjectEncoder) error {
if w.pair == nil {
return nil
}
if w.pair.Local != nil {
e.AddString("localProtocol", w.pair.Local.Protocol.String())
e.AddString("localCandidateType", w.pair.Local.Typ.String())
e.AddString("localAdddress", w.pair.Local.Address)
e.AddUint16("localPort", w.pair.Local.Port)
}
if w.pair.Remote != nil {
e.AddString("remoteProtocol", w.pair.Remote.Protocol.String())
e.AddString("remoteCandidateType", w.pair.Remote.Typ.String())
}
return nil
}
// -------------------------------------------------------------------
type SimulcastTrackInfo struct {
Mid string
Rid string
@@ -617,7 +642,7 @@ func (t *PCTransport) handleConnectionFailed(forceShortConn bool) {
isShort, duration = t.IsShortConnection(time.Now())
if isShort {
pair, err := t.getSelectedPair()
t.params.Logger.Debugw("short ICE connection", "error", err, "pair", pair, "duration", duration)
t.params.Logger.Debugw("short ICE connection", "error", err, "pair", wrappedICECandidatePairLogger{pair}, "duration", duration)
}
}
@@ -653,6 +678,9 @@ func (t *PCTransport) onPeerConnectionStateChange(state webrtc.PeerConnectionSta
t.params.Handler.OnInitialConnected()
t.maybeNotifyFullyEstablished()
} else {
pair, err := t.getSelectedPair()
t.params.Logger.Infow("ice reconnected", "error", err, "pair", wrappedICECandidatePairLogger{pair})
}
case webrtc.PeerConnectionStateFailed:
t.clearConnTimer()
@@ -911,8 +939,8 @@ func (t *PCTransport) HasEverConnected() bool {
return !t.firstConnectedAt.IsZero()
}
func (t *PCTransport) GetICEConnectionDetails() *types.ICEConnectionDetails {
return t.connectionDetails
func (t *PCTransport) GetICEConnectionInfo() *types.ICEConnectionInfo {
return t.connectionDetails.GetInfo()
}
func (t *PCTransport) WriteRTCP(pkts []rtcp.Packet) error {
pkg/rtc/transportmanager.go
+6 -6
View File
@@ -482,15 +482,15 @@ func (t *TransportManager) SubscriberAsPrimary() bool {
return t.params.SubscriberAsPrimary
}
func (t *TransportManager) GetICEConnectionDetails() []*types.ICEConnectionDetails {
details := make([]*types.ICEConnectionDetails, 0, 2)
func (t *TransportManager) GetICEConnectionInfo() []*types.ICEConnectionInfo {
infos := make([]*types.ICEConnectionInfo, 0, 2)
for _, pc := range []*PCTransport{t.publisher, t.subscriber} {
cd := pc.GetICEConnectionDetails()
if cd.HasCandidates() {
details = append(details, cd.Clone())
info := pc.GetICEConnectionInfo()
if info.HasCandidates() {
infos = append(infos, info)
}
}
return details
return infos
}
func (t *TransportManager) getTransport(isPrimary bool) *PCTransport {
pkg/rtc/types/ice.go
+25 -19
View File
@@ -47,43 +47,49 @@ type ICECandidateExtended struct {
Trickle bool
}
type ICEConnectionDetails struct {
// --------------------------------------------
type ICEConnectionInfo struct {
Local []*ICECandidateExtended
Remote []*ICECandidateExtended
Transport livekit.SignalTarget
Type ICEConnectionType
lock sync.Mutex
logger logger.Logger
}
func (i *ICEConnectionInfo) HasCandidates() bool {
return len(i.Local) > 0 || len(i.Remote) > 0
}
// --------------------------------------------
type ICEConnectionDetails struct {
ICEConnectionInfo
lock sync.Mutex
logger logger.Logger
}
func NewICEConnectionDetails(transport livekit.SignalTarget, l logger.Logger) *ICEConnectionDetails {
d := &ICEConnectionDetails{
Transport: transport,
Type: ICEConnectionTypeUnknown,
logger: l,
ICEConnectionInfo: ICEConnectionInfo{
Transport: transport,
Type: ICEConnectionTypeUnknown,
},
logger: l,
}
return d
}
func (d *ICEConnectionDetails) HasCandidates() bool {
func (d *ICEConnectionDetails) GetInfo() *ICEConnectionInfo {
d.lock.Lock()
defer d.lock.Unlock()
return len(d.Local) > 0 || len(d.Remote) > 0
}
// Clone returns a copy of the ICEConnectionDetails, where fields can be read without locking
func (d *ICEConnectionDetails) Clone() *ICEConnectionDetails {
d.lock.Lock()
defer d.lock.Unlock()
clone := &ICEConnectionDetails{
info := &ICEConnectionInfo{
Transport: d.Transport,
Type: d.Type,
logger: d.logger,
Local: make([]*ICECandidateExtended, 0, len(d.Local)),
Remote: make([]*ICECandidateExtended, 0, len(d.Remote)),
}
for _, c := range d.Local {
clone.Local = append(clone.Local, &ICECandidateExtended{
info.Local = append(info.Local, &ICECandidateExtended{
Local: c.Local,
Filtered: c.Filtered,
Selected: c.Selected,
@@ -91,14 +97,14 @@ func (d *ICEConnectionDetails) Clone() *ICEConnectionDetails {
})
}
for _, c := range d.Remote {
clone.Remote = append(clone.Remote, &ICECandidateExtended{
info.Remote = append(info.Remote, &ICECandidateExtended{
Remote: c.Remote,
Filtered: c.Filtered,
Selected: c.Selected,
Trickle: c.Trickle,
})
}
return clone
return info
}
func (d *ICEConnectionDetails) AddLocalCandidate(c *webrtc.ICECandidate, filtered, trickle bool) {
pkg/rtc/types/interfaces.go
+1 -1
View File
@@ -321,7 +321,7 @@ type LocalParticipant interface {
GetBufferFactory() *buffer.Factory
GetPlayoutDelayConfig() *livekit.PlayoutDelay
GetPendingTrack(trackID livekit.TrackID) *livekit.TrackInfo
GetICEConnectionDetails() []*ICEConnectionDetails
GetICEConnectionInfo() []*ICEConnectionInfo
HasConnected() bool
GetEnabledPublishCodecs() []*livekit.Codec
pkg/rtc/types/typesfakes/fake_local_participant.go
+40 -40
View File
@@ -306,15 +306,15 @@ type FakeLocalParticipant struct {
getICEConfigReturnsOnCall map[int]struct {
result1 *livekit.ICEConfig
}
GetICEConnectionDetailsStub func() []*types.ICEConnectionDetails
getICEConnectionDetailsMutex sync.RWMutex
getICEConnectionDetailsArgsForCall []struct {
GetICEConnectionInfoStub func() []*types.ICEConnectionInfo
getICEConnectionInfoMutex sync.RWMutex
getICEConnectionInfoArgsForCall []struct {
}
getICEConnectionDetailsReturns struct {
result1 []*types.ICEConnectionDetails
getICEConnectionInfoReturns struct {
result1 []*types.ICEConnectionInfo
}
getICEConnectionDetailsReturnsOnCall map[int]struct {
result1 []*types.ICEConnectionDetails
getICEConnectionInfoReturnsOnCall map[int]struct {
result1 []*types.ICEConnectionInfo
}
GetLoggerStub func() logger.Logger
getLoggerMutex sync.RWMutex
@@ -2568,15 +2568,15 @@ func (fake *FakeLocalParticipant) GetICEConfigReturnsOnCall(i int, result1 *live
}{result1}
}
func (fake *FakeLocalParticipant) GetICEConnectionDetails() []*types.ICEConnectionDetails {
fake.getICEConnectionDetailsMutex.Lock()
ret, specificReturn := fake.getICEConnectionDetailsReturnsOnCall[len(fake.getICEConnectionDetailsArgsForCall)]
fake.getICEConnectionDetailsArgsForCall = append(fake.getICEConnectionDetailsArgsForCall, struct {
func (fake *FakeLocalParticipant) GetICEConnectionInfo() []*types.ICEConnectionInfo {
fake.getICEConnectionInfoMutex.Lock()
ret, specificReturn := fake.getICEConnectionInfoReturnsOnCall[len(fake.getICEConnectionInfoArgsForCall)]
fake.getICEConnectionInfoArgsForCall = append(fake.getICEConnectionInfoArgsForCall, struct {
}{})
stub := fake.GetICEConnectionDetailsStub
fakeReturns := fake.getICEConnectionDetailsReturns
fake.recordInvocation("GetICEConnectionDetails", []interface{}{})
fake.getICEConnectionDetailsMutex.Unlock()
stub := fake.GetICEConnectionInfoStub
fakeReturns := fake.getICEConnectionInfoReturns
fake.recordInvocation("GetICEConnectionInfo", []interface{}{})
fake.getICEConnectionInfoMutex.Unlock()
if stub != nil {
return stub()
}
@@ -2586,38 +2586,38 @@ func (fake *FakeLocalParticipant) GetICEConnectionDetails() []*types.ICEConnecti
return fakeReturns.result1
}
func (fake *FakeLocalParticipant) GetICEConnectionDetailsCallCount() int {
fake.getICEConnectionDetailsMutex.RLock()
defer fake.getICEConnectionDetailsMutex.RUnlock()
return len(fake.getICEConnectionDetailsArgsForCall)
func (fake *FakeLocalParticipant) GetICEConnectionInfoCallCount() int {
fake.getICEConnectionInfoMutex.RLock()
defer fake.getICEConnectionInfoMutex.RUnlock()
return len(fake.getICEConnectionInfoArgsForCall)
}
func (fake *FakeLocalParticipant) GetICEConnectionDetailsCalls(stub func() []*types.ICEConnectionDetails) {
fake.getICEConnectionDetailsMutex.Lock()
defer fake.getICEConnectionDetailsMutex.Unlock()
fake.GetICEConnectionDetailsStub = stub
func (fake *FakeLocalParticipant) GetICEConnectionInfoCalls(stub func() []*types.ICEConnectionInfo) {
fake.getICEConnectionInfoMutex.Lock()
defer fake.getICEConnectionInfoMutex.Unlock()
fake.GetICEConnectionInfoStub = stub
}
func (fake *FakeLocalParticipant) GetICEConnectionDetailsReturns(result1 []*types.ICEConnectionDetails) {
fake.getICEConnectionDetailsMutex.Lock()
defer fake.getICEConnectionDetailsMutex.Unlock()
fake.GetICEConnectionDetailsStub = nil
fake.getICEConnectionDetailsReturns = struct {
result1 []*types.ICEConnectionDetails
func (fake *FakeLocalParticipant) GetICEConnectionInfoReturns(result1 []*types.ICEConnectionInfo) {
fake.getICEConnectionInfoMutex.Lock()
defer fake.getICEConnectionInfoMutex.Unlock()
fake.GetICEConnectionInfoStub = nil
fake.getICEConnectionInfoReturns = struct {
result1 []*types.ICEConnectionInfo
}{result1}
}
func (fake *FakeLocalParticipant) GetICEConnectionDetailsReturnsOnCall(i int, result1 []*types.ICEConnectionDetails) {
fake.getICEConnectionDetailsMutex.Lock()
defer fake.getICEConnectionDetailsMutex.Unlock()
fake.GetICEConnectionDetailsStub = nil
if fake.getICEConnectionDetailsReturnsOnCall == nil {
fake.getICEConnectionDetailsReturnsOnCall = make(map[int]struct {
result1 []*types.ICEConnectionDetails
func (fake *FakeLocalParticipant) GetICEConnectionInfoReturnsOnCall(i int, result1 []*types.ICEConnectionInfo) {
fake.getICEConnectionInfoMutex.Lock()
defer fake.getICEConnectionInfoMutex.Unlock()
fake.GetICEConnectionInfoStub = nil
if fake.getICEConnectionInfoReturnsOnCall == nil {
fake.getICEConnectionInfoReturnsOnCall = make(map[int]struct {
result1 []*types.ICEConnectionInfo
})
}
fake.getICEConnectionDetailsReturnsOnCall[i] = struct {
result1 []*types.ICEConnectionDetails
fake.getICEConnectionInfoReturnsOnCall[i] = struct {
result1 []*types.ICEConnectionInfo
}{result1}
}
@@ -6969,8 +6969,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
defer fake.getEnabledPublishCodecsMutex.RUnlock()
fake.getICEConfigMutex.RLock()
defer fake.getICEConfigMutex.RUnlock()
fake.getICEConnectionDetailsMutex.RLock()
defer fake.getICEConnectionDetailsMutex.RUnlock()
fake.getICEConnectionInfoMutex.RLock()
defer fake.getICEConnectionInfoMutex.RUnlock()
fake.getLoggerMutex.RLock()
defer fake.getLoggerMutex.RUnlock()
fake.getPacerMutex.RLock()
pkg/sfu/forwarder.go
+3 -1
View File
@@ -196,7 +196,9 @@ type refInfo struct {
}
func (r refInfo) MarshalLogObject(e zapcore.ObjectEncoder) error {
e.AddObject("senderReport", buffer.WrappedRTCPSenderReportStateLogger{r.senderReport})
e.AddObject("senderReport", buffer.WrappedRTCPSenderReportStateLogger{
RTCPSenderReportState: r.senderReport,
})
e.AddUint64("tsOffset", r.tsOffset)
e.AddBool("isTSOffsetValid", r.isTSOffsetValid)
return nil