Create a raw lookup for hostvars that does not template the data

When using hostvars to get extra connection-specific vars for connection
plugins, use this raw lookup to avoid prematurely templating all of the
hostvar data (triggering unnecessary lookups).

Fixes #17024

lib/ansible/executor/task_executor.py

@@ -437,7 +437,17 @@ class TaskExecutor:
         # get the connection and the handler for this execution
         if not self._connection or not getattr(self._connection, 'connected', False) or self._play_context.remote_addr != self._connection._play_context.remote_addr:
             self._connection = self._get_connection(variables=variables, templar=templar)
-            self._connection.set_host_overrides(host=self._host, hostvars=variables.get('hostvars', {}).get(self._host.name, {}))
+            hostvars = variables.get('hostvars', None)
+            if hostvars:
+                try:
+                    target_hostvars = hostvars.raw_get(self._host.name)
+                except:
+                    # FIXME: this should catch the j2undefined error here
+                    #        specifically instead of all exceptions
+                    target_hostvars = dict()
+            else:
+                target_hostvars = dict()
+            self._connection.set_host_overrides(host=self._host, hostvars=target_hostvars)
         else:
             # if connection is reused, its _play_context is no longer valid and needs
             # to be replaced with the one templated above, in case other data changed

lib/ansible/vars/hostvars.py

@@ -68,13 +68,19 @@ class HostVars(collections.Mapping):
             host = self._inventory.get_host(host_name)
         return host
 
-    def __getitem__(self, host_name):
+    def raw_get(self, host_name):
+        '''
+        Similar to __getitem__, however the returned data is not run through
+        the templating engine to expand variables in the hostvars.
+        '''
         host = self._find_host(host_name)
         if host is None:
             raise j2undefined
 
-        data = self._variable_manager.get_vars(loader=self._loader, host=host, include_hostvars=False)
+        return self._variable_manager.get_vars(loader=self._loader, host=host, include_hostvars=False)
 
+    def __getitem__(self, host_name):
+        data = self.raw_get(host_name)
         sha1_hash = sha1(str(data).encode('utf-8')).hexdigest()
         if sha1_hash in self._cached_result:
             result = self._cached_result[sha1_hash]