fixing repmgr repl_status columns

repmgr repl_status had the column time_lag which was documented to be
the time a standby is behind master. In fact it only works like this
when viewed on the standby and not on the master: there it only was the
time of the last status update. We dropped that column and replaced it
by a new column „communication_time_lag“ which is the content of the
repl_status column on the master. On the standby we contain the time of
the last update in shared mem though refer always to the correct time
nonetheless where repl_status is queried. We also added a new column,
„replication_time_lag“, which refers to the apply delay.

repmgr.c

@@ -2089,6 +2089,35 @@ create_schema(PGconn *conn)
 	}
 	PQclear(res);
 
+	/* to avoid confusion of the time_lag field and provide a consistent UI we
+	 * use these functions for providing the latest update timestamp */
+	sqlquery_snprintf(sqlquery,
+					  "CREATE FUNCTION %s.repmgr_update_last_updated() RETURNS TIMESTAMP WITH TIME ZONE "
+					  "AS '$libdir/repmgr_funcs', 'repmgr_update_last_updated' "
+					  " LANGUAGE C STRICT", repmgr_schema);
+	res = PQexec(conn, sqlquery);
+	if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
+	{
+		fprintf(stderr, "Cannot create the function repmgr_update_last_updated: %s\n",
+		        PQerrorMessage(conn));
+		return false;
+	}
+	PQclear(res);
+
+
+	sqlquery_snprintf(sqlquery,
+					  "CREATE FUNCTION %s.repmgr_get_last_updated() RETURNS TIMESTAMP WITH TIME ZONE "
+					  "AS '$libdir/repmgr_funcs', 'repmgr_get_last_updated' "
+					  "LANGUAGE C STRICT", repmgr_schema);
+	res = PQexec(conn, sqlquery);
+	if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
+	{
+		fprintf(stderr, "Cannot create the function repmgr_get_last_updated: %s\n",
+		        PQerrorMessage(conn));
+		return false;
+	}
+	PQclear(res);
+
 
 	/* ... the tables */
 	sqlquery_snprintf(sqlquery, "CREATE TABLE %s.repl_nodes (        "
@@ -2113,6 +2142,7 @@ create_schema(PGconn *conn)
 	                  "  primary_node                   INTEGER NOT NULL, "
 	                  "  standby_node                   INTEGER NOT NULL, "
 	                  "  last_monitor_time              TIMESTAMP WITH TIME ZONE NOT NULL, "
+	                  "  last_apply_time                TIMESTAMP WITH TIME ZONE, "
 	                  "  last_wal_primary_location      TEXT NOT NULL,   "
 	                  "  last_wal_standby_location      TEXT,  "
 	                  "  replication_lag                BIGINT NOT NULL, "
@@ -2133,12 +2163,13 @@ create_schema(PGconn *conn)
 	                  " SELECT primary_node, standby_node, name AS standby_name, last_monitor_time, "
 	                  "        last_wal_primary_location, last_wal_standby_location, "
 	                  "        pg_size_pretty(replication_lag) replication_lag, "
+	                  "        age(now(), last_apply_time) AS replication_time_lag, "
 	                  "        pg_size_pretty(apply_lag) apply_lag, "
-	                  "        age(now(), last_monitor_time) AS time_lag "
+	                  "        age(now(), CASE WHEN pg_is_in_recovery() THEN %s.repmgr_get_last_updated() ELSE last_monitor_time END) AS communication_time_lag "
 	                  "   FROM %s.repl_monitor JOIN %s.repl_nodes ON standby_node = id "
 	                  "  WHERE (standby_node, last_monitor_time) IN (SELECT standby_node, MAX(last_monitor_time) "
 	                  "                                                FROM %s.repl_monitor GROUP BY 1)",
-	                  repmgr_schema, repmgr_schema, repmgr_schema, repmgr_schema);
+	                  repmgr_schema, repmgr_schema, repmgr_schema, repmgr_schema, repmgr_schema);
 	log_debug(_("master register: %s\n"), sqlquery);
 
 	res = PQexec(conn, sqlquery);

repmgrd.c

@@ -530,6 +530,7 @@ StandbyMonitor(void)
 	char last_wal_primary_location[MAXLEN];
 	char last_wal_standby_received[MAXLEN];
 	char last_wal_standby_applied[MAXLEN];
+	char last_wal_standby_applied_timestamp[MAXLEN];
 
 	unsigned long long int lsn_primary;
 	unsigned long long int lsn_standby_received;
@@ -637,7 +638,7 @@ StandbyMonitor(void)
 	sqlquery_snprintf(
 	    sqlquery,
 	    "SELECT CURRENT_TIMESTAMP, pg_last_xlog_receive_location(), "
-	    "pg_last_xlog_replay_location()");
+	    "pg_last_xlog_replay_location(), pg_last_xact_replay_timestamp()");
 
 	res = PQexec(myLocalConn, sqlquery);
 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
@@ -651,6 +652,7 @@ StandbyMonitor(void)
 	strncpy(monitor_standby_timestamp, PQgetvalue(res, 0, 0), MAXLEN);
 	strncpy(last_wal_standby_received , PQgetvalue(res, 0, 1), MAXLEN);
 	strncpy(last_wal_standby_applied , PQgetvalue(res, 0, 2), MAXLEN);
