fix: rolling update process (#422)

Signed-off-by: Ihor Hrytskiv <[email protected]>

Author: ihor-hrytskiv

internal/controller/dragonfly_controller.go

@@ -86,7 +86,7 @@ func (r *DragonflyReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 			return ctrl.Result{}, fmt.Errorf("failed to get statefulset: %w", err)
 		}
 
-		if result, err := dfi.allPodsHealthy(ctx, statefulSet.Status.UpdateRevision); !result.IsZero() || err != nil {
+		if result, err := dfi.allPodsHealthyAndHaveRole(ctx, statefulSet.Status.UpdateRevision); !result.IsZero() || err != nil {
 			return result, err
 		}
 
@@ -95,6 +95,11 @@ func (r *DragonflyReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 			return ctrl.Result{}, fmt.Errorf("failed to get replicas: %w", err)
 		}
 
+		if len(replicas.Items) != int(dfi.df.Spec.Replicas)-1 {
+			dfi.log.Info("waiting for all replicas to be configured", "expected", int(dfi.df.Spec.Replicas)-1, "current", len(replicas.Items))
+			return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
+		}
+
 		// We want to update the replicas first then the master
 		// We want to have at most one updated replica in full sync phase at a time
 		// if not, requeue
@@ -104,7 +109,7 @@ func (r *DragonflyReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 
 		// if we are here it means that all latest replicas are in stable sync
 		// delete older version replicas
-		if result, err := dfi.updatedReplicas(ctx, replicas, statefulSet.Status.UpdateRevision); !result.IsZero() || err != nil {
+		if result, err := dfi.updateReplicas(ctx, replicas, statefulSet.Status.UpdateRevision); !result.IsZero() || err != nil {
 			return result, err
 		}
 

internal/controller/dragonfly_instance.go

@@ -748,8 +748,8 @@ func (dfi *DragonflyInstance) deleteRoleLabel(ctx context.Context, pod *corev1.P
 	return nil
 }
 
-// allPodsHealthy checks whether all pods are healthy, and deletes pods that are outdated and failed to start
-func (dfi *DragonflyInstance) allPodsHealthy(ctx context.Context, updateRevision string) (ctrl.Result, error) {
+// allPodsHealthyAndHaveRole checks whether all pods are healthy, and deletes pods that are outdated and failed to start
+func (dfi *DragonflyInstance) allPodsHealthyAndHaveRole(ctx context.Context, updateRevision string) (ctrl.Result, error) {
 	pods, err := dfi.getPods(ctx)
 	if err != nil {
 		return ctrl.Result{}, fmt.Errorf("failed to get dragonfly pods: %w", err)
@@ -773,6 +773,11 @@ func (dfi *DragonflyInstance) allPodsHealthy(ctx context.Context, updateRevision
 			dfi.log.Info("waiting for pod to finish startup", "pod", pod.Name)
 			return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
 		}
+
+		if !roleExists(&pod) {
+			dfi.log.Info("waiting for pod to be assigned a role", "pod", pod.Name)
+			return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
+		}
 	}
 
 	return ctrl.Result{}, nil
@@ -801,8 +806,12 @@ func (dfi *DragonflyInstance) verifyUpdatedReplicas(ctx context.Context, replica
 	return ctrl.Result{}, nil
 }
 
-// updatedReplicas updates the replicas to the latest version
-func (dfi *DragonflyInstance) updatedReplicas(ctx context.Context, replicas *corev1.PodList, updateRevision string) (ctrl.Result, error) {
+// updateReplicas updates the replicas to the latest version
+func (dfi *DragonflyInstance) updateReplicas(ctx context.Context, replicas *corev1.PodList, updateRevision string) (ctrl.Result, error) {
+	_, err := dfi.getMaster(ctx)
+	if err != nil {
+		return ctrl.Result{}, fmt.Errorf("failed to get master before deleting replica: %w", err)
+	}
 	for _, replica := range replicas.Items {
 		if !isPodOnLatestVersion(&replica, updateRevision) {
 			dfi.log.Info("deleting replica", "pod", replica.Name)