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.

Conflicts:
        oslo_concurrency/processutils.py
        oslo_concurrency/tests/unit/test_processutils.py

Change-Id: I22b2d7bde8797276f7670bc289d915dab5122481
Closes-bug: #1470868
(cherry picked from commit ab78480)

oslo_concurrency/processutils.py

@@ -244,16 +244,17 @@ 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
-            end_time = time.time() - start_time
-            LOG.log(loglevel, 'CMD "%s" returned: %s in %0.3fs' %
-                    (sanitized_cmd, _returncode, end_time))
-
-            if on_completion:
-                on_completion(obj)
+            try:
+                result = obj.communicate(process_input)
+
+                obj.stdin.close()  # pylint: disable=E1101
+                _returncode = obj.returncode  # pylint: disable=E1101
+                end_time = time.time() - start_time
+                LOG.log(loglevel, 'CMD "%s" returned: %s in %0.3fs',
+                        sanitized_cmd, _returncode, end_time)
+            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

@@ -73,6 +73,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)
+
     def test_execute_with_preexec_fn(self):
         # NOTE(dims): preexec_fn is set to a callable object, this object
         # will be called in the child process just before the child is