windows_exporter/internal/collector/process/process_worker.go

// 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 process

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"runtime/debug"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/prometheus-community/windows_exporter/internal/mi"
	"github.com/prometheus/client_golang/prometheus"
	"golang.org/x/sys/windows"
)

type processWorkerRequest struct {
	ch                       chan<- prometheus.Metric
	name                     string
	appPoolName              string
	performanceCounterValues perfDataCounterValues
	waitGroup                *sync.WaitGroup
}

func (c *Collector) collect(ch chan<- prometheus.Metric, maxScrapeDuration time.Duration) error {
	err := c.perfDataCollector.Collect(&c.perfDataObject)
	if err != nil {
		return fmt.Errorf("failed to collect metrics: %w", err)
	}

	requests := make([]processWorkerRequest, 0, len(c.perfDataObject))

	var workerProcessIDs []uint32

	for _, process := range c.perfDataObject {
		// Duplicate processes are suffixed #, and an index number. Remove those.
		name, _, _ := strings.Cut(process.Name, ":") // Process V2

		// Duplicate processes are suffixed #, and an index number. Remove those.
		name, _, _ = strings.Cut(name, "#") // Process V1

		if c.config.ProcessExclude.MatchString(name) || !c.config.ProcessInclude.MatchString(name) {
			continue
		}

		if process.ProcessID == 0 && name != "Idle" {
			c.logger.LogAttrs(context.Background(), slog.LevelDebug, "Skipping process with PID 0",
				slog.String("name", name),
				slog.String("process_name", process.Name),
				slog.Any("process", fmt.Sprintf("%+v", process)),
			)

			continue
		}

		if c.config.EnableWorkerProcess && isWorkerProcess(name) {
			workerProcessIDs = append(workerProcessIDs, uint32(process.ProcessID))
		}

		requests = append(requests, processWorkerRequest{
			ch:                       ch,
			name:                     name,
			performanceCounterValues: process,
		})
	}

	var appPools map[uint32]string

	if len(workerProcessIDs) > 0 {
		appPools, err = resolveAppPools(c.logger, workerProcessIDs, workerProcessAppPool, c.workerProcessWMIQuery(maxScrapeDuration))
	}

	wg := &sync.WaitGroup{}

	for _, req := range requests {
		req.appPoolName = appPools[uint32(req.performanceCounterValues.ProcessID)]
		req.waitGroup = wg

		wg.Add(1)

		c.workerCh <- req
	}

	wg.Wait()

	return err
}

// workerProcessWMIQuery returns the root\WebAdministration query, or nil if it was unavailable during Build.
func (c *Collector) workerProcessWMIQuery(maxScrapeDuration time.Duration) func() ([]WorkerProcess, error) {
	miSession, miQuery := c.miSession, c.workerProcessMIQueryQuery
	if miSession == nil {
		return nil
	}

	return func() ([]WorkerProcess, error) {
		var workerProcesses []WorkerProcess

		err := miSession.Query(&workerProcesses, mi.NamespaceRootWebAdministration, miQuery, maxScrapeDuration)

		return workerProcesses, err
	}
}

// resolveAppPools maps IIS worker process IDs to application pool names.
//
// The command line of the worker process is the primary source: it needs no
// optional feature and no WMI round trip. WMI is queried at most once, for the
// worker processes whose command line couldn't be read or has no -ap argument,
// and only if queryWMI isn't nil. Unresolved processes have no entry.
func resolveAppPools(
	logger *slog.Logger,
	pids []uint32,
	readCommandLine func(pid uint32) (string, error),
	queryWMI func() ([]WorkerProcess, error),
) (map[uint32]string, error) {
	appPools := make(map[uint32]string, len(pids))

	var unresolved []uint32

	for _, pid := range pids {
		appPool, err := readCommandLine(pid)
		if err == nil {
			appPools[pid] = appPool

			continue
		}

		logger.LogAttrs(context.Background(), slog.LevelDebug, "Failed to read the IIS application pool from the worker process command line",
			slog.Uint64("pid", uint64(pid)),
			slog.Any("err", err),
		)

		// OpenProcess fails with ERROR_INVALID_PARAMETER for processes that exited after the
		// performance counter snapshot. WAS no longer reports them either.
		if !errors.Is(err, windows.ERROR_INVALID_PARAMETER) {
			unresolved = append(unresolved, pid)
		}
	}

	if len(unresolved) == 0 || queryWMI == nil {
		return appPools, nil
	}

	workerProcesses, err := queryWMI()
	if err != nil {
		return appPools, fmt.Errorf("WMI query for collector.process.iis failed: %w", err)
	}

	for _, pid := range unresolved {
		for _, wp := range workerProcesses {
			if wp.ProcessId == uint64(pid) && wp.AppPoolName != "" {
				appPools[pid] = wp.AppPoolName

				break
			}
		}
	}

	return appPools, nil
}

