Merge pull request 'Point HA peer URLs at per-pod ClusterIP Services' (#7) from benvin/ha-peer-clusterip into main

Reviewed-on: #7

internal/controller/helpers.go

@@ -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,

internal/controller/keacluster_controller.go

@@ -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 {
+	// Services first: the per-pod ClusterIP Services must exist (and have their
-		return r.fail(ctx, &cluster, "ConfigError", err)
+	// 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).
+// peers builds the HA peer list. Each URL points at the peer's per-pod ClusterIP
-func (r *KeaClusterReconciler) peers(c *v1alpha1.KeaCluster) []kea.Peer {
+// 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 {

internal/controller/keacluster_test.go

@@ -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.
+	// Anycast + headless + per-pod peer services.
-	for _, name := range []string{"pxe", "pxe-headless"} {
+	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)