diff --git a/go.mod b/go.mod index bd908f8e6..a6eb44451 100644 --- a/go.mod +++ b/go.mod @@ -19,7 +19,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 github.com/livekit/mediatransportutil v0.0.0-20240730083616-559fa5ece598 - github.com/livekit/protocol v1.19.4-0.20240813090210-bbd53d6fbfdd + github.com/livekit/protocol v1.20.1-0.20240813123848-0072ee0c6e47 github.com/livekit/psrpc v0.5.3-0.20240616012458-ac39c8549a0a github.com/mackerelio/go-osstat v0.2.5 github.com/magefile/mage v1.15.0 diff --git a/go.sum b/go.sum index d86874dcd..c2ab06e6f 100644 --- a/go.sum +++ b/go.sum @@ -167,8 +167,8 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20240730083616-559fa5ece598 h1:yLlkHk2feSLHstD9n4VKg7YEBR4rLODTI4WE8gNBEnQ= github.com/livekit/mediatransportutil v0.0.0-20240730083616-559fa5ece598/go.mod h1:jwKUCmObuiEDH0iiuJHaGMXwRs3RjrB4G6qqgkr/5oE= -github.com/livekit/protocol v1.19.4-0.20240813090210-bbd53d6fbfdd h1:ejCQXUVF/sj16cRIwR0cN/zf0OBeWfQ9OAy6yEUYohc= -github.com/livekit/protocol v1.19.4-0.20240813090210-bbd53d6fbfdd/go.mod h1:AFuwk3+uIWFeO5ohKjx5w606Djl940+wktaZ441VoCI= +github.com/livekit/protocol v1.20.1-0.20240813123848-0072ee0c6e47 h1:Hsur+//Q0Ll/JfWydKpXxjwDF0FnTzJYOsL0D7s8sAc= +github.com/livekit/protocol v1.20.1-0.20240813123848-0072ee0c6e47/go.mod h1:AFuwk3+uIWFeO5ohKjx5w606Djl940+wktaZ441VoCI= github.com/livekit/psrpc v0.5.3-0.20240616012458-ac39c8549a0a h1:EQAHmcYEGlc6V517cQ3Iy0+jHgP6+tM/B4l2vGuLpQo= github.com/livekit/psrpc v0.5.3-0.20240616012458-ac39c8549a0a/go.mod h1:CQUBSPfYYAaevg1TNCc6/aYsa8DJH4jSRFdCeSZk5u0= github.com/mackerelio/go-osstat v0.2.5 h1:+MqTbZUhoIt4m8qzkVoXUJg1EuifwlAJSk4Yl2GXh+o= diff --git a/pkg/config/config.go b/pkg/config/config.go index d7a856815..a8b03f598 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -242,6 +242,9 @@ type RoomConfig struct { EnableRemoteUnmute bool `yaml:"enable_remote_unmute,omitempty"` PlayoutDelay PlayoutDelayConfig `yaml:"playout_delay,omitempty"` SyncStreams bool `yaml:"sync_streams,omitempty"` + CreateRoomEnabled bool `yaml:"create_room_enabled,omitempty"` + CreateRoomTimeout time.Duration `yaml:"create_room_timeout,omitempty"` + CreateRoomAttempts int `yaml:"create_room_attempts,omitempty"` // deprecated, moved to limits MaxMetadataSize uint32 `yaml:"max_metadata_size,omitempty"` // deprecated, moved to limits @@ -537,8 +540,10 @@ var DefaultConfig = Config{ {Mime: webrtc.MimeTypeVP9}, {Mime: webrtc.MimeTypeAV1}, }, - EmptyTimeout: 5 * 60, - DepartureTimeout: 20, + EmptyTimeout: 5 * 60, + DepartureTimeout: 20, + CreateRoomTimeout: 10 * time.Second, + CreateRoomAttempts: 3, }, Limit: LimitConfig{ MaxMetadataSize: 64000, diff --git a/pkg/routing/interfaces.go b/pkg/routing/interfaces.go index 6286cab74..6fa66f70b 100644 --- a/pkg/routing/interfaces.go +++ b/pkg/routing/interfaces.go @@ -62,6 +62,7 @@ type ParticipantInit struct { ID livekit.ParticipantID SubscriberAllowPause *bool DisableICELite bool + CreateRoom *livekit.CreateRoomRequest } // Router allows multiple nodes to coordinate the participant session @@ -96,12 +97,20 @@ type StartParticipantSignalResults struct { } type MessageRouter interface { + // CreateRoom starts an rtc room + CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (res *livekit.Room, err error) // StartParticipantSignal participant signal connection is ready to start StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (res StartParticipantSignalResults, err error) } -func CreateRouter(rc redis.UniversalClient, node LocalNode, signalClient SignalClient, kps rpc.KeepalivePubSub) Router { - lr := NewLocalRouter(node, signalClient) +func CreateRouter( + rc redis.UniversalClient, + node LocalNode, + signalClient SignalClient, + roomManagerClient RoomManagerClient, + kps rpc.KeepalivePubSub, +) Router { + lr := NewLocalRouter(node, signalClient, roomManagerClient) if rc != nil { return NewRedisRouter(lr, rc, kps) @@ -132,6 +141,7 @@ func (pi *ParticipantInit) ToStartSession(roomName livekit.RoomName, connectionI AdaptiveStream: pi.AdaptiveStream, ParticipantId: string(pi.ID), DisableIceLite: pi.DisableICELite, + CreateRoom: pi.CreateRoom, } if pi.SubscriberAllowPause != nil { subscriberAllowPause := *pi.SubscriberAllowPause diff --git a/pkg/routing/localrouter.go b/pkg/routing/localrouter.go index 85279ead0..182f862a8 100644 --- a/pkg/routing/localrouter.go +++ b/pkg/routing/localrouter.go @@ -30,8 +30,9 @@ var _ Router = (*LocalRouter)(nil) // a router of messages on the same node, basic implementation for local testing type LocalRouter struct { - currentNode LocalNode - signalClient SignalClient + currentNode LocalNode + signalClient SignalClient + roomManagerClient RoomManagerClient lock sync.RWMutex // channels for each participant @@ -40,12 +41,17 @@ type LocalRouter struct { isStarted atomic.Bool } -func NewLocalRouter(currentNode LocalNode, signalClient SignalClient) *LocalRouter { +func NewLocalRouter( + currentNode LocalNode, + signalClient SignalClient, + roomManagerClient RoomManagerClient, +) *LocalRouter { return &LocalRouter{ - currentNode: currentNode, - signalClient: signalClient, - requestChannels: make(map[string]*MessageChannel), - responseChannels: make(map[string]*MessageChannel), + currentNode: currentNode, + signalClient: signalClient, + roomManagerClient: roomManagerClient, + requestChannels: make(map[string]*MessageChannel), + responseChannels: make(map[string]*MessageChannel), } } @@ -89,6 +95,14 @@ func (r *LocalRouter) ListNodes() ([]*livekit.Node, error) { }, nil } +func (r *LocalRouter) CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (res *livekit.Room, err error) { + return r.CreateRoomWithNodeID(ctx, req, livekit.NodeID(r.currentNode.Id)) +} + +func (r *LocalRouter) CreateRoomWithNodeID(ctx context.Context, req *livekit.CreateRoomRequest, nodeID livekit.NodeID) (res *livekit.Room, err error) { + return r.roomManagerClient.CreateRoom(ctx, nodeID, req) +} + func (r *LocalRouter) StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (res StartParticipantSignalResults, err error) { return r.StartParticipantSignalWithNodeID(ctx, roomName, pi, livekit.NodeID(r.currentNode.Id)) } diff --git a/pkg/routing/redisrouter.go b/pkg/routing/redisrouter.go index da40a85a9..85e56dbfd 100644 --- a/pkg/routing/redisrouter.go +++ b/pkg/routing/redisrouter.go @@ -109,6 +109,7 @@ func (r *RedisRouter) RemoveDeadNodes() error { return nil } +// GetNodeForRoom finds the node where the room is hosted at func (r *RedisRouter) GetNodeForRoom(_ context.Context, roomName livekit.RoomName) (*livekit.Node, error) { nodeID, err := r.rc.HGet(r.ctx, NodeRoomKey, string(roomName)).Result() if err == redis.Nil { @@ -161,9 +162,17 @@ func (r *RedisRouter) ListNodes() ([]*livekit.Node, error) { return nodes, nil } +func (r *RedisRouter) CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (res *livekit.Room, err error) { + rtcNode, err := r.GetNodeForRoom(ctx, livekit.RoomName(req.Name)) + if err != nil { + return + } + + return r.CreateRoomWithNodeID(ctx, req, livekit.NodeID(rtcNode.Id)) +} + // StartParticipantSignal signal connection sets up paths to the RTC node, and starts to route messages to that message queue func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit) (res StartParticipantSignalResults, err error) { - // find the node where the room is hosted at rtcNode, err := r.GetNodeForRoom(ctx, roomName) if err != nil { return diff --git a/pkg/routing/roommanager.go b/pkg/routing/roommanager.go new file mode 100644 index 000000000..0df0846c3 --- /dev/null +++ b/pkg/routing/roommanager.go @@ -0,0 +1,61 @@ +// Copyright 2023 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package routing + +import ( + "context" + + "github.com/livekit/livekit-server/pkg/config" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/rpc" + "github.com/livekit/psrpc" + "github.com/livekit/psrpc/pkg/middleware" +) + +//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate + +//counterfeiter:generate . RoomManagerClient +type RoomManagerClient interface { + rpc.TypedRoomManagerClient +} + +type roomManagerClient struct { + config config.RoomConfig + client rpc.TypedRoomManagerClient +} + +func NewRoomManagerClient(clientParams rpc.ClientParams, config config.RoomConfig) (RoomManagerClient, error) { + c, err := rpc.NewTypedRoomManagerClient( + clientParams.Bus, + psrpc.WithClientChannelSize(clientParams.BufferSize), + middleware.WithClientMetrics(clientParams.Observer), + rpc.WithClientLogger(clientParams.Logger), + ) + if err != nil { + return nil, err + } + + return &roomManagerClient{ + config: config, + client: c, + }, nil +} + +func (c *roomManagerClient) CreateRoom(ctx context.Context, nodeID livekit.NodeID, req *livekit.CreateRoomRequest, opts ...psrpc.RequestOption) (*livekit.Room, error) { + return c.client.CreateRoom(ctx, nodeID, req, append(opts, psrpc.WithRequestInterceptors(middleware.NewRPCRetryInterceptor(middleware.RetryOptions{ + MaxAttempts: c.config.CreateRoomAttempts, + Timeout: c.config.CreateRoomTimeout, + })))...) +} diff --git a/pkg/routing/routingfakes/fake_room_manager_client.go b/pkg/routing/routingfakes/fake_room_manager_client.go new file mode 100644 index 000000000..a1df758d3 --- /dev/null +++ b/pkg/routing/routingfakes/fake_room_manager_client.go @@ -0,0 +1,125 @@ +// Code generated by counterfeiter. DO NOT EDIT. +package routingfakes + +import ( + "context" + "sync" + + "github.com/livekit/livekit-server/pkg/routing" + "github.com/livekit/protocol/livekit" + "github.com/livekit/psrpc" +) + +type FakeRoomManagerClient struct { + CreateRoomStub func(context.Context, livekit.NodeID, *livekit.CreateRoomRequest, ...psrpc.RequestOption) (*livekit.Room, error) + createRoomMutex sync.RWMutex + createRoomArgsForCall []struct { + arg1 context.Context + arg2 livekit.NodeID + arg3 *livekit.CreateRoomRequest + arg4 []psrpc.RequestOption + } + createRoomReturns struct { + result1 *livekit.Room + result2 error + } + createRoomReturnsOnCall map[int]struct { + result1 *livekit.Room + result2 error + } + invocations map[string][][]interface{} + invocationsMutex sync.RWMutex +} + +func (fake *FakeRoomManagerClient) CreateRoom(arg1 context.Context, arg2 livekit.NodeID, arg3 *livekit.CreateRoomRequest, arg4 ...psrpc.RequestOption) (*livekit.Room, error) { + fake.createRoomMutex.Lock() + ret, specificReturn := fake.createRoomReturnsOnCall[len(fake.createRoomArgsForCall)] + fake.createRoomArgsForCall = append(fake.createRoomArgsForCall, struct { + arg1 context.Context + arg2 livekit.NodeID + arg3 *livekit.CreateRoomRequest + arg4 []psrpc.RequestOption + }{arg1, arg2, arg3, arg4}) + stub := fake.CreateRoomStub + fakeReturns := fake.createRoomReturns + fake.recordInvocation("CreateRoom", []interface{}{arg1, arg2, arg3, arg4}) + fake.createRoomMutex.Unlock() + if stub != nil { + return stub(arg1, arg2, arg3, arg4...) + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *FakeRoomManagerClient) CreateRoomCallCount() int { + fake.createRoomMutex.RLock() + defer fake.createRoomMutex.RUnlock() + return len(fake.createRoomArgsForCall) +} + +func (fake *FakeRoomManagerClient) CreateRoomCalls(stub func(context.Context, livekit.NodeID, *livekit.CreateRoomRequest, ...psrpc.RequestOption) (*livekit.Room, error)) { + fake.createRoomMutex.Lock() + defer fake.createRoomMutex.Unlock() + fake.CreateRoomStub = stub +} + +func (fake *FakeRoomManagerClient) CreateRoomArgsForCall(i int) (context.Context, livekit.NodeID, *livekit.CreateRoomRequest, []psrpc.RequestOption) { + fake.createRoomMutex.RLock() + defer fake.createRoomMutex.RUnlock() + argsForCall := fake.createRoomArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4 +} + +func (fake *FakeRoomManagerClient) CreateRoomReturns(result1 *livekit.Room, result2 error) { + fake.createRoomMutex.Lock() + defer fake.createRoomMutex.Unlock() + fake.CreateRoomStub = nil + fake.createRoomReturns = struct { + result1 *livekit.Room + result2 error + }{result1, result2} +} + +func (fake *FakeRoomManagerClient) CreateRoomReturnsOnCall(i int, result1 *livekit.Room, result2 error) { + fake.createRoomMutex.Lock() + defer fake.createRoomMutex.Unlock() + fake.CreateRoomStub = nil + if fake.createRoomReturnsOnCall == nil { + fake.createRoomReturnsOnCall = make(map[int]struct { + result1 *livekit.Room + result2 error + }) + } + fake.createRoomReturnsOnCall[i] = struct { + result1 *livekit.Room + result2 error + }{result1, result2} +} + +func (fake *FakeRoomManagerClient) Invocations() map[string][][]interface{} { + fake.invocationsMutex.RLock() + defer fake.invocationsMutex.RUnlock() + fake.createRoomMutex.RLock() + defer fake.createRoomMutex.RUnlock() + copiedInvocations := map[string][][]interface{}{} + for key, value := range fake.invocations { + copiedInvocations[key] = value + } + return copiedInvocations +} + +func (fake *FakeRoomManagerClient) recordInvocation(key string, args []interface{}) { + fake.invocationsMutex.Lock() + defer fake.invocationsMutex.Unlock() + if fake.invocations == nil { + fake.invocations = map[string][][]interface{}{} + } + if fake.invocations[key] == nil { + fake.invocations[key] = [][]interface{}{} + } + fake.invocations[key] = append(fake.invocations[key], args) +} + +var _ routing.RoomManagerClient = new(FakeRoomManagerClient) diff --git a/pkg/routing/routingfakes/fake_router.go b/pkg/routing/routingfakes/fake_router.go index 88fa9bf58..ca8377511 100644 --- a/pkg/routing/routingfakes/fake_router.go +++ b/pkg/routing/routingfakes/fake_router.go @@ -22,6 +22,20 @@ type FakeRouter struct { clearRoomStateReturnsOnCall map[int]struct { result1 error } + CreateRoomStub func(context.Context, *livekit.CreateRoomRequest) (*livekit.Room, error) + createRoomMutex sync.RWMutex + createRoomArgsForCall []struct { + arg1 context.Context + arg2 *livekit.CreateRoomRequest + } + createRoomReturns struct { + result1 *livekit.Room + result2 error + } + createRoomReturnsOnCall map[int]struct { + result1 *livekit.Room + result2 error + } DrainStub func() drainMutex sync.RWMutex drainArgsForCall []struct { @@ -200,6 +214,71 @@ func (fake *FakeRouter) ClearRoomStateReturnsOnCall(i int, result1 error) { }{result1} } +func (fake *FakeRouter) CreateRoom(arg1 context.Context, arg2 *livekit.CreateRoomRequest) (*livekit.Room, error) { + fake.createRoomMutex.Lock() + ret, specificReturn := fake.createRoomReturnsOnCall[len(fake.createRoomArgsForCall)] + fake.createRoomArgsForCall = append(fake.createRoomArgsForCall, struct { + arg1 context.Context + arg2 *livekit.CreateRoomRequest + }{arg1, arg2}) + stub := fake.CreateRoomStub + fakeReturns := fake.createRoomReturns + fake.recordInvocation("CreateRoom", []interface{}{arg1, arg2}) + fake.createRoomMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *FakeRouter) CreateRoomCallCount() int { + fake.createRoomMutex.RLock() + defer fake.createRoomMutex.RUnlock() + return len(fake.createRoomArgsForCall) +} + +func (fake *FakeRouter) CreateRoomCalls(stub func(context.Context, *livekit.CreateRoomRequest) (*livekit.Room, error)) { + fake.createRoomMutex.Lock() + defer fake.createRoomMutex.Unlock() + fake.CreateRoomStub = stub +} + +func (fake *FakeRouter) CreateRoomArgsForCall(i int) (context.Context, *livekit.CreateRoomRequest) { + fake.createRoomMutex.RLock() + defer fake.createRoomMutex.RUnlock() + argsForCall := fake.createRoomArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeRouter) CreateRoomReturns(result1 *livekit.Room, result2 error) { + fake.createRoomMutex.Lock() + defer fake.createRoomMutex.Unlock() + fake.CreateRoomStub = nil + fake.createRoomReturns = struct { + result1 *livekit.Room + result2 error + }{result1, result2} +} + +func (fake *FakeRouter) CreateRoomReturnsOnCall(i int, result1 *livekit.Room, result2 error) { + fake.createRoomMutex.Lock() + defer fake.createRoomMutex.Unlock() + fake.CreateRoomStub = nil + if fake.createRoomReturnsOnCall == nil { + fake.createRoomReturnsOnCall = make(map[int]struct { + result1 *livekit.Room + result2 error + }) + } + fake.createRoomReturnsOnCall[i] = struct { + result1 *livekit.Room + result2 error + }{result1, result2} +} + func (fake *FakeRouter) Drain() { fake.drainMutex.Lock() fake.drainArgsForCall = append(fake.drainArgsForCall, struct { @@ -768,6 +847,8 @@ func (fake *FakeRouter) Invocations() map[string][][]interface{} { defer fake.invocationsMutex.RUnlock() fake.clearRoomStateMutex.RLock() defer fake.clearRoomStateMutex.RUnlock() + fake.createRoomMutex.RLock() + defer fake.createRoomMutex.RUnlock() fake.drainMutex.RLock() defer fake.drainMutex.RUnlock() fake.getNodeForRoomMutex.RLock() diff --git a/pkg/service/interfaces.go b/pkg/service/interfaces.go index c55998fbc..decaa13ee 100644 --- a/pkg/service/interfaces.go +++ b/pkg/service/interfaces.go @@ -73,7 +73,9 @@ type IngressStore interface { //counterfeiter:generate . RoomAllocator type RoomAllocator interface { - CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (*livekit.Room, bool, error) + CreateRoomEnabled() bool + SelectRoomNode(ctx context.Context, roomName livekit.RoomName, nodeID livekit.NodeID) error + CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (*livekit.Room, *livekit.RoomInternal, bool, error) ValidateCreateRoom(ctx context.Context, roomName livekit.RoomName) error } diff --git a/pkg/service/roomallocator.go b/pkg/service/roomallocator.go index 2670985e5..474944a34 100644 --- a/pkg/service/roomallocator.go +++ b/pkg/service/roomallocator.go @@ -53,12 +53,16 @@ func NewRoomAllocator(conf *config.Config, router routing.Router, rs ObjectStore }, nil } +func (r *StandardRoomAllocator) CreateRoomEnabled() bool { + return r.config.Room.CreateRoomEnabled +} + // CreateRoom creates a new room from a request and allocates it to a node to handle // it'll also monitor its state, and cleans it up when appropriate -func (r *StandardRoomAllocator) CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (*livekit.Room, bool, error) { +func (r *StandardRoomAllocator) CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (*livekit.Room, *livekit.RoomInternal, bool, error) { token, err := r.roomStore.LockRoom(ctx, livekit.RoomName(req.Name), 5*time.Second) if err != nil { - return nil, false, err + return nil, nil, false, err } defer func() { _ = r.roomStore.UnlockRoom(ctx, livekit.RoomName(req.Name), token) @@ -78,12 +82,12 @@ func (r *StandardRoomAllocator) CreateRoom(ctx context.Context, req *livekit.Cre internal = &livekit.RoomInternal{} applyDefaultRoomConfig(rm, internal, &r.config.Room) } else if err != nil { - return nil, false, err + return nil, nil, false, err } req, err = r.applyNamedRoomConfiguration(req) if err != nil { - return nil, false, err + return nil, nil, false, err } if req.EmptyTimeout > 0 { @@ -121,48 +125,51 @@ func (r *StandardRoomAllocator) CreateRoom(ctx context.Context, req *livekit.Cre } if err = r.roomStore.StoreRoom(ctx, rm, internal); err != nil { - return nil, false, err + return nil, nil, false, err } + return rm, internal, created, nil +} + +func (r *StandardRoomAllocator) SelectRoomNode(ctx context.Context, roomName livekit.RoomName, nodeID livekit.NodeID) error { // check if room already assigned - existing, err := r.router.GetNodeForRoom(ctx, livekit.RoomName(rm.Name)) + existing, err := r.router.GetNodeForRoom(ctx, roomName) if !errors.Is(err, routing.ErrNotFound) && err != nil { - return nil, false, err + return err } // if already assigned and still available, keep it on that node if err == nil && selector.IsAvailable(existing) { // if node hosting the room is full, deny entry if selector.LimitsReached(r.config.Limit, existing.Stats) { - return nil, false, routing.ErrNodeLimitReached + return routing.ErrNodeLimitReached } - return rm, created, nil + return nil } // select a new node - nodeID := livekit.NodeID(req.NodeId) if nodeID == "" { nodes, err := r.router.ListNodes() if err != nil { - return nil, false, err + return err } node, err := r.selector.SelectNode(nodes) if err != nil { - return nil, false, err + return err } nodeID = livekit.NodeID(node.Id) } - logger.Infow("selected node for room", "room", rm.Name, "roomID", rm.Sid, "selectedNodeID", nodeID) - err = r.router.SetNodeForRoom(ctx, livekit.RoomName(rm.Name), nodeID) + logger.Infow("selected node for room", "room", roomName, "selectedNodeID", nodeID) + err = r.router.SetNodeForRoom(ctx, roomName, nodeID) if err != nil { - return nil, false, err + return err } - return rm, true, nil + return nil } func (r *StandardRoomAllocator) ValidateCreateRoom(ctx context.Context, roomName livekit.RoomName) error { diff --git a/pkg/service/roomallocator_test.go b/pkg/service/roomallocator_test.go index dc917fbcf..41736ba1b 100644 --- a/pkg/service/roomallocator_test.go +++ b/pkg/service/roomallocator_test.go @@ -39,13 +39,15 @@ func TestCreateRoom(t *testing.T) { ra, conf := newTestRoomAllocator(t, conf, node) - room, _, err := ra.CreateRoom(context.Background(), &livekit.CreateRoomRequest{Name: "myroom"}) + room, _, _, err := ra.CreateRoom(context.Background(), &livekit.CreateRoomRequest{Name: "myroom"}) require.NoError(t, err) require.Equal(t, conf.Room.EmptyTimeout, room.EmptyTimeout) require.Equal(t, conf.Room.DepartureTimeout, room.DepartureTimeout) require.NotEmpty(t, room.EnabledCodecs) }) +} +func SelectRoomNode(t *testing.T) { t.Run("reject new participants when track limit has been reached", func(t *testing.T) { conf, err := config.NewConfig("", true, nil, nil) require.NoError(t, err) @@ -58,7 +60,7 @@ func TestCreateRoom(t *testing.T) { ra, _ := newTestRoomAllocator(t, conf, node) - _, _, err = ra.CreateRoom(context.Background(), &livekit.CreateRoomRequest{Name: "low-limit-room"}) + err = ra.SelectRoomNode(context.Background(), "low-limit-room", "") require.ErrorIs(t, err, routing.ErrNodeLimitReached) }) @@ -74,7 +76,7 @@ func TestCreateRoom(t *testing.T) { ra, _ := newTestRoomAllocator(t, conf, node) - _, _, err = ra.CreateRoom(context.Background(), &livekit.CreateRoomRequest{Name: "low-limit-room"}) + err = ra.SelectRoomNode(context.Background(), "low-limit-room", "") require.ErrorIs(t, err, routing.ErrNodeLimitReached) }) } diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 7024d9fe9..2615028d6 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -36,6 +36,7 @@ import ( "github.com/livekit/protocol/utils/guid" "github.com/livekit/protocol/utils/must" "github.com/livekit/psrpc" + "github.com/livekit/psrpc/pkg/middleware" "github.com/livekit/livekit-server/pkg/clientconfiguration" "github.com/livekit/livekit-server/pkg/config" @@ -70,6 +71,8 @@ type RoomManager struct { serverInfo *livekit.ServerInfo currentNode routing.LocalNode router routing.Router + roomAllocator RoomAllocator + roomManagerServer rpc.TypedRoomManagerServer roomStore ObjectStore telemetry telemetry.TelemetryService clientConfManager clientconfiguration.ClientConfigurationManager @@ -96,6 +99,7 @@ func NewLocalRoomManager( roomStore ObjectStore, currentNode routing.LocalNode, router routing.Router, + roomAllocator RoomAllocator, telemetry telemetry.TelemetryService, clientConfManager clientconfiguration.ClientConfigurationManager, agentClient agent.Client, @@ -111,11 +115,12 @@ func NewLocalRoomManager( return nil, err } - return &RoomManager{ + r := &RoomManager{ config: conf, rtcConfig: rtcConf, currentNode: currentNode, router: router, + roomAllocator: roomAllocator, roomStore: roomStore, telemetry: telemetry, clientConfManager: clientConfManager, @@ -139,7 +144,17 @@ func NewLocalRoomManager( Region: conf.Region, NodeId: currentNode.Id, }, - }, nil + } + + r.roomManagerServer, err = rpc.NewTypedRoomManagerServer(r, bus, rpc.WithServerLogger(logger.GetLogger()), middleware.WithServerMetrics(rpc.PSRPCMetricsObserver{}), psrpc.WithServerChannelSize(conf.PSRPC.BufferSize)) + if err != nil { + return nil, err + } + if err := r.roomManagerServer.RegisterAllNodeTopics(livekit.NodeID(currentNode.Id)); err != nil { + return nil, err + } + + return r, nil } func (r *RoomManager) GetRoom(_ context.Context, roomName livekit.RoomName) *rtc.Room { @@ -229,6 +244,7 @@ func (r *RoomManager) Stop() { room.Close(types.ParticipantCloseReasonRoomManagerStop) } + r.roomManagerServer.Kill() r.roomServers.Kill() r.agentDispatchServers.Kill() r.participantServers.Kill() @@ -249,17 +265,27 @@ func (r *RoomManager) Stop() { } } +func (r *RoomManager) CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (*livekit.Room, error) { + room, err := r.getOrCreateRoom(ctx, req) + if err != nil { + return nil, err + } + defer room.Release() + + return room.ToProto(), nil +} + // StartSession starts WebRTC session when a new participant is connected, takes place on RTC node func (r *RoomManager) StartSession( ctx context.Context, - roomName livekit.RoomName, + createRoom *livekit.CreateRoomRequest, pi routing.ParticipantInit, requestSource routing.MessageSource, responseSink routing.MessageSink, ) error { sessionStartTime := time.Now() - room, err := r.getOrCreateRoom(ctx, roomName) + room, err := r.getOrCreateRoom(ctx, createRoom) if err != nil { return err } @@ -288,7 +314,7 @@ func (r *RoomManager) StartSession( // // It is possible that the client did not get that send request. So, send it again. logger.Infow("cannot restart a closed participant", - "room", roomName, + "room", room.Name(), "nodeID", r.currentNode.Id, "participant", pi.Identity, "reason", pi.ReconnectReason, @@ -320,7 +346,7 @@ func (r *RoomManager) StartSession( "reason", pi.ReconnectReason, "numParticipants", room.GetParticipantCount(), ) - iceConfig := r.getIceConfig(roomName, participant) + iceConfig := r.getIceConfig(room.Name(), participant) if err = room.ResumeParticipant( participant, requestSource, @@ -369,7 +395,7 @@ func (r *RoomManager) StartSession( } logger.Debugw("starting RTC session", - "room", roomName, + "room", room.Name(), "nodeID", r.currentNode.Id, "participant", pi.Identity, "clientInfo", logger.Proto(pi.Client), @@ -464,7 +490,7 @@ func (r *RoomManager) StartSession( if err != nil { return err } - iceConfig := r.setIceConfig(roomName, participant) + iceConfig := r.setIceConfig(room.Name(), participant) // join room opts := rtc.ParticipantOptions{ @@ -477,7 +503,7 @@ func (r *RoomManager) StartSession( return err } - participantTopic := rpc.FormatParticipantTopic(roomName, participant.Identity()) + participantTopic := rpc.FormatParticipantTopic(room.Name(), participant.Identity()) participantServer := must.Get(rpc.NewTypedParticipantServer(r, r.bus)) killParticipantServer := r.participantServers.Replace(participantTopic, participantServer) if err := participantServer.RegisterAllParticipantTopics(participantTopic); err != nil { @@ -487,7 +513,7 @@ func (r *RoomManager) StartSession( return err } - if err = r.roomStore.StoreParticipant(ctx, roomName, participant.ToProto()); err != nil { + if err = r.roomStore.StoreParticipant(ctx, room.Name(), participant.ToProto()); err != nil { pLogger.Errorw("could not store participant", err) } @@ -508,7 +534,7 @@ func (r *RoomManager) StartSession( participant.OnClose(func(p types.LocalParticipant) { killParticipantServer() - if err := r.roomStore.DeleteParticipant(ctx, roomName, p.Identity()); err != nil { + if err := r.roomStore.DeleteParticipant(ctx, room.Name(), p.Identity()); err != nil { pLogger.Errorw("could not delete participant", err) } @@ -524,7 +550,7 @@ func (r *RoomManager) StartSession( } }) participant.OnICEConfigChanged(func(participant types.LocalParticipant, iceConfig *livekit.ICEConfig) { - r.iceConfigCache.Put(iceConfigCacheKey{roomName, participant.Identity()}, iceConfig) + r.iceConfigCache.Put(iceConfigCacheKey{room.Name(), participant.Identity()}, iceConfig) }) go r.rtcSessionWorker(room, participant, requestSource) @@ -532,7 +558,9 @@ func (r *RoomManager) StartSession( } // create the actual room object, to be used on RTC node -func (r *RoomManager) getOrCreateRoom(ctx context.Context, roomName livekit.RoomName) (*rtc.Room, error) { +func (r *RoomManager) getOrCreateRoom(ctx context.Context, createRoom *livekit.CreateRoomRequest) (*rtc.Room, error) { + roomName := livekit.RoomName(createRoom.Name) + r.lock.RLock() lastSeenRoom := r.rooms[roomName] r.lock.RUnlock() @@ -542,7 +570,7 @@ func (r *RoomManager) getOrCreateRoom(ctx context.Context, roomName livekit.Room } // create new room, get details first - ri, internal, err := r.roomStore.LoadRoom(ctx, roomName, true) + ri, internal, created, err := r.roomAllocator.CreateRoom(ctx, createRoom) if err != nil { return nil, err } @@ -617,6 +645,21 @@ func (r *RoomManager) getOrCreateRoom(ctx context.Context, roomName livekit.Room r.telemetry.RoomStarted(ctx, newRoom.ToProto()) prometheus.RoomStarted() + if created && createRoom.GetEgress().GetRoom() != nil { + // ensure room name matches + createRoom.Egress.Room.RoomName = createRoom.Name + _, err = r.egressLauncher.StartEgress(ctx, &rpc.StartEgressRequest{ + Request: &rpc.StartEgressRequest_RoomComposite{ + RoomComposite: createRoom.Egress.Room, + }, + RoomId: ri.Sid, + }) + if err != nil { + newRoom.Release() + return nil, err + } + } + return newRoom, nil } diff --git a/pkg/service/roomservice.go b/pkg/service/roomservice.go index 8c817b860..42cecc74b 100644 --- a/pkg/service/roomservice.go +++ b/pkg/service/roomservice.go @@ -36,7 +36,6 @@ import ( type RoomService struct { limitConf config.LimitConfig apiConf config.APIConfig - psrpcConf rpc.PSRPCConfig router routing.MessageRouter roomAllocator RoomAllocator roomStore ServiceStore @@ -50,7 +49,6 @@ type RoomService struct { func NewRoomService( limitConf config.LimitConfig, apiConf config.APIConfig, - psrpcConf rpc.PSRPCConfig, router routing.MessageRouter, roomAllocator RoomAllocator, serviceStore ServiceStore, @@ -63,7 +61,6 @@ func NewRoomService( svc = &RoomService{ limitConf: limitConf, apiConf: apiConf, - psrpcConf: psrpcConf, router: router, roomAllocator: roomAllocator, roomStore: serviceStore, @@ -88,11 +85,24 @@ func (s *RoomService) CreateRoom(ctx context.Context, req *livekit.CreateRoomReq return nil, fmt.Errorf("%w: max length %d", ErrRoomNameExceedsLimits, s.limitConf.MaxRoomNameLength) } - rm, created, err := s.roomAllocator.CreateRoom(ctx, req) + if s.roomAllocator.CreateRoomEnabled() { + err := s.roomAllocator.SelectRoomNode(ctx, livekit.RoomName(req.Name), livekit.NodeID(req.NodeId)) + if err != nil { + return nil, err + } + + return s.router.CreateRoom(ctx, req) + } + + rm, _, created, err := s.roomAllocator.CreateRoom(ctx, req) if err != nil { err = errors.Wrap(err, "could not create room") return nil, err } + err = s.roomAllocator.SelectRoomNode(ctx, livekit.RoomName(req.Name), livekit.NodeID(req.NodeId)) + if err != nil { + return nil, err + } done, err := s.startRoom(ctx, livekit.RoomName(req.Name)) if err != nil { @@ -153,11 +163,18 @@ func (s *RoomService) DeleteRoom(ctx context.Context, req *livekit.DeleteRoomReq return nil, err } - done, err := s.startRoom(ctx, livekit.RoomName(req.Room)) - if err != nil { - return nil, err + if s.roomAllocator.CreateRoomEnabled() { + _, err := s.router.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: req.Room}) + if err != nil { + return nil, err + } + } else { + done, err := s.startRoom(ctx, livekit.RoomName(req.Room)) + if err != nil { + return nil, err + } + defer done() } - defer done() _, err = s.roomClient.DeleteRoom(ctx, s.topicFormatter.RoomTopic(ctx, livekit.RoomName(req.Room)), req) if err != nil { diff --git a/pkg/service/roomservice_test.go b/pkg/service/roomservice_test.go index 71a6a4f73..8c9d19fc3 100644 --- a/pkg/service/roomservice_test.go +++ b/pkg/service/roomservice_test.go @@ -111,7 +111,6 @@ func newTestRoomService(limitConf config.LimitConfig) *TestRoomService { svc, err := service.NewRoomService( limitConf, config.APIConfig{ExecutionTimeout: 2}, - rpc.PSRPCConfig{}, router, allocator, store, diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index aaba72990..f92da4a8e 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -179,6 +179,10 @@ func (s *RTCService) validate(r *http.Request) (livekit.RoomName, routing.Partic Client: s.ParseClientInfo(r), Grants: claims, Region: region, + CreateRoom: &livekit.CreateRoomRequest{ + Name: string(roomName), + ConfigName: GetRoomConfiguration(r.Context()), + }, } if pi.Reconnect { pi.ID = livekit.ParticipantID(participantID) @@ -214,14 +218,13 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } - // for logger - loggerFields := []interface{}{ + loggerFields := []any{ "participant", pi.Identity, + "pID", pi.ID, "room", roomName, "remote", false, } - - l := utils.GetLogger(r.Context()) + pLogger := utils.GetLogger(r.Context()).WithValues(loggerFields...) // give it a few attempts to start session var cr connectionResult @@ -234,8 +237,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { break } if i < 2 { - fieldsWithAttempt := append(loggerFields, "attempt", i) - l.Warnw("failed to start connection, retrying", err, fieldsWithAttempt...) + pLogger.Warnw("failed to start connection, retrying", err, "attempt", i) } } @@ -257,13 +259,6 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { signalStats.ResolveParticipant(join.GetParticipant()) } - pLogger := rtc.LoggerWithParticipant( - rtc.LoggerWithRoom(l, roomName, livekit.RoomID(cr.Room.Sid)), - pi.Identity, - pi.ID, - false, - ) - closedByClient := atomic.NewBool(false) done := make(chan struct{}) // function exits when websocket terminates, it'll close the event reading off of request sink and response source as well @@ -519,8 +514,14 @@ func (s *RTCService) startConnection( var cr connectionResult var err error - cr.Room, _, err = s.roomAllocator.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: string(roomName), ConfigName: GetRoomConfiguration(ctx)}) - if err != nil { + if !s.roomAllocator.CreateRoomEnabled() { + cr.Room, _, _, err = s.roomAllocator.CreateRoom(ctx, pi.CreateRoom) + if err != nil { + return cr, nil, err + } + } + + if err := s.roomAllocator.SelectRoomNode(ctx, roomName, ""); err != nil { return cr, nil, err } diff --git a/pkg/service/servicefakes/fake_room_allocator.go b/pkg/service/servicefakes/fake_room_allocator.go index 134b9649b..e41349e15 100644 --- a/pkg/service/servicefakes/fake_room_allocator.go +++ b/pkg/service/servicefakes/fake_room_allocator.go @@ -10,7 +10,7 @@ import ( ) type FakeRoomAllocator struct { - CreateRoomStub func(context.Context, *livekit.CreateRoomRequest) (*livekit.Room, bool, error) + CreateRoomStub func(context.Context, *livekit.CreateRoomRequest) (*livekit.Room, *livekit.RoomInternal, bool, error) createRoomMutex sync.RWMutex createRoomArgsForCall []struct { arg1 context.Context @@ -18,13 +18,38 @@ type FakeRoomAllocator struct { } createRoomReturns struct { result1 *livekit.Room - result2 bool - result3 error + result2 *livekit.RoomInternal + result3 bool + result4 error } createRoomReturnsOnCall map[int]struct { result1 *livekit.Room - result2 bool - result3 error + result2 *livekit.RoomInternal + result3 bool + result4 error + } + SelectRoomNodeStub func(context.Context, livekit.RoomName, livekit.NodeID) error + selectRoomNodeMutex sync.RWMutex + selectRoomNodeArgsForCall []struct { + arg1 context.Context + arg2 livekit.RoomName + arg3 livekit.NodeID + } + selectRoomNodeReturns struct { + result1 error + } + selectRoomNodeReturnsOnCall map[int]struct { + result1 error + } + StartRoomEnabledStub func() bool + startRoomEnabledMutex sync.RWMutex + startRoomEnabledArgsForCall []struct { + } + startRoomEnabledReturns struct { + result1 bool + } + startRoomEnabledReturnsOnCall map[int]struct { + result1 bool } ValidateCreateRoomStub func(context.Context, livekit.RoomName) error validateCreateRoomMutex sync.RWMutex @@ -42,7 +67,7 @@ type FakeRoomAllocator struct { invocationsMutex sync.RWMutex } -func (fake *FakeRoomAllocator) CreateRoom(arg1 context.Context, arg2 *livekit.CreateRoomRequest) (*livekit.Room, bool, error) { +func (fake *FakeRoomAllocator) CreateRoom(arg1 context.Context, arg2 *livekit.CreateRoomRequest) (*livekit.Room, *livekit.RoomInternal, bool, error) { fake.createRoomMutex.Lock() ret, specificReturn := fake.createRoomReturnsOnCall[len(fake.createRoomArgsForCall)] fake.createRoomArgsForCall = append(fake.createRoomArgsForCall, struct { @@ -57,9 +82,9 @@ func (fake *FakeRoomAllocator) CreateRoom(arg1 context.Context, arg2 *livekit.Cr return stub(arg1, arg2) } if specificReturn { - return ret.result1, ret.result2, ret.result3 + return ret.result1, ret.result2, ret.result3, ret.result4 } - return fakeReturns.result1, fakeReturns.result2, fakeReturns.result3 + return fakeReturns.result1, fakeReturns.result2, fakeReturns.result3, fakeReturns.result4 } func (fake *FakeRoomAllocator) CreateRoomCallCount() int { @@ -68,7 +93,7 @@ func (fake *FakeRoomAllocator) CreateRoomCallCount() int { return len(fake.createRoomArgsForCall) } -func (fake *FakeRoomAllocator) CreateRoomCalls(stub func(context.Context, *livekit.CreateRoomRequest) (*livekit.Room, bool, error)) { +func (fake *FakeRoomAllocator) CreateRoomCalls(stub func(context.Context, *livekit.CreateRoomRequest) (*livekit.Room, *livekit.RoomInternal, bool, error)) { fake.createRoomMutex.Lock() defer fake.createRoomMutex.Unlock() fake.CreateRoomStub = stub @@ -81,33 +106,152 @@ func (fake *FakeRoomAllocator) CreateRoomArgsForCall(i int) (context.Context, *l return argsForCall.arg1, argsForCall.arg2 } -func (fake *FakeRoomAllocator) CreateRoomReturns(result1 *livekit.Room, result2 bool, result3 error) { +func (fake *FakeRoomAllocator) CreateRoomReturns(result1 *livekit.Room, result2 *livekit.RoomInternal, result3 bool, result4 error) { fake.createRoomMutex.Lock() defer fake.createRoomMutex.Unlock() fake.CreateRoomStub = nil fake.createRoomReturns = struct { result1 *livekit.Room - result2 bool - result3 error - }{result1, result2, result3} + result2 *livekit.RoomInternal + result3 bool + result4 error + }{result1, result2, result3, result4} } -func (fake *FakeRoomAllocator) CreateRoomReturnsOnCall(i int, result1 *livekit.Room, result2 bool, result3 error) { +func (fake *FakeRoomAllocator) CreateRoomReturnsOnCall(i int, result1 *livekit.Room, result2 *livekit.RoomInternal, result3 bool, result4 error) { fake.createRoomMutex.Lock() defer fake.createRoomMutex.Unlock() fake.CreateRoomStub = nil if fake.createRoomReturnsOnCall == nil { fake.createRoomReturnsOnCall = make(map[int]struct { result1 *livekit.Room - result2 bool - result3 error + result2 *livekit.RoomInternal + result3 bool + result4 error }) } fake.createRoomReturnsOnCall[i] = struct { result1 *livekit.Room - result2 bool - result3 error - }{result1, result2, result3} + result2 *livekit.RoomInternal + result3 bool + result4 error + }{result1, result2, result3, result4} +} + +func (fake *FakeRoomAllocator) SelectRoomNode(arg1 context.Context, arg2 livekit.RoomName, arg3 livekit.NodeID) error { + fake.selectRoomNodeMutex.Lock() + ret, specificReturn := fake.selectRoomNodeReturnsOnCall[len(fake.selectRoomNodeArgsForCall)] + fake.selectRoomNodeArgsForCall = append(fake.selectRoomNodeArgsForCall, struct { + arg1 context.Context + arg2 livekit.RoomName + arg3 livekit.NodeID + }{arg1, arg2, arg3}) + stub := fake.SelectRoomNodeStub + fakeReturns := fake.selectRoomNodeReturns + fake.recordInvocation("SelectRoomNode", []interface{}{arg1, arg2, arg3}) + fake.selectRoomNodeMutex.Unlock() + if stub != nil { + return stub(arg1, arg2, arg3) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeRoomAllocator) SelectRoomNodeCallCount() int { + fake.selectRoomNodeMutex.RLock() + defer fake.selectRoomNodeMutex.RUnlock() + return len(fake.selectRoomNodeArgsForCall) +} + +func (fake *FakeRoomAllocator) SelectRoomNodeCalls(stub func(context.Context, livekit.RoomName, livekit.NodeID) error) { + fake.selectRoomNodeMutex.Lock() + defer fake.selectRoomNodeMutex.Unlock() + fake.SelectRoomNodeStub = stub +} + +func (fake *FakeRoomAllocator) SelectRoomNodeArgsForCall(i int) (context.Context, livekit.RoomName, livekit.NodeID) { + fake.selectRoomNodeMutex.RLock() + defer fake.selectRoomNodeMutex.RUnlock() + argsForCall := fake.selectRoomNodeArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 +} + +func (fake *FakeRoomAllocator) SelectRoomNodeReturns(result1 error) { + fake.selectRoomNodeMutex.Lock() + defer fake.selectRoomNodeMutex.Unlock() + fake.SelectRoomNodeStub = nil + fake.selectRoomNodeReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeRoomAllocator) SelectRoomNodeReturnsOnCall(i int, result1 error) { + fake.selectRoomNodeMutex.Lock() + defer fake.selectRoomNodeMutex.Unlock() + fake.SelectRoomNodeStub = nil + if fake.selectRoomNodeReturnsOnCall == nil { + fake.selectRoomNodeReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.selectRoomNodeReturnsOnCall[i] = struct { + result1 error + }{result1} +} + +func (fake *FakeRoomAllocator) CreateRoomEnabled() bool { + fake.startRoomEnabledMutex.Lock() + ret, specificReturn := fake.startRoomEnabledReturnsOnCall[len(fake.startRoomEnabledArgsForCall)] + fake.startRoomEnabledArgsForCall = append(fake.startRoomEnabledArgsForCall, struct { + }{}) + stub := fake.StartRoomEnabledStub + fakeReturns := fake.startRoomEnabledReturns + fake.recordInvocation("StartRoomEnabled", []interface{}{}) + fake.startRoomEnabledMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeRoomAllocator) StartRoomEnabledCallCount() int { + fake.startRoomEnabledMutex.RLock() + defer fake.startRoomEnabledMutex.RUnlock() + return len(fake.startRoomEnabledArgsForCall) +} + +func (fake *FakeRoomAllocator) StartRoomEnabledCalls(stub func() bool) { + fake.startRoomEnabledMutex.Lock() + defer fake.startRoomEnabledMutex.Unlock() + fake.StartRoomEnabledStub = stub +} + +func (fake *FakeRoomAllocator) StartRoomEnabledReturns(result1 bool) { + fake.startRoomEnabledMutex.Lock() + defer fake.startRoomEnabledMutex.Unlock() + fake.StartRoomEnabledStub = nil + fake.startRoomEnabledReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeRoomAllocator) StartRoomEnabledReturnsOnCall(i int, result1 bool) { + fake.startRoomEnabledMutex.Lock() + defer fake.startRoomEnabledMutex.Unlock() + fake.StartRoomEnabledStub = nil + if fake.startRoomEnabledReturnsOnCall == nil { + fake.startRoomEnabledReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.startRoomEnabledReturnsOnCall[i] = struct { + result1 bool + }{result1} } func (fake *FakeRoomAllocator) ValidateCreateRoom(arg1 context.Context, arg2 livekit.RoomName) error { @@ -177,6 +321,10 @@ func (fake *FakeRoomAllocator) Invocations() map[string][][]interface{} { defer fake.invocationsMutex.RUnlock() fake.createRoomMutex.RLock() defer fake.createRoomMutex.RUnlock() + fake.selectRoomNodeMutex.RLock() + defer fake.selectRoomNodeMutex.RUnlock() + fake.startRoomEnabledMutex.RLock() + defer fake.startRoomEnabledMutex.RUnlock() fake.validateCreateRoomMutex.RLock() defer fake.validateCreateRoomMutex.RUnlock() copiedInvocations := map[string][][]interface{}{} diff --git a/pkg/service/servicefakes/fake_session_handler.go b/pkg/service/servicefakes/fake_session_handler.go index 552386918..15d5a00d5 100644 --- a/pkg/service/servicefakes/fake_session_handler.go +++ b/pkg/service/servicefakes/fake_session_handler.go @@ -12,11 +12,11 @@ import ( ) type FakeSessionHandler struct { - HandleSessionStub func(context.Context, livekit.RoomName, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) error + HandleSessionStub func(context.Context, *livekit.CreateRoomRequest, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) error handleSessionMutex sync.RWMutex handleSessionArgsForCall []struct { arg1 context.Context - arg2 livekit.RoomName + arg2 *livekit.CreateRoomRequest arg3 routing.ParticipantInit arg4 livekit.ConnectionID arg5 routing.MessageSource @@ -43,12 +43,12 @@ type FakeSessionHandler struct { invocationsMutex sync.RWMutex } -func (fake *FakeSessionHandler) HandleSession(arg1 context.Context, arg2 livekit.RoomName, arg3 routing.ParticipantInit, arg4 livekit.ConnectionID, arg5 routing.MessageSource, arg6 routing.MessageSink) error { +func (fake *FakeSessionHandler) HandleSession(arg1 context.Context, arg2 *livekit.CreateRoomRequest, arg3 routing.ParticipantInit, arg4 livekit.ConnectionID, arg5 routing.MessageSource, arg6 routing.MessageSink) error { fake.handleSessionMutex.Lock() ret, specificReturn := fake.handleSessionReturnsOnCall[len(fake.handleSessionArgsForCall)] fake.handleSessionArgsForCall = append(fake.handleSessionArgsForCall, struct { arg1 context.Context - arg2 livekit.RoomName + arg2 *livekit.CreateRoomRequest arg3 routing.ParticipantInit arg4 livekit.ConnectionID arg5 routing.MessageSource @@ -73,13 +73,13 @@ func (fake *FakeSessionHandler) HandleSessionCallCount() int { return len(fake.handleSessionArgsForCall) } -func (fake *FakeSessionHandler) HandleSessionCalls(stub func(context.Context, livekit.RoomName, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) error) { +func (fake *FakeSessionHandler) HandleSessionCalls(stub func(context.Context, *livekit.CreateRoomRequest, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) error) { fake.handleSessionMutex.Lock() defer fake.handleSessionMutex.Unlock() fake.HandleSessionStub = stub } -func (fake *FakeSessionHandler) HandleSessionArgsForCall(i int) (context.Context, livekit.RoomName, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) { +func (fake *FakeSessionHandler) HandleSessionArgsForCall(i int) (context.Context, *livekit.CreateRoomRequest, routing.ParticipantInit, livekit.ConnectionID, routing.MessageSource, routing.MessageSink) { fake.handleSessionMutex.RLock() defer fake.handleSessionMutex.RUnlock() argsForCall := fake.handleSessionArgsForCall[i] diff --git a/pkg/service/signal.go b/pkg/service/signal.go index 69c04baab..6d5523944 100644 --- a/pkg/service/signal.go +++ b/pkg/service/signal.go @@ -39,7 +39,7 @@ type SessionHandler interface { HandleSession( ctx context.Context, - roomName livekit.RoomName, + createRoom *livekit.CreateRoomRequest, pi routing.ParticipantInit, connectionID livekit.ConnectionID, requestSource routing.MessageSource, @@ -94,7 +94,7 @@ func (s *defaultSessionHandler) Logger(ctx context.Context) logger.Logger { func (s *defaultSessionHandler) HandleSession( ctx context.Context, - roomName livekit.RoomName, + createRoom *livekit.CreateRoomRequest, pi routing.ParticipantInit, connectionID livekit.ConnectionID, requestSource routing.MessageSource, @@ -102,7 +102,7 @@ func (s *defaultSessionHandler) HandleSession( ) error { prometheus.IncrementParticipantRtcInit(1) - rtcNode, err := s.router.GetNodeForRoom(ctx, roomName) + rtcNode, err := s.router.GetNodeForRoom(ctx, livekit.RoomName(createRoom.Name)) if err != nil { return err } @@ -115,12 +115,12 @@ func (s *defaultSessionHandler) HandleSession( return err } - return s.roomManager.StartSession(ctx, roomName, pi, requestSource, responseSink) + return s.roomManager.StartSession(ctx, createRoom, pi, requestSource, responseSink) } func (s *SignalServer) Start() error { logger.Debugw("starting relay signal server", "topic", s.nodeID) - return s.server.RegisterRelaySignalTopic(s.nodeID) + return s.server.RegisterAllNodeTopics(s.nodeID) } func (r *SignalServer) Stop() { @@ -182,7 +182,18 @@ func (r *signalService) RelaySignal(stream psrpc.ServerStream[*rpc.RelaySignalRe // copy the incoming rpc headers to avoid dropping any session vars. ctx := metadata.NewContextWithIncomingHeader(context.Background(), metadata.IncomingHeader(stream.Context())) - err = r.sessionHandler.HandleSession(ctx, livekit.RoomName(ss.RoomName), *pi, livekit.ConnectionID(ss.ConnectionId), reqChan, sink) + createRoom := ss.CreateRoom + if createRoom == nil { + createRoom = &livekit.CreateRoomRequest{ + Name: ss.RoomName, + } + + if pi.Grants != nil && pi.Grants.Video != nil { + createRoom.ConfigName = pi.Grants.Video.RoomConfiguration + } + } + + err = r.sessionHandler.HandleSession(ctx, createRoom, *pi, livekit.ConnectionID(ss.ConnectionId), reqChan, sink) if err != nil { sink.Close() l.Errorw("could not handle new participant", err) diff --git a/pkg/service/signal_test.go b/pkg/service/signal_test.go index 9da50679f..1e3ac03f2 100644 --- a/pkg/service/signal_test.go +++ b/pkg/service/signal_test.go @@ -67,7 +67,7 @@ func TestSignal(t *testing.T) { LoggerStub: func(context.Context) logger.Logger { return logger.GetLogger() }, HandleSessionStub: func( ctx context.Context, - roomName livekit.RoomName, + createRoom *livekit.CreateRoomRequest, pi routing.ParticipantInit, connectionID livekit.ConnectionID, requestSource routing.MessageSource, @@ -124,7 +124,7 @@ func TestSignal(t *testing.T) { LoggerStub: func(context.Context) logger.Logger { return logger.GetLogger() }, HandleSessionStub: func( ctx context.Context, - roomName livekit.RoomName, + createRoom *livekit.CreateRoomRequest, pi routing.ParticipantInit, connectionID livekit.ConnectionID, requestSource routing.MessageSource, diff --git a/pkg/service/wire.go b/pkg/service/wire.go index 56d601bf6..dcabb5da8 100644 --- a/pkg/service/wire.go +++ b/pkg/service/wire.go @@ -85,6 +85,8 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live getSignalRelayConfig, NewDefaultSignalServer, routing.NewSignalClient, + getRoomConfig, + routing.NewRoomManagerClient, rpc.NewKeepalivePubSub, getPSRPCConfig, getPSRPCClientParams, @@ -111,6 +113,8 @@ func InitializeRouter(conf *config.Config, currentNode routing.LocalNode) (routi getPSRPCConfig, getPSRPCClientParams, routing.NewSignalClient, + getRoomConfig, + routing.NewRoomManagerClient, rpc.NewKeepalivePubSub, routing.CreateRouter, ) @@ -239,6 +243,10 @@ func getLimitConf(config *config.Config) config.LimitConfig { return config.Limit } +func getRoomConfig(config *config.Config) config.RoomConfig { + return config.Room +} + func getSignalRelayConfig(config *config.Config) config.SignalRelayConfig { return config.SignalRelay } diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index f27f4dde6..07b48e683 100644 --- a/pkg/service/wire_gen.go +++ b/pkg/service/wire_gen.go @@ -38,7 +38,6 @@ import ( func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*LivekitServer, error) { limitConfig := getLimitConf(conf) apiConfig := config.DefaultAPIConfig() - psrpcConfig := getPSRPCConfig(conf) universalClient, err := createRedisClient(conf) if err != nil { return nil, err @@ -50,12 +49,18 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live if err != nil { return nil, err } + psrpcConfig := getPSRPCConfig(conf) clientParams := getPSRPCClientParams(psrpcConfig, messageBus) + roomConfig := getRoomConfig(conf) + roomManagerClient, err := routing.NewRoomManagerClient(clientParams, roomConfig) + if err != nil { + return nil, err + } keepalivePubSub, err := rpc.NewKeepalivePubSub(clientParams) if err != nil { return nil, err } - router := routing.CreateRouter(universalClient, currentNode, signalClient, keepalivePubSub) + router := routing.CreateRouter(universalClient, currentNode, signalClient, roomManagerClient, keepalivePubSub) objectStore := createStore(universalClient) roomAllocator, err := NewRoomAllocator(conf, router, objectStore) if err != nil { @@ -96,7 +101,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live if err != nil { return nil, err } - roomService, err := NewRoomService(limitConfig, apiConfig, psrpcConfig, router, roomAllocator, objectStore, client, rtcEgressLauncher, topicFormatter, roomClient, participantClient) + roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, client, rtcEgressLauncher, topicFormatter, roomClient, participantClient) if err != nil { return nil, err } @@ -128,7 +133,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live timedVersionGenerator := utils.NewDefaultTimedVersionGenerator() turnAuthHandler := NewTURNAuthHandler(keyProvider) forwardStats := createForwardStats(conf) - roomManager, err := NewLocalRoomManager(conf, objectStore, currentNode, router, telemetryService, clientConfigurationManager, client, agentStore, rtcEgressLauncher, timedVersionGenerator, turnAuthHandler, messageBus, forwardStats) + roomManager, err := NewLocalRoomManager(conf, objectStore, currentNode, router, roomAllocator, telemetryService, clientConfigurationManager, client, agentStore, rtcEgressLauncher, timedVersionGenerator, turnAuthHandler, messageBus, forwardStats) if err != nil { return nil, err } @@ -162,11 +167,16 @@ func InitializeRouter(conf *config.Config, currentNode routing.LocalNode) (routi } psrpcConfig := getPSRPCConfig(conf) clientParams := getPSRPCClientParams(psrpcConfig, messageBus) + roomConfig := getRoomConfig(conf) + roomManagerClient, err := routing.NewRoomManagerClient(clientParams, roomConfig) + if err != nil { + return nil, err + } keepalivePubSub, err := rpc.NewKeepalivePubSub(clientParams) if err != nil { return nil, err } - router := routing.CreateRouter(universalClient, currentNode, signalClient, keepalivePubSub) + router := routing.CreateRouter(universalClient, currentNode, signalClient, roomManagerClient, keepalivePubSub) return router, nil } @@ -293,6 +303,10 @@ func getLimitConf(config2 *config.Config) config.LimitConfig { return config2.Limit } +func getRoomConfig(config2 *config.Config) config.RoomConfig { + return config2.Room +} + func getSignalRelayConfig(config2 *config.Config) config.SignalRelayConfig { return config2.SignalRelay } diff --git a/test/agent_test.go b/test/agent_test.go index 9031b9505..ad50c53d9 100644 --- a/test/agent_test.go +++ b/test/agent_test.go @@ -173,6 +173,7 @@ func TestAgentMultiNode(t *testing.T) { ac1, err := newAgentClient(agentToken(), defaultServerPort) require.NoError(t, err) ac2, err := newAgentClient(agentToken(), defaultServerPort) + require.NoError(t, err) defer ac1.close() defer ac2.close() ac1.Run(livekit.JobType_JT_ROOM, "default")