Add toggle to fix module_defaults with module-as-redirected-action on a per-module basis (#77265) (#77899)

* If there is a platform specific handler, prefer the resolved module over the resolved action when loading module_defaults

Add a toggle for action plugins to prefer the resolved module when loading module_defaults

Allow moving away from modules intercepted as actions pattern

Fixes #77059

(cherry picked from commit 621e782ed0c119d2c84124d006fdf253c082449a)

3 files changed, 45 insertions, 12 deletions

diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index d027dfaeef..eceab1b260 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -26,7 +26,7 @@ from ansible.playbook.conditional import Conditional
 from ansible.playbook.task import Task
 from ansible.plugins.loader import become_loader, cliconf_loader, connection_loader, httpapi_loader, netconf_loader, terminal_loader
 from ansible.template import Templar
-from ansible.utils.collection_loader import AnsibleCollectionConfig
+from ansible.utils.collection_loader import AnsibleCollectionConfig, AnsibleCollectionRef
 from ansible.utils.listify import listify_lookup_plugin_terms
 from ansible.utils.unsafe_proxy import to_unsafe_text, wrap_var
 from ansible.vars.clean import namespace_facts, clean_facts
@@ -590,11 +590,16 @@ class TaskExecutor:
             cvars['ansible_python_interpreter'] = sys.executable
 
         # get handler
-        self._handler = self._get_action_handler(connection=self._connection, templar=templar)
+        self._handler, module_context = self._get_action_handler_with_module_context(connection=self._connection, templar=templar)
+
+        if module_context is not None:
+            module_defaults_fqcn = module_context.resolved_fqcn
+        else:
+            module_defaults_fqcn = self._task.resolved_action
 
         # Apply default params for action/module, if present
         self._task.args = get_action_args_with_defaults(
-            self._task.resolved_action, self._task.args, self._task.module_defaults, templar,
+            module_defaults_fqcn, self._task.args, self._task.module_defaults, templar,
             action_groups=self._task._parent._play._action_groups
         )
 
@@ -1093,7 +1098,12 @@ class TaskExecutor:
         '''
         Returns the correct action plugin to handle the requestion task action
         '''
+        return self._get_action_handler_with_module_context(connection, templar)[0]
 
+    def _get_action_handler_with_module_context(self, connection, templar):
+        '''
+        Returns the correct action plugin to handle the requestion task action and the module context
+        '''
         module_collection, separator, module_name = self._task.action.rpartition(".")
         module_prefix = module_name.split('_')[0]
         if module_collection:
@@ -1106,8 +1116,16 @@ class TaskExecutor:
 
         collections = self._task.collections
 
+        # Check if the module has specified an action handler
+        module = self._shared_loader_obj.module_loader.find_plugin_with_context(
+            self._task.action, collection_list=collections
+        )
+        if not module.resolved or not module.action_plugin:
+            module = None
+        if module is not None:
+            handler_name = module.action_plugin
         # let action plugin override module, fallback to 'normal' action plugin otherwise
-        if self._shared_loader_obj.action_loader.has_plugin(self._task.action, collection_list=collections):
+        elif self._shared_loader_obj.action_loader.has_plugin(self._task.action, collection_list=collections):
             handler_name = self._task.action
         elif all((module_prefix in C.NETWORK_GROUP_MODULES, self._shared_loader_obj.action_loader.has_plugin(network_action, collection_list=collections))):
             handler_name = network_action
@@ -1133,7 +1151,7 @@ class TaskExecutor:
         if not handler:
             raise AnsibleError("the handler '%s' was not found" % handler_name)
 
-        return handler
+        return handler, module
 
 
 def start_connection(play_context, variables, task_uuid):
diff --git a/lib/ansible/playbook/base.py b/lib/ansible/playbook/base.py
index 4bd5a7c287..e8cfe741b6 100644
--- a/lib/ansible/playbook/base.py
+++ b/lib/ansible/playbook/base.py
@@ -507,9 +507,13 @@ class FieldAttributeBase(metaclass=BaseMeta):
         return fq_group_name, resolved_actions
 
     def _resolve_action(self, action_name, mandatory=True):
-        context = action_loader.find_plugin_with_context(action_name)
-        if not context.resolved:
-            context = module_loader.find_plugin_with_context(action_name)
+        context = module_loader.find_plugin_with_context(action_name)
+        if context.resolved and not context.action_plugin:
+            prefer = action_loader.find_plugin_with_context(action_name)
+            if prefer.resolved:
+                context = prefer
+        elif not context.resolved:
+            context = action_loader.find_plugin_with_context(action_name)
 
         if context.resolved:
             return context.resolved_fqcn
diff --git a/lib/ansible/plugins/loader.py b/lib/ansible/plugins/loader.py
index 413cd44cf6..4122f9edc7 100644
--- a/lib/ansible/plugins/loader.py
+++ b/lib/ansible/plugins/loader.py
@@ -126,6 +126,7 @@ class PluginLoadContext(object):
         self.deprecation_warnings = []
         self.resolved = False
         self._resolved_fqcn = None
+        self.action_plugin = None
 
     @property
     def resolved_fqcn(self):
@@ -166,13 +167,14 @@ class PluginLoadContext(object):
         self.deprecation_warnings.append(warning_text)
         return self
 
-    def resolve(self, resolved_name, resolved_path, resolved_collection, exit_reason):
+    def resolve(self, resolved_name, resolved_path, resolved_collection, exit_reason, action_plugin):
         self.pending_redirect = None
         self.plugin_resolved_name = resolved_name
         self.plugin_resolved_path = resolved_path
         self.plugin_resolved_collection = resolved_collection
         self.exit_reason = exit_reason
         self.resolved = True
+        self.action_plugin = action_plugin
         return self
 
     def redirect(self, redirect_name):
@@ -231,8 +233,12 @@ class PluginLoader:
 
         self._searched_paths = set()
 
+    @property
+    def type(self):
+        return AnsibleCollectionRef.legacy_plugin_dir_to_plugin_type(self.subdir)
+
     def __repr__(self):
-        return 'PluginLoader(type={0})'.format(AnsibleCollectionRef.legacy_plugin_dir_to_plugin_type(self.subdir))
+        return 'PluginLoader(type={0})'.format(self.type)
 
     def _clear_caches(self):
 
@@ -459,6 +465,7 @@ class PluginLoader:
         # check collection metadata to see if any special handling is required for this plugin
         routing_metadata = self._query_collection_routing_meta(acr, plugin_type, extension=extension)
 
+        action_plugin = None
         # TODO: factor this into a wrapper method
         if routing_metadata:
             deprecation = routing_metadata.get('deprecation', None)
@@ -497,6 +504,9 @@ class PluginLoader:
                 return plugin_load_context.redirect(redirect)
                 # TODO: non-FQCN case, do we support `.` prefix for current collection, assume it with no dots, require it for subdirs in current, or ?
 
+            if self.type == 'modules':
+                action_plugin = routing_metadata.get('action_plugin')
+
         n_resource = to_native(acr.resource, errors='strict')
         # we want this before the extension is added
         full_name = '{0}.{1}'.format(acr.n_python_package_name, n_resource)
@@ -519,7 +529,7 @@ class PluginLoader:
         # FIXME: and is file or file link or ...
         if os.path.exists(n_resource_path):
             return plugin_load_context.resolve(
-                full_name, to_text(n_resource_path), acr.collection, 'found exact match for {0} in {1}'.format(full_name, acr.collection))
+                full_name, to_text(n_resource_path), acr.collection, 'found exact match for {0} in {1}'.format(full_name, acr.collection), action_plugin)
 
         if extension:
             # the request was extension-specific, don't try for an extensionless match
@@ -540,7 +550,8 @@ class PluginLoader:
             pass
 
         return plugin_load_context.resolve(
-            full_name, to_text(found_files[0]), acr.collection, 'found fuzzy extension match for {0} in {1}'.format(full_name, acr.collection))
+            full_name, to_text(found_files[0]), acr.collection,
+            'found fuzzy extension match for {0} in {1}'.format(full_name, acr.collection), action_plugin)
 
     def find_plugin(self, name, mod_type='', ignore_deprecated=False, check_aliases=False, collection_list=None):
         ''' Find a plugin named name '''