Added logical slot setting, pgbouncer switchover config, added lag monitoring, pgbouncer new reload logic with config exchange

Author: fresh-borzoni

configs/pgbouncer_new.ini

@@ -0,0 +1,26 @@
+[databases]
+testdb_rw = host=postgres_target port=5432 auth_user=testuser dbname=testdb
+testdb_ro = host=postgres_target_ro_1,postgres_target_ro_2 port=5432 auth_user=testuser dbname=testdb
+
+[pgbouncer]
+listen_addr = 0.0.0.0
+listen_port = 6432
+unix_socket_dir = 
+user = postgres
+auth_file = /etc/pgbouncer/userlist.txt
+auth_type = trust
+pool_mode = transaction
+max_client_conn = 1000
+default_pool_size = 100
+ignore_startup_parameters = extra_float_digits
+server_round_robin = 1
+max_prepared_statements = 1000
+
+# Log settings
+admin_users = testuser
+
+# Connection sanity checks, timeouts
+
+# TLS settings
+
+# Dangerous timeouts

configs/switchover-config.yaml

@@ -17,12 +17,16 @@ database:
   name: testdb
   user: testuser
   password: testpass
+  logical_slot: dms_slot
+  logical_publication: dms_pub
+  logical_subscription: dms_sub
 connectors:
   - name: postgres-connector
     slot_name: debezium
     publication_name: dbz_publication
 pgbouncer:
   config_file: pgbouncer.ini
+  switchover_config_file: pgbouncer_new.ini
   admin_port: 6433
   admin_user: testuser # PgBouncer admin user
   admin_password: testpass # PgBouncer admin password

docker-compose.yml

@@ -159,6 +159,7 @@ services:
     image: edoburu/pgbouncer:v1.23.1-p2
     volumes:
       - ./configs/pgbouncer.ini:/etc/pgbouncer/pgbouncer.ini
+      - ./configs/pgbouncer_new.ini:/etc/pgbouncer/pgbouncer_new.ini
       - ./configs/userlist.txt:/etc/pgbouncer/userlist.txt
     command: ["/usr/bin/pgbouncer", "/etc/pgbouncer/pgbouncer.ini"]
     ports:

lib/config.sh

