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
        openstack/common/processutils.py

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

2 files changed, 33 insertions, 0 deletions

diff --git a/openstack/common/processutils.py b/openstack/common/processutils.py
index 894d3932..ce24e897 100644
--- a/openstack/common/processutils.py
+++ b/openstack/common/processutils.py
@@ -116,6 +116,17 @@ def execute(*cmd, **kwargs):
     :type shell:            boolean
     :param loglevel:        log level for execute commands.
     :type loglevel:         int.  (Should be logging.DEBUG or logging.INFO)
+    :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
@@ -132,6 +143,8 @@ def execute(*cmd, **kwargs):
     root_helper = kwargs.pop('root_helper', '')
     shell = kwargs.pop('shell', False)
     loglevel = kwargs.pop('loglevel', logging.DEBUG)
+    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
@@ -173,6 +186,10 @@ def execute(*cmd, **kwargs):
                                    preexec_fn=preexec_fn,
                                    shell=shell,
                                    env=env_variables)
+
+            if on_execute:
+                on_execute(obj)
+
             result = None
             for _i in six.moves.range(20):
                 # NOTE(russellb) 20 is an arbitrary number of retries to
@@ -190,6 +207,10 @@ def execute(*cmd, **kwargs):
             obj.stdin.close()  # pylint: disable=E1101
             _returncode = obj.returncode  # pylint: disable=E1101
             LOG.log(loglevel, 'Result was %s' % _returncode)
+
+            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/tests/unit/test_processutils.py b/tests/unit/test_processutils.py
index 47be604e..1e64fb4d 100644
--- a/tests/unit/test_processutils.py
+++ b/tests/unit/test_processutils.py
@@ -57,6 +57,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):