Merge lp:~jtv/maas/bug-1059453 into lp:maas/trunk

[Jeroen T. Vermeulen]

Run celeryd synchronously, so Upstart doesn't need to track fork()s (many of which precede the operative one as maas-provision starts up).

Discussed with Julian and many others. The "maas-provision start-cluster-controller" command will now keep running until either the cluster controller is rejected by an administrator, or celeryd exits for whatever reason.

Jeroen

[Gavin Panella (allenap)]

[1]

+    return_code = check_call(command, env=env, preexec_fn=Become(uid, gid))
+    if return_code != 0:
+        raise SystemExit(return_code)

So, we've gone full-circle :) Which means we also get back to my
original comment: just exec here. It's absolutely pointless leaving an
idle Python interpreter around just to wait for a return code.

Also, check_call will raise a CalledProcessError if the process exits
with a non-zero code, so the conditional branch above will never be
taken.

review: Needs Fixing

[Jeroen T. Vermeulen (jtv)]

Changes made. I've got a final test run going that I fully expect to pass; I've set things up to commit and push after a successful run.

[lp:~jtv/maas/bug-1059453]

1142. By Jeroen T. Vermeulen on 2012-10-03

Review changes: don't fork, just exec.

[Gavin Panella (allenap)]

review: Approve

=== modified file 'src/provisioningserver/start_cluster_controller.py'
--- src/provisioningserver/start_cluster_controller.py	2012-10-02 10:20:17 +0000
+++ src/provisioningserver/start_cluster_controller.py	2012-10-03 11:41:38 +0000
@@ -137,12 +137,12 @@
         '-Q', get_cluster_uuid(),
         ]
 
-    pid = os.fork()
-    if pid == 0:
-        # Child process.  Become the right user, and start celery.
-        os.setuid(uid)
-        os.setgid(gid)
-        os.execvpe(command[0], command, env)
+    # Change gid first, just in case changing the uid might deprive
+    # us of the privileges required to setgid.
+    os.setgid(gid)
+    os.setuid(uid)
+
+    os.execvpe(command[0], command, env=env)
 
 
 def request_refresh(server_url):
@@ -161,9 +161,12 @@
     This starts up celeryd, listening to the broker that the region
     controller pointed us to, and on the appropriate queue.
     """
+    # Get the region controller to send out credentials.  If it arrives
+    # before celeryd has started up, we should find the message waiting
+    # in our queue.  Even if we're new and the queue did not exist yet,
+    # the arriving task will create the queue.
+    request_refresh(server_url)
     start_celery(connection_details, user=user, group=group)
-    sleep(10)
-    request_refresh(server_url)
 
 
 def run(args):

=== modified file 'src/provisioningserver/tests/test_start_cluster_controller.py'
--- src/provisioningserver/tests/test_start_cluster_controller.py	2012-10-02 10:20:17 +0000
+++ src/provisioningserver/tests/test_start_cluster_controller.py	2012-10-03 11:41:38 +0000
@@ -87,10 +87,9 @@
         self.patch(start_cluster_controller, 'getpwnam')
         start_cluster_controller.getpwnam.pw_uid = randint(3000, 4000)
         start_cluster_controller.getpwnam.pw_gid = randint(3000, 4000)
-        self.patch(os, 'fork').side_effect = Executing()
-        self.patch(os, 'execvpe').side_effect = Executing()
         self.patch(os, 'setuid')
         self.patch(os, 'setgid')
+        self.patch(os, 'execvpe').side_effect = Executing()
         get_uuid = self.patch(start_cluster_controller, 'get_cluster_uuid')
         get_uuid.return_value = factory.getRandomUUID()
 
@@ -133,36 +132,19 @@
         """Prepare to return "request pending" from API request."""
         self.prepare_response(httplib.ACCEPTED)
 
-    def pretend_to_fork_into_child(self):
-        """Make `fork` act as if it's returning into the child process.
-
-        The start_cluster_controller child process then executes celeryd,
-        so this call also patches up the call that does that so it pretends
-        to be successful.
-        """
-        self.patch(os, 'fork').return_value = 0
-        self.patch(os, 'execvpe')
-
-    def pretend_to_fork_into_parent(self):
-        """Make `fork` act as if it's returning into the parent process."""
-        self.patch(os, 'fork').return_value = randint(2, 65535)
-
     def test_run_command(self):
         # We can't really run the script, but we can verify that (with
         # the right system functions patched out) we can run it
         # directly.
-        self.pretend_to_fork_into_child()
-        self.patch(start_cluster_controller, 'sleep')
+        start_cluster_controller.sleep.side_effect = None
         self.prepare_success_response()
         parser = ArgumentParser()
         start_cluster_controller.add_arguments(parser)
-        start_cluster_controller.run(parser.parse_args((make_url(),)))
-        self.assertEqual(1, os.fork.call_count)
+        self.assertRaises(
+            Executing,
+            start_cluster_controller.run,
+            parser.parse_args((make_url(),)))
         self.assertEqual(1, os.execvpe.call_count)
-        os.setuid.assert_called_once_with(
-            start_cluster_controller.getpwnam.return_value.pw_uid)
-        os.setgid.assert_called_once_with(
-            start_cluster_controller.getpwnam.return_value.pw_gid)
 
     def test_uses_given_url(self):
         url = make_url('region')
@@ -227,13 +209,14 @@
             (server_url, connection_details))
 
     def test_start_up_calls_refresh_secrets(self):
+        start_cluster_controller.sleep.side_effect = None
         url = make_url('region')
         connection_details = self.make_connection_details()
-        self.pretend_to_fork_into_parent()
-        self.patch(start_cluster_controller, 'sleep')
         self.prepare_success_response()
 
-        start_cluster_controller.start_up(
+        self.assertRaises(
+            Executing,
+            start_cluster_controller.start_up,
             url, connection_details,
             factory.make_name('user'), factory.make_name('group'))
 
@@ -246,13 +229,14 @@
         self.assertEqual("refresh_workers", post["op"])
 
     def test_start_up_ignores_failure_on_refresh_secrets(self):
-        self.pretend_to_fork_into_parent()
-        self.patch(start_cluster_controller, 'sleep')
+        start_cluster_controller.sleep.side_effect = None
         self.patch(MAASDispatcher, 'dispatch_query').side_effect = URLError(
             "Simulated HTTP failure.")
 
-        start_cluster_controller.start_up(
+        self.assertRaises(
+            Executing,
+            start_cluster_controller.start_up,
             make_url(), self.make_connection_details(),
             factory.make_name('user'), factory.make_name('group'))
 
-        self.assertEqual(1, os.fork.call_count)
+        self.assertEqual(1, os.execvpe.call_count)

=== modified file 'src/provisioningserver/utils.py'
--- src/provisioningserver/utils.py	2012-09-25 05:21:24 +0000
+++ src/provisioningserver/utils.py	2012-10-03 11:41:38 +0000
@@ -376,7 +376,7 @@
         try:
             self.setup()
             self.execute(argv)
-        except CalledProcessError, error:
+        except CalledProcessError as error:
             # Print error.cmd and error.output too?
             raise SystemExit(error.returncode)
         except KeyboardInterrupt: