Add sort by option to node selector (#599)

* Add sort by option to node selector

Allow the final decision of a selected node to be based on different
sorting options rather than just random.

Add a `sysload`, `cpuload`, `rooms`, `clients`, `tracks`, and
`bytespersec` sorting options to select the appropriate room when
multiple are suitable for the chosen selector.

Add tests for sysload sort_by option

Closes: #598

* NodeSelector: rename "random" selector to "any"

Since the selector introduces limits but the final selection is now
based on the sort by value, the random selector name does not make
sense. Rename this selector to "any" to accurately reflect that it will
select any available no with no additional selection limits.

Allow the old setting "random" to point to the new selector name, but
print a deprecation warning message.

config-sample.yaml

@@ -165,8 +165,11 @@ keys:
 
 # # node selector
 # node_selector:
-#   # default: random. valid values: random, sysload, regionaware
+#   # default: any. valid values: any, sysload, cpuload, regionaware
 #   kind: sysload
+#   # priority used for selection of node when multiple are available
+#   # default: random. valid values: random, sysload, cpuload, rooms, clients, tracks, bytespersec
+#   sort_by: sysload
 #   # used in sysload and regionaware
 #   # do not assign room to node if load per CPU exceeds sysload_limit
 #   sysload_limit: 0.7

pkg/config/config.go

@@ -162,6 +162,7 @@ type WebHookConfig struct {
 
 type NodeSelectorConfig struct {
 	Kind         string         `yaml:"kind"`
+	SortBy       string         `yaml:"sort_by"`
 	CPULoadLimit float32        `yaml:"cpu_load_limit"`
 	SysloadLimit float32        `yaml:"sysload_limit"`
 	Regions      []RegionConfig `yaml:"regions"`
@@ -231,7 +232,8 @@ func NewConfig(confString string, c *cli.Context) (*Config, error) {
 			Enabled: false,
 		},
 		NodeSelector: NodeSelectorConfig{
-			Kind:         "random",
+			Kind:         "any",
+			SortBy:       "random",
 			SysloadLimit: 0.9,
 			CPULoadLimit: 0.9,
 		},

pkg/routing/selector/any.go

@@ -0,0 +1,19 @@
+package selector
+
+import (
+	"github.com/livekit/protocol/livekit"
+)
+
+// AnySelector selects any available node with no limitations
+type AnySelector struct {
+	SortBy string
+}
+
+func (s *AnySelector) SelectNode(nodes []*livekit.Node) (*livekit.Node, error) {
+	nodes = GetAvailableNodes(nodes)
+	if len(nodes) == 0 {
+		return nil, ErrNoAvailableNodes
+	}
+
+	return SelectSortedNode(nodes, s.SortBy)
+}

pkg/routing/selector/cpuload.go

@@ -1,15 +1,14 @@
 package selector
 
 import (
-	"github.com/thoas/go-funk"
-
 	"github.com/livekit/protocol/livekit"
 )
 
 // CPULoadSelector eliminates nodes that have CPU usage higher than CPULoadLimit
-// then selects a node randomly from nodes that are not overloaded
+// then selects a node from nodes that are not overloaded
 type CPULoadSelector struct {
 	CPULoadLimit float32
+	SortBy       string
 }
 
 func (s *CPULoadSelector) filterNodes(nodes []*livekit.Node) ([]*livekit.Node, error) {
@@ -37,6 +36,5 @@ func (s *CPULoadSelector) SelectNode(nodes []*livekit.Node) (*livekit.Node, erro
 		return nil, err
 	}
 
-	idx := funk.RandomInt(0, len(nodes))
-	return nodes[idx], nil
+	return SelectSortedNode(nodes, s.SortBy)
 }

pkg/routing/selector/cpuload_test.go

@@ -11,7 +11,7 @@ import (
 )
 
 func TestCPULoadSelector_SelectNode(t *testing.T) {
-	sel := selector.CPULoadSelector{CPULoadLimit: 0.8}
+	sel := selector.CPULoadSelector{CPULoadLimit: 0.8, SortBy: "random"}
 
 	var nodes []*livekit.Node
 	_, err := sel.SelectNode(nodes)

pkg/routing/selector/errors.go

@@ -6,4 +6,6 @@ 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")
+	ErrSortByNotSet               = errors.New("sort by option cannot be blank")
+	ErrSortByUnknown              = errors.New("unknown sort by option")
 )

pkg/routing/selector/interfaces.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 
 	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/logger"
 
 	"github.com/livekit/livekit-server/pkg/config"
 )
@@ -18,26 +19,31 @@ type NodeSelector interface {
 func CreateNodeSelector(conf *config.Config) (NodeSelector, error) {
 	kind := conf.NodeSelector.Kind
 	if kind == "" {
-		kind = "random"
+		kind = "any"
 	}
 	switch kind {
+	case "any":
+		return &AnySelector{conf.NodeSelector.SortBy}, nil
 	case "cpuload":
 		return &CPULoadSelector{
 			CPULoadLimit: conf.NodeSelector.CPULoadLimit,
+			SortBy: conf.NodeSelector.SortBy,
 		}, nil
 	case "sysload":
 		return &SystemLoadSelector{
 			SysloadLimit: conf.NodeSelector.SysloadLimit,
+			SortBy: conf.NodeSelector.SortBy,
 		}, nil
 	case "regionaware":
-		s, err := NewRegionAwareSelector(conf.Region, conf.NodeSelector.Regions)
+		s, err := NewRegionAwareSelector(conf.Region, conf.NodeSelector.Regions, conf.NodeSelector.SortBy)
 		if err != nil {
 			return nil, err
 		}
 		s.SysloadLimit = conf.NodeSelector.SysloadLimit
 		return s, nil
 	case "random":
-		return &RandomSelector{}, nil
+		logger.Warnw("random node selector is deprecated, please switch to \"any\" or another selector", nil)
+		return &AnySelector{conf.NodeSelector.SortBy}, nil
 	default:
 		return nil, ErrUnsupportedSelector
 	}

pkg/routing/selector/random.go

@@ -1,21 +0,0 @@
-package selector
-
-import (
-	"github.com/thoas/go-funk"
-
-	"github.com/livekit/protocol/livekit"
-)
-
-// RandomSelector selects an available node at random
-type RandomSelector struct {
-}
-
-func (s *RandomSelector) SelectNode(nodes []*livekit.Node) (*livekit.Node, error) {
-	nodes = GetAvailableNodes(nodes)
-	if len(nodes) == 0 {
-		return nil, ErrNoAvailableNodes
-	}
-
-	idx := funk.RandomInt(0, len(nodes))
-	return nodes[idx], nil
-}

pkg/routing/selector/regionaware.go

@@ -3,8 +3,6 @@ package selector
 import (
 	"math"
 
-	"github.com/thoas/go-funk"
-
 	"github.com/livekit/protocol/livekit"
 
 	"github.com/livekit/livekit-server/pkg/config"
@@ -16,9 +14,10 @@ type RegionAwareSelector struct {
 	CurrentRegion   string
 	regionDistances map[string]float64
 	regions         []config.RegionConfig
+	SortBy          string
 }
 
-func NewRegionAwareSelector(currentRegion string, regions []config.RegionConfig) (*RegionAwareSelector, error) {
+func NewRegionAwareSelector(currentRegion string, regions []config.RegionConfig, sortBy string) (*RegionAwareSelector, error) {
 	if currentRegion == "" {
 		return nil, ErrCurrentRegionNotSet
 	}
@@ -27,6 +26,7 @@ func NewRegionAwareSelector(currentRegion string, regions []config.RegionConfig)
 		CurrentRegion:   currentRegion,
 		regionDistances: make(map[string]float64),
 		regions:         regions,
+		SortBy:          sortBy,
 	}
 
 	var currentRC *config.RegionConfig
@@ -80,8 +80,7 @@ func (s *RegionAwareSelector) SelectNode(nodes []*livekit.Node) (*livekit.Node,
 		nodes = nearestNodes
 	}
 
-	idx := funk.RandomInt(0, len(nodes))
-	return nodes[idx], nil
+	return SelectSortedNode(nodes, s.SortBy)
 }
 
 // haversine(θ) function

pkg/routing/selector/regionaware_test.go

@@ -18,6 +18,7 @@ const (
 	regionWest    = "us-west"
 	regionEast    = "us-east"
 	regionSeattle = "seattle"
+	sortBy        = "random"
 )
 
 func TestRegionAwareRouting(t *testing.T) {
@@ -42,7 +43,7 @@ func TestRegionAwareRouting(t *testing.T) {
 		nodes := []*livekit.Node{
 			newTestNodeInRegion("", false),
 		}
-		s, err := selector.NewRegionAwareSelector(regionEast, nil)
+		s, err := selector.NewRegionAwareSelector(regionEast, nil, sortBy)
 		require.NoError(t, err)
 
 		node, err := s.SelectNode(nodes)
@@ -58,7 +59,7 @@ func TestRegionAwareRouting(t *testing.T) {
 			expectedNode,
 			newTestNodeInRegion(regionEast, false),
 		}
-		s, err := selector.NewRegionAwareSelector(regionEast, rc)
+		s, err := selector.NewRegionAwareSelector(regionEast, rc, sortBy)
 		require.NoError(t, err)
 		s.SysloadLimit = loadLimit
 
@@ -75,7 +76,7 @@ func TestRegionAwareRouting(t *testing.T) {
 			newTestNodeInRegion(regionWest, true),
 			newTestNodeInRegion(regionEast, false),
 		}
-		s, err := selector.NewRegionAwareSelector(regionEast, rc)
+		s, err := selector.NewRegionAwareSelector(regionEast, rc, sortBy)
 		require.NoError(t, err)
 		s.SysloadLimit = loadLimit
 
@@ -91,7 +92,7 @@ func TestRegionAwareRouting(t *testing.T) {
 			expectedNode,
 			newTestNodeInRegion(regionEast, true),
 		}
-		s, err := selector.NewRegionAwareSelector(regionSeattle, rc)
+		s, err := selector.NewRegionAwareSelector(regionSeattle, rc, sortBy)
 		require.NoError(t, err)
 		s.SysloadLimit = loadLimit
 
@@ -109,7 +110,7 @@ func TestRegionAwareRouting(t *testing.T) {
 			expectedNode,
 			expectedNode,
 		}
-		s, err := selector.NewRegionAwareSelector(regionSeattle, rc)
+		s, err := selector.NewRegionAwareSelector(regionSeattle, rc, sortBy)
 		require.NoError(t, err)
 		s.SysloadLimit = loadLimit
 
@@ -122,7 +123,7 @@ func TestRegionAwareRouting(t *testing.T) {
 		nodes := []*livekit.Node{
 			newTestNodeInRegion(regionWest, true),
 		}
-		s, err := selector.NewRegionAwareSelector(regionEast, rc)
+		s, err := selector.NewRegionAwareSelector(regionEast, rc, sortBy)
 		require.NoError(t, err)
 
 		node, err := s.SelectNode(nodes)

pkg/routing/selector/sortby_test.go

@@ -0,0 +1,50 @@
+package selector_test
+
+import (
+	"testing"
+
+	"github.com/livekit/protocol/livekit"
+
+	"github.com/livekit/livekit-server/pkg/routing/selector"
+)
+
+func SortByTest(t *testing.T, sortBy string) {
+	sel := selector.SystemLoadSelector{SortBy: sortBy}
+	nodes := []*livekit.Node{nodeLoadLow, nodeLoadMedium, nodeLoadHigh}
+
+	for i := 0; i < 5; i++ {
+		node, err := sel.SelectNode(nodes)
+		if err != nil {
+			t.Error(err)
+		}
+		if node != nodeLoadLow {
+			t.Error("selected the wrong node for SortBy:", sortBy)
+		}
+	}
+}
+
+func TestSortByErrors(t *testing.T) {
+	sel := selector.SystemLoadSelector{}
+	nodes := []*livekit.Node{nodeLoadLow, nodeLoadMedium, nodeLoadHigh}
+
+	// Test unset sort by option error
+	_, err := sel.SelectNode(nodes)
+	if err != selector.ErrSortByNotSet {
+		t.Error("shouldn't allow empty sortBy")
+	}
+
+	// Test unknown sort by option error
+	sel.SortBy = "testFail"
+	_, err = sel.SelectNode(nodes)
+	if err != selector.ErrSortByUnknown {
+		t.Error("shouldn't allow unknown sortBy")
+	}
+}
+
+func TestSortBy(t *testing.T) {
+	sortByTests := []string{"sysload", "cpuload", "rooms", "clients", "tracks", "bytespersec"}
+
+	for _, sortBy := range sortByTests {
+		SortByTest(t, sortBy)
+	}
+}

pkg/routing/selector/sysload.go

@@ -1,15 +1,14 @@
 package selector
 
 import (
-	"github.com/thoas/go-funk"
-
 	"github.com/livekit/protocol/livekit"
 )
 
 // SystemLoadSelector eliminates nodes that surpass has a per-cpu node higher than SysloadLimit
-// then selects a node randomly from nodes that are not overloaded
+// then selects a node from nodes that are not overloaded
 type SystemLoadSelector struct {
 	SysloadLimit float32
+	SortBy       string
 }
 
 func (s *SystemLoadSelector) filterNodes(nodes []*livekit.Node) ([]*livekit.Node, error) {
@@ -20,12 +19,7 @@ func (s *SystemLoadSelector) filterNodes(nodes []*livekit.Node) ([]*livekit.Node
 
 	nodesLowLoad := make([]*livekit.Node, 0)
 	for _, node := range nodes {
-		stats := node.Stats
-		numCpus := stats.NumCpus
-		if numCpus == 0 {
-			numCpus = 1
-		}
-		if stats.LoadAvgLast1Min/float32(numCpus) < s.SysloadLimit {
+		if GetNodeSysload(node) < s.SysloadLimit {
 			nodesLowLoad = append(nodesLowLoad, node)
 		}
 	}
@@ -41,6 +35,5 @@ func (s *SystemLoadSelector) SelectNode(nodes []*livekit.Node) (*livekit.Node, e
 		return nil, err
 	}
 
-	idx := funk.RandomInt(0, len(nodes))
-	return nodes[idx], nil
+	return SelectSortedNode(nodes, s.SortBy)
 }

pkg/routing/selector/sysload_test.go

@@ -19,6 +19,28 @@ var (
 			NumCpus:         1,
 			CpuLoad:         0.1,
 			LoadAvgLast1Min: 0.0,
+			NumRooms: 1,
+			NumClients: 2,
+			NumTracksIn: 4,
+			NumTracksOut: 8,
+			BytesInPerSec: 1000,
+			BytesOutPerSec: 2000,
+		},
+	}
+
+	nodeLoadMedium = &livekit.Node{
+		State: livekit.NodeState_SERVING,
+		Stats: &livekit.NodeStats{
+			UpdatedAt:       time.Now().Unix(),
+			NumCpus:         1,
+			CpuLoad:         0.5,
+			LoadAvgLast1Min: 0.5,
+			NumRooms: 5,
+			NumClients: 10,
+			NumTracksIn: 20,
+			NumTracksOut: 200,
+			BytesInPerSec: 5000,
+			BytesOutPerSec: 10000,
 		},
 	}
 
@@ -29,12 +51,18 @@ var (
 			NumCpus:         1,
 			CpuLoad:         0.99,
 			LoadAvgLast1Min: 2.0,
+			NumRooms: 10,
+			NumClients: 20,
+			NumTracksIn: 40,
+			NumTracksOut: 800,
+			BytesInPerSec: 10000,
+			BytesOutPerSec: 40000,
 		},
 	}
 )
 
 func TestSystemLoadSelector_SelectNode(t *testing.T) {
-	sel := selector.SystemLoadSelector{SysloadLimit: 1.0}
+	sel := selector.SystemLoadSelector{SysloadLimit: 1.0, SortBy: "random"}
 
 	var nodes []*livekit.Node
 	_, err := sel.SelectNode(nodes)

pkg/routing/selector/utils.go

@@ -1,6 +1,7 @@
 package selector
 
 import (
+	"sort"
 	"time"
 
 	"github.com/thoas/go-funk"
@@ -24,6 +25,15 @@ func GetAvailableNodes(nodes []*livekit.Node) []*livekit.Node {
 	}).([]*livekit.Node)
 }
 
+func GetNodeSysload(node *livekit.Node) float32 {
+	stats := node.Stats
+	numCpus := stats.NumCpus
+	if numCpus == 0 {
+		numCpus = 1
+	}
+	return stats.LoadAvgLast1Min / float32(numCpus)
+}
+
 // TODO: check remote node configured limit, instead of this node's config
 func LimitsReached(limitConfig config.LimitConfig, nodeStats *livekit.NodeStats) bool {
 	if nodeStats == nil {
@@ -39,3 +49,48 @@ func LimitsReached(limitConfig config.LimitConfig, nodeStats *livekit.NodeStats)
 
 	return false
 }
+
+func SelectSortedNode(nodes []*livekit.Node, sortBy string) (*livekit.Node, error) {
+	if sortBy == "" {
+		return nil, ErrSortByNotSet
+	}
+
+	// Return a node based on what it should be sorted by for priority
+	switch sortBy {
+	case "random":
+		idx := funk.RandomInt(0, len(nodes))
+		return nodes[idx], nil
+	case "sysload":
+		sort.Slice(nodes, func(i, j int) bool {
+			return GetNodeSysload(nodes[i]) < GetNodeSysload(nodes[j])
+		})
+		return nodes[0], nil
+	case "cpuload":
+		sort.Slice(nodes, func(i, j int) bool {
+			return nodes[i].Stats.CpuLoad < nodes[j].Stats.CpuLoad
+		})
+		return nodes[0], nil
+	case "rooms":
+		sort.Slice(nodes, func(i, j int) bool {
+			return nodes[i].Stats.NumRooms < nodes[j].Stats.NumRooms
+		})
+		return nodes[0], nil
+	case "clients":
+		sort.Slice(nodes, func(i, j int) bool {
+			return nodes[i].Stats.NumClients < nodes[j].Stats.NumClients
+		})
+		return nodes[0], nil
+	case "tracks":
+		sort.Slice(nodes, func(i, j int) bool {
+			return nodes[i].Stats.NumTracksIn + nodes[i].Stats.NumTracksOut < nodes[j].Stats.NumTracksIn + nodes[j].Stats.NumTracksOut
+		})
+		return nodes[0], nil
+	case "bytespersec":
+		sort.Slice(nodes, func(i, j int) bool {
+			return nodes[i].Stats.BytesInPerSec + nodes[i].Stats.BytesOutPerSec < nodes[j].Stats.BytesInPerSec + nodes[j].Stats.BytesOutPerSec
+		})
+		return nodes[0], nil
+	default:
+		return nil, ErrSortByUnknown
+	}
+}