Merge pull request #6587 from tonistiigi/fix-push-progress

exporter: fix reporting push progress under export vertex

Author: tonistiigi

client/client_test.go

@@ -163,6 +163,7 @@ var allTests = []func(t *testing.T, sb integration.Sandbox){
 	testHostnameLookup,
 	testHostnameSpecifying,
 	testPushByDigest,
+	testPushProgressSameVertex,
 	testPullWithDigestCheck,
 	testBasicInlineCacheImportExport,
 	testExportBusyboxLocal,
@@ -1316,6 +1317,99 @@ func testPushByDigest(t *testing.T, sb integration.Sandbox) {
 	require.Greater(t, desc.Size, int64(0))
 }
 
+func testPushProgressSameVertex(t *testing.T, sb integration.Sandbox) {
+	workers.CheckFeatureCompat(t, sb, workers.FeatureDirectPush)
+	requiresLinux(t)
+	c, err := New(sb.Context(), sb.Address())
+	require.NoError(t, err)
+	defer c.Close()
+
+	registry, err := sb.NewRegistry()
+	if errors.Is(err, integration.ErrRequirements) {
+		t.Skip(err.Error())
+	}
+	require.NoError(t, err)
+
+	st := llb.Scratch().File(llb.Mkfile("foo", 0600, []byte("data")))
+	def, err := st.Marshal(sb.Context())
+	require.NoError(t, err)
+
+	imageName := registry + "/foo/bar:latest"
+
+	var vertexes []*Vertex
+	var statuses []*VertexStatus
+	ch := make(chan *SolveStatus)
+	eg, ctx := errgroup.WithContext(sb.Context())
+	eg.Go(func() error {
+		for {
+			select {
+			case <-ctx.Done():
+				return context.Cause(ctx)
+			case ss, ok := <-ch:
+				if !ok {
+					return nil
+				}
+				vertexes = append(vertexes, ss.Vertexes...)
+				statuses = append(statuses, ss.Statuses...)
+			}
+		}
+	})
+	eg.Go(func() error {
+		_, err := c.Solve(ctx, def, SolveOpt{
+			Exports: []ExportEntry{
+				{
+					Type: "image",
+					Attrs: map[string]string{
+						"name": imageName,
+						"push": "true",
+					},
+				},
+			},
+		}, ch)
+		return err
+	})
+	require.NoError(t, eg.Wait())
+
+	exportVertexes := map[digest.Digest]struct{}{}
+	pushVertexes := map[digest.Digest]struct{}{}
+	pushManifestPrefix := "pushing manifest for " + imageName + "@"
+
+	for _, v := range vertexes {
+		if v.Name == "exporting to image" {
+			exportVertexes[v.Digest] = struct{}{}
+		}
+	}
+
+	for _, st := range statuses {
+		switch {
+		case st.ID == "pushing layers":
+			pushVertexes[st.Vertex] = struct{}{}
+		case strings.HasPrefix(st.ID, pushManifestPrefix):
+			pushVertexes[st.Vertex] = struct{}{}
+		}
+	}
+
+	require.Len(t, exportVertexes, 1, "expected exactly one export vertex in status stream")
+	require.NotEmpty(t, pushVertexes, "expected at least one push status in status stream")
+
+	var exportVertex digest.Digest
+	for v := range exportVertexes {
+		exportVertex = v
+	}
+	for v := range pushVertexes {
+		require.Equal(t, exportVertex, v, "push statuses should use the same vertex as exporting to image")
+	}
+
+	hasPushManifest := false
+	for _, st := range statuses {
+		if strings.HasPrefix(st.ID, pushManifestPrefix) {
+			hasPushManifest = true
+			require.Equal(t, exportVertex, st.Vertex, "pushing manifest should use the same vertex as exporting to image")
+		}
+	}
+	require.True(t, hasPushManifest, "expected pushing manifest status in status stream")
+}
+
 func testPullWithDigestCheck(t *testing.T, sb integration.Sandbox) {
 	workers.CheckFeatureCompat(t, sb, workers.FeatureDirectPush)
 	requiresLinux(t)

cmd/buildctl/build_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -192,6 +193,28 @@ func testBuildMetadataFile(t *testing.T, sb integration.Sandbox) {
 	}
 }
 
+func testBuildPushProgress(t *testing.T, sb integration.Sandbox) {
+	integration.SkipOnPlatform(t, "windows")
+	registry, err := sb.NewRegistry()
+	if errors.Is(err, integration.ErrRequirements) {
+		t.Skip(err.Error())
+	}
+	require.NoError(t, err)
+
+	st := llb.Scratch().File(llb.Mkfile("foo", 0600, []byte("data")))
+	rdr, err := marshal(sb.Context(), st)
+	require.NoError(t, err)
+
+	imageName := registry + "/foo/bar:latest"
+	cmd := sb.Cmd("build", "--progress=plain", "--output", "type=image,name="+imageName+",push=true")
+	cmd.Stdin = rdr
+
+	dt, err := cmd.CombinedOutput()
+	require.NoError(t, err, string(dt))
+	require.Contains(t, string(dt), "pushing layers")
+	require.Contains(t, string(dt), "pushing manifest for "+imageName+"@")
+}
+
 func marshal(ctx context.Context, st llb.State) (io.Reader, error) {
 	def, err := st.Marshal(ctx)
 	if err != nil {

cmd/buildctl/buildctl_test.go

@@ -23,6 +23,7 @@ func TestCLIIntegration(t *testing.T) {
 		testBuildLocalExporter,
 		testBuildContainerdExporter,
 		testBuildMetadataFile,
+		testBuildPushProgress,
 		testPrune,
 		testUsage,
 	),

solver/llbsolver/export.go

@@ -152,6 +152,10 @@ func runInlineCacheExporter(ctx context.Context, e exporter.ExporterInstance, in
 	return res, done(err)
 }
 
+func exporterVertexID(sessionID string, exporterIndex int) string {
+	return fmt.Sprint(sessionID, "-export-", exporterIndex)
+}
+
 func (s *Solver) runExporters(ctx context.Context, ref string, exporters []exporter.ExporterInstance, inlineCacheExporter inlineCacheExporter, job *solver.Job, cached *result.Result[solver.CachedResult], inp *exporter.Source) (exporterResponse map[string]string, finalizers []exporter.FinalizeFunc, descrefs []exporter.DescriptorReference, err error) {
 	warnings, err := verifier.CheckInvalidPlatforms(ctx, inp)
 	if err != nil {
@@ -164,8 +168,8 @@ func (s *Solver) runExporters(ctx context.Context, ref string, exporters []expor
 	descs := make([]exporter.DescriptorReference, len(exporters))
 	var inlineCacheMu sync.Mutex
 	for i, exp := range exporters {
+		id := exporterVertexID(job.SessionID, i)
 		eg.Go(func() error {
-			id := fmt.Sprint(job.SessionID, "-export-", i)
 			return inBuilderContext(ctx, job, exp.Name(), id, func(ctx context.Context, _ solver.JobContext) error {
 				span, ctx := tracing.StartSpan(ctx, exp.Name())
 				defer span.End()

solver/llbsolver/solver.go

@@ -345,12 +345,16 @@ func (s *Solver) Solve(ctx context.Context, id string, sessionID string, req fro
 	// Image Export has already created layers in the content store,
 	// so cache exporters can see and reuse them.
 	eg, egCtx := errgroup.WithContext(ctx)
-	for _, finalize := range finalizers {
+	for i, finalize := range finalizers {
 		if finalize == nil {
 			continue
 		}
+		name := exp.Exporters[i].Name()
+		id := exporterVertexID(j.SessionID, i)
 		eg.Go(func() error {
-			return finalize(egCtx)
+			return inBuilderContext(egCtx, j, name, id, func(ctx context.Context, _ solver.JobContext) error {
+				return finalize(ctx)
+			})
 		})
 	}
 	var cacheExporterResponse map[string]string