metrics: measure context transfers for local source operations

Measure the transfer size and duration of context transfers for various
categories of local source transfers from the progress stream that's
returned during the build.

Local source transfers are split into one of four categories:
* context
* dockerfile
* dockerignore
* namedcontext

Named contexts that are different names will be categorized under the
same metric.

Signed-off-by: Jonathan A. Sternberg <jonathan.sternberg@docker.com>

commands/build.go

@@ -312,6 +312,7 @@ func runBuild(ctx context.Context, dockerCli command.Cli, options buildOptions)
 	if _, err := console.ConsoleFromFile(os.Stderr); err == nil {
 		term = true
 	}
+	attributes := buildMetricAttributes(dockerCli, b, &options)
 
 	ctx2, cancel := context.WithCancel(context.TODO())
 	defer cancel()
@@ -325,6 +326,7 @@ func runBuild(ctx context.Context, dockerCli command.Cli, options buildOptions)
 			fmt.Sprintf("building with %q instance using %s driver", b.Name, b.Driver),
 			fmt.Sprintf("%s:%s", b.Driver, b.Name),
 		),
+		progress.WithMetrics(mp, attributes),
 		progress.WithOnClose(func() {
 			printWarnings(os.Stderr, printer.Warnings(), progressMode)
 		}),
@@ -333,7 +335,6 @@ func runBuild(ctx context.Context, dockerCli command.Cli, options buildOptions)
 		return err
 	}
 
-	attributes := buildMetricAttributes(dockerCli, b, &options)
 	done := timeBuildCommand(mp, attributes)
 	var resp *client.SolveResponse
 	var retErr error

util/progress/metricwriter.go

@@ -0,0 +1,151 @@
+package progress
+
+import (
+	"context"
+	"regexp"
+	"strings"
+	"time"
+
+	"github.com/docker/buildx/util/metricutil"
+	"github.com/moby/buildkit/client"
+	"github.com/opencontainers/go-digest"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+)
+
+type metricWriter struct {
+	recorders []metricRecorder
+	attrs     attribute.Set
+}
+
+func newMetrics(mp metric.MeterProvider, attrs attribute.Set) *metricWriter {
+	meter := metricutil.Meter(mp)
+	return &metricWriter{
+		recorders: []metricRecorder{
+			newLocalSourceTransferMetricRecorder(meter, attrs),
+		},
+		attrs: attrs,
+	}
+}
+
+func (mw *metricWriter) Write(ss *client.SolveStatus) {
+	for _, recorder := range mw.recorders {
+		recorder.Record(ss)
+	}
+}
+
+type metricRecorder interface {
+	Record(ss *client.SolveStatus)
+}
+
+type (
+	localSourceTransferState struct {
+		// Attributes holds the attributes specific to this context transfer.
+		Attributes attribute.Set
+
+		// LastTransferSize contains the last byte count for the transfer.
+		LastTransferSize int64
+	}
+	localSourceTransferMetricRecorder struct {
+		// BaseAttributes holds the set of base attributes for all metrics produced.
+		BaseAttributes attribute.Set
+
+		// State contains the state for individual digests that are being processed.
+		State map[digest.Digest]*localSourceTransferState
+
+		// TransferSize holds the metric for the number of bytes transferred.
+		TransferSize metric.Int64Counter
+
+		// Duration holds the metric for the total time taken to perform the transfer.
+		Duration metric.Float64Counter
+	}
+)
+
+func newLocalSourceTransferMetricRecorder(meter metric.Meter, attrs attribute.Set) *localSourceTransferMetricRecorder {
+	mr := &localSourceTransferMetricRecorder{
+		BaseAttributes: attrs,
+		State:          make(map[digest.Digest]*localSourceTransferState),
+	}
+	mr.TransferSize, _ = meter.Int64Counter("source.local.transfer.io",
+		metric.WithDescription("Measures the number of bytes transferred between the client and server for the context."),
+		metric.WithUnit("By"))
+
+	mr.Duration, _ = meter.Float64Counter("source.local.transfer.time",
+		metric.WithDescription("Measures the length of time spent transferring the context."),
+		metric.WithUnit("ms"))
+	return mr
+}
+
+func (mr *localSourceTransferMetricRecorder) Record(ss *client.SolveStatus) {
+	for _, v := range ss.Vertexes {
+		state, ok := mr.State[v.Digest]
+		if !ok {
+			attr := detectLocalSourceType(v.Name)
+			if !attr.Valid() {
+				// Not a context transfer operation so just ignore.
+				continue
+			}
+
+			state = &localSourceTransferState{
+				Attributes: attribute.NewSet(attr),
+			}
+			mr.State[v.Digest] = state
+		}
+
+		if v.Started != nil && v.Completed != nil {
+			dur := float64(v.Completed.Sub(*v.Started)) / float64(time.Millisecond)
+			mr.Duration.Add(context.Background(), dur,
+				metric.WithAttributeSet(mr.BaseAttributes),
+				metric.WithAttributeSet(state.Attributes),
+			)
+		}
+	}
+
+	for _, status := range ss.Statuses {
+		state, ok := mr.State[status.Vertex]
+		if !ok {
+			continue
+		}
+
+		if strings.HasPrefix(status.Name, "transferring") {
+			diff := status.Current - state.LastTransferSize
+			if diff > 0 {
+				mr.TransferSize.Add(context.Background(), diff,
+					metric.WithAttributeSet(mr.BaseAttributes),
+					metric.WithAttributeSet(state.Attributes),
+				)
+			}
+		}
+	}
+}
+
+var reLocalSourceType = regexp.MustCompile(
+	strings.Join([]string{
+		`(?P<context>\[internal] load build context)`,
+		`(?P<dockerfile>load build definition)`,
+		`(?P<dockerignore>load \.dockerignore)`,
+		`(?P<namedcontext>\[context .+] load from client)`,
+	}, "|"),
+)
+
+func detectLocalSourceType(vertexName string) attribute.KeyValue {
+	match := reLocalSourceType.FindStringSubmatch(vertexName)
+	if match == nil {
+		return attribute.KeyValue{}
+	}
+
+	for i, source := range reLocalSourceType.SubexpNames() {
+		if len(source) == 0 {
+			// Not a subexpression.
+			continue
+		}
+
+		// Did we find a match for this subexpression?
+		if len(match[i]) > 0 {
+			// Use the match name which corresponds to the name of the source.
+			return attribute.String("source.local.type", source)
+		}
+	}
+	// No matches found.
+	return attribute.KeyValue{}
+}

