From d24d8ef3596045ab9ce55c1823095da57b4f1203 Mon Sep 17 00:00:00 2001 From: Jorge Solorzano Date: Sun, 1 Nov 2026 07:36:38 +0100 Subject: [PATCH 1/4] process cluster events outside of the queue lock FIFO.Pop holds the queue lock while calling the process function. Processing the cluster event inside of it blocked queueClusterEvent for this worker until the event was done, e.g. for the whole rolling update of a cluster. Since the postgresql informer handler queues the events, a single busy worker stalled the events of all clusters, leaving the other workers idle. Pop the event and process it after the lock is released. Now that events can pile up in a queue, update events get a key per resource version. Otherwise a pending update would be replaced by a later one and the changes of the first one, based on the diff of old and new spec, get lost. A delete event therefore discards pending events by cluster UID. Also guard the worker assignment map, which is accessed from the informer handlers, the periodic resync and the API server. Co-Authored-By: Claude Opus 5.5 --- pkg/controller/controller.go | 30 ++++++++++----- pkg/controller/postgresql.go | 39 +++++++++---------- pkg/controller/postgresql_test.go | 62 +++++++++++++++++++++++++++++++ pkg/controller/util.go | 4 ++ 4 files changed, 107 insertions(+), 28 deletions(-) diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go index 0d4638318..ae76d3d39 100644 --- a/pkg/controller/controller.go +++ b/pkg/controller/controller.go @@ -52,6 +52,7 @@ type Controller struct { controllerID string curWorkerID uint32 //initialized with 0 curWorkerCluster sync.Map + clusterWorkersMu sync.Mutex clusterWorkers map[spec.NamespacedName]uint32 clustersMu sync.RWMutex clusters map[spec.NamespacedName]*cluster.Cluster @@ -367,15 +368,8 @@ func (c *Controller) initController() { c.clusterEventQueues = make([]*cache.FIFO, c.opConfig.Workers) c.workerLogs = make(map[uint32]ringlog.RingLogger, c.opConfig.Workers) for i := range c.clusterEventQueues { - keyFn := func(obj interface{}) (string, error) { - e, ok := obj.(ClusterEvent) - if !ok { - return "", fmt.Errorf("could not cast to cluster event") - } - return queueClusterKey(e.EventType, e.UID), nil - } - c.clusterEventStores[i] = cache.NewStore(keyFn) - c.clusterEventQueues[i] = cache.NewFIFO(keyFn) + c.clusterEventStores[i] = cache.NewStore(clusterEventKeyFunc) + c.clusterEventQueues[i] = cache.NewFIFO(clusterEventKeyFunc) } c.apiserver = apiserver.New(c, c.opConfig.APIPort, c.logger.Logger) @@ -483,6 +477,24 @@ func queueClusterKey(eventType EventType, uid types.UID) string { return fmt.Sprintf("%s-%s", eventType, uid) } +// clusterEventKey returns the queue key of a cluster event. Events with the same key replace each other +// while waiting in the queue. Update events must not be merged, because the cluster update is based on +// the diff between their old and new spec, so their key also contains the resource version of the new spec. +func clusterEventKey(e ClusterEvent) string { + if e.EventType == EventUpdate && e.NewSpec != nil { + return fmt.Sprintf("%s-%s", queueClusterKey(e.EventType, e.UID), e.NewSpec.ResourceVersion) + } + return queueClusterKey(e.EventType, e.UID) +} + +func clusterEventKeyFunc(obj interface{}) (string, error) { + e, ok := obj.(ClusterEvent) + if !ok { + return "", fmt.Errorf("could not cast to cluster event") + } + return clusterEventKey(e), nil +} + func (c *Controller) kubeNodesInformer(stopCh <-chan struct{}, wg *sync.WaitGroup) { defer wg.Done() diff --git a/pkg/controller/postgresql.go b/pkg/controller/postgresql.go index d2c7b535f..d0fd770ff 100644 --- a/pkg/controller/postgresql.go +++ b/pkg/controller/postgresql.go @@ -363,16 +363,12 @@ func (c *Controller) processClusterEventsQueue(idx int, stopCh <-chan struct{}, }() for { - _, err := (*c.clusterEventQueues[idx]).Pop(cache.PopProcessFunc(func(obj interface{}, isInitialList bool) error { - event, ok := obj.(ClusterEvent) - if !ok { - c.logger.Errorf("could not cast to cluster event") - return nil // skip event to keep processing - } - c.processEvent(event, isInitialList) - if err := c.clusterEventStores[idx].Delete(obj); err != nil { - c.logger.Errorf("failed to delete key from lookup store: %v", err) - } + var isInitialList bool + // Pop holds the queue lock while calling the process function, so the event is only taken from + // the queue there and processed afterwards. Otherwise queueClusterEvent would block until the + // worker finishes, stalling the informer handler and with it the events of all other clusters. + obj, err := (*c.clusterEventQueues[idx]).Pop(cache.PopProcessFunc(func(_ interface{}, initialList bool) error { + isInitialList = initialList return nil })) if err != nil { @@ -382,6 +378,15 @@ func (c *Controller) processClusterEventsQueue(idx int, stopCh <-chan struct{}, c.logger.Errorf("error when processing cluster events queue: %v", err) continue } + event, ok := obj.(ClusterEvent) + if !ok { + c.logger.Errorf("could not cast to cluster event") + continue // skip event to keep processing + } + if err := c.clusterEventStores[idx].Delete(obj); err != nil { + c.logger.Errorf("failed to delete key from lookup store: %v", err) + } + c.processEvent(event, isInitialList) } } @@ -489,18 +494,14 @@ func (c *Controller) queueClusterEvent(informerOldSpec, informerNewSpec *acidv1. return } // A delete event discards all prior requests for that cluster. - for _, evType := range []EventType{EventAdd, EventSync, EventUpdate, EventRepair} { - obj, exists, err := c.clusterEventStores[workerID].GetByKey(queueClusterKey(evType, uid)) - if err != nil { - lg.Warningf("could not get event from the lookup store: %v", err) - continue - } - - if !exists { + for _, obj := range c.clusterEventStores[workerID].List() { + queuedEvent, ok := obj.(ClusterEvent) + if !ok || queuedEvent.UID != uid || queuedEvent.EventType == EventDelete { continue } + evType := queuedEvent.EventType - err = (*c.clusterEventQueues[workerID]).Delete(obj) + err := (*c.clusterEventQueues[workerID]).Delete(obj) if err != nil { lg.Warningf("could not delete event from the queue: %v", err) } else { diff --git a/pkg/controller/postgresql_test.go b/pkg/controller/postgresql_test.go index 68dea4bd9..6d4605e70 100644 --- a/pkg/controller/postgresql_test.go +++ b/pkg/controller/postgresql_test.go @@ -3,12 +3,15 @@ package controller import ( "fmt" "reflect" + "sort" "testing" "time" acidv1 "github.com/zalando/postgres-operator/v2/pkg/apis/acid.zalan.do/v1" "github.com/zalando/postgres-operator/v2/pkg/spec" + "github.com/zalando/postgres-operator/v2/pkg/util" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/tools/cache" ) var ( @@ -148,3 +151,62 @@ func TestMeetsClusterDeleteAnnotations(t *testing.T) { } } } + +func TestQueueClusterEvents(t *testing.T) { + c := NewController(&spec.ControllerConfig{}, "queue-test") + c.opConfig.Workers = 1 + c.opConfig.EnableOwnerReferences = util.True() // allows deletion without annotations + c.clusterEventStores = []cache.Store{cache.NewStore(clusterEventKeyFunc)} + c.clusterEventQueues = []*cache.FIFO{cache.NewFIFO(clusterEventKeyFunc)} + + pgWithVersion := func(resourceVersion string) *acidv1.Postgresql { + return &acidv1.Postgresql{ObjectMeta: metav1.ObjectMeta{ + Name: "acid-test", Namespace: "default", UID: "uid-1", ResourceVersion: resourceVersion}} + } + sorted := func(types []EventType) []EventType { + sort.Slice(types, func(i, j int) bool { return types[i] < types[j] }) + return types + } + stored := func() []EventType { + var types []EventType + for _, obj := range c.clusterEventStores[0].List() { + types = append(types, obj.(ClusterEvent).EventType) + } + return sorted(types) + } + // a closed queue still hands out the remaining events + drainQueue := func() []EventType { + var types []EventType + c.clusterEventQueues[0].Close() + for { + obj, err := c.clusterEventQueues[0].Pop(func(interface{}, bool) error { return nil }) + if err != nil { + return sorted(types) + } + types = append(types, obj.(ClusterEvent).EventType) + } + } + + // updates are not merged, because each one is applied based on its own old and new spec + c.queueClusterEvent(pgWithVersion("1"), pgWithVersion("2"), EventUpdate) + c.queueClusterEvent(pgWithVersion("2"), pgWithVersion("3"), EventUpdate) + // syncs replace each other + c.queueClusterEvent(nil, pgWithVersion("3"), EventSync) + c.queueClusterEvent(nil, pgWithVersion("3"), EventSync) + + expected := []EventType{EventSync, EventUpdate, EventUpdate} + if got := stored(); !reflect.DeepEqual(got, expected) { + t.Fatalf("expected pending events %v, got %v", expected, got) + } + + // a delete discards all pending events of the cluster + c.queueClusterEvent(pgWithVersion("3"), nil, EventDelete) + + expected = []EventType{EventDelete} + if got := stored(); !reflect.DeepEqual(got, expected) { + t.Fatalf("expected pending events %v after delete, got %v", expected, got) + } + if got := drainQueue(); !reflect.DeepEqual(got, expected) { + t.Fatalf("expected queued events %v after delete, got %v", expected, got) + } +} diff --git a/pkg/controller/util.go b/pkg/controller/util.go index eeacfac10..8e9d7798f 100644 --- a/pkg/controller/util.go +++ b/pkg/controller/util.go @@ -35,6 +35,10 @@ func (c *Controller) makeClusterConfig() cluster.Config { } func (c *Controller) clusterWorkerID(clusterName spec.NamespacedName) uint32 { + // called from the informer handlers, the periodic resync and the API server + c.clusterWorkersMu.Lock() + defer c.clusterWorkersMu.Unlock() + workerID, ok := c.clusterWorkers[clusterName] if ok { return workerID From 51722b5a411e88efd4764846fa508142cf2a3236 Mon Sep 17 00:00:00 2001 From: Jorge Solorzano Date: Sun, 1 Nov 2026 07:37:19 +0100 Subject: [PATCH 2/4] do not lose pod events while waiting on a pod Pod events are sent to the subscriber of a pod without blocking. With an unbuffered channel an event got dropped whenever the subscriber was not waiting on the channel at that very moment, e.g. when two events arrive back to back. A dropped delete event left the worker in waitForPodDeletion until pod_deletion_wait_timeout (10m) expired and the update failed, although the pod was recreated fine. Buffer the subscriber channel and log dropped events. Also recognize a deletion the informer missed: handle tombstones in the pod delete handler and treat an update replacing the pod with a new instance (different UID) as deleted. As events are buffered now, wait for the master label of the candidate after a switchover instead of any role label, skipping earlier replica events. Co-Authored-By: Claude Opus 5.5 --- pkg/cluster/cluster.go | 8 +++- pkg/cluster/pod.go | 4 +- pkg/cluster/pod_test.go | 97 +++++++++++++++++++++++++++++++++++++++++ pkg/cluster/types.go | 4 ++ pkg/cluster/util.go | 15 ++++++- pkg/controller/pod.go | 5 +++ 6 files changed, 129 insertions(+), 4 deletions(-) diff --git a/pkg/cluster/cluster.go b/pkg/cluster/cluster.go index 27e430207..200d5aa03 100644 --- a/pkg/cluster/cluster.go +++ b/pkg/cluster/cluster.go @@ -1382,8 +1382,10 @@ func (c *Cluster) processPodEvent(obj interface{}, isInInitialList bool) error { select { case subscriber <- event: default: - // ending up here when there is no receiver on the channel (i.e. waitForPodLabel finished) + // ending up here when the subscriber's buffer is full, i.e. the subscriber finished waiting + // but has not unregistered yet or is not consuming events anymore // avoids blocking channel: https://gobyexample.com/non-blocking-channel-operations + c.logger.Warningf("dropping %s event of pod %q: subscriber is not receiving events", event.EventType, event.PodName) } } // hold lock for the time of processing the event to avoid race condition @@ -1912,7 +1914,9 @@ func (c *Cluster) Switchover(curMaster *v1.Pod, candidate spec.NamespacedName, s } c.logger.Debugf("successfully switched over from %q to %q", curMaster.Name, candidate) c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeNormal, "Switchover", "Successfully switched over from %q to %q", curMaster.Name, candidate) - _, err = c.waitForPodLabel(ch, stopCh, nil) + // pod events of the candidate are buffered since subscribing, so skip the ones still labeled as replica + master := Master + _, err = c.waitForPodLabel(ch, stopCh, &master) if err != nil { err = fmt.Errorf("could not get master pod label: %v", err) } diff --git a/pkg/cluster/pod.go b/pkg/cluster/pod.go index 3acdabd53..098064855 100644 --- a/pkg/cluster/pod.go +++ b/pkg/cluster/pod.go @@ -189,7 +189,9 @@ func (c *Cluster) registerPodSubscriber(podName spec.NamespacedName) chan PodEve c.podSubscribersMu.Lock() defer c.podSubscribersMu.Unlock() - ch := make(chan PodEvent) + // buffered, because events are sent without blocking and would be lost if the subscriber + // is not waiting on the channel at that very moment, e.g. when two events arrive back to back + ch := make(chan PodEvent, podSubscriberBufferSize) if _, ok := c.podSubscribers[podName]; ok { panic("pod '" + podName.String() + "' is already subscribed") } diff --git a/pkg/cluster/pod_test.go b/pkg/cluster/pod_test.go index 8c7170368..76e36ee0c 100644 --- a/pkg/cluster/pod_test.go +++ b/pkg/cluster/pod_test.go @@ -20,6 +20,7 @@ import ( v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" k8sfake "k8s.io/client-go/kubernetes/fake" k8stesting "k8s.io/client-go/testing" "k8s.io/client-go/tools/record" @@ -541,3 +542,99 @@ func TestAllPodsRunning(t *testing.T) { }) } } + +func TestPodEventsBufferedForSubscriber(t *testing.T) { + podName := spec.NamespacedName{Namespace: "default", Name: "acid-test-cluster-0"} + oldPod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: podName.Name, Namespace: podName.Namespace, UID: "old", + Labels: map[string]string{"spilo-role": "master"}}} + newPod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: podName.Name, Namespace: podName.Namespace, UID: "new"}} + labeledPod := newPod.DeepCopy() + labeledPod.Labels = map[string]string{"spilo-role": "master"} + + opConfig := config.Config{} + opConfig.PodRoleLabel = "spilo-role" + opConfig.PodDeletionWaitTimeout = &metav1.Duration{Duration: time.Second} + opConfig.PodLabelWaitTimeout = &metav1.Duration{Duration: time.Second} + c := New(Config{OpConfig: opConfig}, k8sutil.KubernetesClient{}, + acidv1.Postgresql{ObjectMeta: metav1.ObjectMeta{Name: "acid-test-cluster", Namespace: podName.Namespace}}, logger, record.NewFakeRecorder(2)) + + ch := c.registerPodSubscriber(podName) + defer c.unregisterPodSubscriber(podName) + + // events delivered back to back while nobody is waiting on the channel yet must not get lost + for _, event := range []PodEvent{ + {PodName: types.NamespacedName(podName), EventType: PodEventUpdate, PrevPod: oldPod, CurPod: oldPod}, + {PodName: types.NamespacedName(podName), EventType: PodEventDelete, CurPod: oldPod}, + {PodName: types.NamespacedName(podName), EventType: PodEventAdd, CurPod: newPod}, + {PodName: types.NamespacedName(podName), EventType: PodEventUpdate, PrevPod: newPod, CurPod: labeledPod}, + } { + if err := c.processPodEvent(event, false); err != nil { + t.Fatalf("could not process pod event: %v", err) + } + } + + if err := c.waitForPodDeletion(ch); err != nil { + t.Fatalf("pod deletion not observed: %v", err) + } + pod, err := c.waitForPodLabel(ch, make(chan struct{}), nil) + if err != nil { + t.Fatalf("pod label not observed: %v", err) + } + if pod.UID != "new" { + t.Errorf("expected the new pod, got pod with uid %q", pod.UID) + } +} + +func TestIsPodGone(t *testing.T) { + oldPod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: "old"}} + newPod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: "new"}} + + for _, tt := range []struct { + name string + event PodEvent + expected bool + }{ + {"delete", PodEvent{EventType: PodEventDelete, CurPod: oldPod}, true}, + {"update of the same pod", PodEvent{EventType: PodEventUpdate, PrevPod: oldPod, CurPod: oldPod}, false}, + {"update replacing the pod", PodEvent{EventType: PodEventUpdate, PrevPod: oldPod, CurPod: newPod}, true}, + {"add", PodEvent{EventType: PodEventAdd, CurPod: newPod}, false}, + } { + if got := isPodGone(tt.event); got != tt.expected { + t.Errorf("%s: expected %v, got %v", tt.name, tt.expected, got) + } + } +} + +func TestWaitForMasterLabelSkipsReplicaEvents(t *testing.T) { + podName := spec.NamespacedName{Namespace: "default", Name: "acid-test-cluster-1"} + replicaPod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: podName.Name, Namespace: podName.Namespace, + Labels: map[string]string{"spilo-role": "replica"}}} + masterPod := replicaPod.DeepCopy() + masterPod.Labels["spilo-role"] = "master" + + opConfig := config.Config{} + opConfig.PodRoleLabel = "spilo-role" + opConfig.PodLabelWaitTimeout = &metav1.Duration{Duration: time.Second} + c := New(Config{OpConfig: opConfig}, k8sutil.KubernetesClient{}, + acidv1.Postgresql{ObjectMeta: metav1.ObjectMeta{Name: "acid-test-cluster", Namespace: podName.Namespace}}, logger, record.NewFakeRecorder(2)) + + ch := c.registerPodSubscriber(podName) + defer c.unregisterPodSubscriber(podName) + + // e.g. Patroni updating the member status annotation of the candidate before it gets promoted + for _, pod := range []*v1.Pod{replicaPod, masterPod} { + event := PodEvent{PodName: types.NamespacedName(podName), EventType: PodEventUpdate, PrevPod: replicaPod, CurPod: pod} + if err := c.processPodEvent(event, false); err != nil { + t.Fatalf("could not process pod event: %v", err) + } + } + + master := Master + pod, err := c.waitForPodLabel(ch, make(chan struct{}), &master) + if err != nil { + t.Fatalf("master label not observed: %v", err) + } + if role := PostgresRole(pod.Labels["spilo-role"]); role != Master { + t.Errorf("expected pod labeled %q, got %q", Master, role) + } +} diff --git a/pkg/cluster/types.go b/pkg/cluster/types.go index da3fd29f6..fcdc29a74 100644 --- a/pkg/cluster/types.go +++ b/pkg/cluster/types.go @@ -35,6 +35,10 @@ const ( PodEventDelete PodEventType = "DELETE" ) +// podSubscriberBufferSize is the number of pod events buffered for a subscriber waiting on a pod. +// A pod emits only a handful of events while it is being recreated. +const podSubscriberBufferSize = 100 + // PodEvent describes the event for a single Pod type PodEvent struct { ResourceVersion string diff --git a/pkg/cluster/util.go b/pkg/cluster/util.go index ce4d4f205..bb3c63793 100644 --- a/pkg/cluster/util.go +++ b/pkg/cluster/util.go @@ -363,12 +363,25 @@ func (c *Cluster) waitForPodLabel(podEvents chan PodEvent, stopCh chan struct{}, } } +// isPodGone tells if the event shows that the pod the subscriber waits on has been deleted. +// Besides the delete event itself, an update replacing the pod with a new instance of the same name +// also proves it, e.g. when the informer missed the deletion and only observed the new pod after a relist. +func isPodGone(podEvent PodEvent) bool { + switch podEvent.EventType { + case PodEventDelete: + return true + case PodEventUpdate: + return podEvent.PrevPod != nil && podEvent.CurPod != nil && podEvent.PrevPod.UID != podEvent.CurPod.UID + } + return false +} + func (c *Cluster) waitForPodDeletion(podEvents chan PodEvent) error { timeout := time.After(c.OpConfig.PodDeletionWaitTimeout.Duration) for { select { case podEvent := <-podEvents: - if podEvent.EventType == PodEventDelete { + if isPodGone(podEvent) { return nil } case <-timeout: diff --git a/pkg/controller/pod.go b/pkg/controller/pod.go index f3d1c882d..c3dd393c9 100644 --- a/pkg/controller/pod.go +++ b/pkg/controller/pod.go @@ -7,6 +7,7 @@ import ( "github.com/zalando/postgres-operator/v2/pkg/spec" "github.com/zalando/postgres-operator/v2/pkg/util" "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/tools/cache" ) func (c *Controller) dispatchPodEvent(clusterName spec.NamespacedName, event cluster.PodEvent) { @@ -39,6 +40,10 @@ func (c *Controller) podUpdate(prev, cur interface{}) { } func (c *Controller) podDelete(obj interface{}) { + // the informer hands over a tombstone when it missed the deletion and only noticed it on a relist + if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok { + obj = tombstone.Obj + } if pod, ok := obj.(*v1.Pod); ok { c.preparePodEventForDispatch(pod, nil, cluster.PodEventDelete) From 09691db099ee8cb945b5c04a89bac490cb5acb8e Mon Sep 17 00:00:00 2001 From: Jorge Solorzano Date: Sun, 1 Nov 2026 07:37:31 +0100 Subject: [PATCH 3/4] retry Postgres connections with exponential backoff The connection was retried every 15s (the connect timeout). Right after a pod is (re)created the first attempt usually fails, so every rolling update spent 15s waiting before syncing roles. Start retrying after 1s and double the interval up to 15s, keeping the overall retry timeout of 2m. Co-Authored-By: Claude Opus 5.5 --- pkg/cluster/database.go | 4 ++- pkg/util/constants/postgresql.go | 5 ++-- pkg/util/retryutil/retry_util.go | 38 +++++++++++++++++++++++++++ pkg/util/retryutil/retry_util_test.go | 38 +++++++++++++++++++++++++++ 4 files changed, 82 insertions(+), 3 deletions(-) diff --git a/pkg/cluster/database.go b/pkg/cluster/database.go index d46dc6067..11fa7d32d 100644 --- a/pkg/cluster/database.go +++ b/pkg/cluster/database.go @@ -135,7 +135,9 @@ func (c *Cluster) initDbConnWithName(dbname string) error { var conn *sql.DB connstring := c.pgConnectionString(dbname) - finalerr := retryutil.Retry(constants.PostgresConnectTimeout, constants.PostgresConnectRetryTimeout, + // right after a pod (re)start the first attempts usually fail, so start retrying quickly + finalerr := retryutil.RetryWithBackoff(constants.PostgresConnectRetryInitialInterval, + constants.PostgresConnectTimeout, constants.PostgresConnectRetryTimeout, func() (bool, error) { var err error conn, err = sql.Open("postgres", connstring) diff --git a/pkg/util/constants/postgresql.go b/pkg/util/constants/postgresql.go index 8bd7508a7..ebbb18338 100644 --- a/pkg/util/constants/postgresql.go +++ b/pkg/util/constants/postgresql.go @@ -10,8 +10,9 @@ const ( PatroniPGParametersParameterName = "parameters" - PostgresConnectRetryTimeout = 2 * time.Minute - PostgresConnectTimeout = 15 * time.Second + PostgresConnectRetryTimeout = 2 * time.Minute + PostgresConnectTimeout = 15 * time.Second + PostgresConnectRetryInitialInterval = 1 * time.Second ShmVolumeName = "dshm" ShmVolumePath = "/dev/shm" diff --git a/pkg/util/retryutil/retry_util.go b/pkg/util/retryutil/retry_util.go index 868ba6e98..3fc09956f 100644 --- a/pkg/util/retryutil/retry_util.go +++ b/pkg/util/retryutil/retry_util.go @@ -33,6 +33,44 @@ func Retry(interval time.Duration, timeout time.Duration, f func() (bool, error) return RetryWorker(interval, timeout, tick, f) } +// RetryWithBackoff is a wrapper around RetryWithBackoffWorker that sleeps for real +func RetryWithBackoff(initialInterval, maxInterval, timeout time.Duration, f func() (bool, error)) error { + if timeout < initialInterval { + return fmt.Errorf("timeout(%s) should be greater than initial interval(%v)", timeout, initialInterval) + } + return RetryWithBackoffWorker(initialInterval, maxInterval, timeout, time.Sleep, f) +} + +// RetryWithBackoffWorker calls ConditionFunc like RetryWorker, but starts with initialInterval +// between the calls and doubles it after every failed call up to maxInterval. +// It gives up once the time spent waiting would exceed the timeout. +func RetryWithBackoffWorker( + initialInterval time.Duration, + maxInterval time.Duration, + timeout time.Duration, + sleep func(time.Duration), + f func() (bool, error)) error { + + interval := initialInterval + var waited time.Duration + + for i := 1; ; i++ { + ok, err := f() + if err != nil { + return err + } + if ok { + return nil + } + if waited+interval > timeout { + return fmt.Errorf("still failing after %d retries", i) + } + sleep(interval) + waited += interval + interval = min(2*interval, maxInterval) + } +} + // RetryWorker calls ConditionFunc until either: // * it returns boolean true // * a timeout expires diff --git a/pkg/util/retryutil/retry_util_test.go b/pkg/util/retryutil/retry_util_test.go index 56061767f..95192f679 100644 --- a/pkg/util/retryutil/retry_util_test.go +++ b/pkg/util/retryutil/retry_util_test.go @@ -2,7 +2,9 @@ package retryutil import ( "errors" + "reflect" "testing" + "time" ) type mockTicker struct { @@ -61,3 +63,39 @@ func TestRetryWorkerError(t *testing.T) { t.Errorf("Wrong result, expected: %#v, got: %#v", fail, result) } } + +func TestRetryWithBackoffWorker(t *testing.T) { + fail := errors.New("Error") + + tests := []struct { + name string + succeedAfter int + err error + expectedSleep []time.Duration + expectErr bool + }{ + {"succeeds at once", 1, nil, nil, false}, + {"interval doubles up to the max", 5, nil, []time.Duration{1, 2, 4, 5}, false}, + {"gives up at the timeout", 100, nil, []time.Duration{1, 2, 4, 5, 5}, true}, + {"stops on error", 100, fail, nil, true}, + } + + for _, tt := range tests { + var calls int + var slept []time.Duration + result := RetryWithBackoffWorker(1, 5, 20, func(d time.Duration) { slept = append(slept, d) }, func() (bool, error) { + calls++ + if tt.err != nil { + return false, tt.err + } + return calls >= tt.succeedAfter, nil + }) + + if (result != nil) != tt.expectErr { + t.Errorf("%s: unexpected result %v", tt.name, result) + } + if !reflect.DeepEqual(slept, tt.expectedSleep) { + t.Errorf("%s: expected sleeps %v, got %v", tt.name, tt.expectedSleep, slept) + } + } +} From ba9091eaa5466ef5ebba1aed9dea8d82a95418fb Mon Sep 17 00:00:00 2001 From: Jorge Solorzano Date: Sun, 1 Nov 2026 07:37:47 +0100 Subject: [PATCH 4/4] run cluster repair scans on the repair period The resync ticker fired every resync period only, where a sync was always due, so the repair scan of failed clusters effectively never ran every repair_period. Tick at the shorter of both periods and decide whether a sync or a repair is due. Record the time of the tick as the last sync or repair time instead of the time after listing and queueing all clusters, and tolerate ticks arriving slightly early, so a due sync is not postponed by a whole tick. A repair period that is not positive disables the repair scans. Co-Authored-By: Claude Opus 5.5 --- pkg/controller/postgresql.go | 50 ++++++++++++++------- pkg/controller/postgresql_test.go | 73 +++++++++++++++++++++++++++++++ 2 files changed, 106 insertions(+), 17 deletions(-) diff --git a/pkg/controller/postgresql.go b/pkg/controller/postgresql.go index d0fd770ff..5103f1939 100644 --- a/pkg/controller/postgresql.go +++ b/pkg/controller/postgresql.go @@ -25,14 +25,17 @@ import ( "github.com/zalando/postgres-operator/v2/pkg/util/ringlog" ) +// clusterResync queues a sync of all clusters every resync period and a repair scan +// of the clusters in a failed state every repair period. func (c *Controller) clusterResync(stopCh <-chan struct{}, wg *sync.WaitGroup) { defer wg.Done() - ticker := time.NewTicker(c.opConfig.ResyncPeriod.Duration) + ticker := time.NewTicker(c.resyncTickPeriod()) + defer ticker.Stop() for { select { - case <-ticker.C: - if err := c.clusterListAndSync(); err != nil { + case tickTime := <-ticker.C: + if err := c.clusterListAndSync(tickTime); err != nil { c.logger.Errorf("could not list clusters: %v", err) } case <-stopCh: @@ -41,6 +44,15 @@ func (c *Controller) clusterResync(stopCh <-chan struct{}, wg *sync.WaitGroup) { } } +// resyncTickPeriod returns how often to check whether a sync or a repair is due, which is the shorter +// of both periods. A repair period that is not positive disables the repair scans. +func (c *Controller) resyncTickPeriod() time.Duration { + if c.opConfig.RepairPeriod.Duration > 0 { + return min(c.opConfig.ResyncPeriod.Duration, c.opConfig.RepairPeriod.Duration) + } + return c.opConfig.ResyncPeriod.Duration +} + // clusterListFunc obtains a list of all PostgreSQL clusters func (c *Controller) listClusters(options metav1.ListOptions) (*acidv1.PostgresqlList, error) { var pgList acidv1.PostgresqlList @@ -63,19 +75,21 @@ func (c *Controller) listClusters(options metav1.ListOptions) (*acidv1.Postgresq } // clusterListAndSync lists all manifests and decides whether to run the sync or repair. -func (c *Controller) clusterListAndSync() error { +func (c *Controller) clusterListAndSync(currentTime time.Time) error { var ( err error event EventType ) - currentTime := time.Now().Unix() - timeFromPreviousSync := currentTime - atomic.LoadInt64(&c.lastClusterSyncTime) - timeFromPreviousRepair := currentTime - atomic.LoadInt64(&c.lastClusterRepairTime) + timeFromPreviousSync := currentTime.Sub(time.Unix(atomic.LoadInt64(&c.lastClusterSyncTime), 0)) + timeFromPreviousRepair := currentTime.Sub(time.Unix(atomic.LoadInt64(&c.lastClusterRepairTime), 0)) + // ticks arrive with a small delay and the last times are stored in seconds, so a tick can come slightly + // before a period elapsed. Tolerate that, otherwise the due sync or repair is postponed by a whole tick. + tolerance := c.resyncTickPeriod() / 10 - if timeFromPreviousSync >= int64(c.opConfig.ResyncPeriod.Seconds()) { + if timeFromPreviousSync >= c.opConfig.ResyncPeriod.Duration-tolerance { event = EventSync - } else if timeFromPreviousRepair >= int64(c.opConfig.RepairPeriod.Seconds()) { + } else if c.opConfig.RepairPeriod.Duration > 0 && timeFromPreviousRepair >= c.opConfig.RepairPeriod.Duration-tolerance { event = EventRepair } if event != "" { @@ -83,16 +97,17 @@ func (c *Controller) clusterListAndSync() error { if list, err = c.listClusters(metav1.ListOptions{ResourceVersion: "0"}); err != nil { return err } - c.queueEvents(list, event) + c.queueEvents(list, event, currentTime) } else { - c.logger.Infof("not enough time passed since the last sync (%v seconds) or repair (%v seconds)", - timeFromPreviousSync, timeFromPreviousRepair) + c.logger.Infof("not enough time passed since the last sync (%v) or repair (%v)", + timeFromPreviousSync.Round(time.Second), timeFromPreviousRepair.Round(time.Second)) } return nil } -// queueEvents queues a sync or repair event for every cluster with a valid manifest -func (c *Controller) queueEvents(list *acidv1.PostgresqlList, event EventType) { +// queueEvents queues a sync or repair event for every cluster with a valid manifest. +// The scheduled time is recorded as the time of the last sync or repair. +func (c *Controller) queueEvents(list *acidv1.PostgresqlList, event EventType, scheduledTime time.Time) { var activeClustersCnt, failedClustersCnt, clustersToRepair int for i, pg := range list.Items { // XXX: check the cluster status field instead @@ -126,9 +141,9 @@ func (c *Controller) queueEvents(list *acidv1.PostgresqlList, event EventType) { c.logger.Infof("no clusters running") } if event == EventRepair || event == EventSync { - atomic.StoreInt64(&c.lastClusterRepairTime, time.Now().Unix()) + atomic.StoreInt64(&c.lastClusterRepairTime, scheduledTime.Unix()) if event == EventSync { - atomic.StoreInt64(&c.lastClusterSyncTime, time.Now().Unix()) + atomic.StoreInt64(&c.lastClusterSyncTime, scheduledTime.Unix()) } } } @@ -140,6 +155,7 @@ func (c *Controller) acquireInitialListOfClusters() error { clusterName spec.NamespacedName ) + startTime := time.Now() if list, err = c.listClusters(metav1.ListOptions{ResourceVersion: "0"}); err != nil { return err } @@ -154,7 +170,7 @@ func (c *Controller) acquireInitialListOfClusters() error { c.logger.Debugf("added new cluster: %q", clusterName) } // initiate initial sync of all clusters. - c.queueEvents(list, EventSync) + c.queueEvents(list, EventSync, startTime) return nil } diff --git a/pkg/controller/postgresql_test.go b/pkg/controller/postgresql_test.go index 6d4605e70..1868d3b65 100644 --- a/pkg/controller/postgresql_test.go +++ b/pkg/controller/postgresql_test.go @@ -8,9 +8,13 @@ import ( "time" acidv1 "github.com/zalando/postgres-operator/v2/pkg/apis/acid.zalan.do/v1" + fakeacidv1 "github.com/zalando/postgres-operator/v2/pkg/generated/clientset/versioned/fake" "github.com/zalando/postgres-operator/v2/pkg/spec" "github.com/zalando/postgres-operator/v2/pkg/util" + "github.com/zalando/postgres-operator/v2/pkg/util/k8sutil" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + k8stesting "k8s.io/client-go/testing" "k8s.io/client-go/tools/cache" ) @@ -210,3 +214,72 @@ func TestQueueClusterEvents(t *testing.T) { t.Fatalf("expected queued events %v after delete, got %v", expected, got) } } + +func TestClusterListAndSync(t *testing.T) { + healthy := acidv1.Postgresql{ObjectMeta: metav1.ObjectMeta{Name: "acid-healthy", Namespace: "default", UID: "uid-healthy"}, + Status: acidv1.PostgresStatus{PostgresClusterStatus: acidv1.ClusterStatusRunning}} + failed := acidv1.Postgresql{ObjectMeta: metav1.ObjectMeta{Name: "acid-failed", Namespace: "default", UID: "uid-failed"}, + Status: acidv1.PostgresStatus{PostgresClusterStatus: acidv1.ClusterStatusUpdateFailed}} + now := time.Now() + + tests := []struct { + name string + repairPeriod time.Duration + sinceSync time.Duration + sinceRepair time.Duration + expectedTick time.Duration + expectedEvent EventType + expectedCount int + }{ + {"sync due", 5 * time.Minute, 30 * time.Minute, 5 * time.Minute, 5 * time.Minute, EventSync, 2}, + {"sync due with tick arriving early", 5 * time.Minute, 30*time.Minute - time.Second, 5 * time.Minute, 5 * time.Minute, EventSync, 2}, + {"repair due between syncs", 5 * time.Minute, 10 * time.Minute, 5 * time.Minute, 5 * time.Minute, EventRepair, 1}, + {"repair due with tick arriving early", 5 * time.Minute, 10 * time.Minute, 5*time.Minute - time.Second, 5 * time.Minute, EventRepair, 1}, + {"nothing due", 5 * time.Minute, 2 * time.Minute, 2 * time.Minute, 5 * time.Minute, "", 0}, + {"repair disabled", 0, 10 * time.Minute, 10 * time.Minute, 30 * time.Minute, "", 0}, + {"repair period longer than resync", time.Hour, 30 * time.Minute, 30 * time.Minute, 30 * time.Minute, EventSync, 2}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + c := NewController(&spec.ControllerConfig{}, "") + c.opConfig.Workers = 1 + c.opConfig.ResyncPeriod = &metav1.Duration{Duration: 30 * time.Minute} + c.opConfig.RepairPeriod = &metav1.Duration{Duration: tt.repairPeriod} + // the list kind of the CRD is registered as "postgresqlList", which the fake object tracker cannot list + acidClient := fakeacidv1.NewSimpleClientset() + acidClient.PrependReactor("list", "postgresqls", func(k8stesting.Action) (bool, runtime.Object, error) { + return true, &acidv1.PostgresqlList{Items: []acidv1.Postgresql{*healthy.DeepCopy(), *failed.DeepCopy()}}, nil + }) + c.KubeClient = k8sutil.KubernetesClient{PostgresqlsGetter: acidClient.AcidV1()} + c.clusterEventStores = []cache.Store{cache.NewStore(clusterEventKeyFunc)} + c.clusterEventQueues = []*cache.FIFO{cache.NewFIFO(clusterEventKeyFunc)} + c.lastClusterSyncTime = now.Add(-tt.sinceSync).Unix() + c.lastClusterRepairTime = now.Add(-tt.sinceRepair).Unix() + + if tick := c.resyncTickPeriod(); tick != tt.expectedTick { + t.Errorf("expected tick period %v, got %v", tt.expectedTick, tick) + } + if err := c.clusterListAndSync(now); err != nil { + t.Fatalf("clusterListAndSync failed: %v", err) + } + + queued := c.clusterEventStores[0].List() + if len(queued) != tt.expectedCount { + t.Fatalf("expected %d queued events, got %d", tt.expectedCount, len(queued)) + } + for _, obj := range queued { + if e := obj.(ClusterEvent); e.EventType != tt.expectedEvent { + t.Errorf("expected %s event, got %s for %s", tt.expectedEvent, e.EventType, e.NewSpec.Name) + } + } + // the time of the tick is recorded, not the time when queueing finished + if tt.expectedEvent == EventSync && c.lastClusterSyncTime != now.Unix() { + t.Errorf("expected last sync time %d, got %d", now.Unix(), c.lastClusterSyncTime) + } + if tt.expectedEvent != "" && c.lastClusterRepairTime != now.Unix() { + t.Errorf("expected last repair time %d, got %d", now.Unix(), c.lastClusterRepairTime) + } + }) + } +}