From: Kyotaro Horiguchi Date: Thu, 21 Dec 2017 21:23:25 +0900 Subject: [PATCH 2/6] Add monitoring aid for max_slot_wal_keep_size Adds two columns "status" and "remain" in pg_replication_slot. Setting max_slot_wal_keep_size, replication connections may lose sync by a long delay. The "status" column shows whether the slot is reconnectable or not, or about to lose reserving WAL segments. The "remain" column shows the remaining bytes of WAL that can be advance until the slot loses required WAL records. --- contrib/test_decoding/expected/ddl.out | 4 +- contrib/test_decoding/sql/ddl.sql | 2 + src/backend/access/transam/xlog.c | 117 ++++++++++++++++++++++++++++++--- src/backend/catalog/system_views.sql | 4 +- src/backend/replication/slotfuncs.c | 21 +++++- src/include/access/xlog.h | 2 + src/include/catalog/pg_proc.dat | 6 +- src/test/regress/expected/rules.out | 6 +- 8 files changed, 144 insertions(+), 18 deletions(-) diff --git a/contrib/test_decoding/expected/ddl.out b/contrib/test_decoding/expected/ddl.out index 2c999fd3eb..cf0318f697 100644 --- a/contrib/test_decoding/expected/ddl.out +++ b/contrib/test_decoding/expected/ddl.out @@ -723,8 +723,8 @@ SELECT pg_drop_replication_slot('regression_slot'); (1 row) /* check that the slot is gone */ +\x SELECT * FROM pg_replication_slots; - slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn ------------+--------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+--------------------- (0 rows) +\x diff --git a/contrib/test_decoding/sql/ddl.sql b/contrib/test_decoding/sql/ddl.sql index 856495c952..0f2b9992f7 100644 --- a/contrib/test_decoding/sql/ddl.sql +++ b/contrib/test_decoding/sql/ddl.sql @@ -387,4 +387,6 @@ SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'inc SELECT pg_drop_replication_slot('regression_slot'); /* check that the slot is gone */ +\x SELECT * FROM pg_replication_slots; +\x diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index fcb076100f..d0cc2e0f6d 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -873,7 +873,8 @@ static void checkTimeLineSwitch(XLogRecPtr lsn, TimeLineID newTLI, static void LocalSetXLogInsertAllowed(void); static void CreateEndOfRecoveryRecord(void); static void CheckPointGuts(XLogRecPtr checkPointRedo, int flags); -static XLogSegNo GetOldestKeepSegment(XLogRecPtr currpos, XLogRecPtr minSlotPtr); +static XLogSegNo GetOldestKeepSegment(XLogRecPtr currpos, XLogRecPtr minSlotPtr, + XLogRecPtr targetLSN, uint64 *restBytes); static void KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo); static XLogRecPtr XLogGetReplicationSlotMinimumLSN(void); @@ -9290,6 +9291,63 @@ CreateRestartPoint(int flags) return true; } +/* + * Returns availability of the record at given targetLSN. + * + * Returns three kinds of value in string. + * "streaming" means the WAL record at targetLSN is available. + * "keeping" means it is still available but about to be removed at the next + * checkpoint. + * "losing" means it is already removed. This state is not definite since + * delayed ack from walreceiver can advance restart_lsn later. + * "lost" means it is already removed and no walsenders are running on it. The + * slot is no longer recoverable. + * + * If restBytes is not NULL, sets the remaining LSN bytes until the segment + * for targetLSN will be removed. + */ +char * +GetLsnAvailability(pid_t walsender_pid, XLogRecPtr targetLSN, uint64 *restBytes) +{ + XLogRecPtr currpos; + XLogRecPtr slotPtr; + XLogSegNo currSeg; /* segid of currpos */ + XLogSegNo targetSeg; /* segid of targetLSN */ + XLogSegNo oldestSeg; /* oldest segid kept by max_wal_size */ + XLogSegNo oldestSlotSeg;/* oldest segid kept by slot */ + + Assert(!XLogRecPtrIsInvalid(targetLSN)); + Assert(restBytes); + + currpos = GetXLogWriteRecPtr(); + + /* oldest segment currently needed by slots */ + XLByteToSeg(targetLSN, targetSeg, wal_segment_size); + slotPtr = XLogGetReplicationSlotMinimumLSN(); + oldestSlotSeg = GetOldestKeepSegment(currpos, slotPtr, targetLSN, + restBytes); + + /* oldest segment by max_wal_size */ + XLByteToSeg(currpos, currSeg, wal_segment_size); + oldestSeg = currSeg - + ConvertToXSegs(max_wal_size_mb, wal_segment_size) + 1; + + /* targetSeg is within max_wal_size */ + if (oldestSeg <= targetSeg) + return "streaming"; + + /* targetSeg is being retained by slots */ + if (oldestSlotSeg <= targetSeg) + return "keeping"; + + /* targetSeg is no longer protected. We ignore the possible availability */ + + if (walsender_pid != 0) + return "losing"; + + return "lost"; +} + /* * Returns minimum segment number that the next checkpoint must leave * considering wal_keep_segments, replication slots and @@ -9297,13 +9355,19 @@ CreateRestartPoint(int flags) * * currLSN is the current insert location. * minSlotLSN is the minimum restart_lsn of all active slots. + * targetLSN is used when restBytes is not NULL. + * + * If restBytes is not NULL, sets the remaining LSN bytes until the segment + * for targetLSN will be removed. */ static XLogSegNo -GetOldestKeepSegment(XLogRecPtr currLSN, XLogRecPtr minSlotLSN) +GetOldestKeepSegment(XLogRecPtr currLSN, XLogRecPtr minSlotLSN, + XLogRecPtr targetLSN, uint64 *restBytes) { XLogSegNo currSeg; XLogSegNo minSlotSeg; uint64 keepSegs = 0; /* # of segments actually kept */ + uint64 limitSegs = 0; /* # of maximum segments possibly kept */ XLByteToSeg(currLSN, currSeg, wal_segment_size); XLByteToSeg(minSlotLSN, minSlotSeg, wal_segment_size); @@ -9318,8 +9382,6 @@ GetOldestKeepSegment(XLogRecPtr currLSN, XLogRecPtr minSlotLSN) /* Cap keepSegs by max_slot_wal_keep_size */ if (max_slot_wal_keep_size_mb >= 0) { - uint64 limitSegs; - limitSegs = ConvertToXSegs(max_slot_wal_keep_size_mb, wal_segment_size); /* Reduce it if slots already reserves too many. */ @@ -9327,9 +9389,45 @@ GetOldestKeepSegment(XLogRecPtr currLSN, XLogRecPtr minSlotLSN) keepSegs = limitSegs; } - /* but, keep at least wal_keep_segments segments if any */ - if (wal_keep_segments > 0 && keepSegs < wal_keep_segments) - keepSegs = wal_keep_segments; + if (wal_keep_segments > 0) + { + /* but, keep at least wal_keep_segments segments if any */ + if (keepSegs < wal_keep_segments) + keepSegs = wal_keep_segments; + + /* ditto for limitSegs */ + if (limitSegs < wal_keep_segments) + limitSegs = wal_keep_segments; + } + + /* + * If requested, calculate the remaining LSN bytes until the slot gives up + * keeping WAL records. + */ + if (restBytes) + { + uint64 fragbytes; + XLogSegNo targetSeg; + + *restBytes = 0; + + if (max_slot_wal_keep_size_mb >= 0) + { + XLByteToSeg(targetLSN, targetSeg, wal_segment_size); + + /* avoid underflow */ + if (currSeg <= targetSeg + limitSegs) + { + /* + * This slot still has all required segments. Calculate how + * many LSN bytes the slot has until it loses targetLSN. + */ + fragbytes = wal_segment_size - (currLSN % wal_segment_size); + XLogSegNoOffsetToRecPtr(targetSeg + limitSegs - currSeg, fragbytes, + wal_segment_size, *restBytes); + } + } + } /* avoid underflow, don't go below 1 */ if (currSeg <= keepSegs) @@ -9359,7 +9457,8 @@ KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo) /* * We should keep certain number of WAL segments after this checkpoint. */ - minSegNo = GetOldestKeepSegment(recptr, slotminptr); + minSegNo = + GetOldestKeepSegment(recptr, slotminptr, InvalidXLogRecPtr, NULL); /* * Warn the checkpoint is going to flush the segments required by @@ -9393,7 +9492,7 @@ KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo) if (slot_names) { ereport(WARNING, - (errmsg ("some replication slots have lost required WAL segments"), + (errmsg ("some replication slots lost required WAL segments"), errdetail_plural( "Slot %s lost %ld segment(s).", "Slots %s lost at most %ld segment(s).", diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index ea4c85e395..6a9491e64a 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -849,7 +849,9 @@ CREATE VIEW pg_replication_slots AS L.xmin, L.catalog_xmin, L.restart_lsn, - L.confirmed_flush_lsn + L.confirmed_flush_lsn, + L.wal_status, + L.remain FROM pg_get_replication_slots() AS L LEFT JOIN pg_database D ON (L.datoid = D.oid); diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c index 808a6f5b83..2229a46154 100644 --- a/src/backend/replication/slotfuncs.c +++ b/src/backend/replication/slotfuncs.c @@ -221,7 +221,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS) Datum pg_get_replication_slots(PG_FUNCTION_ARGS) { -#define PG_GET_REPLICATION_SLOTS_COLS 11 +#define PG_GET_REPLICATION_SLOTS_COLS 13 ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; TupleDesc tupdesc; Tuplestorestate *tupstore; @@ -343,6 +343,25 @@ pg_get_replication_slots(PG_FUNCTION_ARGS) else nulls[i++] = true; + if (max_slot_wal_keep_size_mb < 0) + { + values[i++] = CStringGetTextDatum("streaming"); + nulls[i++] = true; + } + else if (restart_lsn == InvalidXLogRecPtr) + { + values[i++] = CStringGetTextDatum("unknown"); + nulls[i++] = true; + } + else + { + uint64 remaining_bytes; + + values[i++] = CStringGetTextDatum( + GetLsnAvailability(active_pid, restart_lsn, &remaining_bytes)); + values[i++] = Int64GetDatum(remaining_bytes); + } + tuplestore_putvalues(tupstore, tupdesc, values, nulls); } LWLockRelease(ReplicationSlotControlLock); diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index b355452072..68ca21f780 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -305,6 +305,8 @@ extern void ShutdownXLOG(int code, Datum arg); extern void InitXLOGAccess(void); extern void CreateCheckPoint(int flags); extern bool CreateRestartPoint(int flags); +extern char *GetLsnAvailability(pid_t walsender_pid, XLogRecPtr targetLSN, + uint64 *restBytes); extern void XLogPutNextOid(Oid nextOid); extern XLogRecPtr XLogRestorePoint(const char *rpName); extern void UpdateFullPageWrites(void); diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 0902dce5f1..300d868980 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -9844,9 +9844,9 @@ proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f', proretset => 't', provolatile => 's', prorettype => 'record', proargtypes => '', - proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn}', - proargmodes => '{o,o,o,o,o,o,o,o,o,o,o}', - proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn}', + proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8}', + proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o}', + proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,remain}', prosrc => 'pg_get_replication_slots' }, { oid => '3786', descr => 'set up a logical replication slot', proname => 'pg_create_logical_replication_slot', provolatile => 'v', diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 210e9cd146..74c44891a4 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -1461,8 +1461,10 @@ pg_replication_slots| SELECT l.slot_name, l.xmin, l.catalog_xmin, l.restart_lsn, - l.confirmed_flush_lsn - FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn) + l.confirmed_flush_lsn, + l.wal_status, + l.remain + FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, remain) LEFT JOIN pg_database d ON ((l.datoid = d.oid))); pg_roles| SELECT pg_authid.rolname, pg_authid.rolsuper, -- 2.16.3 ----Next_Part(Tue_Jul_30_21_30_45_2019_680)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v15-0003-Add-primary_slot_name-to-init_from_backup-in-TAP-tes.patch"