dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-28 09:24:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Configurable RoomService execution timeout (#1206)

Browse Source 
* 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.
This commit is contained in:
David Zhao
2022-12-04 10:13:09 -08:00
committed by GitHub
parent d146ec7a1f
commit 12ae179be2
6 changed files with 44 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config-sample.yaml
+1 -1
View File
@@ -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.
pkg/config/config.go
+16
View File
@@ -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{
pkg/service/roomservice.go
+21 -22
View File
@@ -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)
}
}
}
pkg/service/roomservice_test.go
+3 -1
View File
@@ -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)
}
pkg/service/wire.go
+1
View File
@@ -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,
pkg/service/wire_gen.go
+2 -1
View File
@@ -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
}