resources: CNI network usage sampling support

Signed-off-by: Tonis Tiigi <[email protected]>

Author: tonistiigi

executor/resources/monitor.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/moby/buildkit/executor/resources/types"
+	"github.com/moby/buildkit/util/network"
 	"golang.org/x/sys/unix"
 )
 
@@ -34,6 +35,7 @@ type cgroupRecord struct {
 	err        error
 	done       chan struct{}
 	monitor    *Monitor
+	netSampler NetworkSampler
 }
 
 func (r *cgroupRecord) Wait() error {
@@ -89,6 +91,13 @@ func (r *cgroupRecord) sample() (*types.Sample, error) {
 		IOStat:     io,
 		PIDsStat:   pids,
 	}
+	if r.netSampler != nil {
+		net, err := r.netSampler.Sample()
+		if err != nil {
+			return nil, err
+		}
+		sample.NetStat = net
+	}
 	return sample, nil
 }
 
@@ -117,24 +126,34 @@ type Monitor struct {
 	records map[string]*cgroupRecord
 }
 
-func (m *Monitor) RecordNamespace(ns string, release func(context.Context) error) (types.Recorder, error) {
+type NetworkSampler interface {
+	Sample() (*network.Sample, error)
+}
+
+type RecordOpt struct {
+	Release        func(context.Context) error
+	NetworkSampler NetworkSampler
+}
+
+func (m *Monitor) RecordNamespace(ns string, opt RecordOpt) (types.Recorder, error) {
 	isClosed := false
 	select {
 	case <-m.closed:
 		isClosed = true
 	default:
 	}
 	if !isCgroupV2 || isClosed {
-		if err := release(context.TODO()); err != nil {
+		if err := opt.Release(context.TODO()); err != nil {
 			return nil, err
 		}
 		return &nopRecord{}, nil
 	}
 	r := &cgroupRecord{
-		ns:      ns,
-		release: release,
-		done:    make(chan struct{}),
-		monitor: m,
+		ns:         ns,
+		release:    opt.Release,
+		done:       make(chan struct{}),
+		monitor:    m,
+		netSampler: opt.NetworkSampler,
 	}
 	m.mu.Lock()
 	m.records[ns] = r

executor/resources/types/types.go

@@ -1,6 +1,10 @@
 package types
 
-import "time"
+import (
+	"time"
+
+	"github.com/moby/buildkit/util/network"
+)
 
 type Recorder interface {
 	Wait() error
@@ -9,11 +13,12 @@ type Recorder interface {
 
 // Sample represents a wrapper for sampled data of cgroupv2 controllers
 type Sample struct {
-	Timestamp  time.Time   `json:"timestamp"`
-	CPUStat    *CPUStat    `json:"cpuStat,omitempty"`
-	MemoryStat *MemoryStat `json:"memoryStat,omitempty"`
-	IOStat     *IOStat     `json:"ioStat,omitempty"`
-	PIDsStat   *PIDsStat   `json:"pidsStat,omitempty"`
+	Timestamp  time.Time       `json:"timestamp"`
+	CPUStat    *CPUStat        `json:"cpuStat,omitempty"`
+	MemoryStat *MemoryStat     `json:"memoryStat,omitempty"`
+	IOStat     *IOStat         `json:"ioStat,omitempty"`
+	PIDsStat   *PIDsStat       `json:"pidsStat,omitempty"`
+	NetStat    *network.Sample `json:"netStat,omitempty"`
 }
 
 // CPUStat represents the sampling state of the cgroupv2 CPU controller

executor/runcexecutor/executor.go

@@ -174,7 +174,12 @@ func (w *runcExecutor) Run(ctx context.Context, id string, root executor.Mount,
 	if err != nil {
 		return nil, err
 	}
-	defer namespace.Close()
+	doReleaseNetwork := true
+	defer func() {
+		if doReleaseNetwork {
+			namespace.Close()
+		}
+	}()
 
 	if meta.NetMode == pb.NetMode_HOST {
 		bklog.G(ctx).Info("enabling HostNetworking")
@@ -304,19 +309,31 @@ func (w *runcExecutor) Run(ctx context.Context, id string, root executor.Mount,
 			}
 		})
 	}, true)
+
+	releaseContainer := func(ctx context.Context) error {
+		err := w.runc.Delete(ctx, id, &runc.DeleteOpts{})
+		err1 := namespace.Close()
+		if err == nil {
+			err = err1
+		}
+		return err
+	}
+	doReleaseNetwork = false
+
 	err = exitError(ctx, err)
 	if err != nil {
-		w.runc.Delete(context.TODO(), id, &runc.DeleteOpts{})
+		releaseContainer(context.TODO())
 		return nil, err
 	}
 
 	cgroupPath := spec.Linux.CgroupsPath
 	if cgroupPath != "" {
-		return w.resmon.RecordNamespace(cgroupPath, func(ctx context.Context) error {
-			return w.runc.Delete(ctx, id, &runc.DeleteOpts{})
+		return w.resmon.RecordNamespace(cgroupPath, resources.RecordOpt{
+			Release:        releaseContainer,
+			NetworkSampler: namespace,
 		})
 	}
-	return nil, w.runc.Delete(context.TODO(), id, &runc.DeleteOpts{})
+	return nil, releaseContainer(context.TODO())
 }
 
 func exitError(ctx context.Context, err error) error {

util/network/cniprovider/cni.go

@@ -242,42 +242,97 @@ func (c *cniProvider) newNS(ctx context.Context, hostname string) (*cniNS, error
 			cni.WithArgs("IgnoreUnknown", "1"))
 	}
 
-	if _, err := c.CNI.Setup(context.TODO(), id, nativeID, nsOpts...); err != nil {
+	cniRes, err := c.CNI.Setup(context.TODO(), id, nativeID, nsOpts...)
+	if err != nil {
 		deleteNetNS(nativeID)
 		return nil, errors.Wrap(err, "CNI setup error")
 	}
 	trace.SpanFromContext(ctx).AddEvent("finished setting up network namespace")
 	bklog.G(ctx).Debugf("finished setting up network namespace %s", id)
 
-	return &cniNS{
+	vethName := ""
+	for k := range cniRes.Interfaces {
+		if strings.HasPrefix(k, "veth") {
+			if vethName != "" {
+				// invalid config
+				vethName = ""
+				break
+			}
+			vethName = k
+		}
+	}
+
+	ns := &cniNS{
 		nativeID: nativeID,
 		id:       id,
 		handle:   c.CNI,
 		opts:     nsOpts,
-	}, nil
+		vethName: vethName,
+	}
+
+	if ns.vethName != "" {
+		sample, err := ns.sample()
+		if err == nil && sample != nil {
+			ns.canSample = true
+			ns.offsetSample = sample
+		}
+	}
+
+	return ns, nil
 }
 
 type cniNS struct {
-	pool     *cniPool
-	handle   cni.CNI
-	id       string
-	nativeID string
-	opts     []cni.NamespaceOpts
-	lastUsed time.Time
+	pool         *cniPool
+	handle       cni.CNI
+	id           string
+	nativeID     string
+	opts         []cni.NamespaceOpts
+	lastUsed     time.Time
+	vethName     string
+	canSample    bool
+	offsetSample *network.Sample
+	prevSample   *network.Sample
 }
 
 func (ns *cniNS) Set(s *specs.Spec) error {
 	return setNetNS(s, ns.nativeID)
 }
 
 func (ns *cniNS) Close() error {
+	if ns.prevSample != nil {
+		ns.offsetSample = ns.prevSample
+	}
 	if ns.pool == nil {
 		return ns.release()
 	}
 	ns.pool.put(ns)
 	return nil
 }
 
+func (ns *cniNS) Sample() (*network.Sample, error) {
+	if !ns.canSample {
+		return nil, nil
+	}
+	s, err := ns.sample()
+	if err != nil {
+		return nil, err
+	}
+	if s == nil {
+		return nil, nil
+	}
+	if ns.offsetSample != nil {
+		s.TxBytes -= ns.offsetSample.TxBytes
+		s.RxBytes -= ns.offsetSample.RxBytes
+		s.TxPackets -= ns.offsetSample.TxPackets
+		s.RxPackets -= ns.offsetSample.RxPackets
+		s.TxErrors -= ns.offsetSample.TxErrors
+		s.RxErrors -= ns.offsetSample.RxErrors
+		s.TxDropped -= ns.offsetSample.TxDropped
+		s.RxDropped -= ns.offsetSample.RxDropped
+	}
+	return s, nil
+}
+
 func (ns *cniNS) release() error {
 	bklog.L.Debugf("releasing cni network namespace %s", ns.id)
 	err := ns.handle.Remove(context.TODO(), ns.id, ns.nativeID, ns.opts...)

util/network/cniprovider/cni_linux.go

@@ -0,0 +1,70 @@
+package cniprovider
+
+import (
+	"path/filepath"
+	"strconv"
+	"strings"
+	"syscall"
+
+	"github.com/moby/buildkit/util/network"
+	"github.com/pkg/errors"
+)
+
+func (ns *cniNS) sample() (*network.Sample, error) {
+	dirfd, err := syscall.Open(filepath.Join("/sys/class/net", ns.vethName, "statistics"), syscall.O_RDONLY, 0)
+	if err != nil {
+		if errors.Is(err, syscall.ENOENT) || errors.Is(err, syscall.ENOTDIR) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	defer syscall.Close(dirfd)
+
+	buf := make([]byte, 32)
+	stat := &network.Sample{}
+
+	for _, name := range []string{"tx_bytes", "rx_bytes", "tx_packets", "rx_packets", "tx_errors", "rx_errors", "tx_dropped", "rx_dropped"} {
+		n, err := readFileAt(dirfd, name, buf)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to read %s", name)
+		}
+		switch name {
+		case "tx_bytes":
+			stat.TxBytes = n
+		case "rx_bytes":
+			stat.RxBytes = n
+		case "tx_packets":
+			stat.TxPackets = n
+		case "rx_packets":
+			stat.RxPackets = n
+		case "tx_errors":
+			stat.TxErrors = n
+		case "rx_errors":
+			stat.RxErrors = n
+		case "tx_dropped":
+			stat.TxDropped = n
+		case "rx_dropped":
+			stat.RxDropped = n
+		}
+	}
+	ns.prevSample = stat
+	return stat, nil
+}
+
+func readFileAt(dirfd int, filename string, buf []byte) (int64, error) {
+	fd, err := syscall.Openat(dirfd, filename, syscall.O_RDONLY, 0)
+	if err != nil {
+		return 0, err
+	}
+	defer syscall.Close(fd)
+
+	n, err := syscall.Read(fd, buf[:])
+	if err != nil {
+		return 0, err
+	}
+	nn, err := strconv.ParseInt(strings.TrimSpace(string(buf[:n])), 10, 64)
+	if err != nil {
+		return 0, err
+	}
+	return nn, nil
+}

util/network/cniprovider/cni_nolinux.go

@@ -0,0 +1,12 @@
+//go:build !linux
+// +build !linux
+
+package cniprovider
+
+import (
+	"github.com/moby/buildkit/util/network"
+)
+
+func (ns *cniNS) sample() (*network.Sample, error) {
+	return nil, nil
+}

util/network/host.go

@@ -35,3 +35,7 @@ func (h *hostNS) Set(s *specs.Spec) error {
 func (h *hostNS) Close() error {
 	return nil
 }
+
+func (h *hostNS) Sample() (*Sample, error) {
+	return nil, nil
+}

util/network/network.go

@@ -7,6 +7,17 @@ import (
 	specs "github.com/opencontainers/runtime-spec/specs-go"
 )
 
+type Sample struct {
+	RxBytes   int64 `json:"rxBytes,omitempty"`
+	RxPackets int64 `json:"rxPackets,omitempty"`
+	RxErrors  int64 `json:"rxErrors,omitempty"`
+	RxDropped int64 `json:"rxDropped,omitempty"`
+	TxBytes   int64 `json:"txBytes,omitempty"`
+	TxPackets int64 `json:"txPackets,omitempty"`
+	TxErrors  int64 `json:"txErrors,omitempty"`
+	TxDropped int64 `json:"txDropped,omitempty"`
+}
+
 // Provider interface for Network
 type Provider interface {
 	io.Closer
@@ -18,4 +29,6 @@ type Namespace interface {
 	io.Closer
 	// Set the namespace on the spec
 	Set(*specs.Spec) error
+
+	Sample() (*Sample, error)
 }

util/network/none.go

@@ -31,3 +31,7 @@ func (h *noneNS) Set(s *specs.Spec) error {
 func (h *noneNS) Close() error {
 	return nil
 }
+
+func (h *noneNS) Sample() (*Sample, error) {
+	return nil, nil
+}