From 5a54862ae53abeaadbd3937d5f36df4701538d45 Mon Sep 17 00:00:00 2001
From: Derek McGowan <derek@mcgstyle.net>
Date: Wed, 3 Jan 2018 10:29:08 -0800
Subject: [PATCH] Update namespace empty check to use buckets

Directly get and check whether a bucket is empty.
This prevents unnecessarily loading all records of
the buckets into memory just to check existence.
Also added checks for content and snapshots.

Signed-off-by: Derek McGowan <derek@mcgstyle.net>
---
 metadata/buckets.go    | 13 ++++++-----
 metadata/namespaces.go | 49 ++++++++++++++++++++++++------------------
 2 files changed, 34 insertions(+), 28 deletions(-)

diff --git a/metadata/buckets.go b/metadata/buckets.go
index 9325f1698..45309724f 100644
--- a/metadata/buckets.go
+++ b/metadata/buckets.go
@@ -106,13 +106,8 @@ func imagesBucketPath(namespace string) [][]byte {
 	return [][]byte{bucketKeyVersion, []byte(namespace), bucketKeyObjectImages}
 }
 
-func withImagesBucket(tx *bolt.Tx, namespace string, fn func(bkt *bolt.Bucket) error) error {
-	bkt, err := createBucketIfNotExists(tx, imagesBucketPath(namespace)...)
-	if err != nil {
-		return err
-	}
-
-	return fn(bkt)
+func createImagesBucket(tx *bolt.Tx, namespace string) (*bolt.Bucket, error) {
+	return createBucketIfNotExists(tx, imagesBucketPath(namespace)...)
 }
 
 func getImagesBucket(tx *bolt.Tx, namespace string) *bolt.Bucket {
@@ -143,6 +138,10 @@ func createSnapshotterBucket(tx *bolt.Tx, namespace, snapshotter string) (*bolt.
 	return bkt, nil
 }
 
+func getSnapshottersBucket(tx *bolt.Tx, namespace string) *bolt.Bucket {
+	return getBucket(tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectSnapshots)
+}
+
 func getSnapshotterBucket(tx *bolt.Tx, namespace, snapshotter string) *bolt.Bucket {
 	return getBucket(tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectSnapshots, []byte(snapshotter))
 }
diff --git a/metadata/namespaces.go b/metadata/namespaces.go
index 4b4c4e5fe..4f84051d3 100644
--- a/metadata/namespaces.go
+++ b/metadata/namespaces.go
@@ -133,31 +133,38 @@ func (s *namespaceStore) Delete(ctx context.Context, namespace string) error {
 }
 
 func (s *namespaceStore) namespaceEmpty(ctx context.Context, namespace string) (bool, error) {
-	ctx = namespaces.WithNamespace(ctx, namespace)
-
-	// need to check the various object stores.
-
-	imageStore := NewImageStore(s.tx)
-	images, err := imageStore.List(ctx)
-	if err != nil {
-		return false, err
+	// Get all data buckets
+	buckets := []*bolt.Bucket{
+		getImagesBucket(s.tx, namespace),
+		getBlobsBucket(s.tx, namespace),
+		getContainersBucket(s.tx, namespace),
 	}
-	if len(images) > 0 {
-		return false, nil
+	if snbkt := getSnapshottersBucket(s.tx, namespace); snbkt != nil {
+		if err := snbkt.ForEach(func(k, v []byte) error {
+			if v == nil {
+				buckets = append(buckets, snbkt.Bucket(k))
+			}
+			return nil
+		}); err != nil {
+			return false, err
+		}
 	}
 
-	containerStore := NewContainerStore(s.tx)
-	containers, err := containerStore.List(ctx)
-	if err != nil {
-		return false, err
+	// Ensure data buckets are empty
+	for _, bkt := range buckets {
+		if !isBucketEmpty(bkt) {
+			return false, nil
+		}
 	}
 
-	if len(containers) > 0 {
-		return false, nil
-	}
-
-	// TODO(stevvooe): Need to add check for content store, as well. Still need
-	// to make content store namespace aware.
-
 	return true, nil
 }
+
+func isBucketEmpty(bkt *bolt.Bucket) bool {
+	if bkt == nil {
+		return true
+	}
+
+	k, _ := bkt.Cursor().First()
+	return k == nil
+}