support disk usage metrics for containerd factory

build/build.sh

@@ -61,7 +61,7 @@ fi
 
 # Since github.com/google/cadvisor/cmd is a submodule, we must build from inside that directory
 pushd cmd > /dev/null
-  go build ${GO_FLAGS} -ldflags "${ldflags}" -o "${output_file}" "${repo_path}/cmd"
+  go mod tidy && go build ${GO_FLAGS} -ldflags "${ldflags}" -o "${output_file}" "${repo_path}/cmd"
 popd > /dev/null
 
 exit 0

cmd/go.mod

@@ -128,4 +128,5 @@ require (
 	google.golang.org/grpc v1.72.2 // indirect
 	google.golang.org/protobuf v1.36.6 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
+	k8s.io/cri-api v0.34.3 // indirect
 )

cmd/go.sum

@@ -413,6 +413,8 @@ gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gotest.tools/v3 v3.0.2 h1:kG1BFyqVHuQoVQiR1bWGnfz/fmHvvuiSPIV7rvl360E=
 gotest.tools/v3 v3.0.2/go.mod h1:3SzNCllyD9/Y+b5r9JIKQ474KzkZyqLqEfYqMsX94Bk=
+k8s.io/cri-api v0.34.3 h1:zFdQSHZuQlQXesw9ncjQRUyDpvLng/84Q4qLKd8x2zE=
+k8s.io/cri-api v0.34.3/go.mod h1:4qVUjidMg7/Z9YGZpqIDygbkPWkg3mkS1PvOx/kpHTE=
 k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk=
 k8s.io/klog/v2 v2.130.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
 k8s.io/utils v0.0.0-20250502105355-0f33e8f1c979 h1:jgJW5IePPXLGB8e/1wvd0Ich9QE97RvvF3a8J3fP/Lg=

container/containerd/client.go

@@ -16,9 +16,12 @@ package containerd
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"net"
+	"path"
+	"strings"
 	"sync"
 	"time"
 
@@ -31,6 +34,9 @@ import (
 	"google.golang.org/grpc/backoff"
 	"google.golang.org/grpc/credentials/insecure"
 	emptypb "google.golang.org/protobuf/types/known/emptypb"
+	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
+
+	"github.com/google/cadvisor/container/containerd/config"
 
 	"github.com/google/cadvisor/container/containerd/containers"
 	"github.com/google/cadvisor/container/containerd/pkg/dialer"
@@ -40,6 +46,7 @@ type client struct {
 	containerService containersapi.ContainersClient
 	taskService      tasksapi.TasksClient
 	versionService   versionapi.VersionClient
+	runtimeService   runtimeapi.RuntimeServiceClient
 }
 
 type ContainerdClient interface {
@@ -48,12 +55,18 @@ type ContainerdClient interface {
 	LoadTaskProcess(ctx context.Context, id string) (*tasktypes.Process, error)
 	TaskExitStatus(ctx context.Context, id string) (uint32, error)
 	Version(ctx context.Context) (string, error)
+	RootfsDir(ctx context.Context) (string, error)
 }
 
 var (
 	ErrTaskIsInUnknownState = errors.New("containerd task is in unknown state") // used when process reported in containerd task is in Unknown State
 )
 
+var (
+	statePlugin       = "io.containerd.grpc.v1.cri"
+	taskRuntimePlugin = "io.containerd.runtime.v2.task"
+)
+
 var once sync.Once
 var ctrdClient ContainerdClient = nil
 var ctrdClientErr error = nil
@@ -106,6 +119,7 @@ func Client(address, namespace string) (ContainerdClient, error) {
 			containerService: containersapi.NewContainersClient(conn),
 			taskService:      tasksapi.NewTasksClient(conn),
 			versionService:   versionapi.NewVersionClient(conn),
+			runtimeService:   runtimeapi.NewRuntimeServiceClient(conn),
 		}
 	})
 	return ctrdClient, ctrdClientErr
@@ -166,6 +180,19 @@ func (c *client) Version(ctx context.Context) (string, error) {
 	return response.Version, nil
 }
 
