From 7e74a6645eee85035c96547bdc11527208388666 Mon Sep 17 00:00:00 2001 From: vignesh Date: Sat, 10 Apr 2021 08:57:05 +0530 Subject: [PATCH v6 2/5] Added total txns and total txn bytes to replication statistics. This adds the statistics about total transactions count and total transaction data logically replicated to the decoding output plugin from ReorderBuffer. Users can query the pg_stat_replication_slots view to check these stats. --- contrib/test_decoding/expected/stats.out | 79 +++++++++++++------ contrib/test_decoding/sql/stats.sql | 48 +++++++---- doc/src/sgml/monitoring.sgml | 23 ++++++ src/backend/catalog/system_views.sql | 2 + src/backend/postmaster/pgstat.c | 6 ++ src/backend/replication/logical/logical.c | 16 ++-- .../replication/logical/reorderbuffer.c | 12 +++ src/backend/utils/adt/pgstatfuncs.c | 8 +- src/include/catalog/pg_proc.dat | 6 +- src/include/pgstat.h | 4 + src/include/replication/reorderbuffer.h | 4 + src/test/regress/expected/rules.out | 4 +- 12 files changed, 163 insertions(+), 49 deletions(-) diff --git a/contrib/test_decoding/expected/stats.out b/contrib/test_decoding/expected/stats.out index bca36fa903..bc8e601eab 100644 --- a/contrib/test_decoding/expected/stats.out +++ b/contrib/test_decoding/expected/stats.out @@ -8,7 +8,7 @@ SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_d CREATE TABLE stats_test(data text); -- function to wait for counters to advance -CREATE FUNCTION wait_for_decode_stats(check_reset bool) RETURNS void AS $$ +CREATE FUNCTION wait_for_decode_stats(check_reset bool, check_spill_txns bool) RETURNS void AS $$ DECLARE start_time timestamptz := clock_timestamp(); updated bool; @@ -16,12 +16,25 @@ BEGIN -- we don't want to wait forever; loop will exit after 30 seconds FOR i IN 1 .. 300 LOOP - -- check to see if all updates have been reset/updated - SELECT CASE WHEN check_reset THEN (spill_txns = 0) - ELSE (spill_txns > 0) - END - INTO updated - FROM pg_stat_replication_slots WHERE slot_name='regression_slot'; + IF check_spill_txns THEN + + -- check to see if all updates have been reset/updated + SELECT CASE WHEN check_reset THEN (spill_txns = 0) + ELSE (spill_txns > 0) + END + INTO updated + FROM pg_stat_replication_slots WHERE slot_name='regression_slot'; + + ELSE + + -- check to see if all updates have been reset/updated + SELECT CASE WHEN check_reset THEN (total_txns = 0) + ELSE (total_txns > 0) + END + INTO updated + FROM pg_stat_replication_slots WHERE slot_name='regression_slot'; + + END IF; exit WHEN updated; @@ -51,16 +64,16 @@ SELECT count(*) FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, -- Check stats, wait for the stats collector to update. We can't test the -- exact stats count as that can vary if any background transaction (say by -- autovacuum) happens in parallel to the main transaction. -SELECT wait_for_decode_stats(false); +SELECT wait_for_decode_stats(false, true); wait_for_decode_stats ----------------------- (1 row) -SELECT slot_name, spill_txns > 0 AS spill_txns, spill_count > 0 AS spill_count FROM pg_stat_replication_slots; - slot_name | spill_txns | spill_count ------------------+------------+------------- - regression_slot | t | t +SELECT slot_name, spill_txns > 0 AS spill_txns, spill_count > 0 AS spill_count, total_txns > 0 AS total_txns, total_bytes > 0 AS total_bytes FROM pg_stat_replication_slots; + slot_name | spill_txns | spill_count | total_txns | total_bytes +-----------------+------------+-------------+------------+------------- + regression_slot | t | t | t | t (1 row) -- reset the slot stats, and wait for stats collector to reset @@ -70,16 +83,16 @@ SELECT pg_stat_reset_replication_slot('regression_slot'); (1 row) -SELECT wait_for_decode_stats(true); +SELECT wait_for_decode_stats(true, true); wait_for_decode_stats ----------------------- (1 row) -SELECT slot_name, spill_txns, spill_count FROM pg_stat_replication_slots; - slot_name | spill_txns | spill_count ------------------+------------+------------- - regression_slot | 0 | 0 +SELECT slot_name, spill_txns, spill_count, total_txns, total_bytes FROM pg_stat_replication_slots; + slot_name | spill_txns | spill_count | total_txns | total_bytes +-----------------+------------+-------------+------------+------------- + regression_slot | 0 | 0 | 0 | 0 (1 row) -- decode and check stats again. @@ -89,16 +102,36 @@ SELECT count(*) FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 5002 (1 row) -SELECT wait_for_decode_stats(false); +SELECT wait_for_decode_stats(false, true); + wait_for_decode_stats +----------------------- + +(1 row) + +SELECT slot_name, spill_txns > 0 AS spill_txns, spill_count > 0 AS spill_count, total_txns > 0 AS total_txns, total_bytes > 0 AS total_bytes FROM pg_stat_replication_slots; + slot_name | spill_txns | spill_count | total_txns | total_bytes +-----------------+------------+-------------+------------+------------- + regression_slot | t | t | t | t +(1 row) + +SELECT pg_stat_reset_replication_slot('regression_slot'); + pg_stat_reset_replication_slot +-------------------------------- + +(1 row) + +-- non-spilled xact +INSERT INTO stats_test values(generate_series(1, 10)); +SELECT wait_for_decode_stats(false, false); wait_for_decode_stats ----------------------- (1 row) -SELECT slot_name, spill_txns > 0 AS spill_txns, spill_count > 0 AS spill_count FROM pg_stat_replication_slots; - slot_name | spill_txns | spill_count ------------------+------------+------------- - regression_slot | t | t +SELECT slot_name, spill_txns = 0 AS spill_txns, spill_count = 0 AS spill_count, total_txns > 0 AS total_txns, total_bytes > 0 AS total_bytes FROM pg_stat_replication_slots; + slot_name | spill_txns | spill_count | total_txns | total_bytes +-----------------+------------+-------------+------------+------------- + regression_slot | f | f | t | t (1 row) -- Ensure stats can be repeatedly accessed using the same stats snapshot. See @@ -117,7 +150,7 @@ SELECT slot_name FROM pg_stat_replication_slots; (1 row) COMMIT; -DROP FUNCTION wait_for_decode_stats(bool); +DROP FUNCTION wait_for_decode_stats(bool, bool); DROP TABLE stats_test; SELECT pg_drop_replication_slot('regression_slot'); pg_drop_replication_slot diff --git a/contrib/test_decoding/sql/stats.sql b/contrib/test_decoding/sql/stats.sql index 51294e48e8..8c34aeced1 100644 --- a/contrib/test_decoding/sql/stats.sql +++ b/contrib/test_decoding/sql/stats.sql @@ -6,7 +6,7 @@ SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_d CREATE TABLE stats_test(data text); -- function to wait for counters to advance -CREATE FUNCTION wait_for_decode_stats(check_reset bool) RETURNS void AS $$ +CREATE FUNCTION wait_for_decode_stats(check_reset bool, check_spill_txns bool) RETURNS void AS $$ DECLARE start_time timestamptz := clock_timestamp(); updated bool; @@ -14,12 +14,25 @@ BEGIN -- we don't want to wait forever; loop will exit after 30 seconds FOR i IN 1 .. 300 LOOP - -- check to see if all updates have been reset/updated - SELECT CASE WHEN check_reset THEN (spill_txns = 0) - ELSE (spill_txns > 0) - END - INTO updated - FROM pg_stat_replication_slots WHERE slot_name='regression_slot'; + IF check_spill_txns THEN + + -- check to see if all updates have been reset/updated + SELECT CASE WHEN check_reset THEN (spill_txns = 0) + ELSE (spill_txns > 0) + END + INTO updated + FROM pg_stat_replication_slots WHERE slot_name='regression_slot'; + + ELSE + + -- check to see if all updates have been reset/updated + SELECT CASE WHEN check_reset THEN (total_txns = 0) + ELSE (total_txns > 0) + END + INTO updated + FROM pg_stat_replication_slots WHERE slot_name='regression_slot'; + + END IF; exit WHEN updated; @@ -46,18 +59,25 @@ SELECT count(*) FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, -- Check stats, wait for the stats collector to update. We can't test the -- exact stats count as that can vary if any background transaction (say by -- autovacuum) happens in parallel to the main transaction. -SELECT wait_for_decode_stats(false); -SELECT slot_name, spill_txns > 0 AS spill_txns, spill_count > 0 AS spill_count FROM pg_stat_replication_slots; +SELECT wait_for_decode_stats(false, true); +SELECT slot_name, spill_txns > 0 AS spill_txns, spill_count > 0 AS spill_count, total_txns > 0 AS total_txns, total_bytes > 0 AS total_bytes FROM pg_stat_replication_slots; -- reset the slot stats, and wait for stats collector to reset SELECT pg_stat_reset_replication_slot('regression_slot'); -SELECT wait_for_decode_stats(true); -SELECT slot_name, spill_txns, spill_count FROM pg_stat_replication_slots; +SELECT wait_for_decode_stats(true, true); +SELECT slot_name, spill_txns, spill_count, total_txns, total_bytes FROM pg_stat_replication_slots; -- decode and check stats again. SELECT count(*) FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'skip-empty-xacts', '1'); -SELECT wait_for_decode_stats(false); -SELECT slot_name, spill_txns > 0 AS spill_txns, spill_count > 0 AS spill_count FROM pg_stat_replication_slots; +SELECT wait_for_decode_stats(false, true); +SELECT slot_name, spill_txns > 0 AS spill_txns, spill_count > 0 AS spill_count, total_txns > 0 AS total_txns, total_bytes > 0 AS total_bytes FROM pg_stat_replication_slots; + +SELECT pg_stat_reset_replication_slot('regression_slot'); + +-- non-spilled xact +INSERT INTO stats_test values(generate_series(1, 10)); +SELECT wait_for_decode_stats(false, false); +SELECT slot_name, spill_txns = 0 AS spill_txns, spill_count = 0 AS spill_count, total_txns > 0 AS total_txns, total_bytes > 0 AS total_bytes FROM pg_stat_replication_slots; -- Ensure stats can be repeatedly accessed using the same stats snapshot. See -- https://postgr.es/m/20210317230447.c7uc4g3vbs4wi32i%40alap3.anarazel.de @@ -66,6 +86,6 @@ SELECT slot_name FROM pg_stat_replication_slots; SELECT slot_name FROM pg_stat_replication_slots; COMMIT; -DROP FUNCTION wait_for_decode_stats(bool); +DROP FUNCTION wait_for_decode_stats(bool, bool); DROP TABLE stats_test; SELECT pg_drop_replication_slot('regression_slot'); diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 8287587f61..d5e2012b38 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -2716,6 +2716,29 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i + + + total_txns bigint + + + Number of decoded transactions sent to the decoding output plugin for + this slot. This counter is used to maintain the top level transactions, + so the counter is not incremented for subtransactions. + + + + + + total_bytesbigint + + + Amount of decoded transactions data sent to the decoding output plugin + while decoding the changes from WAL for this slot. This can be used to + gauge the total amount of data sent during logical decoding. + + + + stats_reset timestamp with time zone diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 451db2ee0a..6d78b33590 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -875,6 +875,8 @@ CREATE VIEW pg_stat_replication_slots AS s.stream_txns, s.stream_count, s.stream_bytes, + s.total_txns, + s.total_bytes, s.stats_reset FROM pg_stat_get_replication_slots() AS s; diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c index 666ce95d08..e1ec7d8b7d 100644 --- a/src/backend/postmaster/pgstat.c +++ b/src/backend/postmaster/pgstat.c @@ -1829,6 +1829,8 @@ pgstat_report_replslot(const PgStat_ReplSlotStats *repSlotStat) msg.m_stream_txns = repSlotStat->stream_txns; msg.m_stream_count = repSlotStat->stream_count; msg.m_stream_bytes = repSlotStat->stream_bytes; + msg.m_total_txns = repSlotStat->total_txns; + msg.m_total_bytes = repSlotStat->total_bytes; pgstat_send(&msg, sizeof(PgStat_MsgReplSlot)); } @@ -5568,6 +5570,8 @@ pgstat_recv_replslot(PgStat_MsgReplSlot *msg, int len) replSlotStats[idx].stream_txns += msg->m_stream_txns; replSlotStats[idx].stream_count += msg->m_stream_count; replSlotStats[idx].stream_bytes += msg->m_stream_bytes; + replSlotStats[idx].total_txns += msg->m_total_txns; + replSlotStats[idx].total_bytes += msg->m_total_bytes; } } @@ -5795,6 +5799,8 @@ pgstat_reset_replslot(int i, TimestampTz ts) replSlotStats[i].stream_txns = 0; replSlotStats[i].stream_count = 0; replSlotStats[i].stream_bytes = 0; + replSlotStats[i].total_txns = 0; + replSlotStats[i].total_bytes = 0; replSlotStats[i].stat_reset_timestamp = ts; } diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 68e210ce12..ed9a7c8489 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -1776,20 +1776,22 @@ UpdateDecodingStats(LogicalDecodingContext *ctx) PgStat_ReplSlotStats repSlotStat; /* - * Nothing to do if we haven't spilled or streamed anything since the last - * time the stats has been sent. + * Nothing to do if we don't have any replication stats to be sent. */ - if (rb->spillBytes <= 0 && rb->streamBytes <= 0) + if (rb->spillBytes <= 0 && rb->streamBytes <= 0 && + rb->totalBytes <= 0 && rb->totalTxns <=0) return; - elog(DEBUG2, "UpdateDecodingStats: updating stats %p %lld %lld %lld %lld %lld %lld", + elog(DEBUG2, "UpdateDecodingStats: updating stats %p %lld %lld %lld %lld %lld %lld %lld %lld", rb, (long long) rb->spillTxns, (long long) rb->spillCount, (long long) rb->spillBytes, (long long) rb->streamTxns, (long long) rb->streamCount, - (long long) rb->streamBytes); + (long long) rb->streamBytes, + (long long) rb->totalTxns, + (long long) rb->totalBytes); namestrcpy(&repSlotStat.slotname, NameStr(ctx->slot->data.name)); repSlotStat.spill_txns = rb->spillTxns; @@ -1798,6 +1800,8 @@ UpdateDecodingStats(LogicalDecodingContext *ctx) repSlotStat.stream_txns = rb->streamTxns; repSlotStat.stream_count = rb->streamCount; repSlotStat.stream_bytes = rb->streamBytes; + repSlotStat.total_txns = rb->totalTxns; + repSlotStat.total_bytes = rb->totalBytes; pgstat_report_replslot(&repSlotStat); rb->spillTxns = 0; @@ -1806,4 +1810,6 @@ UpdateDecodingStats(LogicalDecodingContext *ctx) rb->streamTxns = 0; rb->streamCount = 0; rb->streamBytes = 0; + rb->totalTxns = 0; + rb->totalBytes = 0; } diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 52d06285a2..bc251adfda 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -350,6 +350,8 @@ ReorderBufferAllocate(void) buffer->streamTxns = 0; buffer->streamCount = 0; buffer->streamBytes = 0; + buffer->totalTxns = 0; + buffer->totalBytes = 0; buffer->current_restart_decoding_lsn = InvalidXLogRecPtr; @@ -2359,6 +2361,16 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, specinsert = NULL; } + /* + * Update total transaction count and total transaction bytes + * processed. Ensure to not count the streamed transaction multiple + * times. + */ + if (!rbtxn_is_streamed(txn)) + rb->totalTxns++; + + rb->totalBytes += rb->size; + /* clean up the iterator */ ReorderBufferIterTXNFinish(rb, iterstate); iterstate = NULL; diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c index 521ba73614..2680190a40 100644 --- a/src/backend/utils/adt/pgstatfuncs.c +++ b/src/backend/utils/adt/pgstatfuncs.c @@ -2284,7 +2284,7 @@ pg_stat_get_archiver(PG_FUNCTION_ARGS) Datum pg_stat_get_replication_slots(PG_FUNCTION_ARGS) { -#define PG_STAT_GET_REPLICATION_SLOT_COLS 8 +#define PG_STAT_GET_REPLICATION_SLOT_COLS 10 ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; TupleDesc tupdesc; Tuplestorestate *tupstore; @@ -2335,11 +2335,13 @@ pg_stat_get_replication_slots(PG_FUNCTION_ARGS) values[4] = Int64GetDatum(s->stream_txns); values[5] = Int64GetDatum(s->stream_count); values[6] = Int64GetDatum(s->stream_bytes); + values[7] = Int64GetDatum(s->total_txns); + values[8] = Int64GetDatum(s->total_bytes); if (s->stat_reset_timestamp == 0) - nulls[7] = true; + nulls[9] = true; else - values[7] = TimestampTzGetDatum(s->stat_reset_timestamp); + values[9] = TimestampTzGetDatum(s->stat_reset_timestamp); tuplestore_putvalues(tupstore, tupdesc, values, nulls); } diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index f4957653ae..591753fe81 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -5315,9 +5315,9 @@ proname => 'pg_stat_get_replication_slots', prorows => '10', proisstrict => 'f', proretset => 't', provolatile => 's', proparallel => 'r', prorettype => 'record', proargtypes => '', - proallargtypes => '{text,int8,int8,int8,int8,int8,int8,timestamptz}', - proargmodes => '{o,o,o,o,o,o,o,o}', - proargnames => '{slot_name,spill_txns,spill_count,spill_bytes,stream_txns,stream_count,stream_bytes,stats_reset}', + proallargtypes => '{text,int8,int8,int8,int8,int8,int8,int8,int8,timestamptz}', + proargmodes => '{o,o,o,o,o,o,o,o,o,o}', + proargnames => '{slot_name,spill_txns,spill_count,spill_bytes,stream_txns,stream_count,stream_bytes,total_txns,total_bytes,stats_reset}', prosrc => 'pg_stat_get_replication_slots' }, { 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 8e11215058..2aeb3cded4 100644 --- a/src/include/pgstat.h +++ b/src/include/pgstat.h @@ -548,6 +548,8 @@ typedef struct PgStat_MsgReplSlot PgStat_Counter m_stream_txns; PgStat_Counter m_stream_count; PgStat_Counter m_stream_bytes; + PgStat_Counter m_total_txns; + PgStat_Counter m_total_bytes; } PgStat_MsgReplSlot; /* ---------- @@ -924,6 +926,8 @@ typedef struct PgStat_ReplSlotStats PgStat_Counter stream_txns; PgStat_Counter stream_count; PgStat_Counter stream_bytes; + PgStat_Counter total_txns; + PgStat_Counter total_bytes; TimestampTz stat_reset_timestamp; } PgStat_ReplSlotStats; diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index 565a961d6a..a372b70b7d 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -618,6 +618,10 @@ struct ReorderBuffer int64 streamTxns; /* number of transactions streamed */ int64 streamCount; /* streaming invocation counter */ int64 streamBytes; /* amount of data streamed */ + + /* Statistics about all the replicated transactions */ + int64 totalTxns; /* total number of transactions replicated */ + int64 totalBytes; /* total amount of data replicated */ }; diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 186e6c966c..6399f3feef 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -2068,8 +2068,10 @@ pg_stat_replication_slots| SELECT s.slot_name, s.stream_txns, s.stream_count, s.stream_bytes, + s.total_txns, + s.total_bytes, s.stats_reset - FROM pg_stat_get_replication_slots() s(slot_name, spill_txns, spill_count, spill_bytes, stream_txns, stream_count, stream_bytes, stats_reset); + FROM pg_stat_get_replication_slots() s(slot_name, spill_txns, spill_count, spill_bytes, stream_txns, stream_count, stream_bytes, total_txns, total_bytes, stats_reset); pg_stat_slru| SELECT s.name, s.blks_zeroed, s.blks_hit, -- 2.25.1