chore: reuse per-scrape buffers (#2802)

Signed-off-by: Jan-Otto Kröpke <mail@jkroepke.de>
This commit is contained in:
Jan-Otto Kröpke
2026-10-25 02:00:08 +02:00
committed by GitHub
parent 2cf2b245ec
commit 7b55600d6e
10 changed files with 404 additions and 47 deletions
+3
View File
@@ -176,6 +176,9 @@ exercised a live Windows role.
the original `Collection` owns collectors and MI resources. A timed-out call
may still be running: prevent overlap, drain late metrics and do not free its
resources while it runs. Honor the supplied scrape budget in blocking APIs.
The collection reuses a collector's metrics channel for every call, so
`Collect` must not send after it returned, for example from a goroutine it
doesn't wait for: the metric would show up in the next scrape.
`Build` has no budget: bound its MI probe queries with `mi.BuildQueryTimeout`.
A non-positive MI query timeout selects `mi.DefaultQueryTimeout` (5s), which
is too short for slow providers.
+29 -5
View File
@@ -92,6 +92,12 @@ func parseGPUCounterInstanceString(s string) Instance {
type instanceCache struct {
mu sync.Mutex
entries map[string]Instance
// spare is the map that commit replaced. The next scrape reuses it, so that
// the two maps alternate instead of one being allocated per scrape.
spare map[string]Instance
// active counts the scrapes between begin and commit. A replaced map is
// only kept as spare if no scrape may still read it.
active int
}
// instanceScrape parses the instance names of one scrape.
@@ -107,15 +113,24 @@ type instanceScrape struct {
func (c *instanceCache) begin() *instanceScrape {
c.mu.Lock()
prev := c.entries
prev, next := c.entries, c.spare
c.spare = nil
c.active++
c.mu.Unlock()
return &instanceScrape{cache: c, prev: prev, next: make(map[string]Instance, len(prev))}
if next == nil {
next = make(map[string]Instance, len(prev))
} else {
clear(next)
}
return &instanceScrape{cache: c, prev: prev, next: next}
}
func (c *instanceCache) reset() {
c.mu.Lock()
c.entries = nil
c.spare = nil
c.mu.Unlock()
}
@@ -154,7 +169,16 @@ func (s *instanceScrape) commit() {
return
}
s.cache.mu.Lock()
s.cache.entries = next
s.cache.mu.Unlock()
c := s.cache
c.mu.Lock()
defer c.mu.Unlock()
c.active--
if c.active == 0 {
c.spare = c.entries
}
c.entries = next
}
@@ -18,6 +18,7 @@
package gpu
import (
"reflect"
"testing"
"github.com/stretchr/testify/require"
@@ -109,3 +110,61 @@ func TestInstanceCache(t *testing.T) {
cache.reset()
require.Empty(t, cache.entries)
}
func TestInstanceCacheReusesMaps(t *testing.T) {
t.Parallel()
var cache instanceCache
name := "pid_1234_luid_0x00000000_0x00005678_phys_0_eng_0_engtype_3D"
want := parseGPUCounterInstanceString(name)
scrape := cache.begin()
scrape.parse(name)
scrape.commit()
first := cache.entries
scrape = cache.begin()
require.Equal(t, want, scrape.parse(name))
scrape.commit()
// The map of the first scrape is kept for the next one.
require.Equal(t, reflect.ValueOf(first).Pointer(), reflect.ValueOf(cache.spare).Pointer())
scrape = cache.begin()
require.Equal(t, reflect.ValueOf(first).Pointer(), reflect.ValueOf(scrape.next).Pointer())
require.Empty(t, scrape.next)
require.Equal(t, want, scrape.parse(name))
scrape.commit()
require.Equal(t, map[string]Instance{name: want}, cache.entries)
}
func TestInstanceCacheOverlappingScrapes(t *testing.T) {
t.Parallel()
var cache instanceCache
name := "luid_0x00000000_0x00005678_phys_0"
scrape := cache.begin()
scrape.parse(name)
scrape.commit()
// a still reads the map that b's commit replaces, so the map must not be reused.
a := cache.begin()
b := cache.begin()
b.parse(name)
b.commit()
require.Nil(t, cache.spare)
c := cache.begin()
require.Equal(t, parseGPUCounterInstanceString(name), a.parse(name))
require.Equal(t, map[string]Instance{name: parseGPUCounterInstanceString(name)}, a.prev)
a.commit()
c.commit()
}
+82 -9
View File
@@ -56,10 +56,37 @@ type MetricsHTTPHandler struct {
// the exporter itself.
exporterMetricsRegistry *prometheus.Registry
// registries holds idle registries for scrapes of all collectors. Since
// prometheus/client_golang#2170 (not in v1.25.0), a registry sizes the buffers of a
// Gather by the number of metrics of its previous Gather, so reusing it saves growing
// them on every scrape. Filtered scrapes use a new registry each.
registries chan *scrapeRegistry
logger *slog.Logger
timeoutMargin time.Duration
}
// maxIdleRegistries is the number of idle registries kept for reuse. More are only needed
// while scrapes overlap.
const maxIdleRegistries = 4
// scrapeRegistry is a registry that serves one scrape at a time.
type scrapeRegistry struct {
reg *prometheus.Registry
collector scrapeCollector
}
// scrapeCollector collects the metrics of the scrape that currently uses its registry.
type scrapeCollector struct {
handler *collector.Handler
}
func (c *scrapeCollector) Describe(_ chan<- *prometheus.Desc) {}
func (c *scrapeCollector) Collect(ch chan<- prometheus.Metric) {
c.handler.Collect(ch)
}
type Options struct {
DisableExporterMetrics bool
TimeoutMargin float64
@@ -75,6 +102,7 @@ func New(logger *slog.Logger, metricCollectors *collector.Collection, options *O
handler := &MetricsHTTPHandler{
metricCollectors: metricCollectors,
registries: make(chan *scrapeRegistry, maxIdleRegistries),
logger: logger,
timeoutMargin: toDuration(options.TimeoutMargin),
}
@@ -105,7 +133,27 @@ func (c *MetricsHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// The scrape's deadline starts now, so time spent waiting for an earlier scrape counts against it.
scrapeTimeout := c.getScrapeTimeout(logger, r)
handler, err := c.handlerFactory(r.Context(), logger, scrapeTimeout, r.URL.Query()["collect[]"])
requestedCollectors := r.URL.Query()["collect[]"]
collectionHandler, err := c.metricCollectors.NewHandlerWithContext(r.Context(), scrapeTimeout, logger, requestedCollectors)
if err != nil {
err = fmt.Errorf("couldn't create collector handler: %w", err)
} else if len(requestedCollectors) == 0 {
sr := c.getRegistry()
defer c.putRegistry(sr)
sr.collector.handler = collectionHandler
c.newHTTPHandler(logger, sr.reg).ServeHTTP(w, r)
return
}
var handler http.Handler
if err == nil {
handler, err = c.handlerFactory(logger, collectionHandler)
}
if err != nil {
logger.WarnContext(r.Context(), "Couldn't create filtered metrics handler",
slog.Any("err", err),
@@ -163,19 +211,44 @@ func toDuration(seconds float64) time.Duration {
return time.Duration(seconds * float64(time.Second))
}
func (c *MetricsHTTPHandler) handlerFactory(ctx context.Context, logger *slog.Logger, scrapeTimeout time.Duration, requestedCollectors []string) (http.Handler, error) {
// getRegistry returns an idle registry for a scrape of all collectors, or a new one.
func (c *MetricsHTTPHandler) getRegistry() *scrapeRegistry {
select {
case sr := <-c.registries:
return sr
default:
}
sr := &scrapeRegistry{reg: prometheus.NewRegistry()}
sr.reg.MustRegister(version.NewCollector("windows_exporter"), &sr.collector)
return sr
}
// putRegistry keeps a registry whose scrape is done for reuse.
func (c *MetricsHTTPHandler) putRegistry(sr *scrapeRegistry) {
sr.collector.handler = nil
select {
case c.registries <- sr:
default:
}
}
// handlerFactory returns the handler of a filtered scrape, which uses a new registry.
func (c *MetricsHTTPHandler) handlerFactory(logger *slog.Logger, collectionHandler *collector.Handler) (http.Handler, error) {
reg := prometheus.NewRegistry()
reg.MustRegister(version.NewCollector("windows_exporter"))
collectionHandler, err := c.metricCollectors.NewHandlerWithContext(ctx, scrapeTimeout, logger, requestedCollectors)
if err != nil {
return nil, fmt.Errorf("couldn't create collector handler: %w", err)
}
if err := reg.Register(collectionHandler); err != nil {
return nil, fmt.Errorf("couldn't register Prometheus collector: %w", err)
}
return c.newHTTPHandler(logger, reg), nil
}
// newHTTPHandler returns the handler that serves the metrics of reg and of the exporter itself.
func (c *MetricsHTTPHandler) newHTTPHandler(logger *slog.Logger, reg *prometheus.Registry) http.Handler {
// Scrapes are serialized by the collection, so the handler doesn't limit requests in flight.
opts := promhttp.HandlerOpts{
ErrorLog: slog.NewLogLogger(logger.Handler(), slog.LevelError),
@@ -185,7 +258,7 @@ func (c *MetricsHTTPHandler) handlerFactory(ctx context.Context, logger *slog.Lo
}
if c.exporterMetricsRegistry == nil {
return canceledScrapeHandler(logger, reg, opts), nil
return canceledScrapeHandler(logger, reg, opts)
}
// Note that we have to use h.exporterMetricsRegistry here to
@@ -195,7 +268,7 @@ func (c *MetricsHTTPHandler) handlerFactory(ctx context.Context, logger *slog.Lo
return promhttp.InstrumentMetricHandler(
c.exporterMetricsRegistry,
canceledScrapeHandler(logger, mergedGatherer{first: c.exporterMetricsRegistry, second: reg}, opts),
), nil
)
}
// canceledScrapeHandler serves the metrics of gatherer like [promhttp.HandlerFor], but fails the
+43
View File
@@ -24,6 +24,8 @@ import (
"math"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"testing/synctest"
"time"
@@ -33,6 +35,7 @@ import (
"github.com/prometheus-community/windows_exporter/internal/types"
"github.com/prometheus-community/windows_exporter/pkg/collector"
"github.com/prometheus/client_golang/prometheus"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -86,6 +89,46 @@ func TestMetricsHTTPHandler(t *testing.T) {
}
}
// Scrapes of all collectors reuse registries. Repeated and overlapping scrapes,
// mixed with filtered ones, must each serve the metrics once.
func TestMetricsHTTPHandlerReusesRegistries(t *testing.T) {
t.Parallel()
collection := collector.New(collector.Map{
"first": &testCollector{name: "first"},
"second": &testCollector{name: "second"},
})
handler := httphandler.New(slog.New(slog.DiscardHandler), collection, nil)
var wg sync.WaitGroup
for i := range 20 {
wg.Go(func() {
query := ""
if i%5 == 0 {
query = "?collect[]=first"
}
response := httptest.NewRecorder()
handler.ServeHTTP(response, httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/metrics"+query, nil))
body := response.Body.String()
assert.Equal(t, http.StatusOK, response.Code)
assert.Equal(t, 1, strings.Count(body, "\nwindows_test_first 42\n"))
assert.Equal(t, 1, strings.Count(body, "\nwindows_exporter_build_info{"))
if query == "" {
assert.Equal(t, 1, strings.Count(body, "\nwindows_test_second 42\n"))
} else {
assert.NotContains(t, body, "windows_test_second")
}
})
}
wg.Wait()
}
func TestMetricsHTTPHandlerCollectorStatus(t *testing.T) {
t.Parallel()
+42 -6
View File
@@ -143,6 +143,11 @@ type Scrape struct {
committed bool
// used is set by the first MustNewConstMetric call.
used atomic.Bool
// counters and gauges hold the metrics the scrape hands out. They are
// guarded by the lock of the cache.
counters chunk[counterMetric]
gauges chunk[gaugeMetric]
}
// Begin starts a scrape. Call Commit when it is done.
@@ -343,9 +348,10 @@ func mustNewConstMetric[K comparable](s *Scrape, m *seriesMap[K], k K, desc *pro
c.mu.Lock()
if c.resets == s.resets && !s.committed {
if labels, ok := m.get(k, s.generation); ok {
metric := s.newConstMetric(desc, valueType, value, labels)
c.mu.Unlock()
return newConstMetric(desc, valueType, value, labels)
return metric
}
}
c.mu.Unlock()
@@ -368,7 +374,7 @@ func mustNewConstMetric[K comparable](s *Scrape, m *seriesMap[K], k K, desc *pro
// Another goroutine of the scrape may have added the series meanwhile.
if labels, ok := m.get(k, s.generation); ok {
return newConstMetric(desc, valueType, value, labels)
return s.newConstMetric(desc, valueType, value, labels)
}
if *m == nil {
@@ -378,7 +384,7 @@ func mustNewConstMetric[K comparable](s *Scrape, m *seriesMap[K], k K, desc *pro
labels := c.intern(out.GetLabel())
(*m)[k] = series{labels: labels, seen: s.generation}
return newConstMetric(desc, valueType, value, labels)
return s.newConstMetric(desc, valueType, value, labels)
}
func knownValueType(valueType prometheus.ValueType) bool {
@@ -390,20 +396,50 @@ func knownValueType(valueType prometheus.ValueType) bool {
}
}
// Metrics of a scrape are allocated in chunks, which grow from minChunk to
// maxChunk metrics, so a collector with few series doesn't allocate more
// than it needs. A metric that is kept keeps its chunk alive, which is why
// chunks stay small.
const (
minChunk = 8
maxChunk = 128
)
// chunk hands out the metrics of one type of a scrape.
type chunk[T any] struct {
free []T
size int
}
func (c *chunk[T]) next() *T {
if len(c.free) == 0 {
c.size = min(max(2*c.size, minChunk), maxChunk)
c.free = make([]T, c.size)
}
m := &c.free[0]
c.free = c.free[1:]
return m
}
// newConstMetric returns a const metric with shared label pairs. The value
// message lives in the metric, so Write doesn't allocate. Label pairs are
// immutable once written, like those of prometheus.NewConstMetric.
// The caller holds the lock of the cache, which guards the chunks.
//
//nolint:ireturn // Const metrics have no exported type.
func newConstMetric(desc *prometheus.Desc, valueType prometheus.ValueType, value float64, labels []*dto.LabelPair) prometheus.Metric {
func (s *Scrape) newConstMetric(desc *prometheus.Desc, valueType prometheus.ValueType, value float64, labels []*dto.LabelPair) prometheus.Metric {
switch valueType {
case prometheus.CounterValue:
m := &counterMetric{desc: desc, labels: labels, value: value}
m := s.counters.next()
*m = counterMetric{desc: desc, labels: labels, value: value}
m.counter.Value = &m.value
return m
case prometheus.GaugeValue:
m := &gaugeMetric{desc: desc, labels: labels, value: value}
m := s.gauges.next()
*m = gaugeMetric{desc: desc, labels: labels, value: value}
m.gauge.Value = &m.value
return m
+32
View File
@@ -82,6 +82,38 @@ func TestMustNewConstMetric(t *testing.T) {
}
}
// Metrics share chunks, so each must keep its own value, also across chunks
// and after the next scrape reuses the series.
func TestChunkedMetricsKeepTheirValues(t *testing.T) {
t.Parallel()
var cache labelcache.Cache
const n = 300
for scrapeNumber := range 2 {
scrape := cache.Begin()
metrics := make([]prometheus.Metric, 0, 2*n)
for i := range n {
value := float64(scrapeNumber*1000 + i)
metrics = append(metrics,
scrape.MustNewConstMetric(counterDesc, prometheus.CounterValue, value, "p", strconv.Itoa(i), "user"),
scrape.MustNewConstMetric(gaugeDesc, prometheus.GaugeValue, -value, "p", strconv.Itoa(i)),
)
}
scrape.Commit()
for i := range n {
value := float64(scrapeNumber*1000 + i)
require.InDelta(t, value, write(t, metrics[2*i]).GetCounter().GetValue(), 0)
require.InDelta(t, -value, write(t, metrics[2*i+1]).GetGauge().GetValue(), 0)
require.Equal(t, strconv.Itoa(i), write(t, metrics[2*i+1]).GetLabel()[1].GetValue())
}
}
}
// From the second scrape on, a series reuses the label pairs of the previous
// one; series that a scrape didn't report get new pairs.
func TestLabelReuse(t *testing.T) {
+39 -23
View File
@@ -176,7 +176,9 @@ func (c *Collection) collectCollector(ctx context.Context, ch chan<- prometheus.
state := c.state.collector(name)
switch state.tryCollect() {
result, bufCh, doneCh := state.tryCollect()
switch result {
case started:
case notAvailable:
logger.LogAttrs(ctx, slog.LevelDebug, "collector skipped, because it isn't initialized",
@@ -197,20 +199,12 @@ func (c *Collection) collectCollector(ctx context.Context, ch chan<- prometheus.
return pending
}
// bufCh is a buffer channel to store the metrics
// This is needed because once timeout is reached, the prometheus registry channel is closed.
bufCh := make(chan prometheus.Metric, 1000)
errCh := make(chan error, 1)
// bufCh buffers the metrics, because the prometheus registry closes ch once the scrape ends,
// while the collector may still send. The collector state reuses bufCh and doneCh for every call.
t := time.Now()
// execute the collector. The goroutine owns the collector's slot until Collect returns.
go func() {
// Release the slot before closing bufCh, so the next scrape never sees this call as running.
defer close(bufCh)
defer state.release()
errCh <- collect(name, collector, bufCh, maxScrapeDuration)
doneCh <- collect(name, collector, bufCh, maxScrapeDuration)
}()
numMetrics := 0
@@ -219,25 +213,29 @@ func (c *Collection) collectCollector(ctx context.Context, ch chan<- prometheus.
// This is the only goroutine that writes to ch, and it stops before collectCollector returns.
for {
select {
case m, ok := <-bufCh:
if ok {
ch <- m
case m := <-bufCh:
ch <- m
numMetrics++
case err := <-doneCh:
// Collect sent its metrics before its result, so the rest is buffered.
for len(bufCh) > 0 {
ch <- <-bufCh
numMetrics++
continue
}
// Release the slot only once bufCh is empty, so the next call starts with an empty buffer.
state.release()
duration := time.Since(t)
c.sendCollectorDuration(ch, name, duration)
return logCollectorResult(ctx, logger, name, <-errCh, duration, numMetrics)
return logCollectorResult(ctx, logger, name, err, duration, numMetrics)
case <-ctx.Done():
// Drain late metrics, so that Collect doesn't block on a full buffer.
go func() {
for range bufCh {
}
}()
// Drain late metrics, so that Collect doesn't block on a full buffer,
// and release the slot once the call returned and bufCh is empty.
go drainCollect(state, bufCh, doneCh)
duration := time.Since(t)
c.sendCollectorDuration(ch, name, duration)
@@ -262,6 +260,24 @@ func (c *Collection) collectCollector(ctx context.Context, ch chan<- prometheus.
}
}
// drainCollect drops the metrics of a Collect call that outlived its scrape. Once the call returned
// and the buffer is empty, it releases the slot of the collector.
func drainCollect(state *collectorState, bufCh <-chan prometheus.Metric, doneCh <-chan error) {
for {
select {
case <-bufCh:
case <-doneCh: //nolint:errcheck // The scrape already reported the call as timed out.
for len(bufCh) > 0 {
<-bufCh
}
state.release()
return
}
}
}
// collect calls Collector.Collect and turns a panic into an error.
// It begins a scrape of the collector's caches before and ends it after the call.
func collect(name string, collector Collector, ch chan<- prometheus.Metric, maxScrapeDuration time.Duration) (err error) {
+54
View File
@@ -23,6 +23,7 @@ import (
"fmt"
"log/slog"
"runtime"
"strconv"
"strings"
"sync"
"sync/atomic"
@@ -192,6 +193,59 @@ func TestCollectTimedOutCollectorDoesNotOverlap(t *testing.T) {
})
}
// The metrics buffer of a collector is reused, so the late metrics of a timed-out call must not reach
// the next scrape.
func TestCollectLateMetricsDoNotLeak(t *testing.T) {
t.Parallel()
synctest.Test(t, func(t *testing.T) {
// More metrics than the buffer holds, so the late call blocks until they are drained.
late := &countingCollector{fakeCollector: &fakeCollector{name: "late"}, metrics: 1500, firstDelay: 2 * time.Second}
collection := collector.New(collector.Map{"late": late})
handler, err := collection.NewHandler(300*time.Millisecond, slog.New(slog.DiscardHandler), nil)
require.NoError(t, err)
out := scrape(t, handler)
require.Contains(t, out, `windows_exporter_collector_timeout{collector="late"} 1`)
require.NotContains(t, out, "windows_test_late")
// The first call sends its metrics into the buffer after the scrape ended.
time.Sleep(2 * time.Second)
synctest.Wait()
for call := 2; call <= 3; call++ {
out = scrape(t, handler)
require.Contains(t, out, `windows_exporter_collector_success{collector="late"} 1`)
require.Equal(t, late.metrics, strings.Count(out, "windows_test_late{"))
require.Equal(t, late.metrics, strings.Count(out, "} "+strconv.Itoa(call)+"\n"), "scrape %d got metrics of another call", call)
}
})
}
// countingCollector sends metrics whose value is the number of the call. Only the first call waits for firstDelay.
type countingCollector struct {
*fakeCollector
metrics int
firstDelay time.Duration
}
func (c *countingCollector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error {
call := c.calls.Add(1)
if call == 1 {
time.Sleep(c.firstDelay)
}
desc := prometheus.NewDesc("windows_test_"+c.name, "Test metric", []string{"i"}, nil)
for i := range c.metrics {
ch <- prometheus.MustNewConstMetric(desc, prometheus.GaugeValue, float64(call), strconv.Itoa(i))
}
return nil
}
func TestCollectAdmission(t *testing.T) {
t.Parallel()
+21 -4
View File
@@ -27,6 +27,7 @@ import (
"github.com/prometheus-community/windows_exporter/internal/mi"
"github.com/prometheus-community/windows_exporter/internal/scrapecache"
"github.com/prometheus/client_golang/prometheus"
)
var (
@@ -162,23 +163,39 @@ type collectorState struct {
closed bool
// busyLogged is true once a scrape logged that it skipped the current run.
busyLogged bool
// metrics buffers the metrics of a Collect call, and done receives its result.
// Only the holder of the slot uses them, so every call reuses them.
// The holder releases the slot once done was received and metrics is empty.
metrics chan prometheus.Metric
done chan error
}
// metricsBufferSize is the capacity of collectorState.metrics.
const metricsBufferSize = 1000
// tryCollect acquires the slot for a Collect call without waiting.
func (s *collectorState) tryCollect() startResult {
// On success, it returns the buffers of the call.
func (s *collectorState) tryCollect() (startResult, chan prometheus.Metric, chan error) {
s.mu.Lock()
defer s.mu.Unlock()
if s.closing || s.buildErr != nil {
return notAvailable
return notAvailable, nil, nil
}
select {
case s.slot <- struct{}{}:
return started
default:
return busy
return busy, nil, nil
}
if s.metrics == nil {
s.metrics = make(chan prometheus.Metric, metricsBufferSize)
s.done = make(chan error, 1)
}
return started, s.metrics, s.done
}
// release releases the slot acquired by tryCollect.