PostgreSQL database replication - background first-class citizen process walreceiver PG_ stat_ wal_ Receiver view

pg_stat_wal_receiver View

pg_stat_wal_receiver View contains only one row , Displays the servers connected from this receiver WAL Receiver Statistics .（The pg_stat_wal_receiver view will contain only one row, showing statistics about the WAL receiver from that receiver’s connected server.）

pg_stat_wal_receiver The view definition is as follows . It needs attention ：

• receive_start_lsn pg_lsnWAL The first pre write log location and used when the receiver starts receive_start_tli integerWAL The first time line number used when the receiver starts （startup Process settings receiveStart and receiveStartTLI, requirement WalReceiver Stream replication ）;
• written_lsn pg_lsn The last prewrite log location was received and written to disk but not flushed （XLogWalRcvWrite Function call pg_write After that ）;
• flushed_lsn pg_lsn Received the last prewrite log location and flushed to disk , The initial value of this field is WAL The first log location to use when the receiver starts （XLogWalRcvFlush Function call fsync After that ）;
• received_tli integer The timeline number received and flushed to the last prewrite log location on the disk , The initial value of this field is WAL The timeline number of the first log location used when the receiver is started ;
• last_msg_send_time timestamp with time zoneWalReceiver Process from WAL sender The sending time contained in the last message received ;
• last_msg_receipt_time timestamp with time zoneWalReceiver Process from WAL sender Of the last message received WalReceiver Own receiving time ;
• latest_end_lsn pg_lsnWalReceiver Process from WAL sender The current last received prewrite log location ;
• latest_end_time timestamp with time zoneWalReceiver Process from WAL sender The sending time of the received current last pre write log location .

CREATE VIEW pg_stat_wal_receiver AS
    SELECT s.pid, s.status, s.receive_start_lsn, s.receive_start_tli, s.written_lsn, s.flushed_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, s.sender_host, s.sender_port, s.conninfo
    FROM pg_stat_get_wal_receiver() s
    WHERE s.pid IS NOT NULL;

pg_stat_get_wal_receiver function

pg_stat_get_wal_receiver The function will get the corresponding value for the above field , It needs attention ：

• written_lsn pg_lsn The last prewrite log location was received and written to disk but not flushed , by XLogWalRcvWrite Function settings WalRcv->writtenUpto（pg_atomic_uint64 CAS Variable , Atomic reading and writing , Multiple processes can read and write simultaneously ）, And this value is called pg_pwrite Later updated LogstreamResult.Write（Write by XLogRecPtr type , Only by WalReceiver Use ） value ;
• flushed_lsn pg_lsn Received the last prewrite log location and flushed to disk , The initial value of this field is WAL The first log location to use when the receiver starts XLogWalRcvFlush Function call fsync After that WalRcv->flushedUpto, And this value is called issue_xlog_fsync Later updated LogstreamResult.Flush（Flush and flushedUpto All for XLogRecPtr type , Updates need to be obtained walrcv->mutex Spin locked ）, call pg_stat_get_wal_receiver Function is not obtained when getting walrcv->mutex Spin locked ;
• received_tli integer The timeline number received and flushed to the last prewrite log location on the disk , The initial value of this field is WAL The timeline number of the first log location used when the receiver is started ;
• last_msg_receipt_time timestamp with time zoneWalReceiver Process from WAL sender Of the last message received WalReceiver Own receiving time ;last_msg_receipt_time Is to get WalRcv->lastMsgReceiptTime, The variable is in ProcessWalSndrMessage Function , Namely WalReceiver Own time when receiving data , That is to say WalReceiver Receiving time of the terminal ;
• latest_end_time timestamp with time zoneWalReceiver Process from WAL sender Time of the last prewrite log location received ,wal sender Set the sending time to the data packet when the client sends data . That is, after receiving a new XLog The sending time of the packet update .
• last_msg_send_time timestamp with time zoneWalReceiver Process from WAL sender The sending time contained in the last message received ;last_msg_send_time It's what we get WalRcv->lastMsgSendTime, The variable is in ProcessWalSndrMessage Function , Namely WalReceiver The time included in the received data （ namely wal sender Set the sending time to the data packet when the client sends data ）; That is, after receiving the heartbeat packet and XLog Data packets are updated when they are sent .

Above , Setting logic for receiving and sending time , Please check out ProcessWalSndrMessage The relevant code at the beginning of the function . About WAL journal written and flushed The location of , Please check out XLogWalRcvWrite and XLogWalRcvFlush The logic of a function .

