Merge pull request #4803 from jdmeyer3/bug/thread_buffering_issue

fixing subprocess to use system buffer instead of being unbuffered

CHANGELOG.rst

@@ -28,6 +28,12 @@ Changed
   writing very large executions (executions with large results) to the database. #4767
 * Improved development instructions in requirements.txt and dist_utils.py comment headers
   (improvement) #4774
+* Add new ``actionrunner.stream_output_buffer_size`` config option and default it to ``-1``
+  (previously default value was ``0``). This should result in a better performance and smaller
+  CPU utilization for Python runner actions which produce a lot of output.
+  (improvement)
+
+  Reported and contributed by Joshua Meyer (@jdmeyer3) #4803
 
 Fixed
 ~~~~~

conf/st2.conf.sample

@@ -20,6 +20,8 @@ python3_prefix = None
 virtualenv_binary = /usr/bin/virtualenv
 # Python 3 binary which will be used by Python actions for packs which use Python 3 virtual environment.
 python3_binary = /usr/bin/python3
+# Buffer size to use for real time action output streaming. 0 means unbuffered 1 means line buffered, -1 means system default, which usually means fully buffered and any other positive value means use a buffer of (approximately) that size
+stream_output_buffer_size = -1
 # List of virtualenv options to be passsed to "virtualenv" command that creates pack virtualenv.
 virtualenv_opts = --system-site-packages # comma separated list allowed here.
 # True to store and stream action output (stdout and stderr) in real-time.

contrib/runners/python_runner/python_runner/python_runner.py

@@ -248,8 +248,11 @@ def run(self, action_parameters):
         if stdin_params:
             command_string = 'echo %s | %s' % (quote_unix(stdin_params), command_string)
 
-        LOG.debug('Running command: PATH=%s PYTHONPATH=%s %s' % (env['PATH'], env['PYTHONPATH'],
-                                                                 command_string))
+        bufsize = cfg.CONF.actionrunner.stream_output_buffer_size
+
+        LOG.debug('Running command (bufsize=%s): PATH=%s PYTHONPATH=%s %s' % (bufsize, env['PATH'],
+                                                                              env['PYTHONPATH'],
+                                                                              command_string))
         exit_code, stdout, stderr, timed_out = run_command(cmd=args,
                                                            stdin=stdin,
                                                            stdout=subprocess.PIPE,
@@ -261,7 +264,8 @@ def run(self, action_parameters):
                                                            read_stderr_func=read_and_store_stderr,
                                                            read_stdout_buffer=stdout,
                                                            read_stderr_buffer=stderr,
-                                                           stdin_value=stdin_params)
+                                                           stdin_value=stdin_params,
+                                                           bufsize=bufsize)
         LOG.debug('Returning values: %s, %s, %s, %s', exit_code, stdout, stderr, timed_out)
         LOG.debug('Returning.')
         return self._get_output_values(exit_code, stdout, stderr, timed_out)

contrib/runners/python_runner/tests/unit/test_pythonrunner.py

@@ -63,6 +63,8 @@
 
 PRINT_CONFIG_ITEM_ACTION = os.path.join(tests_base.get_resources_path(), 'packs',
                                         'pythonactions/actions/print_config_item_doesnt_exist.py')
+PRINT_TO_STDOUT_STDERR_ACTION = os.path.join(tests_base.get_resources_path(), 'packs',
+                                      'pythonactions/actions/print_to_stdout_and_stderr.py')
 
 
 # Note: runner inherits parent args which doesn't work with tests since test pass additional
