From ed6de2cbfbcdc2c92d72d98f85be13a48f6bfc3c Mon Sep 17 00:00:00 2001 From: Polina Bungina Date: Mon, 2 Nov 2026 11:11:24 +0100 Subject: [PATCH 1/5] Check and log logical replication lag before upgrade for pg17+ --- pkg/cluster/majorversionupgrade.go | 48 ++++++++++++++++++-- pkg/cluster/majorversionupgrade_test.go | 60 +++++++++++++++++++++++++ 2 files changed, 105 insertions(+), 3 deletions(-) create mode 100644 pkg/cluster/majorversionupgrade_test.go diff --git a/pkg/cluster/majorversionupgrade.go b/pkg/cluster/majorversionupgrade.go index dae5db2e8..5ca108196 100644 --- a/pkg/cluster/majorversionupgrade.go +++ b/pkg/cluster/majorversionupgrade.go @@ -16,10 +16,33 @@ import ( ) const ( - majorVersionUpgradeSuccessAnnotation = "last-major-upgrade-success" - majorVersionUpgradeFailureAnnotation = "last-major-upgrade-failure" + majorVersionUpgradeSuccessAnnotation = "last-major-upgrade-success" + majorVersionUpgradeFailureAnnotation = "last-major-upgrade-failure" + logicalReplicationSlotLagThreshold = 2 * 16 * 1024 * 1024 + logicalReplicationSlotLagCheckMinVersion = 17 * 10000 ) +func parseLaggingLogicalReplicationSlots(output string) ([]string, error) { + laggingSlots := []string{} + for _, line := range strings.Split(strings.TrimSpace(output), "\n") { + if line == "" { + continue + } + fields := strings.Split(line, "|") + if len(fields) != 2 { + return nil, fmt.Errorf("unexpected logical replication slot lag output: %q", line) + } + lagBytes, err := strconv.ParseInt(fields[1], 10, 64) + if err != nil { + return nil, fmt.Errorf("invalid logical replication slot lag for slot %q: %w", fields[0], err) + } + if lagBytes > logicalReplicationSlotLagThreshold { + laggingSlots = append(laggingSlots, line) + } + } + return laggingSlots, nil +} + // IsBiggerPostgresVersion Compare two Postgres version numbers func IsBiggerPostgresVersion(old string, new string) bool { oldN, _ := strconv.Atoi(old) @@ -230,6 +253,26 @@ func (c *Cluster) majorVersionUpgrade() error { isUpgradeSuccess := true numberOfPods := len(pods) if allRunning { + podName := &spec.NamespacedName{Namespace: masterPod.Namespace, Name: masterPod.Name} + if c.currentMajorVersion >= logicalReplicationSlotLagCheckMinVersion { + const logicalSlotLagQuery = `SELECT slot_name, pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)::bigint + FROM pg_replication_slots WHERE slot_type = 'logical'` + slotLagOutput, err := c.ExecCommand(podName, "psql", "-X", "-A", "-t", "-v", "ON_ERROR_STOP=1", "-U", c.OpConfig.SuperUsername, "-d", "postgres", "-c", logicalSlotLagQuery) + if err != nil { + c.logger.Errorf("skipping major version upgrade, could not check logical replication slot lag: %v", err) + return fmt.Errorf("could not check logical replication slot lag: %w", err) + } + laggingSlots, err := parseLaggingLogicalReplicationSlots(slotLagOutput) + if err != nil { + c.logger.Errorf("skipping major version upgrade, could not parse logical replication slot lag: %v", err) + return fmt.Errorf("could not parse logical replication slot lag: %w", err) + } + if len(laggingSlots) > 0 { + c.logger.Errorf("skipping major version upgrade, logical replication slot lag exceeds %d bytes (slot_name|lag_bytes): %s", logicalReplicationSlotLagThreshold, strings.Join(laggingSlots, ", ")) + return nil + } + } + c.logger.Infof("healthy cluster ready to upgrade, current: %d desired: %d", c.currentMajorVersion, desiredVersion) if c.currentMajorVersion < desiredVersion { defer func() error { @@ -243,7 +286,6 @@ func (c *Cluster) majorVersionUpgrade() error { return fmt.Errorf("failed to assign critical-operation label: %s", err) } - 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) diff --git a/pkg/cluster/majorversionupgrade_test.go b/pkg/cluster/majorversionupgrade_test.go new file mode 100644 index 000000000..f6ccb4d9c --- /dev/null +++ b/pkg/cluster/majorversionupgrade_test.go @@ -0,0 +1,60 @@ +package cluster + +import ( + "reflect" + "testing" +) + +func TestParseLaggingLogicalReplicationSlots(t *testing.T) { + tests := []struct { + name string + output string + wantSlots []string + wantErr bool + }{ + { + name: "no slots", + wantSlots: []string{}, + }, + { + name: "slots at threshold are allowed", + output: "at_threshold|33554432\n", + wantSlots: []string{}, + }, + { + name: "lagging slot is returned", + output: "lagging_slot|33554433\n", + wantSlots: []string{"lagging_slot|33554433"}, + }, + { + name: "only lagging slots are returned", + output: "healthy_slot|100\nlagging_slot|33554433\n", + wantSlots: []string{"lagging_slot|33554433"}, + }, + { + name: "invalid output is rejected", + output: "unexpected output", + wantErr: true, + }, + { + name: "invalid lag is rejected", + output: "slot|unknown", + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotSlots, err := parseLaggingLogicalReplicationSlots(tt.output) + if (err != nil) != tt.wantErr { + t.Fatalf("parseLaggingLogicalReplicationSlots() error = %v, wantErr %v", err, tt.wantErr) + } + if err != nil { + return + } + if !reflect.DeepEqual(gotSlots, tt.wantSlots) { + t.Errorf("parseLaggingLogicalReplicationSlots() = %v, want %v", gotSlots, tt.wantSlots) + } + }) + } +} From 97091659e321f8242b3a7503bc0f696116476360 Mon Sep 17 00:00:00 2001 From: Polina Bungina Date: Tue, 3 Nov 2026 15:20:55 +0100 Subject: [PATCH 2/5] Address review comments --- pkg/cluster/database.go | 21 +++++++++ pkg/cluster/majorversionupgrade.go | 35 ++------------- pkg/cluster/majorversionupgrade_test.go | 60 ------------------------- 3 files changed, 24 insertions(+), 92 deletions(-) delete mode 100644 pkg/cluster/majorversionupgrade_test.go diff --git a/pkg/cluster/database.go b/pkg/cluster/database.go index d46dc6067..b55596d92 100644 --- a/pkg/cluster/database.go +++ b/pkg/cluster/database.go @@ -109,6 +109,27 @@ func (c *Cluster) pgConnectionString(dbname string) string { constants.PostgresConnectTimeout/time.Second) } +func logicalReplicationSlotLagQuery(threshold int64) string { + return fmt.Sprintf(`WITH slot_lag AS ( + SELECT slot_name, pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) AS lag_bytes + FROM pg_replication_slots + WHERE slot_type = 'logical' + ) + SELECT slot_name, COALESCE(lag_bytes::text, 'unknown') + FROM slot_lag + WHERE lag_bytes IS NULL OR lag_bytes > %d`, threshold) +} + +func (c *Cluster) getLaggingLogicalReplicationSlots(podName *spec.NamespacedName, threshold int64) (string, error) { + query := logicalReplicationSlotLagQuery(threshold) + + output, err := c.ExecCommand(podName, "psql", "-X", "-A", "-t", "-v", "ON_ERROR_STOP=1", "-U", c.OpConfig.SuperUsername, "-d", "postgres", "-c", query) + if err != nil { + return "", fmt.Errorf("could not execute logical replication slot lag query: %w", err) + } + return strings.TrimSpace(output), nil +} + func (c *Cluster) databaseAccessDisabled() bool { if !c.OpConfig.EnableDBAccess { c.logger.Debug("database access is disabled") diff --git a/pkg/cluster/majorversionupgrade.go b/pkg/cluster/majorversionupgrade.go index 5ca108196..1d8b4ba8f 100644 --- a/pkg/cluster/majorversionupgrade.go +++ b/pkg/cluster/majorversionupgrade.go @@ -22,27 +22,6 @@ const ( logicalReplicationSlotLagCheckMinVersion = 17 * 10000 ) -func parseLaggingLogicalReplicationSlots(output string) ([]string, error) { - laggingSlots := []string{} - for _, line := range strings.Split(strings.TrimSpace(output), "\n") { - if line == "" { - continue - } - fields := strings.Split(line, "|") - if len(fields) != 2 { - return nil, fmt.Errorf("unexpected logical replication slot lag output: %q", line) - } - lagBytes, err := strconv.ParseInt(fields[1], 10, 64) - if err != nil { - return nil, fmt.Errorf("invalid logical replication slot lag for slot %q: %w", fields[0], err) - } - if lagBytes > logicalReplicationSlotLagThreshold { - laggingSlots = append(laggingSlots, line) - } - } - return laggingSlots, nil -} - // IsBiggerPostgresVersion Compare two Postgres version numbers func IsBiggerPostgresVersion(old string, new string) bool { oldN, _ := strconv.Atoi(old) @@ -255,20 +234,12 @@ func (c *Cluster) majorVersionUpgrade() error { if allRunning { podName := &spec.NamespacedName{Namespace: masterPod.Namespace, Name: masterPod.Name} if c.currentMajorVersion >= logicalReplicationSlotLagCheckMinVersion { - const logicalSlotLagQuery = `SELECT slot_name, pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)::bigint - FROM pg_replication_slots WHERE slot_type = 'logical'` - slotLagOutput, err := c.ExecCommand(podName, "psql", "-X", "-A", "-t", "-v", "ON_ERROR_STOP=1", "-U", c.OpConfig.SuperUsername, "-d", "postgres", "-c", logicalSlotLagQuery) + laggingSlots, err := c.getLaggingLogicalReplicationSlots(podName, logicalReplicationSlotLagThreshold) if err != nil { - c.logger.Errorf("skipping major version upgrade, could not check logical replication slot lag: %v", err) return fmt.Errorf("could not check logical replication slot lag: %w", err) } - laggingSlots, err := parseLaggingLogicalReplicationSlots(slotLagOutput) - if err != nil { - c.logger.Errorf("skipping major version upgrade, could not parse logical replication slot lag: %v", err) - return fmt.Errorf("could not parse logical replication slot lag: %w", err) - } - if len(laggingSlots) > 0 { - c.logger.Errorf("skipping major version upgrade, logical replication slot lag exceeds %d bytes (slot_name|lag_bytes): %s", logicalReplicationSlotLagThreshold, strings.Join(laggingSlots, ", ")) + if laggingSlots != "" { + c.logger.Errorf("skipping major version upgrade, logical replication slot lag exceeds %d bytes (slot_name|lag_bytes): %s", logicalReplicationSlotLagThreshold, laggingSlots) return nil } } diff --git a/pkg/cluster/majorversionupgrade_test.go b/pkg/cluster/majorversionupgrade_test.go deleted file mode 100644 index f6ccb4d9c..000000000 --- a/pkg/cluster/majorversionupgrade_test.go +++ /dev/null @@ -1,60 +0,0 @@ -package cluster - -import ( - "reflect" - "testing" -) - -func TestParseLaggingLogicalReplicationSlots(t *testing.T) { - tests := []struct { - name string - output string - wantSlots []string - wantErr bool - }{ - { - name: "no slots", - wantSlots: []string{}, - }, - { - name: "slots at threshold are allowed", - output: "at_threshold|33554432\n", - wantSlots: []string{}, - }, - { - name: "lagging slot is returned", - output: "lagging_slot|33554433\n", - wantSlots: []string{"lagging_slot|33554433"}, - }, - { - name: "only lagging slots are returned", - output: "healthy_slot|100\nlagging_slot|33554433\n", - wantSlots: []string{"lagging_slot|33554433"}, - }, - { - name: "invalid output is rejected", - output: "unexpected output", - wantErr: true, - }, - { - name: "invalid lag is rejected", - output: "slot|unknown", - wantErr: true, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - gotSlots, err := parseLaggingLogicalReplicationSlots(tt.output) - if (err != nil) != tt.wantErr { - t.Fatalf("parseLaggingLogicalReplicationSlots() error = %v, wantErr %v", err, tt.wantErr) - } - if err != nil { - return - } - if !reflect.DeepEqual(gotSlots, tt.wantSlots) { - t.Errorf("parseLaggingLogicalReplicationSlots() = %v, want %v", gotSlots, tt.wantSlots) - } - }) - } -} From 4f3db683eb6a128189ce8a8c8fa503b96e18c537 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 4 Nov 2026 12:09:57 +0000 Subject: [PATCH 3/5] Allow longer wait for major upgrade E2E Co-authored-by: hughcapet <27892524+hughcapet@users.noreply.github.com> --- e2e/tests/test_e2e.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/e2e/tests/test_e2e.py b/e2e/tests/test_e2e.py index 2076d3cdf..83de35f15 100644 --- a/e2e/tests/test_e2e.py +++ b/e2e/tests/test_e2e.py @@ -1286,7 +1286,8 @@ def get_annotations(): } k8s.api.custom_objects_api.patch_namespaced_custom_object( "acid.zalan.do", "v1", "default", "postgresqls", "acid-upgrade-test", pg_patch_version_higher_version) - self.eventuallyEqual(lambda: k8s.get_operator_state(), {"0": "idle"}, "Operator does not get in sync") + self.eventuallyEqual(lambda: k8s.get_operator_state(), {"0": "idle"}, "Operator does not get in sync", + retries=180) k8s.wait_for_pod_failover(master_nodes, 'spilo-role=replica,' + cluster_label) k8s.wait_for_pod_start('spilo-role=master,' + cluster_label) From c03dfa27c74fc929803640dd5e2ece77878d8bfe Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 4 Nov 2026 13:05:17 +0000 Subject: [PATCH 4/5] Allow more time for taint test operator sync Co-authored-by: hughcapet <27892524+hughcapet@users.noreply.github.com> --- e2e/tests/test_e2e.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/e2e/tests/test_e2e.py b/e2e/tests/test_e2e.py index 83de35f15..44c8a965a 100644 --- a/e2e/tests/test_e2e.py +++ b/e2e/tests/test_e2e.py @@ -2436,7 +2436,7 @@ def test_taint_based_eviction(self): try: k8s.update_config(patch_toleration_config, step="allow tainted nodes") self.eventuallyEqual(lambda: k8s.get_operator_state(), {"0": "idle"}, - "Operator does not get in sync") + "Operator does not get in sync", retries=180) self.eventuallyEqual(lambda: k8s.count_running_pods(), 2, "No 2 pods running") self.eventuallyEqual(lambda: len(k8s.get_patroni_running_members("acid-minimal-cluster-0")), 2, "Postgres status did not enter running") From 1bf359dfc9cf14051c23d0712aa82eb3d8fb15e9 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 4 Nov 2026 13:05:29 +0000 Subject: [PATCH 5/5] Keep operator alive during taint E2E test Co-authored-by: hughcapet <27892524+hughcapet@users.noreply.github.com> --- e2e/tests/test_e2e.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/e2e/tests/test_e2e.py b/e2e/tests/test_e2e.py index 44c8a965a..4d0dec3db 100644 --- a/e2e/tests/test_e2e.py +++ b/e2e/tests/test_e2e.py @@ -2413,6 +2413,19 @@ def test_taint_based_eviction(self): self.assertNotEqual(master_nodes, []) self.assertNotEqual(replica_nodes, []) + # Keep the operator running if it shares the master's node. + operator_pod = k8s.get_operator_pod() + operator_tolerations = [ + k8s.api.k8s_client.sanitize_for_serialization(toleration) + for toleration in (operator_pod.spec.tolerations or []) + ] + operator_tolerations.append({ + "key": "postgres", + "operator": "Exists", + "effect": "NoExecute" + }) + k8s.patch_pod({"spec": {"tolerations": operator_tolerations}}, operator_pod.metadata.name) + # taint node with postgres=:NoExecute to force failover body = { "spec": {