Coverage for lib/ansible/plugins/strategy/__init__.py : 43%

# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com> 

# 

# This file is part of Ansible 

# 

# Ansible is free software: you can redistribute it and/or modify 

# it under the terms of the GNU General Public License as published by 

# the Free Software Foundation, either version 3 of the License, or 

# (at your option) any later version. 

# 

# Ansible is distributed in the hope that it will be useful, 

# but WITHOUT ANY WARRANTY; without even the implied warranty of 

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

# GNU General Public License for more details. 

# 

# You should have received a copy of the GNU General Public License 

# along with Ansible. If not, see <http://www.gnu.org/licenses/>. 

# Make coding more python3-ish 

from __future__ import (absolute_import, division, print_function) 

__metaclass__ = type 

import cmd 

import functools 

import os 

import pprint 

import sys 

import threading 

import time 

from collections import deque 

from multiprocessing import Lock 

from jinja2.exceptions import UndefinedError 

from ansible import constants as C 

from ansible.errors import AnsibleError, AnsibleParserError, AnsibleUndefinedVariable 

from ansible.executor import action_write_locks 

from ansible.executor.process.worker import WorkerProcess 

from ansible.executor.task_result import TaskResult 

from ansible.inventory.host import Host 

from ansible.module_utils.six.moves import queue as Queue 

from ansible.module_utils.six import iteritems, itervalues, string_types 

from ansible.module_utils._text import to_text 

from ansible.module_utils.connection import Connection, ConnectionError 

from ansible.playbook.helpers import load_list_of_blocks 

from ansible.playbook.included_file import IncludedFile 

from ansible.playbook.task_include import TaskInclude 

from ansible.playbook.role_include import IncludeRole 

from ansible.plugins.loader import action_loader, connection_loader, filter_loader, lookup_loader, module_loader, test_loader 

from ansible.template import Templar 

from ansible.utils.vars import combine_vars 

from ansible.vars.clean import strip_internal_keys 

try: 

from __main__ import display 

except ImportError: 

from ansible.utils.display import Display 

display = Display() 

__all__ = ['StrategyBase'] 

class StrategySentinel: 

pass 

# TODO: this should probably be in the plugins/__init__.py, with 

# a smarter mechanism to set all of the attributes based on 

# the loaders created there 

class SharedPluginLoaderObj: 

''' 

A simple object to make pass the various plugin loaders to 

the forked processes over the queue easier 

''' 

def __init__(self): 

self.action_loader = action_loader 

self.connection_loader = connection_loader 

self.filter_loader = filter_loader 

self.test_loader = test_loader 

self.lookup_loader = lookup_loader 

self.module_loader = module_loader 

_sentinel = StrategySentinel() 

def results_thread_main(strategy): 

while True: 

try: 

result = strategy._final_q.get() 

if isinstance(result, StrategySentinel): 

break 

else: 

strategy._results_lock.acquire() 

strategy._results.append(result) 

strategy._results_lock.release() 

except (IOError, EOFError): 

break 

except Queue.Empty: 

pass 

def debug_closure(func): 

"""Closure to wrap ``StrategyBase._process_pending_results`` and invoke the task debugger""" 

@functools.wraps(func) 

def inner(self, iterator, one_pass=False, max_passes=None): 

status_to_stats_map = ( 

('is_failed', 'failures'), 

('is_unreachable', 'dark'), 

('is_changed', 'changed'), 

('is_skipped', 'skipped'), 

) 

# We don't know the host yet, copy the previous states, for lookup after we process new results 

prev_host_states = iterator._host_states.copy() 

results = func(self, iterator, one_pass=one_pass, max_passes=max_passes) 

_processed_results = [] 

for result in results: 

task = result._task 

host = result._host 

_queued_task_args = self._queued_task_cache.pop((host.name, task._uuid), None) 

task_vars = _queued_task_args['task_vars'] 

play_context = _queued_task_args['play_context'] 

# Try to grab the previous host state, if it doesn't exist use get_host_state to generate an empty state 

try: 

prev_host_state = prev_host_states[host.name] 

except KeyError: 

prev_host_state = iterator.get_host_state(host) 

132 ↛ 133line 132 didn't jump to line 133, because the condition on line 132 was never true while result.needs_debugger(globally_enabled=self.debugger_active): 

next_action = NextAction() 

dbg = Debugger(task, host, task_vars, play_context, result, next_action) 

dbg.cmdloop() 

