Merge pull request #1653 from dmcgowan/fix-pull-race

gc: fix pull race

client.go

@@ -228,7 +228,7 @@ func (c *Client) Pull(ctx context.Context, ref string, opts ...RemoteOpt) (Image
 		handler = images.Handlers(append(pullCtx.BaseHandlers, schema1Converter)...)
 	} else {
 		handler = images.Handlers(append(pullCtx.BaseHandlers,
-			remotes.FetchHandler(store, fetcher),
+			remotes.FetchHandler(store, fetcher, desc),
 			images.ChildrenHandler(store, platforms.Default()))...,
 		)
 	}
@@ -265,6 +265,11 @@ func (c *Client) Pull(ctx context.Context, ref string, opts ...RemoteOpt) (Image
 		imgrec = created
 	}
 
+	// Remove root tag from manifest now that image refers to it
+	if _, err := store.Update(ctx, content.Info{Digest: desc.Digest}, "labels.containerd.io/gc.root"); err != nil {
+		return nil, errors.Wrap(err, "failed to remove manifest root tag")
+	}
+
 	img := &image{
 		client: c,
 		i:      imgrec,

content/helpers.go

@@ -48,7 +48,7 @@ func ReadBlob(ctx context.Context, provider Provider, dgst digest.Digest) ([]byt
 // This is useful when the digest and size are known beforehand.
 //
 // Copy is buffered, so no need to wrap reader in buffered io.
-func WriteBlob(ctx context.Context, cs Ingester, ref string, r io.Reader, size int64, expected digest.Digest) error {
+func WriteBlob(ctx context.Context, cs Ingester, ref string, r io.Reader, size int64, expected digest.Digest, opts ...Opt) error {
 	cw, err := cs.Writer(ctx, ref, size, expected)
 	if err != nil {
 		if !errdefs.IsAlreadyExists(err) {
@@ -59,7 +59,7 @@ func WriteBlob(ctx context.Context, cs Ingester, ref string, r io.Reader, size i
 	}
 	defer cw.Close()
 
-	return Copy(ctx, cw, r, size, expected)
+	return Copy(ctx, cw, r, size, expected, opts...)
 }
 
 // Copy copies data with the expected digest from the reader into the

remotes/docker/schema1/converter.go

@@ -134,11 +134,6 @@ func (c *Converter) Convert(ctx context.Context) (ocispec.Descriptor, error) {
 		Size:      int64(len(b)),
 	}
 
-	ref := remotes.MakeRefKey(ctx, config)
-	if err := content.WriteBlob(ctx, c.contentStore, ref, bytes.NewReader(b), config.Size, config.Digest); err != nil {
-		return ocispec.Descriptor{}, errors.Wrap(err, "failed to write config")
-	}
-
 	layers := make([]ocispec.Descriptor, len(diffIDs))
 	for i, diffID := range diffIDs {
 		layers[i] = c.layerBlobs[diffID]
@@ -152,22 +147,40 @@ func (c *Converter) Convert(ctx context.Context) (ocispec.Descriptor, error) {
 		Layers: layers,
 	}
 
-	b, err = json.Marshal(manifest)
+	mb, err := json.Marshal(manifest)
 	if err != nil {
 		return ocispec.Descriptor{}, errors.Wrap(err, "failed to marshal image")
 	}
 
 	desc := ocispec.Descriptor{
 		MediaType: ocispec.MediaTypeImageManifest,
-		Digest:    digest.Canonical.FromBytes(b),
+		Digest:    digest.Canonical.FromBytes(mb),
-		Size:      int64(len(b)),
+		Size:      int64(len(mb)),
 	}
 
-	ref = remotes.MakeRefKey(ctx, desc)
+	labels := map[string]string{}
-	if err := content.WriteBlob(ctx, c.contentStore, ref, bytes.NewReader(b), desc.Size, desc.Digest); err != nil {
+	labels["containerd.io/gc.root"] = time.Now().UTC().Format(time.RFC3339)
+	labels["containerd.io/gc.ref.content.0"] = manifest.Config.Digest.String()
+	for i, ch := range manifest.Layers {
+		labels[fmt.Sprintf("containerd.io/gc.ref.content.%d", i+1)] = ch.Digest.String()
+	}
+
+	ref := remotes.MakeRefKey(ctx, desc)
+	if err := content.WriteBlob(ctx, c.contentStore, ref, bytes.NewReader(mb), desc.Size, desc.Digest, content.WithLabels(labels)); err != nil {
 		return ocispec.Descriptor{}, errors.Wrap(err, "failed to write config")
 	}
 
+	ref = remotes.MakeRefKey(ctx, config)
+	if err := content.WriteBlob(ctx, c.contentStore, ref, bytes.NewReader(b), config.Size, config.Digest); err != nil {
+		return ocispec.Descriptor{}, errors.Wrap(err, "failed to write config")
+	}
+
+	for _, ch := range manifest.Layers {
+		if _, err := c.contentStore.Update(ctx, content.Info{Digest: ch.Digest}, "labels.containerd.io/gc.root"); err != nil {
+			return ocispec.Descriptor{}, errors.Wrap(err, "failed to remove blob root tag")
+		}
+	}
+
 	return desc, nil
 }
 
@@ -257,14 +270,15 @@ func (c *Converter) fetchBlob(ctx context.Context, desc ocispec.Descriptor) erro
 
 		eg.Go(func() error {
 			defer pw.Close()
-			return content.Copy(ctx, cw, io.TeeReader(rc, pw), desc.Size, desc.Digest)
+			opt := content.WithLabels(map[string]string{
+				"containerd.io/gc.root": time.Now().UTC().Format(time.RFC3339),
+			})
+			return content.Copy(ctx, cw, io.TeeReader(rc, pw), desc.Size, desc.Digest, opt)
 		})
 
 		if err := eg.Wait(); err != nil {
 			return err
 		}
-
-		// TODO: Label blob
 	}
 
 	if desc.Size == 0 {

remotes/handlers.go

@@ -44,7 +44,7 @@ func MakeRefKey(ctx context.Context, desc ocispec.Descriptor) string {
 // FetchHandler returns a handler that will fetch all content into the ingester
 // discovered in a call to Dispatch. Use with ChildrenHandler to do a full
 // recursive fetch.
-func FetchHandler(ingester content.Ingester, fetcher Fetcher) images.HandlerFunc {
+func FetchHandler(ingester content.Ingester, fetcher Fetcher, root ocispec.Descriptor) images.HandlerFunc {
 	return func(ctx context.Context, desc ocispec.Descriptor) (subdescs []ocispec.Descriptor, err error) {
 		ctx = log.WithLogger(ctx, log.G(ctx).WithFields(logrus.Fields{
 			"digest":    desc.Digest,
@@ -56,13 +56,13 @@ func FetchHandler(ingester content.Ingester, fetcher Fetcher) images.HandlerFunc
 		case images.MediaTypeDockerSchema1Manifest:
 			return nil, fmt.Errorf("%v not supported", desc.MediaType)
 		default:
-			err := fetch(ctx, ingester, fetcher, desc)
+			err := fetch(ctx, ingester, fetcher, desc, desc.Digest == root.Digest)
 			return nil, err
 		}
 	}
 }
 
-func fetch(ctx context.Context, ingester content.Ingester, fetcher Fetcher, desc ocispec.Descriptor) error {
+func fetch(ctx context.Context, ingester content.Ingester, fetcher Fetcher, desc ocispec.Descriptor, root bool) error {
 	log.G(ctx).Debug("fetch")
 
 	var (
@@ -104,13 +104,13 @@ func fetch(ctx context.Context, ingester content.Ingester, fetcher Fetcher, desc
 	}
 	defer rc.Close()
 
-	r, opts := commitOpts(desc, rc)
+	r, opts := commitOpts(desc, rc, root)
 	return content.Copy(ctx, cw, r, desc.Size, desc.Digest, opts...)
 }
 
 // commitOpts gets the appropriate content options to alter
 // the content info on commit based on media type.
-func commitOpts(desc ocispec.Descriptor, r io.Reader) (io.Reader, []content.Opt) {
+func commitOpts(desc ocispec.Descriptor, r io.Reader, root bool) (io.Reader, []content.Opt) {
 	var childrenF func(r io.Reader) ([]ocispec.Descriptor, error)
 
 	switch desc.MediaType {
@@ -162,10 +162,13 @@ func commitOpts(desc ocispec.Descriptor, r io.Reader) (io.Reader, []content.Opt)
 			return errors.Wrap(err, "unable to get commit labels")
 		}
 
-		if len(children) > 0 {
+		if len(children) > 0 || root {
 			if info.Labels == nil {
 				info.Labels = map[string]string{}
 			}
+			if root {
+				info.Labels["containerd.io/gc.root"] = time.Now().UTC().Format(time.RFC3339)
+			}
 			for i, ch := range children {
 				info.Labels[fmt.Sprintf("containerd.io/gc.ref.content.%d", i)] = ch.Digest.String()
 			}