From 6b935cba7b541414c8a478402e74f8514738670f Mon Sep 17 00:00:00 2001 From: Andrei Kvapil Date: Sat, 17 Oct 2026 12:23:26 +0500 Subject: [PATCH 1/3] fix(controllers): don't let client Service trafficDistribution wedge reconcile Before Kubernetes 1.34 the apiserver rejects PreferSameZone and PreferSameNode on a Service, and ensureServices runs ahead of bootstrap, scaling and member replacement, so the rejection stopped the whole cluster. An apiserver that does not know the field drops it instead, so the drift check never converged and the Service was patched on every reconcile. Reconcile the client Service without the value when the apiserver rejects it or does not persist it, emit a Warning Event on the EtcdCluster, and back the value off for an hour so it is neither resent every reconcile nor lost once the apiserver is upgraded. Assisted-by: LLM Signed-off-by: Andrei Kvapil --- controllers/client_service.go | 133 +++++++++++ controllers/client_service_compat_test.go | 253 +++++++++++++++++++++ controllers/etcdcluster_controller.go | 80 ++++++- controllers/etcdcluster_controller_test.go | 4 +- main.go | 1 + 5 files changed, 459 insertions(+), 12 deletions(-) create mode 100644 controllers/client_service.go create mode 100644 controllers/client_service_compat_test.go diff --git a/controllers/client_service.go b/controllers/client_service.go new file mode 100644 index 00000000..0820d88e --- /dev/null +++ b/controllers/client_service.go @@ -0,0 +1,133 @@ +/* +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 ( + "sync" + "time" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/types" + + lll "github.com/cozystack/etcd-operator/api/v1alpha2" +) + +const ( + // EventReasonTrafficDistributionRejected is emitted when the apiserver + // refuses spec.trafficDistribution on the client Service, e.g. + // PreferSameZone / PreferSameNode before Kubernetes 1.34. + EventReasonTrafficDistributionRejected = "TrafficDistributionRejected" + // EventReasonTrafficDistributionNotPersisted is emitted when the + // apiserver accepts the write but drops the field, which is what an + // apiserver that does not know spec.trafficDistribution does. + EventReasonTrafficDistributionNotPersisted = "TrafficDistributionNotPersisted" + + // trafficDistributionRetryInterval is how long a value the apiserver + // refused or dropped is left off the client Service before the + // operator tries it again. It bounds the cost of an unsupported value + // to one failed write per interval, and lets the value land on its own + // once the apiserver is upgraded. + trafficDistributionRetryInterval = time.Hour + + trafficDistributionField = "spec.trafficDistribution" +) + +// trafficDistributionBackoff remembers, per client Service, a +// trafficDistribution value the apiserver would not keep, so the operator +// stops writing it on every reconcile. It is in-memory only: after an +// operator restart each unsupported value costs one more failed write. +type trafficDistributionBackoff struct { + mu sync.Mutex + entries map[types.NamespacedName]trafficDistributionBackoffEntry +} + +type trafficDistributionBackoffEntry struct { + value string + until time.Time +} + +// suppressed reports whether value is still backed off for key. An entry +// for a different value (the spec changed) or an expired one does not +// count and is forgotten. +func (b *trafficDistributionBackoff) suppressed(key types.NamespacedName, value string, now time.Time) bool { + b.mu.Lock() + defer b.mu.Unlock() + e, ok := b.entries[key] + if !ok { + return false + } + if e.value != value || !now.Before(e.until) { + delete(b.entries, key) + return false + } + return true +} + +func (b *trafficDistributionBackoff) suppress(key types.NamespacedName, value string, now time.Time) { + b.mu.Lock() + defer b.mu.Unlock() + if b.entries == nil { + b.entries = map[types.NamespacedName]trafficDistributionBackoffEntry{} + } + b.entries[key] = trafficDistributionBackoffEntry{value: value, until: now.Add(trafficDistributionRetryInterval)} +} + +// isTrafficDistributionRejected reports whether err is the apiserver +// refusing spec.trafficDistribution. Other validation failures on the +// Service are not matched: they are not the operator's to work around. +func isTrafficDistributionRejected(err error) bool { + if !errors.IsInvalid(err) { + return false + } + status, ok := err.(errors.APIStatus) + if !ok || status.Status().Details == nil { + return false + } + causes := status.Status().Details.Causes + if len(causes) == 0 { + return false + } + for _, c := range causes { + if c.Field != trafficDistributionField { + return false + } + } + return true +} + +// clientServiceTrafficDistribution returns the trafficDistribution the +// client Service should carry this reconcile: the spec's value, unless that +// value is backed off. +func (r *EtcdClusterReconciler) clientServiceTrafficDistribution(cluster *lll.EtcdCluster, key types.NamespacedName) *string { + cs := cluster.Spec.ClientService + if cs == nil || cs.TrafficDistribution == nil { + return nil + } + if r.tdBackoff.suppressed(key, *cs.TrafficDistribution, r.now()) { + return nil + } + td := *cs.TrafficDistribution + return &td +} + +// trafficDistributionPersisted reports whether the Service returned by the +// apiserver carries the value the operator wrote. +func trafficDistributionPersisted(svc *corev1.Service, want string) bool { + return svc.Spec.TrafficDistribution != nil && *svc.Spec.TrafficDistribution == want +} + +func (r *EtcdClusterReconciler) now() time.Time { + if r.Now != nil { + return r.Now() + } + return time.Now() +} diff --git a/controllers/client_service_compat_test.go b/controllers/client_service_compat_test.go new file mode 100644 index 00000000..7cb148fd --- /dev/null +++ b/controllers/client_service_compat_test.go @@ -0,0 +1,253 @@ +/* +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" + "strings" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/util/validation/field" + "k8s.io/client-go/tools/record" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + + lll "github.com/cozystack/etcd-operator/api/v1alpha2" +) + +// tdApiserver models how an apiserver of a given version treats +// spec.trafficDistribution on a Service: rejecting some values (before 1.34 +// PreferSameZone / PreferSameNode are Invalid), or not knowing the field at +// all and dropping it from what it stores. It counts the writes that carried +// the field so tests can tell a retry from a quiet reconcile. +type tdApiserver struct { + reject map[string]bool + drop bool + + tdWrites int +} + +func (a *tdApiserver) client(t *testing.T, objs ...client.Object) client.Client { + t.Helper() + return fake.NewClientBuilder(). + WithScheme(testScheme(t)). + WithObjects(objs...). + WithStatusSubresource(&lll.EtcdCluster{}). + WithInterceptorFuncs(interceptor.Funcs{ + Create: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.CreateOption) error { + if svc, ok := obj.(*corev1.Service); ok { + if err := a.admit(svc); err != nil { + return err + } + } + return c.Create(ctx, obj, opts...) + }, + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + if svc, ok := obj.(*corev1.Service); ok { + if err := a.admit(svc); err != nil { + return err + } + } + return c.Patch(ctx, obj, patch, opts...) + }, + }). + Build() +} + +// admit rejects or strips trafficDistribution ahead of the write. It runs on +// the object the operator is about to send, which for a merge patch is the +// full desired Service, so stripping the field there is what the operator +// gets back from an apiserver that does not know it. +func (a *tdApiserver) admit(svc *corev1.Service) error { + td := svc.Spec.TrafficDistribution + if td == nil { + return nil + } + a.tdWrites++ + if a.reject[*td] { + return apierrors.NewInvalid(corev1.SchemeGroupVersion.WithKind("Service").GroupKind(), svc.Name, field.ErrorList{ + field.NotSupported(field.NewPath("spec", "trafficDistribution"), *td, []string{corev1.ServiceTrafficDistributionPreferClose}), + }) + } + if a.drop { + svc.Spec.TrafficDistribution = nil + } + return nil +} + +func drainEvents(rec *record.FakeRecorder) []string { + var out []string + for { + select { + case ev := <-rec.Events: + out = append(out, ev) + default: + return out + } + } +} + +func assertOneEvent(t *testing.T, rec *record.FakeRecorder, reason string) { + t.Helper() + evs := drainEvents(rec) + if len(evs) != 1 || !strings.Contains(evs[0], reason) { + t.Fatalf("events = %q, want exactly one %s", evs, reason) + } +} + +// fakeClock is a settable clock for the trafficDistribution backoff. +type fakeClock struct{ t time.Time } + +func (c *fakeClock) now() time.Time { return c.t } + +// A value the apiserver refuses (PreferSameZone before 1.34) must not stop +// the reconcile: ensureServices runs ahead of bootstrap and scaling, so an +// error here would wedge the whole cluster. The client Service is written +// without the field, a Warning Event names the cause, and later reconciles +// don't resend the value until the backoff expires. +func TestEnsureServices_TrafficDistributionRejected(t *testing.T) { + for name, existing := range map[string][]client.Object{ + "create": nil, + "patch": {existingService("test", nil), existingService("test-client", nil)}, + } { + t.Run(name, func(t *testing.T) { + cluster := clusterWithTrafficDistribution(strPtr(corev1.ServiceTrafficDistributionPreferSameZone)) + api := &tdApiserver{reject: map[string]bool{corev1.ServiceTrafficDistributionPreferSameZone: true}} + c := api.client(t, append([]client.Object{cluster}, existing...)...) + rec := record.NewFakeRecorder(10) + clock := &fakeClock{t: time.Unix(1_700_000_000, 0)} + r := &EtcdClusterReconciler{Client: c, Scheme: testScheme(t), Recorder: rec, Now: clock.now} + + if err := r.ensureServices(context.Background(), cluster); err != nil { + t.Fatalf("ensureServices: %v", err) + } + assertTrafficDistribution(t, c, "test-client", nil) + assertOneEvent(t, rec, EventReasonTrafficDistributionRejected) + if api.tdWrites != 1 { + t.Fatalf("writes carrying trafficDistribution = %d, want 1", api.tdWrites) + } + + if err := r.ensureServices(context.Background(), cluster); err != nil { + t.Fatalf("second ensureServices: %v", err) + } + if api.tdWrites != 1 { + t.Fatalf("value resent within the backoff: writes = %d, want 1", api.tdWrites) + } + if evs := drainEvents(rec); len(evs) != 0 { + t.Fatalf("events within the backoff = %q, want none", evs) + } + + // Once the backoff expires the value is tried again, so an + // upgraded apiserver picks it up without an operator restart. + clock.t = clock.t.Add(trafficDistributionRetryInterval) + delete(api.reject, corev1.ServiceTrafficDistributionPreferSameZone) + if err := r.ensureServices(context.Background(), cluster); err != nil { + t.Fatalf("ensureServices after backoff: %v", err) + } + assertTrafficDistribution(t, c, "test-client", strPtr(corev1.ServiceTrafficDistributionPreferSameZone)) + }) + } +} + +// Only a rejection of spec.trafficDistribution alone is worked around; a +// Service invalid for any other reason is still a reconcile error. +func TestEnsureServices_OtherInvalidErrorsPropagate(t *testing.T) { + cluster := clusterWithTrafficDistribution(strPtr(corev1.ServiceTrafficDistributionPreferClose)) + invalid := apierrors.NewInvalid(corev1.SchemeGroupVersion.WithKind("Service").GroupKind(), "test-client", field.ErrorList{ + field.Invalid(field.NewPath("spec", "ports"), nil, "boom"), + field.NotSupported(field.NewPath("spec", "trafficDistribution"), "x", []string{}), + }) + c := fake.NewClientBuilder(). + WithScheme(testScheme(t)). + WithObjects(cluster). + WithInterceptorFuncs(interceptor.Funcs{ + Create: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.CreateOption) error { + if obj.GetName() == "test-client" { + return invalid + } + return c.Create(ctx, obj, opts...) + }, + }). + Build() + rec := record.NewFakeRecorder(10) + r := &EtcdClusterReconciler{Client: c, Scheme: testScheme(t), Recorder: rec} + + if err := r.ensureServices(context.Background(), cluster); !apierrors.IsInvalid(err) { + t.Fatalf("ensureServices error = %v, want the Invalid error", err) + } + if evs := drainEvents(rec); len(evs) != 0 { + t.Fatalf("events = %q, want none", evs) + } +} + +// An apiserver that does not know spec.trafficDistribution (1.29/1.30) +// accepts the write and silently drops the field. Without a check the +// comparison never converges and the Service is patched on every reconcile; +// instead the operator notices the value did not survive, emits a Warning +// Event and stops resending it for the backoff window. +func TestEnsureServices_TrafficDistributionNotPersisted(t *testing.T) { + for name, existing := range map[string][]client.Object{ + "create": nil, + "patch": {existingService("test", nil), existingService("test-client", nil)}, + } { + t.Run(name, func(t *testing.T) { + cluster := clusterWithTrafficDistribution(strPtr(corev1.ServiceTrafficDistributionPreferClose)) + api := &tdApiserver{drop: true} + c := api.client(t, append([]client.Object{cluster}, existing...)...) + rec := record.NewFakeRecorder(10) + clock := &fakeClock{t: time.Unix(1_700_000_000, 0)} + r := &EtcdClusterReconciler{Client: c, Scheme: testScheme(t), Recorder: rec, Now: clock.now} + + for i := range 3 { + if err := r.ensureServices(context.Background(), cluster); err != nil { + t.Fatalf("ensureServices #%d: %v", i, err) + } + } + assertTrafficDistribution(t, c, "test-client", nil) + if api.tdWrites != 1 { + t.Fatalf("writes carrying trafficDistribution = %d, want 1", api.tdWrites) + } + assertOneEvent(t, rec, EventReasonTrafficDistributionNotPersisted) + + clock.t = clock.t.Add(trafficDistributionRetryInterval) + if err := r.ensureServices(context.Background(), cluster); err != nil { + t.Fatalf("ensureServices after backoff: %v", err) + } + if api.tdWrites != 2 { + t.Fatalf("writes after backoff = %d, want 2", api.tdWrites) + } + }) + } +} + +// The backoff is per value: changing spec.clientService.trafficDistribution +// to something the apiserver does support applies it straight away. +func TestEnsureServices_TrafficDistributionBackoffClearedBySpecChange(t *testing.T) { + cluster := clusterWithTrafficDistribution(strPtr(corev1.ServiceTrafficDistributionPreferSameNode)) + api := &tdApiserver{reject: map[string]bool{corev1.ServiceTrafficDistributionPreferSameNode: true}} + c := api.client(t, cluster, existingService("test", nil), existingService("test-client", nil)) + r := &EtcdClusterReconciler{Client: c, Scheme: testScheme(t), Recorder: record.NewFakeRecorder(10)} + + if err := r.ensureServices(context.Background(), cluster); err != nil { + t.Fatalf("ensureServices: %v", err) + } + assertTrafficDistribution(t, c, "test-client", nil) + + cluster.Spec.ClientService.TrafficDistribution = strPtr(corev1.ServiceTrafficDistributionPreferClose) + if err := r.ensureServices(context.Background(), cluster); err != nil { + t.Fatalf("ensureServices after spec change: %v", err) + } + assertTrafficDistribution(t, c, "test-client", strPtr(corev1.ServiceTrafficDistributionPreferClose)) +} diff --git a/controllers/etcdcluster_controller.go b/controllers/etcdcluster_controller.go index 49d8230b..33d3d65e 100644 --- a/controllers/etcdcluster_controller.go +++ b/controllers/etcdcluster_controller.go @@ -34,6 +34,7 @@ import ( "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/client-go/tools/record" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" @@ -74,6 +75,18 @@ type EtcdClusterReconciler struct { // Defaults to "cluster.local" when empty. Override at operator // startup with the --cluster-domain flag. ClusterDomain string + + // Recorder emits Events on the EtcdCluster. Optional: a nil Recorder + // drops them, which unit tests that don't assert on Events rely on. + Recorder record.EventRecorder + + // Now returns the current time; nil means time.Now. Tests inject a + // clock to step past backoff windows. + Now func() time.Time + + // tdBackoff holds client-Service trafficDistribution values the + // apiserver would not keep, so they are not rewritten every reconcile. + tdBackoff trafficDistributionBackoff } //+kubebuilder:rbac:groups=etcd-operator.cozystack.io,resources=etcdclusters,verbs=get;list;watch @@ -82,6 +95,7 @@ type EtcdClusterReconciler struct { //+kubebuilder:rbac:groups=etcd-operator.cozystack.io,resources=etcdmembers,verbs=get;list;watch;create;delete //+kubebuilder:rbac:groups=etcd-operator.cozystack.io,resources=etcdmembers/status,verbs=get;update;patch //+kubebuilder:rbac:groups="",resources=services,verbs=get;list;watch;create;update;patch +//+kubebuilder:rbac:groups="",resources=events,verbs=create;patch //+kubebuilder:rbac:groups=policy,resources=poddisruptionbudgets,verbs=get;list;watch;create;update;patch;delete //+kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch // `delete` is intentionally omitted from the certificates verb list: @@ -1904,7 +1918,7 @@ func (r *EtcdClusterReconciler) ensureServices(ctx context.Context, cluster *lll // self-GCs as they roll) for the adopted pods, which carry the legacy // name in their AnnHeadlessServiceName annotation; the operator does not // manage that one. - if err := r.ensureService(ctx, cluster, cluster.Name, corev1.ServiceSpec{ + if _, err := r.ensureService(ctx, cluster, cluster.Name, corev1.ServiceSpec{ ClusterIP: corev1.ClusterIPNone, PublishNotReadyAddresses: true, Selector: map[string]string{LabelCluster: cluster.Name}, @@ -1917,16 +1931,53 @@ func (r *EtcdClusterReconciler) ensureServices(ctx context.Context, cluster *lll } // Client service — stable endpoint for applications. + clientName := cluster.Name + "-client" + clientKey := types.NamespacedName{Namespace: cluster.Namespace, Name: clientName} clientSpec := corev1.ServiceSpec{ Selector: map[string]string{LabelCluster: cluster.Name}, Ports: []corev1.ServicePort{ {Name: "client", Port: 2379}, }, + TrafficDistribution: r.clientServiceTrafficDistribution(cluster, clientKey), + } + svc, err := r.ensureService(ctx, cluster, clientName, clientSpec) + td := clientSpec.TrafficDistribution + if td == nil { + return err + } + // trafficDistribution is a routing hint, and the apiserver's support + // for it depends on its version: before 1.34 it refuses PreferSameZone + // and PreferSameNode, and an apiserver that does not know the field + // drops it. Neither may stop the reconcile (ensureServices runs ahead + // of bootstrap, scaling and member replacement) or turn it into a + // write on every pass, so the value is reported, backed off and the + // Service reconciled without it. + if err != nil { + if !isTrafficDistributionRejected(err) { + return err + } + r.tdBackoff.suppress(clientKey, *td, r.now()) + r.eventf(cluster, corev1.EventTypeWarning, EventReasonTrafficDistributionRejected, + "Service %s: apiserver rejected trafficDistribution %q, reconciling the Service without it and retrying in %s: %v", + clientName, *td, trafficDistributionRetryInterval, err) + clientSpec.TrafficDistribution = nil + _, err = r.ensureService(ctx, cluster, clientName, clientSpec) + return err } - if cs := cluster.Spec.ClientService; cs != nil { - clientSpec.TrafficDistribution = cs.TrafficDistribution + if !trafficDistributionPersisted(svc, *td) { + r.tdBackoff.suppress(clientKey, *td, r.now()) + r.eventf(cluster, corev1.EventTypeWarning, EventReasonTrafficDistributionNotPersisted, + "Service %s: apiserver did not persist trafficDistribution %q (it does not support the field), retrying in %s", + clientName, *td, trafficDistributionRetryInterval) } - return r.ensureService(ctx, cluster, cluster.Name+"-client", clientSpec) + return nil +} + +func (r *EtcdClusterReconciler) eventf(obj runtime.Object, eventType, reason, format string, args ...any) { + if r.Recorder == nil { + return + } + r.Recorder.Eventf(obj, eventType, reason, format, args...) } // ensureService creates the Service if absent and reconciles drift on the @@ -1936,12 +1987,15 @@ func (r *EtcdClusterReconciler) ensureServices(ctx context.Context, cluster *lll // type, a hand-set TrafficDistribution the spec doesn't ask for) is // preserved — admission webhooks routinely inject these and we don't want // to fight them. +// +// It returns the Service as the apiserver last reported it: the created or +// patched object, or the existing one when nothing needed to change. func (r *EtcdClusterReconciler) ensureService( ctx context.Context, cluster *lll.EtcdCluster, name string, desired corev1.ServiceSpec, -) error { +) (*corev1.Service, error) { svc := &corev1.Service{} err := r.Get(ctx, types.NamespacedName{Namespace: cluster.Namespace, Name: name}, svc) if errors.IsNotFound(err) { @@ -1956,12 +2010,15 @@ func (r *EtcdClusterReconciler) ensureService( Spec: desired, } if err := controllerutil.SetControllerReference(cluster, svc, r.Scheme); err != nil { - return err + return nil, err + } + if err := r.Create(ctx, svc); err != nil { + return nil, err } - return r.Create(ctx, svc) + return svc, nil } if err != nil { - return err + return nil, err } // Drift reconcile on owned fields only. @@ -1992,9 +2049,12 @@ func (r *EtcdClusterReconciler) ensureService( } } if !changed { - return nil + return svc, nil + } + if err := r.Patch(ctx, svc, client.MergeFrom(original)); err != nil { + return nil, err } - return r.Patch(ctx, svc, client.MergeFrom(original)) + return svc, nil } func equalSelectors(a, b map[string]string) bool { diff --git a/controllers/etcdcluster_controller_test.go b/controllers/etcdcluster_controller_test.go index 74ac7caa..c472539e 100644 --- a/controllers/etcdcluster_controller_test.go +++ b/controllers/etcdcluster_controller_test.go @@ -2281,7 +2281,7 @@ func TestEnsureService_ReconcilesOwnedFields(t *testing.T) { c, _ := newTestClient(t, cluster, preexisting) r := &EtcdClusterReconciler{Client: c, Scheme: testScheme(t), EtcdClientFactory: factoryReturning(newFakeEtcd(0xdead))} - if err := r.ensureService(ctx, cluster, "test", corev1.ServiceSpec{ + if _, err := r.ensureService(ctx, cluster, "test", corev1.ServiceSpec{ ClusterIP: corev1.ClusterIPNone, PublishNotReadyAddresses: true, Selector: map[string]string{LabelCluster: "test"}, @@ -3689,7 +3689,7 @@ func TestAdditionalMetadata_AppliedToServiceAndPDB(t *testing.T) { t.Errorf("PDB annotation example.com/note = %q, want bar", pdb.Annotations["example.com/note"]) } - if err := r.ensureService(ctx, cluster, "test-svc", corev1.ServiceSpec{}); err != nil { + if _, err := r.ensureService(ctx, cluster, "test-svc", corev1.ServiceSpec{}); err != nil { t.Fatalf("ensureService: %v", err) } svc := &corev1.Service{} diff --git a/main.go b/main.go index 6f82c523..0051bdfe 100644 --- a/main.go +++ b/main.go @@ -229,6 +229,7 @@ func main() { Scheme: mgr.GetScheme(), CertManagerAvailable: certManagerAvailable, ClusterDomain: clusterDomain, + Recorder: mgr.GetEventRecorderFor("etcd-operator"), }).SetupWithManager(mgr); err != nil { setupLog.Error(err, "unable to create controller", "controller", "EtcdCluster") os.Exit(1) From 86a930ee53fe6fe6ea8413df9c1ac748dafe9f98 Mon Sep 17 00:00:00 2001 From: Andrei Kvapil Date: Sat, 17 Oct 2026 12:23:26 +0500 Subject: [PATCH 2/3] test(api): pin the apiserver's trafficDistribution rejection shape The cluster controller recognises a rejected trafficDistribution by an Invalid error whose causes all name spec.trafficDistribution. Check that against the envtest apiserver, which before 1.34 refuses PreferSameZone. Assisted-by: LLM Signed-off-by: Andrei Kvapil --- api/v1alpha2/client_service_envtest_test.go | 37 +++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/api/v1alpha2/client_service_envtest_test.go b/api/v1alpha2/client_service_envtest_test.go index 4e322294..35cf0e9b 100644 --- a/api/v1alpha2/client_service_envtest_test.go +++ b/api/v1alpha2/client_service_envtest_test.go @@ -16,6 +16,7 @@ import ( "testing" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ctrlclient "sigs.k8s.io/controller-runtime/pkg/client" @@ -81,3 +82,39 @@ func TestClientService_ApiserverPersistsTrafficDistribution(t *testing.T) { t.Fatalf("apiserver did not persist trafficDistribution: %v", got.Spec.TrafficDistribution) } } + +// Before Kubernetes 1.34 the apiserver refuses PreferSameZone on a Service. +// The cluster controller recognises that refusal by its shape (Invalid, every +// cause on spec.trafficDistribution) and reconciles the client Service +// without the field instead of failing; this pins the shape against a real +// apiserver. On 1.34+ the value is accepted and there is nothing to check. +func TestClientService_ApiserverRejectionNamesTrafficDistribution(t *testing.T) { + skipIfNoEnvtest(t) + ctx := context.Background() + + td := corev1.ServiceTrafficDistributionPreferSameZone + svc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{Name: "td-reject-probe", Namespace: "default"}, + Spec: corev1.ServiceSpec{ + Ports: []corev1.ServicePort{{Name: "client", Port: 2379}}, + TrafficDistribution: &td, + }, + } + err := k8s.Create(ctx, svc) + if err == nil { + _ = k8s.Delete(ctx, svc) + t.Skipf("apiserver accepts trafficDistribution=%q (1.34+)", td) + } + if !apierrors.IsInvalid(err) { + t.Fatalf("Create error = %v, want Invalid", err) + } + status, ok := err.(apierrors.APIStatus) + if !ok || status.Status().Details == nil || len(status.Status().Details.Causes) == 0 { + t.Fatalf("Invalid error carries no causes: %v", err) + } + for _, c := range status.Status().Details.Causes { + if c.Field != "spec.trafficDistribution" { + t.Fatalf("cause on %q, want spec.trafficDistribution: %v", c.Field, err) + } + } +} From d4e7e40a09423db9579fdce587366f50061945f6 Mon Sep 17 00:00:00 2001 From: Andrei Kvapil Date: Sat, 17 Oct 2026 12:23:27 +0500 Subject: [PATCH 3/3] docs: describe trafficDistribution on older apiservers Assisted-by: LLM Signed-off-by: Andrei Kvapil --- api/v1alpha2/etcdcluster_types.go | 4 +++- .../crd-bases/etcd-operator.cozystack.io_etcdclusters.yaml | 4 +++- docs/concepts.md | 2 +- 3 files changed, 7 insertions(+), 3 deletions(-) diff --git a/api/v1alpha2/etcdcluster_types.go b/api/v1alpha2/etcdcluster_types.go index 2e88106f..91f846a0 100644 --- a/api/v1alpha2/etcdcluster_types.go +++ b/api/v1alpha2/etcdcluster_types.go @@ -600,7 +600,9 @@ type ClientServiceSpec struct { // 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). + // feature gate, beta and on by default from 1.34). A value the apiserver + // rejects or drops is reported with a Warning Event, left off the + // Service and retried hourly; it does not block the reconcile. // +kubebuilder:validation:Enum=PreferClose;PreferSameZone;PreferSameNode // +optional TrafficDistribution *string `json:"trafficDistribution,omitempty"` diff --git a/charts/etcd-operator/crd-bases/etcd-operator.cozystack.io_etcdclusters.yaml b/charts/etcd-operator/crd-bases/etcd-operator.cozystack.io_etcdclusters.yaml index 8663fa1a..ca1c73e1 100644 --- a/charts/etcd-operator/crd-bases/etcd-operator.cozystack.io_etcdclusters.yaml +++ b/charts/etcd-operator/crd-bases/etcd-operator.cozystack.io_etcdclusters.yaml @@ -1193,7 +1193,9 @@ spec: 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). + feature gate, beta and on by default from 1.34). A value the apiserver + rejects or drops is reported with a Warning Event, left off the + Service and retried hourly; it does not block the reconcile. enum: - PreferClose - PreferSameZone diff --git a/docs/concepts.md b/docs/concepts.md index b0e0103a..08fa4474 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -300,7 +300,7 @@ Semantics: - **Owned only when set.** With the field set, the operator writes it on create and reconciles any drift back to it. With it unset (no `clientService`, or no `trafficDistribution` in it), the operator does not touch the field — a value patched onto the Service by hand survives. - **Client Service only.** The headless Service is never given a traffic distribution. - **Not latched.** The Service carries no consensus state, so unlike the member-level fields a change is applied on the next reconcile rather than waiting for the current `status.observed` target. -- **Apiserver support.** `PreferClose` needs Kubernetes 1.31+ (beta, on by default; GA in 1.33). `PreferSameZone` and `PreferSameNode` need 1.34+, where the `PreferSameTrafficDistribution` feature gate is beta and on by default. An apiserver that does not know the field drops it silently (the setting has no effect); one that knows the field but not the value rejects the Service write, which surfaces as a reconcile error for that cluster. +- **Apiserver support.** `PreferClose` needs Kubernetes 1.31+ (beta, on by default; GA in 1.33). `PreferSameZone` and `PreferSameNode` need 1.34+, where the `PreferSameTrafficDistribution` feature gate is beta and on by default. On an older apiserver the setting never blocks the cluster: if the apiserver rejects the value, or does not know the field and drops it, the operator reconciles the Service without it, emits a Warning Event on the `EtcdCluster` (`TrafficDistributionRejected` or `TrafficDistributionNotPersisted`) and tries the value again an hour later, so it lands on its own once the apiserver is upgraded. Changing the value in the spec retries at once. ## etcd tuning options