manager: limit concurrent housekeeping cgroupfs reads with GOMAXPROCS semaphore

cmd/cadvisor.go

@@ -190,19 +190,17 @@ func main() {
 
 func setMaxProcs() {
 	// TODO(vmarmol): Consider limiting if we have a CPU mask in effect.
-	// Allow as many threads as we have cores unless the user specified a value.
-	var numProcs int
-	if *maxProcs < 1 {
-		numProcs = runtime.NumCPU()
-	} else {
-		numProcs = *maxProcs
-	}
-	runtime.GOMAXPROCS(numProcs)
-
-	// Check if the setting was successful.
-	actualNumProcs := runtime.GOMAXPROCS(0)
-	if actualNumProcs != numProcs {
-		klog.Warningf("Specified max procs of %v but using %v", numProcs, actualNumProcs)
+	// Only override GOMAXPROCS when --max_procs is explicitly set.
+	// Otherwise, honour the Go runtime default which respects the
+	// GOMAXPROCS env var (or falls back to NumCPU).
+	if *maxProcs >= 1 {
+		runtime.GOMAXPROCS(*maxProcs)
+
+		// Check if the setting was successful.
+		actualNumProcs := runtime.GOMAXPROCS(0)
+		if actualNumProcs != *maxProcs {
+			klog.Warningf("Specified max procs of %v but using %v", *maxProcs, actualNumProcs)
+		}
 	}
 }
 

lib/manager/container.go

@@ -23,6 +23,7 @@ import (
 	"math/rand"
 	"os"
 	"path"
+	"runtime"
 	"sort"
 	"strconv"
 	"sync"
@@ -40,6 +41,26 @@ import (
 
 const jitterDefault = 1.0
 
+// updateStatsSem limits the number of concurrent updateStats() calls
+// (cgroupfs reads) to GOMAXPROCS to avoid kernel lock contention on
+// cgroup_rstat_lock when thousands of housekeeping goroutines flush
+// cgroup stats simultaneously.
+//
+// Initialized lazily via initUpdateStatsSem() (called from manager.New)
+// so that it picks up the --max_procs flag value: setMaxProcs() calls
+// runtime.GOMAXPROCS(numProcs) before New(), so runtime.GOMAXPROCS(0)
+// here returns the flag-adjusted value.
+var updateStatsSem chan struct{}
+var updateStatsSemOnce sync.Once
+
+func initUpdateStatsSem() {
+	updateStatsSemOnce.Do(func() {
+		if updateStatsSem == nil {
+			updateStatsSem = make(chan struct{}, runtime.GOMAXPROCS(0))
+		}
+	})
+}
+
 // Housekeeping interval.
 // The netlink cpu-load reader lives in the root binary's utils/cpuload and is
 // wired in via CpuLoadReaderFactory (lib/manager/plugins.go); the kubelet
@@ -351,6 +372,8 @@ func (cd *containerData) housekeepingTick(timer <-chan time.Time, longHousekeepi
 		defer close(finishedChan)
 	case <-timer:
 	}
+	updateStatsSem <- struct{}{}
+	defer func() { <-updateStatsSem }()
 	start := cd.clock.Now()
 	err := cd.updateStats()
 	if err != nil {

lib/manager/container_test.go

@@ -20,8 +20,10 @@ package manager
 
 import (
 	"fmt"
+	"os"
 	"reflect"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -32,11 +34,17 @@ import (
 	itest "github.com/google/cadvisor/lib/model/test"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
 	clock "k8s.io/utils/clock/testing"
 )
 
+func TestMain(m *testing.M) {
+	initUpdateStatsSem()
+	os.Exit(m.Run())
+}
+
 const (
 	containerName        = "/container"
 	testLongHousekeeping = time.Second
@@ -332,6 +340,54 @@ func TestOnDemandHousekeepingRace(t *testing.T) {
 	wg.Wait()
 }
 
+func TestUpdateStatsSemaphore(t *testing.T) {
+	// Save and restore the original semaphore.
+	origSem := updateStatsSem
+	defer func() { updateStatsSem = origSem }()
+
+	const semSize = 1
+	updateStatsSem = make(chan struct{}, semSize)
+
+	const numContainers = 10
+	var maxConcurrent atomic.Int32
+	var curConcurrent atomic.Int32
+
+	statsList := itest.GenerateRandomStats(1, 4, 1*time.Second)
+	stats := statsList[0]
+
+	// Create multiple containers whose GetStats tracks concurrency.
+	var wg sync.WaitGroup
+	for i := 0; i < numContainers; i++ {
+		cd, mockHandler, _, fakeClock := newTestContainerData(t)
+		mockHandler.On("GetStats").Return(stats, nil).Run(func(_ mock.Arguments) {
+			c := curConcurrent.Add(1)
+			for {
+				old := maxConcurrent.Load()
+				if c <= old || maxConcurrent.CompareAndSwap(old, c) {
+					break
+				}
+			}
+			time.Sleep(5 * time.Millisecond)
+			curConcurrent.Add(-1)
+		})
+
+		// Trigger via onDemandChan so housekeepingTick proceeds without
+		// needing the fake clock to be advanced.
+		go func() { cd.OnDemandHousekeeping(0) }()
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			cd.housekeepingTick(fakeClock.NewTimer(time.Minute).C(), testLongHousekeeping)
+		}()
+	}
+
+	wg.Wait()
+
+	assert.LessOrEqual(t, maxConcurrent.Load(), int32(semSize),
+		"concurrent updateStats() calls should not exceed semaphore capacity")
+}
+
 func TestNextHousekeepingInterval(t *testing.T) {
 	base := 1 * time.Second
 	tests := []struct {

lib/manager/manager.go

@@ -166,6 +166,10 @@ type HousekeepingConfig = struct {
 
 // New takes a memory storage and returns a new manager.
 func New(memoryCache *memory.InMemoryCache, sysfs sysfs.SysFs, HousekeepingConfig HousekeepingConfig, includedMetricsSet container.MetricSet, rawContainerCgroupPathPrefixWhiteList, containerEnvMetadataWhiteList []string, perfEventsFile string, resctrlInterval time.Duration) (Manager, error) {
+	// Initialize the housekeeping semaphore now that GOMAXPROCS reflects
+	// the --max_procs flag (setMaxProcs runs before New is called).
+	initUpdateStatsSem()
+
 	if memoryCache == nil {
 		return nil, fmt.Errorf("manager requires memory storage")
 	}