+func (c *client) RootfsDir(ctx context.Context) (string, error) {
+	r, err := c.runtimeService.Status(ctx, &runtimeapi.StatusRequest{Verbose: true})
+	if err != nil {
+		return "", err
+	}
+	configStr := r.Info["config"]
+	config := config.Config{}
+	if err := json.Unmarshal([]byte(configStr), &config); err != nil {
+		return "", err
+	}
+	return path.Join(strings.TrimSuffix(config.StateDir, statePlugin), taskRuntimePlugin), nil
+}
+
 func containerFromProto(containerpb *containersapi.Container) *containers.Container {
 	var runtime containers.RuntimeInfo
 	// TODO: is nil check required for containerpb

container/containerd/client_test.go

@@ -66,6 +66,10 @@ func (c *containerdClientMock) TaskExitStatus(ctx context.Context, id string) (u
 	return c.exitStatus, nil
 }
 
+func (c *containerdClientMock) RootfsDir(ctx context.Context) (string, error) {
+	return "/run/containerd/io.containerd.runtime.v2.task", nil
+}
+
 func mockcontainerdClient(cntrs map[string]*containers.Container, returnErr error) ContainerdClient {
 	tasks := make(map[string]*task.Process)
 

container/containerd/factory.go

@@ -64,6 +64,13 @@ func (f *containerdFactory) NewContainerHandler(name string, metadataEnvAllowLis
 		return
 	}
 
+	ctx, cancel := context.WithTimeout(context.Background(), connectionTimeout)
+	defer cancel()
+	rootfsDir, err := client.RootfsDir(ctx)
+	if err != nil {
+		klog.Warningf("unable to get containerd rootfs dir, err: %v, continue register", err)
+	}
+
 	containerdMetadataEnvAllowList := strings.Split(*containerdEnvMetadataWhiteList, ",")
 
 	// prefer using the unified metadataEnvAllowList
@@ -80,6 +87,7 @@ func (f *containerdFactory) NewContainerHandler(name string, metadataEnvAllowLis
 		inHostNamespace,
 		containerdMetadataEnvAllowList,
 		f.includedMetrics,
+		rootfsDir,
 	)
 }
 

container/containerd/handler.go

@@ -20,6 +20,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"path"
 	"strings"
 	"time"
 
@@ -38,8 +39,10 @@ type containerdContainerHandler struct {
 	machineInfoFactory info.MachineInfoFactory
 	// Absolute path to the cgroup hierarchies of this container.
 	// (e.g.: "cpu" -> "/sys/fs/cgroup/cpu/test")
-	cgroupPaths map[string]string
-	fsInfo      fs.FsInfo
+	cgroupPaths      map[string]string
+	fsInfo           fs.FsInfo
+	fsHandler        common.FsHandler
+	rootfsStorageDir string
 	// Metadata associated with the container.
 	reference info.ContainerReference
 	envs      map[string]string
@@ -65,6 +68,7 @@ func newContainerdContainerHandler(
 	inHostNamespace bool,
 	metadataEnvAllowList []string,
 	includedMetrics container.MetricSet,
+	rootfsDir string,
 ) (container.ContainerHandler, error) {
 	// Create the cgroup paths.
 	cgroupPaths := common.MakeCgroupPaths(cgroupSubsystems, name)
@@ -148,6 +152,10 @@ func newContainerdContainerHandler(
 	}
 	// Add the name and bare ID as aliases of the container.
 	handler.image = cntr.Image
+	if handler.includedMetrics.Has(container.DiskUsageMetrics) {
+		handler.rootfsStorageDir = path.Join(rootfsDir, *ArgContainerdNamespace, id, "rootfs")
+		handler.fsHandler = common.NewFsHandler(common.DefaultPeriod, handler.rootfsStorageDir, "", fsInfo)
+	}
 
 	for _, exposedEnv := range metadataEnvAllowList {
 		if exposedEnv == "" {
@@ -173,9 +181,7 @@ func (h *containerdContainerHandler) ContainerReference() (info.ContainerReferen
 }
 
 func (h *containerdContainerHandler) GetSpec() (info.ContainerSpec, error) {
-	// TODO: Since we dont collect disk usage stats for containerd, we set hasFilesystem
-	// to false. Revisit when we support disk usage stats for containerd
-	hasFilesystem := false
+	hasFilesystem := h.includedMetrics.Has(container.DiskUsageMetrics)
 	hasNet := h.includedMetrics.Has(container.NetworkUsageMetrics)
 	spec, err := common.GetSpec(h.cgroupPaths, h.machineInfoFactory, hasNet, hasFilesystem)
 	spec.Labels = h.labels
@@ -194,6 +200,34 @@ func (h *containerdContainerHandler) getFsStats(stats *info.ContainerStats) erro
 	if h.includedMetrics.Has(container.DiskIOMetrics) {
 		common.AssignDeviceNamesToDiskStats((*common.MachineInfoNamer)(mi), &stats.DiskIo)
 	}
+
+	if h.includedMetrics.Has(container.DiskUsageMetrics) {
+		deviceInfo, err := h.fsInfo.GetDirFsDevice(h.rootfsStorageDir)
+		if err != nil {
+			return fmt.Errorf("unable to determine device info for dir: %v: %v", h.rootfsStorageDir, err)
+		}
+		for _, fileSystem := range mi.Filesystems {
+			if fileSystem.Device == deviceInfo.Device {
+				usage := h.fsHandler.Usage()
+				fsStat := info.FsStats{
+					Device:    deviceInfo.Device,
+					Type:      fileSystem.Type,
+					Limit:     fileSystem.Capacity,
+					BaseUsage: usage.BaseUsageBytes,
+					Usage:     usage.TotalUsageBytes,
+					Inodes:    usage.InodeUsage,
+				}
+				fileSystems, err := h.fsInfo.GetGlobalFsInfo()
+				if err != nil {
+					return fmt.Errorf("unable to obtain diskstats for filesystem %s: %v", fsStat.Device, err)
+				}
+				fs.AddDiskStats(fileSystems, &fileSystem, &fsStat)
+				stats.Filesystem = append(stats.Filesystem, fsStat)
+				break
+			}
+		}
+
+	}
 	return nil
 }
 
@@ -241,9 +275,15 @@ func (h *containerdContainerHandler) Type() container.ContainerType {
 }
 
 func (h *containerdContainerHandler) Start() {
+	if h.fsHandler != nil {
+		h.fsHandler.Start()
+	}
 }
 
 func (h *containerdContainerHandler) Cleanup() {
+	if h.fsHandler != nil {
+		h.fsHandler.Stop()
+	}
 }
 
 func (h *containerdContainerHandler) GetContainerIPAddress() string {

container/containerd/handler_test.go

@@ -16,6 +16,7 @@
 package containerd
 
 import (
+	"context"
 	"testing"
 
 	"github.com/containerd/typeurl/v2"
@@ -121,7 +122,8 @@ func TestHandler(t *testing.T) {
 			map[string]string{"TEST_REGION": "FRA", "TEST_ZONE": "A"},
 		},
 	} {
-		handler, err := newContainerdContainerHandler(ts.client, ts.name, ts.machineInfoFactory, ts.fsInfo, ts.cgroupSubsystems, ts.inHostNamespace, ts.metadataEnvAllowList, ts.includedMetrics)
+		rootfsDir, _ := ts.client.RootfsDir(context.TODO())
+		handler, err := newContainerdContainerHandler(ts.client, ts.name, ts.machineInfoFactory, ts.fsInfo, ts.cgroupSubsystems, ts.inHostNamespace, ts.metadataEnvAllowList, ts.includedMetrics, rootfsDir)
 		if ts.hasErr {
 			as.NotNil(err)
 			if ts.errContains != "" {

container/docker/fs.go

@@ -64,13 +64,13 @@ func FsStats(
 			return nil
 		}
 
-		for _, fs := range mi.Filesystems {
-			if fs.Device == device {
+		for _, fileSystem := range mi.Filesystems {
+			if fileSystem.Device == device {
 				usage := fsHandler.Usage()
 				fsStat := info.FsStats{
 					Device:    device,
-					Type:      fs.Type,
-					Limit:     fs.Capacity,
+					Type:      fileSystem.Type,
+					Limit:     fileSystem.Capacity,
 					BaseUsage: usage.BaseUsageBytes,
 					Usage:     usage.TotalUsageBytes,
 					Inodes:    usage.InodeUsage,
@@ -79,7 +79,7 @@ func FsStats(
 				if err != nil {
 					return fmt.Errorf("unable to obtain diskstats for filesystem %s: %v", fsStat.Device, err)
 				}
-				addDiskStats(fileSystems, &fs, &fsStat)
+				fs.AddDiskStats(fileSystems, &fileSystem, &fsStat)
 				stats.Filesystem = append(stats.Filesystem, fsStat)
 				break
 			}
@@ -89,30 +89,6 @@ func FsStats(
 	return nil
 }
 
-func addDiskStats(fileSystems []fs.Fs, fsInfo *info.FsInfo, fsStats *info.FsStats) {
-	if fsInfo == nil {
-		return
-	}
-
-	for _, fileSys := range fileSystems {
-		if fsInfo.DeviceMajor == fileSys.DiskStats.Major &&
-			fsInfo.DeviceMinor == fileSys.DiskStats.Minor {
-			fsStats.ReadsCompleted = fileSys.DiskStats.ReadsCompleted
-			fsStats.ReadsMerged = fileSys.DiskStats.ReadsMerged
-			fsStats.SectorsRead = fileSys.DiskStats.SectorsRead
-			fsStats.ReadTime = fileSys.DiskStats.ReadTime
-			fsStats.WritesCompleted = fileSys.DiskStats.WritesCompleted
-			fsStats.WritesMerged = fileSys.DiskStats.WritesMerged
-			fsStats.SectorsWritten = fileSys.DiskStats.SectorsWritten
-			fsStats.WriteTime = fileSys.DiskStats.WriteTime
-			fsStats.IoInProgress = fileSys.DiskStats.IoInProgress
-			fsStats.IoTime = fileSys.DiskStats.IoTime
-			fsStats.WeightedIoTime = fileSys.DiskStats.WeightedIoTime
-			break
-		}
-	}
-}
-
 // FsHandler is a composite FsHandler implementation the incorporates
 // the common fs handler, a devicemapper ThinPoolWatcher, and a zfsWatcher
 type FsHandler struct {