From dabae80ca98878b8c2302333a6c6c7aac33589dc Mon Sep 17 00:00:00 2001 From: Brian Coca Date: Tue, 3 Oct 2023 11:02:29 -0400 Subject: [PATCH] try to proxy copy of pl loading --- lib/ansible/executor/process/worker.py | 1 + lib/ansible/executor/task_queue_manager.py | 8 +++++++ lib/ansible/plugins/loader.py | 28 ++++++++++++++++++++++ 3 files changed, 37 insertions(+) diff --git a/lib/ansible/executor/process/worker.py b/lib/ansible/executor/process/worker.py index c043137c95f..c3094d17004 100644 --- a/lib/ansible/executor/process/worker.py +++ b/lib/ansible/executor/process/worker.py @@ -169,6 +169,7 @@ class WorkerProcess(multiprocessing_context.Process): # type: ignore[name-defin # Set the queue on Display so calls to Display.display are proxied over the queue display.set_queue(self._final_q) + self._shared_loader_obj.set_queue(self._final_q) global current_worker current_worker = self diff --git a/lib/ansible/executor/task_queue_manager.py b/lib/ansible/executor/task_queue_manager.py index 3bbf3d592e1..85359c1f176 100644 --- a/lib/ansible/executor/task_queue_manager.py +++ b/lib/ansible/executor/task_queue_manager.py @@ -66,6 +66,12 @@ class DisplaySend: self.args = args self.kwargs = kwargs +class PluginLoaderSend: + def __init__(self, method, *args, **kwargs): + self.method = method + self.args = args + self.kwargs = kwargs + @dataclass class PromptSend: @@ -106,6 +112,8 @@ class FinalQueue(multiprocessing.queues.SimpleQueue): PromptSend(**kwargs), ) + def send_plugin_loader(self, *args, **kwargs): + self.put(PluginLoaderSend(*args, **kwargs)) class AnsibleEndPlay(Exception): def __init__(self, result): diff --git a/lib/ansible/plugins/loader.py b/lib/ansible/plugins/loader.py index dc0eb891a9a..d54c2f0b92e 100644 --- a/lib/ansible/plugins/loader.py +++ b/lib/ansible/plugins/loader.py @@ -34,6 +34,7 @@ from ansible.plugins import get_plugin_class, MODULE_CACHE, PATH_CACHE, PLUGIN_P from ansible.utils.collection_loader import AnsibleCollectionConfig, AnsibleCollectionRef from ansible.utils.collection_loader._collection_finder import _AnsibleCollectionFinder, _get_collection_metadata from ansible.utils.display import Display +from ansible.utils.multiprocessing import context as multiprocessing_context from ansible.utils.plugin_docs import add_fragments # TODO: take the packaging dep, or vendor SpecifierSet? @@ -52,6 +53,29 @@ display = Display() get_with_context_result = namedtuple('get_with_context_result', ['object', 'plugin_load_context']) +_final_q = None + + +def proxy_load(method): + + def proxyit(self, *args, **kwargs): + if sys.modules[__name__]._final_q: + # If _final_q is set, that means we are in a WorkerProcess + # and instead of displaying messages directly from the fork + # we will proxy them through the queue + sys.modules[__name__]._final_q.send_plugin_loader(method.__name__, *args, **kwargs) + return method(self, *args, **kwargs) + + return proxyit + +def set_queue(queue): + """Set the _final_q on PLuginLoader, so that we know to proxy requests over the queue + This is only needed in ansible.executor.process.worker:WorkerProcess._run + """ + if multiprocessing_context.parent_process() is None: + raise RuntimeError('queue cannot be set in parent process') + sys.modules[__name__]._final_q = queue + def get_all_plugin_loaders(): return [(name, obj) for (name, obj) in globals().items() if isinstance(obj, PluginLoader)] @@ -858,6 +882,7 @@ class PluginLoader: def get(self, name, *args, **kwargs): return self.get_with_context(name, *args, **kwargs).object + @proxy_load def get_with_context(self, name, *args, **kwargs): ''' instantiates a plugin of the given name using arguments ''' @@ -932,6 +957,7 @@ class PluginLoader: display.debug(msg) + @proxy_load def all(self, *args, **kwargs): ''' Iterate through all plugins of this type, in configured paths (no collections) @@ -1125,6 +1151,7 @@ class Jinja2Loader(PluginLoader): # FUTURE: now that the resulting plugins are closer, refactor base class method with some extra # hooks so we can avoid all the duplicated plugin metadata logic, and also cache the collection results properly here + @proxy_load def get_with_context(self, name, *args, **kwargs): # pop N/A kwargs to avoid passthrough to parent methods kwargs.pop('class_only', False) @@ -1258,6 +1285,7 @@ class Jinja2Loader(PluginLoader): return get_with_context_result(plugin, context) + @proxy_load def all(self, *args, **kwargs): kwargs.pop('_dedupe', None) path_only = kwargs.pop('path_only', False)