feat(api): set trafficDistribution on the client Service

README.md

@@ -25,6 +25,7 @@ The full design rationale is in [docs/concepts.md](docs/concepts.md).
 - **TLS (BYO Secrets or cert-manager)**: `spec.tls.client` / `spec.tls.peer` enable TLS on each surface independently. Material comes from either user-provided Secrets (`serverSecretRef` / `operatorClientSecretRef` / `secretRef`) or operator-emitted `cert-manager.io/v1` Certificates (`certManager.{serverIssuerRef,operatorClientIssuerRef,issuerRef}`) — mutually exclusive per subtree, enforced by CEL. mTLS is the implicit mode when an operator-client source is supplied; server-TLS-only when it isn't. The whole `tls` subtree is CEL-locked immutable post-create. cert-manager-emitted certs auto-renew via cert-manager; Pod-side rotation is a manual one-at-a-time `kubectl delete pod` either way. See [docs/concepts.md](docs/concepts.md#tls).
 - **Resource sizing**: `spec.resources` (a `corev1.ResourceRequirements`) sets the etcd container's CPU/memory requests and limits. Unset uses a conservative 100m/128Mi-request default. Updates take effect on newly-created members; pair with a `VerticalPodAutoscaler` targeting the cluster for live recommendation/rollout.
 - **Scheduling & extra metadata**: `spec.affinity` and `spec.topologySpreadConstraints` pass through to every member Pod (anti-affinity is not defaulted — set it for production); `spec.additionalMetadata` merges user labels/annotations onto every object the operator creates (member Pods, data PVCs, Services, PDB, `EtcdMember` CRs), with operator-owned keys winning on collision. All three apply on object creation and are latched like the rest of the spec. See [docs/concepts.md](docs/concepts.md#pod-scheduling-and-additional-metadata).
+- **Zone-local clients**: `spec.clientService.trafficDistribution` (`PreferClose` / `PreferSameZone` / `PreferSameNode`) sets `trafficDistribution` on the `<cluster>-client` Service, so in a cluster stretched across zones clients prefer a ready member in their own zone. Unset leaves the field to whoever manages the Service. See [docs/concepts.md](docs/concepts.md#client-service).
 - **Monitoring / autoscaling hooks**: every member Pod always exposes a plaintext `metrics` container port at `2381` (etcd's `/health` + Prometheus `/metrics`) for `VMPodScrape` / `PodMonitor`. The `EtcdCluster` CRD exposes the `/scale` subresource with a populated `status.selector`, making it a valid target for `kubectl scale` and `VerticalPodAutoscaler.targetRef`.
 - **Locking pattern**: `status.observed` snapshots the in-flight target so mid-flight spec edits don't corrupt consensus; `progressDeadline` bounds how long the operator will spend trying to reach a target.
 - **Cluster deletion**: cascading owner refs clean up everything; finalizers detect "the whole cluster is going away" and skip etcd-side removal to avoid deadlock.

api/v1alpha2/client_service_envtest_test.go

@@ -0,0 +1,83 @@
+/*
+Copyright 2023 Timofey Larkin.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+*/
+
+package v1alpha2_test
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
+
+	lll "github.com/cozystack/etcd-operator/api/v1alpha2"
+)
+
+// spec.clientService.trafficDistribution admits exactly the values the core
+// Service API defines; anything else is rejected at admission rather than
+// surfacing later as a failed Service patch inside the reconcile loop.
+func TestClientService_TrafficDistributionEnum(t *testing.T) {
+	skipIfNoEnvtest(t)
+	ctx := context.Background()
+
+	for _, v := range []string{"PreferClose", "PreferSameZone", "PreferSameNode"} {
+		c := validCluster("td-" + strings.ToLower(v))
+		c.Spec.ClientService = &lll.ClientServiceSpec{TrafficDistribution: &v}
+		if err := k8s.Create(ctx, c); err != nil {
+			t.Fatalf("apiserver rejected trafficDistribution=%q: %v", v, err)
+		}
+		t.Cleanup(func() { _ = k8s.Delete(ctx, c) })
+	}
+
+	bogus := "Anywhere"
+	c := validCluster("td-bogus")
+	c.Spec.ClientService = &lll.ClientServiceSpec{TrafficDistribution: &bogus}
+	err := k8s.Create(ctx, c)
+	if err == nil {
+		_ = k8s.Delete(ctx, c)
+		t.Fatalf("apiserver accepted trafficDistribution=%q; expected rejection", bogus)
+	}
+	if !strings.Contains(err.Error(), "spec.clientService.trafficDistribution") {
+		t.Fatalf("error did not name the field: %v", err)
+	}
+}
+
+// The envtest apiserver (pinned to the k8s.io/api minor in go.mod) must
+// persist trafficDistribution on a Service. An apiserver with the field
+// disabled drops it silently, which would leave the operator re-patching
+// the client Service on every reconcile; this guards the version floor the
+// field's docs promise.
+func TestClientService_ApiserverPersistsTrafficDistribution(t *testing.T) {
+	skipIfNoEnvtest(t)
+	ctx := context.Background()
+
+	td := corev1.ServiceTrafficDistributionPreferClose
+	svc := &corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{Name: "td-probe", Namespace: "default"},
+		Spec: corev1.ServiceSpec{
+			Ports:               []corev1.ServicePort{{Name: "client", Port: 2379}},
+			TrafficDistribution: &td,
+		},
+	}
+	if err := k8s.Create(ctx, svc); err != nil {
+		t.Fatalf("Create Service: %v", err)
+	}
+	t.Cleanup(func() { _ = k8s.Delete(ctx, svc) })
+
+	got := &corev1.Service{}
+	if err := k8s.Get(ctx, ctrlclient.ObjectKeyFromObject(svc), got); err != nil {
+		t.Fatalf("Get Service: %v", err)
+	}
+	if got.Spec.TrafficDistribution == nil || *got.Spec.TrafficDistribution != td {
+		t.Fatalf("apiserver did not persist trafficDistribution: %v", got.Spec.TrafficDistribution)
+	}
+}

api/v1alpha2/etcdcluster_types.go

@@ -569,6 +569,41 @@ type EtcdClusterSpec struct {
 	// status.observed.
 	// +optional
 	ImagePullSecrets []corev1.LocalObjectReference `json:"imagePullSecrets,omitempty"`
+
+	// ClientService configures the "<cluster>-client" Service the operator
+	// creates for applications. Absent means the operator manages only the
+	// fields it always owns (selector and the client port) and leaves the
+	// rest of the Service alone. Does not affect the headless Service.
+	//
+	// Unlike the member-level fields above, this is not latched through
+	// status.observed: the Service carries no consensus state, so a change
+	// is applied on the next reconcile.
+	// +optional
+	ClientService *ClientServiceSpec `json:"clientService,omitempty"`
+}
+
+// ClientServiceSpec configures the "<cluster>-client" Service. Grouped in its
+// own struct so further Service-level knobs land next to TrafficDistribution
+// rather than as loose fields on EtcdClusterSpec.
+type ClientServiceSpec struct {
+	// TrafficDistribution sets the client Service's spec.trafficDistribution,
+	// a hint to the Service proxy (kube-proxy, Cilium, ...) on how to route
+	// traffic across the ready etcd members. The common use is a cluster
+	// stretched across zones: "PreferClose" (or "PreferSameZone") keeps
+	// clients talking to a member in their own zone while one is ready.
+	// The values mirror corev1.ServiceSpec.TrafficDistribution.
+	//
+	// When set, the operator writes it on create and reconciles drift back
+	// to it. When absent, the operator does not touch the field, so a value
+	// set on the Service by hand is preserved.
+	//
+	// Requires an apiserver that serves the field: Kubernetes 1.31+ for
+	// "PreferClose" (beta, on by default; GA in 1.33), and 1.34+ for
+	// "PreferSameZone" and "PreferSameNode" (PreferSameTrafficDistribution
+	// feature gate, beta and on by default from 1.34).
+	// +kubebuilder:validation:Enum=PreferClose;PreferSameZone;PreferSameNode
+	// +optional
+	TrafficDistribution *string `json:"trafficDistribution,omitempty"`
 }
 
 // AdditionalMetadata is a set of labels and annotations the operator merges

charts/etcd-operator/crd-bases/etcd-operator.cozystack.io_etcdclusters.yaml

@@ -1166,6 +1166,40 @@ spec:
                       rule: '!has(self.source.pvc) || (has(self.source.pvc.subPath)
                         && size(self.source.pvc.subPath) > 0)'
                 type: object
