container: fix hostprocess memory metrics and containerd bundle handling (#2555)

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

docs/collector.container.md

@@ -10,30 +10,39 @@ Enabled by default? | No
 
 ## Flags
 
-None
+### `--collector.container.enabled`
+
+Comma-separated list of collectors to use. Defaults to all, if not specified. Available collectors: `hcs`, `hostprocess`.
+
+- `hcs` collects containers managed by the Host Compute Service, for example Docker or process-isolated containerd containers.
+- `hostprocess` collects Kubernetes HostProcess containers, which run in Win32 job objects instead of HCS.
+
+### `--collector.container.containerd-state-dir`
+
+Path to the containerd state directory of the `k8s.io` namespace. Kubernetes labels and HostProcess containers are read from the OCI bundles in this directory. Defaults to `C:\ProgramData\containerd\state\io.containerd.runtime.v2.task\k8s.io\`.
 
 ## Metrics
 
-| Name                                                       | Description                            | Type    | Labels                                                   |
-|------------------------------------------------------------|----------------------------------------|---------|----------------------------------------------------------|
-| `windows_container_available`                              | Available                              | counter | `container_id`,`namespace`,`pod`,`container`,            |
-| `windows_container_count`                                  | Number of containers                   | gauge   | `container_id`,`namespace`,`pod`,`container`,            |
-| `windows_container_cpu_usage_seconds_kernelmode`           | Run time in Kernel mode in Seconds     | counter | `container_id`,`namespace`,`pod`,`container`,            |
-| `windows_container_cpu_usage_seconds_usermode`             | Run Time in User mode in Seconds       | counter | `container_id`,`namespace`,`pod`,`container`,            |
-| `windows_container_cpu_usage_seconds_total`                | Total Run time in Seconds              | counter | `container_id`,`namespace`,`pod`,`container`,            |
-| `windows_container_memory_usage_commit_bytes`              | Memory Usage Commit Bytes              | gauge   | `container_id`,`namespace`,`pod`,`container`,            |
-| `windows_container_memory_usage_commit_peak_bytes`         | Memory Usage Commit Peak Bytes         | gauge   | `container_id`,`namespace`,`pod`,`container`,            |
-| `windows_container_memory_usage_private_working_set_bytes` | Memory Usage Private Working Set Bytes | gauge   | `container_id`,`namespace`,`pod`,`container`,            |
-| `windows_container_network_receive_bytes_total`            | Bytes Received on Interface            | counter | `container_id`,`namespace`,`pod`,`container`,`interface` |
-| `windows_container_network_receive_packets_total`          | Packets Received on Interface          | counter | `container_id`,`namespace`,`pod`,`container`,`interface` |
-| `windows_container_network_receive_packets_dropped_total`  | Dropped Incoming Packets on Interface  | counter | `container_id`,`namespace`,`pod`,`container`,`interface` |
-| `windows_container_network_transmit_bytes_total`           | Bytes Sent on Interface                | counter | `container_id`,`namespace`,`pod`,`container`,`interface` |
-| `windows_container_network_transmit_packets_total`         | Packets Sent on Interface              | counter | `container_id`,`namespace`,`pod`,`container`,`interface` |
-| `windows_container_network_transmit_packets_dropped_total` | Dropped Outgoing Packets on Interface  | counter | `container_id`,`namespace`,`pod`,`container`,`interface` |
-| `windows_container_storage_read_count_normalized_total`    | Read Count Normalized                  | counter | `container_id`,`namespace`,`pod`,`container`,            |
-| `windows_container_storage_read_size_bytes_total`          | Read Size Bytes                        | counter | `container_id`,`namespace`,`pod`,`container`,            |
-| `windows_container_storage_write_count_normalized_total`   | Write Count Normalized                 | counter | `container_id`,`namespace`,`pod`,`container`,            |
-| `windows_container_storage_write_size_bytes_total`         | Write Size Bytes                       | counter | `container_id`,`namespace`,`pod`,`container`,            |
+| Name                                                       | Description                            | Type    | Labels                                                     |
+|------------------------------------------------------------|----------------------------------------|---------|------------------------------------------------------------|
+| `windows_container_available`                              | Available                              | gauge   | `container_id`,`namespace`,`pod`,`container`,`hostprocess` |
+| `windows_container_count`                                  | Number of running HCS containers       | gauge   | None                                                       |
+| `windows_container_cpu_usage_seconds_kernelmode`           | Runtime in Kernel mode in Seconds      | counter | `container_id`,`namespace`,`pod`,`container`               |
+| `windows_container_cpu_usage_seconds_usermode`             | Runtime in User mode in Seconds        | counter | `container_id`,`namespace`,`pod`,`container`               |
+| `windows_container_cpu_usage_seconds_total`                | Total Runtime in Seconds               | counter | `container_id`,`namespace`,`pod`,`container`               |
+| `windows_container_memory_usage_commit_bytes`              | Memory Usage Commit Bytes              | gauge   | `container_id`,`namespace`,`pod`,`container`               |
+| `windows_container_memory_usage_commit_peak_bytes`         | Memory Usage Commit Peak Bytes         | gauge   | `container_id`,`namespace`,`pod`,`container`               |
+| `windows_container_memory_usage_private_working_set_bytes` | Memory Usage Private Working Set Bytes | gauge   | `container_id`,`namespace`,`pod`,`container`               |
+| `windows_container_network_receive_bytes_total`            | Bytes Received on Interface            | counter | `container_id`,`namespace`,`pod`,`container`,`interface`   |
+| `windows_container_network_receive_packets_total`          | Packets Received on Interface          | counter | `container_id`,`namespace`,`pod`,`container`,`interface`   |
+| `windows_container_network_receive_packets_dropped_total`  | Dropped Incoming Packets on Interface  | counter | `container_id`,`namespace`,`pod`,`container`,`interface`   |
+| `windows_container_network_transmit_bytes_total`           | Bytes Sent on Interface                | counter | `container_id`,`namespace`,`pod`,`container`,`interface`   |
+| `windows_container_network_transmit_packets_total`         | Packets Sent on Interface              | counter | `container_id`,`namespace`,`pod`,`container`,`interface`   |
+| `windows_container_network_transmit_packets_dropped_total` | Dropped Outgoing Packets on Interface  | counter | `container_id`,`namespace`,`pod`,`container`,`interface`   |
+| `windows_container_storage_read_count_normalized_total`    | Read Count Normalized                  | counter | `container_id`,`namespace`,`pod`,`container`               |
+| `windows_container_storage_read_size_bytes_total`          | Read Size Bytes                        | counter | `container_id`,`namespace`,`pod`,`container`               |
+| `windows_container_storage_write_count_normalized_total`   | Write Count Normalized                 | counter | `container_id`,`namespace`,`pod`,`container`               |
+| `windows_container_storage_write_size_bytes_total`         | Write Size Bytes                       | counter | `container_id`,`namespace`,`pod`,`container`               |
 
 ### Example metric
 _windows_container_network_receive_bytes_total{container_id="docker://1bd30e8b8ac28cbd76a9b697b4d7bb9d760267b0733d1bc55c60024e98d1e43e",interface="822179E7-002C-4280-ABBA-28BCFE401826"} 9.3305343e+07_
@@ -41,10 +50,10 @@ _windows_container_network_receive_bytes_total{container_id="docker://1bd30e8b8a
 This metric means that total _9.3305343e+07_ bytes received on interface _822179E7-002C-4280-ABBA-28BCFE401826_ for container _docker://1bd30e8b8ac28cbd76a9b697b4d7bb9d760267b0733d1bc55c60024e98d1e43e_
 
 ## Useful queries
-Attach labels namespace/pod/container fow windows container metrics.
+Attach labels namespace/pod/container for Windows container metrics.
 ```
-# kube_pod_container_info(a metric of kube-state-metrics) has labels namespace/pod/container/container_id for a container, while windows container metrics only have container_id.
-# Attaching labels namespace/pod/container for windows container metrics, is useful to query for windows pods.
+# kube_pod_container_info(a metric of kube-state-metrics) has labels namespace/pod/container/container_id for a container, while Windows container metrics only have container_id.
+# Attaching labels namespace/pod/container for Windows container metrics, is useful to query for Windows pods.
 windows_container_network_receive_bytes_total * on(container_id) group_left(namespace, pod, container) kube_pod_container_info{container_id!=""}
 ```
 ## Alerting examples

internal/collector/container/container.go

@@ -24,6 +24,7 @@ import (
 	"io/fs"
 	"log/slog"
 	"os"
+	"path/filepath"
 	"slices"
 	"strings"
 	"sync"
@@ -76,7 +77,11 @@ type Collector struct {
 
 	// The annotation caches are keyed by the container ID without the runtime prefix.
 	annotationsCacheHCS map[string]containerInfo
-	annotationsCacheJob map[string]containerInfo
+	// annotationsCacheJob holds every containerd bundle, so that each config.json is parsed only once.
+	annotationsCacheJob map[string]jobContainer
+
+	// stateDirMissingLogged avoids logging a missing containerd state directory on every scrape.
+	stateDirMissingLogged bool
 
 	// Presence
 	containerAvailable *prometheus.Desc
@@ -116,10 +121,25 @@ type containerInfo struct {
 	container string
 }
 
+type jobContainer struct {
+	info        containerInfo
+	hostProcess bool
+}
+
 type ociSpec struct {
 	Annotations map[string]string `json:"annotations"`
 }
 
+// containerInfo returns the Kubernetes metadata of the container.
+func (s ociSpec) containerInfo(id string) containerInfo {
+	return containerInfo{
+		id:        id,
+		namespace: s.Annotations["io.kubernetes.cri.sandbox-namespace"],
+		pod:       s.Annotations["io.kubernetes.cri.sandbox-name"],
+		container: s.Annotations["io.kubernetes.cri.container-name"],
+	}
+}
+
 // New constructs a new Collector.
 func New(config *Config) *Collector {
 	if config == nil {
@@ -182,7 +202,7 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error {
 	}
 
 	c.annotationsCacheHCS = make(map[string]containerInfo)
-	c.annotationsCacheJob = make(map[string]containerInfo)
+	c.annotationsCacheJob = make(map[string]jobContainer)
 
 	// Without the Containers feature, the Host Compute Service is missing and
 	// every scrape would fail. Report the collector as unsupported instead, so
@@ -351,7 +371,7 @@ func (c *Collector) collectHCS(ch chan<- prometheus.Metric) error {
 
 	var countersCount float64
 
-	containerIDs := make([]string, 0, len(containers))
+	containerIDs := make(map[string]struct{}, len(containers))
 	collectErrors := make([]error, 0)
 
 	for _, container := range containers {
@@ -359,46 +379,33 @@ func (c *Collector) collectHCS(ch chan<- prometheus.Metric) error {
 			continue
 		}
 
-		containerIDs = append(containerIDs, container.ID)
+		containerIDs[container.ID] = struct{}{}
 
 		countersCount++
 
-		var (
-			namespace     string
-			podName       string
-			containerName string
-		)
-
-		if _, ok := c.annotationsCacheHCS[container.ID]; !ok {
-			if spec, err := c.getContainerAnnotations(container.ID); err == nil {
-				namespace = spec.Annotations["io.kubernetes.cri.sandbox-namespace"]
-				podName = spec.Annotations["io.kubernetes.cri.sandbox-name"]
-				containerName = spec.Annotations["io.kubernetes.cri.container-name"]
-			}
-
-			c.annotationsCacheHCS[container.ID] = containerInfo{
-				id:        getContainerIdWithPrefix(container),
-				namespace: namespace,
-				pod:       podName,
-				container: containerName,
-			}
+		info, ok := c.annotationsCacheHCS[container.ID]
+		if !ok {
+			// Docker containers have no containerd bundle, so they keep empty Kubernetes labels.
+			spec, _ := c.getContainerAnnotations(container.ID)
+			info = spec.containerInfo(getContainerIdWithPrefix(container))
+			c.annotationsCacheHCS[container.ID] = info
 		}
 
-		if err = c.collectHCSContainer(ch, container, c.annotationsCacheHCS[container.ID]); err != nil {
+		if err = c.collectHCSContainer(ch, container, info); err != nil {
 			if errors.Is(err, hcs.ErrIDNotFound) {
 				c.logger.Debug("err in fetching container statistics",
 					slog.String("container_id", container.ID),
-					slog.String("container_name", c.annotationsCacheHCS[container.ID].container),
-					slog.String("container_pod_name", c.annotationsCacheHCS[container.ID].pod),
-					slog.String("container_namespace", c.annotationsCacheHCS[container.ID].namespace),
+					slog.String("container_name", info.container),
+					slog.String("container_pod_name", info.pod),
+					slog.String("container_namespace", info.namespace),
 					slog.Any("err", err),
 				)
 			} else {
 				c.logger.Error("err in fetching container statistics",
 					slog.String("container_id", container.ID),
-					slog.String("container_name", c.annotationsCacheHCS[container.ID].container),
-					slog.String("container_pod_name", c.annotationsCacheHCS[container.ID].pod),
-					slog.String("container_namespace", c.annotationsCacheHCS[container.ID].namespace),
+					slog.String("container_name", info.container),
+					slog.String("container_pod_name", info.pod),
+					slog.String("container_namespace", info.namespace),
 					slog.Any("err", err),
 				)
 
@@ -419,7 +426,7 @@ func (c *Collector) collectHCS(ch chan<- prometheus.Metric) error {
 
 	// Remove containers that are no longer running
 	for containerID := range c.annotationsCacheHCS {
-		if !slices.Contains(containerIDs, containerID) {
+		if _, ok := containerIDs[containerID]; !ok {
 			delete(c.annotationsCacheHCS, containerID)
 		}
 	}
@@ -637,82 +644,73 @@ func (c *Collector) collectNetworkMetrics(ch chan<- prometheus.Metric) error {
 //
 // Job containers are containers that aren't managed by HCS, e.g host process containers.
 func (c *Collector) collectJobContainers(ch chan<- prometheus.Metric) error {
-	containerDStateFS := os.DirFS(c.config.ContainerDStateDir)
+	// Each directory in the containerd state directory is the OCI bundle of a running task.
+	entries, err := os.ReadDir(c.config.ContainerDStateDir)
+	if err != nil {
+		if !errors.Is(err, fs.ErrNotExist) {
+			return fmt.Errorf("error in reading containerd state directory: %w", err)
+		}
 
-	allContainerIDs := make([]string, 0, len(c.annotationsCacheJob)+len(c.annotationsCacheHCS))
-	jobContainerIDs := make([]string, 0, len(allContainerIDs))
+		if !c.stateDirMissingLogged {
+			c.logger.Warn("containerd state directory does not exist",
+				slog.String("path", c.config.ContainerDStateDir),
+				slog.Any("err", err),
+			)
 
-	if err := fs.WalkDir(containerDStateFS, ".", func(path string, d fs.DirEntry, err error) error {
-		if err != nil {
-			if errors.Is(err, fs.ErrNotExist) {
-				c.logger.Warn("containerd state directory does not exist",
-					slog.String("path", c.config.ContainerDStateDir),
+			c.stateDirMissingLogged = true
+		}
+
+		clear(c.annotationsCacheJob)
+
+		return nil
+	}
+
+	c.stateDirMissingLogged = false
+
+	bundles := make(map[string]struct{}, len(entries))
+	errs := make([]error, 0)
+
+	for _, entry := range entries {
+		if !entry.IsDir() {
+			continue
+		}
+
+		containerID := entry.Name()
+		bundles[containerID] = struct{}{}
+
+		container, ok := c.annotationsCacheJob[containerID]
+		if !ok {
+			spec, err := c.getContainerAnnotations(containerID)
+			if err != nil {
+				// containerd may still be writing the bundle, so retry on the next scrape.
+				c.logger.Debug("error in reading container annotations",
+					slog.String("container_id", containerID),
 					slog.Any("err", err),
 				)
 
-				return nil
+				continue
 			}
 
-			return err
-		}
-
-		if path == "." {
-			return nil
-		}
-
-		if !d.IsDir() {
-			return nil
-		}
-
-		if _, err := os.Stat(path + "\\config.json"); err != nil {
-			containerID := strings.TrimPrefix(strings.Replace(path, c.config.ContainerDStateDir, "", 1), `\`)
-
-			if spec, err := c.getContainerAnnotations(containerID); err == nil {
-				isHostProcess, ok := spec.Annotations["microsoft.com/hostprocess-container"]
-				if ok && isHostProcess == "true" {
-					allContainerIDs = append(allContainerIDs, containerID)
-
-					if _, ok := c.annotationsCacheJob[containerID]; !ok {
-						var (
-							namespace     string
-							podName       string
-							containerName string
-						)
-
-						namespace = spec.Annotations["io.kubernetes.cri.sandbox-namespace"]
-						podName = spec.Annotations["io.kubernetes.cri.sandbox-name"]
-						containerName = spec.Annotations["io.kubernetes.cri.container-name"]
-
-						c.annotationsCacheJob[containerID] = containerInfo{
-							id:        "containerd://" + containerID,
-							namespace: namespace,
-							pod:       podName,
-							container: containerName,
-						}
-					}
-				}
+			container = jobContainer{
+				info:        spec.containerInfo("containerd://" + containerID),
+				hostProcess: spec.Annotations["microsoft.com/hostprocess-container"] == "true",
 			}
+
+			c.annotationsCacheJob[containerID] = container
 		}
 
-		// Skip the directory content
-		return fs.SkipDir
-	}); err != nil {
-		return fmt.Errorf("error in walking containerd state directory: %w", err)
-	}
+		if !container.hostProcess {
+			continue
+		}
 
-	errs := make([]error, 0)
-
-	for _, containerID := range allContainerIDs {
-		if err := c.collectJobContainer(ch, containerID); err != nil {
+		if err := c.collectJobContainer(ch, containerID, container.info); err != nil {
 			errs = append(errs, err)
-		} else {
-			jobContainerIDs = append(jobContainerIDs, containerID)
 		}
 	}
 
 	// Remove containers that are no longer running
 	for containerID := range c.annotationsCacheJob {
-		if !slices.Contains(jobContainerIDs, containerID) {
+		if _, ok := bundles[containerID]; !ok {
 			delete(c.annotationsCacheJob, containerID)
 		}
 	}
@@ -720,7 +718,7 @@ func (c *Collector) collectJobContainers(ch chan<- prometheus.Metric) error {
 	return errors.Join(errs...)
 }
 
-func (c *Collector) collectJobContainer(ch chan<- prometheus.Metric, containerID string) error {
+func (c *Collector) collectJobContainer(ch chan<- prometheus.Metric, containerID string, containerInfo containerInfo) error {
 	jobObjectHandle, err := kernel32.OpenJobObject("Global\\JobContainer_" + containerID)
 	if err != nil {
 		if errors.Is(err, windows.ERROR_FILE_NOT_FOUND) {
@@ -759,13 +757,6 @@ func (c *Collector) collectJobContainer(ch chan<- prometheus.Metric, containerID
 		return fmt.Errorf("error in querying job object memory usage information: %w", err)
 	}
 
-	privateWorkingSetBytes, err := calculatePrivateWorkingSetBytes(jobObjectHandle)
-	if err != nil {
-		c.logger.Debug("error in calculating private working set bytes", slog.Any("err", err))
-	}
-
-	containerInfo := c.annotationsCacheJob[containerID]
-
 	ch <- prometheus.MustNewConstMetric(
 		c.containerAvailable,
 		prometheus.GaugeValue,
@@ -789,18 +780,26 @@ func (c *Collector) collectJobContainer(ch chan<- prometheus.Metric, containerID
 		containerInfo.id, containerInfo.namespace, containerInfo.pod, containerInfo.container,
 	)
 
-	ch <- prometheus.MustNewConstMetric(
-		c.usagePrivateWorkingSetBytes,
-		prometheus.GaugeValue,
-		float64(privateWorkingSetBytes),
+	if privateWorkingSetBytes, err := calculatePrivateWorkingSetBytes(jobObjectHandle); err != nil {
+		c.logger.Debug("error in calculating private working set bytes",
+			slog.String("container_id", containerID),
+			slog.Any("err", err),
+		)
+	} else {
+		ch <- prometheus.MustNewConstMetric(
+			c.usagePrivateWorkingSetBytes,
+			prometheus.GaugeValue,
+			float64(privateWorkingSetBytes),
 
-		containerInfo.id, containerInfo.namespace, containerInfo.pod, containerInfo.container,
-	)
+			containerInfo.id, containerInfo.namespace, containerInfo.pod, containerInfo.container,
+		)
+	}
 
+	// The ThisPeriod* times reset when a job time limit is set, so they are not monotonic.
 	ch <- prometheus.MustNewConstMetric(
 		c.runtimeTotal,
 		prometheus.CounterValue,
-		(float64(jobInfo.BasicInfo.ThisPeriodTotalKernelTime)+float64(jobInfo.BasicInfo.ThisPeriodTotalUserTime))*pdh.TicksToSecondScaleFactor,
+		(float64(jobInfo.BasicInfo.TotalKernelTime)+float64(jobInfo.BasicInfo.TotalUserTime))*pdh.TicksToSecondScaleFactor,
 
 		containerInfo.id, containerInfo.namespace, containerInfo.pod, containerInfo.container,
 	)
@@ -808,7 +807,7 @@ func (c *Collector) collectJobContainer(ch chan<- prometheus.Metric, containerID
 	ch <- prometheus.MustNewConstMetric(
 		c.runtimeUser,
 		prometheus.CounterValue,
-		float64(jobInfo.BasicInfo.ThisPeriodTotalUserTime)*pdh.TicksToSecondScaleFactor,
+		float64(jobInfo.BasicInfo.TotalUserTime)*pdh.TicksToSecondScaleFactor,
 
 		containerInfo.id, containerInfo.namespace, containerInfo.pod, containerInfo.container,
 	)
@@ -816,7 +815,7 @@ func (c *Collector) collectJobContainer(ch chan<- prometheus.Metric, containerID
 	ch <- prometheus.MustNewConstMetric(
 		c.runtimeKernel,
 		prometheus.CounterValue,
-		float64(jobInfo.BasicInfo.ThisPeriodTotalKernelTime)*pdh.TicksToSecondScaleFactor,
+		float64(jobInfo.BasicInfo.TotalKernelTime)*pdh.TicksToSecondScaleFactor,
 
 		containerInfo.id, containerInfo.namespace, containerInfo.pod, containerInfo.container,
 	)
@@ -867,14 +866,16 @@ func getContainerIdWithPrefix(container hcs.Properties) string {
 }
 
 func (c *Collector) getContainerAnnotations(containerID string) (ociSpec, error) {
-	configJSON, err := os.OpenFile(fmt.Sprintf(`%s%s\config.json`, c.config.ContainerDStateDir, containerID), os.O_RDONLY, 0)
+	// Close the file right away: the handle lacks FILE_SHARE_DELETE and would
+	// block containerd from removing the bundle of a stopped container.
+	configJSON, err := os.ReadFile(filepath.Join(c.config.ContainerDStateDir, containerID, "config.json"))
 	if err != nil {
-		return ociSpec{}, fmt.Errorf("error in opening config.json file: %w", err)
+		return ociSpec{}, fmt.Errorf("error in reading config.json file: %w", err)
 	}
 
 	var annotations ociSpec
 
-	if err = json.NewDecoder(configJSON).Decode(&annotations); err != nil {
+	if err = json.Unmarshal(configJSON, &annotations); err != nil {
 		return ociSpec{}, fmt.Errorf("error in decoding config.json file: %w", err)
 	}
 
@@ -882,28 +883,21 @@ func (c *Collector) getContainerAnnotations(containerID string) (ociSpec, error)
 }
 
 func calculatePrivateWorkingSetBytes(jobObjectHandle windows.Handle) (uint64, error) {
-	var pidList kernel32.JobObjectBasicProcessIDList
-
-	retLen := uint32(unsafe.Sizeof(pidList))
-
-	if err := windows.QueryInformationJobObject(
-		jobObjectHandle,
-		windows.JobObjectBasicProcessIdList,
-		uintptr(unsafe.Pointer(&pidList)),
-		retLen, &retLen); err != nil {
-		return 0, err
+	pids, err := kernel32.QueryJobObjectProcessIDs(jobObjectHandle)
+	if err != nil {
+		return 0, fmt.Errorf("error in querying job object process list: %w", err)
 	}
 
-	var (
-		privateWorkingSetBytes uint64
-		vmCounters             kernel32.PROCESS_VM_COUNTERS
-	)
-
-	retLen = uint32(unsafe.Sizeof(vmCounters))
+	var privateWorkingSetBytes uint64
 
 	getMemoryStats := func(pid uint32) (uint64, error) {
 		processHandle, err := windows.OpenProcess(windows.PROCESS_QUERY_LIMITED_INFORMATION, false, pid)
 		if err != nil {
+			// The process exited after the process list was queried.
+			if errors.Is(err, windows.ERROR_INVALID_PARAMETER) {
+				return 0, nil
+			}
+
 			return 0, fmt.Errorf("error in opening process: %w", err)
 		}
 
@@ -911,9 +905,9 @@ func calculatePrivateWorkingSetBytes(jobObjectHandle windows.Handle) (uint64, er
 			_ = windows.Close(fd)
 		}(processHandle)
 
-		var isInJob bool
-
-		if err := kernel32.IsProcessInJob(processHandle, jobObjectHandle, &isInJob); err != nil {
+		// Guard against PID reuse between querying the list and opening the process.
+		isInJob, err := kernel32.IsProcessInJob(processHandle, jobObjectHandle)
+		if err != nil {
 			return 0, fmt.Errorf("error in checking if process is in job: %w", err)
 		}
 
@@ -921,6 +915,10 @@ func calculatePrivateWorkingSetBytes(jobObjectHandle windows.Handle) (uint64, er
 			return 0, nil
 		}
 
+		var vmCounters kernel32.PROCESS_VM_COUNTERS
+
+		retLen := uint32(unsafe.Sizeof(vmCounters))
+
 		if err := windows.NtQueryInformationProcess(
 			processHandle,
 			windows.ProcessVmCounters,
@@ -934,7 +932,7 @@ func calculatePrivateWorkingSetBytes(jobObjectHandle windows.Handle) (uint64, er
 		return uint64(vmCounters.PrivateWorkingSetSize), nil
 	}
 
-	for _, pid := range pidList.PIDs() {
+	for _, pid := range pids {
 		privateWorkingSetSize, err := getMemoryStats(pid)
 		if err != nil {
 			return 0, fmt.Errorf("error in getting private working set bytes: %w", err)

internal/collector/container/container_internal_test.go

@@ -19,27 +19,108 @@ package container
 
 import (
 	"log/slog"
+	"os"
+	"path/filepath"
 	"testing"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/require"
 )
 
+func newTestCollector(stateDir string) *Collector {
+	return &Collector{
+		config:              Config{ContainerDStateDir: stateDir},
+		logger:              slog.New(slog.DiscardHandler),
+		annotationsCacheHCS: map[string]containerInfo{},
+		annotationsCacheJob: map[string]jobContainer{},
+	}
+}
+
+func writeBundle(t *testing.T, stateDir, containerID, config string) {
+	t.Helper()
+
+	bundleDir := filepath.Join(stateDir, containerID)
+
+	require.NoError(t, os.MkdirAll(bundleDir, 0o755))
+
+	if config != "" {
+		require.NoError(t, os.WriteFile(filepath.Join(bundleDir, "config.json"), []byte(config), 0o600))
+	}
+}
+
 // The annotation cache is keyed by the raw container ID, but was pruned by the
 // prefixed ID, so entries of removed containers were never deleted.
 func TestCollectJobContainersPrunesAnnotationsCache(t *testing.T) {
 	t.Parallel()
 
-	c := &Collector{
-		config: Config{ContainerDStateDir: t.TempDir() + `\`},
-		logger: slog.New(slog.DiscardHandler),
-		annotationsCacheJob: map[string]containerInfo{
-			"removed": {id: "containerd://removed"},
-		},
-	}
+	c := newTestCollector(t.TempDir() + `\`)
+	c.annotationsCacheJob["removed"] = jobContainer{info: containerInfo{id: "containerd://removed"}}
 
 	ch := make(chan prometheus.Metric, 100)
 
 	require.NoError(t, c.collectJobContainers(ch))
 	require.Empty(t, c.annotationsCacheJob)
 }
+
+// Every bundle is parsed once and cached, and a bundle without config.json
+// is retried later instead of being cached.
+func TestCollectJobContainersCachesBundles(t *testing.T) {
+	t.Parallel()
+
+	// No trailing separator: the path used to be concatenated without one.
+	stateDir := t.TempDir()
+
+	writeBundle(t, stateDir, "hostprocess", `{"annotations":{
+		"microsoft.com/hostprocess-container":"true",
+		"io.kubernetes.cri.sandbox-namespace":"kube-system",
+		"io.kubernetes.cri.sandbox-name":"pod",
+		"io.kubernetes.cri.container-name":"container"
+	}}`)
+	writeBundle(t, stateDir, "process", `{"annotations":{"io.kubernetes.cri.container-name":"other"}}`)
+	writeBundle(t, stateDir, "creating", "")
+
+	c := newTestCollector(stateDir)
+	ch := make(chan prometheus.Metric, 100)
+
+	// The job objects do not exist, so no metrics are collected.
+	require.NoError(t, c.collectJobContainers(ch))
+	require.Equal(t, map[string]jobContainer{
+		"hostprocess": {
+			info: containerInfo{
+				id:        "containerd://hostprocess",
+				namespace: "kube-system",
+				pod:       "pod",
+				container: "container",
+			},
+			hostProcess: true,
+		},
+		"process": {
+			info: containerInfo{
+				id:        "containerd://process",
+				container: "other",
+			},
+		},
+	}, c.annotationsCacheJob)
+
+	// config.json must be closed, or containerd cannot remove the bundle.
+	require.NoError(t, os.RemoveAll(filepath.Join(stateDir, "process")))
+
+	writeBundle(t, stateDir, "creating", `{"annotations":{}}`)
+
+	require.NoError(t, c.collectJobContainers(ch))
+	require.Contains(t, c.annotationsCacheJob, "creating")
+	require.NotContains(t, c.annotationsCacheJob, "process")
+}
+
+func TestCollectJobContainersMissingStateDir(t *testing.T) {
+	t.Parallel()
+
+	c := newTestCollector(filepath.Join(t.TempDir(), "missing"))
+	c.annotationsCacheJob["removed"] = jobContainer{}
+
+	ch := make(chan prometheus.Metric, 100)
+
+	require.NoError(t, c.collectJobContainers(ch))
+	require.True(t, c.stateDirMissingLogged)
+	require.Empty(t, c.annotationsCacheJob)
+}

internal/headers/kernel32/job.go

@@ -18,6 +18,7 @@
 package kernel32
 
 import (
+	"errors"
 	"unsafe"
 
 	"golang.org/x/sys/windows"
@@ -45,15 +46,60 @@ func OpenJobObject(name string) (windows.Handle, error) {
 	return windows.Handle(handle), nil
 }
 
-func IsProcessInJob(process windows.Handle, job windows.Handle, result *bool) error {
+func IsProcessInJob(process windows.Handle, job windows.Handle) (bool, error) {
+	// IsProcessInJob writes a 4-byte BOOL.
+	var result int32
+
 	ret, _, err := procIsProcessInJob.Call(
 		uintptr(process),
 		uintptr(job),
 		uintptr(unsafe.Pointer(&result)),
 	)
 	if ret == 0 {
-		return err
+		return false, err
 	}
 
-	return nil
+	return result != 0, nil
+}
+
+// QueryJobObjectProcessIDs returns the IDs of all processes assigned to the job.
+func QueryJobObjectProcessIDs(job windows.Handle) ([]uint32, error) {
+	const headerSize = unsafe.Offsetof(JobObjectBasicProcessIDList{}.ProcessIdList)
+
+	headerEntries := int((headerSize + unsafe.Sizeof(uintptr(0)) - 1) / unsafe.Sizeof(uintptr(0)))
+	capacity := 16
+
+	for {
+		// Back the list with []uintptr, so the ULONG_PTR entries are aligned.
+		buf := make([]uintptr, headerEntries+capacity)
+		list := (*JobObjectBasicProcessIDList)(unsafe.Pointer(&buf[0]))
+
+		err := windows.QueryInformationJobObject(
+			job,
+			windows.JobObjectBasicProcessIdList,
+			uintptr(unsafe.Pointer(&buf[0])),
+			uint32(uintptr(len(buf))*unsafe.Sizeof(buf[0])),
+			nil,
+		)
+
+		if errors.Is(err, windows.ERROR_MORE_DATA) {
+			// Processes can join the job between two calls, so leave some headroom.
+			capacity = max(capacity*2, int(list.NumberOfAssignedProcesses)+16)
+
+			continue
+		}
+
+		if err != nil {
+			return nil, err
+		}
+
+		ids := unsafe.Slice(&list.ProcessIdList[0], int(list.NumberOfProcessIdsInList))
+		pids := make([]uint32, len(ids))
+
+		for i, id := range ids {
+			pids[i] = uint32(id)
+		}
+
+		return pids, nil
+	}
 }

internal/headers/kernel32/job_test.go

@@ -0,0 +1,103 @@
+// 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 kernel32_test
+
+import (
+	"os/exec"
+	"testing"
+
+	"github.com/prometheus-community/windows_exporter/internal/headers/kernel32"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sys/windows"
+)
+
+// startJobProcesses starts count idle processes and assigns them to a new job object.
+func startJobProcesses(t *testing.T, count int) (windows.Handle, []uint32) {
+	t.Helper()
+
+	job, err := windows.CreateJobObject(nil, nil)
+	require.NoError(t, err)
+
+	t.Cleanup(func() {
+		_ = windows.TerminateJobObject(job, 0)
+		_ = windows.CloseHandle(job)
+	})
+
+	pids := make([]uint32, 0, count)
+
+	for range count {
+		// cmd.exe blocks on the open stdin pipe until the job is terminated.
+		cmd := exec.CommandContext(t.Context(), "cmd.exe", "/c", "pause")
+
+		stdin, err := cmd.StdinPipe()
+		require.NoError(t, err)
+		require.NoError(t, cmd.Start())
+
+		t.Cleanup(func() {
+			_ = stdin.Close()
+			_ = cmd.Process.Kill()
+			_ = cmd.Wait()
+		})
+
+		process, err := windows.OpenProcess(windows.PROCESS_SET_QUOTA|windows.PROCESS_TERMINATE, false, uint32(cmd.Process.Pid))
+		require.NoError(t, err)
+
+		err = windows.AssignProcessToJobObject(job, process)
+		_ = windows.CloseHandle(process)
+
+		require.NoError(t, err)
+
+		pids = append(pids, uint32(cmd.Process.Pid))
+	}
+
+	return job, pids
+}
+
+// More processes than the initial buffer holds must trigger ERROR_MORE_DATA and a retry.
+func TestQueryJobObjectProcessIDs(t *testing.T) {
+	t.Parallel()
+
+	job, pids := startJobProcesses(t, 20)
+
+	got, err := kernel32.QueryJobObjectProcessIDs(job)
+	require.NoError(t, err)
+	require.ElementsMatch(t, pids, got)
+}
+
+// IsProcessInJob wrote the BOOL into the pointer variable instead of the result,
+// so it always reported false.
+func TestIsProcessInJob(t *testing.T) {
+	t.Parallel()
+
+	job, pids := startJobProcesses(t, 1)
+
+	process, err := windows.OpenProcess(windows.PROCESS_QUERY_LIMITED_INFORMATION, false, pids[0])
+	require.NoError(t, err)
+
+	t.Cleanup(func() {
+		_ = windows.CloseHandle(process)
+	})
+
+	isInJob, err := kernel32.IsProcessInJob(process, job)
+	require.NoError(t, err)
+	require.True(t, isInJob)
+
+	isInJob, err = kernel32.IsProcessInJob(windows.CurrentProcess(), job)
+	require.NoError(t, err)
+	require.False(t, isInJob)
+}

internal/headers/kernel32/types.go

@@ -18,8 +18,6 @@
 package kernel32
 
 import (
-	"unsafe"
-
 	"golang.org/x/sys/windows"
 )
 
@@ -48,17 +46,16 @@ type JobObjectMemoryUsageInformation struct {
 	PeakJobMemoryUsed uint64
 }
 
+// JobObjectBasicProcessIDList is the fixed-size head of JOBOBJECT_BASIC_PROCESS_ID_LIST.
+// The ProcessIdList entries are ULONG_PTR and continue past the declared array;
+// use QueryJobObjectProcessIDs to read them.
+// https://learn.microsoft.com/en-us/windows/win32/api/winnt/ns-winnt-jobobject_basic_process_id_list
 type JobObjectBasicProcessIDList struct {
 	NumberOfAssignedProcesses uint32
 	NumberOfProcessIdsInList  uint32
 	ProcessIdList             [1]uintptr
 }
 
-// PIDs returns all the process Ids in the job object.
-func (p *JobObjectBasicProcessIDList) PIDs() []uint32 {
-	return unsafe.Slice((*uint32)(unsafe.Pointer(&p.ProcessIdList[0])), int(p.NumberOfProcessIdsInList))
-}
-
 type PROCESS_VM_COUNTERS struct {
 	PeakVirtualSize            uintptr
 	VirtualSize                uintptr