pdh: per-counter metric types, no zero for invalid samples, worker panics (#2543)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Jan-Otto Kröpke <mail@jkroepke.de>

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)), "_"), "_")
 }

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"])
+	}
+}

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
 

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)
+}

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)
+		})
+	}
+}