add server implementation for revokeBlockPoolPeering rpc call

Signed-off-by: Rewant Soni <resoni@redhat.com>

deploy/ocs-operator/manifests/provider-role.yaml

@@ -18,6 +18,7 @@ rules:
     - get
     - create
     - update
+    - delete
 - apiGroups:
     - ceph.rook.io
   resources:
@@ -37,6 +38,7 @@ rules:
     - cephblockpools
   verbs:
     - get
+    - list
     - update
     - create
 - apiGroups:
@@ -81,3 +83,4 @@ rules:
   verbs:
     - get
     - create
+    - delete

rbac/provider-role.yaml

@@ -18,6 +18,7 @@ rules:
     - get
     - create
     - update
+    - delete
 - apiGroups:
     - ceph.rook.io
   resources:
@@ -37,6 +38,7 @@ rules:
     - cephblockpools
   verbs:
     - get
+    - list
     - update
     - create
 - apiGroups:
@@ -81,3 +83,4 @@ rules:
   verbs:
     - get
     - create
+    - delete

services/provider/server/cephblockpool.go

@@ -83,6 +83,55 @@ func (c *cephBlockPoolManager) SetBootstrapSecretRef(ctx context.Context, blockP
 	return nil
 }
 
+func (c *cephBlockPoolManager) UnSetBootstrapSecretRef(ctx context.Context, blockPoolName, secretName string, cephBlockPool *rookCephv1.CephBlockPool) error {
+
+	// remove the secret ref
+	_, err := ctrl.CreateOrUpdate(ctx, c.client, cephBlockPool, func() error {
+
+		if cephBlockPool.Spec.Mirroring.Peers == nil {
+			cephBlockPool.Spec.Mirroring.Peers = &rookCephv1.MirroringPeerSpec{SecretNames: []string{secretName}}
+		} else {
+			index := slices.IndexFunc(cephBlockPool.Spec.Mirroring.Peers.SecretNames, func(s string) bool {
+				return s == secretName
+			})
+			if index >= 0 {
+				cephBlockPool.Spec.Mirroring.Peers.SecretNames = append(
+					cephBlockPool.Spec.Mirroring.Peers.SecretNames[:index],
+					cephBlockPool.Spec.Mirroring.Peers.SecretNames[index+1:]...)
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		return fmt.Errorf("failed to unset bootstrap secret ref on CephBlockPool resource with name %q. %v", blockPoolName, err)
+	}
+
+	// delete secret
+	bootstrapSecret := &corev1.Secret{}
+	bootstrapSecret.Name = secretName
+	bootstrapSecret.Namespace = c.namespace
+	err = c.client.Delete(ctx, bootstrapSecret)
+	if err != nil {
+		return fmt.Errorf("failed to delete the bootstrap secret %q. %v", secretName, err)
+	}
+	return nil
+}
+
+func (c *cephBlockPoolManager) DisableBlockPoolMirroring(ctx context.Context, blockPoolName string, cephBlockPool *rookCephv1.CephBlockPool) error {
+
+	_, err := ctrl.CreateOrUpdate(ctx, c.client, cephBlockPool, func() error {
+		if cephBlockPool.Spec.Mirroring.Peers == nil || len(cephBlockPool.Spec.Mirroring.Peers.SecretNames) == 0 {
+			cephBlockPool.Spec.Mirroring.Enabled = false
+			cephBlockPool.Spec.Mirroring.Mode = ""
+		}
+		return nil
+	})
+	if err != nil {
+		return fmt.Errorf("failed to disable mirroring on CephBlockPool resource with name %q. %v", blockPoolName, err)
+	}
+	return nil
+}
+
 func (c *cephBlockPoolManager) GetBlockPoolByName(ctx context.Context, blockPoolName string) (*rookCephv1.CephBlockPool, error) {
 	blockPool := &rookCephv1.CephBlockPool{}
 	blockPool.Name = blockPoolName
@@ -96,3 +145,20 @@ func (c *cephBlockPoolManager) GetBlockPoolByName(ctx context.Context, blockPool
 	}
 	return blockPool, nil
 }
+
+// IsRBDMirrorRequired checks if we require RBDMirror to be deployed or not
+func (c *cephBlockPoolManager) IsRBDMirrorRequired(ctx context.Context) (bool, error) {
+	cephBlockPoolList := &rookCephv1.CephBlockPoolList{}
+	err := c.client.List(ctx, cephBlockPoolList, client.InNamespace(c.namespace))
+	if err != nil {
+		return true, err
+	}
+
+	// if we find a bootstrap secret in any of the blockPools, we require RBDMirror to be deployed
+	for _, cephBlockPool := range cephBlockPoolList.Items {
+		if cephBlockPool.Spec.Mirroring.Peers != nil && len(cephBlockPool.Spec.Mirroring.Peers.SecretNames) > 0 {
+			return true, nil
+		}
+	}
+	return false, nil
+}

services/provider/server/cephrbdmirror.go

@@ -5,6 +5,7 @@ import (
 
 	rookCephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -41,3 +42,13 @@ func (c *cephRBDMirrorManager) Create(ctx context.Context) error {
 	// if any other err/nil return it
 	return err
 }
+
+func (c *cephRBDMirrorManager) Delete(ctx context.Context) error {
+	cephRBDMirrorObj := &rookCephv1.CephRBDMirror{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      rBDMirrorName,
+			Namespace: c.namespace,
+		},
+	}
+	return c.client.Delete(ctx, cephRBDMirrorObj)
+}

services/provider/server/server.go

@@ -736,3 +736,45 @@ func (s *OCSProviderServer) PeerBlockPool(ctx context.Context, req *pb.PeerBlock
 	}
 	return &pb.PeerBlockPoolResponse{}, nil
 }
+
+// RevokeBlockPoolPeering RPC call to delete the bootstrap secret to stop peering
+func (s *OCSProviderServer) RevokeBlockPoolPeering(ctx context.Context, req *pb.RevokeBlockPoolPeeringRequest) (*pb.RevokeBlockPoolPeeringResponse, error) {
+
+	klog.Infof("RevokeBlockPoolPeering request received for CephBlockPool %s and bootstrap secret %s", req.Pool, req.SecretName)
+
+	cephBlockPool, err := s.cephBlockPoolManager.GetBlockPoolByName(ctx, string(req.Pool))
+	if err != nil {
+		return nil, status.Errorf(codes.NotFound, "Failed to find CephBlockPool resource %s: %v", req.Pool, err)
+	}
+
+	// delete secret and unset ref on the blockPool
+	err = s.cephBlockPoolManager.UnSetBootstrapSecretRef(ctx, string(req.Pool), req.SecretName, cephBlockPool)
+	// there might be a case where the bootstrap secret was deleted but request failed after this and there was a retry,
+	// if error is IsNotFound, that means it is safe to proceed as we have deleted the bootstrap secret
+	if err != nil && !kerrors.IsNotFound(err) {
+		return nil, status.Errorf(codes.Internal, "Failed to unset bootstrap secret ref for CephBlockPool resource %s: %v", req.Pool, err)
+	}
+
+	// disable mirroring on blockPool in the req
+	err = s.cephBlockPoolManager.DisableBlockPoolMirroring(ctx, string(req.Pool), cephBlockPool)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "Failed to disable mirroring for CephBlockPool resource %s: %v", req.Pool, err)
+	}
+
+	isRBDMirrorRequired, err := s.cephBlockPoolManager.IsRBDMirrorRequired(ctx)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "Failed to get if rbd mirror is required: %v,", err)
+	}
+
+	if !isRBDMirrorRequired {
+		klog.Infof("No bootstrap secret found for any block pools, removing the rbd mirror instance")
+		err := s.cephRBDMirrorManager.Delete(ctx)
+		// there might be a case where the RBDMirror was deleted but request failed after this and there was a retry,
+		// if error is IsNotFound, that means it is safe to proceed as we have deleted the RBDMirror instance
+		if err != nil && !kerrors.IsNotFound(err) {
+			klog.Errorf("Failed to delete CephRBDMirror instance: %v", err)
+			return nil, status.Errorf(codes.Internal, "Failed to delete CephRBDMirror instance: %v", err)
+		}
+	}
+	return &pb.RevokeBlockPoolPeeringResponse{}, nil
+}