add rpc call for PeerBlockPool

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

@@ -7,10 +7,23 @@ rules:
   - ""
   resources:
   - configmaps
-  - secrets
   - services
   verbs:
   - get
+- apiGroups:
+    - ""
+  resources:
+    - secrets
+  verbs:
+    - get
+    - create
+    - update
+- apiGroups:
+    - ceph.rook.io
+  resources:
+    - cephblockpools/finalizers
+  verbs:
+    - update
 - apiGroups:
   - ceph.rook.io
   resources:
@@ -19,6 +32,13 @@ rules:
   verbs:
   - get
   - list
+- apiGroups:
+    - ceph.rook.io
+  resources:
+    - cephblockpools
+  verbs:
+    - get
+    - update
 - apiGroups:
   - ocs.openshift.io
   resources:
@@ -61,3 +81,10 @@ rules:
   verbs:
   - get
   - list
+- apiGroups:
+    - ceph.rook.io
+  resources:
+    - cephrbdmirrors
+  verbs:
+    - get
+    - create

rbac/provider-role.yaml

@@ -7,10 +7,23 @@ rules:
   - ""
   resources:
   - configmaps
-  - secrets
   - services
   verbs:
   - get
+- apiGroups:
+    - ""
+  resources:
+    - secrets
+  verbs:
+    - get
+    - create
+    - update
+- apiGroups:
+    - ceph.rook.io
+  resources:
+    - cephblockpools/finalizers
+  verbs:
+    - update
 - apiGroups:
   - ceph.rook.io
   resources:
@@ -19,6 +32,13 @@ rules:
   verbs:
   - get
   - list
+- apiGroups:
+    - ceph.rook.io
+  resources:
+    - cephblockpools
+  verbs:
+    - get
+    - update
 - apiGroups:
   - ocs.openshift.io
   resources:
@@ -61,3 +81,10 @@ rules:
   verbs:
   - get
   - list
+- apiGroups:
+    - ceph.rook.io
+  resources:
+    - cephrbdmirrors
+  verbs:
+    - get
+    - create

services/provider/client/client.go

@@ -207,3 +207,20 @@ func (cc *OCSProviderClient) ReportStatus(ctx context.Context, consumerUUID stri
 
 	return cc.Client.ReportStatus(apiCtx, req)
 }
+
+func (cc *OCSProviderClient) PeerBlockPool(ctx context.Context, secretName string, pool, token []byte) (*pb.PeerBlockPoolResponse, error) {
+	if cc.Client == nil || cc.clientConn == nil {
+		return nil, fmt.Errorf("OCS client is closed")
+	}
+
+	req := &pb.PeerBlockPoolRequest{
+		SecretName: secretName,
+		Pool:       pool,
+		Token:      token,
+	}
+
+	apiCtx, cancel := context.WithTimeout(ctx, cc.timeout)
+	defer cancel()
+
+	return cc.Client.PeerBlockPool(apiCtx, req)
+}

services/provider/proto/provider.proto

@@ -38,6 +38,9 @@ service OCSProvider {
 
   rpc ReportStatus(ReportStatusRequest)
   returns (ReportStatusResponse){}
+  // PeerBlockPool RPC call to send the bootstrap secret for the pool
+  rpc PeerBlockPool(PeerBlockPoolRequest)
+  returns (PeerBlockPoolResponse){}
 }
 
 // OnboardConsumerRequest holds the required information to validate the consumer and create StorageConsumer
@@ -172,3 +175,14 @@ message ReportStatusResponse{
     // Contains subscription channel of provider operator for client operator to match
     string desiredClientOperatorChannel = 1;
 }
+
+message PeerBlockPoolRequest{
+  // secretName is the name of the bootstrap secret
+  string secretName = 1;
+  // pool base64 encoded name of the block pool
+  bytes pool = 2;
+  // token is secret information from bootstrap secret
+  bytes token = 3;
+}
+
+message PeerBlockPoolResponse{}

services/provider/server/cephblockpool.go

