unkin/kea-operator
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Stop KeaCluster reconcile hot-loop #10

Merged
benvin merged 3 commits from benvin/fix-reconcile-hotloop into main 2026-10-05 23:53:46 +11:00
Conversation 2 Commits 3 Files Changed 6
+193 -134
6 changed files with 193 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/rbac/role.yaml
-8
View File
@@ -19,14 +19,6 @@ rules:
- patch
- update
- watch
- apiGroups:
- ""
resources:
- pods
verbs:
- get
- list
- watch
- apiGroups:
- apps
resources:
hack/kind/manifests/rbac.yaml
-3
View File
@@ -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"]
internal/controller/keacluster_controller.go
+63 -50
View File
@@ -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"
@@ -23,8 +26,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 +35,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 +58,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 +206,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 +228,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 +242,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 = syncManagedAnnotations(svc.Annotations, mergeAnnotations(c.Spec.Service))
svc.Spec.Selector = commonLabels(c.Name)
svcType := c.Spec.Service.Type
if svcType == "" {
@@ -264,14 +259,64 @@ 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
}
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 {
@@ -405,29 +450,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 +461,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
internal/controller/keacluster_test.go
+128
View File
@@ -315,6 +315,134 @@ 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)
}
}
}
// 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)
}
internal/controller/setup.go
+2 -5
View File
@@ -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
}
internal/kea/client.go
-68
View File
@@ -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
}