From b65c9c869dec403ffc4dfe85562bbd8c784a75ae Mon Sep 17 00:00:00 2001
From: Madhavi Yedla <madhavi@codetantra.in>
Date: Sat, 6 Jun 2026 12:12:22 +0530
Subject: [PATCH] config: add load_distribution strategy (consolidate | spread)

Adds a `load_distribution` service config that controls how the affinity
scorer routes egress jobs across server instances:

- "consolidate" (default, unchanged behavior): packs jobs onto already
  busy servers, keeping idle servers free for heavier egress types
  (room composite, web).
- "spread": routes each job to the server with the most available CPU,
  distributing load evenly. Recommended for track-only deployments.

The spread strategy uses a new Monitor.AvailableCPURatio() as the
affinity score, so the server with the most headroom wins each
selection round. Defaults to "consolidate" when unset.

(cherry picked from commit f2eee2f1b1160a05d268266e08df05c9178c6d77)
---
 pkg/config/service.go    | 21 +++++++++++++++++++++
 pkg/server/server_rpc.go | 13 ++++++++-----
 pkg/stats/monitor.go     | 20 ++++++++++++++++++++
 3 files changed, 49 insertions(+), 5 deletions(-)

diff --git a/pkg/config/service.go b/pkg/config/service.go
index 174c4adaa..8a3f07b99 100644
--- a/pkg/config/service.go
+++ b/pkg/config/service.go
@@ -81,6 +81,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
@@ -98,6 +110,11 @@ type CPUCostConfig struct {
 	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)
 	CpuKillGraceSec    int          `yaml:"cpu_kill_grace_sec"`    // seconds to wait for a graceful EOS drain after sustained high CPU before hard kill (0 = use default)
+
+	// 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) {
@@ -202,6 +219,10 @@ func (c *ServiceConfig) InitDefaults() {
 		c.MaxUploadQueue = maxUploadQueue
 	}
 
+	if c.LoadDistribution == "" {
+		c.LoadDistribution = LoadDistributionConsolidate
+	}
+
 	applyLatencyDefaults(&c.Latency)
 
 	if c.AudioTempoController.Enabled {
diff --git a/pkg/server/server_rpc.go b/pkg/server/server_rpc.go
index 3910dde2f..ed4d46e83 100644
--- a/pkg/server/server_rpc.go
+++ b/pkg/server/server_rpc.go
@@ -207,17 +207,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
 }
 
diff --git a/pkg/stats/monitor.go b/pkg/stats/monitor.go
index c8d4af432..b33346e2e 100644
--- a/pkg/stats/monitor.go
+++ b/pkg/stats/monitor.go
@@ -604,6 +604,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 {