@@ -0,0 +1,81 @@
+package server
+
+import (
+	"context"
+	"fmt"
+	"slices"
+
+	rookCephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	corev1 "k8s.io/api/core/v1"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type cephBlockPoolManager struct {
+	client    client.Client
+	namespace string
+}
+
+func newCephBlockPoolManager(cl client.Client, namespace string) (*cephBlockPoolManager, error) {
+	return &cephBlockPoolManager{
+		client:    cl,
+		namespace: namespace,
+	}, nil
+}
+
+func (c *cephBlockPoolManager) EnableBlockPoolMirroring(ctx context.Context, cephBlockPool *rookCephv1.CephBlockPool) error {
+
+	cephBlockPool.Spec.Mirroring.Enabled = true
+	cephBlockPool.Spec.Mirroring.Mode = "image"
+
+	err := c.client.Update(ctx, cephBlockPool)
+	if err != nil {
+		return fmt.Errorf("failed to enable mirroring on CephBlockPool resource with name %q: %v", cephBlockPool.Name, err)
+	}
+
+	return nil
+
+}
+
+func (c *cephBlockPoolManager) SetBootstrapSecretRef(ctx context.Context, cephBlockPool *rookCephv1.CephBlockPool, secretName string, secretData map[string][]byte) error {
+
+	// create the secret
+	bootstrapSecret := &corev1.Secret{}
+	bootstrapSecret.Name = secretName
+	bootstrapSecret.Namespace = c.namespace
+
+	_, err := ctrl.CreateOrUpdate(ctx, c.client, bootstrapSecret, func() error {
+		bootstrapSecret.Data = secretData
+		return ctrl.SetControllerReference(cephBlockPool, bootstrapSecret, c.client.Scheme())
+	})
+	if err != nil {
+		return fmt.Errorf("failed to create/update the bootstrap secret %q: %v", secretName, err)
+	}
+
+	// set the secret ref
+	if cephBlockPool.Spec.Mirroring.Peers == nil {
+		cephBlockPool.Spec.Mirroring.Peers = &rookCephv1.MirroringPeerSpec{SecretNames: []string{secretName}}
+	} else {
+		if !slices.Contains(cephBlockPool.Spec.Mirroring.Peers.SecretNames, secretName) {
+			cephBlockPool.Spec.Mirroring.Peers.SecretNames = append(cephBlockPool.Spec.Mirroring.Peers.SecretNames, secretName)
+		}
+	}
+
+	err = c.client.Update(ctx, cephBlockPool)
+	if err != nil {
+		return fmt.Errorf("failed to set bootstrap secret ref on CephBlockPool resource with name %q: %v", cephBlockPool.Name, err)
+	}
+
+	return nil
+}
+
+func (c *cephBlockPoolManager) GetBlockPoolByName(ctx context.Context, blockPoolName string) (*rookCephv1.CephBlockPool, error) {
+	blockPool := &rookCephv1.CephBlockPool{}
+	blockPool.Name = blockPoolName
+	blockPool.Namespace = c.namespace
+	err := c.client.Get(ctx, client.ObjectKeyFromObject(blockPool), blockPool)
+	if err != nil {
+		return nil, err
+	}
+	return blockPool, nil
+}

services/provider/server/cephrbdmirror.go

@@ -0,0 +1,43 @@
+package server
+
+import (
+	"context"
+
+	rookCephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+const rBDMirrorName = "rbd-mirror"
+
+type cephRBDMirrorManager struct {
+	client    client.Client
+	namespace string
+}
+
+func newCephRBDMirrorManager(cl client.Client, namespace string) (*cephRBDMirrorManager, error) {
+	return &cephRBDMirrorManager{
+		client:    cl,
+		namespace: namespace,
+	}, nil
+}
+
+func (c *cephRBDMirrorManager) Create(ctx context.Context) error {
+
+	cephRBDMirror := &rookCephv1.CephRBDMirror{}
+	cephRBDMirror.Name = rBDMirrorName
+	cephRBDMirror.Namespace = c.namespace
+	err := c.client.Get(ctx, client.ObjectKeyFromObject(cephRBDMirror), cephRBDMirror)
+
+	// create if not found
+	if err != nil && kerrors.IsNotFound(err) {
+		cephRBDMirror.Spec = rookCephv1.RBDMirroringSpec{Count: 1}
+		err = c.client.Create(ctx, cephRBDMirror)
+		if err != nil {
+			return err
+		}
+	}
+
+	// if any other err/nil return it
+	return err
+}