mirror of
https://github.com/livekit/livekit.git
synced 2026-08-28 20:38:20 +00:00
log: consistent log key (#60)
This commit is contained in:
hn8
+2
-2
@@ -185,7 +185,7 @@ func startServer(c *cli.Context) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
logger.Infow("configured key provider", "num_keys", keyProvider.NumKeys())
|
||||
logger.Infow("configured key provider", "numKeys", keyProvider.NumKeys())
|
||||
|
||||
currentNode, err := routing.NewLocalNode(conf)
|
||||
if err != nil {
|
||||
@@ -218,7 +218,7 @@ func startServer(c *cli.Context) error {
|
||||
|
||||
func createRouterAndStore(config *config.Config, node routing.LocalNode) (router routing.Router, store service.RoomStore, err error) {
|
||||
if config.HasRedis() {
|
||||
logger.Infow("using multi-node routing via redis", "address", config.Redis.Address)
|
||||
logger.Infow("using multi-node routing via redis", "addr", config.Redis.Address)
|
||||
rc := redis.NewClient(&redis.Options{
|
||||
Addr: config.Redis.Address,
|
||||
Username: config.Redis.Username,
|
||||
|
||||
@@ -190,7 +190,7 @@ func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantK
|
||||
if rtcNode.Id != r.currentNode.Id {
|
||||
err = ErrIncorrectRTCNode
|
||||
logger.Errorw("called participant on incorrect node", err,
|
||||
"rtcNode", rtcNode, "currentNode", r.currentNode.Id)
|
||||
"rtcNode", rtcNode, "nodeID", r.currentNode.Id)
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -309,7 +309,7 @@ func (r *RedisRouter) statsWorker() {
|
||||
case <-time.After(statsUpdateInterval):
|
||||
r.currentNode.Stats.UpdatedAt = time.Now().Unix()
|
||||
if err := r.RegisterNode(); err != nil {
|
||||
logger.Errorw("could not update node", err)
|
||||
logger.Errorw("could not update node", err, "nodeID", r.currentNode.Id)
|
||||
}
|
||||
case <-r.ctx.Done():
|
||||
return
|
||||
@@ -320,9 +320,9 @@ func (r *RedisRouter) statsWorker() {
|
||||
// worker that consumes redis messages intended for this node
|
||||
func (r *RedisRouter) redisWorker(startedChan chan struct{}) {
|
||||
defer func() {
|
||||
logger.Debugw("finishing redisWorker", "node", r.currentNode.Id)
|
||||
logger.Debugw("finishing redisWorker", "nodeID", r.currentNode.Id)
|
||||
}()
|
||||
logger.Debugw("starting redisWorker", "node", r.currentNode.Id)
|
||||
logger.Debugw("starting redisWorker", "nodeID", r.currentNode.Id)
|
||||
|
||||
sigChannel := signalNodeChannel(r.currentNode.Id)
|
||||
rtcChannel := rtcNodeChannel(r.currentNode.Id)
|
||||
@@ -373,7 +373,7 @@ func (r *RedisRouter) handleSignalMessage(sm *livekit.SignalNodeMessage) error {
|
||||
switch rmb := sm.Message.(type) {
|
||||
case *livekit.SignalNodeMessage_Response:
|
||||
// logger.Debugw("forwarding signal message",
|
||||
// "connectionId", connectionId,
|
||||
// "connID", connectionId,
|
||||
// "type", fmt.Sprintf("%T", rmb.Response.Message))
|
||||
if err := resSink.WriteMessage(rmb.Response); err != nil {
|
||||
return err
|
||||
@@ -381,7 +381,7 @@ func (r *RedisRouter) handleSignalMessage(sm *livekit.SignalNodeMessage) error {
|
||||
|
||||
case *livekit.SignalNodeMessage_EndSession:
|
||||
// logger.Debugw("received EndSession, closing signal connection",
|
||||
// "connectionId", connectionId)
|
||||
// "connID", connectionId)
|
||||
resSink.Close()
|
||||
}
|
||||
return nil
|
||||
|
||||
@@ -207,8 +207,9 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error {
|
||||
}
|
||||
logger.Debugw("removing peerconnection track",
|
||||
"track", t.params.TrackID,
|
||||
"participantId", t.params.ParticipantID,
|
||||
"destParticipant", sub.Identity())
|
||||
"pIDs", []string{t.params.ParticipantID, sub.ID()},
|
||||
"participant", sub.Identity(),
|
||||
)
|
||||
if err := sub.SubscriberPC().RemoveTrack(sender); err != nil {
|
||||
if err == webrtc.ErrConnectionClosed {
|
||||
// sub closing, can skip removing subscribedtracks
|
||||
@@ -216,7 +217,7 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error {
|
||||
}
|
||||
if _, ok := err.(*rtcerr.InvalidStateError); !ok {
|
||||
logger.Warnw("could not remove remoteTrack from forwarder", err,
|
||||
"sub", sub.Identity())
|
||||
"participant", sub.Identity(), "pID", sub.ID())
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+20
-17
@@ -246,7 +246,7 @@ func (p *ParticipantImpl) OnClose(callback func(types.Participant)) {
|
||||
// HandleOffer an offer from remote participant, used when clients make the initial connection
|
||||
func (p *ParticipantImpl) HandleOffer(sdp webrtc.SessionDescription) (answer webrtc.SessionDescription, err error) {
|
||||
logger.Debugw("answering pub offer", "state", p.State().String(),
|
||||
"participant", p.Identity(),
|
||||
"participant", p.Identity(), "pID", p.ID(),
|
||||
//"sdp", sdp.SDP,
|
||||
)
|
||||
|
||||
@@ -266,7 +266,7 @@ func (p *ParticipantImpl) HandleOffer(sdp webrtc.SessionDescription) (answer web
|
||||
}
|
||||
|
||||
logger.Debugw("sending answer to client",
|
||||
"participant", p.Identity(),
|
||||
"participant", p.Identity(), "pID", p.ID(),
|
||||
//"sdp", sdp.SDP,
|
||||
)
|
||||
err = p.writeMessage(&livekit.SignalResponse{
|
||||
@@ -331,7 +331,7 @@ func (p *ParticipantImpl) HandleAnswer(sdp webrtc.SessionDescription) error {
|
||||
return ErrUnexpectedOffer
|
||||
}
|
||||
logger.Debugw("setting subPC answer",
|
||||
"participant", p.Identity(),
|
||||
"participant", p.Identity(), "pID", p.ID(),
|
||||
//"sdp", sdp.SDP,
|
||||
)
|
||||
|
||||
@@ -438,8 +438,8 @@ func (p *ParticipantImpl) AddSubscriber(op types.Participant) (int, error) {
|
||||
}
|
||||
|
||||
logger.Debugw("subscribing new participant to tracks",
|
||||
"srcParticipant", p.Identity(),
|
||||
"newParticipant", op.Identity(),
|
||||
"participants", []string{p.Identity(), op.Identity()},
|
||||
"pIDs", []string{p.ID(), op.ID()},
|
||||
"numTracks", len(tracks))
|
||||
|
||||
n := 0
|
||||
@@ -547,6 +547,7 @@ func (p *ParticipantImpl) SetTrackMuted(trackId string, muted bool) {
|
||||
if currentMuted != track.IsMuted() && p.onTrackUpdated != nil {
|
||||
logger.Debugw("mute status changed",
|
||||
"participant", p.Identity(),
|
||||
"pID", p.ID(),
|
||||
"track", trackId,
|
||||
"muted", track.IsMuted())
|
||||
p.onTrackUpdated(p, track)
|
||||
@@ -600,7 +601,7 @@ func (p *ParticipantImpl) GetSubscribedTracks() []types.SubscribedTrack {
|
||||
|
||||
// AddSubscribedTrack adds a track to the participant's subscribed list
|
||||
func (p *ParticipantImpl) AddSubscribedTrack(pubId string, subTrack types.SubscribedTrack) {
|
||||
logger.Debugw("added subscribedTrack", "srcParticipant", pubId,
|
||||
logger.Debugw("added subscribedTrack", "pIDs", []string{pubId, p.ID()},
|
||||
"participant", p.Identity(), "track", subTrack.ID())
|
||||
p.lock.Lock()
|
||||
p.subscribedTracks[pubId] = append(p.subscribedTracks[pubId], subTrack)
|
||||
@@ -609,7 +610,7 @@ func (p *ParticipantImpl) AddSubscribedTrack(pubId string, subTrack types.Subscr
|
||||
|
||||
// RemoveSubscribedTrack removes a track to the participant's subscribed list
|
||||
func (p *ParticipantImpl) RemoveSubscribedTrack(pubId string, subTrack types.SubscribedTrack) {
|
||||
logger.Debugw("removed subscribedTrack", "srcParticipant", pubId,
|
||||
logger.Debugw("removed subscribedTrack", "pIDs", []string{pubId, p.ID()},
|
||||
"participant", p.Identity(), "track", subTrack.ID())
|
||||
p.lock.Lock()
|
||||
defer p.lock.Unlock()
|
||||
@@ -628,6 +629,7 @@ func (p *ParticipantImpl) sendIceCandidate(c *webrtc.ICECandidate, target liveki
|
||||
// write candidate
|
||||
logger.Debugw("sending ice candidates",
|
||||
"participant", p.Identity(),
|
||||
"pID", p.ID(),
|
||||
"candidate", c.String())
|
||||
trickle := ToProtoTrickle(ci)
|
||||
trickle.Target = target
|
||||
@@ -644,7 +646,7 @@ func (p *ParticipantImpl) updateState(state livekit.ParticipantInfo_State) {
|
||||
return
|
||||
}
|
||||
p.state.Store(state)
|
||||
logger.Debugw("updating participant state", "state", state.String(), "participant", p.Identity())
|
||||
logger.Debugw("updating participant state", "state", state.String(), "participant", p.Identity(), "pID", p.ID())
|
||||
p.lock.RLock()
|
||||
onStateChange := p.onStateChange
|
||||
p.lock.RUnlock()
|
||||
@@ -664,7 +666,7 @@ func (p *ParticipantImpl) writeMessage(msg *livekit.SignalResponse) error {
|
||||
err := sink.WriteMessage(msg)
|
||||
if err != nil {
|
||||
logger.Warnw("could not send message to participant", err,
|
||||
"id", p.ID(),
|
||||
"pID", p.ID(),
|
||||
"participant", p.Identity(),
|
||||
"message", fmt.Sprintf("%T", msg.Message))
|
||||
return err
|
||||
@@ -675,13 +677,13 @@ func (p *ParticipantImpl) writeMessage(msg *livekit.SignalResponse) error {
|
||||
// when the server has an offer for participant
|
||||
func (p *ParticipantImpl) onOffer(offer webrtc.SessionDescription) {
|
||||
if p.State() == livekit.ParticipantInfo_DISCONNECTED {
|
||||
logger.Debugw("skipping server offer", "participant", p.Identity())
|
||||
logger.Debugw("skipping server offer", "participant", p.Identity(), "pID", p.ID())
|
||||
// skip when disconnected
|
||||
return
|
||||
}
|
||||
|
||||
logger.Debugw("sending server offer to participant",
|
||||
"participant", p.Identity(),
|
||||
"participant", p.Identity(), "pID", p.ID(),
|
||||
//"sdp", offer.SDP,
|
||||
)
|
||||
|
||||
@@ -700,12 +702,13 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w
|
||||
|
||||
logger.Debugw("mediaTrack added",
|
||||
"participant", p.Identity(),
|
||||
"remoteTrack", track.ID(),
|
||||
"pID", p.ID(),
|
||||
"track", track.ID(),
|
||||
"rid", track.RID())
|
||||
|
||||
if !p.CanPublish() {
|
||||
logger.Warnw("no permission to publish mediaTrack", nil,
|
||||
"participant", p.Identity())
|
||||
"participant", p.Identity(), "pID", p.ID())
|
||||
return
|
||||
}
|
||||
|
||||
@@ -771,7 +774,7 @@ func (p *ParticipantImpl) onDataChannel(dc *webrtc.DataChannel) {
|
||||
p.handleDataMessage(livekit.DataPacket_LOSSY, msg.Data)
|
||||
})
|
||||
default:
|
||||
logger.Warnw("unsupported datachannel added", nil, "participant", p.Identity(), "label", dc.Label())
|
||||
logger.Warnw("unsupported datachannel added", nil, "participant", p.Identity(), "pID", p.ID(), "label", dc.Label())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -850,7 +853,7 @@ func (p *ParticipantImpl) handleTrackPublished(track types.PublishedTrack) {
|
||||
|
||||
func (p *ParticipantImpl) handlePublisherICEStateChange(state webrtc.ICEConnectionState) {
|
||||
// logger.Debugw("ICE connection state changed", "state", state.String(),
|
||||
// "participant", p.identity)
|
||||
// "participant", p.identity, "pID", p.ID())
|
||||
if state == webrtc.ICEConnectionStateConnected {
|
||||
p.updateState(livekit.ParticipantInfo_ACTIVE)
|
||||
} else if state == webrtc.ICEConnectionStateFailed {
|
||||
@@ -919,7 +922,7 @@ func (p *ParticipantImpl) downTracksRTCPWorker() {
|
||||
return
|
||||
}
|
||||
logger.Errorw("could not send downtrack reports", err,
|
||||
"participant", p.Identity())
|
||||
"participant", p.Identity(), "pID", p.ID())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -959,7 +962,7 @@ func (p *ParticipantImpl) rtcpSendWorker() {
|
||||
if len(fwdPkts) > 0 {
|
||||
if err := p.publisher.pc.WriteRTCP(fwdPkts); err != nil {
|
||||
logger.Errorw("could not write RTCP to participant", err,
|
||||
"participant", p.Identity())
|
||||
"participant", p.Identity(), "pID", p.ID())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+14
-13
@@ -160,7 +160,7 @@ func (r *Room) Join(participant types.Participant, opts *ParticipantOptions) err
|
||||
// it's important to set this before connection, we don't want to miss out on any publishedTracks
|
||||
participant.OnTrackPublished(r.onTrackPublished)
|
||||
participant.OnStateChange(func(p types.Participant, oldState livekit.ParticipantInfo_State) {
|
||||
logger.Debugw("participant state changed", "state", p.State(), "participant", p.Identity(),
|
||||
logger.Debugw("participant state changed", "state", p.State(), "participant", p.Identity(), "pID", p.ID(),
|
||||
"oldState", oldState)
|
||||
if r.onParticipantChanged != nil {
|
||||
r.onParticipantChanged(participant)
|
||||
@@ -188,9 +188,10 @@ func (r *Room) Join(participant types.Participant, opts *ParticipantOptions) err
|
||||
participant.OnMetadataUpdate(r.onParticipantMetadataUpdate)
|
||||
participant.OnDataPacket(r.onDataPacket)
|
||||
logger.Infow("new participant joined",
|
||||
"id", participant.ID(),
|
||||
"pID", participant.ID(),
|
||||
"participant", participant.Identity(),
|
||||
"roomId", r.Room.Sid)
|
||||
"room", r.Room.Name,
|
||||
"roomID", r.Room.Sid)
|
||||
|
||||
r.participants[participant.Identity()] = participant
|
||||
r.participantOpts[participant.Identity()] = opts
|
||||
@@ -322,7 +323,7 @@ func (r *Room) Close() {
|
||||
if !r.isClosed.TrySet(true) {
|
||||
return
|
||||
}
|
||||
logger.Infow("closing room", "room", r.Room.Sid, "name", r.Room.Name)
|
||||
logger.Infow("closing room", "roomID", r.Room.Sid, "room", r.Room.Name)
|
||||
|
||||
r.statsReporter.RoomEnded()
|
||||
if r.onClose != nil {
|
||||
@@ -384,14 +385,14 @@ func (r *Room) onTrackPublished(participant types.Participant, track types.Publi
|
||||
}
|
||||
|
||||
logger.Debugw("subscribing to new track",
|
||||
"source", participant.Identity(),
|
||||
"remoteTrack", track.ID(),
|
||||
"dest", existingParticipant.Identity())
|
||||
"participants", []string{participant.Identity(), existingParticipant.Identity()},
|
||||
"pIDs", []string{participant.ID(), existingParticipant.ID()},
|
||||
"track", track.ID())
|
||||
if err := track.AddSubscriber(existingParticipant); err != nil {
|
||||
logger.Errorw("could not subscribe to remoteTrack", err,
|
||||
"source", participant.Identity(),
|
||||
"remoteTrack", track.ID(),
|
||||
"dest", existingParticipant.Identity())
|
||||
"participants", []string{participant.Identity(), existingParticipant.Identity()},
|
||||
"pIDs", []string{participant.ID(), existingParticipant.ID()},
|
||||
"track", track.ID())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -458,8 +459,8 @@ func (r *Room) subscribeToExistingTracks(p types.Participant) {
|
||||
if n, err := op.AddSubscriber(p); err != nil {
|
||||
// TODO: log error? or disconnect?
|
||||
logger.Errorw("could not subscribe to participant", err,
|
||||
"dest", p.Identity(),
|
||||
"source", op.Identity())
|
||||
"participants", []string{op.Identity(), p.Identity()},
|
||||
"pIDs", []string{op.ID(), p.ID()})
|
||||
} else {
|
||||
tracksAdded += n
|
||||
}
|
||||
@@ -482,7 +483,7 @@ func (r *Room) broadcastParticipantState(p types.Participant, skipSource bool) {
|
||||
err := op.SendParticipantUpdate(updates)
|
||||
if err != nil {
|
||||
logger.Errorw("could not send update to participant", err,
|
||||
"participant", p.Identity())
|
||||
"participant", p.Identity(), "pID", p.ID())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+18
-12
@@ -113,7 +113,7 @@ func (r *RoomManager) CreateRoom(req *livekit.CreateRoomRequest) (*livekit.Room,
|
||||
nodeId = node.Id
|
||||
}
|
||||
|
||||
logger.Debugw("selected node for room", "room", rm.Name, "node", nodeId)
|
||||
logger.Debugw("selected node for room", "room", rm.Name, "roomID", rm.Sid, "nodeID", nodeId)
|
||||
if err := r.router.SetNodeForRoom(req.Name, nodeId); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -218,7 +218,7 @@ func (r *RoomManager) Stop() {
|
||||
func (r *RoomManager) StartSession(roomName string, pi routing.ParticipantInit, requestSource routing.MessageSource, responseSink routing.MessageSink) {
|
||||
room, err := r.getOrCreateRoom(roomName)
|
||||
if err != nil {
|
||||
logger.Errorw("could not create room", err)
|
||||
logger.Errorw("could not create room", err, "room", roomName)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -229,7 +229,7 @@ func (r *RoomManager) StartSession(roomName string, pi routing.ParticipantInit,
|
||||
if pi.Reconnect {
|
||||
logger.Debugw("resuming RTC session",
|
||||
"room", roomName,
|
||||
"node", r.currentNode.Id,
|
||||
"nodeID", r.currentNode.Id,
|
||||
"participant", pi.Identity,
|
||||
)
|
||||
// close previous sink, and link to new one
|
||||
@@ -270,9 +270,9 @@ func (r *RoomManager) StartSession(roomName string, pi routing.ParticipantInit,
|
||||
|
||||
logger.Debugw("starting RTC session",
|
||||
"room", roomName,
|
||||
"node", r.currentNode.Id,
|
||||
"nodeID", r.currentNode.Id,
|
||||
"participant", pi.Identity,
|
||||
"plan_b", pi.UsePlanB,
|
||||
"planB", pi.UsePlanB,
|
||||
"protocol", pi.ProtocolVersion,
|
||||
)
|
||||
|
||||
@@ -368,7 +368,9 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici
|
||||
defer func() {
|
||||
logger.Debugw("RTC session finishing",
|
||||
"participant", participant.Identity(),
|
||||
"pID", participant.ID(),
|
||||
"room", room.Room.Name,
|
||||
"roomID", room.Room.Sid,
|
||||
)
|
||||
_ = participant.Close()
|
||||
}()
|
||||
@@ -392,32 +394,32 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici
|
||||
case *livekit.SignalRequest_Offer:
|
||||
_, err := participant.HandleOffer(rtc.FromProtoSessionDescription(msg.Offer))
|
||||
if err != nil {
|
||||
logger.Errorw("could not handle offer", err, "participant", participant.Identity())
|
||||
logger.Errorw("could not handle offer", err, "participant", participant.Identity(), "pID", participant.ID())
|
||||
return
|
||||
}
|
||||
case *livekit.SignalRequest_AddTrack:
|
||||
logger.Debugw("add track request", "participant", participant.Identity(),
|
||||
logger.Debugw("add track request", "participant", participant.Identity(), "pID", participant.ID(),
|
||||
"track", msg.AddTrack.Cid)
|
||||
participant.AddTrack(msg.AddTrack)
|
||||
case *livekit.SignalRequest_Answer:
|
||||
if participant.State() == livekit.ParticipantInfo_JOINING {
|
||||
logger.Errorw("cannot negotiate before peer offer", nil, "participant", participant.Identity())
|
||||
logger.Errorw("cannot negotiate before peer offer", nil, "participant", participant.Identity(), "pID", participant.ID())
|
||||
// conn.WriteJSON(jsonError(http.StatusNotAcceptable, "cannot negotiate before peer offer"))
|
||||
return
|
||||
}
|
||||
sd := rtc.FromProtoSessionDescription(msg.Answer)
|
||||
if err := participant.HandleAnswer(sd); err != nil {
|
||||
logger.Errorw("could not handle answer", err, "participant", participant.Identity())
|
||||
logger.Errorw("could not handle answer", err, "participant", participant.Identity(), "pID", participant.ID())
|
||||
}
|
||||
case *livekit.SignalRequest_Trickle:
|
||||
candidateInit, err := rtc.FromProtoTrickle(msg.Trickle)
|
||||
if err != nil {
|
||||
logger.Errorw("could not decode trickle", err, "participant", participant.Identity())
|
||||
logger.Errorw("could not decode trickle", err, "participant", participant.Identity(), "pID", participant.ID())
|
||||
break
|
||||
}
|
||||
// logger.Debugw("adding peer candidate", "participant", participant.ID())
|
||||
// logger.Debugw("adding peer candidate", "participant", participant.Identity())
|
||||
if err := participant.AddICECandidate(candidateInit, msg.Trickle.Target); err != nil {
|
||||
logger.Errorw("could not handle trickle", err, "participant", participant.Identity())
|
||||
logger.Errorw("could not handle trickle", err, "participant", participant.Identity(), "pID", participant.ID())
|
||||
}
|
||||
case *livekit.SignalRequest_Mute:
|
||||
participant.SetTrackMuted(msg.Mute.Sid, msg.Mute.Muted)
|
||||
@@ -425,6 +427,7 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici
|
||||
if err := room.UpdateSubscriptions(participant, msg.Subscription.TrackSids, msg.Subscription.Subscribe); err != nil {
|
||||
logger.Warnw("could not update subscription", err,
|
||||
"participant", participant.Identity(),
|
||||
"pID", participant.ID(),
|
||||
"tracks", msg.Subscription.TrackSids,
|
||||
"subscribe", msg.Subscription.Subscribe)
|
||||
}
|
||||
@@ -436,6 +439,7 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici
|
||||
}
|
||||
logger.Debugw("updating track settings",
|
||||
"participant", participant.Identity(),
|
||||
"pID", participant.ID(),
|
||||
"settings", msg.TrackSetting)
|
||||
subTrack.UpdateSubscriberSettings(!msg.TrackSetting.Disabled, msg.TrackSetting.Quality)
|
||||
}
|
||||
@@ -447,6 +451,7 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici
|
||||
if track.ID() == msg.Simulcast.TrackSid {
|
||||
logger.Debugw("updating simulcast layers",
|
||||
"participant", participant.Identity(),
|
||||
"pID", participant.ID(),
|
||||
"track", track.ID(),
|
||||
"layers", msg.Simulcast.Layers)
|
||||
track.SetSimulcastLayers(msg.Simulcast.Layers)
|
||||
@@ -498,6 +503,7 @@ func (r *RoomManager) handleRTCMessage(roomName, identity string, msg *livekit.R
|
||||
if err := room.UpdateSubscriptions(participant, rm.UpdateSubscriptions.TrackSids, rm.UpdateSubscriptions.Subscribe); err != nil {
|
||||
logger.Warnw("could not update subscription", err,
|
||||
"participant", participant.Identity(),
|
||||
"pID", participant.ID(),
|
||||
"tracks", rm.UpdateSubscriptions.TrackSids,
|
||||
"subscribe", rm.UpdateSubscriptions.Subscribe)
|
||||
}
|
||||
|
||||
@@ -130,7 +130,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
done := make(chan struct{})
|
||||
// function exits when websocket terminates, it'll close the event reading off of response sink as well
|
||||
defer func() {
|
||||
logger.Infow("WS connection closed", "participant", pi.Identity, "connectionId", connId)
|
||||
logger.Infow("WS connection closed", "participant", pi.Identity, "connID", connId)
|
||||
reqSink.Close()
|
||||
close(done)
|
||||
}()
|
||||
@@ -148,9 +148,9 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
logger.Infow("new client WS connected",
|
||||
"connectionId", connId,
|
||||
"room", rm.Sid,
|
||||
"roomName", rm.Name,
|
||||
"connID", connId,
|
||||
"roomID", rm.Sid,
|
||||
"room", rm.Name,
|
||||
"participant", pi.Identity,
|
||||
)
|
||||
|
||||
@@ -170,7 +170,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
if msg == nil {
|
||||
logger.Infow("source closed connection",
|
||||
"participant", pi.Identity,
|
||||
"connectionId", connId)
|
||||
"connID", connId)
|
||||
return
|
||||
}
|
||||
res, ok := msg.(*livekit.SignalResponse)
|
||||
@@ -178,7 +178,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
logger.Errorw("unexpected message type", nil,
|
||||
"type", fmt.Sprintf("%T", msg),
|
||||
"participant", pi.Identity,
|
||||
"connectionId", connId)
|
||||
"connID", connId)
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -206,7 +206,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
if err := reqSink.WriteMessage(req); err != nil {
|
||||
logger.Warnw("error writing to request sink", err,
|
||||
"participant", pi.Identity,
|
||||
"connectionId", connId)
|
||||
"connID", connId)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -151,24 +151,23 @@ func (s *LivekitServer) Start() error {
|
||||
|
||||
go func() {
|
||||
values := []interface{}{
|
||||
"address", s.httpServer.Addr,
|
||||
"node", s.currentNode.Id,
|
||||
"addr", s.httpServer.Addr,
|
||||
"nodeID", s.currentNode.Id,
|
||||
"nodeIP", s.currentNode.Ip,
|
||||
"version", version.Version,
|
||||
}
|
||||
if s.config.RTC.TCPPort != 0 {
|
||||
values = append(values, "rtc.tcp_port", s.config.RTC.TCPPort)
|
||||
values = append(values, "rtc.portTCP", s.config.RTC.TCPPort)
|
||||
}
|
||||
if !s.config.RTC.ForceTCP && s.config.RTC.UDPPort != 0 {
|
||||
values = append(values, "rtc.udp_port", s.config.RTC.UDPPort)
|
||||
values = append(values, "rtc.portUDP", s.config.RTC.UDPPort)
|
||||
} else {
|
||||
values = append(values,
|
||||
"rtc.port_range_start", s.config.RTC.ICEPortRangeStart,
|
||||
"rtc.port_range_end", s.config.RTC.ICEPortRangeEnd,
|
||||
"rtc.portICERange", []uint32{s.config.RTC.ICEPortRangeStart, s.config.RTC.ICEPortRangeEnd},
|
||||
)
|
||||
}
|
||||
if s.config.PrometheusPort != 0 {
|
||||
values = append(values, "prometheus_port", s.config.PrometheusPort)
|
||||
values = append(values, "portPrometheus", s.config.PrometheusPort)
|
||||
}
|
||||
logger.Infow("starting LiveKit server", values...)
|
||||
if err := s.httpServer.Serve(ln); err != http.ErrServerClosed {
|
||||
|
||||
@@ -563,7 +563,7 @@ func (c *RTCClient) processTrack(track *webrtc.TrackRemote) {
|
||||
c.lock.Unlock()
|
||||
|
||||
logger.Debugw("client added track", "participant", c.localParticipant.Identity,
|
||||
"source", pId,
|
||||
"pID", pId,
|
||||
"track", trackId,
|
||||
)
|
||||
|
||||
@@ -593,7 +593,7 @@ func (c *RTCClient) processTrack(track *webrtc.TrackRemote) {
|
||||
numBytes += pkt.MarshalSize()
|
||||
if time.Now().Sub(lastUpdate) > 30*time.Second {
|
||||
logger.Debugw("consumed from participant",
|
||||
"track", trackId, "participant", pId,
|
||||
"track", trackId, "pID", pId,
|
||||
"size", numBytes)
|
||||
lastUpdate = time.Now()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user