kea-operator/internal/controller/keacluster_controller.go

package controller

import (
	"context"
	"fmt"
	"maps"
	"slices"
	"strconv"
	"strings"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"

	v1alpha1 "git.unkin.net/unkin/kea-operator/api/v1alpha1"
	"git.unkin.net/unkin/kea-operator/internal/kea"
)

// KeaClusterReconciler reconciles a KeaCluster.
type KeaClusterReconciler struct {
	client.Client
	Scheme *runtime.Scheme
}

// +kubebuilder:rbac:groups=kea.unkin.net,resources=keaclusters,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=kea.unkin.net,resources=keaclusters/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=kea.unkin.net,resources=keasubnets,verbs=get;list;watch
// +kubebuilder:rbac:groups=kea.unkin.net,resources=keaclientclasses,verbs=get;list;watch
// +kubebuilder:rbac:groups=apps,resources=statefulsets,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=services;configmaps,verbs=get;list;watch;create;update;patch;delete

func (r *KeaClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	l := log.FromContext(ctx)

	var cluster v1alpha1.KeaCluster
	if err := r.Get(ctx, req.NamespacedName, &cluster); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	// Services first: the per-pod ClusterIP Services must exist (and have their
	// ClusterIPs allocated) before the ConfigMap is rendered, because the HA
	// peer URLs baked into kea-dhcp4.conf are those stable ClusterIPs.
	if err := r.reconcileServices(ctx, &cluster); err != nil {
		return r.fail(ctx, &cluster, "ServiceError", err)
	}
	if err := r.reconcileConfigMap(ctx, &cluster); err != nil {
		return r.fail(ctx, &cluster, "ConfigError", err)
	}
	sts, err := r.reconcileStatefulSet(ctx, &cluster)
	if err != nil {
		return r.fail(ctx, &cluster, "WorkloadError", err)
	}

	ready := sts.Status.ReadyReplicas
	desired := int32(1)
	if cluster.Spec.Replicas != nil {
		desired = *cluster.Spec.Replicas
	}

	cluster.Status.ObservedGeneration = cluster.Generation
	cluster.Status.Replicas = sts.Status.Replicas
	cluster.Status.ReadyReplicas = ready
	cluster.Status.ServiceIP = r.serviceIP(ctx, &cluster)
	if ready > 0 {
		cluster.Status.ActivePeer = cluster.Name + "-0"
	}
	if ready >= desired && desired > 0 {
		cluster.Status.Phase = "Ready"
		setReady(&cluster.Status.Conditions, cluster.Generation, true, "Ready", "all replicas ready")
	} else {
		cluster.Status.Phase = "Progressing"
		setReady(&cluster.Status.Conditions, cluster.Generation, false, "Progressing",
			fmt.Sprintf("%d/%d replicas ready", ready, desired))
	}
	if err := r.Status().Update(ctx, &cluster); err != nil {
		l.Error(err, "status update")
	}

	if cluster.Status.Phase != "Ready" {
		return ctrl.Result{RequeueAfter: requeueShort}, nil
	}
	return ctrl.Result{RequeueAfter: requeueLong}, nil
}

func (r *KeaClusterReconciler) fail(ctx context.Context, c *v1alpha1.KeaCluster, reason string, err error) (ctrl.Result, error) {
	c.Status.Phase = "Error"
	setReady(&c.Status.Conditions, c.Generation, false, reason, err.Error())
	_ = r.Status().Update(ctx, c)
	return ctrl.Result{}, err
}

