Cleanup some include logic

* Properly mark hosts with failures in includes as failed
* Don't send callbacks until we're sure we're done, and also fix how
  we increment stats so failures don't show up as ok's
* Fix a bug in the include file logic where a failed include could lead
  to an infinite loop in the task iteration logic

Fixes #12933

lib/ansible/executor/play_iterator.py

@@ -396,6 +396,9 @@ class PlayIterator:
         return None
 
     def _insert_tasks_into_state(self, state, task_list):
+        if state.fail_state != self.FAILED_NONE:
+            return state
+
         if state.run_state == self.ITERATING_TASKS:
             if state.tasks_child_state:
                 state.tasks_child_state = self._insert_tasks_into_state(state.tasks_child_state, task_list)

lib/ansible/plugins/strategy/__init__.py

@@ -199,10 +199,11 @@ class StrategyBase:
                         self._tqm._stats.increment('skipped', host.name)
                         self._tqm.send_callback('v2_runner_on_skipped', task_result)
                     elif result[0] == 'host_task_ok':
-                        self._tqm._stats.increment('ok', host.name)
+                        if task.action != 'include':
-                        if 'changed' in task_result._result and task_result._result['changed']:
+                            self._tqm._stats.increment('ok', host.name)
-                            self._tqm._stats.increment('changed', host.name)
+                            if 'changed' in task_result._result and task_result._result['changed']:
-                        self._tqm.send_callback('v2_runner_on_ok', task_result)
+                                self._tqm._stats.increment('changed', host.name)
+                            self._tqm.send_callback('v2_runner_on_ok', task_result)
 
                         if self._diff and 'diff' in task_result._result:
                             self._tqm.send_callback('v2_on_file_diff', task_result)
@@ -395,10 +396,29 @@ class StrategyBase:
 
         try:
             data = self._loader.load_from_file(included_file._filename)
-            self._tqm.send_callback('v2_playbook_on_include', included_file)
             if data is None:
                 return []
+            elif not isinstance(data, list):
+                raise AnsibleError("included task files must contain a list of tasks")
+
+            block_list = load_list_of_blocks(
+                data,
+                play=included_file._task._block._play,
+                parent_block=included_file._task._block,
+                task_include=included_file._task,
+                role=included_file._task._role,
+                use_handlers=is_handler,
+                loader=self._loader
+            )
+
+            # since we skip incrementing the stats when the task result is
+            # first processed, we do so now for each host in the list
+            for host in included_file._hosts:
+                self._tqm._stats.increment('ok', host.name)
+
         except AnsibleError as e:
+            # mark all of the hosts including this file as failed, send callbacks,
+            # and increment the stats for this host
             for host in included_file._hosts:
                 tr = TaskResult(host=host, task=included_file._task, return_data=dict(failed=True, reason=str(e)))
                 iterator.mark_host_failed(host)
@@ -407,19 +427,6 @@ class StrategyBase:
                 self._tqm.send_callback('v2_runner_on_failed', tr)
             return []
 
-        if not isinstance(data, list):
-            raise AnsibleParserError("included task files must contain a list of tasks", obj=included_file._task._ds)
-
-        block_list = load_list_of_blocks(
-            data,
-            play=included_file._task._block._play,
-            parent_block=included_file._task._block,
-            task_include=included_file._task,
-            role=included_file._task._role,
-            use_handlers=is_handler,
-            loader=self._loader
-        )
-
         # set the vars for this task from those specified as params to the include
         for b in block_list:
             # first make a copy of the including task, so that each has a unique copy to modify
@@ -444,6 +451,8 @@ class StrategyBase:
                 b._task_include.tags = tags
             b._task_include.vars = temp_vars
 
+        # finally, send the callback and return the list of blocks loaded
+        self._tqm.send_callback('v2_playbook_on_include', included_file)
         return block_list
 
     def run_handlers(self, iterator, play_context):

lib/ansible/plugins/strategy/linear.py

@@ -275,25 +275,29 @@ class StrategyModule(StrategyBase):
                         # list of noop tasks, to make sure that they continue running in lock-step
                         try:
                             new_blocks = self._load_included_file(included_file, iterator=iterator)
+
+                            for new_block in new_blocks:
+                                noop_block = Block(parent_block=task._block)
+                                noop_block.block  = [noop_task for t in new_block.block]
+                                noop_block.always = [noop_task for t in new_block.always]
+                                noop_block.rescue = [noop_task for t in new_block.rescue]
+                                for host in hosts_left:
+                                    if host in included_file._hosts:
+                                        task_vars = self._variable_manager.get_vars(loader=self._loader, play=iterator._play, host=host, task=included_file._task)
+                                        final_block = new_block.filter_tagged_tasks(play_context, task_vars)
+                                        all_blocks[host].append(final_block)
+                                    else:
+                                        all_blocks[host].append(noop_block)
+
                         except AnsibleError as e:
                             for host in included_file._hosts:
+                                self._tqm._failed_hosts[host.name] = True
                                 iterator.mark_host_failed(host)
-                            self._display.warning(str(e))
+                            self._display.error(e, wrap_text=False)
                             continue
 
-                        for new_block in new_blocks:
+                    # finally go through all of the hosts and append the
-                            noop_block = Block(parent_block=task._block)
+                    # accumulated blocks to their list of tasks
-                            noop_block.block  = [noop_task for t in new_block.block]
-                            noop_block.always = [noop_task for t in new_block.always]
-                            noop_block.rescue = [noop_task for t in new_block.rescue]
-                            for host in hosts_left:
-                                if host in included_file._hosts:
-                                    task_vars = self._variable_manager.get_vars(loader=self._loader, play=iterator._play, host=host, task=included_file._task)
-                                    final_block = new_block.filter_tagged_tasks(play_context, task_vars)
-                                    all_blocks[host].append(final_block)
-                                else:
-                                    all_blocks[host].append(noop_block)
-
                     for host in hosts_left:
                         iterator.add_tasks(host, all_blocks[host])