feat(replicationState): add an option to add replication state per ns/pvc

Signed-off-by: SkalaNetworks <contact@skala.network>

Author: SkalaNetworks

README.md

@@ -182,6 +182,26 @@ metadata:
 > [!NOTE]
 > While paused, existing `VolumeReplication` objects are frozen in their current state. Unpausing resumes normal reconciliation.
 
+### Replication State
+
+By default, the `replicationState` of the created `VolumeReplication` is set to `primary`.
+You can change this using the `replication.superphenix.net/replicationState` annotation on either the PVC or its namespace.
+
+The annotation on the PVC takes precedence over the annotation on the namespace.
+
+Common values for this annotation are `primary` and `secondary`.
+
+Example — setting the entire namespace as `secondary`:
+
+```yaml
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: my-namespace
+  annotations:
+    replication.superphenix.net/replicationState: "secondary"
+```
+
 ### Excluding PVCs from replication
 
 It is possible to exclude some PVCs from being replicated, even if they have the correct annotations (or their namespace has them).

internal/constants/const.go

@@ -7,6 +7,7 @@ const (
 	PauseAnnotation                        = "replication.superphenix.net/pause"
 	ParentLabel                            = "replication.superphenix.net/parent"
 	StorageClassGroup                      = "replication.superphenix.net/storageClassGroup"
+	ReplicationStateAnnotation             = "replication.superphenix.net/replicationState"
 	StorageProvisionerAnnotation           = "volume.kubernetes.io/storage-provisioner"
 	DeprecatedStorageProvisionerAnnotation = "volume.beta.kubernetes.io/storage-provisioner"
 )

internal/replicator/replicator_test.go

@@ -47,6 +47,11 @@ func TestReconcileVolumeReplication(t *testing.T) {
 		},
 	}
 
