diff --git a/docs/collector.container.md b/docs/collector.container.md index 8079bd63..8142e11a 100644 --- a/docs/collector.container.md +++ b/docs/collector.container.md @@ -15,6 +15,7 @@ Enabled by default? | No 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. + Kubernetes containers with Hyper-V isolation are collected from the CRI endpoint, see [Hyper-V isolated containers](#hyper-v-isolated-containers). - `hostprocess` collects Kubernetes HostProcess containers, which run in Win32 job objects instead of HCS. ### `--collector.container.cri-endpoint` @@ -22,7 +23,8 @@ Comma-separated list of collectors to use. Defaults to all, if not specified. Av Kubernetes Container Runtime Interface (CRI) endpoint. Defaults to `npipe:////./pipe/containerd-containerd`, the endpoint of containerd. The collector reads the running containers and pod sandboxes from this endpoint to add the `namespace`, `pod` and `container` labels, to skip pause containers and to find HostProcess containers. -If the endpoint is not available, for example on hosts without Kubernetes, the HCS containers are exported without Kubernetes labels and HostProcess containers aren't collected. +It also reads the container stats for the writable layer usage and the metrics of Hyper-V isolated containers. +If the endpoint is not available, for example on hosts without Kubernetes, the HCS containers are exported without Kubernetes labels and HostProcess and Hyper-V isolated containers aren't collected. ## Metrics @@ -49,11 +51,35 @@ If the endpoint is not available, for example on hosts without Kubernetes, the H | `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` | +| `windows_container_storage_writable_layer_usage_bytes` | Used bytes of the writable layer | gauge | `container_id`,`namespace`,`pod`,`container` | -`windows_container_start_time_seconds` is the start time reported by HCS. HostProcess containers have no start time, they report their creation time from the CRI endpoint instead. +`windows_container_start_time_seconds` is the start time reported by HCS. HostProcess and Hyper-V isolated containers have no start time in HCS on the host, they report their creation time from the CRI endpoint instead. `windows_container_memory_page_faults_total` is only available for HostProcess containers, because HCS doesn't report page faults. The job object counts page faults in 32 bits, so the counter wraps around after 2^32 page faults. +`windows_container_storage_writable_layer_usage_bytes` is read from the CRI endpoint and only exported for Kubernetes containers. +containerd measures it periodically, so it may lag behind and is missing until the first measurement. + +`windows_container_count` only counts HCS containers. HostProcess and Hyper-V isolated containers aren't counted. + +### Hyper-V isolated containers + +Kubernetes containers with Hyper-V isolation run inside a utility VM. HCS on the host only sees the utility VM, not the containers inside it. +The collector reads their metrics from the container stats of the CRI endpoint instead, if the `hcs` collector is enabled. +containerd fills them from the HCS statistics that the shim queries inside the utility VM, but the CRI API only carries a subset of them: + +| Name | Source | +|------------------------------------------------------------|----------------------------------------| +| `windows_container_available` | always `1`, with `hostprocess="false"` | +| `windows_container_cpu_usage_seconds_total` | `cpu.usage_core_nano_seconds` | +| `windows_container_memory_usage_private_working_set_bytes` | `memory.working_set_bytes` | +| `windows_container_memory_usage_commit_bytes` | `memory.usage_bytes` | +| `windows_container_start_time_seconds` | creation time of the container | +| `windows_container_storage_writable_layer_usage_bytes` | `writable_layer.used_bytes` | + +The user and kernel mode CPU time, the commit peak, page faults, the process count, storage I/O and network metrics aren't available for Hyper-V isolated containers. +The collector identifies them as running Kubernetes containers that are neither HCS containers nor HostProcess containers. + ### Example metric _windows_container_network_receive_bytes_total{container_id="docker://1bd30e8b8ac28cbd76a9b697b4d7bb9d760267b0733d1bc55c60024e98d1e43e",interface="822179E7-002C-4280-ABBA-28BCFE401826"} 9.3305343e+07_ diff --git a/internal/collector/container/container.go b/internal/collector/container/container.go index 5e4a489a..56df15a3 100644 --- a/internal/collector/container/container.go +++ b/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": diff --git a/internal/collector/container/container_internal_test.go b/internal/collector/container/container_internal_test.go index fe4acba4..ef6f9f40 100644 --- a/internal/collector/container/container_internal_test.go +++ b/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() diff --git a/internal/collector/container/container_test.go b/internal/collector/container/container_test.go index cdf206bf..8ec3540a 100644 --- a/internal/collector/container/container_test.go +++ b/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"} diff --git a/internal/headers/cri/cri.go b/internal/headers/cri/cri.go index 7002c901..f8135fb7 100644 --- a/internal/headers/cri/cri.go +++ b/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. diff --git a/internal/headers/cri/cri_internal_test.go b/internal/headers/cri/cri_internal_test.go index bea647a7..61107061 100644 --- a/internal/headers/cri/cri_internal_test.go +++ b/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) } } diff --git a/internal/headers/cri/types.go b/internal/headers/cri/types.go index b52e0d86..582492c5 100644 --- a/internal/headers/cri/types.go +++ b/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 {