Log stats worker when it not closable. (#4313)

pkg/telemetry/events.go

@@ -132,7 +132,6 @@ func (t *telemetryService) ParticipantActive(
 			guard,
 		)
 		if !found {
-			// need to also account for participant count
 			prometheus.AddParticipant()
 		}
 		worker.SetConnected()
@@ -193,6 +192,15 @@ func (t *telemetryService) ParticipantLeft(ctx context.Context,
 			isConnected = worker.IsConnected()
 			if worker.Close(guard) {
 				prometheus.SubParticipant()
+			} else {
+				logger.Infow(
+					"stats worker active",
+					"room", room.Name,
+					"roomID", room.Sid,
+					"participant", participant.Identity,
+					"pID", participant.Sid,
+					"worker", worker,
+				)
 			}
 		}
 

pkg/telemetry/statsworker.go

@@ -19,6 +19,7 @@ import (
 	"sync"
 	"time"
 
+	"go.uber.org/zap/zapcore"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/livekit/livekit-server/pkg/utils"
@@ -51,6 +52,13 @@ func (s *ReferenceCount) Release(guard *ReferenceGuard) bool {
 	return s.count == 0
 }
 
+func (s ReferenceCount) MarshalLogObject(e zapcore.ObjectEncoder) error {
+	e.AddInt("count", s.count)
+	return nil
+}
+
+// ----------------------------------------
+
 // StatsWorker handles participant stats
 type StatsWorker struct {
 	next *StatsWorker
@@ -192,6 +200,20 @@ func (s *StatsWorker) collectStats(
 	return stats
 }
 
+func (s *StatsWorker) MarshalLogObject(e zapcore.ObjectEncoder) error {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
+	e.AddString("room", string(s.roomName))
+	e.AddString("roomID", string(s.roomID))
+	e.AddString("participant", string(s.participantIdentity))
+	e.AddString("pID", string(s.participantID))
+	e.AddBool("isConnected", s.isConnected)
+	e.AddTime("closedAt", s.closedAt)
+	e.AddObject("refCount", s.refCount)
+	return nil
+}
+
 // -------------------------------------------------------------------------
 
 // create a single stream and single video layer post aggregation