From 9f265c73d956454ce74b8fc83f49078fbbe928be Mon Sep 17 00:00:00 2001 From: Enrico Candino Date: Wed, 8 Oct 2025 13:23:15 +0200 Subject: [PATCH] Fix for HA server deletion (#493) * wip * wip * wip * removed todo --- pkg/controller/cluster/cluster.go | 3 + pkg/controller/cluster/statefulset.go | 154 +++++++++++++++++--------- tests/common_test.go | 41 ++++--- 3 files changed, 128 insertions(+), 70 deletions(-) diff --git a/pkg/controller/cluster/cluster.go b/pkg/controller/cluster/cluster.go index 36833447..d7c64358 100644 --- a/pkg/controller/cluster/cluster.go +++ b/pkg/controller/cluster/cluster.go @@ -634,6 +634,9 @@ func (c *ClusterReconciler) server(ctx context.Context, cluster *v1alpha1.Cluste return err } + // Add the finalizer to the StatefulSet so the statefulset controller can handle cleanup. + controllerutil.AddFinalizer(expectedServerStatefulSet, etcdPodFinalizerName) + currentServerStatefulSet := expectedServerStatefulSet.DeepCopy() result, err := controllerutil.CreateOrUpdate(ctx, c.Client, currentServerStatefulSet, func() error { if err := controllerutil.SetControllerReference(cluster, currentServerStatefulSet, c.Scheme); err != nil { diff --git a/pkg/controller/cluster/statefulset.go b/pkg/controller/cluster/statefulset.go index aa5e42f3..b87ef717 100644 --- a/pkg/controller/cluster/statefulset.go +++ b/pkg/controller/cluster/statefulset.go @@ -23,6 +23,7 @@ import ( apps "k8s.io/api/apps/v1" v1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ctrl "sigs.k8s.io/controller-runtime" ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client" @@ -63,35 +64,45 @@ func (p *StatefulSetReconciler) Reconcile(ctx context.Context, req reconcile.Req log := ctrl.LoggerFrom(ctx) log.Info("reconciling statefulset") - s := strings.Split(req.Name, "-") - if len(s) < 1 { - return reconcile.Result{}, nil + var sts apps.StatefulSet + if err := p.Client.Get(ctx, req.NamespacedName, &sts); err != nil { + // we can ignore the IsNotFound error + // if the stateful set was deleted we have already cleaned up the pods + return reconcile.Result{}, ctrlruntimeclient.IgnoreNotFound(err) } - if s[0] != "k3k" { - return reconcile.Result{}, nil + // If the StatefulSet is being deleted, we need to remove the finalizers from its pods + // and remove the finalizer from the StatefulSet itself. + if !sts.DeletionTimestamp.IsZero() { + return p.handleDeletion(ctx, &sts) } - clusterName := s[1] + // get cluster name from the object + clusterKey := clusterNamespacedName(&sts) var cluster v1alpha1.Cluster - if err := p.Client.Get(ctx, types.NamespacedName{Name: clusterName, Namespace: req.Namespace}, &cluster); err != nil { + if err := p.Client.Get(ctx, clusterKey, &cluster); err != nil { if !apierrors.IsNotFound(err) { return reconcile.Result{}, err } } - matchingLabels := ctrlruntimeclient.MatchingLabels(map[string]string{"role": "server"}) - listOpts := &ctrlruntimeclient.ListOptions{Namespace: req.Namespace} - matchingLabels.ApplyToList(listOpts) - - var podList v1.PodList - if err := p.Client.List(ctx, &podList, listOpts); err != nil { - return reconcile.Result{}, ctrlruntimeclient.IgnoreNotFound(err) + podList, err := p.listPods(ctx, &sts) + if err != nil { + return reconcile.Result{}, err } if len(podList.Items) == 1 { - return reconcile.Result{}, nil + serverPod := podList.Items[0] + if !serverPod.DeletionTimestamp.IsZero() { + if controllerutil.RemoveFinalizer(&serverPod, etcdPodFinalizerName) { + if err := p.Client.Update(ctx, &serverPod); err != nil { + return reconcile.Result{}, err + } + } + + return reconcile.Result{}, nil + } } for _, pod := range podList.Items { @@ -107,59 +118,52 @@ func (p *StatefulSetReconciler) handleServerPod(ctx context.Context, cluster v1a log := ctrl.LoggerFrom(ctx) log.Info("handling server pod") - role, found := pod.Labels["role"] - if !found { - return fmt.Errorf("server pod has no role label") - } + if pod.DeletionTimestamp.IsZero() { + if controllerutil.AddFinalizer(pod, etcdPodFinalizerName) { + return p.Client.Update(ctx, pod) + } - if role != "server" { - log.V(1).Info("pod has a different role: " + role) return nil } // if etcd pod is marked for deletion then we need to remove it from the etcd member list before deletion - if !pod.DeletionTimestamp.IsZero() { - // check if cluster is deleted then remove the finalizer from the pod - if cluster.Name == "" { - if controllerutil.RemoveFinalizer(pod, etcdPodFinalizerName) { - if err := p.Client.Update(ctx, pod); err != nil { - return err - } - } - return nil - } - - tlsConfig, err := p.getETCDTLS(ctx, &cluster) - if err != nil { - return err - } - - // remove server from etcd - client, err := clientv3.New(clientv3.Config{ - Endpoints: []string{ - fmt.Sprintf("https://%s.%s:2379", server.ServiceName(cluster.Name), pod.Namespace), - }, - TLS: tlsConfig, - }) - if err != nil { - return err - } - - if err := removePeer(ctx, client, pod.Name, pod.Status.PodIP); err != nil { - return err - } - - // remove our finalizer from the list and update it. + // check if cluster is deleted then remove the finalizer from the pod + if cluster.Name == "" { if controllerutil.RemoveFinalizer(pod, etcdPodFinalizerName) { if err := p.Client.Update(ctx, pod); err != nil { return err } } + + return nil } - if controllerutil.AddFinalizer(pod, etcdPodFinalizerName) { - return p.Client.Update(ctx, pod) + tlsConfig, err := p.getETCDTLS(ctx, &cluster) + if err != nil { + return err + } + + // remove server from etcd + client, err := clientv3.New(clientv3.Config{ + Endpoints: []string{ + fmt.Sprintf("https://%s.%s:2379", server.ServiceName(cluster.Name), pod.Namespace), + }, + TLS: tlsConfig, + }) + if err != nil { + return err + } + + if err := removePeer(ctx, client, pod.Name, pod.Status.PodIP); err != nil { + return err + } + + // remove our finalizer from the list and update it. + if controllerutil.RemoveFinalizer(pod, etcdPodFinalizerName) { + if err := p.Client.Update(ctx, pod); err != nil { + return err + } } return nil @@ -274,3 +278,43 @@ func (p *StatefulSetReconciler) clusterToken(ctx context.Context, cluster *v1alp return string(tokenSecret.Data["token"]), nil } + +func (p *StatefulSetReconciler) handleDeletion(ctx context.Context, sts *apps.StatefulSet) (ctrl.Result, error) { + podList, err := p.listPods(ctx, sts) + if err != nil { + return reconcile.Result{}, err + } + + for _, pod := range podList.Items { + if controllerutil.RemoveFinalizer(&pod, etcdPodFinalizerName) { + if err := p.Client.Update(ctx, &pod); err != nil { + return reconcile.Result{}, err + } + } + } + + if controllerutil.RemoveFinalizer(sts, etcdPodFinalizerName) { + return reconcile.Result{}, p.Client.Update(ctx, sts) + } + + return reconcile.Result{}, nil +} + +func (p *StatefulSetReconciler) listPods(ctx context.Context, sts *apps.StatefulSet) (*v1.PodList, error) { + selector, err := metav1.LabelSelectorAsSelector(sts.Spec.Selector) + if err != nil { + return nil, fmt.Errorf("failed to create selector from statefulset: %w", err) + } + + listOpts := &ctrlruntimeclient.ListOptions{ + Namespace: sts.Namespace, + LabelSelector: selector, + } + + var podList v1.PodList + if err := p.Client.List(ctx, &podList, listOpts); err != nil { + return nil, ctrlruntimeclient.IgnoreNotFound(err) + } + + return &podList, nil +} diff --git a/tests/common_test.go b/tests/common_test.go index dd0f2dd6..d6a2dd39 100644 --- a/tests/common_test.go +++ b/tests/common_test.go @@ -150,34 +150,45 @@ func CreateCluster(cluster *v1alpha1.Cluster) { err := k8sClient.Create(ctx, cluster) Expect(err).To(Not(HaveOccurred())) + By("Waiting for cluster to be ready") + // check that the server Pod and the Kubelet are in Ready state Eventually(func() bool { podList, err := k8s.CoreV1().Pods(cluster.Namespace).List(ctx, v1.ListOptions{}) Expect(err).To(Not(HaveOccurred())) - serverRunning := false - agentRunning := false + // all the servers and agents needs to be in a running phase + var serversReady, agentsReady int for _, pod := range podList.Items { - if pod.Labels == nil { - continue - } - roleLabel := pod.Labels["role"] - typeLabel := pod.Labels["type"] - if roleLabel == "server" { - serverRunning = pod.Status.Phase == corev1.PodRunning - } else if typeLabel == "agent" { - agentRunning = pod.Status.Phase == corev1.PodRunning + if pod.Labels["role"] == "server" { + GinkgoLogr.Info(fmt.Sprintf("server pod=%s/%s status=%s", pod.Namespace, pod.Name, pod.Status.Phase)) + if pod.Status.Phase == corev1.PodRunning { + serversReady++ + } } - if serverRunning && agentRunning { - return true + if pod.Labels["type"] == "agent" { + GinkgoLogr.Info(fmt.Sprintf("agent pod=%s/%s status=%s", pod.Namespace, pod.Name, pod.Status.Phase)) + if pod.Status.Phase == corev1.PodRunning { + agentsReady++ + } } } - return false + expectedServers := int(*cluster.Spec.Servers) + expectedAgents := int(*cluster.Spec.Agents) + + By(fmt.Sprintf("serversReady=%d/%d agentsReady=%d/%d", serversReady, expectedServers, agentsReady, expectedAgents)) + + // the server pods should equal the expected servers, but since in shared mode we also have the kubelet is fine to have more than one + if (serversReady != expectedServers) || (agentsReady < expectedAgents) { + return false + } + + return true }). - WithTimeout(time.Minute * 2). + WithTimeout(time.Minute * 5). WithPolling(time.Second * 5). Should(BeTrue()) }