diff --git a/ansible_mitogen/actions/mitogen_async_status.py b/ansible_mitogen/actions/mitogen_async_status.py new file mode 100644 index 00000000..03c7cad8 --- /dev/null +++ b/ansible_mitogen/actions/mitogen_async_status.py @@ -0,0 +1,64 @@ +# Copyright 2017, David Wilson +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# +# 1. Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# +# 2. Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# +# 3. Neither the name of the copyright holder nor the names of its contributors +# may be used to endorse or promote products derived from this software without +# specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. + +import ansible.plugins.action +import mitogen.core +import ansible_mitogen.helpers + + +class ActionModule(ansible.plugins.action.ActionBase): + def run(self, tmp=None, task_vars=None): + job_id = self._task.args['jid'] + try: + result = self._connection.call( + ansible_mitogen.helpers.get_async_result, + job_id, + ) + except mitogen.core.CallError, e: + return { + 'ansible_job_id': job_id, + 'started': 1, + 'failed': 1, + 'finished': 1, + 'msg': str(e), + } + + if result is None: + return { + 'ansible_job_id': job_id, + 'started': 1, + 'failed': 0, + 'finished': 0, + 'msg': '', + } + + dct = self._parse_returned_data({'stdout': result}) + dct['ansible_job_id'] = job_id + dct['started'] = 1 + dct['finished'] = 1 + return dct diff --git a/ansible_mitogen/helpers.py b/ansible_mitogen/helpers.py index b1dcfe8f..df2b32f7 100644 --- a/ansible_mitogen/helpers.py +++ b/ansible_mitogen/helpers.py @@ -30,15 +30,23 @@ import json import operator import os import pwd +import random import re import stat import subprocess +import threading import time # Prevent accidental import of an Ansible module from hanging on stdin read. import ansible.module_utils.basic ansible.module_utils.basic._ANSIBLE_ARGS = '{}' +#: Mapping of job_id<->result dict +_result_by_job_id = {} + +#: Mapping of job_id<->threading.Thread +_thread_by_job_id = {} + class Exit(Exception): """ @@ -118,6 +126,57 @@ def run_module(module, raw_params=None, args=None): return json.dumps(e.dct) +def _async_main(job_id, module, raw_params, args): + """ + Implementation for the thread that implements asynchronous module + execution. + """ + try: + rc = run_module(module, raw_params, args) + except Exception, e: + rc = mitogen.core.CallError(e) + + _result_by_job_id[job_id] = rc + + +def run_module_async(module, raw_params=None, args=None): + """ + Arrange for an Ansible module to be executed in a thread of the current + process, with results available via :py:func:`get_async_result`. + """ + job_id = '%08x' % random.randint(0, 2**32-1) + _result_by_job_id[job_id] = None + _thread_by_job_id[job_id] = threading.Thread( + target=_async_main, + kwargs={ + 'job_id': job_id, + 'module': module, + 'raw_params': raw_params, + 'args': args, + } + ) + _thread_by_job_id[job_id].start() + return json.dumps({ + 'ansible_job_id': job_id, + 'changed': True + }) + + +def get_async_result(job_id): + """ + Poll for the result of an asynchronous task. + + :param str job_id: + Job ID to poll for. + :returns: + ``None`` if job is still running, JSON-encoded result dictionary if + execution completed normally, or :py:class:`mitogen.core.CallError` if + an exception was thrown. + """ + if not _thread_by_job_id[job_id].isAlive(): + return _result_by_job_id[job_id] + + def get_user_shell(): """ For commands executed directly via an SSH command-line, SSH looks up the @@ -192,6 +251,10 @@ CHMOD_BITS = { def apply_mode_spec(spec, mode): + """ + Given a symbolic file mode change specification in the style of chmod(1) + `spec`, apply changes in the specification to the numeric file mode `mode`. + """ for clause in spec.split(','): match = CHMOD_CLAUSE_PAT.match(clause) who, op, perms = match.groups() diff --git a/ansible_mitogen/mixins.py b/ansible_mitogen/mixins.py index 7d13b2fd..f4a0468f 100644 --- a/ansible_mitogen/mixins.py +++ b/ansible_mitogen/mixins.py @@ -152,12 +152,16 @@ class ActionModuleMixin(ansible.plugins.action.ActionBase): task_vars = task_vars or {} self._update_module_args(module_name, module_args, task_vars) + if wrap_async: + helper = ansible_mitogen.helpers.run_module_async + else: + helper = ansible_mitogen.helpers.run_module # replaces 110 lines js = self.call( - ansible_mitogen.helpers.run_module, + helper, get_command_module_name(module_name), - args=cast(module_args) + args=cast(module_args), ) data = self._parse_returned_data({ @@ -167,9 +171,6 @@ class ActionModuleMixin(ansible.plugins.action.ActionBase): 'stderr': '' }) - if wrap_async: - data['changed'] = True - # pre-split stdout/stderr into lines if needed if 'stdout' in data and 'stdout_lines' not in data: # if the value is 'False', a default won't catch it. diff --git a/ansible_mitogen/strategy/mitogen.py b/ansible_mitogen/strategy/mitogen.py index 765367c1..0260fdc1 100644 --- a/ansible_mitogen/strategy/mitogen.py +++ b/ansible_mitogen/strategy/mitogen.py @@ -50,14 +50,22 @@ except ImportError: # Ansible <2.4 def wrap_action_loader__get(name, *args, **kwargs): """ - While the mitogen stratey is active, trap action_loader.get() calls, + While the mitogen strategy is active, trap action_loader.get() calls, augmenting any fetched class with ActionModuleMixin, which replaces various helper methods inherited from ActionBase with implementations that avoid the use of shell fragments wherever possible. + Additionally catch attempts to instantiate the "normal" action with a task + argument whose action is "async_status", and redirect it to a special + implementation that fetches polls the task result via RPC. + This is used instead of static subclassing as it generalizes to third party action modules outside the Ansible tree. """ + if ( name == 'normal' and 'task' in kwargs and + kwargs['task'].action == 'async_status'): + name = 'mitogen_async_status' + klass = action_loader__get(name, class_only=True) if klass: wrapped_name = 'MitogenActionModule_' + name @@ -243,10 +251,9 @@ class StrategyModule(ansible.plugins.strategy.linear.StrategyModule): Add the mitogen connection plug-in directory to the ModuleLoader path, avoiding the need for manual configuration. """ - # ansible_mitogen base directory: basedir = os.path.dirname(os.path.dirname(__file__)) - conn_dir = os.path.join(basedir, 'connection') - connection_loader.add_directory(conn_dir) + connection_loader.add_directory(os.path.join(basedir, 'connection')) + action_loader.add_directory(os.path.join(basedir, 'actions')) def run(self, iterator, play_context, result=0): self._add_connection_plugin_path() diff --git a/docs/ansible.rst b/docs/ansible.rst index 489341e7..976eccfe 100644 --- a/docs/ansible.rst +++ b/docs/ansible.rst @@ -119,6 +119,10 @@ Behavioural Differences connection to host closed`` to appear in ``stderr`` output of every executed command. This never manifests with the Mitogen extension. +* Asynchronous jobs execute in a thread of the single target Python + interpreter. In future this will be replaced with subprocesses, as it's + likely some use cases spawn many asynchronous jobs. + Configuration -------------