Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
6 changes: 4 additions & 2 deletions pkg/ingester/active_queried_series.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
28 changes: 28 additions & 0 deletions pkg/ingester/active_queried_series_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package ingester
import (
"context"
"fmt"
"runtime"
"sync"
"testing"
"time"
Expand Down Expand Up @@ -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)
})
}
}