From 80c8c6308a0eef11bd938036e91b374410da3ded Mon Sep 17 00:00:00 2001 From: Yvan <8574526@qq,com> Date: Tue, 22 Sep 2026 20:56:28 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AF=B9=E6=8E=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/integration/live/server.go | 57 +++++++++++++++++++++++++++-- internal/logging/zap.go | 2 + internal/logging/zap_test.go | 2 + 3 files changed, 58 insertions(+), 3 deletions(-) diff --git a/internal/integration/live/server.go b/internal/integration/live/server.go index 154240f..b935dbe 100644 --- a/internal/integration/live/server.go +++ b/internal/integration/live/server.go @@ -137,14 +137,23 @@ func New(hub platformws.Hub, streams *bizlive.StreamUsecase, rooms *bizlive.Room if err != nil { return nil, nil, err } + if logger == nil { + logger = slog.Default() + } + logger = logger.With("mod", "live") m := &Manager{hub: hub, streams: streams, roomCatalog: rooms, chat: chat, logger: logger, api: api, rooms: make(map[string]*room), peers: make(map[*melody.Session]*peer), observers: make(map[*melody.Session]struct{}), stop: make(chan struct{})} hub.OnMessage(m.onMessage) hub.OnDisconnect(m.onDisconnect) go m.heartbeatLoop() + logger.Info("live signaling manager initialized", + "stun_configured", len(liveICEServers()) > 0, + "udp_min_port", envPort("KRA_LIVE_UDP_MIN_PORT"), + "udp_max_port", envPort("KRA_LIVE_UDP_MAX_PORT")) return m, m.Close, nil } func (m *Manager) Close() { + m.liveLogger().Info("live signaling manager stopping") m.stopOnce.Do(func() { close(m.stop) }) m.mu.Lock() peers := make([]*peer, 0, len(m.peers)) @@ -160,6 +169,7 @@ func (m *Manager) Close() { func (m *Manager) onMessage(session *melody.Session, raw []byte) { var msg message if err := json.Unmarshal(raw, &msg); err != nil { + m.liveLogger().Warn("live signaling message rejected", "reason", "invalid_json", "message_bytes", len(raw), "remote_addr", sessionRemoteAddr(session)) m.send(session, "Error", map[string]any{"code": "INVALID_JSON", "message": "invalid json"}) return } @@ -167,6 +177,7 @@ func (m *Manager) onMessage(session *melody.Session, raw []byte) { case "Join": var payload joinPayload if err := json.Unmarshal(msg.Payload, &payload); err != nil { + m.liveLogger().Warn("live join rejected", "code", "INVALID_JOIN", "reason", "invalid_payload", "remote_addr", sessionRemoteAddr(session)) m.send(session, "JoinRejected", map[string]any{"code": "INVALID_JOIN", "message": "invalid join payload"}) return } @@ -174,6 +185,7 @@ func (m *Manager) onMessage(session *melody.Session, raw []byte) { case "Offer": var payload sessionDescriptionPayload if err := json.Unmarshal(msg.Payload, &payload); err != nil { + m.liveLogger().Warn("live offer rejected", "code", "INVALID_OFFER", "reason", "invalid_payload", "remote_addr", sessionRemoteAddr(session)) m.send(session, "Error", map[string]any{"code": "INVALID_OFFER", "message": "invalid offer payload"}) return } @@ -203,6 +215,7 @@ func (m *Manager) join(session *melody.Session, payload joinPayload) { role := strings.ToLower(strings.TrimSpace(payload.Role)) streamID := strings.TrimSpace(payload.StreamID) if role == "observer" { + m.liveLogger().Info("live observer joined", "remote_addr", sessionRemoteAddr(session)) m.mu.Lock() m.observers[session] = struct{}{} m.mu.Unlock() @@ -211,11 +224,14 @@ func (m *Manager) join(session *melody.Session, payload joinPayload) { return } if streamID == "" || (role != "publisher" && role != "subscriber") { + m.liveLogger().Warn("live join rejected", "code", "INVALID_JOIN", "role", role, "stream_id", streamID, "remote_addr", sessionRemoteAddr(session)) m.send(session, "JoinRejected", map[string]any{"code": "INVALID_JOIN", "message": "role and streamId are required"}) return } userID, authenticated := sessionUserID(session) + m.liveLogger().Info("live join requested", "role", role, "stream_id", streamID, "user_id", userID, "authenticated", authenticated, "remote_addr", sessionRemoteAddr(session)) if role == "publisher" && !authenticated { + m.liveLogger().Warn("live join rejected", "code", "AUTH_REQUIRED", "role", role, "stream_id", streamID, "remote_addr", sessionRemoteAddr(session)) m.send(session, "JoinRejected", map[string]any{"code": "AUTH_REQUIRED", "message": "authenticated user is required"}) return } @@ -227,10 +243,12 @@ func (m *Manager) join(session *melody.Session, payload joinPayload) { } if role == "publisher" { if err := m.authorizeRoomPublisher(streamID, userID); err != nil { + m.liveLogger().Warn("live join rejected", "code", "PUBLISH_FORBIDDEN", "role", role, "stream_id", streamID, "user_id", userID, "error", err) m.send(session, "JoinRejected", map[string]any{"code": "PUBLISH_FORBIDDEN", "message": err.Error()}) return } } else if err := m.authorizeRoomSubscriber(streamID); err != nil { + m.liveLogger().Warn("live join rejected", "code", "ROOM_UNAVAILABLE", "role", role, "stream_id", streamID, "user_id", userID, "error", err) m.send(session, "JoinRejected", map[string]any{"code": "ROOM_UNAVAILABLE", "message": err.Error()}) return } @@ -240,10 +258,12 @@ func (m *Manager) join(session *melody.Session, payload joinPayload) { occupied := r.publisher != nil r.mu.RUnlock() if occupied { + m.liveLogger().Warn("live join rejected", "code", "DUPLICATE_PUBLISHER", "role", role, "stream_id", streamID, "user_id", userID) m.send(session, "JoinRejected", map[string]any{"code": "DUPLICATE_PUBLISHER", "message": "stream already has a publisher"}) return } if _, err := m.streams.RegisterPublisher(context.Background(), streamID, userID); err != nil { + m.liveLogger().Warn("live join rejected", "code", errorCode(err), "role", role, "stream_id", streamID, "user_id", userID, "error", err) m.send(session, "JoinRejected", map[string]any{"code": errorCode(err), "message": err.Error()}) return } @@ -259,6 +279,7 @@ func (m *Manager) join(session *melody.Session, payload joinPayload) { m.peers[session] = p m.mu.Unlock() m.send(session, "Joined", map[string]any{"role": role, "streamId": streamID, "mode": "pion-sfu"}) + m.liveLogger().Info("live publisher joined", "stream_id", streamID, "user_id", userID, "remote_addr", sessionRemoteAddr(session)) m.sendPublisherStatus(streamID, false, "publisher_waiting_for_offer") m.broadcastStreamList() return @@ -280,6 +301,7 @@ func (m *Manager) join(session *melody.Session, payload joinPayload) { m.closePeer(old, "logged_in_elsewhere") } if _, err := m.streams.RegisterSubscriber(context.Background(), streamID, userID); err != nil { + m.liveLogger().Warn("live join rejected", "code", errorCode(err), "role", role, "stream_id", streamID, "user_id", userID, "error", err) m.send(session, "JoinRejected", map[string]any{"code": errorCode(err), "message": err.Error()}) return } @@ -297,6 +319,7 @@ func (m *Manager) join(session *melody.Session, payload joinPayload) { publicUserID = 0 } m.send(session, "Joined", map[string]any{"role": role, "streamId": streamID, "subscriberId": p.subscriberID, "userId": publicUserID, "mode": "pion-sfu"}) + m.liveLogger().Info("live subscriber joined", "stream_id", streamID, "user_id", publicUserID, "subscriber_id", p.subscriberID, "authenticated", authenticated, "remote_addr", sessionRemoteAddr(session)) m.sendPublisherStatusTo(session, streamID, ready, "publisher_offline") if ready { m.send(session, "PublisherStatus", map[string]any{"streamId": streamID, "ready": true}) @@ -414,15 +437,19 @@ func (m *Manager) authorizeRoomSubscriber(streamID string) error { func (m *Manager) handleOffer(session *melody.Session, payload sessionDescriptionPayload) { p := m.peer(session) if p == nil || (p.role != "publisher" && p.role != "subscriber") { + m.liveLogger().Warn("live offer rejected", "code", "NOT_JOINED", "remote_addr", sessionRemoteAddr(session)) m.send(session, "Error", map[string]any{"code": "NOT_JOINED", "message": "join a stream first"}) return } if strings.TrimSpace(payload.SDP) == "" { + m.liveLogger().Warn("live offer rejected", "code", "INVALID_SDP", "role", p.role, "stream_id", p.streamID, "user_id", p.userID) m.send(session, "Error", map[string]any{"code": "INVALID_SDP", "message": "sdp is required"}) return } + m.liveLogger().Info("live offer received", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "sdp_bytes", len(payload.SDP)) pc, err := m.api.NewPeerConnection(webrtc.Configuration{ICEServers: liveICEServers()}) if err != nil { + m.liveLogger().Error("live peer connection creation failed", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "error", err) m.send(session, "Error", map[string]any{"code": "PEER_CONNECTION", "message": err.Error()}) return } @@ -449,31 +476,37 @@ func (m *Manager) handleOffer(session *melody.Session, payload sessionDescriptio } } if err = pc.SetRemoteDescription(webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: payload.SDP}); err != nil { + m.liveLogger().Error("live remote description failed", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "error", err) m.send(session, "Error", map[string]any{"code": "REMOTE_DESCRIPTION", "message": err.Error()}) return } answer, err := pc.CreateAnswer(nil) if err != nil { + m.liveLogger().Error("live answer creation failed", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "error", err) m.send(session, "Error", map[string]any{"code": "CREATE_ANSWER", "message": err.Error()}) return } if err = pc.SetLocalDescription(answer); err != nil { + m.liveLogger().Error("live local description failed", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "error", err) m.send(session, "Error", map[string]any{"code": "LOCAL_DESCRIPTION", "message": err.Error()}) return } <-webrtc.GatheringCompletePromise(pc) local := pc.LocalDescription() if local == nil { + m.liveLogger().Error("live local description unavailable", "role", p.role, "stream_id", p.streamID, "user_id", p.userID) m.send(session, "Error", map[string]any{"code": "NO_LOCAL_DESCRIPTION", "message": "local description unavailable"}) return } m.send(session, "Answer", map[string]any{"type": local.Type.String(), "sdp": local.SDP}) + m.liveLogger().Info("live answer sent", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "sdp_bytes", len(local.SDP)) if p.role == "publisher" { m.sendPublisherStatus(p.streamID, false, "waiting_for_tracks") } } func (m *Manager) handleICEConnectionStateChange(p *peer, pc *webrtc.PeerConnection, state webrtc.ICEConnectionState) { + m.liveLogger().Info("live ICE state changed", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "state", state.String()) if state != webrtc.ICEConnectionStateFailed && state != webrtc.ICEConnectionStateClosed { return } @@ -488,6 +521,7 @@ func (m *Manager) onPublisherTrack(p *peer, remote *webrtc.TrackRemote) { return } entry := &publishedTrack{id: remote.ID(), kind: remote.Kind(), local: local} + m.liveLogger().Info("live publisher track received", "stream_id", p.streamID, "user_id", p.userID, "track_id", remote.ID(), "kind", strings.ToLower(remote.Kind().String()), "codec", codec.MimeType) p.room.mu.Lock() p.room.tracks[remote.Kind()] = entry viewers := make([]*peer, 0, len(p.room.viewers)) @@ -513,6 +547,7 @@ func (m *Manager) onPublisherTrack(p *peer, remote *webrtc.TrackRemote) { func (m *Manager) onDisconnect(session *melody.Session) { if p := m.peer(session); p != nil { + m.liveLogger().Info("live signaling disconnected", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "remote_addr", sessionRemoteAddr(session)) m.closePeer(p, "disconnected") return } @@ -558,6 +593,7 @@ func (m *Manager) closePeerConnection(p *peer, candidate *webrtc.PeerConnection, return } p.cleanup.Do(func() { + m.liveLogger().Info("live peer closing", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "reason", reason) if pc != nil { _ = pc.Close() } @@ -721,15 +757,30 @@ func (m *Manager) sendPublisherStatusTo(session *melody.Session, streamID string func (m *Manager) send(session *melody.Session, msgType string, payload any) { encoded, err := json.Marshal(map[string]any{"type": msgType, "payload": payload}) if err != nil { + m.liveLogger().Error("live signaling response encoding failed", "message_type", msgType, "error", err) return } - _ = m.hub.Send(session, encoded) + if err = m.hub.Send(session, encoded); err != nil { + m.liveLogger().Warn("live signaling response send failed", "message_type", msgType, "remote_addr", sessionRemoteAddr(session), "error", err) + } } func (m *Manager) log(message string, args ...any) { - if m.logger != nil { - m.logger.Warn(message, args...) + m.liveLogger().Warn(message, args...) +} + +func (m *Manager) liveLogger() *slog.Logger { + if m != nil && m.logger != nil { + return m.logger } + return slog.Default().With("mod", "live") +} + +func sessionRemoteAddr(session *melody.Session) string { + if session == nil || session.Request == nil { + return "" + } + return session.Request.RemoteAddr } func sessionUserID(session *melody.Session) (uint, bool) { diff --git a/internal/logging/zap.go b/internal/logging/zap.go index 49cb89f..b5571ab 100644 --- a/internal/logging/zap.go +++ b/internal/logging/zap.go @@ -435,6 +435,8 @@ func routedLogPaths(module string, level zapcore.Level) []string { paths = append(paths, filepath.Join("timedTask", "task.log")) case "websocket": paths = append(paths, filepath.Join("websocket", "websocket.log")) + case "live": + paths = append(paths, filepath.Join("live", "live.log")) case "error": paths = append(paths, filepath.Join("error", "error.log")) default: diff --git a/internal/logging/zap_test.go b/internal/logging/zap_test.go index 923637d..373fecb 100644 --- a/internal/logging/zap_test.go +++ b/internal/logging/zap_test.go @@ -42,6 +42,7 @@ func TestZapHandlerRoutesHTTPAndErrorLogs(t *testing.T) { logger := slog.New(handler) logger.Info("request", "mod", "http", "request_id", "req-1") logger.Info("handshake", "mod", "websocket") + logger.Info("publisher joined", "mod", "live") logger.Error("failed", "mod", "users") cleanup() @@ -54,6 +55,7 @@ func TestZapHandlerRoutesHTTPAndErrorLogs(t *testing.T) { filepath.Join(root, date, "application.log"), filepath.Join(root, date, "http", "access.log"), filepath.Join(root, date, "websocket", "websocket.log"), + filepath.Join(root, date, "live", "live.log"), filepath.Join(root, date, "users", "application.log"), filepath.Join(root, date, "error", "error.log"), }