Add snapshot and diff service

Remove rootfs service in place of snapshot service. Adds
diff service for extracting and creating diffs. Diff
creation is not yet implemented. This service allows
pulling or creating images without needing root access to
mount. Additionally in the future this will allow containerd
to ensure extractions happen safely in a chroot if needed.

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

rootfs/apply.go

@@ -2,13 +2,9 @@ package rootfs
 
 import (
 	"context"
-	"io"
-	"io/ioutil"
-	"os"
+	"fmt"
 
 	"github.com/containerd/containerd"
-	"github.com/containerd/containerd/archive"
-	"github.com/containerd/containerd/archive/compression"
 	"github.com/containerd/containerd/log"
 	"github.com/containerd/containerd/snapshot"
 	"github.com/opencontainers/go-digest"
@@ -17,134 +13,72 @@ import (
 	"github.com/pkg/errors"
 )
 
-type Unpacker interface {
-	Unpack(ctx context.Context, layers []ocispec.Descriptor) (digest.Digest, error)
+type Applier interface {
+	Apply(context.Context, ocispec.Descriptor, []containerd.Mount) (ocispec.Descriptor, error)
 }
 
-type Mounter interface {
-	Mount(target string, mounts ...containerd.Mount) error
-	Unmount(target string) error
+type Layer struct {
+	Diff ocispec.Descriptor
+	Blob ocispec.Descriptor
 }
 
-// ApplyLayer applies the layer to the provided parent. The resulting snapshot
-// will be stored under its ChainID.
-//
-// The parent *must* be the chainID of the parent layer.
-//
-// The returned digest is the diffID for the applied layer.
-func ApplyLayer(snapshots snapshot.Snapshotter, mounter Mounter, rd io.Reader, parent digest.Digest) (digest.Digest, error) {
-	ctx := context.TODO()
-
-	// create a temporary directory to work from, needs to be on same
-	// filesystem. Probably better if this shared but we'll use a tempdir, for
-	// now.
-	dir, err := ioutil.TempDir("", "unpack-")
-	if err != nil {
-		return "", errors.Wrapf(err, "creating temporary directory failed")
-	}
-	defer os.RemoveAll(dir)
-
-	// TODO(stevvooe): Choose this key WAY more carefully. We should be able to
-	// create collisions for concurrent, conflicting unpack processes but we
-	// would need to have it be a function of the parent diffID and child
-	// layerID (since we don't know the diffID until we are done!).
-	key := dir
-
-	mounts, err := snapshots.Prepare(ctx, key, parent.String())
-	if err != nil {
-		return "", err
-	}
-
-	if err := mounter.Mount(dir, mounts...); err != nil {
-		if err := snapshots.Remove(ctx, key); err != nil {
-			log.L.WithError(err).Error("snapshot rollback failed")
+func ApplyLayers(ctx context.Context, layers []Layer, sn snapshot.Snapshotter, a Applier) (digest.Digest, error) {
+	var chain []digest.Digest
+	for _, layer := range layers {
+		if err := applyLayer(ctx, layer, chain, sn, a); err != nil {
+			// TODO: possibly wait and retry if extraction of same chain id was in progress
+			return "", err
 		}
-		return "", err
+
+		chain = append(chain, layer.Diff.Digest)
 	}
-	defer mounter.Unmount(dir)
-
-	rd, err = compression.DecompressStream(rd)
-	if err != nil {
-		return "", err
-	}
-
-	digester := digest.Canonical.Digester() // used to calculate diffID.
-	rd = io.TeeReader(rd, digester.Hash())
-
-	if _, err := archive.Apply(context.Background(), dir, rd); err != nil {
-		return "", err
-	}
-
-	diffID := digester.Digest()
-
-	chainID := diffID
-	if parent != "" {
-		chainID = identity.ChainID([]digest.Digest{parent, chainID})
-	}
-	if _, err := snapshots.Stat(ctx, chainID.String()); err == nil {
-		return diffID, snapshots.Remove(ctx, key)
-	}
-
-	return diffID, snapshots.Commit(ctx, chainID.String(), key)
+	return identity.ChainID(chain), nil
 }
 
-// Prepare the root filesystem from the set of layers. Snapshots are created
-// for each layer if they don't exist, keyed by their chain id. If the snapshot
-// already exists, it will be skipped.
-//
-// If successful, the chainID for the top-level layer is returned. That
-// identifier can be used to check out a snapshot.
-func Prepare(ctx context.Context, snapshots snapshot.Snapshotter, mounter Mounter, layers []ocispec.Descriptor,
-	// TODO(stevvooe): The following functions are candidate for internal
-	// object functions. We can use these to formulate the beginnings of a
-	// rootfs Controller.
-	//
-	// Just pass them in for now.
-	openBlob func(context.Context, digest.Digest) (io.ReadCloser, error),
-	resolveDiffID func(digest.Digest) digest.Digest,
-	registerDiffID func(diffID, dgst digest.Digest) error) (digest.Digest, error) {
+func applyLayer(ctx context.Context, layer Layer, chain []digest.Digest, sn snapshot.Snapshotter, a Applier) error {
 	var (
-		parent digest.Digest
-		chain  []digest.Digest
+		parent  = identity.ChainID(chain)
+		chainID = identity.ChainID(append(chain, layer.Diff.Digest))
+		diff    ocispec.Descriptor
 	)
 
-	for _, layer := range layers {
-		// This will convert a possibly compressed layer hash to the
-		// uncompressed hash, if we know about it. If we don't, we unpack and
-		// calculate it. If we do have it, we then calculate the chain id for
-		// the application and see if the snapshot is there.
-		diffID := resolveDiffID(layer.Digest)
-		if diffID != "" {
-			chainLocal := append(chain, diffID)
-			chainID := identity.ChainID(chainLocal)
-
-			if _, err := snapshots.Stat(ctx, chainID.String()); err == nil {
-				continue
-			}
-		}
-
-		rc, err := openBlob(ctx, layer.Digest)
-		if err != nil {
-			return "", err
-		}
-		defer rc.Close() // pretty lazy!
-
-		diffID, err = ApplyLayer(snapshots, mounter, rc, parent)
-		if err != nil {
-			return "", err
-		}
-
-		// Register the association between the diffID and the layer's digest.
-		// For uncompressed layers, this will be the same. For compressed
-		// layers, we can look up the diffID from the digest if we've already
-		// unpacked it.
-		if err := registerDiffID(diffID, layer.Digest); err != nil {
-			return "", err
-		}
-
-		chain = append(chain, diffID)
-		parent = identity.ChainID(chain)
+	_, err := sn.Stat(ctx, chainID.String())
+	if err == nil {
+		log.G(ctx).Debugf("Extraction not needed, layer snapshot exists")
+		return nil
+	} else if !snapshot.IsNotExist(err) {
+		return errors.Wrap(err, "failed to stat snapshot")
 	}
 
-	return parent, nil
+	key := fmt.Sprintf("extract %s", chainID)
+
+	// Prepare snapshot with from parent
+	mounts, err := sn.Prepare(ctx, key, parent.String())
+	if err != nil {
+		//TODO: If is snapshot exists error, retry
+		return errors.Wrap(err, "failed to prepare extraction layer")
+	}
+	defer func() {
+		if err != nil {
+			log.G(ctx).WithError(err).WithField("key", key).Infof("Apply failure, attempting cleanup")
+			if rerr := sn.Remove(ctx, key); rerr != nil {
+				log.G(ctx).WithError(rerr).Warnf("Extraction snapshot %q removal failed: %v", key)
+			}
+		}
+	}()
+
+	diff, err = a.Apply(ctx, layer.Blob, mounts)
+	if err != nil {
+		return errors.Wrapf(err, "failed to extract layer %s", layer.Diff.Digest)
+	}
+	if diff.Digest != layer.Diff.Digest {
+		err = errors.Errorf("wrong diff id calculated on extraction %q", diff.Digest)
+		return err
+	}
+
+	if err = sn.Commit(ctx, chainID.String(), key); err != nil {
+		return errors.Wrapf(err, "failed to commit snapshot %s", parent)
+	}
+
+	return nil
 }

rootfs/init.go

@@ -19,6 +19,11 @@ var (
 
 type initializerFunc func(string) error
 
+type Mounter interface {
+	Mount(target string, mounts ...containerd.Mount) error
+	Unmount(target string) error
+}
+
 func InitRootFS(ctx context.Context, name string, parent digest.Digest, readonly bool, snapshotter snapshot.Snapshotter, mounter Mounter) ([]containerd.Mount, error) {
 	_, err := snapshotter.Stat(ctx, name)
 	if err == nil {