scheduled_task: read only published task properties (#2613)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Jan-Otto Kröpke <mail@jkroepke.de>
This commit is contained in:
Jan-Otto Kröpke
2026-10-09 14:43:42 +02:00
committed by GitHub
co-authored by Claude Opus 5.5
parent 809330b07c
commit 708b383e7f
6 changed files with 311 additions and 146 deletions
@@ -0,0 +1,16 @@
# scheduled_task collector
Scrape time is spent waiting on the Task Scheduler service, not on CPU.
- `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.
- 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.
- The fake native task in `scheduled_task_native_test.go` counts reads per
vtable slot; extend those assertions when adding a property read.
- 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.
@@ -87,10 +87,9 @@ const (
SCHED_S_TASK_QUEUED TaskResult = 0x00041325
)
// ScheduledTask holds the task properties published as metrics.
type ScheduledTask struct {
Name string
Path string
Enabled bool
State TaskState
MissedRunsCount float64
LastTaskResult TaskResult
@@ -211,7 +210,7 @@ var TASK_RESULT_STATUSES = []string{
}
func (c *Collector) collect(ch chan<- prometheus.Metric) error {
scheduledTasks, err := getScheduledTasks()
scheduledTasks, err := getScheduledTasks(c.includeTask)
if errors.Is(err, errTasksSkipped) {
// Tasks and folders that can't be read are skipped, the other tasks are still collected.
c.logger.Warn("failed to read some scheduled tasks",
@@ -226,17 +225,17 @@ func (c *Collector) collect(ch chan<- prometheus.Metric) error {
return nil
}
// includeTask reports whether a task path matches include and does not match exclude.
func (c *Collector) includeTask(path string) bool {
return c.config.TaskInclude.MatchString(path) && !c.config.TaskExclude.MatchString(path)
}
func (c *Collector) collectMetrics(ch chan<- prometheus.Metric, scheduledTasks ScheduledTasks) {
for _, task := range scheduledTasks {
if c.config.TaskExclude.MatchString(task.Path) ||
!c.config.TaskInclude.MatchString(task.Path) {
continue
}
for _, state := range TASK_STATES {
var stateValue float64
if strings.ToLower(task.State.String()) == state {
if strings.EqualFold(task.State.String(), state) {
stateValue = 1.0
}
@@ -295,7 +294,8 @@ func (c *Collector) collectMetrics(ch chan<- prometheus.Metric, scheduledTasks S
// errTasksSkipped accompanies readable tasks when some tasks or folders could not be read.
var errTasksSkipped = errors.New("tasks skipped")
func getScheduledTasks() (ScheduledTasks, error) {
// getScheduledTasks returns the tasks whose slash-separated paths pass include.
func getScheduledTasks(include func(path string) bool) (ScheduledTasks, error) {
// COM initialization and every interface call stay on the same OS thread.
runtime.LockOSThread()
@@ -324,7 +324,7 @@ func getScheduledTasks() (ScheduledTasks, error) {
defer root.Release()
tasks := ScheduledTasks{}
if err := fetchTasksRecursively(root, `\`, &tasks); err != nil {
if err := fetchTasksRecursively(root, `\`, include, &tasks); err != nil {
return tasks, fmt.Errorf("%w: %w", errTasksSkipped, err)
}
@@ -332,7 +332,7 @@ func getScheduledTasks() (ScheduledTasks, error) {
}
// fetchTasksInFolder appends readable tasks and reports errors after reading the remaining tasks.
func fetchTasksInFolder(folder *taskschd.TaskFolder, scheduledTasks *ScheduledTasks) error {
func fetchTasksInFolder(folder *taskschd.TaskFolder, include func(path string) bool, scheduledTasks *ScheduledTasks) error {
tasks, err := folder.Tasks()
if err != nil {
return fmt.Errorf("get tasks: %w", err)
@@ -348,13 +348,17 @@ func fetchTasksInFolder(folder *taskschd.TaskFolder, scheduledTasks *ScheduledTa
continue
}
parsedTask, err := parseTask(task)
parsedTask, ok, err := parseTask(task, include)
if err != nil {
errs = append(errs, fmt.Errorf("parse task: %w", err))
continue
}
if !ok {
continue
}
*scheduledTasks = append(*scheduledTasks, parsedTask)
}
@@ -362,9 +366,11 @@ func fetchTasksInFolder(folder *taskschd.TaskFolder, scheduledTasks *ScheduledTa
}
// fetchTasksRecursively appends readable tasks, retaining errors from skipped folders or tasks.
func fetchTasksRecursively(folder *taskschd.TaskFolder, folderPath string, scheduledTasks *ScheduledTasks) error {
func fetchTasksRecursively(
folder *taskschd.TaskFolder, folderPath string, include func(path string) bool, scheduledTasks *ScheduledTasks,
) error {
errs := []error{}
if err := fetchTasksInFolder(folder, scheduledTasks); err != nil {
if err := fetchTasksInFolder(folder, include, scheduledTasks); err != nil {
errs = append(errs, fmt.Errorf("folder %s: %w", folderPath, err))
}
@@ -386,7 +392,7 @@ func fetchTasksRecursively(folder *taskschd.TaskFolder, folderPath string, sched
subfolderPath = path
}
if err := fetchTasksRecursively(subfolder, subfolderPath, scheduledTasks); err != nil {
if err := fetchTasksRecursively(subfolder, subfolderPath, include, scheduledTasks); err != nil {
errs = append(errs, err)
}
}
@@ -394,45 +400,46 @@ func fetchTasksRecursively(folder *taskschd.TaskFolder, folderPath string, sched
return errors.Join(errs...)
}
func parseTask(task *taskschd.RegisteredTask) (ScheduledTask, error) {
name, err := task.Name()
if err != nil {
return ScheduledTask{}, fmt.Errorf("get task name: %w", err)
}
// parseTask reads a task that passes include and reports false for filtered
// tasks. Reading a task property other than Path is a round trip to the Task
// Scheduler service, so filtered tasks are skipped before those reads, and only
// properties that collectMetrics publishes are read.
func parseTask(task *taskschd.RegisteredTask, include func(path string) bool) (ScheduledTask, bool, error) {
path, err := task.Path()
if err != nil {
return ScheduledTask{}, fmt.Errorf("get task path: %w", err)
return ScheduledTask{}, false, fmt.Errorf("get task path: %w", err)
}
enabled, err := task.Enabled()
if err != nil {
return ScheduledTask{}, fmt.Errorf("get task enabled: %w", err)
parsedTask := ScheduledTask{Path: strings.ReplaceAll(path, "\\", "/")}
if !include(parsedTask.Path) {
return ScheduledTask{}, false, nil
}
state, err := task.State()
if err != nil {
return ScheduledTask{}, fmt.Errorf("get task state: %w", err)
return ScheduledTask{}, false, fmt.Errorf("get task state: %w", err)
}
missed, err := task.NumberOfMissedRuns()
if err != nil {
return ScheduledTask{}, fmt.Errorf("get task missed runs: %w", err)
}
parsedTask.State = TaskState(state)
result, err := task.LastTaskResult()
if err != nil {
return ScheduledTask{}, fmt.Errorf("get task last result: %w", err)
return ScheduledTask{}, false, fmt.Errorf("get task last result: %w", err)
}
return ScheduledTask{
Name: name,
Path: strings.ReplaceAll(path, "\\", "/"),
Enabled: enabled,
State: TaskState(state),
MissedRunsCount: float64(missed),
LastTaskResult: TaskResult(result),
}, nil
parsedTask.LastTaskResult = TaskResult(result)
// collectMetrics omits missed runs for tasks that have not run.
if parsedTask.LastTaskResult != SCHED_S_TASK_HAS_NOT_RUN {
missed, err := task.NumberOfMissedRuns()
if err != nil {
return ScheduledTask{}, false, fmt.Errorf("get task missed runs: %w", err)
}
parsedTask.MissedRunsCount = float64(missed)
}
return parsedTask, true, nil
}
func (t TaskState) String() string {
@@ -19,7 +19,6 @@ package scheduled_task
import (
"log/slog"
"regexp"
"testing"
"github.com/prometheus/client_golang/prometheus"
@@ -189,88 +188,6 @@ func TestCollectMetrics(t *testing.T) {
}
}
func TestCollectMetricsFilters(t *testing.T) {
t.Parallel()
tasks := ScheduledTasks{
{Path: "/Included/Task", State: TASK_STATE_READY},
{Path: "/Included/Excluded", State: TASK_STATE_READY},
{Path: "/Other/Task", State: TASK_STATE_READY},
}
for _, tc := range []struct {
name string
config *Config
tasks ScheduledTasks
paths []string
}{
{name: "empty", tasks: ScheduledTasks{}, paths: []string{}},
{name: "defaults", tasks: tasks, paths: []string{"/Included/Task", "/Included/Excluded", "/Other/Task"}},
{
name: "include",
config: &Config{TaskInclude: regexp.MustCompile(`^/Included/`)},
tasks: tasks,
paths: []string{"/Included/Task", "/Included/Excluded"},
},
{
name: "exclude takes precedence",
config: &Config{
TaskInclude: regexp.MustCompile(`^/Included/`),
TaskExclude: regexp.MustCompile(`Excluded$`),
},
tasks: tasks,
paths: []string{"/Included/Task"},
},
{
name: "all excluded",
config: &Config{TaskExclude: regexp.MustCompile(`.*`)},
tasks: tasks,
paths: []string{},
},
} {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
families := gatherTaskMetrics(t, tc.config, tc.tasks)
if len(tc.paths) == 0 {
require.Empty(t, families)
return
}
require.Len(t, families, 4)
for _, family := range families {
paths := make([]string, 0, len(family.GetMetric()))
for _, metric := range family.GetMetric() {
for _, label := range metric.GetLabel() {
if label.GetName() == "task" {
paths = append(paths, label.GetValue())
}
}
}
samplesPerTask := 1
if family.GetName() == "windows_scheduled_task_state" {
samplesPerTask = 5
}
if family.GetName() == "windows_scheduled_task_last_result_status" {
samplesPerTask = 12
}
expected := make([]string, 0, samplesPerTask*len(tc.paths))
for range samplesPerTask {
expected = append(expected, tc.paths...)
}
require.ElementsMatch(t, expected, paths, "%s", family.GetName())
}
})
}
}
func gatherTaskMetrics(t *testing.T, config *Config, tasks ScheduledTasks) map[string]*dto.MetricFamily {
t.Helper()
@@ -18,7 +18,11 @@
package scheduled_task
import (
"regexp"
"slices"
"strings"
"testing"
"unsafe"
"github.com/prometheus-community/windows_exporter/internal/ole"
"github.com/prometheus-community/windows_exporter/internal/ole/taskschd"
@@ -42,7 +46,7 @@ func TestFetchTasksInFolderPartialCollection(t *testing.T) {
folder := fakeTaskFolder(collection, nil, nil)
tasks := ScheduledTasks{}
err := fetchTasksInFolder(folder, &tasks)
err := fetchTasksInFolder(folder, includeAllTasks, &tasks)
require.ErrorIs(t, err, ole.HRESULT(0x80070005))
require.Equal(t, 1, collectionsReleased)
@@ -80,7 +84,7 @@ func TestFetchTasksRecursivelyPartialFolderCollection(t *testing.T) {
root := fakeTaskFolder(emptyTasks, folders, nil)
tasks := ScheduledTasks{}
err := fetchTasksRecursively(root, `\`, &tasks)
err := fetchTasksRecursively(root, `\`, includeAllTasks, &tasks)
require.ErrorIs(t, err, ole.HRESULT(0x80070005))
require.Equal(t, []int64{1, 2, 3}, *attempted)
require.Len(t, tasks, 2)
@@ -91,6 +95,150 @@ func TestFetchTasksRecursivelyPartialFolderCollection(t *testing.T) {
require.Equal(t, 6, collectionsReleased)
}
func TestFetchTasksFilters(t *testing.T) {
for _, tc := range []struct {
name string
config *Config
paths []string
}{
{name: "defaults", paths: []string{"/Included/Task", "/Included/Excluded", "/Other/Task"}},
{
name: "include",
config: &Config{TaskInclude: regexp.MustCompile(`^/Included/`)},
paths: []string{"/Included/Task", "/Included/Excluded"},
},
{
name: "exclude takes precedence",
config: &Config{
TaskInclude: regexp.MustCompile(`^/Included/`),
TaskExclude: regexp.MustCompile(`Excluded$`),
},
paths: []string{"/Included/Task"},
},
{
name: "all excluded",
config: &Config{TaskExclude: regexp.MustCompile(`.*`)},
paths: []string{},
},
} {
t.Run(tc.name, func(t *testing.T) {
var released, collectionsReleased int
fakes := []*fakeTask{
{path: `\Included\Task`, state: int32(TASK_STATE_READY)},
{path: `\Included\Excluded`, state: int32(TASK_STATE_READY)},
{path: `\Other\Task`, state: int32(TASK_STATE_READY)},
}
items := make([]*ole.Object, 0, len(fakes))
for _, task := range fakes {
items = append(items, task.object(&released))
}
collection, _ := fakeTaskCollection(items, false, &collectionsReleased)
tasks := ScheduledTasks{}
err := fetchTasksInFolder(fakeTaskFolder(collection, nil, nil), New(tc.config).includeTask, &tasks)
require.NoError(t, err)
require.Equal(t, len(fakes), released)
// Filtered tasks are skipped before the Task Scheduler service is queried.
for _, task := range fakes {
reads := 0
if slices.Contains(tc.paths, strings.ReplaceAll(task.path, `\`, "/")) {
reads = 1
}
require.Equal(t, 1, task.reads[taskSlotPath], task.path)
require.Equal(t, reads, task.reads[taskSlotState], task.path)
require.Equal(t, reads, task.reads[taskSlotLastResult], task.path)
require.Equal(t, reads, task.reads[taskSlotMissedRuns], task.path)
}
families := gatherTaskMetrics(t, nil, tasks)
if len(tc.paths) == 0 {
require.Empty(t, families)
return
}
require.Len(t, families, 4)
for _, family := range families {
paths := make([]string, 0, len(family.GetMetric()))
for _, metric := range family.GetMetric() {
for _, label := range metric.GetLabel() {
if label.GetName() == "task" {
paths = append(paths, label.GetValue())
}
}
}
samplesPerTask := 1
if family.GetName() == "windows_scheduled_task_state" {
samplesPerTask = 5
}
if family.GetName() == "windows_scheduled_task_last_result_status" {
samplesPerTask = 12
}
expected := make([]string, 0, samplesPerTask*len(tc.paths))
for range samplesPerTask {
expected = append(expected, tc.paths...)
}
require.ElementsMatch(t, expected, paths, "%s", family.GetName())
}
})
}
}
func TestFetchTasksInFolderReadsPublishedProperties(t *testing.T) {
var released, collectionsReleased int
ran := &fakeTask{
path: `\Folder\Ran`, name: "Ran", state: int32(TASK_STATE_READY), enabled: -1, result: 1, missed: 2,
}
notRun := &fakeTask{
path: `\NotRun`, name: "NotRun", state: int32(TASK_STATE_DISABLED),
result: int32(SCHED_S_TASK_HAS_NOT_RUN), missed: 4,
}
collection, _ := fakeTaskCollection(
[]*ole.Object{ran.object(&released), notRun.object(&released)}, false, &collectionsReleased,
)
tasks := ScheduledTasks{}
require.NoError(t, fetchTasksInFolder(fakeTaskFolder(collection, nil, nil), includeAllTasks, &tasks))
require.Equal(t, ScheduledTasks{
{Path: "/Folder/Ran", State: TASK_STATE_READY, LastTaskResult: 1, MissedRunsCount: 2},
{Path: "/NotRun", State: TASK_STATE_DISABLED, LastTaskResult: SCHED_S_TASK_HAS_NOT_RUN},
}, tasks)
// Each published property is read once. Name and Enabled are not published,
// and missed runs are not published for tasks that have not run.
for _, task := range []*fakeTask{ran, notRun} {
missedRuns := 1
if task == notRun {
missedRuns = 0
}
require.Equal(t, 0, task.reads[taskSlotName], task.path)
require.Equal(t, 1, task.reads[taskSlotPath], task.path)
require.Equal(t, 1, task.reads[taskSlotState], task.path)
require.Equal(t, 0, task.reads[taskSlotEnabled], task.path)
require.Equal(t, 1, task.reads[taskSlotLastResult], task.path)
require.Equal(t, missedRuns, task.reads[taskSlotMissedRuns], task.path)
}
missedRuns := gatherTaskMetrics(t, nil, tasks)["windows_scheduled_task_missed_runs"]
require.NotNil(t, missedRuns)
require.Len(t, missedRuns.GetMetric(), 1)
require.Equal(t, "/Folder/Ran", missedRuns.GetMetric()[0].GetLabel()[0].GetValue())
require.InDelta(t, 2, missedRuns.GetMetric()[0].GetGauge().GetValue(), 0)
}
// The fake native collections use the Task Scheduler's one-based VARIANT indices.
func fakeTaskCollection(items []*ole.Object, countError bool, released *int) (*ole.Object, *[]int64) {
var methods [9]uintptr
@@ -156,6 +304,19 @@ func fakeTaskFolder(tasks, folders *ole.Object, released *int) *taskschd.TaskFol
}
func fakeScheduledTask(state int32, released *int) *ole.Object {
return (&fakeTask{state: state}).object(released)
}
// fakeTask is a native IRegisteredTask that counts the reads of each vtable slot.
type fakeTask struct {
path, name string
state int32
enabled int16
result, missed int32
reads [18]int
}
func (f *fakeTask) object(released *int) *ole.Object {
var methods [18]uintptr
methods[2] = windows.NewCallback(func(uintptr) uintptr {
@@ -163,28 +324,74 @@ func fakeScheduledTask(state int32, released *int) *ole.Object {
return 0
})
stringGetter := windows.NewCallback(func(_ uintptr, out *uintptr) uintptr {
*out = 0
methods[taskSlotName] = f.stringGetter(taskSlotName, &f.name)
methods[taskSlotPath] = f.stringGetter(taskSlotPath, &f.path)
methods[taskSlotState] = f.int32Getter(taskSlotState, &f.state)
methods[taskSlotEnabled] = windows.NewCallback(func(_ uintptr, out *int16) uintptr {
f.reads[taskSlotEnabled]++
*out = f.enabled
return 0
})
methods[7], methods[8] = stringGetter, stringGetter
methods[9] = windows.NewCallback(func(_ uintptr, out *int32) uintptr {
*out = state
return 0
})
methods[10] = windows.NewCallback(func(_ uintptr, out *int16) uintptr {
*out = 0
return 0
})
intGetter := windows.NewCallback(func(_ uintptr, out *int32) uintptr {
*out = 0
return 0
})
methods[16], methods[17] = intGetter, intGetter
methods[taskSlotLastResult] = f.int32Getter(taskSlotLastResult, &f.result)
methods[taskSlotMissedRuns] = f.int32Getter(taskSlotMissedRuns, &f.missed)
return &ole.Object{VTable: &methods[0]}
}
// stringGetter returns a caller-owned BSTR, or a null BSTR for an empty value.
func (f *fakeTask) stringGetter(slot int, value *string) uintptr {
return windows.NewCallback(func(_ uintptr, out *uintptr) uintptr {
f.reads[slot]++
*out = 0
if *value == "" {
return 0
}
str, err := windows.UTF16PtrFromString(*value)
if err != nil {
return 0x80070057 // E_INVALIDARG
}
*out, _, _ = sysAllocString.Call(uintptr(unsafe.Pointer(str)))
if *out == 0 {
return 0x8007000e // E_OUTOFMEMORY
}
return 0
})
}
func (f *fakeTask) int32Getter(slot int, value *int32) uintptr {
return windows.NewCallback(func(_ uintptr, out *int32) uintptr {
f.reads[slot]++
*out = *value
return 0
})
}
// IRegisteredTask vtable slots, including the seven IUnknown/IDispatch slots.
const (
taskSlotName = 7
taskSlotPath = 8
taskSlotState = 9
taskSlotEnabled = 10
taskSlotLastResult = 16
taskSlotMissedRuns = 17
)
//nolint:gochecknoglobals
var sysAllocString = windows.NewLazySystemDLL("oleaut32.dll").NewProc("SysAllocString")
func includeAllTasks(string) bool { return true }
// BenchmarkGetScheduledTasks measures the Task Scheduler reads without metric creation.
func BenchmarkGetScheduledTasks(b *testing.B) {
for b.Loop() {
if _, err := getScheduledTasks(includeAllTasks); err != nil {
b.Fatal(err)
}
}
}
+19 -1
View File
@@ -21,8 +21,10 @@ import (
"errors"
"fmt"
"runtime"
"strings"
"syscall"
"unicode/utf16"
"unicode/utf8"
"unsafe"
"golang.org/x/sys/windows"
@@ -234,8 +236,24 @@ func (b bstr) string() string {
}
n, _, _ := sysStringLen.Call(uintptr(unsafe.Pointer(b.ptr)))
units := unsafe.Slice(b.ptr, int(n))
return string(utf16.Decode(unsafe.Slice(b.ptr, int(n))))
for _, unit := range units {
if unit >= utf8.RuneSelf {
return string(utf16.Decode(units))
}
}
// ASCII needs no decoding, so copy it without an intermediate rune slice.
var value strings.Builder
value.Grow(len(units))
for _, unit := range units {
value.WriteByte(byte(unit))
}
return value.String()
}
// String calls a BSTR getter, copies its value, and frees the native string.
+1 -1
View File
@@ -52,7 +52,7 @@ func TestInitialize(t *testing.T) {
}
func TestBSTR(t *testing.T) {
for _, value := range []string{"", "windows_exporter", "before\x00after", "Unicode \U0001F600"} {
for _, value := range []string{"", "windows_exporter", "before\x00after", "Unicode \U0001F600", "Grüße"} {
t.Run(value, func(t *testing.T) {
str, err := newBSTR(value)
require.NoError(t, err)