12349460155

Committed 16 Dec 2024 09:15AM UTC coverage: 44.803%. Remained the same

Build # 12349460155

Build Type

Pull #2773

github

Committed by

web-flow

Commit Message

Merge e6f88d3e3 into c206eb38a

Pull Request Pull Request #2773: Add support for pg17 and remove pg12

Run Details

0 of 4 new or added lines in 2 files covered. (0.0%)

21 existing lines in 1 file now uncovered.

6737 of 15037 relevant lines covered (44.8%)

26.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

11.54

/pkg/cluster/majorversionupgrade.go

package cluster

import (
        "context"
        "encoding/json"
        "fmt"
        "strings"

        "github.com/Masterminds/semver"
        "github.com/zalando/postgres-operator/pkg/spec"
        "github.com/zalando/postgres-operator/pkg/util"
        v1 "k8s.io/api/core/v1"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/types"
)

// VersionMap Map of version numbers
var VersionMap = map[string]int{
        "13": 130000,
        "14": 140000,
        "15": 150000,
        "16": 160000,
        "17": 170000,
}

const (
        majorVersionUpgradeSuccessAnnotation = "last-major-upgrade-success"
        majorVersionUpgradeFailureAnnotation = "last-major-upgrade-failure"
)

// 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" {
                // e.g. current is 13, minimal is 13 allowing 13 to 17 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) isUpgradeAllowedForTeam(owningTeam string) bool {
        allowedTeams := c.OpConfig.MajorVersionUpgradeTeamAllowList

        if len(allowedTeams) == 0 {
                return false
        }

        return util.SliceContains(allowedTeams, owningTeam)
}

func (c *Cluster) annotatePostgresResource(isSuccess bool) error {
        annotations := make(map[string]string)
        currentTime := metav1.Now().Format("2006-01-02T15:04:05Z")
        if isSuccess {
                annotations[majorVersionUpgradeSuccessAnnotation] = currentTime
        } else {
                annotations[majorVersionUpgradeFailureAnnotation] = currentTime
        }
        patchData, err := metaAnnotationsPatch(annotations)
        if err != nil {
                c.logger.Errorf("could not form patch for %s postgresql resource: %v", c.Name, err)
                return err
        }
        _, err = c.KubeClient.Postgresqls(c.Namespace).Patch(context.Background(), c.Name, types.MergePatchType, patchData, metav1.PatchOptions{})
        if err != nil {
                c.logger.Errorf("failed to patch annotations to postgresql resource: %v", err)
                return err
        }
        return nil
}

func (c *Cluster) removeFailuresAnnotation() error {
        annotationToRemove := []map[string]string{
                {
                        "op":   "remove",
                        "path": fmt.Sprintf("/metadata/annotations/%s", majorVersionUpgradeFailureAnnotation),
                },
        }
        removePatch, err := json.Marshal(annotationToRemove)
        if err != nil {
                c.logger.Errorf("could not form removal patch for %s postgresql resource: %v", c.Name, err)
                return err
        }
        _, err = c.KubeClient.Postgresqls(c.Namespace).Patch(context.Background(), c.Name, types.JSONPatchType, removePatch, metav1.PatchOptions{})
        if err != nil {
                c.logger.Errorf("failed to remove annotations from postgresql resource: %v", err)
                return err
        }
        return nil
}

