Check and log logical replication lag before upgrade for pg17+

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)

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)
+			}
+		})
+	}
+}