Region aware routing (#135)

* Region aware routing

* update config, fix spacing on commented out lines

.gitignore

@@ -19,3 +19,9 @@ livekit-server
 # vendor/
 bin/
 proto/
+
+# Mac
+.DS_Store
+
+# IDE
+.idea/jsonSchemas.xml

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: <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: <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

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

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=

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)

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) {

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")

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
 			}

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")
+)

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 {
 }
 

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 = &region
+			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))
+}

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,
+		},
+	}
+}

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

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)

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)
+}

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))
 	})
 }

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
 }

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")
 )

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
 	}
 

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)
 

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
 	}
 }
 

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)