agora inbox for [email protected]
help / color / mirror / Atom feedFrom: Kyotaro Horiguchi <[email protected]>
Subject: [PATCH 2/2] Add monitoring aid for max_replication_slots.
Date: Thu, 7 Sep 2017 19:13:22 +0900
Adds two columns "live" and "distance" in pg_replication_slot.
Setting max_slot_wal_keep_size, long-disconnected slots may lose sync.
The two columns shows how long a slot can live on or how many bytes a
slot have lost if max_slot_wal_keep_size is set.
---
src/backend/access/transam/xlog.c | 80 ++++++++++++++++++++++++++++++++++++
src/backend/catalog/system_views.sql | 4 +-
src/backend/replication/slotfuncs.c | 16 +++++++-
src/include/access/xlog.h | 1 +
src/include/catalog/pg_proc.h | 2 +-
5 files changed, 100 insertions(+), 3 deletions(-)
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index ae70d7d..c4c8307 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -9324,6 +9324,86 @@ CreateRestartPoint(int flags)
}
/*
+ * Check if the record on the given lsn will be preserved at the next
+ * checkpoint.
+ *
+ * Returns true if it will be preserved. If distance is given, the distance
+ * from origin to the beginning of the first segment kept at the next
+ * checkpoint. It means margin when this function returns true and gap of lost
+ * records when false.
+ *
+ * This function should return the consistent result with KeepLogSeg.
+ */
+bool
+GetMarginToSlotSegmentLimit(XLogRecPtr restartLSN, uint64 *distance)
+{
+ XLogRecPtr currpos;
+ XLogRecPtr tailpos;
+ uint64 currSeg;
+ uint64 restByteInSeg;
+ uint64 restartSeg;
+ uint64 tailSeg;
+ uint64 keepSegs;
+
+ currpos = GetXLogWriteRecPtr();
+
+ LWLockAcquire(ControlFileLock, LW_SHARED);
+ tailpos = ControlFile->checkPointCopy.redo;
+ LWLockRelease(ControlFileLock);
+
+ /* Move the pointer to the beginning of the segment*/
+ XLByteToSeg(currpos, currSeg);
+ XLByteToSeg(restartLSN, restartSeg);
+ XLByteToSeg(tailpos, tailSeg);
+ restByteInSeg = 0;
+
+ Assert(wal_keep_segments >= 0);
+ Assert(max_slot_wal_keep_size_mb >= 0);
+
+ /*
+ * WAL are removed by the unit of segment.
+ */
+ keepSegs = wal_keep_segments + ConvertToXSegs(max_slot_wal_keep_size_mb);
+
+ /*
+ * If the latest checkpoint's redo point is older than the current head
+ * minus keep segments, the next checkpoint keeps the redo point's
+ * segment. Elsewise use current head minus number of segments to keep.
+ */
+ if (currSeg < tailSeg + keepSegs)
+ {
+ if (currSeg < keepSegs)
+ tailSeg = 0;
+ else
+ tailSeg = currSeg - keepSegs;
+
+ /* In this case, the margin will be the bytes to the next segment */
+ restByteInSeg = XLogSegSize - (currpos % XLogSegSize);
+ }
+
+ elog(LOG, "rest %lx, curr %lx, tail %lx", restartSeg, currSeg, tailSeg);
+ /* Required sements will be removed at the next checkpoint */
+ if (restartSeg < tailSeg)
+ {
+ /* Calculate how may bytes the slot have lost */
+ if (distance)
+ {
+ uint64 restbytes = (restartSeg + 1) * XLogSegSize - restartLSN;
+ *distance =
+ (tailSeg - restartSeg - 1) * XLogSegSize
+ + restbytes;
+ }
+ return false;
+ }
+
+ /* Margin at the next checkpoint before the slot lose sync */
+ if (distance)
+ *distance = (restartSeg - tailSeg) * XLogSegSize + restByteInSeg;
+
+ return true;
+}
+
+/*
* Retreat *logSegNo to the last segment that we need to retain because of
* either wal_keep_segments or replication slots.
*
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index dc40cde..c55c88b 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -793,7 +793,9 @@ CREATE VIEW pg_replication_slots AS
L.xmin,
L.catalog_xmin,
L.restart_lsn,
- L.confirmed_flush_lsn
+ L.confirmed_flush_lsn,
+ L.live,
+ L.distance
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 d4cbd83..52a9d26 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -182,7 +182,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;
@@ -304,6 +304,20 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
else
nulls[i++] = true;
+ if (max_slot_wal_keep_size_mb > 0 && restart_lsn != InvalidXLogRecPtr)
+ {
+ uint64 distance;
+
+ values[i++] = BoolGetDatum(GetMarginToSlotSegmentLimit(restart_lsn,
+ &distance));
+ values[i++] = Int64GetDatum(distance);
+ }
+ else
+ {
+ values[i++] = BoolGetDatum(true);
+ 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 f596794..30507eb 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -267,6 +267,7 @@ extern void ShutdownXLOG(int code, Datum arg);
extern void InitXLOGAccess(void);
extern void CreateCheckPoint(int flags);
extern bool CreateRestartPoint(int flags);
+extern bool GetMarginToSlotSegmentLimit(XLogRecPtr restartLSN, uint64 *distance);
extern void XLogPutNextOid(Oid nextOid);
extern XLogRecPtr XLogRestorePoint(const char *rpName);
extern void UpdateFullPageWrites(void);
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 8b33b4e..4bad345 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -5293,7 +5293,7 @@ DATA(insert OID = 3779 ( pg_create_physical_replication_slot PGNSP PGUID 12 1 0
DESCR("create a physical replication slot");
DATA(insert OID = 3780 ( pg_drop_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 1 0 2278 "19" _null_ _null_ _null_ _null_ _null_ pg_drop_replication_slot _null_ _null_ _null_ ));
DESCR("drop a replication slot");
-DATA(insert OID = 3781 ( pg_get_replication_slots PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220}" "{o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
+DATA(insert OID = 3781 ( pg_get_replication_slots PGNSP PGUID 12 1 10 0 0 f f f f f t s s 0 0 2249 "" "{19,19,25,26,16,16,23,28,28,3220,3220,16,3220}" "{o,o,o,o,o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,live,distance}" _null_ _null_ pg_get_replication_slots _null_ _null_ _null_ ));
DESCR("information about replication slots currently in use");
DATA(insert OID = 3786 ( pg_create_logical_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f t f v u 3 0 2249 "19 19 16" "{19,19,16,25,3220}" "{i,i,i,o,o}" "{slot_name,plugin,temporary,slot_name,lsn}" _null_ _null_ pg_create_logical_replication_slot _null_ _null_ _null_ ));
DESCR("set up a logical replication slot");
--
2.9.2
----Next_Part(Thu_Sep_07_21_59_56_2017_119)--
Content-Type: text/plain
Content-Disposition: inline
Content-Transfer-Encoding: 8bit
MIME-Version: 1.0
--
Sent via pgsql-hackers mailing list ([email protected])
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
----Next_Part(Thu_Sep_07_21_59_56_2017_119)----
view thread (13+ 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 2/2] Add monitoring aid for max_replication_slots.
In-Reply-To: <no-message-id-265439@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