Merge pull request #849 from dmcgowan/snapshot-service

Add snapshot and diff service

api/services/diff/diff.proto

@@ -0,0 +1,58 @@
+syntax = "proto3";
+
+package containerd.v1;
+
+import "gogoproto/gogo.proto";
+import "google/protobuf/empty.proto";
+import "google/protobuf/timestamp.proto";
+import "github.com/containerd/containerd/api/types/mount/mount.proto";
+import "github.com/containerd/containerd/api/types/descriptor/descriptor.proto";
+
+// Diff service creates and applies diffs
+service Diff {
+	// Apply applies the content associated with the provided digests onto
+	// the provided mounts. Archive content will be extracted and
+	// decompressed if necessary.
+	rpc Apply(ApplyRequest) returns (ApplyResponse);
+
+	// Diff creates a diff between the given mounts and uploads the result
+	// to the content store.
+	rpc Diff(DiffRequest) returns (DiffResponse);
+}
+
+message ApplyRequest {
+	// Diff is the descriptor of the diff to be extracted
+	containerd.v1.types.Descriptor diff = 1;
+
+	repeated containerd.v1.types.Mount mounts = 2;
+}
+
+message ApplyResponse {
+	// Applied is the descriptor for the object which was applied.
+	// If the input was a compressed blob then the result will be
+	// the descriptor for the uncompressed blob.
+	containerd.v1.types.Descriptor applied = 1;
+}
+
+message DiffRequest {
+	// Left are the mounts which represent the older copy
+	// in which is the base of the computed changes.
+	repeated containerd.v1.types.Mount left = 1;
+
+	// Right are the mounts which represents the newer copy
+	// in which changes from the left were made into.
+	repeated containerd.v1.types.Mount right = 2;
+
+	// MediaType is the media type descriptor for the created diff
+	// object
+	string media_type = 3;
+
+	// Ref identifies the pre-commit content store object. This
+	// reference can be used to get the status from the content store.
+	string ref = 5;
+}
+
+message DiffResponse {
+	// Diff is the descriptor of the diff which can be applied
+	containerd.v1.types.Descriptor diff = 3;
+}

api/services/rootfs/rootfs.proto

@@ -1,35 +0,0 @@
-syntax = "proto3";
-
-package containerd.v1;
-
-import "gogoproto/gogo.proto";
-import "github.com/containerd/containerd/api/types/mount/mount.proto";
-import "github.com/containerd/containerd/api/types/descriptor/descriptor.proto";
-
-service RootFS {
-	rpc Unpack(UnpackRequest) returns (UnpackResponse);
-	rpc Prepare(PrepareRequest) returns (MountResponse);
-	rpc Mounts(MountsRequest) returns (MountResponse);
-}
-
-message UnpackRequest {
-	repeated containerd.v1.types.Descriptor layers = 1;
-}
-
-message UnpackResponse {
-	string chainid = 1 [(gogoproto.customtype) = "github.com/opencontainers/go-digest.Digest", (gogoproto.nullable) = false, (gogoproto.customname) = "ChainID"];
-}
-
-message PrepareRequest {
-	string name = 1;
-	string chain_id = 2 [(gogoproto.customtype) = "github.com/opencontainers/go-digest.Digest", (gogoproto.nullable) = false, (gogoproto.customname) = "ChainID"];
-	bool readonly = 3;
-}
-
-message MountsRequest {
-	string name = 1;
-}
-
-message MountResponse {
-	repeated containerd.v1.types.Mount mounts = 1;
-}

api/services/snapshot/snapshots.proto

@@ -0,0 +1,81 @@
+syntax = "proto3";
+
+package containerd.v1.snapshot;
+
+import "gogoproto/gogo.proto";
+import "google/protobuf/empty.proto";
+import "github.com/containerd/containerd/api/types/mount/mount.proto";
+
+// Snapshot service manages snapshots
+service Snapshot {
+	rpc Prepare(PrepareRequest) returns (MountsResponse);
+	rpc View(PrepareRequest) returns (MountsResponse);
+	rpc Mounts(MountsRequest) returns (MountsResponse);
+	rpc Commit(CommitRequest) returns (google.protobuf.Empty);
+	rpc Remove(RemoveRequest) returns (google.protobuf.Empty);
+	rpc Stat(StatRequest) returns (StatResponse);
+	rpc List(ListRequest) returns (stream ListResponse);
+	rpc Usage(UsageRequest) returns (UsageResponse);
+	// "Snapshot" prepares a new set of mounts from existing name
+}
+
+message PrepareRequest {
+	string key = 1;
+	string parent = 2;
+}
+
+message MountsRequest {
+	string key = 1;
+}
+
+message MountsResponse {
+	repeated containerd.v1.types.Mount mounts = 1;
+}
+
+message RemoveRequest {
+	string key = 1;
+}
+
+message CommitRequest {
+	string name = 1;
+	string key = 2;
+}
+
+message StatRequest {
+	string key = 1;
+}
+
+enum Kind {
+	option (gogoproto.goproto_enum_prefix) = false;
+	option (gogoproto.enum_customname) = "Kind";
+
+	ACTIVE = 0 [(gogoproto.enumvalue_customname) = "KindActive"];
+
+	COMMITTED = 1 [(gogoproto.enumvalue_customname) = "KindCommitted"];
+}
+
+message Info {
+	string name = 1;
+	string parent = 2;
+	Kind kind = 3;
+	bool readonly = 4;
+}
+
+message StatResponse {
+	Info info = 1 [(gogoproto.nullable) = false];
+}
+
+message ListRequest{}
+
+message ListResponse {
+	repeated Info info = 1 [(gogoproto.nullable) = false];
+}
+
+message UsageRequest {
+	string key = 1;
+}
+
+message UsageResponse {
+	int64 inodes = 2;
+	int64 size = 1;
+}

cmd/containerd/builtins.go

@@ -3,10 +3,11 @@ package main
 // register containerd builtins here
 import (
 	_ "github.com/containerd/containerd/services/content"
+	_ "github.com/containerd/containerd/services/diff"
 	_ "github.com/containerd/containerd/services/execution"
 	_ "github.com/containerd/containerd/services/healthcheck"
 	_ "github.com/containerd/containerd/services/images"
 	_ "github.com/containerd/containerd/services/metrics"
-	_ "github.com/containerd/containerd/services/rootfs"
+	_ "github.com/containerd/containerd/services/snapshot"
 	_ "github.com/containerd/containerd/services/version"
 )

cmd/containerd/main.go