// buildInput gathers matching subnets/classes and the stable HA peer list.
func (r *KeaClusterReconciler) buildInput(ctx context.Context, c *v1alpha1.KeaCluster) (kea.RenderInput, error) {
	var subnetList v1alpha1.KeaSubnetList
	if err := r.List(ctx, &subnetList, client.InNamespace(c.Namespace)); err != nil {
		return kea.RenderInput{}, err
	}
	var subnets []v1alpha1.KeaSubnet
	for _, s := range subnetList.Items {
		if s.Spec.ClusterRef == "" || s.Spec.ClusterRef == c.Name {
			subnets = append(subnets, s)
		}
	}

	var classList v1alpha1.KeaClientClassList
	if err := r.List(ctx, &classList, client.InNamespace(c.Namespace)); err != nil {
		return kea.RenderInput{}, err
	}
	var classes []v1alpha1.KeaClientClass
	for _, cc := range classList.Items {
		if cc.Spec.ClusterRef == "" || cc.Spec.ClusterRef == c.Name {
			classes = append(classes, cc)
		}
	}

	peers, err := r.peers(ctx, c)
	if err != nil {
		return kea.RenderInput{}, err
	}
	return kea.RenderInput{
		Cluster:       *c,
		Subnets:       subnets,
		ClientClasses: classes,
		Peers:         peers,
	}, nil
}

// peers builds the HA peer list. Each URL points at the peer's per-pod ClusterIP
// Service address (a stable IP literal): kea 2.6's HA hook parses the peer URL
// host as an IP and never resolves DNS, so hostnames are rejected. Using the
// stable ClusterIP (not the pod IP) also keeps the config hash stable across
// pod restarts, so the StatefulSet does not roll-loop.
func (r *KeaClusterReconciler) peers(ctx context.Context, c *v1alpha1.KeaCluster) ([]kea.Peer, error) {
	replicas := int32(1)
	if c.Spec.Replicas != nil {
		replicas = *c.Spec.Replicas
	}
	mode := c.Spec.HA.Mode
	if mode == "" {
		mode = v1alpha1.HAHotStandby
	}
	peers := make([]kea.Peer, 0, replicas)
	for i := int32(0); i < replicas; i++ {
		role := "backup"
		switch {
		case i == 0:
			role = "primary"
		case i == 1 && mode == v1alpha1.HAHotStandby:
			role = "standby"
		case i == 1:
			role = "secondary"
		}
		svcName := peerServiceName(c.Name, int(i))
		var svc corev1.Service
		if err := r.Get(ctx, types.NamespacedName{Namespace: c.Namespace, Name: svcName}, &svc); err != nil {
			return nil, fmt.Errorf("peer service %s: %w", svcName, err)
		}
		ip := svc.Spec.ClusterIP
		if ip == "" || ip == corev1.ClusterIPNone {
			return nil, fmt.Errorf("peer service %s has no ClusterIP allocated yet", svcName)
		}
		peers = append(peers, kea.Peer{
			Name: fmt.Sprintf("server%d", i),
			URL:  peerURL(ip),
			Role: role,
		})
	}
	return peers, nil
}

func (r *KeaClusterReconciler) reconcileConfigMap(ctx context.Context, c *v1alpha1.KeaCluster) error {
	in, err := r.buildInput(ctx, c)
	if err != nil {
		return err
	}
	dhcp4, err := kea.RenderDHCP4(in)
	if err != nil {
		return err
	}
	agent, err := kea.RenderCtrlAgent()
	if err != nil {
		return err
	}

	cm := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: configMapName(c.Name), Namespace: c.Namespace}}
	_, err = ctrl.CreateOrUpdate(ctx, r.Client, cm, func() error {
		cm.Labels = commonLabels(c.Name)
		cm.Data = map[string]string{
			"kea-dhcp4.conf":      dhcp4,
			"kea-ctrl-agent.conf": agent,
			"init.sh":             kea.InitScript(),
		}
		return ctrl.SetControllerReference(c, cm, r.Scheme)
	})
	return err
}

