From a8adb6d7bc9be01c25688cd41777a0a84dad9230 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan-Otto=20Kr=C3=B6pke?= Date: Sun, 11 Oct 2026 13:44:36 +0200 Subject: [PATCH] service: reuse metrics of unchanged services between scrapes (#2675) Co-authored-by: Claude Signed-off-by: Claude --- .../scheduled_task/scheduled_task.go | 14 +- internal/collector/service/service.go | 124 +++++++++++++----- internal/metriccache/metriccache.go | 59 +++++++-- internal/metriccache/metriccache_test.go | 81 ++++++++++++ 4 files changed, 229 insertions(+), 49 deletions(-) diff --git a/internal/collector/scheduled_task/scheduled_task.go b/internal/collector/scheduled_task/scheduled_task.go index ada154c9..4c7688cb 100644 --- a/internal/collector/scheduled_task/scheduled_task.go +++ b/internal/collector/scheduled_task/scheduled_task.go @@ -179,9 +179,6 @@ func (c *Collector) Close() error { func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { c.logger = logger.With(slog.String("collector", Name)) - // Cached metrics refer to the descriptors of an earlier Build. - c.metricCache.Reset() - c.lastResult = prometheus.NewDesc( prometheus.BuildFQName(types.Namespace, Name, "last_result"), "DEPRECATED: use windows_scheduled_task_last_result_status. "+ @@ -211,6 +208,9 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { nil, ) + // Cached metrics refer to the descriptors of an earlier Build. + c.metricCache.Reset() + return nil } @@ -258,11 +258,9 @@ func (c *Collector) collectMetrics(ch chan<- prometheus.Metric, scheduledTasks S missedRunsCount: task.MissedRunsCount, } - metrics, ok := scrape.Load(task.Path, values) - if !ok { - metrics = c.taskMetrics(task) - scrape.Store(task.Path, values, metrics) - } + metrics := scrape.Get(task.Path, values, func() []prometheus.Metric { + return c.taskMetrics(task) + }) for _, metric := range metrics { ch <- metric diff --git a/internal/collector/service/service.go b/internal/collector/service/service.go index 8a2b47c1..22f0bdcf 100644 --- a/internal/collector/service/service.go +++ b/internal/collector/service/service.go @@ -32,6 +32,7 @@ import ( "unsafe" "github.com/alecthomas/kingpin/v2" + "github.com/prometheus-community/windows_exporter/internal/metriccache" "github.com/prometheus-community/windows_exporter/internal/mi" "github.com/prometheus-community/windows_exporter/internal/types" "github.com/prometheus-community/windows_exporter/internal/utils/recovery" @@ -80,6 +81,25 @@ type Collector struct { // allStartModesIncluded is true if ServiceStartModeInclude contains every start mode. // Only then services with an unknown start mode are collected. allStartModesIncluded bool + + // metricCache reuses the metrics of services whose values did not change since + // the last scrape. Each service publishes up to 14 series, and building them is + // most of the allocations of a scrape. The values are still queried on every + // scrape, so it saves allocations, not system calls. + metricCache metriccache.Cache[string, serviceValues] +} + +// serviceValues holds the inputs of the metrics of a service besides its name. +type serviceValues struct { + state uint32 + startMode string + displayName string + runAs string + pathName string + processID uint32 + // processStartTime is the creation time of the process in nanoseconds, + // or 0 if it is unknown and the process metric is not published. + processStartTime uint64 } // serviceEntry is a copy of an [windows.ENUM_SERVICE_STATUS_PROCESS], which @@ -197,6 +217,9 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { nil, ) + // Cached metrics refer to the descriptors of an earlier Build. + c.metricCache.Reset() + c.apiStateValues = map[uint32]string{ windows.SERVICE_CONTINUE_PENDING: "continue pending", windows.SERVICE_PAUSE_PENDING: "pause pending", @@ -245,6 +268,8 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { } func (c *Collector) Close() error { + c.metricCache.Reset() + if c.serviceManagerHandle == nil { return nil } @@ -268,6 +293,8 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error return fmt.Errorf("failed to query all services: %w", err) } + scrape := c.metricCache.Begin() + servicesCh := make(chan serviceEntry, len(services)) for _, service := range services { @@ -282,22 +309,27 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error for range 4 { g.Go(func() error { for service := range servicesCh { - c.collectWorker(ch, service) + c.collectWorker(ch, scrape, service) } return nil }) } - return g.Wait() + err = g.Wait() + + // Services that were not collected in this scrape are dropped from the cache. + scrape.Commit() + + return err } -func (c *Collector) collectWorker(ch chan<- prometheus.Metric, service serviceEntry) { +func (c *Collector) collectWorker(ch chan<- prometheus.Metric, scrape *metriccache.Scrape[string, serviceValues], service serviceEntry) { if c.config.ServiceExclude.MatchString(service.name) || !c.config.ServiceInclude.MatchString(service.name) { return } - if err := c.collectService(ch, service); err != nil { + if err := c.collectService(ch, scrape, service); err != nil { c.logger.Log(context.Background(), slog.LevelWarn, "failed collecting service info", slog.Any("err", err), slog.String("service", service.name), @@ -305,7 +337,7 @@ func (c *Collector) collectWorker(ch chan<- prometheus.Metric, service serviceEn } } -func (c *Collector) collectService(ch chan<- prometheus.Metric, service serviceEntry) error { +func (c *Collector) collectService(ch chan<- prometheus.Metric, scrape *metriccache.Scrape[string, serviceValues], service serviceEntry) error { serviceName := service.name serviceNamePtr, err := windows.UTF16PtrFromString(serviceName) @@ -379,70 +411,100 @@ func (c *Collector) collectService(ch chan<- prometheus.Metric, service serviceE return nil } - ch <- prometheus.MustNewConstMetric( + values := serviceValues{ + state: service.status.CurrentState, + startMode: serviceStartMode, + displayName: serviceConfig.DisplayName, + runAs: serviceConfig.ServiceStartName, + pathName: serviceConfig.BinaryPathName, + processID: service.status.ProcessId, + } + + if values.processID != 0 { + values.processStartTime = c.processStartTime(serviceName, values.processID) + } + + metrics := scrape.Get(serviceName, values, func() []prometheus.Metric { + return c.serviceMetrics(serviceName, values) + }) + + for _, metric := range metrics { + ch <- metric + } + + return nil +} + +// serviceMetrics builds the metrics of a service. +func (c *Collector) serviceMetrics(serviceName string, values serviceValues) []prometheus.Metric { + metrics := make([]prometheus.Metric, 0, 2+len(c.apiStartModeValues)+len(c.apiStateValues)) + + metrics = append(metrics, prometheus.MustNewConstMetric( c.info, prometheus.GaugeValue, 1.0, serviceName, - serviceConfig.DisplayName, - serviceConfig.ServiceStartName, - serviceConfig.BinaryPathName, - ) + values.displayName, + values.runAs, + values.pathName, + )) var ( isCurrentStartMode float64 isCurrentState float64 ) - if serviceStartMode != "" { + if values.startMode != "" { for _, startMode := range c.apiStartModeValues { isCurrentStartMode = 0.0 - if startMode == serviceStartMode { + if startMode == values.startMode { isCurrentStartMode = 1.0 } - ch <- prometheus.MustNewConstMetric( + metrics = append(metrics, prometheus.MustNewConstMetric( c.startMode, prometheus.GaugeValue, isCurrentStartMode, serviceName, startMode, - ) + )) } } for state, stateValue := range c.apiStateValues { isCurrentState = 0.0 - if state == service.status.CurrentState { + if state == values.state { isCurrentState = 1.0 } - ch <- prometheus.MustNewConstMetric( + metrics = append(metrics, prometheus.MustNewConstMetric( c.state, prometheus.GaugeValue, isCurrentState, serviceName, stateValue, - ) + )) } - if service.status.ProcessId == 0 { - return nil + if values.processStartTime == 0 { + return metrics } - processID := strconv.FormatUint(uint64(service.status.ProcessId), 10) + return append(metrics, prometheus.MustNewConstMetric( + c.processID, + prometheus.GaugeValue, + float64(values.processStartTime/1_000_000_000), + serviceName, + strconv.FormatUint(uint64(values.processID), 10), + )) +} - processStartTime, err := c.getProcessStartTime(service.status.ProcessId) +// processStartTime returns the creation time of the process of a service in +// nanoseconds, or 0 if it can't be determined. +func (c *Collector) processStartTime(serviceName string, processID uint32) uint64 { + processStartTime, err := c.getProcessStartTime(processID) if err == nil { - ch <- prometheus.MustNewConstMetric( - c.processID, - prometheus.GaugeValue, - float64(processStartTime/1_000_000_000), - serviceName, - processID, - ) - - return nil + return processStartTime } logLevel := slog.LevelWarn @@ -458,7 +520,7 @@ func (c *Collector) collectService(ch chan<- prometheus.Metric, service serviceE slog.Any("err", err), ) - return nil + return 0 } // queryAllServices returns all service states of the current Windows system diff --git a/internal/metriccache/metriccache.go b/internal/metriccache/metriccache.go index d7142548..77a1334d 100644 --- a/internal/metriccache/metriccache.go +++ b/internal/metriccache/metriccache.go @@ -47,22 +47,28 @@ type entry[V comparable] struct { metrics []prometheus.Metric } -// Scrape collects the entries of one scrape. It must not be shared between -// goroutines. +// Scrape collects the entries of one scrape. Its methods may be called by the +// concurrent workers of one scrape. type Scrape[K, V comparable] struct { cache *Cache[K, V] resets uint64 - prev map[K]entry[V] - next map[K]entry[V] + // prev is the last committed generation. It is never modified, so it is + // read without holding mu. + prev map[K]entry[V] + + mu sync.Mutex + // next collects the entries of this scrape. Commit hands it to the cache + // and sets it to nil, so later calls can't modify a committed generation. + next map[K]entry[V] } // Begin starts a scrape based on the last committed generation. -func (c *Cache[K, V]) Begin() Scrape[K, V] { +func (c *Cache[K, V]) Begin() *Scrape[K, V] { c.mu.Lock() prev, resets := c.entries, c.resets c.mu.Unlock() - return Scrape[K, V]{ + return &Scrape[K, V]{ cache: c, resets: resets, prev: prev, @@ -88,7 +94,11 @@ func (s *Scrape[K, V]) Load(key K, value V) ([]prometheus.Metric, bool) { return nil, false } - s.next[key] = e + s.mu.Lock() + if s.next != nil { + s.next[key] = e + } + s.mu.Unlock() return e.metrics, true } @@ -96,17 +106,46 @@ func (s *Scrape[K, V]) Load(key K, value V) ([]prometheus.Metric, bool) { // Store keeps metrics built from value for key. The caller must not modify // metrics afterwards. func (s *Scrape[K, V]) Store(key K, value V, metrics []prometheus.Metric) { - s.next[key] = entry[V]{value: value, metrics: metrics} + s.mu.Lock() + if s.next != nil { + s.next[key] = entry[V]{value: value, metrics: metrics} + } + s.mu.Unlock() +} + +// Get returns the metrics of the last scrape for key if they were built from +// an equal value. Otherwise it calls build and stores its result. The caller +// must not modify the returned slice. +func (s *Scrape[K, V]) Get(key K, value V, build func() []prometheus.Metric) []prometheus.Metric { + if metrics, ok := s.Load(key, value); ok { + return metrics + } + + metrics := build() + s.Store(key, value, metrics) + + return metrics } // Commit replaces the cached entries with those loaded or stored in this // scrape, so keys that were not seen are dropped. It does nothing if the cache // was reset after the scrape began, for example by a Close or Build that ran -// while a timed-out scrape was still collecting. +// while a timed-out scrape was still collecting. Calls on the scrape after +// Commit don't change the cache. func (s *Scrape[K, V]) Commit() { + s.mu.Lock() + next := s.next + s.next = nil + s.mu.Unlock() + + if next == nil { + // Committed before. + return + } + s.cache.mu.Lock() if s.cache.resets == s.resets { - s.cache.entries = s.next + s.cache.entries = next } s.cache.mu.Unlock() } diff --git a/internal/metriccache/metriccache_test.go b/internal/metriccache/metriccache_test.go index 83207407..baf2c0ed 100644 --- a/internal/metriccache/metriccache_test.go +++ b/internal/metriccache/metriccache_test.go @@ -140,6 +140,87 @@ func TestCacheConcurrentScrapes(t *testing.T) { wg.Wait() } +func TestScrapeGet(t *testing.T) { + t.Parallel() + + var ( + cache metriccache.Cache[string, string] + builds int + ) + + get := func(scrape *metriccache.Scrape[string, string], value string) []prometheus.Metric { + return scrape.Get("a", value, func() []prometheus.Metric { + builds++ + + return build("a", value) + }) + } + + scrape := cache.Begin() + a := get(scrape, "running") + scrape.Commit() + + scrape = cache.Begin() + require.Same(t, a[0], get(scrape, "running")[0]) + require.Equal(t, 1, builds, "unchanged value is reused") + + require.NotSame(t, a[0], get(scrape, "stopped")[0]) + require.Equal(t, 2, builds, "changed value is rebuilt") + scrape.Commit() +} + +func TestScrapeAfterCommit(t *testing.T) { + t.Parallel() + + var cache metriccache.Cache[string, string] + + scrape := cache.Begin() + scrape.Store("a", "running", build("a", "running")) + scrape.Commit() + + // A late call must not modify the committed generation. + scrape.Store("b", "stopped", build("b", "stopped")) + scrape.Commit() + + next := cache.Begin() + _, ok := next.Load("a", "running") + require.True(t, ok) + + _, ok = next.Load("b", "stopped") + require.False(t, ok, "stored after Commit") +} + +func TestScrapeConcurrentWorkers(t *testing.T) { + t.Parallel() + + var cache metriccache.Cache[string, int] + + for range 10 { + scrape := cache.Begin() + + var wg sync.WaitGroup + + // Several workers of one scrape share it, as in the service collector. + for worker := range 4 { + wg.Go(func() { + for k := range 100 { + name := strconv.Itoa(worker*100 + k) + + metrics := scrape.Get(name, k%3, func() []prometheus.Metric { + return build(name, strconv.Itoa(k%3)) + }) + + // require must not be called outside the test goroutine. + assert.Len(t, metrics, 1) + } + }) + } + + wg.Wait() + scrape.Commit() + } +} + func BenchmarkScrape(b *testing.B) { var cache metriccache.Cache[string, int]