From 1b5bb4dddcac17de411303a3f792c7903a9b44ff Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Thu, 12 Sep 2024 13:16:41 +0530 Subject: [PATCH] Log ICE reconnected. (#2999) 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. --- pkg/rtc/room.go | 28 +++---- pkg/rtc/transport.go | 34 +++++++- pkg/rtc/transportmanager.go | 12 +-- pkg/rtc/types/ice.go | 44 +++++----- pkg/rtc/types/interfaces.go | 2 +- .../typesfakes/fake_local_participant.go | 80 +++++++++---------- pkg/sfu/forwarder.go | 4 +- 7 files changed, 120 insertions(+), 84 deletions(-) diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index acb15b81a..481744315 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -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) diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index a3785fac7..21b897d40 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -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 { diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index cc06c8b55..93dfc8b86 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -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 { diff --git a/pkg/rtc/types/ice.go b/pkg/rtc/types/ice.go index 7a6e0bcc4..3b02b4f42 100644 --- a/pkg/rtc/types/ice.go +++ b/pkg/rtc/types/ice.go @@ -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) { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index ad5294a80..31f7196c3 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -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 diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 3b067681f..afe375785 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -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() diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index de2e88690..e14509541 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -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