diff --git a/pkg/operator/ceph/object/bucket/util.go b/pkg/operator/ceph/object/bucket/util.go index 21f5b1297..37f4634a9 100644 --- a/pkg/operator/ceph/object/bucket/util.go +++ b/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) diff --git a/pkg/operator/ceph/object/controller.go b/pkg/operator/ceph/object/controller.go index ee190bc3d..acacb7227 100644 --- a/pkg/operator/ceph/object/controller.go +++ b/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 +} diff --git a/pkg/operator/ceph/object/objectstore.go b/pkg/operator/ceph/object/objectstore.go index c49319c92..8725c4918 100644 --- a/pkg/operator/ceph/object/objectstore.go +++ b/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 ., 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 +} diff --git a/pkg/operator/ceph/object/objectstore_test.go b/pkg/operator/ceph/object/objectstore_test.go index 9ca160eb1..9bf7969ee 100644 --- a/pkg/operator/ceph/object/objectstore_test.go +++ b/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) +}