Handle cases when conn_url is not defined (#1482)

On K8s when one of the Patroni pods in starting there is valid annotation yet, which could cause failure in patronictl.
In addition to that handle cases if port isn't specified in the standby_cluster configuration.

Close https://github.com/zalando/patroni/issues/1100
Close https://github.com/zalando/patroni/issues/1463

patroni/ctl.py

@@ -735,20 +735,21 @@ def output_members(cluster, name, extended=False, fmt='pretty'):
             columns.append(c)
 
     # Show Host as 'host:port' if somebody is running on non-standard port or two nodes are running on the same host
-    append_port = any(m['port'] != 5432 for m in cluster['members']) or\
-        len(set(m['host'] for m in cluster['members'])) < len(cluster['members'])
+    members = [m for m in cluster['members'] if 'host' in m]
+    append_port = any('port' in m and m['port'] != 5432 for m in members) or\
+        len(set(m['host'] for m in cluster['members'])) < len(members)
 
     for m in cluster['members']:
         logging.debug(m)
 
         lag = m.get('lag', '')
-        m.update(cluster=name, member=m['name'], tl=m.get('timeline', ''),
+        m.update(cluster=name, member=m['name'], host=m.get('host'), tl=m.get('timeline', ''),
                  role='' if m['role'] == 'replica' else m['role'].replace('_', ' ').title(),
                  lag_in_mb=round(lag/1024/1024) if isinstance(lag, six.integer_types) else lag,
                  pending_restart='*' if m.get('pending_restart') else '',
                  tags=json.dumps(m['tags']) if m.get('tags') else '')
 
-        if append_port:
+        if append_port and m['host'] and m.get('port'):
             m['host'] = ':'.join([m['host'], str(m['port'])])
 
         if 'scheduled_restart' in m:

patroni/dcs/__init__.py

@@ -140,10 +140,10 @@ class Member(namedtuple('Member', 'index,name,session,data')):
     @property
     def conn_url(self):
         conn_url = self.data.get('conn_url')
-        conn_kwargs = self.data.get('conn_kwargs')
         if conn_url:
             return conn_url
 
+        conn_kwargs = self.data.get('conn_kwargs')
         if conn_kwargs:
             conn_url = uri('postgresql', (conn_kwargs.get('host'), conn_kwargs.get('port', 5432)))
             self.data['conn_url'] = conn_url
@@ -151,16 +151,19 @@ class Member(namedtuple('Member', 'index,name,session,data')):
 
     def conn_kwargs(self, auth=None):
         defaults = {
-            "host": "",
-            "port": "",
-            "database": ""
+            "host": None,
+            "port": None,
+            "database": None
         }
         ret = self.data.get('conn_kwargs')
         if ret:
             defaults.update(ret)
             ret = defaults
         else:
-            r = urlparse(self.conn_url)
+            conn_url = self.conn_url
+            if not conn_url:
+                return {}  # due to the invalid conn_url we don't care about authentication parameters
+            r = urlparse(conn_url)
             ret = {
                 'host': r.hostname,
                 'port': r.port or 5432,

patroni/postgresql/__init__.py

@@ -651,10 +651,10 @@ class Postgresql(object):
         return result
 
     @contextmanager
-    def get_replication_connection_cursor(self, host='localhost', port=5432, database=None, **kwargs):
+    def get_replication_connection_cursor(self, host='localhost', port=5432, **kwargs):
         conn_kwargs = self.config.replication.copy()
-        conn_kwargs.update(host=host, port=int(port), database=database or self._database, connect_timeout=3,
-                           user=conn_kwargs.pop('username'), replication=1, options='-c statement_timeout=2000')
+        conn_kwargs.update(host=host, port=int(port) if port else None, user=conn_kwargs.pop('username'),
+                           connect_timeout=3, replication=1, options='-c statement_timeout=2000')
         with get_connection_cursor(**conn_kwargs) as cur:
             yield cur
 

patroni/postgresql/config.py

@@ -652,7 +652,7 @@ class ConfigHandler(object):
             else:
                 return False
 
-        return all(primary_conninfo.get(p) == str(v) for p, v in wanted_primary_conninfo.items())
+        return all(primary_conninfo.get(p) == str(v) for p, v in wanted_primary_conninfo.items() if v is not None)
 
     def check_recovery_conf(self, member):
         """Returns a tuple. The first boolean element indicates that recovery params don't match

patroni/utils.py

@@ -390,9 +390,12 @@ def cluster_as_json(cluster):
         else:
             role = 'replica'
 
+        member = {'name': m.name, 'role': role, 'state': m.data.get('state', ''), 'api_url': m.api_url}
         conn_kwargs = m.conn_kwargs()
-        member = {'name': m.name, 'host': conn_kwargs['host'], 'port': int(conn_kwargs['port']),
-                  'role': role, 'state': m.data.get('state', ''), 'api_url': m.api_url}
+        if conn_kwargs.get('host'):
+            member['host'] = conn_kwargs['host']
+            if conn_kwargs.get('port'):
+                member['port'] = int(conn_kwargs['port'])
         optional_attributes = ('timeline', 'pending_restart', 'scheduled_restart', 'tags')
         member.update({n: m.data[n] for n in optional_attributes if n in m.data})
 

tests/test_ctl.py

@@ -72,10 +72,9 @@ class TestCtl(unittest.TestCase):
     def test_output_members(self):
         scheduled_at = datetime.now(tzutc) + timedelta(seconds=600)
         cluster = get_cluster_initialized_with_leader(Failover(1, 'foo', 'bar', scheduled_at))
-        self.assertIsNone(output_members(cluster, name='abc', fmt='pretty'))
-        self.assertIsNone(output_members(cluster, name='abc', fmt='json'))
-        self.assertIsNone(output_members(cluster, name='abc', fmt='yaml'))
-        self.assertIsNone(output_members(cluster, name='abc', fmt='tsv'))
+        del cluster.members[1].data['conn_url']
+        for fmt in ('pretty', 'json', 'yaml', 'tsv'):
+            self.assertIsNone(output_members(cluster, name='abc', fmt=fmt))
 
     @patch('patroni.ctl.get_dcs')
     @patch.object(PoolManager, 'request', Mock(return_value=MockResponse()))