From e5c4b2b3eb248c2e66a05226e72e31ec0e69e1a1 Mon Sep 17 00:00:00 2001 From: Anas Khan <83116240+anxkhn@users.noreply.github.com> Date: Fri, 3 Jul 2026 09:50:27 +0530 Subject: [PATCH] Size active queried series worker pool from GOMAXPROCS The active queried series service sized its worker pool with runtime.NumCPU(), which returns the host's logical CPU count and ignores the CPU cgroup quota. Cortex imports go.uber.org/automaxprocs so that GOMAXPROCS matches the container's CPU limit, and reads container-aware parallelism via runtime.GOMAXPROCS(0) elsewhere (for example the resource monitor's CPU limit and the engine's decoding concurrency default). This one call site was inconsistent: in a CPU-limited container on a many-core host it sized the pool off the host cores instead of the CPU budget. Use runtime.GOMAXPROCS(0) so the pool follows the container CPU quota, matching the rest of the codebase. The min(..., 4) cap is unchanged. Add a regression test asserting the worker count follows GOMAXPROCS. Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com> --- CHANGELOG.md | 1 + pkg/ingester/active_queried_series.go | 6 +++-- pkg/ingester/active_queried_series_test.go | 28 ++++++++++++++++++++++ 3 files changed, 33 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a9c2e68ec..9ebc7e263c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -77,6 +77,7 @@ * [BUGFIX] Querier: Fix panic due to request tracker truncating multi-byte UTF-8 character #7640 * [BUGFIX] Ingester: Fix panic (`HistogramProtoToHistogram called with a float histogram`) when ingesting a float native histogram with a zero count (e.g. a staleness marker or empty histogram). The decoder is now selected by histogram type via `IsFloatHistogram()` instead of by count value. #7645 * [BUGFIX] Querier: Fix parquet queryable fallback returning a nil error instead of the actual query error in `LabelValues` and `LabelNames`. #7638 +* [BUGFIX] Ingester: Size the active queried series worker pool from `GOMAXPROCS` instead of `runtime.NumCPU()`, so it respects the container CPU quota (set via automaxprocs) rather than the host core count. #XXXX ## 1.21.0 2026-04-24 diff --git a/pkg/ingester/active_queried_series.go b/pkg/ingester/active_queried_series.go index 96134a0515..32685d0f91 100644 --- a/pkg/ingester/active_queried_series.go +++ b/pkg/ingester/active_queried_series.go @@ -385,8 +385,10 @@ type ActiveQueriedSeriesService struct { // NewActiveQueriedSeriesService creates a new ActiveQueriedSeriesService service. func NewActiveQueriedSeriesService(logger log.Logger, registerer prometheus.Registerer) *ActiveQueriedSeriesService { - // Cap at 4 workers to avoid excessive goroutines - numWorkers := max(min(runtime.NumCPU()/2, 4), 1) + // Cap at 4 workers to avoid excessive goroutines. Use GOMAXPROCS (which + // automaxprocs sets from the CPU cgroup quota) rather than NumCPU, so the + // pool is sized to the container's CPU budget instead of the host cores. + numWorkers := max(min(runtime.GOMAXPROCS(0)/2, 4), 1) m := &ActiveQueriedSeriesService{ updateChan: make(chan activeQueriedSeriesUpdate, 10000), // Buffered channel to avoid blocking diff --git a/pkg/ingester/active_queried_series_test.go b/pkg/ingester/active_queried_series_test.go index 9fdbd27c4f..c036039186 100644 --- a/pkg/ingester/active_queried_series_test.go +++ b/pkg/ingester/active_queried_series_test.go @@ -3,6 +3,7 @@ package ingester import ( "context" "fmt" + "runtime" "sync" "testing" "time" @@ -546,3 +547,30 @@ func TestActiveQueriedSeriesService_NoSendOnClosedChannelOnShutdown(t *testing.T t.Fatalf("producer goroutine panicked during shutdown: %v", panicMsg.Load()) } } + +func TestActiveQueriedSeriesService_NumWorkersFollowsGOMAXPROCS(t *testing.T) { + // The worker pool must be sized from GOMAXPROCS (which automaxprocs derives + // from the CPU cgroup quota) and not from runtime.NumCPU (host cores, which + // ignores cgroup limits). Save and restore GOMAXPROCS around the test. + defer runtime.GOMAXPROCS(runtime.GOMAXPROCS(1)) + + for _, tc := range []struct { + gomaxprocs int + expectedWorkers int + }{ + // GOMAXPROCS(1)/2 == 0, floored to the minimum of 1 worker. Against the + // buggy runtime.NumCPU() this yields min(NumCPU/2, 4) on a multi-core + // host and fails. + {gomaxprocs: 1, expectedWorkers: 1}, + {gomaxprocs: 2, expectedWorkers: 1}, + {gomaxprocs: 4, expectedWorkers: 2}, + // Capped at 4 workers regardless of how many CPUs are available. + {gomaxprocs: 16, expectedWorkers: 4}, + } { + t.Run(fmt.Sprintf("gomaxprocs=%d", tc.gomaxprocs), func(t *testing.T) { + runtime.GOMAXPROCS(tc.gomaxprocs) + svc := NewActiveQueriedSeriesService(log.NewNopLogger(), nil) + assert.Equal(t, tc.expectedWorkers, svc.numWorkers) + }) + } +}