+	// Add namespace to informer
+	_ = NamespaceInformer.Informer().GetIndexer().Add(&corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{Name: nsName},
+	})
+
 	vr := &unstructured.Unstructured{}
 	vr.SetUnstructuredContent(map[string]any{
 		"apiVersion": fmt.Sprintf("%s/%s", VolumeReplicationResource.Group, VolumeReplicationResource.Version),

internal/replicator/utils.go

@@ -28,6 +28,13 @@ 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 {
@@ -90,7 +97,7 @@ func createVolumeReplication(pvc *corev1.PersistentVolumeClaim) error {
 		},
 		"spec": map[string]any{
 			"volumeReplicationClass": getVolumeReplicationClass(pvc),
-			"replicationState":       "primary",
+			"replicationState":       getReplicationState(pvc),
 			"dataSource": map[string]any{
 				"apiGroup": "v1",
 				"kind":     "PersistentVolumeClaim",
@@ -174,27 +181,17 @@ func getPvcProvisioner(pvc *corev1.PersistentVolumeClaim) string {
 }
 
 // isPvcPaused returns whether the replication for a PVC is paused
-// The PVC-level annotation takes precedence over the namespace-level annotation:
-// any explicit value on the PVC (even "false") short-circuits the namespace lookup
 func isPvcPaused(pvc *corev1.PersistentVolumeClaim, namespace string) bool {
-	if pvc != nil {
-		// If the PVC has the annotation specified, it has priority over the one of the namespace
-		if value, ok := pvc.Annotations[constants.PauseAnnotation]; ok {
-			return value == "true"
-		}
+	if pvc == nil {
+		return isNamespacePaused(namespace)
 	}
 
-	return isNamespacePaused(namespace)
+	return getAnnotationValue(pvc, constants.PauseAnnotation) == "true"
 }
 
 // isNamespacePaused returns whether replication is paused at the namespace level
 func isNamespacePaused(namespace string) bool {
-	ns, err := NamespaceInformer.Lister().Get(namespace)
-	if err != nil {
-		return false
-	}
-
-	return ns.Annotations[constants.PauseAnnotation] == "true"
+	return getNamespaceAnnotationValue(namespace, constants.PauseAnnotation) == "true"
 }
 
 // pvcNameMatchesExclusion returns whether a PVC has a name matching the exclusion regex

internal/replicator/utils_test.go

@@ -47,6 +47,11 @@ func TestCreateVolumeReplication(t *testing.T) {
 		},
 	}
 
+	// Add namespace to informer
+	_ = NamespaceInformer.Informer().GetIndexer().Add(&corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{Name: nsName},
+	})
+
 	t.Run("Successful creation", func(t *testing.T) {
 		err := createVolumeReplication(pvc)
 		require.NoError(t, err)
@@ -492,6 +497,11 @@ func TestIsVolumeReplicationCorrect(t *testing.T) {
 		},
 	}
 
+	// Add namespace to informer
+	_ = NamespaceInformer.Informer().GetIndexer().Add(&corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{Name: nsName},
+	})
+
 	tests := []struct {
 		name     string
 		vr       *unstructured.Unstructured
@@ -507,6 +517,7 @@ func TestIsVolumeReplicationCorrect(t *testing.T) {
 					},
 					"spec": map[string]any{
 						"volumeReplicationClass": vrcName,
+						"replicationState":       "primary",
 						"dataSource": map[string]any{
 							"apiGroup": "v1",
 							"kind":     "PersistentVolumeClaim",
@@ -517,6 +528,27 @@ func TestIsVolumeReplicationCorrect(t *testing.T) {
 			},
 			expected: true,
 		},
+		{
+			name: "replicationState mismatch",
+			vr: &unstructured.Unstructured{
+				Object: map[string]any{
+					"metadata": map[string]any{
+						"name":      pvcName,
+						"namespace": nsName,
+					},
+					"spec": map[string]any{
+						"volumeReplicationClass": vrcName,
+						"replicationState":       "secondary",
+						"dataSource": map[string]any{
+							"apiGroup": "v1",
+							"kind":     "PersistentVolumeClaim",
+							"name":     pvcName,
+						},
+					},
+				},
+			},
+			expected: false,
+		},
 		{
 			name: "volumeReplicationClass mismatch",
 			vr: &unstructured.Unstructured{
@@ -833,6 +865,17 @@ func TestIsPvcPaused(t *testing.T) {
 			namespace: pausedNs,
 			expected:  false, // PVC takes precedence, and invalid is not true
 		},
+		{
+			name: "PVC empty pause value, NS paused",
+			pvc: &corev1.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace:   nsName,
+					Annotations: map[string]string{constants.PauseAnnotation: ""},
+				},
+			},
+			namespace: pausedNs,
+			expected:  true, // It falls back to NS now
+		},
 	}
 
 	for _, tt := range tests {

internal/replicator/vrc.go

@@ -90,23 +90,43 @@ func getVolumeReplicationClassSelector(pvc *corev1.PersistentVolumeClaim) string
 	return getAnnotationValue(pvc, constants.VrcSelectorAnnotation)
 }
 
+// getReplicationState returns the replication state to use for a PVC.
+// The replication state is specified through an annotation on the PVC or on its namespace.
+// The annotation on the PVC has priority over the one of the namespace.
+// If no replication state is found on either PVC or namespace, we return "primary" by default.
+func getReplicationState(pvc *corev1.PersistentVolumeClaim) string {
+	state := getAnnotationValue(pvc, constants.ReplicationStateAnnotation)
+	if state == "" {
+		return "primary"
+	}
+	return state
+}
+
 // getAnnotationValue returns the value of an annotation from a PVC or its namespace.
 // The annotation on the PVC has priority over the one of the namespace.
 func getAnnotationValue(pvc *corev1.PersistentVolumeClaim, annotation string) string {
+	if pvc == nil {
+		return ""
+	}
+
 	// If the PVC has the annotation specified, it has priority over the one of the namespace
 	if value, ok := pvc.Annotations[annotation]; ok && value != "" {
 		return value
 	}
 
 	// If the PVC doesn't have the annotation specified, fall back to the namespace
-	namespace, err := NamespaceInformer.Lister().Get(pvc.Namespace)
+	return getNamespaceAnnotationValue(pvc.Namespace, annotation)
+}
+
+// getNamespaceAnnotationValue returns the value of an annotation from a namespace.
+func getNamespaceAnnotationValue(namespace string, annotation string) string {
+	ns, err := NamespaceInformer.Lister().Get(namespace)
 	if err != nil {
-		klog.Errorf("failed to retrieve parent namespace for PVC %s/%s: %s", pvc.Namespace, pvc.Name, err.Error())
+		klog.Errorf("failed to retrieve namespace %s: %s", namespace, err.Error())
 		return ""
 	}
 
-	// If the namespace doesn't have the annotation, this will return an empty string
-	return namespace.Annotations[annotation]
+	return ns.Annotations[annotation]
 }
 
 // filterVrcFromSelector returns a VolumeReplicationClass that is in a specific StorageClass Group

internal/replicator/vrc_test.go

@@ -709,3 +709,84 @@ func TestFilterVrcFromSelector(t *testing.T) {
 		require.Nil(t, list)
 	})
 }
+
+func TestGetReplicationState(t *testing.T) {
+	_, _, informerFactory := setupTestEnvironment()
+	nsName := "test-ns"
+
+	tests := []struct {
+		name           string
+		pvc            *corev1.PersistentVolumeClaim
+		namespace      *corev1.Namespace
+		expectedResult string
+	}{
+		{
+			name: "Default value (primary)",
+			pvc: &corev1.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-pvc",
+					Namespace: nsName,
+				},
+			},
+			expectedResult: "primary",
+		},
+		{
+			name: "Override from Namespace",
+			pvc: &corev1.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-pvc",
+					Namespace: nsName,
+				},
+			},
+			namespace: &corev1.Namespace{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: nsName,
+					Annotations: map[string]string{
+						constants.ReplicationStateAnnotation: "secondary",
+					},
+				},
+			},
+			expectedResult: "secondary",
+		},
+		{
+			name: "Override from PVC",
+			pvc: &corev1.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-pvc",
+					Namespace: nsName,
+					Annotations: map[string]string{
+						constants.ReplicationStateAnnotation: "primary",
+					},
+				},
+			},
+			namespace: &corev1.Namespace{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: nsName,
+					Annotations: map[string]string{
+						constants.ReplicationStateAnnotation: "secondary",
+					},
+				},
+			},
+			expectedResult: "primary",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			clearNamespaceIndexer(t)
+			if tt.namespace != nil {
+				err := informerFactory.Core().V1().Namespaces().Informer().GetIndexer().Add(tt.namespace)
+				require.NoError(t, err)
+			} else {
+				// Add a default namespace if not provided
+				err := informerFactory.Core().V1().Namespaces().Informer().GetIndexer().Add(&corev1.Namespace{
+					ObjectMeta: metav1.ObjectMeta{Name: nsName},
+				})
+				require.NoError(t, err)
+			}
+
+			result := getReplicationState(tt.pvc)
+			require.Equal(t, tt.expectedResult, result)
+		})
+	}
+}