func (c *Collector) collectWorker(requests <-chan processWorkerRequest) {
	defer c.workerWG.Done()

	for req := range requests {
		(func() {
			defer req.waitGroup.Done()
			defer func() {
				if r := recover(); r != nil {
					c.logger.Error("Worker panic",
						slog.Any("panic", r),
						slog.String("stack", string(debug.Stack())),
					)
				}
			}()

			ch := req.ch
			name := req.name
			data := req.performanceCounterValues

			pid := uint64(data.ProcessID)
			parentPID := strconv.FormatUint(uint64(data.CreatingProcessID), 10)

			if req.appPoolName != "" {
				name = name + "_" + req.appPoolName
			}

			cmdLine, processOwner, processGroupID, err := c.getProcessInformation(uint32(pid))
			if err != nil {
				c.logger.LogAttrs(context.Background(), slog.LevelDebug, "Failed to get process information",
					slog.Uint64("pid", pid),
					slog.Any("err", err),
				)
			}

			pidString := strconv.FormatUint(pid, 10)

			ch <- prometheus.MustNewConstMetric(
				c.info,
				prometheus.GaugeValue,
				1.0,
				name, pidString, parentPID, strconv.Itoa(int(processGroupID)), processOwner, cmdLine,
			)

			startTime := data.ElapsedTime
			if c.config.CounterVersion == 2 {
				// For V2, the ElapsedTime is in seconds, so we need to convert it to a timestamp.
				// The start time is the current time minus the elapsed time.
				startTime = float64(time.Now().UnixMicro())/1e6 - data.ElapsedTime
			}

			ch <- prometheus.MustNewConstMetric(
				c.startTime,
				prometheus.GaugeValue,
				startTime,
				name, pidString,
			)

			ch <- prometheus.MustNewConstMetric(
				c.handleCount,
				prometheus.GaugeValue,
				data.HandleCount,
				name, pidString,
			)

			ch <- prometheus.MustNewConstMetric(
				c.cpuTimeTotal,
				prometheus.CounterValue,
				data.PercentPrivilegedTime,
				name, pidString, "privileged",
			)

			ch <- prometheus.MustNewConstMetric(
				c.cpuTimeTotal,
				prometheus.CounterValue,
				data.PercentUserTime,
				name, pidString, "user",
			)

			ch <- prometheus.MustNewConstMetric(
				c.ioBytesTotal,
				prometheus.CounterValue,
				data.IoOtherBytesPerSec,
				name, pidString, "other",
			)

			ch <- prometheus.MustNewConstMetric(
				c.ioOperationsTotal,
				prometheus.CounterValue,
				data.IoOtherOperationsPerSec,
				name, pidString, "other",
			)

			ch <- prometheus.MustNewConstMetric(
				c.ioBytesTotal,
				prometheus.CounterValue,
				data.IoReadBytesPerSec,
				name, pidString, "read",
			)

			ch <- prometheus.MustNewConstMetric(
				c.ioOperationsTotal,
				prometheus.CounterValue,
				data.IoReadOperationsPerSec,
				name, pidString, "read",
			)

			ch <- prometheus.MustNewConstMetric(
				c.ioBytesTotal,
				prometheus.CounterValue,
				data.IoWriteBytesPerSec,
				name, pidString, "write",
			)

			ch <- prometheus.MustNewConstMetric(
				c.ioOperationsTotal,
				prometheus.CounterValue,
				data.IoWriteOperationsPerSec,
				name, pidString, "write",
			)

			ch <- prometheus.MustNewConstMetric(
				c.pageFaultsTotal,
				prometheus.CounterValue,
				data.PageFaultsPerSec,
				name, pidString,
			)

			ch <- prometheus.MustNewConstMetric(
				c.pageFileBytes,
				prometheus.GaugeValue,
				data.PageFileBytes,
				name, pidString,
			)

			ch <- prometheus.MustNewConstMetric(
				c.poolBytes,
				prometheus.GaugeValue,
				data.PoolNonPagedBytes,
				name, pidString, "nonpaged",
			)

			ch <- prometheus.MustNewConstMetric(
				c.poolBytes,
				prometheus.GaugeValue,
				data.PoolPagedBytes,
				name, pidString, "paged",
			)

			ch <- prometheus.MustNewConstMetric(
				c.priorityBase,
				prometheus.GaugeValue,
				data.PriorityBase,
				name, pidString,
			)

			ch <- prometheus.MustNewConstMetric(
				c.privateBytes,
				prometheus.GaugeValue,
				data.PrivateBytes,
				name, pidString,
			)

			ch <- prometheus.MustNewConstMetric(
				c.threadCount,
				prometheus.GaugeValue,
				data.ThreadCount,
				name, pidString,
			)

			ch <- prometheus.MustNewConstMetric(
				c.virtualBytes,
				prometheus.GaugeValue,
				data.VirtualBytes,
				name, pidString,
			)

			ch <- prometheus.MustNewConstMetric(
				c.workingSetPrivate,
				prometheus.GaugeValue,
				data.WorkingSetPrivate,
				name, pidString,
			)

			ch <- prometheus.MustNewConstMetric(
				c.workingSetPeak,
				prometheus.GaugeValue,
				data.WorkingSetPeak,
				name, pidString,
			)

			ch <- prometheus.MustNewConstMetric(
				c.workingSet,
				prometheus.GaugeValue,
				data.WorkingSet,
				name, pidString,
			)
		})()
	}
}