/*
Execute upgrade when mode is set to manual or full or when the owning team is allowed for upgrade (and mode is "off").

Manual upgrade means, it is triggered by the user via manifest version change
Full upgrade means, operator also determines the minimal version used accross all clusters and upgrades violators.
*/
func (c *Cluster) majorVersionUpgrade() error {

        if c.OpConfig.MajorVersionUpgradeMode == "off" && !c.isUpgradeAllowedForTeam(c.Spec.TeamID) {
                return nil
        }

        desiredVersion := c.GetDesiredMajorVersionAsInt()

        if c.currentMajorVersion >= desiredVersion {
                if _, exists := c.ObjectMeta.Annotations[majorVersionUpgradeFailureAnnotation]; exists { // if failure annotation exists, remove it
                        c.removeFailuresAnnotation()
                        c.logger.Infof("removing failure annotation as the cluster is already up to date")
                }
                c.logger.Infof("cluster version up to date. current: %d, min desired: %d", c.currentMajorVersion, desiredVersion)
                return nil
        }

        if !isInMainternanceWindow(c.Spec.MaintenanceWindows) {
                c.logger.Infof("skipping major version upgrade, not in maintenance window")
                return nil
        }

        pods, err := c.listPods()
        if err != nil {
                return err
        }

        allRunning := true

        var masterPod *v1.Pod

        for i, pod := range pods {
                ps, _ := c.patroni.GetMemberData(&pod)

                if ps.Role == "standby_leader" {
                        c.logger.Errorf("skipping major version upgrade for %s/%s standby cluster. Re-deploy standby cluster with the required Postgres version specified", c.Namespace, c.Name)
                        return nil
                }

                if ps.State != "running" {
                        allRunning = false
                        c.logger.Infof("identified non running pod, potentially skipping major version upgrade")
                }

                if ps.Role == "master" {
                        masterPod = &pods[i]
                        c.currentMajorVersion = ps.ServerVersion
                }
        }

        if masterPod == nil {
                c.logger.Infof("no master in the cluster, skipping major version upgrade")
                return nil
        }

        // Recheck version with newest data from Patroni
        if c.currentMajorVersion >= desiredVersion {
                if _, exists := c.ObjectMeta.Annotations[majorVersionUpgradeFailureAnnotation]; exists { // if failure annotation exists, remove it
                        c.removeFailuresAnnotation()
                        c.logger.Infof("removing failure annotation as the cluster is already up to date")
                }
                c.logger.Infof("recheck cluster version is already up to date. current: %d, min desired: %d", c.currentMajorVersion, desiredVersion)
                return nil
        }

        if _, exists := c.ObjectMeta.Annotations[majorVersionUpgradeFailureAnnotation]; exists {
                c.logger.Infof("last major upgrade failed, skipping upgrade")
                return nil
        }

        members, err := c.patroni.GetClusterMembers(masterPod)
        if err != nil {
                c.logger.Error("could not get cluster members data from Patroni API, skipping major version upgrade")
                return err
        }
        patroniData, err := c.patroni.GetMemberData(masterPod)
        if err != nil {
                c.logger.Error("could not get members data from Patroni API, skipping major version upgrade")
                return err
        }
        patroniVer, err := semver.NewVersion(patroniData.Patroni.Version)
        if err != nil {
                c.logger.Error("error parsing Patroni version")
                patroniVer, _ = semver.NewVersion("3.0.4")
        }
        verConstraint, _ := semver.NewConstraint(">= 3.0.4")
        checkStreaming, _ := verConstraint.Validate(patroniVer)

        for _, member := range members {
                if PostgresRole(member.Role) == Leader {
                        continue
                }
                if checkStreaming && member.State != "streaming" {
                        c.logger.Infof("skipping major version upgrade, replica %s is not streaming from primary", member.Name)
                        return nil
                }
                if member.Lag > 16*1024*1024 {
                        c.logger.Infof("skipping major version upgrade, replication lag on member %s is too high", member.Name)
                        return nil
                }
        }

        isUpgradeSuccess := true
        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)
                        c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeNormal, "Major Version Upgrade", "starting major version upgrade on pod %s of %d pods", masterPod.Name, numberOfPods)
                        upgradeCommand := fmt.Sprintf("set -o pipefail && /usr/bin/python3 /scripts/inplace_upgrade.py %d 2>&1 | tee last_upgrade.log", numberOfPods)

                        c.logger.Debug("checking if the spilo image runs with root or non-root (check for user id=0)")
                        resultIdCheck, errIdCheck := c.ExecCommand(podName, "/bin/bash", "-c", "/usr/bin/id -u")
                        if errIdCheck != nil {
                                c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Major Version Upgrade", "checking user id to run upgrade from %d to %d FAILED: %v", c.currentMajorVersion, desiredVersion, errIdCheck)
                        }

                        resultIdCheck = strings.TrimSuffix(resultIdCheck, "\n")
                        var result, scriptErrMsg string
                        if resultIdCheck != "0" {
                                c.logger.Infof("user id was identified as: %s, hence default user is non-root already", resultIdCheck)
                                result, err = c.ExecCommand(podName, "/bin/bash", "-c", upgradeCommand)
                                scriptErrMsg, _ = c.ExecCommand(podName, "/bin/bash", "-c", "tail -n 1 last_upgrade.log")
                        } else {
                                c.logger.Infof("user id was identified as: %s, using su to reach the postgres user", resultIdCheck)
                                result, err = c.ExecCommand(podName, "/bin/su", "postgres", "-c", upgradeCommand)
                                scriptErrMsg, _ = c.ExecCommand(podName, "/bin/bash", "-c", "tail -n 1 last_upgrade.log")
                        }
                        if err != nil {
                                isUpgradeSuccess = false
                                c.annotatePostgresResource(isUpgradeSuccess)
                                c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Major Version Upgrade", "upgrade from %d to %d FAILED: %v", c.currentMajorVersion, desiredVersion, scriptErrMsg)
                                return fmt.Errorf(scriptErrMsg)
                        }

                        c.annotatePostgresResource(isUpgradeSuccess)
                        c.logger.Infof("upgrade action triggered and command completed: %s", result[:100])
                        c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeNormal, "Major Version Upgrade", "upgrade from %d to %d finished", c.currentMajorVersion, desiredVersion)
                }
        }

        return nil
}