@@ -406,6 +408,31 @@ def test_action_stdout_and_stderr_is_stored_in_the_db(self, mock_spawn, mock_pop
         self.assertEqual(output_dbs[1].data, mock_stderr[1])
         self.assertEqual(output_dbs[2].data, mock_stderr[2])
 
+    def test_real_time_output_streaming_bufsize(self):
+        # Test various values for bufsize and verify it works / doesn't hang the process
+        cfg.CONF.set_override(name='stream_output', group='actionrunner', override=True)
+
+        bufsize_values = [-100, -2, -1, 0, 1, 2, 1024, 2048, 4096, 10000]
+
+        for index, bufsize in enumerate(bufsize_values, 1):
+            cfg.CONF.set_override(name='stream_output_buffer_size', override=bufsize,
+                                  group='actionrunner')
+
+            output_dbs = ActionExecutionOutput.get_all()
+            self.assertEqual(len(output_dbs), (index - 1) * 4)
+
+            runner = self._get_mock_runner_obj()
+            runner.entry_point = PRINT_TO_STDOUT_STDERR_ACTION
+            runner.pre_run()
+            (_, output, _) = runner.run({'stdout_count': 2, 'stderr_count': 2})
+
+            self.assertEqual(output['stdout'], 'stdout line 0\nstdout line 1\n')
+            self.assertEqual(output['stderr'], 'stderr line 0\nstderr line 1\n')
+            self.assertEqual(output['exit_code'], 0)
+
+            output_dbs = ActionExecutionOutput.get_all()
+            self.assertEqual(len(output_dbs), (index) * 4)
+
     @mock.patch('st2common.util.concurrency.subprocess_popen')
     def test_stdout_interception_and_parsing(self, mock_popen):
         values = {'delimiter': ACTION_OUTPUT_RESULT_DELIMITER}

scripts/travis/prepare-integration.sh

@@ -15,9 +15,21 @@ source ./virtualenv/bin/activate
 python ./st2client/setup.py develop
 st2 --version
 
+# Clean up old screen log files
+rm -f logs/screen-*.log
+
 # start dev environment in screens
 ./tools/launchdev.sh start -x
 
+# Give processes some time to start and check logs to see if all the services
+# started or if there was any error / failure
+echo "Giving screen processes some time to start..."
+sleep 10
+
+echo " === START: Catting screen process log files. ==="
+cat logs/screen-*.log
+echo " === END: Catting screen process log files. ==="
+
 # This script runs as root on Travis which means other processes which don't run
 # as root can't write to logs/ directory and tests fail
 chmod 777 logs/

st2common/st2common/config.py

@@ -376,7 +376,13 @@ def register_opts(ignore_errors=False):
                  'creates pack virtualenv.'),
         cfg.BoolOpt(
             'stream_output', default=True,
-            help='True to store and stream action output (stdout and stderr) in real-time.')
+            help='True to store and stream action output (stdout and stderr) in real-time.'),
+        cfg.IntOpt(
+            'stream_output_buffer_size', default=-1,
+            help=('Buffer size to use for real time action output streaming. 0 means unbuffered '
+                  '1 means line buffered, -1 means system default, which usually means fully '
+                  'buffered and any other positive value means use a buffer of (approximately) '
+                  'that size'))
     ]
 
     do_register_opts(action_runner_opts, group='actionrunner')

st2common/st2common/util/green/shell.py

@@ -38,7 +38,8 @@
 def run_command(cmd, stdin=None, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=False,
                 cwd=None, env=None, timeout=60, preexec_func=None, kill_func=None,
                 read_stdout_func=None, read_stderr_func=None,
-                read_stdout_buffer=None, read_stderr_buffer=None, stdin_value=None):
+                read_stdout_buffer=None, read_stderr_buffer=None, stdin_value=None,
+                bufsize=0):
     """
     Run the provided command in a subprocess and wait until it completes.
 
@@ -82,6 +83,8 @@ def run_command(cmd, stdin=None, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                  using live read mode.
     :type read_stdout_func: ``func``
 
+    :param bufsize: Buffer size argument to pass to subprocess.popen function.
+    :type bufsize: ``int``
 
     :rtype: ``tuple`` (exit_code, stdout, stderr, timed_out)
     """
@@ -107,7 +110,8 @@ def run_command(cmd, stdin=None, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
     # GreenPipe so it doesn't block
     LOG.debug('Creating subprocess.')
     process = concurrency.subprocess_popen(args=cmd, stdin=stdin, stdout=stdout, stderr=stderr,
-                                           env=env, cwd=cwd, shell=shell, preexec_fn=preexec_func)
+                                           env=env, cwd=cwd, shell=shell, preexec_fn=preexec_func,
+                                           bufsize=bufsize)
 
     if read_stdout_func:
         LOG.debug('Spawning read_stdout_func function')

