Merge pull request #11975 from thotz/bucket-notification-use-obc-from-rook

object: check obc provisioner for bucket notification

pkg/operator/ceph/object/notification/controller_test.go

@@ -57,6 +57,7 @@ var (
 	multipleCreateBucketName = "multi-create"
 	multipleDeleteBucketName = "multi-delete"
 	multipleBothBucketName   = "multi-both"
+	otherClusterBucketName   = "other-cluster"
 	startEvent               = string(cephv1.ReconcileStarted)
 	finishedEvent            = string(cephv1.ReconcileSucceeded)
 	failedEvent              = string(cephv1.ReconcileFailed)

pkg/operator/ceph/object/notification/obc_label_controller.go

@@ -44,7 +44,9 @@ import (
 )
 
 const (
-	notificationLabelPrefix = "bucket-notification-"
+	notificationLabelPrefix   = "bucket-notification-"
+	bucketProvisionerLabelKey = "bucket-provisioner"
+	bucketProvisionerLabelVal = "ceph.rook.io-bucket"
 )
 
 // ReconcileOBCLabels reconciles a ObjectBucketClaim labels
@@ -119,11 +121,18 @@ func (r *ReconcileOBCLabels) reconcile(request reconcile.Request) (reconcile.Res
 	if err := r.client.Get(r.opManagerContext, bucketName, &ob); err != nil {
 		return reconcile.Result{}, errors.Wrapf(err, "failed to retrieve ObjectBucket %q", bucketName)
 	}
+
+	// validate if the bucket is provisioned by the ceph provisioner
+	if !strings.Contains(ob.Labels[bucketProvisionerLabelKey], bucketProvisionerLabelVal) {
+		logger.Debugf("ObjectBucket %q was not provisioned by the ceph object store provisioner and tagged with provisioner %q. ignoring",
+			bucketName, ob.Labels[bucketProvisionerLabelKey])
+		return reconcile.Result{}, nil
+	}
+	// validate object store name
 	objectStoreName, err := getCephObjectStoreName(ob)
 	if err != nil {
 		return reconcile.Result{}, errors.Wrapf(err, "failed to get object store from ObjectBucket %q", bucketName)
 	}
-
 	// Populate clusterInfo during each reconcile
 	clusterInfo, clusterSpec, err := getReadyCluster(r.client, r.opManagerContext, *r.context, objectStoreName.Namespace)
 	if err != nil {

pkg/operator/ceph/object/notification/obc_label_controller_test.go

@@ -30,11 +30,16 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 )
 
-func createOBResources(name string) (*bktv1alpha1.ObjectBucketClaim, *bktv1alpha1.ObjectBucket) {
+const (
+	testProvisionerOtherCluster = "other-provisioner.io-bucket"
+)
+
+func createOBResources(name, provisioner string) (*bktv1alpha1.ObjectBucketClaim, *bktv1alpha1.ObjectBucket) {
 	return &bktv1alpha1.ObjectBucketClaim{
 			ObjectMeta: metav1.ObjectMeta{
 				Name:      name,
 				Namespace: testNamespace,
+				Labels:    map[string]string{bucketProvisionerLabelKey: provisioner},
 			},
 			TypeMeta: metav1.TypeMeta{
 				Kind: "ObjectBucketClaim",
@@ -50,6 +55,7 @@ func createOBResources(name string) (*bktv1alpha1.ObjectBucketClaim, *bktv1alpha
 			ObjectMeta: metav1.ObjectMeta{
 				Name:      name,
 				Namespace: testNamespace,
+				Labels:    map[string]string{bucketProvisionerLabelKey: provisioner},
 			},
 			TypeMeta: metav1.TypeMeta{
 				Kind: "ObjectBucket",
@@ -130,7 +136,7 @@ func TestCephBucketNotificationOBCLabelController(t *testing.T) {
 		},
 	}
 
-	obc, ob := createOBResources(testBucketName)
+	obc, ob := createOBResources(testBucketName, bucketProvisionerLabelVal)
 	obc.Labels = setNotificationLabels([]string{testNotificationName})
 
 	t.Run("provision OBC with notification label with no ob", func(t *testing.T) {
@@ -226,7 +232,7 @@ func TestCephBucketNotificationOBCLabelController(t *testing.T) {
 
 	t.Run("reconcile with already existing label for the obc", func(t *testing.T) {
 		resetValues()
-		noChangeOBC, noChangeOB := createOBResources(noChangeBucketName)
+		noChangeOBC, noChangeOB := createOBResources(noChangeBucketName, bucketProvisionerLabelVal)
 		noChangeOBC.Labels = setNotificationLabels([]string{testNotificationName})
 		noChangeOBC.Spec.ObjectBucketName = noChangeBucketName
 		noChangeOBC.Status.Phase = bktv1alpha1.ObjectBucketClaimStatusPhaseBound
@@ -249,7 +255,7 @@ func TestCephBucketNotificationOBCLabelController(t *testing.T) {
 
 	t.Run("delete notification from the obc", func(t *testing.T) {
 		resetValues()
-		deleteOBC, deleteOB := createOBResources(deleteBucketName)
+		deleteOBC, deleteOB := createOBResources(deleteBucketName, bucketProvisionerLabelVal)
 		deleteOBC.Spec.GenerateBucketName = deleteBucketName
 		deleteOBC.Spec.ObjectBucketName = deleteBucketName
 		deleteOBC.Status.Phase = bktv1alpha1.ObjectBucketClaimStatusPhaseBound
@@ -271,7 +277,7 @@ func TestCephBucketNotificationOBCLabelController(t *testing.T) {
 
 	t.Run("provision OBC with multiple notification labels", func(t *testing.T) {
 		resetValues()
-		multipleCreateOBC, multipleCreateOB := createOBResources(multipleCreateBucketName)
+		multipleCreateOBC, multipleCreateOB := createOBResources(multipleCreateBucketName, bucketProvisionerLabelVal)
 		multipleCreateOBC.Labels = setNotificationLabels([]string{testNotificationName + "-1", testNotificationName + "-2"})
 		multipleCreateOBC.Spec.ObjectBucketName = multipleCreateBucketName
 		multipleCreateOBC.Status.Phase = bktv1alpha1.ObjectBucketClaimStatusPhaseBound
@@ -298,7 +304,7 @@ func TestCephBucketNotificationOBCLabelController(t *testing.T) {
 
 	t.Run("delete multiple notifications from the obc", func(t *testing.T) {
 		resetValues()
-		multipleDeleteOBC, multipleDeleteOB := createOBResources(multipleDeleteBucketName)
+		multipleDeleteOBC, multipleDeleteOB := createOBResources(multipleDeleteBucketName, bucketProvisionerLabelVal)
 		multipleDeleteOBC.Spec.GenerateBucketName = multipleDeleteBucketName
 		multipleDeleteOBC.Spec.ObjectBucketName = multipleDeleteBucketName
 		multipleDeleteOBC.Status.Phase = bktv1alpha1.ObjectBucketClaimStatusPhaseBound
@@ -319,7 +325,7 @@ func TestCephBucketNotificationOBCLabelController(t *testing.T) {
 	})
 	t.Run("provision OBC with multiple delete and create of notifications", func(t *testing.T) {
 		resetValues()
-		multipleBothOBC, multipleBothOB := createOBResources(multipleBothBucketName)
+		multipleBothOBC, multipleBothOB := createOBResources(multipleBothBucketName, bucketProvisionerLabelVal)
 		multipleBothOBC.Labels = setNotificationLabels([]string{testNotificationName + "-1", testNotificationName + "-2"})
 		multipleBothOBC.Spec.ObjectBucketName = multipleBothBucketName
 		multipleBothOBC.Status.Phase = bktv1alpha1.ObjectBucketClaimStatusPhaseBound
@@ -346,4 +352,25 @@ func TestCephBucketNotificationOBCLabelController(t *testing.T) {
 			[]string{multipleDeleteBucketName + testNotificationName + "-1", multipleDeleteBucketName + testNotificationName + "-2"})
 		verifyEvents(t, []string{startEvent, finishedEvent})
 	})
+
+	t.Run("OBC does not belong to the ceph cluster", func(t *testing.T) {
+		resetValues()
+		otherClusterOBC, otherClusterOB := createOBResources(otherClusterBucketName, testProvisionerOtherCluster)
+		otherClusterOBC.Labels = setNotificationLabels([]string{testNotificationName})
+		otherClusterOBC.Spec.ObjectBucketName = otherClusterBucketName
+		otherClusterOBC.Status.Phase = bktv1alpha1.ObjectBucketClaimStatusPhaseBound
+		objects := []runtime.Object{
+			cephCluster,
+			otherClusterOBC,
+			otherClusterOB,
+		}
+
+		res, err := testOBCLabelReconciler(objects, otherClusterBucketName)
+		assert.NoError(t, err)
+		assert.False(t, res.Requeue)
+		assert.False(t, getWasInvoked)
+		assert.Equal(t, 0, len(createdNotifications))
+		assert.Equal(t, 0, len(deletedNotifications))
+		verifyEvents(t, []string{startEvent, finishedEvent})
+	})
 }