containerd integration: produce progress events polling ctrd's content.Store

[thaJeztah]

• upstreaming produce progress events polling ctrd's content.Store rumpl/moby#12

- Description for the changelog

- A picture of a cute animal (not mandatory but encouraged)

[thaJeztah]

Minor fix needed;

daemon/containerd/progress.go:17: File is not `goimports`-ed (goimports)
	"github.com/opencontainers/image-spec/specs-go/v1"
daemon/containerd/progress.go:100:2: `name` is unused (structcheck)
	name     string
	^

[AkihiroSuda]

(the snapshotter has to be eventually configurable, in a follow-up PR)

[AkihiroSuda]

Should we return an error for schema1?

[cpuguy83]

What's the purpose of this? Seems like this races with your goroutine above.

[cpuguy83]

Seems like we can close the channel (from the goroutine) rather than sending?

[ndeloof]

closing channel would be a valid alternative, indeed

[thaJeztah]

Discussing; we can include rumpl#16 here, but also need to have a look at

moby/plugin/fetch_linux.go

Lines 198 to 285 in f068067

	func withFetchProgress(cs content.Store, out progress.Output, ref reference.Named) images.HandlerFunc {
	return func(ctx context.Context, desc specs.Descriptor) ([]specs.Descriptor, error) {
	switch desc.MediaType {
	case specs.MediaTypeImageManifest, images.MediaTypeDockerSchema2Manifest:
	tn := reference.TagNameOnly(ref)
	tagged := tn.(reference.Tagged)
	progress.Messagef(out, tagged.Tag(), "Pulling from %s", reference.FamiliarName(ref))
	progress.Messagef(out, "", "Digest: %s", desc.Digest.String())
	return nil, nil
	case
	images.MediaTypeDockerSchema2LayerGzip,
	images.MediaTypeDockerSchema2Layer,
	specs.MediaTypeImageLayer,
	specs.MediaTypeImageLayerGzip:
	default:
	return nil, nil
	}
	id := stringid.TruncateID(desc.Digest.String())
	if _, err := cs.Info(ctx, desc.Digest); err == nil {
	out.WriteProgress(progress.Progress{ID: id, Action: "Already exists", LastUpdate: true})
	return nil, nil
	}
	progress.Update(out, id, "Waiting")
	key := remotes.MakeRefKey(ctx, desc)
	go func() {
	timer := time.NewTimer(100 * time.Millisecond)
	if !timer.Stop() {
	<-timer.C
	}
	defer timer.Stop()
	var pulling bool
	var ctxErr error
	for {
	timer.Reset(100 * time.Millisecond)
	select {
	case <-ctx.Done():
	ctxErr = ctx.Err()
	// make sure we can still fetch from the content store
	// TODO: Might need to add some sort of timeout
	ctx = context.Background()
	case <-timer.C:
	}
	s, err := cs.Status(ctx, key)
	if err != nil {
	if !c8derrdefs.IsNotFound(err) {
	logrus.WithError(err).WithField("layerDigest", desc.Digest.String()).Error("Error looking up status of plugin layer pull")
	progress.Update(out, id, err.Error())
	return
	}
	if _, err := cs.Info(ctx, desc.Digest); err == nil {
	progress.Update(out, id, "Download complete")
	return
	}
	if ctxErr != nil {
	progress.Update(out, id, ctxErr.Error())
	return
	}
	continue
	}
	if !pulling {
	progress.Update(out, id, "Pulling fs layer")
	pulling = true
	}
	if s.Offset == s.Total {
	out.WriteProgress(progress.Progress{ID: id, Action: "Download complete", Current: s.Offset, LastUpdate: true})
	return
	}
	out.WriteProgress(progress.Progress{ID: id, Action: "Downloading", Current: s.Offset, Total: s.Total})
	}
	}()
	return nil, nil
	}
	}

to see if we can share the existing implementation

[rumpl]

This one can be closed, superseeded by #44756