@@ -18,9 +18,10 @@ import (
 
 	"github.com/Sirupsen/logrus"
 	contentapi "github.com/containerd/containerd/api/services/content"
+	diffapi "github.com/containerd/containerd/api/services/diff"
 	api "github.com/containerd/containerd/api/services/execution"
 	imagesapi "github.com/containerd/containerd/api/services/images"
-	rootfsapi "github.com/containerd/containerd/api/services/rootfs"
+	snapshotapi "github.com/containerd/containerd/api/services/snapshot"
 	versionapi "github.com/containerd/containerd/api/services/version"
 	"github.com/containerd/containerd/content"
 	"github.com/containerd/containerd/images"
@@ -417,14 +418,16 @@ func interceptor(ctx gocontext.Context,
 		ctx = log.WithModule(ctx, "execution")
 	case contentapi.ContentServer:
 		ctx = log.WithModule(ctx, "content")
-	case rootfsapi.RootFSServer:
-		ctx = log.WithModule(ctx, "rootfs")
 	case imagesapi.ImagesServer:
 		ctx = log.WithModule(ctx, "images")
 	case grpc_health_v1.HealthServer:
 		// No need to change the context
 	case versionapi.VersionServer:
 		ctx = log.WithModule(ctx, "version")
+	case snapshotapi.SnapshotServer:
+		ctx = log.WithModule(ctx, "snapshot")
+	case diffapi.DiffServer:
+		ctx = log.WithModule(ctx, "diff")
 	default:
 		log.G(ctx).Warnf("unknown GRPC server type: %#v\n", info.Server)
 	}

cmd/ctr/run.go

@@ -7,14 +7,13 @@ import (
 	"os"
 	"runtime"
 
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
-
 	"github.com/Sirupsen/logrus"
 	"github.com/containerd/console"
+	"github.com/containerd/containerd"
 	"github.com/containerd/containerd/api/services/execution"
-	rootfsapi "github.com/containerd/containerd/api/services/rootfs"
+	mounttypes "github.com/containerd/containerd/api/types/mount"
 	"github.com/containerd/containerd/images"
+	"github.com/containerd/containerd/snapshot"
 	"github.com/opencontainers/image-spec/identity"
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/pkg/errors"
@@ -59,7 +58,7 @@ var runCommand = cli.Command{
 	Action: func(context *cli.Context) error {
 		var (
 			err         error
-			resp        *rootfsapi.MountResponse
+			mounts      []containerd.Mount
 			imageConfig ocispec.Image
 
 			ctx = gocontext.Background()
@@ -87,7 +86,7 @@ var runCommand = cli.Command{
 			return err
 		}
 
-		rootfsClient, err := getRootFSService(context)
+		snapshotter, err := getSnapshotter(context)
 		if err != nil {
 			return err
 		}
@@ -111,20 +110,15 @@ var runCommand = cli.Command{
 				return err
 			}
 
-			if _, err := rootfsClient.Prepare(gocontext.TODO(), &rootfsapi.PrepareRequest{
-				Name:    id,
-				ChainID: identity.ChainID(diffIDs),
-			}); err != nil {
-				if grpc.Code(err) != codes.AlreadyExists {
+			mounts, err = snapshotter.Prepare(ctx, id, identity.ChainID(diffIDs).String())
+			if err != nil {
+				if !snapshot.IsExist(err) {
+					return err
+				}
+				mounts, err = snapshotter.Mounts(ctx, id)
+				if err != nil {
 					return err
 				}
-			}
-
-			resp, err = rootfsClient.Mounts(gocontext.TODO(), &rootfsapi.MountsRequest{
-				Name: id,
-			})
-			if err != nil {
-				return err
 			}
 
 			ic, err := image.Config(ctx, content)
@@ -153,8 +147,13 @@ var runCommand = cli.Command{
 		if err != nil {
 			return err
 		}
-		if resp != nil {
-			create.Rootfs = resp.Mounts
+		for _, m := range mounts {
+			create.Rootfs = append(create.Rootfs, &mounttypes.Mount{
+				Type:    m.Type,
+				Source:  m.Source,
+				Options: m.Options,
+			})
+
 		}
 		var con console.Console
 		if create.Terminal {

cmd/ctr/utils.go

@@ -15,13 +15,15 @@ import (
 	contentapi "github.com/containerd/containerd/api/services/content"
 	"github.com/containerd/containerd/api/services/execution"
 	imagesapi "github.com/containerd/containerd/api/services/images"
-	rootfsapi "github.com/containerd/containerd/api/services/rootfs"
+	snapshotapi "github.com/containerd/containerd/api/services/snapshot"
 	versionservice "github.com/containerd/containerd/api/services/version"
 	"github.com/containerd/containerd/api/types/container"
 	"github.com/containerd/containerd/content"
 	"github.com/containerd/containerd/images"
 	contentservice "github.com/containerd/containerd/services/content"
 	imagesservice "github.com/containerd/containerd/services/images"
+	snapshotservice "github.com/containerd/containerd/services/snapshot"
+	"github.com/containerd/containerd/snapshot"
 	"github.com/urfave/cli"
 	"google.golang.org/grpc"
 )
@@ -44,12 +46,12 @@ func getContentStore(context *cli.Context) (content.Store, error) {
 	return contentservice.NewStoreFromClient(contentapi.NewContentClient(conn)), nil
 }
 
-func getRootFSService(context *cli.Context) (rootfsapi.RootFSClient, error) {
+func getSnapshotter(context *cli.Context) (snapshot.Snapshotter, error) {
 	conn, err := getGRPCConnection(context)
 	if err != nil {
 		return nil, err
 	}
-	return rootfsapi.NewRootFSClient(conn), nil
+	return snapshotservice.NewSnapshotterFromClient(snapshotapi.NewSnapshotClient(conn)), nil
 }
 
 func getImageStore(clicontext *cli.Context) (images.Store, error) {

cmd/dist/common.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"context"
 	"crypto/tls"
+	"encoding/json"
 	"fmt"
 	"net"
 	"net/http"
@@ -18,8 +19,10 @@ import (
 	"github.com/containerd/containerd/images"
 	"github.com/containerd/containerd/remotes"
 	"github.com/containerd/containerd/remotes/docker"
+	"github.com/containerd/containerd/rootfs"
 	contentservice "github.com/containerd/containerd/services/content"
 	imagesservice "github.com/containerd/containerd/services/images"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/pkg/errors"
 	"github.com/urfave/cli"
 	"google.golang.org/grpc"
@@ -151,3 +154,36 @@ func passwordPrompt() (string, error) {
 	}
 	return string(line), nil
 }
+
+func getImageLayers(ctx context.Context, image images.Image, cs content.Store) ([]rootfs.Layer, error) {
+	p, err := content.ReadBlob(ctx, cs, image.Target.Digest)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to read manifest blob")
+	}
+
+	var manifest ocispec.Manifest
+	if err := json.Unmarshal(p, &manifest); err != nil {
+		return nil, errors.Wrap(err, "failed to unmarshal manifest")
+	}
+
+	diffIDs, err := image.RootFS(ctx, cs)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to resolve rootfs")
+	}
+
+	if len(diffIDs) != len(manifest.Layers) {
+		return nil, errors.Errorf("mismatched image rootfs and manifest layers")
+	}
+
+	layers := make([]rootfs.Layer, len(diffIDs))
+	for i := range diffIDs {
+		layers[i].Diff = ocispec.Descriptor{
+			// TODO: derive media type from compressed type
+			MediaType: ocispec.MediaTypeImageLayer,
+			Digest:    diffIDs[i],
+		}
+		layers[i].Blob = manifest.Layers[i]
+	}
+
+	return layers, nil
+}

cmd/dist/pull.go

@@ -2,19 +2,19 @@ package main
 
 import (
 	"context"
-	"encoding/json"
 	"os"
 	"text/tabwriter"
 	"time"
 
-	rootfsapi "github.com/containerd/containerd/api/services/rootfs"
-	"github.com/containerd/containerd/content"
+	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"
-	rootfsservice "github.com/containerd/containerd/services/rootfs"
-	"github.com/opencontainers/image-spec/identity"
+	"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"
@@ -101,43 +101,31 @@ command. As part of this process, we do the following:
 			// TODO(stevvooe): This section unpacks the layers and resolves the
 			// root filesystem chainid for the image. For now, we just print
 			// it, but we should keep track of this in the metadata storage.
-
 			image, err := imageStore.Get(ctx, resolvedImageName)
 			if err != nil {
-				log.G(ctx).Fatal(err)
+				log.G(ctx).WithError(err).Fatal("Failed to get image")
 			}
 
-			p, err := content.ReadBlob(ctx, cs, image.Target.Digest)
+			layers, err := getImageLayers(ctx, image, cs)
 			if err != nil {
-				log.G(ctx).Fatal(err)
-			}
-
-			var manifest ocispec.Manifest
-			if err := json.Unmarshal(p, &manifest); err != nil {
-				log.G(ctx).Fatal(err)
+				log.G(ctx).WithError(err).Fatal("Failed to get rootfs layers")
 			}
 
 			conn, err := connectGRPC(clicontext)
 			if err != nil {
 				log.G(ctx).Fatal(err)
 			}
-			rootfs := rootfsservice.NewUnpackerFromClient(rootfsapi.NewRootFSClient(conn))
+			snapshotter := snapshotservice.NewSnapshotterFromClient(snapshotapi.NewSnapshotClient(conn))
+			applier := diffservice.NewApplierFromClient(diffapi.NewDiffClient(conn))
 
 			log.G(ctx).Info("unpacking rootfs")
-			chainID, err := rootfs.Unpack(ctx, manifest.Layers)
+
+			chainID, err := rootfs.ApplyLayers(ctx, layers, snapshotter, applier)
 			if err != nil {
 				log.G(ctx).Fatal(err)
 			}
 
-			diffIDs, err := image.RootFS(ctx, cs)
-			if err != nil {
-				log.G(ctx).WithError(err).Fatal("failed resolving rootfs")
-			}
-
-			expectedChainID := identity.ChainID(diffIDs)
-			if expectedChainID != chainID {
-				log.G(ctx).Fatal("rootfs service did not match chainid")
-			}
+			log.G(ctx).Infof("Unpacked chain id: %s", chainID)
 		}()
 
 		var (

cmd/dist/rootfs.go

@@ -1,17 +1,17 @@
 package main
 
 import (
-	"context"
-	"encoding/json"
 	"fmt"
-	"io/ioutil"
 	"os"
 	"strings"
 
-	rootfsapi "github.com/containerd/containerd/api/services/rootfs"
-	"github.com/containerd/containerd/content"
+	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"
-	rootfsservice "github.com/containerd/containerd/services/rootfs"
+	"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"
@@ -42,23 +42,32 @@ var rootfsUnpackCommand = cli.Command{
 
 		log.G(ctx).Infof("unpacking layers from manifest %s", dgst.String())
 
-		conn, err := connectGRPC(clicontext)
-		if err != nil {
-			return err
-		}
-
 		cs, err := resolveContentStore(clicontext)
 		if err != nil {
 			return err
 		}
 
-		m, err := resolveManifest(ctx, cs, dgst)
-		if err != nil {
-			return err
+		image := images.Image{
+			Target: ocispec.Descriptor{
+				MediaType: ocispec.MediaTypeImageManifest,
+				Digest:    dgst,
+			},
 		}
 
-		unpacker := rootfsservice.NewUnpackerFromClient(rootfsapi.NewRootFSClient(conn))
-		chainID, err := unpacker.Unpack(ctx, m.Layers)
+		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.NewApplierFromClient(diffapi.NewDiffClient(conn))
+
+		chainID, err := rootfs.ApplyLayers(ctx, layers, snapshotter, applier)
 		if err != nil {
 			return err
 		}
@@ -95,48 +104,17 @@ var rootfsPrepareCommand = cli.Command{
 			return err
 		}
 
-		rclient := rootfsapi.NewRootFSClient(conn)
+		snapshotter := snapshotservice.NewSnapshotterFromClient(snapshotapi.NewSnapshotClient(conn))
 
-		ir := &rootfsapi.PrepareRequest{
-			Name:    target,
-			ChainID: dgst,
-		}
-
-		resp, err := rclient.Prepare(ctx, ir)
+		mounts, err := snapshotter.Prepare(ctx, target, dgst.String())
 		if err != nil {
 			return err
 		}
 
-		for _, m := range resp.Mounts {
+		for _, m := range mounts {
 			fmt.Fprintf(os.Stdout, "mount -t %s %s %s -o %s\n", m.Type, m.Source, target, strings.Join(m.Options, ","))
 		}
 
 		return nil
 	},
 }
-
-func resolveManifest(ctx context.Context, provider content.Provider, dgst digest.Digest) (ocispec.Manifest, error) {
-	p, err := readAll(ctx, provider, dgst)
-	if err != nil {
-		return ocispec.Manifest{}, err
-	}
-
-	// TODO(stevvooe): This assumption that we get a manifest is unfortunate.
-	// Need to provide way to resolve what the type of the target is.
-	var manifest ocispec.Manifest
-	if err := json.Unmarshal(p, &manifest); err != nil {
-		return ocispec.Manifest{}, err
-	}
-
-	return manifest, nil
-}
-
-func readAll(ctx context.Context, provider content.Provider, dgst digest.Digest) ([]byte, error) {
-	rc, err := provider.Reader(ctx, dgst)
-	if err != nil {
-		return nil, err
-	}
-	defer rc.Close()
-
-	return ioutil.ReadAll(rc)
-}

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 {

services/diff/client.go

@@ -0,0 +1,56 @@
+package diff
+
+import (
+	"context"
+
+	"github.com/containerd/containerd"
+	diffapi "github.com/containerd/containerd/api/services/diff"
+	"github.com/containerd/containerd/api/types/descriptor"
+	mounttypes "github.com/containerd/containerd/api/types/mount"
+	"github.com/containerd/containerd/rootfs"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+)
+
+// NewApplierFromClient returns a new Applier which communicates
+// over a GRPC connection.
+func NewApplierFromClient(client diffapi.DiffClient) rootfs.Applier {
+	return &remoteApplier{
+		client: client,
+	}
+}
+
+type remoteApplier struct {
+	client diffapi.DiffClient
+}
+
+func (r *remoteApplier) Apply(ctx context.Context, diff ocispec.Descriptor, mounts []containerd.Mount) (ocispec.Descriptor, error) {
+	req := &diffapi.ApplyRequest{
+		Diff:   fromDescriptor(diff),
+		Mounts: fromMounts(mounts),
+	}
+	resp, err := r.client.Apply(ctx, req)
+	if err != nil {
+		return ocispec.Descriptor{}, err
+	}
+	return toDescriptor(resp.Applied), nil
+}
+
+func fromDescriptor(d ocispec.Descriptor) *descriptor.Descriptor {
+	return &descriptor.Descriptor{
+		MediaType: d.MediaType,
+		Digest:    d.Digest,
+		Size_:     d.Size,
+	}
+}
+
+func fromMounts(mounts []containerd.Mount) []*mounttypes.Mount {
+	apiMounts := make([]*mounttypes.Mount, len(mounts))
+	for i, m := range mounts {
+		apiMounts[i] = &mounttypes.Mount{
+			Type:    m.Type,
+			Source:  m.Source,
+			Options: m.Options,
+		}
+	}
+	return apiMounts
+}

services/diff/service.go

@@ -0,0 +1,138 @@
+package diff
+
+import (
+	"io"
+	"io/ioutil"
+	"os"
+
+	"github.com/containerd/containerd"
+	diffapi "github.com/containerd/containerd/api/services/diff"
+	"github.com/containerd/containerd/api/types/descriptor"
+	mounttypes "github.com/containerd/containerd/api/types/mount"
+	"github.com/containerd/containerd/archive"
+	"github.com/containerd/containerd/archive/compression"
+	"github.com/containerd/containerd/content"
+	"github.com/containerd/containerd/plugin"
+	"github.com/containerd/containerd/snapshot"
+	digest "github.com/opencontainers/go-digest"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+	"github.com/pkg/errors"
+	"golang.org/x/net/context"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	plugin.Register("diff-grpc", &plugin.Registration{
+		Type: plugin.GRPCPlugin,
+		Init: func(ic *plugin.InitContext) (interface{}, error) {
+			return newService(ic.Content, ic.Snapshotter)
+		},
+	})
+}
+
+type service struct {
+	store       content.Store
+	snapshotter snapshot.Snapshotter
+}
+
+func newService(store content.Store, snapshotter snapshot.Snapshotter) (*service, error) {
+	return &service{
+		store:       store,
+		snapshotter: snapshotter,
+	}, nil
+}
+
+func (s *service) Register(gs *grpc.Server) error {
+	diffapi.RegisterDiffServer(gs, s)
+	return nil
+}
+
+func (s *service) Apply(ctx context.Context, er *diffapi.ApplyRequest) (*diffapi.ApplyResponse, error) {
+	desc := toDescriptor(er.Diff)
+	// TODO: Check for supported media types
+
+	mounts := toMounts(er.Mounts)
+
+	dir, err := ioutil.TempDir("", "extract-")
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create temporary directory")
+	}
+	defer os.RemoveAll(dir)
+
+	if err := containerd.MountAll(mounts, dir); err != nil {
+		return nil, errors.Wrap(err, "failed to mount")
+	}
+	defer containerd.Unmount(dir, 0)
+
+	r, err := s.store.Reader(ctx, desc.Digest)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get reader from content store")
+	}
+	defer r.Close()
+
+	// TODO: only decompress stream if media type is compressed
+	ds, err := compression.DecompressStream(r)
+	if err != nil {
+		return nil, err
+	}
+	defer ds.Close()
+
+	digester := digest.Canonical.Digester()
+	rc := &readCounter{
+		r: io.TeeReader(ds, digester.Hash()),
+	}
+
+	if _, err := archive.Apply(ctx, dir, rc); err != nil {
+		return nil, err
+	}
+
+	// Read any trailing data
+	if _, err := io.Copy(ioutil.Discard, rc); err != nil {
+		return nil, err
+	}
+
+	resp := &diffapi.ApplyResponse{
+		Applied: &descriptor.Descriptor{
+			MediaType: ocispec.MediaTypeImageLayer,
+			Digest:    digester.Digest(),
+			Size_:     rc.c,
+		},
+	}
+
+	return resp, nil
+}
+
+func (s *service) Diff(context.Context, *diffapi.DiffRequest) (*diffapi.DiffResponse, error) {
+	return nil, errors.New("not implemented")
+}
+
+type readCounter struct {
+	r io.Reader
+	c int64
+}
+
+func (rc *readCounter) Read(p []byte) (n int, err error) {
+	n, err = rc.r.Read(p)
+	rc.c += int64(n)
+	return
+}
+
+func toDescriptor(d *descriptor.Descriptor) ocispec.Descriptor {
+	return ocispec.Descriptor{
+		MediaType: d.MediaType,
+		Digest:    d.Digest,
+		Size:      d.Size_,
+	}
+}
+
+func toMounts(apim []*mounttypes.Mount) []containerd.Mount {
+	mounts := make([]containerd.Mount, len(apim))
+	for i, m := range apim {
+		mounts[i] = containerd.Mount{
+			Type:    m.Type,
+			Source:  m.Source,
+			Options: m.Options,
+		}
+	}
+	return mounts
+}

services/rootfs/preparer.go

@@ -1,39 +0,0 @@
-package rootfs
-
-import (
-	"context"
-
-	rootfsapi "github.com/containerd/containerd/api/services/rootfs"
-	containerd_v1_types "github.com/containerd/containerd/api/types/descriptor"
-	"github.com/containerd/containerd/rootfs"
-	digest "github.com/opencontainers/go-digest"
-	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
-)
-
-func NewUnpackerFromClient(client rootfsapi.RootFSClient) rootfs.Unpacker {
-	return remoteUnpacker{
-		client: client,
-	}
-}
-
-type remoteUnpacker struct {
-	client rootfsapi.RootFSClient
-}
-
-func (rp remoteUnpacker) Unpack(ctx context.Context, layers []ocispec.Descriptor) (digest.Digest, error) {
-	pr := rootfsapi.UnpackRequest{
-		Layers: make([]*containerd_v1_types.Descriptor, len(layers)),
-	}
-	for i, l := range layers {
-		pr.Layers[i] = &containerd_v1_types.Descriptor{
-			MediaType: l.MediaType,
-			Digest:    l.Digest,
-			Size_:     l.Size,
-		}
-	}
-	resp, err := rp.client.Unpack(ctx, &pr)
-	if err != nil {
-		return "", err
-	}
-	return resp.ChainID, nil
-}

services/rootfs/service.go

@@ -1,114 +0,0 @@
-package rootfs
-
-import (
-	"github.com/containerd/containerd"
-	rootfsapi "github.com/containerd/containerd/api/services/rootfs"
-	containerd_v1_types "github.com/containerd/containerd/api/types/mount"
-	"github.com/containerd/containerd/content"
-	"github.com/containerd/containerd/log"
-	"github.com/containerd/containerd/plugin"
-	"github.com/containerd/containerd/rootfs"
-	"github.com/containerd/containerd/snapshot"
-	digest "github.com/opencontainers/go-digest"
-	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
-	"golang.org/x/net/context"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
-)
-
-func init() {
-	plugin.Register("rootfs-grpc", &plugin.Registration{
-		Type: plugin.GRPCPlugin,
-		Init: func(ic *plugin.InitContext) (interface{}, error) {
-			return NewService(ic.Content, ic.Snapshotter)
-		},
-	})
-}
-
-type Service struct {
-	store       content.Store
-	snapshotter snapshot.Snapshotter
-}
-
-func NewService(store content.Store, snapshotter snapshot.Snapshotter) (*Service, error) {
-	return &Service{
-		store:       store,
-		snapshotter: snapshotter,
-	}, nil
-}
-
-func (s *Service) Register(gs *grpc.Server) error {
-	rootfsapi.RegisterRootFSServer(gs, s)
-	return nil
-}
-
-func (s *Service) Unpack(ctx context.Context, pr *rootfsapi.UnpackRequest) (*rootfsapi.UnpackResponse, error) {
-	layers := make([]ocispec.Descriptor, len(pr.Layers))
-	for i, l := range pr.Layers {
-		layers[i] = ocispec.Descriptor{
-			MediaType: l.MediaType,
-			Digest:    l.Digest,
-			Size:      l.Size_,
-		}
-	}
-	log.G(ctx).Infof("Preparing %#v", layers)
-	chainID, err := rootfs.Prepare(ctx, s.snapshotter, mounter{}, layers, s.store.Reader, emptyResolver, noopRegister)
-	if err != nil {
-		log.G(ctx).Errorf("Rootfs Prepare failed!: %v", err)
-		return nil, err
-	}
-	log.G(ctx).Infof("ChainID %#v", chainID)
-	return &rootfsapi.UnpackResponse{
-		ChainID: chainID,
-	}, nil
-}
-
-func (s *Service) Prepare(ctx context.Context, ir *rootfsapi.PrepareRequest) (*rootfsapi.MountResponse, error) {
-	mounts, err := rootfs.InitRootFS(ctx, ir.Name, ir.ChainID, ir.Readonly, s.snapshotter, mounter{})
-	if err != nil {
-		return nil, grpc.Errorf(codes.AlreadyExists, "%v", err)
-	}
-	return &rootfsapi.MountResponse{
-		Mounts: apiMounts(mounts),
-	}, nil
-}
-
-func (s *Service) Mounts(ctx context.Context, mr *rootfsapi.MountsRequest) (*rootfsapi.MountResponse, error) {
-	mounts, err := s.snapshotter.Mounts(ctx, mr.Name)
-	if err != nil {
-		return nil, err
-	}
-	return &rootfsapi.MountResponse{
-		Mounts: apiMounts(mounts),
-	}, nil
-}
-
-func apiMounts(mounts []containerd.Mount) []*containerd_v1_types.Mount {
-	am := make([]*containerd_v1_types.Mount, len(mounts))
-	for i, m := range mounts {
-		am[i] = &containerd_v1_types.Mount{
-			Type:    m.Type,
-			Source:  m.Source,
-			Options: m.Options,
-		}
-	}
-	return am
-}
-
-type mounter struct{}
-
-func (mounter) Mount(dir string, mounts ...containerd.Mount) error {
-	return containerd.MountAll(mounts, dir)
-}
-
-func (mounter) Unmount(dir string) error {
-	return containerd.Unmount(dir, 0)
-}
-
-func emptyResolver(digest.Digest) digest.Digest {
-	return digest.Digest("")
-}
-
-func noopRegister(digest.Digest, digest.Digest) error {
-	return nil
-}

services/snapshot/client.go

@@ -0,0 +1,158 @@
+package snapshot
+
+import (
+	"context"
+	"io"
+	"strings"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+
+	"github.com/containerd/containerd"
+	snapshotapi "github.com/containerd/containerd/api/services/snapshot"
+	"github.com/containerd/containerd/snapshot"
+	"github.com/pkg/errors"
+)
+
+// NewSnapshotterFromClient returns a new Snapshotter which communicates
+// over a GRPC connection.
+func NewSnapshotterFromClient(client snapshotapi.SnapshotClient) snapshot.Snapshotter {
+	return &remoteSnapshotter{
+		client: client,
+	}
+}
+
+type remoteSnapshotter struct {
+	client snapshotapi.SnapshotClient
+}
+
+func (r *remoteSnapshotter) Stat(ctx context.Context, key string) (snapshot.Info, error) {
+	resp, err := r.client.Stat(ctx, &snapshotapi.StatRequest{Key: key})
+	if err != nil {
+		return snapshot.Info{}, rewriteGRPCError(err)
+	}
+	return toInfo(resp.Info), nil
+}
+
+func (r *remoteSnapshotter) Usage(ctx context.Context, key string) (snapshot.Usage, error) {
+	resp, err := r.client.Usage(ctx, &snapshotapi.UsageRequest{Key: key})
+	if err != nil {
+		return snapshot.Usage{}, rewriteGRPCError(err)
+	}
+	return toUsage(resp), nil
+}
+
+func (r *remoteSnapshotter) Mounts(ctx context.Context, key string) ([]containerd.Mount, error) {
+	resp, err := r.client.Mounts(ctx, &snapshotapi.MountsRequest{Key: key})
+	if err != nil {
+		return nil, rewriteGRPCError(err)
+	}
+	return toMounts(resp), nil
+}
+
+func (r *remoteSnapshotter) Prepare(ctx context.Context, key, parent string) ([]containerd.Mount, error) {
+	resp, err := r.client.Prepare(ctx, &snapshotapi.PrepareRequest{Key: key, Parent: parent})
+	if err != nil {
+		return nil, rewriteGRPCError(err)
+	}
+	return toMounts(resp), nil
+}
+
+func (r *remoteSnapshotter) View(ctx context.Context, key, parent string) ([]containerd.Mount, error) {
+	resp, err := r.client.View(ctx, &snapshotapi.PrepareRequest{Key: key, Parent: parent})
+	if err != nil {
+		return nil, rewriteGRPCError(err)
+	}
+	return toMounts(resp), nil
+}
+
+func (r *remoteSnapshotter) Commit(ctx context.Context, name, key string) error {
+	_, err := r.client.Commit(ctx, &snapshotapi.CommitRequest{
+		Name: name,
+		Key:  key,
+	})
+	return rewriteGRPCError(err)
+}
+
+func (r *remoteSnapshotter) Remove(ctx context.Context, key string) error {
+	_, err := r.client.Remove(ctx, &snapshotapi.RemoveRequest{Key: key})
+	return rewriteGRPCError(err)
+}
+
+func (r *remoteSnapshotter) Walk(ctx context.Context, fn func(context.Context, snapshot.Info) error) error {
+	sc, err := r.client.List(ctx, &snapshotapi.ListRequest{})
+	if err != nil {
+		rewriteGRPCError(err)
+	}
+	for {
+		resp, err := sc.Recv()
+		if err != nil {
+			if err == io.EOF {
+				return nil
+			}
+			return rewriteGRPCError(err)
+		}
+		if resp == nil {
+			return nil
+		}
+		for _, info := range resp.Info {
+			if err := fn(ctx, toInfo(info)); err != nil {
+				return err
+			}
+		}
+	}
+}
+
+func rewriteGRPCError(err error) error {
+	switch grpc.Code(errors.Cause(err)) {
+	case codes.AlreadyExists:
+		return snapshot.ErrSnapshotExist
+	case codes.NotFound:
+		return snapshot.ErrSnapshotNotExist
+	case codes.FailedPrecondition:
+		desc := grpc.ErrorDesc(errors.Cause(err))
+		if strings.Contains(desc, snapshot.ErrSnapshotNotActive.Error()) {
+			return snapshot.ErrSnapshotNotActive
+		}
+		if strings.Contains(desc, snapshot.ErrSnapshotNotCommitted.Error()) {
+			return snapshot.ErrSnapshotNotCommitted
+		}
+	}
+
+	return err
+}
+
+func toKind(kind snapshotapi.Kind) snapshot.Kind {
+	if kind == snapshotapi.KindActive {
+		return snapshot.KindActive
+	}
+	return snapshot.KindCommitted
+}
+
+func toInfo(info snapshotapi.Info) snapshot.Info {
+	return snapshot.Info{
+		Name:     info.Name,
+		Parent:   info.Parent,
+		Kind:     toKind(info.Kind),
+		Readonly: info.Readonly,
+	}
+}
+
+func toUsage(resp *snapshotapi.UsageResponse) snapshot.Usage {
+	return snapshot.Usage{
+		Inodes: resp.Inodes,
+		Size:   resp.Size_,
+	}
+}
+
+func toMounts(resp *snapshotapi.MountsResponse) []containerd.Mount {
+	mounts := make([]containerd.Mount, len(resp.Mounts))
+	for i, m := range resp.Mounts {
+		mounts[i] = containerd.Mount{
+			Type:    m.Type,
+			Source:  m.Source,
+			Options: m.Options,
+		}
+	}
+	return mounts
+}

services/snapshot/service.go

@@ -0,0 +1,204 @@
+package snapshot
+
+import (
+	gocontext "context"
+
+	"github.com/containerd/containerd"
+	snapshotapi "github.com/containerd/containerd/api/services/snapshot"
+	mounttypes "github.com/containerd/containerd/api/types/mount"
+	"github.com/containerd/containerd/log"
+	"github.com/containerd/containerd/plugin"
+	"github.com/containerd/containerd/snapshot"
+	protoempty "github.com/golang/protobuf/ptypes/empty"
+	"golang.org/x/net/context"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+)
+
+func init() {
+	plugin.Register("snapshots-grpc", &plugin.Registration{
+		Type: plugin.GRPCPlugin,
+		Init: func(ic *plugin.InitContext) (interface{}, error) {
+			return newService(ic.Snapshotter)
+		},
+	})
+}
+
+var empty = &protoempty.Empty{}
+
+type service struct {
+	snapshotter snapshot.Snapshotter
+}
+
+func newService(snapshotter snapshot.Snapshotter) (*service, error) {
+	return &service{
+		snapshotter: snapshotter,
+	}, nil
+}
+
+func (s *service) Register(gs *grpc.Server) error {
+	snapshotapi.RegisterSnapshotServer(gs, s)
+	return nil
+}
+
+func (s *service) Prepare(ctx context.Context, pr *snapshotapi.PrepareRequest) (*snapshotapi.MountsResponse, error) {
+	log.G(ctx).WithField("parent", pr.Parent).WithField("key", pr.Key).Debugf("Preparing snapshot")
+	// TODO: Apply namespace
+	// TODO: Lookup snapshot id from metadata store
+	mounts, err := s.snapshotter.Prepare(ctx, pr.Key, pr.Parent)
+	if err != nil {
+		return nil, grpcError(err)
+	}
+	return fromMounts(mounts), nil
+}
+
+func (s *service) View(ctx context.Context, pr *snapshotapi.PrepareRequest) (*snapshotapi.MountsResponse, error) {
+	log.G(ctx).WithField("parent", pr.Parent).WithField("key", pr.Key).Debugf("Preparing view snapshot")
+	// TODO: Apply namespace
+	// TODO: Lookup snapshot id from metadata store
+	mounts, err := s.snapshotter.View(ctx, pr.Key, pr.Parent)
+	if err != nil {
+		return nil, grpcError(err)
+	}
+	return fromMounts(mounts), nil
+}
+
+func (s *service) Mounts(ctx context.Context, mr *snapshotapi.MountsRequest) (*snapshotapi.MountsResponse, error) {
+	log.G(ctx).WithField("key", mr.Key).Debugf("Getting snapshot mounts")
+	// TODO: Apply namespace
+	// TODO: Lookup snapshot id from metadata store
+	mounts, err := s.snapshotter.Mounts(ctx, mr.Key)
+	if err != nil {
+		return nil, grpcError(err)
+	}
+	return fromMounts(mounts), nil
+}
+
+func (s *service) Commit(ctx context.Context, cr *snapshotapi.CommitRequest) (*protoempty.Empty, error) {
+	log.G(ctx).WithField("key", cr.Key).WithField("name", cr.Name).Debugf("Committing snapshot")
+	// TODO: Apply namespace
+	// TODO: Lookup snapshot id from metadata store
+	if err := s.snapshotter.Commit(ctx, cr.Name, cr.Key); err != nil {
+		return nil, grpcError(err)
+	}
+	return empty, nil
+}
+
+func (s *service) Remove(ctx context.Context, rr *snapshotapi.RemoveRequest) (*protoempty.Empty, error) {
+	log.G(ctx).WithField("key", rr.Key).Debugf("Removing snapshot")
+	// TODO: Apply namespace
+	// TODO: Lookup snapshot id from metadata store
+	if err := s.snapshotter.Remove(ctx, rr.Key); err != nil {
+		return nil, grpcError(err)
+	}
+	return empty, nil
+}
+
+func (s *service) Stat(ctx context.Context, sr *snapshotapi.StatRequest) (*snapshotapi.StatResponse, error) {
+	log.G(ctx).WithField("key", sr.Key).Debugf("Statting snapshot")
+	// TODO: Apply namespace
+	info, err := s.snapshotter.Stat(ctx, sr.Key)
+	if err != nil {
+		return nil, grpcError(err)
+	}
+
+	return &snapshotapi.StatResponse{Info: fromInfo(info)}, nil
+}
+
+func (s *service) List(sr *snapshotapi.ListRequest, ss snapshotapi.Snapshot_ListServer) error {
+	// TODO: Apply namespace
+
+	var (
+		buffer    []snapshotapi.Info
+		sendBlock = func(block []snapshotapi.Info) error {
+			return ss.Send(&snapshotapi.ListResponse{
+				Info: block,
+			})
+		}
+	)
+	err := s.snapshotter.Walk(ss.Context(), func(ctx gocontext.Context, info snapshot.Info) error {
+		buffer = append(buffer, fromInfo(info))
+
+		if len(buffer) >= 100 {
+			if err := sendBlock(buffer); err != nil {
+				return err
+			}
+
+			buffer = buffer[:0]
+		}
+
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+	if len(buffer) > 0 {
+		// Send remaining infos
+		if err := sendBlock(buffer); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (s *service) Usage(ctx context.Context, ur *snapshotapi.UsageRequest) (*snapshotapi.UsageResponse, error) {
+	// TODO: Apply namespace
+	usage, err := s.snapshotter.Usage(ctx, ur.Key)
+	if err != nil {
+		return nil, grpcError(err)
+	}
+
+	return fromUsage(usage), nil
+}
+
+func grpcError(err error) error {
+	if snapshot.IsNotExist(err) {
+		return grpc.Errorf(codes.NotFound, err.Error())
+	}
+	if snapshot.IsExist(err) {
+		return grpc.Errorf(codes.AlreadyExists, err.Error())
+	}
+	if snapshot.IsNotActive(err) || snapshot.IsNotCommitted(err) {
+		return grpc.Errorf(codes.FailedPrecondition, err.Error())
+	}
+
+	return err
+}
+
+func fromKind(kind snapshot.Kind) snapshotapi.Kind {
+	if kind == snapshot.KindActive {
+		return snapshotapi.KindActive
+	}
+	return snapshotapi.KindCommitted
+}
+
+func fromInfo(info snapshot.Info) snapshotapi.Info {
+	return snapshotapi.Info{
+		Name:     info.Name,
+		Parent:   info.Parent,
+		Kind:     fromKind(info.Kind),
+		Readonly: info.Readonly,
+	}
+}
+
+func fromUsage(usage snapshot.Usage) *snapshotapi.UsageResponse {
+	return &snapshotapi.UsageResponse{
+		Inodes: usage.Inodes,
+		Size_:  usage.Size,
+	}
+}
+
+func fromMounts(mounts []containerd.Mount) *snapshotapi.MountsResponse {
+	resp := &snapshotapi.MountsResponse{
+		Mounts: make([]*mounttypes.Mount, len(mounts)),
+	}
+	for i, m := range mounts {
+		resp.Mounts[i] = &mounttypes.Mount{
+			Type:    m.Type,
+			Source:  m.Source,
+			Options: m.Options,
+		}
+	}
+	return resp
+}

snapshot/storage/proto/record.pb.go

@@ -63,7 +63,7 @@ func (Kind) EnumDescriptor() ([]byte, []int) { return fileDescriptorRecord, []in
 type Snapshot struct {
 	ID       uint64 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"`
 	Parent   string `protobuf:"bytes,2,opt,name=parent,proto3" json:"parent,omitempty"`
-	Kind     Kind   `protobuf:"varint,4,opt,name=kind,proto3,enum=containerd.v1.Kind" json:"kind,omitempty"`
+	Kind     Kind   `protobuf:"varint,4,opt,name=kind,proto3,enum=containerd.snapshot.v1.Kind" json:"kind,omitempty"`
 	Readonly bool   `protobuf:"varint,5,opt,name=readonly,proto3" json:"readonly,omitempty"`
 	// inodes stores the number inodes in use for the snapshot.
 	//
@@ -81,8 +81,8 @@ func (*Snapshot) ProtoMessage()               {}
 func (*Snapshot) Descriptor() ([]byte, []int) { return fileDescriptorRecord, []int{0} }
 
 func init() {
-	proto1.RegisterType((*Snapshot)(nil), "containerd.v1.Snapshot")
-	proto1.RegisterEnum("containerd.v1.Kind", Kind_name, Kind_value)
+	proto1.RegisterType((*Snapshot)(nil), "containerd.snapshot.v1.Snapshot")
+	proto1.RegisterEnum("containerd.snapshot.v1.Kind", Kind_name, Kind_value)
 }
 func (m *Snapshot) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
@@ -511,26 +511,26 @@ func init() {
 }
 
 var fileDescriptorRecord = []byte{
-	// 330 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x54, 0x90, 0xbd, 0x4e, 0xf3, 0x30,
-	0x18, 0x85, 0xe3, 0x7c, 0xf9, 0x42, 0x6b, 0xd1, 0xaa, 0x18, 0x54, 0x59, 0x19, 0x8c, 0xc5, 0x42,
-	0xc4, 0x90, 0x08, 0xb8, 0x82, 0xfe, 0x0d, 0x55, 0x55, 0x21, 0x85, 0x8a, 0x3d, 0x8d, 0xad, 0xd4,
-	0x82, 0xda, 0x95, 0x63, 0x2a, 0xc1, 0xc4, 0x88, 0x7a, 0x0f, 0x9d, 0x60, 0xe3, 0x0e, 0xb8, 0x82,
-	0x8e, 0x8c, 0x4c, 0x88, 0xe6, 0x4a, 0x50, 0xd2, 0x8a, 0x9f, 0xed, 0x9c, 0xe3, 0x47, 0x8f, 0xa5,
-	0x17, 0x76, 0x53, 0x61, 0x26, 0xb7, 0xe3, 0x20, 0x51, 0xd3, 0x30, 0x51, 0xd2, 0xc4, 0x42, 0x72,
-	0xcd, 0x7e, 0xc7, 0x4c, 0xc6, 0xb3, 0x6c, 0xa2, 0x4c, 0x98, 0x19, 0xa5, 0xe3, 0x94, 0x87, 0x33,
-	0xad, 0x8c, 0x0a, 0x35, 0x4f, 0x94, 0x66, 0x41, 0x59, 0x50, 0xed, 0x87, 0x0f, 0xe6, 0xa7, 0xde,
-	0x41, 0xaa, 0x52, 0xb5, 0xc1, 0x8a, 0xb4, 0x81, 0x8e, 0x5e, 0x00, 0xac, 0x5c, 0x6e, 0x65, 0xa8,
-	0x09, 0x6d, 0xc1, 0x30, 0xa0, 0xc0, 0x77, 0xda, 0x6e, 0xfe, 0x71, 0x68, 0xf7, 0xbb, 0x91, 0x2d,
-	0x18, 0x6a, 0x42, 0x77, 0x16, 0x6b, 0x2e, 0x0d, 0xb6, 0x29, 0xf0, 0xab, 0xd1, 0xb6, 0xa1, 0x63,
-	0xe8, 0x5c, 0x0b, 0xc9, 0xb0, 0x43, 0x81, 0x5f, 0x3f, 0xdb, 0x0f, 0xfe, 0x7c, 0x18, 0x0c, 0x84,
-	0x64, 0x51, 0x09, 0x20, 0x0f, 0x56, 0x34, 0x8f, 0x99, 0x92, 0x37, 0x77, 0xf8, 0x3f, 0x05, 0x7e,
-	0x25, 0xfa, 0xee, 0x85, 0x5c, 0x48, 0xc5, 0x78, 0x86, 0x5d, 0x0a, 0xfc, 0x7f, 0xd1, 0xb6, 0x21,
-	0x04, 0x9d, 0x4c, 0xdc, 0x73, 0xbc, 0x53, 0xae, 0x65, 0x3e, 0x89, 0xa0, 0x33, 0xd8, 0xf8, 0xdc,
-	0x56, 0x67, 0xd4, 0xbf, 0xea, 0x35, 0x2c, 0xaf, 0xbe, 0x58, 0x52, 0x58, 0xac, 0xad, 0xc4, 0x88,
-	0x39, 0x47, 0x14, 0x56, 0x3b, 0x17, 0xc3, 0x61, 0x7f, 0x34, 0xea, 0x75, 0x1b, 0xc0, 0xdb, 0x5b,
-	0x2c, 0x69, 0xad, 0x78, 0xee, 0xa8, 0xe9, 0x54, 0x18, 0xc3, 0x99, 0xb7, 0xfb, 0xf8, 0x44, 0xac,
-	0xd7, 0x67, 0x52, 0xba, 0xda, 0x78, 0xb5, 0x26, 0xd6, 0xfb, 0x9a, 0x58, 0x0f, 0x39, 0x01, 0xab,
-	0x9c, 0x80, 0xb7, 0x9c, 0x80, 0xcf, 0x9c, 0x80, 0xb1, 0x5b, 0x9e, 0xe8, 0xfc, 0x2b, 0x00, 0x00,
-	0xff, 0xff, 0xce, 0x7e, 0x6f, 0x85, 0x8f, 0x01, 0x00, 0x00,
+	// 334 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x6c, 0x90, 0xbb, 0x4e, 0xf3, 0x30,
+	0x1c, 0xc5, 0xe3, 0x7c, 0xf9, 0x42, 0x6b, 0x41, 0x55, 0x2c, 0x54, 0x59, 0x11, 0x32, 0x16, 0x53,
+	0xc4, 0x90, 0x70, 0x79, 0x82, 0xde, 0x86, 0xaa, 0xaa, 0x90, 0x42, 0xc5, 0x9e, 0xc6, 0x56, 0x6a,
+	0x41, 0xed, 0xca, 0x31, 0x95, 0x60, 0x62, 0x44, 0x7d, 0x87, 0x4e, 0xf0, 0x08, 0x4c, 0x3c, 0x41,
+	0x47, 0x46, 0x26, 0x44, 0xf3, 0x24, 0x28, 0x69, 0xb9, 0x0c, 0x6c, 0xe7, 0xf7, 0xf7, 0x4f, 0xe7,
+	0x48, 0x86, 0x9d, 0x54, 0x98, 0xf1, 0xcd, 0x28, 0x48, 0xd4, 0x24, 0x4c, 0x94, 0x34, 0xb1, 0x90,
+	0x5c, 0xb3, 0xdf, 0x31, 0x93, 0xf1, 0x34, 0x1b, 0x2b, 0x13, 0x66, 0x46, 0xe9, 0x38, 0xe5, 0xe1,
+	0x54, 0x2b, 0xa3, 0x42, 0xcd, 0x13, 0xa5, 0x59, 0x50, 0x02, 0x6a, 0xfc, 0xf8, 0xc1, 0x97, 0x1f,
+	0xcc, 0x4e, 0xbc, 0xbd, 0x54, 0xa5, 0x6a, 0xed, 0x17, 0x69, 0x6d, 0x1f, 0x3e, 0x03, 0x58, 0xb9,
+	0xd8, 0x58, 0xa8, 0x01, 0x6d, 0xc1, 0x30, 0xa0, 0xc0, 0x77, 0x5a, 0x6e, 0xfe, 0x7e, 0x60, 0xf7,
+	0x3a, 0x91, 0x2d, 0x18, 0x6a, 0x40, 0x77, 0x1a, 0x6b, 0x2e, 0x0d, 0xb6, 0x29, 0xf0, 0xab, 0xd1,
+	0x86, 0xd0, 0x31, 0x74, 0xae, 0x84, 0x64, 0xd8, 0xa1, 0xc0, 0xaf, 0x9d, 0xee, 0x07, 0x7f, 0x2f,
+	0x07, 0x7d, 0x21, 0x59, 0x54, 0x9a, 0xc8, 0x83, 0x15, 0xcd, 0x63, 0xa6, 0xe4, 0xf5, 0x2d, 0xfe,
+	0x4f, 0x81, 0x5f, 0x89, 0xbe, 0xb9, 0x58, 0x11, 0x52, 0x31, 0x9e, 0x61, 0x97, 0x02, 0xff, 0x5f,
+	0xb4, 0x21, 0x84, 0xa0, 0x93, 0x89, 0x3b, 0x8e, 0xb7, 0xca, 0x6b, 0x99, 0x8f, 0x22, 0xe8, 0xf4,
+	0xd7, 0x7d, 0x6e, 0xb3, 0x3d, 0xec, 0x5d, 0x76, 0xeb, 0x96, 0x57, 0x9b, 0x2f, 0x28, 0x2c, 0xae,
+	0xcd, 0xc4, 0x88, 0x19, 0x47, 0x14, 0x56, 0xdb, 0xe7, 0x83, 0x41, 0x6f, 0x38, 0xec, 0x76, 0xea,
+	0xc0, 0xdb, 0x9d, 0x2f, 0xe8, 0x4e, 0xf1, 0xdc, 0x56, 0x93, 0x89, 0x30, 0x86, 0x33, 0x6f, 0xfb,
+	0xe1, 0x91, 0x58, 0x2f, 0x4f, 0xa4, 0xec, 0x6a, 0xe1, 0xe5, 0x8a, 0x58, 0x6f, 0x2b, 0x62, 0xdd,
+	0xe7, 0x04, 0x2c, 0x73, 0x02, 0x5e, 0x73, 0x02, 0x3e, 0x72, 0x02, 0x46, 0x6e, 0xf9, 0x57, 0x67,
+	0x9f, 0x01, 0x00, 0x00, 0xff, 0xff, 0xae, 0x6e, 0x6e, 0xcd, 0xa1, 0x01, 0x00, 0x00,
 }

snapshot/storage/proto/record.proto

@@ -1,6 +1,6 @@
 syntax = "proto3";
 
-package containerd.v1;
+package containerd.snapshot.v1;
 
 import "gogoproto/gogo.proto";