@@ -35,11 +35,17 @@ load_config() {
     DB_NAME=$(yq e '.database.name' "$CONFIG_FILE")
     DB_USER=$(yq e '.database.user' "$CONFIG_FILE")
     DB_PASSWORD=$(yq e '.database.password' "$CONFIG_FILE")
+
+    DB_LOGICAL_SLOT=$(yq e '.database.logical_slot' "$CONFIG_FILE")
+    DB_LOGICAL_SUB=$(yq e '.database.logical_subscription' "$CONFIG_FILE")
+    DB_LOGICAL_PUB=$(yq e '.database.logical_publication' "$CONFIG_FILE")
 
     KAFKA_CONNECT_URL=($(yq e '.kafka.connect_url' "$CONFIG_FILE"))
     CONNECTORS=($(yq e '.connectors[].name' "$CONFIG_FILE"))
 
-    PGBOUNCER_CONFIG="${PROJECT_ROOT}/configs/$(yq e '.pgbouncer.config_file' "$CONFIG_FILE")"
+    PGBOUNCER_CONFIG=$(yq e '.pgbouncer.config_file' "$CONFIG_FILE")
+    PGBOUNCER_SWITCHOVER_CONFIG=$(yq e '.pgbouncer.switchover_config_file' "$CONFIG_FILE")
+
     PGBOUNCER_PORT=$(yq e '.pgbouncer.admin_port' "$CONFIG_FILE")
     PGBOUNCER_ADMIN_USER=$(yq e '.pgbouncer.admin_user' "$CONFIG_FILE")
     PGBOUNCER_ADMIN_PASSWORD=$(yq e '.pgbouncer.admin_password' "$CONFIG_FILE")

lib/pgbouncer.sh

@@ -36,6 +36,44 @@ pause_pool() {
     return 1
 }
 
+resume_pgbouncer() {
+    log_warn "Resuming PgBouncer..."
+
+    paused_dbs=$(PGPASSWORD=${PGBOUNCER_ADMIN_PASSWORD} psql -h localhost \
+        -p ${PGBOUNCER_PORT} \
+        -U ${PGBOUNCER_ADMIN_USER} \
+        -d ${PGBOUNCER_ADMIN_DB} \
+        -c "SHOW DATABASES" | awk -F'|' '
+        NR==1 {
+            for (i=1; i<=NF; i++) { 
+                gsub(" ", "", $i)
+                if ($i == "paused") paused_col = i
+                if ($i == "name") name_col = i
+            }
+        }
+        NR>2 {
+            val = $paused_col
+            gsub(" ", "", val)
+            if (val == "1") {
+                gsub(" ", "", $name_col)
+                print $name_col
+            }
+        }')
+
+    if [ -n "$paused_dbs" ]; then
+        for db in $paused_dbs; do
+            echo "Resuming database: $db"
+            PGPASSWORD=${PGBOUNCER_ADMIN_PASSWORD} psql -h localhost \
+                -p ${PGBOUNCER_PORT} \
+                -U ${PGBOUNCER_ADMIN_USER} \
+                -d ${PGBOUNCER_ADMIN_DB} \
+                -c "RESUME $db"
+        done
+    else
+        echo "No paused databases found"
+    fi
+}
+
 resume_pool() {
     local pool_name=$1
     log_warn "Resuming PgBouncer pool ${pool_name}..."
@@ -53,17 +91,18 @@ update_pgbouncer_pool() {
     log_warn "Updating PgBouncer pool ${pool_name} to point to ${new_host}..."
     pause_pool "$pool_name"
 
-    awk -v pool="$pool_name" -v host="$new_host" '
-        $1 == pool {sub(/host=[^ ]*/, "host=" host)} 1
-    ' "$PGBOUNCER_CONFIG" > "$PGBOUNCER_CONFIG.new" && \
-    mv "$PGBOUNCER_CONFIG.new" "$PGBOUNCER_CONFIG"
-    
+    PGPASSWORD=${PGBOUNCER_ADMIN_PASSWORD} psql -h localhost \
+        -p ${PGBOUNCER_PORT} \
+        -U ${PGBOUNCER_ADMIN_USER} \
+        -d ${PGBOUNCER_ADMIN_DB} \
+        -c "SET  conffile = '/etc/pgbouncer/$PGBOUNCER_SWITCHOVER_CONFIG';"
+
     PGPASSWORD=${PGBOUNCER_ADMIN_PASSWORD} psql -h localhost \
         -p ${PGBOUNCER_PORT} \
         -U ${PGBOUNCER_ADMIN_USER} \
         -d ${PGBOUNCER_ADMIN_DB} \
         -c "RELOAD;"
-    
+
     resume_pool "$pool_name"
 }
 
@@ -89,7 +128,7 @@ check_ro_pool_status() {
 
 get_current_pool_host() {
     local pool_name=$1
-    PGPASSWORD=${PGBOUNCER_ADMIN_PASSWORD} psql -h localhost \
+    PGPASSWORD=${PGBOUNCER_ADMIN_PASSWORD} psql -h "$SOURCE_HOST" \
         -p ${PGBOUNCER_PORT} \
         -U ${PGBOUNCER_ADMIN_USER} \
         -d ${PGBOUNCER_ADMIN_DB} \

lib/replication.sh

@@ -68,26 +68,26 @@ switch_replication_direction() {
 
     # Drop subscription in target (old subscriber)
     PGPASSWORD=$DB_PASSWORD psql -h "$TARGET_HOST" -p "$TARGET_PORT" -U "$DB_USER" -d "$DB_NAME" -c \
-        "DROP SUBSCRIPTION IF EXISTS dms_sub;"
+        "DROP SUBSCRIPTION IF EXISTS $DB_LOGICAL_SUB;"
 
     PGPASSWORD=$DB_PASSWORD psql -h "$TARGET_HOST" -p "$TARGET_PORT" -U "$DB_USER" -d "$DB_NAME" -c \
-        "SELECT pg_create_logical_replication_slot('dms_slot', 'pgoutput');"
+        "SELECT pg_create_logical_replication_slot('$DB_LOGICAL_SLOT', 'pgoutput');"
 
     # Create publication in target (new publisher)
     PGPASSWORD=$DB_PASSWORD psql -h "$TARGET_HOST" -p "$TARGET_PORT" -U "$DB_USER" -d "$DB_NAME" -c "
-        DROP PUBLICATION IF EXISTS dms_pub;
-        CREATE PUBLICATION dms_pub FOR ALL TABLES;"
+        DROP PUBLICATION IF EXISTS $DB_LOGICAL_PUB;
+        CREATE PUBLICATION $DB_LOGICAL_PUB FOR ALL TABLES;"
 
     # Drop old publication in source
     PGPASSWORD=$DB_PASSWORD psql -h "$SOURCE_HOST" -p "$SOURCE_PORT" -U "$DB_USER" -d "$DB_NAME" -c "
-        DROP PUBLICATION IF EXISTS dms_pub;"
+        DROP PUBLICATION IF EXISTS $DB_LOGICAL_PUB;"
 
     # Create subscription in source (new subscriber)
     PGPASSWORD=$DB_PASSWORD psql -h "$SOURCE_HOST" -p "$SOURCE_PORT" -U "$DB_USER" -d "$DB_NAME" -c "
-        CREATE SUBSCRIPTION dms_sub 
+        CREATE SUBSCRIPTION $DB_LOGICAL_SUB 
         CONNECTION 'host=${TARGET_INTERNAL_NAME} port=5432 dbname=${DB_NAME} user=${DB_USER} password=${DB_PASSWORD}'
-        PUBLICATION dms_pub 
-        WITH (copy_data = false, create_slot = false, slot_name = 'dms_slot');"
+        PUBLICATION $DB_LOGICAL_PUB 
+        WITH (copy_data = false, create_slot = false, slot_name = '$DB_LOGICAL_SLOT');"
 }
 
 wait_for_slots_catchup() {
@@ -106,7 +106,7 @@ wait_for_slots_catchup() {
         sleep 5
         return 0
     fi
-    
+
     # Start checks in parallel
     for slot in "${slots_to_migrate[@]}"; do
         check_slot_lag "$slot" &
@@ -156,6 +156,32 @@ check_slot_lag() {
     return 1
 }
 
+wait_for_logical_replica_sync() {
+   log_info "Waiting for logical replica to catchup..."
+
+    local sent_lsn
+    sent_lsn=$(PGPASSWORD=$DB_PASSWORD psql -h "$SOURCE_HOST" -p "$SOURCE_PORT" -U "$DB_USER" -d "$DB_NAME" -Atc "
+        SELECT sent_lsn 
+        FROM pg_stat_replication 
+        WHERE application_name = '$DB_LOGICAL_SUB'")
+
+    # wait 5 secs at max for catchup
+    for ((i=0; i<5; i++)); do
+        local diff_bytes
+        diff_bytes=$(PGPASSWORD=$DB_PASSWORD psql -h "$TARGET_HOST" -p "$TARGET_PORT" -U "$DB_USER" -d "$DB_NAME" -Atc "
+            SELECT pg_wal_lsn_diff('$sent_lsn', received_lsn) 
+            FROM pg_stat_subscription 
+            WHERE subname = '$DB_LOGICAL_SUB'")
+
+        if [ "$diff_bytes" -le 0 ]; then
+            log_info "Logical replica caught up"
+            return 0
+        fi
+        sleep 1
+    done
+    return 1
+}
+
 get_active_slots() {
     PGPASSWORD=$DB_PASSWORD psql -h "$SOURCE_HOST" -p "$SOURCE_PORT" -U "$DB_USER" -d "$DB_NAME" -tAc "
         SELECT slot_name, 
@@ -167,7 +193,25 @@ get_active_slots() {
 discover_replication_mappings() {
     local mappings=()
     local unmapped=()
-    
+
+    # Display logical replica for upgrade settings
+    echo -e "\nLogical Replication Configuration:"
+    printf "%-25s %-30s\n" "Setting" "Value"
+    echo "------------------------------------------------------------------------"
+    printf "%-25s %-30s\n" "Logical Slot" "$DB_LOGICAL_SLOT"
+    printf "%-25s %-30s\n" "Logical Subscription" "$DB_LOGICAL_SUB"
+    printf "%-25s %-30s\n" "Logical Publication" "$DB_LOGICAL_PUB"
+
+    echo -e "\nProceed with these settings? [y/N] "
+    read -r response
+
+    if [[ "$response" =~ ^[Yy]$ ]]; then
+        echo "Proceeding with configuration..."
+    else
+        echo "Operation cancelled."
+        return 1
+    fi
+
     while IFS='|' read -r slot lag; do
         slot=$(echo "$slot" | xargs)
         lag=$(echo "$lag" | xargs)

scripts/switchover.sh

@@ -58,6 +58,7 @@ perform_switchover() {
 
     local rw_pool_name=$(yq e '.pgbouncer.pools.read_write.name' "$CONFIG_FILE")
     pause_pool "$rw_pool_name"
+    wait_for_logical_replica_sync
 
     sync_sequences
 
@@ -84,18 +85,21 @@ swap_roles() {
     local orig_source_port="$SOURCE_PORT"
     local -a orig_source_replicas=("${SOURCE_REPLICAS[@]}")
     local orig_source_internal="$SOURCE_INTERNAL_NAME"
+    local orig_pgb_config="$PGBOUNCER_CONFIG"
 
     # Swap source to target
     SOURCE_HOST="$TARGET_HOST"
     SOURCE_PORT="$TARGET_PORT"
     SOURCE_REPLICAS=("${TARGET_REPLICAS[@]}")
     SOURCE_INTERNAL_NAME="$TARGET_INTERNAL_NAME"
+    PGBOUNCER_CONFIG="$PGBOUNCER_SWITCHOVER_CONFIG"
 
     # Swap target to original source
     TARGET_HOST="$orig_source_host"
     TARGET_PORT="$orig_source_port"
     TARGET_REPLICAS=("${orig_source_replicas[@]}")
     TARGET_INTERNAL_NAME="$orig_source_internal"
+    PGBOUNCER_SWITCHOVER_CONFIG="$orig_pgb_config"
 
     log_info "Roles swapped: new source=${SOURCE_HOST}, new target=${orig_source_host}"
 }