You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
mitogen/ansible_mitogen/services.py

560 lines
20 KiB
Python

# Copyright 2019, 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.
# !mitogen: minify_safe
"""
Classes in this file define Mitogen 'services' that run (initially) within the
connection multiplexer process that is forked off the top-level controller
process.
Once a worker process connects to a multiplexer process
(Connection._connect()), it communicates with these services to establish new
connections, grant access to files by children, and register for notification
when a child has completed a job.
"""
from __future__ import absolute_import
from __future__ import unicode_literals
import logging
import os
import os.path
import sys
ansible: enable forking when requested and for async jobs. Closes #105. References #155. mitogen/service.py: Refactor services to support individually exposed methods with different security policies for each method. - @mitogen.service.expose() to expose a method and set its policy - @mitogen.service.arg_spec() to validate input. - Require basic service message format to be a tuple of `(method, kwargs)`, where kwargs is always a dict. - Update DeduplicatingService to match the new scheme. ansible_mitogen/connection.py: - Rename 'method' to 'method_name' to disambiguate it from the service.call()'s method= argument. ansible_mitogen/planner.py: - Generate an ID for every job, sync or not, and fetch job results from JobResultService rather than via the initiating function call's return value. - Planner subclasses now get to select whether their Runner should run in a forked process. The base implementation requests this if the 'mitogen_isolation_mode=fork' task variable is present. ansible_mitogen/runner.py: Teach runners to deliver their result via JobResultService executing in their indirect parent mux process. ansible_mitogen/plugins/actions/mitogen_async_status.py: Split the implementation up into methods, and more compatibly emulate Ansible's existing output. ansible_mitogen/process.py: Mux processes now host JobResultService. ansible_mitogen/services.py: Update existing services to the new mitogen.service scheme, and implement JobResultService: * listen() method for synchronous jobs. planner.invoke() registers a Sender with the service prior to invoking the job, then sleeps waiting for the service to write the job result to the corresponding Receiver. * Non-blocking get() method for implementing mitogen_async_status action. * Child-accessible push() method for delivering task results. ansible_mitogen/target.py: New helpers for spawning a virginal subprocess on startup, from which asynchronous and mitogen_task_isolation=fork jobs are forked. Necessary to avoid a task inheriting potentially polluted/monkey-patched parent environment, since remaining jobs continue to run in the original child process. docs/ansible.rst: Add/merge/remove some behaviours/risks. tests/ansible/integration: New tests for forking/async.
7 years ago
import threading
import ansible.constants
import mitogen
import mitogen.service
import mitogen.utils
import ansible_mitogen.loaders
import ansible_mitogen.module_finder
ansible: enable forking when requested and for async jobs. Closes #105. References #155. mitogen/service.py: Refactor services to support individually exposed methods with different security policies for each method. - @mitogen.service.expose() to expose a method and set its policy - @mitogen.service.arg_spec() to validate input. - Require basic service message format to be a tuple of `(method, kwargs)`, where kwargs is always a dict. - Update DeduplicatingService to match the new scheme. ansible_mitogen/connection.py: - Rename 'method' to 'method_name' to disambiguate it from the service.call()'s method= argument. ansible_mitogen/planner.py: - Generate an ID for every job, sync or not, and fetch job results from JobResultService rather than via the initiating function call's return value. - Planner subclasses now get to select whether their Runner should run in a forked process. The base implementation requests this if the 'mitogen_isolation_mode=fork' task variable is present. ansible_mitogen/runner.py: Teach runners to deliver their result via JobResultService executing in their indirect parent mux process. ansible_mitogen/plugins/actions/mitogen_async_status.py: Split the implementation up into methods, and more compatibly emulate Ansible's existing output. ansible_mitogen/process.py: Mux processes now host JobResultService. ansible_mitogen/services.py: Update existing services to the new mitogen.service scheme, and implement JobResultService: * listen() method for synchronous jobs. planner.invoke() registers a Sender with the service prior to invoking the job, then sleeps waiting for the service to write the job result to the corresponding Receiver. * Non-blocking get() method for implementing mitogen_async_status action. * Child-accessible push() method for delivering task results. ansible_mitogen/target.py: New helpers for spawning a virginal subprocess on startup, from which asynchronous and mitogen_task_isolation=fork jobs are forked. Necessary to avoid a task inheriting potentially polluted/monkey-patched parent environment, since remaining jobs continue to run in the original child process. docs/ansible.rst: Add/merge/remove some behaviours/risks. tests/ansible/integration: New tests for forking/async.
7 years ago
import ansible_mitogen.target
LOG = logging.getLogger(__name__)
# Force load of plugin to ensure ConfigManager has definitions loaded. Done
# during module import to ensure a single-threaded environment; PluginLoader
# is not thread-safe.
ansible_mitogen.loaders.shell_loader.get('sh')
if sys.version_info[0] == 3:
def reraise(tp, value, tb):
if value is None:
value = tp()
if value.__traceback__ is not tb:
raise value.with_traceback(tb)
raise value
else:
exec(
"def reraise(tp, value, tb=None):\n"
" raise tp, value, tb\n"
)
def _get_candidate_temp_dirs():
try:
# >=2.5
options = ansible.constants.config.get_plugin_options('shell', 'sh')
remote_tmp = options.get('remote_tmp') or ansible.constants.DEFAULT_REMOTE_TMP
system_tmpdirs = options.get('system_tmpdirs', ('/var/tmp', '/tmp'))
except AttributeError:
# 2.3
remote_tmp = ansible.constants.DEFAULT_REMOTE_TMP
system_tmpdirs = ('/var/tmp', '/tmp')
return mitogen.utils.cast([remote_tmp] + list(system_tmpdirs))
def key_from_dict(**kwargs):
"""
Return a unique string representation of a dict as quickly as possible.
Used to generated deduplication keys from a request.
"""
out = []
stack = [kwargs]
while stack:
obj = stack.pop()
if isinstance(obj, dict):
stack.extend(sorted(obj.items()))
elif isinstance(obj, (list, tuple)):
stack.extend(obj)
else:
out.append(str(obj))
return ''.join(out)
ansible: enable forking when requested and for async jobs. Closes #105. References #155. mitogen/service.py: Refactor services to support individually exposed methods with different security policies for each method. - @mitogen.service.expose() to expose a method and set its policy - @mitogen.service.arg_spec() to validate input. - Require basic service message format to be a tuple of `(method, kwargs)`, where kwargs is always a dict. - Update DeduplicatingService to match the new scheme. ansible_mitogen/connection.py: - Rename 'method' to 'method_name' to disambiguate it from the service.call()'s method= argument. ansible_mitogen/planner.py: - Generate an ID for every job, sync or not, and fetch job results from JobResultService rather than via the initiating function call's return value. - Planner subclasses now get to select whether their Runner should run in a forked process. The base implementation requests this if the 'mitogen_isolation_mode=fork' task variable is present. ansible_mitogen/runner.py: Teach runners to deliver their result via JobResultService executing in their indirect parent mux process. ansible_mitogen/plugins/actions/mitogen_async_status.py: Split the implementation up into methods, and more compatibly emulate Ansible's existing output. ansible_mitogen/process.py: Mux processes now host JobResultService. ansible_mitogen/services.py: Update existing services to the new mitogen.service scheme, and implement JobResultService: * listen() method for synchronous jobs. planner.invoke() registers a Sender with the service prior to invoking the job, then sleeps waiting for the service to write the job result to the corresponding Receiver. * Non-blocking get() method for implementing mitogen_async_status action. * Child-accessible push() method for delivering task results. ansible_mitogen/target.py: New helpers for spawning a virginal subprocess on startup, from which asynchronous and mitogen_task_isolation=fork jobs are forked. Necessary to avoid a task inheriting potentially polluted/monkey-patched parent environment, since remaining jobs continue to run in the original child process. docs/ansible.rst: Add/merge/remove some behaviours/risks. tests/ansible/integration: New tests for forking/async.
7 years ago
class Error(Exception):
pass
class ContextService(mitogen.service.Service):
"""
Used by workers to fetch the single Context instance corresponding to a
connection configuration, creating the matching connection if it does not
exist.
For connection methods and their parameters, see:
https://mitogen.readthedocs.io/en/latest/api.html#context-factories
This concentrates connections in the top-level process, which may become a
bottleneck. The bottleneck can be removed using per-CPU connection
processes and arranging for the worker to select one according to a hash of
the connection parameters (sharding).
"""
max_interpreters = int(os.getenv('MITOGEN_MAX_INTERPRETERS', '20'))
def __init__(self, *args, **kwargs):
super(ContextService, self).__init__(*args, **kwargs)
self._lock = threading.Lock()
#: Records the :meth:`get` result dict for successful calls, returned
#: for identical subsequent calls. Keyed by :meth:`key_from_dict`.
self._response_by_key = {}
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
#: List of :class:`mitogen.core.Latch` awaiting the result for a
#: particular key.
self._latches_by_key = {}
#: Mapping of :class:`mitogen.core.Context` -> reference count. Each
#: call to :meth:`get` increases this by one. Calls to :meth:`put`
#: decrease it by one.
self._refs_by_context = {}
#: List of contexts in creation order by via= parameter. When
#: :attr:`max_interpreters` is reached, the most recently used context
#: is destroyed to make room for any additional context.
self._lru_by_via = {}
#: :func:`key_from_dict` result by Context.
self._key_by_context = {}
#: Mapping of Context -> parent Context
self._via_by_context = {}
@mitogen.service.expose(mitogen.service.AllowParents())
@mitogen.service.arg_spec({
'stack': list,
})
def reset(self, stack):
"""
Return a reference, forcing close and discard of the underlying
connection. Used for 'meta: reset_connection' or when some other error
is detected.
:returns:
:data:`True` if a connection was found to discard, otherwise
:data:`False`.
"""
LOG.debug('%r.reset(%r)', self, stack)
l = mitogen.core.Latch()
context = None
with self._lock:
for i, spec in enumerate(stack):
key = key_from_dict(via=context, **spec)
response = self._response_by_key.get(key)
if response is None:
LOG.debug('%r: could not find connection to shut down; '
'failed at hop %d', self, i)
return False
context = response['context']
mitogen.core.listen(context, 'disconnect', l.put)
self._shutdown_unlocked(context)
# The timeout below is to turn a hang into a crash in case there is any
# possible race between 'disconnect' signal subscription, and the child
# abruptly disconnecting.
l.get(timeout=30.0)
return True
ansible: enable forking when requested and for async jobs. Closes #105. References #155. mitogen/service.py: Refactor services to support individually exposed methods with different security policies for each method. - @mitogen.service.expose() to expose a method and set its policy - @mitogen.service.arg_spec() to validate input. - Require basic service message format to be a tuple of `(method, kwargs)`, where kwargs is always a dict. - Update DeduplicatingService to match the new scheme. ansible_mitogen/connection.py: - Rename 'method' to 'method_name' to disambiguate it from the service.call()'s method= argument. ansible_mitogen/planner.py: - Generate an ID for every job, sync or not, and fetch job results from JobResultService rather than via the initiating function call's return value. - Planner subclasses now get to select whether their Runner should run in a forked process. The base implementation requests this if the 'mitogen_isolation_mode=fork' task variable is present. ansible_mitogen/runner.py: Teach runners to deliver their result via JobResultService executing in their indirect parent mux process. ansible_mitogen/plugins/actions/mitogen_async_status.py: Split the implementation up into methods, and more compatibly emulate Ansible's existing output. ansible_mitogen/process.py: Mux processes now host JobResultService. ansible_mitogen/services.py: Update existing services to the new mitogen.service scheme, and implement JobResultService: * listen() method for synchronous jobs. planner.invoke() registers a Sender with the service prior to invoking the job, then sleeps waiting for the service to write the job result to the corresponding Receiver. * Non-blocking get() method for implementing mitogen_async_status action. * Child-accessible push() method for delivering task results. ansible_mitogen/target.py: New helpers for spawning a virginal subprocess on startup, from which asynchronous and mitogen_task_isolation=fork jobs are forked. Necessary to avoid a task inheriting potentially polluted/monkey-patched parent environment, since remaining jobs continue to run in the original child process. docs/ansible.rst: Add/merge/remove some behaviours/risks. tests/ansible/integration: New tests for forking/async.
7 years ago
@mitogen.service.expose(mitogen.service.AllowParents())
@mitogen.service.arg_spec({
'context': mitogen.core.Context
ansible: enable forking when requested and for async jobs. Closes #105. References #155. mitogen/service.py: Refactor services to support individually exposed methods with different security policies for each method. - @mitogen.service.expose() to expose a method and set its policy - @mitogen.service.arg_spec() to validate input. - Require basic service message format to be a tuple of `(method, kwargs)`, where kwargs is always a dict. - Update DeduplicatingService to match the new scheme. ansible_mitogen/connection.py: - Rename 'method' to 'method_name' to disambiguate it from the service.call()'s method= argument. ansible_mitogen/planner.py: - Generate an ID for every job, sync or not, and fetch job results from JobResultService rather than via the initiating function call's return value. - Planner subclasses now get to select whether their Runner should run in a forked process. The base implementation requests this if the 'mitogen_isolation_mode=fork' task variable is present. ansible_mitogen/runner.py: Teach runners to deliver their result via JobResultService executing in their indirect parent mux process. ansible_mitogen/plugins/actions/mitogen_async_status.py: Split the implementation up into methods, and more compatibly emulate Ansible's existing output. ansible_mitogen/process.py: Mux processes now host JobResultService. ansible_mitogen/services.py: Update existing services to the new mitogen.service scheme, and implement JobResultService: * listen() method for synchronous jobs. planner.invoke() registers a Sender with the service prior to invoking the job, then sleeps waiting for the service to write the job result to the corresponding Receiver. * Non-blocking get() method for implementing mitogen_async_status action. * Child-accessible push() method for delivering task results. ansible_mitogen/target.py: New helpers for spawning a virginal subprocess on startup, from which asynchronous and mitogen_task_isolation=fork jobs are forked. Necessary to avoid a task inheriting potentially polluted/monkey-patched parent environment, since remaining jobs continue to run in the original child process. docs/ansible.rst: Add/merge/remove some behaviours/risks. tests/ansible/integration: New tests for forking/async.
7 years ago
})
def put(self, context):
"""
Return a reference, making it eligable for recycling once its reference
count reaches zero.
"""
LOG.debug('decrementing reference count for %r', context)
self._lock.acquire()
try:
if self._refs_by_context.get(context, 0) == 0:
LOG.warning('%r.put(%r): refcount was 0. shutdown_all called?',
self, context)
return
self._refs_by_context[context] -= 1
finally:
self._lock.release()
def _produce_response(self, key, response):
"""
Reply to every waiting request matching a configuration key with a
response dictionary, deleting the list of waiters when done.
:param str key:
Result of :meth:`key_from_dict`
:param dict response:
Response dictionary
:returns:
Number of waiters that were replied to.
"""
self._lock.acquire()
try:
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
latches = self._latches_by_key.pop(key)
count = len(latches)
for latch in latches:
latch.put(response)
finally:
self._lock.release()
return count
def _forget_context_unlocked(self, context):
key = self._key_by_context.get(context)
if key is None:
LOG.debug('%r: attempt to forget unknown %r', self, context)
return
self._response_by_key.pop(key, None)
self._latches_by_key.pop(key, None)
self._key_by_context.pop(context, None)
self._refs_by_context.pop(context, None)
self._via_by_context.pop(context, None)
self._lru_by_via.pop(context, None)
def _shutdown_unlocked(self, context, lru=None, new_context=None):
"""
Arrange for `context` to be shut down, and optionally add `new_context`
to the LRU list while holding the lock.
"""
LOG.info('%r._shutdown_unlocked(): shutting down %r', self, context)
context.shutdown()
via = self._via_by_context.get(context)
if via:
lru = self._lru_by_via.get(via)
if lru:
if context in lru:
lru.remove(context)
if new_context:
lru.append(new_context)
self._forget_context_unlocked(context)
def _update_lru_unlocked(self, new_context, spec, via):
"""
Update the LRU ("MRU"?) list associated with the connection described
by `kwargs`, destroying the most recently created context if the list
is full. Finally add `new_context` to the list.
"""
self._via_by_context[new_context] = via
lru = self._lru_by_via.setdefault(via, [])
if len(lru) < self.max_interpreters:
lru.append(new_context)
return
for context in reversed(lru):
if self._refs_by_context[context] == 0:
break
else:
LOG.warning('via=%r reached maximum number of interpreters, '
'but they are all marked as in-use.', via)
return
self._shutdown_unlocked(context, lru=lru, new_context=new_context)
def _update_lru(self, new_context, spec, via):
self._lock.acquire()
try:
self._update_lru_unlocked(new_context, spec, via)
finally:
self._lock.release()
@mitogen.service.expose(mitogen.service.AllowParents())
def dump(self):
"""
For testing, return a list of dicts describing every currently
connected context.
"""
return [
{
'context_name': context.name,
'via': getattr(self._via_by_context.get(context),
'name', None),
'refs': self._refs_by_context.get(context),
}
for context, key in sorted(self._key_by_context.items(),
key=lambda c_k: c_k[0].context_id)
]
@mitogen.service.expose(mitogen.service.AllowParents())
def shutdown_all(self):
"""
For testing use, arrange for all connections to be shut down.
"""
self._lock.acquire()
try:
for context in list(self._key_by_context):
self._shutdown_unlocked(context)
finally:
self._lock.release()
def _on_context_disconnect(self, context):
"""
Respond to Context disconnect event by deleting any record of the no
longer reachable context. This method runs in the Broker thread and
must not to block.
"""
self._lock.acquire()
try:
LOG.info('%r: Forgetting %r due to stream disconnect', self, context)
self._forget_context_unlocked(context)
finally:
self._lock.release()
ALWAYS_PRELOAD = (
'ansible.module_utils.basic',
'ansible.module_utils.json_utils',
'ansible.release',
'ansible_mitogen.runner',
'ansible_mitogen.target',
'mitogen.fork',
'mitogen.service',
)
def _send_module_forwards(self, context):
if hasattr(self.router.responder, 'forward_modules'):
self.router.responder.forward_modules(context, self.ALWAYS_PRELOAD)
_candidate_temp_dirs = None
def _get_candidate_temp_dirs(self):
"""
Return a list of locations to try to create the single temporary
directory used by the run. This simply caches the (expensive) plugin
load of :func:`_get_candidate_temp_dirs`.
"""
if self._candidate_temp_dirs is None:
self._candidate_temp_dirs = _get_candidate_temp_dirs()
return self._candidate_temp_dirs
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
def _connect(self, key, spec, via=None):
"""
Actual connect implementation. Arranges for the Mitogen connection to
be created and enqueues an asynchronous call to start the forked task
parent in the remote context.
:param key:
Deduplication key representing the connection configuration.
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
:param spec:
Connection specification.
:returns:
Dict like::
{
'context': mitogen.core.Context or None,
'via': mitogen.core.Context or None,
issue #186: rework async/forked tasks again. The controller must know the ID of the forked child in order to propagate dependencies to it, so forking+starting the module run cannot happen entirely on the target, without some additional mechanism to wait-and-repropagate the deps as they arrive on the target. Rework things so that init_child() also handles starting the fork parent, and returns it along with the context's home directory in a single round trip. Now master knows the identity of the fork parent, it can directly create fork children and call run_module_async() in them. This necessitates 2 roundtrips to start an asynchronous task. This whole thing sucks and entirely needs simplified, but for now things almost work, so keeping it. connection.py: * Expect ContextService to return the entire dict return value of init_child(). Store the fork_contxt from the return value. planner.py: * Rework Planner to store the invocation as an instance attribute, to simplify method calls. * Add Planner.get_push_files() and Planner.get_module_deps(). * Add _propagate_deps() which takes a Planner and ensures the deps it describes are sent to a (non forked or forked) context. * Move async task logic out of target.py and into invoke() / _invoke_*(). process.py: * Services no longer need references to each other. planner.py handles sending module deps with one extra RPC. services.py: * Return "init_child_result" key instead of simple "home_dir" key. * Get rid of dep propagation from ModuleDepService, it lives in planner.py now. target.py: * Get rid of async task start logic, lives in planner.py now.
6 years ago
'init_child_result': {
'fork_context': mitogen.core.Context,
'home_dir': str or None,
},
'msg': str or None
}
issue #186: rework async/forked tasks again. The controller must know the ID of the forked child in order to propagate dependencies to it, so forking+starting the module run cannot happen entirely on the target, without some additional mechanism to wait-and-repropagate the deps as they arrive on the target. Rework things so that init_child() also handles starting the fork parent, and returns it along with the context's home directory in a single round trip. Now master knows the identity of the fork parent, it can directly create fork children and call run_module_async() in them. This necessitates 2 roundtrips to start an asynchronous task. This whole thing sucks and entirely needs simplified, but for now things almost work, so keeping it. connection.py: * Expect ContextService to return the entire dict return value of init_child(). Store the fork_contxt from the return value. planner.py: * Rework Planner to store the invocation as an instance attribute, to simplify method calls. * Add Planner.get_push_files() and Planner.get_module_deps(). * Add _propagate_deps() which takes a Planner and ensures the deps it describes are sent to a (non forked or forked) context. * Move async task logic out of target.py and into invoke() / _invoke_*(). process.py: * Services no longer need references to each other. planner.py handles sending module deps with one extra RPC. services.py: * Return "init_child_result" key instead of simple "home_dir" key. * Get rid of dep propagation from ModuleDepService, it lives in planner.py now. target.py: * Get rid of async task start logic, lives in planner.py now.
6 years ago
Where `context` is a reference to the newly constructed context,
`init_child_result` is the result of executing
:func:`ansible_mitogen.target.init_child` in that context, `msg` is
an error message and the remaining fields are :data:`None`, or
`msg` is :data:`None` and the remaining fields are set.
"""
try:
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
method = getattr(self.router, spec['method'])
except AttributeError:
raise Error('unsupported method: %(method)s' % spec)
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
context = method(via=via, unidirectional=True, **spec['kwargs'])
if via and spec.get('enable_lru'):
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
self._update_lru(context, spec, via)
# Forget the context when its disconnect event fires.
mitogen.core.listen(context, 'disconnect',
lambda: self._on_context_disconnect(context))
self._send_module_forwards(context)
init_child_result = context.call(
ansible_mitogen.target.init_child,
log_level=LOG.getEffectiveLevel(),
candidate_temp_dirs=self._get_candidate_temp_dirs(),
)
if os.environ.get('MITOGEN_DUMP_THREAD_STACKS'):
from mitogen import debug
context.call(debug.dump_to_logger)
self._key_by_context[context] = key
self._refs_by_context[context] = 0
return {
'context': context,
'via': via,
issue #186: rework async/forked tasks again. The controller must know the ID of the forked child in order to propagate dependencies to it, so forking+starting the module run cannot happen entirely on the target, without some additional mechanism to wait-and-repropagate the deps as they arrive on the target. Rework things so that init_child() also handles starting the fork parent, and returns it along with the context's home directory in a single round trip. Now master knows the identity of the fork parent, it can directly create fork children and call run_module_async() in them. This necessitates 2 roundtrips to start an asynchronous task. This whole thing sucks and entirely needs simplified, but for now things almost work, so keeping it. connection.py: * Expect ContextService to return the entire dict return value of init_child(). Store the fork_contxt from the return value. planner.py: * Rework Planner to store the invocation as an instance attribute, to simplify method calls. * Add Planner.get_push_files() and Planner.get_module_deps(). * Add _propagate_deps() which takes a Planner and ensures the deps it describes are sent to a (non forked or forked) context. * Move async task logic out of target.py and into invoke() / _invoke_*(). process.py: * Services no longer need references to each other. planner.py handles sending module deps with one extra RPC. services.py: * Return "init_child_result" key instead of simple "home_dir" key. * Get rid of dep propagation from ModuleDepService, it lives in planner.py now. target.py: * Get rid of async task start logic, lives in planner.py now.
6 years ago
'init_child_result': init_child_result,
'msg': None,
}
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
def _wait_or_start(self, spec, via=None):
latch = mitogen.core.Latch()
key = key_from_dict(via=via, **spec)
self._lock.acquire()
try:
response = self._response_by_key.get(key)
if response is not None:
self._refs_by_context[response['context']] += 1
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
latch.put(response)
return latch
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
latches = self._latches_by_key.setdefault(key, [])
first = len(latches) == 0
latches.append(latch)
finally:
self._lock.release()
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
if first:
# I'm the first requestee, so I will create the connection.
try:
response = self._connect(key, spec, via=via)
count = self._produce_response(key, response)
# Only record the response for non-error results.
self._response_by_key[key] = response
# Set the reference count to the number of waiters.
self._refs_by_context[response['context']] += count
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
except Exception:
self._produce_response(key, sys.exc_info())
return latch
disconnect_msg = (
'Channel was disconnected while connection attempt was in progress; '
'this may be caused by an abnormal Ansible exit, or due to an '
'unreliable target.'
)
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
@mitogen.service.expose(mitogen.service.AllowParents())
@mitogen.service.arg_spec({
'stack': list
})
def get(self, stack):
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
"""
Return a Context referring to an established connection with the given
configuration, establishing new connections as necessary.
:param list stack:
Connection descriptions. Each element is a dict containing 'method'
and 'kwargs' keys describing the Router method and arguments.
Subsequent elements are proxied via the previous.
:returns dict:
* context: mitogen.parent.Context or None.
issue #186: rework async/forked tasks again. The controller must know the ID of the forked child in order to propagate dependencies to it, so forking+starting the module run cannot happen entirely on the target, without some additional mechanism to wait-and-repropagate the deps as they arrive on the target. Rework things so that init_child() also handles starting the fork parent, and returns it along with the context's home directory in a single round trip. Now master knows the identity of the fork parent, it can directly create fork children and call run_module_async() in them. This necessitates 2 roundtrips to start an asynchronous task. This whole thing sucks and entirely needs simplified, but for now things almost work, so keeping it. connection.py: * Expect ContextService to return the entire dict return value of init_child(). Store the fork_contxt from the return value. planner.py: * Rework Planner to store the invocation as an instance attribute, to simplify method calls. * Add Planner.get_push_files() and Planner.get_module_deps(). * Add _propagate_deps() which takes a Planner and ensures the deps it describes are sent to a (non forked or forked) context. * Move async task logic out of target.py and into invoke() / _invoke_*(). process.py: * Services no longer need references to each other. planner.py handles sending module deps with one extra RPC. services.py: * Return "init_child_result" key instead of simple "home_dir" key. * Get rid of dep propagation from ModuleDepService, it lives in planner.py now. target.py: * Get rid of async task start logic, lives in planner.py now.
6 years ago
* init_child_result: Result of :func:`init_child`.
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
* msg: StreamError exception text or None.
* method_name: string failing method name.
"""
via = None
for spec in stack:
try:
result = self._wait_or_start(spec, via=via).get()
if isinstance(result, tuple): # exc_info()
reraise(*result)
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
via = result['context']
except mitogen.core.ChannelError:
return {
'context': None,
'init_child_result': None,
'method_name': spec['method'],
'msg': self.disconnect_msg,
}
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
except mitogen.core.StreamError as e:
return {
'context': None,
issue #186: rework async/forked tasks again. The controller must know the ID of the forked child in order to propagate dependencies to it, so forking+starting the module run cannot happen entirely on the target, without some additional mechanism to wait-and-repropagate the deps as they arrive on the target. Rework things so that init_child() also handles starting the fork parent, and returns it along with the context's home directory in a single round trip. Now master knows the identity of the fork parent, it can directly create fork children and call run_module_async() in them. This necessitates 2 roundtrips to start an asynchronous task. This whole thing sucks and entirely needs simplified, but for now things almost work, so keeping it. connection.py: * Expect ContextService to return the entire dict return value of init_child(). Store the fork_contxt from the return value. planner.py: * Rework Planner to store the invocation as an instance attribute, to simplify method calls. * Add Planner.get_push_files() and Planner.get_module_deps(). * Add _propagate_deps() which takes a Planner and ensures the deps it describes are sent to a (non forked or forked) context. * Move async task logic out of target.py and into invoke() / _invoke_*(). process.py: * Services no longer need references to each other. planner.py handles sending module deps with one extra RPC. services.py: * Return "init_child_result" key instead of simple "home_dir" key. * Get rid of dep propagation from ModuleDepService, it lives in planner.py now. target.py: * Get rid of async task start logic, lives in planner.py now.
6 years ago
'init_child_result': None,
ansible: connection delegation v1 This implements the first edition of Connection Delegation, where delegating connection establishment is initially single-threaded. ansible_mitogen/strategy.py: ansible_mitogen/plugins/connection/*: Begin splitting connection.Connection into subclasses, exposing them directly as "mitogen_ssh", "mitogen_local", etc. connection types. This is far from removing strategy.py, but it's a tiny start. ansible_mitogen/connection.py: * config_from_play_context() and config_from_host_vars() build up a huge dictionary containing either more or less PlayContext contents, or our best attempt at reconstructing a host's connection config from its hostvars, where that config is not the current WorkerProcess target. They both produce the same format with the same keys, allowing remaining code to have a single input format. These dicts contain fields named after how Ansible refers to them, e.g. "sudo_exe". * _config_from_via() parses a basic connection specification like "username@inventory_name" into one of the aforementioned dicts. * _stack_from_config() produces a list of dicts describing the order in which (Mitogen) connections should be established, such that each element is proxied via= the previous element. The dicts produced by this function use Mitogen keyword arguments, the former di. These dicts contain fields named after how Mitogen refers to them, e.g. "sudo_path". * Pass the stack to ContextService, which is responsible for actual setup of the full chain. ansible_mitogen/services.py: Teach get() to walk the supplied stack, establishing each connection in turn, creating refounts for it before continuing. TODO: refcounting is broken in a variety of cases.
7 years ago
'method_name': spec['method'],
'msg': str(e),
}
return result
class ModuleDepService(mitogen.service.Service):
"""
Scan a new-style module and produce a cached mapping of module_utils names
to their resolved filesystem paths.
"""
invoker_class = mitogen.service.SerializedInvoker
issue #186: rework async/forked tasks again. The controller must know the ID of the forked child in order to propagate dependencies to it, so forking+starting the module run cannot happen entirely on the target, without some additional mechanism to wait-and-repropagate the deps as they arrive on the target. Rework things so that init_child() also handles starting the fork parent, and returns it along with the context's home directory in a single round trip. Now master knows the identity of the fork parent, it can directly create fork children and call run_module_async() in them. This necessitates 2 roundtrips to start an asynchronous task. This whole thing sucks and entirely needs simplified, but for now things almost work, so keeping it. connection.py: * Expect ContextService to return the entire dict return value of init_child(). Store the fork_contxt from the return value. planner.py: * Rework Planner to store the invocation as an instance attribute, to simplify method calls. * Add Planner.get_push_files() and Planner.get_module_deps(). * Add _propagate_deps() which takes a Planner and ensures the deps it describes are sent to a (non forked or forked) context. * Move async task logic out of target.py and into invoke() / _invoke_*(). process.py: * Services no longer need references to each other. planner.py handles sending module deps with one extra RPC. services.py: * Return "init_child_result" key instead of simple "home_dir" key. * Get rid of dep propagation from ModuleDepService, it lives in planner.py now. target.py: * Get rid of async task start logic, lives in planner.py now.
6 years ago
def __init__(self, *args, **kwargs):
super(ModuleDepService, self).__init__(*args, **kwargs)
self._cache = {}
def _get_builtin_names(self, builtin_path, resolved):
return [
mitogen.core.to_text(fullname)
for fullname, path, is_pkg in resolved
if os.path.abspath(path).startswith(builtin_path)
]
def _get_custom_tups(self, builtin_path, resolved):
return [
(mitogen.core.to_text(fullname),
mitogen.core.to_text(path),
is_pkg)
for fullname, path, is_pkg in resolved
if not os.path.abspath(path).startswith(builtin_path)
]
@mitogen.service.expose(policy=mitogen.service.AllowParents())
@mitogen.service.arg_spec({
'module_name': mitogen.core.UnicodeType,
'module_path': mitogen.core.FsPathTypes,
'search_path': tuple,
'builtin_path': mitogen.core.FsPathTypes,
'context': mitogen.core.Context,
})
def scan(self, module_name, module_path, search_path, builtin_path, context):
key = (module_name, search_path)
if key not in self._cache:
resolved = ansible_mitogen.module_finder.scan(
module_name=module_name,
module_path=module_path,
search_path=tuple(search_path) + (builtin_path,),
)
builtin_path = os.path.abspath(builtin_path)
builtin = self._get_builtin_names(builtin_path, resolved)
custom = self._get_custom_tups(builtin_path, resolved)
self._cache[key] = {
'builtin': builtin,
'custom': custom,
}
return self._cache[key]