From 46965037904e601b90382cde5a67b5de425836e0 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Thu, 31 Mar 2022 14:57:55 +0530 Subject: [PATCH] Include region in ParticipantInfo (#585) --- go.mod | 2 +- go.sum | 4 +- pkg/routing/interfaces.go | 3 ++ pkg/routing/localrouter.go | 4 ++ pkg/routing/routingfakes/fake_router.go | 65 +++++++++++++++++++++++++ pkg/rtc/participant.go | 2 + pkg/service/roommanager.go | 1 + pkg/service/rtcservice.go | 3 ++ 8 files changed, 81 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index f473e3413..28a7a4ec1 100644 --- a/go.mod +++ b/go.mod @@ -13,7 +13,7 @@ require ( github.com/google/wire v0.5.0 github.com/gorilla/websocket v1.4.2 github.com/hashicorp/golang-lru v0.5.4 - github.com/livekit/protocol v0.13.0 + github.com/livekit/protocol v0.13.1-0.20220331092121-f5e8e205306a github.com/mackerelio/go-osstat v0.2.1 github.com/magefile/mage v1.11.0 github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0 diff --git a/go.sum b/go.sum index 9cd02870f..963fdb5b9 100644 --- a/go.sum +++ b/go.sum @@ -132,8 +132,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY/mKrcvA= github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts= -github.com/livekit/protocol v0.13.0 h1:AdXMeUzdHeJaqBl/EBO2jwaSLNmtW95LSuNeymBuP2c= -github.com/livekit/protocol v0.13.0/go.mod h1:3pHsWUtQmWaH8mG0cXrQWpbf3Vo+kj0U+In77CEXu90= +github.com/livekit/protocol v0.13.1-0.20220331092121-f5e8e205306a h1:634c+R4IECT+TmDp672o4DfWOSMOm93E9DcYulcxltQ= +github.com/livekit/protocol v0.13.1-0.20220331092121-f5e8e205306a/go.mod h1:3pHsWUtQmWaH8mG0cXrQWpbf3Vo+kj0U+In77CEXu90= github.com/mackerelio/go-osstat v0.2.1 h1:5AeAcBEutEErAOlDz6WCkEvm6AKYgHTUQrfwm5RbeQc= github.com/mackerelio/go-osstat v0.2.1/go.mod h1:UzRL8dMCCTqG5WdRtsxbuljMpZt9PCAGXqxPst5QtaY= github.com/magefile/mage v1.11.0 h1:C/55Ywp9BpgVVclD3lRnSYCwXTYxmSppIgLeDYlNuls= diff --git a/pkg/routing/interfaces.go b/pkg/routing/interfaces.go index 303990c48..f51811505 100644 --- a/pkg/routing/interfaces.go +++ b/pkg/routing/interfaces.go @@ -35,6 +35,7 @@ type ParticipantInit struct { AutoSubscribe bool Client *livekit.ClientInfo Grants *auth.ClaimGrants + Region string } type NewParticipantCallback func(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit, requestSource MessageSource, responseSink MessageSink) @@ -55,6 +56,8 @@ type Router interface { SetNodeForRoom(ctx context.Context, roomName livekit.RoomName, nodeId livekit.NodeID) error ClearRoomState(ctx context.Context, roomName livekit.RoomName) error + GetRegion() string + Start() error Drain() Stop() diff --git a/pkg/routing/localrouter.go b/pkg/routing/localrouter.go index 958ce4030..bd17df269 100644 --- a/pkg/routing/localrouter.go +++ b/pkg/routing/localrouter.go @@ -165,6 +165,10 @@ func (r *LocalRouter) Stop() { r.rtcMessageChan.Close() } +func (r *LocalRouter) GetRegion() string { + return r.currentNode.Region +} + func (r *LocalRouter) statsWorker() { for { if !r.isStarted.Load() { diff --git a/pkg/routing/routingfakes/fake_router.go b/pkg/routing/routingfakes/fake_router.go index 0826793cb..190996099 100644 --- a/pkg/routing/routingfakes/fake_router.go +++ b/pkg/routing/routingfakes/fake_router.go @@ -40,6 +40,16 @@ type FakeRouter struct { result1 *livekit.Node result2 error } + GetRegionStub func() string + getRegionMutex sync.RWMutex + getRegionArgsForCall []struct { + } + getRegionReturns struct { + result1 string + } + getRegionReturnsOnCall map[int]struct { + result1 string + } ListNodesStub func() ([]*livekit.Node, error) listNodesMutex sync.RWMutex listNodesArgsForCall []struct { @@ -320,6 +330,59 @@ func (fake *FakeRouter) GetNodeForRoomReturnsOnCall(i int, result1 *livekit.Node }{result1, result2} } +func (fake *FakeRouter) GetRegion() string { + fake.getRegionMutex.Lock() + ret, specificReturn := fake.getRegionReturnsOnCall[len(fake.getRegionArgsForCall)] + fake.getRegionArgsForCall = append(fake.getRegionArgsForCall, struct { + }{}) + stub := fake.GetRegionStub + fakeReturns := fake.getRegionReturns + fake.recordInvocation("GetRegion", []interface{}{}) + fake.getRegionMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeRouter) GetRegionCallCount() int { + fake.getRegionMutex.RLock() + defer fake.getRegionMutex.RUnlock() + return len(fake.getRegionArgsForCall) +} + +func (fake *FakeRouter) GetRegionCalls(stub func() string) { + fake.getRegionMutex.Lock() + defer fake.getRegionMutex.Unlock() + fake.GetRegionStub = stub +} + +func (fake *FakeRouter) GetRegionReturns(result1 string) { + fake.getRegionMutex.Lock() + defer fake.getRegionMutex.Unlock() + fake.GetRegionStub = nil + fake.getRegionReturns = struct { + result1 string + }{result1} +} + +func (fake *FakeRouter) GetRegionReturnsOnCall(i int, result1 string) { + fake.getRegionMutex.Lock() + defer fake.getRegionMutex.Unlock() + fake.GetRegionStub = nil + if fake.getRegionReturnsOnCall == nil { + fake.getRegionReturnsOnCall = make(map[int]struct { + result1 string + }) + } + fake.getRegionReturnsOnCall[i] = struct { + result1 string + }{result1} +} + func (fake *FakeRouter) ListNodes() ([]*livekit.Node, error) { fake.listNodesMutex.Lock() ret, specificReturn := fake.listNodesReturnsOnCall[len(fake.listNodesArgsForCall)] @@ -947,6 +1010,8 @@ func (fake *FakeRouter) Invocations() map[string][][]interface{} { defer fake.drainMutex.RUnlock() fake.getNodeForRoomMutex.RLock() defer fake.getNodeForRoomMutex.RUnlock() + fake.getRegionMutex.RLock() + defer fake.getRegionMutex.RUnlock() fake.listNodesMutex.RLock() defer fake.listNodesMutex.RUnlock() fake.onNewParticipantRTCMutex.RLock() diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 2bb03246c..bec69de9e 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -63,6 +63,7 @@ type ParticipantParams struct { Grants *auth.ClaimGrants InitialVersion uint32 ClientConf *livekit.ClientConfiguration + Region string } type ParticipantImpl struct { @@ -357,6 +358,7 @@ func (p *ParticipantImpl) ToProto() *livekit.ParticipantInfo { JoinedAt: p.ConnectedAt().Unix(), Version: p.version.Inc(), Permission: grants.Video.ToPermission(), + Region: p.params.Region, } info.Tracks = p.UpTrackManager.ToProto() if p.params.Grants != nil { diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 69aa9724a..0b9774d04 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -255,6 +255,7 @@ func (r *RoomManager) StartSession(ctx context.Context, roomName livekit.RoomNam Grants: pi.Grants, Logger: pLogger, ClientConf: clientConf, + Region: pi.Region, }) if err != nil { logger.Errorw("could not create participant", err) diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index cfdbb2eb7..0d09dd942 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -108,7 +108,9 @@ func (s *RTCService) validate(r *http.Request) (livekit.RoomName, routing.Partic claims.Identity += "#" + publishParam } + region := "" if router, ok := s.router.(routing.Router); ok { + region = router.GetRegion() if foundNode, err := router.GetNodeForRoom(r.Context(), roomName); err == nil { if selector.LimitsReached(s.limits, foundNode.Stats) { return "", routing.ParticipantInit{}, http.StatusServiceUnavailable, rtc.ErrLimitExceeded @@ -123,6 +125,7 @@ func (s *RTCService) validate(r *http.Request) (livekit.RoomName, routing.Partic AutoSubscribe: true, Client: s.ParseClientInfo(r), Grants: claims, + Region: region, } if autoSubParam != "" {