k3s/pkg/etcd/metadata_controller.go

package etcd

import (
	"context"
	"os"
	"strconv"
	"sync"
	"time"

	"github.com/k3s-io/k3s/pkg/util"
	controllerv1 "github.com/rancher/wrangler/v3/pkg/generated/controllers/core/v1"
	"github.com/sirupsen/logrus"
	v1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/client-go/util/retry"
	nodeutil "k8s.io/kubernetes/pkg/controller/util/node"
)

func registerMetadataHandlers(ctx context.Context, etcd *ETCD) {
	nodes := etcd.config.Runtime.Core.Core().V1().Node()
	h := &metadataHandler{
		etcd:           etcd,
		nodeController: nodes,
		ctx:            ctx,
		once:           &sync.Once{},
	}

	logrus.Infof("Starting managed etcd node metadata controller")
	nodes.OnChange(ctx, "managed-etcd-metadata-controller", h.sync)
}

type metadataHandler struct {
	etcd           *ETCD
	nodeController controllerv1.NodeController
	ctx            context.Context
	once           *sync.Once
}

func (m *metadataHandler) sync(key string, node *v1.Node) (*v1.Node, error) {
	if node == nil {
		return nil, nil
	}

	nodeName := os.Getenv("NODE_NAME")
	if nodeName == "" {
		logrus.Debug("waiting for node name to be assigned for managed etcd node metadata controller")
		m.nodeController.EnqueueAfter(key, 5*time.Second)
		return node, nil
	}

	if key == nodeName {
		return m.handleSelf(node)
	}

	return node, nil
}

// checkReset ensures that member removal annotations are cleared when the cluster is reset.
// This is done here instead of in the member controller, as the member removal controller is
// not guaranteed to run on the node that was reset.
func (m *metadataHandler) checkReset() {
	if resetDone, _ := m.etcd.IsReset(); resetDone {
		labelSelector := labels.Set{util.ETCDRoleLabelKey: "true"}.String()
		nodes, err := m.nodeController.List(metav1.ListOptions{LabelSelector: labelSelector})
		if err != nil {
			logrus.Errorf("Failed to list etcd nodes: %v", err)
			return
		}
		for _, n := range nodes.Items {
			node := &n
			err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
				_, remove := node.Annotations[removalAnnotation]
				_, removed := node.Annotations[removedNodeNameAnnotation]
				if remove || removed {
					node = node.DeepCopy()
					delete(node.Annotations, removalAnnotation)
					delete(node.Annotations, removedNodeNameAnnotation)
					node, err = m.nodeController.Update(node)
					return err
				}
				return nil
			})
			if err != nil {
				logrus.Errorf("Failed to clear removal annotations from node %s after cluster reset: %v", node.Name, err)
			} else {
				logrus.Infof("Cleared etcd member removal annotations from node %s after cluster reset", node.Name)
			}
		}
		if err := m.etcd.clearReset(); err != nil {
			logrus.Errorf("Failed to delete etcd cluster-reset file: %v", err)
		}
	}
}

func (m *metadataHandler) handleSelf(node *v1.Node) (*v1.Node, error) {
	patch := util.NewPatchList()
	patcher := util.NewPatcher[*v1.Node](m.nodeController)

	if m.etcd.config.DisableETCD {
		if node.Annotations[NodeNameAnnotation] == "" &&
			node.Annotations[NodeAddressAnnotation] == "" &&
			node.Labels[util.ETCDRoleLabelKey] == "" {
			return node, nil
		}

		if position, _ := nodeutil.GetNodeCondition(&node.Status, etcdStatusType); position >= 0 {
			patch.Remove("status", "conditions", strconv.Itoa(position))
		}
		if _, ok := node.Annotations[NodeNameAnnotation]; ok {
			patch.Remove("metadata", "annotations", NodeNameAnnotation)
		}
		if _, ok := node.Annotations[NodeAddressAnnotation]; ok {
			patch.Remove("metadata", "annotations", NodeAddressAnnotation)
		}
		if _, ok := node.Labels[util.ETCDRoleLabelKey]; ok {
			patch.Remove("metadata", "labels", util.ETCDRoleLabelKey)
		}

		return patcher.Patch(m.ctx, patch, node.Name)
	}

	m.once.Do(m.checkReset)

	if node.Annotations[NodeNameAnnotation] == m.etcd.name &&
		node.Annotations[NodeAddressAnnotation] == m.etcd.address &&
		node.Labels[util.ETCDRoleLabelKey] == "true" {
		return node, nil
	}

	patch.Add(m.etcd.name, "metadata", "annotations", NodeNameAnnotation)
	patch.Add(m.etcd.address, "metadata", "annotations", NodeAddressAnnotation)
	patch.Add("true", "metadata", "labels", util.ETCDRoleLabelKey)

	return patcher.Patch(m.ctx, patch, node.Name)
}