service: reuse metrics of unchanged services between scrapes (#2675)

Co-authored-by: Claude <noreply@anthropic.com>
Signed-off-by: Claude <noreply@anthropic.com>

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

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

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

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]