Skip to content

Commit

Permalink
groupreplication: add controller logic for volume group replication
Browse files Browse the repository at this point in the history 
added controller logic for volume group replication

Signed-off-by: Nikhil-Ladha <[email protected]>
  • Loading branch information
@Nikhil-Ladha
Nikhil-Ladha committed Nov 29, 2024
1 parent b9361a3 commit 4f5e624
Show file tree
Hide file tree
Showing 14 changed files with 1,355 additions and 143 deletions.
9 changes: 7 additions & 2 deletions api/replication.storage/v1alpha1/volumegroupreplication_types.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,16 +21,21 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

const (
VolumeGroupReplicationNameAnnotation = "replication.storage.openshift.io/volume-group-replication-name"
)

// VolumeGroupReplicationSpec defines the desired state of VolumeGroupReplication
type VolumeGroupReplicationSpec struct {
// volumeGroupReplicationClassName is the volumeGroupReplicationClass name for this VolumeGroupReplication resource
// +kubebuilder:validation:Required
// +kubebuilder:validation:XValidation:rule="self == oldSelf",message="volumeGroupReplicationClassName is immutable"
VolumeGroupReplicationClassName string `json:"volumeGroupReplicationClassName"`

// volumeReplicationClassName is the volumeReplicationClass name for VolumeReplication object
// volumeReplicationClassName is the volumeReplicationClass name for the VolumeReplication object
// created for this volumeGroupReplication
// +kubebuilder:validation:Required
// +kubebuilder:validation:XValidation:rule="self == oldSelf",message="volumReplicationClassName is immutable"
// +kubebuilder:validation:XValidation:rule="self == oldSelf",message="volumeReplicationClassName is immutable"
VolumeReplicationClassName string `json:"volumeReplicationClassName"`

// Name of the VolumeReplication object created for this volumeGroupReplication
Expand Down
9 changes: 6 additions & 3 deletions api/replication.storage/v1alpha1/volumegroupreplicationcontent_types.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,10 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

const (
VolumeGroupReplicationContentNameAnnotation = "replication.storage.openshift.io/volumegroupreplication-content-name"
)

// VolumeGroupReplicationContentSpec defines the desired state of VolumeGroupReplicationContent
type VolumeGroupReplicationContentSpec struct {
// VolumeGroupreplicationRef specifies the VolumeGroupReplication object to which this
Expand Down Expand Up @@ -52,9 +56,8 @@ type VolumeGroupReplicationContentSpec struct {
// +optional
VolumeGroupReplicationClassName string `json:"volumeGroupReplicationClassName"`

// Source specifies whether the snapshot is (or should be) dynamically provisioned
// Source specifies whether the volume is (or should be) dynamically provisioned
// or already exists, and just requires a Kubernetes object representation.
// This field is immutable after creation.
// Required.
Source VolumeGroupReplicationContentSource `json:"source"`
}
Expand All @@ -68,7 +71,7 @@ type VolumeGroupReplicationContentSource struct {

// VolumeGroupReplicationContentStatus defines the status of VolumeGroupReplicationContent
type VolumeGroupReplicationContentStatus struct {
// PersistentVolumeRefList is the list of of PV for the group replication
// PersistentVolumeRefList is the list of PV for the group replication
// The maximum number of allowed PV in the group is 100.
// +optional
PersistentVolumeRefList []corev1.LocalObjectReference `json:"persistentVolumeRefList,omitempty"`
Expand Down
6 changes: 4 additions & 2 deletions cmd/manager/main.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -212,8 +212,10 @@ func main() {
os.Exit(1)
}
if err = (&replicationController.VolumeGroupReplicationContentReconciler{
Client: mgr.GetClient(),
Scheme: mgr.GetScheme(),
Client: mgr.GetClient(),
Scheme: mgr.GetScheme(),
Connpool: connPool,
Timeout: defaultTimeout,
}).SetupWithManager(mgr); err != nil {
setupLog.Error(err, "unable to create controller", "controller", "VolumeGroupReplicationContent")
os.Exit(1)
Expand Down
87 changes: 63 additions & 24 deletions internal/controller/replication.storage/finalizers.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ import (

"github.com/go-logr/logr"
corev1 "k8s.io/api/core/v1"
"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
Expand Down Expand Up @@ -65,43 +66,44 @@ func (r *VolumeReplicationReconciler) removeFinalizerFromVR(logger logr.Logger,
return nil
}

// addFinalizerToPVC adds the VR finalizer on the PersistentVolumeClaim.
func (r *VolumeReplicationReconciler) addFinalizerToPVC(logger logr.Logger, pvc *corev1.PersistentVolumeClaim) error {
if !slices.Contains(pvc.ObjectMeta.Finalizers, pvcReplicationFinalizer) {
logger.Info("adding finalizer to PersistentVolumeClaim object", "Finalizer", pvcReplicationFinalizer)
pvc.ObjectMeta.Finalizers = append(pvc.ObjectMeta.Finalizers, pvcReplicationFinalizer)
if err := r.Client.Update(context.TODO(), pvc); err != nil {
// AddFinalizerToPVC adds the VR finalizer on the PersistentVolumeClaim.
func AddFinalizerToPVC(client client.Client, logger logr.Logger, pvc *corev1.PersistentVolumeClaim,
replicationFinalizer string) error {
if !slices.Contains(pvc.ObjectMeta.Finalizers, replicationFinalizer) {
logger.Info("adding finalizer to PersistentVolumeClaim object", "Finalizer", replicationFinalizer)
pvc.ObjectMeta.Finalizers = append(pvc.ObjectMeta.Finalizers, replicationFinalizer)
if err := client.Update(context.TODO(), pvc); err != nil {
return fmt.Errorf("failed to add finalizer (%s) to PersistentVolumeClaim resource"+
" (%s/%s) %w",
pvcReplicationFinalizer, pvc.Namespace, pvc.Name, err)
replicationFinalizer, pvc.Namespace, pvc.Name, err)
}
}

return nil
}

// removeFinalizerFromPVC removes the VR finalizer on PersistentVolumeClaim.
func (r *VolumeReplicationReconciler) removeFinalizerFromPVC(logger logr.Logger, pvc *corev1.PersistentVolumeClaim,
) error {
if slices.Contains(pvc.ObjectMeta.Finalizers, pvcReplicationFinalizer) {
logger.Info("removing finalizer from PersistentVolumeClaim object", "Finalizer", pvcReplicationFinalizer)
pvc.ObjectMeta.Finalizers = util.RemoveFromSlice(pvc.ObjectMeta.Finalizers, pvcReplicationFinalizer)
if err := r.Client.Update(context.TODO(), pvc); err != nil {
// RemoveFinalizerFromPVC removes the VR finalizer on PersistentVolumeClaim.
func RemoveFinalizerFromPVC(client client.Client, logger logr.Logger, pvc *corev1.PersistentVolumeClaim,
replicationFinalizer string) error {
if slices.Contains(pvc.ObjectMeta.Finalizers, replicationFinalizer) {
logger.Info("removing finalizer from PersistentVolumeClaim object", "Finalizer", replicationFinalizer)
pvc.ObjectMeta.Finalizers = util.RemoveFromSlice(pvc.ObjectMeta.Finalizers, replicationFinalizer)
if err := client.Update(context.TODO(), pvc); err != nil {
return fmt.Errorf("failed to remove finalizer (%s) from PersistentVolumeClaim resource"+
" (%s/%s), %w",
pvcReplicationFinalizer, pvc.Namespace, pvc.Name, err)
replicationFinalizer, pvc.Namespace, pvc.Name, err)
}
}

return nil
}

// addFinalizerToVGR adds the VR finalizer on the VolumeGroupReplication.
func (r *VolumeReplicationReconciler) addFinalizerToVGR(logger logr.Logger, vgr *replicationv1alpha1.VolumeGroupReplication) error {
// AddFinalizerToVGR adds the VGR finalizer on the VolumeGroupReplication resource
func AddFinalizerToVGR(client client.Client, logger logr.Logger, vgr *replicationv1alpha1.VolumeGroupReplication) error {
if !slices.Contains(vgr.ObjectMeta.Finalizers, vgrReplicationFinalizer) {
logger.Info("adding finalizer to VolumeGroupReplication object", "Finalizer", vgrReplicationFinalizer)
logger.Info("adding finalizer to volumeGroupReplication object", "Finalizer", vgrReplicationFinalizer)
vgr.ObjectMeta.Finalizers = append(vgr.ObjectMeta.Finalizers, vgrReplicationFinalizer)
if err := r.Client.Update(context.TODO(), vgr); err != nil {
if err := client.Update(context.TODO(), vgr); err != nil {
return fmt.Errorf("failed to add finalizer (%s) to VolumeGroupReplication resource"+
" (%s/%s) %w",
vgrReplicationFinalizer, vgr.Namespace, vgr.Name, err)
Expand All @@ -111,13 +113,18 @@ func (r *VolumeReplicationReconciler) addFinalizerToVGR(logger logr.Logger, vgr
return nil
}

// removeFinalizerFromVGR removes the VR finalizer on VolumeGroupReplication.
func (r *VolumeReplicationReconciler) removeFinalizerFromVGR(logger logr.Logger, vgr *replicationv1alpha1.VolumeGroupReplication,
) error {
// RemoveFinalizerFromVGR removes the VGR finalizer from the VolumeGroupReplication instance.
func RemoveFinalizerFromVGR(client client.Client, logger logr.Logger, vgr *replicationv1alpha1.VolumeGroupReplication) error {
if slices.Contains(vgr.ObjectMeta.Finalizers, vgrReplicationFinalizer) {
logger.Info("removing finalizer from VolumeGroupReplication object", "Finalizer", vgrReplicationFinalizer)
logger.Info("removing finalizer from volumeGroupReplication object", "Finalizer", vgrReplicationFinalizer)
// Check if owner annotations are removed from the VGR resource
if vgr.Annotations[replicationv1alpha1.VolumeGroupReplicationContentNameAnnotation] != "" ||
vgr.Annotations[replicationv1alpha1.VolumeReplicationNameAnnotation] != "" {
return fmt.Errorf("failed to remove finalizer from volumeGroupReplication object"+
",dependent resources are not yet deleted (%s/%s)", vgr.Namespace, vgr.Name)
}
vgr.ObjectMeta.Finalizers = util.RemoveFromSlice(vgr.ObjectMeta.Finalizers, vgrReplicationFinalizer)
if err := r.Client.Update(context.TODO(), vgr); err != nil {
if err := client.Update(context.TODO(), vgr); err != nil {
return fmt.Errorf("failed to remove finalizer (%s) from VolumeGroupReplication resource"+
" (%s/%s), %w",
vgrReplicationFinalizer, vgr.Namespace, vgr.Name, err)
Expand All @@ -126,3 +133,35 @@ func (r *VolumeReplicationReconciler) removeFinalizerFromVGR(logger logr.Logger,

return nil
}

// addFinalizerToVGRContent adds the VGR finalizer on the VolumeGroupReplicationContent resource
func (r *VolumeGroupReplicationContentReconciler) addFinalizerToVGRContent(logger logr.Logger,
vgrContent *replicationv1alpha1.VolumeGroupReplicationContent) error {
if !slices.Contains(vgrContent.ObjectMeta.Finalizers, vgrReplicationFinalizer) {
logger.Info("adding finalizer to volumeGroupReplicationContent object", "Finalizer", vgrReplicationFinalizer)
vgrContent.ObjectMeta.Finalizers = append(vgrContent.ObjectMeta.Finalizers, vgrReplicationFinalizer)
if err := r.Client.Update(context.TODO(), vgrContent); err != nil {
return fmt.Errorf("failed to add finalizer (%s) to VolumeGroupReplicationContent resource"+
" (%s/%s) %w",
vgrReplicationFinalizer, vgrContent.Namespace, vgrContent.Name, err)
}
}

return nil
}

// removeFinalizerFromVGRContent removes the VGR finalizer from the VolumeGroupReplicationContent instance.
func (r *VolumeGroupReplicationContentReconciler) removeFinalizerFromVGRContent(logger logr.Logger,
vgrContent *replicationv1alpha1.VolumeGroupReplicationContent) error {
if slices.Contains(vgrContent.ObjectMeta.Finalizers, vgrReplicationFinalizer) {
logger.Info("removing finalizer from volumeGroupReplicationContent object", "Finalizer", vgrReplicationFinalizer)
vgrContent.ObjectMeta.Finalizers = util.RemoveFromSlice(vgrContent.ObjectMeta.Finalizers, vgrReplicationFinalizer)
if err := r.Client.Update(context.TODO(), vgrContent); err != nil {
return fmt.Errorf("failed to remove finalizer (%s) from VolumeGroupReplicationContent resource"+
" (%s/%s), %w",
vgrReplicationFinalizer, vgrContent.Namespace, vgrContent.Name, err)
}
}

return nil
}
10 changes: 8 additions & 2 deletions internal/controller/replication.storage/parameters.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,10 @@ const (
// Driver.
replicationParameterPrefix = "replication.storage.openshift.io/"

prefixedReplicationSecretNameKey = replicationParameterPrefix + "replication-secret-name" // name key for secret
prefixedReplicationSecretNamespaceKey = replicationParameterPrefix + "replication-secret-namespace" // namespace key secret
prefixedReplicationSecretNameKey = replicationParameterPrefix + "replication-secret-name" // name key for secret
prefixedReplicationSecretNamespaceKey = replicationParameterPrefix + "replication-secret-namespace" // namespace key secret
prefixedGroupReplicationSecretNameKey = replicationParameterPrefix + "group-replication-secret-name" // name key for secret
prefixedGroupReplicationSecretNamespaceKey = replicationParameterPrefix + "group-replication-secret-namespace" // namespace key secret
)

// filterPrefixedParameters removes all the reserved keys from the
Expand All @@ -53,10 +55,14 @@ func validatePrefixedParameters(param map[string]string) error {
if strings.HasPrefix(k, replicationParameterPrefix) {
switch k {
case prefixedReplicationSecretNameKey:
fallthrough
case prefixedGroupReplicationSecretNameKey:
if v == "" {
return errors.New("secret name cannot be empty")
}
case prefixedReplicationSecretNamespaceKey:
fallthrough
case prefixedGroupReplicationSecretNamespaceKey:
if v == "" {
return errors.New("secret namespace cannot be empty")
}
Expand Down
39 changes: 23 additions & 16 deletions internal/controller/replication.storage/pvc.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ import (
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/client"

replicationv1alpha1 "github.com/csi-addons/kubernetes-csi-addons/api/replication.storage/v1alpha1"
)
Expand Down Expand Up @@ -59,22 +60,27 @@ func (r VolumeReplicationReconciler) getPVCDataSource(logger logr.Logger, req ty
return pvc, pv, nil
}

// annotatePVCWithOwner will add the VolumeReplication details to the PVC annotations.
func (r *VolumeReplicationReconciler) annotatePVCWithOwner(ctx context.Context, logger logr.Logger, reqOwnerName string, pvc *corev1.PersistentVolumeClaim) error {
// AnnotatePVCWithOwner will add the VolumeReplication/VolumeGroupReplication details to the PVC annotations.
func AnnotatePVCWithOwner(client client.Client, logger logr.Logger, reqOwnerName string,
pvc *corev1.PersistentVolumeClaim, pvcAnnotation string) error {
if pvc.ObjectMeta.Annotations == nil {
pvc.ObjectMeta.Annotations = map[string]string{}
}

currentOwnerName := pvc.ObjectMeta.Annotations[replicationv1alpha1.VolumeReplicationNameAnnotation]
if pvc.ObjectMeta.Annotations[replicationv1alpha1.VolumeReplicationNameAnnotation] != "" &&
pvc.ObjectMeta.Annotations[replicationv1alpha1.VolumeGroupReplicationNameAnnotation] != "" {
logger.Info("PVC can't be part of both VolumeGroupReplication and VolumeReplication")
return fmt.Errorf("PVC %q can't be owned by both VolumeReplication and VolumeGroupReplication", pvc.Name)
}

currentOwnerName := pvc.ObjectMeta.Annotations[pvcAnnotation]
if currentOwnerName == "" {
logger.Info("setting owner on PVC annotation", "Name", pvc.Name, "owner", reqOwnerName)
pvc.ObjectMeta.Annotations[replicationv1alpha1.VolumeReplicationNameAnnotation] = reqOwnerName
err := r.Update(ctx, pvc)
pvc.ObjectMeta.Annotations[pvcAnnotation] = reqOwnerName
err := client.Update(context.TODO(), pvc)
if err != nil {
logger.Error(err, "Failed to update PVC annotation", "Name", pvc.Name)

return fmt.Errorf("failed to update PVC %q annotation for VolumeReplication: %w",
pvc.Name, err)
return fmt.Errorf("failed to update PVC %q annotation for replication: %w", pvc.Name, err)
}

return nil
Expand All @@ -86,22 +92,23 @@ func (r *VolumeReplicationReconciler) annotatePVCWithOwner(ctx context.Context,
"current owner", currentOwnerName,
"requested owner", reqOwnerName)

return fmt.Errorf("PVC %q not owned by VolumeReplication %q",
return fmt.Errorf("PVC %q not owned by correct VolumeReplication/VolumeGroupReplication %q",
pvc.Name, reqOwnerName)
}

return nil
}

// removeOwnerFromPVCAnnotation removes the VolumeReplication owner from the PVC annotations.
func (r *VolumeReplicationReconciler) removeOwnerFromPVCAnnotation(ctx context.Context, logger logr.Logger, pvc *corev1.PersistentVolumeClaim) error {
if _, ok := pvc.ObjectMeta.Annotations[replicationv1alpha1.VolumeReplicationNameAnnotation]; ok {
logger.Info("removing owner annotation from PersistentVolumeClaim object", "Annotation", replicationv1alpha1.VolumeReplicationNameAnnotation)
delete(pvc.ObjectMeta.Annotations, replicationv1alpha1.VolumeReplicationNameAnnotation)
if err := r.Client.Update(ctx, pvc); err != nil {
// RemoveOwnerFromPVCAnnotation removes the VolumeReplication/VolumeGroupReplication owner from the PVC annotations.
func RemoveOwnerFromPVCAnnotation(client client.Client, logger logr.Logger, pvc *corev1.PersistentVolumeClaim,
pvcAnnotation string) error {
if _, ok := pvc.ObjectMeta.Annotations[pvcAnnotation]; ok {
logger.Info("removing owner annotation from PersistentVolumeClaim object", "Annotation", pvcAnnotation)
delete(pvc.ObjectMeta.Annotations, pvcAnnotation)
if err := client.Update(context.TODO(), pvc); err != nil {
return fmt.Errorf("failed to remove annotation %q from PersistentVolumeClaim "+
"%q %w",
replicationv1alpha1.VolumeReplicationNameAnnotation, pvc.Name, err)
pvcAnnotation, pvc.Name, err)
}
}

Expand Down
16 changes: 10 additions & 6 deletions internal/controller/replication.storage/pvc_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ package controller

import (
"context"
"fmt"
"testing"

replicationv1alpha1 "github.com/csi-addons/kubernetes-csi-addons/api/replication.storage/v1alpha1"
Expand Down Expand Up @@ -178,6 +179,7 @@ func TestGetVolumeHandle(t *testing.T) {
func TestVolumeReplicationReconciler_annotatePVCWithOwner(t *testing.T) {
t.Parallel()
vrName := "test-vr"
vrNamespace := "test-ns"

testcases := []struct {
name string
Expand All @@ -196,7 +198,7 @@ func TestVolumeReplicationReconciler_annotatePVCWithOwner(t *testing.T) {
Name: "pvc-name",
Namespace: mockNamespace,
Annotations: map[string]string{
replicationv1alpha1.VolumeReplicationNameAnnotation: vrName,
replicationv1alpha1.VolumeReplicationNameAnnotation: fmt.Sprintf("%s/%s", vrName, vrNamespace),
},
},
},
Expand All @@ -220,13 +222,15 @@ func TestVolumeReplicationReconciler_annotatePVCWithOwner(t *testing.T) {
for _, tc := range testcases {
volumeReplication := &replicationv1alpha1.VolumeReplication{}
mockVolumeReplicationObj.DeepCopyInto(volumeReplication)
volumeReplication.Name = vrName

testPVC := &corev1.PersistentVolumeClaim{}
tc.pvc.DeepCopyInto(testPVC)

ctx := context.TODO()
reconciler := createFakeVolumeReplicationReconciler(t, testPVC, volumeReplication)
err := reconciler.annotatePVCWithOwner(ctx, log.FromContext(context.TODO()), vrName, testPVC)
reqOwner := fmt.Sprintf("%s/%s", volumeReplication.Name, volumeReplication.Namespace)
err := AnnotatePVCWithOwner(reconciler.Client, log.FromContext(context.TODO()), reqOwner, testPVC, replicationv1alpha1.VolumeReplicationNameAnnotation)
if tc.errorExpected {
assert.Error(t, err)
} else {
Expand All @@ -241,14 +245,14 @@ func TestVolumeReplicationReconciler_annotatePVCWithOwner(t *testing.T) {
err = reconciler.Get(ctx, pvcNamespacedName, testPVC)
assert.NoError(t, err)

assert.Equal(t, testPVC.ObjectMeta.Annotations[replicationv1alpha1.VolumeReplicationNameAnnotation], vrName)
assert.Equal(t, testPVC.ObjectMeta.Annotations[replicationv1alpha1.VolumeReplicationNameAnnotation], reqOwner)
}

err = reconciler.removeOwnerFromPVCAnnotation(context.TODO(), log.FromContext(context.TODO()), testPVC)
err = RemoveOwnerFromPVCAnnotation(reconciler.Client, log.FromContext(context.TODO()), testPVC, replicationv1alpha1.VolumeReplicationNameAnnotation)
assert.NoError(t, err)

// try calling delete again, it should not fail
err = reconciler.removeOwnerFromPVCAnnotation(context.TODO(), log.FromContext(context.TODO()), testPVC)
err = RemoveOwnerFromPVCAnnotation(reconciler.Client, log.FromContext(context.TODO()), testPVC, replicationv1alpha1.VolumeReplicationNameAnnotation)
assert.NoError(t, err)

}
Expand All @@ -262,6 +266,6 @@ func TestVolumeReplicationReconciler_annotatePVCWithOwner(t *testing.T) {
}
volumeReplication := &replicationv1alpha1.VolumeReplication{}
reconciler := createFakeVolumeReplicationReconciler(t, pvc, volumeReplication)
err := reconciler.removeOwnerFromPVCAnnotation(context.TODO(), log.FromContext(context.TODO()), pvc)
err := RemoveOwnerFromPVCAnnotation(reconciler.Client, log.FromContext(context.TODO()), pvc, replicationv1alpha1.VolumeReplicationNameAnnotation)
assert.NoError(t, err)
}
Loading

0 comments on commit 4f5e624

Please sign in to comment.