feat: support PDB in redisreplication (#1166)

<!--
Please read
https://github.com/OT-CONTAINER-KIT/redis-operator/blob/master/CONTRIBUTING.md
before submitting
your pull request. Please fill in each section below to help us better
prioritize your pull request. Thanks!
-->

**Description**

<!-- Please provide a summary of the change here. -->

Support PDB in redisreplication

**Type of change**

<!-- Please delete options that are not relevant. -->

* Bug fix (non-breaking change which fixes an issue)
* New feature (non-breaking change which adds functionality)
* Breaking change (fix or feature that would cause existing
functionality to not work as expected)

**Checklist**

- [ ] Tests have been added/modified and all tests pass.
- [ ] Functionality/bugs have been confirmed to be unchanged or fixed.
- [x] I have performed a self-review of my own code.
- [ ] Documentation has been updated or added where necessary.

**Additional Context**

<!-- 
    Is there anything else you'd like reviewers to know? 
    For example, any other related issues or testing carried out.
-->

api/v1beta2/redisreplication_types.go

@@ -1,31 +1,33 @@
 package v1beta2
 
 import (
+	common "github.com/OT-CONTAINER-KIT/redis-operator/api"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 type RedisReplicationSpec struct {
-	Size                          *int32                     `json:"clusterSize"`
-	KubernetesConfig              KubernetesConfig           `json:"kubernetesConfig"`
-	RedisExporter                 *RedisExporter             `json:"redisExporter,omitempty"`
-	RedisConfig                   *RedisConfig               `json:"redisConfig,omitempty"`
-	Storage                       *Storage                   `json:"storage,omitempty"`
-	NodeSelector                  map[string]string          `json:"nodeSelector,omitempty"`
-	PodSecurityContext            *corev1.PodSecurityContext `json:"podSecurityContext,omitempty"`
-	SecurityContext               *corev1.SecurityContext    `json:"securityContext,omitempty"`
-	PriorityClassName             string                     `json:"priorityClassName,omitempty"`
-	Affinity                      *corev1.Affinity           `json:"affinity,omitempty"`
-	Tolerations                   *[]corev1.Toleration       `json:"tolerations,omitempty"`
-	TLS                           *TLSConfig                 `json:"TLS,omitempty"`
-	ACL                           *ACLConfig                 `json:"acl,omitempty"`
-	ReadinessProbe                *corev1.Probe              `json:"readinessProbe,omitempty" protobuf:"bytes,11,opt,name=readinessProbe"`
-	LivenessProbe                 *corev1.Probe              `json:"livenessProbe,omitempty" protobuf:"bytes,12,opt,name=livenessProbe"`
-	InitContainer                 *InitContainer             `json:"initContainer,omitempty"`
-	Sidecars                      *[]Sidecar                 `json:"sidecars,omitempty"`
-	ServiceAccountName            *string                    `json:"serviceAccountName,omitempty"`
-	TerminationGracePeriodSeconds *int64                     `json:"terminationGracePeriodSeconds,omitempty" protobuf:"varint,4,opt,name=terminationGracePeriodSeconds"`
-	EnvVars                       *[]corev1.EnvVar           `json:"env,omitempty"`
+	Size                          *int32                           `json:"clusterSize"`
+	KubernetesConfig              KubernetesConfig                 `json:"kubernetesConfig"`
+	RedisExporter                 *RedisExporter                   `json:"redisExporter,omitempty"`
+	RedisConfig                   *RedisConfig                     `json:"redisConfig,omitempty"`
+	Storage                       *Storage                         `json:"storage,omitempty"`
+	NodeSelector                  map[string]string                `json:"nodeSelector,omitempty"`
+	PodSecurityContext            *corev1.PodSecurityContext       `json:"podSecurityContext,omitempty"`
+	SecurityContext               *corev1.SecurityContext          `json:"securityContext,omitempty"`
+	PriorityClassName             string                           `json:"priorityClassName,omitempty"`
+	Affinity                      *corev1.Affinity                 `json:"affinity,omitempty"`
+	Tolerations                   *[]corev1.Toleration             `json:"tolerations,omitempty"`
+	TLS                           *TLSConfig                       `json:"TLS,omitempty"`
+	PodDisruptionBudget           *common.RedisPodDisruptionBudget `json:"pdb,omitempty"`
+	ACL                           *ACLConfig                       `json:"acl,omitempty"`
+	ReadinessProbe                *corev1.Probe                    `json:"readinessProbe,omitempty" protobuf:"bytes,11,opt,name=readinessProbe"`
+	LivenessProbe                 *corev1.Probe                    `json:"livenessProbe,omitempty" protobuf:"bytes,12,opt,name=livenessProbe"`
+	InitContainer                 *InitContainer                   `json:"initContainer,omitempty"`
+	Sidecars                      *[]Sidecar                       `json:"sidecars,omitempty"`
+	ServiceAccountName            *string                          `json:"serviceAccountName,omitempty"`
+	TerminationGracePeriodSeconds *int64                           `json:"terminationGracePeriodSeconds,omitempty" protobuf:"varint,4,opt,name=terminationGracePeriodSeconds"`
+	EnvVars                       *[]corev1.EnvVar                 `json:"env,omitempty"`
 }
 
 func (cr *RedisReplicationSpec) GetReplicationCounts(t string) int32 {

config/crd/bases/redis.redis.opstreelabs.in_redisreplications.yaml

@@ -6240,6 +6240,19 @@ spec:
                 additionalProperties:
                   type: string
                 type: object
+              pdb:
+                description: RedisPodDisruptionBudget configure a PodDisruptionBudget
+                  on the resource (leader/follower)
+                properties:
+                  enabled:
+                    type: boolean
+                  maxUnavailable:
+                    format: int32
+                    type: integer
+                  minAvailable:
+                    format: int32
+                    type: integer
+                type: object
               podSecurityContext:
                 description: |-
                   PodSecurityContext holds pod-level security attributes and common container settings.

pkg/controllers/redisreplication/redisreplication_controller.go

@@ -46,6 +46,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 			{typ: "finalizer", rec: r.reconcileFinalizer},
 			{typ: "statefulset", rec: r.reconcileStatefulSet},
 			{typ: "service", rec: r.reconcileService},
+			{typ: "poddisruptionbudget", rec: r.reconcilePDB},
 			{typ: "redis", rec: r.reconcileRedis},
 			{typ: "status", rec: r.reconcileStatus},
 		}
@@ -131,6 +132,13 @@ func (r *Reconciler) reconcileAnnotation(ctx context.Context, instance *redisv1b
 	return intctrlutil.Reconciled()
 }
 
+func (r *Reconciler) reconcilePDB(ctx context.Context, instance *redisv1beta2.RedisReplication) (ctrl.Result, error) {
+	if err := k8sutils.ReconcileReplicationPodDisruptionBudget(ctx, instance, instance.Spec.PodDisruptionBudget, r.K8sClient); err != nil {
+		return intctrlutil.RequeueAfter(ctx, time.Second*60, "")
+	}
+	return intctrlutil.Reconciled()
+}
+
 func (r *Reconciler) reconcileStatefulSet(ctx context.Context, instance *redisv1beta2.RedisReplication) (ctrl.Result, error) {
 	if err := k8sutils.CreateReplicationRedis(ctx, instance, r.K8sClient); err != nil {
 		return intctrlutil.RequeueAfter(ctx, time.Second*60, "")

pkg/k8sutils/poddisruption.go

@@ -26,7 +26,7 @@ func ReconcileRedisPodDisruptionBudget(ctx context.Context, cr *redisv1beta2.Red
 		return CreateOrUpdatePodDisruptionBudget(ctx, pdbDef, cl)
 	} else {
 		// Check if one exists, and delete it.
-		_, err := GetPodDisruptionBudget(ctx, cr.Namespace, pdbName, cl)
+		_, err := getPodDisruptionBudget(ctx, cr.Namespace, pdbName, cl)
 		if err == nil {
 			return deletePodDisruptionBudget(ctx, cr.Namespace, pdbName, cl)
 		} else if err != nil && errors.IsNotFound(err) {
@@ -48,7 +48,29 @@ func ReconcileSentinelPodDisruptionBudget(ctx context.Context, cr *redisv1beta2.
 		return CreateOrUpdatePodDisruptionBudget(ctx, pdbDef, cl)
 	} else {
 		// Check if one exists, and delete it.
-		_, err := GetPodDisruptionBudget(ctx, cr.Namespace, pdbName, cl)
+		_, err := getPodDisruptionBudget(ctx, cr.Namespace, pdbName, cl)
+		if err == nil {
+			return deletePodDisruptionBudget(ctx, cr.Namespace, pdbName, cl)
+		} else if err != nil && errors.IsNotFound(err) {
+			log.FromContext(ctx).V(1).Info("Reconciliation Successful, no PodDisruptionBudget Found.")
+			// Its ok if its not found, as we're deleting anyway
+			return nil
+		}
+		return err
+	}
+}
+
+func ReconcileReplicationPodDisruptionBudget(ctx context.Context, cr *redisv1beta2.RedisReplication, pdbParams *commonapi.RedisPodDisruptionBudget, cl kubernetes.Interface) error {
+	pdbName := cr.ObjectMeta.Name + "-replication"
+	if pdbParams != nil && pdbParams.Enabled {
+		labels := getRedisLabels(cr.ObjectMeta.Name, replication, "replication", cr.GetObjectMeta().GetLabels())
+		annotations := generateStatefulSetsAnots(cr.ObjectMeta, cr.Spec.KubernetesConfig.IgnoreAnnotations)
+		pdbMeta := generateObjectMetaInformation(pdbName, cr.Namespace, labels, annotations)
+		pdbDef := generateReplicationPodDisruptionBudgetDef(ctx, cr, "replication", pdbMeta, pdbParams)
+		return CreateOrUpdatePodDisruptionBudget(ctx, pdbDef, cl)
+	} else {
+		// Check if one exists, and delete it.
+		_, err := getPodDisruptionBudget(ctx, cr.Namespace, pdbName, cl)
 		if err == nil {
 			return deletePodDisruptionBudget(ctx, cr.Namespace, pdbName, cl)
 		} else if err != nil && errors.IsNotFound(err) {
@@ -87,6 +109,33 @@ func generatePodDisruptionBudgetDef(ctx context.Context, cr *redisv1beta2.RedisC
 	return pdbTemplate
 }
 
+// generatePodDisruptionBudgetDef will create a PodDisruptionBudget definition
+func generateReplicationPodDisruptionBudgetDef(ctx context.Context, cr *redisv1beta2.RedisReplication, role string, pdbMeta metav1.ObjectMeta, pdbParams *commonapi.RedisPodDisruptionBudget) *policyv1.PodDisruptionBudget {
+	lblSelector := LabelSelectors(map[string]string{
+		"app":  fmt.Sprintf("%s-%s", cr.ObjectMeta.Name, role),
+		"role": role,
+	})
+	pdbTemplate := &policyv1.PodDisruptionBudget{
+		TypeMeta:   generateMetaInformation("PodDisruptionBudget", "policy/v1"),
+		ObjectMeta: pdbMeta,
+		Spec: policyv1.PodDisruptionBudgetSpec{
+			Selector: lblSelector,
+		},
+	}
+	if pdbParams.MinAvailable != nil {
+		pdbTemplate.Spec.MinAvailable = &intstr.IntOrString{Type: intstr.Int, IntVal: *pdbParams.MinAvailable}
+	}
+	if pdbParams.MaxUnavailable != nil {
+		pdbTemplate.Spec.MaxUnavailable = &intstr.IntOrString{Type: intstr.Int, IntVal: *pdbParams.MaxUnavailable}
+	}
+	// If we don't have a value for either, assume quorum: (N/2)+1
+	if pdbTemplate.Spec.MaxUnavailable == nil && pdbTemplate.Spec.MinAvailable == nil {
+		pdbTemplate.Spec.MinAvailable = &intstr.IntOrString{Type: intstr.Int, IntVal: (*cr.Spec.Size / 2) + 1}
+	}
+	AddOwnerRefToObject(pdbTemplate, redisReplicationAsOwner(cr))
+	return pdbTemplate
+}
+
 // generatePodDisruptionBudgetDef will create a PodDisruptionBudget definition
 func generateSentinelPodDisruptionBudgetDef(ctx context.Context, cr *redisv1beta2.RedisSentinel, role string, pdbMeta metav1.ObjectMeta, pdbParams *commonapi.RedisPodDisruptionBudget) *policyv1.PodDisruptionBudget {
 	lblSelector := LabelSelectors(map[string]string{
@@ -116,7 +165,7 @@ func generateSentinelPodDisruptionBudgetDef(ctx context.Context, cr *redisv1beta
 
 // CreateOrUpdateService method will create or update Redis service
 func CreateOrUpdatePodDisruptionBudget(ctx context.Context, pdbDef *policyv1.PodDisruptionBudget, cl kubernetes.Interface) error {
-	storedPDB, err := GetPodDisruptionBudget(ctx, pdbDef.Namespace, pdbDef.Name, cl)
+	storedPDB, err := getPodDisruptionBudget(ctx, pdbDef.Namespace, pdbDef.Name, cl)
 	if err != nil {
 		if err := patch.DefaultAnnotator.SetLastAppliedAnnotation(pdbDef); err != nil { //nolint
 			log.FromContext(ctx).Error(err, "Unable to patch redis PodDisruptionBudget with comparison object")
@@ -204,8 +253,8 @@ func deletePodDisruptionBudget(ctx context.Context, namespace string, pdbName st
 	return nil
 }
 
-// GetPodDisruptionBudget is a method to get PodDisruptionBudgets in Kubernetes
-func GetPodDisruptionBudget(ctx context.Context, namespace string, pdb string, cl kubernetes.Interface) (*policyv1.PodDisruptionBudget, error) {
+// getPodDisruptionBudget is a method to get PodDisruptionBudgets in Kubernetes
+func getPodDisruptionBudget(ctx context.Context, namespace string, pdb string, cl kubernetes.Interface) (*policyv1.PodDisruptionBudget, error) {
 	getOpts := metav1.GetOptions{
 		TypeMeta: generateMetaInformation("PodDisruptionBudget", "policy/v1"),
 	}