From 9e688023aa968cefea5bd5e1d7f0b763b52ee22a Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy Date: Tue, 19 Mar 2024 17:20:45 +0000 Subject: [PATCH v12 2/7] Track last_inactive_at for replication slots Up until now, postgres doesn't track metric that's the time at which the slot became inactive. This commit adds new metric last_inactive_at of type timestamptz to ReplicationSlotPersistentData. Whenever a slot becomes inactive, the current timestamp is persisted to disk. This metric is useful in the following ways: - To improve replication slot monitoring tools. For instance, one can build a monitoring tool that signals a) when replication slots is lying inactive for a day or so using last_inactive_at metric - To implement timeout-based inactive replication slot management capability in postgres. Increases SLOT_VERSION due to the added new metric. --- doc/src/sgml/system-views.sgml | 11 +++++++++++ src/backend/catalog/system_views.sql | 3 ++- src/backend/replication/slot.c | 25 ++++++++++++++++++++++++- src/backend/replication/slotfuncs.c | 7 ++++++- src/include/catalog/pg_proc.dat | 6 +++--- src/include/replication/slot.h | 3 +++ src/test/regress/expected/rules.out | 5 +++-- 7 files changed, 52 insertions(+), 8 deletions(-) diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml index e685921847..ab43032d74 100644 --- a/doc/src/sgml/system-views.sgml +++ b/doc/src/sgml/system-views.sgml @@ -2750,6 +2750,17 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx ID of role + + + + last_inactive_at timestamptz + + + The time at which the slot became inactive. + NULL if the slot is currently actively being + used. + + diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index cd22dad959..2fa4272006 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1026,7 +1026,8 @@ CREATE VIEW pg_replication_slots AS L.conflicting, L.failover, L.synced, - L.invalidation_reason + L.invalidation_reason, + L.last_inactive_at FROM pg_get_replication_slots() AS L LEFT JOIN pg_database D ON (L.datoid = D.oid); diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 4f1a17f6ce..fa40c5f4f1 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -129,7 +129,7 @@ StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1), sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize #define SLOT_MAGIC 0x1051CA1 /* format identifier */ -#define SLOT_VERSION 5 /* version for new files */ +#define SLOT_VERSION 6 /* version for new files */ /* Control array for replication slot management */ ReplicationSlotCtlData *ReplicationSlotCtl = NULL; @@ -398,6 +398,7 @@ ReplicationSlotCreate(const char *name, bool db_specific, slot->data.two_phase_at = InvalidXLogRecPtr; slot->data.failover = failover; slot->data.synced = synced; + slot->data.last_inactive_at = 0; /* and then data only present in shared memory */ slot->just_dirtied = false; @@ -622,6 +623,17 @@ retry: if (SlotIsLogical(s)) pgstat_acquire_replslot(s); + if (s->data.persistency == RS_PERSISTENT) + { + SpinLockAcquire(&s->mutex); + s->data.last_inactive_at = 0; + SpinLockRelease(&s->mutex); + + /* Write this slot to disk */ + ReplicationSlotMarkDirty(); + ReplicationSlotSave(); + } + if (am_walsender) { ereport(log_replication_commands ? LOG : DEBUG1, @@ -691,6 +703,17 @@ ReplicationSlotRelease(void) ConditionVariableBroadcast(&slot->active_cv); } + if (slot->data.persistency == RS_PERSISTENT) + { + SpinLockAcquire(&slot->mutex); + slot->data.last_inactive_at = GetCurrentTimestamp(); + SpinLockRelease(&slot->mutex); + + /* Write this slot to disk */ + ReplicationSlotMarkDirty(); + ReplicationSlotSave(); + } + MyReplicationSlot = NULL; /* might not have been set when we've been a plain slot */ diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c index b5a638edea..95802bf2c9 100644 --- a/src/backend/replication/slotfuncs.c +++ b/src/backend/replication/slotfuncs.c @@ -239,7 +239,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS) Datum pg_get_replication_slots(PG_FUNCTION_ARGS) { -#define PG_GET_REPLICATION_SLOTS_COLS 18 +#define PG_GET_REPLICATION_SLOTS_COLS 19 ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; XLogRecPtr currlsn; int slotno; @@ -436,6 +436,11 @@ pg_get_replication_slots(PG_FUNCTION_ARGS) else values[i++] = CStringGetTextDatum(SlotInvalidationCauses[cause]); + if (slot_contents.data.last_inactive_at > 0) + values[i++] = TimestampTzGetDatum(slot_contents.data.last_inactive_at); + else + nulls[i++] = true; + Assert(i == PG_GET_REPLICATION_SLOTS_COLS); tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 1689009d4f..318bd2abc8 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -11130,9 +11130,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,text,int8,bool,bool,bool,bool,text}', - proargmodes => '{o,o,o,o,o,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,safe_wal_size,two_phase,conflicting,failover,synced,invalidation_reason}', + proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,bool,bool,text,timestamptz}', + proargmodes => '{o,o,o,o,o,o,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,safe_wal_size,two_phase,conflicting,failover,synced,invalidation_reason,last_inactive_at}', 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/include/replication/slot.h b/src/include/replication/slot.h index 7f25a083ee..10f8ba67bc 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -127,6 +127,9 @@ typedef struct ReplicationSlotPersistentData * for logical slots on the primary server. */ bool failover; + + /* When did this slot become inactive last time? */ + TimestampTz last_inactive_at; } ReplicationSlotPersistentData; /* diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 19c44c0cb7..88fbd6a53c 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -1476,8 +1476,9 @@ pg_replication_slots| SELECT l.slot_name, l.conflicting, l.failover, l.synced, - l.invalidation_reason - 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, safe_wal_size, two_phase, conflicting, failover, synced, invalidation_reason) + l.invalidation_reason, + l.last_inactive_at + 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, safe_wal_size, two_phase, conflicting, failover, synced, invalidation_reason, last_inactive_at) LEFT JOIN pg_database d ON ((l.datoid = d.oid))); pg_roles| SELECT pg_authid.rolname, pg_authid.rolsuper, -- 2.34.1