if next_action.result == NextAction.REDO: 

# rollback host state 

self._tqm.clear_failed_hosts() 

iterator._host_states[host.name] = prev_host_state 

for method, what in status_to_stats_map: 

if getattr(result, method)(): 

self._tqm._stats.decrement(what, host.name) 

self._tqm._stats.decrement('ok', host.name) 

# redo 

self._queue_task(host, task, task_vars, play_context) 

_processed_results.extend(debug_closure(func)(self, iterator, one_pass)) 

break 

elif next_action.result == NextAction.CONTINUE: 

_processed_results.append(result) 

break 

elif next_action.result == NextAction.EXIT: 

# Matches KeyboardInterrupt from bin/ansible 

sys.exit(99) 

else: 

_processed_results.append(result) 

return _processed_results 

return inner 

class StrategyBase: 

''' 

This is the base class for strategy plugins, which contains some common 

code useful to all strategies like running handlers, cleanup actions, etc. 

''' 

def __init__(self, tqm): 

self._tqm = tqm 

self._inventory = tqm.get_inventory() 

self._workers = tqm.get_workers() 

self._notified_handlers = tqm._notified_handlers 

self._listening_handlers = tqm._listening_handlers 

self._variable_manager = tqm.get_variable_manager() 

self._loader = tqm.get_loader() 

self._final_q = tqm._final_q 

self._step = getattr(tqm._options, 'step', False) 

self._diff = getattr(tqm._options, 'diff', False) 

self.flush_cache = getattr(tqm._options, 'flush_cache', False) 

# the task cache is a dictionary of tuples of (host.name, task._uuid) 

# used to find the original task object of in-flight tasks and to store 

# the task args/vars and play context info used to queue the task. 

self._queued_task_cache = {} 

# Backwards compat: self._display isn't really needed, just import the global display and use that. 

self._display = display 

# internal counters 

self._pending_results = 0 

self._cur_worker = 0 

# this dictionary is used to keep track of hosts that have 

# outstanding tasks still in queue 

self._blocked_hosts = dict() 

self._results = deque() 

self._results_lock = threading.Condition(threading.Lock()) 

# create the result processing thread for reading results in the background 

self._results_thread = threading.Thread(target=results_thread_main, args=(self,)) 

self._results_thread.daemon = True 

self._results_thread.start() 

# holds the list of active (persistent) connections to be shutdown at 

# play completion 

self._active_connections = dict() 

self.debugger_active = C.ENABLE_TASK_DEBUGGER 

def cleanup(self): 

# close active persistent connections 

for sock in itervalues(self._active_connections): 

try: 

conn = Connection(sock) 

conn.reset() 

except ConnectionError as e: 

# most likely socket is already closed 

display.debug("got an error while closing persistent connection: %s" % e) 

self._final_q.put(_sentinel) 

self._results_thread.join() 

def run(self, iterator, play_context, result=0): 

# execute one more pass through the iterator without peeking, to 

# make sure that all of the hosts are advanced to their final task. 

# This should be safe, as everything should be ITERATING_COMPLETE by 

# this point, though the strategy may not advance the hosts itself. 

[iterator.get_next_task_for_host(host) for host in self._inventory.get_hosts(iterator._play.hosts) if host.name not in self._tqm._unreachable_hosts] 

# save the failed/unreachable hosts, as the run_handlers() 

# method will clear that information during its execution 

failed_hosts = iterator.get_failed_hosts() 

unreachable_hosts = self._tqm._unreachable_hosts.keys() 

display.debug("running handlers") 

handler_result = self.run_handlers(iterator, play_context) 

240 ↛ 241line 240 didn't jump to line 241, because the condition on line 240 was never true if isinstance(handler_result, bool) and not handler_result: 

result |= self._tqm.RUN_ERROR 

242 ↛ 247line 242 didn't jump to line 247, because the condition on line 242 was never false elif not handler_result: 

result |= handler_result 

# now update with the hosts (if any) that failed or were 

# unreachable during the handler execution phase 

failed_hosts = set(failed_hosts).union(iterator.get_failed_hosts()) 

unreachable_hosts = set(unreachable_hosts).union(self._tqm._unreachable_hosts.keys()) 

# return the appropriate code, depending on the status hosts after the run 

251 ↛ 252line 251 didn't jump to line 252, because the condition on line 251 was never true if not isinstance(result, bool) and result != self._tqm.RUN_OK: 

return result 

