From f8ae219eab458d28c1151197a166e62bd635a875 Mon Sep 17 00:00:00 2001 From: "Brint E. Kriebel" Date: Thu, 14 Apr 2022 01:25:36 -0700 Subject: [PATCH] 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 | 5 ++- pkg/config/config.go | 4 +- pkg/routing/selector/any.go | 19 ++++++++ pkg/routing/selector/cpuload.go | 8 ++-- pkg/routing/selector/cpuload_test.go | 2 +- pkg/routing/selector/errors.go | 2 + pkg/routing/selector/interfaces.go | 12 ++++-- pkg/routing/selector/random.go | 21 --------- pkg/routing/selector/regionaware.go | 9 ++-- pkg/routing/selector/regionaware_test.go | 13 +++--- pkg/routing/selector/sortby_test.go | 50 +++++++++++++++++++++ pkg/routing/selector/sysload.go | 15 ++----- pkg/routing/selector/sysload_test.go | 30 ++++++++++++- pkg/routing/selector/utils.go | 55 ++++++++++++++++++++++++ 14 files changed, 190 insertions(+), 55 deletions(-) create mode 100644 pkg/routing/selector/any.go delete mode 100644 pkg/routing/selector/random.go create mode 100644 pkg/routing/selector/sortby_test.go diff --git a/config-sample.yaml b/config-sample.yaml index 007a81976..1e67f6d93 100644 --- a/config-sample.yaml +++ b/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 diff --git a/pkg/config/config.go b/pkg/config/config.go index 901591fa4..2fe3d7fb4 100644 --- a/pkg/config/config.go +++ b/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, }, diff --git a/pkg/routing/selector/any.go b/pkg/routing/selector/any.go new file mode 100644 index 000000000..399ad4947 --- /dev/null +++ b/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) +} diff --git a/pkg/routing/selector/cpuload.go b/pkg/routing/selector/cpuload.go index e485bbc4d..61197907b 100644 --- a/pkg/routing/selector/cpuload.go +++ b/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) } diff --git a/pkg/routing/selector/cpuload_test.go b/pkg/routing/selector/cpuload_test.go index 186472dda..c8afd5bcc 100644 --- a/pkg/routing/selector/cpuload_test.go +++ b/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) diff --git a/pkg/routing/selector/errors.go b/pkg/routing/selector/errors.go index ab6155aeb..9c05a269c 100644 --- a/pkg/routing/selector/errors.go +++ b/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") ) diff --git a/pkg/routing/selector/interfaces.go b/pkg/routing/selector/interfaces.go index b678e44d2..aee027564 100644 --- a/pkg/routing/selector/interfaces.go +++ b/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 } diff --git a/pkg/routing/selector/random.go b/pkg/routing/selector/random.go deleted file mode 100644 index 757aae6d7..000000000 --- a/pkg/routing/selector/random.go +++ /dev/null @@ -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 -} diff --git a/pkg/routing/selector/regionaware.go b/pkg/routing/selector/regionaware.go index 9930f6af3..b8ef420bc 100644 --- a/pkg/routing/selector/regionaware.go +++ b/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 diff --git a/pkg/routing/selector/regionaware_test.go b/pkg/routing/selector/regionaware_test.go index 1596f9f79..1645c4526 100644 --- a/pkg/routing/selector/regionaware_test.go +++ b/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) diff --git a/pkg/routing/selector/sortby_test.go b/pkg/routing/selector/sortby_test.go new file mode 100644 index 000000000..1e391c650 --- /dev/null +++ b/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) + } +} diff --git a/pkg/routing/selector/sysload.go b/pkg/routing/selector/sysload.go index 34dbea5a9..821f092ab 100644 --- a/pkg/routing/selector/sysload.go +++ b/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) } diff --git a/pkg/routing/selector/sysload_test.go b/pkg/routing/selector/sysload_test.go index 3b61ec632..1941d8e7c 100644 --- a/pkg/routing/selector/sysload_test.go +++ b/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) diff --git a/pkg/routing/selector/utils.go b/pkg/routing/selector/utils.go index 667f8049e..25ac25485 100644 --- a/pkg/routing/selector/utils.go +++ b/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 + } +}