diff --git a/internal/collector/tcp/tcp.go b/internal/collector/tcp/tcp.go index 39e2c85b..e00ae357 100644 --- a/internal/collector/tcp/tcp.go +++ b/internal/collector/tcp/tcp.go @@ -60,6 +60,9 @@ var ConfigDefaults = Config{ type Collector struct { config Config + // perfDataQuery samples the IPv4 and IPv6 objects together, so the + // collector waits only once for the other performance counter queries. + perfDataQuery *pdh.Query perfDataCollector4 *pdh.Collector[perfDataCounterValues] perfDataCollector6 *pdh.Collector[perfDataCounterValues] perfDataObject4 []perfDataCounterValues @@ -123,6 +126,7 @@ func (c *Collector) Close() error { if slices.Contains(c.config.CollectorsEnabled, subCollectorMetrics) { c.perfDataCollector4.Close() c.perfDataCollector6.Close() + c.perfDataQuery.Close() } return nil @@ -207,12 +211,17 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { if slices.Contains(c.config.CollectorsEnabled, subCollectorMetrics) { var err error - c.perfDataCollector4, err = pdh.NewCollector[perfDataCounterValues](logger.With(slog.String("collector", Name)), pdh.CounterTypeRaw, "TCPv4", nil) + c.perfDataQuery, err = pdh.NewQuery() + if err != nil { + return fmt.Errorf("failed to open TCP performance counter query: %w", err) + } + + c.perfDataCollector4, err = pdh.NewQueryCollector[perfDataCounterValues](logger.With(slog.String("collector", Name)), c.perfDataQuery, pdh.CounterTypeRaw, "TCPv4", nil) if err != nil { errs = append(errs, fmt.Errorf("failed to create TCPv4 collector: %w", err)) } - c.perfDataCollector6, err = pdh.NewCollector[perfDataCounterValues](logger.With(slog.String("collector", Name)), pdh.CounterTypeRaw, "TCPv6", nil) + c.perfDataCollector6, err = pdh.NewQueryCollector[perfDataCounterValues](logger.With(slog.String("collector", Name)), c.perfDataQuery, pdh.CounterTypeRaw, "TCPv6", nil) if err != nil { errs = append(errs, fmt.Errorf("failed to create TCPv6 collector: %w", err)) } @@ -244,6 +253,10 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error func (c *Collector) collect(ch chan<- prometheus.Metric) error { errs := make([]error, 0) + if err := c.perfDataQuery.Collect(); err != nil { + return fmt.Errorf("failed to collect TCP metrics: %w", err) + } + if err := c.perfDataCollector4.Collect(&c.perfDataObject4); err != nil { errs = append(errs, fmt.Errorf("failed to collect TCPv4 metrics. %w", err)) } else if len(c.perfDataObject4) == 0 { diff --git a/internal/collector/udp/udp.go b/internal/collector/udp/udp.go index 613b1f51..46edbd40 100644 --- a/internal/collector/udp/udp.go +++ b/internal/collector/udp/udp.go @@ -41,6 +41,9 @@ var ConfigDefaults = Config{} type Collector struct { config Config + // perfDataQuery samples the IPv4 and IPv6 objects together, so the + // collector waits only once for the other performance counter queries. + perfDataQuery *pdh.Query perfDataCollector4 *pdh.Collector[perfDataCounterValues] perfDataCollector6 *pdh.Collector[perfDataCounterValues] perfDataObject4 []perfDataCounterValues @@ -79,6 +82,7 @@ func (c *Collector) GetName() string { func (c *Collector) Close() error { c.perfDataCollector4.Close() c.perfDataCollector6.Close() + c.perfDataQuery.Close() return nil } @@ -113,12 +117,17 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { var err error - c.perfDataCollector4, err = pdh.NewCollector[perfDataCounterValues](logger.With(slog.String("collector", Name)), pdh.CounterTypeRaw, "UDPv4", nil) + c.perfDataQuery, err = pdh.NewQuery() + if err != nil { + return fmt.Errorf("failed to open UDP performance counter query: %w", err) + } + + c.perfDataCollector4, err = pdh.NewQueryCollector[perfDataCounterValues](logger.With(slog.String("collector", Name)), c.perfDataQuery, pdh.CounterTypeRaw, "UDPv4", nil) if err != nil { errs = append(errs, fmt.Errorf("failed to create UDPv4 collector: %w", err)) } - c.perfDataCollector6, err = pdh.NewCollector[perfDataCounterValues](logger.With(slog.String("collector", Name)), pdh.CounterTypeRaw, "UDPv6", nil) + c.perfDataCollector6, err = pdh.NewQueryCollector[perfDataCounterValues](logger.With(slog.String("collector", Name)), c.perfDataQuery, pdh.CounterTypeRaw, "UDPv6", nil) if err != nil { errs = append(errs, fmt.Errorf("failed to create UDPv6 collector: %w", err)) } @@ -133,6 +142,10 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error } func (c *Collector) collect(ch chan<- prometheus.Metric) error { + if err := c.perfDataQuery.Collect(); err != nil { + return fmt.Errorf("failed to collect UDP metrics: %w", err) + } + errs := make([]error, 0) if err := c.perfDataCollector4.Collect(&c.perfDataObject4); err != nil { diff --git a/internal/pdh/collector.go b/internal/pdh/collector.go index e5cde5c7..139a5437 100644 --- a/internal/pdh/collector.go +++ b/internal/pdh/collector.go @@ -44,6 +44,9 @@ var ( type CounterValues = map[string]map[string]CounterValue type Collector[T any] struct { + // query is the shared query of a collector from [NewQueryCollector], or + // nil if the collector owns handle. + query *Query object string resultType CounterType counters []Counter @@ -113,6 +116,22 @@ type Counter struct { // could not be added or the initial collection failed. The caller owns that // Collector and must Close it. If the Collector is nil, no resources are held. func NewCollector[T any](logger *slog.Logger, resultType CounterType, object string, instances []string) (*Collector[T], error) { + return newStructCollector[T](logger, nil, resultType, object, instances) +} + +// NewQueryCollector creates a collector like [NewCollector] whose counters are +// part of query. [Query.Collect] samples them together with the other +// collectors of the query; [Collector.Collect] reads the latest sample. Close +// the collector before the query. +func NewQueryCollector[T any](logger *slog.Logger, query *Query, resultType CounterType, object string, instances []string) (*Collector[T], error) { + if query == nil { + return nil, ErrPerformanceCounterNotInitialized + } + + return newStructCollector[T](logger, query, resultType, object, instances) +} + +func newStructCollector[T any](logger *slog.Logger, query *Query, resultType CounterType, object string, instances []string) (*Collector[T], error) { valueType := reflect.TypeFor[T]() if valueType.Kind() != reflect.Struct { return nil, fmt.Errorf("expected a struct, got %s", valueType) @@ -167,7 +186,7 @@ func NewCollector[T any](logger *slog.Logger, resultType CounterType, object str }, } - return newCollector(logger, resultType, object, instances, fields, rows, false, errs) + return newCollector(logger, query, resultType, object, instances, fields, rows, false, errs) } // NewDynamicCollector creates a collector for counters that are only known at runtime. @@ -192,10 +211,10 @@ func NewDynamicCollector(logger *slog.Logger, resultType CounterType, object str }, } - return newCollector(logger, resultType, object, instances, fields, rows, true, nil) + return newCollector(logger, nil, resultType, object, instances, fields, rows, true, nil) } -func newCollector[T any](logger *slog.Logger, resultType CounterType, object string, instances []string, +func newCollector[T any](logger *slog.Logger, query *Query, resultType CounterType, object string, instances []string, fields []counterField, rows rowAccessor[T], partialRows bool, errs []error, ) (*Collector[T], error) { if resultType != CounterTypeRaw && resultType != CounterTypeFormatted { @@ -204,7 +223,17 @@ func newCollector[T any](logger *slog.Logger, resultType CounterType, object str var handle pdhQueryHandle - if ret := OpenQuery(0, 0, &handle); ret != ErrorSuccess { + if query != nil { + // Adding counters while the query samples is not safe. + query.mu.Lock() + defer query.mu.Unlock() + + if query.handle == 0 { + return nil, ErrPerformanceCounterNotInitialized + } + + handle = query.handle + } else if ret := OpenQuery(0, 0, &handle); ret != ErrorSuccess { return nil, NewPdhError(ret) } @@ -213,6 +242,7 @@ func newCollector[T any](logger *slog.Logger, resultType CounterType, object str } collector := &Collector[T]{ + query: query, object: object, resultType: resultType, counters: make([]Counter, 0, len(fields)), @@ -326,7 +356,11 @@ func newCollector[T any](logger *slog.Logger, resultType CounterType, object str } if len(collector.counters) == 0 { - collector.Close() + // A shared query stays open for its other collectors, and its lock + // is held here, so only close a query of our own. + if query == nil { + collector.Close() + } errs = append(errs, errors.New("no counters configured")) @@ -340,6 +374,14 @@ func newCollector[T any](logger *slog.Logger, resultType CounterType, object str collector.state = &collectState{} // Collect initial data because some counters need to be read twice to get the correct value. + if query != nil { + if ret := CollectQueryData(handle); ret != ErrorSuccess && ret != NoData { + return collector, fmt.Errorf("failed to collect initial data: %w", NewPdhError(ret)) + } + + return collector, nil + } + var collectValues []T if err := collector.Collect(&collectValues); err != nil && !errors.Is(err, ErrNoData) { return collector, fmt.Errorf("failed to collect initial data: %w", err) @@ -408,6 +450,12 @@ func (c *Collector[T]) Collect(dst *[]T) error { return errors.New("dst must not be nil") } + if c.query != nil { + // Reading the counter arrays while the query samples is not safe. + c.query.mu.Lock() + defer c.query.mu.Unlock() + } + c.mu.Lock() defer c.mu.Unlock() @@ -445,8 +493,11 @@ func (c *Collector[T]) collect(dst *[]T, state *collectState) (err error) { } }() - if ret := CollectQueryData(c.handle); ret != ErrorSuccess { - return fmt.Errorf("failed to collect query data: %w", NewPdhError(ret)) + // A collector in a shared query reads the sample taken by Query.Collect. + if c.query == nil { + if ret := CollectQueryData(c.handle); ret != ErrorSuccess { + return fmt.Errorf("failed to collect query data: %w", NewPdhError(ret)) + } } *dst = (*dst)[:0:0] @@ -787,11 +838,26 @@ func (c *Collector[T]) Close() { return } + if c.query != nil { + c.query.mu.Lock() + defer c.query.mu.Unlock() + } + c.mu.Lock() defer c.mu.Unlock() - if c.handle != 0 { + switch { + case c.handle == 0: + case c.query == nil: CloseQuery(c.handle) + case c.query.handle != 0: + // The query belongs to its other collectors; only remove the + // counters of this one. + for _, counter := range c.counters { + for _, instance := range counter.Instances { + RemoveCounter(instance) + } + } } c.handle = 0 diff --git a/internal/pdh/pdh.go b/internal/pdh/pdh.go index 08f55295..c77615e5 100644 --- a/internal/pdh/pdh.go +++ b/internal/pdh/pdh.go @@ -290,6 +290,7 @@ var ( pdhGetRawCounterValue = libPdhDll.NewProc("PdhGetRawCounterValue") pdhGetRawCounterArrayW = libPdhDll.NewProc("PdhGetRawCounterArrayW") pdhPdhGetCounterTimeBase = libPdhDll.NewProc("PdhGetCounterTimeBase") + pdhRemoveCounter = libPdhDll.NewProc("PdhRemoveCounter") ) // AddCounter adds the specified counter to the query. This is the internationalized version. Preferably, use the @@ -358,6 +359,13 @@ func AddEnglishCounter(hQuery pdhQueryHandle, szFullCounterPath string, dwUserDa return uint32(ret) } +// RemoveCounter removes a counter from its query. +func RemoveCounter(hCounter pdhCounterHandle) uint32 { + ret, _, _ := pdhRemoveCounter.Call(uintptr(hCounter)) + + return uint32(ret) +} + // CloseQuery closes all counters contained in the specified query, closes all handles related to the query, // and frees all memory associated with the query. func CloseQuery(hQuery pdhQueryHandle) uint32 { diff --git a/internal/pdh/query.go b/internal/pdh/query.go new file mode 100644 index 00000000..5209386e --- /dev/null +++ b/internal/pdh/query.go @@ -0,0 +1,85 @@ +// 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 pdh + +import ( + "fmt" + "sync" +) + +// Query is a PDH query that several collectors share, so one +// PdhCollectQueryData samples all of them. Windows serves counter samples of +// a process one at a time, so every query waits for the others during a +// scrape; collectors that are always read together, such as the IPv4 and IPv6 +// objects of a protocol, wait only once in one query. +// +// Create the collectors with [NewQueryCollector]. [Query.Collect] takes the +// sample; [Collector.Collect] of a collector in the query reads the latest +// sample without taking a new one. +type Query struct { + mu sync.Mutex + handle pdhQueryHandle +} + +// NewQuery opens a query for [NewQueryCollector]. The caller must Close it +// after the collectors in it. +func NewQuery() (*Query, error) { + var handle pdhQueryHandle + + if ret := OpenQuery(0, 0, &handle); ret != ErrorSuccess { + return nil, NewPdhError(ret) + } + + return &Query{handle: handle}, nil +} + +// Collect takes one sample of all counters in the query. +func (q *Query) Collect() error { + if q == nil { + return ErrPerformanceCounterNotInitialized + } + + q.mu.Lock() + defer q.mu.Unlock() + + if q.handle == 0 { + return ErrPerformanceCounterNotInitialized + } + + if ret := CollectQueryData(q.handle); ret != ErrorSuccess { + return fmt.Errorf("failed to collect query data: %w", NewPdhError(ret)) + } + + return nil +} + +// Close closes the query and all counters in it. +func (q *Query) Close() { + if q == nil { + return + } + + q.mu.Lock() + defer q.mu.Unlock() + + if q.handle != 0 { + CloseQuery(q.handle) + } + + q.handle = 0 +} diff --git a/internal/pdh/query_test.go b/internal/pdh/query_test.go new file mode 100644 index 00000000..9dd93c31 --- /dev/null +++ b/internal/pdh/query_test.go @@ -0,0 +1,155 @@ +// 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 pdh_test + +import ( + "log/slog" + "sync" + "testing" + + "github.com/prometheus-community/windows_exporter/internal/pdh" + "github.com/stretchr/testify/require" +) + +type systemValues struct { + Processes float64 `perfdata:"Processes"` + Threads float64 `perfdata:"Threads"` +} + +type memoryValues struct { + AvailableBytes float64 `perfdata:"Available Bytes"` +} + +func TestQuery(t *testing.T) { + t.Parallel() + + query, err := pdh.NewQuery() + require.NoError(t, err) + + system, err := pdh.NewQueryCollector[systemValues](slog.New(slog.DiscardHandler), query, pdh.CounterTypeRaw, "System", nil) + require.NoError(t, err) + + memory, err := pdh.NewQueryCollector[memoryValues](slog.New(slog.DiscardHandler), query, pdh.CounterTypeRaw, "Memory", nil) + require.NoError(t, err) + + var ( + systemData []systemValues + memoryData []memoryValues + ) + + // One sample serves both collectors. + require.NoError(t, query.Collect()) + require.NoError(t, system.Collect(&systemData)) + require.NoError(t, memory.Collect(&memoryData)) + require.Len(t, systemData, 1) + require.Len(t, memoryData, 1) + require.NotZero(t, systemData[0].Processes) + require.NotZero(t, memoryData[0].AvailableBytes) + + // Closing one collector removes only its counters. + system.Close() + require.ErrorIs(t, system.Collect(&systemData), pdh.ErrPerformanceCounterNotInitialized) + require.NoError(t, query.Collect()) + require.NoError(t, memory.Collect(&memoryData)) + require.NotZero(t, memoryData[0].AvailableBytes) + + memory.Close() + query.Close() + query.Close() + + require.ErrorIs(t, query.Collect(), pdh.ErrPerformanceCounterNotInitialized) + + _, err = pdh.NewQueryCollector[memoryValues](slog.New(slog.DiscardHandler), query, pdh.CounterTypeRaw, "Memory", nil) + require.ErrorIs(t, err, pdh.ErrPerformanceCounterNotInitialized) + + _, err = pdh.NewQueryCollector[memoryValues](slog.New(slog.DiscardHandler), nil, pdh.CounterTypeRaw, "Memory", nil) + require.ErrorIs(t, err, pdh.ErrPerformanceCounterNotInitialized) +} + +// A collector whose object is missing reports the error, and the other +// collector of the query keeps working. +func TestQueryMissingObject(t *testing.T) { + t.Parallel() + + query, err := pdh.NewQuery() + require.NoError(t, err) + + defer query.Close() + + missing, err := pdh.NewQueryCollector[memoryValues](slog.New(slog.DiscardHandler), query, pdh.CounterTypeRaw, "Does Not Exist", nil) + require.Error(t, err) + + defer missing.Close() + + memory, err := pdh.NewQueryCollector[memoryValues](slog.New(slog.DiscardHandler), query, pdh.CounterTypeRaw, "Memory", nil) + require.NoError(t, err) + + defer memory.Close() + + var data []memoryValues + + require.NoError(t, query.Collect()) + require.NoError(t, memory.Collect(&data)) + require.NotEmpty(t, data) +} + +func TestQueryConcurrentCollect(t *testing.T) { + t.Parallel() + + query, err := pdh.NewQuery() + require.NoError(t, err) + + system, err := pdh.NewQueryCollector[systemValues](slog.New(slog.DiscardHandler), query, pdh.CounterTypeRaw, "System", nil) + require.NoError(t, err) + + memory, err := pdh.NewQueryCollector[memoryValues](slog.New(slog.DiscardHandler), query, pdh.CounterTypeRaw, "Memory", nil) + require.NoError(t, err) + + var wg sync.WaitGroup + + for range 8 { + wg.Go(func() { + var ( + systemData []systemValues + memoryData []memoryValues + ) + + for range 20 { + if err := query.Collect(); err != nil { + t.Error(err) + + return + } + + if err := system.Collect(&systemData); err != nil { + t.Error(err) + } + + if err := memory.Collect(&memoryData); err != nil { + t.Error(err) + } + } + }) + } + + wg.Wait() + + system.Close() + memory.Close() + query.Close() +}