253 ↛ 254line 253 didn't jump to line 254, because the condition on line 253 was never true elif len(unreachable_hosts) > 0: 

return self._tqm.RUN_UNREACHABLE_HOSTS 

255 ↛ 256line 255 didn't jump to line 256, because the condition on line 255 was never true elif len(failed_hosts) > 0: 

return self._tqm.RUN_FAILED_HOSTS 

else: 

return self._tqm.RUN_OK 

def get_hosts_remaining(self, play): 

return [host for host in self._inventory.get_hosts(play.hosts) 

if host.name not in self._tqm._failed_hosts and host.name not in self._tqm._unreachable_hosts] 

def get_failed_hosts(self, play): 

return [host for host in self._inventory.get_hosts(play.hosts) if host.name in self._tqm._failed_hosts] 

def add_tqm_variables(self, vars, play): 

''' 

Base class method to add extra variables/information to the list of task 

vars sent through the executor engine regarding the task queue manager state. 

''' 

vars['ansible_current_hosts'] = [h.name for h in self.get_hosts_remaining(play)] 

vars['ansible_failed_hosts'] = [h.name for h in self.get_failed_hosts(play)] 

def _queue_task(self, host, task, task_vars, play_context): 

''' handles queueing the task up to be sent to a worker ''' 

display.debug("entering _queue_task() for %s/%s" % (host.name, task.action)) 

# Add a write lock for tasks. 

# Maybe this should be added somewhere further up the call stack but 

# this is the earliest in the code where we have task (1) extracted 

# into its own variable and (2) there's only a single code path 

# leading to the module being run. This is called by three 

# functions: __init__.py::_do_handler_run(), linear.py::run(), and 

# free.py::run() so we'd have to add to all three to do it there. 

# The next common higher level is __init__.py::run() and that has 

# tasks inside of play_iterator so we'd have to extract them to do it 

# there. 

if task.action not in action_write_locks.action_write_locks: 

display.debug('Creating lock for %s' % task.action) 

action_write_locks.action_write_locks[task.action] = Lock() 

# and then queue the new task 

try: 

# create a dummy object with plugin loaders set as an easier 

# way to share them with the forked processes 

shared_loader_obj = SharedPluginLoaderObj() 

queued = False 

starting_worker = self._cur_worker 

while True: 

(worker_prc, rslt_q) = self._workers[self._cur_worker] 

if worker_prc is None or not worker_prc.is_alive(): 

self._queued_task_cache[(host.name, task._uuid)] = { 

'host': host, 

'task': task, 

'task_vars': task_vars, 

'play_context': play_context 

} 

worker_prc = WorkerProcess(self._final_q, task_vars, host, task, play_context, self._loader, self._variable_manager, shared_loader_obj) 

self._workers[self._cur_worker][0] = worker_prc 

worker_prc.start() 

display.debug("worker is %d (out of %d available)" % (self._cur_worker + 1, len(self._workers))) 

queued = True 

self._cur_worker += 1 

320 ↛ 322line 320 didn't jump to line 322, because the condition on line 320 was never false if self._cur_worker >= len(self._workers): 

self._cur_worker = 0 

if queued: 

break 

324 ↛ 304line 324 didn't jump to line 304, because the condition on line 324 was never false elif self._cur_worker == starting_worker: 

time.sleep(0.0001) 

self._pending_results += 1 

except (EOFError, IOError, AssertionError) as e: 

# most likely an abort 

display.debug("got an error while queuing: %s" % e) 

return 

display.debug("exiting _queue_task() for %s/%s" % (host.name, task.action)) 

def get_task_hosts(self, iterator, task_host, task): 

335 ↛ 336line 335 didn't jump to line 336, because the condition on line 335 was never true if task.run_once: 

host_list = [host for host in self._inventory.get_hosts(iterator._play.hosts) if host.name not in self._tqm._unreachable_hosts] 

else: 

host_list = [task_host] 

return host_list 

def get_delegated_hosts(self, result, task): 

host_name = result.get('_ansible_delegated_vars', {}).get('ansible_delegated_host', None) 

if host_name is not None: 

actual_host = self._inventory.get_host(host_name) 

if actual_host is None: 

actual_host = Host(name=host_name) 

else: 

actual_host = Host(name=task.delegate_to) 

return [actual_host] 

@debug_closure 

def _process_pending_results(self, iterator, one_pass=False, max_passes=None): 

