metadata: improve deleting a non-empty namespace's error message

Deleting a non-empty namespace fails with

> namespace must be empty: failed precondition

This change improves the error message by listing the types of
the objects in the namespace that prevent deletion.

Signed-off-by: Kazuyoshi Kato <katokazu@amazon.com>

metadata/namespaces.go

@@ -18,6 +18,7 @@ package metadata
 
 import (
 	"context"
+	"strings"
 
 	"github.com/containerd/containerd/errdefs"
 	"github.com/containerd/containerd/identifiers"
@@ -140,10 +141,17 @@ func (s *namespaceStore) Delete(ctx context.Context, namespace string, opts ...n
 		}
 	}
 	bkt := getBucket(s.tx, bucketKeyVersion)
-	if empty, err := s.namespaceEmpty(ctx, namespace); err != nil {
+	types, err := s.listNs(namespace)
+	if err != nil {
 		return err
-	} else if !empty {
-		return errors.Wrapf(errdefs.ErrFailedPrecondition, "namespace %q must be empty", namespace)
+	}
+
+	if len(types) > 0 {
+		return errors.Wrapf(
+			errdefs.ErrFailedPrecondition,
+			"namespace %q must be empty, but it still has %s",
+			namespace, strings.Join(types, ", "),
+		)
 	}
 
 	if err := bkt.DeleteBucket([]byte(namespace)); err != nil {
@@ -157,32 +165,35 @@ func (s *namespaceStore) Delete(ctx context.Context, namespace string, opts ...n
 	return nil
 }
 
-func (s *namespaceStore) namespaceEmpty(ctx context.Context, namespace string) (bool, error) {
-	// Get all data buckets
-	buckets := []*bolt.Bucket{
-		getImagesBucket(s.tx, namespace),
-		getBlobsBucket(s.tx, namespace),
-		getContainersBucket(s.tx, namespace),
+// listNs returns the types of the remaining objects inside the given namespace.
+// It doesn't return exact objects due to performance concerns.
+func (s *namespaceStore) listNs(namespace string) ([]string, error) {
+	var out []string
+
+	if !isBucketEmpty(getImagesBucket(s.tx, namespace)) {
+		out = append(out, "images")
 	}
+	if !isBucketEmpty(getBlobsBucket(s.tx, namespace)) {
+		out = append(out, "blobs")
+	}
+	if !isBucketEmpty(getContainersBucket(s.tx, namespace)) {
+		out = append(out, "containers")
+	}
+
 	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))
+				if !isBucketEmpty(snbkt.Bucket(k)) {
+					out = append(out, "snapshot-"+string(k))
+				}
 			}
 			return nil
 		}); err != nil {
-			return false, err
+			return nil, err
 		}
 	}
 
-	// Ensure data buckets are empty
-	for _, bkt := range buckets {
-		if !isBucketEmpty(bkt) {
-			return false, nil
-		}
-	}
-
-	return true, nil
+	return out, nil
 }
 
 func isBucketEmpty(bkt *bolt.Bucket) bool {

metadata/namespaces_test.go

@@ -0,0 +1,85 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package metadata
+
+import (
+	"context"
+	"testing"
+
+	"github.com/containerd/containerd/containers"
+	"github.com/containerd/containerd/namespaces"
+	"github.com/gogo/protobuf/types"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.etcd.io/bbolt"
+)
+
+func TestCreateDelete(t *testing.T) {
+	ctx, db, cleanup := testDB(t)
+	defer cleanup()
+
+	subtests := []struct {
+		name     string
+		create   func(t *testing.T, ctx context.Context)
+		validate func(t *testing.T, err error)
+	}{
+		{
+			name:   "empty",
+			create: func(t *testing.T, ctx context.Context) {},
+			validate: func(t *testing.T, err error) {
+				require.NoError(t, err)
+			},
+		},
+		{
+			name: "not-empty",
+			create: func(t *testing.T, ctx context.Context) {
+				store := NewContainerStore(db)
+				_, err := store.Create(ctx, containers.Container{
+					ID:      "c1",
+					Runtime: containers.RuntimeInfo{Name: "rt"},
+					Spec:    &types.Any{},
+				})
+				require.NoError(t, err)
+			},
+			validate: func(t *testing.T, err error) {
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), "still has containers")
+			},
+		},
+	}
+
+	for _, subtest := range subtests {
+		ns := subtest.name
+		ctx = namespaces.WithNamespace(ctx, ns)
+
+		t.Run(subtest.name, func(t *testing.T) {
+			err := db.Update(func(tx *bbolt.Tx) error {
+				store := NewNamespaceStore(tx)
+				return store.Create(ctx, ns, nil)
+			})
+			require.NoError(t, err)
+
+			subtest.create(t, ctx)
+
+			err = db.Update(func(tx *bbolt.Tx) error {
+				store := NewNamespaceStore(tx)
+				return store.Delete(ctx, ns)
+			})
+			subtest.validate(t, err)
+		})
+	}
+}