Add missing file system metrics in containerd handler

container/common/fsHandler.go

@@ -37,15 +37,21 @@ type FsUsage struct {
 	InodeUsage      uint64
 }
 
+type FsUsageProvider interface {
+	// Usage returns the fs usage
+	Usage() (*FsUsage, error)
+	// Targets returns where the fs usage metric is collected,it maybe a directory, a file or some
+	// information about the snapshotter(for containerd)
+	Targets() []string
+}
+
 type realFsHandler struct {
 	sync.RWMutex
-	lastUpdate time.Time
-	usage      FsUsage
-	period     time.Duration
-	minPeriod  time.Duration
-	rootfs     string
-	extraDir   string
-	fsInfo     fs.FsInfo
+	lastUpdate    time.Time
+	usage         FsUsage
+	period        time.Duration
+	minPeriod     time.Duration
+	usageProvider FsUsageProvider
 	// Tells the container to stop.
 	stopChan chan struct{}
 }
@@ -58,56 +64,33 @@ const DefaultPeriod = time.Minute
 
 var _ FsHandler = &realFsHandler{}
 
-func NewFsHandler(period time.Duration, rootfs, extraDir string, fsInfo fs.FsInfo) FsHandler {
+func NewFsHandler(period time.Duration, provider FsUsageProvider) FsHandler {
 	return &realFsHandler{
-		lastUpdate: time.Time{},
-		usage:      FsUsage{},
-		period:     period,
-		minPeriod:  period,
-		rootfs:     rootfs,
-		extraDir:   extraDir,
-		fsInfo:     fsInfo,
-		stopChan:   make(chan struct{}, 1),
+		lastUpdate:    time.Time{},
+		usage:         FsUsage{},
+		period:        period,
+		minPeriod:     period,
+		usageProvider: provider,
+		stopChan:      make(chan struct{}, 1),
 	}
 }
 
 func (fh *realFsHandler) update() error {
-	var (
-		rootUsage, extraUsage fs.UsageInfo
-		rootErr, extraErr     error
-	)
-	// TODO(vishh): Add support for external mounts.
-	if fh.rootfs != "" {
-		rootUsage, rootErr = fh.fsInfo.GetDirUsage(fh.rootfs)
-	}
 
-	if fh.extraDir != "" {
-		extraUsage, extraErr = fh.fsInfo.GetDirUsage(fh.extraDir)
+	usage, err := fh.usageProvider.Usage()
+
+	if err != nil {
+		return err
 	}
 
-	// Wait to handle errors until after all operartions are run.
-	// An error in one will not cause an early return, skipping others
 	fh.Lock()
 	defer fh.Unlock()
 	fh.lastUpdate = time.Now()
-	if fh.rootfs != "" && rootErr == nil {
-		fh.usage.InodeUsage = rootUsage.Inodes
-		fh.usage.BaseUsageBytes = rootUsage.Bytes
-		fh.usage.TotalUsageBytes = rootUsage.Bytes
-	}
-	if fh.extraDir != "" && extraErr == nil {
-		if fh.rootfs != "" {
-			fh.usage.TotalUsageBytes += extraUsage.Bytes
-		} else {
-			// rootfs is empty, totalUsageBytes use extra usage bytes
-			fh.usage.TotalUsageBytes = extraUsage.Bytes
-		}
-	}
 
-	// Combine errors into a single error to return
-	if rootErr != nil || extraErr != nil {
-		return fmt.Errorf("rootDiskErr: %v, extraDiskErr: %v", rootErr, extraErr)
-	}
+	fh.usage.InodeUsage = usage.InodeUsage
+	fh.usage.BaseUsageBytes = usage.BaseUsageBytes
+	fh.usage.TotalUsageBytes = usage.TotalUsageBytes
+
 	return nil
 }
 
@@ -130,7 +113,8 @@ func (fh *realFsHandler) trackUsage() {
 			// if the long duration is persistent either because of slow
 			// disk or lots of containers.
 			longOp = longOp + time.Second
-			klog.V(2).Infof("fs: disk usage and inodes count on following dirs took %v: %v; will not log again for this container unless duration exceeds %v", duration, []string{fh.rootfs, fh.extraDir}, longOp)
+			klog.V(2).Infof(`fs: disk usage and inodes count on targets took %v: %v; `+
+				`will not log again for this container unless duration exceeds %v`, duration, fh.usageProvider.Targets(), longOp)
 		}
 		select {
 		case <-fh.stopChan:
@@ -153,3 +137,56 @@ func (fh *realFsHandler) Usage() FsUsage {
 	defer fh.RUnlock()
 	return fh.usage
 }
+
+type fsUsageProvider struct {
+	fsInfo fs.FsInfo
+	rootFs string
+	// The directory consumed by the container but outside rootFs, e.g. directory of saving logs
+	extraDir string
+}
+
+func NewGeneralFsUsageProvider(fsInfo fs.FsInfo, rootFs, extraDir string) FsUsageProvider {
+	return &fsUsageProvider{
+		fsInfo:   fsInfo,
+		rootFs:   rootFs,
+		extraDir: extraDir,
+	}
+}
+
+func (f *fsUsageProvider) Targets() []string {
+	return []string{f.rootFs, f.extraDir}
+}
+
+func (f *fsUsageProvider) Usage() (*FsUsage, error) {
+	var (
+		rootUsage, extraUsage fs.UsageInfo
+		rootErr, extraErr     error
+	)
+
+	if f.rootFs != "" {
+		rootUsage, rootErr = f.fsInfo.GetDirUsage(f.rootFs)
+	}
+
+	if f.extraDir != "" {
+		extraUsage, extraErr = f.fsInfo.GetDirUsage(f.extraDir)
+	}
+
+	usage := &FsUsage{}
+
+	if f.rootFs != "" && rootErr == nil {
+		usage.InodeUsage = rootUsage.Inodes
+		usage.BaseUsageBytes = rootUsage.Bytes
+		usage.TotalUsageBytes = rootUsage.Bytes
+	}
+
+	if f.extraDir != "" && extraErr == nil {
+		usage.TotalUsageBytes += extraUsage.Bytes
+	}
+
+	// Combine errors into a single error to return
+	if rootErr != nil || extraErr != nil {
+		return nil, fmt.Errorf("failed to obtain filesystem usage; rootDiskErr: %v, extraDiskErr: %v", rootErr, extraErr)
+	}
+
+	return usage, nil
+}

container/containerd/client.go

@@ -22,8 +22,10 @@ import (
 	"time"
 
 	containersapi "github.com/containerd/containerd/api/services/containers/v1"
+	snapshotapi "github.com/containerd/containerd/api/services/snapshots/v1"
 	tasksapi "github.com/containerd/containerd/api/services/tasks/v1"
 	versionapi "github.com/containerd/containerd/api/services/version/v1"
+	types "github.com/containerd/containerd/api/types"
 	"github.com/containerd/containerd/containers"
 	"github.com/containerd/containerd/errdefs"
 	"github.com/containerd/containerd/pkg/dialer"
@@ -37,14 +39,17 @@ type client struct {
 	containerService containersapi.ContainersClient
 	taskService      tasksapi.TasksClient
 	versionService   versionapi.VersionClient
+	snapshotService  snapshotapi.SnapshotsClient
 	criService       criapi.RuntimeServiceClient
 }
 
 type ContainerdClient interface {
 	LoadContainer(ctx context.Context, id string) (*containers.Container, error)
 	TaskPid(ctx context.Context, id string) (uint32, error)
 	Version(ctx context.Context) (string, error)
+	SnapshotMounts(ctx context.Context, snapshotter, key string) ([]*types.Mount, error)
 	ContainerStatus(ctx context.Context, id string) (*criapi.ContainerStatus, error)
+	ContainerStats(ctx context.Context, id string) (*criapi.ContainerStats, error)
 }
 
 var once sync.Once
@@ -95,6 +100,7 @@ func Client(address, namespace string) (ContainerdClient, error) {
 			containerService: containersapi.NewContainersClient(conn),
 			taskService:      tasksapi.NewTasksClient(conn),
 			versionService:   versionapi.NewVersionClient(conn),
+			snapshotService:  snapshotapi.NewSnapshotsClient(conn),
 			criService:       criapi.NewRuntimeServiceClient(conn),
 		}
 	})
@@ -129,6 +135,17 @@ func (c *client) Version(ctx context.Context) (string, error) {
 	return response.Version, nil
 }
 
+func (c *client) SnapshotMounts(ctx context.Context, snapshotter, key string) ([]*types.Mount, error) {
+	response, err := c.snapshotService.Mounts(ctx, &snapshotapi.MountsRequest{
+		Snapshotter: snapshotter,
+		Key:         key,
+	})
+	if err != nil {
+		return nil, errdefs.FromGRPC(err)
+	}
+	return response.Mounts, nil
+}
+
 func (c *client) ContainerStatus(ctx context.Context, id string) (*criapi.ContainerStatus, error) {
 	response, err := c.criService.ContainerStatus(ctx, &criapi.ContainerStatusRequest{
 		ContainerId: id,
@@ -140,6 +157,16 @@ func (c *client) ContainerStatus(ctx context.Context, id string) (*criapi.Contai
 	return response.Status, nil
 }
 
+func (c *client) ContainerStats(ctx context.Context, id string) (*criapi.ContainerStats, error) {
+	response, err := c.criService.ContainerStats(ctx, &criapi.ContainerStatsRequest{
+		ContainerId: id,
+	})
+	if err != nil {
+		return nil, err
+	}
+	return response.Stats, nil
+}
+
 func containerFromProto(containerpb containersapi.Container) *containers.Container {
 	var runtime containers.RuntimeInfo
 	if containerpb.Runtime != nil {

container/containerd/client_test.go

@@ -18,13 +18,15 @@ import (
 	"context"
 	"fmt"
 
+	types "github.com/containerd/containerd/api/types"
 	"github.com/containerd/containerd/containers"
 	criapi "github.com/google/cadvisor/cri-api/pkg/apis/runtime/v1alpha2"
 )
 
 type containerdClientMock struct {
 	cntrs     map[string]*containers.Container
 	status    *criapi.ContainerStatus
+	stats     *criapi.ContainerStats
 	returnErr error
 }
 
@@ -51,10 +53,19 @@ func (c *containerdClientMock) ContainerStatus(ctx context.Context, id string) (
 	return c.status, nil
 }
 
-func mockcontainerdClient(cntrs map[string]*containers.Container, status *criapi.ContainerStatus, returnErr error) ContainerdClient {
+func (c *containerdClientMock) ContainerStats(ctx context.Context, id string) (*criapi.ContainerStats, error) {
+	return c.stats, nil
+}
+
+func (c *containerdClientMock) SnapshotMounts(ctx context.Context, snapshotter, key string) ([]*types.Mount, error) {
+	return nil, nil
+}
+
+func mockcontainerdClient(cntrs map[string]*containers.Container, status *criapi.ContainerStatus, stats *criapi.ContainerStats, returnErr error) ContainerdClient {
 	return &containerdClientMock{
 		cntrs:     cntrs,
 		status:    status,
+		stats:     stats,
 		returnErr: returnErr,
 	}
 }

container/containerd/factory_test.go

@@ -58,7 +58,7 @@ func TestCanHandleAndAccept(t *testing.T) {
 	testContainers["40af7cdcbe507acad47a5a62025743ad3ddc6ab93b77b21363aa1c1d641047c9"] = testContainer
 
 	f := &containerdFactory{
-		client:             mockcontainerdClient(testContainers, nil, nil),
+		client:             mockcontainerdClient(testContainers, nil, nil, nil),
 		cgroupSubsystems:   containerlibcontainer.CgroupSubsystems{},
 		fsInfo:             nil,
 		machineInfoFactory: nil,