Add load_distribution strategy (consolidate | spread)

pkg/config/service.go

@@ -79,6 +79,18 @@ const (
 	MemorySourceCgroup MemorySource = "cgroup"
 )
 
+// LoadDistribution controls how egress jobs are routed across server instances.
+type LoadDistribution string
+
+const (
+	// LoadDistributionConsolidate packs jobs onto already-busy servers, keeping idle
+	// servers free for heavier egress types (room composite, web). Default behavior.
+	LoadDistributionConsolidate LoadDistribution = "consolidate"
+	// LoadDistributionSpread routes each job to the server with the most available CPU,
+	// distributing load evenly. Recommended for track-only deployments.
+	LoadDistributionSpread LoadDistribution = "spread"
+)
+
 type CPUCostConfig struct {
 	MaxCpuUtilization         float64 `yaml:"max_cpu_utilization"` // maximum allowed CPU utilization when deciding to accept a request. Default to 80%
 	MaxMemory                 float64 `yaml:"max_memory"`          // maximum allowed memory usage in GB. 0 to disable
@@ -95,6 +107,11 @@ type CPUCostConfig struct {
 	// Memory source configuration (cgroup-aware memory accounting)
 	MemorySource       MemorySource `yaml:"memory_source"`         // memory measurement source: proc_rss, cgroup
 	MemoryKillGraceSec int          `yaml:"memory_kill_grace_sec"` // grace period in update cycles before kill (0 = immediate)
+
+	// Load distribution strategy across egress server instances.
+	// "consolidate" (default): packs jobs onto busy servers, keeps idle servers free for heavier egress types.
+	// "spread": routes to the server with the most available CPU. Recommended for track_only deployments.
+	LoadDistribution LoadDistribution `yaml:"load_distribution"`
 }
 
 func NewServiceConfig(confString string) (*ServiceConfig, error) {
@@ -196,6 +213,10 @@ func (c *ServiceConfig) InitDefaults() {
 		c.MaxUploadQueue = maxUploadQueue
 	}
 
+	if c.LoadDistribution == "" {
+		c.LoadDistribution = LoadDistributionConsolidate
+	}
+
 	applyLatencyDefaults(&c.Latency)
 
 	if c.AudioTempoController.Enabled {

pkg/server/server_rpc.go

@@ -208,17 +208,20 @@ func (s *Server) processEnded(req *rpc.StartEgressRequest, info *livekit.EgressI
 
 func (s *Server) StartEgressAffinity(_ context.Context, req *rpc.StartEgressRequest) float32 {
 	if s.IsDisabled() || !s.monitor.CanAcceptRequest(req) {
-		// cannot accept
 		return -1
 	}
 
+	if s.conf.LoadDistribution == config.LoadDistributionSpread {
+		// least-loaded: return available CPU ratio so the server with the most
+		// headroom wins each selection round, distributing load evenly.
+		return s.monitor.AvailableCPURatio()
+	}
+
+	// consolidate (default): pack jobs onto already-busy servers to keep idle
+	// servers free for heavier egress types (room composite, web).
 	if s.activeRequests.Load() == 0 {
-		// group multiple track and track composite requests.
-		// if this instance is idle and another is already handling some, the request will go to that server.
-		// this avoids having many instances with one track request each, taking availability from room composite.
 		return 0.5
 	}
-	// already handling a request and has available cpu
 	return 1
 }
 

pkg/stats/monitor.go

@@ -573,6 +573,26 @@ func (m *Monitor) GetAvailableCPU() float64 {
 	return available
 }
 
+// AvailableCPURatio returns the fraction of total CPU that is currently available,
+// clamped to [0, 1]. Used by the least-loaded affinity scorer.
+func (m *Monitor) AvailableCPURatio() float32 {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	total, available, _, _ := m.getCPUUsageLocked()
+	if total == 0 {
+		return 0
+	}
+	ratio := float32(available / total)
+	if ratio < 0 {
+		return 0
+	}
+	if ratio > 1 {
+		return 1
+	}
+	return ratio
+}
+
 func (m *Monitor) getCPUUsageLocked() (total, available, pending, used float64) {
 	total = m.cpuStats.NumCPU()
 	if m.requests.Load() == 0 {