Fix cluster event processing

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

pkg/cluster/database.go

@@ -156,7 +156,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)

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

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

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

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:

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

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)