From 4bc33d1497c302b8669b1f1d9d43f2f806029693 Mon Sep 17 00:00:00 2001
From: Michael Paquier <michael@otacoo.com>
Date: Fri, 18 Dec 2015 22:44:21 +0900
Subject: [PATCH] Add system view and function to report WAL receiver activity

---
 doc/src/sgml/monitoring.sgml          |  91 ++++++++++++++++++++
 src/backend/catalog/system_views.sql  |  16 ++++
 src/backend/replication/walreceiver.c | 154 ++++++++++++++++++++++++++++++++++
 src/include/catalog/pg_proc.h         |   2 +
 src/include/replication/walreceiver.h |   2 +
 src/test/regress/expected/rules.out   |  13 +++
 6 files changed, 278 insertions(+)

diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c503636..85459d0 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -301,6 +301,14 @@ postgres   27093  0.0  0.0  30096  2752 ?        Ss   11:34   0:00 postgres: ser
      </row>
 
      <row>
+      <entry><structname>pg_stat_wal_receiver</><indexterm><primary>pg_stat_wal_receiver</primary></indexterm></entry>
+      <entry>Only one row, showing statistics about the WAL receiver from
+       that receiver's connected server.
+       See <xref linkend="pg-stat-wal-receiver-view"> for details.
+      </entry>
+     </row>
+
+     <row>
       <entry><structname>pg_stat_ssl</><indexterm><primary>pg_stat_ssl</primary></indexterm></entry>
       <entry>One row per connection (regular and replication), showing information about
        SSL used on this connection.
@@ -833,6 +841,89 @@ postgres   27093  0.0  0.0  30096  2752 ?        Ss   11:34   0:00 postgres: ser
    listed; no information is available about downstream standby servers.
   </para>
 
+  <table id="pg-stat-wal-receiver-view" xreflabel="pg_stat_wal_receiver">
+   <title><structname>pg_stat_wal_receiver</structname> View</title>
+   <tgroup cols="3">
+    <thead>
+    <row>
+      <entry>Column</entry>
+      <entry>Type</entry>
+      <entry>Description</entry>
+     </row>
+    </thead>
+
+   <tbody>
+    <row>
+     <entry><structfield>pid</></entry>
+     <entry><type>integer</></entry>
+     <entry>Process ID of the WAL receiver process</entry>
+    </row>
+    <row>
+     <entry><structfield>status</></entry>
+     <entry><type>text</></entry>
+     <entry>Activity status of the WAL receiver process</entry>
+    </row>
+    <row>
+     <entry><structfield>receive_start_lsn</></entry>
+     <entry><type>pg_lsn</></entry>
+     <entry>First transaction log position used when WAL receiver is
+      started</entry>
+    </row>
+    <row>
+     <entry><structfield>receive_start_tli</></entry>
+     <entry><type>integer</></entry>
+     <entry>First timeline number used when WAL receiver is started</entry>
+    </row>
+    <row>
+     <entry><structfield>received_lsn</></entry>
+     <entry><type>pg_lsn</></entry>
+     <entry>Last transaction log position already received and flushed to
+      disk, the initial value of this field being the first log position used
+      when WAL receiver is started</entry>
+    </row>
+    <row>
+     <entry><structfield>received_tli</></entry>
+     <entry><type>integer</></entry>
+     <entry>Timeline number of last transaction log position received and
+      flushed to disk, the initial value of this field being the timeline
+      number of the first log position used when WAL receiver is started
+     </entry>
+    </row>
+    <row>
+     <entry><structfield>last_msg_send_time</></entry>
+     <entry><type>timestamp with time zone</></entry>
+     <entry>Send time of last message received from origin WAL sender</entry>
+    </row>
+    <row>
+     <entry><structfield>last_msg_receipt_time</></entry>
+     <entry><type>timestamp with time zone</></entry>
+     <entry>Receipt time of last message received from origin WAL sender</entry>
+    </row>
+    <row>
+     <entry><structfield>latest_end_lsn</></entry>
+     <entry><type>pg_lsn</></entry>
+     <entry>Last transaction log position reported to origin WAL sender</entry>
+    </row>
+    <row>
+     <entry><structfield>latest_end_time</></entry>
+     <entry><type>timestamp with time zone</></entry>
+     <entry>Time of last transaction log position reported to origin WAL sender</entry>
+    </row>
+    <row>
+     <entry><structfield>slot_name</></entry>
+     <entry><type>text</></entry>
+     <entry>Replication slot name used by this WAL receiver</entry>
+    </row>
+   </tbody>
+   </tgroup>
+  </table>
+
+  <para>
+   The <structname>pg_stat_wal_receiver</structname> view will contain only
+   one row, showing statistics about the WAL receiver from that receiver's
+   connected server.
+  </para>
+
   <table id="pg-stat-ssl-view" xreflabel="pg_stat_ssl">
    <title><structname>pg_stat_ssl</structname> View</title>
    <tgroup cols="3">
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 2052afd..506a884 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -662,6 +662,22 @@ CREATE VIEW pg_stat_replication AS
     WHERE S.usesysid = U.oid AND
             S.pid = W.pid;
 
