From 18d7cc55d947f5c1acbeb5e2b46effc721b83236 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan-Otto=20Kr=C3=B6pke?= Date: Sat, 10 Oct 2026 16:43:36 +0200 Subject: [PATCH] scheduled_task: reuse metrics of unchanged tasks between scrapes (#2637) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Claude Opus 5.5 Signed-off-by: Jan-Otto Kröpke --- internal/collector/scheduled_task/AGENTS.md | 18 +- .../scheduled_task/scheduled_task.go | 126 ++++++++----- .../scheduled_task_internal_test.go | 167 ++++++++++++++++++ internal/metriccache/metriccache.go | 112 ++++++++++++ internal/metriccache/metriccache_test.go | 164 +++++++++++++++++ 5 files changed, 545 insertions(+), 42 deletions(-) create mode 100644 internal/metriccache/metriccache.go create mode 100644 internal/metriccache/metriccache_test.go diff --git a/internal/collector/scheduled_task/AGENTS.md b/internal/collector/scheduled_task/AGENTS.md index ad9b759d..2c527366 100644 --- a/internal/collector/scheduled_task/AGENTS.md +++ b/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. diff --git a/internal/collector/scheduled_task/scheduled_task.go b/internal/collector/scheduled_task/scheduled_task.go index 9a1d2063..ada154c9 100644 --- a/internal/collector/scheduled_task/scheduled_task.go +++ b/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. diff --git a/internal/collector/scheduled_task/scheduled_task_internal_test.go b/internal/collector/scheduled_task/scheduled_task_internal_test.go index 2d91ecc5..9f616db8 100644 --- a/internal/collector/scheduled_task/scheduled_task_internal_test.go +++ b/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 +} diff --git a/internal/metriccache/metriccache.go b/internal/metriccache/metriccache.go new file mode 100644 index 00000000..d7142548 --- /dev/null +++ b/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() +} diff --git a/internal/metriccache/metriccache_test.go b/internal/metriccache/metriccache_test.go new file mode 100644 index 00000000..83207407 --- /dev/null +++ b/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() + } +}