Datum pg_stat_get_wal_receiver(PG_FUNCTION_ARGS) {
    
	TupleDesc	tupdesc;
	Datum	   *values;
	bool	   *nulls;
	int			pid;
	bool		ready_to_display;
	WalRcvState state;
	XLogRecPtr	receive_start_lsn, written_lsn, flushed_lsn, latest_end_lsn;
	TimeLineID	receive_start_tli, received_tli;
	TimestampTz last_send_time, last_receipt_time, latest_end_time;
	char		sender_host[NI_MAXHOST], slotname[NAMEDATALEN], conninfo[MAXCONNINFO];
	int			sender_port = 0;
	
	SpinLockAcquire(&WalRcv->mutex);  /* Take a lock to ensure value consistency */
	pid = (int) WalRcv->pid;
	ready_to_display = WalRcv->ready_to_display;
	state = WalRcv->walRcvState;
	receive_start_lsn = WalRcv->receiveStart;
	receive_start_tli = WalRcv->receiveStartTLI;
	flushed_lsn = WalRcv->flushedUpto;
	received_tli = WalRcv->receivedTLI;
	last_send_time = WalRcv->lastMsgSendTime;
	last_receipt_time = WalRcv->lastMsgReceiptTime;
	latest_end_lsn = WalRcv->latestWalEnd;
	latest_end_time = WalRcv->latestWalEndTime;
	strlcpy(slotname, (char *) WalRcv->slotname, sizeof(slotname));
	strlcpy(sender_host, (char *) WalRcv->sender_host, sizeof(sender_host));
	sender_port = WalRcv->sender_port;
	strlcpy(conninfo, (char *) WalRcv->conninfo, sizeof(conninfo));
	SpinLockRelease(&WalRcv->mutex);

	/* No WAL receiver (or not ready yet), just return a tuple with NULL values */
	if (pid == 0 || !ready_to_display) PG_RETURN_NULL();  // wal receiver process pid It's not set up yet , perhaps ready_to_display It's not set up yet 

	/* Read "writtenUpto" without holding a spinlock. Note that it may not be consistent with the other shared variables of the WAL receiver protected by a spinlock, but this should not be used for data integrity checks. */
	written_lsn = pg_atomic_read_u64(&WalRcv->writtenUpto);

	/* determine result type */
	if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE) elog(ERROR, "return type must be a row type");
	values = palloc0(sizeof(Datum) * tupdesc->natts);
	nulls = palloc0(sizeof(bool) * tupdesc->natts);
	/* Fetch values */
	values[0] = Int32GetDatum(pid);

	if (!is_member_of_role(GetUserId(), ROLE_PG_READ_ALL_STATS)) {
     /* Only superusers and members of pg_read_all_stats 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, sizeof(bool) * (tupdesc->natts - 1));
	}else{
    
		values[1] = CStringGetTextDatum(WalRcvGetStateString(state));

		if (XLogRecPtrIsInvalid(receive_start_lsn)) nulls[2] = true;  //  Illegal set to null
		else values[2] = LSNGetDatum(receive_start_lsn);
		values[3] = Int32GetDatum(receive_start_tli);
		if (XLogRecPtrIsInvalid(written_lsn)) nulls[4] = true;
		else values[4] = LSNGetDatum(written_lsn);
		if (XLogRecPtrIsInvalid(flushed_lsn)) nulls[5] = true;
		else values[5] = LSNGetDatum(flushed_lsn);
		values[6] = Int32GetDatum(received_tli);
		
		if (last_send_time == 0) nulls[7] = true;
		else values[7] = TimestampTzGetDatum(last_send_time);
		if (last_receipt_time == 0) nulls[8] = true;
		else values[8] = TimestampTzGetDatum(last_receipt_time);
		if (XLogRecPtrIsInvalid(latest_end_lsn)) nulls[9] = true;
		else values[9] = LSNGetDatum(latest_end_lsn);
		if (latest_end_time == 0) nulls[10] = true;
		else values[10] = TimestampTzGetDatum(latest_end_time);
			
		if (*slotname == '\0') nulls[11] = true;
		else values[11] = CStringGetTextDatum(slotname);
		if (*sender_host == '\0') nulls[12] = true;
		else values[12] = CStringGetTextDatum(sender_host);
		if (sender_port == 0) nulls[13] = true;
		else values[13] = Int32GetDatum(sender_port);
		if (*conninfo == '\0') nulls[14] = true;
		else  values[14] = CStringGetTextDatum(conninfo);
	}	
	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls))); /* Returns the record as Datum */
}