Merge pull request #5436 from sp98/obc-finalizer

finalizer for OBC cleanup

pkg/operator/ceph/object/bucket/util.go

@@ -17,9 +17,7 @@ limitations under the License.
 package bucket
 
 import (
-	"fmt"
 	"math/rand"
-	"strings"
 	"time"
 
 	"github.com/coreos/pkg/capnslog"
@@ -35,8 +33,7 @@ import (
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	cephclientset "github.com/rook/rook/pkg/client/clientset/versioned/typed/ceph.rook.io/v1"
-	controllerutil "github.com/rook/rook/pkg/operator/ceph/controller"
-	"github.com/rook/rook/pkg/operator/k8sutil"
+	cephObject "github.com/rook/rook/pkg/operator/ceph/object"
 )
 
 var logger = capnslog.NewPackageLogger("github.com/rook/rook", "op-bucket-prov")
@@ -44,7 +41,6 @@ var logger = capnslog.NewPackageLogger("github.com/rook/rook", "op-bucket-prov")
 const (
 	genUserLen           = 8
 	cephUser             = "cephUser"
-	provisionerName      = "ceph.rook.io/bucket"
 	prefixObjectStoreSvc = "rook-ceph-rgw"
 	accessKeyIdKey       = "accessKeyID"
 	secretSecretKeyKey   = "secretAccessKey"
@@ -55,16 +51,7 @@ const (
 
 func NewBucketController(cfg *rest.Config, p *Provisioner) (*provisioner.Provisioner, error) {
 	const allNamespaces = ""
-	provName := provisionerName
-
-	obcWatchOnNamespace, err := k8sutil.GetOperatorSetting(p.context.Clientset, controllerutil.OperatorSettingConfigMapName, "ROOK_OBC_WATCH_OPERATOR_NAMESPACE", "false")
-	if err != nil {
-		logger.Warningf("failed to verify if obc should watch the operator namespace or all of them, watching all")
-	} else {
-		if strings.EqualFold(obcWatchOnNamespace, "true") {
-			provName = fmt.Sprintf("%s.%s", p.namespace, provisionerName)
-		}
-	}
+	provName := cephObject.GetObjectBucketProvisioner(p.context, p.namespace)
 
 	logger.Infof("ceph bucket provisioner launched watching for provisioner %q", provName)
 	return provisioner.NewProvisioner(cfg, provName, p, allNamespaces)

pkg/operator/ceph/object/controller.go

@@ -20,8 +20,10 @@ import (
 	"context"
 	"fmt"
 	"reflect"
+	"strings"
 
 	"github.com/coreos/pkg/capnslog"
+	bktclient "github.com/kube-object-storage/lib-bucket-provisioner/pkg/client/clientset/versioned"
 	"github.com/pkg/errors"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
@@ -70,6 +72,7 @@ var controllerTypeMeta = metav1.TypeMeta{
 // ReconcileCephObjectStore reconciles a cephObjectStore object
 type ReconcileCephObjectStore struct {
 	client          client.Client
+	bktclient       bktclient.Interface
 	scheme          *runtime.Scheme
 	context         *clusterd.Context
 	cephClusterSpec *cephv1.ClusterSpec
@@ -89,9 +92,10 @@ func newReconciler(mgr manager.Manager, context *clusterd.Context) reconcile.Rec
 	cephv1.AddToScheme(mgr.GetScheme())
 
 	return &ReconcileCephObjectStore{
-		client:  mgr.GetClient(),
-		scheme:  mgrScheme,
-		context: context,
+		client:    mgr.GetClient(),
+		scheme:    mgrScheme,
+		context:   context,
+		bktclient: bktclient.NewForConfigOrDie(context.KubeConfig),
 	}
 }
 
@@ -203,10 +207,18 @@ func (r *ReconcileCephObjectStore) reconcile(request reconcile.Request) (reconci
 	// DELETE: the CR was deleted
 	if !cephObjectStore.GetDeletionTimestamp().IsZero() {
 		logger.Debugf("deleting store %q", cephObjectStore.Name)
+
+		response, ok := r.verifyObjectBucketCleanup(cephObjectStore)
+		if !ok {
+			// If the object store cannot be deleted, requeue the request for deletion to see if the conditions
+			// will eventually be satisfied such as the object buckets being removed
+			return response, nil
+		}
+
 		cfg := clusterConfig{context: r.context, store: cephObjectStore}
-		err := cfg.deleteStore()
+		err = cfg.deleteStore()
 		if err != nil {
-			return reconcile.Result{}, errors.Wrapf(err, "failed to delete store %q. ", cephObjectStore.Name)
+			return reconcile.Result{}, errors.Wrapf(err, "failed to delete store %q", cephObjectStore.Name)
 		}
 
 		// Remove finalizer
@@ -319,3 +331,27 @@ func updateStatus(client client.Client, name types.NamespacedName, status string
 	}
 	logger.Debugf("object store %q status updated to %q", name, status)
 }
+
+func (r *ReconcileCephObjectStore) verifyObjectBucketCleanup(objectstore *cephv1.CephObjectStore) (reconcile.Result, bool) {
+	bktProvsioner := GetObjectBucketProvisioner(r.context, objectstore.Namespace)
+	bktProvsioner = strings.Replace(bktProvsioner, "/", "-", -1)
+	selector := fmt.Sprintf("bucket-provisioner=%s", bktProvsioner)
+	objectBuckets, err := r.bktclient.ObjectbucketV1alpha1().ObjectBuckets().List(metav1.ListOptions{LabelSelector: selector})
+	if err != nil {
+		logger.Errorf("failed to delete object store. failed to list buckets for objectstore %q in namespace %q", objectstore.Name, objectstore.Namespace)
+		return opcontroller.WaitForRequeueIfFinalizerBlocked, false
+	}
+
+	if len(objectBuckets.Items) == 0 {
+		logger.Infof("no buckets found for objectstore %q in namespace %q", objectstore.Name, objectstore.Namespace)
+		return reconcile.Result{}, true
+	}
+
+	bucketNames := make([]string, 0)
+	for _, bucket := range objectBuckets.Items {
+		bucketNames = append(bucketNames, bucket.Name)
+	}
+
+	logger.Errorf("failed to delete object store. buckets for objectstore %q in namespace %q are not cleaned up. remaining buckets: %+v", objectstore.Name, objectstore.Namespace, bucketNames)
+	return opcontroller.WaitForRequeueIfFinalizerBlocked, false
+}

pkg/operator/ceph/object/objectstore.go

@@ -24,16 +24,20 @@ import (
 
 	"github.com/pkg/errors"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
 	ceph "github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/rook/rook/pkg/operator/ceph/config"
+	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
+	"github.com/rook/rook/pkg/operator/k8sutil"
 )
 
 const (
 	rootPool = ".rgw.root"
 
 	// AppName is the name Rook uses for the object store's application
-	AppName = "rook-ceph-rgw"
+	AppName               = "rook-ceph-rgw"
+	bucketProvisionerName = "ceph.rook.io/bucket"
 )
 
 var (
@@ -333,3 +337,17 @@ func poolName(storeName, poolName string) string {
 	// the name of the pool is <instance>.<name>, except for the pool ".rgw.root" that spans object stores
 	return fmt.Sprintf("%s.%s", storeName, poolName)
 }
+
+// GetObjectBucketProvisioner returns the bucket provisioner name appended with operator namespace if OBC is watching on it
+func GetObjectBucketProvisioner(c *clusterd.Context, namespace string) string {
+	provName := bucketProvisionerName
+	obcWatchOnNamespace, err := k8sutil.GetOperatorSetting(c.Clientset, opcontroller.OperatorSettingConfigMapName, "ROOK_OBC_WATCH_OPERATOR_NAMESPACE", "false")
+	if err != nil {
+		logger.Warningf("failed to verify if obc should watch the operator namespace or all of them, watching all")
+	} else {
+		if strings.EqualFold(obcWatchOnNamespace, "true") {
+			provName = fmt.Sprintf("%s.%s", namespace, bucketProvisionerName)
+		}
+	}
+	return provName
+}

pkg/operator/ceph/object/objectstore_test.go

@@ -18,13 +18,18 @@ package object
 
 import (
 	"fmt"
+	"os"
 	"testing"
 
 	"github.com/pkg/errors"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
+	"github.com/rook/rook/pkg/operator/k8sutil"
 	exectest "github.com/rook/rook/pkg/util/exec/test"
 	"github.com/stretchr/testify/assert"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
 )
 
 func TestReconcileRealm(t *testing.T) {
@@ -180,3 +185,34 @@ func deleteStore(t *testing.T, name string, existingStores string, expectedDelet
 	assert.Equal(t, expectedDeleteRootPool, deletedRootPool)
 	assert.Equal(t, true, deletedErasureCodeProfile)
 }
+
+func TestGetObjectBucketProvisioner(t *testing.T) {
+	k8s := fake.NewSimpleClientset()
+	operatorSettingConfigMapName := "rook-ceph-operator-config"
+	testNamespace := "test-namespace"
+	watchOperatorNamespace := map[string]string{"ROOK_OBC_WATCH_OPERATOR_NAMESPACE": "true"}
+	ignoreOperatorNamespace := map[string]string{"ROOK_OBC_WATCH_OPERATOR_NAMESPACE": "false"}
+	context := &clusterd.Context{Clientset: k8s}
+	os.Setenv(k8sutil.PodNamespaceEnvVar, testNamespace)
+
+	cm := &v1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      operatorSettingConfigMapName,
+			Namespace: testNamespace,
+		},
+		Data: watchOperatorNamespace,
+	}
+
+	_, err := k8s.CoreV1().ConfigMaps(testNamespace).Create(cm)
+	assert.NoError(t, err)
+
+	bktprovisioner := GetObjectBucketProvisioner(context, testNamespace)
+	assert.Equal(t, fmt.Sprintf("%s.%s", testNamespace, bucketProvisionerName), bktprovisioner)
+
+	cm.Data = ignoreOperatorNamespace
+	_, err = k8s.CoreV1().ConfigMaps(testNamespace).Update(cm)
+	assert.NoError(t, err)
+
+	bktprovisioner = GetObjectBucketProvisioner(context, testNamespace)
+	assert.Equal(t, bucketProvisionerName, bktprovisioner)
+}