mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-31 00:29:29 +00:00
James ColemanandGitHub
d7f579ee61
There are sometimes good reasons to manage replication slots externally to Patroni. For example, a consumer may wish to manage its own slots (so that it can more easily track when a failover has a occurred and whether it is ahead of or behind the WAL position on the new primary). Additionally tooling like pglogical actually replicates slots to all replicas so that the current position can be maintained on failover targets (this also aids consumers by supplying primitives so that they can verify data hasn't been lost or a split brain occurred relative to the physical cluster). To support these use cases this new feature allows configuring Patroni to entirely ignore sets of slots specified by any subset of name, database, slot type, and plugin.
31 lines
1.8 KiB
Python
31 lines
1.8 KiB
Python
from behave import step, then
|
|
import psycopg2 as pg
|
|
|
|
|
|
@step('I create a logical replication slot {slot_name} on {pg_name:w} with the {plugin:w} plugin')
|
|
def create_logical_replication_slot(context, slot_name, pg_name, plugin):
|
|
try:
|
|
output = context.pctl.query(pg_name, "SELECT pg_create_logical_replication_slot('{0}', '{1}'), current_database()".format(slot_name, plugin))
|
|
print(output.fetchone())
|
|
except pg.Error as e:
|
|
print(e)
|
|
assert False, "Error creating slot {0} on {1} with plugin {2}".format(slot_name, pg_name, plugin)
|
|
|
|
@then('{pg_name:w} has a logical replication slot named {slot_name} with the {plugin:w} plugin')
|
|
def has_logical_replication_slot(context, pg_name, slot_name, plugin):
|
|
try:
|
|
row = context.pctl.query(pg_name, "SELECT slot_type, plugin FROM pg_replication_slots WHERE slot_name = '{0}'".format(slot_name)).fetchone()
|
|
assert row, "Couldn't find replication slot named {0}".format(slot_name)
|
|
assert row[0] == "logical", "Found replication slot named {0} but wasn't a logical slot".format(slot_name)
|
|
assert row[1] == plugin, "Found replication slot named {0} but was using plugin {1} rather than {2}".format(slot_name, row[1], plugin)
|
|
except pg.Error as e:
|
|
assert False, "Error looking for slot {0} on {1} with plugin {2}".format(slot_name, pg_name, plugin)
|
|
|
|
@then('{pg_name:w} does not have a logical replication slot named {slot_name}')
|
|
def has_logical_replication_slot(context, pg_name, slot_name):
|
|
try:
|
|
row = context.pctl.query(pg_name, "SELECT 1 FROM pg_replication_slots WHERE slot_name = '{0}'".format(slot_name)).fetchone()
|
|
assert not row, "Found unexpected replication slot named {0}".format(slot_name)
|
|
except pg.Error as e:
|
|
assert False, "Error looking for slot {0} on {1}".format(slot_name, pg_name)
|