From c012c742bd0ba496dd4243ee6866493b5e136a15 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Fri, 5 Feb 2021 17:46:52 -0800 Subject: [PATCH] fix allowing client negotiations prematurely. --- cmd/cli/client/client.go | 7 +-- pkg/routing/interfaces.go | 1 + pkg/routing/localrouter.go | 4 ++ pkg/routing/redis.go | 3 ++ pkg/routing/redisrouter.go | 38 ++++++++++----- pkg/routing/routingfakes/fake_router.go | 65 +++++++++++++++++++++++++ pkg/routing/utils_test.go | 2 +- pkg/rtc/participant.go | 35 +++++++++---- pkg/service/roommanager.go | 10 ++-- pkg/service/rtcservice.go | 4 +- pkg/service/server.go | 7 ++- test/integration_helpers.go | 12 +++-- test/multinode_test.go | 11 ++--- 13 files changed, 157 insertions(+), 42 deletions(-) diff --git a/cmd/cli/client/client.go b/cmd/cli/client/client.go index e04a8b7ee..cf9f9ead6 100644 --- a/cmd/cli/client/client.go +++ b/cmd/cli/client/client.go @@ -127,7 +127,8 @@ func NewRTCClient(conn *websocket.Conn) (*RTCClient, error) { }) peerConn.OnTrack(func(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) { - logger.Debugw("track received", "label", track.StreamID(), "id", track.ID()) + logger.Debugw("track received", "label", track.StreamID(), "id", track.ID(), + "participant", c.localParticipant.Identity) go c.processTrack(track) }) @@ -499,7 +500,7 @@ func (c *RTCClient) handleAnswer(desc webrtc.SessionDescription) error { } func (c *RTCClient) requestNegotiation() error { - logger.Debugw("requesting negotiation") + logger.Debugw("requesting negotiation", "participant", c.localParticipant.Identity) return c.SendRequest(&livekit.SignalRequest{ Message: &livekit.SignalRequest_Negotiate{ Negotiate: &livekit.NegotiationRequest{}, @@ -508,7 +509,7 @@ func (c *RTCClient) requestNegotiation() error { } func (c *RTCClient) negotiate() error { - logger.Debugw("starting negotiation") + logger.Debugw("starting negotiation", "participant", c.localParticipant.Identity) offer, err := c.PeerConn.CreateOffer(nil) if err != nil { return err diff --git a/pkg/routing/interfaces.go b/pkg/routing/interfaces.go index 6d15cc0e9..04640d111 100644 --- a/pkg/routing/interfaces.go +++ b/pkg/routing/interfaces.go @@ -33,6 +33,7 @@ type Router interface { ClearRoomState(roomName string) error RegisterNode() error UnregisterNode() error + RemoveDeadNodes() error GetNode(nodeId string) (*livekit.Node, error) ListNodes() ([]*livekit.Node, error) diff --git a/pkg/routing/localrouter.go b/pkg/routing/localrouter.go index 62ffd9a28..476fa792d 100644 --- a/pkg/routing/localrouter.go +++ b/pkg/routing/localrouter.go @@ -46,6 +46,10 @@ func (r *LocalRouter) UnregisterNode() error { return nil } +func (r *LocalRouter) RemoveDeadNodes() error { + return nil +} + func (r *LocalRouter) GetNode(nodeId string) (*livekit.Node, error) { if nodeId == r.currentNode.Id { return r.currentNode, nil diff --git a/pkg/routing/redis.go b/pkg/routing/redis.go index 069a8a292..1025149a0 100644 --- a/pkg/routing/redis.go +++ b/pkg/routing/redis.go @@ -58,6 +58,9 @@ func publishRTCMessage(rc *redis.Client, nodeId string, participantKey string, m if err != nil { return err } + + //logger.Debugw("publishing to", "rtcChannel", rtcNodeChannel(nodeId), + // "message", rm.Message) return rc.Publish(redisCtx, rtcNodeChannel(nodeId), data).Err() } diff --git a/pkg/routing/redisrouter.go b/pkg/routing/redisrouter.go index 26fc726a0..5cba088e7 100644 --- a/pkg/routing/redisrouter.go +++ b/pkg/routing/redisrouter.go @@ -16,7 +16,7 @@ import ( const ( // expire participant mappings after a day participantMappingTTL = 24 * time.Hour - statsUpdateInterval = 10 * time.Second + statsUpdateInterval = 2 * time.Second ) // RedisRouter uses Redis pub/sub to route signaling messages across different nodes @@ -25,7 +25,6 @@ const ( type RedisRouter struct { LocalRouter rc *redis.Client - cr *utils.CachedRedis ctx context.Context isStarted utils.AtomicFlag @@ -46,7 +45,6 @@ func NewRedisRouter(currentNode LocalNode, rc *redis.Client) *RedisRouter { signalSinks: make(map[string]*SignalNodeSink), } rr.ctx, rr.cancel = context.WithCancel(context.Background()) - rr.cr = utils.NewCachedRedis(rr.ctx, rr.rc) return rr } @@ -55,7 +53,6 @@ func (r *RedisRouter) RegisterNode() error { if err != nil { return err } - r.cr.ExpireHash(NodesKey, r.currentNode.Id) if err := r.rc.HSet(r.ctx, NodesKey, r.currentNode.Id, data).Err(); err != nil { return errors.Wrap(err, "could not register node") } @@ -67,8 +64,23 @@ func (r *RedisRouter) UnregisterNode() error { return r.rc.HDel(context.Background(), NodesKey, r.currentNode.Id).Err() } +func (r *RedisRouter) RemoveDeadNodes() error { + nodes, err := r.ListNodes() + if err != nil { + return err + } + for _, n := range nodes { + if !IsAvailable(n) { + if err := r.rc.HDel(context.Background(), NodesKey, n.Id).Err(); err != nil { + return err + } + } + } + return nil +} + func (r *RedisRouter) GetNodeForRoom(roomName string) (string, error) { - val, err := r.cr.CachedHGet(NodeRoomKey, roomName) + val, err := r.rc.HGet(r.ctx, NodeRoomKey, roomName).Result() if err != nil { err = errors.Wrap(err, "could not get node for room") } @@ -87,7 +99,7 @@ func (r *RedisRouter) ClearRoomState(roomName string) error { } func (r *RedisRouter) GetNode(nodeId string) (*livekit.Node, error) { - data, err := r.cr.CachedHGet(NodesKey, nodeId) + data, err := r.rc.HGet(r.ctx, NodesKey, nodeId).Result() if err != nil { return nil, err } @@ -221,7 +233,6 @@ func (r *RedisRouter) Stop() { } func (r *RedisRouter) setParticipantRTCNode(participantKey, nodeId string) error { - r.cr.Expire(participantRTCKey(participantKey)) err := r.rc.Set(r.ctx, participantRTCKey(participantKey), nodeId, participantMappingTTL).Err() if err != nil { err = errors.Wrap(err, "could not set rtc node") @@ -230,7 +241,6 @@ func (r *RedisRouter) setParticipantRTCNode(participantKey, nodeId string) error } func (r *RedisRouter) setParticipantSignalNode(connectionId, nodeId string) error { - r.cr.Expire(participantSignalKey(connectionId)) if err := r.rc.Set(r.ctx, participantSignalKey(connectionId), nodeId, participantMappingTTL).Err(); err != nil { return errors.Wrap(err, "could not set signal node") } @@ -276,17 +286,17 @@ func (r *RedisRouter) getOrCreateSignalSink(nodeId string, connectionId string) } func (r *RedisRouter) getParticipantRTCNode(participantKey string) (string, error) { - return r.cr.CachedGet(participantRTCKey(participantKey)) + return r.rc.Get(r.ctx, participantRTCKey(participantKey)).Result() } func (r *RedisRouter) getParticipantSignalNode(connectionId string) (nodeId string, err error) { - return r.cr.CachedGet(participantSignalKey(connectionId)) + return r.rc.Get(r.ctx, participantSignalKey(connectionId)).Result() } // update node stats and cleanup func (r *RedisRouter) statsWorker() { for r.ctx.Err() == nil { - // update every 10 seconds + // update periodically seconds select { case <-time.After(statsUpdateInterval): r.currentNode.Stats.UpdatedAt = time.Now().Unix() @@ -343,6 +353,9 @@ func (r *RedisRouter) handleSignalMessage(sm *livekit.SignalNodeMessage) error { switch rmb := sm.Message.(type) { case *livekit.SignalNodeMessage_Response: + //logger.Debugw("forwarding signal message", + // "connectionId", connectionId, + // "type", fmt.Sprintf("%T", rmb.Response.Message)) // in the event the current node is an Signal node, push to response channels resSink := r.getOrCreateMessageChannel(r.responseChannels, connectionId) if err := resSink.WriteMessage(rmb.Response); err != nil { @@ -350,7 +363,8 @@ func (r *RedisRouter) handleSignalMessage(sm *livekit.SignalNodeMessage) error { } case *livekit.SignalNodeMessage_EndSession: - logger.Debugw("received EndSession, closing signal connection") + logger.Debugw("received EndSession, closing signal connection", + "connectionId", connectionId) resSink := r.getOrCreateMessageChannel(r.responseChannels, connectionId) resSink.Close() } diff --git a/pkg/routing/routingfakes/fake_router.go b/pkg/routing/routingfakes/fake_router.go index 882bc3d20..afd88bb5c 100644 --- a/pkg/routing/routingfakes/fake_router.go +++ b/pkg/routing/routingfakes/fake_router.go @@ -73,6 +73,16 @@ type FakeRouter struct { registerNodeReturnsOnCall map[int]struct { result1 error } + RemoveDeadNodesStub func() error + removeDeadNodesMutex sync.RWMutex + removeDeadNodesArgsForCall []struct { + } + removeDeadNodesReturns struct { + result1 error + } + removeDeadNodesReturnsOnCall map[int]struct { + result1 error + } SetNodeForRoomStub func(string, string) error setNodeForRoomMutex sync.RWMutex setNodeForRoomArgsForCall []struct { @@ -460,6 +470,59 @@ func (fake *FakeRouter) RegisterNodeReturnsOnCall(i int, result1 error) { }{result1} } +func (fake *FakeRouter) RemoveDeadNodes() error { + fake.removeDeadNodesMutex.Lock() + ret, specificReturn := fake.removeDeadNodesReturnsOnCall[len(fake.removeDeadNodesArgsForCall)] + fake.removeDeadNodesArgsForCall = append(fake.removeDeadNodesArgsForCall, struct { + }{}) + stub := fake.RemoveDeadNodesStub + fakeReturns := fake.removeDeadNodesReturns + fake.recordInvocation("RemoveDeadNodes", []interface{}{}) + fake.removeDeadNodesMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeRouter) RemoveDeadNodesCallCount() int { + fake.removeDeadNodesMutex.RLock() + defer fake.removeDeadNodesMutex.RUnlock() + return len(fake.removeDeadNodesArgsForCall) +} + +func (fake *FakeRouter) RemoveDeadNodesCalls(stub func() error) { + fake.removeDeadNodesMutex.Lock() + defer fake.removeDeadNodesMutex.Unlock() + fake.RemoveDeadNodesStub = stub +} + +func (fake *FakeRouter) RemoveDeadNodesReturns(result1 error) { + fake.removeDeadNodesMutex.Lock() + defer fake.removeDeadNodesMutex.Unlock() + fake.RemoveDeadNodesStub = nil + fake.removeDeadNodesReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeRouter) RemoveDeadNodesReturnsOnCall(i int, result1 error) { + fake.removeDeadNodesMutex.Lock() + defer fake.removeDeadNodesMutex.Unlock() + fake.RemoveDeadNodesStub = nil + if fake.removeDeadNodesReturnsOnCall == nil { + fake.removeDeadNodesReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.removeDeadNodesReturnsOnCall[i] = struct { + result1 error + }{result1} +} + func (fake *FakeRouter) SetNodeForRoom(arg1 string, arg2 string) error { fake.setNodeForRoomMutex.Lock() ret, specificReturn := fake.setNodeForRoomReturnsOnCall[len(fake.setNodeForRoomArgsForCall)] @@ -736,6 +799,8 @@ func (fake *FakeRouter) Invocations() map[string][][]interface{} { defer fake.onNewParticipantRTCMutex.RUnlock() fake.registerNodeMutex.RLock() defer fake.registerNodeMutex.RUnlock() + fake.removeDeadNodesMutex.RLock() + defer fake.removeDeadNodesMutex.RUnlock() fake.setNodeForRoomMutex.RLock() defer fake.setNodeForRoomMutex.RUnlock() fake.startMutex.RLock() diff --git a/pkg/routing/utils_test.go b/pkg/routing/utils_test.go index 689f2be4c..ca48993c2 100644 --- a/pkg/routing/utils_test.go +++ b/pkg/routing/utils_test.go @@ -14,7 +14,7 @@ func TestIsAvailable(t *testing.T) { t.Run("still available", func(t *testing.T) { n := &livekit.Node{ Stats: &livekit.NodeStats{ - UpdatedAt: time.Now().Unix() - 10, + UpdatedAt: time.Now().Unix() - 3, }, } assert.True(t, routing.IsAvailable(n)) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 15c5b7187..77b6cf554 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -113,7 +113,8 @@ func NewParticipant(identity string, pc types.PeerConnection, rs routing.Message }, }) if err != nil { - logger.Errorw("could not send trickle", "err", err) + logger.Errorw("could not send trickle", "err", err, + "participant", identity) } if participant.onICECandidate != nil { @@ -246,7 +247,7 @@ func (p *ParticipantImpl) Answer(sdp webrtc.SessionDescription) (answer webrtc.S } p.negotiationCond.L.Unlock() - logger.Debugw("sending to client answer", + logger.Debugw("sending answer to client", "participant", p.Identity(), //"sdp", sdp.SDP, ) @@ -279,7 +280,7 @@ func (p *ParticipantImpl) AddTrack(clientId, name string, trackType livekit.Trac } p.pendingTracks[clientId] = ti - p.responseSink.WriteMessage(&livekit.SignalResponse{ + err := p.responseSink.WriteMessage(&livekit.SignalResponse{ Message: &livekit.SignalResponse_TrackPublished{ TrackPublished: &livekit.TrackPublishedResponse{ Cid: clientId, @@ -287,6 +288,10 @@ func (p *ParticipantImpl) AddTrack(clientId, name string, trackType livekit.Trac }, }, }) + if err != nil { + logger.Errorw("could not write message", "error", err, + "participant", p.identity) + } } func (p *ParticipantImpl) HandleAnswer(sdp webrtc.SessionDescription) error { @@ -323,11 +328,15 @@ func (p *ParticipantImpl) HandleClientNegotiation() { logger.Debugw("allowing participant to negotiate", "participant", p.Identity()) - p.responseSink.WriteMessage(&livekit.SignalResponse{ + err := p.responseSink.WriteMessage(&livekit.SignalResponse{ Message: &livekit.SignalResponse_Negotiate{ Negotiate: &livekit.NegotiationResponse{}, }, }) + if err != nil { + logger.Errorw("could not write message", "error", err, + "participant", p.identity) + } } // AddICECandidate adds candidates for remote peer @@ -369,11 +378,16 @@ func (p *ParticipantImpl) Close() error { func (p *ParticipantImpl) AddSubscriber(op types.Participant) error { p.lock.RLock() tracks := funk.Values(p.publishedTracks).([]types.PublishedTrack) - defer p.lock.RUnlock() + p.lock.RUnlock() + + if len(tracks) == 0 { + return nil + } logger.Debugw("subscribing new participant to tracks", "srcParticipant", p.Identity(), - "dstParticipant", op.Identity()) + "newParticipant", op.Identity(), + "numTracks", len(tracks)) for _, track := range tracks { if err := track.AddSubscriber(op); err != nil { @@ -472,13 +486,13 @@ func (p *ParticipantImpl) negotiate() { return } - logger.Debugw("starting server negotiation", "participant", p.Identity()) p.negotiationCond.L.Lock() for p.negotiationState != negotiationStateNone { p.negotiationCond.Wait() - p.negotiationState = negotiationStateServer } + p.negotiationState = negotiationStateServer p.negotiationCond.L.Unlock() + logger.Debugw("starting server negotiation", "participant", p.Identity()) offer, err := p.peerConn.CreateOffer(nil) if err != nil { @@ -508,8 +522,9 @@ func (p *ParticipantImpl) negotiate() { }, }) if err != nil { - logger.Errorw("could not send offer to peer", - "err", err) + logger.Errorw("could not send offer to participant", + "err", err, + "participant", p.identity) } } diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 3d007e7be..500ad078a 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -253,7 +253,7 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici case *livekit.SignalRequest_Offer: _, err := participant.Answer(rtc.FromProtoSessionDescription(msg.Offer)) if err != nil { - logger.Errorw("could not handle join", "err", err, "participant", participant.ID()) + logger.Errorw("could not handle offer", "err", err, "participant", participant.Identity()) return } case *livekit.SignalRequest_AddTrack: @@ -262,13 +262,13 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici participant.AddTrack(msg.AddTrack.Cid, msg.AddTrack.Name, msg.AddTrack.Type) case *livekit.SignalRequest_Answer: if participant.State() == livekit.ParticipantInfo_JOINING { - logger.Errorw("cannot negotiate before peer offer", "participant", participant.ID()) + logger.Errorw("cannot negotiate before peer offer", "participant", participant.Identity()) //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", "participant", participant.ID(), "err", err) + logger.Errorw("could not handle answer", "participant", participant.Identity(), "err", err) //conn.WriteJSON( // jsonError(http.StatusInternalServerError, "could not handle negotiate", err.Error())) return @@ -277,7 +277,7 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici participant.HandleClientNegotiation() case *livekit.SignalRequest_Trickle: if participant.State() == livekit.ParticipantInfo_JOINING { - logger.Errorw("cannot trickle before peer offer", "participant", participant.ID()) + logger.Errorw("cannot trickle before offer", "participant", participant.Identity()) //conn.WriteJSON(jsonError(http.StatusNotAcceptable, "cannot trickle before peer offer")) return } @@ -285,7 +285,7 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici candidateInit := rtc.FromProtoTrickle(msg.Trickle) //logger.Debugw("adding peer candidate", "participant", participant.ID()) if err := participant.AddICECandidate(candidateInit); err != nil { - logger.Errorw("could not handle trickle", "participant", participant.ID(), "err", err) + logger.Errorw("could not handle trickle", "participant", participant.Identity(), "err", err) //conn.WriteJSON( // jsonError(http.StatusInternalServerError, "could not handle trickle", err.Error())) return diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index b2cc367d6..898635397 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -123,7 +123,9 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { } res, ok := msg.(*livekit.SignalResponse) if !ok { - logger.Errorw("unexpected message type", "type", fmt.Sprintf("%T", msg)) + logger.Errorw("unexpected message type", + "type", fmt.Sprintf("%T", msg), + "participant", identity) continue } diff --git a/pkg/service/server.go b/pkg/service/server.go index 1c5ca4aa3..5e2fc1c5f 100644 --- a/pkg/service/server.go +++ b/pkg/service/server.go @@ -69,7 +69,12 @@ func NewLivekitServer(conf *config.Config, router.OnNewParticipantRTC(roomManager.StartSession) // clean up old rooms on startup - err = roomManager.CleanupRooms() + if err = roomManager.CleanupRooms(); err != nil { + return + } + if err = router.RemoveDeadNodes(); err != nil { + return + } return } diff --git a/test/integration_helpers.go b/test/integration_helpers.go index c358ce688..c57bd2f3c 100644 --- a/test/integration_helpers.go +++ b/test/integration_helpers.go @@ -17,6 +17,7 @@ import ( "github.com/livekit/livekit-server/pkg/logger" "github.com/livekit/livekit-server/pkg/routing" "github.com/livekit/livekit-server/pkg/service" + "github.com/livekit/livekit-server/pkg/utils" "github.com/livekit/livekit-server/proto/livekit" ) @@ -61,8 +62,8 @@ func setupSingleNodeTest(roomName string) *service.LivekitServer { } func setupMultiNodeTest() (*service.LivekitServer, *service.LivekitServer) { - s1 := createMultiNodeServer(nodeId1, defaultServerPort) - s2 := createMultiNodeServer(nodeId2, secondServerPort) + s1 := createMultiNodeServer(utils.NewGuid(nodeId1), defaultServerPort) + s2 := createMultiNodeServer(utils.NewGuid(nodeId2), secondServerPort) go s1.Start() go s2.Start() @@ -153,7 +154,7 @@ func createSingleNodeServer() *service.LivekitServer { } currentNode, err := routing.NewLocalNode(conf) - currentNode.Id = nodeId1 + currentNode.Id = utils.NewGuid(nodeId1) if err != nil { panic(err) } @@ -161,6 +162,7 @@ func createSingleNodeServer() *service.LivekitServer { // local routing and store router := routing.NewLocalRouter(currentNode) roomStore := service.NewLocalRoomStore() + roomStore.DeleteRoom(testRoom) s, err := service.InitializeServer(conf, &StaticKeyProvider{}, roomStore, router, currentNode, &routing.RandomSelector{}) if err != nil { panic(fmt.Sprintf("could not create server: %v", err)) @@ -190,9 +192,13 @@ func createMultiNodeServer(nodeId string, port uint32) *service.LivekitServer { if err = rc.Ping(context.Background()).Err(); err != nil { panic(err) } + if err = rc.Del(context.Background(), routing.NodesKey).Err(); err != nil { + panic(err) + } router := routing.NewRedisRouter(currentNode, rc) roomStore := service.NewRedisRoomStore(rc) + roomStore.DeleteRoom(testRoom) s, err := service.InitializeServer(conf, &StaticKeyProvider{}, roomStore, router, currentNode, &routing.RandomSelector{}) if err != nil { panic(fmt.Sprintf("could not create server: %v", err)) diff --git a/test/multinode_test.go b/test/multinode_test.go index dddc8a22b..e95ece2d6 100644 --- a/test/multinode_test.go +++ b/test/multinode_test.go @@ -15,7 +15,7 @@ func TestMultiNodeRouting(t *testing.T) { return } - logger.Infow("---Starting TestMultiNodeRouting---") + logger.Infow("\n\n---Starting TestMultiNodeRouting---") defer logger.Infow("---Finishing TestMultiNodeRouting---") s1, s2 := setupMultiNodeTest() @@ -24,8 +24,7 @@ func TestMultiNodeRouting(t *testing.T) { // creating room on node 1 _, err := roomClient.CreateRoom(contextWithCreateRoomToken(), &livekit.CreateRoomRequest{ - Name: testRoom, - NodeId: nodeId1, + Name: testRoom, }) assert.NoError(t, err) @@ -80,7 +79,7 @@ func TestConnectWithoutCreation(t *testing.T) { t.SkipNow() return } - logger.Infow("---Starting TestConnectWithoutCreation---") + logger.Infow("\n\n---Starting TestConnectWithoutCreation---") defer logger.Infow("---Finishing TestConnectWithoutCreation---") s1, s2 := setupMultiNodeTest() @@ -100,7 +99,7 @@ func TestMultinodePublishingUponJoining(t *testing.T) { return } - logger.Infow("---Starting TestMultinodePublishingUponJoining---") + logger.Infow("\n\n---Starting TestMultinodePublishingUponJoining---") defer logger.Infow("---Finishing TestMultinodePublishingUponJoining---") s1, s2 := setupMultiNodeTest() @@ -116,7 +115,7 @@ func TestMultinodeReceiveBeforePublish(t *testing.T) { return } - logger.Infow("---Starting TestMultinodeReceiveBeforePublish---") + logger.Infow("\n\n---Starting TestMultinodeReceiveBeforePublish---") defer logger.Infow("---Finishing TestMultinodeReceiveBeforePublish---") s1, s2 := setupMultiNodeTest()