Better errors for delegate_to (#82319)

Handle empty result of templating
  Also skip work when we omit

(cherry picked from commit 6ebefaceb6)

changelogs/fragments/delegate_to_invalid.yml

@@ -0,0 +1,2 @@
+bugfixes:
+  - delegate_to when set to an empty or undefined variable will now give a proper error.

lib/ansible/executor/task_executor.py

@@ -408,11 +408,7 @@ class TaskExecutor:
         """This method is responsible for effectively pre-validating Task.delegate_to and will
         happen before Task.post_validate is executed
         """
-        delegated_vars, delegated_host_name = self._variable_manager.get_delegated_vars_and_hostname(
-            templar,
-            self._task,
-            variables
-        )
+        delegated_vars, delegated_host_name = self._variable_manager.get_delegated_vars_and_hostname(templar, self._task, variables)
         # At the point this is executed it is safe to mutate self._task,
         # since `self._task` is either a copy referred to by `tmp_task` in `_run_loop`
         # or just a singular non-looped task

lib/ansible/vars/manager.py

@@ -524,6 +524,13 @@ class VariableManager:
         delegated_host_name = None
         if task.delegate_to:
             delegated_host_name = templar.template(task.delegate_to, fail_on_undefined=False)
+
+            # no need to do work if omitted
+            if delegated_host_name != self._omit_token:
+
+                if not delegated_host_name:
+                    raise AnsibleError('Empty hostname produced from delegate_to: "%s"' % task.delegate_to)
+
                 delegated_host = self._inventory.get_host(delegated_host_name)
                 if delegated_host is None:
                     for h in self._inventory.get_hosts(ignore_limits=True, ignore_restrictions=True):
@@ -545,6 +552,7 @@ class VariableManager:
                     )
                 }
                 delegated_vars['ansible_delegated_vars'][delegated_host_name]['inventory_hostname'] = variables.get('inventory_hostname')
+
         return delegated_vars, delegated_host_name
 
     def _get_delegated_vars(self, play, task, existing_variables):

test/integration/targets/delegate_to/test_delegate_to.yml

@@ -57,6 +57,25 @@
     - name: remove test file
       file: path={{ output_dir }}/tmp.txt state=absent
 
+    - name: Use omit to thwart delegation
+      ping:
+      delegate_to: "{{ jenkins_install_key_on|default(omit) }}"
+      register: d_omitted
+
+    - name: Use empty to thwart delegation should fail
+      ping:
+      delegate_to: "{{ jenkins_install_key_on }}"
+      when: jenkins_install_key_on != ""
+      vars:
+        jenkins_install_key_on: ''
+      ignore_errors: true
+      register: d_empty
+
+    - name: Ensure previous 2 tests actually did what was expected
+      assert:
+        that:
+          - d_omitted is success
+          - d_empty is failed
 
 - name: verify delegation with per host vars
   hosts: testhost6