Merge branch 'master' of github.com:zalando/patroni into feature/disable-automatic-failover

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':

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)

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:[email protected]:5435/postgres'})
         self.leader = Leader(-1, 28, self.leadermem)
-        self.other = Member(0, 'test1', 28, {'conn_url': 'postgres://replicator:[email protected]:5433/postgres',
+        self.other = Member(0, 'test-1', 28, {'conn_url': 'postgres://replicator:[email protected]:5433/postgres',
                             'tags': {'replicatefrom': 'leader'}})
         self.me = Member(0, 'test0', 28, {'conn_url': 'postgres://replicator:[email protected]: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:[email protected]:5436/postgres'})
+            alias2 = Member(0, 'test.3', 28, {'conn_url': 'postgres://replicator:[email protected]: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):