implement server side rpc call for peering

Signed-off-by: Rewant Soni <[email protected]>
red-hat-storage · Nov 4, 2024 · 3ca9f81 · 3ca9f81
1 parent 2d42e55
commit 3ca9f81
Show file tree

Hide file tree

Showing 3 changed files with 153 additions and 22 deletions.
diff --git a/rbac/provider-role.yaml b/rbac/provider-role.yaml
@@ -75,3 +75,12 @@ rules:
     verbs:
       - get
       - list
+  - apiGroups:
+      - ocs.openshift.io
+    resources:
+      - storageclusterpeers
+      - storageclusterpeers/status
+    verbs:
+      - get
+      - list
+      - update
diff --git a/services/provider/server/server.go b/services/provider/server/server.go
@@ -66,10 +66,11 @@ const (
 
 type OCSProviderServer struct {
 	pb.UnimplementedOCSProviderServer
-	client                client.Client
-	consumerManager       *ocsConsumerManager
-	storageRequestManager *storageRequestManager
-	namespace             string
+	client                    client.Client
+	consumerManager           *ocsConsumerManager
+	storageRequestManager     *storageRequestManager
+	storageClusterPeerManager *storageClusterPeerManager
+	namespace                 string
 }
 
 func NewOCSProviderServer(ctx context.Context, namespace string) (*OCSProviderServer, error) {
@@ -93,11 +94,17 @@ func NewOCSProviderServer(ctx context.Context, namespace string) (*OCSProviderSe
 		return nil, fmt.Errorf("failed to create new StorageRequest instance. %v", err)
 	}
 
+	storageClusterPeerManager, err := newStorageClusterPeerManager(client, namespace)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create new StorageClusterPeer instance. %v", err)
+	}
+
 	return &OCSProviderServer{
-		client:                client,
-		consumerManager:       consumerManager,
-		storageRequestManager: storageRequestManager,
-		namespace:             namespace,
+		client:                    client,
+		consumerManager:           consumerManager,
+		storageRequestManager:     storageRequestManager,
+		storageClusterPeerManager: storageClusterPeerManager,
+		namespace:                 namespace,
 	}, nil
 }
 
