From 12ae179be2904732d97c9dee5edd3597c5175e13 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Sun, 4 Dec 2022 10:13:09 -0800 Subject: [PATCH] Configurable RoomService execution timeout (#1206) * API execution timeout is now configurable In certain environments, it can take longer than the default 2s to fully execute API requests. Making execution timeout a configurable option. * do not expose api to YAML. internal for now. --- config-sample.yaml | 2 +- pkg/config/config.go | 16 ++++++++++++ pkg/service/roomservice.go | 43 ++++++++++++++++----------------- pkg/service/roomservice_test.go | 4 ++- pkg/service/wire.go | 1 + pkg/service/wire_gen.go | 3 ++- 6 files changed, 44 insertions(+), 25 deletions(-) diff --git a/config-sample.yaml b/config-sample.yaml index f6866d6d3..1cb57b590 100644 --- a/config-sample.yaml +++ b/config-sample.yaml @@ -80,7 +80,7 @@ rtc: # low_quality: 500ms # mid_quality: 1s # high_quality: 1s - # # when set, Livekit will collect loopback candidates, it is useful for some VM have public address mapped to its loopback interface. + # # when set, Livekit will collect loopback candidates, it is useful for some VM have public address mapped to its loopback interface. # enable_loopback_candidate: true # # network interface filter. If the machine has more than one network interface and you'd like it to use or skip specific interfaces # # both inclusion and exclusion filters can be used together. If neither is defined (default), all interfaces on the machine will be used. diff --git a/pkg/config/config.go b/pkg/config/config.go index 080c075a6..80e4e508d 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -199,6 +199,22 @@ type IngressConfig struct { RTMPBaseURL string `yaml:"rtmp_base_url"` } +// not exposed to YAML +type APIConfig struct { + // amount of time to wait for API to execute, default 2s + ExecutionTimeout time.Duration + + // amount of time to wait before checking for operation complete + CheckInterval time.Duration +} + +func DefaultAPIConfig() APIConfig { + return APIConfig{ + ExecutionTimeout: 2 * time.Second, + CheckInterval: 100 * time.Millisecond, + } +} + func NewConfig(confString string, strictMode bool, c *cli.Context, baseFlags []cli.Flag) (*Config, error) { // start with defaults conf := &Config{ diff --git a/pkg/service/roomservice.go b/pkg/service/roomservice.go index 8740ec95a..8cd4238d9 100644 --- a/pkg/service/roomservice.go +++ b/pkg/service/roomservice.go @@ -16,14 +16,10 @@ import ( "github.com/livekit/protocol/livekit" ) -const ( - executionTimeout = 2 * time.Second - checkInterval = 50 * time.Millisecond -) - // A rooms service that supports a single node type RoomService struct { - conf config.RoomConfig + roomConf config.RoomConfig + apiConf config.APIConfig router routing.MessageRouter roomAllocator RoomAllocator roomStore ServiceStore @@ -31,15 +27,16 @@ type RoomService struct { } func NewRoomService( - conf config.RoomConfig, + roomConf config.RoomConfig, + apiConf config.APIConfig, router routing.MessageRouter, roomAllocator RoomAllocator, serviceStore ServiceStore, egressLauncher rtc.EgressLauncher, ) (svc *RoomService, err error) { - svc = &RoomService{ - conf: conf, + roomConf: roomConf, + apiConf: apiConf, router: router, roomAllocator: roomAllocator, roomStore: serviceStore, @@ -74,7 +71,7 @@ func (s *RoomService) CreateRoom(ctx context.Context, req *livekit.CreateRoomReq source.Close() // ensure it's created correctly - err = confirmExecution(func() error { + err = s.confirmExecution(func() error { _, _, err := s.roomStore.LoadRoom(ctx, livekit.RoomName(req.Name), false) if err != nil { return ErrOperationFailed @@ -137,7 +134,7 @@ func (s *RoomService) DeleteRoom(ctx context.Context, req *livekit.DeleteRoomReq } // we should not return until when the room is confirmed deleted - err = confirmExecution(func() error { + err = s.confirmExecution(func() error { _, _, err := s.roomStore.LoadRoom(ctx, livekit.RoomName(req.Room), false) if err == nil { return ErrOperationFailed @@ -196,7 +193,7 @@ func (s *RoomService) RemoveParticipant(ctx context.Context, req *livekit.RoomPa return nil, err } - err = confirmExecution(func() error { + err = s.confirmExecution(func() error { _, err := s.roomStore.LoadParticipant(ctx, livekit.RoomName(req.Room), livekit.ParticipantIdentity(req.Identity)) if err == ErrParticipantNotFound { return nil @@ -229,7 +226,7 @@ func (s *RoomService) MutePublishedTrack(ctx context.Context, req *livekit.MuteR } var track *livekit.TrackInfo - err = confirmExecution(func() error { + err = s.confirmExecution(func() error { p, err := s.roomStore.LoadParticipant(ctx, livekit.RoomName(req.Room), livekit.ParticipantIdentity(req.Identity)) if err != nil { return err @@ -260,8 +257,9 @@ func (s *RoomService) MutePublishedTrack(ctx context.Context, req *livekit.MuteR func (s *RoomService) UpdateParticipant(ctx context.Context, req *livekit.UpdateParticipantRequest) (*livekit.ParticipantInfo, error) { AppendLogFields(ctx, "room", req.Room, "participant", req.Identity) - if s.conf.MaxMetadataSize > 0 && len(req.Metadata) > int(s.conf.MaxMetadataSize) { - return nil, twirp.InvalidArgumentError(ErrMetadataExceedsLimits.Error(), strconv.Itoa(int(s.conf.MaxMetadataSize))) + maxMetadataSize := int(s.roomConf.MaxMetadataSize) + if maxMetadataSize > 0 && len(req.Metadata) > maxMetadataSize { + return nil, twirp.InvalidArgumentError(ErrMetadataExceedsLimits.Error(), strconv.Itoa(maxMetadataSize)) } err := s.writeParticipantMessage(ctx, livekit.RoomName(req.Room), livekit.ParticipantIdentity(req.Identity), &livekit.RTCNodeMessage{ @@ -274,7 +272,7 @@ func (s *RoomService) UpdateParticipant(ctx context.Context, req *livekit.Update } var participant *livekit.ParticipantInfo - err = confirmExecution(func() error { + err = s.confirmExecution(func() error { participant, err = s.roomStore.LoadParticipant(ctx, livekit.RoomName(req.Room), livekit.ParticipantIdentity(req.Identity)) if err != nil { return err @@ -333,8 +331,9 @@ func (s *RoomService) SendData(ctx context.Context, req *livekit.SendDataRequest func (s *RoomService) UpdateRoomMetadata(ctx context.Context, req *livekit.UpdateRoomMetadataRequest) (*livekit.Room, error) { AppendLogFields(ctx, "room", req.Room, "size", len(req.Metadata)) - if s.conf.MaxMetadataSize > 0 && len(req.Metadata) > int(s.conf.MaxMetadataSize) { - return nil, twirp.InvalidArgumentError(ErrMetadataExceedsLimits.Error(), strconv.Itoa(int(s.conf.MaxMetadataSize))) + maxMetadataSize := int(s.roomConf.MaxMetadataSize) + if maxMetadataSize > 0 && len(req.Metadata) > maxMetadataSize { + return nil, twirp.InvalidArgumentError(ErrMetadataExceedsLimits.Error(), strconv.Itoa(maxMetadataSize)) } if err := EnsureAdminPermission(ctx, livekit.RoomName(req.Room)); err != nil { @@ -365,7 +364,7 @@ func (s *RoomService) UpdateRoomMetadata(ctx context.Context, req *livekit.Updat return nil, err } - err = confirmExecution(func() error { + err = s.confirmExecution(func() error { room, _, err = s.roomStore.LoadRoom(ctx, livekit.RoomName(req.Room), false) if err != nil { return err @@ -390,8 +389,8 @@ func (s *RoomService) writeParticipantMessage(ctx context.Context, room livekit. return s.router.WriteParticipantRTC(ctx, room, identity, msg) } -func confirmExecution(f func() error) error { - expired := time.After(executionTimeout) +func (s *RoomService) confirmExecution(f func() error) error { + expired := time.After(s.apiConf.ExecutionTimeout) var err error for { select { @@ -402,7 +401,7 @@ func confirmExecution(f func() error) error { if err == nil { return nil } - time.Sleep(checkInterval) + time.Sleep(s.apiConf.CheckInterval) } } } diff --git a/pkg/service/roomservice_test.go b/pkg/service/roomservice_test.go index 0eb6bded7..744e18442 100644 --- a/pkg/service/roomservice_test.go +++ b/pkg/service/roomservice_test.go @@ -107,7 +107,9 @@ func newTestRoomService(conf config.RoomConfig) *TestRoomService { router := &routingfakes.FakeRouter{} allocator := &servicefakes.FakeRoomAllocator{} store := &servicefakes.FakeServiceStore{} - svc, err := service.NewRoomService(conf, router, allocator, store, nil) + svc, err := service.NewRoomService(conf, + config.APIConfig{ExecutionTimeout: 2}, + router, allocator, store, nil) if err != nil { panic(err) } diff --git a/pkg/service/wire.go b/pkg/service/wire.go index 3a5d7621d..d8457bf46 100644 --- a/pkg/service/wire.go +++ b/pkg/service/wire.go @@ -37,6 +37,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live createClientConfiguration, routing.CreateRouter, getRoomConf, + config.DefaultAPIConfig, wire.Bind(new(routing.MessageRouter), new(routing.Router)), wire.Bind(new(livekit.RoomService), new(*RoomService)), telemetry.NewAnalyticsService, diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index af7110598..15354c241 100644 --- a/pkg/service/wire_gen.go +++ b/pkg/service/wire_gen.go @@ -33,6 +33,7 @@ import ( func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*LivekitServer, error) { roomConfig := getRoomConf(conf) + apiConfig := config.DefaultAPIConfig() universalClient, err := createRedisClient(conf) if err != nil { return nil, err @@ -57,7 +58,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live analyticsService := telemetry.NewAnalyticsService(conf, currentNode) telemetryService := telemetry.NewTelemetryService(notifier, analyticsService) rtcEgressLauncher := NewEgressLauncher(rpcClient, egressStore, telemetryService) - roomService, err := NewRoomService(roomConfig, router, roomAllocator, objectStore, rtcEgressLauncher) + roomService, err := NewRoomService(roomConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher) if err != nil { return nil, err }