From 682b6b64befaf654ebbb48ee873b66cc43af3108 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Mon, 5 Oct 2026 22:15:09 +1100 Subject: [PATCH 1/3] Stop KeaCluster reconcile hot-loop Merge Service labels/annotations instead of replacing them, keep allocated NodePorts and set TargetPort so steady-state reconciles are no-ops, and drop the per-reconcile config-reload (running pods never see new config; a config change rolls the StatefulSet). --- config/rbac/role.yaml | 8 -- internal/controller/keacluster_controller.go | 83 ++++++++------------ internal/controller/keacluster_test.go | 70 +++++++++++++++++ internal/controller/setup.go | 7 +- internal/kea/client.go | 68 ---------------- 5 files changed, 105 insertions(+), 131 deletions(-) delete mode 100644 internal/kea/client.go diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml index 5737fe3..ec31023 100644 --- a/config/rbac/role.yaml +++ b/config/rbac/role.yaml @@ -19,14 +19,6 @@ rules: - patch - update - watch -- apiGroups: - - "" - resources: - - pods - verbs: - - get - - list - - watch - apiGroups: - apps resources: diff --git a/internal/controller/keacluster_controller.go b/internal/controller/keacluster_controller.go index adae2b1..5bcf614 100644 --- a/internal/controller/keacluster_controller.go +++ b/internal/controller/keacluster_controller.go @@ -23,8 +23,7 @@ import ( // KeaClusterReconciler reconciles a KeaCluster. type KeaClusterReconciler struct { client.Client - Scheme *runtime.Scheme - Control *kea.ControlClient + Scheme *runtime.Scheme } // +kubebuilder:rbac:groups=kea.unkin.net,resources=keaclusters,verbs=get;list;watch;create;update;patch;delete @@ -33,7 +32,6 @@ type KeaClusterReconciler struct { // +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 -// +kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch func (r *KeaClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { l := log.FromContext(ctx) @@ -57,8 +55,6 @@ func (r *KeaClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) return r.fail(ctx, &cluster, "WorkloadError", err) } - r.reloadReadyPods(ctx, &cluster) - ready := sts.Status.ReadyReplicas desired := int32(1) if cluster.Spec.Replicas != nil { @@ -207,13 +203,11 @@ func (r *KeaClusterReconciler) reconcileServices(ctx context.Context, c *v1alpha // 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 = commonLabels(c.Name) + headless.Labels = mergeInto(headless.Labels, commonLabels(c.Name)) headless.Spec.ClusterIP = corev1.ClusterIPNone headless.Spec.PublishNotReadyAddresses = true headless.Spec.Selector = commonLabels(c.Name) - headless.Spec.Ports = []corev1.ServicePort{ - {Name: "ctrl", Port: kea.CtrlAgentPort, Protocol: corev1.ProtocolTCP}, - } + 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 @@ -231,13 +225,11 @@ func (r *KeaClusterReconciler) reconcileServices(ctx context.Context, c *v1alpha 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 = commonLabels(c.Name) + 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)} - peerSvc.Spec.Ports = []corev1.ServicePort{ - {Name: "ctrl", Port: kea.CtrlAgentPort, TargetPort: intstrFromInt(kea.CtrlAgentPort), Protocol: corev1.ProtocolTCP}, - } + 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 @@ -247,8 +239,8 @@ func (r *KeaClusterReconciler) reconcileServices(ctx context.Context, c *v1alpha // 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 = commonLabels(c.Name) - svc.Annotations = mergeAnnotations(c.Spec.Service) + svc.Labels = mergeInto(svc.Labels, commonLabels(c.Name)) + svc.Annotations = mergeInto(svc.Annotations, mergeAnnotations(c.Spec.Service)) svc.Spec.Selector = commonLabels(c.Name) svcType := c.Spec.Service.Type if svcType == "" { @@ -264,14 +256,37 @@ func (r *KeaClusterReconciler) reconcileServices(ctx context.Context, c *v1alpha if svcType == corev1.ServiceTypeLoadBalancer || svcType == corev1.ServiceTypeNodePort { svc.Spec.ExternalTrafficPolicy = corev1.ServiceExternalTrafficPolicyLocal } - svc.Spec.Ports = []corev1.ServicePort{ - {Name: "dhcp", Port: kea.DHCP4Port, Protocol: corev1.ProtocolUDP}, - } + 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 +} + +// 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 { @@ -405,29 +420,6 @@ func initEnv() []corev1.EnvVar { } } -// reloadReadyPods best-effort hot-reloads config on ready pods via the -// ctrl-agent REST channel, analogous to bind-operator's rndc reconfig. -func (r *KeaClusterReconciler) reloadReadyPods(ctx context.Context, c *v1alpha1.KeaCluster) { - if r.Control == nil { - return - } - var pods corev1.PodList - if err := r.List(ctx, &pods, client.InNamespace(c.Namespace), client.MatchingLabels(commonLabels(c.Name))); err != nil { - return - } - l := log.FromContext(ctx) - for i := range pods.Items { - p := &pods.Items[i] - if p.Status.PodIP == "" || !podReady(p) { - continue - } - url := fmt.Sprintf("http://%s:%d/", p.Status.PodIP, kea.CtrlAgentPort) - if err := r.Control.ConfigReload(ctx, url); err != nil { - l.V(1).Info("config-reload failed", "pod", p.Name, "err", err.Error()) - } - } -} - 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 { @@ -439,15 +431,6 @@ func (r *KeaClusterReconciler) serviceIP(ctx context.Context, c *v1alpha1.KeaClu return svc.Spec.ClusterIP } -func podReady(p *corev1.Pod) bool { - for _, cond := range p.Status.Conditions { - if cond.Type == corev1.PodReady { - return cond.Status == corev1.ConditionTrue - } - } - return false -} - func (r *KeaClusterReconciler) SetupWithManager(mgr ctrl.Manager) error { mapToClusters := func(ctx context.Context, obj client.Object) []reconcile.Request { var list v1alpha1.KeaClusterList diff --git a/internal/controller/keacluster_test.go b/internal/controller/keacluster_test.go index d1afe07..ad63cd0 100644 --- a/internal/controller/keacluster_test.go +++ b/internal/controller/keacluster_test.go @@ -315,6 +315,76 @@ func TestClusterRefFiltersSubnets(t *testing.T) { } } +// TestSteadyStateReconcileLeavesServicesAlone guards the reconcile hot-loop: +// once the apiserver has defaulted the Services (NodePort, TargetPort) and +// another controller has annotated one, a no-change reconcile must not write +// any Service, since each write re-triggers the Owns(Service) watch. +func TestSteadyStateReconcileLeavesServicesAlone(t *testing.T) { + scheme := testScheme(t) + cluster := newClusterFixture() + cluster.Spec.Service.IPAddressPool = "dhcp" + cl := fake.NewClientBuilder(). + WithScheme(scheme). + WithStatusSubresource(&v1alpha1.KeaCluster{}). + WithObjects(withPeers(cluster)...). + Build() + r := &KeaClusterReconciler{Client: cl, Scheme: scheme} + ctx := context.Background() + req := ctrl.Request{NamespacedName: types.NamespacedName{Namespace: "dhcp-system", Name: "pxe"}} + if _, err := r.Reconcile(ctx, req); err != nil { + t.Fatalf("first reconcile: %v", err) + } + + names := []string{"pxe", "pxe-headless", "pxe-peer-0", "pxe-peer-1"} + const foreign = "field.cattle.io/publicEndpoints" + versions := map[string]string{} + for _, name := range names { + var svc corev1.Service + if err := cl.Get(ctx, types.NamespacedName{Namespace: "dhcp-system", Name: name}, &svc); err != nil { + t.Fatal(err) + } + if svc.Spec.Ports[0].TargetPort.IntVal == 0 { + svc.Spec.Ports[0].TargetPort = intstrFromInt(int(svc.Spec.Ports[0].Port)) + } + if name == "pxe" { + svc.Spec.Ports[0].NodePort = 31067 + if svc.Annotations == nil { + svc.Annotations = map[string]string{} + } + svc.Annotations[foreign] = "[]" + } + if err := cl.Update(ctx, &svc); err != nil { + t.Fatal(err) + } + versions[name] = svc.ResourceVersion + } + + if _, err := r.Reconcile(ctx, req); err != nil { + t.Fatalf("second reconcile: %v", err) + } + for _, name := range names { + var svc corev1.Service + if err := cl.Get(ctx, types.NamespacedName{Namespace: "dhcp-system", Name: name}, &svc); err != nil { + t.Fatal(err) + } + if svc.ResourceVersion != versions[name] { + t.Errorf("service %s updated on a no-change reconcile", name) + } + if name != "pxe" { + continue + } + if svc.Annotations[foreign] != "[]" { + t.Errorf("foreign annotation %s dropped: %v", foreign, svc.Annotations) + } + if svc.Annotations["purelb.io/service-group"] != "dhcp" { + t.Errorf("operator annotation missing: %v", svc.Annotations) + } + if svc.Spec.Ports[0].NodePort != 31067 { + t.Errorf("allocated NodePort not preserved: %d", svc.Spec.Ports[0].NodePort) + } + } +} + func contains(hay, needle string) bool { return len(hay) >= len(needle) && (indexOf(hay, needle) >= 0) } diff --git a/internal/controller/setup.go b/internal/controller/setup.go index c4e8f3a..edde059 100644 --- a/internal/controller/setup.go +++ b/internal/controller/setup.go @@ -2,16 +2,13 @@ package controller import ( ctrl "sigs.k8s.io/controller-runtime" - - "git.unkin.net/unkin/kea-operator/internal/kea" ) // SetupAll wires every reconciler into the manager. func SetupAll(mgr ctrl.Manager) error { if err := (&KeaClusterReconciler{ - Client: mgr.GetClient(), - Scheme: mgr.GetScheme(), - Control: kea.NewControlClient(), + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), }).SetupWithManager(mgr); err != nil { return err } diff --git a/internal/kea/client.go b/internal/kea/client.go deleted file mode 100644 index 46b2972..0000000 --- a/internal/kea/client.go +++ /dev/null @@ -1,68 +0,0 @@ -package kea - -import ( - "bytes" - "context" - "encoding/json" - "fmt" - "net/http" - "time" -) - -// ControlClient talks to a kea-ctrl-agent REST endpoint. -type ControlClient struct { - HTTP *http.Client -} - -// NewControlClient returns a ControlClient with a bounded timeout. -func NewControlClient() *ControlClient { - return &ControlClient{HTTP: &http.Client{Timeout: 5 * time.Second}} -} - -type command struct { - Command string `json:"command"` - Service []string `json:"service,omitempty"` - Arguments any `json:"arguments,omitempty"` -} - -type response struct { - Result int `json:"result"` - Text string `json:"text"` -} - -// ConfigReload asks the dhcp4 server behind the agent at baseURL to re-read its -// config file from disk (the hot-reload path, analogous to rndc reconfig). -func (c *ControlClient) ConfigReload(ctx context.Context, baseURL string) error { - return c.send(ctx, baseURL, command{Command: "config-reload", Service: []string{"dhcp4"}}) -} - -func (c *ControlClient) send(ctx context.Context, baseURL string, cmd command) error { - body, err := json.Marshal(cmd) - if err != nil { - return err - } - req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL, bytes.NewReader(body)) - if err != nil { - return err - } - req.Header.Set("Content-Type", "application/json") - - resp, err := c.HTTP.Do(req) - if err != nil { - return err - } - defer resp.Body.Close() - if resp.StatusCode != http.StatusOK { - return fmt.Errorf("kea control %q: http %d", cmd.Command, resp.StatusCode) - } - var results []response - if err := json.NewDecoder(resp.Body).Decode(&results); err != nil { - return fmt.Errorf("decode kea control response: %w", err) - } - for _, r := range results { - if r.Result != 0 { - return fmt.Errorf("kea control %q failed: %s", cmd.Command, r.Text) - } - } - return nil -} -- 2.47.3 From 703eeb0eed47a8deb449168212ca77da86f29a61 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Mon, 5 Oct 2026 22:21:00 +1100 Subject: [PATCH 2/3] Remove stale operator-owned Service annotations --- internal/controller/keacluster_controller.go | 32 ++++++++++- internal/controller/keacluster_test.go | 58 ++++++++++++++++++++ 2 files changed, 89 insertions(+), 1 deletion(-) diff --git a/internal/controller/keacluster_controller.go b/internal/controller/keacluster_controller.go index 5bcf614..1e8e421 100644 --- a/internal/controller/keacluster_controller.go +++ b/internal/controller/keacluster_controller.go @@ -3,7 +3,10 @@ package controller import ( "context" "fmt" + "maps" + "slices" "strconv" + "strings" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" @@ -240,7 +243,7 @@ func (r *KeaClusterReconciler) reconcileServices(ctx context.Context, c *v1alpha 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 = mergeInto(svc.Annotations, mergeAnnotations(c.Spec.Service)) + svc.Annotations = syncManagedAnnotations(svc.Annotations, mergeAnnotations(c.Spec.Service)) svc.Spec.Selector = commonLabels(c.Name) svcType := c.Spec.Service.Type if svcType == "" { @@ -274,6 +277,33 @@ func mergeInto(dst, src map[string]string) map[string]string { 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) { diff --git a/internal/controller/keacluster_test.go b/internal/controller/keacluster_test.go index ad63cd0..11583fb 100644 --- a/internal/controller/keacluster_test.go +++ b/internal/controller/keacluster_test.go @@ -385,6 +385,64 @@ func TestSteadyStateReconcileLeavesServicesAlone(t *testing.T) { } } +// TestRemovedServiceAnnotationIsDropped ensures annotations dropped from the +// spec are removed from the Service while foreign annotations survive. +func TestRemovedServiceAnnotationIsDropped(t *testing.T) { + scheme := testScheme(t) + cluster := newClusterFixture() + cluster.Spec.Service.IPAddressPool = "dhcp" + cluster.Spec.Service.Annotations = map[string]string{"example.com/x": "1"} + cl := fake.NewClientBuilder(). + WithScheme(scheme). + WithStatusSubresource(&v1alpha1.KeaCluster{}). + WithObjects(withPeers(cluster)...). + Build() + r := &KeaClusterReconciler{Client: cl, Scheme: scheme} + ctx := context.Background() + key := types.NamespacedName{Namespace: "dhcp-system", Name: "pxe"} + req := ctrl.Request{NamespacedName: key} + if _, err := r.Reconcile(ctx, req); err != nil { + t.Fatalf("first reconcile: %v", err) + } + + const foreign = "field.cattle.io/publicEndpoints" + var svc corev1.Service + if err := cl.Get(ctx, key, &svc); err != nil { + t.Fatal(err) + } + if svc.Annotations["example.com/x"] != "1" || svc.Annotations["purelb.io/service-group"] != "dhcp" { + t.Fatalf("operator annotations missing: %v", svc.Annotations) + } + svc.Annotations[foreign] = "[]" + if err := cl.Update(ctx, &svc); err != nil { + t.Fatal(err) + } + + var kc v1alpha1.KeaCluster + if err := cl.Get(ctx, key, &kc); err != nil { + t.Fatal(err) + } + kc.Spec.Service.Annotations = nil + kc.Spec.Service.IPAddressPool = "" + if err := cl.Update(ctx, &kc); err != nil { + t.Fatal(err) + } + if _, err := r.Reconcile(ctx, req); err != nil { + t.Fatalf("second reconcile: %v", err) + } + if err := cl.Get(ctx, key, &svc); err != nil { + t.Fatal(err) + } + for _, k := range []string{"example.com/x", "purelb.io/service-group", managedAnnotationsKey} { + if _, ok := svc.Annotations[k]; ok { + t.Errorf("stale annotation %s kept: %v", k, svc.Annotations) + } + } + if svc.Annotations[foreign] != "[]" { + t.Errorf("foreign annotation dropped: %v", svc.Annotations) + } +} + func contains(hay, needle string) bool { return len(hay) >= len(needle) && (indexOf(hay, needle) >= 0) } -- 2.47.3 From ab0f810c07868c413ee6e26b15b6841ba1036ac2 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Mon, 5 Oct 2026 22:21:00 +1100 Subject: [PATCH 3/3] Drop unused pods grant from kind RBAC --- hack/kind/manifests/rbac.yaml | 3 --- 1 file changed, 3 deletions(-) diff --git a/hack/kind/manifests/rbac.yaml b/hack/kind/manifests/rbac.yaml index af9753e..9c7a291 100644 --- a/hack/kind/manifests/rbac.yaml +++ b/hack/kind/manifests/rbac.yaml @@ -16,9 +16,6 @@ rules: - apiGroups: [""] resources: ["services", "configmaps", "secrets", "serviceaccounts"] verbs: ["get", "list", "watch", "create", "update", "patch", "delete"] - - apiGroups: [""] - resources: ["pods"] - verbs: ["get", "list", "watch"] - apiGroups: ["apps"] resources: ["statefulsets", "deployments"] verbs: ["get", "list", "watch", "create", "update", "patch", "delete"] -- 2.47.3