+              clientService:
+                description: |-
+                  ClientService configures the "<cluster>-client" Service the operator
+                  creates for applications. Absent means the operator manages only the
+                  fields it always owns (selector and the client port) and leaves the
+                  rest of the Service alone. Does not affect the headless Service.
+
+                  Unlike the member-level fields above, this is not latched through
+                  status.observed: the Service carries no consensus state, so a change
+                  is applied on the next reconcile.
+                properties:
+                  trafficDistribution:
+                    description: |-
+                      TrafficDistribution sets the client Service's spec.trafficDistribution,
+                      a hint to the Service proxy (kube-proxy, Cilium, ...) on how to route
+                      traffic across the ready etcd members. The common use is a cluster
+                      stretched across zones: "PreferClose" (or "PreferSameZone") keeps
+                      clients talking to a member in their own zone while one is ready.
+                      The values mirror corev1.ServiceSpec.TrafficDistribution.
+
+                      When set, the operator writes it on create and reconciles drift back
+                      to it. When absent, the operator does not touch the field, so a value
+                      set on the Service by hand is preserved.
+
+                      Requires an apiserver that serves the field: Kubernetes 1.31+ for
+                      "PreferClose" (beta, on by default; GA in 1.33), and 1.34+ for
+                      "PreferSameZone" and "PreferSameNode" (PreferSameTrafficDistribution
+                      feature gate, beta and on by default from 1.34).
+                    enum:
+                    - PreferClose
+                    - PreferSameZone
+                    - PreferSameNode
+                    type: string
+                type: object
               imagePullSecrets:
                 description: |-
                   ImagePullSecrets is a list of Secret references in the cluster's

