public_git
/
postgres-operator
mirror of https://github.com/zalando/postgres-operator.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

resolve conflict

This commit is contained in:
Felix Kunde 2022-05-17 18:23:28 +02:00
parent 268a86a045
commit 7951e9cb02
3 changed files with 42 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
pkg/cluster/cluster.go
View File

@ -1031,11 +1031,15 @@ func (c *Cluster) processPodEvent(obj interface{}) error {
if !ok {
return fmt.Errorf("could not cast to PodEvent")
}
podName := spec.NamespacedName(event.PodName)
// can only take lock when (un)registerPodSubscriber is finshed
c.podSubscribersMu.RLock()
subscriber, ok := c.podSubscribers[spec.NamespacedName(event.PodName)]
subscriber, ok := c.podSubscribers[podName]
if ok {
if event.EventType == PodEventEnd {
c.unregisterPodSubscriber(podName)
} else {
select {
case subscriber <- event:
default:
@ -1043,6 +1047,7 @@ func (c *Cluster) processPodEvent(obj interface{}) error {
// avoids blocking channel: https://gobyexample.com/non-blocking-channel-operations
}
}
}
// hold lock for the time of processing the event to avoid race condition
// with unregisterPodSubscriber closing the channel (see #1876)
c.podSubscribersMu.RUnlock()
@ -1510,8 +1515,13 @@ func (c *Cluster) Switchover(curMaster *v1.Pod, candidate spec.NamespacedName) e
c.logger.Debugf("switching over from %q to %q", curMaster.Name, candidate)
c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeNormal, "Switchover", "Switching over from %q to %q", curMaster.Name, candidate)
stopCh := make(chan struct{})
ch := c.registerPodSubscriber(candidate)
defer c.unregisterPodSubscriber(candidate)
// create buffered channel so that go routine with processPodEvent does not block when
// consumer waitForPodLabel is already finished. Channel size should be big enough to
// receive an event for all pod label wait timeout + patroni API timeout seconds
chanSize := int(c.OpConfig.PodLabelWaitTimeout.Seconds() + c.OpConfig.PatroniAPICheckTimeout.Seconds())
ch := c.registerBufferedPodSubscriber(candidate, chanSize)
// send special end event to trigger removal of PodEvent channel by processPodEvent
defer c.ReceivePodEvent(PodEvent{PodName: types.NamespacedName(candidate), EventType: PodEventEnd})
defer close(stopCh)
if err = c.patroni.Switchover(curMaster, candidate.Name); err == nil {

30
pkg/cluster/pod.go
View File

@ -136,8 +136,9 @@ func (c *Cluster) deletePods() error {
func (c *Cluster) deletePod(podName spec.NamespacedName) error {
c.setProcessName("deleting pod %q", podName)
ch := c.registerPodSubscriber(podName)
defer c.unregisterPodSubscriber(podName)
ch := c.registerUnbufferedPodSubscriber(podName)
// send special end event to trigger removal of PodEvent channel by processPodEvent
defer c.ReceivePodEvent(PodEvent{PodName: types.NamespacedName(podName), EventType: PodEventEnd})
if err := c.KubeClient.Pods(podName.Namespace).Delete(context.TODO(), podName.Name, c.deleteOptions); err != nil {
return err
@ -155,17 +156,31 @@ func (c *Cluster) unregisterPodSubscriber(podName spec.NamespacedName) {
if !ok {
panic("subscriber for pod '" + podName.String() + "' is not found")
}
delete(c.podSubscribers, podName)
close(ch)
}
func (c *Cluster) registerPodSubscriber(podName spec.NamespacedName) chan PodEvent {
func (c *Cluster) registerUnbufferedPodSubscriber(podName spec.NamespacedName) chan PodEvent {
return c.registerPodSubscriber(podName, 0)
}
func (c *Cluster) registerBufferedPodSubscriber(podName spec.NamespacedName, chanSize int) chan PodEvent {
return c.registerPodSubscriber(podName, chanSize)
}
func (c *Cluster) registerPodSubscriber(podName spec.NamespacedName, chanSize int) chan PodEvent {
c.logger.Debugf("subscribing to pod %q", podName)
var ch chan PodEvent
c.podSubscribersMu.Lock()
defer c.podSubscribersMu.Unlock()
ch := make(chan PodEvent)
// buffered or unbuffered PodEvent channel
if chanSize > 0 {
ch = make(chan PodEvent, chanSize)
} else {
ch = make(chan PodEvent)
}
if _, ok := c.podSubscribers[podName]; ok {
panic("pod '" + podName.String() + "' is already subscribed")
}
@ -401,8 +416,9 @@ func (c *Cluster) getPatroniMemberData(pod *v1.Pod) (patroni.MemberData, error)
func (c *Cluster) recreatePod(podName spec.NamespacedName) (*v1.Pod, error) {
stopCh := make(chan struct{})
ch := c.registerPodSubscriber(podName)
defer c.unregisterPodSubscriber(podName)
ch := c.registerUnbufferedPodSubscriber(podName)
// send special end event to trigger removal of PodEvent channel by processPodEvent
defer c.ReceivePodEvent(PodEvent{PodName: types.NamespacedName(podName), EventType: PodEventEnd})
defer close(stopCh)
err := retryutil.Retry(1*time.Second, 5*time.Second,

1
pkg/cluster/types.go
View File

@ -32,6 +32,7 @@ const (
PodEventAdd PodEventType = "ADD"
PodEventUpdate PodEventType = "UPDATE"
PodEventDelete PodEventType = "DELETE"
PodEventEnd PodEventType = "END"
)
// PodEvent describes the event for a single Pod