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