processutils: ensure on_completion callback is always called

If the subprocess.Popen.communicate method raises an exception,
the on_completion callback is never invoked. If a caller is
trying to use on_execute + on_completion to track lifecycle
of a process this creates a problem, as they cannot reliably
detect completion.

Change-Id: I22b2d7bde8797276f7670bc289d915dab5122481
Closes-bug: #1470868

oslo_concurrency/processutils.py

@@ -239,15 +239,16 @@ def execute(*cmd, **kwargs):
             if on_execute:
                 on_execute(obj)
 
-            result = obj.communicate(process_input)
-
-            obj.stdin.close()  # pylint: disable=E1101
-            _returncode = obj.returncode  # pylint: disable=E1101
-            LOG.log(loglevel, 'CMD "%s" returned: %s in %0.3fs',
-                    sanitized_cmd, _returncode, watch.elapsed())
-
-            if on_completion:
-                on_completion(obj)
+            try:
+                result = obj.communicate(process_input)
+
+                obj.stdin.close()  # pylint: disable=E1101
+                _returncode = obj.returncode  # pylint: disable=E1101
+                LOG.log(loglevel, 'CMD "%s" returned: %s in %0.3fs',
+                        sanitized_cmd, _returncode, watch.elapsed())
+            finally:
+                if on_completion:
+                    on_completion(obj)
 
             if not ignore_exit_code and _returncode not in check_exit_code:
                 (stdout, stderr) = result

oslo_concurrency/tests/unit/test_processutils.py

@@ -20,6 +20,7 @@
 import multiprocessing
 import os
 import stat
+import subprocess
 import sys
 import tempfile
 
@@ -78,6 +79,24 @@ def test_execute_with_callback(self):
         self.assertEqual(1, on_execute_callback.call_count)
         self.assertEqual(1, on_completion_callback.call_count)
 
+    @mock.patch.object(subprocess.Popen, "communicate")
+    def test_execute_with_callback_and_errors(self, mock_comm):
+        on_execute_callback = mock.Mock()
+        on_completion_callback = mock.Mock()
+
+        def fake_communicate(*args):
+            raise IOError("Broken pipe")
+
+        mock_comm.side_effect = fake_communicate
+
+        self.assertRaises(IOError,
+                          processutils.execute,
+                          "/bin/true",
+                          on_execute=on_execute_callback,
+                          on_completion=on_completion_callback)
+        self.assertEqual(1, on_execute_callback.call_count)
+        self.assertEqual(1, on_completion_callback.call_count)
+
 
 class ProcessExecutionErrorTest(test_base.BaseTestCase):