From 3485634c4942c1571f61373e97cdef64ddf5132d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan-Otto=20Kr=C3=B6pke?= Date: Thu, 8 Oct 2026 14:48:34 +0200 Subject: [PATCH] pdh: per-counter metric types, no zero for invalid samples, worker panics (#2543) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Claude Opus 5.5 Signed-off-by: Jan-Otto Kröpke --- .../performancecounter/performancecounter.go | 46 +- .../performancecounter_test_test.go | 50 +- internal/pdh/collector.go | 596 +++++++++++------- internal/pdh/collector_internal_test.go | 187 ++++++ internal/pdh/collector_test.go | 58 ++ 5 files changed, 681 insertions(+), 256 deletions(-) create mode 100644 internal/pdh/collector_internal_test.go diff --git a/internal/collector/performancecounter/performancecounter.go b/internal/collector/performancecounter/performancecounter.go index fddd81b3..08fe041d 100644 --- a/internal/collector/performancecounter/performancecounter.go +++ b/internal/collector/performancecounter/performancecounter.go @@ -22,6 +22,7 @@ import ( "fmt" "log/slog" "maps" + "math" "regexp" "slices" "strings" @@ -272,10 +273,12 @@ func (c *Collector) collectObject(ch chan<- prometheus.Metric, perfDataObject Ob return fmt.Errorf("failed to collect data: %w", err) } + metricTypes := counterMetricTypes(perfDataObject) + var errs []error for _, row := range rows { - for _, counter := range perfDataObject.Counters { + for j, counter := range perfDataObject.Counters { valueIndex, ok := perfDataObject.valueIndex[counter.Name] if !ok { errs = append(errs, fmt.Errorf("%s not found in collected data", counter.Name)) @@ -284,7 +287,12 @@ func (c *Collector) collectObject(ch chan<- prometheus.Metric, perfDataObject Ob } collectedCounterValue := row.Values[valueIndex] - metricType := row.MetricType + + // The counter has no valid value for this instance in this sample. + // Omit it instead of reporting zero, which looks like a counter reset. + if math.IsNaN(collectedCounterValue) { + continue + } labels := make(prometheus.Labels, len(counter.Labels)+1) @@ -294,13 +302,6 @@ func (c *Collector) collectObject(ch chan<- prometheus.Metric, perfDataObject Ob maps.Copy(labels, counter.Labels) - switch counter.Type { - case "counter": - metricType = prometheus.CounterValue - case "gauge": - metricType = prometheus.GaugeValue - } - ch <- prometheus.MustNewConstMetric( prometheus.NewDesc( counter.Metric, @@ -308,7 +309,7 @@ func (c *Collector) collectObject(ch chan<- prometheus.Metric, perfDataObject Ob nil, labels, ), - metricType, + metricTypes[j], collectedCounterValue, ) } @@ -317,6 +318,31 @@ func (c *Collector) collectObject(ch chan<- prometheus.Metric, perfDataObject Ob return errors.Join(errs...) } +// counterMetricTypes returns the metric type of every counter of the object. +// An explicit type in the configuration takes precedence over the type that +// PDH reports for the counter. +func counterMetricTypes(perfDataObject Object) []prometheus.ValueType { + metricTypes := make([]prometheus.ValueType, len(perfDataObject.Counters)) + + for i, counter := range perfDataObject.Counters { + switch counter.Type { + case "counter": + metricTypes[i] = prometheus.CounterValue + case "gauge": + metricTypes[i] = prometheus.GaugeValue + default: + metricType, ok := perfDataObject.collector.MetricType(counter.Name) + if !ok { + metricType = prometheus.GaugeValue + } + + metricTypes[i] = metricType + } + } + + return metricTypes +} + func sanitizeMetricName(name string) string { return strings.Trim(reNonAlphaNum.ReplaceAllString(strings.ToLower(stringReplacer.Replace(name)), "_"), "_") } diff --git a/internal/collector/performancecounter/performancecounter_test_test.go b/internal/collector/performancecounter/performancecounter_test_test.go index 4aa31f59..9d20fe7d 100644 --- a/internal/collector/performancecounter/performancecounter_test_test.go +++ b/internal/collector/performancecounter/performancecounter_test_test.go @@ -24,11 +24,13 @@ import ( "net/http/httptest" "regexp" "testing" + "time" "github.com/prometheus-community/windows_exporter/internal/collector/performancecounter" "github.com/prometheus-community/windows_exporter/internal/pdh" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promhttp" + dto "github.com/prometheus/client_model/go" "github.com/stretchr/testify/require" ) @@ -130,8 +132,8 @@ windows_performancecounter_collector_duration_seconds\{collector="processor_info windows_performancecounter_collector_success\{collector="processor_information_formatted"} 1 # HELP windows_performancecounter_processor_information_processor_time windows_exporter: custom Performance Counter metric # TYPE windows_performancecounter_processor_information_processor_time gauge -windows_performancecounter_processor_information_processor_time\{core="0,0",state="active"} [0-9]+ -windows_performancecounter_processor_information_processor_time\{core="0,0",state="idle"} [0-9]+ +windows_performancecounter_processor_information_processor_time\{core="0,0",state="active"} [0-9.e+-]+ +windows_performancecounter_processor_information_processor_time\{core="0,0",state="idle"} [0-9.e+-]+ .*`), }, { @@ -225,6 +227,13 @@ windows_performancecounter_memory_available_bytes_lowercase [0-9.e+-]+`), require.NoError(t, err) + if tc.counterType == pdh.CounterTypeFormatted { + // Formatted rate counters are computed from the last two samples. Right + // after Build, the interval can be too short for a valid value, and + // invalid values are omitted instead of being reported as zero. + time.Sleep(time.Second) + } + registry := prometheus.NewRegistry() registry.MustRegister(collectorAdapter{*perfDataCollector}) @@ -354,3 +363,40 @@ func TestCollectorClose(t *testing.T) { // A closed collector must no longer hold usable native counter queries. require.ErrorIs(t, c.Collect(metrics, 0), pdh.ErrPerformanceCounterNotInitialized) } + +// TestCollectorMetricTypePerCounter checks that each counter keeps its own +// metric type on every scrape. The type used to be taken from whichever +// counter created the instance row first, in random map order. +func TestCollectorMetricTypePerCounter(t *testing.T) { + t.Parallel() + + c := performancecounter.New(&performancecounter.Config{ + Objects: []performancecounter.Object{{ + Name: "process", + Object: "Process", + Instances: []string{"*"}, + Counters: []performancecounter.Counter{ + {Name: "Thread Count"}, + {Name: "IO Read Bytes/sec"}, + }, + }}, + }) + require.NoError(t, c.Build(slog.New(slog.DiscardHandler), nil)) + t.Cleanup(func() { require.NoError(t, c.Close()) }) + + registry := prometheus.NewRegistry() + registry.MustRegister(collectorAdapter{*c}) + + for range 20 { + families, err := registry.Gather() + require.NoError(t, err) + + types := make(map[string]dto.MetricType, len(families)) + for _, family := range families { + types[family.GetName()] = family.GetType() + } + + require.Equal(t, dto.MetricType_GAUGE, types["windows_performancecounter_process_thread_count"]) + require.Equal(t, dto.MetricType_COUNTER, types["windows_performancecounter_process_io_read_bytes_sec"]) + } +} diff --git a/internal/pdh/collector.go b/internal/pdh/collector.go index 9f90d230..98ba4fe6 100644 --- a/internal/pdh/collector.go +++ b/internal/pdh/collector.go @@ -21,7 +21,9 @@ import ( "errors" "fmt" "log/slog" + "math" "reflect" + "runtime/debug" "slices" "strconv" "strings" @@ -43,7 +45,8 @@ type CounterValues = map[string]map[string]CounterValue type Collector[T any] struct { object string - counters map[string]Counter + resultType CounterType + counters []Counter handle pdhQueryHandle totalCounterRequested bool mu sync.RWMutex @@ -51,22 +54,27 @@ type Collector[T any] struct { rows rowAccessor[T] + // partialRows keeps instances for which some counters have no valid value + // in a sample and sets those values to NaN. Otherwise, such instances are + // left out of the sample, so a missing value is never reported as zero. + partialRows bool + collectCh chan *[]T errorCh chan error } // Row holds the values of one instance collected by a collector from [NewDynamicCollector]. type Row struct { - Name string - MetricType prometheus.ValueType + Name string // Values holds one value per counter, in the order the counters were passed to [NewDynamicCollector]. + // A value is NaN if the counter has no valid value for the instance in this sample. Values []float64 } // rowAccessor creates rows of type T and sets their counter values. // field is the index the counterField of the counter was created with. type rowAccessor[T any] struct { - newRow func(instance string, metricType prometheus.ValueType) T + newRow func(instance string) T setValue func(row *T, field int, value float64) } @@ -79,8 +87,10 @@ type counterField struct { } type Counter struct { - Name string - Desc string + Name string + Desc string + // MetricType is the Prometheus metric type of the counter. It is zero + // until the counter info of one of its instances has been read. MetricType prometheus.ValueType Instances map[string]pdhCounterHandle Type uint32 @@ -92,24 +102,25 @@ type Counter struct { // NewCollector creates a collector for the counters declared by the perfdata tags of the struct T. // -// The optional fields Name (string) and MetricType (prometheus.ValueType) of T -// receive the instance name and the metric type of the counter. +// The optional field Name (string) of T receives the instance name. An +// instance is only collected if every counter has a valid value for it in that +// sample. Use [Collector.MetricType] for the metric type of a counter. +// +// If an error is returned together with a non-nil Collector, some counters +// 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) { valueType := reflect.TypeFor[T]() if valueType.Kind() != reflect.Struct { return nil, fmt.Errorf("expected a struct, got %s", valueType) } - nameIndex, metricTypeIndex := -1, -1 + nameIndex := -1 if f, ok := valueType.FieldByName("Name"); ok && f.Type.Kind() == reflect.String { nameIndex = f.Index[0] } - if f, ok := valueType.FieldByName("MetricType"); ok && f.Type == reflect.TypeFor[prometheus.ValueType]() { - metricTypeIndex = f.Index[0] - } - var errs []error fields := make([]counterField, 0, valueType.NumField()) @@ -134,17 +145,11 @@ func NewCollector[T any](logger *slog.Logger, resultType CounterType, object str } rows := rowAccessor[T]{ - newRow: func(instance string, metricType prometheus.ValueType) T { + newRow: func(instance string) T { var row T - rv := reflect.ValueOf(&row).Elem() - if nameIndex != -1 { - rv.Field(nameIndex).SetString(instance) - } - - if metricTypeIndex != -1 { - rv.Field(metricTypeIndex).SetInt(int64(metricType)) + reflect.ValueOf(&row).Elem().Field(nameIndex).SetString(instance) } return row @@ -154,11 +159,13 @@ func NewCollector[T any](logger *slog.Logger, resultType CounterType, object str }, } - return newCollector(logger, resultType, object, instances, fields, rows, errs) + return newCollector(logger, resultType, object, instances, fields, rows, false, errs) } // NewDynamicCollector creates a collector for counters that are only known at runtime. // Row.Values of the collected rows holds one value per entry of counters, in the same order. +// Unlike [NewCollector], an instance is kept if only some of its counters have +// a valid value; the others are NaN. Errors are returned as described for [NewCollector]. func NewDynamicCollector(logger *slog.Logger, resultType CounterType, object string, instances []string, counters []string) (*Collector[Row], error) { fields := make([]counterField, len(counters)) for i, counter := range counters { @@ -166,20 +173,24 @@ func NewDynamicCollector(logger *slog.Logger, resultType CounterType, object str } rows := rowAccessor[Row]{ - newRow: func(instance string, metricType prometheus.ValueType) Row { - return Row{Name: instance, MetricType: metricType, Values: make([]float64, len(counters))} + newRow: func(instance string) Row { + return Row{Name: instance, Values: make([]float64, len(counters))} }, setValue: func(row *Row, field int, value float64) { row.Values[field] = value }, } - return newCollector(logger, resultType, object, instances, fields, rows, nil) + return newCollector(logger, resultType, object, instances, fields, rows, true, nil) } func newCollector[T any](logger *slog.Logger, resultType CounterType, object string, instances []string, - fields []counterField, rows rowAccessor[T], errs []error, + fields []counterField, rows rowAccessor[T], partialRows bool, errs []error, ) (*Collector[T], error) { + if resultType != CounterTypeRaw && resultType != CounterTypeFormatted { + return nil, fmt.Errorf("invalid result type: %v", resultType) + } + var handle pdhQueryHandle if ret := OpenQuery(0, 0, &handle); ret != ErrorSuccess { @@ -190,33 +201,37 @@ func newCollector[T any](logger *slog.Logger, resultType CounterType, object str instances = []string{InstanceEmpty} } - if resultType != CounterTypeRaw && resultType != CounterTypeFormatted { - return nil, fmt.Errorf("invalid result type: %v", resultType) - } - collector := &Collector[T]{ object: object, - counters: make(map[string]Counter, len(fields)), + resultType: resultType, + counters: make([]Counter, 0, len(fields)), handle: handle, totalCounterRequested: slices.Contains(instances, InstanceTotal), mu: sync.RWMutex{}, logger: logger, rows: rows, + partialRows: partialRows, } + counterIndex := make(map[string]int, len(fields)) + for _, f := range fields { counterName, secondValue := strings.CutSuffix(f.tag, ",secondvalue") - counter, ok := collector.counters[counterName] + i, ok := counterIndex[counterName] if !ok { - counter = Counter{ + i = len(collector.counters) + counterIndex[counterName] = i + collector.counters = append(collector.counters, Counter{ Name: counterName, Instances: make(map[string]pdhCounterHandle, len(instances)), FieldIndexSecondValue: -1, FieldIndexValue: -1, - } + }) } + counter := &collector.counters[i] + if secondValue { counter.FieldIndexSecondValue = f.index } else { @@ -224,8 +239,6 @@ func newCollector[T any](logger *slog.Logger, resultType CounterType, object str } if len(counter.Instances) != 0 { - collector.counters[counterName] = counter - continue } @@ -255,7 +268,7 @@ func newCollector[T any](logger *slog.Logger, resultType CounterType, object str counter.Instances[instance] = counterHandle - if counter.Type != 0 { + if counter.MetricType != 0 { continue } @@ -289,11 +302,7 @@ func newCollector[T any](logger *slog.Logger, resultType CounterType, object str } counter.Type = counterInfo.DwType - if val, ok := SupportedCounterTypes[counter.Type]; ok { - counter.MetricType = val - } else { - counter.MetricType = prometheus.GaugeValue - } + counter.MetricType = metricType(resultType, counter.Type) if counter.Type == PERF_ELAPSED_TIME { if ret := GetCounterTimeBase(counterHandle, &counter.Frequency); ret != ErrorSuccess && ret != NoData { @@ -303,26 +312,24 @@ func newCollector[T any](logger *slog.Logger, resultType CounterType, object str } } } + } - collector.counters[counterName] = counter + if len(collector.counters) == 0 { + collector.Close() + + errs = append(errs, errors.New("no counters configured")) + + return nil, errors.Join(errs...) } if err := errors.Join(errs...); err != nil { return collector, fmt.Errorf("failed to initialize collector: %w", err) } - if len(collector.counters) == 0 { - return nil, errors.New("no counters configured") - } - collector.collectCh = make(chan *[]T) collector.errorCh = make(chan error) - if resultType == CounterTypeRaw { - go collector.collectWorkerRaw() - } else { - go collector.collectWorkerFormatted() - } + go collector.collectWorker() // Collect initial data because some counters need to be read twice to get the correct value. var collectValues []T @@ -333,6 +340,18 @@ func newCollector[T any](logger *slog.Logger, resultType CounterType, object str return collector, nil } +// metricType returns the Prometheus metric type for a PDH counter type. +// Formatted values are already rates or ratios, so they are always gauges. +func metricType(resultType CounterType, counterType uint32) prometheus.ValueType { + if resultType == CounterTypeRaw { + if val, ok := SupportedCounterTypes[counterType]; ok { + return val + } + } + + return prometheus.GaugeValue +} + func (c *Collector[T]) Describe() map[string]string { if c == nil { return map[string]string{} @@ -350,6 +369,27 @@ func (c *Collector[T]) Describe() map[string]string { return desc } +// MetricType returns the Prometheus metric type of the counter named +// counterName, derived from its PDH counter type. Formatted values are always +// gauges. It returns false if the counter is unknown or none of its instances +// could be added. +func (c *Collector[T]) MetricType(counterName string) (prometheus.ValueType, bool) { + if c == nil { + return 0, false + } + + c.mu.RLock() + defer c.mu.RUnlock() + + for _, counter := range c.counters { + if counter.Name == counterName { + return counter.MetricType, counter.MetricType != 0 + } + } + + return 0, false +} + // Collect replaces the content of dst with one row per collected instance. func (c *Collector[T]) Collect(dst *[]T) error { if c == nil { @@ -372,228 +412,294 @@ func (c *Collector[T]) Collect(dst *[]T) error { return <-c.errorCh } -func (c *Collector[T]) collectWorkerRaw() { - var ( - err error - itemCount uint32 - items []RawCounterItem - bytesNeeded uint32 - ) - - // buf starts empty so that the first call only queries the required size with a nil buffer. - // PdhGetRawCounterArrayW writes 8 bytes into a non-nil buffer, even if lpdwBufferSize is smaller, - // which corrupts the neighboring heap memory of a tiny allocation. - var buf []byte +func (c *Collector[T]) collectWorker() { + var state collectState for dst := range c.collectCh { - err = (func() error { - if ret := CollectQueryData(c.handle); ret != ErrorSuccess { - return fmt.Errorf("failed to collect query data: %w", NewPdhError(ret)) - } - - *dst = (*dst)[:0:0] - - indexMap := map[string]int{} - nameCache := map[string]string{} - - for _, counter := range c.counters { - instances: - for _, instance := range counter.Instances { - // Get the info with the current buffer size - bytesNeeded = uint32(len(buf)) - - for { - ret := GetRawCounterArray(instance, &bytesNeeded, &itemCount, unsafe.SliceData(buf)) - - if ret == ErrorSuccess { - break - } - - if err := NewPdhError(ret); ret != MoreData { - if isKnownCounterDataError(err) { - continue instances - } - - return fmt.Errorf("GetRawCounterArray: %w", err) - } - - if bytesNeeded <= uint32(len(buf)) { - return fmt.Errorf("GetRawCounterArray reports buffer too small (%d), but buffer is large enough (%d): %w", uint32(len(buf)), bytesNeeded, NewPdhError(ret)) - } - - buf = make([]byte, bytesNeeded) - } - - items = unsafe.Slice((*RawCounterItem)(unsafe.Pointer(unsafe.SliceData(buf))), itemCount) - - for _, item := range items { - if item.RawValue.CStatus != CstatusValidData && item.RawValue.CStatus != CstatusNewData { - c.logger.Debug("skipping counter item with invalid data status", - slog.String("counter", counter.Name), - slog.String("instance", windows.UTF16PtrToString(item.SzName)), - slog.Uint64("status", uint64(item.RawValue.CStatus)), - ) - - continue - } - - instanceName := decodeInstanceName(nameCache, item.SzName) - - if strings.HasSuffix(instanceName, InstanceTotal) && !c.totalCounterRequested { - continue - } - - if instanceName == "" || instanceName == "*" { - instanceName = InstanceEmpty - } - - var ( - index int - ok bool - ) - - if index, ok = indexMap[instanceName]; !ok { - index = len(*dst) - indexMap[instanceName] = index - - var metricsType prometheus.ValueType - if metricsType, ok = SupportedCounterTypes[counter.Type]; !ok { - metricsType = prometheus.GaugeValue - } - - *dst = append(*dst, c.rows.newRow(instanceName, metricsType)) - } - - row := &(*dst)[index] - - // This is a workaround for the issue with the elapsed time counter type. - // Source: https://github.com/prometheus-community/windows_exporter/pull/335/files#diff-d5d2528f559ba2648c2866aec34b1eaa5c094dedb52bd0ff22aa5eb83226bd8dR76-R83 - // Ref: https://learn.microsoft.com/en-us/windows/win32/perfctrs/calculating-counter-values - switch counter.Type { - case PERF_ELAPSED_TIME: - c.rows.setValue(row, counter.FieldIndexValue, float64((item.RawValue.SecondValue-item.RawValue.FirstValue)/counter.Frequency)) - case PERF_100NSEC_TIMER, PERF_PRECISION_100NS_TIMER: - c.rows.setValue(row, counter.FieldIndexValue, float64(item.RawValue.FirstValue)*TicksToSecondScaleFactor) - default: - if counter.FieldIndexSecondValue != -1 { - c.rows.setValue(row, counter.FieldIndexSecondValue, float64(item.RawValue.SecondValue)) - } - - if counter.FieldIndexValue != -1 { - c.rows.setValue(row, counter.FieldIndexValue, float64(item.RawValue.FirstValue)) - } - } - } - } - } - - if len(*dst) == 0 { - return ErrNoData - } - - return nil - })() - - c.errorCh <- err + c.errorCh <- c.collect(dst, &state) } } -func (c *Collector[T]) collectWorkerFormatted() { - var ( - err error - itemCount uint32 - items []FmtCounterValueItemDouble - bytesNeeded uint32 - ) - +// collectState holds the buffers that the collect worker reuses between samples. +type collectState struct { // buf starts empty so that the first call only queries the required size with a nil buffer. - // See collectWorkerRaw. - var buf []byte + // PdhGetRawCounterArrayW writes 8 bytes into a non-nil buffer, even if lpdwBufferSize is smaller, + // which corrupts the neighboring heap memory of a tiny allocation. + buf []byte - for dst := range c.collectCh { - err = (func() error { - if ret := CollectQueryData(c.handle); ret != ErrorSuccess { - return fmt.Errorf("failed to collect query data: %w", NewPdhError(ret)) + // valid records for every row and counter whether the counter had a valid value. + valid []bool +} + +// collect replaces the content of dst with one sample. +func (c *Collector[T]) collect(dst *[]T, state *collectState) (err error) { + defer func() { + if r := recover(); r != nil { + err = c.panicError(r) + } + }() + + if ret := CollectQueryData(c.handle); ret != ErrorSuccess { + return fmt.Errorf("failed to collect query data: %w", NewPdhError(ret)) + } + + *dst = (*dst)[:0:0] + + rows := rowSet[T]{ + c: c, + dst: dst, + index: map[string]int{}, + nameCache: map[string]string{}, + valid: state.valid[:0], + } + + for counterIndex := range c.counters { + for _, instance := range c.counters[counterIndex].Instances { + itemCount, ok, err := c.getCounterArray(instance, &state.buf) + if err != nil { + return err } - *dst = (*dst)[:0:0] + if !ok { + continue + } - indexMap := map[string]int{} - nameCache := map[string]string{} + if c.resultType == CounterTypeRaw { + rows.addRawItems(counterIndex, state.buf, itemCount) + } else { + rows.addFormattedItems(counterIndex, state.buf, itemCount) + } + } + } - for _, counter := range c.counters { - instances: - for _, instance := range counter.Instances { - // Get the info with the current buffer size - bytesNeeded = uint32(len(buf)) + rows.finish() - for { - ret := GetFormattedCounterArrayDouble(instance, &bytesNeeded, &itemCount, unsafe.SliceData(buf)) + state.valid = rows.valid - if ret == ErrorSuccess { - break - } + if len(*dst) == 0 { + return ErrNoData + } - if err := NewPdhError(ret); ret != MoreData { - if isKnownCounterDataError(err) { - continue instances - } + return nil +} - return fmt.Errorf("GetFormattedCounterArrayDouble: %w", err) - } +// getCounterArray reads the counter array of instance into buf and returns the number of items. +// ok is false if the counter has no data in this sample. +func (c *Collector[T]) getCounterArray(instance pdhCounterHandle, buf *[]byte) (uint32, bool, error) { + getArray, name := GetRawCounterArray, "GetRawCounterArray" + if c.resultType == CounterTypeFormatted { + getArray, name = GetFormattedCounterArrayDouble, "GetFormattedCounterArrayDouble" + } - if bytesNeeded <= uint32(len(buf)) { - return fmt.Errorf("GetFormattedCounterArrayDouble reports buffer too small (%d), but buffer is large enough (%d): %w", uint32(len(buf)), bytesNeeded, NewPdhError(ret)) - } + var itemCount uint32 - buf = make([]byte, bytesNeeded) - } + // Get the info with the current buffer size + bytesNeeded := uint32(len(*buf)) - items = unsafe.Slice((*FmtCounterValueItemDouble)(unsafe.Pointer(unsafe.SliceData(buf))), itemCount) + for { + ret := getArray(instance, &bytesNeeded, &itemCount, unsafe.SliceData(*buf)) - for _, item := range items { - if item.FmtValue.CStatus != CstatusValidData && item.FmtValue.CStatus != CstatusNewData { - continue - } + if ret == ErrorSuccess { + return itemCount, true, nil + } - instanceName := decodeInstanceName(nameCache, item.SzName) + if err := NewPdhError(ret); ret != MoreData { + if isKnownCounterDataError(err) { + return 0, false, nil + } - if strings.HasSuffix(instanceName, InstanceTotal) && !c.totalCounterRequested { - continue - } + return 0, false, fmt.Errorf("%s: %w", name, err) + } - if instanceName == "" || instanceName == "*" { - instanceName = InstanceEmpty - } + if bytesNeeded <= uint32(len(*buf)) { + return 0, false, fmt.Errorf("%s reports buffer too small (%d), but buffer is large enough (%d): %w", name, uint32(len(*buf)), bytesNeeded, NewPdhError(ret)) + } - var ( - index int - ok bool - ) + *buf = make([]byte, bytesNeeded) + } +} - if index, ok = indexMap[instanceName]; !ok { - index = len(*dst) - indexMap[instanceName] = index +// setRawValue stores the value of a raw counter item in row. It returns false +// if the value cannot be computed. +func (c *Collector[T]) setRawValue(row *T, counter *Counter, value RawCounter) bool { + // This is a workaround for the issue with the elapsed time counter type. + // Source: https://github.com/prometheus-community/windows_exporter/pull/335/files#diff-d5d2528f559ba2648c2866aec34b1eaa5c094dedb52bd0ff22aa5eb83226bd8dR76-R83 + // Ref: https://learn.microsoft.com/en-us/windows/win32/perfctrs/calculating-counter-values + switch counter.Type { + case PERF_ELAPSED_TIME: + // A zero frequency would divide by zero. + if counter.Frequency <= 0 { + return false + } - *dst = append(*dst, c.rows.newRow(instanceName, prometheus.GaugeValue)) - } + if counter.FieldIndexValue != -1 { + c.rows.setValue(row, counter.FieldIndexValue, float64(value.SecondValue-value.FirstValue)/float64(counter.Frequency)) + } + case PERF_100NSEC_TIMER, PERF_PRECISION_100NS_TIMER: + if counter.FieldIndexValue != -1 { + c.rows.setValue(row, counter.FieldIndexValue, float64(value.FirstValue)*TicksToSecondScaleFactor) + } + default: + if counter.FieldIndexSecondValue != -1 { + c.rows.setValue(row, counter.FieldIndexSecondValue, float64(value.SecondValue)) + } - if counter.FieldIndexValue != -1 { - c.rows.setValue(&(*dst)[index], counter.FieldIndexValue, item.FmtValue.DoubleValue) - } - } + if counter.FieldIndexValue != -1 { + c.rows.setValue(row, counter.FieldIndexValue, float64(value.FirstValue)) + } + } + + return true +} + +// rowSet appends one row per instance to dst and tracks which counters have +// a valid value for each row. +type rowSet[T any] struct { + c *Collector[T] + dst *[]T + index map[string]int + nameCache map[string]string + // valid has len(c.counters) entries per row. + valid []bool +} + +// row returns the index of the row for the instance named by szName, +// appending the row if needed. ok is false if the item has no valid data or +// its instance is not collected. +func (r *rowSet[T]) row(counter *Counter, szName *uint16, status uint32) (int, bool) { + if status != CstatusValidData && status != CstatusNewData { + r.c.logger.Debug("skipping counter item with invalid data status", + slog.String("counter", counter.Name), + slog.String("instance", windows.UTF16PtrToString(szName)), + slog.Uint64("status", uint64(status)), + ) + + return 0, false + } + + instanceName := decodeInstanceName(r.nameCache, szName) + + if strings.HasSuffix(instanceName, InstanceTotal) && !r.c.totalCounterRequested { + return 0, false + } + + if instanceName == "" || instanceName == "*" { + instanceName = InstanceEmpty + } + + if index, ok := r.index[instanceName]; ok { + return index, true + } + + index := len(*r.dst) + r.index[instanceName] = index + + *r.dst = append(*r.dst, r.c.rows.newRow(instanceName)) + + n := len(r.c.counters) + r.valid = slices.Grow(r.valid, n)[:len(r.valid)+n] + clear(r.valid[len(r.valid)-n:]) + + return index, true +} + +func (r *rowSet[T]) setValid(row, counterIndex int) { + r.valid[row*len(r.c.counters)+counterIndex] = true +} + +// addRawItems stores the values of the raw counter items in buf. +func (r *rowSet[T]) addRawItems(counterIndex int, buf []byte, itemCount uint32) { + counter := &r.c.counters[counterIndex] + items := unsafe.Slice((*RawCounterItem)(unsafe.Pointer(unsafe.SliceData(buf))), itemCount) + + for _, item := range items { + row, ok := r.row(counter, item.SzName, item.RawValue.CStatus) + if ok && r.c.setRawValue(&(*r.dst)[row], counter, item.RawValue) { + r.setValid(row, counterIndex) + } + } +} + +// addFormattedItems stores the values of the formatted counter items in buf. +func (r *rowSet[T]) addFormattedItems(counterIndex int, buf []byte, itemCount uint32) { + counter := &r.c.counters[counterIndex] + items := unsafe.Slice((*FmtCounterValueItemDouble)(unsafe.Pointer(unsafe.SliceData(buf))), itemCount) + + for _, item := range items { + row, ok := r.row(counter, item.SzName, item.FmtValue.CStatus) + if !ok { + continue + } + + if counter.FieldIndexValue != -1 { + r.c.rows.setValue(&(*r.dst)[row], counter.FieldIndexValue, item.FmtValue.DoubleValue) + } + + r.setValid(row, counterIndex) + } +} + +// finish handles rows that lack a valid value for some counter. Counters +// without any instance, e.g. those skipped by perfdata_min_build, are not +// required. Without partial rows, such rows are removed. Otherwise, the +// missing values are set to NaN. +func (r *rowSet[T]) finish() { + rows := *r.dst + n := len(r.c.counters) + kept := 0 + + for row := range rows { + complete := true + + for counterIndex := range r.c.counters { + counter := &r.c.counters[counterIndex] + + if len(counter.Instances) == 0 || r.valid[row*n+counterIndex] { + continue + } + + complete = false + + if !r.c.partialRows { + break + } + + for _, field := range []int{counter.FieldIndexValue, counter.FieldIndexSecondValue} { + if field != -1 { + r.c.rows.setValue(&rows[row], field, math.NaN()) } } + } - if len(*dst) == 0 { - return ErrNoData - } + if !complete && !r.c.partialRows { + continue + } - return nil - })() - - c.errorCh <- err + rows[kept] = rows[row] + kept++ } + + if dropped := len(rows) - kept; dropped > 0 { + r.c.logger.Debug("omitting instances without a valid value for every counter", + slog.String("object", r.c.object), + slog.Int("instances", dropped), + ) + + clear(rows[kept:]) + *r.dst = rows[:kept] + } +} + +// panicError logs a panic recovered in the collect worker and returns it as an +// error. The worker goroutine has no other recovery, so a panic would end the process. +func (c *Collector[T]) panicError(r any) error { + c.logger.Error("recovered from panic while collecting performance counters", + slog.String("object", c.object), + slog.Any("panic", r), + slog.String("stack", string(debug.Stack())), + ) + + return fmt.Errorf("panic while collecting performance counters of %s: %v", c.object, r) } func (c *Collector[T]) Close() { @@ -604,7 +710,9 @@ func (c *Collector[T]) Close() { c.mu.Lock() defer c.mu.Unlock() - CloseQuery(c.handle) + if c.handle != 0 { + CloseQuery(c.handle) + } c.handle = 0 diff --git a/internal/pdh/collector_internal_test.go b/internal/pdh/collector_internal_test.go new file mode 100644 index 00000000..8198d013 --- /dev/null +++ b/internal/pdh/collector_internal_test.go @@ -0,0 +1,187 @@ +// 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 ( + "log/slog" + "math" + "reflect" + "testing" + + "github.com/prometheus/client_golang/prometheus" + "github.com/stretchr/testify/require" + "golang.org/x/sys/windows" +) + +type rowSetValues struct { + Name string + A float64 `perfdata:"A"` + B float64 `perfdata:"B"` + BSecond float64 `perfdata:"B,secondvalue"` + Optional float64 `perfdata:"Optional"` +} + +// newRowSetTestCollector returns a collector with the counters A and B, which +// have instances, and Optional, which was skipped at Build and has none. +func newRowSetTestCollector(partialRows bool) *Collector[rowSetValues] { + return &Collector[rowSetValues]{ + partialRows: partialRows, + logger: slog.New(slog.DiscardHandler), + rows: rowAccessor[rowSetValues]{ + newRow: func(instance string) rowSetValues { + return rowSetValues{Name: instance} + }, + setValue: func(row *rowSetValues, field int, value float64) { + reflect.ValueOf(row).Elem().Field(field).SetFloat(value) + }, + }, + counters: []Counter{ + {Name: "A", Instances: map[string]pdhCounterHandle{"*": 1}, FieldIndexValue: 1, FieldIndexSecondValue: -1}, + {Name: "B", Instances: map[string]pdhCounterHandle{"*": 2}, FieldIndexValue: 2, FieldIndexSecondValue: 3}, + {Name: "Optional", Instances: map[string]pdhCounterHandle{}, FieldIndexValue: 4, FieldIndexSecondValue: -1}, + }, + } +} + +// fillRows adds the instance "complete" with valid values for A and B, and the +// instance "partial" with a valid value for A only, as if its item for B had an +// invalid status. +func fillRows(t *testing.T, c *Collector[rowSetValues]) []rowSetValues { + t.Helper() + + var dst []rowSetValues + + rows := rowSet[rowSetValues]{ + c: c, + dst: &dst, + index: map[string]int{}, + nameCache: map[string]string{}, + } + + for _, name := range []string{"complete", "partial"} { + row, ok := rows.row(&c.counters[0], windows.StringToUTF16Ptr(name), CstatusValidData) + require.True(t, ok) + require.True(t, c.setRawValue(&dst[row], &c.counters[0], RawCounter{FirstValue: 1})) + rows.setValid(row, 0) + } + + _, ok := rows.row(&c.counters[1], windows.StringToUTF16Ptr("partial"), CstatusInvalidData) + require.False(t, ok) + + row, ok := rows.row(&c.counters[1], windows.StringToUTF16Ptr("complete"), CstatusNewData) + require.True(t, ok) + require.True(t, c.setRawValue(&dst[row], &c.counters[1], RawCounter{FirstValue: 2, SecondValue: 3})) + rows.setValid(row, 1) + + rows.finish() + + return dst +} + +func TestRowSetOmitsIncompleteInstances(t *testing.T) { + t.Parallel() + + rows := fillRows(t, newRowSetTestCollector(false)) + + // "partial" has no valid value for B. Without partial rows, it must be left + // out instead of reporting B as zero. Optional has no instances, so it is + // not required. + require.Equal(t, []rowSetValues{{Name: "complete", A: 1, B: 2, BSecond: 3}}, rows) +} + +func TestRowSetPartialRows(t *testing.T) { + t.Parallel() + + rows := fillRows(t, newRowSetTestCollector(true)) + require.Len(t, rows, 2) + + require.Equal(t, rowSetValues{Name: "complete", A: 1, B: 2, BSecond: 3}, rows[0]) + + require.Equal(t, "partial", rows[1].Name) + require.InDelta(t, 1, rows[1].A, 0) + require.True(t, math.IsNaN(rows[1].B), "B: %v", rows[1].B) + require.True(t, math.IsNaN(rows[1].BSecond), "BSecond: %v", rows[1].BSecond) + require.Zero(t, rows[1].Optional) +} + +func TestSetRawValueElapsedTime(t *testing.T) { + t.Parallel() + + c := newRowSetTestCollector(false) + + var values rowSetValues + + counter := &Counter{Type: PERF_ELAPSED_TIME, Frequency: 10_000_000, FieldIndexValue: 1, FieldIndexSecondValue: -1} + + // 2.5 s at 10 MHz. Integer division truncated this to 2. + require.True(t, c.setRawValue(&values, counter, RawCounter{FirstValue: 100_000_000, SecondValue: 125_000_000})) + require.InDelta(t, 2.5, values.A, 0) + + // A zero frequency panicked with an integer divide by zero. + counter.Frequency = 0 + require.False(t, c.setRawValue(&values, counter, RawCounter{FirstValue: 1, SecondValue: 2})) + + // A counter with only a second value field must not set field -1. + counter = &Counter{Type: PERF_ELAPSED_TIME, Frequency: 1, FieldIndexValue: -1, FieldIndexSecondValue: 3} + require.True(t, c.setRawValue(&values, counter, RawCounter{FirstValue: 1, SecondValue: 2})) +} + +func TestMetricTypeForCounterType(t *testing.T) { + t.Parallel() + + require.Equal(t, prometheus.GaugeValue, metricType(CounterTypeRaw, PERF_COUNTER_RAWCOUNT)) + require.Equal(t, prometheus.CounterValue, metricType(CounterTypeRaw, PERF_COUNTER_BULK_COUNT)) + require.Equal(t, prometheus.GaugeValue, metricType(CounterTypeRaw, 0xFFFFFFFF)) + require.Equal(t, prometheus.GaugeValue, metricType(CounterTypeFormatted, PERF_COUNTER_BULK_COUNT)) +} + +type processThreads struct { + Name string + ThreadCount float64 `perfdata:"Thread Count"` +} + +// TestCollectRecoversPanic checks that a panic in the worker goroutine is +// returned as an error instead of terminating the process. +func TestCollectRecoversPanic(t *testing.T) { + t.Parallel() + + c, err := NewCollector[processThreads](slog.New(slog.DiscardHandler), CounterTypeRaw, "Process", InstancesAll) + require.NoError(t, err) + + t.Cleanup(c.Close) + + var dst []processThreads + + require.NoError(t, c.Collect(&dst)) + + // Point the counter at a field that does not exist, so that reflect panics. + c.mu.Lock() + c.counters[0].FieldIndexValue = 42 + c.mu.Unlock() + + err = c.Collect(&dst) + require.ErrorContains(t, err, "panic while collecting performance counters of Process") + + // The worker survives and serves the next request. + c.mu.Lock() + c.counters[0].FieldIndexValue = 1 + c.mu.Unlock() + + require.NoError(t, c.Collect(&dst)) + require.NotEmpty(t, dst) +} diff --git a/internal/pdh/collector_test.go b/internal/pdh/collector_test.go index 05fb5cfd..347bb177 100644 --- a/internal/pdh/collector_test.go +++ b/internal/pdh/collector_test.go @@ -23,6 +23,7 @@ import ( "time" "github.com/prometheus-community/windows_exporter/internal/pdh" + "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/require" ) @@ -169,3 +170,60 @@ func TestDynamicCollector(t *testing.T) { require.NotZerof(t, row.Values[0], "instance: %s, counter: %s", row.Name, counters[0]) } } + +// TestNewCollectorInvalidInput checks that invalid input returns no collector, +// so there is nothing for the caller to close. +func TestNewCollectorInvalidInput(t *testing.T) { + t.Parallel() + + logger := slog.New(slog.DiscardHandler) + + collector, err := pdh.NewCollector[process](logger, "invalid", "Process", pdh.InstancesAll) + require.ErrorContains(t, err, "invalid result type") + require.Nil(t, collector) + + collector2, err := pdh.NewCollector[struct{ Name string }](logger, pdh.CounterTypeRaw, "Process", pdh.InstancesAll) + require.ErrorContains(t, err, "no counters configured") + require.Nil(t, collector2) +} + +type processIO struct { + Name string + ThreadCount float64 `perfdata:"Thread Count"` + IOReadBytesPS float64 `perfdata:"IO Read Bytes/sec"` +} + +// TestCollectorMetricType checks that every counter carries its own metric +// type, independent of the other counters of the object. +func TestCollectorMetricType(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + resultType pdh.CounterType + threadCount prometheus.ValueType + ioReadBytes prometheus.ValueType + }{ + {pdh.CounterTypeRaw, prometheus.GaugeValue, prometheus.CounterValue}, + {pdh.CounterTypeFormatted, prometheus.GaugeValue, prometheus.GaugeValue}, + } { + t.Run(string(tc.resultType), func(t *testing.T) { + t.Parallel() + + collector, err := pdh.NewCollector[processIO](slog.New(slog.DiscardHandler), tc.resultType, "Process", pdh.InstancesAll) + require.NoError(t, err) + + t.Cleanup(collector.Close) + + metricType, ok := collector.MetricType("Thread Count") + require.True(t, ok) + require.Equal(t, tc.threadCount, metricType) + + metricType, ok = collector.MetricType("IO Read Bytes/sec") + require.True(t, ok) + require.Equal(t, tc.ioReadBytes, metricType) + + _, ok = collector.MetricType("unknown") + require.False(t, ok) + }) + } +}