Update dist pull to use client

Replaced pull unpacker with boolean to call unpack.
Added unpack and target to image type.
Updated progress logic for pull.
Added list images to client.
Updated rootfs unpacker to use client.

Signed-off-by: Derek McGowan <derek@mcgstyle.net>

cmd/dist/fetch.go

@@ -9,15 +9,15 @@ import (
 	"text/tabwriter"
 	"time"
 
-	contentapi "github.com/containerd/containerd/api/services/content"
+	"github.com/containerd/containerd"
+	"github.com/containerd/containerd/content"
 	"github.com/containerd/containerd/images"
 	"github.com/containerd/containerd/log"
 	"github.com/containerd/containerd/progress"
 	"github.com/containerd/containerd/remotes"
-	contentservice "github.com/containerd/containerd/services/content"
+	digest "github.com/opencontainers/go-digest"
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/urfave/cli"
-	"golang.org/x/sync/errgroup"
 )
 
 var fetchCommand = cli.Command{
@@ -47,174 +47,210 @@ Most of this is experimental and there are few leaps to make this work.`,
 		ctx, cancel := appContext(clicontext)
 		defer cancel()
 
-		conn, err := connectGRPC(clicontext)
-		if err != nil {
-			return err
-		}
-
-		resolver, err := getResolver(ctx, clicontext)
-		if err != nil {
-			return err
-		}
-
-		ongoing := newJobs()
-
-		content := contentservice.NewStoreFromClient(contentapi.NewContentClient(conn))
-
-		// TODO(stevvooe): Need to replace this with content store client.
-		cs, err := resolveContentStore(clicontext)
-		if err != nil {
-			return err
-		}
-
-		eg, ctx := errgroup.WithContext(ctx)
-
-		resolved := make(chan struct{})
-		eg.Go(func() error {
-			ongoing.add(ref)
-			name, desc, err := resolver.Resolve(ctx, ref)
-			if err != nil {
-				return err
-			}
-			fetcher, err := resolver.Fetcher(ctx, name)
-			if err != nil {
-				return err
-			}
-			log.G(ctx).WithField("image", name).Debug("fetching")
-			close(resolved)
-
-			return images.Dispatch(ctx,
-				images.Handlers(images.HandlerFunc(func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
-					ongoing.add(remotes.MakeRefKey(ctx, desc))
-					return nil, nil
-				}),
-					remotes.FetchHandler(content, fetcher),
-					images.ChildrenHandler(content),
-				),
-				desc)
-		})
-
-		errs := make(chan error)
-		go func() {
-			defer close(errs)
-			errs <- eg.Wait()
-		}()
-
-		ticker := time.NewTicker(100 * time.Millisecond)
-		fw := progress.NewWriter(os.Stdout)
-		start := time.Now()
-		defer ticker.Stop()
-		var done bool
-
-		for {
-			select {
-			case <-ticker.C:
-				fw.Flush()
-
-				tw := tabwriter.NewWriter(fw, 1, 8, 1, ' ', 0)
-
-				statuses := map[string]statusInfo{}
-
-				activeSeen := map[string]struct{}{}
-				if !done {
-					active, err := cs.Status(ctx, "")
-					if err != nil {
-						log.G(ctx).WithError(err).Error("active check failed")
-						continue
-					}
-					// update status of active entries!
-					for _, active := range active {
-						statuses[active.Ref] = statusInfo{
-							Ref:       active.Ref,
-							Status:    "downloading",
-							Offset:    active.Offset,
-							Total:     active.Total,
-							StartedAt: active.StartedAt,
-							UpdatedAt: active.UpdatedAt,
-						}
-						activeSeen[active.Ref] = struct{}{}
-					}
-				}
-
-				js := ongoing.jobs()
-				// now, update the items in jobs that are not in active
-				for _, j := range js {
-					if _, ok := activeSeen[j]; ok {
-						continue
-					}
-					status := "done"
-
-					if j == ref {
-						select {
-						case <-resolved:
-							status = "resolved"
-						default:
-							status = "resolving"
-						}
-					}
-
-					statuses[j] = statusInfo{
-						Ref:    j,
-						Status: status, // for now!
-					}
-				}
-
-				var ordered []statusInfo
-				for _, j := range js {
-					ordered = append(ordered, statuses[j])
-				}
-
-				display(tw, ordered, start)
-				tw.Flush()
-
-				if done {
-					fw.Flush()
-					return nil
-				}
-			case err := <-errs:
-				if err != nil {
-					return err
-				}
-				done = true
-			case <-ctx.Done():
-				done = true // allow ui to update once more
-			}
-		}
+		_, err := fetch(ctx, ref, clicontext)
+		return err
 	},
 }
 
+func fetch(ctx context.Context, ref string, clicontext *cli.Context) (containerd.Image, error) {
+	client, err := getClient(clicontext)
+	if err != nil {
+		return nil, err
+	}
+
+	resolver, err := getResolver(ctx, clicontext)
+	if err != nil {
+		return nil, err
+	}
+
+	ongoing := newJobs(ref)
+
+	pctx, stopProgress := context.WithCancel(ctx)
+	progress := make(chan struct{})
+
+	go func() {
+		showProgress(pctx, ongoing, client.ContentStore(), os.Stdout)
+		close(progress)
+	}()
+
+	h := images.HandlerFunc(func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
+		ongoing.add(desc)
+		return nil, nil
+	})
+
+	log.G(pctx).WithField("image", ref).Debug("fetching")
+
+	img, err := client.Pull(pctx, ref, containerd.WithResolver(resolver), containerd.WithImageHandler(h))
+	stopProgress()
+	if err != nil {
+		return nil, err
+	}
+
+	<-progress
+	return img, nil
+}
+
+func showProgress(ctx context.Context, ongoing *jobs, cs content.Store, out io.Writer) {
+	var (
+		ticker   = time.NewTicker(100 * time.Millisecond)
+		fw       = progress.NewWriter(out)
+		start    = time.Now()
+		statuses = map[string]statusInfo{}
+		done     bool
+	)
+	defer ticker.Stop()
+
+outer:
+	for {
+		select {
+		case <-ticker.C:
+			fw.Flush()
+
+			tw := tabwriter.NewWriter(fw, 1, 8, 1, ' ', 0)
+
+			resolved := "resolved"
+			if !ongoing.isResolved() {
+				resolved = "resolving"
+			}
+			statuses[ongoing.name] = statusInfo{
+				Ref:    ongoing.name,
+				Status: resolved,
+			}
+			keys := []string{ongoing.name}
+
+			activeSeen := map[string]struct{}{}
+			if !done {
+				active, err := cs.Status(ctx, "")
+				if err != nil {
+					log.G(ctx).WithError(err).Error("active check failed")
+					continue
+				}
+				// update status of active entries!
+				for _, active := range active {
+					statuses[active.Ref] = statusInfo{
+						Ref:       active.Ref,
+						Status:    "downloading",
+						Offset:    active.Offset,
+						Total:     active.Total,
+						StartedAt: active.StartedAt,
+						UpdatedAt: active.UpdatedAt,
+					}
+					activeSeen[active.Ref] = struct{}{}
+				}
+			}
+
+			// now, update the items in jobs that are not in active
+			for _, j := range ongoing.jobs() {
+				key := remotes.MakeRefKey(ctx, j)
+				keys = append(keys, key)
+				if _, ok := activeSeen[key]; ok {
+					continue
+				}
+
+				status, ok := statuses[key]
+				if !done && (!ok || status.Status == "downloading") {
+					info, err := cs.Info(ctx, j.Digest)
+					if err != nil {
+						if !content.IsNotFound(err) {
+							log.G(ctx).WithError(err).Errorf("failed to get content info")
+							continue outer
+						} else {
+							statuses[key] = statusInfo{
+								Ref:    key,
+								Status: "waiting",
+							}
+						}
+					} else if info.CommittedAt.After(start) {
+						statuses[key] = statusInfo{
+							Ref:       key,
+							Status:    "done",
+							Offset:    info.Size,
+							Total:     info.Size,
+							UpdatedAt: info.CommittedAt,
+						}
+					} else {
+						statuses[key] = statusInfo{
+							Ref:    key,
+							Status: "exists",
+						}
+					}
+				} else if done {
+					if ok {
+						if status.Status != "done" && status.Status != "exists" {
+							status.Status = "done"
+							statuses[key] = status
+						}
+					} else {
+						statuses[key] = statusInfo{
+							Ref:    key,
+							Status: "done",
+						}
+					}
+				}
+			}
+
+			var ordered []statusInfo
+			for _, key := range keys {
+				ordered = append(ordered, statuses[key])
+			}
+
+			display(tw, ordered, start)
+			tw.Flush()
+
+			if done {
+				fw.Flush()
+				return
+			}
+		case <-ctx.Done():
+			done = true // allow ui to update once more
+		}
+	}
+}
+
 // jobs provides a way of identifying the download keys for a particular task
 // encountering during the pull walk.
 //
 // This is very minimal and will probably be replaced with something more
 // featured.
 type jobs struct {
-	added map[string]struct{}
-	refs  []string
-	mu    sync.Mutex
+	name     string
+	added    map[digest.Digest]struct{}
+	descs    []ocispec.Descriptor
+	mu       sync.Mutex
+	resolved bool
 }
 
-func newJobs() *jobs {
-	return &jobs{added: make(map[string]struct{})}
+func newJobs(name string) *jobs {
+	return &jobs{
+		name:  name,
+		added: map[digest.Digest]struct{}{},
+	}
 }
 
-func (j *jobs) add(ref string) {
+func (j *jobs) add(desc ocispec.Descriptor) {
 	j.mu.Lock()
 	defer j.mu.Unlock()
+	j.resolved = true
 
-	if _, ok := j.added[ref]; ok {
+	if _, ok := j.added[desc.Digest]; ok {
 		return
 	}
-	j.refs = append(j.refs, ref)
-	j.added[ref] = struct{}{}
+	j.descs = append(j.descs, desc)
+	j.added[desc.Digest] = struct{}{}
 }
 
-func (j *jobs) jobs() []string {
+func (j *jobs) jobs() []ocispec.Descriptor {
 	j.mu.Lock()
 	defer j.mu.Unlock()
 
-	var jobs []string
-	return append(jobs, j.refs...)
+	var descs []ocispec.Descriptor
+	return append(descs, j.descs...)
+}
+
+func (j *jobs) isResolved() bool {
+	j.mu.Lock()
+	defer j.mu.Unlock()
+	return j.resolved
 }
 
 type statusInfo struct {

cmd/dist/pull.go

@@ -1,23 +1,10 @@
 package main
 
 import (
-	"context"
-	"os"
-	"text/tabwriter"
-	"time"
+	"fmt"
 
-	diffapi "github.com/containerd/containerd/api/services/diff"
-	snapshotapi "github.com/containerd/containerd/api/services/snapshot"
-	"github.com/containerd/containerd/images"
 	"github.com/containerd/containerd/log"
-	"github.com/containerd/containerd/progress"
-	"github.com/containerd/containerd/remotes"
-	"github.com/containerd/containerd/rootfs"
-	diffservice "github.com/containerd/containerd/services/diff"
-	snapshotservice "github.com/containerd/containerd/services/snapshot"
-	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/urfave/cli"
-	"golang.org/x/sync/errgroup"
 )
 
 var pullCommand = cli.Command{
@@ -42,175 +29,17 @@ command. As part of this process, we do the following:
 		ctx, cancel := appContext(clicontext)
 		defer cancel()
 
-		cs, err := resolveContentStore(clicontext)
+		img, err := fetch(ctx, ref, clicontext)
 		if err != nil {
 			return err
 		}
 
-		imageStore, err := resolveImageStore(clicontext)
-		if err != nil {
-			return err
-		}
-
-		resolver, err := getResolver(ctx, clicontext)
-		if err != nil {
-			return err
-		}
-		ongoing := newJobs()
-
-		eg, ctx := errgroup.WithContext(ctx)
-
-		var resolvedImageName string
-		resolved := make(chan struct{})
-		eg.Go(func() error {
-			ongoing.add(ref)
-			name, desc, err := resolver.Resolve(ctx, ref)
-			if err != nil {
-				log.G(ctx).WithError(err).Error("failed to resolve")
-				return err
-			}
-			fetcher, err := resolver.Fetcher(ctx, name)
-			if err != nil {
-				return err
-			}
-
-			log.G(ctx).WithField("image", name).Debug("fetching")
-			resolvedImageName = name
-			close(resolved)
-
-			eg.Go(func() error {
-				return imageStore.Put(ctx, name, desc)
-			})
-
-			return images.Dispatch(ctx,
-				images.Handlers(images.HandlerFunc(func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
-					ongoing.add(remotes.MakeRefKey(ctx, desc))
-					return nil, nil
-				}),
-					remotes.FetchHandler(cs, fetcher),
-					images.ChildrenHandler(cs)),
-				desc)
-
-		})
-
-		errs := make(chan error)
-		go func() {
-			defer close(errs)
-			errs <- eg.Wait()
-		}()
-
-		defer func() {
-			// we need new ctx here, since we run on return.
-			ctx, cancel := appContext(clicontext)
-			defer cancel()
-			image, err := imageStore.Get(ctx, resolvedImageName)
-			if err != nil {
-				log.G(ctx).WithError(err).Fatal("failed to get image")
-			}
-
-			layers, err := getImageLayers(ctx, image, cs)
-			if err != nil {
-				log.G(ctx).WithError(err).Fatal("failed to get rootfs layers")
-			}
-
-			conn, err := connectGRPC(clicontext)
-			if err != nil {
-				log.G(ctx).Fatal(err)
-			}
-			snapshotter := snapshotservice.NewSnapshotterFromClient(snapshotapi.NewSnapshotClient(conn))
-			applier := diffservice.NewDiffServiceFromClient(diffapi.NewDiffClient(conn))
-
-			log.G(ctx).Info("unpacking rootfs")
-
-			chainID, err := rootfs.ApplyLayers(ctx, layers, snapshotter, applier)
-			if err != nil {
-				log.G(ctx).Fatal(err)
-			}
-
-			log.G(ctx).Infof("Unpacked chain id: %s", chainID)
-		}()
-
-		var (
-			ticker = time.NewTicker(100 * time.Millisecond)
-			fw     = progress.NewWriter(os.Stdout)
-			start  = time.Now()
-			done   bool
-		)
-		defer ticker.Stop()
-
-		for {
-			select {
-			case <-ticker.C:
-				fw.Flush()
-
-				tw := tabwriter.NewWriter(fw, 1, 8, 1, ' ', 0)
-				js := ongoing.jobs()
-				statuses := map[string]statusInfo{}
-
-				activeSeen := map[string]struct{}{}
-				if !done {
-					active, err := cs.Status(ctx, "")
-					if err != nil {
-						log.G(ctx).WithError(err).Error("active check failed")
-						continue
-					}
-					// update status of active entries!
-					for _, active := range active {
-						statuses[active.Ref] = statusInfo{
-							Ref:       active.Ref,
-							Status:    "downloading",
-							Offset:    active.Offset,
-							Total:     active.Total,
-							StartedAt: active.StartedAt,
-							UpdatedAt: active.UpdatedAt,
-						}
-						activeSeen[active.Ref] = struct{}{}
-					}
-				}
-
-				// now, update the items in jobs that are not in active
-				for _, j := range js {
-					if _, ok := activeSeen[j]; ok {
-						continue
-					}
-					status := "done"
-
-					if j == ref {
-						select {
-						case <-resolved:
-							status = "resolved"
-						default:
-							status = "resolving"
-						}
-					}
-
-					statuses[j] = statusInfo{
-						Ref:    j,
-						Status: status, // for now!
-					}
-				}
-
-				var ordered []statusInfo
-				for _, j := range js {
-					ordered = append(ordered, statuses[j])
-				}
-
-				display(tw, ordered, start)
-				tw.Flush()
-
-				if done {
-					fw.Flush()
-					return nil
-				}
-			case err := <-errs:
-				if err != nil {
-					return err
-				}
-				done = true
-			case <-ctx.Done():
-				done = true // allow ui to update once more
-			}
-		}
+		log.G(ctx).WithField("image", ref).Debug("unpacking")
 
+		// TODO: Show unpack status
+		fmt.Printf("unpacking %s...", img.Target().Digest)
+		err = img.Unpack(ctx)
+		fmt.Println("done")
+		return err
 	},
 }

cmd/dist/rootfs.go

@@ -1,19 +1,15 @@
 package main
 
 import (
+	"errors"
 	"fmt"
 	"os"
 	"strings"
 
-	diffapi "github.com/containerd/containerd/api/services/diff"
 	snapshotapi "github.com/containerd/containerd/api/services/snapshot"
-	"github.com/containerd/containerd/images"
 	"github.com/containerd/containerd/log"
-	"github.com/containerd/containerd/rootfs"
-	diffservice "github.com/containerd/containerd/services/diff"
 	snapshotservice "github.com/containerd/containerd/services/snapshot"
 	digest "github.com/opencontainers/go-digest"
-	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/urfave/cli"
 )
 
@@ -40,39 +36,39 @@ var rootfsUnpackCommand = cli.Command{
 			return err
 		}
 
-		log.G(ctx).Infof("unpacking layers from manifest %s", dgst.String())
+		log.G(ctx).Debugf("unpacking layers from manifest %s", dgst.String())
 
-		cs, err := resolveContentStore(clicontext)
+		client, err := getClient(clicontext)
 		if err != nil {
 			return err
 		}
 
-		image := images.Image{
-			Target: ocispec.Descriptor{
-				MediaType: ocispec.MediaTypeImageManifest,
-				Digest:    dgst,
-			},
-		}
+		// TODO: Support unpack by name
 
-		layers, err := getImageLayers(ctx, image, cs)
-		if err != nil {
-			log.G(ctx).WithError(err).Fatal("Failed to get rootfs layers")
-		}
-
-		conn, err := connectGRPC(clicontext)
-		if err != nil {
-			log.G(ctx).Fatal(err)
-		}
-
-		snapshotter := snapshotservice.NewSnapshotterFromClient(snapshotapi.NewSnapshotClient(conn))
-		applier := diffservice.NewDiffServiceFromClient(diffapi.NewDiffClient(conn))
-
-		chainID, err := rootfs.ApplyLayers(ctx, layers, snapshotter, applier)
+		images, err := client.ListImages(ctx)
 		if err != nil {
 			return err
 		}
 
-		log.G(ctx).Infof("chain ID: %s", chainID.String())
+		var unpacked bool
+		for _, image := range images {
+			if image.Target().Digest == dgst {
+				fmt.Printf("unpacking %s (%s)...", dgst, image.Target().MediaType)
+				if err := image.Unpack(ctx); err != nil {
+					fmt.Println()
+					return err
+				}
+				fmt.Println("done")
+				unpacked = true
+				break
+			}
+		}
+		if !unpacked {
+			return errors.New("manifest not found")
+		}
+
+		// TODO: Get rootfs from Image
+		//log.G(ctx).Infof("chain ID: %s", chainID.String())
 
 		return nil
 	},