+	strncpy(last_wal_standby_applied_timestamp, PQgetvalue(res, 0, 3), MAXLEN);
 	PQclear(res);
 
 	/* Get primary xlog info */
@@ -678,9 +680,10 @@ StandbyMonitor(void)
 	sqlquery_snprintf(sqlquery,
 	                  "INSERT INTO %s.repl_monitor "
 	                  "VALUES(%d, %d, '%s'::timestamp with time zone, "
-	                  " '%s', '%s', "
+	                  " '%s'::timestamp with time zone, '%s', '%s', "
 	                  " %lld, %lld)", repmgr_schema,
 	                  primary_options.node, local_options.node, monitor_standby_timestamp,
+	                  last_wal_standby_applied_timestamp,
 	                  last_wal_primary_location,
 	                  last_wal_standby_received,
 	                  (lsn_primary - lsn_standby_received),

sql/repmgr_funcs.c

@@ -15,6 +15,7 @@
 #include "storage/shmem.h"
 #include "storage/spin.h"
 #include "utils/builtins.h"
+#include "utils/timestamp.h"
 
 /* same definition as the one in xlog_internal.h */
 #define MAXFNAMELEN 	64
@@ -28,6 +29,7 @@ typedef struct repmgrSharedState
 {
     LWLockId    lock;           		/* protects search/modification */
     char		location[MAXFNAMELEN];	/* last known xlog location */
+	TimestampTz last_updated;
 } repmgrSharedState;
 
 /* Links to shared memory state */
@@ -49,6 +51,12 @@ Datum repmgr_get_last_standby_location(PG_FUNCTION_ARGS);
 PG_FUNCTION_INFO_V1(repmgr_update_standby_location);
 PG_FUNCTION_INFO_V1(repmgr_get_last_standby_location);
 
+Datum repmgr_update_last_updated(PG_FUNCTION_ARGS);
+Datum repmgr_get_last_updated(PG_FUNCTION_ARGS);
+
+PG_FUNCTION_INFO_V1(repmgr_update_last_updated);
+PG_FUNCTION_INFO_V1(repmgr_get_last_updated);
+
 
 /*
  * Module load callback
@@ -187,3 +195,38 @@ repmgr_update_standby_location(PG_FUNCTION_ARGS)
 
 	PG_RETURN_BOOL(repmgr_set_standby_location(locationstr));
 }
+
+/* update and return last updated with current timestamp */
+Datum
+repmgr_update_last_updated(PG_FUNCTION_ARGS)
+{
+	TimestampTz last_updated = GetCurrentTimestamp();
+
+	/* Safety check... */
+	if (!shared_state)
+		PG_RETURN_NULL();
+
+	LWLockAcquire(shared_state->lock, LW_SHARED);
+	shared_state->last_updated = last_updated;
+	LWLockRelease(shared_state->lock);
+
+	PG_RETURN_TIMESTAMPTZ(last_updated);
+}
+
+
+/* get last updated timestamp */
+Datum
+repmgr_get_last_updated(PG_FUNCTION_ARGS)
+{
+	TimestampTz last_updated;
+
+    /* Safety check... */
+    if (!shared_state)
+        PG_RETURN_NULL();
+
+	LWLockAcquire(shared_state->lock, LW_EXCLUSIVE);
+	last_updated = shared_state->last_updated;
+	LWLockRelease(shared_state->lock);
+
+	PG_RETURN_TIMESTAMPTZ(last_updated);
+}

sql/repmgr_funcs.sql.in

@@ -1,6 +1,6 @@
 /*
  * repmgr_function.sql
- * Copyright (c) 2ndQuadrant, 2010
+ * Copyright (c) 2ndQuadrant, 2010-2014
  *
  */
 
@@ -13,3 +13,11 @@ LANGUAGE C STRICT;
 CREATE FUNCTION repmgr_get_last_standby_location() RETURNS text
 AS 'MODULE_PATHNAME', 'repmgr_get_last_standby_location'
 LANGUAGE C STRICT;
+
+CREATE FUNCTION repmgr_update_last_updated() RETURNS TIMESTAMP WITH TIME ZONE
+AS 'MODULE_PATHNAME', 'repmgr_update_last_updated'
+LANGUAGE C STRICT;
+
+CREATE FUNCTION repmgr_get_last_updated() RETURNS TIMESTAMP WITH TIME ZONE
+AS 'MODULE_PATHNAME', 'repmgr_get_last_updated'
+LANGUAGE C STRICT;

sql/uninstall_repmgr_funcs.sql

@@ -1,2 +1,11 @@
+/*
+ * uninstall_repmgr_funcs.sql
+ * Copyright (c) 2ndQuadrant, 2010-2014
+ *
+ */
+
 DROP FUNCTION repmgr_update_standby_location(text);
 DROP FUNCTION repmgr_get_last_standby_location();
+
+DROP FUNCTION repmgr_update_last_updated();
+DROP FUNCTION repmgr_get_last_updated();