resolve conflict with latest config sync

This commit is contained in:
Felix Kunde
2021-10-18 17:23:46 +02:00
46 changed files with 876 additions and 571 deletions
+3 -6
View File
@@ -504,12 +504,6 @@ var PostgresCRDResourceValidation = apiextv1.CustomResourceValidation{
"version": {
Type: "string",
Enum: []apiextv1.JSON{
{
Raw: []byte(`"9.3"`),
},
{
Raw: []byte(`"9.4"`),
},
{
Raw: []byte(`"9.5"`),
},
@@ -528,6 +522,9 @@ var PostgresCRDResourceValidation = apiextv1.CustomResourceValidation{
{
Raw: []byte(`"13"`),
},
{
Raw: []byte(`"14"`),
},
},
},
"parameters": {
@@ -44,8 +44,8 @@ type PostgresUsersConfiguration struct {
// MajorVersionUpgradeConfiguration defines how to execute major version upgrades of Postgres.
type MajorVersionUpgradeConfiguration struct {
MajorVersionUpgradeMode string `json:"major_version_upgrade_mode" default:"off"` // off - no actions, manual - manifest triggers action, full - manifest and minimal version violation trigger upgrade
MinimalMajorVersion string `json:"minimal_major_version" default:"9.5"`
TargetMajorVersion string `json:"target_major_version" default:"13"`
MinimalMajorVersion string `json:"minimal_major_version" default:"9.6"`
TargetMajorVersion string `json:"target_major_version" default:"14"`
}
// KubernetesMetaConfiguration defines k8s conf required for all Postgres clusters and the operator itself
+8 -9
View File
@@ -30,13 +30,12 @@ import (
)
const (
pgBinariesLocationTemplate = "/usr/lib/postgresql/%v/bin"
patroniPGBinariesParameterName = "bin_dir"
patroniPGParametersParameterName = "parameters"
patroniPGHBAConfParameterName = "pg_hba"
localHost = "127.0.0.1/32"
connectionPoolerContainer = "connection-pooler"
pgPort = 5432
pgBinariesLocationTemplate = "/usr/lib/postgresql/%v/bin"
patroniPGBinariesParameterName = "bin_dir"
patroniPGHBAConfParameterName = "pg_hba"
localHost = "127.0.0.1/32"
connectionPoolerContainer = "connection-pooler"
pgPort = 5432
)
type pgUser struct {
@@ -277,11 +276,11 @@ PatroniInitDBParams:
local, bootstrap := getLocalAndBoostrapPostgreSQLParameters(pg.Parameters)
if len(local) > 0 {
config.PgLocalConfiguration[patroniPGParametersParameterName] = local
config.PgLocalConfiguration[constants.PatroniPGParametersParameterName] = local
}
if len(bootstrap) > 0 {
config.Bootstrap.DCS.PGBootstrapConfiguration = make(map[string]interface{})
config.Bootstrap.DCS.PGBootstrapConfiguration[patroniPGParametersParameterName] = bootstrap
config.Bootstrap.DCS.PGBootstrapConfiguration[constants.PatroniPGParametersParameterName] = bootstrap
}
}
// Patroni gives us a choice of writing pg_hba.conf to either the bootstrap section or to the local postgresql one.
+2 -1
View File
@@ -15,6 +15,7 @@ var VersionMap = map[string]int{
"11": 110000,
"12": 120000,
"13": 130000,
"14": 140000,
}
// IsBiggerPostgresVersion Compare two Postgres version numbers
@@ -33,7 +34,7 @@ func (c *Cluster) GetDesiredMajorVersionAsInt() int {
func (c *Cluster) GetDesiredMajorVersion() string {
if c.Config.OpConfig.MajorVersionUpgradeMode == "full" {
// current is 9.5, minimal is 11 allowing 11 to 13 clusters, everything below is upgraded
// e.g. current is 9.6, minimal is 11 allowing 11 to 14 clusters, everything below is upgraded
if IsBiggerPostgresVersion(c.Spec.PgVersion, c.Config.OpConfig.MinimalMajorVersion) {
c.logger.Infof("overwriting configured major version %s to %s", c.Spec.PgVersion, c.Config.OpConfig.TargetMajorVersion)
return c.Config.OpConfig.TargetMajorVersion
+8 -8
View File
@@ -59,11 +59,8 @@ func (c *Cluster) deleteStreams() error {
func (c *Cluster) syncPostgresConfig() error {
desiredPostgresConfig := make(map[string]interface{})
slots := make(map[string]map[string]string)
// if streams are defined wal_level must be switched to logical and slots have to be defined
desiredPostgresConfig["postgresql"] = map[string]interface{}{patroniPGParametersParameterName: map[string]string{"wal_level": "logical"}}
desiredPostgresConfig := c.Spec.Patroni
slots := desiredPostgresConfig.Slots
for _, stream := range c.Spec.Streams {
slotName := c.getLogicalReplicationSlot(stream.Database)
@@ -83,24 +80,27 @@ func (c *Cluster) syncPostgresConfig() error {
for slotName, slot := range slots {
c.logger.Debugf("creating logical replication slot %q in database %q", slotName, slot["database"])
}
desiredPostgresConfig["slots"] = slots
desiredPostgresConfig.Slots = slots
} else {
return nil
}
// if streams are defined wal_level must be switched to logical and slots have to be defined
desiredPgParameters := map[string]string{"wal_level": "logical"}
pods, err := c.listPods()
if err != nil || len(pods) == 0 {
c.logger.Warnf("could not list pods of the statefulset: %v", err)
}
for i, pod := range pods {
podName := util.NameFromMeta(pods[i].ObjectMeta)
effectivePostgresConfig, err := c.patroni.GetConfig(&pod)
effectivePostgresConfig, effectivePgParameters, err := c.patroni.GetConfig(&pod)
if err != nil {
c.logger.Warningf("could not get Postgres config from pod %s: %v", podName, err)
continue
}
_, err = c.checkAndSetGlobalPostgreSQLConfiguration(&pod, effectivePostgresConfig, desiredPostgresConfig)
_, err = c.checkAndSetGlobalPostgreSQLConfiguration(&pod, effectivePostgresConfig, desiredPostgresConfig, effectivePgParameters, desiredPgParameters)
if err != nil {
c.logger.Warningf("could not set PostgreSQL configuration options for pod %s: %v", podName, err)
continue
+66 -45
View File
@@ -275,7 +275,6 @@ func (c *Cluster) syncStatefulSet() error {
instanceRestartRequired bool
)
desiredPostgresConfig := make(map[string]interface{})
podsToRecreate := make([]v1.Pod, 0)
switchoverCandidates := make([]spec.NamespacedName, 0)
@@ -400,49 +399,27 @@ func (c *Cluster) syncStatefulSet() error {
}
// get Postgres config, compare with manifest and update via Patroni PATCH endpoint if it differs
// Patroni's config endpoint is just a "proxy" to DCS. It is enough to patch it only once and it doesn't matter which pod is used.
if len(c.Spec.Parameters) > 0 {
desiredPostgresConfig["postgresql"] = map[string]interface{}{patroniPGParametersParameterName: c.Spec.Parameters}
}
if c.Spec.Patroni.LoopWait > 0 {
desiredPostgresConfig["loop_wait"] = c.Spec.Patroni.LoopWait
}
if c.Spec.Patroni.MaximumLagOnFailover > 0 {
desiredPostgresConfig["maximum_lag_on_failover"] = c.Spec.Patroni.MaximumLagOnFailover
}
if c.Spec.Patroni.PgHba != nil {
desiredPostgresConfig["pg_hba"] = c.Spec.Patroni.PgHba
}
if c.Spec.Patroni.RetryTimeout > 0 {
desiredPostgresConfig["retry_timeout"] = c.Spec.Patroni.RetryTimeout
}
if c.Spec.Patroni.Slots != nil {
desiredPostgresConfig["slots"] = c.Spec.Patroni.Slots
}
if c.Spec.Patroni.TTL > 0 {
desiredPostgresConfig["ttl"] = c.Spec.Patroni.TTL
}
if effectivePostgresConfig["synchronous_mode"] != c.Spec.Patroni.SynchronousMode {
desiredPostgresConfig["synchronous_mode"] = c.Spec.Patroni.SynchronousMode
}
if effectivePostgresConfig["synchronous_mode_strict"] != c.Spec.Patroni.SynchronousModeStrict {
desiredPostgresConfig["synchronous_mode_strict"] = c.Spec.Patroni.SynchronousModeStrict
}
// Patroni's config endpoint is just a "proxy" to DCS. It is enough to patch it only once and it doesn't matter which pod is used
for i, pod := range pods {
emptyPatroniConfig := acidv1.Patroni{}
podName := util.NameFromMeta(pods[i].ObjectMeta)
effectivePostgresConfig, err := c.patroni.GetConfig(&pod)
patroniConfig, pgParameters, err := c.patroni.GetConfig(&pod)
if err != nil {
c.logger.Warningf("could not get Postgres config from pod %s: %v", podName, err)
continue
}
instanceRestartRequired, err = c.checkAndSetGlobalPostgreSQLConfiguration(&pod, effectivePostgresConfig, desiredPostgresConfig)
if err != nil {
c.logger.Warningf("could not set PostgreSQL configuration options for pod %s: %v", podName, err)
continue
// 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)
if err != nil {
c.logger.Warningf("could not set PostgreSQL configuration options for pod %s: %v", podName, err)
continue
}
break
}
break
}
// if the config update requires a restart, call Patroni restart for replicas first, then master
@@ -529,20 +506,64 @@ func (c *Cluster) AnnotationsToPropagate(annotations map[string]string) map[stri
// checkAndSetGlobalPostgreSQLConfiguration checks whether cluster-wide API parameters
// (like max_connections) have changed and if necessary sets it via the Patroni API
func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration(pod *v1.Pod, effectivePostgresConfig, desiredPostgresConfig map[string]interface{}) (bool, error) {
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)
configUpdateRequired := false
for parameter, value := range desiredPostgresConfig {
if !reflect.DeepEqual(effectivePostgresConfig[parameter], value) {
configUpdateRequired = true
break
// 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 !configUpdateRequired {
if len(parametersToSet) > 0 {
configToSet["postgresql"] = map[string]interface{}{constants.PatroniPGParametersParameterName: parametersToSet}
}
// compare other options from config with c.Spec.Patroni from manifest
if desiredPatroniConfig.LoopWait > 0 && desiredPatroniConfig.LoopWait != effectivePatroniConfig.LoopWait {
configToSet["loop_wait"] = desiredPatroniConfig.LoopWait
}
if desiredPatroniConfig.MaximumLagOnFailover > 0 && desiredPatroniConfig.MaximumLagOnFailover != effectivePatroniConfig.MaximumLagOnFailover {
configToSet["maximum_lag_on_failover"] = desiredPatroniConfig.MaximumLagOnFailover
}
if desiredPatroniConfig.PgHba != nil && !reflect.DeepEqual(desiredPatroniConfig.PgHba, effectivePatroniConfig.PgHba) {
configToSet["pg_hba"] = desiredPatroniConfig.PgHba
}
if desiredPatroniConfig.RetryTimeout > 0 && desiredPatroniConfig.RetryTimeout != effectivePatroniConfig.RetryTimeout {
configToSet["retry_timeout"] = desiredPatroniConfig.RetryTimeout
}
if desiredPatroniConfig.SynchronousMode != effectivePatroniConfig.SynchronousMode {
configToSet["synchronous_mode"] = desiredPatroniConfig.SynchronousMode
}
if desiredPatroniConfig.SynchronousModeStrict != effectivePatroniConfig.SynchronousModeStrict {
configToSet["synchronous_mode_strict"] = desiredPatroniConfig.SynchronousModeStrict
}
if desiredPatroniConfig.TTL > 0 && desiredPatroniConfig.TTL != effectivePatroniConfig.TTL {
configToSet["ttl"] = desiredPatroniConfig.TTL
}
// check if specified slots exist in config and if they differ
slotsToSet := make(map[string]map[string]string)
for slotName, desiredSlot := range desiredPatroniConfig.Slots {
if effectiveSlot, exists := effectivePatroniConfig.Slots[slotName]; exists {
if reflect.DeepEqual(desiredSlot, effectiveSlot) {
continue
}
}
slotsToSet[slotName] = desiredSlot
}
if len(slotsToSet) > 0 {
configToSet["slots"] = slotsToSet
}
if len(configToSet) == 0 {
return false, nil
}
configToSetJson, err := json.Marshal(desiredPostgresConfig)
configToSetJson, err := json.Marshal(configToSet)
if err != nil {
c.logger.Debugf("could not convert config patch to JSON: %v", err)
}
@@ -552,7 +573,7 @@ func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration(pod *v1.Pod, effectiv
podName := util.NameFromMeta(pod.ObjectMeta)
c.logger.Debugf("patching Postgres config via Patroni API on pod %s with following options: %s",
podName, configToSetJson)
if err = c.patroni.SetConfig(pod, desiredPostgresConfig); err != nil {
if err = c.patroni.SetConfig(pod, configToSet); err != nil {
return true, fmt.Errorf("could not patch postgres parameters within pod %s: %v", podName, err)
}
+1 -1
View File
@@ -128,7 +128,7 @@ func (c *Cluster) syncUnderlyingEBSVolume() error {
err = c.VolumeResizer.ModifyVolume(volume.VolumeID, modifyType, modifySize, modifyIops, modifyThroughput)
if err != nil {
errors = append(errors, fmt.Sprintf("modify volume failed: volume=%s size=%d iops=%d throughput=%d", volume.VolumeID, volume.Size, volume.Iops, volume.Throughput))
errors = append(errors, fmt.Sprintf("modify failed, showing current EBS values: volume-id=%s size=%d iops=%d throughput=%d", volume.VolumeID, volume.Size, volume.Iops, volume.Throughput))
}
}
}
+4 -3
View File
@@ -39,7 +39,7 @@ func (c *Controller) importConfigurationFromCRD(fromCRD *acidv1.OperatorConfigur
result.EnableSpiloWalPathCompat = fromCRD.EnableSpiloWalPathCompat
result.EtcdHost = fromCRD.EtcdHost
result.KubernetesUseConfigMaps = fromCRD.KubernetesUseConfigMaps
result.DockerImage = util.Coalesce(fromCRD.DockerImage, "registry.opensource.zalan.do/acid/spilo-13:2.1-p1")
result.DockerImage = util.Coalesce(fromCRD.DockerImage, "registry.opensource.zalan.do/acid/spilo-14:2.1-p2")
result.Workers = util.CoalesceUInt32(fromCRD.Workers, 8)
result.MinInstances = fromCRD.MinInstances
result.MaxInstances = fromCRD.MaxInstances
@@ -56,8 +56,8 @@ func (c *Controller) importConfigurationFromCRD(fromCRD *acidv1.OperatorConfigur
// major version upgrade config
result.MajorVersionUpgradeMode = util.Coalesce(fromCRD.MajorVersionUpgrade.MajorVersionUpgradeMode, "off")
result.MinimalMajorVersion = util.Coalesce(fromCRD.MajorVersionUpgrade.MinimalMajorVersion, "9.5")
result.TargetMajorVersion = util.Coalesce(fromCRD.MajorVersionUpgrade.TargetMajorVersion, "13")
result.MinimalMajorVersion = util.Coalesce(fromCRD.MajorVersionUpgrade.MinimalMajorVersion, "9.6")
result.TargetMajorVersion = util.Coalesce(fromCRD.MajorVersionUpgrade.TargetMajorVersion, "14")
// kubernetes config
result.CustomPodAnnotations = fromCRD.Kubernetes.CustomPodAnnotations
@@ -113,6 +113,7 @@ func (c *Controller) importConfigurationFromCRD(fromCRD *acidv1.OperatorConfigur
result.MasterPodMoveTimeout = util.CoalesceDuration(time.Duration(fromCRD.Kubernetes.MasterPodMoveTimeout), "10m")
result.EnablePodAntiAffinity = fromCRD.Kubernetes.EnablePodAntiAffinity
result.PodAntiAffinityTopologyKey = util.Coalesce(fromCRD.Kubernetes.PodAntiAffinityTopologyKey, "kubernetes.io/hostname")
result.PodToleration = fromCRD.Kubernetes.PodToleration
// Postgres Pod resources
result.DefaultCPURequest = util.Coalesce(fromCRD.PostgresPodResources.DefaultCPURequest, "100m")
+3 -3
View File
@@ -152,7 +152,7 @@ type Config struct {
WatchedNamespace string `name:"watched_namespace"` // special values: "*" means 'watch all namespaces', the empty string "" means 'watch a namespace where operator is deployed to'
KubernetesUseConfigMaps bool `name:"kubernetes_use_configmaps" default:"false"`
EtcdHost string `name:"etcd_host" default:""` // special values: the empty string "" means Patroni will use K8s as a DCS
DockerImage string `name:"docker_image" default:"registry.opensource.zalan.do/acid/spilo-13:2.1-p1"`
DockerImage string `name:"docker_image" default:"registry.opensource.zalan.do/acid/spilo-14:2.1-p2"`
SidecarImages map[string]string `name:"sidecar_docker_images"` // deprecated in favour of SidecarContainers
SidecarContainers []v1.Container `name:"sidecars"`
PodServiceAccountName string `name:"pod_service_account_name" default:"postgres-pod"`
@@ -212,8 +212,8 @@ type Config struct {
EnablePgVersionEnvVar bool `name:"enable_pgversion_env_var" default:"true"`
EnableSpiloWalPathCompat bool `name:"enable_spilo_wal_path_compat" default:"false"`
MajorVersionUpgradeMode string `name:"major_version_upgrade_mode" default:"off"`
MinimalMajorVersion string `name:"minimal_major_version" default:"9.5"`
TargetMajorVersion string `name:"target_major_version" default:"13"`
MinimalMajorVersion string `name:"minimal_major_version" default:"9.6"`
TargetMajorVersion string `name:"target_major_version" default:"14"`
}
// MustMarshal marshals the config or panics
+2
View File
@@ -8,6 +8,8 @@ const (
PostgresDataMount = "/home/postgres/pgdata"
PostgresDataPath = PostgresDataMount + "/pgroot"
PatroniPGParametersParameterName = "parameters"
PostgresConnectRetryTimeout = 2 * time.Minute
PostgresConnectTimeout = 15 * time.Second
+50 -42
View File
@@ -10,9 +10,11 @@ import (
"strconv"
"time"
"github.com/zalando/postgres-operator/pkg/util/constants"
httpclient "github.com/zalando/postgres-operator/pkg/util/httpclient"
"github.com/sirupsen/logrus"
acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
v1 "k8s.io/api/core/v1"
)
@@ -31,7 +33,7 @@ type Interface interface {
SetPostgresParameters(server *v1.Pod, options map[string]string) error
GetMemberData(server *v1.Pod) (MemberData, error)
Restart(server *v1.Pod) error
GetConfig(server *v1.Pod) (map[string]interface{}, error)
GetConfig(server *v1.Pod) (acidv1.Patroni, map[string]string, error)
SetConfig(server *v1.Pod, config map[string]interface{}) error
}
@@ -109,28 +111,23 @@ func (p *Patroni) httpPostOrPatch(method string, url string, body *bytes.Buffer)
}
func (p *Patroni) httpGet(url string) (string, error) {
request, err := http.NewRequest("GET", url, nil)
if err != nil {
return "", fmt.Errorf("could not create request: %v", err)
}
p.logger.Debugf("making GET http request: %s", url)
p.logger.Debugf("making GET http request: %s", request.URL.String())
resp, err := p.httpClient.Do(request)
response, err := p.httpClient.Get(url)
if err != nil {
return "", fmt.Errorf("could not make request: %v", err)
}
bodyBytes, err := ioutil.ReadAll(resp.Body)
defer response.Body.Close()
bodyBytes, err := ioutil.ReadAll(response.Body)
if err != nil {
return "", fmt.Errorf("could not read response: %v", err)
}
if err := resp.Body.Close(); err != nil {
return "", fmt.Errorf("could not close request: %v", err)
if response.StatusCode != http.StatusOK {
return string(bodyBytes), fmt.Errorf("patroni returned '%d'", response.StatusCode)
}
if resp.StatusCode != http.StatusOK {
return string(bodyBytes), fmt.Errorf("patroni returned '%d'", resp.StatusCode)
}
return string(bodyBytes), nil
}
@@ -194,30 +191,43 @@ type MemberData struct {
Patroni MemberDataPatroni `json:"patroni"`
}
func (p *Patroni) GetConfigOrStatus(server *v1.Pod, path string) (map[string]interface{}, error) {
result := make(map[string]interface{})
func (p *Patroni) GetConfig(server *v1.Pod) (acidv1.Patroni, map[string]string, error) {
var (
patroniConfig acidv1.Patroni
pgConfig map[string]interface{}
)
apiURLString, err := apiURL(server)
if err != nil {
return result, err
return patroniConfig, nil, err
}
body, err := p.httpGet(apiURLString + path)
err = json.Unmarshal([]byte(body), &result)
body, err := p.httpGet(apiURLString + configPath)
if err != nil {
return result, err
return patroniConfig, nil, err
}
err = json.Unmarshal([]byte(body), &patroniConfig)
if err != nil {
return patroniConfig, nil, err
}
return result, err
// unmarshalling postgresql parameters needs a detour
err = json.Unmarshal([]byte(body), &pgConfig)
if err != nil {
return patroniConfig, nil, err
}
pgParameters := make(map[string]string)
if _, exists := pgConfig["postgresql"]; exists {
effectivePostgresql := pgConfig["postgresql"].(map[string]interface{})
effectivePgParameters := effectivePostgresql[constants.PatroniPGParametersParameterName].(map[string]interface{})
for parameter, value := range effectivePgParameters {
strValue := fmt.Sprintf("%v", value)
pgParameters[parameter] = strValue
}
}
return patroniConfig, pgParameters, err
}
func (p *Patroni) GetStatus(server *v1.Pod) (map[string]interface{}, error) {
return p.GetConfigOrStatus(server, statusPath)
}
func (p *Patroni) GetConfig(server *v1.Pod) (map[string]interface{}, error) {
return p.GetConfigOrStatus(server, configPath)
}
//Restart method restarts instance via Patroni POST API call.
// Restart method restarts instance via Patroni POST API call.
func (p *Patroni) Restart(server *v1.Pod) error {
buf := &bytes.Buffer{}
err := json.NewEncoder(buf).Encode(map[string]interface{}{"restart_pending": true})
@@ -228,9 +238,13 @@ func (p *Patroni) Restart(server *v1.Pod) error {
if err != nil {
return err
}
status, err := p.GetStatus(server)
pending_restart, ok := status["pending_restart"]
if !ok || !pending_restart.(bool) {
memberData, err := p.GetMemberData(server)
if err != nil {
return err
}
// do restart only when it is pending
if !memberData.PendingRestart {
return nil
}
return p.httpPostOrPatch(http.MethodPost, apiURLString+restartPath, buf)
@@ -243,19 +257,13 @@ func (p *Patroni) GetMemberData(server *v1.Pod) (MemberData, error) {
if err != nil {
return MemberData{}, err
}
response, err := p.httpClient.Get(apiURLString)
body, err := p.httpGet(apiURLString + statusPath)
if err != nil {
return MemberData{}, fmt.Errorf("could not perform Get request: %v", err)
}
defer response.Body.Close()
body, err := ioutil.ReadAll(response.Body)
if err != nil {
return MemberData{}, fmt.Errorf("could not read response: %v", err)
return MemberData{}, err
}
data := MemberData{}
err = json.Unmarshal(body, &data)
err = json.Unmarshal([]byte(body), &data)
if err != nil {
return MemberData{}, err
}
+125 -10
View File
@@ -6,14 +6,19 @@ import (
"fmt"
"io/ioutil"
"net/http"
"reflect"
"testing"
"github.com/golang/mock/gomock"
"github.com/sirupsen/logrus"
"github.com/zalando/postgres-operator/mocks"
acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
v1 "k8s.io/api/core/v1"
)
var logger = logrus.New().WithField("test", "patroni")
func newMockPod(ip string) *v1.Pod {
return &v1.Pod{
Status: v1.PodStatus{
@@ -80,31 +85,141 @@ func TestApiURL(t *testing.T) {
}
}
func TestPatroniAPI(t *testing.T) {
func TestGetMemberData(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
json := `{"state": "running", "postmaster_start_time": "2021-02-19 14:31:50.053 CET", "role": "master", "server_version": 90621, "cluster_unlocked": false, "xlog": {"location": 55978296057856}, "timeline": 6, "database_system_identifier": "6462555844314089962", "pending_restart": true, "patroni": {"version": "2.0.1", "scope": "acid-rest92-standby"}}`
expectedMemberData := MemberData{
State: "running",
Role: "master",
ServerVersion: 130004,
PendingRestart: true,
Patroni: MemberDataPatroni{
Version: "2.1.1",
Scope: "acid-test-cluster",
},
}
json := `{"state": "running", "postmaster_start_time": "2021-02-19 14:31:50.053 CET", "role": "master", "server_version": 130004, "cluster_unlocked": false, "xlog": {"location": 123456789}, "timeline": 1, "database_system_identifier": "6462555844314089962", "pending_restart": true, "patroni": {"version": "2.1.1", "scope": "acid-test-cluster"}}`
r := ioutil.NopCloser(bytes.NewReader([]byte(json)))
response := http.Response{
Status: "200",
Body: r,
StatusCode: 200,
Body: r,
}
mockClient := mocks.NewMockHTTPClient(ctrl)
mockClient.EXPECT().Get(gomock.Any()).Return(&response, nil)
p := New(nil, mockClient)
p := New(logger, mockClient)
pod := v1.Pod{
Status: v1.PodStatus{
PodIP: "192.168.100.1",
},
memberData, err := p.GetMemberData(newMockPod("192.168.100.1"))
if !reflect.DeepEqual(expectedMemberData, memberData) {
t.Errorf("Patroni member data differs: expected: %#v, got: %#v", expectedMemberData, memberData)
}
_, err := p.GetMemberData(&pod)
if err != nil {
t.Errorf("Could not read Patroni data: %v", err)
}
}
func TestGetConfig(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
expectedPatroniConfig := acidv1.Patroni{
TTL: 30,
LoopWait: 10,
RetryTimeout: 10,
MaximumLagOnFailover: 33554432,
Slots: map[string]map[string]string{
"cdc": {
"database": "foo",
"plugin": "wal2json",
"type": "logical",
},
},
}
expectedPgParameters := map[string]string{
"archive_mode": "on",
"archive_timeout": "1800s",
"autovacuum_analyze_scale_factor": "0.02",
"autovacuum_max_workers": "5",
"autovacuum_vacuum_scale_factor": "0.05",
"checkpoint_completion_target": "0.9",
"hot_standby": "on",
"log_autovacuum_min_duration": "0",
"log_checkpoints": "on",
"log_connections": "on",
"log_disconnections": "on",
"log_line_prefix": "%t [%p]: [%l-1] %c %x %d %u %a %h ",
"log_lock_waits": "on",
"log_min_duration_statement": "500",
"log_statement": "ddl",
"log_temp_files": "0",
"max_connections": "100",
"max_replication_slots": "10",
"max_wal_senders": "10",
"tcp_keepalives_idle": "900",
"tcp_keepalives_interval": "100",
"track_functions": "all",
"wal_level": "hot_standby",
"wal_log_hints": "on",
}
configJson := `{"loop_wait": 10, "maximum_lag_on_failover": 33554432, "postgresql": {"parameters": {"archive_mode": "on", "archive_timeout": "1800s", "autovacuum_analyze_scale_factor": 0.02, "autovacuum_max_workers": 5, "autovacuum_vacuum_scale_factor": 0.05, "checkpoint_completion_target": 0.9, "hot_standby": "on", "log_autovacuum_min_duration": 0, "log_checkpoints": "on", "log_connections": "on", "log_disconnections": "on", "log_line_prefix": "%t [%p]: [%l-1] %c %x %d %u %a %h ", "log_lock_waits": "on", "log_min_duration_statement": 500, "log_statement": "ddl", "log_temp_files": 0, "max_connections": 100, "max_replication_slots": 10, "max_wal_senders": 10, "tcp_keepalives_idle": 900, "tcp_keepalives_interval": 100, "track_functions": "all", "wal_level": "hot_standby", "wal_log_hints": "on"}, "use_pg_rewind": true, "use_slots": true}, "retry_timeout": 10, "slots": {"cdc": {"database": "foo", "plugin": "wal2json", "type": "logical"}}, "ttl": 30}`
r := ioutil.NopCloser(bytes.NewReader([]byte(configJson)))
response := http.Response{
StatusCode: 200,
Body: r,
}
mockClient := mocks.NewMockHTTPClient(ctrl)
mockClient.EXPECT().Get(gomock.Any()).Return(&response, nil)
p := New(logger, mockClient)
patroniConfig, pgParameters, err := p.GetConfig(newMockPod("192.168.100.1"))
if err != nil {
t.Errorf("Could not read Patroni config endpoint: %v", err)
}
if !reflect.DeepEqual(expectedPatroniConfig, patroniConfig) {
t.Errorf("Patroni config differs: expected: %#v, got: %#v", expectedPatroniConfig, patroniConfig)
}
if !reflect.DeepEqual(expectedPgParameters, pgParameters) {
t.Errorf("Postgre parameters differ: expected: %#v, got: %#v", expectedPgParameters, pgParameters)
}
}
func TestSetPostgresParameters(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
parametersToSet := map[string]string{
"max_connections": "50",
"wal_level": "logical",
}
configJson := `{"loop_wait": 10, "maximum_lag_on_failover": 33554432, "postgresql": {"parameters": {"archive_mode": "on", "archive_timeout": "1800s", "autovacuum_analyze_scale_factor": 0.02, "autovacuum_max_workers": 5, "autovacuum_vacuum_scale_factor": 0.05, "checkpoint_completion_target": 0.9, "hot_standby": "on", "log_autovacuum_min_duration": 0, "log_checkpoints": "on", "log_connections": "on", "log_disconnections": "on", "log_line_prefix": "%t [%p]: [%l-1] %c %x %d %u %a %h ", "log_lock_waits": "on", "log_min_duration_statement": 500, "log_statement": "ddl", "log_temp_files": 0, "max_connections": 50, "max_replication_slots": 10, "max_wal_senders": 10, "tcp_keepalives_idle": 900, "tcp_keepalives_interval": 100, "track_functions": "all", "wal_level": "logical", "wal_log_hints": "on"}, "use_pg_rewind": true, "use_slots": true}, "retry_timeout": 10, "slots": {"cdc": {"database": "foo", "plugin": "wal2json", "type": "logical"}}, "ttl": 30}`
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)
p := New(logger, mockClient)
err := p.SetPostgresParameters(newMockPod("192.168.100.1"), parametersToSet)
if err != nil {
t.Errorf("could not call patch Patroni config: %v", err)
}
}