Add leases api

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

metadata/buckets.go

@@ -38,6 +38,7 @@ var (
 	bucketKeyObjectContent    = []byte("content")    // stores content references
 	bucketKeyObjectBlob       = []byte("blob")       // stores content links
 	bucketKeyObjectIngest     = []byte("ingest")     // stores ingest links
+	bucketKeyObjectLeases     = []byte("leases")     // stores leases
 
 	bucketKeyDigest      = []byte("digest")
 	bucketKeyMediaType   = []byte("mediatype")
@@ -53,6 +54,7 @@ var (
 	bucketKeySnapshotter = []byte("snapshotter")
 	bucketKeyTarget      = []byte("target")
 	bucketKeyExtensions  = []byte("extensions")
+	bucketKeyCreatedAt   = []byte("createdat")
 )
 
 func getBucket(tx *bolt.Tx, keys ...[]byte) *bolt.Bucket {

metadata/content.go

@@ -391,27 +391,31 @@ func (nw *namespacedWriter) Commit(ctx context.Context, size int64, expected dig
 				return err
 			}
 		}
-		return nw.commit(ctx, tx, size, expected, opts...)
+		dgst, err := nw.commit(ctx, tx, size, expected, opts...)
+		if err != nil {
+			return err
+		}
+		return addContentLease(ctx, tx, dgst)
 	})
 }
 
