From 864088b03462755f58fd98696f741c7479f1ac9c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan-Otto=20Kr=C3=B6pke?= Date: Thu, 8 Oct 2026 22:24:54 +0200 Subject: [PATCH] container: read Kubernetes metadata from the CRI API (#2557) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Claude Opus 5.5 Signed-off-by: Jan-Otto Kröpke --- .github/linters/.codespellrc | 2 +- docs/collector.container.md | 9 +- go.mod | 2 +- internal/collector/container/container.go | 346 +++++++-------- .../container/container_internal_test.go | 183 ++++---- .../collector/container/container_test.go | 4 +- internal/headers/cri/cri.go | 204 +++++++++ internal/headers/cri/cri_internal_test.go | 312 ++++++++++++++ internal/headers/cri/pipe.go | 266 ++++++++++++ internal/headers/cri/pipe_internal_test.go | 396 ++++++++++++++++++ internal/headers/cri/types.go | 251 +++++++++++ 11 files changed, 1721 insertions(+), 254 deletions(-) create mode 100644 internal/headers/cri/cri.go create mode 100644 internal/headers/cri/cri_internal_test.go create mode 100644 internal/headers/cri/pipe.go create mode 100644 internal/headers/cri/pipe_internal_test.go create mode 100644 internal/headers/cri/types.go diff --git a/.github/linters/.codespellrc b/.github/linters/.codespellrc index 840eca34..bdc6e912 100644 --- a/.github/linters/.codespellrc +++ b/.github/linters/.codespellrc @@ -1,2 +1,2 @@ [codespell] -ignore-words-list = notin,entires +ignore-words-list = notin,entires,te diff --git a/docs/collector.container.md b/docs/collector.container.md index 7c2ccd9c..37e58e6d 100644 --- a/docs/collector.container.md +++ b/docs/collector.container.md @@ -5,7 +5,7 @@ The container collector exposes metrics about containers running on a Hyper-V sy ||| -|- Metric name prefix | `container` -Data source | [HCS](https://learn.microsoft.com/en-us/virtualization/api/hcs/overview) +Data source | [HCS](https://learn.microsoft.com/en-us/virtualization/api/hcs/overview), [Kubernetes CRI](https://kubernetes.io/docs/concepts/architecture/cri/) Enabled by default? | No ## Flags @@ -17,9 +17,12 @@ Comma-separated list of collectors to use. Defaults to all, if not specified. Av - `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` +### `--collector.container.cri-endpoint` -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\`. +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. ## Metrics diff --git a/go.mod b/go.mod index 9ee72b6f..bac7d9ba 100644 --- a/go.mod +++ b/go.mod @@ -12,6 +12,7 @@ require ( github.com/stretchr/testify v1.12.1 go.yaml.in/yaml/v3 v3.0.5 golang.org/x/sys v0.48.0 + google.golang.org/protobuf v1.36.12 ) require ( @@ -35,5 +36,4 @@ require ( golang.org/x/sync v0.23.0 // indirect golang.org/x/text v0.42.0 // indirect golang.org/x/time v0.16.0 // indirect - google.golang.org/protobuf v1.36.12 // indirect ) diff --git a/internal/collector/container/container.go b/internal/collector/container/container.go index 8f9bfb34..f3eeabcf 100644 --- a/internal/collector/container/container.go +++ b/internal/collector/container/container.go @@ -18,13 +18,10 @@ package container import ( - "encoding/json" + "context" "errors" "fmt" - "io/fs" "log/slog" - "os" - "path/filepath" "slices" "strings" "sync" @@ -32,6 +29,7 @@ import ( "unsafe" "github.com/alecthomas/kingpin/v2" + "github.com/prometheus-community/windows_exporter/internal/headers/cri" "github.com/prometheus-community/windows_exporter/internal/headers/hcn" "github.com/prometheus-community/windows_exporter/internal/headers/hcs" "github.com/prometheus-community/windows_exporter/internal/headers/kernel32" @@ -49,11 +47,19 @@ const ( subCollectorHostprocess = "hostprocess" JobObjectMemoryUsageInformation = 28 + + // defaultCRITimeout limits the CRI calls if the scrape has no timeout. + defaultCRITimeout = 10 * time.Second + + // grpcUnimplemented is returned by containerd if its CRI plugin is disabled. + grpcUnimplemented = 12 ) type Config struct { - CollectorsEnabled []string `yaml:"enabled"` - ContainerDStateDir string `yaml:"containerd-state-dir"` + CollectorsEnabled []string `yaml:"enabled"` + // CRIEndpoint is the Kubernetes Container Runtime Interface (CRI) endpoint. + // It provides the Kubernetes metadata and the HostProcess containers. + CRIEndpoint string `yaml:"cri-endpoint"` } //nolint:gochecknoglobals @@ -62,7 +68,7 @@ var ConfigDefaults = Config{ subCollectorHCS, subCollectorHostprocess, }, - ContainerDStateDir: `C:\ProgramData\containerd\state\io.containerd.runtime.v2.task\k8s.io\`, + CRIEndpoint: "npipe:////./pipe/containerd-containerd", } // A Collector is a Prometheus Collector for containers metrics. @@ -71,17 +77,16 @@ type Collector struct { logger *slog.Logger - // mu serializes Collect, because the annotation caches are not safe for concurrent use. - // Overlapping Collect calls, e.g. after a scrape timeout, would otherwise write them concurrently. + criClient *cri.Client + + // mu serializes Collect, because the CRI state below is not safe for concurrent use. + // Overlapping Collect calls, e.g. after a scrape timeout, would otherwise write it concurrently. mu sync.Mutex - // The annotation caches are keyed by the container ID without the runtime prefix. - annotationsCacheHCS 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 + // criRuntimeName is the runtime name, e.g. containerd. Kubernetes uses it as container ID prefix. + criRuntimeName string + // criUnavailableLogged avoids logging an unavailable CRI endpoint on every scrape. + criUnavailableLogged bool // Presence containerAvailable *prometheus.Desc @@ -121,23 +126,43 @@ type containerInfo struct { container string } -type jobContainer struct { - info containerInfo - hostProcess bool +// kubernetesContainers holds the Kubernetes metadata of the running containers, read from the CRI endpoint. +type kubernetesContainers struct { + // containers maps the container IDs without runtime prefix to their metadata. + containers map[string]containerInfo + // sandboxes holds the IDs of all pod sandboxes. Their pause containers are not exported. + sandboxes map[string]struct{} } -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"], +func newKubernetesContainers(runtimeName string, sandboxes []cri.PodSandbox, containers []cri.Container) kubernetesContainers { + k := kubernetesContainers{ + containers: make(map[string]containerInfo, len(containers)), + sandboxes: make(map[string]struct{}, len(sandboxes)), } + + pods := make(map[string]cri.PodSandbox, len(sandboxes)) + + for _, sandbox := range sandboxes { + k.sandboxes[sandbox.ID] = struct{}{} + pods[sandbox.ID] = sandbox + } + + for _, container := range containers { + if container.State != cri.ContainerRunning { + continue + } + + pod := pods[container.PodSandboxID] + + k.containers[container.ID] = containerInfo{ + id: runtimeName + "://" + container.ID, + namespace: pod.Namespace, + pod: pod.Name, + container: container.Name, + } + } + + return k } // New constructs a new Collector. @@ -150,6 +175,10 @@ func New(config *Config) *Collector { config.CollectorsEnabled = ConfigDefaults.CollectorsEnabled } + if config.CRIEndpoint == "" { + config.CRIEndpoint = ConfigDefaults.CRIEndpoint + } + c := &Collector{ config: *config, } @@ -171,9 +200,9 @@ func NewWithFlags(app *kingpin.Application) *Collector { ).Default(strings.Join(ConfigDefaults.CollectorsEnabled, ",")).StringVar(&collectorsEnabled) app.Flag( - "collector.container.containerd-state-dir", - "Path to the containerd state directory. Defaults to C:\\ProgramData\\containerd\\state\\io.containerd.runtime.v2.task\\k8s.io\\", - ).Default(ConfigDefaults.ContainerDStateDir).StringVar(&c.config.ContainerDStateDir) + "collector.container.cri-endpoint", + "Kubernetes CRI endpoint, used for Kubernetes labels and HostProcess containers.", + ).Default(ConfigDefaults.CRIEndpoint).StringVar(&c.config.CRIEndpoint) app.Action(func(*kingpin.ParseContext) error { c.config.CollectorsEnabled = strings.Split(collectorsEnabled, ",") @@ -189,6 +218,11 @@ func (c *Collector) GetName() string { } func (c *Collector) Close() error { + if c.criClient != nil { + c.criClient.Close() + c.criClient = nil + } + return nil } @@ -201,9 +235,6 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { } } - c.annotationsCacheHCS = 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 // it is skipped once at startup rather than logging a warning per scrape. @@ -214,6 +245,17 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { } } + _ = c.Close() + + criClient, err := cri.NewClient(c.config.CRIEndpoint) + if err != nil { + return fmt.Errorf("invalid CRI endpoint: %w", err) + } + + c.criClient = criClient + c.criRuntimeName = "" + c.criUnavailableLogged = false + c.containerAvailable = prometheus.NewDesc( prometheus.BuildFQName(types.Namespace, Name, "available"), "Available", @@ -328,20 +370,33 @@ func (c *Collector) Build(logger *slog.Logger, _ *mi.Session) error { // Collect sends the metric values for each metric // to the provided prometheus Metric channel. -func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error { +func (c *Collector) Collect(ch chan<- prometheus.Metric, maxScrapeDuration time.Duration) error { c.mu.Lock() defer c.mu.Unlock() + if maxScrapeDuration <= 0 { + maxScrapeDuration = defaultCRITimeout + } + + ctx, cancel := context.WithTimeout(context.Background(), maxScrapeDuration) + defer cancel() + errs := make([]error, 0) + // Without Kubernetes metadata, HCS containers are still exported, but without Kubernetes labels. + kubernetes, err := c.getKubernetesContainers(ctx) + if err != nil { + errs = append(errs, err) + } + if slices.Contains(c.config.CollectorsEnabled, subCollectorHCS) { - if err := c.collectHCS(ch); err != nil { + if err := c.collectHCS(ch, kubernetes); err != nil { errs = append(errs, err) } } if slices.Contains(c.config.CollectorsEnabled, subCollectorHostprocess) { - if err := c.collectJobContainers(ch); err != nil { + if err := c.collectJobContainers(ch, kubernetes); err != nil { errs = append(errs, err) } } @@ -349,7 +404,64 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error return errors.Join(errs...) } -func (c *Collector) collectHCS(ch chan<- prometheus.Metric) error { +// getKubernetesContainers reads the running containers and pod sandboxes from the CRI endpoint. +// If the endpoint is not available, e.g. on hosts without Kubernetes, it returns no containers. +func (c *Collector) getKubernetesContainers(ctx context.Context) (kubernetesContainers, error) { + if c.criRuntimeName == "" { + version, err := c.criClient.Version(ctx) + if err != nil { + return kubernetesContainers{}, c.handleCRIError(err) + } + + c.criRuntimeName = version.RuntimeName + if c.criRuntimeName == "" { + c.criRuntimeName = "containerd" + } + } + + sandboxes, err := c.criClient.ListPodSandboxes(ctx) + if err != nil { + return kubernetesContainers{}, c.handleCRIError(err) + } + + containers, err := c.criClient.ListRunningContainers(ctx) + if err != nil { + return kubernetesContainers{}, c.handleCRIError(err) + } + + if c.criUnavailableLogged { + c.logger.InfoContext(ctx, "CRI endpoint is available", slog.String("cri_endpoint", c.config.CRIEndpoint)) + + c.criUnavailableLogged = false + } + + return newKubernetesContainers(c.criRuntimeName, sandboxes, containers), nil +} + +// handleCRIError ignores errors of a missing CRI endpoint, which is expected on hosts without Kubernetes. +func (c *Collector) handleCRIError(err error) error { + // The runtime may be replaced, so query its name again. + c.criRuntimeName = "" + + var statusErr *cri.StatusError + + if !errors.Is(err, windows.ERROR_FILE_NOT_FOUND) && (!errors.As(err, &statusErr) || statusErr.Code != grpcUnimplemented) { + return fmt.Errorf("error in fetching Kubernetes metadata from CRI endpoint %s: %w", c.config.CRIEndpoint, err) + } + + if !c.criUnavailableLogged { + c.logger.Info("CRI endpoint is not available, Kubernetes labels and HostProcess containers are not collected", + slog.String("cri_endpoint", c.config.CRIEndpoint), + slog.Any("err", err), + ) + + c.criUnavailableLogged = true + } + + return nil +} + +func (c *Collector) collectHCS(ch chan<- prometheus.Metric, kubernetes kubernetesContainers) error { // Types Container is passed to get the containers compute systems only containers, err := hcs.GetContainers() if err != nil { @@ -358,8 +470,6 @@ func (c *Collector) collectHCS(ch chan<- prometheus.Metric) error { count := len(containers) if count == 0 { - clear(c.annotationsCacheHCS) - ch <- prometheus.MustNewConstMetric( c.containersCount, prometheus.GaugeValue, @@ -371,7 +481,7 @@ func (c *Collector) collectHCS(ch chan<- prometheus.Metric) error { var countersCount float64 - containerIDs := make(map[string]struct{}, len(containers)) + hcsContainers := make(map[string]containerInfo, len(containers)) collectErrors := make([]error, 0) for _, container := range containers { @@ -379,18 +489,21 @@ func (c *Collector) collectHCS(ch chan<- prometheus.Metric) error { continue } - containerIDs[container.ID] = struct{}{} - countersCount++ - 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 + // Skip pause containers + if _, ok := kubernetes.sandboxes[container.ID]; ok { + continue } + info, ok := kubernetes.containers[container.ID] + if !ok { + // Containers not managed by Kubernetes, e.g. Docker containers, have no Kubernetes labels. + info = containerInfo{id: getContainerIdWithPrefix(container)} + } + + hcsContainers[container.ID] = info + if err = c.collectHCSContainer(ch, container, info); err != nil { if errors.Is(err, hcs.ErrIDNotFound) { c.logger.Debug("err in fetching container statistics", @@ -422,17 +535,8 @@ func (c *Collector) collectHCS(ch chan<- prometheus.Metric) error { countersCount, ) - networkErr := c.collectNetworkMetrics(ch) - - // Remove containers that are no longer running - for containerID := range c.annotationsCacheHCS { - if _, ok := containerIDs[containerID]; !ok { - delete(c.annotationsCacheHCS, containerID) - } - } - - if networkErr != nil { - return fmt.Errorf("error in fetching container network statistics: %w", networkErr) + if err := c.collectNetworkMetrics(ch, hcsContainers, kubernetes.sandboxes); err != nil { + return fmt.Errorf("error in fetching container network statistics: %w", err) } if len(collectErrors) > 0 { @@ -443,18 +547,6 @@ func (c *Collector) collectHCS(ch chan<- prometheus.Metric) error { } func (c *Collector) collectHCSContainer(ch chan<- prometheus.Metric, containerDetails hcs.Properties, containerInfo containerInfo) error { - // Skip if the container is a pause container - if containerInfo.pod != "" && containerInfo.container == "" { - c.logger.Debug("skipping pause container", - slog.String("container_id", containerDetails.ID), - slog.String("container_name", containerInfo.container), - slog.String("pod_name", containerInfo.pod), - slog.String("namespace", containerInfo.namespace), - ) - - return nil - } - containerStats, err := hcs.GetContainerStatistics(containerDetails.ID) if err != nil { return fmt.Errorf("error fetching container statistics: %w", err) @@ -550,8 +642,8 @@ func (c *Collector) collectHCSContainer(ch chan<- prometheus.Metric, containerDe return nil } -// collectNetworkMetrics collects network metrics for containers. -func (c *Collector) collectNetworkMetrics(ch chan<- prometheus.Metric) error { +// collectNetworkMetrics collects network metrics for the HCS containers. +func (c *Collector) collectNetworkMetrics(ch chan<- prometheus.Metric, containers map[string]containerInfo, sandboxes map[string]struct{}) error { endpoints, err := hcn.ListEndpoints() if err != nil { return fmt.Errorf("error in fetching HCN endpoints: %w", err) @@ -576,19 +668,16 @@ func (c *Collector) collectNetworkMetrics(ch chan<- prometheus.Metric) error { } for _, containerId := range endpoint.SharedContainers { - containerInfo, ok := c.annotationsCacheHCS[containerId] - + containerInfo, ok := containers[containerId] if !ok { - c.logger.Debug("Unknown container " + containerId + " for endpoint " + endpoint.ID) + // Pause containers share the endpoint of their pod. + if _, ok := sandboxes[containerId]; !ok { + c.logger.Debug("Unknown container " + containerId + " for endpoint " + endpoint.ID) + } continue } - // Skip if the container is a pause container - if containerInfo.pod != "" && containerInfo.container == "" { - continue - } - endpointId := strings.ToUpper(endpoint.ID) ch <- prometheus.MustNewConstMetric( @@ -643,81 +732,21 @@ func (c *Collector) collectNetworkMetrics(ch chan<- prometheus.Metric) error { // https://learn.microsoft.com/en-us/windows/win32/procthread/job-objects // // Job containers are containers that aren't managed by HCS, e.g host process containers. -func (c *Collector) collectJobContainers(ch chan<- prometheus.Metric) error { - // 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) - } - - if !c.stateDirMissingLogged { - c.logger.Warn("containerd state directory does not exist", - slog.String("path", c.config.ContainerDStateDir), - slog.Any("err", err), - ) - - c.stateDirMissingLogged = true - } - - clear(c.annotationsCacheJob) - - return nil - } - - c.stateDirMissingLogged = false - - bundles := make(map[string]struct{}, len(entries)) +// 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 { 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), - ) - - continue - } - - container = jobContainer{ - info: spec.containerInfo("containerd://" + containerID), - hostProcess: spec.Annotations["microsoft.com/hostprocess-container"] == "true", - } - - c.annotationsCacheJob[containerID] = container - } - - if !container.hostProcess { - continue - } - - if err := c.collectJobContainer(ch, containerID, container.info); err != nil { + for containerID, info := range kubernetes.containers { + if err := c.collectJobContainer(ch, containerID, info); err != nil { errs = append(errs, err) } } - // Remove containers that are no longer running - for containerID := range c.annotationsCacheJob { - if _, ok := bundles[containerID]; !ok { - delete(c.annotationsCacheJob, containerID) - } - } - return 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) if err != nil { @@ -865,23 +894,6 @@ func getContainerIdWithPrefix(container hcs.Properties) string { } } -func (c *Collector) getContainerAnnotations(containerID string) (ociSpec, error) { - // 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 reading config.json file: %w", err) - } - - var annotations ociSpec - - if err = json.Unmarshal(configJSON, &annotations); err != nil { - return ociSpec{}, fmt.Errorf("error in decoding config.json file: %w", err) - } - - return annotations, nil -} - func calculatePrivateWorkingSetBytes(jobObjectHandle windows.Handle) (uint64, error) { pids, err := kernel32.QueryJobObjectProcessIDs(jobObjectHandle) if err != nil { diff --git a/internal/collector/container/container_internal_test.go b/internal/collector/container/container_internal_test.go index f25d6dc6..e79e04a1 100644 --- a/internal/collector/container/container_internal_test.go +++ b/internal/collector/container/container_internal_test.go @@ -18,109 +18,132 @@ package container import ( + "errors" + "fmt" "log/slog" - "os" - "path/filepath" "testing" + "github.com/prometheus-community/windows_exporter/internal/headers/cri" "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/require" + "golang.org/x/sys/windows" ) -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 TestNewKubernetesContainers(t *testing.T) { + t.Parallel() + + k := newKubernetesContainers("containerd", + []cri.PodSandbox{ + {ID: "sandbox-1", Name: "pod", Namespace: "default"}, + {ID: "sandbox-2", Name: "stopped", Namespace: "kube-system", State: cri.SandboxNotReady}, + }, + []cri.Container{ + {ID: "abc", PodSandboxID: "sandbox-1", Name: "nanoserver", State: cri.ContainerRunning}, + {ID: "orphan", PodSandboxID: "unknown", Name: "orphan", State: cri.ContainerRunning}, + {ID: "exited", PodSandboxID: "sandbox-1", Name: "init", State: cri.ContainerExited}, + }, + ) + + require.Equal(t, map[string]containerInfo{ + "abc": { + id: "containerd://abc", + namespace: "default", + pod: "pod", + container: "nanoserver", + }, + "orphan": { + id: "containerd://orphan", + container: "orphan", + }, + }, k.containers) + require.Equal(t, map[string]struct{}{"sandbox-1": {}, "sandbox-2": {}}, k.sandboxes) +} + +func TestHandleCRIError(t *testing.T) { + t.Parallel() + + for name, tc := range map[string]struct { + err error + unavailable bool + }{ + "pipe not found": { + err: fmt.Errorf("dial: %w", windows.ERROR_FILE_NOT_FOUND), + unavailable: true, + }, + "CRI plugin disabled": { + err: fmt.Errorf("CRI Version: %w", &cri.StatusError{Code: grpcUnimplemented}), + unavailable: true, + }, + "other gRPC status": { + err: fmt.Errorf("CRI Version: %w", &cri.StatusError{Code: 14}), + }, + "other error": { + err: errors.New("timeout"), + }, + } { + t.Run(name, func(t *testing.T) { + t.Parallel() + + c := &Collector{ + config: Config{CRIEndpoint: ConfigDefaults.CRIEndpoint}, + logger: slog.New(slog.DiscardHandler), + criRuntimeName: "containerd", + } + + err := c.handleCRIError(tc.err) + require.Empty(t, c.criRuntimeName) + require.Equal(t, tc.unavailable, c.criUnavailableLogged) + + if tc.unavailable { + require.NoError(t, err) + } else { + require.ErrorIs(t, err, tc.err) + } + }) } } -func writeBundle(t *testing.T, stateDir, containerID, config string) { - t.Helper() +// A missing CRI endpoint is expected on hosts without Kubernetes. +func TestGetKubernetesContainersWithoutCRI(t *testing.T) { + t.Parallel() - bundleDir := filepath.Join(stateDir, containerID) + 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)) - require.NoError(t, os.MkdirAll(bundleDir, 0o755)) + t.Cleanup(func() { _ = c.Close() }) - if config != "" { - require.NoError(t, os.WriteFile(filepath.Join(bundleDir, "config.json"), []byte(config), 0o600)) + for range 2 { + k, err := c.getKubernetesContainers(t.Context()) + require.NoError(t, err) + require.Empty(t, k.containers) + require.True(t, c.criUnavailableLogged) } } -// 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) { +// Containers without a job object are not job containers and are skipped. +func TestCollectJobContainersSkipsHCSContainers(t *testing.T) { t.Parallel() - c := newTestCollector(t.TempDir() + `\`) - c.annotationsCacheJob["removed"] = jobContainer{info: containerInfo{id: "containerd://removed"}} - + c := &Collector{logger: slog.New(slog.DiscardHandler)} 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, + require.NoError(t, c.collectJobContainers(ch, kubernetesContainers{ + containers: map[string]containerInfo{ + "windows-exporter-test-no-job-object": {id: "containerd://windows-exporter-test-no-job-object"}, }, - "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") + })) + require.Empty(t, ch) } -func TestCollectJobContainersMissingStateDir(t *testing.T) { +func TestBuildInvalidCRIEndpoint(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) + c := New(&Config{ + CollectorsEnabled: []string{subCollectorHostprocess}, + CRIEndpoint: "unix:///run/containerd/containerd.sock", + }) + require.Error(t, c.Build(slog.New(slog.DiscardHandler), nil)) } diff --git a/internal/collector/container/container_test.go b/internal/collector/container/container_test.go index ae4bccc3..471d8204 100644 --- a/internal/collector/container/container_test.go +++ b/internal/collector/container/container_test.go @@ -31,6 +31,6 @@ func BenchmarkCollector(b *testing.B) { 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", "hostprocess": "true"}) - testutils.RequireFixtureMetric(t, metrics, container.Name, "windows_container_available", prometheus.Labels{"container": "nanoserver", "hostprocess": "false"}) + 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"}) } diff --git a/internal/headers/cri/cri.go b/internal/headers/cri/cri.go new file mode 100644 index 00000000..7002c901 --- /dev/null +++ b/internal/headers/cri/cri.go @@ -0,0 +1,204 @@ +// 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 cri is a minimal client for the Kubernetes Container Runtime Interface (CRI) v1. +// +// gRPC is HTTP/2 with length-prefixed protobuf messages, so the client uses +// net/http with unencrypted HTTP/2 and decodes the few messages it needs by hand. +// This avoids linking google.golang.org/grpc. +// https://github.com/kubernetes/cri-api/blob/master/pkg/apis/runtime/v1/api.proto +package cri + +import ( + "bytes" + "context" + "encoding/binary" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/url" + "strconv" +) + +const ( + // maxMessageSize limits the size of a response, like the gRPC default receive limit. + maxMessageSize = 16 << 20 + + runtimeService = "/runtime.v1.RuntimeService/" +) + +var ErrInvalidResponse = errors.New("invalid gRPC response") + +// StatusError is a non-OK gRPC status returned by the runtime. +type StatusError struct { + Code int + Message string +} + +func (e *StatusError) Error() string { + return fmt.Sprintf("gRPC status %d: %s", e.Code, e.Message) +} + +// Client calls the CRI runtime service. It keeps one HTTP/2 connection open and is safe for concurrent use. +type Client struct { + transport *http.Transport + client *http.Client +} + +// NewClient creates a client for a CRI endpoint like npipe:////./pipe/containerd-containerd. +// It does not connect until the first call. +func NewClient(endpoint string) (*Client, error) { + path, err := pipePath(endpoint) + if err != nil { + return nil, err + } + + protocols := new(http.Protocols) + protocols.SetUnencryptedHTTP2(true) + + transport := &http.Transport{ + Protocols: protocols, + DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { + return dialPipe(ctx, path) + }, + DisableCompression: true, + } + + return &Client{ + transport: transport, + client: &http.Client{Transport: transport}, + }, nil +} + +// Close closes the connection to the runtime. +func (c *Client) Close() { + c.transport.CloseIdleConnections() +} + +// Version returns the name and version of the runtime. +func (c *Client) Version(ctx context.Context) (Version, error) { + resp, err := c.call(ctx, "Version", nil) + if err != nil { + return Version{}, err + } + + return decodeVersion(resp) +} + +// ListRunningContainers returns all running containers. +func (c *Client) ListRunningContainers(ctx context.Context) ([]Container, error) { + resp, err := c.call(ctx, "ListContainers", encodeListRunningContainersRequest()) + if err != nil { + return nil, err + } + + return decodeListContainersResponse(resp) +} + +// ListPodSandboxes returns all pod sandboxes. +func (c *Client) ListPodSandboxes(ctx context.Context) ([]PodSandbox, error) { + resp, err := c.call(ctx, "ListPodSandbox", nil) + if err != nil { + return nil, err + } + + return decodeListPodSandboxResponse(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. + body := make([]byte, 5+len(msg)) + binary.BigEndian.PutUint32(body[1:5], uint32(len(msg))) + copy(body[5:], msg) + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "http://localhost"+runtimeService+method, bytes.NewReader(body)) + if err != nil { + return nil, err + } + + req.Header.Set("Content-Type", "application/grpc") + req.Header.Set("TE", "trailers") + + resp, err := c.client.Do(req) + if err != nil { + return nil, fmt.Errorf("CRI %s: %w", method, err) + } + + defer func() { + _ = resp.Body.Close() + }() + + data, err := io.ReadAll(io.LimitReader(resp.Body, 5+maxMessageSize+1)) + if err != nil { + return nil, fmt.Errorf("CRI %s: %w", method, err) + } + + if len(data) > 5+maxMessageSize { + return nil, fmt.Errorf("CRI %s: %w: message exceeds %d bytes", method, ErrInvalidResponse, maxMessageSize) + } + + if err := grpcStatus(resp); err != nil { + return nil, fmt.Errorf("CRI %s: %w", method, err) + } + + if len(data) < 5 || data[0] != 0 { + return nil, fmt.Errorf("CRI %s: %w: missing or compressed message", method, ErrInvalidResponse) + } + + if int(binary.BigEndian.Uint32(data[1:5])) != len(data)-5 { + return nil, fmt.Errorf("CRI %s: %w: message length mismatch", method, ErrInvalidResponse) + } + + return data[5:], nil +} + +// grpcStatus returns the gRPC status of a response, which is sent as trailer. +// Responses without a message carry the status in the headers. +func grpcStatus(resp *http.Response) error { + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("%w: HTTP status %s", ErrInvalidResponse, resp.Status) + } + + header := resp.Trailer + if header.Get("Grpc-Status") == "" { + header = resp.Header + } + + status := header.Get("Grpc-Status") + if status == "" { + return fmt.Errorf("%w: missing grpc-status", ErrInvalidResponse) + } + + code, err := strconv.Atoi(status) + if err != nil { + return fmt.Errorf("%w: invalid grpc-status %q", ErrInvalidResponse, status) + } + + if code == 0 { + return nil + } + + message := header.Get("Grpc-Message") + if unescaped, err := url.PathUnescape(message); err == nil { + message = unescaped + } + + return &StatusError{Code: code, Message: message} +} diff --git a/internal/headers/cri/cri_internal_test.go b/internal/headers/cri/cri_internal_test.go new file mode 100644 index 00000000..00ca418f --- /dev/null +++ b/internal/headers/cri/cri_internal_test.go @@ -0,0 +1,312 @@ +// 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 cri + +import ( + "encoding/binary" + "io" + "net/http" + "strings" + "testing" + + "github.com/stretchr/testify/require" + "golang.org/x/sys/windows" + "google.golang.org/protobuf/encoding/protowire" +) + +// message builds protobuf messages for test fixtures. +type message []byte + +func (m message) string(num protowire.Number, v string) message { + m = protowire.AppendTag(m, num, protowire.BytesType) + + return protowire.AppendString(m, v) +} + +func (m message) message(num protowire.Number, v message) message { + m = protowire.AppendTag(m, num, protowire.BytesType) + + return protowire.AppendBytes(m, v) +} + +func (m message) varint(num protowire.Number, v uint64) message { + m = protowire.AppendTag(m, num, protowire.VarintType) + + return protowire.AppendVarint(m, v) +} + +func (m message) fixed64(num protowire.Number, v uint64) message { + m = protowire.AppendTag(m, num, protowire.Fixed64Type) + + return protowire.AppendFixed64(m, v) +} + +func labels(num protowire.Number, key, value string) message { + return message{}.message(num, message{}.string(1, key).string(2, value)) +} + +// A ListContainersResponse as containerd sends it, including fields the client skips. +func containersResponse() message { + nanoserver := message{}. + string(1, "abc"). + string(2, "sandbox-1"). + message(3, message{}.string(1, "nanoserver").varint(2, 2)). + message(4, message{}.string(1, "mcr.microsoft.com/windows/nanoserver")). + string(5, "sha256:1234"). + varint(6, uint64(ContainerRunning)). + varint(7, 1700000000000000000) + nanoserver = append(nanoserver, labels(8, "io.kubernetes.pod.name", "pod")...) + nanoserver = append(nanoserver, labels(9, "io.kubernetes.container.restartCount", "2")...) + + // Unknown fields of all wire types must be skipped. + nanoserver = nanoserver.fixed64(100, 42).varint(101, 1).string(102, "unknown") + + return message{}. + message(1, nanoserver). + message(1, message{}.string(1, "def").message(3, message{}.string(1, "other"))) +} + +func sandboxesResponse() message { + return message{}. + message(1, message{}. + string(1, "sandbox-1"). + message(2, message{}.string(1, "pod").string(2, "uid-1").string(3, "default").varint(4, 1)). + varint(3, uint64(SandboxReady)). + varint(4, 1700000000000000000)). + message(1, message{}. + string(1, "sandbox-2"). + message(2, message{}.string(1, "old").string(3, "kube-system")). + varint(3, uint64(SandboxNotReady))) +} + +func TestDecodeListContainersResponse(t *testing.T) { + t.Parallel() + + containers, err := decodeListContainersResponse(containersResponse()) + require.NoError(t, err) + require.Equal(t, []Container{ + {ID: "abc", PodSandboxID: "sandbox-1", Name: "nanoserver", State: ContainerRunning}, + {ID: "def", Name: "other", State: ContainerCreated}, + }, containers) +} + +func TestDecodeListPodSandboxResponse(t *testing.T) { + t.Parallel() + + sandboxes, err := decodeListPodSandboxResponse(sandboxesResponse()) + require.NoError(t, err) + require.Equal(t, []PodSandbox{ + {ID: "sandbox-1", Name: "pod", UID: "uid-1", Namespace: "default", State: SandboxReady}, + {ID: "sandbox-2", Name: "old", Namespace: "kube-system", State: SandboxNotReady}, + }, sandboxes) +} + +func TestDecodeInvalidMessage(t *testing.T) { + t.Parallel() + + // A truncated container message. + truncated := containersResponse() + truncated = truncated[:len(truncated)-3] + + _, err := decodeListContainersResponse(truncated) + require.Error(t, err) + + // A nested message with an invalid length. + _, err = decodeListPodSandboxResponse(message{}.message(1, message{0x12, 0x7f})) + require.Error(t, err) +} + +func TestEncodeListRunningContainersRequest(t *testing.T) { + t.Parallel() + + want := message{}.message(1, message{}.message(2, message{}.varint(1, uint64(ContainerRunning)))) + require.Equal(t, []byte(want), encodeListRunningContainersRequest()) +} + +// grpcHandler serves a unary gRPC method like grpc-go does. +func grpcHandler(t *testing.T, wantRequest []byte, response []byte) http.HandlerFunc { + t.Helper() + + return func(w http.ResponseWriter, r *http.Request) { + if r.ProtoMajor != 2 || r.Header.Get("Content-Type") != "application/grpc" || r.Header.Get("TE") != "trailers" { + http.Error(w, "not a gRPC request", http.StatusUnsupportedMediaType) + + return + } + + body, err := io.ReadAll(r.Body) + if err != nil || len(body) < 5 || int(binary.BigEndian.Uint32(body[1:5])) != len(body)-5 || string(body[5:]) != string(wantRequest) { + w.Header().Set("Content-Type", "application/grpc") + w.Header().Set("Grpc-Status", "3") + w.Header().Set("Grpc-Message", "unexpected request") + + return + } + + w.Header().Set("Content-Type", "application/grpc") + w.Header().Set("Trailer", "Grpc-Status, Grpc-Message") + + frame := make([]byte, 5+len(response)) + binary.BigEndian.PutUint32(frame[1:5], uint32(len(response))) + copy(frame[5:], response) + + _, _ = w.Write(frame) + + w.Header().Set("Grpc-Status", "0") + } +} + +func startServer(t *testing.T, handler http.Handler) string { + t.Helper() + + l := listenPipe(t) + + protocols := new(http.Protocols) + protocols.SetUnencryptedHTTP2(true) + + server := &http.Server{Handler: handler, Protocols: protocols} //nolint:gosec // test server + + go func() { _ = server.Serve(l) }() + + t.Cleanup(func() { _ = server.Close() }) + + return "npipe://" + strings.ReplaceAll(l.path, `\`, "/") +} + +func TestClient(t *testing.T) { + t.Parallel() + + mux := http.NewServeMux() + mux.Handle(runtimeService+"Version", grpcHandler(t, nil, + 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())) + + client, err := NewClient(startServer(t, mux)) + require.NoError(t, err) + + t.Cleanup(client.Close) + + // Repeated calls reuse the HTTP/2 connection. + for range 3 { + version, err := client.Version(t.Context()) + require.NoError(t, err) + require.Equal(t, Version{RuntimeName: "containerd", RuntimeVersion: "v2.1.4", RuntimeAPIVersion: "v1"}, version) + + containers, err := client.ListRunningContainers(t.Context()) + require.NoError(t, err) + require.Len(t, containers, 2) + require.Equal(t, "nanoserver", containers[0].Name) + + sandboxes, err := client.ListPodSandboxes(t.Context()) + require.NoError(t, err) + require.Len(t, sandboxes, 2) + require.Equal(t, "default", sandboxes[0].Namespace) + } +} + +func TestClientStatusError(t *testing.T) { + t.Parallel() + + mux := http.NewServeMux() + mux.HandleFunc(runtimeService+"Version", func(w http.ResponseWriter, _ *http.Request) { + // A trailers-only response, as sent for errors. + w.Header().Set("Content-Type", "application/grpc") + w.Header().Set("Grpc-Status", "12") + w.Header().Set("Grpc-Message", "unknown service runtime.v1.RuntimeService%2FVersion") + }) + + client, err := NewClient(startServer(t, mux)) + require.NoError(t, err) + + t.Cleanup(client.Close) + + _, err = client.Version(t.Context()) + + var statusErr *StatusError + + require.ErrorAs(t, err, &statusErr) + require.Equal(t, 12, statusErr.Code) + require.Equal(t, "unknown service runtime.v1.RuntimeService/Version", statusErr.Message) + + // A non-gRPC response, here a 404 for an unregistered method, is rejected. + _, err = client.ListPodSandboxes(t.Context()) + require.ErrorIs(t, err, ErrInvalidResponse) +} + +func TestClientMessageSize(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + payloadSize int + declaredSize int + invalid bool + }{ + {name: "at limit", payloadSize: maxMessageSize, declaredSize: maxMessageSize}, + {name: "one byte over limit", payloadSize: maxMessageSize + 1, declaredSize: maxMessageSize + 1, invalid: true}, + { + name: "truncated oversized response", + payloadSize: maxMessageSize + 2, + declaredSize: maxMessageSize + 1, + invalid: true, + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + handler := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/grpc") + w.Header().Set("Grpc-Status", "0") + + frame := make([]byte, 5+tc.payloadSize) + binary.BigEndian.PutUint32(frame[1:5], uint32(tc.declaredSize)) + + _, _ = w.Write(frame) + }) + + client, err := NewClient(startServer(t, handler)) + require.NoError(t, err) + + t.Cleanup(client.Close) + + data, err := client.call(t.Context(), "Version", nil) + if tc.invalid { + require.ErrorIs(t, err, ErrInvalidResponse) + require.Nil(t, data) + + return + } + + require.NoError(t, err) + require.Len(t, data, tc.payloadSize) + }) + } +} + +func TestClientPipeNotFound(t *testing.T) { + t.Parallel() + + client, err := NewClient("npipe:////./pipe/windows_exporter-cri-test-does-not-exist") + require.NoError(t, err) + + t.Cleanup(client.Close) + + _, err = client.Version(t.Context()) + require.ErrorIs(t, err, windows.ERROR_FILE_NOT_FOUND) +} diff --git a/internal/headers/cri/pipe.go b/internal/headers/cri/pipe.go new file mode 100644 index 00000000..203f8bd9 --- /dev/null +++ b/internal/headers/cri/pipe.go @@ -0,0 +1,266 @@ +// 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 cri + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "os" + "runtime" + "strings" + "sync" + "time" + + "golang.org/x/sys/windows" +) + +// pipePath converts a CRI endpoint like npipe:////./pipe/containerd-containerd +// into a named pipe path like \\.\pipe\containerd-containerd. +func pipePath(endpoint string) (string, error) { + path, ok := strings.CutPrefix(endpoint, "npipe://") + if !ok { + return "", fmt.Errorf("unsupported CRI endpoint %q: only npipe:// endpoints are supported", endpoint) + } + + path = strings.ReplaceAll(path, "/", `\`) + + if !strings.HasPrefix(path, `\\.\pipe\`) || len(path) == len(`\\.\pipe\`) { + return "", fmt.Errorf("invalid CRI endpoint %q: expected npipe:////./pipe/", endpoint) + } + + return path, nil +} + +// dialPipe connects to the named pipe. The handle is opened for overlapped I/O, +// so the returned connection supports deadlines and can be closed while a read is pending. +func dialPipe(ctx context.Context, path string) (net.Conn, error) { + name, err := windows.UTF16PtrFromString(path) + if err != nil { + return nil, err + } + + for { + // SECURITY_ANONYMOUS prevents the pipe server from impersonating the exporter. + handle, err := windows.CreateFile( + name, + windows.GENERIC_READ|windows.GENERIC_WRITE, + 0, + nil, + windows.OPEN_EXISTING, + windows.FILE_FLAG_OVERLAPPED|windows.SECURITY_SQOS_PRESENT|windows.SECURITY_ANONYMOUS, + 0, + ) + if err == nil { + return newReadAheadConn(&pipeConn{File: os.NewFile(uintptr(handle), path)}), nil + } + + // All pipe instances are busy until the server creates the next one. + if !errors.Is(err, windows.ERROR_PIPE_BUSY) { + return nil, &net.OpError{Op: "dial", Net: "pipe", Addr: pipeAddr(path), Err: err} + } + + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(10 * time.Millisecond): + } + } +} + +type pipeAddr string + +func (a pipeAddr) Network() string { return "pipe" } +func (a pipeAddr) String() string { return string(a) } + +// pipeConn adapts a named pipe file to net.Conn. +type pipeConn struct { + *os.File + + readMu sync.Mutex + mu sync.Mutex + closed bool +} + +func (c *pipeConn) LocalAddr() net.Addr { return pipeAddr(c.Name()) } +func (c *pipeConn) RemoteAddr() net.Addr { return pipeAddr(c.Name()) } + +// Read uses event-based overlapped I/O because os.File.Read and os.File.Write +// update a shared file offset concurrently on Windows pipes. Serializing both +// directions would deadlock HTTP/2. Control keeps the handle alive until the +// read completes; File.Close cancels pending pipe I/O, including this read. +func (c *pipeConn) Read(p []byte) (int, error) { + c.readMu.Lock() + defer c.readMu.Unlock() + + if len(p) == 0 { + return 0, nil + } + + raw, err := c.SyscallConn() + if err != nil { + return 0, err + } + + event, err := windows.CreateEvent(nil, 1, 0, nil) + if err != nil { + return 0, err + } + + defer func() { _ = windows.CloseHandle(event) }() + + // The low bit suppresses notifications to the Go runtime's completion port, + // which only accepts OVERLAPPED structures created by its own poller. + overlapped := windows.Overlapped{HEvent: event | 1} + + var pinner runtime.Pinner + pinner.Pin(&overlapped) + pinner.Pin(&p[0]) + + defer pinner.Unpin() + + var ( + n uint32 + readErr error + ) + + err = raw.Control(func(handle uintptr) { + c.mu.Lock() + if c.closed { + c.mu.Unlock() + + readErr = os.ErrClosed + + return + } + + readErr = windows.ReadFile(windows.Handle(handle), p, &n, &overlapped) + c.mu.Unlock() + + if errors.Is(readErr, windows.ERROR_IO_PENDING) { + readErr = windows.GetOverlappedResult(windows.Handle(handle), &overlapped, &n, true) + } + }) + if err != nil { + return 0, err + } + + switch { + case errors.Is(readErr, windows.ERROR_BROKEN_PIPE), readErr == nil && n == 0: + readErr = io.EOF + case errors.Is(readErr, windows.ERROR_OPERATION_ABORTED): + readErr = os.ErrClosed + } + + return int(n), readErr +} + +func (c *pipeConn) Close() error { + c.mu.Lock() + if c.closed { + c.mu.Unlock() + + return os.ErrClosed + } + + // Prevent a new read from starting after File.Close cancels pending I/O. + c.closed = true + c.mu.Unlock() + + return c.File.Close() +} + +// Read deadlines are unsupported by the event-based reader. Close cancels it. +func (c *pipeConn) SetReadDeadline(time.Time) error { return nil } + +func (c *pipeConn) SetDeadline(t time.Time) error { return c.SetWriteDeadline(t) } + +// readAheadConn reads the connection eagerly in the background. +// +// Both HTTP/2 peers write their preface before they read. A write to a named +// pipe without buffer, like the one containerd listens on, blocks until the +// peer reads it, so both sides would block forever on their first write. +type readAheadConn struct { + net.Conn + + mu sync.Mutex + cond *sync.Cond + buf []byte + err error +} + +func newReadAheadConn(conn net.Conn) *readAheadConn { + c := &readAheadConn{Conn: conn} + c.cond = sync.NewCond(&c.mu) + + go c.readLoop() + + return c +} + +func (c *readAheadConn) readLoop() { + buf := make([]byte, 32*1024) + + for { + n, err := c.Conn.Read(buf) + + c.mu.Lock() + c.buf = append(c.buf, buf[:n]...) + c.err = err + c.cond.Broadcast() + c.mu.Unlock() + + if err != nil { + return + } + } +} + +func (c *readAheadConn) Read(p []byte) (int, error) { + c.mu.Lock() + defer c.mu.Unlock() + + for len(c.buf) == 0 && c.err == nil { + c.cond.Wait() + } + + if len(c.buf) == 0 { + return 0, c.err + } + + n := copy(p, c.buf) + c.buf = c.buf[n:] + + if len(c.buf) == 0 { + c.buf = nil + } + + return n, nil +} + +// SetDeadline only applies to writes. A read deadline would stop the background reader for good. +func (c *readAheadConn) SetDeadline(t time.Time) error { + return c.SetWriteDeadline(t) +} + +// SetReadDeadline is not supported. Closing the connection unblocks pending reads. +func (c *readAheadConn) SetReadDeadline(time.Time) error { + return nil +} diff --git a/internal/headers/cri/pipe_internal_test.go b/internal/headers/cri/pipe_internal_test.go new file mode 100644 index 00000000..d60ddd9d --- /dev/null +++ b/internal/headers/cri/pipe_internal_test.go @@ -0,0 +1,396 @@ +// 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 cri + +import ( + "errors" + "fmt" + "io" + "net" + "os" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + "golang.org/x/sys/windows" +) + +// pipeCounter makes pipe names unique, the wall clock is too coarse for parallel tests. +var pipeCounter atomic.Int64 //nolint:gochecknoglobals + +// pipeListener accepts connections on a named pipe without buffer, like the one containerd creates. +// Like go-winio, it always keeps one pipe instance waiting, so clients never see a missing pipe. +type pipeListener struct { + path string + + mu sync.Mutex + next windows.Handle + closed chan struct{} +} + +func listenPipe(t *testing.T) *pipeListener { + t.Helper() + + l := &pipeListener{ + path: fmt.Sprintf(`\\.\pipe\windows_exporter-cri-test-%d-%d`, os.Getpid(), pipeCounter.Add(1)), + closed: make(chan struct{}), + } + + var err error + + l.next, err = l.createInstance() + require.NoError(t, err) + + t.Cleanup(func() { _ = l.Close() }) + + return l +} + +func (l *pipeListener) createInstance() (windows.Handle, error) { + name, err := windows.UTF16PtrFromString(l.path) + if err != nil { + return 0, err + } + + return windows.CreateNamedPipe( + name, + windows.PIPE_ACCESS_DUPLEX|windows.FILE_FLAG_OVERLAPPED, + windows.PIPE_TYPE_BYTE|windows.PIPE_READMODE_BYTE|windows.PIPE_WAIT, + windows.PIPE_UNLIMITED_INSTANCES, + 0, // no output buffer + 0, // no input buffer + 0, + nil, + ) +} + +func (l *pipeListener) Accept() (net.Conn, error) { + l.mu.Lock() + handle := l.next + l.next = 0 + l.mu.Unlock() + + if handle == 0 { + return nil, net.ErrClosed + } + + if err := l.connect(handle); err != nil { + _ = windows.CloseHandle(handle) + + return nil, err + } + + next, err := l.createInstance() + if err != nil { + _ = windows.CloseHandle(handle) + + return nil, err + } + + l.mu.Lock() + select { + case <-l.closed: + _ = windows.CloseHandle(next) + default: + l.next = next + } + l.mu.Unlock() + + return &pipeConn{File: os.NewFile(uintptr(handle), l.path)}, nil +} + +// connect waits for a client to connect, or for the listener to be closed. +func (l *pipeListener) connect(handle windows.Handle) error { + event, err := windows.CreateEvent(nil, 1, 0, nil) + if err != nil { + return err + } + + defer func() { + _ = windows.CloseHandle(event) + }() + + overlapped := windows.Overlapped{HEvent: event} + + err = windows.ConnectNamedPipe(handle, &overlapped) + if errors.Is(err, windows.ERROR_PIPE_CONNECTED) { + return nil + } else if !errors.Is(err, windows.ERROR_IO_PENDING) { + return err + } + + for { + select { + case <-l.closed: + _ = windows.CancelIoEx(handle, &overlapped) + _, _ = windows.WaitForSingleObject(event, windows.INFINITE) + + return net.ErrClosed + default: + } + + result, err := windows.WaitForSingleObject(event, 50) + if err != nil { + return err + } + + if result == windows.WAIT_OBJECT_0 { + return nil + } + } +} + +func (l *pipeListener) Close() error { + l.mu.Lock() + defer l.mu.Unlock() + + select { + case <-l.closed: + return nil + default: + } + + close(l.closed) + + if l.next != 0 { + _ = windows.CloseHandle(l.next) + l.next = 0 + } + + return nil +} + +func (l *pipeListener) Addr() net.Addr { return pipeAddr(l.path) } + +func TestPipePath(t *testing.T) { + t.Parallel() + + for endpoint, want := range map[string]string{ + "npipe:////./pipe/containerd-containerd": `\\.\pipe\containerd-containerd`, + `npipe://\\.\pipe\containerd-containerd`: `\\.\pipe\containerd-containerd`, + "npipe:////./pipe/": "", + "npipe://./pipe/containerd-containerd": "", + "unix:///run/containerd/containerd.sock": "", + `\\.\pipe\containerd-containerd`: "", + } { + got, err := pipePath(endpoint) + if want == "" { + require.Error(t, err, endpoint) + + continue + } + + require.NoError(t, err, endpoint) + require.Equal(t, want, got, endpoint) + } +} + +// Closing a pipe must cancel an event-based read while Control holds the handle. +func TestPipeConnClose(t *testing.T) { + t.Parallel() + + l := listenPipe(t) + accepted := make(chan net.Conn, 1) + acceptErr := make(chan error, 1) + + go func() { + conn, err := l.Accept() + if err != nil { + acceptErr <- err + + return + } + + accepted <- conn + }() + + client, err := dialPipe(t.Context(), l.path) + require.NoError(t, err) + + t.Cleanup(func() { _ = client.Close() }) + + var server net.Conn + + select { + case server = <-accepted: + case err := <-acceptErr: + t.Fatal(err) + case <-time.After(5 * time.Second): + t.Fatal("pipe was not accepted") + } + + t.Cleanup(func() { _ = server.Close() }) + + readErr := make(chan error, 1) + + go func() { + _, err := server.Read(make([]byte, 1)) + readErr <- err + }() + + // Sending a byte proves the reader can complete an overlapped operation. + require.NoError(t, client.SetWriteDeadline(time.Now().Add(5*time.Second))) + + _, err = client.Write([]byte{1}) + require.NoError(t, err) + + select { + case err := <-readErr: + require.NoError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("pipe read did not complete") + } + + go func() { + _, err := server.Read(make([]byte, 1)) + readErr <- err + }() + + closed := make(chan error, 1) + + go func() { closed <- server.Close() }() + + select { + case err := <-readErr: + require.ErrorIs(t, err, os.ErrClosed) + case <-time.After(5 * time.Second): + t.Fatal("pending pipe read was not canceled by Close") + } + + select { + case err := <-closed: + require.NoError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("Close did not release the pipe handle") + } +} + +// Both peers write first. On a pipe without buffer, this blocks both writes +// until the peer reads, unless the client reads ahead. +func TestReadAheadConnPreventsDeadlock(t *testing.T) { + t.Parallel() + + l := listenPipe(t) + + type result struct { + conn net.Conn + err error + } + + accepted := make(chan result, 1) + + go func() { + conn, err := l.Accept() + accepted <- result{conn, err} + }() + + var ( + client net.Conn + err error + ) + + require.Eventually(t, func() bool { + client, err = dialPipe(t.Context(), l.path) + + return err == nil + }, 5*time.Second, 10*time.Millisecond) + + defer client.Close() + + server := <-accepted + require.NoError(t, server.err) + + defer server.conn.Close() + + done := make(chan error, 2) + + go func() { + // Server: write the preface, then read the peer's. + if _, err := server.conn.Write([]byte("server-preface")); err != nil { + done <- err + + return + } + + buf := make([]byte, len("client-preface")) + + _, err := io.ReadFull(server.conn, buf) + done <- err + }() + + go func() { + // Client: write the preface, then read the peer's. + if _, err := client.Write([]byte("client-preface")); err != nil { + done <- err + + return + } + + buf := make([]byte, len("server-preface")) + if _, err := io.ReadFull(client, buf); err != nil { + done <- err + + return + } + + if string(buf) != "server-preface" { + done <- fmt.Errorf("unexpected preface %q", buf) + + return + } + + done <- nil + }() + + for range 2 { + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("deadlock: both peers are blocked writing") + } + } +} + +// Closing the connection must stop the background reader and fail pending reads. +func TestReadAheadConnClose(t *testing.T) { + t.Parallel() + + client, server := net.Pipe() + defer server.Close() + + conn := newReadAheadConn(client) + + readErr := make(chan error, 1) + + go func() { + _, err := conn.Read(make([]byte, 1)) + readErr <- err + }() + + require.NoError(t, conn.Close()) + + select { + case err := <-readErr: + require.ErrorIs(t, err, io.ErrClosedPipe) + case <-time.After(5 * time.Second): + t.Fatal("pending read was not unblocked by Close") + } +} diff --git a/internal/headers/cri/types.go b/internal/headers/cri/types.go new file mode 100644 index 00000000..8b7d088f --- /dev/null +++ b/internal/headers/cri/types.go @@ -0,0 +1,251 @@ +// 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 cri + +import ( + "google.golang.org/protobuf/encoding/protowire" +) + +// ContainerState is the runtime.v1.ContainerState enum. +type ContainerState int32 + +const ( + ContainerCreated ContainerState = 0 + ContainerRunning ContainerState = 1 + ContainerExited ContainerState = 2 + ContainerUnknown ContainerState = 3 +) + +// PodSandboxState is the runtime.v1.PodSandboxState enum. +type PodSandboxState int32 + +const ( + SandboxReady PodSandboxState = 0 + SandboxNotReady PodSandboxState = 1 +) + +// Version is the subset of runtime.v1.VersionResponse used by the exporter. +type Version struct { + // RuntimeName is the runtime name, e.g. containerd. Kubernetes uses it as container ID prefix. + RuntimeName string + RuntimeVersion string + RuntimeAPIVersion string +} + +// Container is the subset of runtime.v1.Container used by the exporter. +type Container struct { + ID string + PodSandboxID string + // Name is the container name from the Kubernetes pod spec. + Name string + State ContainerState +} + +// PodSandbox is the subset of runtime.v1.PodSandbox used by the exporter. +type PodSandbox struct { + ID string + Name string + UID string + Namespace string + State PodSandboxState +} + +// field is a decoded protobuf field. value holds varints, bytes length-delimited values. +type field struct { + num protowire.Number + typ protowire.Type + value uint64 + bytes []byte +} + +func (f field) is(num protowire.Number, typ protowire.Type) bool { + return f.num == num && f.typ == typ +} + +// eachField calls fn for each field of the protobuf message b. +func eachField(b []byte, fn func(field) error) error { + for len(b) > 0 { + num, typ, n := protowire.ConsumeTag(b) + if n < 0 { + return protowire.ParseError(n) + } + + b = b[n:] + f := field{num: num, typ: typ} + + switch typ { + case protowire.VarintType: + f.value, n = protowire.ConsumeVarint(b) + case protowire.BytesType: + f.bytes, n = protowire.ConsumeBytes(b) + default: + n = protowire.ConsumeFieldValue(num, typ, b) + } + + if n < 0 { + return protowire.ParseError(n) + } + + b = b[n:] + + if err := fn(f); err != nil { + return err + } + } + + return nil +} + +func decodeVersion(b []byte) (Version, error) { + var version Version + + err := eachField(b, func(f field) error { + switch { + case f.is(2, protowire.BytesType): + version.RuntimeName = string(f.bytes) + case f.is(3, protowire.BytesType): + version.RuntimeVersion = string(f.bytes) + case f.is(4, protowire.BytesType): + version.RuntimeAPIVersion = string(f.bytes) + } + + return nil + }) + + return version, err +} + +func decodeListContainersResponse(b []byte) ([]Container, error) { + var containers []Container + + err := eachField(b, func(f field) error { + if !f.is(1, protowire.BytesType) { + return nil + } + + container, err := decodeContainer(f.bytes) + if err != nil { + return err + } + + containers = append(containers, container) + + return nil + }) + + return containers, err +} + +func decodeContainer(b []byte) (Container, error) { + var container Container + + err := eachField(b, func(f field) error { + switch { + case f.is(1, protowire.BytesType): + container.ID = string(f.bytes) + case f.is(2, protowire.BytesType): + container.PodSandboxID = string(f.bytes) + case f.is(3, protowire.BytesType): // ContainerMetadata + return eachField(f.bytes, func(f field) error { + if f.is(1, protowire.BytesType) { + container.Name = string(f.bytes) + } + + return nil + }) + case f.is(6, protowire.VarintType): + container.State = ContainerState(f.value) + } + + return nil + }) + + return container, err +} + +func decodeListPodSandboxResponse(b []byte) ([]PodSandbox, error) { + var sandboxes []PodSandbox + + err := eachField(b, func(f field) error { + if !f.is(1, protowire.BytesType) { + return nil + } + + sandbox, err := decodePodSandbox(f.bytes) + if err != nil { + return err + } + + sandboxes = append(sandboxes, sandbox) + + return nil + }) + + return sandboxes, err +} + +func decodePodSandbox(b []byte) (PodSandbox, error) { + var sandbox PodSandbox + + err := eachField(b, func(f field) error { + switch { + case f.is(1, protowire.BytesType): + sandbox.ID = string(f.bytes) + case f.is(2, protowire.BytesType): // PodSandboxMetadata + return eachField(f.bytes, func(f field) error { + switch { + case f.is(1, protowire.BytesType): + sandbox.Name = string(f.bytes) + case f.is(2, protowire.BytesType): + sandbox.UID = string(f.bytes) + case f.is(3, protowire.BytesType): + sandbox.Namespace = string(f.bytes) + } + + return nil + }) + case f.is(3, protowire.VarintType): + sandbox.State = PodSandboxState(f.value) + } + + return nil + }) + + return sandbox, err +} + +// encodeListRunningContainersRequest encodes a ListContainersRequest +// with filter.state.state set to CONTAINER_RUNNING. +func encodeListRunningContainersRequest() []byte { + var state []byte // ContainerStateValue + + state = protowire.AppendTag(state, 1, protowire.VarintType) + state = protowire.AppendVarint(state, uint64(ContainerRunning)) + + var filter []byte // ContainerFilter + + filter = protowire.AppendTag(filter, 2, protowire.BytesType) + filter = protowire.AppendBytes(filter, state) + + var req []byte // ListContainersRequest + + req = protowire.AppendTag(req, 1, protowire.BytesType) + req = protowire.AppendBytes(req, filter) + + return req +}