Replace self-execute with multiprocessing.Process (#994)

In addition to that transfer postmaster pid to Patroni process with the help of multiprocessing.Pipe instead of using stdin-stdout pipes.

Closes https://github.com/zalando/patroni/issues/992

patroni/__init__.py

@@ -153,34 +153,11 @@ def patroni_main():
         logging.shutdown()
 
 
-def pg_ctl_start(args):
-    import subprocess
-    if os.name != 'nt':
-        os.setsid()
-    postmaster = subprocess.Popen(args)
-    print(postmaster.pid)
-
-
-def call_self(args, **kwargs):
-    """This function executes Patroni once again with provided arguments.
-
-    :args: list of arguments to call Patroni with.
-    :returns: `Popen` object"""
-
-    exe = [sys.executable]
-    if not getattr(sys, 'frozen', False):  # Binary distribution?
-        exe.append(sys.argv[0])
-
-    import subprocess
-    return subprocess.Popen(exe + args, **kwargs)
-
-
 def main():
     if os.getpid() != 1:
-        if len(sys.argv) > 5 and sys.argv[1] == 'pg_ctl_start':
-            return pg_ctl_start(sys.argv[2:])
         return patroni_main()
 
+    # Patroni started with PID=1, it looks like we are in the container
     pid = 0
 
     # Looks like we are in a docker, so we will act like init
@@ -209,6 +186,8 @@ def main():
     signal.signal(signal.SIGABRT, passtochild)
     signal.signal(signal.SIGTERM, passtochild)
 
-    patroni = call_self(sys.argv[1:])
+    import multiprocessing
+    patroni = multiprocessing.Process(target=patroni_main)
+    patroni.start()
     pid = patroni.pid
-    patroni.wait()
+    patroni.join()

patroni/postmaster.py

@@ -1,12 +1,11 @@
 import logging
+import multiprocessing
 import os
 import psutil
 import re
 import signal
 import subprocess
 
-from patroni import call_self
-
 logger = logging.getLogger(__name__)
 
 STOP_SIGNALS = {
@@ -16,6 +15,18 @@ STOP_SIGNALS = {
 }
 
 
+def pg_ctl_start(conn, cmdline, env):
+    if os.name != 'nt':
+        os.setsid()
+    try:
+        postmaster = subprocess.Popen(cmdline, close_fds=True, env=env)
+        conn.send(postmaster.pid)
+    except Exception:
+        logger.exception('Failed to execute %s', cmdline)
+        conn.send(None)
+    conn.close()
+
+
 class PostmasterProcess(psutil.Process):
 
     def __init__(self, pid):
@@ -159,10 +170,13 @@ class PostmasterProcess(psutil.Process):
             pass
         cmdline = [pgcommand, '-D', data_dir, '--config-file={}'.format(conf)] + options
         logger.debug("Starting postgres: %s", " ".join(cmdline))
-        proc = call_self(['pg_ctl_start'] + cmdline, close_fds=(os.name != 'nt'),
-                         stdout=subprocess.PIPE, env=env)
-        pid = int(proc.stdout.readline().strip())
-        proc.wait()
+        parent_conn, child_conn = multiprocessing.Pipe(False)
+        proc = multiprocessing.Process(target=pg_ctl_start, args=(child_conn, cmdline, env))
+        proc.start()
+        pid = parent_conn.recv()
+        proc.join()
+        if pid is None:
+            return
         logger.info('postmaster pid=%s', pid)
 
         # TODO: In an extremely unlikely case, the process could have exited and the pid reassigned. The start

tests/test_patroni.py

@@ -67,16 +67,12 @@ class TestPatroni(unittest.TestCase):
                     patroni_main()
 
     @patch('os.getpid')
-    @patch('subprocess.Popen', )
+    @patch('multiprocessing.Process')
     @patch('patroni.patroni_main', Mock())
-    def test_patroni_main(self, mock_popen, mock_getpid):
+    def test_patroni_main(self, mock_process, mock_getpid):
         mock_getpid.return_value = 2
         _main()
 
-        with patch('sys.frozen', Mock(return_value=True), create=True), patch('os.setsid', Mock()):
-            sys.argv = ['/patroni', 'pg_ctl_start', 'postgres', '-D', '/data', '--max_connections=100']
-            _main()
-
         mock_getpid.return_value = 1
 
         def mock_signal(signo, handler):
@@ -94,10 +90,10 @@ class TestPatroni(unittest.TestCase):
             if signo == signal.SIGHUP:
                 ref['passtochild'] = handler
 
-        def mock_wait():
+        def mock_join():
             ref['passtochild'](0, None)
 
-        mock_popen.return_value.wait = mock_wait
+        mock_process.return_value.join = mock_join
         with patch('signal.signal', mock_sighup), patch('os.kill', Mock()):
             self.assertIsNone(_main())
 

tests/test_postgresql.py

@@ -633,7 +633,7 @@ class TestPostgresql(unittest.TestCase):
         self.assertFalse(self.p.bootstrap(config))
 
         mock_cancellable_subprocess_call.return_value = 0
-        with patch('subprocess.Popen', Mock(side_effect=Exception("42"))),\
+        with patch('multiprocessing.Process', Mock(side_effect=Exception("42"))),\
                 patch('os.path.isfile', Mock(return_value=True)),\
                 patch('os.unlink', Mock()),\
                 patch.object(Postgresql, 'save_configuration_files', Mock()),\

tests/test_postmaster.py

@@ -87,13 +87,16 @@ class TestPostmasterProcess(unittest.TestCase):
     def test_start(self, mock_frompidfile, mock_frompid, mock_popen):
         mock_frompidfile.return_value._is_postmaster_process.return_value = False
         mock_frompid.return_value = "proc 123"
-        mock_popen.return_value.stdout.readline.return_value = '123'
+        mock_popen.return_value.pid = 123
         self.assertEqual(PostmasterProcess.start('true', '/tmp', '/tmp/test.conf', []), "proc 123")
         mock_frompid.assert_called_with(123)
 
         mock_frompidfile.side_effect = psutil.NoSuchProcess(123)
         self.assertEqual(PostmasterProcess.start('true', '/tmp', '/tmp/test.conf', []), "proc 123")
 
+        mock_popen.side_effect = Exception
+        self.assertIsNone(PostmasterProcess.start('true', '/tmp', '/tmp/test.conf', []))
+
     @patch('psutil.Process.__init__', Mock(side_effect=psutil.NoSuchProcess(123)))
     def test_read_postmaster_pidfile(self):
         with patch.object(builtins, 'open', Mock(side_effect=IOError)):