container: collect Hyper-V isolated containers from CRI stats (#2568)

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

internal/collector/container/container.go

@@ -118,10 +118,11 @@ type Collector struct {
 	droppedPacketsOutgoing *prometheus.Desc
 
 	// Storage
-	readCountNormalized  *prometheus.Desc
-	readSizeBytes        *prometheus.Desc
-	writeCountNormalized *prometheus.Desc
-	writeSizeBytes       *prometheus.Desc
+	readCountNormalized     *prometheus.Desc
+	readSizeBytes           *prometheus.Desc
+	writeCountNormalized    *prometheus.Desc
+	writeSizeBytes          *prometheus.Desc
+	writableLayerUsageBytes *prometheus.Desc
 }
 
 type containerInfo struct {
@@ -390,6 +391,12 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error {
 		[]string{"container_id", "namespace", "pod", "container"},
 		nil,
 	)
+	c.writableLayerUsageBytes = prometheus.NewDesc(
+		prometheus.BuildFQName(types.Namespace, Name, "storage_writable_layer_usage_bytes"),
+		"Used bytes of the writable layer, as reported by the CRI endpoint",
+		[]string{"container_id", "namespace", "pod", "container"},
+		nil,
+	)
 
 	return nil
 }
@@ -415,18 +422,30 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, maxScrapeDuration time.
 		errs = append(errs, err)
 	}
 