func (r *KeaClusterReconciler) reconcileServices(ctx context.Context, c *v1alpha1.KeaCluster) error {
	// Headless service for stable per-pod DNS (ctrl-agent discovery).
	headless := &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: headlessName(c.Name), Namespace: c.Namespace}}
	if _, err := ctrl.CreateOrUpdate(ctx, r.Client, headless, func() error {
		headless.Labels = mergeInto(headless.Labels, commonLabels(c.Name))
		headless.Spec.ClusterIP = corev1.ClusterIPNone
		headless.Spec.PublishNotReadyAddresses = true
		headless.Spec.Selector = commonLabels(c.Name)
		setPorts(headless, corev1.ServicePort{Name: "ctrl", Port: kea.CtrlAgentPort, TargetPort: intstrFromInt(kea.CtrlAgentPort), Protocol: corev1.ProtocolTCP})
		return ctrl.SetControllerReference(c, headless, r.Scheme)
	}); err != nil {
		return err
	}

	// Per-pod ClusterIP Services: one stable IP per HA peer. Kea's HA hook needs
	// an IP literal for each peer URL, and a ClusterIP survives pod restarts, so
	// it is safe to bake into the (roll-triggering) config hash. Not-ready
	// addresses are published so peers are routable during HA bootstrap.
	replicas := int32(1)
	if c.Spec.Replicas != nil {
		replicas = *c.Spec.Replicas
	}
	for i := int32(0); i < replicas; i++ {
		ord := int(i)
		peerSvc := &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: peerServiceName(c.Name, ord), Namespace: c.Namespace}}
		if _, err := ctrl.CreateOrUpdate(ctx, r.Client, peerSvc, func() error {
			peerSvc.Labels = mergeInto(peerSvc.Labels, commonLabels(c.Name))
			peerSvc.Spec.Type = corev1.ServiceTypeClusterIP
			peerSvc.Spec.PublishNotReadyAddresses = true
			peerSvc.Spec.Selector = map[string]string{statefulSetPodNameLabel: podName(c.Name, ord)}
			setPorts(peerSvc, corev1.ServicePort{Name: "ctrl", Port: kea.CtrlAgentPort, TargetPort: intstrFromInt(kea.CtrlAgentPort), Protocol: corev1.ProtocolTCP})
			return ctrl.SetControllerReference(c, peerSvc, r.Scheme)
		}); err != nil {
			return err
		}
	}

	// Anycast DHCP service (LoadBalancer via PureLB by default).
	svc := &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: serviceName(c.Name), Namespace: c.Namespace}}
	_, err := ctrl.CreateOrUpdate(ctx, r.Client, svc, func() error {
		svc.Labels = mergeInto(svc.Labels, commonLabels(c.Name))
		svc.Annotations = syncManagedAnnotations(svc.Annotations, mergeAnnotations(c.Spec.Service))
		svc.Spec.Selector = commonLabels(c.Name)
		svcType := c.Spec.Service.Type
		if svcType == "" {
			svcType = corev1.ServiceTypeLoadBalancer
		}
		svc.Spec.Type = svcType
		if c.Spec.Service.LoadBalancerIP != "" {
			svc.Spec.LoadBalancerIP = c.Spec.Service.LoadBalancerIP
		}
		if c.Spec.Service.LoadBalancerClass != nil {
			svc.Spec.LoadBalancerClass = c.Spec.Service.LoadBalancerClass
		}
		if svcType == corev1.ServiceTypeLoadBalancer || svcType == corev1.ServiceTypeNodePort {
			svc.Spec.ExternalTrafficPolicy = corev1.ServiceExternalTrafficPolicyLocal
		}
		setPorts(svc, corev1.ServicePort{Name: "dhcp", Port: kea.DHCP4Port, TargetPort: intstrFromInt(kea.DHCP4Port), Protocol: corev1.ProtocolUDP})
		return ctrl.SetControllerReference(c, svc, r.Scheme)
	})
	return err
}

// mergeInto sets src keys on dst without dropping keys owned by other
// controllers (e.g. Rancher's field.cattle.io/publicEndpoints).
func mergeInto(dst, src map[string]string) map[string]string {
	if dst == nil && len(src) > 0 {
		dst = make(map[string]string, len(src))
	}
	for k, v := range src {
		dst[k] = v
	}
	return dst
}

const managedAnnotationsKey = "kea.unkin.net/managed-annotations"

// syncManagedAnnotations applies want and removes keys this operator set
// previously but no longer wants, leaving foreign keys untouched.
func syncManagedAnnotations(have, want map[string]string) map[string]string {
	if have == nil {
		have = map[string]string{}
	}
	for _, k := range strings.Split(have[managedAnnotationsKey], ",") {
		if _, ok := want[k]; !ok {
			delete(have, k)
		}
	}
	delete(have, managedAnnotationsKey)
	keys := slices.Sorted(maps.Keys(want))
	for _, k := range keys {
		have[k] = want[k]
	}
	if len(keys) > 0 {
		have[managedAnnotationsKey] = strings.Join(keys, ",")
	}
	if len(have) == 0 {
		return nil
	}
	return have
}

