Merge pull request #1707 from dmcgowan/snapshot-storage-db

Open snapshot database only once

snapshot/storage/bolt.go

@@ -30,21 +30,6 @@ var (
 	ErrNoTransaction = errors.New("no transaction in context")
 )
 
-type boltFileTransactor struct {
-	db *bolt.DB
-	tx *bolt.Tx
-}
-
-func (bft *boltFileTransactor) Rollback() error {
-	defer bft.db.Close()
-	return bft.tx.Rollback()
-}
-
-func (bft *boltFileTransactor) Commit() error {
-	defer bft.db.Close()
-	return bft.tx.Commit()
-}
-
 // parentKey returns a composite key of the parent and child identifiers. The
 // parts of the key are separated by a zero byte.
 func parentKey(parent, child uint64) []byte {
@@ -405,11 +390,11 @@ func CommitActive(ctx context.Context, key, name string, usage snapshot.Usage, o
 }
 
 func withSnapshotBucket(ctx context.Context, key string, fn func(context.Context, *bolt.Bucket, *bolt.Bucket) error) error {
-	t, ok := ctx.Value(transactionKey{}).(*boltFileTransactor)
+	tx, ok := ctx.Value(transactionKey{}).(*bolt.Tx)
 	if !ok {
 		return ErrNoTransaction
 	}
-	bkt := t.tx.Bucket(bucketKeyStorageVersion)
+	bkt := tx.Bucket(bucketKeyStorageVersion)
 	if bkt == nil {
 		return errors.Wrap(errdefs.ErrNotFound, "bucket does not exist")
 	}
@@ -426,11 +411,11 @@ func withSnapshotBucket(ctx context.Context, key string, fn func(context.Context
 }
 
 func withBucket(ctx context.Context, fn func(context.Context, *bolt.Bucket, *bolt.Bucket) error) error {
-	t, ok := ctx.Value(transactionKey{}).(*boltFileTransactor)
+	tx, ok := ctx.Value(transactionKey{}).(*bolt.Tx)
 	if !ok {
 		return ErrNoTransaction
 	}
-	bkt := t.tx.Bucket(bucketKeyStorageVersion)
+	bkt := tx.Bucket(bucketKeyStorageVersion)
 	if bkt == nil {
 		return errors.Wrap(errdefs.ErrNotFound, "bucket does not exist")
 	}
@@ -438,12 +423,12 @@ func withBucket(ctx context.Context, fn func(context.Context, *bolt.Bucket, *bol
 }
 
 func createBucketIfNotExists(ctx context.Context, fn func(context.Context, *bolt.Bucket, *bolt.Bucket) error) error {
-	t, ok := ctx.Value(transactionKey{}).(*boltFileTransactor)
+	tx, ok := ctx.Value(transactionKey{}).(*bolt.Tx)
 	if !ok {
 		return ErrNoTransaction
 	}
 
-	bkt, err := t.tx.CreateBucketIfNotExists(bucketKeyStorageVersion)
+	bkt, err := tx.CreateBucketIfNotExists(bucketKeyStorageVersion)
 	if err != nil {
 		return errors.Wrap(err, "failed to create version bucket")
 	}

snapshot/storage/metastore.go

@@ -7,6 +7,7 @@ package storage
 
 import (
 	"context"
+	"sync"
 
 	"github.com/boltdb/bolt"
 	"github.com/containerd/containerd/snapshot"
@@ -46,6 +47,9 @@ type Snapshot struct {
 // complexities of a driver implementation.
 type MetaStore struct {
 	dbfile string
+
+	dbL sync.Mutex
+	db  *bolt.DB
 }
 
 // NewMetaStore returns a snapshot MetaStore for storage of metadata related to
@@ -63,22 +67,33 @@ type transactionKey struct{}
 // TransactionContext creates a new transaction context. The writable value
 // should be set to true for transactions which are expected to mutate data.
 func (ms *MetaStore) TransactionContext(ctx context.Context, writable bool) (context.Context, Transactor, error) {
-	db, err := bolt.Open(ms.dbfile, 0600, nil)
-	if err != nil {
-		return ctx, nil, errors.Wrap(err, "failed to open database file")
+	ms.dbL.Lock()
+	if ms.db == nil {
+		db, err := bolt.Open(ms.dbfile, 0600, nil)
+		if err != nil {
+			ms.dbL.Unlock()
+			return ctx, nil, errors.Wrap(err, "failed to open database file")
+		}
+		ms.db = db
 	}
+	ms.dbL.Unlock()
 
-	tx, err := db.Begin(writable)
+	tx, err := ms.db.Begin(writable)
 	if err != nil {
 		return ctx, nil, errors.Wrap(err, "failed to start transaction")
 	}
 
-	t := &boltFileTransactor{
-		db: db,
-		tx: tx,
-	}
+	ctx = context.WithValue(ctx, transactionKey{}, tx)
 
-	ctx = context.WithValue(ctx, transactionKey{}, t)
-
-	return ctx, t, nil
+	return ctx, tx, nil
+}
+
+// Close closes the metastore and any underlying database connections
+func (ms *MetaStore) Close() error {
+	ms.dbL.Lock()
+	defer ms.dbL.Unlock()
+	if ms.db == nil {
+		return nil
+	}
+	return ms.db.Close()
 }

snapshot/testsuite/helpers.go

@@ -122,3 +122,33 @@ func checkSnapshots(ctx context.Context, sn snapshot.Snapshotter, work string, a
 	return nil
 
 }
+
+// checkInfo checks that the infos are the same
+func checkInfo(si1, si2 snapshot.Info) error {
+	if si1.Kind != si2.Kind {
+		return errors.Errorf("Expected kind %v, got %v", si1.Kind, si2.Kind)
+	}
+	if si1.Name != si2.Name {
+		return errors.Errorf("Expected name %v, got %v", si1.Name, si2.Name)
+	}
+	if si1.Parent != si2.Parent {
+		return errors.Errorf("Expected Parent %v, got %v", si1.Parent, si2.Parent)
+	}
+	if len(si1.Labels) != len(si2.Labels) {
+		return errors.Errorf("Expected %d labels, got %d", len(si1.Labels), len(si2.Labels))
+	}
+	for k, l1 := range si1.Labels {
+		l2 := si2.Labels[k]
+		if l1 != l2 {
+			return errors.Errorf("Expected label %v, got %v", l1, l2)
+		}
+	}
+	if si1.Created != si2.Created {
+		return errors.Errorf("Expected Created %v, got %v", si1.Created, si2.Created)
+	}
+	if si1.Updated != si2.Updated {
+		return errors.Errorf("Expected Updated %v, got %v", si1.Updated, si2.Updated)
+	}
+
+	return nil
+}

snapshot/testsuite/issues.go

@@ -2,6 +2,8 @@ package testsuite
 
 import (
 	"context"
+	"fmt"
+	"strings"
 	"testing"
 	"time"
 
@@ -151,6 +153,54 @@ func checkDirectoryPermissionOnCommit(ctx context.Context, t *testing.T, sn snap
 	}
 }
 
+// checkStatInWalk ensures that a stat can be called during a walk
+func checkStatInWalk(ctx context.Context, t *testing.T, sn snapshot.Snapshotter, work string) {
+	prefix := "stats-in-walk-"
+	if err := createNamedSnapshots(ctx, sn, prefix); err != nil {
+		t.Fatal(err)
+	}
+
+	err := sn.Walk(ctx, func(ctx context.Context, si snapshot.Info) error {
+		if !strings.HasPrefix(si.Name, prefix) {
+			// Only stat snapshots from this test
+			return nil
+		}
+		si2, err := sn.Stat(ctx, si.Name)
+		if err != nil {
+			return err
+		}
+
+		return checkInfo(si, si2)
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func createNamedSnapshots(ctx context.Context, snapshotter snapshot.Snapshotter, ns string) error {
+	c1 := fmt.Sprintf("%sc1", ns)
+	c2 := fmt.Sprintf("%sc2", ns)
+	if _, err := snapshotter.Prepare(ctx, c1+"-a", ""); err != nil {
+		return err
+	}
+	if err := snapshotter.Commit(ctx, c1, c1+"-a"); err != nil {
+		return err
+	}
+	if _, err := snapshotter.Prepare(ctx, c2+"-a", c1); err != nil {
+		return err
+	}
+	if err := snapshotter.Commit(ctx, c2, c2+"-a"); err != nil {
+		return err
+	}
+	if _, err := snapshotter.Prepare(ctx, fmt.Sprintf("%sa1", ns), c2); err != nil {
+		return err
+	}
+	if _, err := snapshotter.View(ctx, fmt.Sprintf("%sv1", ns), c2); err != nil {
+		return err
+	}
+	return nil
+}
+
 // More issues to test
 //
 // checkRemoveAfterCommit
@@ -170,3 +220,6 @@ func checkDirectoryPermissionOnCommit(ctx context.Context, t *testing.T, sn snap
 //
 // checkChmod
 // See https://github.com/docker/machine/issues/3327
+//
+// checkRemoveInWalk
+// Allow mutations during walk without deadlocking

snapshot/testsuite/testsuite.go

@@ -42,6 +42,8 @@ func SnapshotterSuite(t *testing.T, name string, snapshotterFn func(ctx context.
 	//t.Run("Rename", makeTest(name, snapshotterFn, checkRename))
 
 	t.Run("ViewReadonly", makeTest(name, snapshotterFn, checkSnapshotterViewReadonly))
+
+	t.Run("StatInWalk", makeTest(name, snapshotterFn, checkStatInWalk))
 }
 
 func makeTest(name string, snapshotterFn func(ctx context.Context, root string) (snapshot.Snapshotter, func() error, error), fn func(ctx context.Context, t *testing.T, snapshotter snapshot.Snapshotter, work string)) func(t *testing.T) {