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

Integrate logger components (#1933)

Browse Source 
* Integrate logger components

Dividing into the following components
* pub - publisher
* pub.sfu
* sub - subscriber
* transport
* transport.pion
* transport.cc
* api
* webhook

* update go modules
This commit is contained in:
David Zhao
2023-08-03 13:31:17 -07:00
committed by GitHub
parent adf971f5b1
commit debd75fa15
15 changed files with 107 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/server/main.go
+1 -1
View File
@@ -202,7 +202,7 @@ func getConfig(c *cli.Context) (*config.Config, error) {
if err != nil {
return nil, err
}
config.InitLoggerFromConfig(conf.Logging)
config.InitLoggerFromConfig(&conf.Logging)
if c.String("config") == "" && c.String("config-body") == "" && conf.Development {
// use single port UDP when no config is provided
go.mod
+1 -1
View File
@@ -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-20230716190407-fc4944cbc33a
github.com/livekit/protocol v1.5.11-0.20230803081122-d791f31b6e0f
github.com/livekit/protocol v1.6.0
github.com/livekit/psrpc v0.3.2
github.com/mackerelio/go-osstat v0.2.4
github.com/magefile/mage v1.15.0
go.sum
+2 -2
View File
@@ -124,8 +124,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-20230716190407-fc4944cbc33a h1:JWpPHcMFuw0fP4swE89CfMgeUXiSN5IKvCJL/5HLI3A=
github.com/livekit/mediatransportutil v0.0.0-20230716190407-fc4944cbc33a/go.mod h1:xirUXW8xnLGmfCwUeAv/nj1VGo1OO1BmgxrYP7jK/14=
github.com/livekit/protocol v1.5.11-0.20230803081122-d791f31b6e0f h1:nhkzJVWnRePDo6imcFnuIljwU8MTBjcRdzfdLaVdWXw=
github.com/livekit/protocol v1.5.11-0.20230803081122-d791f31b6e0f/go.mod h1:SUS9foM1xBzw/AFrgTJuFX/oSuwlnIbHmpdiPdCvwEM=
github.com/livekit/protocol v1.6.0 h1:19S+vFZqnivKIOpyR3DEK/mSaykQ3UEf7H2G/mBOE54=
github.com/livekit/protocol v1.6.0/go.mod h1:SUS9foM1xBzw/AFrgTJuFX/oSuwlnIbHmpdiPdCvwEM=
github.com/livekit/psrpc v0.3.2 h1:eAaJhASme33gtoBhCRLH9jsnWcdm1tHWf0WzaDk56ew=
github.com/livekit/psrpc v0.3.2/go.mod h1:n6JntEg+zT6Ji8InoyTpV7wusPNwGqqtxmHlkNhDN0U=
github.com/mackerelio/go-osstat v0.2.4 h1:qxGbdPkFo65PXOb/F/nhDKpF2nGmGaCFDLXoZjJTtUs=
pkg/config/config.go
+9 -4
View File
@@ -29,7 +29,6 @@ import (
"github.com/livekit/mediatransportutil/pkg/rtcconfig"
"github.com/livekit/protocol/logger"
"github.com/livekit/protocol/logger/pionlogger"
redisLiveKit "github.com/livekit/protocol/redis"
)
@@ -525,6 +524,13 @@ func NewConfig(confString string, strictMode bool, c *cli.Context, baseFlags []c
if conf.Logging.Level == "" && conf.Development {
conf.Logging.Level = "debug"
}
if conf.Logging.PionLevel != "" {
if conf.Logging.ComponentLevels == nil {
conf.Logging.ComponentLevels = map[string]string{}
}
conf.Logging.ComponentLevels["transport.pion"] = conf.Logging.PionLevel
conf.Logging.ComponentLevels["pion"] = conf.Logging.PionLevel
}
if conf.Development {
conf.Environment = "dev"
@@ -818,7 +824,6 @@ func SetLogger(l logger.Logger) {
logger.SetLogger(l, "livekit")
}
func InitLoggerFromConfig(config LoggingConfig) {
pionlogger.SetLogLevel(config.PionLevel)
logger.InitFromConfig(config.Config, "livekit")
func InitLoggerFromConfig(config *LoggingConfig) {
logger.InitFromConfig(&config.Config, "livekit")
}
pkg/rtc/mediatracksubscriptions.go
+2 -1
View File
@@ -22,6 +22,7 @@ import (
"github.com/pion/webrtc/v3"
"go.uber.org/atomic"
sutils "github.com/livekit/livekit-server/pkg/utils"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
@@ -133,7 +134,7 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr *
PlayoutDelayLimit: sub.GetPlayoutDelayConfig(),
Pacer: sub.GetPacer(),
Trailer: trailer,
Logger: LoggerWithTrack(sub.GetLogger(), trackID, t.params.IsRelayed),
Logger: LoggerWithTrack(sub.GetLogger().WithComponent(sutils.ComponentSub), trackID, t.params.IsRelayed),
})
if err != nil {
return nil, err
pkg/rtc/participant.go
+39 -32
View File
@@ -41,6 +41,7 @@ import (
"github.com/livekit/livekit-server/pkg/sfu/streamallocator"
"github.com/livekit/livekit-server/pkg/telemetry"
"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
sutils "github.com/livekit/livekit-server/pkg/utils"
"github.com/livekit/mediatransportutil/pkg/twcc"
"github.com/livekit/protocol/auth"
"github.com/livekit/protocol/livekit"
@@ -187,6 +188,10 @@ type ParticipantImpl struct {
supervisor *supervisor.ParticipantSupervisor
tracksQuality map[livekit.TrackID]livekit.ConnectionQuality
// loggers for publisher and subscriber
pubLogger logger.Logger
subLogger logger.Logger
}
func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) {
@@ -213,6 +218,8 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) {
params.Telemetry),
supervisor: supervisor.NewParticipantSupervisor(supervisor.ParticipantSupervisorParams{Logger: params.Logger}),
tracksQuality: make(map[livekit.TrackID]livekit.ConnectionQuality),
pubLogger: params.Logger.WithComponent(sutils.ComponentPub),
subLogger: params.Logger.WithComponent(sutils.ComponentSub),
}
p.version.Store(params.InitialVersion)
p.timedVersion.Update(params.VersionGenerator.New())
@@ -1106,7 +1113,7 @@ func (p *ParticipantImpl) setupTransportManager() error {
AllowUDPUnstableFallback: p.params.AllowUDPUnstableFallback,
TURNSEnabled: p.params.TURNSEnabled,
AllowPlayoutDelay: p.params.PlayoutDelay.GetEnabled() && p.SupportSyncStreamID(),
Logger: p.params.Logger,
Logger: p.params.Logger.WithComponent(sutils.ComponentTransport),
})
if err != nil {
return err
@@ -1161,7 +1168,7 @@ func (p *ParticipantImpl) setupTransportManager() error {
func (p *ParticipantImpl) setupUpTrackManager() {
p.UpTrackManager = NewUpTrackManager(UpTrackManagerParams{
SID: p.params.SID,
Logger: p.params.Logger,
Logger: p.pubLogger,
VersionGenerator: p.params.VersionGenerator,
})
@@ -1182,7 +1189,7 @@ func (p *ParticipantImpl) setupUpTrackManager() {
func (p *ParticipantImpl) setupSubscriptionManager() {
p.SubscriptionManager = NewSubscriptionManager(SubscriptionManagerParams{
Participant: p,
Logger: p.params.Logger.WithoutSampler(),
Logger: p.subLogger.WithoutSampler(),
TrackResolver: p.params.TrackResolver,
Telemetry: p.params.Telemetry,
OnTrackSubscribed: p.onTrackSubscribed,
@@ -1269,7 +1276,7 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w
publishedTrack, isNewTrack := p.mediaTrackReceived(track, rtpReceiver)
if publishedTrack == nil {
p.params.Logger.Warnw("webrtc Track published but can't find MediaTrack", nil,
p.pubLogger.Warnw("webrtc Track published but can't find MediaTrack", nil,
"kind", track.Kind().String(),
"webrtcTrackID", track.ID(),
"rid", track.RID(),
@@ -1280,7 +1287,7 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w
}
if !p.CanPublishSource(publishedTrack.Source()) {
p.params.Logger.Warnw("no permission to publish mediaTrack", nil,
p.pubLogger.Warnw("no permission to publish mediaTrack", nil,
"source", publishedTrack.Source(),
)
p.removePublishedTrack(publishedTrack)
@@ -1290,7 +1297,7 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w
p.setIsPublisher(true)
p.dirty.Store(true)
p.params.Logger.Infow("mediaTrack published",
p.pubLogger.Infow("mediaTrack published",
"kind", track.Kind().String(),
"trackID", publishedTrack.ID(),
"webrtcTrackID", track.ID(),
@@ -1318,7 +1325,7 @@ func (p *ParticipantImpl) onDataMessage(kind livekit.DataPacket_Kind, data []byt
dp := livekit.DataPacket{}
if err := proto.Unmarshal(data, &dp); err != nil {
p.params.Logger.Warnw("could not parse data packet", err)
p.pubLogger.Warnw("could not parse data packet", err)
return
}
@@ -1336,7 +1343,7 @@ func (p *ParticipantImpl) onDataMessage(kind livekit.DataPacket_Kind, data []byt
onDataPacket(p, &dp)
}
default:
p.params.Logger.Warnw("received unsupported data packet", nil, "payload", payload)
p.pubLogger.Warnw("received unsupported data packet", nil, "payload", payload)
}
p.setIsPublisher(true)
@@ -1445,7 +1452,7 @@ func (p *ParticipantImpl) subscriberRTCPWorker() {
if IsEOF(err) {
return
}
p.params.Logger.Errorw("could not send down track reports", err)
p.subLogger.Errorw("could not send down track reports", err)
}
pkts = pkts[:0]
@@ -1462,7 +1469,7 @@ func (p *ParticipantImpl) subscriberRTCPWorker() {
if IsEOF(err) {
return
}
p.params.Logger.Errorw("could not send down track reports", err)
p.subLogger.Errorw("could not send down track reports", err)
}
}
@@ -1545,7 +1552,7 @@ func (p *ParticipantImpl) onSubscribedMaxQualityChange(trackID livekit.TrackID,
)
}
p.params.Logger.Infow(
p.pubLogger.Infow(
"sending max subscribed quality",
"trackID", trackID,
"qualities", subscribedQualities,
@@ -1565,7 +1572,7 @@ func (p *ParticipantImpl) addPendingTrackLocked(req *livekit.AddTrackRequest) *l
if req.Sid != "" {
track := p.GetPublishedTrack(livekit.TrackID(req.Sid))
if track == nil {
p.params.Logger.Infow("could not find existing track for multi-codec simulcast", "trackID", req.Sid)
p.pubLogger.Infow("could not find existing track for multi-codec simulcast", "trackID", req.Sid)
return nil
}
@@ -1616,17 +1623,17 @@ func (p *ParticipantImpl) addPendingTrackLocked(req *livekit.AddTrackRequest) *l
} else {
p.pendingTracks[req.Cid].trackInfos = append(p.pendingTracks[req.Cid].trackInfos, ti)
}
p.params.Logger.Infow("pending track queued", "trackID", ti.Sid, "track", ti.String(), "request", req.String())
p.pubLogger.Infow("pending track queued", "trackID", ti.Sid, "track", ti.String(), "request", req.String())
return nil
}
p.pendingTracks[req.Cid] = &pendingTrackInfo{trackInfos: []*livekit.TrackInfo{ti}}
p.params.Logger.Infow("pending track added", "trackID", ti.Sid, "track", ti.String(), "request", req.String())
p.pubLogger.Infow("pending track added", "trackID", ti.Sid, "track", ti.String(), "request", req.String())
return ti
}
func (p *ParticipantImpl) sendTrackPublished(cid string, ti *livekit.TrackInfo) {
p.params.Logger.Debugw("sending track published", "cid", cid, "trackInfo", ti.String())
p.pubLogger.Debugw("sending track published", "cid", cid, "trackInfo", ti.String())
_ = p.writeMessage(&livekit.SignalResponse{
Message: &livekit.SignalResponse_TrackPublished{
TrackPublished: &livekit.TrackPublishedResponse{
@@ -1678,7 +1685,7 @@ func (p *ParticipantImpl) setTrackMuted(trackID livekit.TrackID, muted bool) {
}
if !isPending && track == nil {
p.params.Logger.Warnw("could not locate track", nil, "trackID", trackID)
p.pubLogger.Warnw("could not locate track", nil, "trackID", trackID)
}
}
@@ -1686,7 +1693,7 @@ func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpRecei
p.pendingTracksLock.Lock()
newTrack := false
p.params.Logger.Debugw(
p.pubLogger.Debugw(
"media track received",
"kind", track.Kind().String(),
"trackID", track.ID(),
@@ -1696,7 +1703,7 @@ func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpRecei
)
mid := p.TransportManager.GetPublisherMid(rtpReceiver)
if mid == "" {
p.params.Logger.Warnw("could not get mid for track", nil, "trackID", track.ID())
p.pubLogger.Warnw("could not get mid for track", nil, "trackID", track.ID())
return nil, false
}
@@ -1735,10 +1742,10 @@ func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpRecei
}
func (p *ParticipantImpl) addMigrateMutedTrack(cid string, ti *livekit.TrackInfo) *MediaTrack {
p.params.Logger.Infow("add migrate muted track", "cid", cid, "trackID", ti.Sid, "track", ti.String())
p.pubLogger.Infow("add migrate muted track", "cid", cid, "trackID", ti.Sid, "track", ti.String())
rtpReceiver := p.TransportManager.GetPublisherRTPReceiver(ti.Mid)
if rtpReceiver == nil {
p.params.Logger.Errorw("could not find receiver for migrated track", nil, "trackID", ti.Sid)
p.pubLogger.Errorw("could not find receiver for migrated track", nil, "trackID", ti.Sid)
return nil
}
@@ -1801,7 +1808,7 @@ func (p *ParticipantImpl) addMediaTrack(signalCid string, sdpCid string, ti *liv
AudioConfig: p.params.AudioConfig,
VideoConfig: p.params.VideoConfig,
Telemetry: p.params.Telemetry,
Logger: LoggerWithTrack(p.params.Logger, livekit.TrackID(ti.Sid), false),
Logger: LoggerWithTrack(p.pubLogger, livekit.TrackID(ti.Sid), false),
SubscriberConfig: p.params.Config.Subscriber,
PLIThrottleConfig: p.params.PLIThrottleConfig,
SimTracks: p.params.SimTracks,
@@ -1816,7 +1823,7 @@ func (p *ParticipantImpl) addMediaTrack(signalCid string, sdpCid string, ti *liv
pti := p.pendingTracks[signalCid]
if pti != nil {
if p.pendingPublishingTracks[livekit.TrackID(ti.Sid)] != nil {
p.params.Logger.Infow("unexpected pending publish track", "trackID", ti.Sid)
p.pubLogger.Infow("unexpected pending publish track", "trackID", ti.Sid)
}
p.pendingPublishingTracks[livekit.TrackID(ti.Sid)] = &pendingTrackInfo{
trackInfos: []*livekit.TrackInfo{pti.trackInfos[0]},
@@ -1855,7 +1862,7 @@ func (p *ParticipantImpl) addMediaTrack(signalCid string, sdpCid string, ti *liv
if !p.IsClosed() {
// unpublished events aren't necessary when participant is closed
p.params.Logger.Infow("unpublished track", "trackID", ti.Sid, "trackInfo", ti)
p.pubLogger.Infow("unpublished track", "trackID", ti.Sid, "trackInfo", ti)
p.lock.RLock()
onTrackUnpublished := p.onTrackUnpublished
p.lock.RUnlock()
@@ -1952,7 +1959,7 @@ func (p *ParticipantImpl) getPendingTrack(clientId string, kind livekit.TrackTyp
// if still not found, we are done
if pendingInfo == nil {
p.params.Logger.Errorw("track info not published prior to track", nil, "clientId", clientId)
p.pubLogger.Errorw("track info not published prior to track", nil, "clientId", clientId)
return signalCid, nil
}
@@ -2030,7 +2037,7 @@ func (p *ParticipantImpl) getPublishedTrackBySignalCid(clientId string) types.Me
func (p *ParticipantImpl) getPublishedTrackBySdpCid(clientId string) types.MediaTrack {
for _, publishedTrack := range p.GetPublishedTracks() {
if publishedTrack.(types.LocalMediaTrack).HasSdpCid(clientId) {
p.params.Logger.Debugw("found track by sdp cid", "sdpCid", clientId, "trackID", publishedTrack.ID())
p.pubLogger.Debugw("found track by SDP cid", "sdpCid", clientId, "trackID", publishedTrack.ID())
return publishedTrack
}
}
@@ -2048,13 +2055,13 @@ func (p *ParticipantImpl) publisherRTCPWorker() {
// read from rtcpChan
for pkts := range p.rtcpCh {
if pkts == nil {
p.params.Logger.Debugw("exiting publisher RTCP worker")
p.pubLogger.Debugw("exiting publisher RTCP worker")
return
}
if err := p.TransportManager.WritePublisherRTCP(pkts); err != nil {
if !IsEOF(err) {
p.params.Logger.Errorw("could not write RTCP to participant", err)
p.pubLogger.Errorw("could not write RTCP to participant", err)
}
}
}
@@ -2106,7 +2113,7 @@ func (p *ParticipantImpl) setDowntracksConnected() {
func (p *ParticipantImpl) CacheDownTrack(trackID livekit.TrackID, rtpTransceiver *webrtc.RTPTransceiver, downTrack sfu.DownTrackState) {
p.lock.Lock()
if existing := p.cachedDownTracks[trackID]; existing != nil && existing.transceiver != rtpTransceiver {
p.params.Logger.Infow("cached transceiver changed", "trackID", trackID)
p.subLogger.Infow("cached transceiver changed", "trackID", trackID)
}
p.cachedDownTracks[trackID] = &downTrackState{transceiver: rtpTransceiver, downTrack: downTrack}
p.lock.Unlock()
@@ -2162,7 +2169,7 @@ func (p *ParticipantImpl) IssueFullReconnect(reason types.ParticipantCloseReason
func (p *ParticipantImpl) onPublicationError(trackID livekit.TrackID) {
if p.params.ReconnectOnPublicationError {
p.params.Logger.Infow("issuing full reconnect on publication error", "trackID", trackID)
p.pubLogger.Infow("issuing full reconnect on publication error", "trackID", trackID)
p.IssueFullReconnect(types.ParticipantCloseReasonPublicationError)
}
}
@@ -2186,7 +2193,7 @@ func (p *ParticipantImpl) onSubscriptionError(trackID livekit.TrackID, fatal boo
})
if p.params.ReconnectOnSubscriptionError && fatal {
p.params.Logger.Infow("issuing full reconnect on subscription error", "trackID", trackID)
p.subLogger.Infow("issuing full reconnect on subscription error", "trackID", trackID)
p.IssueFullReconnect(types.ParticipantCloseReasonSubscriptionError)
}
}
@@ -2201,7 +2208,7 @@ func (p *ParticipantImpl) onAnyTransportNegotiationFailed() {
func (p *ParticipantImpl) UpdateSubscribedQuality(nodeID livekit.NodeID, trackID livekit.TrackID, maxQualities []types.SubscribedCodecQuality) error {
track := p.GetPublishedTrack(trackID)
if track == nil {
p.params.Logger.Warnw("could not find track", nil, "trackID", trackID)
p.pubLogger.Warnw("could not find track", nil, "trackID", trackID)
return errors.New("could not find published track")
}
@@ -2212,7 +2219,7 @@ func (p *ParticipantImpl) UpdateSubscribedQuality(nodeID livekit.NodeID, trackID
func (p *ParticipantImpl) UpdateMediaLoss(nodeID livekit.NodeID, trackID livekit.TrackID, fractionalLoss uint32) error {
track := p.GetPublishedTrack(trackID)
if track == nil {
p.params.Logger.Warnw("could not find track", nil, "trackID", trackID)
p.pubLogger.Warnw("could not find track", nil, "trackID", trackID)
return errors.New("could not find published track")
}
pkg/rtc/room.go
+8 -3
View File
@@ -27,6 +27,7 @@ import (
"google.golang.org/protobuf/proto"
"github.com/livekit/livekit-server/pkg/sfu/connectionquality"
sutils "github.com/livekit/livekit-server/pkg/utils"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/livekit/protocol/utils"
@@ -113,9 +114,13 @@ func NewRoom(
egressLauncher EgressLauncher,
) *Room {
r := &Room{
protoRoom: proto.Clone(room).(*livekit.Room),
internal: internal,
Logger: LoggerWithRoom(logger.GetLogger(), livekit.RoomName(room.Name), livekit.RoomID(room.Sid)),
protoRoom: proto.Clone(room).(*livekit.Room),
internal: internal,
Logger: LoggerWithRoom(
logger.GetLogger().WithComponent(sutils.ComponentRoom),
livekit.RoomName(room.Name),
livekit.RoomID(room.Sid),
),
config: config,
audioConfig: audioConfig,
telemetry: telemetry,
pkg/rtc/room_test.go
+1 -4
View File
@@ -24,7 +24,6 @@ import (
"github.com/livekit/livekit-server/version"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/livekit/protocol/webhook"
"github.com/livekit/livekit-server/pkg/config"
@@ -43,9 +42,7 @@ const (
)
func init() {
config.InitLoggerFromConfig(config.LoggingConfig{
Config: logger.Config{Level: "debug"},
})
config.InitLoggerFromConfig(&config.DefaultConfig.Logging)
// allow immediate closure in testing
RoomDepartureGrace = 1
roomUpdateInterval = defaultDelay
pkg/rtc/subscribedtrack.go
+2 -1
View File
@@ -22,6 +22,7 @@ import (
"github.com/pion/webrtc/v3"
"go.uber.org/atomic"
sutils "github.com/livekit/livekit-server/pkg/utils"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
@@ -64,7 +65,7 @@ type SubscribedTrack struct {
func NewSubscribedTrack(params SubscribedTrackParams) *SubscribedTrack {
s := &SubscribedTrack{
params: params,
logger: params.Subscriber.GetLogger().WithValues(
logger: params.Subscriber.GetLogger().WithComponent(sutils.ComponentSub).WithValues(
"trackID", params.DownTrack.ID(),
"publisherID", params.PublisherID,
"publisher", params.PublisherIdentity,
pkg/rtc/transport.go
+2 -1
View File
@@ -34,6 +34,7 @@ import (
"github.com/pkg/errors"
"go.uber.org/atomic"
sutils "github.com/livekit/livekit-server/pkg/utils"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/livekit/protocol/logger/pionlogger"
@@ -391,7 +392,7 @@ func NewPCTransport(params TransportParams) (*PCTransport, error) {
if params.IsSendSide {
t.streamAllocator = streamallocator.NewStreamAllocator(streamallocator.StreamAllocatorParams{
Config: params.CongestionControlConfig,
Logger: params.Logger,
Logger: params.Logger.WithComponent(sutils.ComponentCongestionControl),
})
t.streamAllocator.Start()
t.pacer = pacer.NewPassThrough(params.Logger)
pkg/service/roommanager.go
+5 -1
View File
@@ -335,7 +335,11 @@ func (r *RoomManager) StartSession(
rtcConf := *r.rtcConfig
rtcConf.SetBufferFactory(room.GetBufferFactory())
sid := livekit.ParticipantID(utils.NewGuid(utils.ParticipantPrefix))
pLogger := rtc.LoggerWithParticipant(room.Logger, pi.Identity, sid, false)
pLogger := rtc.LoggerWithParticipant(
rtc.LoggerWithRoom(logger.GetLogger(), room.Name(), room.ID()),
pi.Identity,
sid,
false)
// default allow forceTCP
allowFallback := true
if r.config.RTC.AllowTCPFallback != nil {
pkg/service/server.go
+2 -1
View File
@@ -37,6 +37,7 @@ 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"
@@ -102,7 +103,7 @@ func NewLivekitServer(conf *config.Config,
middlewares = append(middlewares, NewAPIKeyAuthMiddleware(keyProvider))
}
twirpLoggingHook := TwirpLogger(logger.GetLogger())
twirpLoggingHook := TwirpLogger(logger.GetLogger().WithComponent(sutils.ComponentAPI))
twirpRequestStatusHook := TwirpRequestStatusReporter()
roomServer := livekit.NewRoomServiceServer(roomService, twirpLoggingHook)
egressServer := livekit.NewEgressServer(egressService, twirp.WithServerHooks(
pkg/sfu/buffer/buffer.go
+4 -3
View File
@@ -30,6 +30,7 @@ import (
"go.uber.org/atomic"
"github.com/livekit/livekit-server/pkg/sfu/audio"
sutils "github.com/livekit/livekit-server/pkg/utils"
"github.com/livekit/mediatransportutil"
"github.com/livekit/mediatransportutil/pkg/bucket"
"github.com/livekit/mediatransportutil/pkg/nack"
@@ -124,7 +125,7 @@ func NewBuffer(ssrc uint32, vp, ap *sync.Pool) *Buffer {
videoPool: vp,
audioPool: ap,
pliThrottle: int64(500 * time.Millisecond),
logger: l,
logger: l.WithComponent(sutils.ComponentPub).WithComponent(sutils.ComponentSFU),
}
b.extPackets.SetMinCapacity(7)
return b
@@ -134,9 +135,9 @@ func (b *Buffer) SetLogger(logger logger.Logger) {
b.Lock()
defer b.Unlock()
b.logger = logger
b.logger = logger.WithComponent(sutils.ComponentSFU)
if b.rtpStats != nil {
b.rtpStats.SetLogger(logger)
b.rtpStats.SetLogger(b.logger)
}
}
pkg/utils/logging.go
+28
View File
@@ -0,0 +1,28 @@
/*
* Copyright 2023 LiveKit, Inc
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package utils
const (
ComponentPub = "pub"
ComponentSub = "sub"
ComponentRoom = "room"
ComponentAPI = "api"
ComponentTransport = "transport"
ComponentSFU = "sfu"
// transport subcomponents
ComponentCongestionControl = "cc"
)
test/integration_helpers.go
+1 -3
View File
@@ -55,9 +55,7 @@ const (
var roomClient livekit.RoomService
func init() {
config.InitLoggerFromConfig(config.LoggingConfig{
Config: logger.Config{Level: "debug"},
})
config.InitLoggerFromConfig(&config.DefaultConfig.Logging)
prometheus.Init("test", livekit.NodeType_SERVER, "test")
}