public_git/postgres-operator
1
0
Fork 0
mirror of https://github.com/zalando/postgres-operator.git synced 2026-10-05 21:45:21 +02:00
Code Issues Projects Releases Wiki Activity

Merge branch 'master' into bump-v1.8.1

Browse Source
This commit is contained in:
Felix Kunde
2022-05-19 11:34:56 +02:00
parent 170c9a826c 6268fb826a
commit 6aa3ce7693
16 changed files with 238 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/cluster/cluster.go
+15 -33
View File
@@ -1032,12 +1032,20 @@ func (c *Cluster) processPodEvent(obj interface{}) error {
return fmt.Errorf("could not cast to PodEvent")
}
// can only take lock when (un)registerPodSubscriber is finshed
c.podSubscribersMu.RLock()
subscriber, ok := c.podSubscribers[spec.NamespacedName(event.PodName)]
c.podSubscribersMu.RUnlock()
if ok {
subscriber <- event
select {
case subscriber <- event:
default:
// ending up here when there is no receiver on the channel (i.e. waitForPodLabel finished)
// 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()
return nil
}
@@ -1501,34 +1509,16 @@ func (c *Cluster) Switchover(curMaster *v1.Pod, candidate spec.NamespacedName) e
var err error
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)
var wg sync.WaitGroup
podLabelErr := make(chan error)
stopCh := make(chan struct{})
wg.Add(1)
go func() {
defer wg.Done()
ch := c.registerPodSubscriber(candidate)
defer c.unregisterPodSubscriber(candidate)
role := Master
select {
case <-stopCh:
case podLabelErr <- func() (err2 error) {
_, err2 = c.waitForPodLabel(ch, stopCh, &role)
return
}():
}
}()
ch := c.registerPodSubscriber(candidate)
defer c.unregisterPodSubscriber(candidate)
defer close(stopCh)
if err = c.patroni.Switchover(curMaster, candidate.Name); err == nil {
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)
if err = <-podLabelErr; err != nil {
_, err = c.waitForPodLabel(ch, stopCh, nil)
if err != nil {
err = fmt.Errorf("could not get master pod label: %v", err)
}
} else {
@@ -1536,14 +1526,6 @@ func (c *Cluster) Switchover(curMaster *v1.Pod, candidate spec.NamespacedName) e
c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeNormal, "Switchover", "Switchover from %q to %q FAILED: %v", curMaster.Name, candidate, err)
}
// signal the role label waiting goroutine to close the shop and go home
close(stopCh)
// wait until the goroutine terminates, since unregisterPodSubscriber
// must be called before the outer return; otherwise we risk subscribing to the same pod twice.
wg.Wait()
// close the label waiting channel no sooner than the waiting goroutine terminates.
close(podLabelErr)
return err
}
pkg/cluster/k8sres.go
+1 -2
View File
@@ -940,7 +940,6 @@ func (c *Cluster) generateSpiloPodEnvVars(
func appendEnvVars(envs []v1.EnvVar, appEnv ...v1.EnvVar) []v1.EnvVar {
collectedEnvs := envs
for _, env := range appEnv {
env.Name = strings.ToUpper(env.Name)
if !isEnvVarPresent(collectedEnvs, env.Name) {
collectedEnvs = append(collectedEnvs, env)
}
@@ -950,7 +949,7 @@ func appendEnvVars(envs []v1.EnvVar, appEnv ...v1.EnvVar) []v1.EnvVar {
func isEnvVarPresent(envs []v1.EnvVar, key string) bool {
for _, env := range envs {
if env.Name == key {
if strings.EqualFold(env.Name, key) {
return true
}
}
pkg/cluster/k8sres_test.go
+6 -6
View File
@@ -504,7 +504,7 @@ func TestGenerateSpiloPodEnvVars(t *testing.T) {
expectedS3BucketConfigMap := []ExpectedValue{
{
envIndex: 17,
envVarConstant: "WAL_S3_BUCKET",
envVarConstant: "wal_s3_bucket",
envVarValue: "global-s3-bucket-configmap",
},
}
@@ -518,7 +518,7 @@ func TestGenerateSpiloPodEnvVars(t *testing.T) {
expectedCustomVariableSecret := []ExpectedValue{
{
envIndex: 16,
envVarConstant: "CUSTOM_VARIABLE",
envVarConstant: "custom_variable",
envVarValueRef: &v1.EnvVarSource{
SecretKeyRef: &v1.SecretKeySelector{
LocalObjectReference: v1.LocalObjectReference{
@@ -532,7 +532,7 @@ func TestGenerateSpiloPodEnvVars(t *testing.T) {
expectedCustomVariableConfigMap := []ExpectedValue{
{
envIndex: 16,
envVarConstant: "CUSTOM_VARIABLE",
envVarConstant: "custom_variable",
envVarValue: "configmap-test",
},
}
@@ -573,14 +573,14 @@ func TestGenerateSpiloPodEnvVars(t *testing.T) {
},
{
envIndex: 20,
envVarConstant: "CLONE_AWS_ENDPOINT",
envVarConstant: "clone_aws_endpoint",
envVarValue: "s3.eu-west-1.amazonaws.com",
},
}
expectedCloneEnvSecret := []ExpectedValue{
{
envIndex: 20,
envVarConstant: "CLONE_AWS_ACCESS_KEY_ID",
envVarConstant: "clone_aws_access_key_id",
envVarValueRef: &v1.EnvVarSource{
SecretKeyRef: &v1.SecretKeySelector{
LocalObjectReference: v1.LocalObjectReference{
@@ -599,7 +599,7 @@ func TestGenerateSpiloPodEnvVars(t *testing.T) {
},
{
envIndex: 20,
envVarConstant: "STANDBY_GOOGLE_APPLICATION_CREDENTIALS",
envVarConstant: "standby_google_application_credentials",
envVarValueRef: &v1.EnvVarSource{
SecretKeyRef: &v1.SecretKeySelector{
LocalObjectReference: v1.LocalObjectReference{
pkg/cluster/pod.go
+11 -9
View File
@@ -67,7 +67,7 @@ func (c *Cluster) markRollingUpdateFlagForPod(pod *v1.Pod, msg string) error {
return fmt.Errorf("could not form patch for pod's rolling update flag: %v", err)
}
err = retryutil.Retry(c.OpConfig.PatroniAPICheckInterval, c.OpConfig.PatroniAPICheckTimeout,
err = retryutil.Retry(1*time.Second, 5*time.Second,
func() (bool, error) {
_, err2 := c.KubeClient.Pods(pod.Namespace).Patch(
context.TODO(),
@@ -151,12 +151,13 @@ func (c *Cluster) unregisterPodSubscriber(podName spec.NamespacedName) {
c.podSubscribersMu.Lock()
defer c.podSubscribersMu.Unlock()
if _, ok := c.podSubscribers[podName]; !ok {
ch, ok := c.podSubscribers[podName]
if !ok {
panic("subscriber for pod '" + podName.String() + "' is not found")
}
close(c.podSubscribers[podName])
delete(c.podSubscribers, podName)
close(ch)
}
func (c *Cluster) registerPodSubscriber(podName spec.NamespacedName) chan PodEvent {
@@ -399,11 +400,12 @@ 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)
stopChan := make(chan struct{})
defer close(stopCh)
err := retryutil.Retry(c.OpConfig.PatroniAPICheckInterval, c.OpConfig.PatroniAPICheckTimeout,
err := retryutil.Retry(1*time.Second, 5*time.Second,
func() (bool, error) {
err2 := c.KubeClient.Pods(podName.Namespace).Delete(
context.TODO(),
@@ -421,7 +423,7 @@ func (c *Cluster) recreatePod(podName spec.NamespacedName) (*v1.Pod, error) {
if err := c.waitForPodDeletion(ch); err != nil {
return nil, err
}
pod, err := c.waitForPodLabel(ch, stopChan, nil)
pod, err := c.waitForPodLabel(ch, stopCh, nil)
if err != nil {
return nil, err
}
@@ -446,7 +448,7 @@ func (c *Cluster) recreatePods(pods []v1.Pod, switchoverCandidates []spec.Namesp
continue
}
podName := util.NameFromMeta(pod.ObjectMeta)
podName := util.NameFromMeta(pods[i].ObjectMeta)
newPod, err := c.recreatePod(podName)
if err != nil {
return fmt.Errorf("could not recreate replica pod %q: %v", util.NameFromMeta(pod.ObjectMeta), err)
@@ -520,13 +522,13 @@ func (c *Cluster) getSwitchoverCandidate(master *v1.Pod) (spec.NamespacedName, e
// if sync_standby replicas were found assume synchronous_mode is enabled and ignore other candidates list
if len(syncCandidates) > 0 {
sort.Slice(syncCandidates, func(i, j int) bool {
return util.IntFromIntStr(syncCandidates[i].Lag) < util.IntFromIntStr(syncCandidates[j].Lag)
return syncCandidates[i].Lag < syncCandidates[j].Lag
})
return spec.NamespacedName{Namespace: master.Namespace, Name: syncCandidates[0].Name}, nil
}
if len(candidates) > 0 {
sort.Slice(candidates, func(i, j int) bool {
return util.IntFromIntStr(candidates[i].Lag) < util.IntFromIntStr(candidates[j].Lag)
return candidates[i].Lag < candidates[j].Lag
})
return spec.NamespacedName{Namespace: master.Namespace, Name: candidates[0].Name}, nil
}
pkg/cluster/util.go
+2 -2
View File
@@ -316,7 +316,7 @@ func (c *Cluster) annotationsSet(annotations map[string]string) map[string]strin
return nil
}
func (c *Cluster) waitForPodLabel(podEvents chan PodEvent, stopChan chan struct{}, role *PostgresRole) (*v1.Pod, error) {
func (c *Cluster) waitForPodLabel(podEvents chan PodEvent, stopCh chan struct{}, role *PostgresRole) (*v1.Pod, error) {
timeout := time.After(c.OpConfig.PodLabelWaitTimeout)
for {
select {
@@ -332,7 +332,7 @@ func (c *Cluster) waitForPodLabel(podEvents chan PodEvent, stopChan chan struct{
}
case <-timeout:
return nil, fmt.Errorf("pod label wait timeout")
case <-stopChan:
case <-stopCh:
return nil, fmt.Errorf("pod label wait cancelled")
}
}
pkg/controller/controller.go
+1 -1
View File
@@ -451,7 +451,7 @@ func (c *Controller) Run(stopCh <-chan struct{}, wg *sync.WaitGroup) {
panic("could not acquire initial list of clusters")
}
wg.Add(5)
wg.Add(5 + util.Bool2Int(c.opConfig.EnablePostgresTeamCRD))
go c.runPodInformer(stopCh, wg)
go c.runPostgresqlInformer(stopCh, wg)
go c.clusterResync(stopCh, wg)
pkg/controller/postgresql.go
+1 -1
View File
@@ -225,7 +225,7 @@ func (c *Controller) processEvent(event ClusterEvent) {
switch event.EventType {
case EventAdd:
if clusterFound {
lg.Infof("recieved add event for already existing Postgres cluster")
lg.Infof("received add event for already existing Postgres cluster")
return
}
pkg/util/patroni/patroni.go
+22 -6
View File
@@ -5,6 +5,7 @@ import (
"encoding/json"
"fmt"
"io/ioutil"
"math"
"net"
"net/http"
"strconv"
@@ -16,7 +17,6 @@ import (
"github.com/sirupsen/logrus"
acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/util/intstr"
)
const (
@@ -185,11 +185,27 @@ type ClusterMembers struct {
// ClusterMember cluster member data from Patroni API
type ClusterMember struct {
Name string `json:"name"`
Role string `json:"role"`
State string `json:"state"`
Timeline int `json:"timeline"`
Lag intstr.IntOrString `json:"lag,omitempty"`
Name string `json:"name"`
Role string `json:"role"`
State string `json:"state"`
Timeline int `json:"timeline"`
Lag ReplicationLag `json:"lag,omitempty"`
}
type ReplicationLag uint64
// UnmarshalJSON converts member lag (can be int or string) into uint64
func (rl *ReplicationLag) UnmarshalJSON(data []byte) error {
var lagUInt64 uint64
if data[0] == '"' {
*rl = math.MaxUint64
return nil
}
if err := json.Unmarshal(data, &lagUInt64); err != nil {
return err
}
*rl = ReplicationLag(lagUInt64)
return nil
}
// MemberDataPatroni child element
pkg/util/patroni/patroni_test.go
+15 -4
View File
@@ -5,6 +5,7 @@ import (
"errors"
"fmt"
"io/ioutil"
"math"
"net/http"
"reflect"
"testing"
@@ -15,7 +16,6 @@ import (
acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/util/intstr"
)
var logger = logrus.New().WithField("test", "patroni")
@@ -101,16 +101,27 @@ func TestGetClusterMembers(t *testing.T) {
Role: "sync_standby",
State: "running",
Timeline: 1,
Lag: intstr.IntOrString{IntVal: 0},
Lag: 0,
}, {
Name: "acid-test-cluster-2",
Role: "replica",
State: "running",
Timeline: 1,
Lag: intstr.IntOrString{Type: 1, StrVal: "unknown"},
Lag: math.MaxUint64,
}, {
Name: "acid-test-cluster-3",
Role: "replica",
State: "running",
Timeline: 1,
Lag: 3000000000,
}}
json := `{"members": [{"name": "acid-test-cluster-0", "role": "leader", "state": "running", "api_url": "http://192.168.100.1:8008/patroni", "host": "192.168.100.1", "port": 5432, "timeline": 1}, {"name": "acid-test-cluster-1", "role": "sync_standby", "state": "running", "api_url": "http://192.168.100.2:8008/patroni", "host": "192.168.100.2", "port": 5432, "timeline": 1, "lag": 0}, {"name": "acid-test-cluster-2", "role": "replica", "state": "running", "api_url": "http://192.168.100.3:8008/patroni", "host": "192.168.100.3", "port": 5432, "timeline": 1, "lag": "unknown"}]}`
json := `{"members": [
{"name": "acid-test-cluster-0", "role": "leader", "state": "running", "api_url": "http://192.168.100.1:8008/patroni", "host": "192.168.100.1", "port": 5432, "timeline": 1},
{"name": "acid-test-cluster-1", "role": "sync_standby", "state": "running", "api_url": "http://192.168.100.2:8008/patroni", "host": "192.168.100.2", "port": 5432, "timeline": 1, "lag": 0},
{"name": "acid-test-cluster-2", "role": "replica", "state": "running", "api_url": "http://192.168.100.3:8008/patroni", "host": "192.168.100.3", "port": 5432, "timeline": 1, "lag": "unknown"},
{"name": "acid-test-cluster-3", "role": "replica", "state": "running", "api_url": "http://192.168.100.3:8008/patroni", "host": "192.168.100.3", "port": 5432, "timeline": 1, "lag": 3000000000}
]}`
r := ioutil.NopCloser(bytes.NewReader([]byte(json)))
response := http.Response{
pkg/util/util.go
+6 -7
View File
@@ -8,7 +8,6 @@ import (
"encoding/base64"
"encoding/hex"
"fmt"
"math"
"math/big"
"math/rand"
"reflect"
@@ -324,18 +323,18 @@ func testNil(values ...*int32) bool {
return false
}
// Convert int to IntOrString type
// ToIntStr converts int to IntOrString type
func ToIntStr(val int) *intstr.IntOrString {
b := intstr.FromInt(val)
return &b
}
// Get int from IntOrString and return max int if string
func IntFromIntStr(intOrStr intstr.IntOrString) int {
if intOrStr.Type == 1 {
return math.MaxInt
// Bool2Int converts bool to int
func Bool2Int(flag bool) int {
if flag {
return 1
}
return intOrStr.IntValue()
return 0
}
// MaxInt32 : Return maximum of two integers provided via pointers. If one value