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

resolve conflict with master

Browse Source
This commit is contained in:
Felix Kunde
2021-02-25 12:13:15 +01:00
parent cfeb60dc6d 217f306650
commit ccaa24e1a2
14 changed files with 336 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/cluster/cluster.go
+31 -24
View File
@@ -83,15 +83,16 @@ type Cluster struct {
deleteOptions metav1.DeleteOptions
podEventsQueue *cache.FIFO
teamsAPIClient teams.Interface
oauthTokenGetter OAuthTokenGetter
KubeClient k8sutil.KubernetesClient //TODO: move clients to the better place?
currentProcess Process
processMu sync.RWMutex // protects the current operation for reporting, no need to hold the master mutex
specMu sync.RWMutex // protects the spec for reporting, no need to hold the master mutex
ConnectionPooler map[PostgresRole]*ConnectionPoolerObjects
EBSVolumes map[string]volumes.VolumeProperties
VolumeResizer volumes.VolumeResizer
teamsAPIClient teams.Interface
oauthTokenGetter OAuthTokenGetter
KubeClient k8sutil.KubernetesClient //TODO: move clients to the better place?
currentProcess Process
processMu sync.RWMutex // protects the current operation for reporting, no need to hold the master mutex
specMu sync.RWMutex // protects the spec for reporting, no need to hold the master mutex
ConnectionPooler map[PostgresRole]*ConnectionPoolerObjects
EBSVolumes map[string]volumes.VolumeProperties
VolumeResizer volumes.VolumeResizer
currentMajorVersion int
}
type compareStatefulsetResult struct {
@@ -128,15 +129,16 @@ func New(cfg Config, kubeClient k8sutil.KubernetesClient, pgSpec acidv1.Postgres
Secrets: make(map[types.UID]*v1.Secret),
Services: make(map[PostgresRole]*v1.Service),
Endpoints: make(map[PostgresRole]*v1.Endpoints)},
userSyncStrategy: users.DefaultUserSyncStrategy{PasswordEncryption: passwordEncryption},
deleteOptions: metav1.DeleteOptions{PropagationPolicy: &deletePropagationPolicy},
podEventsQueue: podEventsQueue,
KubeClient: kubeClient,
userSyncStrategy: users.DefaultUserSyncStrategy{PasswordEncryption: passwordEncryption},
deleteOptions: metav1.DeleteOptions{PropagationPolicy: &deletePropagationPolicy},
podEventsQueue: podEventsQueue,
KubeClient: kubeClient,
currentMajorVersion: 0,
}
cluster.logger = logger.WithField("pkg", "cluster").WithField("cluster-name", cluster.clusterName())
cluster.teamsAPIClient = teams.NewTeamsAPI(cfg.OpConfig.TeamsAPIUrl, logger)
cluster.oauthTokenGetter = newSecretOauthTokenGetter(&kubeClient, cfg.OpConfig.OAuthTokenSecretName)
cluster.patroni = patroni.New(cluster.logger)
cluster.patroni = patroni.New(cluster.logger, nil)
cluster.eventRecorder = eventRecorder
cluster.EBSVolumes = make(map[string]volumes.VolumeProperties)
@@ -359,7 +361,7 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *appsv1.StatefulSet) *compa
}
if !reflect.DeepEqual(c.Statefulset.Annotations, statefulSet.Annotations) {
match = false
reasons = append(reasons, "new statefulset's annotations does not match the current one")
reasons = append(reasons, "new statefulset's annotations do not match the current one")
}
needsRollUpdate, reasons = c.compareContainers("initContainers", c.Statefulset.Spec.Template.Spec.InitContainers, statefulSet.Spec.Template.Spec.InitContainers, needsRollUpdate, reasons)
@@ -614,17 +616,14 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
logNiceDiff(c.logger, oldSpec, newSpec)
if oldSpec.Spec.PostgresqlParam.PgVersion > newSpec.Spec.PostgresqlParam.PgVersion {
c.logger.Warningf("postgresql version change(%q -> %q) has no effect",
oldSpec.Spec.PostgresqlParam.PgVersion, newSpec.Spec.PostgresqlParam.PgVersion)
c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "PostgreSQL", "postgresql version change(%q -> %q) has no effect",
oldSpec.Spec.PostgresqlParam.PgVersion, newSpec.Spec.PostgresqlParam.PgVersion)
// we need that hack to generate statefulset with the old version
newSpec.Spec.PostgresqlParam.PgVersion = oldSpec.Spec.PostgresqlParam.PgVersion
} else if oldSpec.Spec.PostgresqlParam.PgVersion < newSpec.Spec.PostgresqlParam.PgVersion {
c.logger.Infof("postgresql version increased (%q -> %q), major version upgrade can be done manually after StatefulSet Sync",
if IsBiggerPostgresVersion(oldSpec.Spec.PostgresqlParam.PgVersion, c.GetDesiredMajorVersion()) {
c.logger.Infof("postgresql version increased (%s -> %s), depending on config manual upgrade needed",
oldSpec.Spec.PostgresqlParam.PgVersion, newSpec.Spec.PostgresqlParam.PgVersion)
syncStatetfulSet = true
} else {
c.logger.Infof("postgresql major version unchanged or smaller, no changes needed")
// sticking with old version, this will also advance GetDesiredVersion next time.
newSpec.Spec.PostgresqlParam.PgVersion = oldSpec.Spec.PostgresqlParam.PgVersion
}
// Service
@@ -781,6 +780,14 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
updateFailed = true
}
if !updateFailed {
// Major version upgrade must only fire after success of earlier operations and should stay last
if err := c.majorVersionUpgrade(); err != nil {
c.logger.Errorf("major version upgrade failed: %v", err)
updateFailed = true
}
}
return nil
}
pkg/cluster/k8sres.go
+1 -1
View File
@@ -733,7 +733,7 @@ func (c *Cluster) generateSpiloPodEnvVars(uid types.UID, spiloConfiguration stri
},
}
if c.OpConfig.EnablePgVersionEnvVar {
envVars = append(envVars, v1.EnvVar{Name: "PGVERSION", Value: c.Spec.PgVersion})
envVars = append(envVars, v1.EnvVar{Name: "PGVERSION", Value: c.GetDesiredMajorVersion()})
}
// Spilo expects cluster labels as JSON
if clusterLabels, err := json.Marshal(labels.Set(c.OpConfig.ClusterLabels)); err != nil {
pkg/cluster/majorversionupgrade.go
+97
View File
@@ -0,0 +1,97 @@
package cluster
import (
"fmt"
"github.com/zalando/postgres-operator/pkg/spec"
v1 "k8s.io/api/core/v1"
)
// VersionMap Map of version numbers
var VersionMap = map[string]int{
"9.5": 90500,
"9.6": 90600,
"10": 100000,
"11": 110000,
"12": 120000,
"13": 130000,
}
// IsBiggerPostgresVersion Compare two Postgres version numbers
func IsBiggerPostgresVersion(old string, new string) bool {
oldN, _ := VersionMap[old]
newN, _ := VersionMap[new]
return newN > oldN
}
// GetDesiredMajorVersionAsInt Convert string to comparable integer of PG version
func (c *Cluster) GetDesiredMajorVersionAsInt() int {
return VersionMap[c.GetDesiredMajorVersion()]
}
// GetDesiredMajorVersion returns major version to use, incl. potential auto upgrade
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
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
}
}
return c.Spec.PgVersion
}
func (c *Cluster) majorVersionUpgrade() error {
if c.OpConfig.MajorVersionUpgradeMode == "off" {
return nil
}
desiredVersion := c.GetDesiredMajorVersionAsInt()
if c.currentMajorVersion >= desiredVersion {
c.logger.Infof("cluster version up to date. current: %d desired: %d", c.currentMajorVersion, desiredVersion)
return nil
}
pods, err := c.listPods()
if err != nil {
return err
}
allRunning := true
var masterPod *v1.Pod
for _, pod := range pods {
ps, _ := c.patroni.GetMemberData(&pod)
if ps.State != "running" {
allRunning = false
c.logger.Infof("identified non running pod, potentially skipping major version upgrade")
}
if ps.Role == "master" {
masterPod = &pod
c.currentMajorVersion = ps.ServerVersion
}
}
numberOfPods := len(pods)
if allRunning && masterPod != nil {
c.logger.Infof("healthy cluster ready to upgrade, current: %d desired: %d", c.currentMajorVersion, desiredVersion)
if c.currentMajorVersion < desiredVersion {
podName := &spec.NamespacedName{Namespace: masterPod.Namespace, Name: masterPod.Name}
c.logger.Infof("triggering major version upgrade on pod %s of %d pods", masterPod.Name, numberOfPods)
upgradeCommand := fmt.Sprintf("/usr/bin/python3 /scripts/inplace_upgrade.py %d 2>&1 | tee last_upgrade.log", numberOfPods)
_, err := c.ExecCommand(podName, "/bin/su", "postgres", "-c", upgradeCommand)
if err != nil {
return err
}
}
}
return nil
}
pkg/cluster/pod.go
+9 -3
View File
@@ -14,6 +14,7 @@ import (
"github.com/zalando/postgres-operator/pkg/spec"
"github.com/zalando/postgres-operator/pkg/util"
"github.com/zalando/postgres-operator/pkg/util/patroni"
"github.com/zalando/postgres-operator/pkg/util/retryutil"
)
@@ -378,12 +379,17 @@ func (c *Cluster) isSafeToRecreatePods(pods []v1.Pod) bool {
*/
for _, pod := range pods {
c.logger.Debugf("name=%s phase=%s ip=%s", pod.Name, pod.Status.Phase, pod.Status.PodIP)
var state string
}
for _, pod := range pods {
var data patroni.MemberData
err := retryutil.Retry(1*time.Second, 5*time.Second,
func() (bool, error) {
var err error
state, err = c.patroni.GetPatroniMemberState(&pod)
data, err = c.patroni.GetMemberData(&pod)
if err != nil {
return false, err
}
@@ -394,7 +400,7 @@ func (c *Cluster) isSafeToRecreatePods(pods []v1.Pod) bool {
if err != nil {
c.logger.Errorf("failed to get Patroni state for pod: %s", err)
return false
} else if state == "creating replica" {
} else if data.State == "creating replica" {
c.logger.Warningf("cannot re-create replica %s: it is currently being initialized", pod.Name)
return false
}
pkg/cluster/sync.go
+9 -4
View File
@@ -118,6 +118,11 @@ func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error {
return fmt.Errorf("could not sync connection pooler: %v", err)
}
// Major version upgrade must only run after success of all earlier operations, must remain last item in sync
if err := c.majorVersionUpgrade(); err != nil {
c.logger.Errorf("major version upgrade failed: %v", err)
}
return err
}
@@ -502,7 +507,7 @@ func (c *Cluster) syncSecrets() error {
for secretUsername, secretSpec := range secrets {
if secret, err = c.KubeClient.Secrets(secretSpec.Namespace).Create(context.TODO(), secretSpec, metav1.CreateOptions{}); err == nil {
c.Secrets[secret.UID] = secret
c.logger.Debugf("created new secret %q, uid: %q", util.NameFromMeta(secret.ObjectMeta), secret.UID)
c.logger.Debugf("created new secret %s, uid: %s", util.NameFromMeta(secret.ObjectMeta), secret.UID)
continue
}
if k8sutil.ResourceAlreadyExists(err) {
@@ -511,7 +516,7 @@ func (c *Cluster) syncSecrets() error {
return fmt.Errorf("could not get current secret: %v", err)
}
if secretUsername != string(secret.Data["username"]) {
c.logger.Errorf("secret %s does not contain the role %q", secretSpec.Name, secretUsername)
c.logger.Errorf("secret %s does not contain the role %s", secretSpec.Name, secretUsername)
continue
}
c.Secrets[secret.UID] = secret
@@ -530,7 +535,7 @@ func (c *Cluster) syncSecrets() error {
if pwdUser.Password != string(secret.Data["password"]) &&
pwdUser.Origin == spec.RoleOriginInfrastructure {
c.logger.Debugf("updating the secret %q from the infrastructure roles", secretSpec.Name)
c.logger.Debugf("updating the secret %s from the infrastructure roles", secretSpec.Name)
if _, err = c.KubeClient.Secrets(secretSpec.Namespace).Update(context.TODO(), secretSpec, metav1.UpdateOptions{}); err != nil {
return fmt.Errorf("could not update infrastructure role secret for role %q: %v", secretUsername, err)
}
@@ -540,7 +545,7 @@ func (c *Cluster) syncSecrets() error {
userMap[secretUsername] = pwdUser
}
} else {
return fmt.Errorf("could not create secret for user %q: %v", secretUsername, err)
return fmt.Errorf("could not create secret for user %s: %v", secretUsername, err)
}
}