st2tests/st2tests/resources/packs/pythonactions/actions/pascal_row.py

@@ -36,6 +36,8 @@ def run(self, **kwargs):
 
     @staticmethod
     def _compute_pascal_row(row_index=0):
+        print('Pascal row action')
+
         if row_index == 'a':
             return False, 'This is suppose to fail don\'t worry!!'
         elif row_index == 'b':

st2tests/st2tests/resources/packs/pythonactions/actions/print_to_stdout_and_stderr.py

@@ -0,0 +1,29 @@
+# Copyright 2019 Extreme Networks, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import absolute_import
+
+import sys
+
+from st2common.runners.base_action import Action
+from six.moves import range
+
+
+class PrintToStdoutAndStderrAction(Action):
+    def run(self, stdout_count=3, stderr_count=3):
+        for index in range(0, stdout_count):
+            sys.stdout.write('stdout line %s\n' % (index))
+
+        for index in range(0, stderr_count):
+            sys.stderr.write('stderr line %s\n' % (index))

tools/launchdev.sh

@@ -245,15 +245,18 @@ function st2start(){
         screen -ls | grep mistral | cut -d. -f1 | awk '{print $1}' | xargs kill
     fi
 
+    # NOTE: We can't rely on latest version of screen with "-Logfile path"
+    # option so we need to use screen config file per screen window
+
     # Run the st2 API server
     echo 'Starting screen session st2-api...'
     if [ "${use_gunicorn}" = true ]; then
         echo '  using gunicorn to run st2-api...'
         export ST2_CONFIG_PATH=${ST2_CONF}
-        screen -d -m -S st2-api ${VIRTUALENV}/bin/gunicorn \
+        screen -L -c tools/screen-configs/st2api.conf -d -m -S st2-api ${VIRTUALENV}/bin/gunicorn \
             st2api.wsgi:application -k eventlet -b "$BINDING_ADDRESS:9101" --workers 1
     else
-        screen -d -m -S st2-api ${VIRTUALENV}/bin/python \
+        screen -L -c tools/screen-configs/st2api.conf -d -m -S st2-api ${VIRTUALENV}/bin/python \
             ./st2api/bin/st2api \
             --config-file $ST2_CONF
     fi
@@ -262,10 +265,10 @@ function st2start(){
     if [ "${use_gunicorn}" = true ]; then
         echo '  using gunicorn to run st2-stream'
         export ST2_CONFIG_PATH=${ST2_CONF}
-        screen -d -m -S st2-stream ${VIRTUALENV}/bin/gunicorn \
+        screen -L -c tools/screen-configs/st2stream.conf -d -m -S st2-stream ${VIRTUALENV}/bin/gunicorn \
             st2stream.wsgi:application -k eventlet -b "$BINDING_ADDRESS:9102" --workers 1
     else
-        screen -d -m -S st2-stream ${VIRTUALENV}/bin/python \
+        screen -L -c tools/screen-configs/st2stream.conf -d -m -S st2-stream ${VIRTUALENV}/bin/python \
             ./st2stream/bin/st2stream \
             --config-file $ST2_CONF
     fi
@@ -278,7 +281,7 @@ function st2start(){
         WORKFLOW_ENGINE_NAME=st2-workflow-$i
         WORKFLOW_ENGINE_SCREENS+=($WORKFLOW_ENGINE_NAME)
         echo '  starting '$WORKFLOW_ENGINE_NAME'...'
-        screen -d -m -S $WORKFLOW_ENGINE_NAME ${VIRTUALENV}/bin/python \
+        screen -L -c tools/screen-configs/st2workflowengine.conf -d -m -S $WORKFLOW_ENGINE_NAME ${VIRTUALENV}/bin/python \
             ./st2actions/bin/st2workflowengine \
             --config-file $ST2_CONF
     done
@@ -291,14 +294,14 @@ function st2start(){
         RUNNER_NAME=st2-actionrunner-$i
         RUNNER_SCREENS+=($RUNNER_NAME)
         echo '  starting '$RUNNER_NAME'...'
-        screen -d -m -S $RUNNER_NAME ${VIRTUALENV}/bin/python \
+        screen -L -c tools/screen-configs/st2actionrunner.conf -d -m -S $RUNNER_NAME ${VIRTUALENV}/bin/python \
             ./st2actions/bin/st2actionrunner \
             --config-file $ST2_CONF
     done
 
     # Run the garbage collector service
     echo 'Starting screen session st2-garbagecollector'
-    screen -d -m -S st2-garbagecollector ${VIRTUALENV}/bin/python \
+    screen -L -c tools/screen-configs/st2garbagecollector.conf -d -m -S st2-garbagecollector ${VIRTUALENV}/bin/python \
         ./st2reactor/bin/st2garbagecollector \
         --config-file $ST2_CONF
 
@@ -310,38 +313,38 @@ function st2start(){
         SCHEDULER_NAME=st2-scheduler-$i
         SCHEDULER_SCREENS+=($SCHEDULER_NAME)
         echo '  starting '$SCHEDULER_NAME'...'
-        screen -d -m -S $SCHEDULER_NAME ${VIRTUALENV}/bin/python \
+        screen -L -c tools/screen-configs/st2scheduler.conf -d -m -S $SCHEDULER_NAME ${VIRTUALENV}/bin/python \
             ./st2actions/bin/st2scheduler \
             --config-file $ST2_CONF
     done
 
     # Run the sensor container server
     echo 'Starting screen session st2-sensorcontainer'
-    screen -d -m -S st2-sensorcontainer ${VIRTUALENV}/bin/python \
+    screen -L -c tools/screen-configs/st2sensorcontainer.conf -d -m -S st2-sensorcontainer ${VIRTUALENV}/bin/python \
         ./st2reactor/bin/st2sensorcontainer \
         --config-file $ST2_CONF
 
     # Run the rules engine server
     echo 'Starting screen session st2-rulesengine...'
-    screen -d -m -S st2-rulesengine ${VIRTUALENV}/bin/python \
+    screen -L -c tools/screen-configs/st2rulesengine.conf -d -m -S st2-rulesengine ${VIRTUALENV}/bin/python \
         ./st2reactor/bin/st2rulesengine \
         --config-file $ST2_CONF
 
     # Run the timer engine server
     echo 'Starting screen session st2-timersengine...'
-    screen -d -m -S st2-timersengine ${VIRTUALENV}/bin/python \
+    screen -L -c tools/screen-configs/st2timersengine.conf -d -m -S st2-timersengine ${VIRTUALENV}/bin/python \
         ./st2reactor/bin/st2timersengine \
         --config-file $ST2_CONF
 
     # Run the results tracker
     echo 'Starting screen session st2-resultstracker...'
-    screen -d -m -S st2-resultstracker ${VIRTUALENV}/bin/python \
+    screen -L -c tools/screen-configs/st2resultstracker.conf -d -m -S st2-resultstracker ${VIRTUALENV}/bin/python \
         ./st2actions/bin/st2resultstracker \
         --config-file $ST2_CONF
 
     # Run the actions notifier
     echo 'Starting screen session st2-notifier...'
-    screen -d -m -S st2-notifier ${VIRTUALENV}/bin/python \
+    screen -L -c tools/screen-configs/st2notifier.conf -d -m -S st2-notifier ${VIRTUALENV}/bin/python \
         ./st2actions/bin/st2notifier \
         --config-file $ST2_CONF
 
@@ -350,10 +353,10 @@ function st2start(){
     if [ "${use_gunicorn}" = true ]; then
         echo '  using gunicorn to run st2-auth...'
         export ST2_CONFIG_PATH=${ST2_CONF}
-        screen -d -m -S st2-auth ${VIRTUALENV}/bin/gunicorn \
+        screen -L -c tools/screen-configs/st2auth.conf -d -m -S st2-auth ${VIRTUALENV}/bin/gunicorn \
             st2auth.wsgi:application -k eventlet -b "$BINDING_ADDRESS:9100" --workers 1
     else
-        screen -d -m -S st2-auth ${VIRTUALENV}/bin/python \
+        screen -L -c tools/screen-configs/st2auth.conf -d -m -S st2-auth ${VIRTUALENV}/bin/python \
             ./st2auth/bin/st2auth \
             --config-file $ST2_CONF
     fi
@@ -364,26 +367,25 @@ function st2start(){
         sudo mkdir -p $EXPORTS_DIR
         sudo chown -R ${CURRENT_USER}:${CURRENT_USER_GROUP} $EXPORTS_DIR
         echo 'Starting screen session st2-exporter...'
-        screen -d -m -S st2-exporter ${VIRTUALENV}/bin/python \
+        screen -L -d -m -S st2-exporter ${VIRTUALENV}/bin/python \
             ./st2exporter/bin/st2exporter \
             --config-file $ST2_CONF
     fi
 
     if [ "${include_mistral}" = true ]; then
-
         LOGDIR=${ST2_REPO}/logs
 
         # Run mistral-server
         echo 'Starting screen session mistral-server...'
-        screen -d -m -S mistral-server ${MISTRAL_REPO}/.venv/bin/python \
+        screen -L -Logfile logs/screen-mistral-server.log -d -m -S mistral-server ${MISTRAL_REPO}/.venv/bin/python \
             ${MISTRAL_REPO}/.venv/bin/mistral-server \
             --server engine,executor \
             --config-file $MISTRAL_CONF \
             --log-file "$LOGDIR/mistral-server.log"
 
         # Run mistral-api
         echo 'Starting screen session mistral-api...'
-        screen -d -m -S mistral-api ${MISTRAL_REPO}/.venv/bin/python \
+        screen -L -Logfile logs/screen-mistral-server.log -d -m -S mistral-api ${MISTRAL_REPO}/.venv/bin/python \
             ${MISTRAL_REPO}/.venv/bin/mistral-server \
             --server api \
             --config-file $MISTRAL_CONF \

tools/screen-configs/st2actionrunner.conf

@@ -0,0 +1,6 @@
+logfile logs/screen-st2actionrunner.log
+logfile flush 1
+log on
+logtstamp after 1
+logtstamp string \"[ %t: %Y-%m-%d %c:%s ]\012\"
+logtstamp on

tools/screen-configs/st2api.conf

@@ -0,0 +1,6 @@
+logfile logs/screen-st2api.log
+logfile flush 1
+log on
+logtstamp after 1
+logtstamp string \"[ %t: %Y-%m-%d %c:%s ]\012\"
+logtstamp on

tools/screen-configs/st2auth.conf

@@ -0,0 +1,6 @@
+logfile logs/screen-st2auth.log
+logfile flush 1
+log on
+logtstamp after 1
+logtstamp string \"[ %t: %Y-%m-%d %c:%s ]\012\"
+logtstamp on

tools/screen-configs/st2garbagecollector.conf

@@ -0,0 +1,6 @@
+logfile logs/screen-st2garbagecollector.log
+logfile flush 1
+log on
+logtstamp after 1
+logtstamp string \"[ %t: %Y-%m-%d %c:%s ]\012\"
+logtstamp on

tools/screen-configs/st2notifier.conf

@@ -0,0 +1,6 @@
+logfile logs/screen-st2notifier.log
+logfile flush 1
+log on
+logtstamp after 1
+logtstamp string \"[ %t: %Y-%m-%d %c:%s ]\012\"
+logtstamp on

tools/screen-configs/st2resultstracker.conf

@@ -0,0 +1,6 @@
+logfile logs/screen-st2resultstracker.log
+logfile flush 1
+log on
+logtstamp after 1
+logtstamp string \"[ %t: %Y-%m-%d %c:%s ]\012\"
+logtstamp on

tools/screen-configs/st2rulesengine.conf

@@ -0,0 +1,6 @@
+logfile logs/screen-st2rulesengine.log
+logfile flush 1
+log on
+logtstamp after 1
+logtstamp string \"[ %t: %Y-%m-%d %c:%s ]\012\"
+logtstamp on

tools/screen-configs/st2scheduler.conf

@@ -0,0 +1,6 @@
+logfile logs/screen-st2scheduler.log
+logfile flush 1
+log on
+logtstamp after 1
+logtstamp string \"[ %t: %Y-%m-%d %c:%s ]\012\"
+logtstamp on