// setPorts replaces the Service ports, keeping any apiserver-allocated
// NodePort so a steady-state reconcile is a no-op.
func setPorts(svc *corev1.Service, want ...corev1.ServicePort) {
	for i := range want {
		for _, have := range svc.Spec.Ports {
			if have.Name == want[i].Name {
				want[i].NodePort = have.NodePort
			}
		}
	}
	svc.Spec.Ports = want
}

func mergeAnnotations(s v1alpha1.ClusterServiceSpec) map[string]string {
	out := map[string]string{}
	for k, v := range s.Annotations {
		out[k] = v
	}
	if s.IPAddressPool != "" {
		out["purelb.io/service-group"] = s.IPAddressPool
	}
	if len(out) == 0 {
		return nil
	}
	return out
}

func (r *KeaClusterReconciler) reconcileStatefulSet(ctx context.Context, c *v1alpha1.KeaCluster) (*appsv1.StatefulSet, error) {
	hash, err := configHash(ctx, r.Client, c.Namespace, configMapName(c.Name))
	if err != nil {
		return nil, err
	}
	image := c.Spec.Image
	if image == "" {
		image = kea.DefaultImage
	}
	replicas := int32(1)
	if c.Spec.Replicas != nil {
		replicas = *c.Spec.Replicas
	}

	sts := &appsv1.StatefulSet{ObjectMeta: metav1.ObjectMeta{Name: stsName(c.Name), Namespace: c.Namespace}}
	_, err = ctrl.CreateOrUpdate(ctx, r.Client, sts, func() error {
		sts.Labels = commonLabels(c.Name)
		sts.Spec.ServiceName = headlessName(c.Name)
		sts.Spec.Replicas = int32ptr(replicas)
		sts.Spec.Selector = &metav1.LabelSelector{MatchLabels: commonLabels(c.Name)}
		sts.Spec.PodManagementPolicy = appsv1.ParallelPodManagement
		sts.Spec.Template = r.podTemplate(c, image, hash)
		return ctrl.SetControllerReference(c, sts, r.Scheme)
	})
	if err != nil {
		return nil, err
	}
	return sts, nil
}

func (r *KeaClusterReconciler) podTemplate(c *v1alpha1.KeaCluster, image, hash string) corev1.PodTemplateSpec {
	volProjected := corev1.Volume{
		Name: "kea-etc",
		VolumeSource: corev1.VolumeSource{
			ConfigMap: &corev1.ConfigMapVolumeSource{
				LocalObjectReference: corev1.LocalObjectReference{Name: configMapName(c.Name)},
				DefaultMode:          int32ptr(0o755),
			},
		},
	}
	volRun := corev1.Volume{Name: "run", VolumeSource: corev1.VolumeSource{EmptyDir: &corev1.EmptyDirVolumeSource{}}}

	mounts := []corev1.VolumeMount{
		{Name: "kea-etc", MountPath: kea.ConfigDir, ReadOnly: true},
		{Name: "run", MountPath: kea.RunDir},
	}

	// The initContainer finalises the per-pod config and bounded-waits for HA
	// peer DNS; the main containers then exec kea directly with no wrapper shell.
	initC := corev1.Container{
		Name:         kea.ContainerInit,
		Image:        image,
		Command:      []string{"/bin/sh", kea.InitScriptPath},
		Env:          initEnv(),
		Resources:    c.Spec.Resources,
		VolumeMounts: mounts,
	}

	dhcp4 := corev1.Container{
		Name:      kea.ContainerDHCP4,
		Image:     image,
		Command:   []string{kea.DHCP4Bin, "-c", kea.DHCP4ConfPath},
		Resources: c.Spec.Resources,
		Ports: []corev1.ContainerPort{
			{Name: "dhcp", ContainerPort: kea.DHCP4Port, Protocol: corev1.ProtocolUDP},
		},
		VolumeMounts: mounts,
	}
	agent := corev1.Container{
		Name:      kea.ContainerCtrlAgent,
		Image:     image,
		Command:   []string{kea.CtrlAgentBin, "-c", kea.CtrlAgentConfPath},
		Resources: c.Spec.Resources,
		Ports: []corev1.ContainerPort{
			{Name: "ctrl", ContainerPort: kea.CtrlAgentPort, Protocol: corev1.ProtocolTCP},
		},
		VolumeMounts: mounts,
		ReadinessProbe: &corev1.Probe{
			ProbeHandler:        corev1.ProbeHandler{TCPSocket: &corev1.TCPSocketAction{Port: intstrFromInt(kea.CtrlAgentPort)}},
			InitialDelaySeconds: 5, PeriodSeconds: 10,
		},
	}

	labels := commonLabels(c.Name)
	return corev1.PodTemplateSpec{
		ObjectMeta: metav1.ObjectMeta{
			Labels:      labels,
			Annotations: map[string]string{"kea.unkin.net/config-hash": hash},
		},
		Spec: corev1.PodSpec{
			InitContainers: []corev1.Container{initC},
			Containers:     []corev1.Container{dhcp4, agent},
			Volumes:        []corev1.Volume{volProjected, volRun},
			NodeSelector:   c.Spec.NodeSelector,
			Tolerations:    c.Spec.Tolerations,
			Affinity:       c.Spec.Affinity,
		},
	}
}

