mirror of
https://github.com/prometheus-community/windows_exporter.git
synced 2026-10-17 10:09:05 +02:00
container: read Kubernetes metadata from the CRI API (#2557)
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Jan-Otto Kröpke <mail@jkroepke.de>
This commit is contained in:
Jan-Otto Kröpke
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
5c148173e7
commit
864088b034
@@ -1,2 +1,2 @@
|
||||
[codespell]
|
||||
ignore-words-list = notin,entires
|
||||
ignore-words-list = notin,entires,te
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
|
||||
@@ -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"})
|
||||
}
|
||||
|
||||
@@ -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}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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/<name>", 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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user