feat(replications): update live replicationState when changed

internal/replicator/handlers.go

@@ -12,12 +12,14 @@ import (
 )
 
 // namespaceUpdate is called whenever an update is detected on a namespace
-// We check if the volumeReplicationClass annotation has changed, and if it has,
+// We check if annotations have changed, and if it has,
 // we propagate the update to every PVC inside the namespace
 func (c *Controller) namespaceUpdate(oldNs, newNs *corev1.Namespace) {
-	// Don't continue if the class haven't changed or if the annotations weren't deleted
+	// Don't continue if the annotations have not changed/were not deleted
 	if oldNs.Annotations[constants.VrcValueAnnotation] == newNs.Annotations[constants.VrcValueAnnotation] &&
-		oldNs.Annotations[constants.VrcSelectorAnnotation] == newNs.Annotations[constants.VrcSelectorAnnotation] {
+		oldNs.Annotations[constants.VrcSelectorAnnotation] == newNs.Annotations[constants.VrcSelectorAnnotation] &&
+		oldNs.Annotations[constants.PauseAnnotation] == newNs.Annotations[constants.PauseAnnotation] &&
+		oldNs.Annotations[constants.ReplicationStateAnnotation] == newNs.Annotations[constants.ReplicationStateAnnotation] {
 		return
 	}
 

internal/replicator/replicator.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/cache"
@@ -62,8 +63,10 @@ func (c *Controller) processNextItem() bool {
 // - if the VolumeReplication exists
 //   - check if the PVC has a matching VolumeReplicationClass
 //   - and if it doesn't, delete the VolumeReplication
-//   - check if the definition of the VolumeReplication is correct
+//   - check if the target of the VolumeReplication is correct
 //   - and if it doesn't, delete it, and it will be re-created on the next sync
+//   - check if the replicationState of the VolumeReplication is correct
+//   - and if isn't, live update the VolumeReplication
 //
 // - if the VolumeReplication doesn't exist
 //   - and if a corresponding VolumeReplicationClass exists, create the VolumeReplication
@@ -112,21 +115,33 @@ func reconcileVolumeReplication(key string) {
 
 	// The VolumeReplication exists, we need to check:
 	//  - if the PVC still has a matching VolumeReplicationClass
-	//    - and if it doesn't, we need to delete the VolumeReplication
-	//  - if the definition of the VolumeReplication is correct
-	//    - and if it isn't, we need to delete the VolumeReplication
+	//    - and if it doesn't, we need to delete the VolumeReplication (we can't live update the class)
+	//  - if the target of the VolumeReplication is correct
+	//    - and if it isn't, we need to delete the VolumeReplication (we can't live update the target)
+	//  - if the replicationState of the VolumeReplication is correct
+	//    - and if it isn't, we live update the VR
 	if volumeReplication != nil {
 		vrcExists := replicationClass != ""
 		vrCorrect := isVolumeReplicationCorrect(pvc, volumeReplication)
 
 		if !vrcExists || !vrCorrect {
 			klog.Infof("deleting VolumeReplication %s as it doesn't conform anymore, vrcExists(%t), vrCorrect(%t)", key, vrcExists, vrCorrect)
 
-			// If we're meant to update the VolumeReplication (!vrCorrect), we delete it here, and it will trigger an
+			// If we're meant to re-create the VolumeReplication (!vrCorrect), we delete it here, and it will trigger an
 			// event that will bring us back in this function to re-create it with the correct definition
 			cleanupVolumeReplication(name, namespace)
 			return
 		}
+
+		// Check if the replicationState needs an update
+		expectedState := getReplicationState(pvc)
+		currentState, _, _ := unstructured.NestedString(volumeReplication.Object, "spec", "replicationState")
+		if currentState != expectedState {
+			klog.Infof("updating VolumeReplication %s with new replication state %s (was %s)", key, expectedState, currentState)
+			if err = updateVolumeReplication(pvc, volumeReplication); err != nil {
+				klog.Errorf("failed to update VolumeReplication %s: %s", key, err.Error())
+			}
+		}
 	}
 
 	// No volume replication object was found for this PVC, we need to create it

internal/replicator/replicator_test.go

@@ -337,6 +337,35 @@ func TestReconcileVolumeReplication(t *testing.T) {
 				require.True(t, created, "VR should have been created")
 			},
 		},
+		{
+			name: "VR exists, replicationState mismatch -> update VR",
+			setup: func() {
+				pvcSecondary := pvc.DeepCopy()
+				pvcSecondary.Annotations[constants.ReplicationStateAnnotation] = "secondary"
+				err := PvcInformer.Informer().GetIndexer().Add(pvcSecondary)
+				require.NoError(t, err)
+				err = VolumeReplicationInformer.Informer().GetIndexer().Add(vr)
+				require.NoError(t, err)
+			},
+			verify: func(t *testing.T) {
+				actions := dynamicClient.Actions()
+				updated := slices.ContainsFunc(actions, func(action k8s_testing.Action) bool {
+					if action.GetVerb() != "update" {
+						return false
+					}
+					updateAction := action.(k8s_testing.UpdateAction)
+					obj := updateAction.GetObject().(*unstructured.Unstructured)
+					state, _, _ := unstructured.NestedString(obj.Object, "spec", "replicationState")
+					return state == "secondary"
+				})
+				require.True(t, updated, "VR should have been updated with new replication state")
+
+				deleted := slices.ContainsFunc(actions, func(action k8s_testing.Action) bool {
+					return action.GetVerb() == "delete"
+				})
+				require.False(t, deleted, "VR should not have been deleted")
+			},
+		},
 	}
 
 	for _, tt := range tests {

internal/replicator/utils.go

@@ -28,13 +28,6 @@ func isVolumeReplicationCorrect(pvc *corev1.PersistentVolumeClaim, vr *unstructu
 		return false
 	}
 
-	// Check that the replicationState correspond to the one inherited from the PVC/NS
-	replicationState, _, _ := unstructured.NestedString(vr.Object, "spec", "replicationState")
-	if getReplicationState(pvc) != replicationState {
-		klog.Infof("VolumeReplication %s has a replication state mismatch with its parent (got %s)", key, replicationState)
-		return false
-	}
-
 	// Check that the dataSource points to the PVC
 	dataSource, _, _ := unstructured.NestedNullCoercingStringMap(vr.Object, "spec", "dataSource")
 	if dataSource["apiGroup"] != "v1" || dataSource["kind"] != "PersistentVolumeClaim" || dataSource["name"] != pvc.Name {
@@ -56,6 +49,19 @@ func cleanupVolumeReplication(name, namespace string) {
 	}
 }
 
+// updateVolumeReplication updates the replicationState of a VolumeReplication
+func updateVolumeReplication(pvc *corev1.PersistentVolumeClaim, vr *unstructured.Unstructured) error {
+	// Update the replicationState
+	err := unstructured.SetNestedField(vr.Object, getReplicationState(pvc), "spec", "replicationState")
+	if err != nil {
+		return err
+	}
+
+	resourceInterface := k8s.DynamicClientSet.Resource(VolumeReplicationResource).Namespace(pvc.Namespace)
+	_, err = resourceInterface.Update(context.Background(), vr, metav1.UpdateOptions{})
+	return err
+}
+
 // getPersistentVolumeClaim returns a PersistentVolumeClaim from its key
 func getPersistentVolumeClaim(key string) (*corev1.PersistentVolumeClaim, error) {
 	pvc, exists, err := PvcInformer.Informer().GetIndexer().GetByKey(key)

internal/replicator/utils_test.go

@@ -548,7 +548,7 @@ func TestIsVolumeReplicationCorrect(t *testing.T) {
 					},
 				},
 			},
-			expected: false,
+			expected: true,
 		},
 		{
 			name: "volumeReplicationClass mismatch",