util/progress/printer.go

@@ -6,11 +6,14 @@ import (
 	"sync"
 
 	"github.com/containerd/console"
+	"github.com/docker/buildx/util/confutil"
 	"github.com/docker/buildx/util/logutil"
 	"github.com/moby/buildkit/client"
 	"github.com/moby/buildkit/util/progress/progressui"
 	"github.com/opencontainers/go-digest"
 	"github.com/sirupsen/logrus"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
 )
 
 type Printer struct {
@@ -24,6 +27,7 @@ type Printer struct {
 	warnings     []client.VertexWarning
 	logMu        sync.Mutex
 	logSourceMap map[digest.Digest]interface{}
+	metrics      *metricWriter
 
 	// TODO: remove once we can use result context to pass build ref
 	//  see https://github.com/docker/buildx/pull/1861
@@ -49,6 +53,9 @@ func (p *Printer) Unpause() {
 
 func (p *Printer) Write(s *client.SolveStatus) {
 	p.status <- s
+	if p.metrics != nil {
+		p.metrics.Write(s)
+	}
 }
 
 func (p *Printer) Warnings() []client.VertexWarning {
@@ -96,7 +103,8 @@ func NewPrinter(ctx context.Context, out console.File, mode progressui.DisplayMo
 	}
 
 	pw := &Printer{
-		ready: make(chan struct{}),
+		ready:   make(chan struct{}),
+		metrics: opt.mw,
 	}
 	go func() {
 		for {
@@ -147,6 +155,7 @@ func (p *Printer) BuildRefs() map[string]string {
 
 type printerOpts struct {
 	displayOpts []progressui.DisplayOpt
+	mw          *metricWriter
 
 	onclose func()
 }
@@ -165,6 +174,14 @@ func WithDesc(text string, console string) PrinterOpt {
 	}
 }
 
+func WithMetrics(mp metric.MeterProvider, attrs attribute.Set) PrinterOpt {
+	return func(opt *printerOpts) {
+		if confutil.IsExperimental() {
+			opt.mw = newMetrics(mp, attrs)
+		}
+	}
+}
+
 func WithOnClose(onclose func()) PrinterOpt {
 	return func(opt *printerOpts) {
 		opt.onclose = onclose