From 126bb8867b7cdac039907d869c41e1f432aafe61 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Fri, 21 Jan 2022 09:57:36 -0800 Subject: [PATCH] Ability to disable auto-create (#361) * setting to disable autocreate * improve test reliability * update comment * update to address feedback --- config-sample.yaml | 2 + magefile.go | 4 +- pkg/config/config.go | 3 ++ pkg/config/config_test.go | 9 +++++ pkg/service/errors.go | 1 + pkg/service/roomservice.go | 38 ++++++++++++++++++ pkg/service/roomservice_test.go | 1 + pkg/service/rtcservice.go | 23 ++++++++++- pkg/service/wire_gen.go | 2 +- test/integration_helpers.go | 14 +++---- test/multinode_test.go | 12 ++++++ test/scenarios.go | 21 ++++++++-- test/singlenode_test.go | 70 +++++++++++++++++++++++++++++---- 13 files changed, 178 insertions(+), 22 deletions(-) diff --git a/config-sample.yaml b/config-sample.yaml index 0a0d8ca27..7a765f119 100644 --- a/config-sample.yaml +++ b/config-sample.yaml @@ -81,6 +81,8 @@ keys: # Each room created will inherit these settings. If rooms are created explicitly with CreateRoom, they will take # precedence over defaults # room: +# # allow rooms to be automatically created when participants join, defaults to true +# # auto_create: false # # number of seconds to leave a room open when it's empty # empty_timeout: 300 # # limit number of participants that can be in a room, 0 for no limit diff --git a/magefile.go b/magefile.go index 2558a2cc3..ee486b94a 100644 --- a/magefile.go +++ b/magefile.go @@ -118,7 +118,7 @@ func PublishDocker() error { // run unit tests, skipping integration func Test() error { mg.Deps(generateWire, macULimit) - cmd := exec.Command("go", "test", "-short", "./...") + cmd := exec.Command("go", "test", "-short", "./...", "-count=1") connectStd(cmd) return cmd.Run() } @@ -127,7 +127,7 @@ func Test() error { func TestAll() error { mg.Deps(generateWire, macULimit) // "-v", "-race", - cmd := exec.Command("go", "test", "./...", "-count=1", "-timeout=4m") + cmd := exec.Command("go", "test", "./...", "-count=1", "-timeout=4m", "-v") connectStd(cmd) return cmd.Run() } diff --git a/pkg/config/config.go b/pkg/config/config.go index 6e885e97a..b6a52a708 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -103,6 +103,8 @@ type RedisConfig struct { } type RoomConfig struct { + // enable rooms to be automatically created + AutoCreate bool `yaml:"auto_create"` EnabledCodecs []CodecSpec `yaml:"enabled_codecs"` MaxParticipants uint32 `yaml:"max_participants"` EmptyTimeout uint32 `yaml:"empty_timeout"` @@ -187,6 +189,7 @@ func NewConfig(confString string, c *cli.Context) (*Config, error) { }, Redis: RedisConfig{}, Room: RoomConfig{ + AutoCreate: true, // by default only enable opus and VP8 EnabledCodecs: []CodecSpec{ {Mime: webrtc.MimeTypeOpus}, diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 35463ef3a..3efa36f72 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -13,3 +13,12 @@ func TestConfig_UnmarshalKeys(t *testing.T) { require.NoError(t, conf.unmarshalKeys("key1: secret1")) require.Equal(t, "secret1", conf.Keys["key1"]) } + +func TestConfig_DefaultsKept(t *testing.T) { + const content = `room: + empty_timeout: 10` + conf, err := NewConfig(content, nil) + require.NoError(t, err) + require.Equal(t, true, conf.Room.AutoCreate) + require.Equal(t, uint32(10), conf.Room.EmptyTimeout) +} diff --git a/pkg/service/errors.go b/pkg/service/errors.go index 992e45549..cac476193 100644 --- a/pkg/service/errors.go +++ b/pkg/service/errors.go @@ -9,4 +9,5 @@ var ( ErrParticipantNotFound = errors.New("participant does not exist") ErrTrackNotFound = errors.New("track is not found") ErrWebHookMissingAPIKey = errors.New("api_key is required to use webhooks") + ErrOperationFailed = errors.New("operation cannot be completed") ) diff --git a/pkg/service/roomservice.go b/pkg/service/roomservice.go index 0717cf841..755407ee7 100644 --- a/pkg/service/roomservice.go +++ b/pkg/service/roomservice.go @@ -2,6 +2,7 @@ package service import ( "context" + "time" "github.com/livekit/protocol/livekit" "github.com/pkg/errors" @@ -11,6 +12,11 @@ import ( "github.com/livekit/livekit-server/pkg/routing" ) +const ( + executionTimeout = 2 * time.Second + checkInterval = 50 * time.Millisecond +) + // A rooms service that supports a single node type RoomService struct { router routing.MessageRouter @@ -75,6 +81,21 @@ func (s *RoomService) DeleteRoom(ctx context.Context, req *livekit.DeleteRoomReq return nil, err } + // we should not return until when the room is confirmed deleted + err = confirmExecution(func() error { + _, err := s.roomStore.LoadRoom(ctx, livekit.RoomName(req.Room)) + if err == nil { + return ErrOperationFailed + } else if err != ErrRoomNotFound { + return err + } else { + return nil + } + }) + if err != nil { + return nil, err + } + return &livekit.DeleteRoomResponse{}, nil } @@ -252,3 +273,20 @@ func (s *RoomService) writeRoomMessage(ctx context.Context, room livekit.RoomNam return s.router.WriteRoomRTC(ctx, room, identity, msg) } + +func confirmExecution(f func() error) error { + expired := time.After(executionTimeout) + var err error + for { + select { + case <-expired: + return err + default: + err = f() + if err == nil { + return nil + } + time.Sleep(checkInterval) + } + } +} diff --git a/pkg/service/roomservice_test.go b/pkg/service/roomservice_test.go index 3c035dc90..c392997f4 100644 --- a/pkg/service/roomservice_test.go +++ b/pkg/service/roomservice_test.go @@ -24,6 +24,7 @@ func TestDeleteRoom(t *testing.T) { }, } ctx := context.WithValue(context.Background(), grantsKey, grant) + svc.store.LoadRoomReturns(nil, service.ErrRoomNotFound) _, err := svc.DeleteRoom(ctx, &livekit.DeleteRoomRequest{ Room: "testroom", }) diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index d330e669d..c2c0cf8ed 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -1,6 +1,7 @@ package service import ( + "context" "fmt" "io" "net/http" @@ -23,18 +24,28 @@ import ( type RTCService struct { router routing.MessageRouter roomAllocator RoomAllocator + store RoomStore upgrader websocket.Upgrader currentNode routing.LocalNode + config *config.Config isDev bool limits config.LimitConfig } -func NewRTCService(conf *config.Config, ra RoomAllocator, router routing.MessageRouter, currentNode routing.LocalNode) *RTCService { +func NewRTCService( + conf *config.Config, + ra RoomAllocator, + store RoomStore, + router routing.MessageRouter, + currentNode routing.LocalNode, +) *RTCService { s := &RTCService{ router: router, roomAllocator: ra, + store: store, upgrader: websocket.Upgrader{}, currentNode: currentNode, + config: conf, isDev: conf.Development, limits: conf.Limit, } @@ -130,6 +141,16 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } + // when autocreate is disabled, we'll check to ensure it's already created + if !s.config.Room.AutoCreate { + _, err := s.store.LoadRoom(context.Background(), roomName) + if err == ErrRoomNotFound { + handleError(w, 404, err.Error()) + } else if err != nil { + handleError(w, 500, err.Error()) + } + } + // create room if it doesn't exist, also assigns an RTC node for the room rm, err := s.roomAllocator.CreateRoom(r.Context(), &livekit.CreateRoomRequest{Name: string(roomName)}) if err != nil { diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index 7c2853612..dfdf424b0 100644 --- a/pkg/service/wire_gen.go +++ b/pkg/service/wire_gen.go @@ -50,7 +50,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live analyticsService := telemetry.NewAnalyticsService(conf, currentNode) telemetryService := telemetry.NewTelemetryService(notifier, analyticsService) recordingService := NewRecordingService(messageBus, telemetryService) - rtcService := NewRTCService(conf, roomAllocator, router, currentNode) + rtcService := NewRTCService(conf, roomAllocator, roomStore, router, currentNode) roomManager, err := NewLocalRoomManager(conf, roomStore, currentNode, router, telemetryService) if err != nil { return nil, err diff --git a/test/integration_helpers.go b/test/integration_helpers.go index b44416efd..47adf93bd 100644 --- a/test/integration_helpers.go +++ b/test/integration_helpers.go @@ -46,9 +46,9 @@ func init() { serverlogger.InitFromConfig(config.LoggingConfig{Level: "debug"}) } -func setupSingleNodeTest(name string, roomName string) (*service.LivekitServer, func()) { +func setupSingleNodeTest(name string) (*service.LivekitServer, func()) { logger.Infow("----------------STARTING TEST----------------", "test", name) - s := createSingleNodeServer() + s := createSingleNodeServer(nil) go func() { if err := s.Start(); err != nil { logger.Errorw("server returned error", err) @@ -57,11 +57,6 @@ func setupSingleNodeTest(name string, roomName string) (*service.LivekitServer, waitForServerToStart(s) - // create test room - _, err := roomClient.CreateRoom(contextWithToken(createRoomToken()), &livekit.CreateRoomRequest{Name: roomName}) - if err != nil { - panic(err) - } return s, func() { s.Stop(true) logger.Infow("----------------FINISHING TEST----------------", "test", name) @@ -132,7 +127,7 @@ func waitUntilConnected(t *testing.T, clients ...*testclient.RTCClient) { } } -func createSingleNodeServer() *service.LivekitServer { +func createSingleNodeServer(configUpdater func(*config.Config)) *service.LivekitServer { var err error conf, err := config.NewConfig("", nil) if err != nil { @@ -140,6 +135,9 @@ func createSingleNodeServer() *service.LivekitServer { } conf.Development = true conf.Keys = map[string]string{testApiKey: testApiSecret} + if configUpdater != nil { + configUpdater(conf) + } currentNode, err := routing.NewLocalNode(conf) if err != nil { diff --git a/test/multinode_test.go b/test/multinode_test.go index 08cf8cab0..75b97b741 100644 --- a/test/multinode_test.go +++ b/test/multinode_test.go @@ -151,3 +151,15 @@ func TestMultiNodeRoomList(t *testing.T) { roomServiceListRoom(t) } + +func TestMultiNodeJoinAfterClose(t *testing.T) { + if testing.Short() { + t.SkipNow() + return + } + + _, _, finish := setupMultiNodeTest("TestMultiNodeJoinAfterClose") + defer finish() + + scenarioJoinClosedRoom(t) +} diff --git a/test/scenarios.go b/test/scenarios.go index e997db671..067042443 100644 --- a/test/scenarios.go +++ b/test/scenarios.go @@ -4,13 +4,12 @@ import ( "testing" "time" + "github.com/livekit/livekit-server/pkg/testutils" + testclient "github.com/livekit/livekit-server/test/client" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils" "github.com/stretchr/testify/require" - - "github.com/livekit/livekit-server/pkg/testutils" - testclient "github.com/livekit/livekit-server/test/client" ) // a scenario with lots of clients connecting, publishing, and leaving at random periods @@ -141,6 +140,22 @@ func scenarioDataPublish(t *testing.T) { }) } +func scenarioJoinClosedRoom(t *testing.T) { + c1 := createRTCClient("jcr1", defaultServerPort, nil) + waitUntilConnected(t, c1) + + // close room with room client + _, err := roomClient.DeleteRoom(contextWithToken(createRoomToken()), &livekit.DeleteRoomRequest{ + Room: testRoom, + }) + require.NoError(t, err) + + // now join again + c2 := createRTCClient("jcr2", defaultServerPort, nil) + waitUntilConnected(t, c2) + stopClients(c2) +} + func publishTracksForClients(t *testing.T, clients ...*testclient.RTCClient) []*testclient.TrackWriter { logger.Infow("publishing tracks for clients") var writers []*testclient.TrackWriter diff --git a/test/singlenode_test.go b/test/singlenode_test.go index 37337b60c..1b62439e8 100644 --- a/test/singlenode_test.go +++ b/test/singlenode_test.go @@ -8,8 +8,10 @@ import ( "testing" "time" + "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/protocol/auth" "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" "github.com/pion/webrtc/v3" "github.com/stretchr/testify/require" "github.com/thoas/go-funk" @@ -25,7 +27,7 @@ func TestClientCouldConnect(t *testing.T) { return } - _, finish := setupSingleNodeTest("TestClientCouldConnect", testRoom) + _, finish := setupSingleNodeTest("TestClientCouldConnect") defer finish() c1 := createRTCClient("c1", defaultServerPort, nil) @@ -44,7 +46,7 @@ func TestClientConnectDuplicate(t *testing.T) { return } - _, finish := setupSingleNodeTest("TestClientCouldConnect", testRoom) + _, finish := setupSingleNodeTest("TestClientCouldConnect") defer finish() grant := &auth.VideoGrant{RoomJoin: true, Room: testRoom} @@ -120,7 +122,7 @@ func TestSinglePublisher(t *testing.T) { return } - s, finish := setupSingleNodeTest("TestSinglePublisher", testRoom) + s, finish := setupSingleNodeTest("TestSinglePublisher") defer finish() c1 := createRTCClient("c1", defaultServerPort, nil) @@ -203,7 +205,7 @@ func Test_WhenAutoSubscriptionDisabled_ClientShouldNotReceiveAnyPublishedTracks( return } - _, finish := setupSingleNodeTest("Test_WhenAutoSubscriptionDisabled_ClientShouldNotReceiveAnyPublishedTracks", testRoom) + _, finish := setupSingleNodeTest("Test_WhenAutoSubscriptionDisabled_ClientShouldNotReceiveAnyPublishedTracks") defer finish() opts := testclient.Options{AutoSubscribe: false} @@ -228,7 +230,7 @@ func Test_RenegotiationWithDifferentCodecs(t *testing.T) { return } - _, finish := setupSingleNodeTest("TestRenegotiationWithDifferentCodecs", testRoom) + _, finish := setupSingleNodeTest("TestRenegotiationWithDifferentCodecs") defer finish() c1 := createRTCClient("c1", defaultServerPort, nil) @@ -302,7 +304,7 @@ func TestSingleNodeRoomList(t *testing.T) { t.SkipNow() return } - _, finish := setupSingleNodeTest("TestSingleNodeRoomList", testRoom) + _, finish := setupSingleNodeTest("TestSingleNodeRoomList") defer finish() roomServiceListRoom(t) @@ -314,7 +316,7 @@ func TestSingleNodeCORS(t *testing.T) { t.SkipNow() return } - s, finish := setupSingleNodeTest("TestSingleNodeCORS", testRoom) + s, finish := setupSingleNodeTest("TestSingleNodeCORS") defer finish() req, err := http.NewRequest("POST", fmt.Sprintf("http://localhost:%d", s.HTTPPort()), nil) @@ -325,3 +327,57 @@ func TestSingleNodeCORS(t *testing.T) { require.NoError(t, err) require.Equal(t, "testhost.com", res.Header.Get("Access-Control-Allow-Origin")) } + +func TestSingleNodeJoinAfterClose(t *testing.T) { + if testing.Short() { + t.SkipNow() + return + } + + _, finish := setupSingleNodeTest("TestJoinAfterClose") + defer finish() + + scenarioJoinClosedRoom(t) +} + +func TestAutoCreate(t *testing.T) { + disableAutoCreate := func(conf *config.Config) { + conf.Room.AutoCreate = false + } + t.Run("cannot join if room isn't created", func(t *testing.T) { + s := createSingleNodeServer(disableAutoCreate) + go func() { + if err := s.Start(); err != nil { + logger.Errorw("server returned error", err) + } + }() + defer s.Stop(true) + + waitForServerToStart(s) + + token := joinToken(testRoom, "start-before-create") + _, err := testclient.NewWebSocketConn(fmt.Sprintf("ws://localhost:%d", defaultServerPort), token, nil) + require.Error(t, err) + }) + + t.Run("join with explicit createRoom", func(t *testing.T) { + s := createSingleNodeServer(disableAutoCreate) + go func() { + if err := s.Start(); err != nil { + logger.Errorw("server returned error", err) + } + }() + defer s.Stop(true) + + waitForServerToStart(s) + + // explicitly create + _, err := roomClient.CreateRoom(contextWithToken(createRoomToken()), &livekit.CreateRoomRequest{Name: testRoom}) + require.NoError(t, err) + + c1 := createRTCClient("join-after-create", defaultServerPort, nil) + waitUntilConnected(t, c1) + + c1.Stop() + }) +}