+CREATE VIEW pg_stat_wal_receiver AS
+    SELECT
+            s.pid,
+            s.status,
+            s.receive_start_lsn,
+            s.receive_start_tli,
+            s.received_lsn,
+            s.received_tli,
+            s.last_msg_send_time,
+            s.last_msg_receipt_time,
+            s.latest_end_lsn,
+            s.latest_end_time,
+            s.slot_name
+    FROM pg_stat_get_wal_receiver() s
+    WHERE s.pid IS NOT NULL;
+
 CREATE VIEW pg_stat_ssl AS
     SELECT
             S.pid,
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index 81f1529..7b36e02 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -46,9 +46,12 @@
 #include <signal.h>
 #include <unistd.h>
 
+#include "access/htup_details.h"
 #include "access/timeline.h"
 #include "access/transam.h"
 #include "access/xlog_internal.h"
+#include "catalog/pg_type.h"
+#include "funcapi.h"
 #include "libpq/pqformat.h"
 #include "libpq/pqsignal.h"
 #include "miscadmin.h"
@@ -57,7 +60,9 @@
 #include "storage/ipc.h"
 #include "storage/pmsignal.h"
 #include "storage/procarray.h"
+#include "utils/builtins.h"
 #include "utils/guc.h"
+#include "utils/pg_lsn.h"
 #include "utils/ps_status.h"
 #include "utils/resowner.h"
 #include "utils/timestamp.h"
@@ -1215,3 +1220,152 @@ ProcessWalSndrMessage(XLogRecPtr walEnd, TimestampTz sendTime)
 		pfree(receipttime);
 	}
 }
