Point HA peer URLs at per-pod ClusterIP Services #7

Merged
benvin merged 1 commits from benvin/ha-peer-clusterip into main 2026-08-09 19:07:05 +10:00
3 changed files with 149 additions and 17 deletions
+16 -2
View File
@@ -29,6 +29,10 @@ const (
clusterLabel = "kea.unkin.net/cluster"
roleLabel = "kea.unkin.net/role"
// statefulSetPodNameLabel is stamped on every StatefulSet pod by Kubernetes;
// the per-pod ClusterIP Service selects a single pod through it.
statefulSetPodNameLabel = "statefulset.kubernetes.io/pod-name"
finalizer = "kea.unkin.net/finalizer"
defaultOperatorImage = "git.unkin.net/unkin/kea-operator:latest"
@@ -39,10 +43,20 @@ func headlessName(cluster string) string { return cluster + "-headless" }
func serviceName(cluster string) string { return cluster }
func configMapName(cluster string) string { return cluster + "-config" }
func stsName(cluster string) string { return cluster }
func peerDNS(cluster, ns string, ordinal int) string {
return fmt.Sprintf("http://%s-%d.%s.%s:%d/", cluster, ordinal, headlessName(cluster), ns, 8000)
// podName is the StatefulSet pod name for an ordinal.
func podName(cluster string, ordinal int) string { return fmt.Sprintf("%s-%d", cluster, ordinal) }
// peerServiceName is the per-pod ClusterIP Service fronting one HA peer's
// ctrl-agent. Its stable ClusterIP is what the HA hook peer URL points at.
func peerServiceName(cluster string, ordinal int) string {
return fmt.Sprintf("%s-peer-%d", cluster, ordinal)
}
// peerURL builds an HA peer control-channel URL. Kea 2.6's HA hook parses the
// host as an IP literal (no DNS), so this must be a stable IP address.
func peerURL(ip string) string { return fmt.Sprintf("http://%s:%d/", ip, 8000) }
func commonLabels(cluster string) map[string]string {
return map[string]string{
managedByLabel: managedByValue,
+54 -9
View File
@@ -43,12 +43,15 @@ func (r *KeaClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request)
return ctrl.Result{}, client.IgnoreNotFound(err)
}
if err := r.reconcileConfigMap(ctx, &cluster); err != nil {
return r.fail(ctx, &cluster, "ConfigError", 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)
@@ -118,16 +121,24 @@ func (r *KeaClusterReconciler) buildInput(ctx context.Context, c *v1alpha1.KeaCl
}
}
peers, err := r.peers(ctx, c)
if err != nil {
return kea.RenderInput{}, err
}
return kea.RenderInput{
Cluster: *c,
Subnets: subnets,
ClientClasses: classes,
Peers: r.peers(c),
Peers: peers,
}, nil
}
// peers returns stable HA peer identities (DNS only, no pod IPs).
func (r *KeaClusterReconciler) peers(c *v1alpha1.KeaCluster) []kea.Peer {
// 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
@@ -147,13 +158,22 @@ func (r *KeaClusterReconciler) peers(c *v1alpha1.KeaCluster) []kea.Peer {
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: peerDNS(c.Name, c.Namespace, int(i)),
URL: peerURL(ip),
Role: role,
})
}
return peers
return peers, nil
}
func (r *KeaClusterReconciler) reconcileConfigMap(ctx context.Context, c *v1alpha1.KeaCluster) error {
@@ -184,7 +204,7 @@ func (r *KeaClusterReconciler) reconcileConfigMap(ctx context.Context, c *v1alph
}
func (r *KeaClusterReconciler) reconcileServices(ctx context.Context, c *v1alpha1.KeaCluster) error {
// Headless service for stable per-pod DNS (HA peer URLs, ctrl-agent).
// 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)
@@ -199,6 +219,31 @@ func (r *KeaClusterReconciler) reconcileServices(ctx context.Context, c *v1alpha
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 = 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},
}
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 {
+79 -6
View File
@@ -40,6 +40,79 @@ func newClusterFixture() *v1alpha1.KeaCluster {
}
}
// peerServiceFixtures stands in for the per-pod ClusterIP Services with the
// ClusterIPs the apiserver would allocate (the fake client does not allocate),
// so peers() can read them when rendering the HA config.
func peerServiceFixtures() []client.Object {
return []client.Object{
&corev1.Service{
ObjectMeta: metav1.ObjectMeta{Name: "pxe-peer-0", Namespace: "dhcp-system"},
Spec: corev1.ServiceSpec{Type: corev1.ServiceTypeClusterIP, ClusterIP: "10.96.10.10"},
},
&corev1.Service{
ObjectMeta: metav1.ObjectMeta{Name: "pxe-peer-1", Namespace: "dhcp-system"},
Spec: corev1.ServiceSpec{Type: corev1.ServiceTypeClusterIP, ClusterIP: "10.96.10.11"},
},
}
}
func withPeers(objs ...client.Object) []client.Object {
return append(objs, peerServiceFixtures()...)
}
// TestPeerURLsUseClusterIPs is the regression guard for the HA bootstrap fix:
// kea 2.6's HA hook rejects hostnames, so the rendered peer URLs must be the
// per-pod ClusterIP literals, never the headless DNS names.
func TestPeerURLsUseClusterIPs(t *testing.T) {
scheme := testScheme(t)
cl := fake.NewClientBuilder().
WithScheme(scheme).
WithStatusSubresource(&v1alpha1.KeaCluster{}).
WithObjects(withPeers(newClusterFixture())...).
Build()
r := &KeaClusterReconciler{Client: cl, Scheme: scheme}
if _, err := r.Reconcile(context.Background(), ctrl.Request{NamespacedName: types.NamespacedName{Namespace: "dhcp-system", Name: "pxe"}}); err != nil {
t.Fatalf("reconcile: %v", err)
}
var cm corev1.ConfigMap
if err := cl.Get(context.Background(), types.NamespacedName{Namespace: "dhcp-system", Name: "pxe-config"}, &cm); err != nil {
t.Fatalf("configmap not created: %v", err)
}
conf := cm.Data["kea-dhcp4.conf"]
for _, want := range []string{"http://10.96.10.10:8000/", "http://10.96.10.11:8000/"} {
if !contains(conf, want) {
t.Errorf("HA peer URL %q missing from rendered config", want)
}
}
if contains(conf, "kea-headless") {
t.Errorf("HA peer URLs must not use headless DNS names (kea's HA hook rejects hostnames)")
}
// The per-pod ClusterIP Services must exist.
for _, name := range []string{"pxe-peer-0", "pxe-peer-1"} {
var svc corev1.Service
if err := cl.Get(context.Background(), types.NamespacedName{Namespace: "dhcp-system", Name: name}, &svc); err != nil {
t.Errorf("per-pod service %s not present: %v", name, err)
}
}
}
// TestPeersRequeueWithoutClusterIP proves the render blocks (returns an error to
// requeue) until the peer ClusterIPs are allocated, rather than emitting a
// hostname the HA hook would reject.
func TestPeersRequeueWithoutClusterIP(t *testing.T) {
scheme := testScheme(t)
cl := fake.NewClientBuilder().
WithScheme(scheme).
WithStatusSubresource(&v1alpha1.KeaCluster{}).
WithObjects(newClusterFixture()).
Build()
r := &KeaClusterReconciler{Client: cl, Scheme: scheme}
if _, err := r.Reconcile(context.Background(), ctrl.Request{NamespacedName: types.NamespacedName{Namespace: "dhcp-system", Name: "pxe"}}); err == nil {
t.Fatal("expected reconcile to error while peer ClusterIPs are unallocated")
}
}
func TestKeaClusterReconcileCreatesWorkload(t *testing.T) {
scheme := testScheme(t)
cluster := newClusterFixture()
@@ -50,7 +123,7 @@ func TestKeaClusterReconcileCreatesWorkload(t *testing.T) {
cl := fake.NewClientBuilder().
WithScheme(scheme).
WithStatusSubresource(&v1alpha1.KeaCluster{}).
WithObjects(cluster, subnet).
WithObjects(withPeers(cluster, subnet)...).
Build()
r := &KeaClusterReconciler{Client: cl, Scheme: scheme}
@@ -82,8 +155,8 @@ func TestKeaClusterReconcileCreatesWorkload(t *testing.T) {
t.Errorf("missing config-hash annotation")
}
// Anycast + headless services.
for _, name := range []string{"pxe", "pxe-headless"} {
// Anycast + headless + per-pod peer services.
for _, name := range []string{"pxe", "pxe-headless", "pxe-peer-0", "pxe-peer-1"} {
var svc corev1.Service
if err := cl.Get(context.Background(), types.NamespacedName{Namespace: "dhcp-system", Name: name}, &svc); err != nil {
t.Errorf("service %s not created: %v", name, err)
@@ -102,7 +175,7 @@ func TestReconcileWiresInitContainer(t *testing.T) {
cl := fake.NewClientBuilder().
WithScheme(scheme).
WithStatusSubresource(&v1alpha1.KeaCluster{}).
WithObjects(cluster).
WithObjects(withPeers(cluster)...).
Build()
r := &KeaClusterReconciler{Client: cl, Scheme: scheme}
if _, err := r.Reconcile(context.Background(), ctrl.Request{NamespacedName: types.NamespacedName{Namespace: "dhcp-system", Name: "pxe"}}); err != nil {
@@ -186,7 +259,7 @@ func TestConfigHashChangesWithSubnets(t *testing.T) {
scheme := testScheme(t)
hashFor := func(objs ...client.Object) string {
base := []client.Object{newClusterFixture()}
base := withPeers(newClusterFixture())
cl := fake.NewClientBuilder().
WithScheme(scheme).
WithStatusSubresource(&v1alpha1.KeaCluster{}).
@@ -226,7 +299,7 @@ func TestClusterRefFiltersSubnets(t *testing.T) {
}
cl := fake.NewClientBuilder().WithScheme(scheme).
WithStatusSubresource(&v1alpha1.KeaCluster{}).
WithObjects(cluster, mine, other).Build()
WithObjects(withPeers(cluster, mine, other)...).Build()
r := &KeaClusterReconciler{Client: cl, Scheme: scheme}
if _, err := r.Reconcile(context.Background(), ctrl.Request{NamespacedName: types.NamespacedName{Namespace: "dhcp-system", Name: "pxe"}}); err != nil {
t.Fatal(err)