Always call on_start callback when starting Patroni (#262)

When Patroni was "joining" already running postgres it was not calling
callbacks, what in some cases causing issues (callback could be used to
change routing/load-balancer or assign/remove floating (service) ip.

In addition to that we should `start` postgres instead of `restart`-ing
it when doing recovery, because in this case 'on_start' callback should
be called, instead of 'on_restart'

patroni/ha.py

@@ -9,6 +9,7 @@ import pytz
 from multiprocessing.pool import ThreadPool
 from patroni.async_executor import AsyncExecutor
 from patroni.exceptions import DCSError, PostgresConnectionException
+from patroni.postgresql import ACTION_ON_START
 from patroni.utils import sleep
 
 logger = logging.getLogger(__name__)
@@ -587,6 +588,8 @@ class Ha(object):
                 # stops PostgreSQL, therefore, we only reload replication slots if no
                 # asynchronous processes are running (should be always the case for the master)
                 if not self._async_executor.busy:
+                    if not self.state_handler.cb_called:
+                        self.state_handler.call_nowait(ACTION_ON_START)
                     self.state_handler.sync_replication_slots(self.cluster)
         except DCSError:
             logger.error('Error communicating with DCS')

patroni/postgresql.py

@@ -76,6 +76,7 @@ class Postgresql(object):
 
         self._pgpass = config.get('pgpass') or os.path.join(os.path.expanduser('~'), 'pgpass')
         self.callback = config.get('callbacks') or {}
+        self.__cb_called = False
         config_base_name = config.get('config_base_name', 'postgresql')
         self._postgresql_conf = os.path.join(self._data_dir, config_base_name + '.conf')
         self._postgresql_base_conf_name = config_base_name + '.base.conf'
@@ -459,8 +460,15 @@ class Postgresql(object):
         except Exception:
             return False
 
+    @property
+    def cb_called(self):
+        return self.__cb_called
+
     def call_nowait(self, cb_name):
         """ pick a callback command and call it without waiting for it to finish """
+        if cb_name in (ACTION_ON_START, ACTION_ON_STOP, ACTION_ON_RESTART, ACTION_ON_ROLE_CHANGE):
+            self.__cb_called = True
+
         if not self.callback or cb_name not in self.callback:
             return False
         cmd = self.callback[cb_name]
@@ -788,7 +796,7 @@ class Postgresql(object):
             self._need_rewind = False
         else:
             self.write_recovery_conf(primary_conninfo)
-            ret = self.restart()
+            ret = self.start() if recovery else self.restart()
             self.set_role('replica')
 
         if change_role:

tests/test_api.py

@@ -179,7 +179,6 @@ class TestRestApiHandler(unittest.TestCase):
             MockRestApiServer(RestApiHandler, 'POST /reload HTTP/1.0' + self._authorization)
         self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /reload HTTP/1.0' + self._authorization))
 
-    #@patch.object(MockPatroni, 'dcs')
     def test_do_POST_restart(self):
         request = 'POST /restart HTTP/1.0' + self._authorization
         self.assertIsNotNone(MockRestApiServer(RestApiHandler, request))
@@ -221,7 +220,6 @@ class TestRestApiHandler(unittest.TestCase):
                 request = make_request('{"role": "master", "postgres_version": "9.5.2"}')
                 MockRestApiServer(RestApiHandler, request)
 
-    #@patch.object(MockPatroni, 'dcs')
     def test_do_DELETE_restart(self):
         for retval in (True, False):
             with patch.object(MockHa, 'delete_future_restart', Mock(return_value=retval)):

tests/test_ctl.py

@@ -29,9 +29,9 @@ def test_rw_config():
         os.rmdir(CONFIG_FILE_PATH)
 
 
-@patch('patroni.ctl.load_config', Mock(return_value={'postgresql': {'data_dir': '.', 'parameters': {}, 'retry_timeout': 5},
-                                                     'restapi': {'auth': 'u:p', 'listen': ''},
-                                                     'etcd': {'host': 'localhost:4001'}}))
+@patch('patroni.ctl.load_config',
+       Mock(return_value={'postgresql': {'data_dir': '.', 'parameters': {}, 'retry_timeout': 5},
+                          'restapi': {'auth': 'u:p', 'listen': ''}, 'etcd': {'host': 'localhost:4001'}}))
 class TestCtl(unittest.TestCase):
 
     @patch('socket.getaddrinfo', socket_getaddrinfo)