fix(controllers): keep reconciling when trafficDistribution is unsupported

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)
+		}
+	}
+}

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"`

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

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()
+}