Include region in ParticipantInfo (#585)

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

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=

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()

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() {

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()

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 {

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)

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 != "" {