+
+/*
+ * Return a string constant representing the state. This is used
+ * in system functions and views, and should *not* be translated.
+ */
+static const char *
+WalRcvGetStateString(WalRcvState state)
+{
+	switch (state)
+	{
+		case WALRCV_STOPPED:
+			return "stopped";
+		case WALRCV_STARTING:
+			return "starting";
+		case WALRCV_STREAMING:
+			return "streaming";
+		case WALRCV_WAITING:
+			return "waiting";
+		case WALRCV_RESTARTING:
+			return "restarting";
+		case WALRCV_STOPPING:
+			return "stopping";
+	}
+	return "UNKNOWN";
+}
+
+/*
+ * Returns activity of WAL receiver, including pid, state and xlog locations
+ * received from the WAL sender of another server.
+ */
+Datum
+pg_stat_get_wal_receiver(PG_FUNCTION_ARGS)
+{
+#define PG_STAT_GET_WAL_RECEIVER_COLS	11
+	TupleDesc	tupdesc;
+	Datum		values[PG_STAT_GET_WAL_RECEIVER_COLS];
+	bool		nulls[PG_STAT_GET_WAL_RECEIVER_COLS];
+	WalRcvData *walrcv = WalRcv;
+	WalRcvState state;
+	XLogRecPtr	receive_start_lsn;
+	TimeLineID	receive_start_tli;
+	XLogRecPtr	received_lsn;
+	TimeLineID	received_tli;
+	TimestampTz	last_send_time;
+	TimestampTz	last_receipt_time;
+	XLogRecPtr	latest_end_lsn;
+	TimestampTz	latest_end_time;
+	char	   *slotname;
+
+	/* No WAL receiver, just return a tuple with NULL values */
+	if (walrcv->pid == 0)
+		PG_RETURN_NULL();
+
+	/* Initialise values and NULL flags arrays */
+	MemSet(values, 0, sizeof(values));
+	MemSet(nulls, 0, sizeof(nulls));
+
+	/* Initialise attributes information in the tuple descriptor */
+	tupdesc = CreateTemplateTupleDesc(PG_STAT_GET_WAL_RECEIVER_COLS, false);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 1, "pid",
+					   INT4OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 2, "status",
+					   TEXTOID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 3, "receive_start_lsn",
+					   LSNOID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 4, "receive_start_tli",
+					   INT4OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 5, "received_lsn",
+					   LSNOID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 6, "received_tli",
+					   INT4OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 7, "last_msg_send_time",
+					   TIMESTAMPTZOID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 8, "last_msg_receipt_time",
+					   TIMESTAMPTZOID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 9, "latest_end_lsn",
+					   LSNOID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 10, "latest_end_time",
+					   TIMESTAMPTZOID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 11, "slot_name",
+					   TEXTOID, -1, 0);
+
+	BlessTupleDesc(tupdesc);
+
+	/* Take a lock to ensure value consistency */
+	SpinLockAcquire(&walrcv->mutex);
+	state = walrcv->walRcvState;
+	receive_start_lsn = walrcv->receiveStart;
+	receive_start_tli = walrcv->receiveStartTLI;
+	received_lsn = walrcv->receivedUpto;
+	received_tli = walrcv->receivedTLI;
+	last_send_time = walrcv->lastMsgSendTime;
+	last_receipt_time = walrcv->lastMsgReceiptTime;
+	latest_end_lsn = walrcv->latestWalEnd;
+	latest_end_time = walrcv->latestWalEndTime;
+	slotname = pstrdup(walrcv->slotname);
+	SpinLockRelease(&walrcv->mutex);
+
+	/* Fetch values */
+	values[0] = Int32GetDatum(walrcv->pid);
+
+	if (!superuser())
+	{
+		/*
+		 * Only superusers can see details. Other users only get the pid
+		 * value to know whether it is a WAL receiver, but no details.
+		 */
+		MemSet(&nulls[1], true, PG_STAT_GET_WAL_RECEIVER_COLS - 1);
+	}
+	else
+	{
+		values[1] = CStringGetTextDatum(WalRcvGetStateString(state));
+
+		if (XLogRecPtrIsInvalid(receive_start_lsn))
+			nulls[2] = true;
+		else
+			values[2] = LSNGetDatum(receive_start_lsn);
+		values[3] = Int32GetDatum(receive_start_tli);
+		if (XLogRecPtrIsInvalid(received_lsn))
+			nulls[4] = true;
+		else
+			values[4] = LSNGetDatum(received_lsn);
+		values[5] = Int32GetDatum(received_tli);
+		if (last_send_time == 0)
+			nulls[6] = true;
+		else
+			values[6] = TimestampTzGetDatum(last_send_time);
+		if (last_receipt_time == 0)
+			nulls[7] = true;
+		else
+			values[7] = TimestampTzGetDatum(last_receipt_time);
+		if (XLogRecPtrIsInvalid(latest_end_lsn))
+			nulls[8] = true;
+		else
+			values[8] = LSNGetDatum(latest_end_lsn);
+		if (latest_end_time == 0)
+			nulls[9] = true;
+		else
+			values[9] = TimestampTzGetDatum(latest_end_time);
+		if (*slotname == '\0')
+			nulls[10] = true;
+		else
+			values[10] = CStringGetTextDatum(slotname);
+	}
+
+	/* Returns the record as Datum */
+	PG_RETURN_DATUM(HeapTupleGetDatum(
+						  heap_form_tuple(tupdesc, values, nulls)));
+}
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 9250545..86b09a1 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -2787,6 +2787,8 @@ DATA(insert OID = 2022 (  pg_stat_get_activity			PGNSP PGUID 12 1 100 0 0 f f f
 DESCR("statistics: information about currently active backends");
 DATA(insert OID = 3099 (  pg_stat_get_wal_senders	PGNSP PGUID 12 1 10 0 0 f f f f f t s r 0 0 2249 "" "{23,25,3220,3220,3220,3220,23,25}" "{o,o,o,o,o,o,o,o}" "{pid,state,sent_location,write_location,flush_location,replay_location,sync_priority,sync_state}" _null_ _null_ pg_stat_get_wal_senders _null_ _null_ _null_ ));
 DESCR("statistics: information about currently active replication");
+DATA(insert OID = 3317 (  pg_stat_get_wal_receiver	PGNSP PGUID 12 1 0 0 0 f f f f f f s r 0 0 2249 "" "{23,25,3220,23,3220,23,1184,1184,3220,1184,25}" "{o,o,o,o,o,o,o,o,o,o,o}" "{pid,status,receive_start_lsn,receive_start_tli,received_lsn,received_tli,last_msg_send_time,last_msg_receipt_time,latest_end_lsn,latest_end_time,slot_name}" _null_ _null_ pg_stat_get_wal_receiver _null_ _null_ _null_ ));
+DESCR("statistics: information about WAL receiver");
 DATA(insert OID = 2026 (  pg_backend_pid				PGNSP PGUID 12 1 0 0 0 f f f f t f s r 0 0 23 "" _null_ _null_ _null_ _null_ _null_ pg_backend_pid _null_ _null_ _null_ ));
 DESCR("statistics: current backend PID");
 DATA(insert OID = 1937 (  pg_stat_get_backend_pid		PGNSP PGUID 12 1 0 0 0 f f f f t f s r 1 0 23 "23" _null_ _null_ _null_ _null_ _null_ pg_stat_get_backend_pid _null_ _null_ _null_ ));
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index db40d9d..6eacb09 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -14,6 +14,7 @@
 
 #include "access/xlog.h"
 #include "access/xlogdefs.h"
+#include "fmgr.h"
 #include "storage/latch.h"
 #include "storage/spin.h"
 #include "pgtime.h"
@@ -148,6 +149,7 @@ extern PGDLLIMPORT walrcv_disconnect_type walrcv_disconnect;
 
 /* prototypes for functions in walreceiver.c */
 extern void WalReceiverMain(void) pg_attribute_noreturn();
+extern Datum pg_stat_get_wal_receiver(PG_FUNCTION_ARGS);
 
 /* prototypes for functions in walreceiverfuncs.c */
 extern Size WalRcvShmemSize(void);
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 80374e4..28b061f 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1848,6 +1848,19 @@ pg_stat_user_tables| SELECT pg_stat_all_tables.relid,
     pg_stat_all_tables.autoanalyze_count
    FROM pg_stat_all_tables
   WHERE ((pg_stat_all_tables.schemaname <> ALL (ARRAY['pg_catalog'::name, 'information_schema'::name])) AND (pg_stat_all_tables.schemaname !~ '^pg_toast'::text));
+pg_stat_wal_receiver| SELECT s.pid,
+    s.status,
+    s.receive_start_lsn,
+    s.receive_start_tli,
+    s.received_lsn,
+    s.received_tli,
+    s.last_msg_send_time,
+    s.last_msg_receipt_time,
+    s.latest_end_lsn,
+    s.latest_end_time,
+    s.slot_name
+   FROM pg_stat_get_wal_receiver() s(pid, status, receive_start_lsn, receive_start_tli, received_lsn, received_tli, last_msg_send_time, last_msg_receipt_time, latest_end_lsn, latest_end_time, slot_name)
+  WHERE (s.pid IS NOT NULL);
 pg_stat_xact_all_tables| SELECT c.oid AS relid,
     n.nspname AS schemaname,
     c.relname,
-- 
2.6.4

