From c745f0a31836180ca88d5704c7beaaa927c4f7e3 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Wed, 6 Oct 2021 21:40:04 -0700 Subject: [PATCH] Region aware routing (#135) * Region aware routing * update config, fix spacing on commented out lines --- .gitignore | 6 + config-sample.yaml | 151 ++++++++++-------- go.mod | 2 + go.sum | 4 +- magefile.go | 2 + pkg/config/config.go | 7 +- pkg/routing/errors.go | 1 - pkg/routing/redisrouter.go | 3 +- pkg/routing/selector/errors.go | 9 ++ .../{selectorrandom.go => selector/random.go} | 3 +- pkg/routing/selector/regionaware.go | 115 +++++++++++++ pkg/routing/selector/regionaware_test.go | 111 +++++++++++++ .../sysload.go} | 14 +- .../sysload_test.go} | 7 +- pkg/routing/selector/utils.go | 24 +++ pkg/routing/{ => selector}/utils_test.go | 9 +- pkg/routing/utils.go | 17 -- pkg/service/errors.go | 1 + pkg/service/roomallocator.go | 3 +- pkg/service/roomallocator_test.go | 3 +- pkg/service/utils.go | 22 ++- pkg/service/wire_gen.go | 5 +- 22 files changed, 406 insertions(+), 113 deletions(-) create mode 100644 pkg/routing/selector/errors.go rename pkg/routing/{selectorrandom.go => selector/random.go} (84%) create mode 100644 pkg/routing/selector/regionaware.go create mode 100644 pkg/routing/selector/regionaware_test.go rename pkg/routing/{selectorsystemload.go => selector/sysload.go} (66%) rename pkg/routing/{selectorsystemload_test.go => selector/sysload_test.go} (89%) create mode 100644 pkg/routing/selector/utils.go rename pkg/routing/{ => selector}/utils_test.go (73%) diff --git a/.gitignore b/.gitignore index ffd84986d..51271e375 100644 --- a/.gitignore +++ b/.gitignore @@ -19,3 +19,9 @@ livekit-server # vendor/ bin/ proto/ + +# Mac +.DS_Store + +# IDE +.idea/jsonSchemas.xml \ No newline at end of file diff --git a/config-sample.yaml b/config-sample.yaml index a5d604f45..1f5bf9571 100644 --- a/config-sample.yaml +++ b/config-sample.yaml @@ -9,9 +9,9 @@ log_level: info # clients could connect to any node and be routed to the same room redis: address: redis.host:6379 -# db: 0 -# username: myuser -# password: mypassword + # db: 0 + # username: myuser + # password: mypassword # WebRTC configuration rtc: @@ -34,24 +34,24 @@ rtc: # port_range_start & end must not be set for this config to take effect # udp_port: 7882 # optional settings -# # when using REMB, the max bitrate that the SFU would accept, defaults to 3Mbps -# max_bitrate: 3145728 -# # number of packets to buffer in the SFU, defaults to 500 -# packet_buffer_size: 500 -# # optional STUN servers for LiveKit clients to use. Clients will be configured to use these STUN servers automatically. -# # by default LiveKit clients use Google's public STUN servers -# stun_servers: -# - server1 -# # minimum amount of time between pli/fir rtcp packets being sent to an individual -# # producer. Increasing these times can lead to longer black screens when participants join, -# # while reducing them can lead to higher producer bitrates. -# pli_throttle: -# low_quality: 500ms -# mid_quality: 1s -# high_quality: 1s + # # when using REMB, the max bitrate that the SFU would accept, defaults to 3Mbps + # max_bitrate: 3145728 + # # number of packets to buffer in the SFU, defaults to 500 + # packet_buffer_size: 500 + # # optional STUN servers for LiveKit clients to use. Clients will be configured to use these STUN servers automatically. + # # by default LiveKit clients use Google's public STUN servers + # stun_servers: + # - server1 + # # minimum amount of time between pli/fir rtcp packets being sent to an individual + # # producer. Increasing these times can lead to longer black screens when participants join, + # # while reducing them can lead to higher producer bitrates. + # pli_throttle: + # low_quality: 500ms + # mid_quality: 1s + # high_quality: 1s # when enabled, LiveKit will expose prometheus metrics on :6789/metrics -#prometheus_port: 6789 +# prometheus_port: 6789 # API key / secret pairs. # Keys are used for JWT authentication, server APIs would require a keypair in order to generate access tokens @@ -63,68 +63,77 @@ keys: # Default room config # Each room created will inherit these settings. If rooms are created explicitly with CreateRoom, they will take # precedence over defaults -#room: -# # 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 -# max_participants: 0 -# # only accept specific codecs for clients publishing to this room -# # this is useful to standardize codecs across clients -# # other supported codecs are video/h264, video/vp9 -# enabled_codecs: -# - mime: audio/opus -# - mime: video/vp8 -# # allow tracks to be unmuted remotely, defaults to false -# # tracks can always be muted from the Room Service APIs -# enable_remote_unmute: true +# room: +# # 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 +# max_participants: 0 +# # only accept specific codecs for clients publishing to this room +# # this is useful to standardize codecs across clients +# # other supported codecs are video/h264, video/vp9 +# enabled_codecs: +# - mime: audio/opus +# - mime: video/vp8 +# # allow tracks to be unmuted remotely, defaults to false +# # tracks can always be muted from the Room Service APIs +# enable_remote_unmute: true # Webhooks # when configured, LiveKit notifies your URL handler with room events -#webhook: -# # the API key to use in order to sign the message -# # this must match one of the keys LiveKit is configured with -# api_key: -# # list of URLs to be notified of room events -# urls: -# - https://your-host.com/handler +# webhook: +# # the API key to use in order to sign the message +# # this must match one of the keys LiveKit is configured with +# api_key: +# # list of URLs to be notified of room events +# urls: +# - https://your-host.com/handler # customize audio level sensitivity -#audio: -# # minimum level to be considered active, 0-127, where 0 is loudest -# # defaults to 30 -# active_level: 30 -# # percentile to measure, a participant is considered active if it has exceeded the -# # ActiveLevel more than MinPercentile% of the time -# # defaults to 40 -# min_percentile: 40 -# # frequency in ms to notify changes to clients, defaults to 500 -# update_interval: 500 -# # to prevent speaker updates from too jumpy, smooth out values over N samples -# smooth_intervals: 4 +# audio: +# # minimum level to be considered active, 0-127, where 0 is loudest +# # defaults to 30 +# active_level: 30 +# # percentile to measure, a participant is considered active if it has exceeded the +# # ActiveLevel more than MinPercentile% of the time +# # defaults to 40 +# min_percentile: 40 +# # frequency in ms to notify changes to clients, defaults to 500 +# update_interval: 500 +# # to prevent speaker updates from too jumpy, smooth out values over N samples +# smooth_intervals: 4 # turn server -#turn: -# # Uses TLS. Requires cert and key pem files by either: -# # - using turn.secretName if deploying with our helm chart, or -# # - setting LIVEKIT_TURN_CERT and LIVEKIT_TURN_KEY env vars with file locations, or -# # - using cert_file and key_file below -# # defaults to false -# enabled: false -# # defaults to 3478 - recommended to 443 if not running HTTP3/QUIC server -# # only 53/80/443 are allowed if less than 1024 -# udp_port: 3478 -# # defaults to 5349 - if not using a load balancer, this must be set to 443 -# tls_port: 5349 -# # needs to match tls cert domain -# domain: turn.myhost.com -# # optional -# # cert_file: /path/to/cert.pem -# # key_file: /path/to/key.pem +# turn: +# # Uses TLS. Requires cert and key pem files by either: +# # - using turn.secretName if deploying with our helm chart, or +# # - setting LIVEKIT_TURN_CERT and LIVEKIT_TURN_KEY env vars with file locations, or +# # - using cert_file and key_file below +# # defaults to false +# enabled: false +# # defaults to 3478 - recommended to 443 if not running HTTP3/QUIC server +# # only 53/80/443 are allowed if less than 1024 +# udp_port: 3478 +# # defaults to 5349 - if not using a load balancer, this must be set to 443 +# tls_port: 5349 +# # needs to match tls cert domain +# domain: turn.myhost.com +# # optional +# # cert_file: /path/to/cert.pem +# # key_file: /path/to/key.pem + +# Region of the current node. Required if using regionaware node selector +# region: us-west-2 # # node selector # node_selector: -# # default: random. valid values: random, sysload +# # default: random. valid values: random, sysload, regionaware # kind: sysload -# # used in sysload node selector +# # used in sysload and regionaware # # do not assign room to node if load per CPU exceeds sysload_limit # sysload_limit: 0.7 +# # used in regionaware +# # list of regions and their lat/lon coordinates +# regions: +# - name: us-west-2 +# lat: 44.19434095976287 +# lon: -123.0674908379146 diff --git a/go.mod b/go.mod index 459ed406e..bec821c1d 100644 --- a/go.mod +++ b/go.mod @@ -47,3 +47,5 @@ require ( ) replace github.com/pion/ion-sfu => github.com/livekit/ion-sfu v1.20.11 + +replace github.com/livekit/protocol => github.com/livekit/protocol v0.9.6-0.20211006235141-c785be2288fd diff --git a/go.sum b/go.sum index 989b2b934..6c0373c38 100644 --- a/go.sum +++ b/go.sum @@ -246,8 +246,8 @@ github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts= github.com/livekit/ion-sfu v1.20.11 h1:QhfksN1jcYn9VvVLKHSxrlaPNpSxx1j57tsuncF5DbI= github.com/livekit/ion-sfu v1.20.11/go.mod h1:IVcCb8yMl5qHq+8InP5v9HgrjgnPZny8GhFS86YOtug= -github.com/livekit/protocol v0.9.4 h1:1vA49MwawrriOS3VvPqYfLiFI94XWvYGaWfkHN6ah6Y= -github.com/livekit/protocol v0.9.4/go.mod h1:MEKn847Iu/2U8ClZyUmEm2oHn8k9fnSHy81Wv8kkSDo= +github.com/livekit/protocol v0.9.6-0.20211006235141-c785be2288fd h1:Q7wRRzJGqK6i3hAhgYtD82h2TypwSrySTrtk6AKWtI4= +github.com/livekit/protocol v0.9.6-0.20211006235141-c785be2288fd/go.mod h1:MEKn847Iu/2U8ClZyUmEm2oHn8k9fnSHy81Wv8kkSDo= github.com/lucsky/cuid v1.0.2 h1:z4XlExeoderxoPj2/dxKOyPxe9RCOu7yNq9/XWxIUMQ= github.com/lucsky/cuid v1.0.2/go.mod h1:QaaJqckboimOmhRSJXSx/+IT+VTfxfPGSo/6mfgUfmE= github.com/lyft/protoc-gen-validate v0.0.13/go.mod h1:XbGvPuh87YZc5TdIa2/I4pLk0QoUACkjt2znoq26NVQ= diff --git a/magefile.go b/magefile.go index 775b50060..1edb8e6b7 100644 --- a/magefile.go +++ b/magefile.go @@ -124,6 +124,7 @@ func PublishDocker() error { // run unit tests, skipping integration func Test() error { + mg.Deps(generateWire) cmd := exec.Command("go", "test", "-short", "./...") connectStd(cmd) return cmd.Run() @@ -131,6 +132,7 @@ func Test() error { // run all tests including integration func TestAll() error { + mg.Deps(generateWire) // "-v", "-race", cmd := exec.Command("go", "test", "./...", "-count=1", "-timeout=3m") connectStd(cmd) diff --git a/pkg/config/config.go b/pkg/config/config.go index 2383d74a2..dfa33c369 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -5,6 +5,7 @@ import ( "os" "time" + "github.com/livekit/livekit-server/pkg/routing/selector" "github.com/mitchellh/go-homedir" "github.com/pion/webrtc/v3" "github.com/pkg/errors" @@ -29,6 +30,7 @@ type Config struct { NodeSelector NodeSelectorConfig `yaml:"node_selector"` KeyFile string `yaml:"key_file"` Keys map[string]string `yaml:"keys"` + Region string `yaml:"region"` LogLevel string `yaml:"log_level"` Development bool `yaml:"development"` @@ -109,8 +111,9 @@ type WebHookConfig struct { } type NodeSelectorConfig struct { - Kind string `yaml:"kind"` - SysloadLimit float32 `yaml:"sysload_limit"` + Kind string `yaml:"kind"` + SysloadLimit float32 `yaml:"sysload_limit"` + Regions []selector.RegionConfig `yaml:"regions"` } func NewConfig(confString string, c *cli.Context) (*Config, error) { diff --git a/pkg/routing/errors.go b/pkg/routing/errors.go index 15fd80525..c0a2929f8 100644 --- a/pkg/routing/errors.go +++ b/pkg/routing/errors.go @@ -6,7 +6,6 @@ var ( ErrNotFound = errors.New("could not find object") ErrIPNotSet = errors.New("ip address is required and not set") ErrHandlerNotDefined = errors.New("handler not defined") - ErrNoAvailableNodes = errors.New("could not find any available nodes") ErrIncorrectRTCNode = errors.New("current node isn't the RTC node for the room") ErrNodeNotFound = errors.New("could not locate the node") ErrInvalidRouterMessage = errors.New("invalid router message") diff --git a/pkg/routing/redisrouter.go b/pkg/routing/redisrouter.go index 364a20460..0be9cdc81 100644 --- a/pkg/routing/redisrouter.go +++ b/pkg/routing/redisrouter.go @@ -5,6 +5,7 @@ import ( "time" "github.com/go-redis/redis/v8" + "github.com/livekit/livekit-server/pkg/routing/selector" "github.com/livekit/protocol/logger" livekit "github.com/livekit/protocol/proto" "github.com/livekit/protocol/utils" @@ -64,7 +65,7 @@ func (r *RedisRouter) RemoveDeadNodes() error { return err } for _, n := range nodes { - if !IsAvailable(n) { + if !selector.IsAvailable(n) { if err := r.rc.HDel(context.Background(), NodesKey, n.Id).Err(); err != nil { return err } diff --git a/pkg/routing/selector/errors.go b/pkg/routing/selector/errors.go new file mode 100644 index 000000000..ab6155aeb --- /dev/null +++ b/pkg/routing/selector/errors.go @@ -0,0 +1,9 @@ +package selector + +import "errors" + +var ( + ErrNoAvailableNodes = errors.New("could not find any available nodes") + ErrCurrentRegionNotSet = errors.New("current region cannot be blank") + ErrCurrentRegionUnknownLatLon = errors.New("unknown lat and lon for the current region") +) diff --git a/pkg/routing/selectorrandom.go b/pkg/routing/selector/random.go similarity index 84% rename from pkg/routing/selectorrandom.go rename to pkg/routing/selector/random.go index deac3c556..d67240f8f 100644 --- a/pkg/routing/selectorrandom.go +++ b/pkg/routing/selector/random.go @@ -1,10 +1,11 @@ -package routing +package selector import ( livekit "github.com/livekit/protocol/proto" "github.com/thoas/go-funk" ) +// RandomSelector selects an available node at random type RandomSelector struct { } diff --git a/pkg/routing/selector/regionaware.go b/pkg/routing/selector/regionaware.go new file mode 100644 index 000000000..cfdce0e40 --- /dev/null +++ b/pkg/routing/selector/regionaware.go @@ -0,0 +1,115 @@ +package selector + +import ( + "math" + + livekit "github.com/livekit/protocol/proto" + "github.com/thoas/go-funk" +) + +// RegionConfig lists available regions and their latitude/longitude, so the selector would prefer +// regions that are closer +type RegionConfig struct { + Name string `yaml:"name"` + Lat float64 `yaml:"lat"` + Lon float64 `yaml:"lon"` +} + +// RegionAwareSelector prefers available nodes that are closest to the region of the current instance +type RegionAwareSelector struct { + SystemLoadSelector + CurrentRegion string + regionDistances map[string]float64 + regions []RegionConfig +} + +func NewRegionAwareSelector(currentRegion string, regions []RegionConfig) (*RegionAwareSelector, error) { + if currentRegion == "" { + return nil, ErrCurrentRegionNotSet + } + // build internal map of distances + s := &RegionAwareSelector{ + CurrentRegion: currentRegion, + regionDistances: make(map[string]float64), + regions: regions, + } + + var currentRC *RegionConfig + + for _, region := range regions { + if region.Name == currentRegion { + currentRC = ®ion + break + } + } + + if currentRC == nil && len(regions) > 0 { + return nil, ErrCurrentRegionUnknownLatLon + } + + if currentRC != nil { + for _, region := range regions { + s.regionDistances[region.Name] = distanceBetween(currentRC.Lat, currentRC.Lon, region.Lat, region.Lon) + } + } + + return s, nil +} + +func (s *RegionAwareSelector) SelectNode(nodes []*livekit.Node, room *livekit.Room) (*livekit.Node, error) { + nodes, err := s.SystemLoadSelector.filterNodes(nodes) + if err != nil { + return nil, err + } + + // find nodes nearest to current region + var nearestNodes []*livekit.Node + nearestRegion := s.CurrentRegion + minDist := math.MaxFloat64 + for _, node := range nodes { + if node.Region == nearestRegion { + nearestNodes = append(nearestNodes, node) + continue + } + if dist, ok := s.regionDistances[node.Region]; ok { + if dist < minDist { + minDist = dist + nearestRegion = node.Region + nearestNodes = nearestNodes[:0] + nearestNodes = append(nearestNodes, node) + } + } + } + + if len(nearestNodes) > 0 { + nodes = nearestNodes + } + + idx := funk.RandomInt(0, len(nodes)) + return nodes[idx], nil +} + +// haversin(θ) function +func hsin(theta float64) float64 { + return math.Pow(math.Sin(theta/2), 2) +} + +// Haversin Distance Formula +// http://en.wikipedia.org/wiki/Haversine_formula +// from https://gist.github.com/cdipaolo/d3f8db3848278b49db68 +func distanceBetween(lat1, lon1, lat2, lon2 float64) float64 { + // convert to radians + // must cast radius as float to multiply later + var la1, lo1, la2, lo2, r float64 + la1 = lat1 * math.Pi / 180 + lo1 = lon1 * math.Pi / 180 + la2 = lat2 * math.Pi / 180 + lo2 = lon2 * math.Pi / 180 + + r = 6378100 // Earth radius in METERS + + // calculate + h := hsin(la2-la1) + math.Cos(la1)*math.Cos(la2)*hsin(lo2-lo1) + + return 2 * r * math.Asin(math.Sqrt(h)) +} diff --git a/pkg/routing/selector/regionaware_test.go b/pkg/routing/selector/regionaware_test.go new file mode 100644 index 000000000..3763640b5 --- /dev/null +++ b/pkg/routing/selector/regionaware_test.go @@ -0,0 +1,111 @@ +package selector_test + +import ( + "testing" + "time" + + "github.com/livekit/livekit-server/pkg/routing/selector" + livekit "github.com/livekit/protocol/proto" + "github.com/livekit/protocol/utils" + "github.com/stretchr/testify/require" +) + +const ( + loadLimit = 0.5 + regionWest = "us-west" + regionEast = "us-east" + regionSeattle = "seattle" +) + +func TestRegionAwareRouting(t *testing.T) { + rc := []selector.RegionConfig{ + { + Name: regionWest, + Lat: 37.64046607830567, + Lon: -120.88026233189062, + }, + { + Name: regionEast, + Lat: 40.68914362140307, + Lon: -74.04445748616385, + }, + { + Name: regionSeattle, + Lat: 47.620426730945454, + Lon: -122.34938468973702, + }, + } + t.Run("works without region config", func(t *testing.T) { + nodes := []*livekit.Node{ + newTestNodeInRegion("", false), + } + s, err := selector.NewRegionAwareSelector(regionEast, nil) + require.NoError(t, err) + + node, err := s.SelectNode(nodes, nil) + require.NoError(t, err) + require.NotNil(t, node) + }) + + t.Run("picks available nodes in same region", func(t *testing.T) { + expectedNode := newTestNodeInRegion(regionEast, true) + nodes := []*livekit.Node{ + newTestNodeInRegion(regionSeattle, true), + newTestNodeInRegion(regionWest, true), + expectedNode, + newTestNodeInRegion(regionEast, false), + } + s, err := selector.NewRegionAwareSelector(regionEast, rc) + require.NoError(t, err) + s.SysloadLimit = loadLimit + + node, err := s.SelectNode(nodes, nil) + require.NoError(t, err) + require.Equal(t, expectedNode, node) + }) + + t.Run("picks closest node in a diff region", func(t *testing.T) { + expectedNode := newTestNodeInRegion(regionWest, true) + nodes := []*livekit.Node{ + newTestNodeInRegion(regionSeattle, false), + expectedNode, + newTestNodeInRegion(regionEast, true), + } + s, err := selector.NewRegionAwareSelector(regionSeattle, rc) + require.NoError(t, err) + s.SysloadLimit = loadLimit + + node, err := s.SelectNode(nodes, nil) + require.NoError(t, err) + require.Equal(t, expectedNode, node) + }) + + t.Run("functions when current region is full", func(t *testing.T) { + nodes := []*livekit.Node{ + newTestNodeInRegion(regionWest, true), + } + s, err := selector.NewRegionAwareSelector(regionEast, rc) + require.NoError(t, err) + + node, err := s.SelectNode(nodes, nil) + require.NoError(t, err) + require.NotNil(t, node) + }) +} + +func newTestNodeInRegion(region string, available bool) *livekit.Node { + load := float32(0.4) + if !available { + load = 1.0 + } + return &livekit.Node{ + Id: utils.NewGuid(utils.NodePrefix), + Region: region, + State: livekit.NodeState_SERVING, + Stats: &livekit.NodeStats{ + UpdatedAt: time.Now().Unix(), + NumCpus: 1, + LoadAvgLast1Min: load, + }, + } +} diff --git a/pkg/routing/selectorsystemload.go b/pkg/routing/selector/sysload.go similarity index 66% rename from pkg/routing/selectorsystemload.go rename to pkg/routing/selector/sysload.go index 9f8f97683..22a2af95a 100644 --- a/pkg/routing/selectorsystemload.go +++ b/pkg/routing/selector/sysload.go @@ -1,15 +1,17 @@ -package routing +package selector import ( livekit "github.com/livekit/protocol/proto" "github.com/thoas/go-funk" ) +// SystemLoadSelector eliminates nodes that surpass has a per-cpu node higher than SysloadLimit +// then selects a node randomly from nodes that are not overloaded type SystemLoadSelector struct { SysloadLimit float32 } -func (s *SystemLoadSelector) SelectNode(nodes []*livekit.Node, room *livekit.Room) (*livekit.Node, error) { +func (s *SystemLoadSelector) filterNodes(nodes []*livekit.Node) ([]*livekit.Node, error) { nodes = GetAvailableNodes(nodes) if len(nodes) == 0 { return nil, ErrNoAvailableNodes @@ -28,6 +30,14 @@ func (s *SystemLoadSelector) SelectNode(nodes []*livekit.Node, room *livekit.Roo if len(nodesLowLoad) > 0 { nodes = nodesLowLoad } + return nodes, nil +} + +func (s *SystemLoadSelector) SelectNode(nodes []*livekit.Node, room *livekit.Room) (*livekit.Node, error) { + nodes, err := s.filterNodes(nodes) + if err != nil { + return nil, err + } idx := funk.RandomInt(0, len(nodes)) return nodes[idx], nil diff --git a/pkg/routing/selectorsystemload_test.go b/pkg/routing/selector/sysload_test.go similarity index 89% rename from pkg/routing/selectorsystemload_test.go rename to pkg/routing/selector/sysload_test.go index 4c3b90003..7667128bd 100644 --- a/pkg/routing/selectorsystemload_test.go +++ b/pkg/routing/selector/sysload_test.go @@ -1,13 +1,12 @@ -package routing_test +package selector_test import ( "testing" "time" + "github.com/livekit/livekit-server/pkg/routing/selector" livekit "github.com/livekit/protocol/proto" "github.com/stretchr/testify/require" - - "github.com/livekit/livekit-server/pkg/routing" ) var ( @@ -31,7 +30,7 @@ var ( ) func TestSystemLoadSelector_SelectNode(t *testing.T) { - selector := routing.SystemLoadSelector{SysloadLimit: 1.0} + selector := selector.SystemLoadSelector{SysloadLimit: 1.0} nodes := []*livekit.Node{} _, err := selector.SelectNode(nodes, nil) diff --git a/pkg/routing/selector/utils.go b/pkg/routing/selector/utils.go new file mode 100644 index 000000000..7e77cad17 --- /dev/null +++ b/pkg/routing/selector/utils.go @@ -0,0 +1,24 @@ +package selector + +import ( + "time" + + livekit "github.com/livekit/protocol/proto" + "github.com/thoas/go-funk" +) + +const ( + AvailableSeconds = 5 +) + +// checks if a node has been updated recently to be considered for selection +func IsAvailable(node *livekit.Node) bool { + delta := time.Now().Unix() - node.Stats.UpdatedAt + return int(delta) < AvailableSeconds +} + +func GetAvailableNodes(nodes []*livekit.Node) []*livekit.Node { + return funk.Filter(nodes, func(node *livekit.Node) bool { + return IsAvailable(node) && node.State == livekit.NodeState_SERVING + }).([]*livekit.Node) +} diff --git a/pkg/routing/utils_test.go b/pkg/routing/selector/utils_test.go similarity index 73% rename from pkg/routing/utils_test.go rename to pkg/routing/selector/utils_test.go index 715b0e19f..ec0cc064b 100644 --- a/pkg/routing/utils_test.go +++ b/pkg/routing/selector/utils_test.go @@ -1,13 +1,12 @@ -package routing_test +package selector_test import ( "testing" "time" + "github.com/livekit/livekit-server/pkg/routing/selector" livekit "github.com/livekit/protocol/proto" "github.com/stretchr/testify/require" - - "github.com/livekit/livekit-server/pkg/routing" ) func TestIsAvailable(t *testing.T) { @@ -17,7 +16,7 @@ func TestIsAvailable(t *testing.T) { UpdatedAt: time.Now().Unix() - 3, }, } - require.True(t, routing.IsAvailable(n)) + require.True(t, selector.IsAvailable(n)) }) t.Run("expired", func(t *testing.T) { @@ -26,6 +25,6 @@ func TestIsAvailable(t *testing.T) { UpdatedAt: time.Now().Unix() - 20, }, } - require.False(t, routing.IsAvailable(n)) + require.False(t, selector.IsAvailable(n)) }) } diff --git a/pkg/routing/utils.go b/pkg/routing/utils.go index b3011251e..2c9b5d2ee 100644 --- a/pkg/routing/utils.go +++ b/pkg/routing/utils.go @@ -3,25 +3,8 @@ package routing import ( "errors" "strings" - "time" - - livekit "github.com/livekit/protocol/proto" - "github.com/thoas/go-funk" ) -// checks if a node has been updated recently to be considered for selection -func IsAvailable(node *livekit.Node) bool { - delta := time.Now().Unix() - node.Stats.UpdatedAt - limit := statsUpdateInterval.Seconds() * 2 - return float64(delta) < limit -} - -func GetAvailableNodes(nodes []*livekit.Node) []*livekit.Node { - return funk.Filter(nodes, func(node *livekit.Node) bool { - return IsAvailable(node) && node.State == livekit.NodeState_SERVING - }).([]*livekit.Node) -} - func participantKey(roomName, identity string) string { return roomName + "|" + identity } diff --git a/pkg/service/errors.go b/pkg/service/errors.go index 992e45549..2d6c79e9b 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") + ErrUnsupportedSelector = errors.New("unsupported node selector") ) diff --git a/pkg/service/roomallocator.go b/pkg/service/roomallocator.go index f3f6556c8..11353eca1 100644 --- a/pkg/service/roomallocator.go +++ b/pkg/service/roomallocator.go @@ -4,6 +4,7 @@ import ( "context" "time" + "github.com/livekit/livekit-server/pkg/routing/selector" "github.com/livekit/protocol/logger" livekit "github.com/livekit/protocol/proto" "github.com/livekit/protocol/utils" @@ -70,7 +71,7 @@ func (r *RoomAllocator) CreateRoom(ctx context.Context, req *livekit.CreateRoomR } // keep it on that node - if err == nil && routing.IsAvailable(node) { + if err == nil && selector.IsAvailable(node) { return rm, nil } diff --git a/pkg/service/roomallocator_test.go b/pkg/service/roomallocator_test.go index 44d717fdf..76e17e540 100644 --- a/pkg/service/roomallocator_test.go +++ b/pkg/service/roomallocator_test.go @@ -4,6 +4,7 @@ import ( "context" "testing" + "github.com/livekit/livekit-server/pkg/routing/selector" livekit "github.com/livekit/protocol/proto" "github.com/stretchr/testify/require" @@ -31,7 +32,7 @@ func newTestRoomAllocator(t *testing.T) (*service.RoomAllocator, *config.Config) router := &routingfakes.FakeRouter{} conf, err := config.NewConfig("", nil) require.NoError(t, err) - selector := &routing.RandomSelector{} + selector := &selector.RandomSelector{} node, err := routing.NewLocalNode(conf) require.NoError(t, err) diff --git a/pkg/service/utils.go b/pkg/service/utils.go index df95cd095..beb08a259 100644 --- a/pkg/service/utils.go +++ b/pkg/service/utils.go @@ -9,6 +9,7 @@ import ( "github.com/go-redis/redis/v8" "github.com/google/wire" + "github.com/livekit/livekit-server/pkg/routing/selector" "github.com/livekit/protocol/auth" "github.com/livekit/protocol/logger" livekit "github.com/livekit/protocol/proto" @@ -78,14 +79,27 @@ func CreateWebhookNotifier(conf *config.Config, provider auth.KeyProvider) (webh return webhook.NewNotifier(wc.APIKey, secret, wc.URLs), nil } -func CreateNodeSelector(conf *config.Config) routing.NodeSelector { - switch conf.NodeSelector.Kind { +func CreateNodeSelector(conf *config.Config) (routing.NodeSelector, error) { + kind := conf.NodeSelector.Kind + if kind == "" { + kind = "random" + } + switch kind { case "sysload": - return &routing.SystemLoadSelector{ + return &selector.SystemLoadSelector{ SysloadLimit: conf.NodeSelector.SysloadLimit, + }, nil + case "regionaware": + s, err := selector.NewRegionAwareSelector(conf.Region, conf.NodeSelector.Regions) + if err != nil { + return nil, err } + s.SysloadLimit = conf.NodeSelector.SysloadLimit + return s, nil + case "random": + return &selector.RandomSelector{}, nil default: - return &routing.RandomSelector{} + return nil, ErrUnsupportedSelector } } diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index 3e20cee70..ff419d8ce 100644 --- a/pkg/service/wire_gen.go +++ b/pkg/service/wire_gen.go @@ -18,7 +18,10 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live return nil, err } router := createRouter(client, currentNode) - nodeSelector := CreateNodeSelector(conf) + nodeSelector, err := CreateNodeSelector(conf) + if err != nil { + return nil, err + } roomStore := createStore(client) roomAllocator := NewRoomAllocator(conf, router, nodeSelector, roomStore) roomService, err := NewRoomService(roomAllocator, roomStore, router)