Use the task loop to calculate multiple delegated hosts

Due to the way we're now calculating delegate_to, if that value is based
on a loop variable ('item') we need to calculate all of the possible
delegated_to variables for that loop.

Fixes #12499

lib/ansible/executor/process/worker.py

@@ -112,14 +112,9 @@ class WorkerProcess(multiprocessing.Process):
                 # the task handles updating parent/child objects as needed.
                 task.set_loader(self._loader)
 
-                # apply the given task's information to the connection info,
-                # which may override some fields already set by the play or
-                # the options specified on the command line
-                new_play_context = play_context.set_task_and_variable_override(task=task, variables=job_vars)
-
                 # execute the task and build a TaskResult from the result
                 debug("running TaskExecutor() for %s/%s" % (host, task))
-                executor_result = TaskExecutor(host, task, job_vars, new_play_context, self._new_stdin, self._loader, shared_loader_obj).run()
+                executor_result = TaskExecutor(host, task, job_vars, play_context, self._new_stdin, self._loader, shared_loader_obj).run()
                 debug("done running TaskExecutor() for %s/%s" % (host, task))
                 task_result = TaskResult(host, task, executor_result)
 

lib/ansible/executor/task_executor.py

@@ -261,6 +261,11 @@ class TaskExecutor:
 
         templar = Templar(loader=self._loader, shared_loader_obj=self._shared_loader_obj, variables=variables)
 
+        # apply the given task's information to the connection info,
+        # which may override some fields already set by the play or
+        # the options specified on the command line
+        self._play_context = self._play_context.set_task_and_variable_override(task=self._task, variables=variables, templar=templar)
+
         # fields set from the play/task may be based on variables, so we have to
         # do the same kind of post validation step on it here before we use it.
         # We also add "magic" variables back into the variables dict to make sure

lib/ansible/playbook/play_context.py

@@ -267,7 +267,7 @@ class PlayContext(Base):
             elif isinstance(options.skip_tags, string_types):
                 self.skip_tags.update(options.skip_tags.split(','))
 
-    def set_task_and_variable_override(self, task, variables):
+    def set_task_and_variable_override(self, task, variables, templar):
         '''
         Sets attributes from the task if they are set, which will override
         those from the play.
@@ -288,7 +288,15 @@ class PlayContext(Base):
         # If the value 'ansible_delegated_vars' is in the variables, it means
         # we have a delegated-to host, so we check there first before looking
         # at the variables in general
-        delegated_vars = variables.get('ansible_delegated_vars', dict())
+        if task.delegate_to is not None:
+            # In the case of a loop, the delegated_to host may have been
+            # templated based on the loop variable, so we try and locate
+            # the host name in the delegated variable dictionary here
+            delegated_host_name = templar.template(task.delegate_to)
+            delegated_vars = variables.get('ansible_delegated_vars', dict()).get(delegated_host_name, dict())
+        else:
+            delegated_vars = dict()
+
         for (attr, variable_names) in iteritems(MAGIC_VARIABLE_MAPPING):
             for variable_name in variable_names:
                 if isinstance(delegated_vars, dict) and variable_name in delegated_vars:

lib/ansible/vars/__init__.py

@@ -36,9 +36,11 @@ from ansible.cli import CLI
 from ansible.errors import AnsibleError, AnsibleParserError, AnsibleUndefinedVariable, AnsibleFileNotFound
 from ansible.inventory.host import Host
 from ansible.parsing import DataLoader
+from ansible.plugins import lookup_loader
 from ansible.plugins.cache import FactCache
 from ansible.template import Templar
 from ansible.utils.debug import debug
+from ansible.utils.listify import listify_lookup_plugin_terms
 from ansible.utils.vars import combine_vars
 from ansible.vars.hostvars import HostVars
 from ansible.vars.unsafe_proxy import UnsafeProxy
@@ -333,7 +335,38 @@ class VariableManager:
                 # as we're fetching vars before post_validate has been called on
                 # the task that has been passed in
                 templar = Templar(loader=loader, variables=all_vars)
-                delegated_host_name = templar.template(task.delegate_to)
+
+                items = []
+                if task.loop is not None:
+                    if task.loop in lookup_loader:
+                        #TODO: remove convert_bare true and deprecate this in with_ 
+                        try:
+                            loop_terms = listify_lookup_plugin_terms(terms=task.loop_args, templar=templar, loader=loader, fail_on_undefined=True, convert_bare=True)
+                        except AnsibleUndefinedVariable as e:
+                            if 'has no attribute' in str(e):
+                                loop_terms = []
+                                self._display.deprecated("Skipping task due to undefined attribute, in the future this will be a fatal error.")
+                            else:
+                                raise
+                        items = lookup_loader.get(task.loop, loader=loader, templar=templar).run(terms=loop_terms, variables=all_vars)
+                    else:
+                        raise AnsibleError("Unexpected failure in finding the lookup named '%s' in the available lookup plugins" % task.loop)
+                else:
+                    items = [None]
+
+                vars_copy = all_vars.copy()
+                delegated_host_vars = dict()
+                for item in items:
+                    # update the variables with the item value for templating, in case we need it
+                    if item is not None:
+                        vars_copy['item'] = item
+
+                    templar.set_available_variables(vars_copy)
+                    delegated_host_name = templar.template(task.delegate_to, fail_on_undefined=False)
+                    if delegated_host_name in delegated_host_vars:
+                        # no need to repeat ourselves, as the delegate_to value
+                        # does not appear to be tied to the loop item variable
+                        continue
 
                     # a dictionary of variables to use if we have to create a new host below
                     new_delegated_host_vars = dict(
@@ -365,7 +398,15 @@ class VariableManager:
 
                     # now we go fetch the vars for the delegated-to host and save them in our
                     # master dictionary of variables to be used later in the TaskExecutor/PlayContext
-                all_vars['ansible_delegated_vars'] = self.get_vars(loader=loader, play=play, host=delegated_host, task=task, include_delegate_to=False, include_hostvars=False)
+                    delegated_host_vars[delegated_host_name] = self.get_vars(
+                        loader=loader,
+                        play=play,
+                        host=delegated_host,
+                        task=task,
+                        include_delegate_to=False,
+                        include_hostvars=False,
+                    )
+                all_vars['ansible_delegated_vars'] = delegated_host_vars
 
         if self._inventory is not None:
             all_vars['inventory_dir'] = self._inventory.basedir()

test/units/playbook/test_play_context.py

@@ -99,8 +99,10 @@ class TestPlayContext(unittest.TestCase):
             ansible_ssh_port = 4321,
         )
 
+        mock_templar = MagicMock()
+
         play_context = PlayContext(play=mock_play, options=options)
-        play_context = play_context.set_task_and_variable_override(task=mock_task, variables=all_vars)
+        play_context = play_context.set_task_and_variable_override(task=mock_task, variables=all_vars, templar=mock_templar)
         self.assertEqual(play_context.connection, 'mock_inventory')
         self.assertEqual(play_context.remote_user, 'mocktask')
         self.assertEqual(play_context.port, 4321)

test/units/vars/test_variable_manager.py

@@ -171,6 +171,7 @@ class TestVariableManager(unittest.TestCase):
 
         mock_task = MagicMock()
         mock_task._role = None
+        mock_task.loop = None
         mock_task.get_vars.return_value = dict(foo="bar")
 
         v = VariableManager()