diff --git a/AGENTS.md b/AGENTS.md index 0aa242fc..9a3a8eb7 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -127,6 +127,14 @@ exercised a live Windows role. resources while it runs. Honor the supplied scrape budget in blocking APIs. `Build` has no budget: bound its MI probe queries with `mi.BuildQueryTimeout`, never `0` (no MI timeout), so one stalled provider cannot block startup. +- Collectors that reuse metrics or label pairs between scrapes keep an + `internal/metriccache` or `internal/labelcache` field and list it in + `ScrapeCaches()` (`internal/scrapecache`). The collection resets these caches + before `Build` and after `Close`, and begins a scrape before and ends it after + each `Collect`; `Collect` gets it with `Scrape()`. Do not call `Begin`, + `Commit` or `Reset` in collectors. A direct `Collect` call, as in most tests, + runs without the caches; use `testutils.Collect` or `scrapecache.Begin` to + test the cached path. - Release resources after partial initialization and on error paths. A non-nil PDH collector returned with an error still belongs to the caller and must be closed. Sub-collector cleanup closures must read the initialized field when diff --git a/internal/collector/gpu/gpu.go b/internal/collector/gpu/gpu.go index 0d19bce5..30c8cad0 100644 --- a/internal/collector/gpu/gpu.go +++ b/internal/collector/gpu/gpu.go @@ -29,6 +29,7 @@ import ( "github.com/prometheus-community/windows_exporter/internal/labelcache" "github.com/prometheus-community/windows_exporter/internal/mi" "github.com/prometheus-community/windows_exporter/internal/pdh" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus-community/windows_exporter/internal/types" "github.com/prometheus/client_golang/prometheus" ) @@ -141,13 +142,17 @@ func (c *Collector) GetName() string { return Name } +// ScrapeCaches returns the caches that the collection resets and begins a scrape of around Collect. +func (c *Collector) ScrapeCaches() []scrapecache.Cache { + return []scrapecache.Cache{&c.labels} +} + func (c *Collector) Close() error { c.gpuEnginePerfDataCollector.Close() c.gpuAdapterMemoryPerfDataCollector.Close() c.gpuLocalAdapterMemoryPerfDataCollector.Close() c.gpuNonLocalAdapterMemoryPerfDataCollector.Close() c.gpuProcessMemoryPerfDataCollector.Close() - c.labels.Reset() c.instances.reset() return nil @@ -409,8 +414,7 @@ func (c *Collector) getGPUDevice(luid string) (gpuDevice, bool) { func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error { errs := make([]error, 0) - scrape := c.labels.Begin() - defer scrape.Commit() + scrape := c.labels.Scrape() instances := c.instances.begin() defer instances.commit() diff --git a/internal/collector/hyperv/hyperv.go b/internal/collector/hyperv/hyperv.go index 114d98d2..3b068cdc 100644 --- a/internal/collector/hyperv/hyperv.go +++ b/internal/collector/hyperv/hyperv.go @@ -30,6 +30,7 @@ import ( "github.com/prometheus-community/windows_exporter/internal/labelcache" "github.com/prometheus-community/windows_exporter/internal/mi" "github.com/prometheus-community/windows_exporter/internal/osversion" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus-community/windows_exporter/internal/utils/recovery" "github.com/prometheus/client_golang/prometheus" ) @@ -184,13 +185,17 @@ func (c *Collector) GetName() string { return Name } +// ScrapeCaches returns the caches that the collection resets and begins a scrape of around Collect. +func (c *Collector) ScrapeCaches() []scrapecache.Cache { + return []scrapecache.Cache{&c.labels} +} + func (c *Collector) Close() error { for _, fn := range c.closeFns { fn() } c.closeFns = nil - c.labels.Reset() return nil } @@ -359,8 +364,7 @@ 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 { - scrape := c.labels.Begin() - defer scrape.Commit() + scrape := c.labels.Scrape() var g recovery.Group diff --git a/internal/collector/mscluster/mscluster.go b/internal/collector/mscluster/mscluster.go index edd753c3..01a5e0b1 100644 --- a/internal/collector/mscluster/mscluster.go +++ b/internal/collector/mscluster/mscluster.go @@ -30,6 +30,7 @@ import ( "github.com/alecthomas/kingpin/v2" "github.com/prometheus-community/windows_exporter/internal/mi" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus-community/windows_exporter/internal/utils/recovery" "github.com/prometheus/client_golang/prometheus" ) @@ -122,6 +123,17 @@ func (c *Collector) GetName() string { return Name } +// ScrapeCaches returns the caches that the collection resets and begins a scrape of around Collect. +func (c *Collector) ScrapeCaches() []scrapecache.Cache { + return []scrapecache.Cache{ + &c.clusterMetricCache, + &c.networkMetricCache, + &c.nodeMetricCache, + &c.resourceMetricCache, + &c.resourceGroupMetricCache, + } +} + func (c *Collector) Close() error { c.lifecycleMu.Lock() defer c.lifecycleMu.Unlock() @@ -129,7 +141,7 @@ func (c *Collector) Close() error { return c.closeSources() } -// closeSources releases the ClusAPI sources and drops the cached metrics. A +// closeSources releases the ClusAPI sources. A // source whose Close fails is dropped as well: the clusapi.Cluster has already // released its handle, and a later Build must be able to create new sources // (Grafana Alloy rebuilds collectors in place). @@ -155,12 +167,6 @@ func (c *Collector) closeSources() error { closeSource(c.resourceGroupSource, func() { c.resourceGroupSource = nil }) closeSource(c.sharedVolumesSource, func() { c.sharedVolumesSource = nil }) - c.clusterMetricCache.Reset() - c.networkMetricCache.Reset() - c.nodeMetricCache.Reset() - c.resourceMetricCache.Reset() - c.resourceGroupMetricCache.Reset() - return errors.Join(errs...) } diff --git a/internal/collector/mscluster/mscluster_cluster.go b/internal/collector/mscluster/mscluster_cluster.go index 83ed6f87..a0fd3346 100644 --- a/internal/collector/mscluster/mscluster_cluster.go +++ b/internal/collector/mscluster/mscluster_cluster.go @@ -679,9 +679,6 @@ func (c *Collector) buildClusterDescriptors() { []string{"name"}, nil, ) - - // Cached metrics refer to the descriptors of an earlier build. - c.clusterMetricCache.Reset() } func (c *Collector) collectCluster(ch chan<- prometheus.Metric, maxScrapeDuration time.Duration) error { @@ -695,11 +692,7 @@ func (c *Collector) collectCluster(ch chan<- prometheus.Metric, maxScrapeDuratio } func (c *Collector) publishCluster(ch chan<- prometheus.Metric, cluster clusapi.Object, build uint16, resultErr error) error { - scrape := c.clusterMetricCache.Begin() - if cluster.Name == "" { - commitScrape(scrape, 0, resultErr) - return resultErr } @@ -797,8 +790,7 @@ func (c *Collector) publishCluster(ch chan<- prometheus.Metric, cluster clusapi. {name: "DetectedCloudPlatform", desc: c.clusterDetectedCloudPlatform, minBuild: ws2022, older: neverPublish}, } - _, resultErr = publishCachedObjectFields(ch, scrape, nil, "cluster", cluster, fields, build, resultErr) - commitScrape(scrape, 1, resultErr) + _, resultErr = publishCachedObjectFields(ch, c.clusterMetricCache.Scrape(), nil, "cluster", cluster, fields, build, resultErr) return resultErr } diff --git a/internal/collector/mscluster/mscluster_cluster_native_test.go b/internal/collector/mscluster/mscluster_cluster_native_test.go index 95665cb8..d4053090 100644 --- a/internal/collector/mscluster/mscluster_cluster_native_test.go +++ b/internal/collector/mscluster/mscluster_cluster_native_test.go @@ -28,6 +28,7 @@ import ( "github.com/prometheus-community/windows_exporter/internal/headers/clusapi" "github.com/prometheus-community/windows_exporter/internal/osversion" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus/client_golang/prometheus" dto "github.com/prometheus/client_model/go" ) @@ -290,7 +291,11 @@ func BenchmarkClusterPublication(b *testing.B) { b.ReportAllocs() for b.Loop() { - if err := c.publishCluster(ch, cluster, osversion.LTSC2022, nil); err != nil { + cacheScrape := scrapecache.Begin(c) + err := c.publishCluster(ch, cluster, osversion.LTSC2022, nil) + cacheScrape.End(err) + + if err != nil { b.Fatal(err) } diff --git a/internal/collector/mscluster/mscluster_network.go b/internal/collector/mscluster/mscluster_network.go index addc44ff..41bbc578 100644 --- a/internal/collector/mscluster/mscluster_network.go +++ b/internal/collector/mscluster/mscluster_network.go @@ -102,9 +102,6 @@ func (c *Collector) buildNetworkDescriptors() { []string{"name"}, nil, ) - - // Cached metrics refer to the descriptors of an earlier build. - c.networkMetricCache.Reset() } // Collect sends the metric values for each metric @@ -128,7 +125,7 @@ func (c *Collector) publishNetworks(ch chan<- prometheus.Metric, networks []clus {name: "State", desc: c.networkState}, } - scrape := c.networkMetricCache.Begin() + scrape := c.networkMetricCache.Scrape() var key []byte @@ -142,7 +139,5 @@ func (c *Collector) publishNetworks(ch chan<- prometheus.Metric, networks []clus key, resultErr = publishCachedObjectFields(ch, scrape, key, "network", network, fields, 0, resultErr) } - commitScrape(scrape, len(networks), resultErr) - return resultErr } diff --git a/internal/collector/mscluster/mscluster_network_native_test.go b/internal/collector/mscluster/mscluster_network_native_test.go index ead87ea7..29d133dd 100644 --- a/internal/collector/mscluster/mscluster_network_native_test.go +++ b/internal/collector/mscluster/mscluster_network_native_test.go @@ -26,6 +26,7 @@ import ( "time" "github.com/prometheus-community/windows_exporter/internal/headers/clusapi" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus/client_golang/prometheus" dto "github.com/prometheus/client_model/go" ) @@ -175,7 +176,11 @@ func BenchmarkNetworkPublication(b *testing.B) { b.ReportAllocs() for b.Loop() { - if err := c.publishNetworks(ch, networks, nil); err != nil { + cacheScrape := scrapecache.Begin(c) + err := c.publishNetworks(ch, networks, nil) + cacheScrape.End(err) + + if err != nil { b.Fatal(err) } diff --git a/internal/collector/mscluster/mscluster_node.go b/internal/collector/mscluster/mscluster_node.go index d11ea585..eea41cd5 100644 --- a/internal/collector/mscluster/mscluster_node.go +++ b/internal/collector/mscluster/mscluster_node.go @@ -175,9 +175,6 @@ func (c *Collector) buildNodeDescriptors() { []string{"name"}, nil, ) - - // Cached metrics refer to the descriptors of an earlier build. - c.nodeMetricCache.Reset() } // Collect sends the metric values for each metric @@ -214,7 +211,7 @@ func (c *Collector) publishNodes(ch chan<- prometheus.Metric, nodes []clusapi.Ob } nodeNames := make([]string, 0, len(nodes)) - scrape := c.nodeMetricCache.Begin() + scrape := c.nodeMetricCache.Scrape() var key []byte @@ -227,7 +224,5 @@ func (c *Collector) publishNodes(ch chan<- prometheus.Metric, nodes []clusapi.Ob nodeNames = append(nodeNames, node.Name) } - commitScrape(scrape, len(nodes), resultErr) - return nodeNames, resultErr } diff --git a/internal/collector/mscluster/mscluster_node_native_test.go b/internal/collector/mscluster/mscluster_node_native_test.go index d707931c..a3464b22 100644 --- a/internal/collector/mscluster/mscluster_node_native_test.go +++ b/internal/collector/mscluster/mscluster_node_native_test.go @@ -28,6 +28,7 @@ import ( "github.com/prometheus-community/windows_exporter/internal/headers/clusapi" "github.com/prometheus-community/windows_exporter/internal/osversion" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus/client_golang/prometheus" dto "github.com/prometheus/client_model/go" ) @@ -268,7 +269,11 @@ func BenchmarkNodePublication(b *testing.B) { b.ReportAllocs() for b.Loop() { - if _, err := c.publishNodes(ch, nodes, osversion.LTSC2022, nil); err != nil { + cacheScrape := scrapecache.Begin(c) + _, err := c.publishNodes(ch, nodes, osversion.LTSC2022, nil) + cacheScrape.End(err) + + if err != nil { b.Fatal(err) } diff --git a/internal/collector/mscluster/mscluster_objects.go b/internal/collector/mscluster/mscluster_objects.go index d6c05324..c713159b 100644 --- a/internal/collector/mscluster/mscluster_objects.go +++ b/internal/collector/mscluster/mscluster_objects.go @@ -183,13 +183,3 @@ func publishCachedObjectFields(ch chan<- prometheus.Metric, scrape *metriccache. return key, resultErr } - -// commitScrape commits a scrape unless it collected nothing because reading -// the objects failed, so a failed read keeps the cached metrics. -func commitScrape[K, V comparable](scrape *metriccache.Scrape[K, V], collected int, err error) { - if collected == 0 && err != nil { - return - } - - scrape.Commit() -} diff --git a/internal/collector/mscluster/mscluster_resource.go b/internal/collector/mscluster/mscluster_resource.go index 51c22b26..6964067b 100644 --- a/internal/collector/mscluster/mscluster_resource.go +++ b/internal/collector/mscluster/mscluster_resource.go @@ -210,9 +210,6 @@ func (c *Collector) buildResourceDescriptors() { []string{"type", "owner_group", "name"}, nil, ) - - // Cached metrics refer to the descriptors of an earlier build. - c.resourceMetricCache.Reset() } // Collect sends the metric values for each metric @@ -283,7 +280,7 @@ func (c *Collector) publishResources(ch chan<- prometheus.Metric, resources []cl } nodeNamesValue := string(nodeNamesKey) - scrape := c.resourceMetricCache.Begin() + scrape := c.resourceMetricCache.Scrape() for _, resource := range resources { if !resource.IdentityValid { @@ -345,8 +342,6 @@ func (c *Collector) publishResources(ch chan<- prometheus.Metric, resources []cl } } - commitScrape(scrape, len(resources), resultErr) - return resultErr } diff --git a/internal/collector/mscluster/mscluster_resource_native_test.go b/internal/collector/mscluster/mscluster_resource_native_test.go index 13be17ba..50ae4e7e 100644 --- a/internal/collector/mscluster/mscluster_resource_native_test.go +++ b/internal/collector/mscluster/mscluster_resource_native_test.go @@ -26,6 +26,7 @@ import ( "time" "github.com/prometheus-community/windows_exporter/internal/headers/clusapi" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus/client_golang/prometheus" dto "github.com/prometheus/client_model/go" ) @@ -289,7 +290,11 @@ func BenchmarkResourcePublication(b *testing.B) { b.ReportAllocs() for b.Loop() { - if err := c.publishResources(ch, resources, []string{"nodeA", "nodeB"}, nil); err != nil { + cacheScrape := scrapecache.Begin(c) + err := c.publishResources(ch, resources, []string{"nodeA", "nodeB"}, nil) + cacheScrape.End(err) + + if err != nil { b.Fatal(err) } @@ -305,7 +310,9 @@ func TestResourceMetricCache(t *testing.T) { publish := func(resource clusapi.Resource, nodeNames ...string) ([]prometheus.Metric, error) { ch := make(chan prometheus.Metric, 100) + cacheScrape := scrapecache.Begin(c) err := c.publishResources(ch, []clusapi.Resource{resource}, nodeNames, nil) + cacheScrape.End(err) close(ch) diff --git a/internal/collector/mscluster/mscluster_resourcegroup.go b/internal/collector/mscluster/mscluster_resourcegroup.go index c8639041..b2b30d39 100644 --- a/internal/collector/mscluster/mscluster_resourcegroup.go +++ b/internal/collector/mscluster/mscluster_resourcegroup.go @@ -175,9 +175,6 @@ func (c *Collector) buildResourceGroupDescriptors() { []string{"name"}, nil, ) - - // Cached metrics refer to the descriptors of an earlier build. - c.resourceGroupMetricCache.Reset() } // Collect sends the metric values for each metric @@ -209,7 +206,7 @@ func (c *Collector) publishResourceGroups(ch chan<- prometheus.Metric, groups [] {name: "State", desc: c.resourceGroupState}, } - scrape := c.resourceGroupMetricCache.Begin() + scrape := c.resourceGroupMetricCache.Scrape() var nodeNamesKey []byte for _, nodeName := range nodeNames { @@ -257,8 +254,6 @@ func (c *Collector) publishResourceGroups(ch chan<- prometheus.Metric, groups [] } } - commitScrape(scrape, len(groups), resultErr) - return resultErr } diff --git a/internal/collector/mscluster/mscluster_resourcegroup_native_test.go b/internal/collector/mscluster/mscluster_resourcegroup_native_test.go index a21e924e..e93cf251 100644 --- a/internal/collector/mscluster/mscluster_resourcegroup_native_test.go +++ b/internal/collector/mscluster/mscluster_resourcegroup_native_test.go @@ -27,6 +27,7 @@ import ( "github.com/prometheus-community/windows_exporter/internal/headers/clusapi" "github.com/prometheus-community/windows_exporter/internal/osversion" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus/client_golang/prometheus" dto "github.com/prometheus/client_model/go" ) @@ -370,7 +371,11 @@ func BenchmarkResourceGroupPublication(b *testing.B) { b.ReportAllocs() for b.Loop() { - if err := c.publishResourceGroups(ch, groups, []string{"nodeA", "nodeB"}, osversion.LTSC2022, nil); err != nil { + cacheScrape := scrapecache.Begin(c) + err := c.publishResourceGroups(ch, groups, []string{"nodeA", "nodeB"}, osversion.LTSC2022, nil) + cacheScrape.End(err) + + if err != nil { b.Fatal(err) } diff --git a/internal/collector/net/net.go b/internal/collector/net/net.go index 829f3608..c3e93798 100644 --- a/internal/collector/net/net.go +++ b/internal/collector/net/net.go @@ -33,6 +33,7 @@ import ( "github.com/prometheus-community/windows_exporter/internal/metriccache" "github.com/prometheus-community/windows_exporter/internal/mi" "github.com/prometheus-community/windows_exporter/internal/pdh" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus-community/windows_exporter/internal/types" "github.com/prometheus-community/windows_exporter/internal/utils" "github.com/prometheus/client_golang/prometheus" @@ -184,10 +185,13 @@ func (c *Collector) GetName() string { return Name } +// ScrapeCaches returns the caches that the collection resets and begins a scrape of around Collect. +func (c *Collector) ScrapeCaches() []scrapecache.Cache { + return []scrapecache.Cache{&c.labels, &c.nicMetricCache} +} + func (c *Collector) Close() error { c.perfDataCollector.Close() - c.nicMetricCache.Reset() - c.labels.Reset() c.adapterAddressesBuf = nil return nil @@ -299,10 +303,6 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { nil, ) - // Cached metrics refer to the descriptors of an earlier Build. - c.nicMetricCache.Reset() - c.labels.Reset() - var err error c.perfDataCollector, err = pdh.NewCollector[perfDataCounterValues](logger.With(slog.String("collector", Name)), pdh.CounterTypeRaw, "Network Interface", pdh.InstancesAll) @@ -345,8 +345,7 @@ func (c *Collector) collect(ch chan<- prometheus.Metric) error { return fmt.Errorf("failed to collect Network Information metrics: %w", err) } - scrape := c.labels.Begin() - defer scrape.Commit() + scrape := c.labels.Scrape() for _, data := range c.perfDataObject { if c.config.NicExclude.MatchString(data.Name) || !c.config.NicInclude.MatchString(data.Name) { @@ -460,7 +459,7 @@ func (c *Collector) collectNICInfo(ch chan<- prometheus.Metric) error { return err } - scrape := c.nicMetricCache.Begin() + scrape := c.nicMetricCache.Scrape() var addresses []byte @@ -497,8 +496,6 @@ func (c *Collector) collectNICInfo(ch chan<- prometheus.Metric) error { } } - scrape.Commit() - return nil } diff --git a/internal/collector/net/net_bench_test.go b/internal/collector/net/net_bench_test.go index f4642461..d9e7ce4c 100644 --- a/internal/collector/net/net_bench_test.go +++ b/internal/collector/net/net_bench_test.go @@ -23,6 +23,7 @@ import ( "testing" "github.com/prometheus-community/windows_exporter/internal/collector/net" + "github.com/prometheus-community/windows_exporter/internal/utils/testutils" "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -49,6 +50,6 @@ func BenchmarkCollector(b *testing.B) { }) for b.Loop() { - require.NoError(b, c.Collect(metrics, 0)) + require.NoError(b, testutils.Collect(c, metrics, 0)) } } diff --git a/internal/collector/netframework/netframework.go b/internal/collector/netframework/netframework.go index 4c2bf4d0..a0093538 100644 --- a/internal/collector/netframework/netframework.go +++ b/internal/collector/netframework/netframework.go @@ -31,6 +31,7 @@ import ( "github.com/prometheus-community/windows_exporter/internal/labelcache" "github.com/prometheus-community/windows_exporter/internal/mi" "github.com/prometheus-community/windows_exporter/internal/pdh" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus-community/windows_exporter/internal/utils/recovery" "github.com/prometheus/client_golang/prometheus" "golang.org/x/sys/windows" @@ -205,6 +206,11 @@ func (c *Collector) GetName() string { return Name } +// ScrapeCaches returns the caches that the collection resets and begins a scrape of around Collect. +func (c *Collector) ScrapeCaches() []scrapecache.Cache { + return []scrapecache.Cache{&c.labels} +} + func (c *Collector) Close() error { for _, closeFn := range c.closeFns { closeFn() @@ -212,7 +218,6 @@ func (c *Collector) Close() error { c.closeFns = nil c.collectorFns = nil - c.labels.Reset() c.perfQuery.Close() c.perfQuery = nil @@ -362,8 +367,7 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, maxScrapeDuration time. } } - scrape := c.labels.Begin() - defer scrape.Commit() + scrape := c.labels.Scrape() var g recovery.Group diff --git a/internal/collector/printer/printer.go b/internal/collector/printer/printer.go index 84dae2c2..bb12b4a4 100644 --- a/internal/collector/printer/printer.go +++ b/internal/collector/printer/printer.go @@ -28,6 +28,7 @@ import ( "github.com/prometheus-community/windows_exporter/internal/headers/winspool" "github.com/prometheus-community/windows_exporter/internal/metriccache" "github.com/prometheus-community/windows_exporter/internal/mi" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus-community/windows_exporter/internal/types" "github.com/prometheus-community/windows_exporter/internal/utils" "github.com/prometheus/client_golang/prometheus" @@ -129,8 +130,6 @@ func NewWithFlags(app *kingpin.Application) *Collector { } func (c *Collector) Close() error { - c.metricCache.Reset() - return nil } @@ -154,14 +153,16 @@ func (c *Collector) Build(_ *slog.Logger, _ *mi.Session) error { nil, ) - // Cached metrics refer to the descriptors of an earlier Build. - c.metricCache.Reset() - return nil } func (c *Collector) GetName() string { return Name } +// ScrapeCaches returns the caches that the collection resets and begins a scrape of around Collect. +func (c *Collector) ScrapeCaches() []scrapecache.Cache { + return []scrapecache.Cache{&c.metricCache} +} + // Collect reads printers and jobs from the print spooler. // Win32_Printer is not used because the CIMWin32 provider calls DeviceCapabilities // for every printer, which blocks for about a minute on an unreachable network printer. @@ -172,7 +173,7 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error return fmt.Errorf("failed to enumerate printers: %w", err) } - scrape := c.metricCache.Begin() + scrape := c.metricCache.Scrape() var errs []error @@ -218,9 +219,6 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error } } - // Printers that were not collected in this scrape are dropped from the cache. - scrape.Commit() - return errors.Join(errs...) } diff --git a/internal/collector/printer/printer_test.go b/internal/collector/printer/printer_test.go index d101e02c..4d53613e 100644 --- a/internal/collector/printer/printer_test.go +++ b/internal/collector/printer/printer_test.go @@ -51,5 +51,5 @@ func TestCollectorWithoutMISession(t *testing.T) { ch := make(chan prometheus.Metric, 1000) - require.NoError(t, c.Collect(ch, time.Second)) + require.NoError(t, testutils.Collect(c, ch, time.Second)) } diff --git a/internal/collector/process/process.go b/internal/collector/process/process.go index 51b714d9..a172b8ba 100644 --- a/internal/collector/process/process.go +++ b/internal/collector/process/process.go @@ -30,6 +30,7 @@ import ( "github.com/alecthomas/kingpin/v2" "github.com/prometheus-community/windows_exporter/internal/labelcache" "github.com/prometheus-community/windows_exporter/internal/mi" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus-community/windows_exporter/internal/types" "github.com/prometheus-community/windows_exporter/internal/utils" "github.com/prometheus/client_golang/prometheus" @@ -158,6 +159,11 @@ func (c *Collector) GetName() string { return Name } +// ScrapeCaches returns the caches that the collection resets and begins a scrape of around Collect. +func (c *Collector) ScrapeCaches() []scrapecache.Cache { + return []scrapecache.Cache{&c.labels} +} + func (c *Collector) Close() error { c.mu.Lock() defer c.mu.Unlock() @@ -165,7 +171,6 @@ func (c *Collector) Close() error { c.snapshot = snapshotBuffer{} c.processes = nil c.infoCache = nil - c.labels.Reset() // The session belongs to the caller of Build. c.miSession = nil @@ -362,8 +367,7 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, maxScrapeDuration time. infos := c.resolveProcessInfo(processes) - scrape := c.labels.Begin() - defer scrape.Commit() + scrape := c.labels.Scrape() for i, process := range processes { name := process.name diff --git a/internal/collector/scheduled_task/scheduled_task.go b/internal/collector/scheduled_task/scheduled_task.go index 4f435f07..72216bca 100644 --- a/internal/collector/scheduled_task/scheduled_task.go +++ b/internal/collector/scheduled_task/scheduled_task.go @@ -32,6 +32,7 @@ import ( "github.com/prometheus-community/windows_exporter/internal/mi" "github.com/prometheus-community/windows_exporter/internal/ole" "github.com/prometheus-community/windows_exporter/internal/ole/taskschd" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus-community/windows_exporter/internal/types" "github.com/prometheus-community/windows_exporter/internal/utils" "github.com/prometheus/client_golang/prometheus" @@ -172,9 +173,12 @@ func (c *Collector) GetName() string { return Name } -func (c *Collector) Close() error { - c.metricCache.Reset() +// ScrapeCaches returns the caches that the collection resets and begins a scrape of around Collect. +func (c *Collector) ScrapeCaches() []scrapecache.Cache { + return []scrapecache.Cache{&c.metricCache} +} +func (c *Collector) Close() error { return nil } @@ -212,9 +216,6 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { c.folderFilter = newFolderFilter(c.config.TaskInclude, c.config.TaskExclude) - // Cached metrics refer to the descriptors of an earlier Build. - c.metricCache.Reset() - return nil } @@ -261,7 +262,7 @@ func (c *Collector) taskReader() taskReader { } func (c *Collector) collectMetrics(ch chan<- prometheus.Metric, scheduledTasks ScheduledTasks) { - scrape := c.metricCache.Begin() + scrape := c.metricCache.Scrape() for _, task := range scheduledTasks { values := taskValues{ @@ -278,8 +279,6 @@ func (c *Collector) collectMetrics(ch chan<- prometheus.Metric, scheduledTasks S ch <- metric } } - - scrape.Commit() } // taskMetrics builds the metrics of a task. diff --git a/internal/collector/scheduled_task/scheduled_task_internal_test.go b/internal/collector/scheduled_task/scheduled_task_internal_test.go index 42bcc3b8..e64d01ce 100644 --- a/internal/collector/scheduled_task/scheduled_task_internal_test.go +++ b/internal/collector/scheduled_task/scheduled_task_internal_test.go @@ -24,6 +24,7 @@ import ( "testing" "github.com/alecthomas/kingpin/v2" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus/client_golang/prometheus" dto "github.com/prometheus/client_model/go" "github.com/prometheus/common/expfmt" @@ -252,7 +253,9 @@ func TestCollectMetricsCache(t *testing.T) { for i, scrape := range []ScheduledTasks{tasks, tasks, changed, changed} { // Collect once outside the registry, which calls Collect for Describe too. collected := make(chan prometheus.Metric, len(scrape)*19) + cacheScrape := scrapecache.Begin(c) c.collectMetrics(collected, scrape) + cacheScrape.End(nil) close(collected) var sent []prometheus.Metric @@ -305,7 +308,9 @@ func TestCollectMetricsCache(t *testing.T) { _, ok = scrape.Load("/Added", taskValues{state: TASK_STATE_QUEUED, lastTaskResult: SCHED_S_TASK_QUEUED}) require.True(t, ok) - // Build creates new descriptors, so the cache must not return metrics built from the old ones. + // Build creates new descriptors, so the collection resets the caches of + // the collector before it, which must include the metric cache. + scrapecache.Reset(c) require.NoError(t, c.Build(slog.New(slog.DiscardHandler), nil)) scrape = c.metricCache.Begin() @@ -363,7 +368,9 @@ func BenchmarkCollectMetrics(b *testing.B) { c.metricCache.Reset() } + cacheScrape := scrapecache.Begin(c) c.collectMetrics(ch, tasks) + cacheScrape.End(nil) for len(ch) > 0 { <-ch diff --git a/internal/collector/service/service.go b/internal/collector/service/service.go index 4045f594..f5304fa6 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/metriccache" "github.com/prometheus-community/windows_exporter/internal/mi" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus-community/windows_exporter/internal/types" "github.com/prometheus-community/windows_exporter/internal/utils" "github.com/prometheus-community/windows_exporter/internal/utils/recovery" @@ -176,6 +177,11 @@ func (c *Collector) GetName() string { return Name } +// ScrapeCaches returns the caches that the collection resets and begins a scrape of around Collect. +func (c *Collector) ScrapeCaches() []scrapecache.Cache { + return []scrapecache.Cache{&c.metricCache} +} + func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { c.logger = logger.With(slog.String("collector", Name)) @@ -212,9 +218,6 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { nil, ) - // Cached metrics refer to the descriptors of an earlier Build. - c.metricCache.Reset() - c.apiStateValues = map[uint32]string{ windows.SERVICE_CONTINUE_PENDING: "continue pending", windows.SERVICE_PAUSE_PENDING: "pause pending", @@ -269,8 +272,6 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { } 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), @@ -315,7 +316,7 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error ) } - scrape := c.metricCache.Begin() + scrape := c.metricCache.Scrape() servicesCh := make(chan serviceEntry, len(services)) @@ -340,9 +341,6 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error err = g.Wait() - // Services that were not collected in this scrape are dropped from the cache. - scrape.Commit() - return err } diff --git a/internal/collector/service/service_internal_test.go b/internal/collector/service/service_internal_test.go index 1a27a4b4..6833b74f 100644 --- a/internal/collector/service/service_internal_test.go +++ b/internal/collector/service/service_internal_test.go @@ -27,6 +27,7 @@ import ( "testing" "time" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -44,7 +45,12 @@ func (c serviceMetricCollector) Describe(ch chan<- *prometheus.Desc) { } func (c serviceMetricCollector) Collect(ch chan<- prometheus.Metric) { - assert.NoError(c.t, c.collector.Collect(ch, 0)) + // Collect within a scrape of the caches, as the collection does. + cacheScrape := scrapecache.Begin(c.collector) + err := c.collector.Collect(ch, 0) + cacheScrape.End(err) + + assert.NoError(c.t, err) } // serviceMetrics scrapes the collector and returns the value of each series of diff --git a/internal/collector/terminal_services/terminal_services.go b/internal/collector/terminal_services/terminal_services.go index fd41fcc5..0e16851a 100644 --- a/internal/collector/terminal_services/terminal_services.go +++ b/internal/collector/terminal_services/terminal_services.go @@ -31,6 +31,7 @@ import ( "github.com/prometheus-community/windows_exporter/internal/metriccache" "github.com/prometheus-community/windows_exporter/internal/mi" "github.com/prometheus-community/windows_exporter/internal/pdh" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus-community/windows_exporter/internal/types" "github.com/prometheus/client_golang/prometheus" "golang.org/x/sys/windows" @@ -141,14 +142,16 @@ func (c *Collector) GetName() string { return Name } +// ScrapeCaches returns the caches that the collection resets and begins a scrape of around Collect. +func (c *Collector) ScrapeCaches() []scrapecache.Cache { + return []scrapecache.Cache{&c.labels, &c.sessionCache} +} + func (c *Collector) Close() error { c.perfDataCollectorTerminalServicesSession.Close() c.perfDataCollectorBroker.Close() c.perfDataQuery.Close() - c.sessionCache.Reset() - c.labels.Reset() - // hServer is 0 if Build failed before opening the server. if c.hServer == 0 || c.hServer == windows.InvalidHandle { return nil @@ -163,9 +166,6 @@ func (c *Collector) Close() error { func (c *Collector) Build(logger *slog.Logger, miSession *mi.Session) error { c.logger = logger.With(slog.String("collector", Name)) - c.sessionCache.Reset() - c.labels.Reset() - c.sessionInfo = prometheus.NewDesc( prometheus.BuildFQName(types.Namespace, Name, "session_info"), "Terminal Services sessions info", @@ -356,8 +356,7 @@ func (c *Collector) collectTSSessionCounters(ch chan<- prometheus.Metric) error return fmt.Errorf("failed to collect Terminal Services Session metrics: %w", err) } - scrape := c.labels.Begin() - defer scrape.Commit() + scrape := c.labels.Scrape() names := make(map[string]bool) @@ -521,9 +520,7 @@ func (c *Collector) collectWTSSessions(ch chan<- prometheus.Metric) error { return fmt.Errorf("failed to enumerate WTS sessions: %w", err) } - scrape := c.sessionCache.Begin() - // Sessions that were not collected in this scrape are dropped from the cache. - defer scrape.Commit() + scrape := c.sessionCache.Scrape() for _, session := range sessions { // only connect metrics for remote named sessions diff --git a/internal/collector/terminal_services/terminal_services_internal_test.go b/internal/collector/terminal_services/terminal_services_internal_test.go index a13ff46c..787b4ef8 100644 --- a/internal/collector/terminal_services/terminal_services_internal_test.go +++ b/internal/collector/terminal_services/terminal_services_internal_test.go @@ -23,6 +23,7 @@ import ( "testing" "time" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -48,7 +49,10 @@ func TestCollectorWithoutSessionObject(t *testing.T) { for range 2 { ch := make(chan prometheus.Metric, 1024) - require.NoError(t, c.Collect(ch, time.Second)) + cacheScrape := scrapecache.Begin(c) + err := c.Collect(ch, time.Second) + cacheScrape.End(err) + require.NoError(t, err) close(ch) for metric := range ch { diff --git a/internal/collector/wmi/wmi.go b/internal/collector/wmi/wmi.go index 82bb6215..4020a14e 100644 --- a/internal/collector/wmi/wmi.go +++ b/internal/collector/wmi/wmi.go @@ -31,6 +31,7 @@ import ( "github.com/alecthomas/kingpin/v2" "github.com/prometheus-community/windows_exporter/internal/labelcache" "github.com/prometheus-community/windows_exporter/internal/mi" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus-community/windows_exporter/internal/types" "github.com/prometheus/client_golang/prometheus" "go.yaml.in/yaml/v3" @@ -144,9 +145,12 @@ func (c *Collector) GetName() string { return Name } -func (c *Collector) Close() error { - c.labels.Reset() +// ScrapeCaches returns the caches that the collection resets and begins a scrape of around Collect. +func (c *Collector) ScrapeCaches() []scrapecache.Cache { + return []scrapecache.Cache{&c.labels} +} +func (c *Collector) Close() error { return nil } @@ -459,8 +463,7 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, maxScrapeDuration time. deadline = time.Now().Add(maxScrapeDuration) } - scrape := c.labels.Begin() - defer scrape.Commit() + scrape := c.labels.Scrape() var errs []error diff --git a/internal/labelcache/labelcache.go b/internal/labelcache/labelcache.go index a6e94f62..1baef6c3 100644 --- a/internal/labelcache/labelcache.go +++ b/internal/labelcache/labelcache.go @@ -25,10 +25,15 @@ // Label pairs are interned: all series with the same label name and value, // such as the process and process_id labels of the series of one process, // share one pair, so the cache keeps little more than one slice per series. +// +// A collector lists its caches in ScrapeCaches (see internal/scrapecache). +// The collection resets them around Build and Close and begins and ends a +// scrape around each Collect call, which gets it from [Cache.Scrape]. package labelcache import ( "sync" + "sync/atomic" "github.com/prometheus/client_golang/prometheus" dto "github.com/prometheus/client_model/go" @@ -117,6 +122,14 @@ type Cache struct { // resets counts Reset calls, so that a scrape begun before a Reset doesn't // add series afterwards. resets uint64 + + // scrapeMu guards active and the creation of current, so that the + // workers of a scrape create one scrape and none outlives EndScrape. + scrapeMu sync.Mutex + // active is set between BeginScrape and EndScrape. + active bool + // current is the scrape that Scrape created after BeginScrape, or nil. + current atomic.Pointer[Scrape] } // Scrape creates the metrics of one scrape. @@ -128,9 +141,14 @@ type Scrape struct { // metrics without the cache, so they can't add series to the set the // commit decided. committed bool + // used is set by the first MustNewConstMetric call. + used atomic.Bool } // Begin starts a scrape. Call Commit when it is done. +// +// Collectors don't call Begin: the collection begins and ends their scrapes +// through BeginScrape and EndScrape, and Collect gets its scrape from [Cache.Scrape]. func (c *Cache) Begin() *Scrape { c.mu.Lock() defer c.mu.Unlock() @@ -140,6 +158,76 @@ func (c *Cache) Begin() *Scrape { return &Scrape{cache: c, generation: c.generation, resets: c.resets} } +// BeginScrape begins a scrape. The collection calls it before Collect. +func (c *Cache) BeginScrape() { + c.scrapeMu.Lock() + c.current.Store(nil) + c.active = true + c.scrapeMu.Unlock() +} + +// EndScrape ends the scrape begun by BeginScrape. The collection calls it +// once Collect returned err. +// +// A scrape that Collect never asked for leaves the cache unchanged. Otherwise +// the scrape is committed, unless Collect failed before it created a metric: +// then the series of the previous scrapes stay. +func (c *Cache) EndScrape(err error) { + c.scrapeMu.Lock() + c.active = false + s := c.current.Swap(nil) + c.scrapeMu.Unlock() + + if s == nil { + return + } + + if err != nil && !s.used.Load() { + s.discard() + + return + } + + s.Commit() +} + +// Scrape returns the scrape of the running Collect call, which it creates on +// the first call. It is nil outside of a scrape begun by the collection, for +// example when Collect is called directly; a nil Scrape creates metrics +// without the cache. +// +// Only Collect and the workers it waits for may call Scrape: after Collect +// returned, it returns nil or the scrape of a later Collect call. +func (c *Cache) Scrape() *Scrape { + if s := c.current.Load(); s != nil { + return s + } + + c.scrapeMu.Lock() + defer c.scrapeMu.Unlock() + + // Another worker of the scrape may have created it meanwhile. + if s := c.current.Load(); s != nil { + return s + } + + if !c.active { + return nil + } + + s := c.Begin() + c.current.Store(s) + + return s +} + +// discard ends the scrape without dropping any series. +func (s *Scrape) discard() { + s.cache.mu.Lock() + s.committed = true + s.cache.mu.Unlock() +} + // Reset drops all cached label pairs, for example when the collector closes. func (c *Cache) Reset() { c.mu.Lock() @@ -225,6 +313,10 @@ func (c *Cache) intern(labels []*dto.LabelPair) []*dto.LabelPair { // //nolint:ireturn // Const metrics have no exported type. func (s *Scrape) MustNewConstMetric(desc *prometheus.Desc, valueType prometheus.ValueType, value float64, labelValues ...string) prometheus.Metric { + if s != nil && !s.used.Load() { + s.used.Store(true) + } + if s == nil || len(labelValues) > maxLabels || !knownValueType(valueType) { return prometheus.MustNewConstMetric(desc, valueType, value, append([]string(nil), labelValues...)...) } diff --git a/internal/labelcache/labelcache_test.go b/internal/labelcache/labelcache_test.go index adbed585..fa6dbc5b 100644 --- a/internal/labelcache/labelcache_test.go +++ b/internal/labelcache/labelcache_test.go @@ -16,6 +16,7 @@ package labelcache_test import ( + "errors" "runtime" "strconv" "sync" @@ -358,3 +359,71 @@ func TestCommittedScrape(t *testing.T) { require.NotSame(t, late[1], write(t, next.MustNewConstMetric(gaugeDesc, prometheus.GaugeValue, 1, "p", "2")).GetLabel()[1], "late series was not cached") next.Commit() } + +// scrapeWith runs one scrape of the pids the way the collection does around +// a Collect call that returns err, and returns the label pairs of each series. +func scrapeWith(t *testing.T, cache *labelcache.Cache, err error, pids ...string) [][]*dto.LabelPair { + t.Helper() + + cache.BeginScrape() + defer cache.EndScrape(err) + + scrape := cache.Scrape() + labels := make([][]*dto.LabelPair, 0, len(pids)) + + for _, pid := range pids { + labels = append(labels, write(t, scrape.MustNewConstMetric(gaugeDesc, prometheus.GaugeValue, 1, "p", pid)).GetLabel()) + } + + return labels +} + +func TestBeginEndScrape(t *testing.T) { + t.Parallel() + + var cache labelcache.Cache + + require.Nil(t, cache.Scrape(), "no scrape outside of BeginScrape") + + first := scrapeWith(t, &cache, nil, "1", "2") + require.Nil(t, cache.Scrape(), "EndScrape ends the scrape") + + second := scrapeWith(t, &cache, nil, "1", "2") + require.Same(t, first[0][1], second[0][1], "label pairs reused") + require.Same(t, first[1][1], second[1][1]) + + // A Collect call that never asked for the scrape leaves the cache unchanged. + cache.BeginScrape() + cache.EndScrape(nil) + + // A failed Collect that created no metric keeps the series. + scrapeWith(t, &cache, errors.New("read failed")) + + third := scrapeWith(t, &cache, nil, "1", "2") + require.Same(t, first[1][1], third[1][1], "failed empty scrape kept the series") + + // A failed Collect that created metrics commits its series. + scrapeWith(t, &cache, errors.New("partial"), "1") + + fourth := scrapeWith(t, &cache, nil, "1", "2") + require.Same(t, first[0][1], fourth[0][1]) + require.NotSame(t, first[1][1], fourth[1][1], "partial scrape dropped series 2") + + // A successful empty scrape drops every series. + scrapeWith(t, &cache, nil) + + fifth := scrapeWith(t, &cache, nil, "1") + require.NotSame(t, fourth[0][1], fifth[0][1]) + + // Reset drops the series, also those of a running scrape. + cache.BeginScrape() + write(t, cache.Scrape().MustNewConstMetric(gaugeDesc, prometheus.GaugeValue, 1, "p", "1")) + cache.Reset() + cache.EndScrape(nil) + + sixth := scrapeWith(t, &cache, nil, "1") + require.NotSame(t, fifth[0][1], sixth[0][1]) + + // EndScrape without a scrape does nothing. + cache.EndScrape(nil) +} diff --git a/internal/metriccache/metriccache.go b/internal/metriccache/metriccache.go index 77a1334d..f4e662e6 100644 --- a/internal/metriccache/metriccache.go +++ b/internal/metriccache/metriccache.go @@ -20,10 +20,15 @@ // the allocations of collectors that publish many state-set series per object. // Const metrics are immutable and a registry only reads them during Gather, so // sending the same metric again in a later scrape is safe. +// +// A collector lists its caches in ScrapeCaches (see internal/scrapecache). +// The collection resets them around Build and Close and begins and ends a +// scrape around each Collect call, which gets it from [Cache.Scrape]. package metriccache import ( "sync" + "sync/atomic" "github.com/prometheus/client_golang/prometheus" ) @@ -40,6 +45,14 @@ type Cache[K, V comparable] struct { // resets counts Reset calls, so that a scrape begun before a Reset // doesn't commit afterwards. resets uint64 + + // scrapeMu guards active and the creation of current, so that the + // workers of a scrape create one scrape and none outlives EndScrape. + scrapeMu sync.Mutex + // active is set between BeginScrape and EndScrape. + active bool + // current is the scrape that Scrape created after BeginScrape, or nil. + current atomic.Pointer[Scrape[K, V]] } type entry[V comparable] struct { @@ -48,7 +61,8 @@ type entry[V comparable] struct { } // Scrape collects the entries of one scrape. Its methods may be called by the -// concurrent workers of one scrape. +// concurrent workers of one scrape. A nil Scrape builds every metric and +// caches nothing. type Scrape[K, V comparable] struct { cache *Cache[K, V] resets uint64 @@ -63,6 +77,10 @@ type Scrape[K, V comparable] struct { } // Begin starts a scrape based on the last committed generation. +// +// Collectors don't call Begin: the collection begins and ends their scrapes +// through BeginScrape and EndScrape, and Collect gets its scrape from +// [Cache.Scrape]. func (c *Cache[K, V]) Begin() *Scrape[K, V] { c.mu.Lock() prev, resets := c.entries, c.resets @@ -76,6 +94,73 @@ func (c *Cache[K, V]) Begin() *Scrape[K, V] { } } +// BeginScrape begins a scrape. The collection calls it before Collect. +func (c *Cache[K, V]) BeginScrape() { + c.scrapeMu.Lock() + c.current.Store(nil) + c.active = true + c.scrapeMu.Unlock() +} + +// EndScrape ends the scrape begun by BeginScrape. The collection calls it +// once Collect returned err. +// +// A scrape that Collect never asked for leaves the cache unchanged. Otherwise +// the scrape is committed, unless Collect failed before it loaded or stored an +// entry: then the entries of the previous scrape stay, so a failed read of the +// objects doesn't drop them. +func (c *Cache[K, V]) EndScrape(err error) { + c.scrapeMu.Lock() + c.active = false + s := c.current.Swap(nil) + c.scrapeMu.Unlock() + + if s == nil { + return + } + + s.mu.Lock() + empty := len(s.next) == 0 + s.mu.Unlock() + + if err != nil && empty { + s.discard() + + return + } + + s.Commit() +} + +// Scrape returns the scrape of the running Collect call, which it creates on +// the first call. It is nil outside of a scrape begun by the collection, for +// example when Collect is called directly. +// +// Only Collect and the workers it waits for may call Scrape: after Collect +// returned, it returns nil or the scrape of a later Collect call. +func (c *Cache[K, V]) Scrape() *Scrape[K, V] { + if s := c.current.Load(); s != nil { + return s + } + + c.scrapeMu.Lock() + defer c.scrapeMu.Unlock() + + // Another worker of the scrape may have created it meanwhile. + if s := c.current.Load(); s != nil { + return s + } + + if !c.active { + return nil + } + + s := c.Begin() + c.current.Store(s) + + return s +} + // Reset drops all cached metrics, for example after their descriptors changed. // Scrapes begun before Reset don't commit their entries. func (c *Cache[K, V]) Reset() { @@ -89,6 +174,10 @@ func (c *Cache[K, V]) Reset() { // an equal value, and keeps them for the next scrape. The caller must not // modify the returned slice. func (s *Scrape[K, V]) Load(key K, value V) ([]prometheus.Metric, bool) { + if s == nil { + return nil, false + } + e, ok := s.prev[key] if !ok || e.value != value { return nil, false @@ -106,6 +195,10 @@ func (s *Scrape[K, V]) Load(key K, value V) ([]prometheus.Metric, bool) { // Store keeps metrics built from value for key. The caller must not modify // metrics afterwards. func (s *Scrape[K, V]) Store(key K, value V, metrics []prometheus.Metric) { + if s == nil { + return + } + s.mu.Lock() if s.next != nil { s.next[key] = entry[V]{value: value, metrics: metrics} @@ -133,11 +226,11 @@ func (s *Scrape[K, V]) Get(key K, value V, build func() []prometheus.Metric) []p // while a timed-out scrape was still collecting. Calls on the scrape after // Commit don't change the cache. func (s *Scrape[K, V]) Commit() { - s.mu.Lock() - next := s.next - s.next = nil - s.mu.Unlock() + if s == nil { + return + } + next := s.discard() if next == nil { // Committed before. return @@ -149,3 +242,14 @@ func (s *Scrape[K, V]) Commit() { } s.cache.mu.Unlock() } + +// discard ends the scrape without changing the cache and returns its entries. +func (s *Scrape[K, V]) discard() map[K]entry[V] { + s.mu.Lock() + defer s.mu.Unlock() + + next := s.next + s.next = nil + + return next +} diff --git a/internal/metriccache/metriccache_test.go b/internal/metriccache/metriccache_test.go index baf2c0ed..d7ee6467 100644 --- a/internal/metriccache/metriccache_test.go +++ b/internal/metriccache/metriccache_test.go @@ -16,6 +16,7 @@ package metriccache_test import ( + "errors" "strconv" "sync" "testing" @@ -221,6 +222,125 @@ func TestScrapeConcurrentWorkers(t *testing.T) { } } +func TestNilScrape(t *testing.T) { + t.Parallel() + + var cache metriccache.Cache[string, string] + + scrape := cache.Scrape() + require.Nil(t, scrape, "no scrape outside of BeginScrape") + + built := 0 + buildCounted := func() []prometheus.Metric { + built++ + + return build("a", "up") + } + + metrics := scrape.Get("a", "up", buildCounted) + require.Len(t, metrics, 1) + + _, ok := scrape.Load("a", "up") + require.False(t, ok) + scrape.Store("a", "up", metrics) + scrape.Commit() + + scrape.Get("a", "up", buildCounted) + require.Equal(t, 2, built, "a nil scrape caches nothing") +} + +// scrapeWith runs one scrape of keys the way the collection does around a +// Collect call that returns err. +func scrapeWith(cache *metriccache.Cache[string, string], err error, keys ...string) []prometheus.Metric { + cache.BeginScrape() + defer cache.EndScrape(err) + + scrape := cache.Scrape() + metrics := make([]prometheus.Metric, 0, len(keys)) + + for _, key := range keys { + metrics = append(metrics, scrape.Get(key, "up", func() []prometheus.Metric { + return build(key, "up") + })...) + } + + return metrics +} + +func TestBeginEndScrape(t *testing.T) { + t.Parallel() + + var cache metriccache.Cache[string, string] + + first := scrapeWith(&cache, nil, "a", "b") + require.Nil(t, cache.Scrape(), "EndScrape ends the scrape") + + second := scrapeWith(&cache, nil, "a", "b") + require.Same(t, first[0], second[0], "metrics reused") + require.Same(t, first[1], second[1]) + + // A Collect call that never asked for the scrape leaves the cache unchanged. + cache.BeginScrape() + cache.EndScrape(nil) + + // A failed Collect that used no entry keeps the previous ones. + scrapeWith(&cache, errors.New("read failed")) + + third := scrapeWith(&cache, nil, "a", "b") + require.Same(t, first[0], third[0], "failed empty scrape kept the entries") + + // A failed Collect that used entries commits them. + scrapeWith(&cache, errors.New("partial"), "a") + + fourth := scrapeWith(&cache, nil, "a", "b") + require.Same(t, first[0], fourth[0]) + require.NotSame(t, first[1], fourth[1], "partial scrape dropped b") + + // A successful empty scrape drops every entry. + scrapeWith(&cache, nil) + + fifth := scrapeWith(&cache, nil, "a") + require.NotSame(t, fourth[0], fifth[0]) + + // Reset drops the entries, also those of a running scrape. + cache.BeginScrape() + cache.Scrape().Get("a", "up", func() []prometheus.Metric { return build("a", "up") }) + cache.Reset() + cache.EndScrape(nil) + + sixth := scrapeWith(&cache, nil, "a") + require.NotSame(t, fifth[0], sixth[0]) + + // EndScrape without a scrape does nothing. + cache.EndScrape(nil) +} + +func TestScrapeCreatedOnce(t *testing.T) { + t.Parallel() + + var ( + cache metriccache.Cache[string, string] + wg sync.WaitGroup + scrapes [8]*metriccache.Scrape[string, string] + ) + + cache.BeginScrape() + + for i := range scrapes { + wg.Go(func() { + scrapes[i] = cache.Scrape() + }) + } + + wg.Wait() + cache.EndScrape(nil) + + for _, scrape := range scrapes { + require.NotNil(t, scrape) + require.Same(t, scrapes[0], scrape, "the workers of a scrape share one scrape") + } +} + func BenchmarkScrape(b *testing.B) { var cache metriccache.Cache[string, int] diff --git a/internal/scrapecache/scrapecache.go b/internal/scrapecache/scrapecache.go new file mode 100644 index 00000000..e7f05020 --- /dev/null +++ b/internal/scrapecache/scrapecache.go @@ -0,0 +1,86 @@ +// SPDX-License-Identifier: Apache-2.0 +// +// Copyright The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package scrapecache lets the collection own the caches that collectors keep +// between scrapes, such as internal/labelcache and internal/metriccache. +// +// A collector lists its caches with ScrapeCaches. The collection resets them +// before Build and after Close, and begins a scrape of each cache before +// Collect and ends it after Collect returned. Collect calls of one collector +// never overlap, so a scrape always belongs to exactly one Collect call, even +// one that outlives its scrape budget. +// +// Outside of the collection, for example when a test or a library consumer +// calls Collect directly, the caches have no scrape and every metric is built +// without them. +package scrapecache + +// Cache is a cache that a collector keeps between scrapes. +type Cache interface { + // BeginScrape begins the scrape of a Collect call. + BeginScrape() + // EndScrape ends the scrape after Collect returned err. + EndScrape(err error) + // Reset drops the cached entries. Their descriptors may be stale after + // a new Build. + Reset() +} + +// Owner is implemented by collectors that keep caches between scrapes. +type Owner interface { + // ScrapeCaches returns the caches of the collector. The collection may + // call it before Build. + ScrapeCaches() []Cache +} + +// Scrape is a scrape of the caches of a collector. +type Scrape struct { + caches []Cache +} + +// Begin begins a scrape of the caches of collector, if it is an [Owner]. +// Call End with the error of Collect once Collect returned. +func Begin(collector any) Scrape { + owner, ok := collector.(Owner) + if !ok { + return Scrape{} + } + + caches := owner.ScrapeCaches() + for _, cache := range caches { + cache.BeginScrape() + } + + return Scrape{caches: caches} +} + +// End ends the scrape after Collect returned err. +func (s Scrape) End(err error) { + for _, cache := range s.caches { + cache.EndScrape(err) + } +} + +// Reset resets the caches of collector, if it is an [Owner]. +func Reset(collector any) { + owner, ok := collector.(Owner) + if !ok { + return + } + + for _, cache := range owner.ScrapeCaches() { + cache.Reset() + } +} diff --git a/internal/scrapecache/scrapecache_test.go b/internal/scrapecache/scrapecache_test.go new file mode 100644 index 00000000..84b7e64d --- /dev/null +++ b/internal/scrapecache/scrapecache_test.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. + +package scrapecache_test + +import ( + "errors" + "fmt" + "testing" + + "github.com/prometheus-community/windows_exporter/internal/scrapecache" + "github.com/stretchr/testify/require" +) + +type cache struct { + name string + events *[]string +} + +func (c cache) BeginScrape() { + *c.events = append(*c.events, c.name+" begin") +} + +func (c cache) EndScrape(err error) { + *c.events = append(*c.events, fmt.Sprintf("%s end %v", c.name, err)) +} + +func (c cache) Reset() { + *c.events = append(*c.events, c.name+" reset") +} + +type owner struct { + events []string + calls int +} + +func (o *owner) ScrapeCaches() []scrapecache.Cache { + o.calls++ + + return []scrapecache.Cache{cache{name: "a", events: &o.events}, cache{name: "b", events: &o.events}} +} + +func TestOwner(t *testing.T) { + t.Parallel() + + o := &owner{} + + scrapecache.Reset(o) + + cacheScrape := scrapecache.Begin(o) + require.Equal(t, []string{"a reset", "b reset", "a begin", "b begin"}, o.events) + + cacheScrape.End(errors.New("failed")) + require.Equal(t, []string{"a end failed", "b end failed"}, o.events[4:]) + require.Equal(t, 2, o.calls, "Begin asks for the caches once per scrape") +} + +func TestNotAnOwner(t *testing.T) { + t.Parallel() + + scrapecache.Reset(struct{}{}) + scrapecache.Begin(struct{}{}).End(errors.New("ignored")) +} diff --git a/internal/utils/testutils/testutils.go b/internal/utils/testutils/testutils.go index 2fb1e588..949f46cf 100644 --- a/internal/utils/testutils/testutils.go +++ b/internal/utils/testutils/testutils.go @@ -31,6 +31,7 @@ import ( "github.com/prometheus-community/windows_exporter/internal/collector/update" "github.com/prometheus-community/windows_exporter/internal/mi" "github.com/prometheus-community/windows_exporter/internal/pdh" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus-community/windows_exporter/internal/types" "github.com/prometheus-community/windows_exporter/pkg/collector" "github.com/prometheus/client_golang/prometheus" @@ -69,10 +70,22 @@ func FuncBenchmarkCollector[C collector.Collector](b *testing.B, name string, co }) for b.Loop() { - require.NoError(b, c.Collect(metrics, 0)) + require.NoError(b, Collect(c, metrics, 0)) } } +// Collect calls c.Collect within a scrape of the caches of c, as the +// collection does. Calling c.Collect directly collects without the caches. +func Collect(c collector.Collector, ch chan<- prometheus.Metric, maxScrapeDuration time.Duration) (err error) { + cacheScrape := scrapecache.Begin(c) + + defer func() { + cacheScrape.End(err) + }() + + return c.Collect(ch, maxScrapeDuration) +} + // TestCollector validates real Windows collector output. CI lists provisioned // collectors in WINDOWS_EXPORTER_TEST_COLLECTORS so setup failures cannot skip. func TestCollector[C collector.Collector, V any](t *testing.T, fn func(*V) C, conf *V) map[string]*dto.MetricFamily { @@ -89,7 +102,14 @@ func TestCollector[C collector.Collector, V any](t *testing.T, fn func(*V) C, co miSession, err := miApp.NewSession(nil) require.NoError(t, err) t.Cleanup(func() { assert.NoError(t, miSession.Close()) }) - t.Cleanup(func() { assert.NoError(t, c.Close()) }) + t.Cleanup(func() { + assert.NoError(t, c.Close()) + scrapecache.Reset(c) + }) + + // Like the collection, start without cached metrics and collect through + // the caches, so the second scrape publishes cached metrics. + scrapecache.Reset(c) if err := c.Build(logger, miSession); err != nil { if !required && unsupportedBuild(err) { @@ -159,13 +179,13 @@ func collectMetrics(t *testing.T, c collector.Collector, required bool) map[stri wg.Wait() }() - err := c.Collect(ch, 30*time.Second) + err := Collect(c, ch, 30*time.Second) if errors.Is(err, update.ErrNoUpdates) && required { deadline := time.Now().Add(time.Minute) for errors.Is(err, update.ErrNoUpdates) && time.Now().Before(deadline) { time.Sleep(100 * time.Millisecond) - err = c.Collect(ch, 30*time.Second) + err = Collect(c, ch, 30*time.Second) } } diff --git a/pkg/collector/collect.go b/pkg/collector/collect.go index 4fff98bf..aee87e74 100644 --- a/pkg/collector/collect.go +++ b/pkg/collector/collect.go @@ -28,6 +28,7 @@ import ( "time" "github.com/prometheus-community/windows_exporter/internal/pdh" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" "github.com/prometheus-community/windows_exporter/internal/types" "github.com/prometheus/client_golang/prometheus" "golang.org/x/sys/windows" @@ -251,15 +252,32 @@ func (c *Collection) collectCollector(ctx context.Context, ch chan<- prometheus. } // collect calls Collector.Collect and turns a panic into an error. +// It begins a scrape of the collector's caches before and ends it after the call. func collect(name string, collector Collector, ch chan<- prometheus.Metric, maxScrapeDuration time.Duration) (err error) { + var cacheScrape scrapecache.Scrape + + // Registered before the scrape of the caches begins, so a panic in the + // caches or in Collect is turned into an error, and the scrape ends with it. defer func() { if r := recover(); r != nil { err = fmt.Errorf("panic in collector %s: %v. stack: %s", name, r, string(debug.Stack()), ) } + + defer func() { + if r := recover(); r != nil { + err = errors.Join(err, fmt.Errorf("panic in the caches of collector %s: %v. stack: %s", name, r, + string(debug.Stack()), + )) + } + }() + + cacheScrape.End(err) }() + cacheScrape = scrapecache.Begin(collector) + return collector.Collect(ch, maxScrapeDuration) } diff --git a/pkg/collector/scrapecache_test.go b/pkg/collector/scrapecache_test.go new file mode 100644 index 00000000..d3a95f03 --- /dev/null +++ b/pkg/collector/scrapecache_test.go @@ -0,0 +1,388 @@ +// 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 + +import ( + "context" + "errors" + "fmt" + "log/slog" + "strings" + "sync" + "sync/atomic" + "testing" + "testing/synctest" + "time" + + "github.com/prometheus-community/windows_exporter/internal/labelcache" + "github.com/prometheus-community/windows_exporter/internal/metriccache" + "github.com/prometheus-community/windows_exporter/internal/mi" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" + "github.com/prometheus-community/windows_exporter/pkg/collector" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/common/expfmt" + "github.com/stretchr/testify/require" +) + +// eventLog records the calls of a collector and its cache in order. +type eventLog struct { + mu sync.Mutex + events []string +} + +func (l *eventLog) addf(format string, args ...any) { + l.mu.Lock() + l.events = append(l.events, fmt.Sprintf(format, args...)) + l.mu.Unlock() +} + +func (l *eventLog) take() []string { + l.mu.Lock() + defer l.mu.Unlock() + + events := l.events + l.events = nil + + return events +} + +// recordingCache is a scrapecache.Cache that records its calls. +type recordingCache struct { + log *eventLog + active atomic.Bool + beginPanic atomic.Bool + endPanic atomic.Bool +} + +func (c *recordingCache) BeginScrape() { + if c.beginPanic.Load() { + panic("begin panic") + } + + c.active.Store(true) + c.log.addf("begin") +} + +func (c *recordingCache) EndScrape(err error) { + c.active.Store(false) + + if err != nil { + // Keep the first line, a panic error contains the stack. + err = errors.New(strings.SplitN(err.Error(), ".", 2)[0]) + } + + c.log.addf("end %v", err) + + if c.endPanic.Load() { + panic("end panic") + } +} + +func (c *recordingCache) Reset() { + c.log.addf("reset") +} + +// cacheOwner is a collector with a recordingCache. +type cacheOwner struct { + log *eventLog + cache recordingCache + collectErr error + collectPanic bool + delay time.Duration +} + +func newCacheOwner() *cacheOwner { + log := &eventLog{} + + return &cacheOwner{log: log, cache: recordingCache{log: log}} +} + +func (c *cacheOwner) GetName() string { return "test" } + +func (c *cacheOwner) ScrapeCaches() []scrapecache.Cache { + return []scrapecache.Cache{&c.cache} +} + +func (c *cacheOwner) Build(*slog.Logger, *mi.Session) error { + c.log.addf("build") + + return nil +} + +func (c *cacheOwner) Collect(ch chan<- prometheus.Metric, _ time.Duration) error { + c.log.addf("collect in scrape=%t", c.cache.active.Load()) + + time.Sleep(c.delay) + + if c.collectPanic { + panic("collect panic") + } + + ch <- prometheus.MustNewConstMetric(prometheus.NewDesc("windows_test", "Test metric", nil, nil), prometheus.GaugeValue, 1) + + if c.delay > 0 { + c.log.addf("collect returned") + } + + return c.collectErr +} + +func (c *cacheOwner) Close() error { + c.log.addf("close") + + return nil +} + +func TestScrapeCacheLifecycle(t *testing.T) { + owner := newCacheOwner() + collection := collector.New(collector.Map{"test": owner}) + + require.NoError(t, collection.Build(t.Context(), slog.New(slog.DiscardHandler))) + require.Equal(t, []string{"reset", "build"}, owner.log.take(), "caches are reset before Build") + + handler, err := collection.NewHandler(time.Minute, slog.New(slog.DiscardHandler), nil) + require.NoError(t, err) + + scrape(t, handler) + require.Equal(t, []string{"begin", "collect in scrape=true", "end "}, owner.log.take()) + + owner.collectErr = errors.New("collect failed") + + scrape(t, handler) + require.Equal(t, []string{"begin", "collect in scrape=true", "end collect failed"}, owner.log.take(), + "EndScrape gets the error of Collect") + + owner.collectErr = nil + owner.collectPanic = true + + scrape(t, handler) + require.Equal(t, []string{"begin", "collect in scrape=true", "end panic in collector test: collect panic"}, owner.log.take(), + "EndScrape gets the error of a panic") + + require.NoError(t, collection.Close()) + require.Equal(t, []string{"close", "reset"}, owner.log.take(), "caches are reset after Close") +} + +// TestScrapeCachePanics verifies that a panic in a cache fails the collector +// like a panic in Collect, instead of crashing the exporter. +func TestScrapeCachePanics(t *testing.T) { + owner := newCacheOwner() + collection := collector.New(collector.Map{"test": owner}) + + require.NoError(t, collection.Build(t.Context(), slog.New(slog.DiscardHandler))) + t.Cleanup(func() { require.NoError(t, collection.Close()) }) + + handler, err := collection.NewHandler(time.Minute, slog.New(slog.DiscardHandler), nil) + require.NoError(t, err) + + owner.log.take() + owner.cache.endPanic.Store(true) + + out := scrape(t, handler) + require.Contains(t, out, `windows_exporter_collector_success{collector="test"} 0`) + require.Contains(t, out, "windows_test 1", "metrics of Collect are kept") + require.Equal(t, []string{"begin", "collect in scrape=true", "end "}, owner.log.take()) + + owner.cache.endPanic.Store(false) + owner.cache.beginPanic.Store(true) + + out = scrape(t, handler) + require.Contains(t, out, `windows_exporter_collector_success{collector="test"} 0`) + require.Empty(t, owner.log.take(), "Collect is not called when the scrape of the caches fails to begin") + + owner.cache.beginPanic.Store(false) + + out = scrape(t, handler) + require.Contains(t, out, `windows_exporter_collector_success{collector="test"} 1`) + require.Equal(t, []string{"begin", "collect in scrape=true", "end "}, owner.log.take()) +} + +func TestScrapeCacheEndsAfterTimedOutCollect(t *testing.T) { + // The collector's delay and the scrape deadline use synthetic time. + synctest.Test(t, func(t *testing.T) { + owner := newCacheOwner() + owner.delay = 2 * time.Second + + collection := collector.New(collector.Map{"test": owner}) + require.NoError(t, collection.Build(t.Context(), slog.New(slog.DiscardHandler))) + owner.log.take() + + handler, err := collection.NewHandler(300*time.Millisecond, slog.New(slog.DiscardHandler), nil) + require.NoError(t, err) + + out := scrape(t, handler) + require.Contains(t, out, `windows_exporter_collector_timeout{collector="test"} 1`) + require.Equal(t, []string{"begin", "collect in scrape=true"}, owner.log.take(), + "the scrape of the caches outlives the scrape budget") + + time.Sleep(2 * time.Second) + synctest.Wait() + + require.Equal(t, []string{"collect returned", "end "}, owner.log.take(), + "the scrape ends once Collect returned") + + require.NoError(t, collection.Close()) + }) +} + +func TestScrapeCacheDirectCollect(t *testing.T) { + t.Parallel() + + owner := newCacheOwner() + + require.NoError(t, owner.Collect(make(chan prometheus.Metric, 1), time.Second)) + require.Equal(t, []string{"collect in scrape=false"}, owner.log.take(), + "a direct Collect call runs without a scrape") +} + +// cachedCollector publishes one series through a labelcache and one through a +// metriccache, like the built-in collectors. +type cachedCollector struct { + labels labelcache.Cache + metrics metriccache.Cache[string, float64] + + builds atomic.Int32 + counter *prometheus.Desc + state *prometheus.Desc + value float64 +} + +func (c *cachedCollector) GetName() string { return "cached" } + +func (c *cachedCollector) ScrapeCaches() []scrapecache.Cache { + return []scrapecache.Cache{&c.labels, &c.metrics} +} + +func (c *cachedCollector) Build(*slog.Logger, *mi.Session) error { + c.counter = prometheus.NewDesc("windows_test_counter_total", "Test counter", []string{"name", "id"}, nil) + c.state = prometheus.NewDesc("windows_test_state", "Test state", []string{"name", "state"}, nil) + + return nil +} + +func (c *cachedCollector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error { + ch <- c.labels.Scrape().MustNewConstMetric(c.counter, prometheus.CounterValue, c.value, "a", "1") + + for _, metric := range c.metrics.Scrape().Get("a", c.value, func() []prometheus.Metric { + c.builds.Add(1) + + return []prometheus.Metric{ + prometheus.MustNewConstMetric(c.state, prometheus.GaugeValue, 1, "a", "running"), + prometheus.MustNewConstMetric(c.state, prometheus.GaugeValue, c.value, "a", "value"), + } + }) { + ch <- metric + } + + return nil +} + +func (c *cachedCollector) Close() error { return nil } + +// TestScrapeCacheReuse verifies that the collection's scrapes reuse the cached +// metrics and that they publish the same series and values as a Collect call +// without the caches. +func TestScrapeCacheReuse(t *testing.T) { + cached := &cachedCollector{value: 1} + collection := collector.New(collector.Map{"cached": cached}) + + require.NoError(t, collection.Build(context.Background(), slog.New(slog.DiscardHandler))) + t.Cleanup(func() { require.NoError(t, collection.Close()) }) + + handler, err := collection.NewHandler(time.Minute, slog.New(slog.DiscardHandler), nil) + require.NoError(t, err) + + uncached := func() string { + direct := &cachedCollector{value: cached.value} + require.NoError(t, direct.Build(nil, nil)) + + out, err := gatherCollector(direct) + require.NoError(t, err) + + return out + } + + for i, value := range []float64{1, 1, 2, 2} { + cached.value = value + + out := scrape(t, handler) + want := uncached() + + for line := range strings.Lines(want) { + if strings.HasPrefix(line, "windows_test_") { + require.Contains(t, out, line, "scrape %d", i) + } + } + } + + require.Equal(t, int32(2), cached.builds.Load(), "metrics are built once per value") + + // A new collection that builds the same collector again, as Grafana Alloy + // does, doesn't publish metrics built from the old descriptors. + require.NoError(t, collection.Close()) + + collection = collector.New(collector.Map{"cached": cached}) + require.NoError(t, collection.Build(context.Background(), slog.New(slog.DiscardHandler))) + + handler, err = collection.NewHandler(time.Minute, slog.New(slog.DiscardHandler), nil) + require.NoError(t, err) + + scrape(t, handler) + require.Equal(t, int32(3), cached.builds.Load()) +} + +// gatherCollector calls Collect of c directly and returns its metrics in the text format. +func gatherCollector(c collector.Collector) (string, error) { + metrics := make(chan prometheus.Metric, 16) + if err := c.Collect(metrics, time.Second); err != nil { + return "", err + } + + close(metrics) + + registry := prometheus.NewPedanticRegistry() + if err := registry.Register(collectedMetrics(metrics)); err != nil { + return "", err + } + + families, err := registry.Gather() + if err != nil { + return "", err + } + + var out strings.Builder + + for _, family := range families { + if _, err := expfmt.MetricFamilyToText(&out, family); err != nil { + return "", err + } + } + + return out.String(), nil +} + +// collectedMetrics is a prometheus.Collector of already collected metrics. +type collectedMetrics <-chan prometheus.Metric + +func (m collectedMetrics) Describe(chan<- *prometheus.Desc) {} + +func (m collectedMetrics) Collect(ch chan<- prometheus.Metric) { + for metric := range m { + ch <- metric + } +} diff --git a/pkg/collector/state.go b/pkg/collector/state.go index 1541c127..3ab55f72 100644 --- a/pkg/collector/state.go +++ b/pkg/collector/state.go @@ -26,6 +26,7 @@ import ( "sync" "github.com/prometheus-community/windows_exporter/internal/mi" + "github.com/prometheus-community/windows_exporter/internal/scrapecache" ) var ( @@ -238,6 +239,9 @@ func (s *collectorState) build(ctx context.Context, logger *slog.Logger, miSessi } }() + // Cached metrics refer to the descriptors of an earlier Build. + scrapecache.Reset(collector) + return collector.Build(logger, miSession) } @@ -272,5 +276,8 @@ func (s *collectorState) close(ctx context.Context, collector Collector) (err er } }() + // Release the cached metrics, also if Close panics. + defer scrapecache.Reset(collector) + return collector.Close() }