''' 

Reads results off the final queue and takes appropriate action 

based on the result (executing callbacks, updating state, etc.). 

''' 

ret_results = [] 

def get_original_host(host_name): 

# FIXME: this should not need x2 _inventory 

host_name = to_text(host_name) 

364 ↛ 367line 364 didn't jump to line 367, because the condition on line 364 was never false if host_name in self._inventory.hosts: 

return self._inventory.hosts[host_name] 

else: 

return self._inventory.get_host(host_name) 

def search_handler_blocks_by_name(handler_name, handler_blocks): 

for handler_block in handler_blocks: 

for handler_task in handler_block.block: 

if handler_task.name: 

handler_vars = self._variable_manager.get_vars(play=iterator._play, task=handler_task) 

templar = Templar(loader=self._loader, variables=handler_vars) 

try: 

# first we check with the full result of get_name(), which may 

# include the role name (if the handler is from a role). If that 

# is not found, we resort to the simple name field, which doesn't 

# have anything extra added to it. 

target_handler_name = templar.template(handler_task.name) 

if target_handler_name == handler_name: 

return handler_task 

else: 

target_handler_name = templar.template(handler_task.get_name()) 

if target_handler_name == handler_name: 

return handler_task 

except (UndefinedError, AnsibleUndefinedVariable): 

# We skip this handler due to the fact that it may be using 

# a variable in the name that was conditionally included via 

# set_fact or some other method, and we don't want to error 

# out unnecessarily 

continue 

return None 

def search_handler_blocks_by_uuid(handler_uuid, handler_blocks): 

for handler_block in handler_blocks: 

for handler_task in handler_block.block: 

if handler_uuid == handler_task._uuid: 

return handler_task 

return None 

def parent_handler_match(target_handler, handler_name): 

if target_handler: 

if isinstance(target_handler, (TaskInclude, IncludeRole)): 

try: 

handler_vars = self._variable_manager.get_vars(play=iterator._play, task=target_handler) 

templar = Templar(loader=self._loader, variables=handler_vars) 

target_handler_name = templar.template(target_handler.name) 

if target_handler_name == handler_name: 

return True 

else: 

target_handler_name = templar.template(target_handler.get_name()) 

if target_handler_name == handler_name: 

return True 

except (UndefinedError, AnsibleUndefinedVariable): 

pass 

return parent_handler_match(target_handler._parent, handler_name) 

else: 

return False 

cur_pass = 0 

while True: 

try: 

self._results_lock.acquire() 

task_result = self._results.popleft() 

except IndexError: 

break 

finally: 

self._results_lock.release() 

# get the original host and task. We then assign them to the TaskResult for use in callbacks/etc. 

original_host = get_original_host(task_result._host) 

queue_cache_entry = (original_host.name, task_result._task) 

found_task = self._queued_task_cache.get(queue_cache_entry)['task'] 

original_task = found_task.copy(exclude_parent=True, exclude_tasks=True) 

original_task._parent = found_task._parent 

original_task.from_attrs(task_result._task_fields) 

task_result._host = original_host 

task_result._task = original_task 

# get the correct loop var for use later 

if original_task.loop_control: 

loop_var = original_task.loop_control.loop_var 

else: 

loop_var = 'item' 

# send callbacks for 'non final' results 

449 ↛ 450line 449 didn't jump to line 450, because the condition on line 449 was never true if '_ansible_retry' in task_result._result: 

self._tqm.send_callback('v2_runner_retry', task_result) 

continue 

elif '_ansible_item_result' in task_result._result: 

453 ↛ 454line 453 didn't jump to line 454, because the condition on line 453 was never true if task_result.is_failed() or task_result.is_unreachable(): 

self._tqm.send_callback('v2_runner_item_on_failed', task_result) 

455 ↛ 456line 455 didn't jump to line 456, because the condition on line 455 was never true elif task_result.is_skipped(): 

self._tqm.send_callback('v2_runner_item_on_skipped', task_result) 

else: 

if 'diff' in task_result._result: 

459 ↛ 460line 459 didn't jump to line 460, because the condition on line 459 was never true if self._diff: 

self._tqm.send_callback('v2_on_file_diff', task_result) 

self._tqm.send_callback('v2_runner_item_on_ok', task_result) 

continue 

if original_task.register: 

host_list = self.get_task_hosts(iterator, original_host, original_task) 

clean_copy = strip_internal_keys(task_result._result) 

