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) + }) + } +}