Add safety latches to create/ensure VRG manifest work functions

The function to createVRGManifestWork will create a new VRG when a
ManifestWork for it does not exist. Ensure that the MW is not present
before creating one, and return an Exists error for callers to process
if one exists.

The function ensureVRGManifestWork is to ensure ManifestWork for VRG
exists and update its mutable fields appropriately. Added a latch to
ensure that the MW for VRG that needs to be ensured is present, and that
it is reported as delivered to the ManagedCluster (using a check to see
if a view is present).

The changes to ensureVRGManifestWork also includes NOT creating a VRG
from a view, as that should be ensured by the adoptOrphanVRG function
at the start of reconciling a DRPC.

The caller startDeploying, which calls createVRGManifestWork, now checks
for an exists error and proceeds to update the Placement. This is done as
any errors in updating placement would send the reconciler back to creating
the VRG, which as it exists would keep failing. Other callers ensure that
the VRG exists and if so update it, rather than calling create.

Signed-off-by: Shyamsundar Ranganathan <[email protected]>

internal/controller/drplacementcontrol.go

@@ -15,6 +15,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	clrapiv1beta1 "open-cluster-management.io/api/cluster/v1beta1"
 
 	rmn "github.com/ramendr/ramen/api/v1alpha1"
@@ -297,7 +298,7 @@ func (d *DRPCInstance) startDeploying(homeCluster, homeClusterNamespace string)
 	d.setProgression(rmn.ProgressionCreatingMW)
 	// Create VRG first, to leverage user PlacementRule decision to skip placement and move to cleanup
 	err := d.createVRGManifestWork(homeCluster, rmn.Primary)
-	if err != nil {
+	if err != nil && !errors.IsAlreadyExists(err) {
 		return false, err
 	}
 
@@ -1512,28 +1513,49 @@ func (d *DRPCInstance) updatePreferredDecision() {
 	}
 }
 
+// createVRGManifestWork is called to create a new VRG ManifestWork on homeCluster
 func (d *DRPCInstance) createVRGManifestWork(homeCluster string, repState rmn.ReplicationState) error {
-	// TODO: check if VRG MW here as a less expensive way to validate if Namespace exists
 	err := d.ensureNamespaceManifestWork(homeCluster)
 	if err != nil {
 		return fmt.Errorf("createVRGManifestWork couldn't ensure namespace '%s' on cluster %s exists",
 			d.vrgNamespace, homeCluster)
 	}
 
+	// Safety latch to ensure VRG MW is not present
+	vrg, err := d.getVRGFromManifestWork(homeCluster)
+	if (err != nil && !errors.IsNotFound(err)) || vrg != nil {
+		if err != nil {
+			return fmt.Errorf("error (%w) determining ManifestWork for VolumeReplicationGroup resource "+
+				"exists on cluster %s", err, homeCluster)
+		}
+
+		if vrg.Spec.ReplicationState != repState {
+			return fmt.Errorf("ManifestWork for VolumeReplicationGroup resource "+
+				"exists with mismatching state (%s) on cluster %s",
+				vrg.Spec.ReplicationState, homeCluster)
+		}
+
+		return fmt.Errorf("VolumeReplicationGroup ManifestWork for cluster %s in state %s exists (%w)",
+			homeCluster, string(vrg.Spec.ReplicationState), errors.NewAlreadyExists(
+				schema.GroupResource{
+					Group:    rmn.GroupVersion.Group,
+					Resource: "VolumeReplicationGroup",
+				}, vrg.Name))
+	}
+
 	// create VRG ManifestWork
 	d.log.Info("Creating VRG ManifestWork", "ReplicationState", repState,
 		"Last State:", d.getLastDRState(), "cluster", homeCluster)
 
-	vrg := d.newVRG(homeCluster, repState)
-
+	newVRG := d.newVRG(homeCluster, repState)
 	annotations := make(map[string]string)
 
 	annotations[DRPCNameAnnotation] = d.instance.Name
 	annotations[DRPCNamespaceAnnotation] = d.instance.Namespace
 
 	if _, err := d.mwu.CreateOrUpdateVRGManifestWork(
 		d.instance.Name, d.vrgNamespace,
-		homeCluster, vrg, annotations); err != nil {
+		homeCluster, newVRG, annotations); err != nil {
 		d.log.Error(err, "failed to create or update VolumeReplicationGroup manifest")
 
 		return fmt.Errorf("failed to create or update VolumeReplicationGroup manifest in namespace %s (%w)", homeCluster, err)
@@ -1543,30 +1565,37 @@ func (d *DRPCInstance) createVRGManifestWork(homeCluster string, repState rmn.Re
 }
 
 // ensureVRGManifestWork ensures that the VRG ManifestWork exists and matches the current VRG state.
-// TODO: This may be safe only when the VRG is primary - check if callers use this correctly.
 func (d *DRPCInstance) ensureVRGManifestWork(homeCluster string) error {
 	d.log.Info("Ensure VRG ManifestWork",
 		"Last State:", d.getLastDRState(), "cluster", homeCluster)
 
 	mw, mwErr := d.mwu.FindManifestWorkByType(rmnutil.MWTypeVRG, homeCluster)
 	if mwErr != nil {
 		if errors.IsNotFound(mwErr) {
-			cachedVrg := d.vrgs[homeCluster]
-			if cachedVrg == nil {
-				return fmt.Errorf("failed to get vrg from cluster %s", homeCluster)
-			}
-
-			return d.createVRGManifestWork(homeCluster, cachedVrg.Spec.ReplicationState)
+			return fmt.Errorf("failed to find ManifestWork for VolumeReplicationGroup from cluster %s", homeCluster)
 		}
 
-		return fmt.Errorf("ensure VRG ManifestWork failed (%w)", mwErr)
+		return fmt.Errorf("error (%w) in finding ManifestWork for VolumeReplicationGroup from cluster %s",
+			mwErr, homeCluster)
 	}
 
 	vrg, err := rmnutil.ExtractVRGFromManifestWork(mw)
 	if err != nil {
 		return fmt.Errorf("error extracting VRG from ManifestWork for cluster %s. Error: %w", homeCluster, err)
 	}
 
+	// Safety latch to ensure VRG to update is Primary
+	if vrg.Spec.ReplicationState != rmn.Primary {
+		return fmt.Errorf("invalid update for VolumeReplicationGroup in %s spec.replicationState on cluster %s",
+			vrg.Spec.ReplicationState, homeCluster)
+	}
+
+	// Safety latch to ensure a view exists for the existing VRG ManifestWork
+	if d.vrgs[homeCluster] == nil {
+		return fmt.Errorf("missing VolumeReplicationGroup view for cluster %s, while attempting to update an instance",
+			homeCluster)
+	}
+
 	d.updateVRGOptionalFields(vrg, homeCluster)
 
 	return d.mwu.UpdateVRGManifestWork(vrg, mw)