468 ↛ 471line 468 didn't jump to line 471, because the condition on line 468 was never false if 'invocation' in clean_copy: 

del clean_copy['invocation'] 

for target_host in host_list: 

self._variable_manager.set_nonpersistent_facts(target_host, {original_task.register: clean_copy}) 

# all host status messages contain 2 entries: (msg, task_result) 

role_ran = False 

if task_result.is_failed(): 

role_ran = True 

ignore_errors = original_task.ignore_errors 

479 ↛ 480line 479 didn't jump to line 480, because the condition on line 479 was never true if not ignore_errors: 

display.debug("marking %s as failed" % original_host.name) 

if original_task.run_once: 

# if we're using run_once, we have to fail every host here 

for h in self._inventory.get_hosts(iterator._play.hosts): 

if h.name not in self._tqm._unreachable_hosts: 

state, _ = iterator.get_next_task_for_host(h, peek=True) 

iterator.mark_host_failed(h) 

state, new_task = iterator.get_next_task_for_host(h, peek=True) 

else: 

iterator.mark_host_failed(original_host) 

# increment the failed count for this host 

self._tqm._stats.increment('failures', original_host.name) 

# grab the current state and if we're iterating on the rescue portion 

# of a block then we save the failed task in a special var for use 

# within the rescue/always 

state, _ = iterator.get_next_task_for_host(original_host, peek=True) 

if iterator.is_failed(original_host) and state and state.run_state == iterator.ITERATING_COMPLETE: 

self._tqm._failed_hosts[original_host.name] = True 

if state and state.run_state == iterator.ITERATING_RESCUE: 

self._variable_manager.set_nonpersistent_facts( 

original_host, 

dict( 

ansible_failed_task=original_task.serialize(), 

ansible_failed_result=task_result._result, 

), 

) 

else: 

self._tqm._stats.increment('ok', original_host.name) 

512 ↛ 513line 512 didn't jump to line 513, because the condition on line 512 was never true if 'changed' in task_result._result and task_result._result['changed']: 

self._tqm._stats.increment('changed', original_host.name) 

self._tqm.send_callback('v2_runner_on_failed', task_result, ignore_errors=ignore_errors) 

515 ↛ 516line 515 didn't jump to line 516, because the condition on line 515 was never true elif task_result.is_unreachable(): 

self._tqm._unreachable_hosts[original_host.name] = True 

iterator._play._removed_hosts.append(original_host.name) 

self._tqm._stats.increment('dark', original_host.name) 

self._tqm.send_callback('v2_runner_on_unreachable', task_result) 

elif task_result.is_skipped(): 

self._tqm._stats.increment('skipped', original_host.name) 

self._tqm.send_callback('v2_runner_on_skipped', task_result) 

else: 

role_ran = True 

if original_task.loop: 

# this task had a loop, and has more than one result, so 

# loop over all of them instead of a single result 

result_items = task_result._result.get('results', []) 

else: 

result_items = [task_result._result] 

for result_item in result_items: 

534 ↛ 535line 534 didn't jump to line 535, because the condition on line 534 was never true if '_ansible_notify' in result_item: 

if task_result.is_changed(): 

# The shared dictionary for notified handlers is a proxy, which 

# does not detect when sub-objects within the proxy are modified. 

# So, per the docs, we reassign the list so the proxy picks up and 

# notifies all other threads 

for handler_name in result_item['_ansible_notify']: 

found = False 

# Find the handler using the above helper. First we look up the 

# dependency chain of the current task (if it's from a role), otherwise 

# we just look through the list of handlers in the current play/all 

# roles and use the first one that matches the notify name 

target_handler = search_handler_blocks_by_name(handler_name, iterator._play.handlers) 

if target_handler is not None: 

found = True 

if target_handler._uuid not in self._notified_handlers: 

self._notified_handlers[target_handler._uuid] = [] 

if original_host not in self._notified_handlers[target_handler._uuid]: 

self._notified_handlers[target_handler._uuid].append(original_host) 

self._tqm.send_callback('v2_playbook_on_notify', target_handler, original_host) 

else: 

# As there may be more than one handler with the notified name as the 

# parent, so we just keep track of whether or not we found one at all 

for target_handler_uuid in self._notified_handlers: 

target_handler = search_handler_blocks_by_uuid(target_handler_uuid, iterator._play.handlers) 

if target_handler and parent_handler_match(target_handler, handler_name): 

found = True 

if original_host not in self._notified_handlers[target_handler._uuid]: 

