diff --git a/internal/replicator/handlers.go b/internal/replicator/handlers.go index 5ea58cc..7bb8af5 100644 --- a/internal/replicator/handlers.go +++ b/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 } diff --git a/internal/replicator/replicator.go b/internal/replicator/replicator.go index 24d833d..2b54129 100644 --- a/internal/replicator/replicator.go +++ b/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,9 +115,11 @@ 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) @@ -122,11 +127,21 @@ func reconcileVolumeReplication(key string) { 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 diff --git a/internal/replicator/replicator_test.go b/internal/replicator/replicator_test.go index 2deb0ba..c4c6047 100644 --- a/internal/replicator/replicator_test.go +++ b/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 { diff --git a/internal/replicator/utils.go b/internal/replicator/utils.go index fb590e1..e42a3aa 100644 --- a/internal/replicator/utils.go +++ b/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) diff --git a/internal/replicator/utils_test.go b/internal/replicator/utils_test.go index 946e0c9..d806a39 100644 --- a/internal/replicator/utils_test.go +++ b/internal/replicator/utils_test.go @@ -548,7 +548,7 @@ func TestIsVolumeReplicationCorrect(t *testing.T) { }, }, }, - expected: false, + expected: true, }, { name: "volumeReplicationClass mismatch",