feat: adding kata integration

comp/core/workloadmeta/collectors/internal/containerd/container_builder.go

@@ -128,6 +128,7 @@ func buildWorkloadMetaContainer(namespace string, container containerd.Container
 		Ports:         nil, // Not available
 		Runtime:       workloadmeta.ContainerRuntimeContainerd,
 		RuntimeFlavor: runtimeFlavor,
+		SandboxID:     info.SandboxID,
 		State: workloadmeta.ContainerState{
 			Running:    status == containerd.Running,
 			Status:     extractStatus(status),

comp/core/workloadmeta/collectors/internal/crio/containers.go

@@ -44,11 +44,12 @@ func (c *collector) convertContainerToEvent(ctx context.Context, ctr *v1.Contain
 				Labels:      ctr.GetLabels(),
 				Annotations: ctr.GetAnnotations(),
 			},
-			Hostname: hostname,
-			Image:    image,
-			PID:      pid,
-			Ports:    ports,
-			Runtime:  workloadmeta.ContainerRuntimeCRIO,
+			Hostname:  hostname,
+			Image:     image,
+			PID:       pid,
+			Ports:     ports,
+			Runtime:   workloadmeta.ContainerRuntimeCRIO,
+			SandboxID: ctr.GetPodSandboxId(),
 			State:    getContainerState(containerStatus),
 			Resources: workloadmeta.ContainerResources{
 				CPULimit:    cpuLimit,

comp/core/workloadmeta/collectors/internal/podman/podman.go

@@ -204,8 +204,9 @@ func convertToEvent(container *podman.Container) workloadmeta.CollectorEvent {
 			Image:      image,
 			NetworkIPs: networkIPs(container),
 			PID:        container.State.PID,
-			Ports:      ports,
-			Runtime:    workloadmeta.ContainerRuntimePodman,
+			Ports:     ports,
+			Runtime:   workloadmeta.ContainerRuntimePodman,
+			SandboxID: container.Config.Pod,
 			State: workloadmeta.ContainerState{
 				Running:    container.State.State == podman.ContainerStateRunning,
 				Status:     status(container.State.State),

comp/core/workloadmeta/def/types.go

@@ -654,8 +654,11 @@ type Container struct {
 	// CgroupPath is a path to the cgroup of the container.
 	// It can be relative to the cgroup parent.
 	// Linux only.
-	CgroupPath   string
-	RestartCount int
+	CgroupPath string
+	// SandboxID is the identifier of the sandbox this container belongs to.
+	// Populated from containerd's container info SandboxID field.
+	SandboxID string
+	RestartCount  int
 }
 
 // GetID implements Entity#GetID.

pkg/collector/corechecks/containers/kata/check.go

@@ -0,0 +1,334 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+//go:build linux
+
+// Package kata implements the kata_containers check.
+package kata
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"maps"
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"time"
+
+	yaml "go.yaml.in/yaml/v2"
+
+	"github.com/DataDog/datadog-agent/comp/core/autodiscovery/integration"
+	tagger "github.com/DataDog/datadog-agent/comp/core/tagger/def"
+	taggertypes "github.com/DataDog/datadog-agent/comp/core/tagger/types"
+	workloadmeta "github.com/DataDog/datadog-agent/comp/core/workloadmeta/def"
+	"github.com/DataDog/datadog-agent/pkg/aggregator/sender"
+	"github.com/DataDog/datadog-agent/pkg/collector/check"
+	core "github.com/DataDog/datadog-agent/pkg/collector/corechecks"
+	"github.com/DataDog/datadog-agent/pkg/metrics/servicecheck"
+	"github.com/DataDog/datadog-agent/pkg/util/option"
+	prom "github.com/DataDog/datadog-agent/pkg/util/prometheus"
+)
+
+const (
+	// CheckName is the name of the check
+	CheckName = "kata_containers"
+
+	shimSocket            = "shim-monitor.sock"
+	shimDialTimeout       = 5 * time.Second
+	shimReadTimeout       = 10 * time.Second
+	defaultScrapeInterval = 15 * time.Second
+)
+
+var defaultSandboxStoragePaths = []string{"/run/vc/sbs", "/run/kata"}
+var defaultRenameLabels = map[string]string{"version": "go_version"}
+
+// KataConfig holds the check instance configuration
+type KataConfig struct {
+	SandboxStoragePaths []string          `yaml:"sandbox_storage_paths"`
+	RenameLabels        map[string]string `yaml:"rename_labels"`
+	ExcludeLabels       []string          `yaml:"exclude_labels"`
+	Tags                []string          `yaml:"tags"`
+}
+
+// KataCheck collects metrics from Kata Containers sandboxes
+type KataCheck struct {
+	core.CheckBase
+	instance   *KataConfig
+	tagger     tagger.Component
+	store      workloadmeta.Component
+	excludeSet map[string]struct{} // built once at Configure time
+
+	mu                 sync.RWMutex
+	sandboxContainerID map[string]string // sandboxID -> containerID, updated by workloadmeta events
+
+	stopOnce sync.Once
+	stopCh   chan struct{}
+}
+
+// Factory returns a check factory for kata_containers
+func Factory(store workloadmeta.Component, tagger tagger.Component) option.Option[func() check.Check] {
+	return option.New(func() check.Check {
+		return core.NewLongRunningCheckWrapper(&KataCheck{
+			CheckBase:          core.NewCheckBase(CheckName),
+			instance:           &KataConfig{},
+			tagger:             tagger,
+			store:              store,
+			sandboxContainerID: make(map[string]string),
+			stopCh:             make(chan struct{}),
+		})
+	})
+}
+
+// Parse parses the KataConfig and sets defaults
+func (c *KataConfig) Parse(data []byte) error {
+	c.SandboxStoragePaths = defaultSandboxStoragePaths
+	c.RenameLabels = maps.Clone(defaultRenameLabels)
+
+	if err := yaml.Unmarshal(data, c); err != nil {
+		return err
+	}
+
+	if len(c.SandboxStoragePaths) == 0 {
+		c.SandboxStoragePaths = defaultSandboxStoragePaths
+	}
+	if c.RenameLabels == nil {
+		c.RenameLabels = maps.Clone(defaultRenameLabels)
+	}
+
+	return nil
+}
+
+// Configure parses the check configuration
+func (c *KataCheck) Configure(senderManager sender.SenderManager, _ uint64, config, initConfig integration.Data, source string) error {
+	if err := c.CommonConfigure(senderManager, initConfig, config, source); err != nil {
+		return err
+	}
+	if err := c.instance.Parse(config); err != nil {
+		return err
+	}
+
+	c.excludeSet = make(map[string]struct{}, len(c.instance.ExcludeLabels))
+	for _, l := range c.instance.ExcludeLabels {
+		c.excludeSet[l] = struct{}{}
+	}
+
+	return nil
+}
+
+// Run is the long-running event loop: it subscribes to workloadmeta container
+// events to maintain a sandboxID→containerID cache, and periodically scrapes
+// all discovered Kata shim sockets.
+func (c *KataCheck) Run() error {
+	filter := workloadmeta.NewFilterBuilder().
+		AddKind(workloadmeta.KindContainer).
+		Build()
+	containerEventsCh := c.store.Subscribe(CheckName, workloadmeta.NormalPriority, filter)
+	defer c.store.Unsubscribe(containerEventsCh)
+
+	ticker := time.NewTicker(defaultScrapeInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case eventBundle, ok := <-containerEventsCh:
+			if !ok {
+				return nil
+			}
+			c.processContainerEvents(eventBundle)
+		case <-ticker.C:
+			c.runScrape()
+		case <-c.stopCh:
+			return nil
+		}
+	}
+}
+
+// Stop signals the Run loop to exit.
+func (c *KataCheck) Stop() {
+	c.stopOnce.Do(func() { close(c.stopCh) })
+}
+
+// processContainerEvents updates the sandboxContainerID cache from a workloadmeta event bundle.
+func (c *KataCheck) processContainerEvents(eventBundle workloadmeta.EventBundle) {
+	defer eventBundle.Acknowledge()
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	for _, event := range eventBundle.Events {
+		ctr, ok := event.Entity.(*workloadmeta.Container)
+		if !ok || ctr.SandboxID == "" {
+			continue
+		}
+		switch event.Type {
+		case workloadmeta.EventTypeSet:
+			c.sandboxContainerID[ctr.SandboxID] = ctr.ID
+		case workloadmeta.EventTypeUnset:
+			delete(c.sandboxContainerID, ctr.SandboxID)
+		}
+	}
+}
+
+// runScrape discovers sandboxes and scrapes each one.
+func (c *KataCheck) runScrape() {
+	s, err := c.GetSender()
+	if err != nil {
+		_ = c.Warnf("kata_containers: failed to get sender: %v", err)
+		return
+	}
+	defer s.Commit()
+
+	sandboxes := c.discoverSandboxes()
+
+	s.Gauge("kata.running_shim_count", float64(len(sandboxes)), "", c.instance.Tags)
+
+	for sandboxID, socketPath := range sandboxes {
+		baseTags := c.buildBaseTags(sandboxID)
+		c.scrapeSandbox(s, sandboxID, socketPath, baseTags)
+	}
+}
+
+// buildBaseTags returns sandbox_id tag plus any orchestrator tags from the tagger.
+func (c *KataCheck) buildBaseTags(sandboxID string) []string {
+	tags := []string{"sandbox_id:" + sandboxID}
+
+	c.mu.RLock()
+	containerID, ok := c.sandboxContainerID[sandboxID]
+	c.mu.RUnlock()
+
+	if ok {
+		entityID := taggertypes.NewEntityID(taggertypes.ContainerID, containerID)
+		if taggerTags, err := c.tagger.Tag(entityID, taggertypes.OrchestratorCardinality); err == nil {
+			tags = append(tags, taggerTags...)
+		}
+	}
+
+	return tags
+}
+
+// discoverSandboxes scans sandbox storage paths and returns a map of sandboxID → socketPath
+func (c *KataCheck) discoverSandboxes() map[string]string {
+	sandboxes := make(map[string]string)
+
+	for _, basePath := range c.instance.SandboxStoragePaths {
+		if _, err := os.Stat(basePath); os.IsNotExist(err) {
+			continue
+		}
+
+		entries, err := os.ReadDir(basePath)
+		if err != nil {
+			_ = c.Warnf("kata_containers: failed to read directory %s: %v", basePath, err)
+			continue
+		}
+
+		for _, entry := range entries {
+			if !entry.IsDir() {
+				continue
+			}
+			socketPath := filepath.Join(basePath, entry.Name(), shimSocket)
+			if _, err := os.Stat(socketPath); err == nil {
+				sandboxes[entry.Name()] = socketPath
+			}
+		}
+	}
+
+	return sandboxes
+}
+
+// scrapeSandbox scrapes Prometheus metrics from a single sandbox's shim socket.
+// It dials the unix socket directly and issues a raw HTTP/1.1 GET.
+// https://github.com/kata-containers/kata-containers/blob/main/docs/design/kata-2-0-metrics.md#metrics-architecture
+func (c *KataCheck) scrapeSandbox(s sender.Sender, sandboxID, socketPath string, baseTags []string) {
+	conn, err := net.DialTimeout("unix", socketPath, shimDialTimeout)
+	if err != nil {
+		s.ServiceCheck("kata.openmetrics.health", servicecheck.ServiceCheckCritical, "", baseTags,
+			fmt.Sprintf("failed to connect to sandbox %s: %v", sandboxID, err))
+		return
+	}
+	defer conn.Close()
+	conn.SetDeadline(time.Now().Add(shimReadTimeout)) //nolint:errcheck
+
+	fmt.Fprintf(conn, "GET /metrics HTTP/1.0\r\nHost: local\r\n\r\n") //nolint:errcheck
+
+	resp, err := http.ReadResponse(bufio.NewReader(conn), nil)
+	if err != nil {
+		s.ServiceCheck("kata.openmetrics.health", servicecheck.ServiceCheckCritical, "", baseTags,
+			fmt.Sprintf("failed to read response from sandbox %s: %v", sandboxID, err))
+		return
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		s.ServiceCheck("kata.openmetrics.health", servicecheck.ServiceCheckCritical, "", baseTags,
+			fmt.Sprintf("failed to read body from sandbox %s: %v", sandboxID, err))
+		return
+	}
+
+	families, err := prom.ParseMetrics(body)
+	if err != nil {
+		s.ServiceCheck("kata.openmetrics.health", servicecheck.ServiceCheckCritical, "", baseTags,
+			fmt.Sprintf("failed to parse metrics from sandbox %s: %v", sandboxID, err))
+		return
+	}
+
+	for _, family := range families {
+		for _, sample := range family.Samples {
+			rawName := sample.Metric["__name__"]
+			if rawName == "" {
+				rawName = family.Name
+			}
+			metricName := formatMetricName(rawName)
+			tags := c.buildSampleTags(baseTags, sample.Metric)
+
+			switch strings.ToUpper(family.Type) {
+			case "COUNTER":
+				s.Rate(metricName, sample.Value, "", tags)
+			default: // GAUGE, HISTOGRAM, SUMMARY, UNTYPED
+				s.Gauge(metricName, sample.Value, "", tags)
+			}
+		}
+	}
+
+	s.ServiceCheck("kata.openmetrics.health", servicecheck.ServiceCheckOK, "", baseTags, "")
+}
+
+// formatMetricName converts a raw Prometheus metric name to a Datadog metric name.
+// "kata_hypervisor_fds" -> "kata.hypervisor.fds"
+func formatMetricName(rawName string) string {
+	name := rawName
+	if after, ok := strings.CutPrefix(name, "kata_"); ok {
+		name = after
+	}
+	name = strings.ReplaceAll(name, "_", ".")
+	return "kata." + name
+}
+
+// buildSampleTags builds the full tag list for a metric sample from pre-resolved baseTags.
+func (c *KataCheck) buildSampleTags(baseTags []string, metric prom.Metric) []string {
+	tags := make([]string, len(baseTags), len(baseTags)+len(metric)+len(c.instance.Tags))
+	copy(tags, baseTags)
+	tags = append(tags, c.instance.Tags...)
+
+	for k, v := range metric {
+		if k == "__name__" {
+			continue
+		}
+		if _, excluded := c.excludeSet[k]; excluded {
+			continue
+		}
+		labelName := k
+		if renamed, ok := c.instance.RenameLabels[k]; ok {
+			labelName = renamed
+		}
+		tags = append(tags, labelName+":"+v)
+	}
+
+	return tags
+}