Add support for timeout to processutils.execute

This is a standard Python feature since 3.3.

Change-Id: Ib13af5aab0ebbae532f1e46309ad6023ca94d6b9

oslo_concurrency/processutils.py

@@ -265,11 +265,16 @@ def execute(*cmd, **kwargs):
                             prlimits. If this is not set it will default to use
                             sys.executable.
     :type python_exec:      string
+    :param timeout:         Timeout (in seconds) to wait for the process
+                            termination. If timeout is reached,
+                            :class:`subprocess.TimeoutExpired` is raised.
+    :type timeout:          int
     :returns:               (stdout, stderr) from process execution
     :raises:                :class:`UnknownArgumentError` on
                             receiving unknown arguments
     :raises:                :class:`ProcessExecutionError`
     :raises:                :class:`OSError`
+    :raises:                :class:`subprocess.TimeoutExpired`
 
     The *prlimit* parameter can be used to set resource limits on the child
     process.  If this parameter is used, the child process will be spawned by a
@@ -318,6 +323,7 @@ def execute(*cmd, **kwargs):
     preexec_fn = kwargs.pop('preexec_fn', None)
     prlimit = kwargs.pop('prlimit', None)
     python_exec = kwargs.pop('python_exec', sys.executable)
+    timeout = kwargs.pop('timeout', None)
 
     if isinstance(check_exit_code, bool):
         ignore_exit_code = not check_exit_code
@@ -398,14 +404,20 @@ def execute(*cmd, **kwargs):
                 # we have to wrap this call using tpool.
                 if eventlet_patched and os.name == 'nt':
                     result = tpool.execute(obj.communicate,
-                                           process_input)
+                                           process_input,
+                                           timeout=timeout)
                 else:
-                    result = obj.communicate(process_input)
+                    result = obj.communicate(process_input,
+                                             timeout=timeout)
 
                 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())
+            except subprocess.TimeoutExpired:
+                LOG.log(loglevel, 'CMD "%s" reached timeout in %0.3fs',
+                        sanitized_cmd, watch.elapsed())
+                raise
             finally:
                 if on_completion:
                     on_completion(obj)

oslo_concurrency/tests/unit/test_processutils.py

@@ -25,6 +25,7 @@ import stat
 import subprocess
 import sys
 import tempfile
+import time
 from unittest import mock
 
 import fixtures
@@ -87,7 +88,7 @@ class UtilsTest(test_base.BaseTestCase):
         on_execute_callback = mock.Mock()
         on_completion_callback = mock.Mock()
 
-        def fake_communicate(*args):
+        def fake_communicate(*args, timeout=None):
             raise IOError("Broken pipe")
 
         mock_comm.side_effect = fake_communicate
@@ -145,9 +146,9 @@ class UtilsTest(test_base.BaseTestCase):
 
         if use_eventlet:
             mock_tpool.execute.assert_called_once_with(
-                mock_comm, fake_pinput)
+                mock_comm, fake_pinput, timeout=None)
         else:
-            mock_comm.assert_called_once_with(fake_pinput)
+            mock_comm.assert_called_once_with(fake_pinput, timeout=None)
 
     def test_windows_execute_without_eventlet(self):
         self._test_windows_execute()
@@ -513,6 +514,19 @@ grep foo
         self.assertEqual('my description', exc.description)
         self.assertEqual(str(exc), exc_message)
 
+    def test_timeout(self):
+        start = time.time()
+        # FIXME(dtantsur): I'm not sure what fancy mocking is happening in unit
+        # tests here, but I cannot check for a more precise exception because
+        # subprocess.TimeoutException != subprocess.TimeoutException.
+        # Checking the error message instead.
+        self.assertRaisesRegex(Exception,
+                               'timed out after 1 seconds',
+                               processutils.execute,
+                               '/usr/bin/env', 'sh', '-c', 'sleep 10',
+                               timeout=1)
+        self.assertLess(time.time(), start + 5)
+
 
 class ProcessExecutionErrorLoggingTest(test_base.BaseTestCase):
     def setUp(self):

releasenotes/notes/timeout-c3fb65acda04c1c7.yaml

@@ -0,0 +1,5 @@
+---
+features:
+  - |
+    Adds a new ``timeout`` argument to ``processutils.execute``. If set,
+    the process will be aborted if it runs more than ``timeout`` seconds.