REP-1913: Allow Kvisor to collect storage metrics regarding CRI used in the cluster (#629)

Add CRI-O support with --disable-containerd flag for optional containerd features (process tree, image digest resolution). Includes storage stats collection.

Author: krystiancastai

charts/kvisor/templates/agent.yaml

@@ -66,6 +66,7 @@ spec:
           args:
             - "run"
             - "--metrics-http-listen-port={{.Values.agent.metricsHTTPListenPort}}"
+            - "--disable-containerd={{ .Values.agent.disableContainerd }}"
           {{- if eq .Values.mockServer.enabled true }}
             - "--castai-server-insecure=true"
           {{- end }}

charts/kvisor/values.yaml

@@ -145,9 +145,13 @@ agent:
   # Additional environment variables for the agent container via configMaps or secrets.
   envFrom: []
 
-  # Custom socket path for containerd.
+  # Path to the container runtime socket (containerd or CRI-O)
   containerdSocketPath: "/run/containerd/containerd.sock"
 
+  # Disable containerd-specific features (process tree, image digest resolution).
+  # Set to true for CRI-O clusters.
+  disableContainerd: false
+
 controller:
   enabled: true
 

cmd/agent/daemon/app/app.go

@@ -160,13 +160,17 @@ func (a *App) Run(ctx context.Context) error {
 		return err
 	}
 
+	if cfg.DisableContainerd {
+		disableFeaturesRequiringContainerd(cfg, log)
+	}
+
 	criClient, criCloseFn, err := cri.NewRuntimeClient(ctx, cfg.CRIEndpoint)
 	if err != nil {
 		return fmt.Errorf("new CRI runtime client: %w", err)
 	}
 	defer criCloseFn() //nolint:errcheck
 
-	containersClient, err := containers.NewClient(log, cgroupClient, cfg.ContainerdSockPath, procHandler, criClient, cfg.EventLabels, cfg.EventAnnotations)
+	containersClient, err := containers.NewClient(log, cgroupClient, cfg.ContainerdSockPath, cfg.DisableContainerd, procHandler, criClient, cfg.EventLabels, cfg.EventAnnotations)
 	if err != nil {
 		return err
 	}
@@ -400,6 +404,13 @@ func enableBPFStats(cfg *config.Config, log *logging.Logger) func() {
 	return cleanup
 }
 
+func disableFeaturesRequiringContainerd(cfg *config.Config, log *logging.Logger) {
+	if cfg.ProcessTree.Enabled {
+		log.Warn("process tree requires containerd, disabling because containerd is disabled")
+		cfg.ProcessTree.Enabled = false
+	}
+}
+
 func buildEBPFPolicy(log *logging.Logger, cfg *config.Config, signatureEngine *signature.SignatureEngine) *ebpftracer.Policy {
 	// TODO: Allow to build these policies on the fly from the control plane. Ideally we should be able to disable, enable policies and change rate limits dynamically.
 	policy := &ebpftracer.Policy{

cmd/agent/daemon/config/config.go

@@ -23,6 +23,7 @@ type Config struct {
 	PromMetricsExportInterval      time.Duration                   `json:"promMetricsExportInterval"`
 	Version                        string                          `json:"version"`
 	BTFPath                        string                          `json:"BTFPath"`
+	DisableContainerd              bool                            `json:"disableContainerd"`
 	ContainerdSockPath             string                          `json:"containerdSockPath"`
 	HostCgroupsDir                 string                          `json:"hostCgroupsDir"`
 	MetricsHTTPListenPort          int                             `json:"metricsHTTPListenPort"`

cmd/agent/daemon/daemon.go

@@ -36,6 +36,7 @@ func NewRunCommand(version string) *cobra.Command {
 		promMetricsExportInterval = command.Flags().Duration("prom-metrics-export-interval", 5*time.Minute, "Internal prometheus metrics export interval")
 
 		sendLogLevel          = command.Flags().String("send-logs-level", slog.LevelInfo.String(), "send logs level")
+		disableContainerd     = command.Flags().Bool("disable-containerd", false, "Disable containerd-specific features (process tree, image digest). Enable for non-containerd runtimes.")
 		containerdSockPath    = command.Flags().String("containerd-sock", "/run/containerd/containerd.sock", "Path to containerd socket file")
 		metricsHTTPListenPort = command.Flags().Int("metrics-http-listen-port", 6060, "metrics http listen port")
 		hostCgroupsDir        = command.Flags().String("host-cgroups", "/cgroups", "Host /sys/fs/cgroups directory name mounted to container")
@@ -150,6 +151,7 @@ func NewRunCommand(version string) *cobra.Command {
 			PromMetricsExportInterval: *promMetricsExportInterval,
 			Version:                   version,
 			BTFPath:                   *btfPath,
+			DisableContainerd:         *disableContainerd,
 			ContainerdSockPath:        *containerdSockPath,
 			HostCgroupsDir:            *hostCgroupsDir,
 			MetricsHTTPListenPort:     *metricsHTTPListenPort,

cmd/agent/daemon/debug/netflows.go

@@ -29,6 +29,7 @@ func NewNetflowsDebugCommand() *cobra.Command {
 	criEndpoint := cmd.Flags().String("cri-endpoint", "unix:///run/containerd/containerd.sock", "CRI endpoint")
 	hostCgroupsDir := cmd.Flags().String("host-cgroups", "/cgroups", "Host /sys/fs/cgroups directory name mounted to container")
 	containerdSockPath := cmd.Flags().String("containerd-sock", "/run/containerd/containerd.sock", "Path to containerd socket file")
+	disableContainerd := cmd.Flags().Bool("disable-containerd", true, "Disable containerd-specific features")
 	btfPath := cmd.Flags().String("btf-path", "/sys/kernel/btf/vmlinux", "btf file path")
 	waitDuration := cmd.Flags().Duration("wait", 2*time.Second, "Wait duration before scraping netflows")
 	limit := cmd.Flags().Int("limit", 500, "Limit netflows output")
@@ -57,7 +58,7 @@ func NewNetflowsDebugCommand() *cobra.Command {
 			return err
 		}
 
-		containersClient, err := containers.NewClient(log, cgroupClient, *containerdSockPath, procHandler, criClient, []string{}, []string{})
+		containersClient, err := containers.NewClient(log, cgroupClient, *containerdSockPath, *disableContainerd, procHandler, criClient, []string{}, []string{})
 		if err != nil {
 			return err
 		}

pkg/containers/client.go

@@ -12,10 +12,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/castai/kvisor/cmd/agent/daemon/metrics"
-	"github.com/castai/kvisor/pkg/cgroup"
-	"github.com/castai/kvisor/pkg/logging"
-	"github.com/castai/kvisor/pkg/proc"
 	"github.com/containerd/containerd"
 	"github.com/containerd/containerd/api/services/tasks/v1"
 	"github.com/containerd/containerd/content"
@@ -25,6 +21,11 @@ import (
 	"google.golang.org/grpc/backoff"
 	"google.golang.org/grpc/credentials/insecure"
 	criapi "k8s.io/cri-api/pkg/apis/runtime/v1"
+
+	"github.com/castai/kvisor/cmd/agent/daemon/metrics"
+	"github.com/castai/kvisor/pkg/cgroup"
+	"github.com/castai/kvisor/pkg/logging"
+	"github.com/castai/kvisor/pkg/proc"
 )
 
 var (
@@ -82,6 +83,7 @@ type Client struct {
 	cgroupClient                cgroupsClient
 	criRuntimeServiceClient     criClient
 	containerContentStoreClient containerContentStoreClient
+	containerdDisabled          bool
 
 	containerCreatedListeners []ContainerCreatedListener
 	containerDeletedListeners []ContainerDeletedListener
@@ -98,9 +100,27 @@ type Client struct {
 	inactiveContainersDuration time.Duration
 }
 
-func NewClient(log *logging.Logger, cgroupClient *cgroup.Client, containerdSock string, procHandler *proc.Proc, criRuntimeServiceClient criapi.RuntimeServiceClient,
+func NewClient(log *logging.Logger, cgroupClient *cgroup.Client, containerdSock string, disableContainerd bool, procHandler *proc.Proc, criRuntimeServiceClient criapi.RuntimeServiceClient,
 	labels, annotations []string) (*Client, error) {
 
+	client := &Client{
+		log:                        log.WithField("component", "cgroups"),
+		cgroupClient:               cgroupClient,
+		containersByCgroup:         map[uint64]*Container{},
+		containersByID:             map[string]*Container{},
+		procHandler:                procHandler,
+		criRuntimeServiceClient:    criRuntimeServiceClient,
+		forwardedLabels:            labels,
+		forwardedAnnotations:       annotations,
+		inactiveContainersDuration: 2 * time.Minute,
+		containerdDisabled:         disableContainerd,
+	}
+
+	if disableContainerd {
+		log.Info("containerd features disabled")
+		return client, nil
+	}
+
 	backoffConfig := backoff.DefaultConfig
 	backoffConfig.MaxDelay = 3 * time.Second
 	connParams := grpc.ConnectParams{
@@ -128,23 +148,17 @@ func NewClient(log *logging.Logger, cgroupClient *cgroup.Client, containerdSock
 		return nil, fmt.Errorf("failed connecting to containerd client: %w", err)
 	}
 
-	return &Client{
-		log:                         log.WithField("component", "cgroups"),
-		containerdClient:            containerdClient,
-		containerContentStoreClient: containerdClient.ContentStore(),
-		cgroupClient:                cgroupClient,
-		containersByCgroup:          map[uint64]*Container{},
-		containersByID:              map[string]*Container{},
-		procHandler:                 procHandler,
-		criRuntimeServiceClient:     criRuntimeServiceClient,
-		forwardedLabels:             labels,
-		forwardedAnnotations:        annotations,
-		inactiveContainersDuration:  2 * time.Minute,
-	}, nil
+	client.containerdClient = containerdClient
+	client.containerContentStoreClient = containerdClient.ContentStore()
+
+	return client, nil
 }
 
 func (c *Client) Close() error {
-	return c.containerdClient.Close()
+	if c.containerdClient != nil {
+		return c.containerdClient.Close()
+	}
+	return nil
 }
 
 type ContainerProcess struct {
@@ -153,6 +167,10 @@ type ContainerProcess struct {
 }
 
 func (c *Client) LoadContainerTasks(ctx context.Context) ([]ContainerProcess, error) {
+	if c.containerdClient == nil {
+		return nil, nil
+	}
+
 	resp, err := c.containerdClient.TaskService().List(ctx, nil)
 	if err != nil {
 		return nil, err
@@ -324,11 +342,13 @@ func (c *Client) addContainerWithCgroup(container *criapi.Container, cg *cgroup.
 	}
 	cont.markAccessed()
 
-	imageDigest, err := c.findImageDigest(container)
-	if err != nil {
-		c.log.Warnf("finding image digest for container %v: %v", container.Id, err)
+	if !c.containerdDisabled {
+		imageDigest, err := c.findImageDigest(container)
+		if err != nil {
+			c.log.Warnf("finding image digest for container %v: %v", container.Id, err)
+		}
+		cont.ImageDigest = imageDigest
 	}
-	cont.ImageDigest = imageDigest
 
 	getSandboxCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
 	defer cancel()

pkg/containers/client_test.go

@@ -261,6 +261,7 @@ func newTestClient() *Client {
 		inactiveContainersDuration:  1 * time.Minute,
 		containerdClient:            &mockContainerdClient{},
 		containerContentStoreClient: &mockContainerContentStoreClient{},
+		containerdDisabled:          false,
 	}
 }