Add 2 callbacks to processutils.execute()

Add optional on_execute and on_completion callbacks to allow callers of
procesutils.execute() to track process completion asynchronously.

This could be used to cache the pid of long running tasks associated
with an instance and then clear the cache when the process completes.
While the tasks are running should it be required the pid retrieved and
the process can be signaled.

Conflicts:
        oslo_concurrency/processutils.py

Co-Authored-By: abhishekkekane <abhishek.kekane@nttdata.com>
Change-Id: Ifc23325eddb523f6449ba06a2deb0885a8a7009d
(cherry picked from commit d8307005bfac7d680bb1975733928e52e9053b77)

2 files changed, 32 insertions, 0 deletions

diff --git a/oslo_concurrency/processutils.py b/oslo_concurrency/processutils.py
index ba61cba..3328ae8 100644
--- a/oslo_concurrency/processutils.py
+++ b/oslo_concurrency/processutils.py
@@ -144,6 +144,17 @@ def execute(*cmd, **kwargs):
                             last attempt, and LOG_ALL_ERRORS requires
                             logging on each occurence of an error.
     :type log_errors:       integer.
+    :param on_execute:      This function will be called upon process creation
+                            with the object as a argument.  The Purpose of this
+                            is to allow the caller of `processutils.execute` to
+                            track process creation asynchronously.
+    :type on_execute:       function(:class:`subprocess.Popen`)
+    :param on_completion:   This function will be called upon process
+                            completion with the object as a argument.  The
+                            Purpose of this is to allow the caller of
+                            `processutils.execute` to track process completion
+                            asynchronously.
+    :type on_completion:    function(:class:`subprocess.Popen`)
     :returns:               (stdout, stderr) from process execution
     :raises:                :class:`UnknownArgumentError` on
                             receiving unknown arguments
@@ -163,6 +174,8 @@ def execute(*cmd, **kwargs):
     shell = kwargs.pop('shell', False)
     loglevel = kwargs.pop('loglevel', logging.DEBUG)
     log_errors = kwargs.pop('log_errors', None)
+    on_execute = kwargs.pop('on_execute', None)
+    on_completion = kwargs.pop('on_completion', None)
 
     if isinstance(check_exit_code, bool):
         ignore_exit_code = not check_exit_code
@@ -216,6 +229,9 @@ def execute(*cmd, **kwargs):
                                    cwd=cwd,
                                    env=env_variables)
 
+            if on_execute:
+                on_execute(obj)
+
             result = obj.communicate(process_input)
 
             obj.stdin.close()  # pylint: disable=E1101
@@ -223,6 +239,10 @@ def execute(*cmd, **kwargs):
             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)
+
             if not ignore_exit_code and _returncode not in check_exit_code:
                 (stdout, stderr) = result
                 sanitized_stdout = strutils.mask_password(stdout)
diff --git a/oslo_concurrency/tests/unit/test_processutils.py b/oslo_concurrency/tests/unit/test_processutils.py
index 8d723c7..7e77e6d 100644
--- a/oslo_concurrency/tests/unit/test_processutils.py
+++ b/oslo_concurrency/tests/unit/test_processutils.py
@@ -60,6 +60,18 @@ class UtilsTest(test_base.BaseTestCase):
                                                         mock_cpu_count):
         self.assertEqual(1, processutils.get_worker_count())
 
+    def test_execute_with_callback(self):
+        on_execute_callback = mock.Mock()
+        on_completion_callback = mock.Mock()
+        processutils.execute("/bin/true")
+        self.assertEqual(0, on_execute_callback.call_count)
+        self.assertEqual(0, on_completion_callback.call_count)
+
+        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):