managed-container/ocf/loxthelion/managed-container.d/applications/postgresql/replication.sh

# Sourced by managed-container; uses the shared OCF and parameter environment.

pg_remote_query()
{
    local sql="$1"

    pg_replication_exec \
        psql \
        "host=${REJOIN_SOURCE_HOST} port=${REJOIN_SOURCE_PORT} dbname=postgres user=${REPLICATION_USER}" \
        -Atqc "${sql}"
}


# Return exactly one reachable remote writable PostgreSQL primary from the
# validated native peer map as: node|host|port.
# Return 1 when no remote primary is currently reachable, and 2 when more
# than one writable peer is observed (unsafe / split-brain indication).
mc_pg_find_remote_primary()
{
    local self peers line node host port role
    local found_count=0 found_node='' found_host='' found_port=''

    peers="${OCF_RESKEY_replication_peers:-}"
    self="${OCF_RESKEY_CRM_meta_on_node:-$(hostname)}"

    [ -n "${peers}" ] || return 1

    while IFS='|' read -r node host port; do
        [ -n "${node}" ] || continue

        role="$(
            pg_replication_exec \
                psql \
                "host=${host} port=${port} user=${REPLICATION_USER} dbname=postgres connect_timeout=2" \
                -X -w -Atqc 'SELECT pg_is_in_recovery();' \
                2>/dev/null
        )" || continue

        if [ "${role}" = "f" ]; then
            found_count=$((found_count + 1))
            found_node="${node}"
            found_host="${host}"
            found_port="${port}"
        fi
    done <<EOF_PEERS
$(LC_ALL=C awk -v peers="${peers}" -v self="${self}" '
BEGIN {
    n=split(peers, entries, ","); self=tolower(self);
    for(i=1;i<=n;i++) {
        if(split(entries[i], pair, "=")!=2) continue;
        node=pair[1];
        if(tolower(node)==self) continue;
        if(split(pair[2], addr, ":")!=2) continue;
        print node "|" addr[1] "|" addr[2];
    }
}')
EOF_PEERS

    if [ "${found_count}" -gt 1 ]; then
        ocf_log err "Multiple writable PostgreSQL peers detected while selecting a rejoin source"
        return 2
    fi

    [ "${found_count}" -eq 1 ] || return 1
    printf '%s|%s|%s\n' "${found_node}" "${found_host}" "${found_port}"
}

ensure_local_ha_slots()
{
    local slots raw_slot slot slot_type

    #
    # POSTGRES_HA_SLOTS lives in the container environment supplied by
    # postgres.env.  It is the cluster-wide list of physical slots that
    # every writable primary must provide.
    #
    slots="$(
        pg_command '
            printf "%s" "${POSTGRES_HA_SLOTS:-}"
        ' 2>/dev/null
    )" || {
        ocf_exit_reason \
            "Unable to read POSTGRES_HA_SLOTS from PostgreSQL container"
        return "${OCF_ERR_GENERIC}"
    }

    if [ -z "${slots}" ]; then
        ocf_log warn \
            "POSTGRES_HA_SLOTS is empty; no local HA slots will be reconciled"
        return "${OCF_SUCCESS}"
    fi

    while IFS= read -r raw_slot; do
        slot="$(printf '%s' "${raw_slot}" | tr -d '[:space:]')"

        [ -n "${slot}" ] || continue

        case "${slot}" in
            *[!a-z0-9_]*)
                ocf_exit_reason \
                    "Invalid physical replication slot name: ${slot}"
                return "${OCF_ERR_CONFIGURED}"
                ;;
        esac

        slot_type="$(
            pg_command "
                psql \
                    -U '${DB_USER}' \
                    -d postgres \
                    -At \
                    -v ON_ERROR_STOP=1 \
                    -c \"
                        SELECT slot_type
                        FROM pg_replication_slots
                        WHERE slot_name = '${slot}';
                    \"
            " 2>/dev/null
        )" || {
            ocf_exit_reason \
                "Failed querying replication slot ${slot}"
            return "${OCF_ERR_GENERIC}"
        }

        slot_type="$(printf '%s' "${slot_type}" | tr -d '[:space:]')"

        case "${slot_type}" in
            physical)
                ocf_log info \
                    "Physical HA slot already exists: ${slot}"
                ;;

            "")
                ocf_log info \
                    "Creating physical HA slot: ${slot}"

                if ! pg_command "
                    psql \
                        -U '${DB_USER}' \
                        -d postgres \
                        -v ON_ERROR_STOP=1 \
                        -c \"
                            SELECT pg_create_physical_replication_slot('${slot}');
                        \"
                " >/dev/null
                then
                    ocf_exit_reason \
                        "Failed creating physical replication slot ${slot}"
                    return "${OCF_ERR_GENERIC}"
                fi
                ;;

            *)
                ocf_exit_reason \
                    "Replication slot ${slot} exists but is type ${slot_type}"
                return "${OCF_ERR_CONFIGURED}"
                ;;
        esac

    done <<EOF
$(printf '%s\n' "${slots}" | tr ',' '\n')
EOF

    return "${OCF_SUCCESS}"
}

ensure_remote_slot()
{
    local host="${1:-}"
    local slot="${2:-}"
    local result

    [ -n "${host}" ] || return 1
    [ -n "${slot}" ] || return 1

    result="$(
        pg_replication_exec \
        psql \
        "host=${host} port=${REJOIN_SOURCE_PORT} user=${REPLICATION_USER} dbname=postgres" \
        -Atc "
            SELECT slot_type
            FROM pg_replication_slots
            WHERE slot_name = '${slot}';
        " 2>/dev/null
    )" || return 1

    case "${result}" in
        physical)
            return 0
            ;;

        "")
            if ! pg_replication_exec \
                psql \
                "host=${host} port=${REJOIN_SOURCE_PORT} user=${REPLICATION_USER} dbname=postgres" \
                -Atc "
                    SELECT slot_name
                    FROM pg_create_physical_replication_slot('${slot}');
                " >/dev/null
            then
                return 1
            fi

            return 0
            ;;

        *)
            ocf_log err \
                "Replication slot ${slot} exists on ${host} but is not physical"
            return 1
            ;;
    esac
}

candidate_caught_up()
{
    local candidate="${1:-}"
    local result

    [ -n "${candidate}" ] || return 1

    result="$(pg_command "
        psql -U '${DB_USER}' -d '${DB_NAME}' -Atc \"
            SELECT CASE
                WHEN state = 'streaming'
                 AND replay_lsn >= pg_current_wal_flush_lsn()
                THEN 'yes'
                ELSE 'no'
            END
            FROM pg_stat_replication
            WHERE application_name = '${candidate}'
            LIMIT 1;
        \"
    " 2>/dev/null)" || return 1

    [ "${result}" = "yes" ]
}

candidate_is_standby()
{
    local host="${1:-}"
    local result
    local port="${REJOIN_SOURCE_PORT:-5432}"

    [ -n "${host}" ] || return 1

    result="$(
        pg_replication_exec \
        psql \
        "host=${host} port=${port} user=${REPLICATION_USER} dbname=postgres" \
        -Atc "SELECT pg_is_in_recovery();" \
        2>/dev/null
    )" || return 1

    [ "${result}" = "t" ]
}