Implement Cluster Scaling and StatefulSet Management

Author: kvaps

internal/controller/etcdcluster_controller.go

@@ -146,7 +146,7 @@ func (r *EtcdClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 			return ctrl.Result{}, nil
 		}
 
-		return ctrl.Result{}, r.scaleUpFromZero(ctx) // TODO: needs implementing
+		return ctrl.Result{}, r.scaleUpFromZero(ctx, state) // TODO: needs implementing
 	}
 
 	// get status of every endpoint and member list from every endpoint
@@ -174,7 +174,7 @@ func (r *EtcdClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	}
 
 	if !memberReached {
-		return ctrl.Result{}, r.createOrUpdateStatefulSet(ctx)
+		return ctrl.Result{}, r.createOrUpdateStatefulSet(ctx, state)
 	}
 
 	state.setClusterID()
@@ -197,16 +197,46 @@ func (r *EtcdClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		return ctrl.Result{}, fmt.Errorf("cluster has lost quorum")
 	}
 
+	// If endpoints are not found
+	if !state.endpointsFound {
+		if !state.stsExists {
+			return r.createClusterFromScratch(ctx, state)
+		}
+
+		// Update StatefulSet pod template if necessary
+		if !state.statefulSetPodSpecCorrect() {
+			desiredSts := factory.TemplateStatefulSet()
+			state.statefulSet.Spec.Template.Spec = desiredSts.Spec.Template.Spec
+			return ctrl.Result{}, r.patchOrCreateObject(ctx, &state.statefulSet)
+		}
+
+		if !state.statefulSetReady() {
+			return ctrl.Result{}, fmt.Errorf("waiting for statefulset to become ready")
+		}
+
+		if *state.statefulSet.Spec.Replicas > 0 {
+			return ctrl.Result{}, fmt.Errorf("reached an impossible state (no endpoints, but active pods)")
+		}
+
+		if *state.instance.Spec.Replicas == 0 {
+			// Cluster successfully scaled down to zero
+			return ctrl.Result{}, nil
+		}
+
+		// **Updated function call with `state` parameter**
+		return ctrl.Result{}, r.scaleUpFromZero(ctx, state)
+	}
+
 	if state.hasLearners() {
-		return ctrl.Result{}, r.promoteLearners(ctx)
+		return ctrl.Result{}, r.promoteLearners(ctx, state)
 	}
 
-	if err := r.createOrUpdateClusterStateConfigMap(ctx); err != nil {
+	if err := r.createOrUpdateClusterStateConfigMap(ctx, state); err != nil {
 		return ctrl.Result{}, err
 	}
 
 	if !state.statefulSetPodSpecCorrect() {
-		return ctrl.Result{}, r.createOrUpdateStatefulSet(ctx)
+		return ctrl.Result{}, r.createOrUpdateStatefulSet(ctx, state)
 	}
 
 	// if size is different we have to remove statefulset it will be recreated in the next step
@@ -703,26 +733,69 @@ func (r *EtcdClusterReconciler) createClusterFromScratch(ctx context.Context, st
 	panic("not yet implemented")
 }
 
-// TODO!
-// nolint:unused
-func (r *EtcdClusterReconciler) scaleUpFromZero(ctx context.Context) error {
-	return fmt.Errorf("not yet implemented")
+// scaleUpFromZero scales the etcd cluster from zero replicas to the desired replica count.
+func (r *EtcdClusterReconciler) scaleUpFromZero(ctx context.Context, state *observables) error {
+	// Update the cluster state ConfigMap to set ETCD_INITIAL_CLUSTER_STATE to "existing"
+	if err := r.createOrUpdateClusterStateConfigMap(ctx, state); err != nil {
+		return fmt.Errorf("failed to create or update cluster state ConfigMap: %w", err)
+	}
+
+	// Update the StatefulSet to have the desired number of replicas
+	if err := r.createOrUpdateStatefulSet(ctx, state); err != nil {
+		return fmt.Errorf("failed to create or update StatefulSet: %w", err)
+	}
+
+	return nil
 }
 
-// TODO!
-// nolint:unused
-func (r *EtcdClusterReconciler) createOrUpdateClusterStateConfigMap(ctx context.Context) error {
-	return fmt.Errorf("not yet implemented")
+// createOrUpdateClusterStateConfigMap ensures the cluster state ConfigMap is up-to-date.
+func (r *EtcdClusterReconciler) createOrUpdateClusterStateConfigMap(ctx context.Context, state *observables) error {
+	// Create a ConfigMap with ETCD_INITIAL_CLUSTER_STATE set to "existing"
+	cm := factory.TemplateClusterStateConfigMap(state.instance, "existing", int(*state.instance.Spec.Replicas))
+	if err := ctrl.SetControllerReference(state.instance, cm, r.Scheme); err != nil {
+		return fmt.Errorf("cannot set controller reference: %w", err)
+	}
+
+	if err := r.patchOrCreateObject(ctx, cm); err != nil {
+		return fmt.Errorf("failed to create or update cluster state ConfigMap: %w", err)
+	}
+
+	return nil
 }
 
-// TODO!
-// nolint:unused
-func (r *EtcdClusterReconciler) createOrUpdateStatefulSet(ctx context.Context) error {
-	return fmt.Errorf("not yet implemented")
+// createOrUpdateStatefulSet ensures the StatefulSet is up-to-date with the desired state.
+func (r *EtcdClusterReconciler) createOrUpdateStatefulSet(ctx context.Context, state *observables) error {
+	if err := factory.CreateOrUpdateStatefulSet(ctx, state.instance, r.Client); err != nil {
+		return fmt.Errorf("failed to create or update StatefulSet: %w", err)
+	}
+	return nil
 }
 
-// TODO!
-// nolint:unused
-func (r *EtcdClusterReconciler) promoteLearners(ctx context.Context) error {
-	return fmt.Errorf("not yet implemented")
+// promoteLearners promotes any learner members in the etcd cluster to full voting members.
+func (r *EtcdClusterReconciler) promoteLearners(ctx context.Context, state *observables) error {
+	// Get a client to the etcd cluster
+	clusterClient, _, err := factory.NewEtcdClientSet(ctx, state.instance, r.Client)
+	if err != nil {
+		return fmt.Errorf("failed to create etcd client: %w", err)
+	}
+	defer clusterClient.Close()
+
+	// Retrieve the current member list
+	memberListResp, err := clusterClient.MemberList(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to get member list: %w", err)
+	}
+
+	// Promote any learner members to full voting members
+	for _, member := range memberListResp.Members {
+		if member.IsLearner {
+			_, err := clusterClient.MemberPromote(ctx, member.ID)
+			if err != nil {
+				return fmt.Errorf("failed to promote member %s: %w", member.Name, err)
+			}
+			log.Info(ctx, "Promoted learner to member", "member", member.Name)
+		}
+	}
+
+	return nil
 }