terminal_services: fix swapped cpu_time labels, pending broker gauge, shared PDH query and series caching (#2687)

Co-authored-by: Claude <noreply@anthropic.com>
Signed-off-by: Claude <noreply@anthropic.com>

internal/collector/terminal_services/terminal_services.go

@@ -27,6 +27,8 @@ import (
 
 	"github.com/alecthomas/kingpin/v2"
 	"github.com/prometheus-community/windows_exporter/internal/headers/wtsapi32"
+	"github.com/prometheus-community/windows_exporter/internal/labelcache"
+	"github.com/prometheus-community/windows_exporter/internal/metriccache"
 	"github.com/prometheus-community/windows_exporter/internal/mi"
 	"github.com/prometheus-community/windows_exporter/internal/pdh"
 	"github.com/prometheus-community/windows_exporter/internal/types"
@@ -78,6 +80,9 @@ type Collector struct {
 
 	connectionBrokerEnabled bool
 
+	// perfDataQuery samples the session and Connection Broker objects together,
+	// so the collector waits only once for the other performance counter queries.
+	perfDataQuery                            *pdh.Query
 	perfDataCollectorTerminalServicesSession *pdh.Collector[perfDataCounterValuesTerminalServicesSession]
 	perfDataCollectorBroker                  *pdh.Collector[perfDataCounterValuesBroker]
 
@@ -86,21 +91,28 @@ type Collector struct {
 
 	hServer windows.Handle
 
-	sessionInfo                 *prometheus.Desc
-	connectionBrokerPerformance *prometheus.Desc
-	handleCount                 *prometheus.Desc
-	pageFaultsPerSec            *prometheus.Desc
-	pageFileBytes               *prometheus.Desc
-	pageFileBytesPeak           *prometheus.Desc
-	percentCPUTime              *prometheus.Desc
-	poolNonPagedBytes           *prometheus.Desc
-	poolPagedBytes              *prometheus.Desc
-	privateBytes                *prometheus.Desc
-	threadCount                 *prometheus.Desc
-	virtualBytes                *prometheus.Desc
-	virtualBytesPeak            *prometheus.Desc
-	workingSet                  *prometheus.Desc
-	workingSetPeak              *prometheus.Desc
+	// sessionCache reuses the session_info state series of sessions whose
+	// name, user, host and state did not change.
+	sessionCache metriccache.Cache[uint32, sessionValues]
+	// labels reuses the label pairs of the per-session counter series.
+	labels labelcache.Cache
+
+	sessionInfo                        *prometheus.Desc
+	connectionBrokerPerformance        *prometheus.Desc
+	connectionBrokerPendingConnections *prometheus.Desc
+	handleCount                        *prometheus.Desc
+	pageFaultsPerSec                   *prometheus.Desc
+	pageFileBytes                      *prometheus.Desc
+	pageFileBytesPeak                  *prometheus.Desc
+	percentCPUTime                     *prometheus.Desc
+	poolNonPagedBytes                  *prometheus.Desc
+	poolPagedBytes                     *prometheus.Desc
+	privateBytes                       *prometheus.Desc
+	threadCount                        *prometheus.Desc
+	virtualBytes                       *prometheus.Desc
+	virtualBytesPeak                   *prometheus.Desc
+	workingSet                         *prometheus.Desc
+	workingSetPeak                     *prometheus.Desc
 }
 
 func New(config *Config) *Collector {
@@ -126,25 +138,28 @@ func (c *Collector) GetName() string {
 func (c *Collector) Close() error {
 	c.perfDataCollectorTerminalServicesSession.Close()
 	c.perfDataCollectorBroker.Close()
+	c.perfDataQuery.Close()
+
+	c.sessionCache.Reset()
+	c.labels.Reset()
 
 	// hServer is 0 if Build failed before opening the server.
 	if c.hServer == 0 || c.hServer == windows.InvalidHandle {
 		return nil
 	}
 
-	err := wtsapi32.WTSCloseServer(c.hServer)
+	wtsapi32.WTSCloseServer(c.hServer)
 	c.hServer = 0
 
-	if err != nil {
-		return fmt.Errorf("failed to close WTS server: %w", err)
-	}
-
 	return nil
 }
 
 func (c *Collector) Build(logger *slog.Logger, miSession *mi.Session) error {
 	c.logger = logger.With(slog.String("collector", Name))
 
+	c.sessionCache.Reset()
+	c.labels.Reset()
+
 	c.sessionInfo = prometheus.NewDesc(
 		prometheus.BuildFQName(types.Namespace, Name, "session_info"),
 		"Terminal Services sessions info",
@@ -153,10 +168,16 @@ func (c *Collector) Build(logger *slog.Logger, miSession *mi.Session) error {
 	)
 	c.connectionBrokerPerformance = prometheus.NewDesc(
 		prometheus.BuildFQName(types.Namespace, Name, "connection_broker_performance_total"),
-		"The total number of connections handled by the Connection Brokers since the service started.",
+		"The number of connections handled by the Connection Broker since the service started, by result.",
 		[]string{"connection"},
 		nil,
 	)
+	c.connectionBrokerPendingConnections = prometheus.NewDesc(
+		prometheus.BuildFQName(types.Namespace, Name, "connection_broker_pending_connections"),
+		"The number of redirected connections pending completion from the Remote Desktop Session Host servers in the farm.",
+		nil,
+		nil,
+	)
 	c.handleCount = prometheus.NewDesc(
 		prometheus.BuildFQName(types.Namespace, Name, "handles"),
 		"Total number of handles currently opened by this process. This number is the sum of the handles currently opened by each thread in this process.",
@@ -238,15 +259,28 @@ func (c *Collector) Build(logger *slog.Logger, miSession *mi.Session) error {
 
 	var err error
 
+	c.perfDataQuery, err = pdh.NewQuery()
+	if err != nil {
+		return fmt.Errorf("failed to open Terminal Services performance counter query: %w", err)
+	}
+
 	c.connectionBrokerEnabled = isConnectionBrokerServer()
 
 	if c.connectionBrokerEnabled {
-		c.perfDataCollectorBroker, err = pdh.NewCollector[perfDataCounterValuesBroker](c.logger, pdh.CounterTypeRaw, "Remote Desktop Connection Broker Counterset", pdh.InstancesAll)
+		c.perfDataCollectorBroker, err = pdh.NewQueryCollector[perfDataCounterValuesBroker](c.logger, c.perfDataQuery, pdh.CounterTypeRaw, "Remote Desktop Connection Broker Counterset", pdh.InstancesAll)
 		if err != nil {
-			return fmt.Errorf("failed to create Remote Desktop Connection Broker Counterset collector: %w", err)
+			// The service can be installed while its counters are not registered or
+			// unreadable. Keep the session metrics and skip only the broker metrics.
+			c.logger.Warn("failed to create Remote Desktop Connection Broker Counterset collector, skipping Connection Broker performance metrics",
+				slog.Any("err", err),
+			)
+
+			c.perfDataCollectorBroker.Close()
+			c.perfDataCollectorBroker = nil
+			c.connectionBrokerEnabled = false
 		}
 	} else {
-		logger.Debug("host is not a connection broker skipping Connection Broker performance metrics.")
+		c.logger.Debug("host is not a connection broker skipping Connection Broker performance metrics.")
 	}
 
 	c.hServer, err = wtsapi32.WTSOpenServer("")
@@ -254,7 +288,7 @@ func (c *Collector) Build(logger *slog.Logger, miSession *mi.Session) error {
 		return fmt.Errorf("failed to open WTS server: %w", err)
 	}
 
-	c.perfDataCollectorTerminalServicesSession, err = pdh.NewCollector[perfDataCounterValuesTerminalServicesSession](c.logger, pdh.CounterTypeRaw, "Terminal Services Session", pdh.InstancesAll)
+	c.perfDataCollectorTerminalServicesSession, err = pdh.NewQueryCollector[perfDataCounterValuesTerminalServicesSession](c.logger, c.perfDataQuery, pdh.CounterTypeRaw, "Terminal Services Session", pdh.InstancesAll)
 	if err != nil {
 		return fmt.Errorf("failed to create Terminal Services Session collector: %w", err)
 	}
@@ -271,6 +305,15 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error
 		errs = append(errs, fmt.Errorf("failed collecting terminal services session infos: %w", err))
 	}
 
+	// The session and Connection Broker counters read this sample. Unlike
+	// netframework, a failed sample, including pdh.ErrNoData, skips both, so
+	// they don't publish the values of an earlier sample again.
+	if err := c.perfDataQuery.Collect(); err != nil {
+		errs = append(errs, fmt.Errorf("failed to sample session and Connection Broker counters: %w", err))
+
+		return errors.Join(errs...)
+	}
+
 	if err := c.collectTSSessionCounters(ch); err != nil {
 		errs = append(errs, fmt.Errorf("failed collecting terminal services session count metrics: %w", err))
 	}
@@ -291,6 +334,9 @@ func (c *Collector) collectTSSessionCounters(ch chan<- prometheus.Metric) error
 		return fmt.Errorf("failed to collect Terminal Services Session metrics: %w", err)
 	}
 
+	scrape := c.labels.Begin()
+	defer scrape.Commit()
+
 	names := make(map[string]bool)
 
 	for _, data := range c.perfDataObjectTerminalServicesSession {
@@ -306,108 +352,108 @@ func (c *Collector) collectTSSessionCounters(ch chan<- prometheus.Metric) error
 
 		names[n] = true
 
-		ch <- prometheus.MustNewConstMetric(
+		ch <- scrape.MustNewConstMetric(
 			c.handleCount,
 			prometheus.GaugeValue,
 			data.HandleCount,
 			data.Name,
 		)
 
-		ch <- prometheus.MustNewConstMetric(
+		ch <- scrape.MustNewConstMetric(
 			c.pageFaultsPerSec,
 			prometheus.CounterValue,
 			data.PageFaultsPersec,
 			data.Name,
 		)
 
-		ch <- prometheus.MustNewConstMetric(
+		ch <- scrape.MustNewConstMetric(
 			c.pageFileBytes,
 			prometheus.GaugeValue,
 			data.PageFileBytes,
 			data.Name,
 		)
 
-		ch <- prometheus.MustNewConstMetric(
+		ch <- scrape.MustNewConstMetric(
 			c.pageFileBytesPeak,
 			prometheus.GaugeValue,
 			data.PageFileBytesPeak,
 			data.Name,
 		)
 
-		ch <- prometheus.MustNewConstMetric(
+		ch <- scrape.MustNewConstMetric(
 			c.percentCPUTime,
 			prometheus.CounterValue,
 			data.PercentPrivilegedTime,
-			data.Name,
 			"privileged",
+			data.Name,
 		)
 
-		ch <- prometheus.MustNewConstMetric(
+		ch <- scrape.MustNewConstMetric(
 			c.percentCPUTime,
 			prometheus.CounterValue,
 			data.PercentProcessorTime,
-			data.Name,
 			"processor",
+			data.Name,
 		)
 
-		ch <- prometheus.MustNewConstMetric(
+		ch <- scrape.MustNewConstMetric(
 			c.percentCPUTime,
 			prometheus.CounterValue,
 			data.PercentUserTime,
-			data.Name,
 			"user",
+			data.Name,
 		)
 
-		ch <- prometheus.MustNewConstMetric(
+		ch <- scrape.MustNewConstMetric(
 			c.poolNonPagedBytes,
 			prometheus.GaugeValue,
 			data.PoolNonpagedBytes,
 			data.Name,
 		)
 
-		ch <- prometheus.MustNewConstMetric(
+		ch <- scrape.MustNewConstMetric(
 			c.poolPagedBytes,
 			prometheus.GaugeValue,
 			data.PoolPagedBytes,
 			data.Name,
 		)
 
-		ch <- prometheus.MustNewConstMetric(
+		ch <- scrape.MustNewConstMetric(
 			c.privateBytes,
 			prometheus.GaugeValue,
 			data.PrivateBytes,
 			data.Name,
 		)
 
-		ch <- prometheus.MustNewConstMetric(
+		ch <- scrape.MustNewConstMetric(
 			c.threadCount,
 			prometheus.GaugeValue,
 			data.ThreadCount,
 			data.Name,
 		)
 
-		ch <- prometheus.MustNewConstMetric(
+		ch <- scrape.MustNewConstMetric(
 			c.virtualBytes,
 			prometheus.GaugeValue,
 			data.VirtualBytes,
 			data.Name,
 		)
 
-		ch <- prometheus.MustNewConstMetric(
+		ch <- scrape.MustNewConstMetric(
 			c.virtualBytesPeak,
 			prometheus.GaugeValue,
 			data.VirtualBytesPeak,
 			data.Name,
 		)
 
-		ch <- prometheus.MustNewConstMetric(
+		ch <- scrape.MustNewConstMetric(
 			c.workingSet,
 			prometheus.GaugeValue,
 			data.WorkingSet,
 			data.Name,
 		)
 
-		ch <- prometheus.MustNewConstMetric(
+		ch <- scrape.MustNewConstMetric(
 			c.workingSetPeak,
 			prometheus.GaugeValue,
 			data.WorkingSetPeak,
@@ -432,10 +478,9 @@ func (c *Collector) collectCollectionBrokerPerformanceCounter(ch chan<- promethe
 	)
 
 	ch <- prometheus.MustNewConstMetric(
-		c.connectionBrokerPerformance,
-		prometheus.CounterValue,
+		c.connectionBrokerPendingConnections,
+		prometheus.GaugeValue,
 		c.perfDataObjectBroker[0].PendingConnections,
-		"Pending",
 	)
 
 	ch <- prometheus.MustNewConstMetric(
@@ -454,6 +499,10 @@ func (c *Collector) collectWTSSessions(ch chan<- prometheus.Metric) error {
 		return fmt.Errorf("failed to enumerate WTS sessions: %w", err)
 	}
 
+	scrape := c.sessionCache.Begin()
+	// Sessions that were not collected in this scrape are dropped from the cache.
+	defer scrape.Commit()
+
 	for _, session := range sessions {
 		// only connect metrics for remote named sessions
 		n := strings.ReplaceAll(session.SessionName, "#", " ")
@@ -463,27 +512,56 @@ func (c *Collector) collectWTSSessions(ch chan<- prometheus.Metric) error {
 
 		userName := session.UserName
 		if session.DomainName != "" {
-			userName = fmt.Sprintf("%s\\%s", session.DomainName, session.UserName)
+			userName = session.DomainName + "\\" + session.UserName
 		}
 
-		for stateID, stateName := range wtsapi32.WTSSessionStates {
-			isState := 0.0
-			if session.State == stateID {
-				isState = 1.0
-			}
+		values := sessionValues{
+			name:  n,
+			user:  userName,
+			host:  session.HostName,
+			state: session.State,
+		}
 
-			ch <- prometheus.MustNewConstMetric(
-				c.sessionInfo,
-				prometheus.GaugeValue,
-				isState,
-				n,
-				userName,
-				session.HostName,
-				stateName,
-				strconv.Itoa(int(session.SessionID)),
-			)
+		for _, metric := range scrape.Get(session.SessionID, values, func() []prometheus.Metric {
+			return c.sessionInfoMetrics(session.SessionID, values)
+		}) {
+			ch <- metric
 		}
 	}
 
 	return nil
 }
+
+// sessionValues are the inputs of the session_info series of a session.
+type sessionValues struct {
+	name  string
+	user  string
+	host  string
+	state wtsapi32.WTSConnectState
+}
+
+// sessionInfoMetrics builds the session_info state series of a session.
+func (c *Collector) sessionInfoMetrics(sessionID uint32, values sessionValues) []prometheus.Metric {
+	id := strconv.FormatUint(uint64(sessionID), 10)
+	metrics := make([]prometheus.Metric, 0, len(wtsapi32.WTSSessionStates))
+
+	for stateID, stateName := range wtsapi32.WTSSessionStates {
+		isState := 0.0
+		if values.state == stateID {
+			isState = 1.0
+		}
+
+		metrics = append(metrics, prometheus.MustNewConstMetric(
+			c.sessionInfo,
+			prometheus.GaugeValue,
+			isState,
+			values.name,
+			values.user,
+			values.host,
+			stateName,
+			id,
+		))
+	}
+
+	return metrics
+}

internal/collector/terminal_services/terminal_services_test.go

@@ -34,6 +34,14 @@ func BenchmarkCollector(b *testing.B) {
 func TestCollector(t *testing.T) {
 	metrics := testutils.TestCollector(t, terminal_services.New, nil)
 
+	// Pending Connections is a current count, published as the pending
+	// connections gauge, not as a series of the connection counter.
+	for _, metric := range metrics["windows_terminal_services_connection_broker_performance_total"].GetMetric() {
+		for _, label := range metric.GetLabel() {
+			require.False(t, label.GetName() == "connection" && label.GetValue() == "Pending", "connection_broker_performance_total must not have a Pending series")
+		}
+	}
+
 	userName := os.Getenv("WINDOWS_EXPORTER_TEST_RDP_USER")
 	if userName == "" {
 		return
@@ -51,4 +59,10 @@ func TestCollector(t *testing.T) {
 	handles := testutils.RequireFixtureMetric(t, metrics, terminal_services.Name, "windows_terminal_services_handles", prometheus.Labels{"session_name": sessionName})
 	require.NotNil(t, handles)
 	require.Positive(t, handles.GetGauge().GetValue(), "RDP fixture has no session processes")
+
+	cpuTime := testutils.RequireFixtureMetric(t, metrics, terminal_services.Name, "windows_terminal_services_cpu_time_seconds_total", prometheus.Labels{
+		"mode":         "user",
+		"session_name": sessionName,
+	})
+	require.NotNil(t, cpuTime)
 }

internal/headers/wtsapi32/wtsapi32.go

@@ -18,7 +18,6 @@
 package wtsapi32
 
 import (
-	"errors"
 	"fmt"
 	"log/slog"
 	"unsafe"
@@ -145,13 +144,10 @@ func WTSOpenServer(server string) (windows.Handle, error) {
 	return serverHandle, nil
 }
 
-func WTSCloseServer(server windows.Handle) error {
-	r1, _, err := procWTSCloseServer.Call(uintptr(server))
-	if r1 != 1 && !errors.Is(err, windows.ERROR_SUCCESS) {
-		return fmt.Errorf("failed to close server: %w", err)
-	}
-
-	return nil
+// WTSCloseServer closes a handle opened by WTSOpenServer. The function returns
+// void, so neither r1 nor the last error say whether it failed.
+func WTSCloseServer(server windows.Handle) {
+	_, _, _ = procWTSCloseServer.Call(uintptr(server))
 }
 
 func WTSFreeMemoryEx(class WTSTypeClass, pMemory uintptr, numberOfEntries uint32) error {