Merge pull request #1994 from crazy-max/load-progress

build: sublogger to show docker load progress output

build/build.go

@@ -552,7 +552,9 @@ func toSolveOpt(ctx context.Context, node builder.Node, multiDriver bool, opt Op
 						return nil, nil, err
 					}
 					defers = append(defers, cancel)
-					opt.Exports[i].Output = wrapWriteCloser(w)
+					opt.Exports[i].Output = func(_ map[string]string) (io.WriteCloser, error) {
+						return w, nil
+					}
 				}
 			} else if !nodeDriver.Features(ctx)[driver.DockerExporter] {
 				return nil, nil, notSupported(driver.DockerExporter, nodeDriver, "https:/docs.docker.com/go/build-exporters/")
@@ -1609,12 +1611,6 @@ func handleLowercaseDockerfile(dir, p string) string {
 	return p
 }
 
-func wrapWriteCloser(wc io.WriteCloser) func(map[string]string) (io.WriteCloser, error) {
-	return func(map[string]string) (io.WriteCloser, error) {
-		return wc, nil
-	}
-}
-
 var nodeIdentifierMu sync.Mutex
 
 func tryNodeIdentifier(configDir string) (out string) {

util/dockerutil/client.go

@@ -60,8 +60,10 @@ func (c *Client) LoadImage(ctx context.Context, name string, status progress.Wri
 				return
 			}
 
-			prog := progress.WithPrefix(status, "", false)
+			status = progress.ResetTime(status)
-			if err := fromReader(prog, "importing to docker", resp.Body); err != nil {
+			if err := progress.Wrap("importing to docker", status.Write, func(l progress.SubLogger) error {
+				return fromReader(l, resp.Body)
+			}); err != nil {
 				handleErr(err)
 			}
 		},

util/dockerutil/progress.go

@@ -1,47 +1,76 @@
 package dockerutil
 
 import (
-	"errors"
+	"encoding/json"
 	"io"
 	"time"
 
 	"github.com/docker/buildx/util/progress"
-	"github.com/docker/cli/cli/streams"
 	"github.com/docker/docker/pkg/jsonmessage"
 	"github.com/moby/buildkit/client"
-	"github.com/moby/buildkit/identity"
-	"github.com/opencontainers/go-digest"
 )
 
-func fromReader(w progress.Writer, name string, rc io.ReadCloser) error {
+const minTimeDelta = 2 * time.Second
-	dgst := digest.FromBytes([]byte(identity.NewID()))
-	tm := time.Now()
 
-	vtx := client.Vertex{
+func fromReader(l progress.SubLogger, rc io.ReadCloser) error {
-		Digest:  dgst,
+	started := map[string]client.VertexStatus{}
-		Name:    name,
-		Started: &tm,
-	}
 
-	w.Write(&client.SolveStatus{
+	defer func() {
-		Vertexes: []*client.Vertex{&vtx},
+		for _, st := range started {
-	})
+			st := st
-
+			if st.Completed == nil {
-	err := jsonmessage.DisplayJSONMessagesToStream(rc, streams.NewOut(io.Discard), nil)
+				now := time.Now()
-	if err != nil {
+				st.Completed = &now
-		if jerr, ok := err.(*jsonmessage.JSONError); ok {
+				l.SetStatus(&st)
-			err = errors.New(jerr.Message)
 			}
 		}
+	}()
 
-	tm2 := time.Now()
+	dec := json.NewDecoder(rc)
-	vtx2 := vtx
+	var parsedErr error
-	vtx2.Completed = &tm2
+	var jm jsonmessage.JSONMessage
-	if err != nil {
+	for {
-		vtx2.Error = err.Error()
+		if err := dec.Decode(&jm); err != nil {
+			if parsedErr != nil {
+				return parsedErr
+			}
+			if err == io.EOF {
+				break
 			}
-	w.Write(&client.SolveStatus{
-		Vertexes: []*client.Vertex{&vtx2},
-	})
 			return err
 		}
+		if jm.Error != nil {
+			parsedErr = jm.Error
+		}
+		if jm.ID == "" || jm.Progress == nil {
+			continue
+		}
+
+		id := "loading layer " + jm.ID
+		st, ok := started[id]
+		if !ok {
+			now := time.Now()
+			st = client.VertexStatus{
+				ID:      id,
+				Started: &now,
+			}
+		}
+		timeDelta := time.Now().Sub(st.Timestamp)
+		if timeDelta < minTimeDelta {
+			continue
+		}
+		st.Timestamp = time.Now()
+		if jm.Status == "Loading layer" {
+			st.Current = jm.Progress.Current
+			st.Total = jm.Progress.Total
+		}
+		if jm.Error != nil {
+			now := time.Now()
+			st.Completed = &now
+		}
+		started[id] = st
+		l.SetStatus(&st)
+	}
+
+	return nil
+}