Merge pull request moby#3860 from tonistiigi/step-usage-monitoring

Add resource usage monitoring for build steps

Author: tonistiigi

Dockerfile

@@ -216,6 +216,7 @@ RUN --mount=target=/root/.cache,type=cache \
 
 FROM buildkit-export AS buildkit-linux
 COPY --link --from=binaries / /usr/bin/
+ENV BUILDKIT_SETUP_CGROUPV2_ROOT=1
 ENTRYPOINT ["buildkitd"]
 
 FROM binaries AS buildkit-darwin
@@ -255,6 +256,7 @@ ENTRYPOINT ["/docker-entrypoint.sh"]
 # musl is needed to directly use the registry binary that is built on alpine
 ENV BUILDKIT_INTEGRATION_CONTAINERD_EXTRA="containerd-1.6=/opt/containerd-alt-16/bin"
 ENV BUILDKIT_INTEGRATION_SNAPSHOTTER=stargz
+ENV BUILDKIT_SETUP_CGROUPV2_ROOT=1
 ENV CGO_ENABLED=0
 ENV GOTESTSUM_FORMAT=standard-verbose
 COPY --link --from=gotestsum /out/gotestsum /usr/bin/

executor/containerdexecutor/executor.go

@@ -21,6 +21,7 @@ import (
 	"github.com/docker/docker/pkg/idtools"
 	"github.com/moby/buildkit/executor"
 	"github.com/moby/buildkit/executor/oci"
+	resourcestypes "github.com/moby/buildkit/executor/resources/types"
 	gatewayapi "github.com/moby/buildkit/frontend/gateway/pb"
 	"github.com/moby/buildkit/identity"
 	"github.com/moby/buildkit/snapshot"
@@ -78,7 +79,7 @@ func New(client *containerd.Client, root, cgroup string, networkProviders map[pb
 	}
 }
 
-func (w *containerdExecutor) Run(ctx context.Context, id string, root executor.Mount, mounts []executor.Mount, process executor.ProcessInfo, started chan<- struct{}) (err error) {
+func (w *containerdExecutor) Run(ctx context.Context, id string, root executor.Mount, mounts []executor.Mount, process executor.ProcessInfo, started chan<- struct{}) (rec resourcestypes.Recorder, err error) {
 	if id == "" {
 		id = identity.NewID()
 	}
@@ -105,25 +106,25 @@ func (w *containerdExecutor) Run(ctx context.Context, id string, root executor.M
 
 	resolvConf, err := oci.GetResolvConf(ctx, w.root, nil, w.dnsConfig)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	hostsFile, clean, err := oci.GetHostsFile(ctx, w.root, meta.ExtraHosts, nil, meta.Hostname)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	if clean != nil {
 		defer clean()
 	}
 
 	mountable, err := root.Src.Mount(ctx, false)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	rootMounts, release, err := mountable.Mount()
 	if err != nil {
-		return err
+		return nil, err
 	}
 	if release != nil {
 		defer release()
@@ -132,14 +133,14 @@ func (w *containerdExecutor) Run(ctx context.Context, id string, root executor.M
 	lm := snapshot.LocalMounterWithMounts(rootMounts)
 	rootfsPath, err := lm.Mount()
 	if err != nil {
-		return err
+		return nil, err
 	}
 	defer lm.Unmount()
 	defer executor.MountStubsCleaner(ctx, rootfsPath, mounts, meta.RemoveMountStubsRecursive)()
 
 	uid, gid, sgids, err := oci.GetUser(rootfsPath, meta.User)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	identity := idtools.Identity{
@@ -149,21 +150,21 @@ func (w *containerdExecutor) Run(ctx context.Context, id string, root executor.M
 
 	newp, err := fs.RootPath(rootfsPath, meta.Cwd)
 	if err != nil {
-		return errors.Wrapf(err, "working dir %s points to invalid target", newp)
+		return nil, errors.Wrapf(err, "working dir %s points to invalid target", newp)
 	}
 	if _, err := os.Stat(newp); err != nil {
 		if err := idtools.MkdirAllAndChown(newp, 0755, identity); err != nil {
-			return errors.Wrapf(err, "failed to create working directory %s", newp)
+			return nil, errors.Wrapf(err, "failed to create working directory %s", newp)
 		}
 	}
 
 	provider, ok := w.networkProviders[meta.NetMode]
 	if !ok {
-		return errors.Errorf("unknown network mode %s", meta.NetMode)
+		return nil, errors.Errorf("unknown network mode %s", meta.NetMode)
 	}
 	namespace, err := provider.New(ctx, meta.Hostname)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	defer namespace.Close()
 
@@ -179,21 +180,21 @@ func (w *containerdExecutor) Run(ctx context.Context, id string, root executor.M
 	processMode := oci.ProcessSandbox // FIXME(AkihiroSuda)
 	spec, cleanup, err := oci.GenerateSpec(ctx, meta, mounts, id, resolvConf, hostsFile, namespace, w.cgroupParent, processMode, nil, w.apparmorProfile, w.selinux, w.traceSocket, opts...)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	defer cleanup()
 	spec.Process.Terminal = meta.Tty
 	if w.rootless {
 		if err := rootlessspecconv.ToRootless(spec); err != nil {
-			return err
+			return nil, err
 		}
 	}
 
 	container, err := w.client.NewContainer(ctx, id,
 		containerd.WithSpec(spec),
 	)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	defer func() {
@@ -214,7 +215,7 @@ func (w *containerdExecutor) Run(ctx context.Context, id string, root executor.M
 		Options: []string{"rbind"},
 	}}))
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	defer func() {
@@ -225,7 +226,7 @@ func (w *containerdExecutor) Run(ctx context.Context, id string, root executor.M
 
 	if nn, ok := namespace.(OnCreateRuntimer); ok {
 		if err := nn.OnCreateRuntime(task.Pid()); err != nil {
-			return err
+			return nil, err
 		}
 	}
 
@@ -238,7 +239,7 @@ func (w *containerdExecutor) Run(ctx context.Context, id string, root executor.M
 			}
 		})
 	})
-	return err
+	return nil, err
 }
 
 func (w *containerdExecutor) Exec(ctx context.Context, id string, process executor.ProcessInfo) (err error) {

executor/executor.go

@@ -6,6 +6,7 @@ import (
 	"net"
 	"syscall"
 
+	resourcestypes "github.com/moby/buildkit/executor/resources/types"
 	"github.com/moby/buildkit/snapshot"
 	"github.com/moby/buildkit/solver/pb"
 )
@@ -55,7 +56,7 @@ type Executor interface {
 	// Run will start a container for the given process with rootfs, mounts.
 	// `id` is an optional name for the container so it can be referenced later via Exec.
 	// `started` is an optional channel that will be closed when the container setup completes and has started running.
-	Run(ctx context.Context, id string, rootfs Mount, mounts []Mount, process ProcessInfo, started chan<- struct{}) error
+	Run(ctx context.Context, id string, rootfs Mount, mounts []Mount, process ProcessInfo, started chan<- struct{}) (resourcestypes.Recorder, error)
 	// Exec will start a process in container matching `id`. An error will be returned
 	// if the container failed to start (via Run) or has exited before Exec is called.
 	Exec(ctx context.Context, id string, process ProcessInfo) error

executor/resources/cpu.go

@@ -0,0 +1,140 @@
+package resources
+
+import (
+	"bufio"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+
+	"github.com/moby/buildkit/executor/resources/types"
+	"github.com/pkg/errors"
+)
+
+const (
+	cpuUsageUsec     = "usage_usec"
+	cpuUserUsec      = "user_usec"
+	cpuSystemUsec    = "system_usec"
+	cpuNrPeriods     = "nr_periods"
+	cpuNrThrottled   = "nr_throttled"
+	cpuThrottledUsec = "throttled_usec"
+)
+
+func getCgroupCPUStat(cgroupPath string) (*types.CPUStat, error) {
+	cpuStat := &types.CPUStat{}
+
+	// Read cpu.stat file
+	cpuStatFile, err := os.Open(filepath.Join(cgroupPath, "cpu.stat"))
+	if err != nil {
+		if errors.Is(err, os.ErrNotExist) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	defer cpuStatFile.Close()
+
+	scanner := bufio.NewScanner(cpuStatFile)
+	for scanner.Scan() {
+		line := scanner.Text()
+		fields := strings.Fields(line)
+
+		if len(fields) < 2 {
+			continue
+		}
+
+		key := fields[0]
+		value, err := strconv.ParseUint(fields[1], 10, 64)
+		if err != nil {
+			continue
+		}
+
+		switch key {
+		case cpuUsageUsec:
+			cpuStat.UsageNanos = uint64Ptr(value * 1000)
+		case cpuUserUsec:
+			cpuStat.UserNanos = uint64Ptr(value * 1000)
+		case cpuSystemUsec:
+			cpuStat.SystemNanos = uint64Ptr(value * 1000)
+		case cpuNrPeriods:
+			cpuStat.NrPeriods = new(uint32)
+			*cpuStat.NrPeriods = uint32(value)
+		case cpuNrThrottled:
+			cpuStat.NrThrottled = new(uint32)
+			*cpuStat.NrThrottled = uint32(value)
+		case cpuThrottledUsec:
+			cpuStat.ThrottledNanos = uint64Ptr(value * 1000)
+		}
+	}
+
+	if err := scanner.Err(); err != nil {
+		return nil, err
+	}
+
+	// Read cpu.pressure file
+	pressure, err := parsePressureFile(filepath.Join(cgroupPath, "cpu.pressure"))
+	if err == nil {
+		cpuStat.Pressure = pressure
+	}
+
+	return cpuStat, nil
+}
+func parsePressureFile(filename string) (*types.Pressure, error) {
+	content, err := os.ReadFile(filename)
+	if err != nil {
+		if errors.Is(err, os.ErrNotExist) { // pressure file requires CONFIG_PSI
+			return nil, nil
+		}
+		return nil, err
+	}
+
+	lines := strings.Split(string(content), "\n")
+
+	pressure := &types.Pressure{}
+	for _, line := range lines {
+		// Skip empty lines
+		if len(strings.TrimSpace(line)) == 0 {
+			continue
+		}
+
+		fields := strings.Fields(line)
+		prefix := fields[0]
+		pressureValues := &types.PressureValues{}
+
+		for i := 1; i < len(fields); i++ {
+			keyValue := strings.Split(fields[i], "=")
+			key := keyValue[0]
+			valueStr := keyValue[1]
+
+			if key == "total" {
+				totalValue, err := strconv.ParseUint(valueStr, 10, 64)
+				if err != nil {
+					return nil, err
+				}
+				pressureValues.Total = &totalValue
+			} else {
+				value, err := strconv.ParseFloat(valueStr, 64)
+				if err != nil {
+					return nil, err
+				}
+
+				switch key {
+				case "avg10":
+					pressureValues.Avg10 = &value
+				case "avg60":
+					pressureValues.Avg60 = &value
+				case "avg300":
+					pressureValues.Avg300 = &value
+				}
+			}
+		}
+
+		switch prefix {
+		case "some":
+			pressure.Some = pressureValues
+		case "full":
+			pressure.Full = pressureValues
+		}
+	}
+
+	return pressure, nil
+}