agora inbox for [email protected]
help / color / mirror / Atom feedFrom: Kyotaro Horiguchi <[email protected]>
Subject: [PATCH v17 2/6] Add monitoring aid for max_slot_wal_keep_size
Date: Thu, 21 Dec 2017 21:23:25 +0900
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 | 234 ++++++++++++++++++++++++-
src/backend/catalog/system_views.sql | 4 +-
src/backend/replication/slotfuncs.c | 17 +-
src/include/access/xlog.h | 3 +
src/include/catalog/pg_proc.dat | 6 +-
src/test/regress/expected/rules.out | 6 +-
8 files changed, 259 insertions(+), 17 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 ba6b9b0d4f..bb6bfda529 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -872,7 +872,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, int64 *restBytes);
static void KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo);
static XLogRecPtr XLogGetReplicationSlotMinimumLSN(void);
@@ -3900,6 +3901,55 @@ XLogGetLastRemovedSegno(void)
return lastRemovedSegNo;
}
+/*
+ * Return the oldest WAL segment file.
+ *
+ * The returned value is XLogGetLastRemovedSegno() + 1 when the function
+ * returns a valid value. Otherwise this function scans over WAL files and
+ * finds the oldest segment at the first time, which could be very slow.
+ */
+XLogSegNo
+FindOldestXLogFileSegNo(void)
+{
+ static XLogSegNo lastFoundOldestSeg = 0;
+ DIR *xldir;
+ struct dirent *xlde;
+ XLogSegNo segno = XLogGetLastRemovedSegno();
+
+ if (segno > 0)
+ return segno + 1;
+
+ if (lastFoundOldestSeg > 0)
+ return lastFoundOldestSeg;
+
+ xldir = AllocateDir(XLOGDIR);
+ while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL)
+ {
+ TimeLineID tli;
+ XLogSegNo fsegno;
+
+ /* Ignore files that are not XLOG segments */
+ if (!IsXLogFileName(xlde->d_name) &&
+ !IsPartialXLogFileName(xlde->d_name))
+ continue;
+
+ XLogFromFileName(xlde->d_name, &tli, &fsegno, wal_segment_size);
+
+ /*
+ * Get minimum segment ignoring timeline ID, the same way with
+ * RemoveOldXlogFiles().
+ */
+ if (segno == 0 || fsegno < segno)
+ segno = fsegno;
+ }
+
+ FreeDir(xldir);
+
+ lastFoundOldestSeg = segno;
+
+ return segno;
+}
+
/*
* Update the last removed segno pointer in shared memory, to reflect
* that the given XLOG file has been removed.
@@ -9324,6 +9374,124 @@ CreateRestartPoint(int flags)
return true;
}
+/*
+ * Detect availability of the record at given targetLSN.
+ *
+ * targetLSN is restart_lsn of a slot.
+ * walsender_pid is the slot's walsender PID.
+ * restBytes is the pointer to uint64 variable, to store the remaining bytes
+ * until the slot goes into "losing" state.
+ *
+ * Returns in four kinds of strings.
+ *
+ * "normal" means targetLSN is available because it is in the range of
+ * max_wal_size.
+ *
+ * "keeping" means it is still available by preserving extra segments beyond
+ * max_wal_size.
+ *
+ * "losing" means it is being removed or already removed but the walsender
+ * using the given slot is keeping repliation stream yet. The state may return
+ * to "keeping" or "normal" state if the walsender advances restart_lsn.
+ *
+ * "lost" means it is definitly lost. The walsender worked on the slot has
+ * been stopped.
+ *
+ * returns NULL if restart_lsn is invalid.
+ *
+ * -1 is stored to restBytes if the values is useless.
+ */
+char *
+GetLsnAvailability(XLogRecPtr restart_lsn, pid_t walsender_pid,
+ int64 *restBytes)
+{
+ XLogRecPtr currpos;
+ XLogRecPtr slotPtr;
+ XLogSegNo currSeg; /* segid of currpos */
+ XLogSegNo restartSeg; /* segid of restart_lsn */
+ XLogSegNo oldestSeg; /* actual oldest segid */
+ XLogSegNo oldestSegMaxWalSize; /* oldest segid kept by max_wal_size */
+ XLogSegNo oldestSlotSeg;/* oldest segid kept by slot */
+ uint64 keepSegs;
+
+ Assert(restBytes);
+
+ /* the case where the slot has never been activated */
+ if (XLogRecPtrIsInvalid(restart_lsn))
+ {
+ *restBytes = -1;
+ return NULL;
+ }
+
+ /*
+ * slot limitation is not activated, WAL files are kept unlimitedlllly in
+ * the case.
+ */
+ if (max_slot_wal_keep_size_mb < 0)
+ {
+ *restBytes = -1;
+ return "normal";
+ }
+
+ currpos = GetXLogWriteRecPtr();
+
+ /* calculate oldest segment currently needed by slots */
+ XLByteToSeg(restart_lsn, restartSeg, wal_segment_size);
+ slotPtr = XLogGetReplicationSlotMinimumLSN();
+ oldestSlotSeg = GetOldestKeepSegment(currpos, slotPtr, restart_lsn,
+ restBytes);
+
+ /* find the oldest segment file actually exists */
+ oldestSeg = FindOldestXLogFileSegNo();
+
+ /* calculate oldest segment by max_wal_size */
+ XLByteToSeg(currpos, currSeg, wal_segment_size);
+ keepSegs = ConvertToXSegs(max_wal_size_mb, wal_segment_size) + 1;
+
+ if (currSeg > keepSegs)
+ oldestSegMaxWalSize = currSeg - keepSegs;
+ else
+ oldestSegMaxWalSize = 1;
+
+
+ /*
+ * If max_slot_wal_keep_size has changed after the last call, the segment
+ * that would been kept by the current setting might have been lost by the
+ * previous setting. No point in showing normal or keeping status values if
+ * the restartSeg is known to be lost.
+ */
+ if (restartSeg >= oldestSeg)
+ {
+ /*
+ * show "normal" when restartSeg is within max_wal_size. If
+ * max_slot_wal_keep_size is smaller than max_wal_size, there's no
+ * point in showing the status.
+ */
+ if (max_slot_wal_keep_size_mb >= max_wal_size_mb &&
+ oldestSegMaxWalSize <= restartSeg)
+ return "normal";
+
+ /* being retained by slots */
+ if (oldestSlotSeg <= restartSeg)
+ return "keeping";
+ }
+
+ /* it is useless for the states below */
+ *restBytes = -1;
+
+ /*
+ * The segment is alrady lost or being lost. If the oldest segment is just
+ * after the restartSeg, running walsender may be reading the just removed
+ * segment. The walsender may safely move to the oldest existing segment in
+ * that case.
+ */
+ if (oldestSeg == restartSeg + 1 && walsender_pid != 0)
+ return "losing";
+
+ /* definitely lost. stopped walsender can no longer restart */
+ return "lost";
+}
+
/*
* Returns minimum segment number that the next checkpoint must leave
* considering wal_keep_segments, replication slots and
@@ -9331,13 +9499,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, int64 *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);
@@ -9352,8 +9526,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. */
@@ -9361,9 +9533,54 @@ 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;
+
+ XLByteToSeg(targetLSN, targetSeg, wal_segment_size);
+
+ /* avoid underflow */
+ if (currSeg <= targetSeg + limitSegs)
+ {
+ uint64 restbytes;
+
+ /*
+ * 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);
+
+ /*
+ * not realistic, but make sure that it is not out of the
+ * range of int64. No problem to do so since such large values
+ * have no significant difference.
+ */
+ if (restbytes > PG_INT64_MAX)
+ restbytes = PG_INT64_MAX;
+ *restBytes = restbytes;
+ }
+ }
/* avoid underflow, don't go below 1 */
if (currSeg <= keepSegs)
@@ -9393,7 +9610,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
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index f7800f01a6..2fe346461d 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -854,7 +854,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 46e6dd4d12..7db48aaa14 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;
@@ -276,6 +276,8 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
Oid database;
NameData slot_name;
NameData plugin;
+ char *walstate;
+ int64 remaining_bytes;
int i;
if (!slot->in_use)
@@ -343,6 +345,19 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
else
nulls[i++] = true;
+ walstate =
+ GetLsnAvailability(restart_lsn, active_pid, &remaining_bytes);
+
+ if (walstate)
+ values[i++] = CStringGetTextDatum(walstate);
+ else
+ nulls[i++] = true;
+
+ if (remaining_bytes >= 0)
+ values[i++] = Int64GetDatum(remaining_bytes);
+ else
+ nulls[i++] = true;
+
tuplestore_putvalues(tupstore, tupdesc, values, nulls);
}
LWLockRelease(ReplicationSlotControlLock);
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index c454e9a061..1f92e87a25 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -268,6 +268,7 @@ extern int XLogFileOpen(XLogSegNo segno);
extern void CheckXLogRemoved(XLogSegNo segno, TimeLineID tli);
extern XLogSegNo XLogGetLastRemovedSegno(void);
+extern XLogSegNo FindOldestXLogFileSegNo(void);
extern void XLogSetAsyncXactLSN(XLogRecPtr record);
extern void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn);
@@ -304,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(XLogRecPtr restart_lsn, pid_t walsender_pid,
+ int64 *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 ac8f64b219..3887eb3ce0 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -9873,9 +9873,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 80a07825b9..6fc5251536 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.23.0
----Next_Part(Tue_Dec_24_21_26_14_2019_056)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
filename="v17-0003-Add-primary_slot_name-to-init_from_backup-in-TAP.patch"
view thread (15+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected]
Subject: Re: [PATCH v17 2/6] Add monitoring aid for max_slot_wal_keep_size
In-Reply-To: <no-message-id-203678@localhost>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox