diff --git a/patroni/api.py b/patroni/api.py index 26c15cea..b09d4185 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -367,7 +367,11 @@ class RestApiHandler(BaseHTTPRequestHandler): def get_postgresql_status(self, retry=False): try: - row = self.query("""SELECT to_char(pg_postmaster_start_time(), 'YYYY-MM-DD HH24:MI:SS.MS TZ'), + row = self.query("""WITH replication_info AS ( + SELECT usename, application_name, client_addr, state, sync_state, sync_priority + FROM pg_stat_replication + ) + SELECT to_char(pg_postmaster_start_time(), 'YYYY-MM-DD HH24:MI:SS.MS TZ'), pg_is_in_recovery(), CASE WHEN pg_is_in_recovery() THEN 0 @@ -376,8 +380,10 @@ class RestApiHandler(BaseHTTPRequestHandler): pg_xlog_location_diff(pg_last_xlog_receive_location(), '0/0')::bigint, pg_xlog_location_diff(pg_last_xlog_replay_location(), '0/0')::bigint, to_char(pg_last_xact_replay_timestamp(), 'YYYY-MM-DD HH24:MI:SS.MS TZ'), - pg_is_in_recovery() AND pg_is_xlog_replay_paused()""", retry=retry)[0] - return { + pg_is_in_recovery() AND pg_is_xlog_replay_paused(), + (SELECT json_agg(row_to_json(ri)) FROM replication_info ri)""", retry=retry)[0] + + result = { 'state': self.server.patroni.postgresql.state, 'postmaster_start_time': row[0], 'role': 'replica' if row[1] else 'master', @@ -390,6 +396,11 @@ class RestApiHandler(BaseHTTPRequestHandler): 'location': row[2] }) } + + if row[7]: + result['replication'] = row[7] + + return result except (psycopg2.Error, RetryFailedError, PostgresConnectionException): state = self.server.patroni.postgresql.state if state == 'running': diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 99d8cde0..ce8b6183 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -1,6 +1,8 @@ +from collections import defaultdict import logging import os import psycopg2 +import re import shlex import shutil import subprocess @@ -21,6 +23,22 @@ ACTION_ON_RELOAD = "on_reload" ACTION_ON_ROLE_CHANGE = "on_role_change" +def slot_name_from_member_name(member_name): + """Translate member name to valid PostgreSQL slot name. + + PostgreSQL replication slot names must be valid PostgreSQL names. This function maps the wider space of + member names to valid PostgreSQL names. Names are lowercased, dashes and periods common in hostnames + are replaced with underscores, other characters are encoded as their unicode codepoint. Name is truncated + to 64 characters. Multiple different member names may map to a single slot name.""" + + def replace_char(match): + c = match.group(0) + return '_' if c in '-.' else "u{:04d}".format(ord(c)) + + slot_name = re.sub('[^a-z0-9_]', replace_char, member_name.lower()) + return slot_name[0:64] + + class Postgresql(object): # List of parameters which must be always passed to postmaster as command line options @@ -651,7 +669,7 @@ class Postgresql(object): if primary_conninfo: f.write("primary_conninfo = '{0}'\n".format(primary_conninfo)) if self.use_slots: - f.write("primary_slot_name = '{0}'\n".format(self.name)) + f.write("primary_slot_name = '{0}'\n".format(slot_name_from_member_name(self.name))) for name, value in self.config.get('recovery_conf', {}).items(): if name not in ('standby_mode', 'recovery_target_timeline', 'primary_conninfo', 'primary_slot_name'): f.write("{0} = '{1}'\n".format(name, value)) @@ -887,21 +905,32 @@ $$""".format(name, ' '.join(options)), name, password, password) # the replicatefrom destination member is currently not a member of the cluster (fallback to the # master), or if replicatefrom destination member happens to be the current master if self.role == 'master': - slots = [m.name for m in cluster.members if m.name != self.name and - (m.replicatefrom is None or m.replicatefrom == self.name or - not cluster.has_member(m.replicatefrom))] + slot_members = [m.name for m in cluster.members if m.name != self.name and + (m.replicatefrom is None or m.replicatefrom == self.name or + not cluster.has_member(m.replicatefrom))] else: # only manage slots for replicas that replicate from this one, except for the leader among them - slots = [m.name for m in cluster.members if m.replicatefrom == self.name and - m.name != cluster.leader.name] + slot_members = [m.name for m in cluster.members if m.replicatefrom == self.name and + m.name != cluster.leader.name] + slots = set(slot_name_from_member_name(name) for name in slot_members) + + if len(slots) < len(slot_members): + # Find which names are conflicting for a nicer error message + slot_conflicts = defaultdict(list) + for name in slot_members: + slot_conflicts[slot_name_from_member_name(name)].append(name) + logger.error("Following cluster members share a replication slot name: %s", + "; ".join("{} map to {}".format(", ".join(v), k) + for k, v in slot_conflicts.items() if len(v) > 1)) + # drop unused slots - for slot in set(self._replication_slots) - set(slots): + for slot in set(self._replication_slots) - slots: self.query("""SELECT pg_drop_replication_slot(%s) WHERE EXISTS(SELECT 1 FROM pg_replication_slots WHERE slot_name = %s AND NOT active)""", slot, slot) # create new slots - for slot in set(slots) - set(self._replication_slots): + for slot in slots - set(self._replication_slots): self.query("""SELECT pg_create_physical_replication_slot(%s) WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = %s)""", slot, slot) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 9fb7bb7a..e77d22ba 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -32,8 +32,10 @@ class MockCursor(object): self.results = [(0,)] elif sql == 'SELECT pg_is_in_recovery()': self.results = [(False, )] - elif sql.startswith('SELECT to_char(pg_postmaster_start_time'): - self.results = [('', True, '', '', '', '', False)] + elif sql.startswith('WITH replication_info AS ('): + replication_info = '[{"application_name":"walreceiver","client_addr":"1.2.3.4",' +\ + '"state":"streaming","sync_state":"async","sync_priority":0}]' + self.results = [('', True, '', '', '', '', False, replication_info)] elif sql.startswith('SELECT name, setting'): self.results = [('wal_segment_size', '2048', '8kB', 'integer', 'internal'), ('search_path', 'public', None, 'string', 'user'), @@ -182,7 +184,7 @@ class TestPostgresql(unittest.TestCase): 'restore': 'true'}) self.leadermem = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres'}) self.leader = Leader(-1, 28, self.leadermem) - self.other = Member(0, 'test1', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres', + self.other = Member(0, 'test-1', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres', 'tags': {'replicatefrom': 'leader'}}) self.me = Member(0, 'test0', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres'}) @@ -316,6 +318,15 @@ class TestPostgresql(unittest.TestCase): self.p.schedule_load_slots = False with mock.patch('patroni.postgresql.Postgresql.role', new_callable=PropertyMock(return_value='replica')): self.p.sync_replication_slots(cluster) + with mock.patch('patroni.postgresql.logger.error', new_callable=Mock()) as errorlog_mock: + self.p.query = Mock() + alias1 = Member(0, 'test-3', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres'}) + alias2 = Member(0, 'test.3', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres'}) + cluster.members.extend([alias1, alias2]) + self.p.sync_replication_slots(cluster) + errorlog_mock.assert_called_once() + assert "test-3" in errorlog_mock.call_args[0][1] + assert "test.3" in errorlog_mock.call_args[0][1] @patch.object(MockConnect, 'closed', 2) def test__query(self):