From 5c148173e7588c2bddc559e4f85f5587a5e5a9e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan-Otto=20Kr=C3=B6pke?= Date: Thu, 8 Oct 2026 20:16:03 +0200 Subject: [PATCH] container: fix hostprocess memory metrics and containerd bundle handling (#2555) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Jan-Otto Kröpke Co-authored-by: Claude Opus 5.5 --- docs/collector.container.md | 57 ++-- internal/collector/container/container.go | 258 +++++++++--------- .../container/container_internal_test.go | 95 ++++++- internal/headers/kernel32/job.go | 52 +++- internal/headers/kernel32/job_test.go | 103 +++++++ internal/headers/kernel32/types.go | 11 +- 6 files changed, 405 insertions(+), 171 deletions(-) create mode 100644 internal/headers/kernel32/job_test.go diff --git a/docs/collector.container.md b/docs/collector.container.md index ae974303..7c2ccd9c 100644 --- a/docs/collector.container.md +++ b/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 diff --git a/internal/collector/container/container.go b/internal/collector/container/container.go index 7ae5e41f..8f9bfb34 100644 --- a/internal/collector/container/container.go +++ b/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) diff --git a/internal/collector/container/container_internal_test.go b/internal/collector/container/container_internal_test.go index cc42286a..f25d6dc6 100644 --- a/internal/collector/container/container_internal_test.go +++ b/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) +} diff --git a/internal/headers/kernel32/job.go b/internal/headers/kernel32/job.go index a924c24f..32a45e73 100644 --- a/internal/headers/kernel32/job.go +++ b/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 + } } diff --git a/internal/headers/kernel32/job_test.go b/internal/headers/kernel32/job_test.go new file mode 100644 index 00000000..c856396f --- /dev/null +++ b/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) +} diff --git a/internal/headers/kernel32/types.go b/internal/headers/kernel32/types.go index 8c52f3e5..cc76a882 100644 --- a/internal/headers/kernel32/types.go +++ b/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