mirror of
https://github.com/prometheus-community/windows_exporter.git
synced 2026-10-17 10:09:05 +02:00
service: cache service configurations between scrapes (#2674)
Co-authored-by: Claude <noreply@anthropic.com> Signed-off-by: Claude <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude
parent
930609b0e5
commit
e6d1f00d1f
@@ -28,6 +28,16 @@ Comma separated list of service start modes to include.
|
||||
Possible values: auto, boot, manual, disabled, system.
|
||||
Recommended to keep down number of returned metrics. Default: `auto,boot,manual,disabled,system`
|
||||
|
||||
## Configuration cache
|
||||
|
||||
The service configuration (display name, start mode, path and account, used by
|
||||
`windows_service_info` and `windows_service_start_mode`) is cached between scrapes.
|
||||
The collector watches the registry key `HKLM\SYSTEM\CurrentControlSet\Services`. When a
|
||||
service's own key changes, for example after `sc config`, a change in the Services console
|
||||
or a reinstall, its configuration is queried again on the next scrape. All configurations
|
||||
are queried again at the latest every 10 minutes. The service state and process are queried
|
||||
on every scrape.
|
||||
|
||||
## Metrics
|
||||
|
||||
| Name | Description | Type | Labels |
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
# service collector
|
||||
|
||||
- `EnumServicesStatusEx` returns state and PID of all services in one call. The
|
||||
configuration needs `OpenService`, `QueryServiceConfig` and
|
||||
`CloseServiceHandle` per service, which costs CPU in services.exe as well, so
|
||||
it is cached in [`config_cache.go`](config_cache.go). Never cache state or PID.
|
||||
- `NotifyServiceStatusChange` does not report configuration changes. The cache
|
||||
watches `HKLM\SYSTEM\CurrentControlSet\Services` with
|
||||
`RegNotifyChangeKeyValue` instead. Arm it with `REG_NOTIFY_THREAD_AGNOSTIC`:
|
||||
without it the registration ends when the arming OS thread exits, and Go
|
||||
goroutines are not bound to one.
|
||||
- The subtree notification fires every few seconds on a normal machine (W32Time,
|
||||
Tcpip, VSS, driver `Enum` keys, Hyper-V). Treat it as a hint only and compare
|
||||
the last write time of each service's own key, which changes only for its own
|
||||
values and direct subkeys.
|
||||
- Registry write times have a resolution of one clock tick, so two writes in
|
||||
the same tick leave the time unchanged. Only cache a configuration once its
|
||||
key is older than `configCacheMinKeyAge`, and never cache an unavailable one
|
||||
(for example a missing MUI display name), which can change without a write.
|
||||
- The cache tests use a temporary key below `HKEY_CURRENT_USER`, so they run
|
||||
without elevation. Registry write times have a coarse resolution; sleep before
|
||||
a write that must get a newer time.
|
||||
- `TestCollectorPicksUpConfigChanges` creates a real service and changes it
|
||||
through the SCM. It needs an elevated process: it skips locally without one
|
||||
and fails in CI, where `service` is in `WINDOWS_EXPORTER_TEST_COLLECTORS`.
|
||||
@@ -0,0 +1,322 @@
|
||||
// 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.
|
||||
|
||||
//go:build windows
|
||||
|
||||
package service
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"golang.org/x/sys/windows"
|
||||
"golang.org/x/sys/windows/registry"
|
||||
"golang.org/x/sys/windows/svc/mgr"
|
||||
)
|
||||
|
||||
const (
|
||||
// servicesKeyPath is the registry key the service control manager stores the
|
||||
// service configurations in. ChangeServiceConfig writes the values of its subkeys.
|
||||
servicesKeyPath = `SYSTEM\CurrentControlSet\Services`
|
||||
|
||||
// REG_NOTIFY_THREAD_AGNOSTIC keeps the notification alive if the OS thread
|
||||
// that armed it exits. Go may run the next arm on any thread.
|
||||
servicesKeyNotifyFilter = windows.REG_NOTIFY_CHANGE_NAME |
|
||||
windows.REG_NOTIFY_CHANGE_LAST_SET |
|
||||
windows.REG_NOTIFY_THREAD_AGNOSTIC
|
||||
|
||||
// configCacheMaxAge bounds how long a cached configuration is used, in case
|
||||
// a change notification is missed.
|
||||
configCacheMaxAge = 10 * time.Minute
|
||||
|
||||
// configCacheMinKeyAge is how long a service key must be unchanged before its
|
||||
// configuration is cached. Registry write times have a resolution of one
|
||||
// clock tick, so a second write in the same tick would not change it.
|
||||
configCacheMinKeyAge = time.Second
|
||||
)
|
||||
|
||||
var errConfigCacheClosed = errors.New("service configuration cache is closed")
|
||||
|
||||
// cachedServiceConfig is the result of queryServiceConfig.
|
||||
type cachedServiceConfig struct {
|
||||
config mgr.Config
|
||||
available bool
|
||||
|
||||
// lastWrite is the last write time of the service's registry key, read
|
||||
// before the configuration was queried.
|
||||
lastWrite time.Time
|
||||
}
|
||||
|
||||
// serviceConfigCache caches the service configurations between scrapes.
|
||||
//
|
||||
// NotifyServiceStatusChange doesn't report configuration changes, so the cache
|
||||
// watches the services registry key instead. The notification fires for any
|
||||
// write below it, and drivers and services write to deeper subkeys (Parameters,
|
||||
// Enum, State, ...) every few seconds. So a notification only triggers a check
|
||||
// of the last write time of each cached service key, which changes only when
|
||||
// the values of the key itself change or a direct subkey is created or deleted.
|
||||
// Only services whose key changed are queried again. The state and process of a
|
||||
// service are never cached.
|
||||
type serviceConfigCache struct {
|
||||
mu sync.Mutex
|
||||
|
||||
// entries is nil while the cache is disabled.
|
||||
entries map[string]cachedServiceConfig
|
||||
|
||||
// generation is incremented on every check, so that a configuration queried
|
||||
// before a check is not stored after it.
|
||||
generation uint64
|
||||
expiresAt time.Time
|
||||
|
||||
key registry.Key
|
||||
changed windows.Handle
|
||||
}
|
||||
|
||||
// init opens the services registry key below root and starts watching it for changes.
|
||||
// The cache stays disabled if init fails.
|
||||
func (cache *serviceConfigCache) init(root registry.Key, path string) error {
|
||||
key, err := registry.OpenKey(root, path, registry.NOTIFY|registry.QUERY_VALUE|registry.ENUMERATE_SUB_KEYS)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to open registry key %s: %w", path, err)
|
||||
}
|
||||
|
||||
// Auto-reset event, so that waiting on it also resets it.
|
||||
changed, err := windows.CreateEvent(nil, 0, 0, nil)
|
||||
if err != nil {
|
||||
_ = key.Close()
|
||||
|
||||
return fmt.Errorf("failed to create event: %w", err)
|
||||
}
|
||||
|
||||
cache.mu.Lock()
|
||||
defer cache.mu.Unlock()
|
||||
|
||||
cache.key = key
|
||||
cache.changed = changed
|
||||
|
||||
if err := cache.armLocked(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
cache.entries = make(map[string]cachedServiceConfig)
|
||||
cache.expiresAt = time.Now().Add(configCacheMaxAge)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// armLocked requests a single notification for the next change below the services key.
|
||||
func (cache *serviceConfigCache) armLocked() error {
|
||||
err := windows.RegNotifyChangeKeyValue(windows.Handle(cache.key), true, servicesKeyNotifyFilter, cache.changed, true)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to watch registry key: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// lastWrite returns the last write time of the registry key of a service.
|
||||
func (cache *serviceConfigCache) lastWrite(name string) (time.Time, error) {
|
||||
cache.mu.Lock()
|
||||
defer cache.mu.Unlock()
|
||||
|
||||
return cache.lastWriteLocked(name)
|
||||
}
|
||||
|
||||
// lastWriteLocked is lastWrite for callers that hold the lock, which keeps
|
||||
// close from closing the services key while it is in use.
|
||||
func (cache *serviceConfigCache) lastWriteLocked(name string) (time.Time, error) {
|
||||
if cache.key == 0 {
|
||||
return time.Time{}, errConfigCacheClosed
|
||||
}
|
||||
|
||||
key, err := registry.OpenKey(cache.key, name, registry.QUERY_VALUE)
|
||||
if err != nil {
|
||||
return time.Time{}, err
|
||||
}
|
||||
|
||||
defer key.Close()
|
||||
|
||||
info, err := key.Stat()
|
||||
if err != nil {
|
||||
return time.Time{}, err
|
||||
}
|
||||
|
||||
return info.ModTime(), nil
|
||||
}
|
||||
|
||||
// refresh drops the cached configurations of services whose registry key changed
|
||||
// since they were queried, or all of them if the cache expired. It returns the
|
||||
// number of dropped configurations and whether the cache expired. If the watch
|
||||
// can't be armed again, the cache is disabled.
|
||||
func (cache *serviceConfigCache) refresh() (int, bool, error) {
|
||||
cache.mu.Lock()
|
||||
defer cache.mu.Unlock()
|
||||
|
||||
if cache.entries == nil {
|
||||
return 0, false, nil
|
||||
}
|
||||
|
||||
if !time.Now().Before(cache.expiresAt) {
|
||||
dropped := len(cache.entries)
|
||||
|
||||
clear(cache.entries)
|
||||
cache.generation++
|
||||
cache.expiresAt = time.Now().Add(configCacheMaxAge)
|
||||
|
||||
return dropped, true, nil
|
||||
}
|
||||
|
||||
event, err := windows.WaitForSingleObject(cache.changed, 0)
|
||||
if err != nil {
|
||||
cache.entries = nil
|
||||
|
||||
return 0, false, fmt.Errorf("failed to check registry key for changes: %w", err)
|
||||
}
|
||||
|
||||
if event != windows.WAIT_OBJECT_0 {
|
||||
return 0, false, nil
|
||||
}
|
||||
|
||||
// Arm before checking the keys, so that a change from now on is not missed.
|
||||
if err := cache.armLocked(); err != nil {
|
||||
cache.entries = nil
|
||||
|
||||
return 0, false, err
|
||||
}
|
||||
|
||||
cache.generation++
|
||||
|
||||
var dropped int
|
||||
|
||||
for name, entry := range cache.entries {
|
||||
// A failure, for example because the service was deleted, also drops the entry.
|
||||
lastWrite, err := cache.lastWriteLocked(name)
|
||||
if err == nil && lastWrite.Equal(entry.lastWrite) {
|
||||
continue
|
||||
}
|
||||
|
||||
delete(cache.entries, name)
|
||||
|
||||
dropped++
|
||||
}
|
||||
|
||||
return dropped, false, nil
|
||||
}
|
||||
|
||||
// get returns the cached configuration of a service and the current generation.
|
||||
// ok is false on a miss.
|
||||
func (cache *serviceConfigCache) get(name string) (cachedServiceConfig, uint64, bool) {
|
||||
cache.mu.Lock()
|
||||
defer cache.mu.Unlock()
|
||||
|
||||
entry, ok := cache.entries[name]
|
||||
|
||||
return entry, cache.generation, ok
|
||||
}
|
||||
|
||||
// put stores a configuration that was queried in the given generation.
|
||||
// It is dropped if the cache was checked or disabled in the meantime.
|
||||
func (cache *serviceConfigCache) put(name string, generation uint64, entry cachedServiceConfig) {
|
||||
cache.mu.Lock()
|
||||
defer cache.mu.Unlock()
|
||||
|
||||
if cache.entries == nil || generation != cache.generation {
|
||||
return
|
||||
}
|
||||
|
||||
cache.entries[name] = entry
|
||||
}
|
||||
|
||||
// enabled reports whether the cache is in use.
|
||||
func (cache *serviceConfigCache) enabled() bool {
|
||||
cache.mu.Lock()
|
||||
defer cache.mu.Unlock()
|
||||
|
||||
return cache.entries != nil
|
||||
}
|
||||
|
||||
func (cache *serviceConfigCache) close() error {
|
||||
cache.mu.Lock()
|
||||
defer cache.mu.Unlock()
|
||||
|
||||
cache.entries = nil
|
||||
|
||||
var errs []error
|
||||
|
||||
// Closing the key also signals the pending notification.
|
||||
if cache.key != 0 {
|
||||
if err := cache.key.Close(); err != nil {
|
||||
errs = append(errs, fmt.Errorf("failed to close registry key: %w", err))
|
||||
}
|
||||
|
||||
cache.key = 0
|
||||
}
|
||||
|
||||
if cache.changed != 0 {
|
||||
if err := windows.CloseHandle(cache.changed); err != nil {
|
||||
errs = append(errs, fmt.Errorf("failed to close event: %w", err))
|
||||
}
|
||||
|
||||
cache.changed = 0
|
||||
}
|
||||
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
|
||||
// getCachedServiceConfig returns the configuration of a service from the cache,
|
||||
// or queries and caches it on a miss.
|
||||
func (c *Collector) getCachedServiceConfig(serviceName string) (mgr.Config, bool, error) {
|
||||
entry, generation, ok := c.configCache.get(serviceName)
|
||||
if ok {
|
||||
return entry.config, entry.available, nil
|
||||
}
|
||||
|
||||
if !c.configCache.enabled() {
|
||||
return c.queryServiceConfig(serviceName)
|
||||
}
|
||||
|
||||
// Read the last write time first. If the key changes while the configuration
|
||||
// is queried, the next check sees a newer time and queries it again.
|
||||
lastWrite, lastWriteErr := c.configCache.lastWrite(serviceName)
|
||||
|
||||
config, available, err := c.queryServiceConfig(serviceName)
|
||||
if err != nil {
|
||||
return mgr.Config{}, false, err
|
||||
}
|
||||
|
||||
switch {
|
||||
case lastWriteErr != nil:
|
||||
c.logger.Debug("Failed to read the last write time of the service registry key. The service configuration is not cached",
|
||||
slog.String("service", serviceName),
|
||||
slog.Any("err", lastWriteErr),
|
||||
)
|
||||
case !available:
|
||||
// An unavailable configuration, for example a display name from a missing
|
||||
// MUI file, can become available without a change to the service key.
|
||||
case time.Since(lastWrite) < configCacheMinKeyAge:
|
||||
// Cached on a later scrape, once a second write in the same tick is ruled out.
|
||||
default:
|
||||
c.configCache.put(serviceName, generation, cachedServiceConfig{
|
||||
config: config,
|
||||
available: available,
|
||||
lastWrite: lastWrite,
|
||||
})
|
||||
}
|
||||
|
||||
return config, available, nil
|
||||
}
|
||||
@@ -0,0 +1,266 @@
|
||||
// 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.
|
||||
|
||||
//go:build windows
|
||||
|
||||
package service
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"golang.org/x/sys/windows/registry"
|
||||
"golang.org/x/sys/windows/svc/mgr"
|
||||
)
|
||||
|
||||
// newTestConfigCache returns a cache that watches a temporary key below
|
||||
// HKEY_CURRENT_USER with the subkeys svcA and svcB, each with a cached entry.
|
||||
func newTestConfigCache(t *testing.T) (*serviceConfigCache, registry.Key) {
|
||||
t.Helper()
|
||||
|
||||
path := fmt.Sprintf(`Software\windows_exporter_test\%s_%d`, t.Name(), time.Now().UnixNano())
|
||||
|
||||
root, _, err := registry.CreateKey(registry.CURRENT_USER, path, registry.ALL_ACCESS)
|
||||
require.NoError(t, err)
|
||||
|
||||
t.Cleanup(func() {
|
||||
_ = registry.DeleteKey(root, `svcA\Parameters`)
|
||||
|
||||
for _, name := range []string{"svcA", "svcB"} {
|
||||
_ = registry.DeleteKey(root, name)
|
||||
}
|
||||
|
||||
_ = root.Close()
|
||||
_ = registry.DeleteKey(registry.CURRENT_USER, path)
|
||||
// Fails while other tests still use it.
|
||||
_ = registry.DeleteKey(registry.CURRENT_USER, `Software\windows_exporter_test`)
|
||||
})
|
||||
|
||||
for _, name := range []string{"svcA", "svcB"} {
|
||||
key, _, err := registry.CreateKey(root, name, registry.ALL_ACCESS)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, key.SetStringValue("DisplayName", name))
|
||||
require.NoError(t, key.Close())
|
||||
}
|
||||
|
||||
cache := &serviceConfigCache{}
|
||||
require.NoError(t, cache.init(registry.CURRENT_USER, path))
|
||||
t.Cleanup(func() { assert.NoError(t, cache.close()) })
|
||||
|
||||
for _, name := range []string{"svcA", "svcB"} {
|
||||
lastWrite, err := cache.lastWrite(name)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, generation, _ := cache.get(name)
|
||||
cache.put(name, generation, cachedServiceConfig{
|
||||
config: mgr.Config{DisplayName: name},
|
||||
available: true,
|
||||
lastWrite: lastWrite,
|
||||
})
|
||||
}
|
||||
|
||||
// Registry write times have a coarse resolution. Make sure the next write
|
||||
// gets a different time than the keys were created with.
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
return cache, root
|
||||
}
|
||||
|
||||
// waitForCheck refreshes the cache until it handled a change notification and
|
||||
// returns the number of dropped entries.
|
||||
func waitForCheck(t *testing.T, cache *serviceConfigCache) int {
|
||||
t.Helper()
|
||||
|
||||
_, before, _ := cache.get("")
|
||||
|
||||
var dropped int
|
||||
|
||||
require.Eventually(t, func() bool {
|
||||
n, _, err := cache.refresh()
|
||||
if !assert.NoError(t, err) {
|
||||
return true
|
||||
}
|
||||
|
||||
dropped += n
|
||||
_, after, _ := cache.get("")
|
||||
|
||||
return after != before
|
||||
}, 5*time.Second, 10*time.Millisecond)
|
||||
|
||||
return dropped
|
||||
}
|
||||
|
||||
func cached(cache *serviceConfigCache, name string) bool {
|
||||
_, _, ok := cache.get(name)
|
||||
|
||||
return ok
|
||||
}
|
||||
|
||||
func TestConfigCacheKeepsEntriesWithoutNotification(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newTestConfigCache(t)
|
||||
|
||||
dropped, _, err := cache.refresh()
|
||||
require.NoError(t, err)
|
||||
assert.Zero(t, dropped)
|
||||
assert.True(t, cached(cache, "svcA"))
|
||||
assert.True(t, cached(cache, "svcB"))
|
||||
}
|
||||
|
||||
func TestConfigCacheDropsChangedService(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, root := newTestConfigCache(t)
|
||||
|
||||
key, err := registry.OpenKey(root, "svcA", registry.SET_VALUE)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, key.SetDWordValue("Start", 4))
|
||||
require.NoError(t, key.Close())
|
||||
|
||||
assert.Equal(t, 1, waitForCheck(t, cache))
|
||||
assert.False(t, cached(cache, "svcA"))
|
||||
assert.True(t, cached(cache, "svcB"))
|
||||
|
||||
// The watch is armed again and catches a second change.
|
||||
key, err = registry.OpenKey(root, "svcB", registry.SET_VALUE)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, key.SetStringValue("DisplayName", "changed"))
|
||||
require.NoError(t, key.Close())
|
||||
|
||||
assert.Equal(t, 1, waitForCheck(t, cache))
|
||||
assert.False(t, cached(cache, "svcB"))
|
||||
}
|
||||
|
||||
func TestConfigCacheKeepsServiceOnDeeperWrite(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, root := newTestConfigCache(t)
|
||||
|
||||
// Services and drivers write below their own key at runtime, for example to
|
||||
// Parameters or Enum. That doesn't change the configuration.
|
||||
key, _, err := registry.CreateKey(root, `svcA\Parameters`, registry.ALL_ACCESS)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, key.SetDWordValue("RunTime", 1))
|
||||
require.NoError(t, key.Close())
|
||||
|
||||
key, err = registry.OpenKey(root, `svcA\Parameters`, registry.SET_VALUE)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, key.SetDWordValue("RunTime", 2))
|
||||
require.NoError(t, key.Close())
|
||||
|
||||
// Creating the Parameters subkey changes the write time of svcA, a value
|
||||
// change below it doesn't. Handle and drain the notifications of the setup.
|
||||
waitForCheck(t, cache)
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
_, _, err = cache.refresh()
|
||||
require.NoError(t, err)
|
||||
|
||||
lastWrite, err := cache.lastWrite("svcA")
|
||||
require.NoError(t, err)
|
||||
|
||||
_, generation, _ := cache.get("svcA")
|
||||
cache.put("svcA", generation, cachedServiceConfig{available: true, lastWrite: lastWrite})
|
||||
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
key, err = registry.OpenKey(root, `svcA\Parameters`, registry.SET_VALUE)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, key.SetDWordValue("RunTime", 3))
|
||||
require.NoError(t, key.Close())
|
||||
|
||||
assert.Zero(t, waitForCheck(t, cache))
|
||||
assert.True(t, cached(cache, "svcA"))
|
||||
assert.True(t, cached(cache, "svcB"))
|
||||
}
|
||||
|
||||
func TestConfigCacheDropsDeletedService(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, root := newTestConfigCache(t)
|
||||
|
||||
require.NoError(t, registry.DeleteKey(root, "svcB"))
|
||||
|
||||
assert.Equal(t, 1, waitForCheck(t, cache))
|
||||
assert.True(t, cached(cache, "svcA"))
|
||||
assert.False(t, cached(cache, "svcB"))
|
||||
}
|
||||
|
||||
func TestConfigCacheDropsPutFromBeforeCheck(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, root := newTestConfigCache(t)
|
||||
|
||||
_, generation, _ := cache.get("svcC")
|
||||
|
||||
key, err := registry.OpenKey(root, "svcA", registry.SET_VALUE)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, key.SetDWordValue("Start", 4))
|
||||
require.NoError(t, key.Close())
|
||||
|
||||
waitForCheck(t, cache)
|
||||
|
||||
cache.put("svcC", generation, cachedServiceConfig{available: true})
|
||||
assert.False(t, cached(cache, "svcC"))
|
||||
}
|
||||
|
||||
func TestConfigCacheExpires(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newTestConfigCache(t)
|
||||
|
||||
cache.mu.Lock()
|
||||
cache.expiresAt = time.Now().Add(-time.Second)
|
||||
cache.mu.Unlock()
|
||||
|
||||
dropped, expired, err := cache.refresh()
|
||||
require.NoError(t, err)
|
||||
assert.True(t, expired)
|
||||
assert.Equal(t, 2, dropped)
|
||||
assert.False(t, cached(cache, "svcA"))
|
||||
assert.False(t, cached(cache, "svcB"))
|
||||
}
|
||||
|
||||
func TestConfigCacheAfterClose(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newTestConfigCache(t)
|
||||
|
||||
require.NoError(t, cache.close())
|
||||
assert.False(t, cache.enabled())
|
||||
assert.False(t, cached(cache, "svcA"))
|
||||
|
||||
_, generation, _ := cache.get("svcA")
|
||||
cache.put("svcA", generation, cachedServiceConfig{available: true})
|
||||
assert.False(t, cached(cache, "svcA"))
|
||||
|
||||
dropped, _, err := cache.refresh()
|
||||
require.NoError(t, err)
|
||||
assert.Zero(t, dropped)
|
||||
require.NoError(t, cache.close())
|
||||
}
|
||||
|
||||
func TestConfigCacheInitFailureDisablesCache(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache := &serviceConfigCache{}
|
||||
require.Error(t, cache.init(registry.CURRENT_USER, `Software\windows_exporter_test\does_not_exist`))
|
||||
assert.False(t, cache.enabled())
|
||||
require.NoError(t, cache.close())
|
||||
}
|
||||
@@ -38,6 +38,7 @@ import (
|
||||
"github.com/prometheus-community/windows_exporter/internal/utils/recovery"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"golang.org/x/sys/windows"
|
||||
"golang.org/x/sys/windows/registry"
|
||||
"golang.org/x/sys/windows/svc/mgr"
|
||||
)
|
||||
|
||||
@@ -78,6 +79,9 @@ type Collector struct {
|
||||
queryAllServicesMu sync.Mutex
|
||||
queryAllServicesBuffer []byte
|
||||
|
||||
// configCache caches the service configurations between scrapes.
|
||||
configCache serviceConfigCache
|
||||
|
||||
// allStartModesIncluded is true if ServiceStartModeInclude contains every start mode.
|
||||
// Only then services with an unknown start mode are collected.
|
||||
allStartModesIncluded bool
|
||||
@@ -264,12 +268,24 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error {
|
||||
|
||||
c.serviceManagerHandle = &mgr.Mgr{Handle: handle}
|
||||
|
||||
if err := c.configCache.init(registry.LOCAL_MACHINE, servicesKeyPath); err != nil {
|
||||
c.logger.Warn("Failed to watch the service configurations for changes. Service configurations are queried on every scrape",
|
||||
slog.Any("err", err),
|
||||
)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Collector) Close() error {
|
||||
c.metricCache.Reset()
|
||||
|
||||
if err := c.configCache.close(); err != nil {
|
||||
c.logger.Warn("Failed to close the service configuration cache",
|
||||
slog.Any("err", err),
|
||||
)
|
||||
}
|
||||
|
||||
if c.serviceManagerHandle == nil {
|
||||
return nil
|
||||
}
|
||||
@@ -293,6 +309,21 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error
|
||||
return fmt.Errorf("failed to query all services: %w", err)
|
||||
}
|
||||
|
||||
switch dropped, expired, err := c.configCache.refresh(); {
|
||||
case err != nil:
|
||||
c.logger.Warn("Failed to watch the service configurations for changes. Service configurations are queried on every scrape",
|
||||
slog.Any("err", err),
|
||||
)
|
||||
case expired:
|
||||
c.logger.Debug("Service configuration cache expired",
|
||||
slog.Int("count", dropped),
|
||||
)
|
||||
case dropped > 0:
|
||||
c.logger.Debug("Dropped changed service configurations from the cache",
|
||||
slog.Int("count", dropped),
|
||||
)
|
||||
}
|
||||
|
||||
scrape := c.metricCache.Begin()
|
||||
|
||||
servicesCh := make(chan serviceEntry, len(services))
|
||||
@@ -340,41 +371,9 @@ func (c *Collector) collectWorker(ch chan<- prometheus.Metric, scrape *metriccac
|
||||
func (c *Collector) collectService(ch chan<- prometheus.Metric, scrape *metriccache.Scrape[string, serviceValues], service serviceEntry) error {
|
||||
serviceName := service.name
|
||||
|
||||
serviceNamePtr, err := windows.UTF16PtrFromString(serviceName)
|
||||
serviceConfig, configAvailable, err := c.getCachedServiceConfig(serviceName)
|
||||
if err != nil {
|
||||
return fmt.Errorf("invalid service name: %w", err)
|
||||
}
|
||||
|
||||
// Open connection for service handler.
|
||||
serviceHandle, err := windows.OpenService(c.serviceManagerHandle.Handle, serviceNamePtr, windows.SERVICE_QUERY_CONFIG)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to open service: %w", err)
|
||||
}
|
||||
|
||||
// Create handle for each service.
|
||||
serviceManager := &mgr.Service{Name: serviceName, Handle: serviceHandle}
|
||||
defer func(serviceManager *mgr.Service) {
|
||||
if err := serviceManager.Close(); err != nil {
|
||||
c.logger.Log(context.Background(), slog.LevelWarn, "failed to close service handle",
|
||||
slog.Any("err", err),
|
||||
slog.String("service", serviceName),
|
||||
)
|
||||
}
|
||||
}(serviceManager)
|
||||
|
||||
// Get Service Configuration.
|
||||
serviceConfig, err := c.getServiceConfig(serviceManager)
|
||||
configAvailable := err == nil
|
||||
|
||||
if err != nil {
|
||||
if !errors.Is(err, windows.ERROR_FILE_NOT_FOUND) && !errors.Is(err, windows.ERROR_MUI_FILE_NOT_FOUND) {
|
||||
return fmt.Errorf("failed to get service configuration: %w", err)
|
||||
}
|
||||
|
||||
c.logger.Log(context.Background(), slog.LevelDebug, "failed collecting service config",
|
||||
slog.Any("err", err),
|
||||
slog.String("service", serviceName),
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// serviceStartMode stays empty if the service configuration is not available.
|
||||
@@ -598,6 +597,49 @@ func (c *Collector) queryAllServices() ([]serviceEntry, error) {
|
||||
return entries, nil
|
||||
}
|
||||
|
||||
// queryServiceConfig opens the service and queries its configuration.
|
||||
// configAvailable is false if the service has no readable configuration.
|
||||
func (c *Collector) queryServiceConfig(serviceName string) (mgr.Config, bool, error) {
|
||||
serviceNamePtr, err := windows.UTF16PtrFromString(serviceName)
|
||||
if err != nil {
|
||||
return mgr.Config{}, false, fmt.Errorf("invalid service name: %w", err)
|
||||
}
|
||||
|
||||
// Open connection for service handler.
|
||||
serviceHandle, err := windows.OpenService(c.serviceManagerHandle.Handle, serviceNamePtr, windows.SERVICE_QUERY_CONFIG)
|
||||
if err != nil {
|
||||
return mgr.Config{}, false, fmt.Errorf("failed to open service: %w", err)
|
||||
}
|
||||
|
||||
// Create handle for each service.
|
||||
serviceManager := &mgr.Service{Name: serviceName, Handle: serviceHandle}
|
||||
defer func(serviceManager *mgr.Service) {
|
||||
if err := serviceManager.Close(); err != nil {
|
||||
c.logger.Log(context.Background(), slog.LevelWarn, "failed to close service handle",
|
||||
slog.Any("err", err),
|
||||
slog.String("service", serviceName),
|
||||
)
|
||||
}
|
||||
}(serviceManager)
|
||||
|
||||
// Get Service Configuration.
|
||||
serviceConfig, err := c.getServiceConfig(serviceManager)
|
||||
if err != nil {
|
||||
if !errors.Is(err, windows.ERROR_FILE_NOT_FOUND) && !errors.Is(err, windows.ERROR_MUI_FILE_NOT_FOUND) {
|
||||
return mgr.Config{}, false, fmt.Errorf("failed to get service configuration: %w", err)
|
||||
}
|
||||
|
||||
c.logger.Log(context.Background(), slog.LevelDebug, "failed collecting service config",
|
||||
slog.Any("err", err),
|
||||
slog.String("service", serviceName),
|
||||
)
|
||||
|
||||
return mgr.Config{}, false, nil
|
||||
}
|
||||
|
||||
return serviceConfig, true, nil
|
||||
}
|
||||
|
||||
func (c *Collector) getProcessStartTime(pid uint32) (uint64, error) {
|
||||
handle, err := windows.OpenProcess(windows.PROCESS_QUERY_LIMITED_INFORMATION, false, pid)
|
||||
if err != nil {
|
||||
|
||||
@@ -0,0 +1,178 @@
|
||||
// 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.
|
||||
|
||||
//go:build windows
|
||||
|
||||
package service
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os"
|
||||
"regexp"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"golang.org/x/sys/windows"
|
||||
"golang.org/x/sys/windows/svc/mgr"
|
||||
)
|
||||
|
||||
type serviceMetricCollector struct {
|
||||
t assert.TestingT
|
||||
collector *Collector
|
||||
}
|
||||
|
||||
func (c serviceMetricCollector) Describe(ch chan<- *prometheus.Desc) {
|
||||
prometheus.DescribeByCollect(c, ch)
|
||||
}
|
||||
|
||||
func (c serviceMetricCollector) Collect(ch chan<- prometheus.Metric) {
|
||||
assert.NoError(c.t, c.collector.Collect(ch, 0))
|
||||
}
|
||||
|
||||
// serviceMetrics scrapes the collector and returns the value of each series of
|
||||
// the service as "<metric>{<label values>}". It only reports failures to t, so
|
||||
// that it can be retried inside EventuallyWithT.
|
||||
func serviceMetrics(t assert.TestingT, c *Collector, name string) map[string]float64 {
|
||||
registry := prometheus.NewPedanticRegistry()
|
||||
// require must not be used inside EventuallyWithT.
|
||||
//nolint:testifylint
|
||||
if !assert.NoError(t, registry.Register(serviceMetricCollector{t: t, collector: c})) {
|
||||
return nil
|
||||
}
|
||||
|
||||
families, err := registry.Gather()
|
||||
//nolint:testifylint
|
||||
if !assert.NoError(t, err) {
|
||||
return nil
|
||||
}
|
||||
|
||||
metrics := make(map[string]float64)
|
||||
|
||||
for _, family := range families {
|
||||
for _, metric := range family.GetMetric() {
|
||||
var (
|
||||
labels []string
|
||||
matches bool
|
||||
)
|
||||
|
||||
for _, label := range metric.GetLabel() {
|
||||
if label.GetName() == "name" {
|
||||
matches = label.GetValue() == name
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
labels = append(labels, label.GetName()+"="+label.GetValue())
|
||||
}
|
||||
|
||||
if matches {
|
||||
metrics[fmt.Sprintf("%s{%s}", family.GetName(), strings.Join(labels, ","))] = metric.GetGauge().GetValue()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return metrics
|
||||
}
|
||||
|
||||
// TestCollectorPicksUpConfigChanges creates a service, changes its configuration
|
||||
// through the service control manager and checks that the cached configuration
|
||||
// is replaced. It needs an elevated process, which the Windows CI runners are.
|
||||
func TestCollectorPicksUpConfigChanges(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
required := slices.Contains(strings.Split(os.Getenv("WINDOWS_EXPORTER_TEST_COLLECTORS"), ","), Name)
|
||||
if !windows.GetCurrentProcessToken().IsElevated() {
|
||||
if required {
|
||||
t.Fatal("creating a test service needs an elevated process")
|
||||
}
|
||||
|
||||
t.Skip("creating a test service needs an elevated process")
|
||||
}
|
||||
|
||||
manager, err := mgr.Connect()
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { assert.NoError(t, manager.Disconnect()) })
|
||||
|
||||
name := fmt.Sprintf("windows_exporter_test_%d", time.Now().UnixNano())
|
||||
|
||||
// The service is never started.
|
||||
service, err := manager.CreateService(name, `C:\Windows\System32\cmd.exe`, mgr.Config{
|
||||
StartType: mgr.StartManual,
|
||||
DisplayName: "before",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
deleted := false
|
||||
|
||||
t.Cleanup(func() {
|
||||
if !deleted {
|
||||
assert.NoError(t, service.Delete())
|
||||
assert.NoError(t, service.Close())
|
||||
}
|
||||
})
|
||||
|
||||
c := New(&Config{
|
||||
ServiceInclude: regexp.MustCompile("^" + regexp.QuoteMeta(name) + "$"),
|
||||
ServiceExclude: regexp.MustCompile("^$"),
|
||||
})
|
||||
require.NoError(t, c.Build(slog.New(slog.DiscardHandler), nil))
|
||||
t.Cleanup(func() { assert.NoError(t, c.Close()) })
|
||||
require.True(t, c.configCache.enabled(), "the configuration cache must be enabled")
|
||||
|
||||
// A configuration is only cached once its key is older than a clock tick.
|
||||
time.Sleep(configCacheMinKeyAge + 100*time.Millisecond)
|
||||
|
||||
metrics := serviceMetrics(t, c, name)
|
||||
assert.InDelta(t, 1.0, metrics["windows_service_start_mode{start_mode=manual}"], 0)
|
||||
assert.Contains(t, metrics, `windows_service_info{display_name=before,path_name=C:\Windows\System32\cmd.exe,run_as=LocalSystem}`)
|
||||
|
||||
_, _, ok := c.configCache.get(name)
|
||||
require.True(t, ok, "the configuration must be cached after a scrape")
|
||||
|
||||
config, err := service.Config()
|
||||
require.NoError(t, err)
|
||||
|
||||
config.StartType = mgr.StartDisabled
|
||||
config.DisplayName = "after"
|
||||
require.NoError(t, service.UpdateConfig(config))
|
||||
|
||||
// The change notification is delivered asynchronously.
|
||||
require.EventuallyWithT(t, func(collect *assert.CollectT) {
|
||||
metrics := serviceMetrics(collect, c, name)
|
||||
assert.InDelta(collect, 1.0, metrics["windows_service_start_mode{start_mode=disabled}"], 0)
|
||||
assert.InDelta(collect, 0.0, metrics["windows_service_start_mode{start_mode=manual}"], 0)
|
||||
assert.Contains(collect, metrics, `windows_service_info{display_name=after,path_name=C:\Windows\System32\cmd.exe,run_as=LocalSystem}`)
|
||||
}, 5*time.Second, 50*time.Millisecond)
|
||||
|
||||
// The service control manager removes the service once the last handle is closed.
|
||||
require.NoError(t, service.Delete())
|
||||
|
||||
deleted = true
|
||||
|
||||
require.NoError(t, service.Close())
|
||||
|
||||
require.EventuallyWithT(t, func(collect *assert.CollectT) {
|
||||
assert.Empty(collect, serviceMetrics(collect, c, name))
|
||||
|
||||
_, _, ok := c.configCache.get(name)
|
||||
assert.False(collect, ok, "the configuration of a deleted service must be dropped")
|
||||
}, 5*time.Second, 50*time.Millisecond)
|
||||
}
|
||||
Reference in New Issue
Block a user