controllers: disable mirroring when peer cluster is down

allow the mirroring to be disabled when the peer site is down

Signed-off-by: Rewant Soni <[email protected]>

controllers/mirroring/mirroring_controller.go

@@ -209,20 +209,13 @@ func (r *MirroringReconciler) reconcilePhases(clientMappingConfig *corev1.Config
 		return ctrl.Result{RequeueAfter: 3 * time.Second}, nil
 	}
 
-	ocsClient, err := providerClient.NewProviderClient(r.ctx, storageClusterPeer.Spec.ApiEndpoint, util.OcsClientTimeout)
-	if err != nil {
-		return ctrl.Result{}, fmt.Errorf("failed to create a new provider client: %v", err)
-	}
-	defer ocsClient.Close()
-
 	errorOccurred := false
 
 	if errored := r.reconcileRbdMirror(clientMappingConfig, shouldMirror); errored {
 		errorOccurred = true
 	}
 
 	if errored := r.reconcileBlockPoolMirroring(
-		ocsClient,
 		clientMappingConfig,
 		storageClusterPeer,
 		shouldMirror,
@@ -231,7 +224,6 @@ func (r *MirroringReconciler) reconcilePhases(clientMappingConfig *corev1.Config
 	}
 
 	if errored := r.reconcileRadosNamespaceMirroring(
-		ocsClient,
 		clientMappingConfig,
 		storageClusterPeer,
 		shouldMirror,
@@ -296,7 +288,6 @@ func (r *MirroringReconciler) reconcileRbdMirror(clientMappingConfig *corev1.Con
 }
 
 func (r *MirroringReconciler) reconcileBlockPoolMirroring(
-	ocsClient *providerClient.OCSProviderClient,
 	clientMappingConfig *corev1.ConfigMap,
 	storageClusterPeer *ocsv1.StorageClusterPeer,
 	shouldMirror bool,
@@ -322,6 +313,13 @@ func (r *MirroringReconciler) reconcileBlockPoolMirroring(
 	}
 
 	if len(blockPoolByName) > 0 {
+		ocsClient, err := r.newProviderClient(storageClusterPeer)
+		if err != nil {
+			errorOccurred = true
+			r.log.Error(err, "failed to create providerClient")
+			return errorOccurred
+		}
+		defer ocsClient.Close()
 		// fetch BlockPoolsInfo
 		response, err := ocsClient.GetBlockPoolsInfo(
 			r.ctx,
@@ -430,7 +428,6 @@ func (r *MirroringReconciler) reconcileBlockPoolMirroring(
 }
 
 func (r *MirroringReconciler) reconcileRadosNamespaceMirroring(
-	ocsClient *providerClient.OCSProviderClient,
 	clientMappingConfig *corev1.ConfigMap,
 	storageClusterPeer *ocsv1.StorageClusterPeer,
 	shouldMirror bool,
@@ -480,6 +477,13 @@ func (r *MirroringReconciler) reconcileRadosNamespaceMirroring(
 	}
 
 	if len(peerClientIDs) > 0 {
+		ocsClient, err := r.newProviderClient(storageClusterPeer)
+		if err != nil {
+			errorOccurred = true
+			r.log.Error(err, "failed to create providerClient")
+			return errorOccurred
+		}
+		defer ocsClient.Close()
 		response, err := ocsClient.GetStorageClientsInfo(
 			r.ctx,
 			storageClusterPeer.Status.PeerInfo.StorageClusterUid,
@@ -548,6 +552,14 @@ func (r *MirroringReconciler) reconcileRadosNamespaceMirroring(
 	return errorOccurred
 }
 
+func (r *MirroringReconciler) newProviderClient(storageClusterPeer *ocsv1.StorageClusterPeer) (*providerClient.OCSProviderClient, error) {
+	ocsClient, err := providerClient.NewProviderClient(r.ctx, storageClusterPeer.Spec.ApiEndpoint, util.OcsClientTimeout)
+	if err != nil {
+		return nil, err
+	}
+	return ocsClient, nil
+}
+
 func (r *MirroringReconciler) get(obj client.Object) error {
 	return r.Client.Get(r.ctx, client.ObjectKeyFromObject(obj), obj)
 }