Re: Monitoring logical replication

Hi Saheed!

I monitor our replication this way:

1. Every 10 seconds i fetch the current LSN and write it into a table, 
next with the current timestamp. Further I fetch confirmend LSNs from 
the replication slots and delete old entries in lsn2data table.

calculate_logical_replication_lag.php:

<?php

$path = realpath(dirname(__FILE__) . "/../inc");
set_include_path($path . PATH_SEPARATOR . get_include_path());

require_once('config.php');
$config_int['syslogprefix'] = basename(__FILE__);
require_once('logging.php');

$dbuser="replication_lag_user";
$dbpass="XXXXXXXXXXXXXXXXXXXX";
if (!$dbconn = pg_pconnect('host='.$config_int['dbhost'].' 
dbname='.$config_int['dbname'].' user='.$dbuser.' password='.$dbpass)) {
         print "Sorry, database connection failed";
         exit;
}

$accuracy = 10; // in seconds

//
// Preparations:
//
// CREATE TABLE lsn2data(
//    lsn pg_lsn PRIMARY KEY,
//    seen timestamp NOT NULL DEFAULT NOW()
// );
// CREATE ROLE replication_lag_user WITH LOGIN PASSWORD 
'XXXXXXXXXXXXXXXXXXX';
// GRANT ALL ON TABLE lsn2data TO replication_lag_user;
//
// CREATE OR REPLACE FUNCTION get_replication_lag() RETURNS TABLE 
(subscriber name, lag bigint) AS
// $BODY$
// DECLARE
//     subscriber name;
// BEGIN
//     FOR subscriber IN
//         SELECT slot_name FROM pg_replication_slots
//     LOOP
//         RETURN QUERY SELECT slot_name, EXTRACT(EPOCH FROM 
NOW()-seen)::bigint lag from lsn2data,pg_replication_slots WHERE 
slot_name=subscriber AND lsn < confirmed_flush_lsn ORDER BY lsn DESC 
LIMIT 1;
//     END LOOP;
//     RETURN;
// END
// $BODY$
// LANGUAGE plpgsql;
//
while (1) {
         $dbq = pg_query("INSERT INTO lsn2data (lsn) VALUES 
(pg_current_wal_lsn())");
         if ($dbq === FALSE) {
                 mylog(LOG_ERROR, "SQL query error: 
".pg_last_error()."\n");
                 exit(1);
         }

         $dbq = pg_query("DELETE FROM lsn2data WHERE lsn < (".
                           "SELECT lsn FROM lsn2data WHERE lsn < (".
                             "SELECT confirmed_flush_lsn FROM 
pg_replication_slots ORDER BY confirmed_flush_lsn ASC LIMIT 1".
                           ") ORDER BY lsn DESC LIMIT 1".
                         ")"
         );
         if ($dbq === FALSE) {
                 mylog(LOG_ERROR, "SQL query error: 
".pg_last_error()."\n");
                 exit(1);
         }
         sleep($accuracy);
}

2. I graph the replications lags (converted from LSN to seconds) in my 
check_mk monitoring:

#!/bin/bash

#
# Managed by Puppet: 
modules/base/files/monitoring/check_logical_replication_lag.sh
#
# Check the logical replication lag and export performance data for each 
subscriber
#

# exit on error
#set -e

#Make sure this script only runs one at a time
(

   ME=$0
   MEBASE=`basename $0`

   mylog () {
     echo "$MEBASE: $1"
     logger -t "$MEBASE" "$1"
   }

   flock -x -w 1 200
   if [ $? != "0" ]; then
     #echo "ERROR: $0 is already running ... exit"
     logger -t "$MEBASE" "ERROR: $0 is already running ... exit"
     exit 1
   fi

   # Do stuff

# Variablen fuer Monitoring
CMK_SPOOLDIR=/var/lib/check_mk_agent/spool
CMK_NAME=$MEBASE
CMK_SPOOLFILE=600_`basename ${CMK_NAME}`.txt
CMK_HEADER="<<<local>>>"
TMP_FILE="/tmp/logical_replication_lag.csv"

# Schwellwerte
warn=300
crit=600

final_output="$CMK_HEADER\nP $CMK_NAME "

# move to a directory where user postgresl may reside (sudo)
cd /tmp

# Lag auslesen. Waehrend dem initialen aufsynchen eines Subscribers gibt 
es temporaere Subscriptions, mit dem Namen reg_xxx1-pid-sync-pid.
# Damit diese nicht getrackt werden gibt es die huebsche LIKE Clause.
rm -f "$TMP_FILE"
sudo -u postgres psql regdns -c "COPY (SELECT subscriber,lag FROM 
get_replication_lag() WHERE subscriber LIKE '%\_____' ORDER BY 2 DESC) 
TO '$TMP_FILE' With CSV" 2>&1> /dev/null
LC=$(sudo -u postgres psql -t regdns -c "SELECT count(*) FROM 
get_replication_lag();" | tr -d ' ')

if [ $LC == "0" ]; then
         echo -e "$CMK_HEADER\n0 $CMK_NAME - No Slaves with Replication 
found - maybe we are a slave?" > $CMK_SPOOLDIR/$CMK_SPOOLFILE
         exit 0;
fi

grep $(hostname | cut -d '-' -f2) "$TMP_FILE" > /dev/null
if [ $? != "0" ]; then
         echo -e "$CMK_HEADER\n2 $CMK_NAME - Postgres Output does not 
seem valid. Please check script $ME and output in $TMP_FILE" > 
$CMK_SPOOLDIR/$CMK_SPOOLFILE
         exit 1;
fi

# CSV in Array einlesen
IFS=$'\n' read -d '' -r -a input_file < "$TMP_FILE"

# Auswerten
maxlag=0
for i in "${input_file[@]}"; do
         node=`echo $i | awk -F  "," '{print $1}' | tr -- _ -`
         lag=`echo $i | awk -F  "," '{print $2}'`
         final_output="$final_output$node=$lag;$warn;$crit|"
         # 
https://unix.stackexchange.com/questions/186663/is-there-a-unix-command-that-gives-the-minimum-maximum-of-two-numbers
         maxlag=$(( maxlag > lag ? maxlag : lag ))
done
final_output="${final_output}max-lag=$maxlag;$warn;$crit"

# Letztes Pipe Zeichen rausschneiden
#final_output=`echo $final_output | rev | cut -c 2- | rev`

# Spool File schreiben
echo -e $final_output > $CMK_SPOOLDIR/$CMK_SPOOLFILE
logger -t "$MEBASE" "$final_output"


) 200>/tmp/`basename $0`.exclusivelock


3. During initial sync I check the status on the subscriber. Once it has 
synced all tables of the publication, it will send me an email.
#
# Managed by Puppet: 
modules/pdns/templates/check_pglogical_subscription.sh.erb
#

#
# This script checks and eventually creates the subscription, and wait 
until the initial sync is finished
#

PUB=regdns2020_pub
SLEEP=5
PREFIX=check_pglogical_subscription.sh
NUMTAB=175

SECONDS=0
date
while true; do
         echo "SELECT * from pg_subscription;" | sudo -u postgres psql -t 
regdns | grep -q $PUB
         if [ $? -eq 0 ]; then
                 echo "OK: Host is subscribed to '$PUB'. Checking for 
table count ..."
                 break
         fi
         echo "ERROR: Host is not subscribed to '$PUB'. Subscribing to 
master ..."
         logger -t $PREFIX "ERROR: Host is not subscribed to '$PUB'. 
Subscribing to master ..."
         echo "CREATE SUBSCRIPTION `hostname -s|tr -- - _` CONNECTION 
'host=XXXXX dbname=XXXX user=XXXXX password=XXXXXX PUBLICATION 
regdns2020_pub;" | sudo -u postgres psql regdns && touch 
/etc/regdns.schema_subscription.created
         echo "Re-Checking in $SLEEP seconds ..."
         logger -t $PREFIX "Re-Checking in $SLEEP seconds ..."
         sleep $SLEEP
done

while true; do
         COUNT=$(echo "SELECT count(*) from pg_subscription_rel;" | sudo 
-u postgres psql -t regdns | head -1 | xargs)
         if [ $COUNT -eq $NUMTAB ]; then
                 echo "OK: Subscription '$PUB' contains $NUMTAB tables - 
that is OK. Checking for initial-sync status ..."
                 logger -t $PREFIX "OK: Subscription '$PUB' contains 
$NUMTAB tables - that is OK. Checking for initial-sync status ..."
                 break
         fi
         echo "ERROR: Subscription '$PUB' contains $COUNT tables, but 
should contain $NUMTAB table. Re-Checking in $SLEEP seconds ..."
         logger -t $PREFIX  "ERROR: Subscription '$PUB' contains $COUNT 
tables, but should contain $NUMTAB table. Re-Checking in $SLEEP seconds 
..."
         sleep $SLEEP
done

while true; do
         COUNTFIN=$(echo "SELECT count(*) from pg_subscription_rel WHERE 
srsubstate='r';" | sudo -u postgres psql -t regdns | head -1 | xargs)
         if [ $COUNTFIN -eq $NUMTAB ]; then
                 echo "OK: Initial sync of $COUNTFIN/$NUMTAB tables 
finished in $SECONDS seconds."
                 logger -t $PREFIX "OK: Initial sync of $COUNTFIN/$NUMTAB 
tables finished in $SECONDS seconds."
                 echo "OK: Initial sync of $COUNTFIN/$NUMTAB tables 
finished in $SECONDS seconds." | mailx -s "$HOST $SECONDS seconds to 
subscribe" -- root
                 break
         fi
         echo "PROGRESS: Initial sync of $COUNTFIN/$NUMTAB tables 
finished. Re-Checking in $SLEEP seconds ..."
         logger -t $PREFIX  "PROGRESS: Initial sync of $COUNTFIN/$NUMTAB
tables finished. Re-Checking in $SLEEP seconds ..."
         sleep $SLEEP
done



regards
Klaus




Am 2023-10-07 17:31, schrieb Shaheed Haque:
> Hi,
> 
> I've been playing with logical replication (currently on PG14),
> specifically in an AWS RDS Postgres context, but NOT using AWS' own
> replication tooling. I'm generally familiar with the challenges of
> distributed systems (such causality, time synchronisation etc), but
> not especially familiar with PG.
> 
> In looking at how to tell how a given subscriber has caught up with
> its publisher, there is plenty of advice around the Web, for example
> https://dba.stackexchange.com/questions/314324/monitor-logical-replication-using-lsn.
> Like this example, much advice ends up talking about using separate
> queries on the publisher and the subscriber to compare LSNs. First, (I
> think) I understand the core difficulty that comparing LSNs is
> inherently racy, but given that, I'm a bit unclear as to why a single
> query on the publisher is not enough...IIUC:
> 
>     * Changes sent from the publisher to the subscriber are identified by
> LSN.
>     * The publisher knows it's own current latest LSN
> (pg_current_wal_lsn()), but this seems not to be exposed at the
> subscriber.
>     * The subscriber knows what it has applied locally and even tells the
> publisher (pg_stat_subscription.latest_end_lsn), but it does not seem
> to be exposed at the publisher.
> 
> Have I missed something? Is there a way to track the LSN delta (given
> that this is known to be racy) just by querying one end?
> 
> Second, how do folk "know" when replication is "done". For example, if
> the two LSNs continued to match for 1 * replication lag? Or N *
> replication lag? What would be a plausible N?
> 
> Third, as we know when logical replication is started, the initial
> table state is captured in a snapshot, and sent across using COPY
> TABLE under the covers. Now, let's say that the publisher is idle
> (i.e. no SQL writes to the user's schema...obviously pg_catalog might
> change as replication is configured and enabled) and that the
> replication starts with the publisher as LSN_start. How could one know
> when the copying is done:
> 
>     * I initially assumed that the publisher's LSN would not change from
> LSN_start, but as the copying proceeds, I see that it DOES change
> (presumably because there are updates happening to pg_catalog, such as
> the temporary slots coming and going).
>     * Is there some kind of singleton state on either publisher or
> subscriber that could be checked to know? (At the moment, I am
> counting the records in all copied tables).
> 
> I realise that the knowledge that the publisher is "idle" is a special
> case, but right now, my test for being "done" is:
> 
>     * Number of records in copied tables matches AND the publisher's
> pg_stat_subscription matches the subscriber's
> pg_stat_subscription.latest_end_lsn.
> 
> Plus or minus the bit about replication lag, is there a better way?
> 
> Thanks, Shaheed