-func (nw *namespacedWriter) commit(ctx context.Context, tx *bolt.Tx, size int64, expected digest.Digest, opts ...content.Opt) error {
+func (nw *namespacedWriter) commit(ctx context.Context, tx *bolt.Tx, size int64, expected digest.Digest, opts ...content.Opt) (digest.Digest, error) {
 	var base content.Info
 	for _, opt := range opts {
 		if err := opt(&base); err != nil {
-			return err
+			return "", err
 		}
 	}
 	if err := validateInfo(&base); err != nil {
-		return err
+		return "", err
 	}
 
 	status, err := nw.Writer.Status()
 	if err != nil {
-		return err
+		return "", err
 	}
 	if size != 0 && size != status.Offset {
-		return errors.Errorf("%q failed size validation: %v != %v", nw.ref, status.Offset, size)
+		return "", errors.Errorf("%q failed size validation: %v != %v", nw.ref, status.Offset, size)
 	}
 	size = status.Offset
 
@@ -419,32 +423,32 @@ func (nw *namespacedWriter) commit(ctx context.Context, tx *bolt.Tx, size int64,
 
 	if err := nw.Writer.Commit(ctx, size, expected); err != nil {
 		if !errdefs.IsAlreadyExists(err) {
-			return err
+			return "", err
 		}
 		if getBlobBucket(tx, nw.namespace, actual) != nil {
-			return errors.Wrapf(errdefs.ErrAlreadyExists, "content %v", actual)
+			return "", errors.Wrapf(errdefs.ErrAlreadyExists, "content %v", actual)
 		}
 	}
 
 	bkt, err := createBlobBucket(tx, nw.namespace, actual)
 	if err != nil {
-		return err
+		return "", err
 	}
 
 	commitTime := time.Now().UTC()
 
 	sizeEncoded, err := encodeInt(size)
 	if err != nil {
-		return err
+		return "", err
 	}
 
 	if err := boltutil.WriteTimestamps(bkt, commitTime, commitTime); err != nil {
-		return err
+		return "", err
 	}
 	if err := boltutil.WriteLabels(bkt, base.Labels); err != nil {
-		return err
+		return "", err
 	}
-	return bkt.Put(bucketKeySize, sizeEncoded)
+	return actual, bkt.Put(bucketKeySize, sizeEncoded)
 }
 
 func (nw *namespacedWriter) Status() (content.Status, error) {

metadata/gc.go

@@ -46,6 +46,55 @@ func scanRoots(ctx context.Context, tx *bolt.Tx, nc chan<- gc.Node) error {
 		nbkt := v1bkt.Bucket(k)
 		ns := string(k)
 
+		lbkt := nbkt.Bucket(bucketKeyObjectLeases)
+		if lbkt != nil {
+			if err := lbkt.ForEach(func(k, v []byte) error {
+				if v != nil {
+					return nil
+				}
+				libkt := lbkt.Bucket(k)
+
+				cbkt := libkt.Bucket(bucketKeyObjectContent)
+				if cbkt != nil {
+					if err := cbkt.ForEach(func(k, v []byte) error {
+						select {
+						case nc <- gcnode(ResourceContent, ns, string(k)):
+						case <-ctx.Done():
+							return ctx.Err()
+						}
+						return nil
+					}); err != nil {
+						return err
+					}
+				}
+
+				sbkt := libkt.Bucket(bucketKeyObjectSnapshots)
+				if sbkt != nil {
+					if err := sbkt.ForEach(func(sk, sv []byte) error {
+						if sv != nil {
+							return nil
+						}
+						snbkt := sbkt.Bucket(sk)
+
+						return snbkt.ForEach(func(k, v []byte) error {
+							select {
+							case nc <- gcnode(ResourceSnapshot, ns, fmt.Sprintf("%s/%s", sk, k)):
+							case <-ctx.Done():
+								return ctx.Err()
+							}
+							return nil
+						})
+					}); err != nil {
+						return err
+					}
+				}
+
+				return nil
+			}); err != nil {
+				return err
+			}
+		}
+
 		ibkt := nbkt.Bucket(bucketKeyObjectImages)
 		if ibkt != nil {
 			if err := ibkt.ForEach(func(k, v []byte) error {

metadata/gc_test.go

@@ -34,14 +34,22 @@ func TestGCRoots(t *testing.T) {
 		addSnapshot("ns1", "overlay", "sn1", "", nil),
 		addSnapshot("ns1", "overlay", "sn2", "", nil),
 		addSnapshot("ns1", "overlay", "sn3", "", labelmap(string(labelGCRoot), "always")),
+		addLeaseSnapshot("ns2", "l1", "overlay", "sn5"),
+		addLeaseSnapshot("ns2", "l2", "overlay", "sn6"),
+		addLeaseContent("ns2", "l1", dgst(4)),
+		addLeaseContent("ns2", "l2", dgst(5)),
 	}
 
 	expected := []gc.Node{
 		gcnode(ResourceContent, "ns1", dgst(1).String()),
 		gcnode(ResourceContent, "ns1", dgst(2).String()),
 		gcnode(ResourceContent, "ns2", dgst(2).String()),
+		gcnode(ResourceContent, "ns2", dgst(4).String()),
+		gcnode(ResourceContent, "ns2", dgst(5).String()),
 		gcnode(ResourceSnapshot, "ns1", "overlay/sn2"),
 		gcnode(ResourceSnapshot, "ns1", "overlay/sn3"),
+		gcnode(ResourceSnapshot, "ns2", "overlay/sn5"),
+		gcnode(ResourceSnapshot, "ns2", "overlay/sn6"),
 	}
 
 	if err := db.Update(func(tx *bolt.Tx) error {
@@ -374,6 +382,26 @@ func addContent(ns string, dgst digest.Digest, labels map[string]string) alterFu
 	}
 }
 
+func addLeaseSnapshot(ns, lid, snapshotter, name string) alterFunc {
+	return func(bkt *bolt.Bucket) error {
+		sbkt, err := createBuckets(bkt, ns, string(bucketKeyObjectLeases), lid, string(bucketKeyObjectSnapshots), snapshotter)
+		if err != nil {
+			return err
+		}
+		return sbkt.Put([]byte(name), nil)
+	}
+}
+
+func addLeaseContent(ns, lid string, dgst digest.Digest) alterFunc {
+	return func(bkt *bolt.Bucket) error {
+		cbkt, err := createBuckets(bkt, ns, string(bucketKeyObjectLeases), lid, string(bucketKeyObjectContent))
+		if err != nil {
+			return err
+		}
+		return cbkt.Put([]byte(dgst.String()), nil)
+	}
+}
+
 func addContainer(ns, name, snapshotter, snapshot string, labels map[string]string) alterFunc {
 	return func(bkt *bolt.Bucket) error {
 		cbkt, err := createBuckets(bkt, ns, string(bucketKeyObjectContainers), name)

metadata/leases.go

@@ -0,0 +1,201 @@
+package metadata
+
+import (
+	"context"
+	"time"
+
+	"github.com/boltdb/bolt"
+	"github.com/containerd/containerd/errdefs"
+	"github.com/containerd/containerd/leases"
+	"github.com/containerd/containerd/metadata/boltutil"
+	"github.com/containerd/containerd/namespaces"
+	digest "github.com/opencontainers/go-digest"
+	"github.com/pkg/errors"
+)
+
+// Lease retains resources to prevent garbage collection before
+// the resources can be fully referenced.
+type Lease struct {
+	ID        string
+	CreatedAt time.Time
+	Labels    map[string]string
+
+	Content   []string
+	Snapshots map[string][]string
+}
+
+// LeaseManager manages the create/delete lifecyle of leases
+// and also returns existing leases
+type LeaseManager struct {
+	tx *bolt.Tx
+}
+
+// NewLeaseManager creates a new lease manager for managing leases using
+// the provided database transaction.
+func NewLeaseManager(tx *bolt.Tx) *LeaseManager {
+	return &LeaseManager{
+		tx: tx,
+	}
+}
+
+// Create creates a new lease using the provided lease
+func (lm *LeaseManager) Create(ctx context.Context, lid string, labels map[string]string) (Lease, error) {
+	namespace, err := namespaces.NamespaceRequired(ctx)
+	if err != nil {
+		return Lease{}, err
+	}
+
+	topbkt, err := createBucketIfNotExists(lm.tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectLeases)
+	if err != nil {
+		return Lease{}, err
+	}
+
+	txbkt, err := topbkt.CreateBucket([]byte(lid))
+	if err != nil {
+		if err == bolt.ErrBucketExists {
+			err = errdefs.ErrAlreadyExists
+		}
+		return Lease{}, err
+	}
+
+	t := time.Now().UTC()
+	createdAt, err := t.MarshalBinary()
+	if err != nil {
+		return Lease{}, err
+	}
+	if err := txbkt.Put(bucketKeyCreatedAt, createdAt); err != nil {
+		return Lease{}, err
+	}
+
+	if labels != nil {
+		if err := boltutil.WriteLabels(txbkt, labels); err != nil {
+			return Lease{}, err
+		}
+	}
+
+	return Lease{
+		ID:        lid,
+		CreatedAt: t,
+		Labels:    labels,
+	}, nil
+}
+
+// Delete delets the lease with the provided lease ID
+func (lm *LeaseManager) Delete(ctx context.Context, lid string) error {
+	namespace, err := namespaces.NamespaceRequired(ctx)
+	if err != nil {
+		return err
+	}
+
+	topbkt := getBucket(lm.tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectLeases)
+	if topbkt == nil {
+		return nil
+	}
+	if err := topbkt.DeleteBucket([]byte(lid)); err != nil && err != bolt.ErrBucketNotFound {
+		return err
+	}
+	return nil
+}
+
+// List lists all active leases
+func (lm *LeaseManager) List(ctx context.Context, includeResources bool, filter ...string) ([]Lease, error) {
+	namespace, err := namespaces.NamespaceRequired(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	var leases []Lease
+
+	topbkt := getBucket(lm.tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectLeases)
+	if topbkt == nil {
+		return leases, nil
+	}
+
+	if err := topbkt.ForEach(func(k, v []byte) error {
+		if v != nil {
+			return nil
+		}
+		txbkt := topbkt.Bucket(k)
+
+		l := Lease{
+			ID: string(k),
+		}
+
+		if v := txbkt.Get(bucketKeyCreatedAt); v != nil {
+			t := &l.CreatedAt
+			if err := t.UnmarshalBinary(v); err != nil {
+				return err
+			}
+		}
+
+		labels, err := boltutil.ReadLabels(txbkt)
+		if err != nil {
+			return err
+		}
+		l.Labels = labels
+
+		// TODO: Read Snapshots
+		// TODO: Read Content
+
+		leases = append(leases, l)
+
+		return nil
+	}); err != nil {
+		return nil, err
+	}
+
+	return leases, nil
+}
+
+func addSnapshotLease(ctx context.Context, tx *bolt.Tx, snapshotter, key string) error {
+	lid, ok := leases.Lease(ctx)
+	if !ok {
+		return nil
+	}
+
+	namespace, ok := namespaces.Namespace(ctx)
+	if !ok {
+		panic("namespace must already be required")
+	}
+
+	bkt := getBucket(tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectLeases, []byte(lid))
+	if bkt == nil {
+		return errors.Wrap(errdefs.ErrNotFound, "lease does not exist")
+	}
+
+	bkt, err := bkt.CreateBucketIfNotExists(bucketKeyObjectSnapshots)
+	if err != nil {
+		return err
+	}
+
+	bkt, err = bkt.CreateBucketIfNotExists([]byte(snapshotter))
+	if err != nil {
+		return err
+	}
+
+	return bkt.Put([]byte(key), nil)
+}
+
+func addContentLease(ctx context.Context, tx *bolt.Tx, dgst digest.Digest) error {
+	lid, ok := leases.Lease(ctx)
+	if !ok {
+		return nil
+	}
+
+	namespace, ok := namespaces.Namespace(ctx)
+	if !ok {
+		panic("namespace must already be required")
+	}
+
+	bkt := getBucket(tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectLeases, []byte(lid))
+	if bkt == nil {
+		return errors.Wrap(errdefs.ErrNotFound, "lease does not exist")
+	}
+
+	bkt, err := bkt.CreateBucketIfNotExists(bucketKeyObjectContent)
+	if err != nil {
+		return err
+	}
+
+	return bkt.Put([]byte(dgst.String()), nil)
+}

metadata/leases_test.go

@@ -0,0 +1,90 @@
+package metadata
+
+import (
+	"testing"
+
+	"github.com/boltdb/bolt"
+	"github.com/containerd/containerd/errdefs"
+	"github.com/pkg/errors"
+)
+
+func TestLeases(t *testing.T) {
+	ctx, db, cancel := testEnv(t)
+	defer cancel()
+
+	testCases := []struct {
+		ID    string
+		Cause error
+	}{
+		{
+			ID: "tx1",
+		},
+		{
+			ID:    "tx1",
+			Cause: errdefs.ErrAlreadyExists,
+		},
+		{
+			ID: "tx2",
+		},
+	}
+
+	var leases []Lease
+
+	for _, tc := range testCases {
+		if err := db.Update(func(tx *bolt.Tx) error {
+			lease, err := NewLeaseManager(tx).Create(ctx, tc.ID, nil)
+			if err != nil {
+				if tc.Cause != nil && errors.Cause(err) == tc.Cause {
+					return nil
+				}
+				return err
+			}
+			leases = append(leases, lease)
+			return nil
+		}); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	var listed []Lease
+	// List leases, check same
+	if err := db.View(func(tx *bolt.Tx) error {
+		var err error
+		listed, err = NewLeaseManager(tx).List(ctx, false)
+		return err
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	if len(listed) != len(leases) {
+		t.Fatalf("Expected %d lease, got %d", len(leases), len(listed))
+	}
+	for i := range listed {
+		if listed[i].ID != leases[i].ID {
+			t.Fatalf("Expected lease ID %s, got %s", leases[i].ID, listed[i].ID)
+		}
+		if listed[i].CreatedAt != leases[i].CreatedAt {
+			t.Fatalf("Expected lease created at time %s, got %s", leases[i].CreatedAt, listed[i].CreatedAt)
+		}
+	}
+
+	for _, tc := range testCases {
+		if err := db.Update(func(tx *bolt.Tx) error {
+			return NewLeaseManager(tx).Delete(ctx, tc.ID)
+		}); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	if err := db.View(func(tx *bolt.Tx) error {
+		var err error
+		listed, err = NewLeaseManager(tx).List(ctx, false)
+		return err
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	if len(listed) > 0 {
+		t.Fatalf("Expected no leases, found %d: %v", len(listed), listed)
+	}
+}

metadata/snapshot.go

@@ -326,6 +326,10 @@ func (s *snapshotter) createSnapshot(ctx context.Context, key, parent string, re
 			return err
 		}
 
+		if err := addSnapshotLease(ctx, tx, s.name, key); err != nil {
+			return err
+		}
+
 		// TODO: Consider doing this outside of transaction to lessen
 		// metadata lock time
 		if readonly {