diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 999d984..b35b667 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -962,11 +962,17 @@ CREATE VIEW pg_stat_replication_slots AS s.spill_txns, s.spill_count, s.spill_bytes, + s.failed_spill_txns, + s.failed_spill_bytes, s.stream_txns, s.stream_count, s.stream_bytes, + s.failed_stream_txns, + s.failed_stream_bytes, s.total_txns, s.total_bytes, + s.failed_total_txns, + s.failed_total_bytes, s.stats_reset FROM pg_replication_slots as r, LATERAL pg_stat_get_replication_slot(slot_name) as s diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c index ce8888c..de98eb7 100644 --- a/src/backend/postmaster/pgstat.c +++ b/src/backend/postmaster/pgstat.c @@ -1819,11 +1819,17 @@ pgstat_report_replslot(const PgStat_StatReplSlotEntry *repSlotStat) msg.m_spill_txns = repSlotStat->spill_txns; msg.m_spill_count = repSlotStat->spill_count; msg.m_spill_bytes = repSlotStat->spill_bytes; + msg.m_failed_spill_txns = repSlotStat->failed_spill_txns; + msg.m_failed_spill_bytes = repSlotStat->failed_spill_bytes; msg.m_stream_txns = repSlotStat->stream_txns; msg.m_stream_count = repSlotStat->stream_count; msg.m_stream_bytes = repSlotStat->stream_bytes; + msg.m_failed_stream_txns = repSlotStat->failed_stream_txns; + msg.m_failed_stream_bytes = repSlotStat->failed_stream_bytes; msg.m_total_txns = repSlotStat->total_txns; msg.m_total_bytes = repSlotStat->total_bytes; + msg.m_failed_total_txns = repSlotStat->failed_total_txns; + msg.m_failed_total_bytes = repSlotStat->failed_total_bytes; pgstat_send(&msg, sizeof(PgStat_MsgReplSlot)); } @@ -5509,11 +5515,17 @@ pgstat_recv_replslot(PgStat_MsgReplSlot *msg, int len) slotent->spill_txns += msg->m_spill_txns; slotent->spill_count += msg->m_spill_count; slotent->spill_bytes += msg->m_spill_bytes; + slotent->failed_spill_txns += msg->m_failed_spill_txns; + slotent->failed_spill_bytes += msg->m_failed_spill_bytes; slotent->stream_txns += msg->m_stream_txns; slotent->stream_count += msg->m_stream_count; slotent->stream_bytes += msg->m_stream_bytes; + slotent->failed_stream_txns += msg->m_failed_stream_txns; + slotent->failed_stream_bytes += msg->m_failed_stream_bytes; slotent->total_txns += msg->m_total_txns; slotent->total_bytes += msg->m_total_bytes; + slotent->failed_total_txns += msg->m_failed_total_txns; + slotent->failed_total_bytes += msg->m_failed_total_bytes; } } } @@ -5760,11 +5772,17 @@ pgstat_reset_replslot(PgStat_StatReplSlotEntry *slotent, TimestampTz ts) slotent->spill_txns = 0; slotent->spill_count = 0; slotent->spill_bytes = 0; + slotent->failed_spill_txns = 0; + slotent->failed_spill_bytes = 0; slotent->stream_txns = 0; slotent->stream_count = 0; slotent->stream_bytes = 0; + slotent->failed_stream_txns = 0; + slotent->failed_stream_bytes = 0; slotent->total_txns = 0; slotent->total_bytes = 0; + slotent->failed_total_txns = 0; + slotent->failed_total_bytes = 0; slotent->stat_reset_timestamp = ts; } diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index d536a5f..159223f 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -196,6 +196,12 @@ StartupDecodingContext(List *output_plugin_options, LWLockRelease(ProcArrayLock); } + slot->failedSpillTxns = 0; + slot->failedSpillBytes = 0; + slot->failedStreamTxns = 0; + slot->failedStreamBytes = 0; + slot->failedTotalTxns = 0; + slot->failedTotalBytes = 0; ctx->slot = slot; ctx->reader = XLogReaderAllocate(wal_segment_size, NULL, xl_routine, ctx); @@ -1794,11 +1800,17 @@ UpdateDecodingStats(LogicalDecodingContext *ctx) repSlotStat.spill_txns = rb->spillTxns; repSlotStat.spill_count = rb->spillCount; repSlotStat.spill_bytes = rb->spillBytes; + repSlotStat.failed_spill_txns = 0; + repSlotStat.failed_spill_bytes = 0; repSlotStat.stream_txns = rb->streamTxns; repSlotStat.stream_count = rb->streamCount; repSlotStat.stream_bytes = rb->streamBytes; + repSlotStat.failed_stream_txns = 0; + repSlotStat.failed_stream_bytes = 0; repSlotStat.total_txns = rb->totalTxns; repSlotStat.total_bytes = rb->totalBytes; + repSlotStat.failed_total_txns = 0; + repSlotStat.failed_total_bytes = 0; pgstat_report_replslot(&repSlotStat); diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index b8c5e2a..ca35c42 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -2051,6 +2051,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, PG_TRY(); { ReorderBufferChange *change; + ReplicationSlot *slot = MyReplicationSlot; if (using_subtxn) BeginInternalSubTransaction(streaming ? "stream" : "replay"); @@ -2409,9 +2410,13 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, * which we have already accounted in ReorderBufferIterTXNNext. */ if (!rbtxn_is_streamed(txn)) + { rb->totalTxns++; + slot->failedTotalTxns++; + } rb->totalBytes += txn->total_size; + slot->failedTotalBytes += txn->total_size; /* * Done with current changes, send the last message for this set of @@ -3582,11 +3587,17 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn) /* update the statistics iff we have spilled anything */ if (spilled) { + ReplicationSlot *slot = MyReplicationSlot; + int spill_txn; + rb->spillCount += 1; rb->spillBytes += size; + slot->failedSpillBytes += size; /* don't consider already serialized transactions */ - rb->spillTxns += (rbtxn_is_serialized(txn) || rbtxn_is_serialized_clear(txn)) ? 0 : 1; + spill_txn = (rbtxn_is_serialized(txn) || rbtxn_is_serialized_clear(txn)) ? 0 : 1; + rb->spillTxns += spill_txn; + slot->failedSpillTxns += spill_txn; /* update the decoding stats */ UpdateDecodingStats((LogicalDecodingContext *) rb->private_data); @@ -3861,6 +3872,7 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn) CommandId command_id; Size stream_bytes; bool txn_is_streamed; + ReplicationSlot *slot = MyReplicationSlot; /* We can never reach here for a subtransaction. */ Assert(txn->toptxn == NULL); @@ -3956,9 +3968,11 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn) rb->streamCount += 1; rb->streamBytes += stream_bytes; + slot->failedStreamBytes += stream_bytes; /* Don't consider already streamed transaction. */ rb->streamTxns += (txn_is_streamed) ? 0 : 1; + slot->failedStreamTxns += (txn_is_streamed) ? 0 : 1; /* update the decoding stats */ UpdateDecodingStats((LogicalDecodingContext *) rb->private_data); diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 8c18b4e..ac5e953 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -44,6 +44,7 @@ #include "common/string.h" #include "miscadmin.h" #include "pgstat.h" +#include "replication/logical.h" #include "replication/slot.h" #include "storage/fd.h" #include "storage/proc.h" @@ -471,6 +472,11 @@ ReplicationSlotRelease(void) Assert(slot != NULL && slot->active_pid != 0); + if (SlotIsLogical(slot)) + { + UpdateDecodingFailedStats(); + } + if (slot->data.persistency == RS_EPHEMERAL) { /* @@ -1839,3 +1845,40 @@ RestoreSlotFromDisk(const char *name) (errmsg("too many replication slots active before shutdown"), errhint("Increase max_replication_slots and try again."))); } + +/* + * Report failed stats for a slot. + */ +void +UpdateDecodingFailedStats() +{ + PgStat_StatReplSlotEntry repSlotStat; + ReplicationSlot *slot = MyReplicationSlot; + + Assert(MyReplicationSlot != NULL); + + namestrcpy(&repSlotStat.slotname, NameStr(slot->data.name)); + repSlotStat.spill_txns = 0; + repSlotStat.spill_count = 0; + repSlotStat.spill_bytes = 0; + repSlotStat.failed_spill_txns = slot->failedSpillTxns; + repSlotStat.failed_spill_bytes = slot->failedSpillBytes; + repSlotStat.stream_txns = 0; + repSlotStat.stream_count = 0; + repSlotStat.stream_bytes = 0; + repSlotStat.failed_stream_txns = slot->failedStreamTxns; + repSlotStat.failed_stream_bytes = slot->failedStreamBytes; + repSlotStat.total_txns = 0; + repSlotStat.total_bytes = 0; + repSlotStat.failed_total_txns = slot->failedTotalTxns; + repSlotStat.failed_total_bytes = slot->failedTotalBytes; + + pgstat_report_replslot(&repSlotStat); + + slot->failedSpillTxns = 0; + slot->failedSpillBytes = 0; + slot->failedStreamTxns = 0; + slot->failedStreamBytes = 0; + slot->failedTotalTxns = 0; + slot->failedTotalBytes = 0; +} diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c index f0e09ea..f518964 100644 --- a/src/backend/utils/adt/pgstatfuncs.c +++ b/src/backend/utils/adt/pgstatfuncs.c @@ -2313,7 +2313,7 @@ pg_stat_get_archiver(PG_FUNCTION_ARGS) Datum pg_stat_get_replication_slot(PG_FUNCTION_ARGS) { -#define PG_STAT_GET_REPLICATION_SLOT_COLS 10 +#define PG_STAT_GET_REPLICATION_SLOT_COLS 16 text *slotname_text = PG_GETARG_TEXT_P(0); NameData slotname; TupleDesc tupdesc; @@ -2336,17 +2336,29 @@ pg_stat_get_replication_slot(PG_FUNCTION_ARGS) INT8OID, -1, 0); TupleDescInitEntry(tupdesc, (AttrNumber) 4, "spill_bytes", INT8OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber) 5, "stream_txns", + TupleDescInitEntry(tupdesc, (AttrNumber) 5, "failed_spill_txns", INT8OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber) 6, "stream_count", + TupleDescInitEntry(tupdesc, (AttrNumber) 6, "failed_spill_bytes", INT8OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber) 7, "stream_bytes", + TupleDescInitEntry(tupdesc, (AttrNumber) 7, "stream_txns", INT8OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber) 8, "total_txns", + TupleDescInitEntry(tupdesc, (AttrNumber) 8, "stream_count", INT8OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber) 9, "total_bytes", + TupleDescInitEntry(tupdesc, (AttrNumber) 9, "stream_bytes", INT8OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber) 10, "stats_reset", + TupleDescInitEntry(tupdesc, (AttrNumber) 10, "failed_stream_txns", + INT8OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber) 11, "failed_stream_bytes", + INT8OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber) 12, "total_txns", + INT8OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber) 13, "total_bytes", + INT8OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber) 14, "failed_total_txns", + INT8OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber) 15, "failed_total_bytes", + INT8OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber) 16, "stats_reset", TIMESTAMPTZOID, -1, 0); BlessTupleDesc(tupdesc); @@ -2366,16 +2378,22 @@ pg_stat_get_replication_slot(PG_FUNCTION_ARGS) values[1] = Int64GetDatum(slotent->spill_txns); values[2] = Int64GetDatum(slotent->spill_count); values[3] = Int64GetDatum(slotent->spill_bytes); - values[4] = Int64GetDatum(slotent->stream_txns); - values[5] = Int64GetDatum(slotent->stream_count); - values[6] = Int64GetDatum(slotent->stream_bytes); - values[7] = Int64GetDatum(slotent->total_txns); - values[8] = Int64GetDatum(slotent->total_bytes); + values[4] = Int64GetDatum(slotent->failed_spill_txns); + values[5] = Int64GetDatum(slotent->failed_spill_bytes); + values[6] = Int64GetDatum(slotent->stream_txns); + values[7] = Int64GetDatum(slotent->stream_count); + values[8] = Int64GetDatum(slotent->stream_bytes); + values[9] = Int64GetDatum(slotent->failed_stream_txns); + values[10] = Int64GetDatum(slotent->failed_stream_bytes); + values[11] = Int64GetDatum(slotent->total_txns); + values[12] = Int64GetDatum(slotent->total_bytes); + values[13] = Int64GetDatum(slotent->failed_total_txns); + values[14] = Int64GetDatum(slotent->failed_total_bytes); if (slotent->stat_reset_timestamp == 0) - nulls[9] = true; + nulls[15] = true; else - values[9] = TimestampTzGetDatum(slotent->stat_reset_timestamp); + values[15] = TimestampTzGetDatum(slotent->stat_reset_timestamp); /* Returns the record as Datum */ PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls))); diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index fde251f..11cf01b 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -5313,9 +5313,9 @@ proname => 'pg_stat_get_replication_slot', prorows => '1', proisstrict => 'f', proretset => 't', provolatile => 's', proparallel => 'r', prorettype => 'record', proargtypes => 'text', - proallargtypes => '{text,text,int8,int8,int8,int8,int8,int8,int8,int8,timestamptz}', - proargmodes => '{i,o,o,o,o,o,o,o,o,o,o}', - proargnames => '{slot_name,slot_name,spill_txns,spill_count,spill_bytes,stream_txns,stream_count,stream_bytes,total_txns,total_bytes,stats_reset}', + proallargtypes => '{text,text,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,timestamptz}', + proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}', + proargnames => '{slot_name,slot_name,spill_txns,spill_count,spill_bytes,failed_spill_txns,failed_spill_bytes,stream_txns,stream_count,stream_bytes,failed_stream_txns,failed_stream_bytes,total_txns,total_bytes,failed_total_txns,failed_total_bytes,stats_reset}', prosrc => 'pg_stat_get_replication_slot' }, { oid => '6118', descr => 'statistics: information about subscription', proname => 'pg_stat_get_subscription', prorows => '10', proisstrict => 'f', diff --git a/src/include/pgstat.h b/src/include/pgstat.h index 9612c0a..5064408 100644 --- a/src/include/pgstat.h +++ b/src/include/pgstat.h @@ -532,11 +532,17 @@ typedef struct PgStat_MsgReplSlot PgStat_Counter m_spill_txns; PgStat_Counter m_spill_count; PgStat_Counter m_spill_bytes; + PgStat_Counter m_failed_spill_txns; + PgStat_Counter m_failed_spill_bytes; PgStat_Counter m_stream_txns; PgStat_Counter m_stream_count; PgStat_Counter m_stream_bytes; + PgStat_Counter m_failed_stream_txns; + PgStat_Counter m_failed_stream_bytes; PgStat_Counter m_total_txns; PgStat_Counter m_total_bytes; + PgStat_Counter m_failed_total_txns; + PgStat_Counter m_failed_total_bytes; } PgStat_MsgReplSlot; @@ -900,11 +906,17 @@ typedef struct PgStat_StatReplSlotEntry PgStat_Counter spill_txns; PgStat_Counter spill_count; PgStat_Counter spill_bytes; + PgStat_Counter failed_spill_txns; + PgStat_Counter failed_spill_bytes; PgStat_Counter stream_txns; PgStat_Counter stream_count; PgStat_Counter stream_bytes; + PgStat_Counter failed_stream_txns; + PgStat_Counter failed_stream_bytes; PgStat_Counter total_txns; PgStat_Counter total_bytes; + PgStat_Counter failed_total_txns; + PgStat_Counter failed_total_bytes; TimestampTz stat_reset_timestamp; } PgStat_StatReplSlotEntry; diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 2eb7e3a..8d0bef6 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -164,6 +164,20 @@ typedef struct ReplicationSlot XLogRecPtr candidate_xmin_lsn; XLogRecPtr candidate_restart_valid; XLogRecPtr candidate_restart_lsn; + + /* + * Statistics about failed transactions. + * + * These failed statistics works to make a distinction between + * successful and unsuccessful transactions, by utilizing other + * corresponding stats of transactions respectively. + */ + int64 failedSpillTxns; + int64 failedSpillBytes; + int64 failedStreamTxns; + int64 failedStreamBytes; + int64 failedTotalTxns; + int64 failedTotalBytes; } ReplicationSlot; #define SlotIsPhysical(slot) ((slot)->data.database == InvalidOid) @@ -223,5 +237,6 @@ extern void StartupReplicationSlots(void); extern void CheckPointReplicationSlots(void); extern void CheckSlotRequirements(void); +extern void UpdateDecodingFailedStats(void); #endif /* SLOT_H */ diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index e5ab112..41724bb 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -2054,14 +2054,20 @@ pg_stat_replication_slots| SELECT s.slot_name, s.spill_txns, s.spill_count, s.spill_bytes, + s.failed_spill_txns, + s.failed_spill_bytes, s.stream_txns, s.stream_count, s.stream_bytes, + s.failed_stream_txns, + s.failed_stream_bytes, s.total_txns, s.total_bytes, + s.failed_total_txns, + s.failed_total_bytes, s.stats_reset FROM pg_replication_slots r, - LATERAL pg_stat_get_replication_slot((r.slot_name)::text) s(slot_name, spill_txns, spill_count, spill_bytes, stream_txns, stream_count, stream_bytes, total_txns, total_bytes, stats_reset) + LATERAL pg_stat_get_replication_slot((r.slot_name)::text) s(slot_name, spill_txns, spill_count, spill_bytes, failed_spill_txns, failed_spill_bytes, stream_txns, stream_count, stream_bytes, failed_stream_txns, failed_stream_bytes, total_txns, total_bytes, failed_total_txns, failed_total_bytes, stats_reset) WHERE (r.datoid IS NOT NULL); pg_stat_slru| SELECT s.name, s.blks_zeroed,