diff --git a/internal/collector/container/container.go b/internal/collector/container/container.go index 4b57310f..351d9333 100644 --- a/internal/collector/container/container.go +++ b/internal/collector/container/container.go @@ -26,6 +26,7 @@ import ( "os" "slices" "strings" + "sync" "time" "unsafe" @@ -69,6 +70,11 @@ type Collector struct { logger *slog.Logger + // mu serializes Collect, because the annotation caches are not safe for concurrent use. + // Overlapping Collect calls, e.g. after a scrape timeout, would otherwise write them concurrently. + mu sync.Mutex + + // The annotation caches are keyed by the container ID without the runtime prefix. annotationsCacheHCS map[string]containerInfo annotationsCacheJob map[string]containerInfo @@ -293,6 +299,9 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { // Collect sends the metric values for each metric // to the provided prometheus Metric channel. func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error { + c.mu.Lock() + defer c.mu.Unlock() + errs := make([]error, 0) if slices.Contains(c.config.CollectorsEnabled, subCollectorHCS) { @@ -319,6 +328,8 @@ func (c *Collector) collectHCS(ch chan<- prometheus.Metric) error { count := len(containers) if count == 0 { + clear(c.annotationsCacheHCS) + ch <- prometheus.MustNewConstMetric( c.containersCount, prometheus.GaugeValue, @@ -394,17 +405,19 @@ func (c *Collector) collectHCS(ch chan<- prometheus.Metric) error { countersCount, ) - if err = c.collectNetworkMetrics(ch); err != nil { - return fmt.Errorf("error in fetching container network statistics: %w", err) - } + networkErr := c.collectNetworkMetrics(ch) // Remove containers that are no longer running - for _, containerID := range c.annotationsCacheHCS { - if !slices.Contains(containerIDs, containerID.id) { - delete(c.annotationsCacheHCS, containerID.id) + for containerID := range c.annotationsCacheHCS { + if !slices.Contains(containerIDs, containerID) { + delete(c.annotationsCacheHCS, containerID) } } + if networkErr != nil { + return fmt.Errorf("error in fetching container network statistics: %w", networkErr) + } + if len(collectErrors) > 0 { return fmt.Errorf("errors while fetching container statistics: %w", errors.Join(collectErrors...)) } @@ -688,9 +701,9 @@ func (c *Collector) collectJobContainers(ch chan<- prometheus.Metric) error { } // Remove containers that are no longer running - for _, containerID := range c.annotationsCacheJob { - if !slices.Contains(jobContainerIDs, containerID.id) { - delete(c.annotationsCacheJob, containerID.id) + for containerID := range c.annotationsCacheJob { + if !slices.Contains(jobContainerIDs, containerID) { + delete(c.annotationsCacheJob, containerID) } } diff --git a/internal/collector/container/container_internal_test.go b/internal/collector/container/container_internal_test.go new file mode 100644 index 00000000..cc42286a --- /dev/null +++ b/internal/collector/container/container_internal_test.go @@ -0,0 +1,45 @@ +// 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 container + +import ( + "log/slog" + "testing" + + "github.com/prometheus/client_golang/prometheus" + "github.com/stretchr/testify/require" +) + +// The annotation cache is keyed by the raw container ID, but was pruned by the +// prefixed ID, so entries of removed containers were never deleted. +func TestCollectJobContainersPrunesAnnotationsCache(t *testing.T) { + t.Parallel() + + c := &Collector{ + config: Config{ContainerDStateDir: t.TempDir() + `\`}, + logger: slog.New(slog.DiscardHandler), + annotationsCacheJob: map[string]containerInfo{ + "removed": {id: "containerd://removed"}, + }, + } + + ch := make(chan prometheus.Metric, 100) + + require.NoError(t, c.collectJobContainers(ch)) + require.Empty(t, c.annotationsCacheJob) +} diff --git a/internal/collector/dfsr/dfsr.go b/internal/collector/dfsr/dfsr.go index 7228f54c..4bb9619c 100644 --- a/internal/collector/dfsr/dfsr.go +++ b/internal/collector/dfsr/dfsr.go @@ -163,6 +163,16 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { logger.Info("dfsr collector is in an experimental state! Metrics for this collector have not been tested.") + subCollectors := []string{"connection", "folder", "volume"} + + for _, collector := range c.config.CollectorsEnabled { + if !slices.Contains(subCollectors, collector) { + return fmt.Errorf("unknown sub collector: %s. Possible values: %s", collector, + strings.Join(subCollectors, ", "), + ) + } + } + // connection c.connectionBandwidthSavingsUsingDFSReplicationTotal = prometheus.NewDesc( prometheus.BuildFQName(types.Namespace, Name, "connection_bandwidth_savings_using_dfs_replication_bytes_total"), diff --git a/internal/collector/dhcp/dhcp.go b/internal/collector/dhcp/dhcp.go index 9dcfb5c1..93fa1154 100644 --- a/internal/collector/dhcp/dhcp.go +++ b/internal/collector/dhcp/dhcp.go @@ -153,6 +153,16 @@ func (c *Collector) Close() error { func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { c.logger = logger.With(slog.String("collector", Name)) + subCollectors := []string{subCollectorServerMetrics, subCollectorScopeMetrics} + + for _, collector := range c.config.CollectorsEnabled { + if !slices.Contains(subCollectors, collector) { + return fmt.Errorf("unknown sub collector: %s. Possible values: %s", collector, + strings.Join(subCollectors, ", "), + ) + } + } + var err error if slices.Contains(c.config.CollectorsEnabled, subCollectorScopeMetrics) { diff --git a/internal/collector/exchange/exchange.go b/internal/collector/exchange/exchange.go index e82e637b..ad7894a0 100644 --- a/internal/collector/exchange/exchange.go +++ b/internal/collector/exchange/exchange.go @@ -21,13 +21,15 @@ import ( "errors" "fmt" "log/slog" + "maps" "os" + "slices" "strings" - "sync" "time" "github.com/alecthomas/kingpin/v2" "github.com/prometheus-community/windows_exporter/internal/mi" + "github.com/prometheus-community/windows_exporter/internal/utils/recovery" "github.com/prometheus/client_golang/prometheus" ) @@ -169,6 +171,8 @@ func (c *Collector) Close() error { fn() } + c.closeFns = nil + return nil } @@ -232,12 +236,24 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { }, } - errs := make([]error, 0, len(c.config.CollectorsEnabled)) + collectorsEnabled := slices.Compact(slices.Sorted(slices.Values(c.config.CollectorsEnabled))) - for _, name := range c.config.CollectorsEnabled { + for _, name := range collectorsEnabled { if _, ok := subCollectors[name]; !ok { - return fmt.Errorf("unknown collector: %s", name) + return fmt.Errorf("unknown sub collector: %s. Possible values: %s", name, + strings.Join(slices.Sorted(maps.Keys(subCollectors)), ", "), + ) } + } + + c.collectorFns = make([]func(ch chan<- prometheus.Metric) error, 0, len(collectorsEnabled)) + c.closeFns = make([]func(), 0, len(collectorsEnabled)) + + errs := make([]error, 0, len(collectorsEnabled)) + + for _, name := range collectorsEnabled { + // Register close before build, so that Close also releases a partially built sub collector. + c.closeFns = append(c.closeFns, subCollectors[name].close) if err := subCollectors[name].build(); err != nil { errs = append(errs, fmt.Errorf("failed to build %s collector: %w", name, err)) @@ -246,7 +262,6 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { } c.collectorFns = append(c.collectorFns, subCollectors[name].collect) - c.closeFns = append(c.closeFns, subCollectors[name].close) } return errors.Join(errs...) @@ -254,32 +269,15 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { // Collect collects exchange metrics and sends them to prometheus. func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error { - errCh := make(chan error, len(c.collectorFns)) - errs := make([]error, 0, len(c.collectorFns)) - - wg := sync.WaitGroup{} + var g recovery.Group for _, fn := range c.collectorFns { - wg.Add(1) - - go func(fn func(ch chan<- prometheus.Metric) error) { - defer wg.Done() - - if err := fn(ch); err != nil { - errCh <- err - } - }(fn) + g.Go(func() error { + return fn(ch) + }) } - wg.Wait() - - close(errCh) - - for err := range errCh { - errs = append(errs, err) - } - - return errors.Join(errs...) + return g.Wait() } // toLabelName converts strings to lowercase and replaces all whitespaces and dots with underscores. diff --git a/internal/collector/file/file.go b/internal/collector/file/file.go index bd2d6402..4752586d 100644 --- a/internal/collector/file/file.go +++ b/internal/collector/file/file.go @@ -18,20 +18,19 @@ package file import ( - "errors" "fmt" "io/fs" "log/slog" "os" "path/filepath" "strings" - "sync" "time" "github.com/alecthomas/kingpin/v2" "github.com/bmatcuk/doublestar/v4" "github.com/prometheus-community/windows_exporter/internal/mi" "github.com/prometheus-community/windows_exporter/internal/types" + "github.com/prometheus-community/windows_exporter/internal/utils/recovery" "github.com/prometheus/client_golang/prometheus" ) @@ -139,32 +138,19 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { // Collect sends the metric values for each metric // to the provided prometheus Metric channel. func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error { - wg := sync.WaitGroup{} - errCh := make(chan error, len(c.config.FilePatterns)) + var g recovery.Group for _, filePattern := range c.config.FilePatterns { - wg.Add(1) - - go func(filePattern string) { - defer wg.Done() - + g.Go(func() error { if err := c.collectGlobFilePath(ch, filePattern); err != nil { - errCh <- fmt.Errorf("failed collecting metrics for file pattern %s: %w", filePattern, err) + return fmt.Errorf("failed collecting metrics for file pattern %s: %w", filePattern, err) } - }(filePattern) + + return nil + }) } - wg.Wait() - - close(errCh) - - errs := make([]error, 0, len(c.config.FilePatterns)) - - for err := range errCh { - errs = append(errs, err) - } - - return errors.Join(errs...) + return g.Wait() } func (c *Collector) collectGlobFilePath(ch chan<- prometheus.Metric, filePattern string) error { diff --git a/internal/collector/hyperv/hyperv.go b/internal/collector/hyperv/hyperv.go index b64e570f..180f3bee 100644 --- a/internal/collector/hyperv/hyperv.go +++ b/internal/collector/hyperv/hyperv.go @@ -21,14 +21,15 @@ import ( "errors" "fmt" "log/slog" - "sort" + "maps" + "slices" "strings" - "sync" "time" "github.com/alecthomas/kingpin/v2" "github.com/prometheus-community/windows_exporter/internal/mi" "github.com/prometheus-community/windows_exporter/internal/osversion" + "github.com/prometheus-community/windows_exporter/internal/utils/recovery" "github.com/prometheus/client_golang/prometheus" ) @@ -160,6 +161,8 @@ func (c *Collector) Close() error { fn() } + c.closeFns = nil + return nil } @@ -276,16 +279,21 @@ func (c *Collector) Build(logger *slog.Logger, miSession *mi.Session) error { buildNumber := osversion.Build() + // Sort a copy, to not modify the slice of the caller or ConfigDefaults. // Result must order, to prevent test failures. - sort.Strings(c.config.CollectorsEnabled) + collectorsEnabled := slices.Compact(slices.Sorted(slices.Values(c.config.CollectorsEnabled))) - errs := make([]error, 0, len(c.config.CollectorsEnabled)) - - for _, name := range c.config.CollectorsEnabled { + for _, name := range collectorsEnabled { if _, ok := subCollectors[name]; !ok { - return fmt.Errorf("unknown collector: %s", name) + return fmt.Errorf("unknown sub collector: %s. Possible values: %s", name, + strings.Join(slices.Sorted(maps.Keys(subCollectors)), ", "), + ) } + } + errs := make([]error, 0, len(collectorsEnabled)) + + for _, name := range collectorsEnabled { if buildNumber < subCollectors[name].minBuildNumber { c.logger.Warn(fmt.Sprintf( "collector %s requires windows build version %d. Current build version: %d", @@ -295,6 +303,9 @@ func (c *Collector) Build(logger *slog.Logger, miSession *mi.Session) error { continue } + // Register close before build, so that Close also releases a partially built sub collector. + c.closeFns = append(c.closeFns, subCollectors[name].close) + if err := subCollectors[name].build(); err != nil { errs = append(errs, fmt.Errorf("failed to build %s collector: %w", name, err)) @@ -302,7 +313,6 @@ func (c *Collector) Build(logger *slog.Logger, miSession *mi.Session) error { } c.collectorFns = append(c.collectorFns, subCollectors[name].collect) - c.closeFns = append(c.closeFns, subCollectors[name].close) } return errors.Join(errs...) @@ -311,30 +321,13 @@ func (c *Collector) Build(logger *slog.Logger, miSession *mi.Session) error { // Collect sends the metric values for each metric // to the provided prometheus Metric channel. func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error { - errCh := make(chan error, len(c.collectorFns)) - errs := make([]error, 0, len(c.collectorFns)) - - wg := sync.WaitGroup{} + var g recovery.Group for _, fn := range c.collectorFns { - wg.Add(1) - - go func(fn func(ch chan<- prometheus.Metric) error) { - defer wg.Done() - - if err := fn(ch); err != nil { - errCh <- err - } - }(fn) + g.Go(func() error { + return fn(ch) + }) } - wg.Wait() - - close(errCh) - - for err := range errCh { - errs = append(errs, err) - } - - return errors.Join(errs...) + return g.Wait() } diff --git a/internal/collector/iis/iis.go b/internal/collector/iis/iis.go index e3d519cc..1921ec92 100644 --- a/internal/collector/iis/iis.go +++ b/internal/collector/iis/iis.go @@ -159,6 +159,7 @@ func (c *Collector) Close() error { c.perfDataCollectorHttpServiceRequestQueues.Close() c.perfDataCollectorAppPoolWAS.Close() c.w3SVCW3WPPerfDataCollector.Close() + c.w3SVCW3WPPerfDataCollectorV8.Close() c.serviceCachePerfDataCollector.Close() return nil diff --git a/internal/collector/iis/iis_w3svc_w3wp.go b/internal/collector/iis/iis_w3svc_w3wp.go index 790f923e..01a48a9a 100644 --- a/internal/collector/iis/iis_w3svc_w3wp.go +++ b/internal/collector/iis/iis_w3svc_w3wp.go @@ -408,7 +408,7 @@ func (c *Collector) collectW3SVCW3WP(ch chan<- prometheus.Metric) error { func (c *Collector) collectW3SVCW3WPv8(ch chan<- prometheus.Metric) error { err := c.w3SVCW3WPPerfDataCollectorV8.Collect(&c.perfDataObjectW3SVCW3WPV8) if err != nil { - return fmt.Errorf("failed to collect APP_POOL_WAS metrics: %w", err) + return fmt.Errorf("failed to collect W3SVC_W3WP IIS 8+ metrics: %w", err) } c.perfDataObjectW3SVCW3WPV8 = deduplicateIISNames(c.perfDataObjectW3SVCW3WPV8) @@ -502,7 +502,7 @@ func (c *Collector) collectW3SVCW3WPv8(ch chan<- prometheus.Metric) error { func (c *Collector) collectW3SVCW3WPv7(ch chan<- prometheus.Metric) error { err := c.w3SVCW3WPPerfDataCollector.Collect(&c.perfDataObjectW3SVCW3WP) if err != nil { - return fmt.Errorf("failed to collect APP_POOL_WAS metrics: %w", err) + return fmt.Errorf("failed to collect W3SVC_W3WP metrics: %w", err) } c.perfDataObjectW3SVCW3WP = deduplicateIISNames(c.perfDataObjectW3SVCW3WP) diff --git a/internal/collector/lifecycle_test.go b/internal/collector/lifecycle_test.go new file mode 100644 index 00000000..97c59b9f --- /dev/null +++ b/internal/collector/lifecycle_test.go @@ -0,0 +1,256 @@ +// 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 collector_test contains tests that apply to every collector. +package collector_test + +import ( + "log/slog" + "runtime" + "slices" + "strings" + "sync" + "testing" + "time" + + "github.com/alecthomas/kingpin/v2" + "github.com/prometheus-community/windows_exporter/internal/collector/dfsr" + "github.com/prometheus-community/windows_exporter/internal/collector/dhcp" + "github.com/prometheus-community/windows_exporter/internal/collector/dns" + "github.com/prometheus-community/windows_exporter/internal/collector/exchange" + "github.com/prometheus-community/windows_exporter/internal/collector/hyperv" + "github.com/prometheus-community/windows_exporter/internal/collector/logical_disk" + "github.com/prometheus-community/windows_exporter/internal/collector/mscluster" + "github.com/prometheus-community/windows_exporter/internal/collector/mssql" + "github.com/prometheus-community/windows_exporter/internal/collector/net" + "github.com/prometheus-community/windows_exporter/internal/collector/netframework" + "github.com/prometheus-community/windows_exporter/internal/collector/tcp" + "github.com/prometheus-community/windows_exporter/internal/collector/update" + "github.com/prometheus-community/windows_exporter/internal/mi" + "github.com/prometheus-community/windows_exporter/pkg/collector" + "github.com/prometheus/client_golang/prometheus" + "github.com/stretchr/testify/require" +) + +func newMISession(t *testing.T) *mi.Session { + t.Helper() + + app, err := mi.ApplicationInitialize() + require.NoError(t, err) + + t.Cleanup(func() { require.NoError(t, app.Close()) }) + + session, err := app.NewSession(nil) + require.NoError(t, err) + + t.Cleanup(func() { require.NoError(t, session.Close()) }) + + return session +} + +// TestBuildCloseCycle builds, collects and closes every collector repeatedly and +// fails if goroutines grow with every cycle. Grafana Alloy embeds the collectors +// and rebuilds them on every configuration reload, so a leak per cycle grows +// without bound there. Collectors whose Build fails on this host, e.g. because a +// role is not installed, are closed after the failed Build, which must release +// everything a partial Build allocated. The test does not run in parallel, +// because goroutine counts are process-wide. +func TestBuildCloseCycle(t *testing.T) { + const cycles = 5 + + logger := slog.New(slog.DiscardHandler) + miSession := newMISession(t) + + for _, name := range collector.Available() { + t.Run(name, func(t *testing.T) { + if name == update.Name { + t.Skip("the update worker finishes a running Windows Update search before it exits") + } + + cycle := func() { + app := kingpin.New("windows_exporter", "") + c := collector.BuildersWithFlags[name](app) + + _, err := app.Parse(nil) + require.NoError(t, err) + + if err := c.Build(logger, miSession); err == nil { + collect(t, c) + } + + require.NoError(t, c.Close()) + } + + // The first cycle loads DLLs and may start goroutines that live as long as the process. + cycle() + + before := runtime.NumGoroutine() + + for range cycles { + cycle() + } + + // Closed PDH collectors stop their worker goroutines asynchronously. + deadline := time.Now().Add(10 * time.Second) + + for runtime.NumGoroutine()-before >= cycles && time.Now().Before(deadline) { + time.Sleep(50 * time.Millisecond) + } + + require.Less(t, runtime.NumGoroutine()-before, cycles, + "goroutines grew with every Build/Close cycle of collector %s", name, + ) + }) + } +} + +func collect(t *testing.T, c collector.Collector) { + t.Helper() + + ch := make(chan prometheus.Metric) + + var wg sync.WaitGroup + + wg.Go(func() { + for range ch { + } + }) + + // Errors are expected for collectors without data on this host. + _ = c.Collect(ch, 10*time.Second) + + close(ch) + wg.Wait() +} + +func TestBuildRejectsUnknownSubCollector(t *testing.T) { + t.Parallel() + + logger := slog.New(slog.DiscardHandler) + miSession := newMISession(t) + unknown := []string{"does_not_exist"} + + for _, c := range []collector.Collector{ + dfsr.New(&dfsr.Config{CollectorsEnabled: unknown}), + dhcp.New(&dhcp.Config{CollectorsEnabled: unknown}), + dns.New(&dns.Config{CollectorsEnabled: unknown}), + exchange.New(&exchange.Config{CollectorsEnabled: unknown}), + hyperv.New(&hyperv.Config{CollectorsEnabled: unknown}), + logical_disk.New(&logical_disk.Config{CollectorsEnabled: unknown}), + mscluster.New(&mscluster.Config{CollectorsEnabled: unknown}), + mssql.New(&mssql.Config{CollectorsEnabled: unknown}), + net.New(&net.Config{CollectorsEnabled: unknown}), + netframework.New(&netframework.Config{CollectorsEnabled: unknown}), + tcp.New(&tcp.Config{CollectorsEnabled: unknown}), + } { + t.Run(c.GetName(), func(t *testing.T) { + t.Parallel() + + err := c.Build(logger, miSession) + require.ErrorContains(t, err, "unknown sub collector: does_not_exist. Possible values: ") + require.NoError(t, c.Close()) + }) + } +} + +// Build must not sort the CollectorsEnabled slice of the caller in place. +// It may be the slice of the exported ConfigDefaults. +func TestBuildDoesNotModifyCollectorsEnabled(t *testing.T) { + t.Parallel() + + logger := slog.New(slog.DiscardHandler) + miSession := newMISession(t) + + for _, tc := range []struct { + enabled []string + new func(enabled []string) collector.Collector + }{ + { + enabled: []string{"Autodiscover", "ActiveSync"}, + new: func(enabled []string) collector.Collector { + return exchange.New(&exchange.Config{CollectorsEnabled: enabled}) + }, + }, + { + enabled: []string{"host", "datastore"}, + new: func(enabled []string) collector.Collector { + return hyperv.New(&hyperv.Config{CollectorsEnabled: enabled}) + }, + }, + { + enabled: []string{"locks", "info"}, + new: func(enabled []string) collector.Collector { + return mssql.New(&mssql.Config{CollectorsEnabled: enabled}) + }, + }, + { + enabled: []string{"clrmemory", "clrexceptions"}, + new: func(enabled []string) collector.Collector { + return netframework.New(&netframework.Config{CollectorsEnabled: enabled}) + }, + }, + } { + c := tc.new(tc.enabled) + + t.Run(c.GetName(), func(t *testing.T) { + t.Parallel() + + want := slices.Clone(tc.enabled) + + // Build errors are expected if the role is not installed. + _ = c.Build(logger, miSession) + + require.NoError(t, c.Close()) + require.Equal(t, want, tc.enabled) + }) + } +} + +// New must apply the default sub collectors if CollectorsEnabled is nil, +// otherwise library users that don't set it get no metrics. +func TestNewDefaultsCollectorsEnabled(t *testing.T) { + t.Parallel() + + c := logical_disk.New(&logical_disk.Config{}) + + require.NoError(t, c.Build(slog.New(slog.DiscardHandler), nil)) + + t.Cleanup(func() { require.NoError(t, c.Close()) }) + + ch := make(chan prometheus.Metric) + + var ( + wg sync.WaitGroup + descs []string + ) + + wg.Go(func() { + for metric := range ch { + descs = append(descs, metric.Desc().String()) + } + }) + + err := c.Collect(ch, 10*time.Second) + + close(ch) + wg.Wait() + + require.NoError(t, err) + require.True(t, slices.ContainsFunc(descs, func(desc string) bool { + return strings.Contains(desc, `fqName: "windows_logical_disk_read_bytes_total"`) + }), "metrics sub collector is not enabled by default") +} diff --git a/internal/collector/logical_disk/logical_disk.go b/internal/collector/logical_disk/logical_disk.go index 65362e57..9599791a 100644 --- a/internal/collector/logical_disk/logical_disk.go +++ b/internal/collector/logical_disk/logical_disk.go @@ -132,6 +132,10 @@ func New(config *Config) *Collector { config.VolumeInclude = ConfigDefaults.VolumeInclude } + if config.CollectorsEnabled == nil { + config.CollectorsEnabled = ConfigDefaults.CollectorsEnabled + } + c := &Collector{ config: *config, } @@ -191,6 +195,8 @@ func (c *Collector) GetName() string { } func (c *Collector) Close() error { + c.perfDataCollector.Close() + return nil } diff --git a/internal/collector/memory/memory.go b/internal/collector/memory/memory.go index d9c63938..d76283c2 100644 --- a/internal/collector/memory/memory.go +++ b/internal/collector/memory/memory.go @@ -109,6 +109,8 @@ func (c *Collector) GetName() string { } func (c *Collector) Close() error { + c.perfDataCollector.Close() + return nil } diff --git a/internal/collector/mscluster/mscluster.go b/internal/collector/mscluster/mscluster.go index d1c0f3f9..d16e92a0 100644 --- a/internal/collector/mscluster/mscluster.go +++ b/internal/collector/mscluster/mscluster.go @@ -23,11 +23,11 @@ import ( "log/slog" "slices" "strings" - "sync" "time" "github.com/alecthomas/kingpin/v2" "github.com/prometheus-community/windows_exporter/internal/mi" + "github.com/prometheus-community/windows_exporter/internal/utils/recovery" "github.com/prometheus/client_golang/prometheus" ) @@ -128,6 +128,25 @@ func (c *Collector) Build(_ *slog.Logger, miSession *mi.Session) error { return nil } + subCollectors := []string{ + subCollectorCluster, + subCollectorNetwork, + subCollectorNode, + subCollectorResource, + subCollectorResourceGroup, + subCollectorSharedVolumes, + subCollectorVirtualDisk, + subCollectorStoragePool, + } + + for _, name := range c.config.CollectorsEnabled { + if !slices.Contains(subCollectors, name) { + return fmt.Errorf("unknown sub collector: %s. Possible values: %s", name, + strings.Join(subCollectors, ", "), + ) + } + } + if miSession == nil { return errors.New("miSession is nil") } @@ -194,104 +213,94 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, maxScrapeDuration time. return nil } - errCh := make(chan error, 8) + var g recovery.Group - wg := sync.WaitGroup{} - wg.Add(8) - - go func() { - defer wg.Done() - - if slices.Contains(c.config.CollectorsEnabled, subCollectorCluster) { + if slices.Contains(c.config.CollectorsEnabled, subCollectorCluster) { + g.Go(func() error { if err := c.collectCluster(ch, maxScrapeDuration); err != nil { - errCh <- fmt.Errorf("failed to collect cluster metrics: %w", err) + return fmt.Errorf("failed to collect cluster metrics: %w", err) } - } - }() - go func() { - defer wg.Done() - - if slices.Contains(c.config.CollectorsEnabled, subCollectorNetwork) { - if err := c.collectNetwork(ch, maxScrapeDuration); err != nil { - errCh <- fmt.Errorf("failed to collect network metrics: %w", err) - } - } - }() - - go func() { - defer wg.Done() - - nodeNames := make([]string, 0) - - if slices.Contains(c.config.CollectorsEnabled, subCollectorNode) { - var err error - - nodeNames, err = c.collectNode(ch, maxScrapeDuration) - if err != nil { - errCh <- fmt.Errorf("failed to collect node metrics: %w", err) - } - } - - go func() { - defer wg.Done() - - if slices.Contains(c.config.CollectorsEnabled, subCollectorResource) { - if err := c.collectResource(ch, maxScrapeDuration, nodeNames); err != nil { - errCh <- fmt.Errorf("failed to collect resource metrics: %w", err) - } - } - }() - - go func() { - defer wg.Done() - - if slices.Contains(c.config.CollectorsEnabled, subCollectorResourceGroup) { - if err := c.collectResourceGroup(ch, maxScrapeDuration, nodeNames); err != nil { - errCh <- fmt.Errorf("failed to collect resource group metrics: %w", err) - } - } - }() - }() - - go func() { - defer wg.Done() - - if slices.Contains(c.config.CollectorsEnabled, subCollectorSharedVolumes) { - if err := c.collectSharedVolumes(ch, maxScrapeDuration); err != nil { - errCh <- fmt.Errorf("failed to collect shared_volumes metrics: %w", err) - } - } - }() - - go func() { - defer wg.Done() - - if slices.Contains(c.config.CollectorsEnabled, subCollectorVirtualDisk) { - if err := c.collectVirtualDisk(ch, maxScrapeDuration); err != nil { - errCh <- fmt.Errorf("failed to collect virtualdisk metrics: %w", err) - } - } - }() - - go func() { - defer wg.Done() - - if slices.Contains(c.config.CollectorsEnabled, subCollectorStoragePool) { - if err := c.collectStoragePool(ch, maxScrapeDuration); err != nil { - errCh <- fmt.Errorf("failed to collect storagepool metrics: %w", err) - } - } - }() - - wg.Wait() - close(errCh) - - errs := make([]error, 0) - - for err := range errCh { - errs = append(errs, err) + return nil + }) } - return errors.Join(errs...) + if slices.Contains(c.config.CollectorsEnabled, subCollectorNetwork) { + g.Go(func() error { + if err := c.collectNetwork(ch, maxScrapeDuration); err != nil { + return fmt.Errorf("failed to collect network metrics: %w", err) + } + + return nil + }) + } + + // The resource and resource group collectors need the node names. + g.Go(func() error { + var ( + nodeNames []string + err error + ) + + if slices.Contains(c.config.CollectorsEnabled, subCollectorNode) { + nodeNames, err = c.collectNode(ch, maxScrapeDuration) + if err != nil { + err = fmt.Errorf("failed to collect node metrics: %w", err) + } + } + + if slices.Contains(c.config.CollectorsEnabled, subCollectorResource) { + g.Go(func() error { + if err := c.collectResource(ch, maxScrapeDuration, nodeNames); err != nil { + return fmt.Errorf("failed to collect resource metrics: %w", err) + } + + return nil + }) + } + + if slices.Contains(c.config.CollectorsEnabled, subCollectorResourceGroup) { + g.Go(func() error { + if err := c.collectResourceGroup(ch, maxScrapeDuration, nodeNames); err != nil { + return fmt.Errorf("failed to collect resource group metrics: %w", err) + } + + return nil + }) + } + + return err + }) + + if slices.Contains(c.config.CollectorsEnabled, subCollectorSharedVolumes) { + g.Go(func() error { + if err := c.collectSharedVolumes(ch, maxScrapeDuration); err != nil { + return fmt.Errorf("failed to collect shared_volumes metrics: %w", err) + } + + return nil + }) + } + + if slices.Contains(c.config.CollectorsEnabled, subCollectorVirtualDisk) { + g.Go(func() error { + if err := c.collectVirtualDisk(ch, maxScrapeDuration); err != nil { + return fmt.Errorf("failed to collect virtualdisk metrics: %w", err) + } + + return nil + }) + } + + if slices.Contains(c.config.CollectorsEnabled, subCollectorStoragePool) { + g.Go(func() error { + if err := c.collectStoragePool(ch, maxScrapeDuration); err != nil { + return fmt.Errorf("failed to collect storagepool metrics: %w", err) + } + + return nil + }) + } + + return g.Wait() } diff --git a/internal/collector/msmq/msmq.go b/internal/collector/msmq/msmq.go index 384d4bcd..8c7b319c 100644 --- a/internal/collector/msmq/msmq.go +++ b/internal/collector/msmq/msmq.go @@ -73,6 +73,8 @@ func (c *Collector) GetName() string { } func (c *Collector) Close() error { + c.perfDataCollector.Close() + return nil } diff --git a/internal/collector/mssql/mssql.go b/internal/collector/mssql/mssql.go index 250b5fbd..44dc6e93 100644 --- a/internal/collector/mssql/mssql.go +++ b/internal/collector/mssql/mssql.go @@ -22,15 +22,16 @@ import ( "errors" "fmt" "log/slog" - "sort" + "maps" + "slices" "strings" - "sync" "time" "github.com/alecthomas/kingpin/v2" "github.com/prometheus-community/windows_exporter/internal/mi" "github.com/prometheus-community/windows_exporter/internal/pdh" "github.com/prometheus-community/windows_exporter/internal/types" + "github.com/prometheus-community/windows_exporter/internal/utils/recovery" "github.com/prometheus/client_golang/prometheus" "golang.org/x/sys/windows/registry" ) @@ -151,19 +152,14 @@ func (c *Collector) Close() error { fn() } + c.closeFns = nil + return nil } func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { c.logger = logger.With(slog.String("collector", Name)) - instances, err := c.getMSSQLInstances() - if err != nil { - return fmt.Errorf("couldn't get SQL instances: %w", err) - } - - c.mssqlInstances = instances - subCollectors := map[string]struct { build func() error collect func(ch chan<- prometheus.Metric) error @@ -236,24 +232,38 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { }, } - c.collectorFns = make([]func(ch chan<- prometheus.Metric) error, 0, len(c.config.CollectorsEnabled)) - c.closeFns = make([]func(), 0, len(c.config.CollectorsEnabled)) + // Sort a copy, to not modify the slice of the caller or ConfigDefaults. // Result must order, to prevent test failures. - sort.Strings(c.config.CollectorsEnabled) + collectorsEnabled := slices.Compact(slices.Sorted(slices.Values(c.config.CollectorsEnabled))) - errs := make([]error, 0, len(c.config.CollectorsEnabled)) - - for _, name := range c.config.CollectorsEnabled { + for _, name := range collectorsEnabled { if _, ok := subCollectors[name]; !ok { - return fmt.Errorf("unknown collector: %s", name) + return fmt.Errorf("unknown sub collector: %s. Possible values: %s", name, + strings.Join(slices.Sorted(maps.Keys(subCollectors)), ", "), + ) } + } + + instances, err := c.getMSSQLInstances() + if err != nil { + return fmt.Errorf("couldn't get SQL instances: %w", err) + } + + c.mssqlInstances = instances + c.collectorFns = make([]func(ch chan<- prometheus.Metric) error, 0, len(collectorsEnabled)) + c.closeFns = make([]func(), 0, len(collectorsEnabled)) + + errs := make([]error, 0, len(collectorsEnabled)) + + for _, name := range collectorsEnabled { + // Register close before build, so that Close also releases a partially built sub collector. + c.closeFns = append(c.closeFns, subCollectors[name].close) if err := subCollectors[name].build(); err != nil { errs = append(errs, fmt.Errorf("failed to build %s collector: %w", name, err)) } c.collectorFns = append(c.collectorFns, subCollectors[name].collect) - c.closeFns = append(c.closeFns, subCollectors[name].close) } c.mssqlScrapeDurationDesc = prometheus.NewDesc( @@ -279,31 +289,15 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error return fmt.Errorf("no SQL instances found: %w", pdh.ErrNoData) } - errCh := make(chan error, len(c.collectorFns)) - errs := make([]error, 0, len(c.collectorFns)) + var g recovery.Group - wg := sync.WaitGroup{} for _, fn := range c.collectorFns { - wg.Add(1) - - go func(fn func(ch chan<- prometheus.Metric) error) { - defer wg.Done() - - if err := fn(ch); err != nil { - errCh <- err - } - }(fn) + g.Go(func() error { + return fn(ch) + }) } - wg.Wait() - - close(errCh) - - for err := range errCh { - errs = append(errs, err) - } - - return errors.Join(errs...) + return g.Wait() } func (c *Collector) getMSSQLInstances() ([]mssqlInstance, error) { diff --git a/internal/collector/mssql/mssql_database.go b/internal/collector/mssql/mssql_database.go index 51612d96..04382099 100644 --- a/internal/collector/mssql/mssql_database.go +++ b/internal/collector/mssql/mssql_database.go @@ -822,4 +822,8 @@ func (c *Collector) closeDatabases() { for _, collector := range c.databasesPerfDataCollectors { collector.Close() } + + for _, collector := range c.databasesPerfDataCollectors2019 { + collector.Close() + } } diff --git a/internal/collector/netframework/netframework.go b/internal/collector/netframework/netframework.go index 7c1a7fd4..3940c94b 100644 --- a/internal/collector/netframework/netframework.go +++ b/internal/collector/netframework/netframework.go @@ -21,13 +21,14 @@ import ( "errors" "fmt" "log/slog" - "sort" + "maps" + "slices" "strings" - "sync" "time" "github.com/alecthomas/kingpin/v2" "github.com/prometheus-community/windows_exporter/internal/mi" + "github.com/prometheus-community/windows_exporter/internal/utils/recovery" "github.com/prometheus/client_golang/prometheus" ) @@ -140,6 +141,10 @@ func New(config *Config) *Collector { config = &ConfigDefaults } + if config.CollectorsEnabled == nil { + config.CollectorsEnabled = ConfigDefaults.CollectorsEnabled + } + c := &Collector{ config: *config, } @@ -229,14 +234,19 @@ func (c *Collector) Build(_ *slog.Logger, miSession *mi.Session) error { }, } + // Sort a copy, to not modify the slice of the caller or ConfigDefaults. // Result must order, to prevent test failures. - sort.Strings(c.config.CollectorsEnabled) + collectorsEnabled := slices.Compact(slices.Sorted(slices.Values(c.config.CollectorsEnabled))) - for _, name := range c.config.CollectorsEnabled { + for _, name := range collectorsEnabled { if _, ok := subCollectors[name]; !ok { - return fmt.Errorf("unknown collector: %s", name) + return fmt.Errorf("unknown sub collector: %s. Possible values: %s", name, + strings.Join(slices.Sorted(maps.Keys(subCollectors)), ", "), + ) } + } + for _, name := range collectorsEnabled { subCollectors[name].build() c.collectorFns = append(c.collectorFns, subCollectors[name].collect) @@ -248,30 +258,13 @@ func (c *Collector) Build(_ *slog.Logger, miSession *mi.Session) error { // Collect sends the metric values for each metric // to the provided prometheus Metric channel. func (c *Collector) Collect(ch chan<- prometheus.Metric, maxScrapeDuration time.Duration) error { - errCh := make(chan error, len(c.collectorFns)) - errs := make([]error, 0, len(c.collectorFns)) - - wg := sync.WaitGroup{} + var g recovery.Group for _, fn := range c.collectorFns { - wg.Add(1) - - go func(fn func(ch chan<- prometheus.Metric, maxScrapeDuration time.Duration) error) { - defer wg.Done() - - if err := fn(ch, maxScrapeDuration); err != nil { - errCh <- err - } - }(fn) + g.Go(func() error { + return fn(ch, maxScrapeDuration) + }) } - wg.Wait() - - close(errCh) - - for err := range errCh { - errs = append(errs, err) - } - - return errors.Join(errs...) + return g.Wait() } diff --git a/internal/collector/nps/nps.go b/internal/collector/nps/nps.go index 26757340..7107a507 100644 --- a/internal/collector/nps/nps.go +++ b/internal/collector/nps/nps.go @@ -93,6 +93,9 @@ func (c *Collector) GetName() string { } func (c *Collector) Close() error { + c.accessPerfDataCollector.Close() + c.accountingPerfDataCollector.Close() + return nil } diff --git a/internal/collector/physical_disk/physical_disk.go b/internal/collector/physical_disk/physical_disk.go index c878f2f9..2c139482 100644 --- a/internal/collector/physical_disk/physical_disk.go +++ b/internal/collector/physical_disk/physical_disk.go @@ -126,6 +126,8 @@ func (c *Collector) GetName() string { } func (c *Collector) Close() error { + c.perfDataCollector.Close() + return nil } diff --git a/internal/collector/process/process.go b/internal/collector/process/process.go index 454a98aa..255d7b09 100644 --- a/internal/collector/process/process.go +++ b/internal/collector/process/process.go @@ -23,6 +23,7 @@ import ( "fmt" "log/slog" "regexp" + "slices" "strconv" "strings" "sync" @@ -171,7 +172,9 @@ func (c *Collector) Close() error { c.mu.Lock() defer c.mu.Unlock() - c.perfDataCollector.Close() + if c.perfDataCollector != nil { + c.perfDataCollector.Close() + } if c.workerCh != nil { close(c.workerCh) @@ -196,6 +199,9 @@ func (c *Collector) Build(logger *slog.Logger, miSession *mi.Session) error { c.config.CounterVersion = 2 if errors.Is(err, pdh.NewPdhError(pdh.CstatusNoObject)) { + // NewCollector returns a partially built collector on errors. Release it before falling back to V1. + c.perfDataCollector.Close() + c.perfDataCollector, err = registry.NewCollector[perfDataCounterValues]("Process", pdh.InstancesAll) c.config.CounterVersion = 1 } @@ -216,7 +222,9 @@ func (c *Collector) Build(logger *slog.Logger, miSession *mi.Session) error { c.mu = sync.RWMutex{} c.lookupCache = sync.Map{} - if c.config.ProcessInclude.String() == "^(?:.*)$" && c.config.ProcessExclude.String() == "^(?:)$" { + // The flags wrap the expressions in ^(?:...)$, the defaults of [New] are [types.RegExpAny] and [types.RegExpEmpty]. + if slices.Contains([]string{"^(?:.+)$", "^(?:.*)$", types.RegExpAny.String()}, c.config.ProcessInclude.String()) && + slices.Contains([]string{"^(?:)$", types.RegExpEmpty.String()}, c.config.ProcessExclude.String()) { logger.Warn("No filters specified for process collector. This will generate a very large number of metrics!") } diff --git a/internal/collector/scheduled_task/scheduled_task.go b/internal/collector/scheduled_task/scheduled_task.go index 9f5dc41e..69965860 100644 --- a/internal/collector/scheduled_task/scheduled_task.go +++ b/internal/collector/scheduled_task/scheduled_task.go @@ -49,6 +49,7 @@ var ConfigDefaults = Config{ type Collector struct { config Config + logger *slog.Logger lastResult *prometheus.Desc lastResultStatus *prometheus.Desc @@ -161,7 +162,9 @@ func (c *Collector) Close() error { return nil } -func (c *Collector) Build(_ *slog.Logger, _ *mi.Session) error { +func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { + c.logger = logger.With(slog.String("collector", Name)) + c.lastResult = prometheus.NewDesc( prometheus.BuildFQName(types.Namespace, Name, "last_result"), "DEPRECATED: use windows_scheduled_task_last_result_status. "+ @@ -209,7 +212,12 @@ var TASK_RESULT_STATUSES = []string{ func (c *Collector) collect(ch chan<- prometheus.Metric) error { scheduledTasks, err := getScheduledTasks() - if err != nil { + 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", + slog.Any("err", err), + ) + } else if err != nil { return fmt.Errorf("get scheduled tasks: %w", err) } @@ -297,6 +305,9 @@ var taskSchedulerCLSID = ole.GUID{ // S_FALSE is returned by CoInitialize if it was already called on this thread. const S_FALSE = 0x00000001 +// errTasksSkipped is returned with the readable tasks if some tasks or folders could not be read. +var errTasksSkipped = errors.New("tasks skipped") + func getScheduledTasks() (ScheduledTasks, error) { var scheduledTasks ScheduledTasks @@ -322,15 +333,18 @@ func getScheduledTasks() (ScheduledTasks, error) { } defer taskSchedulerObj.Release() - taskServiceObj := taskSchedulerObj.MustQueryInterface(ole.IID_IDispatch) + taskServiceObj, err := taskSchedulerObj.QueryInterface(ole.IID_IDispatch) + if err != nil { + return scheduledTasks, fmt.Errorf("IID_IDispatch: %w", err) + } + + defer taskServiceObj.Release() _, err = oleutil.CallMethod(taskServiceObj, "Connect") if err != nil { return scheduledTasks, err } - defer taskServiceObj.Release() - res, err := oleutil.CallMethod(taskServiceObj, "GetFolder", `\`) if err != nil { return scheduledTasks, err @@ -339,45 +353,61 @@ func getScheduledTasks() (ScheduledTasks, error) { rootFolderObj := res.ToIDispatch() defer rootFolderObj.Release() - err = fetchTasksRecursively(rootFolderObj, &scheduledTasks) + if err = fetchTasksRecursively(rootFolderObj, `\`, &scheduledTasks); err != nil { + return scheduledTasks, fmt.Errorf("%w: %w", errTasksSkipped, err) + } - return scheduledTasks, err + return scheduledTasks, nil } +// fetchTasksInFolder appends the tasks of folder to scheduledTasks. +// A task that can't be read is skipped and its error is returned after all other tasks are read. func fetchTasksInFolder(folder *ole.IDispatch, scheduledTasks *ScheduledTasks) error { res, err := oleutil.CallMethod(folder, "GetTasks", 1) if err != nil { - return err + return fmt.Errorf("get tasks: %w", err) } tasks := res.ToIDispatch() defer tasks.Release() + errs := make([]error, 0) + err = oleutil.ForEach(tasks, func(v *ole.VARIANT) error { task := v.ToIDispatch() defer task.Release() parsedTask, err := parseTask(task) if err != nil { - return err + errs = append(errs, fmt.Errorf("parse task: %w", err)) + + return nil } *scheduledTasks = append(*scheduledTasks, parsedTask) return nil }) + if err != nil { + errs = append(errs, fmt.Errorf("enumerate tasks: %w", err)) + } - return err + return errors.Join(errs...) } -func fetchTasksRecursively(folder *ole.IDispatch, scheduledTasks *ScheduledTasks) error { +// fetchTasksRecursively appends the tasks of folder and its sub folders to scheduledTasks. +// A folder or task that can't be read is skipped and its error is returned after all other +// folders are read. +func fetchTasksRecursively(folder *ole.IDispatch, folderPath string, scheduledTasks *ScheduledTasks) error { + errs := make([]error, 0) + if err := fetchTasksInFolder(folder, scheduledTasks); err != nil { - return err + errs = append(errs, fmt.Errorf("folder %s: %w", folderPath, err)) } res, err := oleutil.CallMethod(folder, "GetFolders", 1) if err != nil { - return err + return errors.Join(append(errs, fmt.Errorf("folder %s: get sub folders: %w", folderPath, err))...) } subFolders := res.ToIDispatch() @@ -387,10 +417,24 @@ func fetchTasksRecursively(folder *ole.IDispatch, scheduledTasks *ScheduledTasks subFolder := v.ToIDispatch() defer subFolder.Release() - return fetchTasksRecursively(subFolder, scheduledTasks) - }) + subFolderPath := folderPath - return err + if pathVar, err := oleutil.GetProperty(subFolder, "Path"); err == nil { + subFolderPath = pathVar.ToString() + _ = pathVar.Clear() + } + + if err := fetchTasksRecursively(subFolder, subFolderPath, scheduledTasks); err != nil { + errs = append(errs, err) + } + + return nil + }) + if err != nil { + errs = append(errs, fmt.Errorf("folder %s: enumerate sub folders: %w", folderPath, err)) + } + + return errors.Join(errs...) } func parseTask(task *ole.IDispatch) (ScheduledTask, error) { diff --git a/internal/collector/scheduled_task/scheduled_task_internal_test.go b/internal/collector/scheduled_task/scheduled_task_internal_test.go index 4bba5f92..90e69a73 100644 --- a/internal/collector/scheduled_task/scheduled_task_internal_test.go +++ b/internal/collector/scheduled_task/scheduled_task_internal_test.go @@ -18,6 +18,7 @@ package scheduled_task import ( + "log/slog" "regexp" "testing" @@ -274,7 +275,7 @@ func gatherTaskMetrics(t *testing.T, config *Config, tasks ScheduledTasks) map[s t.Helper() c := New(config) - require.NoError(t, c.Build(nil, nil)) + require.NoError(t, c.Build(slog.New(slog.DiscardHandler), nil)) registry := prometheus.NewPedanticRegistry() require.NoError(t, registry.Register(taskMetricCollector{collector: c, tasks: tasks})) diff --git a/internal/collector/service/service.go b/internal/collector/service/service.go index 7f9dfeca..8a2b47c1 100644 --- a/internal/collector/service/service.go +++ b/internal/collector/service/service.go @@ -34,6 +34,7 @@ import ( "github.com/alecthomas/kingpin/v2" "github.com/prometheus-community/windows_exporter/internal/mi" "github.com/prometheus-community/windows_exporter/internal/types" + "github.com/prometheus-community/windows_exporter/internal/utils/recovery" "github.com/prometheus/client_golang/prometheus" "golang.org/x/sys/windows" "golang.org/x/sys/windows/svc/mgr" @@ -73,7 +74,19 @@ type Collector struct { serviceConfigPoolBytes sync.Pool serviceManagerHandle *mgr.Mgr + queryAllServicesMu sync.Mutex queryAllServicesBuffer []byte + + // allStartModesIncluded is true if ServiceStartModeInclude contains every start mode. + // Only then services with an unknown start mode are collected. + allStartModesIncluded bool +} + +// serviceEntry is a copy of an [windows.ENUM_SERVICE_STATUS_PROCESS], which +// does not point into the buffer of queryAllServices. +type serviceEntry struct { + name string + status windows.SERVICE_STATUS_PROCESS } func New(config *Config) *Collector { @@ -202,32 +215,48 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { windows.SERVICE_SYSTEM_START: "system", } + startModes := slices.Sorted(maps.Values(c.apiStartModeValues)) + + for _, startMode := range c.config.ServiceStartModeInclude { + if !slices.Contains(startModes, startMode) { + return fmt.Errorf("unknown start mode: %s. Possible values: %s", startMode, + strings.Join(startModes, ", "), + ) + } + } + + c.allStartModesIncluded = true + + for _, startMode := range startModes { + if !slices.Contains(c.config.ServiceStartModeInclude, startMode) { + c.allStartModesIncluded = false + } + } + // EnumServiceStatusEx requires only SC_MANAGER_ENUM_SERVICE. handle, err := windows.OpenSCManager(nil, nil, windows.SC_MANAGER_ENUMERATE_SERVICE) if err != nil { return fmt.Errorf("failed to open scm: %w", err) } - for _, startMode := range c.config.ServiceStartModeInclude { - if !slices.Contains(slices.Collect(maps.Values(c.apiStartModeValues)), startMode) { - return fmt.Errorf("unknown start mode: %s. Possible values: %s", startMode, - strings.Join(slices.Collect(maps.Values(c.apiStartModeValues)), ", "), - ) - } - } - c.serviceManagerHandle = &mgr.Mgr{Handle: handle} return nil } func (c *Collector) Close() error { + if c.serviceManagerHandle == nil { + return nil + } + if err := c.serviceManagerHandle.Disconnect(); err != nil { c.logger.Warn("Failed to disconnect from scm", slog.Any("err", err), ) } + c.serviceManagerHandle = nil + return nil } @@ -239,18 +268,7 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error return fmt.Errorf("failed to query all services: %w", err) } - servicesCh := make(chan windows.ENUM_SERVICE_STATUS_PROCESS, len(services)) - wg := sync.WaitGroup{} - wg.Add(len(services)) - - for range 4 { - go func(ch chan<- prometheus.Metric, wg *sync.WaitGroup) { - for service := range servicesCh { - c.collectWorker(ch, service) - wg.Done() - } - }(ch, &wg) - } + servicesCh := make(chan serviceEntry, len(services)) for _, service := range services { servicesCh <- service @@ -258,38 +276,45 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error close(servicesCh) - wg.Wait() + // If a worker panics, the remaining workers collect the remaining services. + var g recovery.Group - return nil + for range 4 { + g.Go(func() error { + for service := range servicesCh { + c.collectWorker(ch, service) + } + + return nil + }) + } + + return g.Wait() } -func (c *Collector) collectWorker(ch chan<- prometheus.Metric, service windows.ENUM_SERVICE_STATUS_PROCESS) { - if uintptr(unsafe.Pointer(service.ServiceName)) == uintptr(windows.InvalidHandle) { - c.logger.Log(context.Background(), slog.LevelWarn, "failed collecting service info", - slog.String("err", "ServiceName is 0xffffffffffffffff"), - slog.String("service", fmt.Sprintf("%+v", service)), - ) - +func (c *Collector) collectWorker(ch chan<- prometheus.Metric, service serviceEntry) { + if c.config.ServiceExclude.MatchString(service.name) || !c.config.ServiceInclude.MatchString(service.name) { return } - serviceName := windows.UTF16PtrToString(service.ServiceName) - - if c.config.ServiceExclude.MatchString(serviceName) || !c.config.ServiceInclude.MatchString(serviceName) { - return - } - - if err := c.collectService(ch, serviceName, service); err != nil { + if err := c.collectService(ch, service); err != nil { c.logger.Log(context.Background(), slog.LevelWarn, "failed collecting service info", slog.Any("err", err), - slog.String("service", serviceName), + slog.String("service", service.name), ) } } -func (c *Collector) collectService(ch chan<- prometheus.Metric, serviceName string, service windows.ENUM_SERVICE_STATUS_PROCESS) error { +func (c *Collector) collectService(ch chan<- prometheus.Metric, service serviceEntry) error { + serviceName := service.name + + serviceNamePtr, err := windows.UTF16PtrFromString(serviceName) + if err != nil { + return fmt.Errorf("invalid service name: %w", err) + } + // Open connection for service handler. - serviceHandle, err := windows.OpenService(c.serviceManagerHandle.Handle, service.ServiceName, windows.SERVICE_QUERY_CONFIG) + serviceHandle, err := windows.OpenService(c.serviceManagerHandle.Handle, serviceNamePtr, windows.SERVICE_QUERY_CONFIG) if err != nil { return fmt.Errorf("failed to open service: %w", err) } @@ -307,6 +332,8 @@ func (c *Collector) collectService(ch chan<- prometheus.Metric, serviceName stri // 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) @@ -318,20 +345,35 @@ func (c *Collector) collectService(ch chan<- prometheus.Metric, serviceName stri ) } - serviceStartMode, ok := c.apiStartModeValues[serviceConfig.StartType] - if !ok { - c.logger.Log(context.Background(), slog.LevelWarn, "unknown service start mode", - slog.String("service", serviceName), - slog.Uint64("start_mode", uint64(serviceConfig.StartType)), - ) + // serviceStartMode stays empty if the service configuration is not available. + // The zero StartType of the empty configuration must not be reported as "boot". + var serviceStartMode string - return nil - } + if configAvailable { + var ok bool - if !slices.Contains(c.config.ServiceStartModeInclude, serviceStartMode) { - c.logger.Log(context.Background(), slog.LevelDebug, "service start mode excluded by config", + serviceStartMode, ok = c.apiStartModeValues[serviceConfig.StartType] + if !ok { + c.logger.Log(context.Background(), slog.LevelWarn, "unknown service start mode", + slog.String("service", serviceName), + slog.Uint64("start_mode", uint64(serviceConfig.StartType)), + ) + + return nil + } + + if !slices.Contains(c.config.ServiceStartModeInclude, serviceStartMode) { + c.logger.Log(context.Background(), slog.LevelDebug, "service start mode excluded by config", + slog.String("service", serviceName), + slog.String("start_mode", serviceStartMode), + ) + + return nil + } + } else if !c.allStartModesIncluded { + // A service with an unknown start mode can't match a start mode filter. + c.logger.Log(context.Background(), slog.LevelDebug, "service with unknown start mode excluded by config", slog.String("service", serviceName), - slog.String("start_mode", serviceStartMode), ) return nil @@ -352,24 +394,26 @@ func (c *Collector) collectService(ch chan<- prometheus.Metric, serviceName stri isCurrentState float64 ) - for _, startMode := range c.apiStartModeValues { - isCurrentStartMode = 0.0 - if startMode == serviceStartMode { - isCurrentStartMode = 1.0 - } + if serviceStartMode != "" { + for _, startMode := range c.apiStartModeValues { + isCurrentStartMode = 0.0 + if startMode == serviceStartMode { + isCurrentStartMode = 1.0 + } - ch <- prometheus.MustNewConstMetric( - c.startMode, - prometheus.GaugeValue, - isCurrentStartMode, - serviceName, - startMode, - ) + ch <- prometheus.MustNewConstMetric( + c.startMode, + prometheus.GaugeValue, + isCurrentStartMode, + serviceName, + startMode, + ) + } } for state, stateValue := range c.apiStateValues { isCurrentState = 0.0 - if state == service.ServiceStatusProcess.CurrentState { + if state == service.status.CurrentState { isCurrentState = 1.0 } @@ -382,13 +426,13 @@ func (c *Collector) collectService(ch chan<- prometheus.Metric, serviceName stri ) } - if service.ServiceStatusProcess.ProcessId == 0 { + if service.status.ProcessId == 0 { return nil } - processID := strconv.FormatUint(uint64(service.ServiceStatusProcess.ProcessId), 10) + processID := strconv.FormatUint(uint64(service.status.ProcessId), 10) - processStartTime, err := c.getProcessStartTime(service.ServiceStatusProcess.ProcessId) + processStartTime, err := c.getProcessStartTime(service.status.ProcessId) if err == nil { ch <- prometheus.MustNewConstMetric( c.processID, @@ -419,13 +463,18 @@ func (c *Collector) collectService(ch chan<- prometheus.Metric, serviceName stri // queryAllServices returns all service states of the current Windows system // This is realized by ask Service Manager directly. -func (c *Collector) queryAllServices() ([]windows.ENUM_SERVICE_STATUS_PROCESS, error) { +func (c *Collector) queryAllServices() ([]serviceEntry, error) { var ( additionalBytesNeeded uint32 servicesReturned uint32 err error ) + // The buffer is reused across scrapes. Hold the lock until the services are copied + // out of the buffer, so that an overlapping Collect can't overwrite it in the meantime. + c.queryAllServicesMu.Lock() + defer c.queryAllServicesMu.Unlock() + clear(c.queryAllServicesBuffer) for { @@ -462,12 +511,29 @@ func (c *Collector) queryAllServices() ([]windows.ENUM_SERVICE_STATUS_PROCESS, e } if servicesReturned == 0 { - return []windows.ENUM_SERVICE_STATUS_PROCESS{}, nil + return []serviceEntry{}, nil } services := unsafe.Slice((*windows.ENUM_SERVICE_STATUS_PROCESS)(unsafe.Pointer(&c.queryAllServicesBuffer[0])), int(servicesReturned)) + entries := make([]serviceEntry, 0, len(services)) - return services, nil + for _, service := range services { + if uintptr(unsafe.Pointer(service.ServiceName)) == uintptr(windows.InvalidHandle) { + c.logger.Log(context.Background(), slog.LevelWarn, "failed collecting service info", + slog.String("err", "ServiceName is 0xffffffffffffffff"), + slog.String("service", fmt.Sprintf("%+v", service)), + ) + + continue + } + + entries = append(entries, serviceEntry{ + name: windows.UTF16PtrToString(service.ServiceName), + status: service.ServiceStatusProcess, + }) + } + + return entries, nil } func (c *Collector) getProcessStartTime(pid uint32) (uint64, error) { diff --git a/internal/collector/smbclient/smbclient.go b/internal/collector/smbclient/smbclient.go index 70990397..b9e7e682 100644 --- a/internal/collector/smbclient/smbclient.go +++ b/internal/collector/smbclient/smbclient.go @@ -90,6 +90,8 @@ func (c *Collector) GetName() string { } func (c *Collector) Close() error { + c.perfDataCollector.Close() + return nil } diff --git a/internal/collector/tcp/tcp.go b/internal/collector/tcp/tcp.go index 53a87546..39e2c85b 100644 --- a/internal/collector/tcp/tcp.go +++ b/internal/collector/tcp/tcp.go @@ -129,6 +129,16 @@ func (c *Collector) Close() error { } func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { + subCollectors := []string{subCollectorMetrics, subCollectorConnectionsState} + + for _, collector := range c.config.CollectorsEnabled { + if !slices.Contains(subCollectors, collector) { + return fmt.Errorf("unknown sub collector: %s. Possible values: %s", collector, + strings.Join(subCollectors, ", "), + ) + } + } + labels := []string{"af"} c.connectionFailures = prometheus.NewDesc( diff --git a/internal/collector/terminal_services/terminal_services.go b/internal/collector/terminal_services/terminal_services.go index c47912c3..29829663 100644 --- a/internal/collector/terminal_services/terminal_services.go +++ b/internal/collector/terminal_services/terminal_services.go @@ -121,15 +121,19 @@ func (c *Collector) GetName() string { } func (c *Collector) Close() error { - err := wtsapi32.WTSCloseServer(c.hServer) - if err != nil { - return fmt.Errorf("failed to close WTS server: %w", err) + c.perfDataCollectorTerminalServicesSession.Close() + c.perfDataCollectorBroker.Close() + + // hServer is 0 if Build failed before opening the server. + if c.hServer == 0 || c.hServer == windows.InvalidHandle { + return nil } - c.perfDataCollectorTerminalServicesSession.Close() + err := wtsapi32.WTSCloseServer(c.hServer) + c.hServer = 0 - if c.connectionBrokerEnabled { - c.perfDataCollectorBroker.Close() + if err != nil { + return fmt.Errorf("failed to close WTS server: %w", err) } return nil diff --git a/internal/collector/textfile/textfile.go b/internal/collector/textfile/textfile.go index d5f0e868..4d4760a1 100644 --- a/internal/collector/textfile/textfile.go +++ b/internal/collector/textfile/textfile.go @@ -123,34 +123,52 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { // Given a slice of metric families, determine if any two entries are duplicates. // Duplicates will be detected where the metric name, labels and label values are identical. +// Labels with an empty value are ignored, because convertMetricFamily adds missing labels +// with an empty value, and Prometheus treats them as absent. func duplicateMetricEntry(metricFamilies []*dto.MetricFamily) bool { - uniqueMetrics := make(map[string]map[string]string) + seen := make(map[string]struct{}) for _, metricFamily := range metricFamilies { - metricName := metricFamily.GetName() - for _, metric := range metricFamily.GetMetric() { - metricLabels := metric.GetLabel() - labels := make(map[string]string) + key := seriesKey(metricFamily.GetName(), metric.GetLabel()) - for _, label := range metricLabels { - labels[label.GetName()] = label.GetValue() - } - // Check if key is present before appending - _, mapContainsKey := uniqueMetrics[metricName] - - // Duplicate metric found with identical labels & label values - if mapContainsKey && maps.Equal(uniqueMetrics[metricName], labels) { + if _, ok := seen[key]; ok { return true } - uniqueMetrics[metricName] = labels + seen[key] = struct{}{} } } return false } +// seriesKey returns a key that identifies a series by its metric name and its non-empty labels, +// independent of the label order. +func seriesKey(metricName string, labelPairs []*dto.LabelPair) string { + labels := make(map[string]string, len(labelPairs)) + + for _, label := range labelPairs { + if label.GetValue() != "" { + labels[label.GetName()] = label.GetValue() + } + } + + var sb strings.Builder + + sb.WriteString(metricName) + + // The separator byte is not valid UTF-8, so it can't be part of a label name or value. + for _, name := range slices.Sorted(maps.Keys(labels)) { + sb.WriteByte(model.SeparatorByte) + sb.WriteString(name) + sb.WriteByte(model.SeparatorByte) + sb.WriteString(labels[name]) + } + + return sb.String() +} + func (c *Collector) convertMetricFamily(logger *slog.Logger, metricFamily *dto.MetricFamily, ch chan<- prometheus.Metric) { var valType prometheus.ValueType diff --git a/internal/collector/textfile/textfile_test.go b/internal/collector/textfile/textfile_test.go index 449d5fb7..dc671c2a 100644 --- a/internal/collector/textfile/textfile_test.go +++ b/internal/collector/textfile/textfile_test.go @@ -179,3 +179,66 @@ func TestDuplicateMetricEntry(t *testing.T) { t.Errorf("Unexpected duplicate found in differentValues") } } + +func TestDuplicateMetricEntryAnyPosition(t *testing.T) { + t.Parallel() + + type pair [2]string + + metric := func(labels ...pair) *dto.Metric { + m := &dto.Metric{} + + for _, label := range labels { + m.Label = append(m.Label, &dto.LabelPair{Name: &label[0], Value: &label[1]}) + } + + return m + } + + family := func(name string, metrics ...*dto.Metric) *dto.MetricFamily { + return &dto.MetricFamily{Name: &name, Metric: metrics} + } + + for _, tc := range []struct { + name string + families []*dto.MetricFamily + duplicate bool + }{ + { + name: "duplicate is not adjacent", + families: []*dto.MetricFamily{family("m", metric(pair{"a", "1"}), metric(pair{"a", "2"}), metric(pair{"a", "1"}))}, + duplicate: true, + }, + { + name: "duplicate in another family of the same name", + families: []*dto.MetricFamily{family("m", metric(pair{"a", "1"}), metric(pair{"a", "2"})), family("m", metric(pair{"a", "1"}))}, + duplicate: true, + }, + { + name: "different label order", + families: []*dto.MetricFamily{family("m", metric(pair{"a", "1"}, pair{"b", "2"}), metric(pair{"b", "2"}, pair{"a", "1"}))}, + duplicate: true, + }, + { + name: "empty label value is absent", + families: []*dto.MetricFamily{family("m", metric(pair{"a", "1"}), metric(pair{"a", "1"}, pair{"b", ""}))}, + duplicate: true, + }, + { + name: "same labels in different metrics", + families: []*dto.MetricFamily{family("m", metric(pair{"a", "1"})), family("n", metric(pair{"a", "1"}))}, + }, + { + name: "label value shifted between labels", + families: []*dto.MetricFamily{family("m", metric(pair{"a", "1"}, pair{"b", "2"}), metric(pair{"a", "12"}, pair{"b", ""}))}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + if got := duplicateMetricEntry(tc.families); got != tc.duplicate { + t.Errorf("duplicateMetricEntry() = %t, want %t", got, tc.duplicate) + } + }) + } +} diff --git a/internal/collector/thermalzone/thermalzone.go b/internal/collector/thermalzone/thermalzone.go index f0594a7d..7ce2a14b 100644 --- a/internal/collector/thermalzone/thermalzone.go +++ b/internal/collector/thermalzone/thermalzone.go @@ -69,6 +69,8 @@ func (c *Collector) GetName() string { } func (c *Collector) Close() error { + c.perfDataCollector.Close() + return nil } diff --git a/internal/collector/update/update.go b/internal/collector/update/update.go index 72894845..74b44df1 100644 --- a/internal/collector/update/update.go +++ b/internal/collector/update/update.go @@ -33,6 +33,7 @@ import ( "github.com/go-ole/go-ole/oleutil" "github.com/prometheus-community/windows_exporter/internal/mi" "github.com/prometheus-community/windows_exporter/internal/types" + "github.com/prometheus-community/windows_exporter/internal/utils/recovery" "github.com/prometheus/client_golang/prometheus" ) @@ -111,7 +112,10 @@ func NewWithFlags(app *kingpin.Application) *Collector { } func (c *Collector) Close() error { - c.ctxCancelFn() + if c.ctxCancelFn != nil { + c.ctxCancelFn() + c.ctxCancelFn = nil + } return nil } @@ -151,8 +155,27 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { ctx, cancel := context.WithCancel(context.Background()) + // scheduleUpdateStatus sends exactly one value to initErrCh, nil after a successful initialization. initErrCh := make(chan error, 1) - go c.scheduleUpdateStatus(ctx, logger, initErrCh, c.config.Online) + + go func() { + err := recovery.Run(func() error { + c.scheduleUpdateStatus(ctx, logger, initErrCh, c.config.Online) + + return nil + }) + if err != nil { + logger.Error("Windows Update worker stopped", + slog.Any("err", err), + ) + + // Unblock Build if the panic happened during the initialization. + select { + case initErrCh <- err: + default: + } + } + }() c.ctxCancelFn = cancel @@ -248,8 +271,8 @@ func (c *Collector) scheduleUpdateStatus(ctx context.Context, logger *slog.Logge return } + // ush shares the reference of us, which is released by us.Clear(). ush := us.ToIDispatch() - defer ush.Release() _, err = oleutil.PutProperty(ush, "Online", online) if err != nil { @@ -272,15 +295,19 @@ func (c *Collector) scheduleUpdateStatus(ctx context.Context, logger *slog.Logge return } - close(initErrCh) - - usd := us.ToIDispatch() - defer usd.Release() + initErrCh <- nil var metricsBuf []prometheus.Metric for { - metricsBuf, err = c.fetchUpdates(logger, usd) + // A panic in a single search must not stop the worker. + err = recovery.Run(func() error { + var err error + + metricsBuf, err = c.fetchUpdates(logger, ush) + + return err + }) if err != nil { logger.ErrorContext(ctx, "failed to fetch updates", slog.Any("err", err), @@ -405,6 +432,8 @@ func (c *Collector) getUpdateStatus(updd *ole.IDispatch, item int) (windowsUpdat return windowsUpdate{}, fmt.Errorf("get MsrcSeverity: %w", err) } + defer clearVariant(severity) + categoriesRaw, err := oleutil.GetProperty(updateItem, "Categories") if err != nil { return windowsUpdate{}, fmt.Errorf("get Categories: %w", err) @@ -423,6 +452,8 @@ func (c *Collector) getUpdateStatus(updd *ole.IDispatch, item int) (windowsUpdat return windowsUpdate{}, fmt.Errorf("get Title: %w", err) } + defer clearVariant(title) + // Get the Identity object identityVariant, err := oleutil.GetProperty(updateItem, "Identity") if err != nil { @@ -438,6 +469,8 @@ func (c *Collector) getUpdateStatus(updd *ole.IDispatch, item int) (windowsUpdat return windowsUpdate{}, fmt.Errorf("get UpdateID: %w", err) } + defer clearVariant(updateIDVariant) + revisionVariant, err := oleutil.GetProperty(identity, "RevisionNumber") if err != nil { return windowsUpdate{}, fmt.Errorf("get RevisionNumber: %w", err) @@ -493,6 +526,8 @@ func getUpdateCategory(categories *ole.IDispatch) (string, error) { return fmt.Errorf("get Category item Name: %w", err) } + defer clearVariant(categoryNameRaw) + orderRaw, err := oleutil.GetProperty(category, "Order") if err != nil { return fmt.Errorf("get Category item Order: %w", err) @@ -512,3 +547,11 @@ func getUpdateCategory(categories *ole.IDispatch) (string, error) { return categoryName, nil } + +// clearVariant frees the value of a VARIANT, for example a BSTR. +// Values must be copied, e.g. with [ole.VARIANT.ToString], before. +func clearVariant(v *ole.VARIANT) { + if v != nil { + _ = v.Clear() + } +} diff --git a/internal/collector/wmi/wmi.go b/internal/collector/wmi/wmi.go index 0f21e743..e584a630 100644 --- a/internal/collector/wmi/wmi.go +++ b/internal/collector/wmi/wmi.go @@ -20,6 +20,7 @@ package wmi import ( "errors" "fmt" + "io" "log/slog" "maps" "regexp" @@ -118,7 +119,13 @@ func NewWithFlags(app *kingpin.Application) *Collector { return nil } - if err := yaml.Unmarshal([]byte(queries), &c.config.Queries); err != nil { + // Reject unknown keys, so that a typo like "label_property" fails instead + // of silently dropping the setting. A value without a document, e.g. only + // comments, means no queries, like in the configuration file. + decoder := yaml.NewDecoder(strings.NewReader(queries)) + decoder.KnownFields(true) + + if err := decoder.Decode(&c.config.Queries); err != nil && !errors.Is(err, io.EOF) { return fmt.Errorf("failed to parse queries %s: %w", queries, err) } diff --git a/internal/collector/wmi/wmi_output_test.go b/internal/collector/wmi/wmi_output_test.go index a1f36157..2dfeaaab 100644 --- a/internal/collector/wmi/wmi_output_test.go +++ b/internal/collector/wmi/wmi_output_test.go @@ -345,6 +345,16 @@ func TestNewWithFlags(t *testing.T) { _, err := app.Parse([]string{"--collector.wmi.queries", `[{"name": `}) require.Error(t, err) }) + + t.Run("unknown field", func(t *testing.T) { + t.Parallel() + + app := kingpin.New("test", "") + _ = wmi.NewWithFlags(app) + + _, err := app.Parse([]string{"--collector.wmi.queries", `[{"name":"os","class":"Win32_OperatingSystem","label_property":[{"name":"Name"}],"properties":[{"name":"Primary"}]}]`}) + require.ErrorContains(t, err, "label_property") + }) } // TestCollectConcurrent runs overlapping scrapes, as concurrent HTTP requests diff --git a/internal/utils/recovery/recovery.go b/internal/utils/recovery/recovery.go new file mode 100644 index 00000000..e5f3b853 --- /dev/null +++ b/internal/utils/recovery/recovery.go @@ -0,0 +1,75 @@ +// 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 recovery runs functions in a way that a panic is returned as an error +// instead of terminating the process. A panic in a goroutine cannot be recovered +// by the caller that started it, so every goroutine a collector starts must +// recover on its own. +package recovery + +import ( + "errors" + "fmt" + "runtime/debug" + "sync" +) + +// ErrPanic is wrapped by every error that Run and Group return for a recovered panic. +var ErrPanic = errors.New("panic") + +// Run calls fn and returns its error. If fn panics, Run recovers and returns +// an error wrapping [ErrPanic] that contains the panic value and the stack trace. +func Run(fn func() error) (err error) { + defer func() { + if r := recover(); r != nil { + err = fmt.Errorf("%w: %v\n%s", ErrPanic, r, debug.Stack()) + } + }() + + return fn() +} + +// Group runs functions in goroutines and collects all of their errors. +// A panic in one function is returned as an error by Wait and does not affect +// the other functions. The zero value is ready to use. Go may be called from +// within a function that runs in the group. +type Group struct { + wg sync.WaitGroup + mu sync.Mutex + errs []error +} + +// Go calls fn through [Run] in a new goroutine. +func (g *Group) Go(fn func() error) { + g.wg.Go(func() { + if err := Run(fn); err != nil { + g.mu.Lock() + g.errs = append(g.errs, err) + g.mu.Unlock() + } + }) +} + +// Wait blocks until all functions have returned and returns their errors joined with [errors.Join]. +func (g *Group) Wait() error { + g.wg.Wait() + + g.mu.Lock() + defer g.mu.Unlock() + + return errors.Join(g.errs...) +} diff --git a/internal/utils/recovery/recovery_test.go b/internal/utils/recovery/recovery_test.go new file mode 100644 index 00000000..2ea41437 --- /dev/null +++ b/internal/utils/recovery/recovery_test.go @@ -0,0 +1,92 @@ +// 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 recovery_test + +import ( + "errors" + "sync/atomic" + "testing" + + "github.com/prometheus-community/windows_exporter/internal/utils/recovery" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestRun(t *testing.T) { + t.Parallel() + + errExpected := errors.New("expected") + + require.NoError(t, recovery.Run(func() error { return nil })) + require.ErrorIs(t, recovery.Run(func() error { return errExpected }), errExpected) + + err := recovery.Run(func() error { + panic("boom") + }) + require.ErrorIs(t, err, recovery.ErrPanic) + assert.Contains(t, err.Error(), "boom") + assert.Contains(t, err.Error(), "recovery_test.go", "error must contain the stack trace") +} + +func TestGroup(t *testing.T) { + t.Parallel() + + errExpected := errors.New("expected") + + var ( + g recovery.Group + done atomic.Int32 + ) + + g.Go(func() error { + done.Add(1) + + return nil + }) + g.Go(func() error { + done.Add(1) + + return errExpected + }) + g.Go(func() error { + done.Add(1) + + panic("boom") + }) + g.Go(func() error { + // Nested Go calls must be waited for, too. + g.Go(func() error { + done.Add(1) + + return nil + }) + + done.Add(1) + + return nil + }) + + err := g.Wait() + require.ErrorIs(t, err, errExpected) + require.ErrorIs(t, err, recovery.ErrPanic) + assert.Contains(t, err.Error(), "boom") + assert.Equal(t, int32(5), done.Load()) + + var empty recovery.Group + require.NoError(t, empty.Wait()) +}