scheduled_task: reuse metrics of unchanged tasks between scrapes (#2637)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Jan-Otto Kröpke <mail@jkroepke.de>

internal/collector/scheduled_task/AGENTS.md

@@ -1,11 +1,22 @@
 # scheduled_task collector
 
-Scrape time is spent waiting on the Task Scheduler service, not on CPU.
+Scrape time is spent in the Task Scheduler service. The exporter's own CPU
+time is small, but the Schedule service (`svchost`) uses about as much CPU time
+as the scrape takes. Measure that with the raw `PercentProcessorTime` of
+`Win32_PerfRawData_PerfProc_Process` for the service's PID (100 ns units),
+which an unprivileged user can read.
 
 - `IRegisteredTask` `Name` and `Path` are answered locally from the enumerated
   collection. `State`, `Enabled`, `LastTaskResult` and `NumberOfMissedRuns` each
   make a separate RPC to the Schedule service (roughly 0.1–0.3 ms per call).
   `ITaskFolder::GetTasks` and `GetFolders` add one RPC per folder each.
+  `IRegisteredTaskCollection::get_Item` is local, so enumerating with
+  `_NewEnum` and batched `IEnumVARIANT::Next` saves nothing.
+- Keep the serial walk. Reading folders with concurrent workers, each with its
+  own connection, halved the scrape time but cost the Schedule service about
+  40% more CPU time in total with four workers. That is a bad trade on small
+  VMs (#2643). Re-opening folders by path instead of walking the folder
+  objects also adds a round trip per folder.
 - Read only the properties that are published, apply the include/exclude
   filter to the path before any service read, and keep property reads
   conditional on whether the metric that uses them is emitted.
@@ -14,3 +25,8 @@ Scrape time is spent waiting on the Task Scheduler service, not on CPU.
 - To find slow calls, time each call type over a full enumeration on a live
   system. Run it as `SYSTEM` as well as a user: unprivileged accounts see fewer
   tasks.
+- Each task publishes 19 const metrics, and building them is most of the
+  allocations of a scrape. `collectMetrics` reuses the metrics of the previous
+  scrape through `internal/metriccache` while a task's path, state, last result
+  and missed runs are unchanged. Add new inputs of a task's metrics to
+  `taskValues`, or the cache returns stale series.

internal/collector/scheduled_task/scheduled_task.go

@@ -27,6 +27,7 @@ import (
 	"time"
 
 	"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/ole"
 	"github.com/prometheus-community/windows_exporter/internal/ole/taskschd"
@@ -55,6 +56,11 @@ type Collector struct {
 	lastResultStatus *prometheus.Desc
 	missedRuns       *prometheus.Desc
 	state            *prometheus.Desc
+
+	// metricCache reuses the metrics of tasks whose values did not change since
+	// the last scrape. Each task publishes 19 series, and building them is most
+	// of the allocations of a scrape.
+	metricCache metriccache.Cache[string, taskValues]
 }
 
 // TaskState ...
@@ -97,6 +103,13 @@ type ScheduledTask struct {
 
 type ScheduledTasks []ScheduledTask
 
+// taskValues holds the inputs of the metrics of a task besides its path.
+type taskValues struct {
+	state           TaskState
+	lastTaskResult  TaskResult
+	missedRunsCount float64
+}
+
 func New(config *Config) *Collector {
 	if config == nil {
 		config = &ConfigDefaults
@@ -158,12 +171,17 @@ func (c *Collector) GetName() string {
 }
 
 func (c *Collector) Close() error {
+	c.metricCache.Reset()
+
 	return nil
 }
 
 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. "+
@@ -231,64 +249,90 @@ func (c *Collector) includeTask(path string) bool {
 }
 
 func (c *Collector) collectMetrics(ch chan<- prometheus.Metric, scheduledTasks ScheduledTasks) {
+	scrape := c.metricCache.Begin()
+
 	for _, task := range scheduledTasks {
-		for _, state := range TASK_STATES {
-			var stateValue float64
-
-			if strings.EqualFold(task.State.String(), state) {
-				stateValue = 1.0
-			}
-
-			ch <- prometheus.MustNewConstMetric(
-				c.state,
-				prometheus.GaugeValue,
-				stateValue,
-				task.Path,
-				state,
-			)
+		values := taskValues{
+			state:           task.State,
+			lastTaskResult:  task.LastTaskResult,
+			missedRunsCount: task.MissedRunsCount,
 		}
 
-		resultStatus := task.LastTaskResult.String()
-
-		for _, status := range TASK_RESULT_STATUSES {
-			var statusValue float64
-
-			if resultStatus == status {
-				statusValue = 1
-			}
-
-			ch <- prometheus.MustNewConstMetric(
-				c.lastResultStatus,
-				prometheus.GaugeValue,
-				statusValue,
-				task.Path,
-				status,
-			)
+		metrics, ok := scrape.Load(task.Path, values)
+		if !ok {
+			metrics = c.taskMetrics(task)
+			scrape.Store(task.Path, values, metrics)
 		}
 
-		if task.LastTaskResult == SCHED_S_TASK_HAS_NOT_RUN {
-			continue
+		for _, metric := range metrics {
+			ch <- metric
+		}
+	}
+
+	scrape.Commit()
+}
+
+// taskMetrics builds the metrics of a task.
+func (c *Collector) taskMetrics(task ScheduledTask) []prometheus.Metric {
+	metrics := make([]prometheus.Metric, 0, len(TASK_STATES)+len(TASK_RESULT_STATUSES)+2)
+
+	for _, state := range TASK_STATES {
+		var stateValue float64
+
+		if strings.EqualFold(task.State.String(), state) {
+			stateValue = 1.0
 		}
 
-		lastResult := 0.0
-		if task.LastTaskResult == SCHED_S_SUCCESS {
-			lastResult = 1.0
+		metrics = append(metrics, prometheus.MustNewConstMetric(
+			c.state,
+			prometheus.GaugeValue,
+			stateValue,
+			task.Path,
+			state,
+		))
+	}
+
+	resultStatus := task.LastTaskResult.String()
+
+	for _, status := range TASK_RESULT_STATUSES {
+		var statusValue float64
+
+		if resultStatus == status {
+			statusValue = 1
 		}
 
-		ch <- prometheus.MustNewConstMetric(
+		metrics = append(metrics, prometheus.MustNewConstMetric(
+			c.lastResultStatus,
+			prometheus.GaugeValue,
+			statusValue,
+			task.Path,
+			status,
+		))
+	}
+
+	if task.LastTaskResult == SCHED_S_TASK_HAS_NOT_RUN {
+		return metrics
+	}
+
+	lastResult := 0.0
+	if task.LastTaskResult == SCHED_S_SUCCESS {
+		lastResult = 1.0
+	}
+
+	return append(metrics,
+		prometheus.MustNewConstMetric(
 			c.lastResult,
 			prometheus.GaugeValue,
 			lastResult,
 			task.Path,
-		)
-
-		ch <- prometheus.MustNewConstMetric(
+		),
+		prometheus.MustNewConstMetric(
 			c.missedRuns,
 			prometheus.GaugeValue,
 			task.MissedRunsCount,
 			task.Path,
-		)
-	}
+		),
+	)
 }
 
 // errTasksSkipped accompanies readable tasks when some tasks or folders could not be read.

internal/collector/scheduled_task/scheduled_task_internal_test.go

@@ -18,11 +18,14 @@
 package scheduled_task
 
 import (
+	"fmt"
 	"log/slog"
+	"strings"
 	"testing"
 
 	"github.com/prometheus/client_golang/prometheus"
 	dto "github.com/prometheus/client_model/go"
+	"github.com/prometheus/common/expfmt"
 	"github.com/stretchr/testify/require"
 )
 
@@ -220,3 +223,167 @@ func (c taskMetricCollector) Describe(ch chan<- *prometheus.Desc) {
 func (c taskMetricCollector) Collect(ch chan<- prometheus.Metric) {
 	c.collector.collectMetrics(ch, c.tasks)
 }
+
+func TestCollectMetricsCache(t *testing.T) {
+	t.Parallel()
+
+	tasks := ScheduledTasks{
+		{Path: "/Unchanged", State: TASK_STATE_READY, MissedRunsCount: 1, LastTaskResult: SCHED_S_SUCCESS},
+		{Path: "/StateChanges", State: TASK_STATE_READY, LastTaskResult: SCHED_S_SUCCESS},
+		{Path: "/ResultChanges", State: TASK_STATE_READY, LastTaskResult: SCHED_S_TASK_HAS_NOT_RUN},
+		{Path: "/MissedRunsChange", State: TASK_STATE_READY, LastTaskResult: 1},
+		{Path: "/Deleted", State: TASK_STATE_DISABLED, LastTaskResult: SCHED_S_TASK_DISABLED},
+	}
+
+	changed := ScheduledTasks{
+		tasks[0],
+		{Path: "/StateChanges", State: TASK_STATE_RUNNING, LastTaskResult: SCHED_S_SUCCESS},
+		{Path: "/ResultChanges", State: TASK_STATE_READY, LastTaskResult: SCHED_S_SUCCESS},
+		{Path: "/MissedRunsChange", State: TASK_STATE_READY, MissedRunsCount: 2, LastTaskResult: 1},
+		{Path: "/Added", State: TASK_STATE_QUEUED, LastTaskResult: SCHED_S_TASK_QUEUED},
+	}
+
+	c := New(nil)
+	require.NoError(t, c.Build(slog.New(slog.DiscardHandler), nil))
+
+	var previous map[string][]prometheus.Metric
+
+	for i, scrape := range []ScheduledTasks{tasks, tasks, changed, changed} {
+		// Collect once outside the registry, which calls Collect for Describe too.
+		collected := make(chan prometheus.Metric, len(scrape)*19)
+		c.collectMetrics(collected, scrape)
+		close(collected)
+
+		var sent []prometheus.Metric
+		for metric := range collected {
+			sent = append(sent, metric)
+		}
+
+		cached := exposition(t, func(ch chan<- prometheus.Metric) {
+			for _, metric := range sent {
+				ch <- metric
+			}
+		})
+
+		uncached := exposition(t, func(ch chan<- prometheus.Metric) {
+			for _, task := range scrape {
+				for _, metric := range c.taskMetrics(task) {
+					ch <- metric
+				}
+			}
+		})
+
+		require.Equal(t, uncached, cached, "scrape %d", i)
+
+		// Unchanged tasks send the metrics of the previous scrape again.
+		current := make(map[string][]prometheus.Metric, len(scrape))
+
+		for _, task := range scrape {
+			n := 19
+			if task.LastTaskResult == SCHED_S_TASK_HAS_NOT_RUN {
+				n = 17
+			}
+
+			current[task.Path], sent = sent[:n], sent[n:]
+
+			// Scrapes 1 and 3 repeat the previous one, scrape 2 changes all tasks but /Unchanged.
+			old, ok := previous[task.Path]
+			wantReused := ok && (i%2 == 1 || task.Path == "/Unchanged")
+			require.Equal(t, wantReused, ok && old[0] == current[task.Path][0], "scrape %d task %s", i, task.Path)
+		}
+
+		require.Empty(t, sent)
+
+		previous = current
+	}
+
+	// Deleted tasks are dropped from the cache.
+	scrape := c.metricCache.Begin()
+	_, ok := scrape.Load("/Deleted", taskValues{state: TASK_STATE_DISABLED, lastTaskResult: SCHED_S_TASK_DISABLED})
+	require.False(t, ok)
+	_, ok = scrape.Load("/Added", taskValues{state: TASK_STATE_QUEUED, lastTaskResult: SCHED_S_TASK_QUEUED})
+	require.True(t, ok)
+
+	// Build creates new descriptors, so the cache must not return metrics built from the old ones.
+	require.NoError(t, c.Build(slog.New(slog.DiscardHandler), nil))
+
+	scrape = c.metricCache.Begin()
+	_, ok = scrape.Load("/Added", taskValues{state: TASK_STATE_QUEUED, lastTaskResult: SCHED_S_TASK_QUEUED})
+	require.False(t, ok)
+}
+
+// exposition gathers the metrics sent by collect and returns them in the text format.
+func exposition(t *testing.T, collect func(ch chan<- prometheus.Metric)) string {
+	t.Helper()
+
+	registry := prometheus.NewPedanticRegistry()
+	require.NoError(t, registry.Register(collectorFunc(collect)))
+
+	families, err := registry.Gather()
+	require.NoError(t, err)
+
+	var text strings.Builder
+
+	for _, family := range families {
+		_, err := expfmt.MetricFamilyToText(&text, family)
+		require.NoError(t, err)
+	}
+
+	return text.String()
+}
+
+type collectorFunc func(ch chan<- prometheus.Metric)
+
+func (f collectorFunc) Describe(ch chan<- *prometheus.Desc) {
+	prometheus.DescribeByCollect(f, ch)
+}
+
+func (f collectorFunc) Collect(ch chan<- prometheus.Metric) {
+	f(ch)
+}
+
+// BenchmarkCollectMetrics measures building and sending the metrics of 265
+// tasks, roughly the number of tasks on a Windows 11 workstation. With a warm
+// cache the task values did not change since the last scrape; with a cold cache
+// every metric is built again.
+func BenchmarkCollectMetrics(b *testing.B) {
+	for _, warm := range []bool{true, false} {
+		b.Run(fmt.Sprintf("warm=%t", warm), func(b *testing.B) {
+			c := New(nil)
+			require.NoError(b, c.Build(slog.New(slog.DiscardHandler), nil))
+
+			tasks := benchmarkTasks(265)
+			ch := make(chan prometheus.Metric, len(tasks)*19)
+
+			b.ReportAllocs()
+
+			for b.Loop() {
+				if !warm {
+					c.metricCache.Reset()
+				}
+
+				c.collectMetrics(ch, tasks)
+
+				for len(ch) > 0 {
+					<-ch
+				}
+			}
+		})
+	}
+}
+
+func benchmarkTasks(n int) ScheduledTasks {
+	results := []TaskResult{SCHED_S_SUCCESS, SCHED_S_TASK_READY, SCHED_S_TASK_HAS_NOT_RUN, 1}
+	tasks := make(ScheduledTasks, n)
+
+	for i := range tasks {
+		tasks[i] = ScheduledTask{
+			Path:            fmt.Sprintf("/Microsoft/Windows/Folder%d/Task%d", i/4, i),
+			State:           TaskState(i % 5),
+			MissedRunsCount: float64(i % 3),
+			LastTaskResult:  results[i%len(results)],
+		}
+	}
+
+	return tasks
+}

internal/metriccache/metriccache.go

@@ -0,0 +1,112 @@
+// SPDX-License-Identifier: Apache-2.0
+//
+// Copyright The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package metriccache reuses const metrics across scrapes for objects whose
+// values rarely change.
+//
+// Building a const metric allocates its label pairs and value, which dominates
+// the allocations of collectors that publish many state-set series per object.
+// Const metrics are immutable and a registry only reads them during Gather, so
+// sending the same metric again in a later scrape is safe.
+package metriccache
+
+import (
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// Cache holds the metrics built for each key during the last committed scrape,
+// together with the input value they were built from.
+//
+// The zero value is ready to use. A Cache may be used by concurrent scrapes:
+// each scrape reads the last committed generation, which is never modified,
+// and the last scrape to commit wins.
+type Cache[K, V comparable] struct {
+	mu      sync.Mutex
+	entries map[K]entry[V]
+	// resets counts Reset calls, so that a scrape begun before a Reset
+	// doesn't commit afterwards.
+	resets uint64
+}
+
+type entry[V comparable] struct {
+	value   V
+	metrics []prometheus.Metric
+}
+
+// Scrape collects the entries of one scrape. It must not be shared between
+// goroutines.
+type Scrape[K, V comparable] struct {
+	cache  *Cache[K, V]
+	resets uint64
+	prev   map[K]entry[V]
+	next   map[K]entry[V]
+}
+
+// Begin starts a scrape based on the last committed generation.
+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]{
+		cache:  c,
+		resets: resets,
+		prev:   prev,
+		next:   make(map[K]entry[V], len(prev)),
+	}
+}
+
+// Reset drops all cached metrics, for example after their descriptors changed.
+// Scrapes begun before Reset don't commit their entries.
+func (c *Cache[K, V]) Reset() {
+	c.mu.Lock()
+	c.entries = nil
+	c.resets++
+	c.mu.Unlock()
+}
+
+// Load returns the metrics of the last scrape for key if they were built from
+// an equal value, and keeps them for the next scrape. The caller must not
+// modify the returned slice.
+func (s *Scrape[K, V]) Load(key K, value V) ([]prometheus.Metric, bool) {
+	e, ok := s.prev[key]
+	if !ok || e.value != value {
+		return nil, false
+	}
+
+	s.next[key] = e
+
+	return e.metrics, true
+}
+
+// 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}
+}
+
+// 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.
+func (s *Scrape[K, V]) Commit() {
+	s.cache.mu.Lock()
+	if s.cache.resets == s.resets {
+		s.cache.entries = s.next
+	}
+	s.cache.mu.Unlock()
+}

internal/metriccache/metriccache_test.go

@@ -0,0 +1,164 @@
+// SPDX-License-Identifier: Apache-2.0
+//
+// Copyright The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package metriccache_test
+
+import (
+	"strconv"
+	"sync"
+	"testing"
+
+	"github.com/prometheus-community/windows_exporter/internal/metriccache"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+//nolint:gochecknoglobals
+var desc = prometheus.NewDesc("test_state", "test", []string{"name", "state"}, nil)
+
+func build(name, state string) []prometheus.Metric {
+	return []prometheus.Metric{
+		prometheus.MustNewConstMetric(desc, prometheus.GaugeValue, 1, name, state),
+	}
+}
+
+func TestCache(t *testing.T) {
+	t.Parallel()
+
+	var cache metriccache.Cache[string, string]
+
+	scrape := cache.Begin()
+	_, ok := scrape.Load("a", "running")
+	require.False(t, ok, "empty cache")
+
+	a := build("a", "running")
+	scrape.Store("a", "running", a)
+	scrape.Store("b", "stopped", build("b", "stopped"))
+
+	_, ok = scrape.Load("a", "running")
+	require.False(t, ok, "entries are visible after Commit")
+
+	scrape.Commit()
+
+	scrape = cache.Begin()
+	metrics, ok := scrape.Load("a", "running")
+	require.True(t, ok)
+	require.Same(t, a[0], metrics[0])
+
+	_, ok = scrape.Load("a", "stopped")
+	require.False(t, ok, "changed value")
+
+	// b is neither loaded nor stored in this scrape.
+	scrape.Commit()
+
+	scrape = cache.Begin()
+	_, ok = scrape.Load("b", "stopped")
+	require.False(t, ok, "keys not seen in the last scrape are dropped")
+
+	metrics, ok = scrape.Load("a", "running")
+	require.True(t, ok, "loaded keys are kept")
+	require.Same(t, a[0], metrics[0])
+	scrape.Commit()
+
+	cache.Reset()
+
+	scrape = cache.Begin()
+	_, ok = scrape.Load("a", "running")
+	require.False(t, ok, "reset")
+}
+
+func TestCacheResetDuringScrape(t *testing.T) {
+	t.Parallel()
+
+	var cache metriccache.Cache[string, string]
+
+	// A scrape that is still running when the collector is closed or rebuilt
+	// must not put its metrics back into the cache.
+	scrape := cache.Begin()
+	scrape.Store("a", "running", build("a", "running"))
+	cache.Reset()
+	scrape.Commit()
+
+	scrape = cache.Begin()
+	_, ok := scrape.Load("a", "running")
+	require.False(t, ok)
+
+	// Scrapes begun after the Reset commit as usual.
+	scrape.Store("a", "running", build("a", "running"))
+	scrape.Commit()
+
+	scrape = cache.Begin()
+	_, ok = scrape.Load("a", "running")
+	require.True(t, ok)
+}
+
+func TestCacheConcurrentScrapes(t *testing.T) {
+	t.Parallel()
+
+	var (
+		cache metriccache.Cache[string, int]
+		wg    sync.WaitGroup
+	)
+
+	for i := range 8 {
+		wg.Go(func() {
+			for j := range 100 {
+				scrape := cache.Begin()
+
+				for k := range 10 {
+					name := strconv.Itoa(k)
+					value := (i + j) % 3
+
+					metrics, ok := scrape.Load(name, value)
+					if !ok {
+						metrics = build(name, strconv.Itoa(value))
+						scrape.Store(name, value, metrics)
+					}
+
+					// require must not be called outside the test goroutine.
+					assert.Len(t, metrics, 1)
+				}
+
+				scrape.Commit()
+			}
+		})
+	}
+
+	wg.Wait()
+}
+
+func BenchmarkScrape(b *testing.B) {
+	var cache metriccache.Cache[string, int]
+
+	names := make([]string, 1000)
+	for i := range names {
+		names[i] = strconv.Itoa(i)
+	}
+
+	b.ReportAllocs()
+
+	for b.Loop() {
+		scrape := cache.Begin()
+
+		for _, name := range names {
+			if _, ok := scrape.Load(name, 0); !ok {
+				scrape.Store(name, 0, build(name, "0"))
+			}
+		}
+
+		scrape.Commit()
+	}
+}