From 7b55600d6eac6cf56367afe0af92c652feaaa71c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan-Otto=20Kr=C3=B6pke?= Date: Sun, 25 Oct 2026 02:00:08 +0200 Subject: [PATCH] chore: reuse per-scrape buffers (#2802) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Jan-Otto Kröpke --- AGENTS.md | 3 + internal/collector/gpu/utils.go | 34 ++++++- internal/collector/gpu/utils_internal_test.go | 59 ++++++++++++ internal/httphandler/httphandler.go | 91 +++++++++++++++++-- internal/httphandler/httphandler_test.go | 43 +++++++++ internal/labelcache/labelcache.go | 48 ++++++++-- internal/labelcache/labelcache_test.go | 32 +++++++ pkg/collector/collect.go | 62 ++++++++----- pkg/collector/collect_test.go | 54 +++++++++++ pkg/collector/state.go | 25 ++++- 10 files changed, 404 insertions(+), 47 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 0c0d6818..2ab63724 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -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. diff --git a/internal/collector/gpu/utils.go b/internal/collector/gpu/utils.go index dc540eaa..b36b41f7 100644 --- a/internal/collector/gpu/utils.go +++ b/internal/collector/gpu/utils.go @@ -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 } diff --git a/internal/collector/gpu/utils_internal_test.go b/internal/collector/gpu/utils_internal_test.go index 1ea883f1..442a0eb7 100644 --- a/internal/collector/gpu/utils_internal_test.go +++ b/internal/collector/gpu/utils_internal_test.go @@ -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() +} diff --git a/internal/httphandler/httphandler.go b/internal/httphandler/httphandler.go index 9eb27e6d..347d37bd 100644 --- a/internal/httphandler/httphandler.go +++ b/internal/httphandler/httphandler.go @@ -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 diff --git a/internal/httphandler/httphandler_test.go b/internal/httphandler/httphandler_test.go index d685f881..a0ead64d 100644 --- a/internal/httphandler/httphandler_test.go +++ b/internal/httphandler/httphandler_test.go @@ -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() diff --git a/internal/labelcache/labelcache.go b/internal/labelcache/labelcache.go index 1baef6c3..ad43109d 100644 --- a/internal/labelcache/labelcache.go +++ b/internal/labelcache/labelcache.go @@ -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 diff --git a/internal/labelcache/labelcache_test.go b/internal/labelcache/labelcache_test.go index fa6dbc5b..50325eeb 100644 --- a/internal/labelcache/labelcache_test.go +++ b/internal/labelcache/labelcache_test.go @@ -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) { diff --git a/pkg/collector/collect.go b/pkg/collector/collect.go index a1d53443..a92be7e5 100644 --- a/pkg/collector/collect.go +++ b/pkg/collector/collect.go @@ -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) { diff --git a/pkg/collector/collect_test.go b/pkg/collector/collect_test.go index 43a9772e..bd791dcc 100644 --- a/pkg/collector/collect_test.go +++ b/pkg/collector/collect_test.go @@ -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() diff --git a/pkg/collector/state.go b/pkg/collector/state.go index 3ab55f72..f03adbff 100644 --- a/pkg/collector/state.go +++ b/pkg/collector/state.go @@ -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.