From 3fe124c87f58748b42e119bf76c221983673d686 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Sat, 2 Dec 2023 15:07:31 -0800 Subject: [PATCH] Log cleanup pass (#2285) * Log cleanup pass Demoted a bunch of logs to DEBUG, consolidated logs. * use context logger and fix context var usage * moved common error types, fixed tests --- go.mod | 2 +- go.sum | 4 +- pkg/routing/errors.go | 9 +- pkg/routing/interfaces.go | 9 +- pkg/routing/localrouter.go | 13 +- pkg/routing/redisrouter.go | 20 +- pkg/routing/routingfakes/fake_router.go | 48 +- pkg/rtc/participant.go | 6 +- pkg/rtc/participant_signal.go | 4 +- pkg/rtc/subscriptionmanager.go | 2 +- pkg/rtc/transport.go | 53 +- pkg/rtc/uptrackmanager.go | 8 +- pkg/service/agentservice.go | 35 +- pkg/service/auth.go | 8 +- pkg/service/roomservice.go | 6 +- pkg/service/rtcservice.go | 35 +- pkg/service/server.go | 3 +- pkg/service/servicefakes/fake_sipstore.go | 975 +++++++++++++++++++++ pkg/service/twirp.go | 32 +- pkg/service/utils.go | 5 +- pkg/sfu/buffer/buffer.go | 5 +- pkg/sfu/downtrack.go | 12 +- pkg/sfu/streamallocator/streamallocator.go | 2 +- pkg/utils/context.go | 24 +- 24 files changed, 1171 insertions(+), 149 deletions(-) create mode 100644 pkg/service/servicefakes/fake_sipstore.go diff --git a/go.mod b/go.mod index 8a05272f4..4e163b610 100644 --- a/go.mod +++ b/go.mod @@ -18,7 +18,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 github.com/livekit/mediatransportutil v0.0.0-20231128042044-05525c8278cb - github.com/livekit/protocol v1.9.3-0.20231130173607-ec88d89da1d3 + github.com/livekit/protocol v1.9.4-0.20231202181655-afa0350bcd0f github.com/livekit/psrpc v0.5.2 github.com/mackerelio/go-osstat v0.2.4 github.com/magefile/mage v1.15.0 diff --git a/go.sum b/go.sum index 5193e6f52..e60920f90 100644 --- a/go.sum +++ b/go.sum @@ -125,8 +125,8 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20231128042044-05525c8278cb h1:KiGg4k+kYQD9NjKixaSDMMeYOO2//XBM4IROTI1Itjo= github.com/livekit/mediatransportutil v0.0.0-20231128042044-05525c8278cb/go.mod h1:GBzn9xL+mivI1pW+tyExcKgbc0VOc29I9yJsNcAVaAc= -github.com/livekit/protocol v1.9.3-0.20231130173607-ec88d89da1d3 h1:am72beYtXZM71MRr+12lkG3IyqKxzrCa6slsbKrwMe8= -github.com/livekit/protocol v1.9.3-0.20231130173607-ec88d89da1d3/go.mod h1:8f342d5nvfNp9YAEfJokSR+zbNFpaivgU0h6vwaYhes= +github.com/livekit/protocol v1.9.4-0.20231202181655-afa0350bcd0f h1:6XPC53t/XEcfIe8BUwKkeFmgTLKPfF77JmQ7nydqoOs= +github.com/livekit/protocol v1.9.4-0.20231202181655-afa0350bcd0f/go.mod h1:8f342d5nvfNp9YAEfJokSR+zbNFpaivgU0h6vwaYhes= github.com/livekit/psrpc v0.5.2 h1:+MvG8Otm/J6MTg2MP/uuMbrkxOWsrj2hDhu/I1VIU1U= github.com/livekit/psrpc v0.5.2/go.mod h1:cQjxg1oCxYHhxxv6KJH1gSvdtCHQoRZCHgPdm5N8v2g= github.com/mackerelio/go-osstat v0.2.4 h1:qxGbdPkFo65PXOb/F/nhDKpF2nGmGaCFDLXoZjJTtUs= diff --git a/pkg/routing/errors.go b/pkg/routing/errors.go index 4b6af0686..28a0dd1ac 100644 --- a/pkg/routing/errors.go +++ b/pkg/routing/errors.go @@ -14,7 +14,9 @@ package routing -import "errors" +import ( + "errors" +) var ( ErrNotFound = errors.New("could not find object") @@ -26,4 +28,9 @@ var ( ErrInvalidRouterMessage = errors.New("invalid router message") ErrChannelClosed = errors.New("channel closed") ErrChannelFull = errors.New("channel is full") + + // errors when starting signal connection + ErrRequestChannelClosed = errors.New("request channel closed") + ErrCouldNotMigrateParticipant = errors.New("could not migrate participant") + ErrClientInfoNotSet = errors.New("client info not set") ) diff --git a/pkg/routing/interfaces.go b/pkg/routing/interfaces.go index 7a450bd16..9411ed2c3 100644 --- a/pkg/routing/interfaces.go +++ b/pkg/routing/interfaces.go @@ -107,9 +107,16 @@ type Router interface { OnRTCMessage(callback RTCMessageCallback) } +type StartParticipantSignalResults struct { + ConnectionID livekit.ConnectionID + RequestSink MessageSink + ResponseSource MessageSource + NodeID livekit.NodeID +} + type MessageRouter interface { // StartParticipantSignal participant signal connection is ready to start - StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (connectionID livekit.ConnectionID, reqSink MessageSink, resSource MessageSource, err error) + StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (res StartParticipantSignalResults, err error) // Write a message to a participant or room WriteParticipantRTC(ctx context.Context, roomName livekit.RoomName, identity livekit.ParticipantIdentity, msg *livekit.RTCNodeMessage) error diff --git a/pkg/routing/localrouter.go b/pkg/routing/localrouter.go index 01687ef3f..d83f5ea60 100644 --- a/pkg/routing/localrouter.go +++ b/pkg/routing/localrouter.go @@ -97,18 +97,25 @@ func (r *LocalRouter) ListNodes() ([]*livekit.Node, error) { }, nil } -func (r *LocalRouter) StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (connectionID livekit.ConnectionID, reqSink MessageSink, resSource MessageSource, err error) { +func (r *LocalRouter) StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (res StartParticipantSignalResults, err error) { return r.StartParticipantSignalWithNodeID(ctx, roomName, pi, livekit.NodeID(r.currentNode.Id)) } -func (r *LocalRouter) StartParticipantSignalWithNodeID(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit, nodeID livekit.NodeID) (connectionID livekit.ConnectionID, reqSink MessageSink, resSource MessageSource, err error) { - connectionID, reqSink, resSource, err = r.signalClient.StartParticipantSignal(ctx, roomName, pi, nodeID) +func (r *LocalRouter) StartParticipantSignalWithNodeID(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit, nodeID livekit.NodeID) (res StartParticipantSignalResults, err error) { + connectionID, reqSink, resSource, err := r.signalClient.StartParticipantSignal(ctx, roomName, pi, nodeID) if err != nil { logger.Errorw("could not handle new participant", err, "room", roomName, "participant", pi.Identity, "connID", connectionID, ) + } else { + return StartParticipantSignalResults{ + ConnectionID: connectionID, + RequestSink: reqSink, + ResponseSource: resSource, + NodeID: nodeID, + }, nil } return } diff --git a/pkg/routing/redisrouter.go b/pkg/routing/redisrouter.go index ace768a11..5a920b9c5 100644 --- a/pkg/routing/redisrouter.go +++ b/pkg/routing/redisrouter.go @@ -156,7 +156,7 @@ func (r *RedisRouter) ListNodes() ([]*livekit.Node, error) { } // StartParticipantSignal signal connection sets up paths to the RTC node, and starts to route messages to that message queue -func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (connectionID livekit.ConnectionID, reqSink MessageSink, resSource MessageSource, err error) { +func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (res StartParticipantSignalResults, err error) { // find the node where the room is hosted at rtcNode, err := r.GetNodeForRoom(ctx, roomName) if err != nil { @@ -164,33 +164,33 @@ func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livek } if r.usePSRPCSignal { - connectionID, reqSink, resSource, err = r.StartParticipantSignalWithNodeID(ctx, roomName, pi, livekit.NodeID(rtcNode.Id)) + res, err = r.StartParticipantSignalWithNodeID(ctx, roomName, pi, livekit.NodeID(rtcNode.Id)) if err != nil { return } // map signal & rtc nodes - err = r.setParticipantSignalNode(connectionID, r.currentNode.Id) + err = r.setParticipantSignalNode(res.ConnectionID, r.currentNode.Id) return } - connectionID = livekit.ConnectionID(utils.NewGuid("CO_")) + res.ConnectionID = livekit.ConnectionID(utils.NewGuid("CO_")) pKey := ParticipantKeyLegacy(roomName, pi.Identity) pKeyB62 := ParticipantKey(roomName, pi.Identity) // map signal & rtc nodes - if err = r.setParticipantSignalNode(connectionID, r.currentNode.Id); err != nil { + if err = r.setParticipantSignalNode(res.ConnectionID, r.currentNode.Id); err != nil { return } // index by connectionID, since there may be multiple connections for the participant // set up response channel before sending StartSession and be ready to receive responses. - resChan := r.getOrCreateMessageChannel(r.responseChannels, string(connectionID)) + resChan := r.getOrCreateMessageChannel(r.responseChannels, string(res.ConnectionID)) - sink := NewRTCNodeSink(r.rc, livekit.NodeID(rtcNode.Id), connectionID, pKey, pKeyB62) + sink := NewRTCNodeSink(r.rc, livekit.NodeID(rtcNode.Id), res.ConnectionID, pKey, pKeyB62) // serialize claims - ss, err := pi.ToStartSession(roomName, connectionID) + ss, err := pi.ToStartSession(roomName, res.ConnectionID) if err != nil { return } @@ -201,7 +201,9 @@ func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livek return } - return connectionID, sink, resChan, nil + res.RequestSink = sink + res.ResponseSource = resChan + return res, nil } func (r *RedisRouter) WriteParticipantRTC(_ context.Context, roomName livekit.RoomName, identity livekit.ParticipantIdentity, msg *livekit.RTCNodeMessage) error { diff --git a/pkg/routing/routingfakes/fake_router.go b/pkg/routing/routingfakes/fake_router.go index 190996099..b52f36bb0 100644 --- a/pkg/routing/routingfakes/fake_router.go +++ b/pkg/routing/routingfakes/fake_router.go @@ -115,7 +115,7 @@ type FakeRouter struct { startReturnsOnCall map[int]struct { result1 error } - StartParticipantSignalStub func(context.Context, livekit.RoomName, routing.ParticipantInit) (livekit.ConnectionID, routing.MessageSink, routing.MessageSource, error) + StartParticipantSignalStub func(context.Context, livekit.RoomName, routing.ParticipantInit) (routing.StartParticipantSignalResults, error) startParticipantSignalMutex sync.RWMutex startParticipantSignalArgsForCall []struct { arg1 context.Context @@ -123,16 +123,12 @@ type FakeRouter struct { arg3 routing.ParticipantInit } startParticipantSignalReturns struct { - result1 livekit.ConnectionID - result2 routing.MessageSink - result3 routing.MessageSource - result4 error + result1 routing.StartParticipantSignalResults + result2 error } startParticipantSignalReturnsOnCall map[int]struct { - result1 livekit.ConnectionID - result2 routing.MessageSink - result3 routing.MessageSource - result4 error + result1 routing.StartParticipantSignalResults + result2 error } StopStub func() stopMutex sync.RWMutex @@ -725,7 +721,7 @@ func (fake *FakeRouter) StartReturnsOnCall(i int, result1 error) { }{result1} } -func (fake *FakeRouter) StartParticipantSignal(arg1 context.Context, arg2 livekit.RoomName, arg3 routing.ParticipantInit) (livekit.ConnectionID, routing.MessageSink, routing.MessageSource, error) { +func (fake *FakeRouter) StartParticipantSignal(arg1 context.Context, arg2 livekit.RoomName, arg3 routing.ParticipantInit) (routing.StartParticipantSignalResults, error) { fake.startParticipantSignalMutex.Lock() ret, specificReturn := fake.startParticipantSignalReturnsOnCall[len(fake.startParticipantSignalArgsForCall)] fake.startParticipantSignalArgsForCall = append(fake.startParticipantSignalArgsForCall, struct { @@ -741,9 +737,9 @@ func (fake *FakeRouter) StartParticipantSignal(arg1 context.Context, arg2 liveki return stub(arg1, arg2, arg3) } if specificReturn { - return ret.result1, ret.result2, ret.result3, ret.result4 + return ret.result1, ret.result2 } - return fakeReturns.result1, fakeReturns.result2, fakeReturns.result3, fakeReturns.result4 + return fakeReturns.result1, fakeReturns.result2 } func (fake *FakeRouter) StartParticipantSignalCallCount() int { @@ -752,7 +748,7 @@ func (fake *FakeRouter) StartParticipantSignalCallCount() int { return len(fake.startParticipantSignalArgsForCall) } -func (fake *FakeRouter) StartParticipantSignalCalls(stub func(context.Context, livekit.RoomName, routing.ParticipantInit) (livekit.ConnectionID, routing.MessageSink, routing.MessageSource, error)) { +func (fake *FakeRouter) StartParticipantSignalCalls(stub func(context.Context, livekit.RoomName, routing.ParticipantInit) (routing.StartParticipantSignalResults, error)) { fake.startParticipantSignalMutex.Lock() defer fake.startParticipantSignalMutex.Unlock() fake.StartParticipantSignalStub = stub @@ -765,36 +761,30 @@ func (fake *FakeRouter) StartParticipantSignalArgsForCall(i int) (context.Contex return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 } -func (fake *FakeRouter) StartParticipantSignalReturns(result1 livekit.ConnectionID, result2 routing.MessageSink, result3 routing.MessageSource, result4 error) { +func (fake *FakeRouter) StartParticipantSignalReturns(result1 routing.StartParticipantSignalResults, result2 error) { fake.startParticipantSignalMutex.Lock() defer fake.startParticipantSignalMutex.Unlock() fake.StartParticipantSignalStub = nil fake.startParticipantSignalReturns = struct { - result1 livekit.ConnectionID - result2 routing.MessageSink - result3 routing.MessageSource - result4 error - }{result1, result2, result3, result4} + result1 routing.StartParticipantSignalResults + result2 error + }{result1, result2} } -func (fake *FakeRouter) StartParticipantSignalReturnsOnCall(i int, result1 livekit.ConnectionID, result2 routing.MessageSink, result3 routing.MessageSource, result4 error) { +func (fake *FakeRouter) StartParticipantSignalReturnsOnCall(i int, result1 routing.StartParticipantSignalResults, result2 error) { fake.startParticipantSignalMutex.Lock() defer fake.startParticipantSignalMutex.Unlock() fake.StartParticipantSignalStub = nil if fake.startParticipantSignalReturnsOnCall == nil { fake.startParticipantSignalReturnsOnCall = make(map[int]struct { - result1 livekit.ConnectionID - result2 routing.MessageSink - result3 routing.MessageSource - result4 error + result1 routing.StartParticipantSignalResults + result2 error }) } fake.startParticipantSignalReturnsOnCall[i] = struct { - result1 livekit.ConnectionID - result2 routing.MessageSink - result3 routing.MessageSource - result4 error - }{result1, result2, result3, result4} + result1 routing.StartParticipantSignalResults + result2 error + }{result1, result2} } func (fake *FakeRouter) Stop() { diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 46acee7e0..3f82ec85b 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -584,7 +584,6 @@ func (p *ParticipantImpl) HandleSignalSourceClose() { if !p.TransportManager.HasPublisherEverConnected() && !p.TransportManager.HasSubscriberEverConnected() { reason := types.ParticipantCloseReasonJoinFailed - p.params.Logger.Infow("closing disconnected participant", "reason", reason) _ = p.Close(false, reason, false) } } @@ -858,7 +857,7 @@ func (p *ParticipantImpl) MaybeStartMigration(force bool, onStart func()) bool { if p.IsClosed() || p.IsDisconnected() { return } - p.subLogger.Infow("closing subscriber peer connection to aid migration") + p.subLogger.Debugw("closing subscriber peer connection to aid migration") // // Close all down tracks before closing subscriber peer connection. @@ -1444,7 +1443,6 @@ func (p *ParticipantImpl) setupDisconnectTimer() { return } reason := types.ParticipantCloseReasonPeerConnectionDisconnected - p.params.Logger.Infow("closing disconnected participant", "reason", reason) _ = p.Close(true, reason, false) }) p.lock.Unlock() @@ -1693,7 +1691,7 @@ func (p *ParticipantImpl) addPendingTrackLocked(req *livekit.AddTrackRequest) *l } p.pendingTracks[req.Cid] = &pendingTrackInfo{trackInfos: []*livekit.TrackInfo{ti}} - p.pubLogger.Infow("pending track added", "trackID", ti.Sid, "track", logger.Proto(ti), "request", logger.Proto(req)) + p.pubLogger.Debugw("pending track added", "trackID", ti.Sid, "track", logger.Proto(ti), "request", logger.Proto(req)) return ti } diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 5214df47e..863fd67d6 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -289,7 +289,9 @@ func (p *ParticipantImpl) writeMessage(msg *livekit.SignalResponse) error { func (p *ParticipantImpl) CloseSignalConnection(reason types.SignallingCloseReason) { sink := p.getResponseSink() if sink != nil { - p.params.Logger.Infow("closing signal connection", "reason", reason, "connID", sink.ConnectionID()) + if reason != types.SignallingCloseReasonParticipantClose { + p.params.Logger.Infow("closing signal connection", "reason", reason, "connID", sink.ConnectionID()) + } sink.Close() p.SetResponseSink(nil) } diff --git a/pkg/rtc/subscriptionmanager.go b/pkg/rtc/subscriptionmanager.go index 602e786fd..eec97e026 100644 --- a/pkg/rtc/subscriptionmanager.go +++ b/pkg/rtc/subscriptionmanager.go @@ -924,7 +924,7 @@ func (s *trackSubscription) handleSourceTrackRemoved() { } // source track removed, we would unsubscribe - s.logger.Infow("unsubscribing from track since source track was removed") + s.logger.Debugw("unsubscribing from track since source track was removed") s.desired = false s.setChangedNotifierLocked(nil) diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index b72b1411e..4230b9dfd 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -466,10 +466,10 @@ func (t *PCTransport) setICEStartedAt(at time.Time) { } else if tcpICETimeout > maxTcpICEConnectTimeout { tcpICETimeout = maxTcpICEConnectTimeout } - t.params.Logger.Debugw("set tcp ice connect timer", "timeout", tcpICETimeout, "signalRTT", signalingRTT) + t.params.Logger.Debugw("set TCP ICE connect timer", "timeout", tcpICETimeout, "signalRTT", signalingRTT) t.tcpICETimer = time.AfterFunc(tcpICETimeout, func() { if t.pc.ICEConnectionState() == webrtc.ICEConnectionStateChecking { - t.params.Logger.Infow("tcp ice connect timeout", "timeout", tcpICETimeout, "signalRTT", signalingRTT) + t.params.Logger.Infow("TCP ICE connect timeout", "timeout", tcpICETimeout, "signalRTT", signalingRTT) t.handleConnectionFailed(true) } }) @@ -497,7 +497,7 @@ func (t *PCTransport) setICEConnectedAt(at time.Time) { if connTimeoutAfterICE > maxConnectTimeoutAfterICE { connTimeoutAfterICE = maxConnectTimeoutAfterICE } - t.params.Logger.Debugw("setting connection timer after ice connected", "timeout", connTimeoutAfterICE, "iceDuration", iceDuration) + t.params.Logger.Debugw("setting connection timer after ICE connected", "timeout", connTimeoutAfterICE, "iceDuration", iceDuration) t.connectAfterICETimer = time.AfterFunc(connTimeoutAfterICE, func() { state := t.pc.ConnectionState() // if pc is still checking or connected but not fully established after timeout, then fire connection fail @@ -546,12 +546,12 @@ func (t *PCTransport) IsShortConnection(at time.Time) (bool, time.Duration) { } func (t *PCTransport) getSelectedPair() (*webrtc.ICECandidatePair, error) { - sctp := t.pc.SCTP() - if sctp == nil { + s := t.pc.SCTP() + if s == nil { return nil, errors.New("no SCTP") } - dtlsTransport := sctp.Transport() + dtlsTransport := s.Transport() if dtlsTransport == nil { return nil, errors.New("no DTLS transport") } @@ -629,11 +629,6 @@ func (t *PCTransport) onICEConnectionStateChange(state webrtc.ICEConnectionState switch state { case webrtc.ICEConnectionStateConnected: t.setICEConnectedAt(time.Now()) - if pair, err := t.getSelectedPair(); err != nil { - t.params.Logger.Errorw("error getting selected ICE candidate pair", err) - } else { - t.params.Logger.Infow("selected ICE candidate pair", "pair", pair) - } case webrtc.ICEConnectionStateChecking: t.setICEStartedAt(time.Now()) @@ -1268,7 +1263,7 @@ func (t *PCTransport) preparePC(previousAnswer webrtc.SessionDescription) error // trying to replicate previous setup, read from previous answer and use that role. // se := webrtc.SettingEngine{} - se.SetAnsweringDTLSRole(lksdp.ExtractDTLSRole(parsed)) + _ = se.SetAnsweringDTLSRole(lksdp.ExtractDTLSRole(parsed)) api := webrtc.NewAPI( webrtc.WithSettingEngine(se), webrtc.WithMediaEngine(t.me), @@ -1446,9 +1441,11 @@ func (t *PCTransport) processEvents() { err := t.handleEvent(&event) if err != nil { - t.params.Logger.Errorw("error handling event", err, "event", event.String()) - if onNegotiationFailed := t.getOnNegotiationFailed(); onNegotiationFailed != nil { - onNegotiationFailed() + if !t.isClosed.Load() { + t.params.Logger.Errorw("error handling event", err, "event", event.String()) + if onNegotiationFailed := t.getOnNegotiationFailed(); onNegotiationFailed != nil { + onNegotiationFailed() + } } break } @@ -1480,7 +1477,7 @@ func (t *PCTransport) handleEvent(e *event) error { return nil } -func (t *PCTransport) handleICEGatheringComplete(e *event) error { +func (t *PCTransport) handleICEGatheringComplete(_ *event) error { if t.params.IsOfferer { return t.handleICEGatheringCompleteOfferer() } else { @@ -1609,17 +1606,25 @@ func (t *PCTransport) handleRemoteICECandidate(e *event) error { return nil } -func (t *PCTransport) handleLogICECandidates(e *event) error { +func (t *PCTransport) handleLogICECandidates(_ *event) error { lc := t.allowedLocalCandidates.Get() rc := t.allowedRemoteCandidates.Get() + var fields []interface{} if len(lc) != 0 || len(rc) != 0 { - t.params.Logger.Infow( - "ice candidates", + fields = append(fields, "lc", lc, "rc", rc, - "lc (filtered)", t.filteredLocalCandidates.Get(), - "rc (filtered)", t.filteredRemoteCandidates.Get(), + "lc_filtered", t.filteredLocalCandidates.Get(), + "rc_filtered", t.filteredRemoteCandidates.Get(), ) + + } + if pair, err := t.getSelectedPair(); err == nil { + fields = append(fields, "selected_pair", pair) + } + + if len(fields) > 0 { + t.params.Logger.Infow("ice candidates", fields...) } return nil @@ -1711,7 +1716,7 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error { // when there's an ongoing negotiation, let it finish and not disrupt its state if t.negotiationState == NegotiationStateRemote { - t.params.Logger.Infow("skipping negotiation, trying again later") + t.params.Logger.Debugw("skipping negotiation, trying again later") t.setNegotiationState(NegotiationStateRetry) return nil } else if t.negotiationState == NegotiationStateRetry { @@ -1801,7 +1806,7 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error { return ErrNoOfferHandler } -func (t *PCTransport) handleSendOffer(e *event) error { +func (t *PCTransport) handleSendOffer(_ *event) error { return t.createAndSendOffer(nil) } @@ -2030,7 +2035,7 @@ func (t *PCTransport) doICERestart() error { } } -func (t *PCTransport) handleICERestart(e *event) error { +func (t *PCTransport) handleICERestart(_ *event) error { return t.doICERestart() } diff --git a/pkg/rtc/uptrackmanager.go b/pkg/rtc/uptrackmanager.go index 1c079dc34..ce5177661 100644 --- a/pkg/rtc/uptrackmanager.go +++ b/pkg/rtc/uptrackmanager.go @@ -111,7 +111,7 @@ func (u *UpTrackManager) SetPublishedTrackMuted(trackID livekit.TrackID, muted b track.SetMuted(muted) if currentMuted != track.IsMuted() { - u.params.Logger.Infow("publisher mute status changed", "trackID", trackID, "muted", track.IsMuted()) + u.params.Logger.Debugw("publisher mute status changed", "trackID", trackID, "muted", track.IsMuted()) if u.onTrackUpdated != nil { u.onTrackUpdated(track) } @@ -142,7 +142,7 @@ func (u *UpTrackManager) GetPublishedTracks() []types.MediaTrack { func (u *UpTrackManager) UpdateSubscriptionPermission( subscriptionPermission *livekit.SubscriptionPermission, timedVersion utils.TimedVersion, - resolverByIdentity func(participantIdentity livekit.ParticipantIdentity) types.LocalParticipant, + _ func(participantIdentity livekit.ParticipantIdentity) types.LocalParticipant, // TODO: separate PR to remove this argument resolverBySid func(participantID livekit.ParticipantID) types.LocalParticipant, ) error { u.lock.Lock() @@ -203,7 +203,7 @@ func (u *UpTrackManager) UpdateSubscriptionPermission( } u.lock.Unlock() - u.maybeRevokeSubscriptions(resolverByIdentity) + u.maybeRevokeSubscriptions() return nil } @@ -381,7 +381,7 @@ func (u *UpTrackManager) getAllowedSubscribersLocked(trackID livekit.TrackID) [] return allowed } -func (u *UpTrackManager) maybeRevokeSubscriptions(resolver func(participantIdentity livekit.ParticipantIdentity) types.LocalParticipant) { +func (u *UpTrackManager) maybeRevokeSubscriptions() { u.lock.Lock() defer u.lock.Unlock() diff --git a/pkg/service/agentservice.go b/pkg/service/agentservice.go index f2ba672b2..9e7eef0e8 100644 --- a/pkg/service/agentservice.go +++ b/pkg/service/agentservice.go @@ -28,6 +28,7 @@ import ( "google.golang.org/protobuf/types/known/emptypb" "github.com/livekit/livekit-server/pkg/rtc" + "github.com/livekit/livekit-server/pkg/utils" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/rpc" @@ -65,6 +66,7 @@ type worker struct { jobType livekit.JobType status livekit.WorkerStatus activeJobs int + logger logger.Logger } type availability struct { @@ -102,18 +104,18 @@ func (s *AgentService) ServeHTTP(writer http.ResponseWriter, r *http.Request) { // require a claim claims := GetGrants(r.Context()) if claims == nil || claims.Video == nil || !claims.Video.Agent { - handleError(writer, http.StatusUnauthorized, rtc.ErrPermissionDenied) + handleError(writer, r, http.StatusUnauthorized, rtc.ErrPermissionDenied) return } // upgrade conn, err := s.upgrader.Upgrade(writer, r, nil) if err != nil { - handleError(writer, http.StatusInternalServerError, err) + handleError(writer, r, http.StatusInternalServerError, err) return } - s.HandleConnection(conn) + s.HandleConnection(r.Context(), conn) } func NewAgentHandler(agentServer rpc.AgentInternalServer, roomTopic, publisherTopic string) *AgentHandler { @@ -128,11 +130,12 @@ func NewAgentHandler(agentServer rpc.AgentInternalServer, roomTopic, publisherTo } } -func (s *AgentHandler) HandleConnection(conn *websocket.Conn) { +func (s *AgentHandler) HandleConnection(ctx context.Context, conn *websocket.Conn) { sigConn := NewWSSignalConnection(conn) w := &worker{ conn: conn, sigConn: sigConn, + logger: utils.GetLogger(ctx), } s.mu.Lock() @@ -177,9 +180,9 @@ func (s *AgentHandler) HandleConnection(conn *websocket.Conn) { websocket.CloseNormalClosure, websocket.CloseNoStatusReceived, ) { - logger.Infow("exit ws read loop for closed connection", "wsError", err) + w.logger.Infow("Agent worker closed WS connection", "wsError", err) } else { - logger.Errorw("error reading from websocket", err) + w.logger.Errorw("error reading from websocket", err) } return } @@ -199,7 +202,7 @@ func (s *AgentHandler) HandleConnection(conn *websocket.Conn) { func (s *AgentHandler) handleRegister(worker *worker, msg *livekit.RegisterWorkerRequest) { if err := s.doHandleRegister(worker, msg); err != nil { - logger.Errorw("failed to register worker", err, "workerID", msg.WorkerId, "jobType", msg.Type) + worker.logger.Errorw("failed to register worker", err, "workerID", msg.WorkerId, "jobType", msg.Type) worker.conn.Close() } } @@ -225,7 +228,7 @@ func (s *AgentHandler) doHandleRegister(worker *worker, msg *livekit.RegisterWor if !s.roomRegistered { err := s.agentServer.RegisterJobRequestTopic(s.roomTopic) if err != nil { - logger.Errorw("failed to register room agents", err) + worker.logger.Errorw("failed to register room agents", err) } else { s.roomRegistered = true } @@ -240,7 +243,7 @@ func (s *AgentHandler) doHandleRegister(worker *worker, msg *livekit.RegisterWor if !s.publisherRegistered { err := s.agentServer.RegisterJobRequestTopic(s.publisherTopic) if err != nil { - logger.Errorw("failed to register publisher agents", err) + worker.logger.Errorw("failed to register publisher agents", err) } else { s.publisherRegistered = true } @@ -260,7 +263,7 @@ func (s *AgentHandler) doHandleRegister(worker *worker, msg *livekit.RegisterWor }, }) if err != nil { - logger.Errorw("failed to write server message", err) + worker.logger.Errorw("failed to write server message", err) } return nil @@ -282,9 +285,9 @@ func (s *AgentHandler) handleAvailability(w *worker, msg *livekit.AvailabilityRe func (s *AgentHandler) handleJobUpdate(w *worker, msg *livekit.JobStatusUpdate) { switch msg.Status { case livekit.JobStatus_JS_SUCCESS: - logger.Debugw("job complete", "jobID", msg.JobId) + w.logger.Debugw("job complete", "jobID", msg.JobId) case livekit.JobStatus_JS_FAILED: - logger.Warnw("job failed", errors.New(msg.Error), "jobID", msg.JobId) + w.logger.Warnw("job failed", errors.New(msg.Error), "jobID", msg.JobId) } w.mu.Lock() @@ -307,7 +310,7 @@ func (s *AgentHandler) handleStatus(w *worker, msg *livekit.UpdateWorkerStatus) s.agentServer.DeregisterJobRequestTopic(s.roomTopic) } else if !s.roomRegistered && s.roomAvailableLocked() { if err := s.agentServer.RegisterJobRequestTopic(s.roomTopic); err != nil { - logger.Errorw("failed to register room agents", err) + w.logger.Errorw("failed to register room agents", err) } else { s.roomRegistered = true } @@ -318,7 +321,7 @@ func (s *AgentHandler) handleStatus(w *worker, msg *livekit.UpdateWorkerStatus) s.agentServer.DeregisterJobRequestTopic(s.publisherTopic) } else if !s.publisherRegistered && s.publisherAvailableLocked() { if err := s.agentServer.RegisterJobRequestTopic(s.publisherTopic); err != nil { - logger.Errorw("failed to register publisher agents", err) + w.logger.Errorw("failed to register publisher agents", err) } else { s.publisherRegistered = true } @@ -388,7 +391,7 @@ func (s *AgentHandler) JobRequest(ctx context.Context, job *livekit.Job) (*empty Availability: &livekit.AvailabilityRequest{Job: job}, }}) if err != nil { - logger.Errorw("failed to send availability request", err, "workerID", selected.id) + selected.logger.Errorw("failed to send availability request", err, "workerID", selected.id) } select { @@ -400,7 +403,7 @@ func (s *AgentHandler) JobRequest(ctx context.Context, job *livekit.Job) (*empty Assignment: &livekit.JobAssignment{Job: job}, }}) if err != nil { - logger.Errorw("failed to assign job", err, "workerID", selected.id) + selected.logger.Errorw("failed to assign job", err, "workerID", selected.id) } else { selected.mu.Lock() selected.activeJobs++ diff --git a/pkg/service/auth.go b/pkg/service/auth.go index 1b5829e30..5633b2889 100644 --- a/pkg/service/auth.go +++ b/pkg/service/auth.go @@ -62,7 +62,7 @@ func (m *APIKeyAuthMiddleware) ServeHTTP(w http.ResponseWriter, r *http.Request, if authHeader != "" { if !strings.HasPrefix(authHeader, bearerPrefix) { - handleError(w, http.StatusUnauthorized, ErrMissingAuthorization) + handleError(w, r, http.StatusUnauthorized, ErrMissingAuthorization) return } @@ -75,19 +75,19 @@ func (m *APIKeyAuthMiddleware) ServeHTTP(w http.ResponseWriter, r *http.Request, if authToken != "" { v, err := auth.ParseAPIToken(authToken) if err != nil { - handleError(w, http.StatusUnauthorized, ErrInvalidAuthorizationToken) + handleError(w, r, http.StatusUnauthorized, ErrInvalidAuthorizationToken) return } secret := m.provider.GetSecret(v.APIKey()) if secret == "" { - handleError(w, http.StatusUnauthorized, errors.New("invalid API key: "+v.APIKey())) + handleError(w, r, http.StatusUnauthorized, errors.New("invalid API key: "+v.APIKey())) return } grants, err := v.Verify(secret) if err != nil { - handleError(w, http.StatusUnauthorized, errors.New("invalid token: "+authToken+", error: "+err.Error())) + handleError(w, r, http.StatusUnauthorized, errors.New("invalid token: "+authToken+", error: "+err.Error())) return } diff --git a/pkg/service/roomservice.go b/pkg/service/roomservice.go index 174ea3391..2076b5ca4 100644 --- a/pkg/service/roomservice.go +++ b/pkg/service/roomservice.go @@ -93,15 +93,15 @@ func (s *RoomService) CreateRoom(ctx context.Context, req *livekit.CreateRoomReq } // actually start the room on an RTC node, to ensure metadata & empty timeout functionality - _, sink, source, err := s.router.StartParticipantSignal(ctx, + res, err := s.router.StartParticipantSignal(ctx, livekit.RoomName(req.Name), routing.ParticipantInit{}, ) if err != nil { return nil, err } - defer sink.Close() - defer source.Close() + defer res.RequestSink.Close() + defer res.ResponseSource.Close() // ensure it's created correctly err = s.confirmExecution(func() error { diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index 86ff5f630..a273edecf 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -29,6 +29,7 @@ import ( "github.com/gorilla/websocket" "github.com/ua-parser/uap-go/uaparser" + "go.uber.org/atomic" "golang.org/x/exp/maps" "github.com/livekit/livekit-server/pkg/config" @@ -39,7 +40,6 @@ import ( "github.com/livekit/livekit-server/pkg/telemetry/prometheus" "github.com/livekit/livekit-server/pkg/utils" "github.com/livekit/protocol/livekit" - "github.com/livekit/protocol/logger" putil "github.com/livekit/protocol/utils" "github.com/livekit/psrpc" ) @@ -97,7 +97,7 @@ func NewRTCService( func (s *RTCService) Validate(w http.ResponseWriter, r *http.Request) { _, _, code, err := s.validate(r) if err != nil { - handleError(w, code, err) + handleError(w, r, code, err) return } _, _ = w.Write([]byte("success")) @@ -202,7 +202,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { roomName, pi, code, err := s.validate(r) if err != nil { - handleError(w, code, err) + handleError(w, r, code, err) return } @@ -213,6 +213,8 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { "remote", false, } + l := utils.GetLogger(r.Context()) + // give it a few attempts to start session var cr connectionResult var initialResponse *livekit.SignalResponse @@ -229,13 +231,13 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { } if i < 2 { fieldsWithAttempt := append(loggerFields, "attempt", i) - logger.Warnw("failed to start connection, retrying", err, fieldsWithAttempt...) + l.Warnw("failed to start connection, retrying", err, fieldsWithAttempt...) } } if err != nil { prometheus.IncrementParticipantJoinFail(1) - handleError(w, http.StatusInternalServerError, err, loggerFields...) + handleError(w, r, http.StatusInternalServerError, err, loggerFields...) return } @@ -254,16 +256,20 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { } pLogger := rtc.LoggerWithParticipant( - rtc.LoggerWithRoom(logger.GetLogger(), roomName, livekit.RoomID(cr.Room.Sid)), + rtc.LoggerWithRoom(l, roomName, livekit.RoomID(cr.Room.Sid)), pi.Identity, pi.ID, false, ) + closedByClient := atomic.NewBool(false) done := make(chan struct{}) // function exits when websocket terminates, it'll close the event reading off of request sink and response source as well defer func() { - pLogger.Infow("finishing WS connection", "connID", cr.ConnectionID) + pLogger.Infow("finishing WS connection", + "connID", cr.ConnectionID, + "closedByClient", closedByClient.Load(), + ) cr.ResponseSource.Close() cr.RequestSink.Close() close(done) @@ -276,7 +282,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { // upgrade only once the basics are good to go conn, err := s.upgrader.Upgrade(w, r, nil) if err != nil { - handleError(w, http.StatusInternalServerError, err, loggerFields...) + handleError(w, r, http.StatusInternalServerError, err, loggerFields...) return } @@ -305,6 +311,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { "reconnect", pi.Reconnect, "reconnectReason", pi.ReconnectReason, "adaptiveStream", pi.AdaptiveStream, + "selectedNodeID", cr.NodeID, ) // handle responses @@ -366,7 +373,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { req, count, err := sigConn.ReadRequest() if err != nil { // normal/expected closure - if err == io.EOF || + if errors.Is(err, io.EOF) || strings.HasSuffix(err.Error(), "use of closed network connection") || strings.HasSuffix(err.Error(), "connection reset by peer") || websocket.IsCloseError( @@ -376,7 +383,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { websocket.CloseNormalClosure, websocket.CloseNoStatusReceived, ) { - pLogger.Infow("exit ws read loop for closed connection", "connID", cr.ConnectionID, "wsError", err) + closedByClient.Store(true) } else { pLogger.Errorw("error reading from websocket", err, "connID", cr.ConnectionID) } @@ -512,10 +519,8 @@ func (s *RTCService) DrainConnections(interval time.Duration) { } type connectionResult struct { - Room *livekit.Room - ConnectionID livekit.ConnectionID - RequestSink routing.MessageSink - ResponseSource routing.MessageSource + routing.StartParticipantSignalResults + Room *livekit.Room } func (s *RTCService) startConnection( @@ -533,7 +538,7 @@ func (s *RTCService) startConnection( } // this needs to be started first *before* using router functions on this node - cr.ConnectionID, cr.RequestSink, cr.ResponseSource, err = s.router.StartParticipantSignal(ctx, roomName, pi) + cr.StartParticipantSignalResults, err = s.router.StartParticipantSignal(ctx, roomName, pi) if err != nil { return cr, nil, err } diff --git a/pkg/service/server.go b/pkg/service/server.go index f66d848b5..fcb09a648 100644 --- a/pkg/service/server.go +++ b/pkg/service/server.go @@ -37,7 +37,6 @@ import ( "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/routing" - sutils "github.com/livekit/livekit-server/pkg/utils" "github.com/livekit/livekit-server/version" "github.com/livekit/protocol/auth" "github.com/livekit/protocol/livekit" @@ -107,7 +106,7 @@ func NewLivekitServer(conf *config.Config, middlewares = append(middlewares, NewAPIKeyAuthMiddleware(keyProvider)) } - twirpLoggingHook := TwirpLogger(logger.GetLogger().WithComponent(sutils.ComponentAPI)) + twirpLoggingHook := TwirpLogger() twirpRequestStatusHook := TwirpRequestStatusReporter() roomServer := livekit.NewRoomServiceServer(roomService, twirpLoggingHook) egressServer := livekit.NewEgressServer(egressService, twirp.WithServerHooks( diff --git a/pkg/service/servicefakes/fake_sipstore.go b/pkg/service/servicefakes/fake_sipstore.go new file mode 100644 index 000000000..069eff951 --- /dev/null +++ b/pkg/service/servicefakes/fake_sipstore.go @@ -0,0 +1,975 @@ +// Code generated by counterfeiter. DO NOT EDIT. +package servicefakes + +import ( + "context" + "sync" + + "github.com/livekit/livekit-server/pkg/service" + "github.com/livekit/protocol/livekit" +) + +type FakeSIPStore struct { + DeleteSIPDispatchRuleStub func(context.Context, *livekit.SIPDispatchRuleInfo) error + deleteSIPDispatchRuleMutex sync.RWMutex + deleteSIPDispatchRuleArgsForCall []struct { + arg1 context.Context + arg2 *livekit.SIPDispatchRuleInfo + } + deleteSIPDispatchRuleReturns struct { + result1 error + } + deleteSIPDispatchRuleReturnsOnCall map[int]struct { + result1 error + } + DeleteSIPParticipantStub func(context.Context, *livekit.SIPParticipantInfo) error + deleteSIPParticipantMutex sync.RWMutex + deleteSIPParticipantArgsForCall []struct { + arg1 context.Context + arg2 *livekit.SIPParticipantInfo + } + deleteSIPParticipantReturns struct { + result1 error + } + deleteSIPParticipantReturnsOnCall map[int]struct { + result1 error + } + DeleteSIPTrunkStub func(context.Context, *livekit.SIPTrunkInfo) error + deleteSIPTrunkMutex sync.RWMutex + deleteSIPTrunkArgsForCall []struct { + arg1 context.Context + arg2 *livekit.SIPTrunkInfo + } + deleteSIPTrunkReturns struct { + result1 error + } + deleteSIPTrunkReturnsOnCall map[int]struct { + result1 error + } + ListSIPDispatchRuleStub func(context.Context) ([]*livekit.SIPDispatchRuleInfo, error) + listSIPDispatchRuleMutex sync.RWMutex + listSIPDispatchRuleArgsForCall []struct { + arg1 context.Context + } + listSIPDispatchRuleReturns struct { + result1 []*livekit.SIPDispatchRuleInfo + result2 error + } + listSIPDispatchRuleReturnsOnCall map[int]struct { + result1 []*livekit.SIPDispatchRuleInfo + result2 error + } + ListSIPParticipantStub func(context.Context) ([]*livekit.SIPParticipantInfo, error) + listSIPParticipantMutex sync.RWMutex + listSIPParticipantArgsForCall []struct { + arg1 context.Context + } + listSIPParticipantReturns struct { + result1 []*livekit.SIPParticipantInfo + result2 error + } + listSIPParticipantReturnsOnCall map[int]struct { + result1 []*livekit.SIPParticipantInfo + result2 error + } + ListSIPTrunkStub func(context.Context) ([]*livekit.SIPTrunkInfo, error) + listSIPTrunkMutex sync.RWMutex + listSIPTrunkArgsForCall []struct { + arg1 context.Context + } + listSIPTrunkReturns struct { + result1 []*livekit.SIPTrunkInfo + result2 error + } + listSIPTrunkReturnsOnCall map[int]struct { + result1 []*livekit.SIPTrunkInfo + result2 error + } + LoadSIPDispatchRuleStub func(context.Context, string) (*livekit.SIPDispatchRuleInfo, error) + loadSIPDispatchRuleMutex sync.RWMutex + loadSIPDispatchRuleArgsForCall []struct { + arg1 context.Context + arg2 string + } + loadSIPDispatchRuleReturns struct { + result1 *livekit.SIPDispatchRuleInfo + result2 error + } + loadSIPDispatchRuleReturnsOnCall map[int]struct { + result1 *livekit.SIPDispatchRuleInfo + result2 error + } + LoadSIPParticipantStub func(context.Context, string) (*livekit.SIPParticipantInfo, error) + loadSIPParticipantMutex sync.RWMutex + loadSIPParticipantArgsForCall []struct { + arg1 context.Context + arg2 string + } + loadSIPParticipantReturns struct { + result1 *livekit.SIPParticipantInfo + result2 error + } + loadSIPParticipantReturnsOnCall map[int]struct { + result1 *livekit.SIPParticipantInfo + result2 error + } + LoadSIPTrunkStub func(context.Context, string) (*livekit.SIPTrunkInfo, error) + loadSIPTrunkMutex sync.RWMutex + loadSIPTrunkArgsForCall []struct { + arg1 context.Context + arg2 string + } + loadSIPTrunkReturns struct { + result1 *livekit.SIPTrunkInfo + result2 error + } + loadSIPTrunkReturnsOnCall map[int]struct { + result1 *livekit.SIPTrunkInfo + result2 error + } + StoreSIPDispatchRuleStub func(context.Context, *livekit.SIPDispatchRuleInfo) error + storeSIPDispatchRuleMutex sync.RWMutex + storeSIPDispatchRuleArgsForCall []struct { + arg1 context.Context + arg2 *livekit.SIPDispatchRuleInfo + } + storeSIPDispatchRuleReturns struct { + result1 error + } + storeSIPDispatchRuleReturnsOnCall map[int]struct { + result1 error + } + StoreSIPParticipantStub func(context.Context, *livekit.SIPParticipantInfo) error + storeSIPParticipantMutex sync.RWMutex + storeSIPParticipantArgsForCall []struct { + arg1 context.Context + arg2 *livekit.SIPParticipantInfo + } + storeSIPParticipantReturns struct { + result1 error + } + storeSIPParticipantReturnsOnCall map[int]struct { + result1 error + } + StoreSIPTrunkStub func(context.Context, *livekit.SIPTrunkInfo) error + storeSIPTrunkMutex sync.RWMutex + storeSIPTrunkArgsForCall []struct { + arg1 context.Context + arg2 *livekit.SIPTrunkInfo + } + storeSIPTrunkReturns struct { + result1 error + } + storeSIPTrunkReturnsOnCall map[int]struct { + result1 error + } + invocations map[string][][]interface{} + invocationsMutex sync.RWMutex +} + +func (fake *FakeSIPStore) DeleteSIPDispatchRule(arg1 context.Context, arg2 *livekit.SIPDispatchRuleInfo) error { + fake.deleteSIPDispatchRuleMutex.Lock() + ret, specificReturn := fake.deleteSIPDispatchRuleReturnsOnCall[len(fake.deleteSIPDispatchRuleArgsForCall)] + fake.deleteSIPDispatchRuleArgsForCall = append(fake.deleteSIPDispatchRuleArgsForCall, struct { + arg1 context.Context + arg2 *livekit.SIPDispatchRuleInfo + }{arg1, arg2}) + stub := fake.DeleteSIPDispatchRuleStub + fakeReturns := fake.deleteSIPDispatchRuleReturns + fake.recordInvocation("DeleteSIPDispatchRule", []interface{}{arg1, arg2}) + fake.deleteSIPDispatchRuleMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeSIPStore) DeleteSIPDispatchRuleCallCount() int { + fake.deleteSIPDispatchRuleMutex.RLock() + defer fake.deleteSIPDispatchRuleMutex.RUnlock() + return len(fake.deleteSIPDispatchRuleArgsForCall) +} + +func (fake *FakeSIPStore) DeleteSIPDispatchRuleCalls(stub func(context.Context, *livekit.SIPDispatchRuleInfo) error) { + fake.deleteSIPDispatchRuleMutex.Lock() + defer fake.deleteSIPDispatchRuleMutex.Unlock() + fake.DeleteSIPDispatchRuleStub = stub +} + +func (fake *FakeSIPStore) DeleteSIPDispatchRuleArgsForCall(i int) (context.Context, *livekit.SIPDispatchRuleInfo) { + fake.deleteSIPDispatchRuleMutex.RLock() + defer fake.deleteSIPDispatchRuleMutex.RUnlock() + argsForCall := fake.deleteSIPDispatchRuleArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeSIPStore) DeleteSIPDispatchRuleReturns(result1 error) { + fake.deleteSIPDispatchRuleMutex.Lock() + defer fake.deleteSIPDispatchRuleMutex.Unlock() + fake.DeleteSIPDispatchRuleStub = nil + fake.deleteSIPDispatchRuleReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeSIPStore) DeleteSIPDispatchRuleReturnsOnCall(i int, result1 error) { + fake.deleteSIPDispatchRuleMutex.Lock() + defer fake.deleteSIPDispatchRuleMutex.Unlock() + fake.DeleteSIPDispatchRuleStub = nil + if fake.deleteSIPDispatchRuleReturnsOnCall == nil { + fake.deleteSIPDispatchRuleReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.deleteSIPDispatchRuleReturnsOnCall[i] = struct { + result1 error + }{result1} +} + +func (fake *FakeSIPStore) DeleteSIPParticipant(arg1 context.Context, arg2 *livekit.SIPParticipantInfo) error { + fake.deleteSIPParticipantMutex.Lock() + ret, specificReturn := fake.deleteSIPParticipantReturnsOnCall[len(fake.deleteSIPParticipantArgsForCall)] + fake.deleteSIPParticipantArgsForCall = append(fake.deleteSIPParticipantArgsForCall, struct { + arg1 context.Context + arg2 *livekit.SIPParticipantInfo + }{arg1, arg2}) + stub := fake.DeleteSIPParticipantStub + fakeReturns := fake.deleteSIPParticipantReturns + fake.recordInvocation("DeleteSIPParticipant", []interface{}{arg1, arg2}) + fake.deleteSIPParticipantMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeSIPStore) DeleteSIPParticipantCallCount() int { + fake.deleteSIPParticipantMutex.RLock() + defer fake.deleteSIPParticipantMutex.RUnlock() + return len(fake.deleteSIPParticipantArgsForCall) +} + +func (fake *FakeSIPStore) DeleteSIPParticipantCalls(stub func(context.Context, *livekit.SIPParticipantInfo) error) { + fake.deleteSIPParticipantMutex.Lock() + defer fake.deleteSIPParticipantMutex.Unlock() + fake.DeleteSIPParticipantStub = stub +} + +func (fake *FakeSIPStore) DeleteSIPParticipantArgsForCall(i int) (context.Context, *livekit.SIPParticipantInfo) { + fake.deleteSIPParticipantMutex.RLock() + defer fake.deleteSIPParticipantMutex.RUnlock() + argsForCall := fake.deleteSIPParticipantArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeSIPStore) DeleteSIPParticipantReturns(result1 error) { + fake.deleteSIPParticipantMutex.Lock() + defer fake.deleteSIPParticipantMutex.Unlock() + fake.DeleteSIPParticipantStub = nil + fake.deleteSIPParticipantReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeSIPStore) DeleteSIPParticipantReturnsOnCall(i int, result1 error) { + fake.deleteSIPParticipantMutex.Lock() + defer fake.deleteSIPParticipantMutex.Unlock() + fake.DeleteSIPParticipantStub = nil + if fake.deleteSIPParticipantReturnsOnCall == nil { + fake.deleteSIPParticipantReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.deleteSIPParticipantReturnsOnCall[i] = struct { + result1 error + }{result1} +} + +func (fake *FakeSIPStore) DeleteSIPTrunk(arg1 context.Context, arg2 *livekit.SIPTrunkInfo) error { + fake.deleteSIPTrunkMutex.Lock() + ret, specificReturn := fake.deleteSIPTrunkReturnsOnCall[len(fake.deleteSIPTrunkArgsForCall)] + fake.deleteSIPTrunkArgsForCall = append(fake.deleteSIPTrunkArgsForCall, struct { + arg1 context.Context + arg2 *livekit.SIPTrunkInfo + }{arg1, arg2}) + stub := fake.DeleteSIPTrunkStub + fakeReturns := fake.deleteSIPTrunkReturns + fake.recordInvocation("DeleteSIPTrunk", []interface{}{arg1, arg2}) + fake.deleteSIPTrunkMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeSIPStore) DeleteSIPTrunkCallCount() int { + fake.deleteSIPTrunkMutex.RLock() + defer fake.deleteSIPTrunkMutex.RUnlock() + return len(fake.deleteSIPTrunkArgsForCall) +} + +func (fake *FakeSIPStore) DeleteSIPTrunkCalls(stub func(context.Context, *livekit.SIPTrunkInfo) error) { + fake.deleteSIPTrunkMutex.Lock() + defer fake.deleteSIPTrunkMutex.Unlock() + fake.DeleteSIPTrunkStub = stub +} + +func (fake *FakeSIPStore) DeleteSIPTrunkArgsForCall(i int) (context.Context, *livekit.SIPTrunkInfo) { + fake.deleteSIPTrunkMutex.RLock() + defer fake.deleteSIPTrunkMutex.RUnlock() + argsForCall := fake.deleteSIPTrunkArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeSIPStore) DeleteSIPTrunkReturns(result1 error) { + fake.deleteSIPTrunkMutex.Lock() + defer fake.deleteSIPTrunkMutex.Unlock() + fake.DeleteSIPTrunkStub = nil + fake.deleteSIPTrunkReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeSIPStore) DeleteSIPTrunkReturnsOnCall(i int, result1 error) { + fake.deleteSIPTrunkMutex.Lock() + defer fake.deleteSIPTrunkMutex.Unlock() + fake.DeleteSIPTrunkStub = nil + if fake.deleteSIPTrunkReturnsOnCall == nil { + fake.deleteSIPTrunkReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.deleteSIPTrunkReturnsOnCall[i] = struct { + result1 error + }{result1} +} + +func (fake *FakeSIPStore) ListSIPDispatchRule(arg1 context.Context) ([]*livekit.SIPDispatchRuleInfo, error) { + fake.listSIPDispatchRuleMutex.Lock() + ret, specificReturn := fake.listSIPDispatchRuleReturnsOnCall[len(fake.listSIPDispatchRuleArgsForCall)] + fake.listSIPDispatchRuleArgsForCall = append(fake.listSIPDispatchRuleArgsForCall, struct { + arg1 context.Context + }{arg1}) + stub := fake.ListSIPDispatchRuleStub + fakeReturns := fake.listSIPDispatchRuleReturns + fake.recordInvocation("ListSIPDispatchRule", []interface{}{arg1}) + fake.listSIPDispatchRuleMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *FakeSIPStore) ListSIPDispatchRuleCallCount() int { + fake.listSIPDispatchRuleMutex.RLock() + defer fake.listSIPDispatchRuleMutex.RUnlock() + return len(fake.listSIPDispatchRuleArgsForCall) +} + +func (fake *FakeSIPStore) ListSIPDispatchRuleCalls(stub func(context.Context) ([]*livekit.SIPDispatchRuleInfo, error)) { + fake.listSIPDispatchRuleMutex.Lock() + defer fake.listSIPDispatchRuleMutex.Unlock() + fake.ListSIPDispatchRuleStub = stub +} + +func (fake *FakeSIPStore) ListSIPDispatchRuleArgsForCall(i int) context.Context { + fake.listSIPDispatchRuleMutex.RLock() + defer fake.listSIPDispatchRuleMutex.RUnlock() + argsForCall := fake.listSIPDispatchRuleArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeSIPStore) ListSIPDispatchRuleReturns(result1 []*livekit.SIPDispatchRuleInfo, result2 error) { + fake.listSIPDispatchRuleMutex.Lock() + defer fake.listSIPDispatchRuleMutex.Unlock() + fake.ListSIPDispatchRuleStub = nil + fake.listSIPDispatchRuleReturns = struct { + result1 []*livekit.SIPDispatchRuleInfo + result2 error + }{result1, result2} +} + +func (fake *FakeSIPStore) ListSIPDispatchRuleReturnsOnCall(i int, result1 []*livekit.SIPDispatchRuleInfo, result2 error) { + fake.listSIPDispatchRuleMutex.Lock() + defer fake.listSIPDispatchRuleMutex.Unlock() + fake.ListSIPDispatchRuleStub = nil + if fake.listSIPDispatchRuleReturnsOnCall == nil { + fake.listSIPDispatchRuleReturnsOnCall = make(map[int]struct { + result1 []*livekit.SIPDispatchRuleInfo + result2 error + }) + } + fake.listSIPDispatchRuleReturnsOnCall[i] = struct { + result1 []*livekit.SIPDispatchRuleInfo + result2 error + }{result1, result2} +} + +func (fake *FakeSIPStore) ListSIPParticipant(arg1 context.Context) ([]*livekit.SIPParticipantInfo, error) { + fake.listSIPParticipantMutex.Lock() + ret, specificReturn := fake.listSIPParticipantReturnsOnCall[len(fake.listSIPParticipantArgsForCall)] + fake.listSIPParticipantArgsForCall = append(fake.listSIPParticipantArgsForCall, struct { + arg1 context.Context + }{arg1}) + stub := fake.ListSIPParticipantStub + fakeReturns := fake.listSIPParticipantReturns + fake.recordInvocation("ListSIPParticipant", []interface{}{arg1}) + fake.listSIPParticipantMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *FakeSIPStore) ListSIPParticipantCallCount() int { + fake.listSIPParticipantMutex.RLock() + defer fake.listSIPParticipantMutex.RUnlock() + return len(fake.listSIPParticipantArgsForCall) +} + +func (fake *FakeSIPStore) ListSIPParticipantCalls(stub func(context.Context) ([]*livekit.SIPParticipantInfo, error)) { + fake.listSIPParticipantMutex.Lock() + defer fake.listSIPParticipantMutex.Unlock() + fake.ListSIPParticipantStub = stub +} + +func (fake *FakeSIPStore) ListSIPParticipantArgsForCall(i int) context.Context { + fake.listSIPParticipantMutex.RLock() + defer fake.listSIPParticipantMutex.RUnlock() + argsForCall := fake.listSIPParticipantArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeSIPStore) ListSIPParticipantReturns(result1 []*livekit.SIPParticipantInfo, result2 error) { + fake.listSIPParticipantMutex.Lock() + defer fake.listSIPParticipantMutex.Unlock() + fake.ListSIPParticipantStub = nil + fake.listSIPParticipantReturns = struct { + result1 []*livekit.SIPParticipantInfo + result2 error + }{result1, result2} +} + +func (fake *FakeSIPStore) ListSIPParticipantReturnsOnCall(i int, result1 []*livekit.SIPParticipantInfo, result2 error) { + fake.listSIPParticipantMutex.Lock() + defer fake.listSIPParticipantMutex.Unlock() + fake.ListSIPParticipantStub = nil + if fake.listSIPParticipantReturnsOnCall == nil { + fake.listSIPParticipantReturnsOnCall = make(map[int]struct { + result1 []*livekit.SIPParticipantInfo + result2 error + }) + } + fake.listSIPParticipantReturnsOnCall[i] = struct { + result1 []*livekit.SIPParticipantInfo + result2 error + }{result1, result2} +} + +func (fake *FakeSIPStore) ListSIPTrunk(arg1 context.Context) ([]*livekit.SIPTrunkInfo, error) { + fake.listSIPTrunkMutex.Lock() + ret, specificReturn := fake.listSIPTrunkReturnsOnCall[len(fake.listSIPTrunkArgsForCall)] + fake.listSIPTrunkArgsForCall = append(fake.listSIPTrunkArgsForCall, struct { + arg1 context.Context + }{arg1}) + stub := fake.ListSIPTrunkStub + fakeReturns := fake.listSIPTrunkReturns + fake.recordInvocation("ListSIPTrunk", []interface{}{arg1}) + fake.listSIPTrunkMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *FakeSIPStore) ListSIPTrunkCallCount() int { + fake.listSIPTrunkMutex.RLock() + defer fake.listSIPTrunkMutex.RUnlock() + return len(fake.listSIPTrunkArgsForCall) +} + +func (fake *FakeSIPStore) ListSIPTrunkCalls(stub func(context.Context) ([]*livekit.SIPTrunkInfo, error)) { + fake.listSIPTrunkMutex.Lock() + defer fake.listSIPTrunkMutex.Unlock() + fake.ListSIPTrunkStub = stub +} + +func (fake *FakeSIPStore) ListSIPTrunkArgsForCall(i int) context.Context { + fake.listSIPTrunkMutex.RLock() + defer fake.listSIPTrunkMutex.RUnlock() + argsForCall := fake.listSIPTrunkArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeSIPStore) ListSIPTrunkReturns(result1 []*livekit.SIPTrunkInfo, result2 error) { + fake.listSIPTrunkMutex.Lock() + defer fake.listSIPTrunkMutex.Unlock() + fake.ListSIPTrunkStub = nil + fake.listSIPTrunkReturns = struct { + result1 []*livekit.SIPTrunkInfo + result2 error + }{result1, result2} +} + +func (fake *FakeSIPStore) ListSIPTrunkReturnsOnCall(i int, result1 []*livekit.SIPTrunkInfo, result2 error) { + fake.listSIPTrunkMutex.Lock() + defer fake.listSIPTrunkMutex.Unlock() + fake.ListSIPTrunkStub = nil + if fake.listSIPTrunkReturnsOnCall == nil { + fake.listSIPTrunkReturnsOnCall = make(map[int]struct { + result1 []*livekit.SIPTrunkInfo + result2 error + }) + } + fake.listSIPTrunkReturnsOnCall[i] = struct { + result1 []*livekit.SIPTrunkInfo + result2 error + }{result1, result2} +} + +func (fake *FakeSIPStore) LoadSIPDispatchRule(arg1 context.Context, arg2 string) (*livekit.SIPDispatchRuleInfo, error) { + fake.loadSIPDispatchRuleMutex.Lock() + ret, specificReturn := fake.loadSIPDispatchRuleReturnsOnCall[len(fake.loadSIPDispatchRuleArgsForCall)] + fake.loadSIPDispatchRuleArgsForCall = append(fake.loadSIPDispatchRuleArgsForCall, struct { + arg1 context.Context + arg2 string + }{arg1, arg2}) + stub := fake.LoadSIPDispatchRuleStub + fakeReturns := fake.loadSIPDispatchRuleReturns + fake.recordInvocation("LoadSIPDispatchRule", []interface{}{arg1, arg2}) + fake.loadSIPDispatchRuleMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *FakeSIPStore) LoadSIPDispatchRuleCallCount() int { + fake.loadSIPDispatchRuleMutex.RLock() + defer fake.loadSIPDispatchRuleMutex.RUnlock() + return len(fake.loadSIPDispatchRuleArgsForCall) +} + +func (fake *FakeSIPStore) LoadSIPDispatchRuleCalls(stub func(context.Context, string) (*livekit.SIPDispatchRuleInfo, error)) { + fake.loadSIPDispatchRuleMutex.Lock() + defer fake.loadSIPDispatchRuleMutex.Unlock() + fake.LoadSIPDispatchRuleStub = stub +} + +func (fake *FakeSIPStore) LoadSIPDispatchRuleArgsForCall(i int) (context.Context, string) { + fake.loadSIPDispatchRuleMutex.RLock() + defer fake.loadSIPDispatchRuleMutex.RUnlock() + argsForCall := fake.loadSIPDispatchRuleArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeSIPStore) LoadSIPDispatchRuleReturns(result1 *livekit.SIPDispatchRuleInfo, result2 error) { + fake.loadSIPDispatchRuleMutex.Lock() + defer fake.loadSIPDispatchRuleMutex.Unlock() + fake.LoadSIPDispatchRuleStub = nil + fake.loadSIPDispatchRuleReturns = struct { + result1 *livekit.SIPDispatchRuleInfo + result2 error + }{result1, result2} +} + +func (fake *FakeSIPStore) LoadSIPDispatchRuleReturnsOnCall(i int, result1 *livekit.SIPDispatchRuleInfo, result2 error) { + fake.loadSIPDispatchRuleMutex.Lock() + defer fake.loadSIPDispatchRuleMutex.Unlock() + fake.LoadSIPDispatchRuleStub = nil + if fake.loadSIPDispatchRuleReturnsOnCall == nil { + fake.loadSIPDispatchRuleReturnsOnCall = make(map[int]struct { + result1 *livekit.SIPDispatchRuleInfo + result2 error + }) + } + fake.loadSIPDispatchRuleReturnsOnCall[i] = struct { + result1 *livekit.SIPDispatchRuleInfo + result2 error + }{result1, result2} +} + +func (fake *FakeSIPStore) LoadSIPParticipant(arg1 context.Context, arg2 string) (*livekit.SIPParticipantInfo, error) { + fake.loadSIPParticipantMutex.Lock() + ret, specificReturn := fake.loadSIPParticipantReturnsOnCall[len(fake.loadSIPParticipantArgsForCall)] + fake.loadSIPParticipantArgsForCall = append(fake.loadSIPParticipantArgsForCall, struct { + arg1 context.Context + arg2 string + }{arg1, arg2}) + stub := fake.LoadSIPParticipantStub + fakeReturns := fake.loadSIPParticipantReturns + fake.recordInvocation("LoadSIPParticipant", []interface{}{arg1, arg2}) + fake.loadSIPParticipantMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *FakeSIPStore) LoadSIPParticipantCallCount() int { + fake.loadSIPParticipantMutex.RLock() + defer fake.loadSIPParticipantMutex.RUnlock() + return len(fake.loadSIPParticipantArgsForCall) +} + +func (fake *FakeSIPStore) LoadSIPParticipantCalls(stub func(context.Context, string) (*livekit.SIPParticipantInfo, error)) { + fake.loadSIPParticipantMutex.Lock() + defer fake.loadSIPParticipantMutex.Unlock() + fake.LoadSIPParticipantStub = stub +} + +func (fake *FakeSIPStore) LoadSIPParticipantArgsForCall(i int) (context.Context, string) { + fake.loadSIPParticipantMutex.RLock() + defer fake.loadSIPParticipantMutex.RUnlock() + argsForCall := fake.loadSIPParticipantArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeSIPStore) LoadSIPParticipantReturns(result1 *livekit.SIPParticipantInfo, result2 error) { + fake.loadSIPParticipantMutex.Lock() + defer fake.loadSIPParticipantMutex.Unlock() + fake.LoadSIPParticipantStub = nil + fake.loadSIPParticipantReturns = struct { + result1 *livekit.SIPParticipantInfo + result2 error + }{result1, result2} +} + +func (fake *FakeSIPStore) LoadSIPParticipantReturnsOnCall(i int, result1 *livekit.SIPParticipantInfo, result2 error) { + fake.loadSIPParticipantMutex.Lock() + defer fake.loadSIPParticipantMutex.Unlock() + fake.LoadSIPParticipantStub = nil + if fake.loadSIPParticipantReturnsOnCall == nil { + fake.loadSIPParticipantReturnsOnCall = make(map[int]struct { + result1 *livekit.SIPParticipantInfo + result2 error + }) + } + fake.loadSIPParticipantReturnsOnCall[i] = struct { + result1 *livekit.SIPParticipantInfo + result2 error + }{result1, result2} +} + +func (fake *FakeSIPStore) LoadSIPTrunk(arg1 context.Context, arg2 string) (*livekit.SIPTrunkInfo, error) { + fake.loadSIPTrunkMutex.Lock() + ret, specificReturn := fake.loadSIPTrunkReturnsOnCall[len(fake.loadSIPTrunkArgsForCall)] + fake.loadSIPTrunkArgsForCall = append(fake.loadSIPTrunkArgsForCall, struct { + arg1 context.Context + arg2 string + }{arg1, arg2}) + stub := fake.LoadSIPTrunkStub + fakeReturns := fake.loadSIPTrunkReturns + fake.recordInvocation("LoadSIPTrunk", []interface{}{arg1, arg2}) + fake.loadSIPTrunkMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *FakeSIPStore) LoadSIPTrunkCallCount() int { + fake.loadSIPTrunkMutex.RLock() + defer fake.loadSIPTrunkMutex.RUnlock() + return len(fake.loadSIPTrunkArgsForCall) +} + +func (fake *FakeSIPStore) LoadSIPTrunkCalls(stub func(context.Context, string) (*livekit.SIPTrunkInfo, error)) { + fake.loadSIPTrunkMutex.Lock() + defer fake.loadSIPTrunkMutex.Unlock() + fake.LoadSIPTrunkStub = stub +} + +func (fake *FakeSIPStore) LoadSIPTrunkArgsForCall(i int) (context.Context, string) { + fake.loadSIPTrunkMutex.RLock() + defer fake.loadSIPTrunkMutex.RUnlock() + argsForCall := fake.loadSIPTrunkArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeSIPStore) LoadSIPTrunkReturns(result1 *livekit.SIPTrunkInfo, result2 error) { + fake.loadSIPTrunkMutex.Lock() + defer fake.loadSIPTrunkMutex.Unlock() + fake.LoadSIPTrunkStub = nil + fake.loadSIPTrunkReturns = struct { + result1 *livekit.SIPTrunkInfo + result2 error + }{result1, result2} +} + +func (fake *FakeSIPStore) LoadSIPTrunkReturnsOnCall(i int, result1 *livekit.SIPTrunkInfo, result2 error) { + fake.loadSIPTrunkMutex.Lock() + defer fake.loadSIPTrunkMutex.Unlock() + fake.LoadSIPTrunkStub = nil + if fake.loadSIPTrunkReturnsOnCall == nil { + fake.loadSIPTrunkReturnsOnCall = make(map[int]struct { + result1 *livekit.SIPTrunkInfo + result2 error + }) + } + fake.loadSIPTrunkReturnsOnCall[i] = struct { + result1 *livekit.SIPTrunkInfo + result2 error + }{result1, result2} +} + +func (fake *FakeSIPStore) StoreSIPDispatchRule(arg1 context.Context, arg2 *livekit.SIPDispatchRuleInfo) error { + fake.storeSIPDispatchRuleMutex.Lock() + ret, specificReturn := fake.storeSIPDispatchRuleReturnsOnCall[len(fake.storeSIPDispatchRuleArgsForCall)] + fake.storeSIPDispatchRuleArgsForCall = append(fake.storeSIPDispatchRuleArgsForCall, struct { + arg1 context.Context + arg2 *livekit.SIPDispatchRuleInfo + }{arg1, arg2}) + stub := fake.StoreSIPDispatchRuleStub + fakeReturns := fake.storeSIPDispatchRuleReturns + fake.recordInvocation("StoreSIPDispatchRule", []interface{}{arg1, arg2}) + fake.storeSIPDispatchRuleMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeSIPStore) StoreSIPDispatchRuleCallCount() int { + fake.storeSIPDispatchRuleMutex.RLock() + defer fake.storeSIPDispatchRuleMutex.RUnlock() + return len(fake.storeSIPDispatchRuleArgsForCall) +} + +func (fake *FakeSIPStore) StoreSIPDispatchRuleCalls(stub func(context.Context, *livekit.SIPDispatchRuleInfo) error) { + fake.storeSIPDispatchRuleMutex.Lock() + defer fake.storeSIPDispatchRuleMutex.Unlock() + fake.StoreSIPDispatchRuleStub = stub +} + +func (fake *FakeSIPStore) StoreSIPDispatchRuleArgsForCall(i int) (context.Context, *livekit.SIPDispatchRuleInfo) { + fake.storeSIPDispatchRuleMutex.RLock() + defer fake.storeSIPDispatchRuleMutex.RUnlock() + argsForCall := fake.storeSIPDispatchRuleArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeSIPStore) StoreSIPDispatchRuleReturns(result1 error) { + fake.storeSIPDispatchRuleMutex.Lock() + defer fake.storeSIPDispatchRuleMutex.Unlock() + fake.StoreSIPDispatchRuleStub = nil + fake.storeSIPDispatchRuleReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeSIPStore) StoreSIPDispatchRuleReturnsOnCall(i int, result1 error) { + fake.storeSIPDispatchRuleMutex.Lock() + defer fake.storeSIPDispatchRuleMutex.Unlock() + fake.StoreSIPDispatchRuleStub = nil + if fake.storeSIPDispatchRuleReturnsOnCall == nil { + fake.storeSIPDispatchRuleReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.storeSIPDispatchRuleReturnsOnCall[i] = struct { + result1 error + }{result1} +} + +func (fake *FakeSIPStore) StoreSIPParticipant(arg1 context.Context, arg2 *livekit.SIPParticipantInfo) error { + fake.storeSIPParticipantMutex.Lock() + ret, specificReturn := fake.storeSIPParticipantReturnsOnCall[len(fake.storeSIPParticipantArgsForCall)] + fake.storeSIPParticipantArgsForCall = append(fake.storeSIPParticipantArgsForCall, struct { + arg1 context.Context + arg2 *livekit.SIPParticipantInfo + }{arg1, arg2}) + stub := fake.StoreSIPParticipantStub + fakeReturns := fake.storeSIPParticipantReturns + fake.recordInvocation("StoreSIPParticipant", []interface{}{arg1, arg2}) + fake.storeSIPParticipantMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeSIPStore) StoreSIPParticipantCallCount() int { + fake.storeSIPParticipantMutex.RLock() + defer fake.storeSIPParticipantMutex.RUnlock() + return len(fake.storeSIPParticipantArgsForCall) +} + +func (fake *FakeSIPStore) StoreSIPParticipantCalls(stub func(context.Context, *livekit.SIPParticipantInfo) error) { + fake.storeSIPParticipantMutex.Lock() + defer fake.storeSIPParticipantMutex.Unlock() + fake.StoreSIPParticipantStub = stub +} + +func (fake *FakeSIPStore) StoreSIPParticipantArgsForCall(i int) (context.Context, *livekit.SIPParticipantInfo) { + fake.storeSIPParticipantMutex.RLock() + defer fake.storeSIPParticipantMutex.RUnlock() + argsForCall := fake.storeSIPParticipantArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeSIPStore) StoreSIPParticipantReturns(result1 error) { + fake.storeSIPParticipantMutex.Lock() + defer fake.storeSIPParticipantMutex.Unlock() + fake.StoreSIPParticipantStub = nil + fake.storeSIPParticipantReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeSIPStore) StoreSIPParticipantReturnsOnCall(i int, result1 error) { + fake.storeSIPParticipantMutex.Lock() + defer fake.storeSIPParticipantMutex.Unlock() + fake.StoreSIPParticipantStub = nil + if fake.storeSIPParticipantReturnsOnCall == nil { + fake.storeSIPParticipantReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.storeSIPParticipantReturnsOnCall[i] = struct { + result1 error + }{result1} +} + +func (fake *FakeSIPStore) StoreSIPTrunk(arg1 context.Context, arg2 *livekit.SIPTrunkInfo) error { + fake.storeSIPTrunkMutex.Lock() + ret, specificReturn := fake.storeSIPTrunkReturnsOnCall[len(fake.storeSIPTrunkArgsForCall)] + fake.storeSIPTrunkArgsForCall = append(fake.storeSIPTrunkArgsForCall, struct { + arg1 context.Context + arg2 *livekit.SIPTrunkInfo + }{arg1, arg2}) + stub := fake.StoreSIPTrunkStub + fakeReturns := fake.storeSIPTrunkReturns + fake.recordInvocation("StoreSIPTrunk", []interface{}{arg1, arg2}) + fake.storeSIPTrunkMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeSIPStore) StoreSIPTrunkCallCount() int { + fake.storeSIPTrunkMutex.RLock() + defer fake.storeSIPTrunkMutex.RUnlock() + return len(fake.storeSIPTrunkArgsForCall) +} + +func (fake *FakeSIPStore) StoreSIPTrunkCalls(stub func(context.Context, *livekit.SIPTrunkInfo) error) { + fake.storeSIPTrunkMutex.Lock() + defer fake.storeSIPTrunkMutex.Unlock() + fake.StoreSIPTrunkStub = stub +} + +func (fake *FakeSIPStore) StoreSIPTrunkArgsForCall(i int) (context.Context, *livekit.SIPTrunkInfo) { + fake.storeSIPTrunkMutex.RLock() + defer fake.storeSIPTrunkMutex.RUnlock() + argsForCall := fake.storeSIPTrunkArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeSIPStore) StoreSIPTrunkReturns(result1 error) { + fake.storeSIPTrunkMutex.Lock() + defer fake.storeSIPTrunkMutex.Unlock() + fake.StoreSIPTrunkStub = nil + fake.storeSIPTrunkReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeSIPStore) StoreSIPTrunkReturnsOnCall(i int, result1 error) { + fake.storeSIPTrunkMutex.Lock() + defer fake.storeSIPTrunkMutex.Unlock() + fake.StoreSIPTrunkStub = nil + if fake.storeSIPTrunkReturnsOnCall == nil { + fake.storeSIPTrunkReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.storeSIPTrunkReturnsOnCall[i] = struct { + result1 error + }{result1} +} + +func (fake *FakeSIPStore) Invocations() map[string][][]interface{} { + fake.invocationsMutex.RLock() + defer fake.invocationsMutex.RUnlock() + fake.deleteSIPDispatchRuleMutex.RLock() + defer fake.deleteSIPDispatchRuleMutex.RUnlock() + fake.deleteSIPParticipantMutex.RLock() + defer fake.deleteSIPParticipantMutex.RUnlock() + fake.deleteSIPTrunkMutex.RLock() + defer fake.deleteSIPTrunkMutex.RUnlock() + fake.listSIPDispatchRuleMutex.RLock() + defer fake.listSIPDispatchRuleMutex.RUnlock() + fake.listSIPParticipantMutex.RLock() + defer fake.listSIPParticipantMutex.RUnlock() + fake.listSIPTrunkMutex.RLock() + defer fake.listSIPTrunkMutex.RUnlock() + fake.loadSIPDispatchRuleMutex.RLock() + defer fake.loadSIPDispatchRuleMutex.RUnlock() + fake.loadSIPParticipantMutex.RLock() + defer fake.loadSIPParticipantMutex.RUnlock() + fake.loadSIPTrunkMutex.RLock() + defer fake.loadSIPTrunkMutex.RUnlock() + fake.storeSIPDispatchRuleMutex.RLock() + defer fake.storeSIPDispatchRuleMutex.RUnlock() + fake.storeSIPParticipantMutex.RLock() + defer fake.storeSIPParticipantMutex.RUnlock() + fake.storeSIPTrunkMutex.RLock() + defer fake.storeSIPTrunkMutex.RUnlock() + copiedInvocations := map[string][][]interface{}{} + for key, value := range fake.invocations { + copiedInvocations[key] = value + } + return copiedInvocations +} + +func (fake *FakeSIPStore) recordInvocation(key string, args []interface{}) { + fake.invocationsMutex.Lock() + defer fake.invocationsMutex.Unlock() + if fake.invocations == nil { + fake.invocations = map[string][][]interface{}{} + } + if fake.invocations[key] == nil { + fake.invocations[key] = [][]interface{}{} + } + fake.invocations[key] = append(fake.invocations[key], args) +} + +var _ service.SIPStore = new(FakeSIPStore) diff --git a/pkg/service/twirp.go b/pkg/service/twirp.go index 50f3f2dc0..b7b31d764 100644 --- a/pkg/service/twirp.go +++ b/pkg/service/twirp.go @@ -25,13 +25,11 @@ import ( "github.com/twitchtv/twirp" "github.com/livekit/livekit-server/pkg/telemetry/prometheus" - "github.com/livekit/protocol/logger" + "github.com/livekit/livekit-server/pkg/utils" ) -var ( - loggerKey = struct{}{} - statusReporterKey = struct{ a int }{42} -) +type twirpLoggerContext struct{} +type statusReporterKey struct{} type twirpRequestFields struct { service string @@ -41,11 +39,10 @@ type twirpRequestFields struct { // logging handling inspired by https://github.com/bakins/twirpzap // License: Apache-2.0 -func TwirpLogger(logger logger.Logger) *twirp.ServerHooks { +func TwirpLogger() *twirp.ServerHooks { loggerPool := &sync.Pool{ New: func() interface{} { return &requestLogger{ - logger: logger, fieldsOrig: make([]interface{}, 0, 30), } }, @@ -65,14 +62,13 @@ func TwirpLogger(logger logger.Logger) *twirp.ServerHooks { type requestLogger struct { twirpRequestFields - logger logger.Logger fieldsOrig []interface{} fields []interface{} startedAt time.Time } func AppendLogFields(ctx context.Context, fields ...interface{}) { - r, ok := ctx.Value(loggerKey).(*requestLogger) + r, ok := ctx.Value(twirpLoggerContext{}).(*requestLogger) if !ok || r == nil { return } @@ -91,13 +87,13 @@ func requestReceived(ctx context.Context, requestLoggerPool *sync.Pool) (context r.fields = append(r.fields, "service", svc) } - ctx = context.WithValue(ctx, loggerKey, r) + ctx = context.WithValue(ctx, twirpLoggerContext{}, r) return ctx, nil } func responseRouted(ctx context.Context) (context.Context, error) { if meth, ok := twirp.MethodName(ctx); ok { - l, ok := ctx.Value(loggerKey).(*requestLogger) + l, ok := ctx.Value(twirpLoggerContext{}).(*requestLogger) if !ok || l == nil { return ctx, nil } @@ -109,7 +105,7 @@ func responseRouted(ctx context.Context) (context.Context, error) { } func responseSent(ctx context.Context, requestLoggerPool *sync.Pool) { - r, ok := ctx.Value(loggerKey).(*requestLogger) + r, ok := ctx.Value(twirpLoggerContext{}).(*requestLogger) if !ok || r == nil { return } @@ -125,7 +121,7 @@ func responseSent(ctx context.Context, requestLoggerPool *sync.Pool) { } serviceMethod := "API " + r.service + "." + r.method - r.logger.Infow(serviceMethod, r.fields...) + utils.GetLogger(ctx).WithComponent(utils.ComponentAPI).Infow(serviceMethod, r.fields...) r.fields = r.fieldsOrig r.error = nil @@ -134,7 +130,7 @@ func responseSent(ctx context.Context, requestLoggerPool *sync.Pool) { } func errorReceived(ctx context.Context, e twirp.Error) context.Context { - r, ok := ctx.Value(loggerKey).(*requestLogger) + r, ok := ctx.Value(twirpLoggerContext{}).(*requestLogger) if !ok || r == nil { return ctx } @@ -160,13 +156,13 @@ func statusReporterRequestReceived(ctx context.Context) (context.Context, error) r.service = svc } - ctx = context.WithValue(ctx, statusReporterKey, r) + ctx = context.WithValue(ctx, statusReporterKey{}, r) return ctx, nil } func statusReporterResponseRouted(ctx context.Context) (context.Context, error) { if meth, ok := twirp.MethodName(ctx); ok { - l, ok := ctx.Value(statusReporterKey).(*twirpRequestFields) + l, ok := ctx.Value(statusReporterKey{}).(*twirpRequestFields) if !ok || l == nil { return ctx, nil } @@ -177,7 +173,7 @@ func statusReporterResponseRouted(ctx context.Context) (context.Context, error) } func statusReporterResponseSent(ctx context.Context) { - r, ok := ctx.Value(statusReporterKey).(*twirpRequestFields) + r, ok := ctx.Value(statusReporterKey{}).(*twirpRequestFields) if !ok || r == nil { return } @@ -205,7 +201,7 @@ func statusReporterResponseSent(ctx context.Context) { } func statusReporterErrorReceived(ctx context.Context, e twirp.Error) context.Context { - r, ok := ctx.Value(statusReporterKey).(*twirpRequestFields) + r, ok := ctx.Value(statusReporterKey{}).(*twirpRequestFields) if !ok || r == nil { return ctx } diff --git a/pkg/service/utils.go b/pkg/service/utils.go index 32076455f..2d1bafd4e 100644 --- a/pkg/service/utils.go +++ b/pkg/service/utils.go @@ -22,8 +22,11 @@ import ( "github.com/livekit/protocol/logger" ) -func handleError(w http.ResponseWriter, status int, err error, keysAndValues ...interface{}) { +func handleError(w http.ResponseWriter, r *http.Request, status int, err error, keysAndValues ...interface{}) { keysAndValues = append(keysAndValues, "status", status) + if r != nil && r.URL != nil { + keysAndValues = append(keysAndValues, "method", r.Method, "path", r.URL.Path) + } logger.GetLogger().WithCallDepth(1).Warnw("error handling request", err, keysAndValues...) w.WriteHeader(status) _, _ = w.Write([]byte(err.Error())) diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 573b85649..3cf102a08 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -350,7 +350,10 @@ func (b *Buffer) Close() error { if b.rtpStats != nil { b.rtpStats.Stop() - b.logger.Infow("rtp stats", "direction", "upstream", "stats", b.rtpStats.ToString()) + b.logger.Debugw("rtp stats", + "direction", "upstream", + "stats", func() interface{} { return b.rtpStats.ToString() }, + ) if b.onFinalRtpStats != nil { b.onFinalRtpStats(b.rtpStats.ToProto()) } diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 30bf71d20..70a8e8fed 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -984,10 +984,14 @@ func (d *DownTrack) CloseWithFlush(flush bool) { d.bindLock.Unlock() d.connectionStats.Close() d.rtpStats.Stop() - rtpStats := d.rtpStats.ToString() - if rtpStats != "" { - d.params.Logger.Infow("rtp stats", "direction", "downstream", "mime", d.mime, "ssrc", d.ssrc, "stats", rtpStats) - } + d.params.Logger.Debugw("rtp stats", + "direction", "downstream", + "mime", d.mime, + "ssrc", d.ssrc, + // evaluate only if log level matches + "stats", func() interface{} { + return d.rtpStats.ToString() + }) d.maxLayerNotifierChMu.Lock() d.maxLayerNotifierChClosed = true diff --git a/pkg/sfu/streamallocator/streamallocator.go b/pkg/sfu/streamallocator/streamallocator.go index 9d65363a2..8fde6e3c7 100644 --- a/pkg/sfu/streamallocator/streamallocator.go +++ b/pkg/sfu/streamallocator/streamallocator.go @@ -1297,7 +1297,7 @@ func (s *StreamAllocator) initProbe(probeGoalDeltaBps int64) { s.channelObserver = s.newChannelObserverProbe() s.channelObserver.SeedEstimate(s.lastReceivedEstimate) - s.params.Logger.Infow( + s.params.Logger.Debugw( "stream allocator: starting probe", "probeClusterId", probeClusterId, "current usage", expectedBandwidthUsage, diff --git a/pkg/utils/context.go b/pkg/utils/context.go index f5262e178..4ad9086d2 100644 --- a/pkg/utils/context.go +++ b/pkg/utils/context.go @@ -16,17 +16,33 @@ package utils -import "context" +import ( + "context" -var attemptKey = struct{}{} + "github.com/livekit/protocol/logger" +) + +type attemptKey struct{} +type loggerKey = struct{} func ContextWithAttempt(ctx context.Context, attempt int) context.Context { - return context.WithValue(ctx, attemptKey, attempt) + return context.WithValue(ctx, attemptKey{}, attempt) } func GetAttempt(ctx context.Context) int { - if attempt, ok := ctx.Value(attemptKey).(int); ok { + if attempt, ok := ctx.Value(attemptKey{}).(int); ok { return attempt } return 0 } + +func ContextWithLogger(ctx context.Context, logger logger.Logger) context.Context { + return context.WithValue(ctx, loggerKey{}, logger) +} + +func GetLogger(ctx context.Context) logger.Logger { + if l, ok := ctx.Value(loggerKey{}).(logger.Logger); ok { + return l + } + return logger.GetLogger() +}