pgstat: add WAL receiver status view & SRF

This new view provides insight into the state of a running WAL receiver
in a HOT standby node.
The information returned includes the PID of the WAL receiver process,
its status (stopped, starting, streaming, etc), start LSN and TLI, last
received LSN and TLI, timestamp of last message send and receipt, latest
end-of-WAL LSN and time, and the name of the slot (if any).

Access to the detailed data is only granted to superusers; others only
get the PID.

Author: Michael Paquier
Reviewer: Haribabu Kommi

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)));
+}