// initEnv is the environment the embedded init.sh reads. Passing paths and
// tunables as env vars (rather than interpolating them into the script text)
// keeps init.sh static, committed and shellcheck-clean.
func initEnv() []corev1.EnvVar {
	return []corev1.EnvVar{
		{Name: kea.EnvPodName, ValueFrom: &corev1.EnvVarSource{
			FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.name"},
		}},
		{Name: kea.EnvRunDir, Value: kea.RunDir},
		{Name: kea.EnvConfigDir, Value: kea.ConfigDir},
		{Name: kea.EnvThisServerPlaceholder, Value: kea.ThisServerPlaceholder},
		{Name: kea.EnvDHCP4Bin, Value: kea.DHCP4Bin},
		{Name: kea.EnvDHCP4Conf, Value: kea.DHCP4ConfPath},
		{Name: kea.EnvCtrlAgentConf, Value: kea.CtrlAgentConfPath},
		{Name: kea.EnvWaitAttempts, Value: strconv.Itoa(kea.WaitAttempts)},
		{Name: kea.EnvWaitSleep, Value: strconv.Itoa(kea.WaitSleepSeconds)},
	}
}

func (r *KeaClusterReconciler) serviceIP(ctx context.Context, c *v1alpha1.KeaCluster) string {
	var svc corev1.Service
	if err := r.Get(ctx, types.NamespacedName{Namespace: c.Namespace, Name: serviceName(c.Name)}, &svc); err != nil {
		return ""
	}
	if len(svc.Status.LoadBalancer.Ingress) > 0 {
		return svc.Status.LoadBalancer.Ingress[0].IP
	}
	return svc.Spec.ClusterIP
}

func (r *KeaClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
	mapToClusters := func(ctx context.Context, obj client.Object) []reconcile.Request {
		var list v1alpha1.KeaClusterList
		if err := r.List(ctx, &list, client.InNamespace(obj.GetNamespace())); err != nil {
			return nil
		}
		var reqs []reconcile.Request
		for _, c := range list.Items {
			reqs = append(reqs, reconcile.Request{NamespacedName: types.NamespacedName{Namespace: c.Namespace, Name: c.Name}})
		}
		return reqs
	}

	return ctrl.NewControllerManagedBy(mgr).
		For(&v1alpha1.KeaCluster{}).
		Owns(&appsv1.StatefulSet{}).
		Owns(&corev1.Service{}).
		Owns(&corev1.ConfigMap{}).
		Watches(&v1alpha1.KeaSubnet{}, handler.EnqueueRequestsFromMapFunc(mapToClusters)).
		Watches(&v1alpha1.KeaClientClass{}, handler.EnqueueRequestsFromMapFunc(mapToClusters)).
		Complete(r)
}