+	// hcsContainers and jobContainers stay nil if their collector is disabled.
+	var (
+		hcsContainers map[string]struct{}
+		jobContainers map[string]struct{}
+	)
+
 	if slices.Contains(c.config.CollectorsEnabled, subCollectorHCS) {
-		if err := c.collectHCS(ch, kubernetes); err != nil {
+		hcsContainers, err = c.collectHCS(ch, kubernetes)
+		if err != nil {
 			errs = append(errs, err)
 		}
 	}
 
 	if slices.Contains(c.config.CollectorsEnabled, subCollectorHostprocess) {
-		if err := c.collectJobContainers(ch, kubernetes); err != nil {
+		jobContainers, err = c.collectJobContainers(ch, kubernetes)
+		if err != nil {
 			errs = append(errs, err)
 		}
 	}
 
+	if err := c.collectCRIStats(ctx, ch, kubernetes, hcsContainers, jobContainers); err != nil {
+		errs = append(errs, err)
+	}
+
 	return errors.Join(errs...)
 }
 
@@ -487,11 +506,13 @@ func (c *Collector) handleCRIError(err error) error {
 	return nil
 }
 
-func (c *Collector) collectHCS(ch chan<- prometheus.Metric, kubernetes kubernetesContainers) error {
+// collectHCS collects the metrics of the containers managed by HCS.
+// It returns the IDs of all HCS containers in any state, or nil if HCS could not be queried.
+func (c *Collector) collectHCS(ch chan<- prometheus.Metric, kubernetes kubernetesContainers) (map[string]struct{}, error) {
 	// Types Container is passed to get the containers compute systems only
 	containers, err := hcs.GetContainers()
 	if err != nil {
-		return fmt.Errorf("error in fetching containers: %w", err)
+		return nil, fmt.Errorf("error in fetching containers: %w", err)
 	}
 
 	count := len(containers)
@@ -502,15 +523,19 @@ func (c *Collector) collectHCS(ch chan<- prometheus.Metric, kubernetes kubernete
 			0,
 		)
 
-		return nil
+		return map[string]struct{}{}, nil
 	}
 
 	var countersCount float64
 
+	hcsIDs := make(map[string]struct{}, len(containers))
 	hcsContainers := make(map[string]containerInfo, len(containers))
 	collectErrors := make([]error, 0)
 
 	for _, container := range containers {
+		// Stopping or paused containers are still HCS containers, not Hyper-V isolated ones.
+		hcsIDs[container.ID] = struct{}{}
+
 		if container.State != "Running" {
 			continue
 		}
@@ -562,14 +587,14 @@ func (c *Collector) collectHCS(ch chan<- prometheus.Metric, kubernetes kubernete
 	)
 
 	if err := c.collectNetworkMetrics(ch, hcsContainers, kubernetes.sandboxes); err != nil {
-		return fmt.Errorf("error in fetching container network statistics: %w", err)
+		return hcsIDs, fmt.Errorf("error in fetching container network statistics: %w", err)
 	}
 
 	if len(collectErrors) > 0 {
-		return fmt.Errorf("errors while fetching container statistics: %w", errors.Join(collectErrors...))
+		return hcsIDs, fmt.Errorf("errors while fetching container statistics: %w", errors.Join(collectErrors...))
 	}
 
-	return nil
+	return hcsIDs, nil
 }
 
 func (c *Collector) collectHCSContainer(ch chan<- prometheus.Metric, containerDetails hcs.Properties, containerInfo containerInfo) error {
@@ -784,28 +809,35 @@ func (c *Collector) collectNetworkMetrics(ch chan<- prometheus.Metric, container
 //
 // Job containers are containers that aren't managed by HCS, e.g host process containers.
 // They are found by the job object, which hcsshim names after the container ID.
-func (c *Collector) collectJobContainers(ch chan<- prometheus.Metric, kubernetes kubernetesContainers) error {
+// It returns the IDs of the job containers.
+func (c *Collector) collectJobContainers(ch chan<- prometheus.Metric, kubernetes kubernetesContainers) (map[string]struct{}, error) {
+	jobContainers := make(map[string]struct{})
 	errs := make([]error, 0)
 
 	for containerID, info := range kubernetes.containers {
-		if err := c.collectJobContainer(ch, containerID, info); err != nil {
+		found, err := c.collectJobContainer(ch, containerID, info)
+		if found {
+			jobContainers[containerID] = struct{}{}
+		}
+
+		if err != nil {
 			errs = append(errs, err)
 		}
 	}
 
-	return errors.Join(errs...)
+	return jobContainers, errors.Join(errs...)
 }
 
 // collectJobContainer collects the metrics of a job container.
-// It does nothing if the container has no job object, i.e. it is not a job container.
-func (c *Collector) collectJobContainer(ch chan<- prometheus.Metric, containerID string, containerInfo containerInfo) error {
-	jobObjectHandle, err := kernel32.OpenJobObject("Global\\JobContainer_" + containerID)
+// It does nothing and returns false if the container has no job object, i.e. it is not a job container.
+func (c *Collector) collectJobContainer(ch chan<- prometheus.Metric, containerID string, containerInfo containerInfo) (bool, error) {
+	jobObjectHandle, err := openJobObject(containerID)
 	if err != nil {
 		if errors.Is(err, windows.ERROR_FILE_NOT_FOUND) {
-			return nil
+			return false, nil
 		}
 
-		return fmt.Errorf("error in opening job object: %w", err)
+		return true, fmt.Errorf("error in opening job object: %w", err)
 	}
 
 	defer func(fd windows.Handle) {
@@ -821,7 +853,7 @@ func (c *Collector) collectJobContainer(ch chan<- prometheus.Metric, containerID
 		uint32(unsafe.Sizeof(jobInfo)),
 		nil,
 	); err != nil {
-		return fmt.Errorf("error in querying job object information: %w", err)
+		return true, fmt.Errorf("error in querying job object information: %w", err)
 	}
 
 	var jobMemoryInfo kernel32.JobObjectMemoryUsageInformation
@@ -834,7 +866,7 @@ func (c *Collector) collectJobContainer(ch chan<- prometheus.Metric, containerID
 		uint32(unsafe.Sizeof(jobMemoryInfo)),
 		nil,
 	); err != nil {
-		return fmt.Errorf("error in querying job object memory usage information: %w", err)
+		return true, fmt.Errorf("error in querying job object memory usage information: %w", err)
 	}
 
 	ch <- prometheus.MustNewConstMetric(
@@ -844,16 +876,8 @@ func (c *Collector) collectJobContainer(ch chan<- prometheus.Metric, containerID
 		containerInfo.id, containerInfo.namespace, containerInfo.pod, containerInfo.container, "true",
 	)
 
-	// Job objects have no start time, so the creation time reported by the CRI endpoint is used.
-	if !containerInfo.createdAt.IsZero() {
-		ch <- prometheus.MustNewConstMetric(
-			c.startTime,
-			prometheus.GaugeValue,
-			float64(containerInfo.createdAt.UnixNano())/1e9,
-
-			containerInfo.id, containerInfo.namespace, containerInfo.pod, containerInfo.container,
-		)
-	}
+	// Job objects have no start time.
+	c.collectCreationTime(ch, containerInfo)
 
 	ch <- prometheus.MustNewConstMetric(
 		c.processes,
@@ -959,9 +983,193 @@ func (c *Collector) collectJobContainer(ch chan<- prometheus.Metric, containerID
 		containerInfo.id, containerInfo.namespace, containerInfo.pod, containerInfo.container,
 	)
 
+	return true, nil
+}
+
+// collectCreationTime collects the creation time reported by the CRI endpoint as start time,
+// for containers whose start time HCS doesn't report.
+func (c *Collector) collectCreationTime(ch chan<- prometheus.Metric, containerInfo containerInfo) {
+	if containerInfo.createdAt.IsZero() {
+		return
+	}
+
+	ch <- prometheus.MustNewConstMetric(
+		c.startTime,
+		prometheus.GaugeValue,
+		float64(containerInfo.createdAt.UnixNano())/1e9,
+
+		containerInfo.id, containerInfo.namespace, containerInfo.pod, containerInfo.container,
+	)
+}
+
+// openJobObject opens the job object of a job container, which hcsshim names after the container ID.
+func openJobObject(containerID string) (windows.Handle, error) {
+	return kernel32.OpenJobObject("Global\\JobContainer_" + containerID)
+}
+
+// isJobContainer reports whether a container has a job object.
+// Errors other than a missing job object are treated as job container.
+func isJobContainer(containerID string) bool {
+	jobObjectHandle, err := openJobObject(containerID)
+	if err != nil {
+		return !errors.Is(err, windows.ERROR_FILE_NOT_FOUND)
+	}
+
+	_ = windows.Close(jobObjectHandle)
+
+	return true
+}
+
+// collectCRIStats collects the container stats of the CRI endpoint for the running Kubernetes containers:
+//   - the writable layer usage of all containers exported by the enabled collectors.
+//   - CPU and memory usage of Hyper-V isolated containers. They run inside a utility VM,
+//     so they are neither visible to HCS on the host nor job containers.
+//     They belong to the hcs collector, which provides the list of HCS containers to exclude.
+//
+// The CRI stats lack the user and kernel mode CPU time, the commit peak, storage I/O and network statistics.
+func (c *Collector) collectCRIStats(
+	ctx context.Context,
+	ch chan<- prometheus.Metric,
+	kubernetes kubernetesContainers,
+	hcsContainers map[string]struct{},
+	jobContainers map[string]struct{},
+) error {
+	exported, hypervContainers := selectCRIStatsContainers(kubernetes, hcsContainers, jobContainers)
+	if len(exported) == 0 {
+		return nil
+	}
+
+	stats, err := c.criClient.ListContainerStats(ctx)
+	if err != nil {
+		err = fmt.Errorf("error in fetching container stats from CRI endpoint %s: %w", c.config.CRIEndpoint, err)
+
+		// Without Hyper-V isolated containers, only the writable layer usage is missing.
+		if len(hypervContainers) == 0 {
+			c.logger.WarnContext(ctx, "writable layer usage is not collected", slog.Any("err", err))
+
+			return nil
+		}
+
+		return err
+	}
+
+	c.collectCRIContainers(ch, stats, kubernetes, exported, hypervContainers)
+
 	return nil
 }
 
+// selectCRIStatsContainers returns the running Kubernetes containers exported by the enabled collectors,
+// and among them the Hyper-V isolated containers, which are neither HCS nor job containers.
+// hcsContainers and jobContainers are nil if their collector is disabled.
+func selectCRIStatsContainers(
+	kubernetes kubernetesContainers,
+	hcsContainers map[string]struct{},
+	jobContainers map[string]struct{},
+) (map[string]struct{}, map[string]struct{}) {
+	exported := make(map[string]struct{}, len(kubernetes.containers))
+	hypervContainers := make(map[string]struct{})
+
+	for containerID := range kubernetes.containers {
+		_, isHCS := hcsContainers[containerID]
+		_, isJob := jobContainers[containerID]
+
+		switch {
+		case isHCS || isJob:
+			exported[containerID] = struct{}{}
+		case hcsContainers == nil:
+			// Without the HCS containers, Hyper-V isolated containers are unknown.
+		case jobContainers == nil && isJobContainer(containerID):
+			// A job container while the hostprocess collector is disabled.
+		default:
+			exported[containerID] = struct{}{}
+			hypervContainers[containerID] = struct{}{}
+		}
+	}
+
+	return exported, hypervContainers
+}
+
+// collectCRIContainers collects the CRI stats of the exported containers.
+func (c *Collector) collectCRIContainers(
+	ch chan<- prometheus.Metric,
+	stats []cri.ContainerStats,
+	kubernetes kubernetesContainers,
+	exported map[string]struct{},
+	hypervContainers map[string]struct{},
+) {
+	for _, stat := range stats {
+		if _, ok := exported[stat.ID]; !ok {
+			continue
+		}
+
+		containerInfo := kubernetes.containers[stat.ID]
+
+		// Containers without task metrics aren't running, e.g. they exited after they were listed.
+		if _, ok := hypervContainers[stat.ID]; ok && (stat.CPU != nil || stat.Memory != nil) {
+			c.collectCRIContainer(ch, stat, containerInfo)
+		}
+
+		// containerd reports 0 bytes without timestamp until it has measured the writable layer.
+		if stat.WritableLayer != nil && stat.WritableLayer.UsedBytes != nil && stat.WritableLayer.Timestamp != 0 {
+			ch <- prometheus.MustNewConstMetric(
+				c.writableLayerUsageBytes,
+				prometheus.GaugeValue,
+				float64(*stat.WritableLayer.UsedBytes),
+
+				containerInfo.id, containerInfo.namespace, containerInfo.pod, containerInfo.container,
+			)
+		}
+	}
+}
+
+// collectCRIContainer collects the CPU and memory usage of a container from its CRI stats.
+// containerd reports the private working set as working set and the commit size as usage.
+func (c *Collector) collectCRIContainer(ch chan<- prometheus.Metric, stat cri.ContainerStats, containerInfo containerInfo) {
+	ch <- prometheus.MustNewConstMetric(
+		c.containerAvailable,
+		prometheus.GaugeValue,
+		1,
+		containerInfo.id, containerInfo.namespace, containerInfo.pod, containerInfo.container, "false",
+	)
+
+	// HCS on the host doesn't report the start time.
+	c.collectCreationTime(ch, containerInfo)
+
+	if stat.CPU != nil && stat.CPU.UsageCoreNanoSeconds != nil {
+		ch <- prometheus.MustNewConstMetric(
+			c.runtimeTotal,
+			prometheus.CounterValue,
+			float64(*stat.CPU.UsageCoreNanoSeconds)/1e9,
+
+			containerInfo.id, containerInfo.namespace, containerInfo.pod, containerInfo.container,
+		)
+	}
+
+	if stat.Memory == nil {
+		return
+	}
+
+	if stat.Memory.WorkingSetBytes != nil {
+		ch <- prometheus.MustNewConstMetric(
+			c.usagePrivateWorkingSetBytes,
+			prometheus.GaugeValue,
+			float64(*stat.Memory.WorkingSetBytes),
+
+			containerInfo.id, containerInfo.namespace, containerInfo.pod, containerInfo.container,
+		)
+	}
+
+	if stat.Memory.UsageBytes != nil {
+		ch <- prometheus.MustNewConstMetric(
+			c.usageCommitBytes,
+			prometheus.GaugeValue,
+			float64(*stat.Memory.UsageBytes),
+
+			containerInfo.id, containerInfo.namespace, containerInfo.pod, containerInfo.container,
+		)
+	}
+}
+
 func getContainerIdWithPrefix(container hcs.Properties) string {
 	switch container.Owner {
 	case "containerd-shim-runhcs-v1.exe":

internal/collector/container/container_internal_test.go

@@ -21,11 +21,14 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"maps"
+	"slices"
 	"testing"
 	"time"
 
 	"github.com/prometheus-community/windows_exporter/internal/headers/cri"
 	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sys/windows"
 )
@@ -132,14 +135,165 @@ func TestCollectJobContainersSkipsHCSContainers(t *testing.T) {
 	c := &Collector{logger: slog.New(slog.DiscardHandler)}
 	ch := make(chan prometheus.Metric, 100)
 
-	require.NoError(t, c.collectJobContainers(ch, kubernetesContainers{
+	jobContainers, err := c.collectJobContainers(ch, kubernetesContainers{
 		containers: map[string]containerInfo{
 			"windows-exporter-test-no-job-object": {id: "containerd://windows-exporter-test-no-job-object"},
 		},
-	}))
+	})
+	require.NoError(t, err)
+	require.Empty(t, jobContainers)
 	require.Empty(t, ch)
 }
 
+func TestSelectCRIStatsContainers(t *testing.T) {
+	t.Parallel()
+
+	kubernetes := kubernetesContainers{
+		containers: map[string]containerInfo{
+			"process-isolated": {},
+			"hostprocess":      {},
+			"hyperv":           {},
+		},
+	}
+	hcsContainers := map[string]struct{}{"process-isolated": {}, "docker": {}}
+	jobContainers := map[string]struct{}{"hostprocess": {}}
+
+	for name, tc := range map[string]struct {
+		hcsContainers map[string]struct{}
+		jobContainers map[string]struct{}
+		exported      []string
+		hyperv        []string
+	}{
+		"all collectors": {
+			hcsContainers: hcsContainers,
+			jobContainers: jobContainers,
+			exported:      []string{"process-isolated", "hostprocess", "hyperv"},
+			hyperv:        []string{"hyperv"},
+		},
+		// Hyper-V isolated containers belong to the hcs collector.
+		"hostprocess only": {
+			jobContainers: jobContainers,
+			exported:      []string{"hostprocess"},
+		},
+		// The test containers have no job object, so they are taken as Hyper-V isolated.
+		"hcs only": {
+			hcsContainers: hcsContainers,
+			exported:      []string{"process-isolated", "hostprocess", "hyperv"},
+			hyperv:        []string{"hostprocess", "hyperv"},
+		},
+		// Without HCS, Hyper-V isolated containers can't be told apart from process-isolated ones.
+		"HCS not available": {
+			jobContainers: jobContainers,
+			exported:      []string{"hostprocess"},
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			exported, hyperv := selectCRIStatsContainers(kubernetes, tc.hcsContainers, tc.jobContainers)
+			require.ElementsMatch(t, tc.exported, slices.Collect(maps.Keys(exported)))
+			require.ElementsMatch(t, tc.hyperv, slices.Collect(maps.Keys(hyperv)))
+		})
+	}
+}
+
+func TestCollectCRIContainers(t *testing.T) {
+	t.Parallel()
+
+	c := New(&Config{
+		CollectorsEnabled: []string{subCollectorHostprocess},
+		CRIEndpoint:       "npipe:////./pipe/windows_exporter-container-test-does-not-exist",
+	})
+	require.NoError(t, c.Build(slog.New(slog.DiscardHandler), nil))
+
+	t.Cleanup(func() { _ = c.Close() })
+
+	ptr := func(v uint64) *uint64 { return &v }
+
+	kubernetes := kubernetesContainers{
+		containers: map[string]containerInfo{
+			"hyperv":           {id: "containerd://hyperv", namespace: "default", pod: "pod", container: "hyperv", createdAt: time.Unix(1700000000, 0)},
+			"process-isolated": {id: "containerd://process-isolated", namespace: "default", pod: "pod", container: "app"},
+			"no-stats":         {id: "containerd://no-stats"},
+		},
+	}
+
+	stats := []cri.ContainerStats{
+		{
+			ID:            "hyperv",
+			CPU:           &cri.CPUUsage{Timestamp: 1, UsageCoreNanoSeconds: ptr(1500000000)},
+			Memory:        &cri.MemoryUsage{Timestamp: 1, WorkingSetBytes: ptr(4096), UsageBytes: ptr(8192)},
+			WritableLayer: &cri.FilesystemUsage{Timestamp: 1, UsedBytes: ptr(1024)},
+		},
+		{
+			// HCS provides the CPU and memory usage of process-isolated containers.
+			ID:            "process-isolated",
+			CPU:           &cri.CPUUsage{Timestamp: 1, UsageCoreNanoSeconds: ptr(1)},
+			WritableLayer: &cri.FilesystemUsage{Timestamp: 1, UsedBytes: ptr(2048)},
+		},
+		{
+			// A container without task metrics, e.g. exited after listing, whose writable layer containerd
+			// has not measured yet. Nothing is exported for it.
+			ID:            "no-stats",
+			WritableLayer: &cri.FilesystemUsage{UsedBytes: ptr(0)},
+		},
+		{
+			// Containers not exported, e.g. pause containers, are skipped.
+			ID:            "sandbox",
+			WritableLayer: &cri.FilesystemUsage{Timestamp: 1, UsedBytes: ptr(1)},
+		},
+	}
+
+	ch := make(chan prometheus.Metric, 100)
+
+	c.collectCRIContainers(ch, stats, kubernetes,
+		map[string]struct{}{"hyperv": {}, "process-isolated": {}, "no-stats": {}},
+		map[string]struct{}{"hyperv": {}, "no-stats": {}},
+	)
+	close(ch)
+
+	type sample struct {
+		desc        *prometheus.Desc
+		containerID string
+		value       float64
+	}
+
+	var samples []sample
+
+	for m := range ch {
+		var metric dto.Metric
+
+		require.NoError(t, m.Write(&metric))
+
+		s := sample{desc: m.Desc()}
+
+		for _, label := range metric.GetLabel() {
+			if label.GetName() == "container_id" {
+				s.containerID = label.GetValue()
+			}
+		}
+
+		switch {
+		case metric.GetGauge() != nil:
+			s.value = metric.GetGauge().GetValue()
+		case metric.GetCounter() != nil:
+			s.value = metric.GetCounter().GetValue()
+		}
+
+		samples = append(samples, s)
+	}
+
+	require.ElementsMatch(t, []sample{
+		{c.containerAvailable, "containerd://hyperv", 1},
+		{c.startTime, "containerd://hyperv", 1700000000},
+		{c.runtimeTotal, "containerd://hyperv", 1.5},
+		{c.usagePrivateWorkingSetBytes, "containerd://hyperv", 4096},
+		{c.usageCommitBytes, "containerd://hyperv", 8192},
+		{c.writableLayerUsageBytes, "containerd://hyperv", 1024},
+		{c.writableLayerUsageBytes, "containerd://process-isolated", 2048},
+	}, samples)
+}
+
 func TestBuildInvalidCRIEndpoint(t *testing.T) {
 	t.Parallel()
 

internal/collector/container/container_test.go

@@ -35,6 +35,7 @@ func TestCollector(t *testing.T) {
 	metrics := testutils.TestCollector(t, container.New, nil)
 	testutils.RequireFixtureMetric(t, metrics, container.Name, "windows_container_available", prometheus.Labels{"container": "hostprocess", "namespace": "default", "hostprocess": "true"})
 	testutils.RequireFixtureMetric(t, metrics, container.Name, "windows_container_available", prometheus.Labels{"container": "nanoserver", "namespace": "default", "hostprocess": "false"})
+	testutils.RequireFixtureMetric(t, metrics, container.Name, "windows_container_storage_writable_layer_usage_bytes", prometheus.Labels{"container": "nanoserver", "namespace": "default"})
 
 	for _, name := range []string{"hostprocess", "nanoserver"} {
 		labels := prometheus.Labels{"container": name, "namespace": "default"}

internal/headers/cri/cri.go

@@ -121,6 +121,16 @@ func (c *Client) ListPodSandboxes(ctx context.Context) ([]PodSandbox, error) {
 	return decodeListPodSandboxResponse(resp)
 }
 
+// ListContainerStats returns the stats of all containers.
+func (c *Client) ListContainerStats(ctx context.Context) ([]ContainerStats, error) {
+	resp, err := c.call(ctx, "ListContainerStats", nil)
+	if err != nil {
+		return nil, err
+	}
+
+	return decodeListContainerStatsResponse(resp)
+}
+
 // call performs a unary gRPC call and returns the response message.
 func (c *Client) call(ctx context.Context, method string, msg []byte) ([]byte, error) {
 	// A gRPC message is prefixed by a compression flag and its length.

internal/headers/cri/cri_internal_test.go

@@ -95,6 +95,63 @@ func sandboxesResponse() message {
 			varint(3, uint64(SandboxNotReady)))
 }
 
+func uint64Value(v uint64) message {
+	return message{}.varint(1, v)
+}
+
+// A ListContainerStatsResponse as containerd sends it on Windows, including fields the client skips.
+func containerStatsResponse() message {
+	attributes := message{}.
+		string(1, "abc").
+		message(2, message{}.string(1, "nanoserver"))
+	attributes = append(attributes, labels(3, "io.kubernetes.pod.name", "pod")...)
+
+	running := message{}.
+		message(1, attributes).
+		message(2, message{}.
+			varint(1, 1700000000000000000).
+			message(2, uint64Value(1500000000)).
+			message(3, uint64Value(20000000))).
+		message(3, message{}.
+			varint(1, 1700000000000000000).
+			message(2, uint64Value(4096)).
+			message(4, uint64Value(8192))).
+		message(4, message{}.
+			varint(1, 1700000000000000001).
+			message(2, message{}.string(1, `C:\ProgramData\containerd\root`)).
+			message(3, uint64Value(1024)).
+			message(4, uint64Value(0)))
+
+	// A container without task metrics, e.g. a created container, only has attributes and a
+	// writable layer which containerd has not measured yet. The value 0 is sent as empty message.
+	created := message{}.
+		message(1, message{}.string(1, "def")).
+		message(4, message{}.message(3, message{}))
+
+	return message{}.message(1, running).message(1, created)
+}
+
+func TestDecodeListContainerStatsResponse(t *testing.T) {
+	t.Parallel()
+
+	ptr := func(v uint64) *uint64 { return &v }
+
+	stats, err := decodeListContainerStatsResponse(containerStatsResponse())
+	require.NoError(t, err)
+	require.Equal(t, []ContainerStats{
+		{
+			ID:            "abc",
+			CPU:           &CPUUsage{Timestamp: 1700000000000000000, UsageCoreNanoSeconds: ptr(1500000000)},
+			Memory:        &MemoryUsage{Timestamp: 1700000000000000000, WorkingSetBytes: ptr(4096), UsageBytes: ptr(8192)},
+			WritableLayer: &FilesystemUsage{Timestamp: 1700000000000000001, UsedBytes: ptr(1024)},
+		},
+		{
+			ID:            "def",
+			WritableLayer: &FilesystemUsage{UsedBytes: ptr(0)},
+		},
+	}, stats)
+}
+
 func TestDecodeListContainersResponse(t *testing.T) {
 	t.Parallel()
 
@@ -133,6 +190,10 @@ func TestDecodeInvalidMessage(t *testing.T) {
 	// A nested message with an invalid length.
 	_, err = decodeListPodSandboxResponse(message{}.message(1, message{0x12, 0x7f}))
 	require.Error(t, err)
+
+	// A UInt64Value with an invalid length.
+	_, err = decodeListContainerStatsResponse(message{}.message(1, message{}.message(2, message{}.message(2, message{0x08}))))
+	require.Error(t, err)
 }
 
 func TestEncodeListRunningContainersRequest(t *testing.T) {
@@ -200,6 +261,7 @@ func TestClient(t *testing.T) {
 		message{}.string(1, "0.1.0").string(2, "containerd").string(3, "v2.1.4").string(4, "v1")))
 	mux.Handle(runtimeService+"ListContainers", grpcHandler(t, encodeListRunningContainersRequest(), containersResponse()))
 	mux.Handle(runtimeService+"ListPodSandbox", grpcHandler(t, nil, sandboxesResponse()))
+	mux.Handle(runtimeService+"ListContainerStats", grpcHandler(t, nil, containerStatsResponse()))
 
 	client, err := NewClient(startServer(t, mux))
 	require.NoError(t, err)
@@ -221,6 +283,13 @@ func TestClient(t *testing.T) {
 		require.NoError(t, err)
 		require.Len(t, sandboxes, 2)
 		require.Equal(t, "default", sandboxes[0].Namespace)
+
+		stats, err := client.ListContainerStats(t.Context())
+		require.NoError(t, err)
+		require.Len(t, stats, 2)
+		require.Equal(t, "abc", stats[0].ID)
+		require.NotNil(t, stats[0].CPU)
+		require.Equal(t, uint64(1500000000), *stats[0].CPU.UsageCoreNanoSeconds)
 	}
 }
 

internal/headers/cri/types.go

@@ -69,6 +69,45 @@ type PodSandbox struct {
 	State     PodSandboxState
 }
 
+// ContainerStats is the subset of runtime.v1.ContainerStats used by the exporter.
+// Nil fields were not reported by the runtime.
+//
+// On Windows, containerd fills them from the HCS statistics of the container,
+// which the shim also reads for Hyper-V isolated containers.
+// https://github.com/containerd/containerd/blob/v2.4.1/internal/cri/server/container_stats_list.go
+type ContainerStats struct {
+	ID            string
+	CPU           *CPUUsage
+	Memory        *MemoryUsage
+	WritableLayer *FilesystemUsage
+}
+
+// CPUUsage is the subset of runtime.v1.CpuUsage used by the exporter.
+type CPUUsage struct {
+	// Timestamp is the time of the sample in nanoseconds since the Unix epoch.
+	Timestamp int64
+	// UsageCoreNanoSeconds is the cumulative CPU time of the container.
+	UsageCoreNanoSeconds *uint64
+}
+
+// MemoryUsage is the subset of runtime.v1.MemoryUsage used by the exporter.
+type MemoryUsage struct {
+	// Timestamp is the time of the sample in nanoseconds since the Unix epoch.
+	Timestamp int64
+	// WorkingSetBytes is the private working set on Windows.
+	WorkingSetBytes *uint64
+	// UsageBytes is the commit size on Windows.
+	UsageBytes *uint64
+}
+
+// FilesystemUsage is the subset of runtime.v1.FilesystemUsage used by the exporter.
+type FilesystemUsage struct {
+	// Timestamp is the time of the sample in nanoseconds since the Unix epoch.
+	// containerd sends 0 if it has not measured the usage yet.
+	Timestamp int64
+	UsedBytes *uint64
+}
+
 // field is a decoded protobuf field. value holds varints, bytes length-delimited values.
 type field struct {
 	num   protowire.Number
@@ -237,6 +276,112 @@ func decodePodSandbox(b []byte) (PodSandbox, error) {
 	return sandbox, err
 }
 
+func decodeListContainerStatsResponse(b []byte) ([]ContainerStats, error) {
+	var stats []ContainerStats
+
+	err := eachField(b, func(f field) error {
+		if !f.is(1, protowire.BytesType) {
+			return nil
+		}
+
+		s, err := decodeContainerStats(f.bytes)
+		if err != nil {
+			return err
+		}
+
+		stats = append(stats, s)
+
+		return nil
+	})
+
+	return stats, err
+}
+
+func decodeContainerStats(b []byte) (ContainerStats, error) {
+	var stats ContainerStats
+
+	err := eachField(b, func(f field) error {
+		var err error
+
+		switch {
+		case f.is(1, protowire.BytesType): // ContainerAttributes
+			err = eachField(f.bytes, func(f field) error {
+				if f.is(1, protowire.BytesType) {
+					stats.ID = string(f.bytes)
+				}
+
+				return nil
+			})
+		case f.is(2, protowire.BytesType):
+			stats.CPU = &CPUUsage{}
+			err = eachField(f.bytes, func(f field) error {
+				var err error
+
+				switch {
+				case f.is(1, protowire.VarintType):
+					stats.CPU.Timestamp = int64(f.value)
+				case f.is(2, protowire.BytesType):
+					stats.CPU.UsageCoreNanoSeconds, err = decodeUInt64Value(f.bytes)
+				}
+
+				return err
+			})
+		case f.is(3, protowire.BytesType):
+			stats.Memory = &MemoryUsage{}
+			err = eachField(f.bytes, func(f field) error {
+				var err error
+
+				switch {
+				case f.is(1, protowire.VarintType):
+					stats.Memory.Timestamp = int64(f.value)
+				case f.is(2, protowire.BytesType):
+					stats.Memory.WorkingSetBytes, err = decodeUInt64Value(f.bytes)
+				case f.is(4, protowire.BytesType):
+					stats.Memory.UsageBytes, err = decodeUInt64Value(f.bytes)
+				}
+
+				return err
+			})
+		case f.is(4, protowire.BytesType):
+			stats.WritableLayer = &FilesystemUsage{}
+			err = eachField(f.bytes, func(f field) error {
+				var err error
+
+				switch {
+				case f.is(1, protowire.VarintType):
+					stats.WritableLayer.Timestamp = int64(f.value)
+				case f.is(3, protowire.BytesType):
+					stats.WritableLayer.UsedBytes, err = decodeUInt64Value(f.bytes)
+				}
+
+				return err
+			})
+		}
+
+		return err
+	})
+
+	return stats, err
+}
+
+// decodeUInt64Value decodes a runtime.v1.UInt64Value. An empty message is the value 0.
+func decodeUInt64Value(b []byte) (*uint64, error) {
+	var value uint64
+
+	err := eachField(b, func(f field) error {
+		if f.is(1, protowire.VarintType) {
+			value = f.value
+		}
+
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &value, nil
+}
+
 // encodeListRunningContainersRequest encodes a ListContainersRequest
 // with filter.state.state set to CONTAINER_RUNNING.
 func encodeListRunningContainersRequest() []byte {