self._notified_handlers[target_handler._uuid].append(original_host) 

self._tqm.send_callback('v2_playbook_on_notify', target_handler, original_host) 

if handler_name in self._listening_handlers: 

for listening_handler_uuid in self._listening_handlers[handler_name]: 

listening_handler = search_handler_blocks_by_uuid(listening_handler_uuid, iterator._play.handlers) 

if listening_handler is not None: 

found = True 

else: 

continue 

if original_host not in self._notified_handlers[listening_handler._uuid]: 

self._notified_handlers[listening_handler._uuid].append(original_host) 

self._tqm.send_callback('v2_playbook_on_notify', listening_handler, original_host) 

# and if none were found, then we raise an error 

if not found: 

msg = ("The requested handler '%s' was not found in either the main handlers list nor in the listening " 

"handlers list" % handler_name) 

if C.ERROR_ON_MISSING_HANDLER: 

raise AnsibleError(msg) 

else: 

display.warning(msg) 

585 ↛ 587line 585 didn't jump to line 587, because the condition on line 585 was never true if 'add_host' in result_item: 

# this task added a new host (add_host module) 

new_host_info = result_item.get('add_host', dict()) 

self._add_host(new_host_info, iterator) 

590 ↛ 592line 590 didn't jump to line 592, because the condition on line 590 was never true elif 'add_group' in result_item: 

# this task added a new group (group_by module) 

self._add_group(original_host, result_item) 

if 'ansible_facts' in result_item: 

# if delegated fact and we are delegating facts, we need to change target host for them 

597 ↛ 598line 597 didn't jump to line 598, because the condition on line 597 was never true if original_task.delegate_to is not None and original_task.delegate_facts: 

host_list = self.get_delegated_hosts(result_item, original_task) 

else: 

host_list = self.get_task_hosts(iterator, original_host, original_task) 

602 ↛ 603line 602 didn't jump to line 603, because the condition on line 602 was never true if original_task.action == 'include_vars': 

for (var_name, var_value) in iteritems(result_item['ansible_facts']): 

# find the host we're actually referring too here, which may 

# be a host that is not really in inventory at all 

for target_host in host_list: 

self._variable_manager.set_host_variable(target_host, var_name, var_value) 

else: 

cacheable = result_item.pop('_ansible_facts_cacheable', False) 

for target_host in host_list: 

611 ↛ 612line 611 didn't jump to line 612, because the condition on line 611 was never true if not original_task.action == 'set_fact' or cacheable: 

self._variable_manager.set_host_facts(target_host, result_item['ansible_facts'].copy()) 

613 ↛ 610line 613 didn't jump to line 610, because the condition on line 613 was never false if original_task.action == 'set_fact': 

self._variable_manager.set_nonpersistent_facts(target_host, result_item['ansible_facts'].copy()) 

616 ↛ 618line 616 didn't jump to line 618, because the condition on line 616 was never true if 'ansible_stats' in result_item and 'data' in result_item['ansible_stats'] and result_item['ansible_stats']['data']: 

if 'per_host' not in result_item['ansible_stats'] or result_item['ansible_stats']['per_host']: 

host_list = self.get_task_hosts(iterator, original_host, original_task) 

else: 

host_list = [None] 

data = result_item['ansible_stats']['data'] 

aggregate = 'aggregate' in result_item['ansible_stats'] and result_item['ansible_stats']['aggregate'] 

for myhost in host_list: 

for k in data.keys(): 

if aggregate: 

self._tqm._stats.update_custom_stats(k, data[k], myhost) 

else: 

self._tqm._stats.set_custom_stats(k, data[k], myhost) 

632 ↛ 633line 632 didn't jump to line 633, because the condition on line 632 was never true if 'diff' in task_result._result: 

if self._diff: 

self._tqm.send_callback('v2_on_file_diff', task_result) 

if not isinstance(original_task, TaskInclude): 

self._tqm._stats.increment('ok', original_host.name) 

if 'changed' in task_result._result and task_result._result['changed']: 

self._tqm._stats.increment('changed', original_host.name) 

# finally, send the ok for this task 

self._tqm.send_callback('v2_runner_on_ok', task_result) 

self._pending_results -= 1 

645 ↛ 650line 645 didn't jump to line 650, because the condition on line 645 was never false if original_host.name in self._blocked_hosts: 

del self._blocked_hosts[original_host.name] 