controllers/client_service_test.go

@@ -0,0 +1,155 @@
+/*
+Copyright 2023 Timofey Larkin.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+*/
+
+package controllers
+
+import (
+	"context"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	lll "github.com/cozystack/etcd-operator/api/v1alpha2"
+)
+
+// clusterWithTrafficDistribution returns a cluster whose client Service asks
+// for td; a nil td leaves spec.clientService unset.
+func clusterWithTrafficDistribution(td *string) *lll.EtcdCluster {
+	cluster := &lll.EtcdCluster{
+		ObjectMeta: metav1.ObjectMeta{Name: "test", Namespace: "ns"},
+	}
+	if td != nil {
+		cluster.Spec.ClientService = &lll.ClientServiceSpec{TrafficDistribution: td}
+	}
+	return cluster
+}
+
+// existingService returns a Service as it would sit on the apiserver before
+// the operator's reconcile, carrying td on spec.trafficDistribution.
+func existingService(name string, td *string) *corev1.Service {
+	svc := &corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "ns"},
+		Spec: corev1.ServiceSpec{
+			Selector:            map[string]string{LabelCluster: "test"},
+			Ports:               []corev1.ServicePort{{Name: "client", Port: 2379}},
+			TrafficDistribution: td,
+		},
+	}
+	if name == "test" {
+		svc.Spec.ClusterIP = corev1.ClusterIPNone
+		svc.Spec.PublishNotReadyAddresses = true
+		svc.Spec.Ports = append(svc.Spec.Ports, corev1.ServicePort{Name: "peer", Port: 2380})
+	}
+	return svc
+}
+
+func strPtr(s string) *string { return &s }
+
+func getTrafficDistribution(t *testing.T, c client.Client, name string) *string {
+	t.Helper()
+	svc := &corev1.Service{}
+	if err := c.Get(context.Background(), types.NamespacedName{Namespace: "ns", Name: name}, svc); err != nil {
+		t.Fatalf("Get Service %s: %v", name, err)
+	}
+	return svc.Spec.TrafficDistribution
+}
+
+func assertTrafficDistribution(t *testing.T, c client.Client, name string, want *string) {
+	t.Helper()
+	got := getTrafficDistribution(t, c, name)
+	switch {
+	case want == nil && got != nil:
+		t.Fatalf("Service %s trafficDistribution = %q, want unset", name, *got)
+	case want != nil && got == nil:
+		t.Fatalf("Service %s trafficDistribution unset, want %q", name, *want)
+	case want != nil && *got != *want:
+		t.Fatalf("Service %s trafficDistribution = %q, want %q", name, *got, *want)
+	}
+}
+
+// On create the client Service carries the requested value; the headless
+// Service never does.
+func TestEnsureServices_TrafficDistributionOnCreate(t *testing.T) {
+	cluster := clusterWithTrafficDistribution(strPtr(corev1.ServiceTrafficDistributionPreferClose))
+	c, _ := newTestClient(t, cluster)
+	r := &EtcdClusterReconciler{Client: c, Scheme: testScheme(t)}
+
+	if err := r.ensureServices(context.Background(), cluster); err != nil {
+		t.Fatalf("ensureServices: %v", err)
+	}
+	assertTrafficDistribution(t, c, "test-client", strPtr(corev1.ServiceTrafficDistributionPreferClose))
+	assertTrafficDistribution(t, c, "test", nil)
+}
+
+// A client Service that drifted (value removed or changed by hand) is
+// brought back to the spec.
+func TestEnsureServices_TrafficDistributionDriftReconciled(t *testing.T) {
+	for name, onService := range map[string]*string{
+		"unset":     nil,
+		"different": strPtr(corev1.ServiceTrafficDistributionPreferSameNode),
+	} {
+		t.Run(name, func(t *testing.T) {
+			cluster := clusterWithTrafficDistribution(strPtr(corev1.ServiceTrafficDistributionPreferClose))
+			c, _ := newTestClient(t, cluster, existingService("test", nil), existingService("test-client", onService))
+			r := &EtcdClusterReconciler{Client: c, Scheme: testScheme(t)}
+
+			if err := r.ensureServices(context.Background(), cluster); err != nil {
+				t.Fatalf("ensureServices: %v", err)
+			}
+			assertTrafficDistribution(t, c, "test-client", strPtr(corev1.ServiceTrafficDistributionPreferClose))
+			assertTrafficDistribution(t, c, "test", nil)
+		})
+	}
+}
+
+// With the field unset — no clientService block, or a block without
+// trafficDistribution — the operator does not own it: a value patched onto
+// the Service by hand survives reconcile, and no value is added.
+func TestEnsureServices_TrafficDistributionUnsetLeavesServiceAlone(t *testing.T) {
+	for name, cluster := range map[string]*lll.EtcdCluster{
+		"no clientService": clusterWithTrafficDistribution(nil),
+		"empty clientService": func() *lll.EtcdCluster {
+			cl := clusterWithTrafficDistribution(nil)
+			cl.Spec.ClientService = &lll.ClientServiceSpec{}
+			return cl
+		}(),
+	} {
+		t.Run(name, func(t *testing.T) {
+			manual := strPtr(corev1.ServiceTrafficDistributionPreferClose)
+			c, _ := newTestClient(t, cluster, existingService("test", nil), existingService("test-client", manual))
+			r := &EtcdClusterReconciler{Client: c, Scheme: testScheme(t)}
+
+			if err := r.ensureServices(context.Background(), cluster); err != nil {
+				t.Fatalf("ensureServices: %v", err)
+			}
+			assertTrafficDistribution(t, c, "test-client", manual)
+			assertTrafficDistribution(t, c, "test", nil)
+		})
+	}
+}
+
+// The headless Service is not the operator's to set traffic distribution on:
+// a value someone put there is neither cleared nor overwritten with the
+// client Service's value.
+func TestEnsureServices_HeadlessTrafficDistributionUntouched(t *testing.T) {
+	cluster := clusterWithTrafficDistribution(strPtr(corev1.ServiceTrafficDistributionPreferClose))
+	manual := strPtr(corev1.ServiceTrafficDistributionPreferSameNode)
+	c, _ := newTestClient(t, cluster, existingService("test", manual), existingService("test-client", nil))
+	r := &EtcdClusterReconciler{Client: c, Scheme: testScheme(t)}
+
+	if err := r.ensureServices(context.Background(), cluster); err != nil {
+		t.Fatalf("ensureServices: %v", err)
+	}
+	assertTrafficDistribution(t, c, "test", manual)
+	assertTrafficDistribution(t, c, "test-client", strPtr(corev1.ServiceTrafficDistributionPreferClose))
+}