Merge pull request #47432 from vvoland/c8d-pull-fslayer

c8d/pull: Progress fixes

daemon/containerd/image_pull.go

@@ -117,6 +117,9 @@ func (i *ImageService) pullTag(ctx context.Context, ref reference.Named, platfor
 			progress.Message(out, "", distribution.DeprecatedSchema1ImageMessage(ref))
 			sentSchema1Deprecation = true
 		}
+		if images.IsLayerType(desc.MediaType) {
+			progress.Update(out, desc.Digest.String(), "Pulling fs layer")
+		}
 		if images.IsManifestType(desc.MediaType) {
 			if !sentPullingFrom {
 				var tagOrDigest string

daemon/containerd/progress.go

@@ -135,6 +135,9 @@ func (p pullProgress) UpdateProgress(ctx context.Context, ongoing *jobs, out pro
 		}
 		key := remotes.MakeRefKey(ctx, j)
 		if info, ok := pulling[key]; ok {
+			if info.Offset == 0 {
+				continue
+			}
 			out.WriteProgress(progress.Progress{
 				ID:      stringid.TruncateID(j.Digest.Encoded()),
 				Action:  "Downloading",

pkg/streamformatter/streamformatter.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"sync"
 
 	"github.com/docker/docker/pkg/jsonmessage"
 	"github.com/docker/docker/pkg/progress"
@@ -109,6 +110,7 @@ type progressOutput struct {
 	sf       formatProgress
 	out      io.Writer
 	newLines bool
+	mu       sync.Mutex
 }
 
 // WriteProgress formats progress information from a ProgressReader.
@@ -120,6 +122,9 @@ func (out *progressOutput) WriteProgress(prog progress.Progress) error {
 		jsonProgress := jsonmessage.JSONProgress{Current: prog.Current, Total: prog.Total, HideCounts: prog.HideCounts, Units: prog.Units}
 		formatted = out.sf.formatProgress(prog.ID, prog.Action, &jsonProgress, prog.Aux)
 	}
+
+	out.mu.Lock()
+	defer out.mu.Unlock()
 	_, err := out.out.Write(formatted)
 	if err != nil {
 		return err