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

Log cleanup pass (#2285)

Browse Source 
* 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
This commit is contained in:
David Zhao
2023-12-02 15:07:31 -08:00
committed by GitHub
parent beecfe3710
commit 3fe124c87f
24 changed files with 1171 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/routing/errors.go
+8 -1
View File
@@ -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")
)
pkg/routing/interfaces.go
+8 -1
View File
@@ -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
pkg/routing/localrouter.go
+10 -3
View File
@@ -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
}
pkg/routing/redisrouter.go
+11 -9
View File
@@ -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 {
pkg/routing/routingfakes/fake_router.go
+19 -29
View File
@@ -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() {