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

reflect latest change in restart logic

Browse Source
This commit is contained in:
Felix Kunde
2021-10-26 17:06:06 +02:00
parent db76b8c642 1eafd688d0
commit a5acfd3b42
9 changed files with 303 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/cluster/sync.go
+82 -41
View File
@@ -6,6 +6,7 @@ import (
"fmt"
"reflect"
"regexp"
"strconv"
"strings"
"time"
@@ -20,6 +21,14 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
var requireMasterRestartWhenDecreased = []string{
"max_connections",
"max_prepared_transactions",
"max_locks_per_transaction",
"max_worker_processes",
"max_wal_senders",
}
// Sync syncs the cluster, making sure the actual Kubernetes objects correspond to what is defined in the manifest.
// Unlike the update, sync does not error out if some objects do not exist and takes care of creating them.
func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error {
@@ -270,11 +279,9 @@ func (c *Cluster) syncPodDisruptionBudget(isUpdate bool) error {
func (c *Cluster) syncStatefulSet() error {
var (
masterPod *v1.Pod
restartTTL uint32
instanceRestartRequired bool
restartWait uint32
restartMasterFirst bool
)
podsToRecreate := make([]v1.Pod, 0)
switchoverCandidates := make([]spec.NamespacedName, 0)
@@ -408,35 +415,43 @@ func (c *Cluster) syncStatefulSet() error {
c.logger.Warningf("could not get Postgres config from pod %s: %v", podName, err)
continue
}
restartWait = patroniConfig.LoopWait
// empty config probably means cluster is not fully initialized yet, e.g. restoring from backup
// do not attempt a restart
if !reflect.DeepEqual(patroniConfig, emptyPatroniConfig) || len(pgParameters) > 0 {
instanceRestartRequired, err = c.checkAndSetGlobalPostgreSQLConfiguration(&pod, patroniConfig, c.Spec.Patroni, pgParameters, c.Spec.Parameters)
restartMasterFirst, err = c.checkAndSetGlobalPostgreSQLConfiguration(&pod, patroniConfig, c.Spec.Patroni, pgParameters, c.Spec.Parameters)
if err != nil {
c.logger.Warningf("could not set PostgreSQL configuration options for pod %s: %v", podName, err)
continue
}
restartTTL = patroniConfig.TTL
// it could take up to LoopWait to apply the config
time.Sleep(time.Duration(restartWait)*time.Second + time.Second*2)
break
}
}
// if the config update requires a restart, call Patroni restart for replicas first, then master
if instanceRestartRequired {
c.logger.Debug("restarting Postgres server within pods")
for i, pod := range pods {
role := PostgresRole(pod.Labels[c.OpConfig.PodRoleLabel])
if role == Master {
masterPod = &pods[i]
continue
}
c.restartInstance(&pod)
time.Sleep(time.Duration(restartTTL) * time.Second)
// restart instances if required
remainingPods := make([]*v1.Pod, 0)
skipRole := Master
if restartMasterFirst {
skipRole = Replica
}
for i, pod := range pods {
role := PostgresRole(pod.Labels[c.OpConfig.PodRoleLabel])
if role == skipRole {
remainingPods = append(remainingPods, &pods[i])
continue
}
c.restartInstance(&pod, restartWait)
}
if masterPod != nil {
c.restartInstance(masterPod)
// in most cases only the master should be left to restart
if len(remainingPods) > 0 {
for _, remainingPod := range remainingPods {
c.restartInstance(remainingPod, restartWait)
}
}
@@ -453,19 +468,27 @@ func (c *Cluster) syncStatefulSet() error {
return nil
}
func (c *Cluster) restartInstance(pod *v1.Pod) {
func (c *Cluster) restartInstance(pod *v1.Pod, restartWait uint32) {
podName := util.NameFromMeta(pod.ObjectMeta)
role := PostgresRole(pod.Labels[c.OpConfig.PodRoleLabel])
c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Update", fmt.Sprintf("restarting Postgres server within %s pod %s", role, pod.Name))
if err := c.patroni.Restart(pod); err != nil {
c.logger.Warningf("could not restart Postgres server within %s pod %s: %v", role, podName, err)
// if the config update requires a restart, call Patroni restart
memberData, err := c.patroni.GetMemberData(pod)
if err != nil {
c.logger.Debugf("could not get member data of %s pod %s - skipping possible restart attempt: %v", role, podName, err)
return
}
c.logger.Debugf("Postgres server successfuly restarted in %s pod %s", role, podName)
c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Update", fmt.Sprintf("Postgres server restart done for %s pod %s", role, pod.Name))
// do restart only when it is pending
if memberData.PendingRestart {
c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Update", fmt.Sprintf("restarting Postgres server within %s pod %s", role, pod.Name))
if err := c.patroni.Restart(pod); err != nil {
c.logger.Warningf("could not restart Postgres server within %s pod %s: %v", role, podName, err)
return
}
time.Sleep(time.Duration(restartWait) * time.Second)
c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Update", fmt.Sprintf("Postgres server restart done for %s pod %s", role, pod.Name))
}
}
// AnnotationsToPropagate get the annotations to update if required
@@ -505,20 +528,10 @@ func (c *Cluster) AnnotationsToPropagate(annotations map[string]string) map[stri
func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration(pod *v1.Pod, effectivePatroniConfig, desiredPatroniConfig acidv1.Patroni, effectivePgParameters, desiredPgParameters map[string]string) (bool, error) {
configToSet := make(map[string]interface{})
parametersToSet := make(map[string]string)
restartMaster := make([]bool, 0)
requiresMasterRestart := false
// compare parameters under postgresql section with c.Spec.Postgresql.Parameters from manifest
for desiredOption, desiredValue := range desiredPgParameters {
effectiveValue := effectivePgParameters[desiredOption]
if isBootstrapOnlyParameter(desiredOption) && (effectiveValue != desiredValue) {
parametersToSet[desiredOption] = desiredValue
}
}
if len(parametersToSet) > 0 {
configToSet["postgresql"] = map[string]interface{}{constants.PatroniPGParametersParameterName: parametersToSet}
}
// compare other options from config with c.Spec.Patroni from manifest
// compare effective and desired Patroni config options
if desiredPatroniConfig.LoopWait > 0 && desiredPatroniConfig.LoopWait != effectivePatroniConfig.LoopWait {
configToSet["loop_wait"] = desiredPatroniConfig.LoopWait
}
@@ -555,6 +568,34 @@ func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration(pod *v1.Pod, effectiv
configToSet["slots"] = slotsToSet
}
// compare parameters under postgresql section with c.Spec.Postgresql.Parameters from manifest
for desiredOption, desiredValue := range desiredPgParameters {
effectiveValue := effectivePgParameters[desiredOption]
if isBootstrapOnlyParameter(desiredOption) && (effectiveValue != desiredValue) {
parametersToSet[desiredOption] = desiredValue
if util.SliceContains(requireMasterRestartWhenDecreased, desiredOption) {
effectiveValueNum, errConv := strconv.Atoi(effectiveValue)
desiredValueNum, errConv2 := strconv.Atoi(desiredValue)
if errConv != nil || errConv2 != nil {
continue
}
if effectiveValueNum > desiredValueNum {
restartMaster = append(restartMaster, true)
continue
}
}
restartMaster = append(restartMaster, false)
}
}
if !util.SliceContains(restartMaster, false) && len(configToSet) == 0 {
requiresMasterRestart = true
}
if len(parametersToSet) > 0 {
configToSet["postgresql"] = map[string]interface{}{constants.PatroniPGParametersParameterName: parametersToSet}
}
if len(configToSet) == 0 {
return false, nil
}
@@ -570,10 +611,10 @@ func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration(pod *v1.Pod, effectiv
c.logger.Debugf("patching Postgres config via Patroni API on pod %s with following options: %s",
podName, configToSetJson)
if err = c.patroni.SetConfig(pod, configToSet); err != nil {
return true, fmt.Errorf("could not patch postgres parameters within pod %s: %v", podName, err)
return requiresMasterRestart, fmt.Errorf("could not patch postgres parameters with a pod %s: %v", podName, err)
}
return true, nil
return requiresMasterRestart, nil
}
func (c *Cluster) syncSecrets() error {
pkg/cluster/sync_test.go
+149
View File
@@ -1,22 +1,40 @@
package cluster
import (
"bytes"
"io/ioutil"
"net/http"
"testing"
"time"
"context"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
"github.com/golang/mock/gomock"
"github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
"github.com/zalando/postgres-operator/mocks"
acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
fakeacidv1 "github.com/zalando/postgres-operator/pkg/generated/clientset/versioned/fake"
"github.com/zalando/postgres-operator/pkg/util/config"
"github.com/zalando/postgres-operator/pkg/util/k8sutil"
"github.com/zalando/postgres-operator/pkg/util/patroni"
"k8s.io/client-go/kubernetes/fake"
)
var patroniLogger = logrus.New().WithField("test", "patroni")
func newMockPod(ip string) *v1.Pod {
return &v1.Pod{
Status: v1.PodStatus{
PodIP: ip,
},
}
}
func newFakeK8sSyncClient() (k8sutil.KubernetesClient, *fake.Clientset) {
acidClientSet := fakeacidv1.NewSimpleClientset()
clientSet := fake.NewSimpleClientset()
@@ -113,3 +131,134 @@ func TestSyncStatefulSetsAnnotations(t *testing.T) {
t.Errorf("%s: inherited annotation not found in desired statefulset: %#v", testName, desiredSts.Annotations)
}
}
func TestCheckAndSetGlobalPostgreSQLConfiguration(t *testing.T) {
testName := "test config comparison"
client, _ := newFakeK8sSyncClient()
clusterName := "acid-test-cluster"
namespace := "default"
ctrl := gomock.NewController(t)
defer ctrl.Finish()
pg := acidv1.Postgresql{
ObjectMeta: metav1.ObjectMeta{
Name: clusterName,
Namespace: namespace,
},
Spec: acidv1.PostgresSpec{
Patroni: acidv1.Patroni{
TTL: 20,
},
PostgresqlParam: acidv1.PostgresqlParam{
Parameters: map[string]string{
"log_min_duration_statement": "200",
"max_connections": "50",
},
},
Volume: acidv1.Volume{
Size: "1Gi",
},
},
}
var cluster = New(
Config{
OpConfig: config.Config{
PodManagementPolicy: "ordered_ready",
Resources: config.Resources{
ClusterLabels: map[string]string{"application": "spilo"},
ClusterNameLabel: "cluster-name",
DefaultCPURequest: "300m",
DefaultCPULimit: "300m",
DefaultMemoryRequest: "300Mi",
DefaultMemoryLimit: "300Mi",
PodRoleLabel: "spilo-role",
ResourceCheckInterval: time.Duration(3),
ResourceCheckTimeout: time.Duration(10),
},
},
}, client, pg, logger, eventRecorder)
// mocking a config after setConfig is called
configJson := `{"postgresql": {"parameters": {"log_min_duration_statement": 200, "max_connections": 50}}}, "ttl": 20}`
r := ioutil.NopCloser(bytes.NewReader([]byte(configJson)))
response := http.Response{
StatusCode: 200,
Body: r,
}
mockClient := mocks.NewMockHTTPClient(ctrl)
mockClient.EXPECT().Do(gomock.Any()).Return(&response, nil).AnyTimes()
p := patroni.New(patroniLogger, mockClient)
cluster.patroni = p
mockPod := newMockPod("192.168.100.1")
// simulate existing config that differs with cluster.Spec
tests := []struct {
subtest string
pod *v1.Pod
patroni acidv1.Patroni
pgParams map[string]string
restartMaster bool
}{
{
subtest: "Patroni and Postgresql.Parameters differ - restart replica first",
pod: mockPod,
patroni: acidv1.Patroni{
TTL: 30, // desired 20
},
pgParams: map[string]string{
"log_min_duration_statement": "500", // desired 200
"max_connections": "100", // desired 50
},
restartMaster: false,
},
{
subtest: "multiple Postgresql.Parameters differ - restart replica first",
pod: mockPod,
patroni: acidv1.Patroni{
TTL: 20,
},
pgParams: map[string]string{
"log_min_duration_statement": "500", // desired 200
"max_connections": "100", // desired 50
},
restartMaster: false,
},
{
subtest: "desired max_connections bigger - restart replica first",
pod: mockPod,
patroni: acidv1.Patroni{
TTL: 20,
},
pgParams: map[string]string{
"log_min_duration_statement": "200",
"max_connections": "30", // desired 50
},
restartMaster: false,
},
{
subtest: "desired max_connections smaller - restart master first",
pod: mockPod,
patroni: acidv1.Patroni{
TTL: 20,
},
pgParams: map[string]string{
"log_min_duration_statement": "200",
"max_connections": "100", // desired 50
},
restartMaster: true,
},
}
for _, tt := range tests {
requireMasterRestart, err := cluster.checkAndSetGlobalPostgreSQLConfiguration(tt.pod, tt.patroni, cluster.Spec.Patroni, tt.pgParams, cluster.Spec.Parameters)
assert.NoError(t, err)
if requireMasterRestart != tt.restartMaster {
t.Errorf("%s - %s: unexpect master restart strategy, got %v, expected %v", testName, tt.subtest, requireMasterRestart, tt.restartMaster)
}
}
}
pkg/util/patroni/patroni.go
+3 -7
View File
@@ -238,16 +238,12 @@ func (p *Patroni) Restart(server *v1.Pod) error {
if err != nil {
return err
}
memberData, err := p.GetMemberData(server)
if err != nil {
if err := p.httpPostOrPatch(http.MethodPost, apiURLString+restartPath, buf); err != nil {
return err
}
p.logger.Infof("Postgres server successfuly restarted in pod %s", server.Name)
// do restart only when it is pending
if !memberData.PendingRestart {
return nil
}
return p.httpPostOrPatch(http.MethodPost, apiURLString+restartPath, buf)
return nil
}
// GetMemberData read member data from patroni API