@@ -535,14 +542,9 @@ func getSubVolumeGroupClusterID(subVolumeGroup *rookCephv1.CephFilesystemSubVolu
 }
 
 func decodeAndValidateTicket(ticket string, pubKey *rsa.PublicKey) (*services.OnboardingTicket, error) {
-	ticketArr := strings.Split(string(ticket), ".")
-	if len(ticketArr) != 2 {
-		return nil, fmt.Errorf("invalid ticket")
-	}
-
-	message, err := base64.StdEncoding.DecodeString(ticketArr[0])
+	message, signature, err := decodeTicket(ticket)
 	if err != nil {
-		return nil, fmt.Errorf("failed to decode onboarding ticket: %v", err)
+		return nil, err
 	}
 
 	var ticketData services.OnboardingTicket
@@ -564,11 +566,6 @@ func decodeAndValidateTicket(ticket string, pubKey *rsa.PublicKey) (*services.On
 		return nil, fmt.Errorf("invalid onboarding ticket subject role")
 	}
 
-	signature, err := base64.StdEncoding.DecodeString(ticketArr[1])
-	if err != nil {
-		return nil, fmt.Errorf("failed to decode onboarding ticket %s signature: %v", ticketData.ID, err)
-	}
-
 	hash := sha256.Sum256(message)
 	err = rsa.VerifyPKCS1v15(pubKey, crypto.SHA256, hash[:], signature)
 	if err != nil {
@@ -584,6 +581,25 @@ func decodeAndValidateTicket(ticket string, pubKey *rsa.PublicKey) (*services.On
 	return &ticketData, nil
 }
 
+func decodeTicket(ticket string) ([]byte, []byte, error) {
+	ticketArr := strings.Split(string(ticket), ".")
+	if len(ticketArr) != 2 {
+		return nil, nil, fmt.Errorf("invalid ticket")
+	}
+
+	message, err := base64.StdEncoding.DecodeString(ticketArr[0])
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to decode onboarding ticket: %v", err)
+	}
+
+	signature, err := base64.StdEncoding.DecodeString(ticketArr[1])
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to decode onboarding ticketsignature: %v", err)
+	}
+
+	return message, signature, nil
+}
+
 // FulfillStorageClaim RPC call to create the StorageClaim CR on
 // provider cluster.
 func (s *OCSProviderServer) FulfillStorageClaim(ctx context.Context, req *pb.FulfillStorageClaimRequest) (*pb.FulfillStorageClaimResponse, error) {
@@ -925,6 +941,51 @@ func extractMonitorIps(data string) ([]string, error) {
 	return ips, nil
 }
 
-func (s *OCSProviderServer) PeerStorageCluster(_ context.Context, _ *pb.PeerStorageClusterRequest) (*pb.PeerStorageClusterResponse, error) {
-	return &pb.PeerStorageClusterResponse{}, nil
+func (s *OCSProviderServer) PeerStorageCluster(ctx context.Context, req *pb.PeerStorageClusterRequest) (*pb.PeerStorageClusterResponse, error) {
+
+	pubKey, err := s.getOnboardingValidationKey(ctx)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to get public key to validate onboarding ticket for StorageCluster %q. %v", req.StorageClusterUID, err)
+	}
+
+	onboardingToken, err := decodeAndValidateTicket(req.OnboardingToken, pubKey)
+	if err != nil {
+		klog.Errorf("failed to validate onboarding ticket for StorageCluster  %q. %v", req.StorageClusterUID, err)
+		return nil, status.Errorf(codes.InvalidArgument, "onboarding ticket is not valid. %v", err)
+	}
+
+	if onboardingToken.SubjectRole != services.PeerRole {
+		err := fmt.Errorf("unsupported ticket role for StorageCluster %q, found %s, expected %s", req.StorageClusterUID, onboardingToken.SubjectRole, services.PeerRole)
+		klog.Error(err)
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	storageCluster, err := util.GetStorageClusterInNamespace(ctx, s.client, s.namespace)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to get storageCluster. %v", err)
+	}
+
+	if storageCluster.UID != onboardingToken.StorageCluster {
+		return nil, status.Errorf(codes.InvalidArgument, "onboarding ticket storageCluster not match existing storageCluster.")
+	}
+
+	klog.Infof("Found StorageCluster %s for PeerStorageCluster", storageCluster.Name)
+
+	storageClusterPeer, err := s.storageClusterPeerManager.FindStorageClusterPeerWithStorageClusterID(ctx, types.UID(req.StorageClusterUID))
+	if err != nil {
+		klog.Error(err)
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	klog.Infof("Found StorageClusterPeer %s for PeerStorageCluster", storageClusterPeer.Name)
+
+	//TODO: Should we store the onboarding token in storageClusterPeer?
+
+	err = s.storageClusterPeerManager.UpdateStorageClusterPeerStatus(ctx, storageClusterPeer, req.StorageClusterUID)
+	if err != nil {
+		klog.Error(err)
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	return &pb.PeerStorageClusterResponse{StorageClusterUID: string(storageCluster.UID)}, nil
 }
diff --git a/services/provider/server/storageclusterpeer.go b/services/provider/server/storageclusterpeer.go
@@ -0,0 +1,61 @@
+package server
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/red-hat-storage/ocs-operator/v4/services"
+	"k8s.io/apimachinery/pkg/types"
+
+	ocsv1 "github.com/red-hat-storage/ocs-operator/api/v4/v1"
+	"k8s.io/klog/v2"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type storageClusterPeerManager struct {
+	client    client.Client
+	namespace string
+}
+
+func newStorageClusterPeerManager(cl client.Client, namespace string) (*storageClusterPeerManager, error) {
+	return &storageClusterPeerManager{
+		client:    cl,
+		namespace: namespace,
+	}, nil
+}
+
+func (s *storageClusterPeerManager) FindStorageClusterPeerWithStorageClusterID(ctx context.Context, peerStorageClusterUID types.UID) (*ocsv1.StorageClusterPeer, error) {
+	storageClusterPeerList := &ocsv1.StorageClusterPeerList{}
+	err := s.client.List(ctx, storageClusterPeerList, client.InNamespace(s.namespace))
+	if err != nil {
+		return nil, err
+	}
+	for i := range storageClusterPeerList.Items {
+		message, _, err := decodeTicket(storageClusterPeerList.Items[i].Spec.OnboardingToken)
+		if err != nil {
+			return nil, err
+		}
+
+		var ticketData services.OnboardingTicket
+		err = json.Unmarshal(message, &ticketData)
+		if err != nil {
+			return nil, fmt.Errorf("failed to unmarshal onboarding ticket message. %v", err)
+		}
+
+		if ticketData.StorageCluster == peerStorageClusterUID {
+			return &storageClusterPeerList.Items[i], nil
+		}
+	}
+	return nil, fmt.Errorf("StorageClusterPeer linked to StorageCluster with uid %q not found", peerStorageClusterUID)
+}
+
+func (s *storageClusterPeerManager) UpdateStorageClusterPeerStatus(ctx context.Context, storageClusterPeer *ocsv1.StorageClusterPeer, storageClusterUID string) error {
+
+	storageClusterPeer.Status.PeerInfo.StorageClusterUid = storageClusterUID
+
+	if err := s.client.Status().Update(ctx, storageClusterPeer); err != nil {
+		return fmt.Errorf("failed to patch Status for StorageClusterPeer %v: %v", storageClusterPeer.Name, err)
+	}
+	klog.Infof("successfully updated Status for StorageConsumer %v", storageClusterPeer.Name)
+	return nil
+}