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

diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index c503636a8855afc0bf34869a3441b70cbddecc5c..85459d04b4509a59c7135acbb0cabb45503a4728 100644 (file)
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -300,6 +300,14 @@ postgres   27093  0.0  0.0  30096  2752 ?        Ss   11:34   0:00 postgres: ser
       
      
 
+     
+      pg_stat_wal_receiverpg_stat_wal_receiver
+      Only one row, showing statistics about the WAL receiver from
+       that receiver's connected server.
+       See  for details.
+      
+     
+
      
       pg_stat_sslpg_stat_ssl
       One row per connection (regular and replication), showing information about
@@ -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.
   
 
+  
+   <structname>pg_stat_wal_receiver</structname> View
+   
+    
+    
+      Column
+      Type
+      Description
+     
+    
+
+   
+    
+     pid
+     integer
+     Process ID of the WAL receiver process
+    
+    
+     status
+     text
+     Activity status of the WAL receiver process
+    
+    
+     receive_start_lsn
+     pg_lsn
+     First transaction log position used when WAL receiver is
+      started
+    
+    
+     receive_start_tli
+     integer
+     First timeline number used when WAL receiver is started
+    
+    
+     received_lsn
+     pg_lsn
+     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
+    
+    
+     received_tli
+     integer
+     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
+     
+    
+    
+     last_msg_send_time
+     timestamp with time zone
+     Send time of last message received from origin WAL sender
+    
+    
+     last_msg_receipt_time
+     timestamp with time zone
+     Receipt time of last message received from origin WAL sender
+    
+    
+     latest_end_lsn
+     pg_lsn
+     Last transaction log position reported to origin WAL sender
+    
+    
+     latest_end_time
+     timestamp with time zone
+     Time of last transaction log position reported to origin WAL sender
+    
+    
+     slot_name
+     text
+     Replication slot name used by this WAL receiver
+    
+   
+   
+  
+
+  
+   The pg_stat_wal_receiver view will contain only
+   one row, showing statistics about the WAL receiver from that receiver's
+   connected server.
+  
+
   
    <structname>pg_stat_ssl</structname> View
    

diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 2052afdc8d7183079460a37cddf5034b0a10bc97..506a884e33f6e1cc32d4955329e47e9ed42221a4 100644 (file)
--- 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 81f152940d82f714518c725ed093978085ef54d3..7b36e02faa944b8bbeb3f9e688bfbb0d912d9c0e 100644 (file)
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -46,9 +46,12 @@
 #include 
 #include 
 
+#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/catversion.h b/src/include/catalog/catversion.h
index 31a132ea98462a3663704bfd396c2ec64e6334f5..eaa7a5189186bade8f8e76d5b7f06c7b2d761a90 100644 (file)
--- a/src/include/catalog/catversion.h
+++ b/src/include/catalog/catversion.h
@@ -53,6 +53,6 @@
  */
 
 /*                         yyyymmddN */
-#define CATALOG_VERSION_NO 201601052
+#define CATALOG_VERSION_NO 201601071
 
 #endif

diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 9250545d3b5af6d05eda5ac30bbf5e03d8d65359..86b09a15eaaefc26372ab08d5f7c67d88a3f819f 100644 (file)
--- 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 db40d9d7603e83d6de4351a8e572d1cf54126264..6eacb095d1be140df3ac60dfbc0b2522df500a55 100644 (file)
--- 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 80374e4d506aba58272935d3c48bf3d2dcd3541f..28b061f9d55359352a3a4796a6d166c0d3ea6bb6 100644 (file)
--- 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,