Support CPU request override for containerized deployments (#1500)
* Support CPU request override for containerized deployments Media processing pods intentionally omit CPU limits to avoid CFS throttling on real-time media workloads. Without a limit, NumCPU() falls back to runtime.NumCPU() which returns the full node CPU count. This could inflate the admission budget — on a 64-core node with a 15-CPU request, the monitor thinks it has 64 CPUs available and accepts too many jobs. Goruntime GOMAXPROC env variable will also be set to too high value which could affect go runtime goroutine scheduling (frequent context switches). CPU requests are a scheduler-level concept with no cgroup representation, so cgroup-based approaches (automaxprocs, Go 1.25 container awareness) can't help. The Kubernetes Downward API can expose requests.cpu as an env var, which this change reads. EffectiveCPURequest() in cpu.go reads the env var once at startup and NumCPU() returns it when set, falling back to platform detection otherwise — monitorProcesses picks this up automatically. A new maxprocs package caps GOMAXPROCS down to ceil(request) via init(), never increasing it beyond the current value so explicit settings and cgroup quotas are respected.
This commit is contained in:
Milos Pesic
@@ -15,6 +15,9 @@
|
||||
package hwstats
|
||||
|
||||
import (
|
||||
"os"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/frostbyte73/core"
|
||||
@@ -27,6 +30,33 @@ import (
|
||||
// This object returns cgroup quota aware cpu stats. On other systems than Linux,
|
||||
// it falls back to full system stats
|
||||
|
||||
const CPURequestEnvVar = "LIVEKIT_CPU_REQUEST"
|
||||
|
||||
var (
|
||||
effectiveCPURequestOnce sync.Once
|
||||
effectiveCPURequest float64
|
||||
)
|
||||
|
||||
// EffectiveCPURequest returns the value of LIVEKIT_CPU_REQUEST when set and valid.
|
||||
// The result is parsed at most once; use for stats normalization and optional
|
||||
// GOMAXPROCS tuning (see utils/hwstats/maxprocs).
|
||||
func EffectiveCPURequest() float64 {
|
||||
effectiveCPURequestOnce.Do(parseCPURequestEnv)
|
||||
return effectiveCPURequest
|
||||
}
|
||||
|
||||
func parseCPURequestEnv() {
|
||||
cpuReq := os.Getenv(CPURequestEnvVar)
|
||||
if cpuReq == "" {
|
||||
return
|
||||
}
|
||||
n, err := strconv.ParseFloat(cpuReq, 64)
|
||||
if err != nil || n <= 0 {
|
||||
return
|
||||
}
|
||||
effectiveCPURequest = n
|
||||
}
|
||||
|
||||
type platformCPUMonitor interface {
|
||||
getCPUIdle() (float64, error)
|
||||
numCPU() float64
|
||||
@@ -139,6 +169,9 @@ func (c *CPUStats) GetCPUIdle() float64 {
|
||||
}
|
||||
|
||||
func (c *CPUStats) NumCPU() float64 {
|
||||
if n := EffectiveCPURequest(); n > 0 {
|
||||
return n
|
||||
}
|
||||
return c.platform.numCPU()
|
||||
}
|
||||
|
||||
@@ -186,7 +219,7 @@ func (c *CPUStats) monitorCPULoad() {
|
||||
}
|
||||
|
||||
func (c *CPUStats) monitorProcesses() {
|
||||
numCPU := c.platform.numCPU()
|
||||
numCPU := c.NumCPU()
|
||||
pageSize := getPageSize()
|
||||
|
||||
fs, err := procfs.NewFS(procfs.DefaultMountPoint)
|
||||
|
||||
145
utils/hwstats/cpu_request_test.go
Normal file
145
utils/hwstats/cpu_request_test.go
Normal file
@@ -0,0 +1,145 @@
|
||||
package hwstats
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"runtime"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/frostbyte73/core"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Mock platform monitor
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
type mockPlatform struct {
|
||||
nCPU float64
|
||||
}
|
||||
|
||||
func (m *mockPlatform) getCPUIdle() (float64, error) { return m.nCPU, nil }
|
||||
func (m *mockPlatform) numCPU() float64 { return m.nCPU }
|
||||
|
||||
func newTestCPUStats(plat platformCPUMonitor) *CPUStats {
|
||||
return &CPUStats{
|
||||
platform: plat,
|
||||
warningThrottle: core.NewThrottle(time.Minute),
|
||||
closeChan: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// NumCPU without override (no env var set in this process)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func TestNumCPU_NoOverride(t *testing.T) {
|
||||
c := newTestCPUStats(&mockPlatform{nCPU: 64})
|
||||
assert.Equal(t, 64.0, c.NumCPU())
|
||||
|
||||
c2 := newTestCPUStats(&mockPlatform{nCPU: 8.5})
|
||||
assert.Equal(t, 8.5, c2.NumCPU())
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Subprocess tests for EffectiveCPURequest and NumCPU with override
|
||||
//
|
||||
// sync.Once means we can only test one env-var value per process.
|
||||
// Each test case spawns a subprocess with the desired env.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const subprocessEnvKey = "HWSTATS_TEST_SUBPROCESS"
|
||||
|
||||
type subprocessResult struct {
|
||||
EffectiveCPU float64 `json:"effective_cpu"`
|
||||
NumCPU float64 `json:"num_cpu"`
|
||||
PlatformCPU float64 `json:"platform_cpu"`
|
||||
}
|
||||
|
||||
func TestEffectiveCPURequest(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
envValue string
|
||||
expectedCPU float64
|
||||
}{
|
||||
{"valid integer", "4", 4},
|
||||
{"valid float", "3.33", 3.33},
|
||||
{"valid large", "15", 15},
|
||||
{"valid fractional", "0.5", 0.5},
|
||||
{"empty", "", 0},
|
||||
{"negative", "-1", 0},
|
||||
{"zero", "0", 0},
|
||||
{"invalid string", "abc", 0},
|
||||
{"invalid mixed", "4abc", 0},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
result := runSubprocess(t, tt.envValue)
|
||||
assert.InDelta(t, tt.expectedCPU, result.EffectiveCPU, 0.001)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func runSubprocess(t *testing.T, cpuRequestEnv string) subprocessResult {
|
||||
t.Helper()
|
||||
|
||||
cmd := exec.Command(os.Args[0], "-test.run=TestSubprocessWorker", "-test.v")
|
||||
|
||||
// Build clean env
|
||||
var env []string
|
||||
for _, e := range os.Environ() {
|
||||
if !strings.HasPrefix(e, "LIVEKIT_CPU_REQUEST=") &&
|
||||
!strings.HasPrefix(e, subprocessEnvKey+"=") {
|
||||
env = append(env, e)
|
||||
}
|
||||
}
|
||||
env = append(env, subprocessEnvKey+"=1")
|
||||
if cpuRequestEnv != "" {
|
||||
env = append(env, "LIVEKIT_CPU_REQUEST="+cpuRequestEnv)
|
||||
}
|
||||
cmd.Env = env
|
||||
|
||||
out, err := cmd.Output()
|
||||
if err != nil {
|
||||
if exitErr, ok := err.(*exec.ExitError); ok {
|
||||
t.Fatalf("subprocess failed: %s\nstderr: %s", err, exitErr.Stderr)
|
||||
}
|
||||
t.Fatalf("subprocess failed: %s", err)
|
||||
}
|
||||
|
||||
var result subprocessResult
|
||||
for _, line := range strings.Split(string(out), "\n") {
|
||||
line = strings.TrimSpace(line)
|
||||
if strings.HasPrefix(line, "{") {
|
||||
require.NoError(t, json.Unmarshal([]byte(line), &result))
|
||||
return result
|
||||
}
|
||||
}
|
||||
t.Fatalf("no JSON result in subprocess output: %s", string(out))
|
||||
return result
|
||||
}
|
||||
|
||||
// TestSubprocessWorker runs inside spawned subprocesses only.
|
||||
func TestSubprocessWorker(t *testing.T) {
|
||||
if os.Getenv(subprocessEnvKey) != "1" {
|
||||
t.Skip("only runs as subprocess")
|
||||
}
|
||||
|
||||
plat := &mockPlatform{nCPU: float64(runtime.NumCPU())}
|
||||
c := newTestCPUStats(plat)
|
||||
|
||||
result := subprocessResult{
|
||||
EffectiveCPU: EffectiveCPURequest(),
|
||||
NumCPU: c.NumCPU(),
|
||||
PlatformCPU: plat.numCPU(),
|
||||
}
|
||||
|
||||
b, _ := json.Marshal(result)
|
||||
fmt.Println(string(b))
|
||||
}
|
||||
77
utils/hwstats/maxprocs/maxprocs.go
Normal file
77
utils/hwstats/maxprocs/maxprocs.go
Normal file
@@ -0,0 +1,77 @@
|
||||
// Copyright 2026 LiveKit, Inc.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
// Package maxprocs caps GOMAXPROCS to the Kubernetes CPU request so the Go
|
||||
// scheduler doesn't create more OS threads than the pod is guaranteed.
|
||||
//
|
||||
// # Why not CPU limits / automaxprocs
|
||||
//
|
||||
// Real-time media workloads (egress, ingress) intentionally omit CPU limits
|
||||
// to avoid CFS throttling — even brief throttle events cause audio glitches
|
||||
// and video frame drops. Without a limit the cgroup quota is "max", so
|
||||
// libraries like uber/automaxprocs fall back to runtime.NumCPU (the full
|
||||
// node). This package uses the CPU request instead, which reflects what the
|
||||
// Kubernetes scheduler actually guarantees to the pod.
|
||||
//
|
||||
// When a CPU limit IS set, the cgroup quota gives a tighter bound and Go
|
||||
// (or automaxprocs) already derives GOMAXPROCS from it. This package still
|
||||
// works in that case — it caps GOMAXPROCS down to the request but never
|
||||
// increases it, so the stricter of (limit, request) wins.
|
||||
//
|
||||
// # Usage
|
||||
//
|
||||
// The CPU request is read from the LIVEKIT_CPU_REQUEST environment variable,
|
||||
// which should be populated e.g via the Kubernetes Downward API:
|
||||
//
|
||||
// env:
|
||||
// - name: LIVEKIT_CPU_REQUEST
|
||||
// valueFrom:
|
||||
// resourceFieldRef:
|
||||
// containerName: <name>
|
||||
// resource: requests.cpu
|
||||
//
|
||||
// Without this variable the package is a no-op. A blank import is
|
||||
// sufficient — init runs at process startup:
|
||||
//
|
||||
// import _ "github.com/livekit/protocol/utils/hwstats/maxprocs"
|
||||
package maxprocs
|
||||
|
||||
import (
|
||||
"math"
|
||||
"runtime"
|
||||
"sync"
|
||||
|
||||
"github.com/livekit/protocol/utils/hwstats"
|
||||
)
|
||||
|
||||
var once sync.Once
|
||||
|
||||
func init() { Install() }
|
||||
|
||||
// Install caps GOMAXPROCS to ceil(LIVEKIT_CPU_REQUEST) when that environment
|
||||
// variable is set and valid. It never increases GOMAXPROCS beyond its current
|
||||
// value. It is a no-op when the variable is absent. Runs at most once per
|
||||
// process.
|
||||
func Install() {
|
||||
once.Do(func() {
|
||||
req := hwstats.EffectiveCPURequest()
|
||||
if req <= 0 {
|
||||
return
|
||||
}
|
||||
cap := int(math.Ceil(req))
|
||||
if runtime.GOMAXPROCS(0) > cap {
|
||||
runtime.GOMAXPROCS(cap)
|
||||
}
|
||||
})
|
||||
}
|
||||
160
utils/hwstats/maxprocs/maxprocs_test.go
Normal file
160
utils/hwstats/maxprocs/maxprocs_test.go
Normal file
@@ -0,0 +1,160 @@
|
||||
package maxprocs_test
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"runtime"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
const subprocessEnvKey = "MAXPROCS_TEST_SUBPROCESS"
|
||||
|
||||
type subprocessResult struct {
|
||||
GOMAXPROCS int `json:"gomaxprocs"`
|
||||
}
|
||||
|
||||
func TestInstall(t *testing.T) {
|
||||
hostCPU := runtime.NumCPU()
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
cpuRequest string // LIVEKIT_CPU_REQUEST value ("" = unset)
|
||||
gomaxprocsEnv string // GOMAXPROCS value ("" = unset)
|
||||
expectedMaxProcs int
|
||||
}{
|
||||
{
|
||||
name: "no env vars set",
|
||||
expectedMaxProcs: hostCPU,
|
||||
},
|
||||
{
|
||||
name: "request caps GOMAXPROCS down",
|
||||
cpuRequest: "4",
|
||||
expectedMaxProcs: 4,
|
||||
},
|
||||
{
|
||||
name: "fractional request rounds up",
|
||||
cpuRequest: "3.33",
|
||||
expectedMaxProcs: 4,
|
||||
},
|
||||
{
|
||||
name: "request of 1",
|
||||
cpuRequest: "1",
|
||||
expectedMaxProcs: 1,
|
||||
},
|
||||
{
|
||||
name: "request larger than host CPUs (no change)",
|
||||
cpuRequest: "9999",
|
||||
expectedMaxProcs: hostCPU,
|
||||
},
|
||||
{
|
||||
name: "explicit GOMAXPROCS lower than request (respected)",
|
||||
cpuRequest: "8",
|
||||
gomaxprocsEnv: "4",
|
||||
expectedMaxProcs: 4,
|
||||
},
|
||||
{
|
||||
name: "explicit GOMAXPROCS higher than request (capped)",
|
||||
cpuRequest: "4",
|
||||
gomaxprocsEnv: "16",
|
||||
expectedMaxProcs: 4,
|
||||
},
|
||||
{
|
||||
name: "explicit GOMAXPROCS equal to request",
|
||||
cpuRequest: "8",
|
||||
gomaxprocsEnv: "8",
|
||||
expectedMaxProcs: 8,
|
||||
},
|
||||
{
|
||||
name: "explicit GOMAXPROCS without request (unchanged)",
|
||||
gomaxprocsEnv: "4",
|
||||
expectedMaxProcs: 4,
|
||||
},
|
||||
{
|
||||
name: "invalid request ignored",
|
||||
cpuRequest: "abc",
|
||||
expectedMaxProcs: hostCPU,
|
||||
},
|
||||
{
|
||||
name: "negative request ignored",
|
||||
cpuRequest: "-2",
|
||||
expectedMaxProcs: hostCPU,
|
||||
},
|
||||
{
|
||||
name: "zero request ignored",
|
||||
cpuRequest: "0",
|
||||
expectedMaxProcs: hostCPU,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
result := runSubprocess(t, tt.cpuRequest, tt.gomaxprocsEnv)
|
||||
assert.Equal(t, tt.expectedMaxProcs, result.GOMAXPROCS)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func runSubprocess(t *testing.T, cpuRequest, gomaxprocs string) subprocessResult {
|
||||
t.Helper()
|
||||
|
||||
cmd := exec.Command(os.Args[0], "-test.run=TestSubprocessWorker", "-test.v")
|
||||
|
||||
// Build clean env: start from current, filter out the vars we control
|
||||
var env []string
|
||||
for _, e := range os.Environ() {
|
||||
if strings.HasPrefix(e, "LIVEKIT_CPU_REQUEST=") ||
|
||||
strings.HasPrefix(e, "GOMAXPROCS=") ||
|
||||
strings.HasPrefix(e, subprocessEnvKey+"=") {
|
||||
continue
|
||||
}
|
||||
env = append(env, e)
|
||||
}
|
||||
env = append(env, subprocessEnvKey+"=1")
|
||||
if cpuRequest != "" {
|
||||
env = append(env, "LIVEKIT_CPU_REQUEST="+cpuRequest)
|
||||
}
|
||||
if gomaxprocs != "" {
|
||||
env = append(env, "GOMAXPROCS="+gomaxprocs)
|
||||
}
|
||||
cmd.Env = env
|
||||
|
||||
out, err := cmd.Output()
|
||||
if err != nil {
|
||||
if exitErr, ok := err.(*exec.ExitError); ok {
|
||||
t.Fatalf("subprocess failed: %s\nstderr: %s", err, exitErr.Stderr)
|
||||
}
|
||||
t.Fatalf("subprocess failed: %s", err)
|
||||
}
|
||||
|
||||
var result subprocessResult
|
||||
for _, line := range strings.Split(string(out), "\n") {
|
||||
line = strings.TrimSpace(line)
|
||||
if strings.HasPrefix(line, "{") {
|
||||
require.NoError(t, json.Unmarshal([]byte(line), &result))
|
||||
return result
|
||||
}
|
||||
}
|
||||
t.Fatalf("no JSON result in subprocess output: %s", string(out))
|
||||
return result
|
||||
}
|
||||
|
||||
// TestSubprocessWorker runs inside spawned subprocesses.
|
||||
// The maxprocs init() has already fired by the time we get here.
|
||||
func TestSubprocessWorker(t *testing.T) {
|
||||
if os.Getenv(subprocessEnvKey) != "1" {
|
||||
t.Skip("only runs as subprocess")
|
||||
}
|
||||
|
||||
// init() already called Install() — just report the result
|
||||
result := subprocessResult{
|
||||
GOMAXPROCS: runtime.GOMAXPROCS(0),
|
||||
}
|
||||
b, _ := json.Marshal(result)
|
||||
fmt.Println(string(b))
|
||||
}
|
||||
Reference in New Issue
Block a user