1	package cluster
2
3	import (
4	"context"
5	"encoding/json"
6	"fmt"
7	"strings"
8
9	"github.com/Masterminds/semver"
10	"github.com/zalando/postgres-operator/pkg/spec"
11	"github.com/zalando/postgres-operator/pkg/util"
12	v1 "k8s.io/api/core/v1"
13	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
14	"k8s.io/apimachinery/pkg/types"
15	)
16
17	// VersionMap Map of version numbers
18	var VersionMap = map[string]int{
19	"13": 130000,
20	"14": 140000,
21	"15": 150000,
22	"16": 160000,
23	"17": 170000,
24	}
25
26	const (
27	majorVersionUpgradeSuccessAnnotation = "last-major-upgrade-success"
28	majorVersionUpgradeFailureAnnotation = "last-major-upgrade-failure"
29	)
30
31	// IsBiggerPostgresVersion Compare two Postgres version numbers
32	func IsBiggerPostgresVersion(old string, new string) bool {	1✔
33	oldN := VersionMap[old]	1✔
34	newN := VersionMap[new]	1✔
35	return newN > oldN	1✔
36	}	1✔
37
38	// GetDesiredMajorVersionAsInt Convert string to comparable integer of PG version
39	func (c *Cluster) GetDesiredMajorVersionAsInt() int {	6✔
40	return VersionMap[c.GetDesiredMajorVersion()]	6✔
41	}	6✔
42
43	// GetDesiredMajorVersion returns major version to use, incl. potential auto upgrade
44	func (c *Cluster) GetDesiredMajorVersion() string {	7✔
45		7✔
46	if c.Config.OpConfig.MajorVersionUpgradeMode == "full" {	7✔
NEW UNCOV 47	// e.g. current is 13, minimal is 13 allowing 13 to 17 clusters, everything below is upgraded	×
48	if IsBiggerPostgresVersion(c.Spec.PgVersion, c.Config.OpConfig.MinimalMajorVersion) {	×
49	c.logger.Infof("overwriting configured major version %s to %s", c.Spec.PgVersion, c.Config.OpConfig.TargetMajorVersion)	×
50	return c.Config.OpConfig.TargetMajorVersion	×
51	}	×
52	}
53
54	return c.Spec.PgVersion	7✔
55	}
56
UNCOV 57	func (c *Cluster) isUpgradeAllowedForTeam(owningTeam string) bool {	×
58	allowedTeams := c.OpConfig.MajorVersionUpgradeTeamAllowList	×
59		×
60	if len(allowedTeams) == 0 {	×
61	return false	×
62	}	×
63
UNCOV 64	return util.SliceContains(allowedTeams, owningTeam)	×
65	}
66
UNCOV 67	func (c *Cluster) annotatePostgresResource(isSuccess bool) error {	×
68	annotations := make(map[string]string)	×
69	currentTime := metav1.Now().Format("2006-01-02T15:04:05Z")	×
70	if isSuccess {	×
71	annotations[majorVersionUpgradeSuccessAnnotation] = currentTime	×
72	} else {	×
73	annotations[majorVersionUpgradeFailureAnnotation] = currentTime	×
74	}	×
75	patchData, err := metaAnnotationsPatch(annotations)	×
76	if err != nil {	×
77	c.logger.Errorf("could not form patch for %s postgresql resource: %v", c.Name, err)	×
78	return err	×
79	}	×
80	_, err = c.KubeClient.Postgresqls(c.Namespace).Patch(context.Background(), c.Name, types.MergePatchType, patchData, metav1.PatchOptions{})	×
81	if err != nil {	×
82	c.logger.Errorf("failed to patch annotations to postgresql resource: %v", err)	×
83	return err	×
84	}	×
85	return nil	×
86	}
87
UNCOV 88	func (c *Cluster) removeFailuresAnnotation() error {	×
89	annotationToRemove := []map[string]string{	×
90	{	×
91	"op": "remove",	×
92	"path": fmt.Sprintf("/metadata/annotations/%s", majorVersionUpgradeFailureAnnotation),	×
93	},	×
94	}	×
95	removePatch, err := json.Marshal(annotationToRemove)	×
96	if err != nil {	×
97	c.logger.Errorf("could not form removal patch for %s postgresql resource: %v", c.Name, err)	×
98	return err	×
99	}	×
100	_, err = c.KubeClient.Postgresqls(c.Namespace).Patch(context.Background(), c.Name, types.JSONPatchType, removePatch, metav1.PatchOptions{})	×
101	if err != nil {	×
102	c.logger.Errorf("failed to remove annotations from postgresql resource: %v", err)	×
103	return err	×
104	}	×
105	return nil	×
106	}
107
108	/*
109	Execute upgrade when mode is set to manual or full or when the owning team is allowed for upgrade (and mode is "off").
110
111	Manual upgrade means, it is triggered by the user via manifest version change
112	Full upgrade means, operator also determines the minimal version used accross all clusters and upgrades violators.
113	*/
114	func (c *Cluster) majorVersionUpgrade() error {	6✔
115		6✔
116	if c.OpConfig.MajorVersionUpgradeMode == "off" && !c.isUpgradeAllowedForTeam(c.Spec.TeamID) {	6✔
UNCOV 117	return nil	×
118	}	×
119
120	desiredVersion := c.GetDesiredMajorVersionAsInt()	6✔
121		6✔
122	if c.currentMajorVersion >= desiredVersion {	12✔
123	if _, exists := c.ObjectMeta.Annotations[majorVersionUpgradeFailureAnnotation]; exists { // if failure annotation exists, remove it	6✔
UNCOV 124	c.removeFailuresAnnotation()	×
125	c.logger.Infof("removing failure annotation as the cluster is already up to date")	×
126	}	×
127	c.logger.Infof("cluster version up to date. current: %d, min desired: %d", c.currentMajorVersion, desiredVersion)	6✔
128	return nil	6✔
129	}
130
UNCOV 131	if !isInMainternanceWindow(c.Spec.MaintenanceWindows) {	×
132	c.logger.Infof("skipping major version upgrade, not in maintenance window")	×
133	return nil	×
134	}	×
135
UNCOV 136	pods, err := c.listPods()	×
137	if err != nil {	×
138	return err	×
139	}	×
140
UNCOV 141	allRunning := true	×
142		×
143	var masterPod *v1.Pod	×
144		×
145	for i, pod := range pods {	×
146	ps, _ := c.patroni.GetMemberData(&pod)	×
147		×
148	if ps.Role == "standby_leader" {	×
149	c.logger.Errorf("skipping major version upgrade for %s/%s standby cluster. Re-deploy standby cluster with the required Postgres version specified", c.Namespace, c.Name)	×
150	return nil	×
151	}	×
152
UNCOV 153	if ps.State != "running" {	×
154	allRunning = false	×
155	c.logger.Infof("identified non running pod, potentially skipping major version upgrade")	×
156	}	×
157
UNCOV 158	if ps.Role == "master" {	×
159	masterPod = &pods[i]	×
160	c.currentMajorVersion = ps.ServerVersion	×
161	}	×
162	}
163
UNCOV 164	if masterPod == nil {	×
165	c.logger.Infof("no master in the cluster, skipping major version upgrade")	×
166	return nil	×
167	}	×
168
169	// Recheck version with newest data from Patroni
UNCOV 170	if c.currentMajorVersion >= desiredVersion {	×
171	if _, exists := c.ObjectMeta.Annotations[majorVersionUpgradeFailureAnnotation]; exists { // if failure annotation exists, remove it	×
172	c.removeFailuresAnnotation()	×
173	c.logger.Infof("removing failure annotation as the cluster is already up to date")	×
174	}	×
175	c.logger.Infof("recheck cluster version is already up to date. current: %d, min desired: %d", c.currentMajorVersion, desiredVersion)	×
176	return nil	×
177	}
178
UNCOV 179	if _, exists := c.ObjectMeta.Annotations[majorVersionUpgradeFailureAnnotation]; exists {	×
180	c.logger.Infof("last major upgrade failed, skipping upgrade")	×
181	return nil	×
182	}	×
183
UNCOV 184	members, err := c.patroni.GetClusterMembers(masterPod)	×
185	if err != nil {	×
186	c.logger.Error("could not get cluster members data from Patroni API, skipping major version upgrade")	×
187	return err	×
188	}	×
189	patroniData, err := c.patroni.GetMemberData(masterPod)	×
190	if err != nil {	×
191	c.logger.Error("could not get members data from Patroni API, skipping major version upgrade")	×
192	return err	×
193	}	×
194	patroniVer, err := semver.NewVersion(patroniData.Patroni.Version)	×
195	if err != nil {	×
196	c.logger.Error("error parsing Patroni version")	×
197	patroniVer, _ = semver.NewVersion("3.0.4")	×
198	}	×
199	verConstraint, _ := semver.NewConstraint(">= 3.0.4")	×
200	checkStreaming, _ := verConstraint.Validate(patroniVer)	×
201		×
202	for _, member := range members {	×
203	if PostgresRole(member.Role) == Leader {	×
204	continue	×
205	}
UNCOV 206	if checkStreaming && member.State != "streaming" {	×
207	c.logger.Infof("skipping major version upgrade, replica %s is not streaming from primary", member.Name)	×
208	return nil	×
209	}	×
210	if member.Lag > 1610241024 {	×
211	c.logger.Infof("skipping major version upgrade, replication lag on member %s is too high", member.Name)	×
212	return nil	×
213	}	×
214	}
215
UNCOV 216	isUpgradeSuccess := true	×
217	numberOfPods := len(pods)	×
218	if allRunning && masterPod != nil {	×
219	c.logger.Infof("healthy cluster ready to upgrade, current: %d desired: %d", c.currentMajorVersion, desiredVersion)	×
220	if c.currentMajorVersion < desiredVersion {	×
221	podName := &spec.NamespacedName{Namespace: masterPod.Namespace, Name: masterPod.Name}	×
222	c.logger.Infof("triggering major version upgrade on pod %s of %d pods", masterPod.Name, numberOfPods)	×
223	c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeNormal, "Major Version Upgrade", "starting major version upgrade on pod %s of %d pods", masterPod.Name, numberOfPods)	×
224	upgradeCommand := fmt.Sprintf("set -o pipefail && /usr/bin/python3 /scripts/inplace_upgrade.py %d 2>&1 \| tee last_upgrade.log", numberOfPods)	×
225		×
226	c.logger.Debug("checking if the spilo image runs with root or non-root (check for user id=0)")	×
227	resultIdCheck, errIdCheck := c.ExecCommand(podName, "/bin/bash", "-c", "/usr/bin/id -u")	×
228	if errIdCheck != nil {	×
229	c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Major Version Upgrade", "checking user id to run upgrade from %d to %d FAILED: %v", c.currentMajorVersion, desiredVersion, errIdCheck)	×
230	}	×
231
UNCOV 232	resultIdCheck = strings.TrimSuffix(resultIdCheck, "\n")	×
233	var result, scriptErrMsg string	×
234	if resultIdCheck != "0" {	×
235	c.logger.Infof("user id was identified as: %s, hence default user is non-root already", resultIdCheck)	×
236	result, err = c.ExecCommand(podName, "/bin/bash", "-c", upgradeCommand)	×
237	scriptErrMsg, _ = c.ExecCommand(podName, "/bin/bash", "-c", "tail -n 1 last_upgrade.log")	×
238	} else {	×
239	c.logger.Infof("user id was identified as: %s, using su to reach the postgres user", resultIdCheck)	×
240	result, err = c.ExecCommand(podName, "/bin/su", "postgres", "-c", upgradeCommand)	×
241	scriptErrMsg, _ = c.ExecCommand(podName, "/bin/bash", "-c", "tail -n 1 last_upgrade.log")	×
242	}	×
243	if err != nil {	×
244	isUpgradeSuccess = false	×
245	c.annotatePostgresResource(isUpgradeSuccess)	×
246	c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Major Version Upgrade", "upgrade from %d to %d FAILED: %v", c.currentMajorVersion, desiredVersion, scriptErrMsg)	×
247	return fmt.Errorf(scriptErrMsg)	×
248	}	×
249
UNCOV 250	c.annotatePostgresResource(isUpgradeSuccess)	×
251	c.logger.Infof("upgrade action triggered and command completed: %s", result[:100])	×
252	c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeNormal, "Major Version Upgrade", "upgrade from %d to %d finished", c.currentMajorVersion, desiredVersion)	×
253	}
254	}
255
UNCOV 256	return nil	×
257	}

zalando / postgres-operator / 12349460155

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous