Line data Source code
1 : /* -------------------------------------------------------------------------
2 : *
3 : * pgstat_replslot.c
4 : * Implementation of replication slot statistics.
5 : *
6 : * This file contains the implementation of replication slot statistics. It is kept
7 : * separate from pgstat.c to enforce the line between the statistics access /
8 : * storage implementation and the details about individual types of
9 : * statistics.
10 : *
11 : * Replication slot stats work a bit different than other variable-numbered
12 : * stats. Slots do not have oids (so they can be created on physical
13 : * replicas). Use the slot index as object id while running. However, the slot
14 : * index can change when restarting. That is addressed by using the name when
15 : * (de-)serializing. After a restart it is possible for slots to have been
16 : * dropped while shut down, which is addressed by not restoring stats for
17 : * slots that cannot be found by name when starting up.
18 : *
19 : * Copyright (c) 2001-2024, PostgreSQL Global Development Group
20 : *
21 : * IDENTIFICATION
22 : * src/backend/utils/activity/pgstat_replslot.c
23 : * -------------------------------------------------------------------------
24 : */
25 :
26 : #include "postgres.h"
27 :
28 : #include "replication/slot.h"
29 : #include "utils/pgstat_internal.h"
30 :
31 :
32 : static int get_replslot_index(const char *name, bool need_lock);
33 :
34 :
35 : /*
36 : * Reset counters for a single replication slot.
37 : *
38 : * Permission checking for this function is managed through the normal
39 : * GRANT system.
40 : */
41 : void
42 8 : pgstat_reset_replslot(const char *name)
43 : {
44 : ReplicationSlot *slot;
45 :
46 : Assert(name != NULL);
47 :
48 8 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
49 :
50 : /* Check if the slot exits with the given name. */
51 8 : slot = SearchNamedReplicationSlot(name, false);
52 :
53 8 : if (!slot)
54 2 : ereport(ERROR,
55 : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
56 : errmsg("replication slot \"%s\" does not exist",
57 : name)));
58 :
59 : /*
60 : * Reset stats if it is a logical slot. Nothing to do for physical slots
61 : * as we collect stats only for logical slots.
62 : */
63 6 : if (SlotIsLogical(slot))
64 6 : pgstat_reset(PGSTAT_KIND_REPLSLOT, InvalidOid,
65 6 : ReplicationSlotIndex(slot));
66 :
67 6 : LWLockRelease(ReplicationSlotControlLock);
68 6 : }
69 :
70 : /*
71 : * Report replication slot statistics.
72 : *
73 : * We can rely on the stats for the slot to exist and to belong to this
74 : * slot. We can only get here if pgstat_create_replslot() or
75 : * pgstat_acquire_replslot() have already been called.
76 : */
77 : void
78 11482 : pgstat_report_replslot(ReplicationSlot *slot, const PgStat_StatReplSlotEntry *repSlotStat)
79 : {
80 : PgStat_EntryRef *entry_ref;
81 : PgStatShared_ReplSlot *shstatent;
82 : PgStat_StatReplSlotEntry *statent;
83 :
84 11482 : entry_ref = pgstat_get_entry_ref_locked(PGSTAT_KIND_REPLSLOT, InvalidOid,
85 11482 : ReplicationSlotIndex(slot), false);
86 11482 : shstatent = (PgStatShared_ReplSlot *) entry_ref->shared_stats;
87 11482 : statent = &shstatent->stats;
88 :
89 : /* Update the replication slot statistics */
90 : #define REPLSLOT_ACC(fld) statent->fld += repSlotStat->fld
91 11482 : REPLSLOT_ACC(spill_txns);
92 11482 : REPLSLOT_ACC(spill_count);
93 11482 : REPLSLOT_ACC(spill_bytes);
94 11482 : REPLSLOT_ACC(stream_txns);
95 11482 : REPLSLOT_ACC(stream_count);
96 11482 : REPLSLOT_ACC(stream_bytes);
97 11482 : REPLSLOT_ACC(total_txns);
98 11482 : REPLSLOT_ACC(total_bytes);
99 : #undef REPLSLOT_ACC
100 :
101 11482 : pgstat_unlock_entry(entry_ref);
102 11482 : }
103 :
104 : /*
105 : * Report replication slot creation.
106 : *
107 : * NB: This gets called with ReplicationSlotAllocationLock already held, be
108 : * careful about calling back into slot.c.
109 : */
110 : void
111 828 : pgstat_create_replslot(ReplicationSlot *slot)
112 : {
113 : PgStat_EntryRef *entry_ref;
114 : PgStatShared_ReplSlot *shstatent;
115 :
116 : Assert(LWLockHeldByMeInMode(ReplicationSlotAllocationLock, LW_EXCLUSIVE));
117 :
118 828 : entry_ref = pgstat_get_entry_ref_locked(PGSTAT_KIND_REPLSLOT, InvalidOid,
119 828 : ReplicationSlotIndex(slot), false);
120 828 : shstatent = (PgStatShared_ReplSlot *) entry_ref->shared_stats;
121 :
122 : /*
123 : * NB: need to accept that there might be stats from an older slot, e.g.
124 : * if we previously crashed after dropping a slot.
125 : */
126 828 : memset(&shstatent->stats, 0, sizeof(shstatent->stats));
127 :
128 828 : pgstat_unlock_entry(entry_ref);
129 828 : }
130 :
131 : /*
132 : * Report replication slot has been acquired.
133 : *
134 : * This guarantees that a stats entry exists during later
135 : * pgstat_report_replslot() calls.
136 : *
137 : * If we previously crashed, no stats data exists. But if we did not crash,
138 : * the stats do belong to this slot:
139 : * - the stats cannot belong to a dropped slot, pgstat_drop_replslot() would
140 : * have been called
141 : * - if the slot was removed while shut down,
142 : * pgstat_replslot_from_serialized_name_cb() returning false would have
143 : * caused the stats to be dropped
144 : */
145 : void
146 1856 : pgstat_acquire_replslot(ReplicationSlot *slot)
147 : {
148 1856 : pgstat_get_entry_ref(PGSTAT_KIND_REPLSLOT, InvalidOid,
149 1856 : ReplicationSlotIndex(slot), true, NULL);
150 1856 : }
151 :
152 : /*
153 : * Report replication slot drop.
154 : */
155 : void
156 706 : pgstat_drop_replslot(ReplicationSlot *slot)
157 : {
158 : Assert(LWLockHeldByMeInMode(ReplicationSlotAllocationLock, LW_EXCLUSIVE));
159 :
160 706 : if (!pgstat_drop_entry(PGSTAT_KIND_REPLSLOT, InvalidOid,
161 706 : ReplicationSlotIndex(slot)))
162 86 : pgstat_request_entry_refs_gc();
163 706 : }
164 :
165 : /*
166 : * Support function for the SQL-callable pgstat* functions. Returns
167 : * a pointer to the replication slot statistics struct.
168 : */
169 : PgStat_StatReplSlotEntry *
170 94 : pgstat_fetch_replslot(NameData slotname)
171 : {
172 : int idx;
173 94 : PgStat_StatReplSlotEntry *slotentry = NULL;
174 :
175 94 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
176 :
177 94 : idx = get_replslot_index(NameStr(slotname), false);
178 :
179 94 : if (idx != -1)
180 90 : slotentry = (PgStat_StatReplSlotEntry *) pgstat_fetch_entry(PGSTAT_KIND_REPLSLOT,
181 : InvalidOid, idx);
182 :
183 94 : LWLockRelease(ReplicationSlotControlLock);
184 :
185 94 : return slotentry;
186 : }
187 :
188 : void
189 150 : pgstat_replslot_to_serialized_name_cb(const PgStat_HashKey *key, const PgStatShared_Common *header, NameData *name)
190 : {
191 : /*
192 : * This is only called late during shutdown. The set of existing slots
193 : * isn't allowed to change at this point, we can assume that a slot exists
194 : * at the offset.
195 : */
196 150 : if (!ReplicationSlotName(key->objid, name))
197 0 : elog(ERROR, "could not find name for replication slot index %llu",
198 : (unsigned long long) key->objid);
199 150 : }
200 :
201 : bool
202 92 : pgstat_replslot_from_serialized_name_cb(const NameData *name, PgStat_HashKey *key)
203 : {
204 92 : int idx = get_replslot_index(NameStr(*name), true);
205 :
206 : /* slot might have been deleted */
207 92 : if (idx == -1)
208 2 : return false;
209 :
210 90 : key->kind = PGSTAT_KIND_REPLSLOT;
211 90 : key->dboid = InvalidOid;
212 90 : key->objid = idx;
213 :
214 90 : return true;
215 : }
216 :
217 : void
218 16 : pgstat_replslot_reset_timestamp_cb(PgStatShared_Common *header, TimestampTz ts)
219 : {
220 16 : ((PgStatShared_ReplSlot *) header)->stats.stat_reset_timestamp = ts;
221 16 : }
222 :
223 : static int
224 186 : get_replslot_index(const char *name, bool need_lock)
225 : {
226 : ReplicationSlot *slot;
227 :
228 : Assert(name != NULL);
229 :
230 186 : slot = SearchNamedReplicationSlot(name, need_lock);
231 :
232 186 : if (!slot)
233 6 : return -1;
234 :
235 180 : return ReplicationSlotIndex(slot);
236 : }
|