# If this is a role task, mark the parent role as being run (if 

# the task was ok or failed, but not skipped or unreachable) 

if original_task._role is not None and role_ran: # TODO: and original_task.action != 'include_role':? 

# lookup the role in the ROLE_CACHE to make sure we're dealing 

# with the correct object and mark it as executed 

for (entry, role_obj) in iteritems(iterator._play.ROLE_CACHE[original_task._role._role_name]): 

654 ↛ 653line 654 didn't jump to line 653, because the condition on line 654 was never false if role_obj._uuid == original_task._role._uuid: 

role_obj._had_task_run[original_host.name] = True 

ret_results.append(task_result) 

659 ↛ 660line 659 didn't jump to line 660, because the condition on line 659 was never true if one_pass or max_passes is not None and (cur_pass + 1) >= max_passes: 

break 

cur_pass += 1 

return ret_results 

def _wait_on_pending_results(self, iterator): 

''' 

Wait for the shared counter to drop to zero, using a short sleep 

between checks to ensure we don't spin lock 

''' 

ret_results = [] 

display.debug("waiting for pending results...") 

while self._pending_results > 0 and not self._tqm._terminated: 

677 ↛ 678line 677 didn't jump to line 678, because the condition on line 677 was never true if self._tqm.has_dead_workers(): 

raise AnsibleError("A worker was found in a dead state") 

results = self._process_pending_results(iterator) 

ret_results.extend(results) 

if self._pending_results > 0: 

time.sleep(C.DEFAULT_INTERNAL_POLL_INTERVAL) 

display.debug("no more pending results, returning what we have") 

return ret_results 

def _add_host(self, host_info, iterator): 

''' 

Helper function to add a new host to inventory based on a task result. 

''' 

if host_info: 

host_name = host_info.get('host_name') 

# Check if host in inventory, add if not 

if host_name not in self._inventory.hosts: 

self._inventory.add_host(host_name, 'all') 

new_host = self._inventory.hosts.get(host_name) 

# Set/update the vars for this host 

new_host.vars = combine_vars(new_host.get_vars(), host_info.get('host_vars', dict())) 

new_groups = host_info.get('groups', []) 

for group_name in new_groups: 

if group_name not in self._inventory.groups: 

self._inventory.add_group(group_name) 

new_group = self._inventory.groups[group_name] 

new_group.add_host(self._inventory.hosts[host_name]) 

# reconcile inventory, ensures inventory rules are followed 

self._inventory.reconcile_inventory() 

def _add_group(self, host, result_item): 

''' 

Helper function to add a group (if it does not exist), and to assign the 

specified host to that group. 

''' 

changed = False 

# the host here is from the executor side, which means it was a 

# serialized/cloned copy and we'll need to look up the proper 

# host object from the master inventory 

real_host = self._inventory.hosts[host.name] 

group_name = result_item.get('add_group') 

parent_group_names = result_item.get('parent_groups', []) 

for name in [group_name] + parent_group_names: 

if name not in self._inventory.groups: 

# create the new group and add it to inventory 

self._inventory.add_group(name) 

changed = True 

group = self._inventory.groups[group_name] 

for parent_group_name in parent_group_names: 

parent_group = self._inventory.groups[parent_group_name] 

parent_group.add_child_group(group) 

if real_host.name not in group.get_hosts(): 

group.add_host(real_host) 

changed = True 

if group_name not in host.get_groups(): 

real_host.add_group(group) 

changed = True 

if changed: 

self._inventory.reconcile_inventory() 

return changed 

def _load_included_file(self, included_file, iterator, is_handler=False): 

''' 

Loads an included YAML file of tasks, applying the optional set of variables. 

''' 

display.debug("loading included file: %s" % included_file._filename) 

try: 

data = self._loader.load_from_file(included_file._filename) 

761 ↛ 762line 761 didn't jump to line 762, because the condition on line 761 was never true if data is None: 

return [] 

763 ↛ 764line 763 didn't jump to line 764, because the condition on line 763 was never true elif not isinstance(data, list): 

raise AnsibleError("included task files must contain a list of tasks") 

ti_copy = included_file._task.copy(exclude_parent=True) 

ti_copy._parent = included_file._task._parent 

temp_vars = ti_copy.vars.copy() 

temp_vars.update(included_file._args) 

# pop tags out of the include args, if they were specified there, and assign 

# them to the include. If the include already had tags specified, we raise an 

# error so that users know not to specify them both ways