[stable-2.7] Don't use the task for a cache, return a special cache var (#47243)

* Don't use task to cache loop results, use hostvars. Fixes #47207

* Avoid a race condition, supply _ansible_loop_cache through get_vars directly

* Add tests

* Add changelog fragment

* Remove unnecessary copy

* Remove unnecessary host from _get_delegated_vars signature.
(cherry picked from commit 77d32b8f5799b8a32f464a22e25b38e5ea4b260c)

Co-authored-by: Matt Martz <matt@sivel.net>

5 files changed, 62 insertions, 9 deletions

diff --git a/changelogs/fragments/delegate_to_loop_hostvars.yaml b/changelogs/fragments/delegate_to_loop_hostvars.yaml
new file mode 100644
index 0000000000..26680f3ae5
--- /dev/null
+++ b/changelogs/fragments/delegate_to_loop_hostvars.yaml
@@ -0,0 +1,2 @@
+bugfixes:
+- delegate_to - When templating ``delegate_to`` in a loop, don't use the task for a cache, return a special cache through ``get_vars`` allowing looping over a hostvar (https://github.com/ansible/ansible/issues/47207)
diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index 86df0b57cd..5f13727469 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -210,7 +210,12 @@ class TaskExecutor:
 
         templar = Templar(loader=self._loader, shared_loader_obj=self._shared_loader_obj, variables=self._job_vars)
         items = None
-        if self._task.loop_with:
+        loop_cache = self._job_vars.get('_ansible_loop_cache')
+        if loop_cache is not None:
+            # _ansible_loop_cache may be set in `get_vars` when calculating `delegate_to`
+            # to avoid reprocessing the loop
+            items = loop_cache
+        elif self._task.loop_with:
             if self._task.loop_with in self._shared_loader_obj.lookup_loader:
                 fail = True
                 if self._task.loop_with == 'first_found':
diff --git a/lib/ansible/vars/manager.py b/lib/ansible/vars/manager.py
index a88b7ae996..2516014f0c 100644
--- a/lib/ansible/vars/manager.py
+++ b/lib/ansible/vars/manager.py
@@ -429,7 +429,7 @@ class VariableManager:
         # if we have a task and we're delegating to another host, figure out the
         # variables for that host now so we don't have to rely on hostvars later
         if task and task.delegate_to is not None and include_delegate_to:
-            all_vars['ansible_delegated_vars'] = self._get_delegated_vars(play, task, all_vars)
+            all_vars['ansible_delegated_vars'], all_vars['_ansible_loop_cache'] = self._get_delegated_vars(play, task, all_vars)
 
         # 'vars' magic var
         if task or play:
@@ -594,16 +594,15 @@ class VariableManager:
                 include_hostvars=False,
             )
 
+        _ansible_loop_cache = None
         if has_loop and cache_items:
-            # delegate_to templating produced a change, update task.loop with templated items,
+            # delegate_to templating produced a change, so we will cache the templated items
+            # in a special private hostvar
             # this ensures that delegate_to+loop doesn't produce different results than TaskExecutor
             # which may reprocess the loop
-            # Set loop_with to None, so we don't do extra unexpected processing on the cached items later
-            # in TaskExecutor
-            task.loop_with = None
-            task.loop = items
+            _ansible_loop_cache = items
 
-        return delegated_host_vars
+        return delegated_host_vars, _ansible_loop_cache
 
     def clear_facts(self, hostname):
         '''
diff --git a/test/integration/targets/delegate_to/runme.sh b/test/integration/targets/delegate_to/runme.sh
index 8d669df198..26b3d54295 100755
--- a/test/integration/targets/delegate_to/runme.sh
+++ b/test/integration/targets/delegate_to/runme.sh
@@ -9,4 +9,6 @@ ansible-playbook test_loop_control.yml -v "$@"
 
 ansible-playbook test_delegate_to_loop_randomness.yml -v "$@"
 
-ansible-playbook delegate_and_nolog.yml -v "$@"
+ansible-playbook delegate_and_nolog.yml -i ../../inventory -v "$@"
+
+ansible-playbook test_delegate_to_loop_caching.yml -i ../../inventory -v "$@"
diff --git a/test/integration/targets/delegate_to/test_delegate_to_loop_caching.yml b/test/integration/targets/delegate_to/test_delegate_to_loop_caching.yml
new file mode 100644
index 0000000000..6ea08f72d8
--- /dev/null
+++ b/test/integration/targets/delegate_to/test_delegate_to_loop_caching.yml
@@ -0,0 +1,45 @@
+- hosts: testhost,testhost2
+  gather_facts: false
+  vars:
+    delegate_to_host: "localhost"
+  tasks:
+    - set_fact:
+        gandalf:
+          shout: 'You shall not pass!'
+      when: inventory_hostname == 'testhost'
+
+    - set_fact:
+        gandalf:
+          speak: 'Run you fools!'
+      when: inventory_hostname == 'testhost2'
+
+    - name: works correctly
+      debug: var=item
+      delegate_to: localhost
+      with_dict: "{{ gandalf }}"
+      register: result1
+
+    - name: shows same item for all hosts
+      debug: var=item
+      delegate_to: "{{ delegate_to_host }}"
+      with_dict: "{{ gandalf }}"
+      register: result2
+
+    - debug:
+        var: result2.results[0].item.value
+
+    - assert:
+        that:
+          - result1.results[0].item.value == 'You shall not pass!'
+          - result2.results[0].item.value == 'You shall not pass!'
+      when: inventory_hostname == 'testhost'
+
+    - assert:
+        that:
+          - result1.results[0].item.value == 'Run you fools!'
+          - result2.results[0].item.value == 'Run you fools!'
+      when: inventory_hostname == 'testhost2'
+
+    - assert:
+        that:
+          - _ansible_loop_cache is undefined