Source code for ansible_runner.runner

import os
import stat
import time
import json
import errno
import signal
from subprocess import Popen, PIPE
import shutil
import codecs
import collections
import datetime
import logging

import six
import pexpect
import psutil

import ansible_runner.plugins

from .utils import OutputEventFilter, cleanup_artifact_dir, ensure_str, collect_new_events
from .exceptions import CallbackError, AnsibleRunnerException
from ansible_runner.output import debug

logger = logging.getLogger('ansible-runner')


[docs]class Runner(object): def __init__(self, config, cancel_callback=None, remove_partials=True, event_handler=None, finished_callback=None, status_handler=None): self.config = config self.cancel_callback = cancel_callback self.event_handler = event_handler self.finished_callback = finished_callback self.status_handler = status_handler self.canceled = False self.timed_out = False self.errored = False self.status = "unstarted" self.rc = None self.remove_partials = remove_partials
[docs] def event_callback(self, event_data): ''' Invoked for every Ansible event to collect stdout with the event data and store it for later use ''' self.last_stdout_update = time.time() job_events_path = os.path.join(self.config.artifact_dir, 'job_events') if not os.path.exists(job_events_path): os.mkdir(job_events_path, 0o700) if 'uuid' in event_data: filename = '{}-partial.json'.format(event_data['uuid']) partial_filename = os.path.join(self.config.artifact_dir, 'job_events', filename) full_filename = os.path.join(self.config.artifact_dir, 'job_events', '{}-{}.json'.format(event_data['counter'], event_data['uuid'])) try: event_data.update(dict(runner_ident=str(self.config.ident))) try: with codecs.open(partial_filename, 'r', encoding='utf-8') as read_file: partial_event_data = json.load(read_file) event_data.update(partial_event_data) if self.remove_partials: os.remove(partial_filename) except IOError: debug("Failed to open ansible stdout callback plugin partial data file {}".format(partial_filename)) if self.event_handler is not None: should_write = self.event_handler(event_data) else: should_write = True for plugin in ansible_runner.plugins: ansible_runner.plugins[plugin].event_handler(self.config, event_data) if should_write: with codecs.open(full_filename, 'w', encoding='utf-8') as write_file: os.chmod(full_filename, stat.S_IRUSR | stat.S_IWUSR) json.dump(event_data, write_file) except IOError as e: debug("Failed writing event data: {}".format(e))
[docs] def status_callback(self, status): self.status = status status_data = dict(status=status, runner_ident=str(self.config.ident)) for plugin in ansible_runner.plugins: ansible_runner.plugins[plugin].status_handler(self.config, status_data) if self.status_handler is not None: self.status_handler(status_data, runner_config=self.config)
[docs] def run(self): ''' Launch the Ansible task configured in self.config (A RunnerConfig object), returns once the invocation is complete ''' self.status_callback('starting') stdout_filename = os.path.join(self.config.artifact_dir, 'stdout') command_filename = os.path.join(self.config.artifact_dir, 'command') try: os.makedirs(self.config.artifact_dir, mode=0o700) except OSError as exc: if exc.errno == errno.EEXIST and os.path.isdir(self.config.artifact_dir): pass else: raise os.close(os.open(stdout_filename, os.O_CREAT, stat.S_IRUSR | stat.S_IWUSR)) if six.PY2: command = [a.decode('utf-8') for a in self.config.command] else: command = self.config.command with codecs.open(command_filename, 'w', encoding='utf-8') as f: os.chmod(command_filename, stat.S_IRUSR | stat.S_IWUSR) json.dump( {'command': command, 'cwd': self.config.cwd, 'env': self.config.env}, f, ensure_ascii=False ) if self.config.ident is not None: cleanup_artifact_dir(os.path.join(self.config.artifact_dir, ".."), self.config.rotate_artifacts) stdout_handle = codecs.open(stdout_filename, 'w', encoding='utf-8') stdout_handle = OutputEventFilter(stdout_handle, self.event_callback, self.config.suppress_ansible_output, output_json=self.config.json_mode) if not isinstance(self.config.expect_passwords, collections.OrderedDict): # We iterate over `expect_passwords.keys()` and # `expect_passwords.values()` separately to map matched inputs to # patterns and choose the proper string to send to the subprocess; # enforce usage of an OrderedDict so that the ordering of elements in # `keys()` matches `values()`. expect_passwords = collections.OrderedDict(self.config.expect_passwords) password_patterns = list(expect_passwords.keys()) password_values = list(expect_passwords.values()) # pexpect needs all env vars to be utf-8 encoded bytes # https://github.com/pexpect/pexpect/issues/512 # Use a copy so as not to cause problems when serializing the job_env. env = { ensure_str(k): ensure_str(v) if k != 'PATH' and isinstance(v, six.text_type) else v for k, v in self.config.env.items() } # Prepare to collect performance data if self.config.resource_profiling: cgroup_path = '{0}/{1}'.format(self.config.resource_profiling_base_cgroup, self.config.ident) import getpass import grp user = getpass.getuser() group = grp.getgrgid(os.getgid()).gr_name cmd = 'cgcreate -a {user}:{group} -t {user}:{group} -g cpuacct,memory,pids:{}'.format(cgroup_path, user=user, group=group) proc = Popen(cmd, stdout=PIPE, stderr=PIPE, shell=True) _, stderr = proc.communicate() if proc.returncode: # Unable to create cgroup logger.error('Unable to create cgroup: {}'.format(stderr)) raise RuntimeError('Unable to create cgroup: {}'.format(stderr)) else: logger.info("Created cgroup '{}'".format(cgroup_path)) self.status_callback('running') self.last_stdout_update = time.time() try: child = pexpect.spawn( command[0], command[1:], cwd=self.config.cwd, env=env, ignore_sighup=True, encoding='utf-8', echo=False, use_poll=self.config.pexpect_use_poll, ) child.logfile_read = stdout_handle except pexpect.exceptions.ExceptionPexpect as e: child = collections.namedtuple( 'MissingProcess', 'exitstatus isalive close' )( exitstatus=127, isalive=lambda: False, close=lambda: None, ) def _decode(x): return x.decode('utf-8') if six.PY2 else x # create the events directory (the callback plugin won't run, so it # won't get created) events_directory = os.path.join(self.config.artifact_dir, 'job_events') if not os.path.exists(events_directory): os.mkdir(events_directory, 0o700) stdout_handle.write(_decode(str(e))) stdout_handle.write(_decode('\n')) job_start = time.time() while child.isalive(): result_id = child.expect(password_patterns, timeout=self.config.pexpect_timeout, searchwindowsize=100) password = password_values[result_id] if password is not None: child.sendline(password) self.last_stdout_update = time.time() if self.cancel_callback: try: self.canceled = self.cancel_callback() except Exception as e: # TODO: logger.exception('Could not check cancel callback - cancelling immediately') #if isinstance(extra_update_fields, dict): # extra_update_fields['job_explanation'] = "System error during job execution, check system logs" raise CallbackError("Exception in Cancel Callback: {}".format(e)) if self.config.job_timeout and not self.canceled and (time.time() - job_start) > self.config.job_timeout: self.timed_out = True # if isinstance(extra_update_fields, dict): # extra_update_fields['job_explanation'] = "Job terminated due to timeout" if self.canceled or self.timed_out or self.errored: Runner.handle_termination(child.pid, is_cancel=self.canceled) if self.config.idle_timeout and (time.time() - self.last_stdout_update) > self.config.idle_timeout: Runner.handle_termination(child.pid, is_cancel=False) self.timed_out = True stdout_handle.flush() stdout_handle.close() child.close() if self.canceled: self.status_callback('canceled') elif child.exitstatus == 0 and not self.timed_out: self.status_callback('successful') elif self.timed_out: self.status_callback('timeout') else: self.status_callback('failed') self.rc = child.exitstatus if not (self.timed_out or self.canceled) else 254 for filename, data in [ ('status', self.status), ('rc', self.rc), ]: artifact_path = os.path.join(self.config.artifact_dir, filename) if not os.path.exists(artifact_path): os.close(os.open(artifact_path, os.O_CREAT, stat.S_IRUSR | stat.S_IWUSR)) with open(artifact_path, 'w') as f: f.write(str(data)) if self.config.directory_isolation_path and self.config.directory_isolation_cleanup: shutil.rmtree(self.config.directory_isolation_path) if self.config.process_isolation and self.config.process_isolation_path_actual: def _delete(retries=15): try: shutil.rmtree(self.config.process_isolation_path_actual) except OSError as e: res = False if e.errno == 16 and retries > 0: time.sleep(1) res = _delete(retries=retries - 1) if not res: raise return True _delete() if self.config.resource_profiling: cmd = 'cgdelete -g cpuacct,memory,pids:{}'.format(cgroup_path) proc = Popen(cmd, stdout=PIPE, stderr=PIPE, shell=True) _, stderr = proc.communicate() if proc.returncode: logger.error('Failed to delete cgroup: {}'.format(stderr)) raise RuntimeError('Failed to delete cgroup: {}'.format(stderr)) if self.finished_callback is not None: try: self.finished_callback(self) except Exception as e: raise CallbackError("Exception in Finished Callback: {}".format(e)) return self.status, self.rc
@property def stdout(self): ''' Returns an open file handle to the stdout representing the Ansible run ''' stdout_path = os.path.join(self.config.artifact_dir, 'stdout') if not os.path.exists(stdout_path): raise AnsibleRunnerException("stdout missing") return open(os.path.join(self.config.artifact_dir, 'stdout'), 'r') @property def events(self): ''' A generator that will return all ansible job events in the order that they were emitted from Ansible Example: { "event":"runner_on_ok", "uuid":"00a50d9c-161a-4b74-b978-9f60becaf209", "stdout":"ok: [localhost] => {\\r\\n \\" msg\\":\\"Test!\\"\\r\\n}", "counter":6, "pid":740, "created":"2018-04-05T18:24:36.096725", "end_line":10, "start_line":7, "event_data":{ "play_pattern":"all", "play":"all", "task":"debug", "task_args":"msg=Test!", "remote_addr":"localhost", "res":{ "msg":"Test!", "changed":false, "_ansible_verbose_always":true, "_ansible_no_log":false }, "pid":740, "play_uuid":"0242ac11-0002-443b-cdb1-000000000006", "task_uuid":"0242ac11-0002-443b-cdb1-000000000008", "event_loop":null, "playbook_uuid":"634edeee-3228-4c17-a1b4-f010fdd42eb2", "playbook":"test.yml", "task_action":"debug", "host":"localhost", "task_path":"/tmp/demo/project/test.yml:3" } } ''' # collection of all the events that were yielded old_events = {} event_path = os.path.join(self.config.artifact_dir, 'job_events') # Wait for events dir to be created now = datetime.datetime.now() while not os.path.exists(event_path): time.sleep(0.05) wait_time = datetime.datetime.now() - now if wait_time.total_seconds() > 60: raise AnsibleRunnerException("events directory is missing: %s" % event_path) while self.status == "running": for event, old_evnts in collect_new_events(event_path, old_events): old_events = old_evnts yield event # collect new events that were written after the playbook has finished for event, old_evnts in collect_new_events(event_path, old_events): old_events = old_evnts yield event @property def stats(self): ''' Returns the final high level stats from the Ansible run Example: {'dark': {}, 'failures': {}, 'skipped': {}, 'ok': {u'localhost': 2}, 'processed': {u'localhost': 1}} ''' last_event = list(filter(lambda x: 'event' in x and x['event'] == 'playbook_on_stats', self.events)) if not last_event: return None last_event = last_event[0]['event_data'] return dict(skipped=last_event.get('skipped',{}), ok=last_event.get('ok',{}), dark=last_event.get('dark',{}), failures=last_event.get('failures',{}), processed=last_event.get('processed',{}), changed=last_event.get('changed',{}))
[docs] def host_events(self, host): ''' Given a host name, this will return all task events executed on that host ''' all_host_events = filter(lambda x: 'event_data' in x and 'host' in x['event_data'] and x['event_data']['host'] == host, self.events) return all_host_events
[docs] @classmethod def handle_termination(cls, pid, pidfile=None, is_cancel=True): ''' Internal method to terminate a subprocess spawned by `pexpect` representing an invocation of runner. :param pid: the process id of the running the job. :param pidfile: the daemon's PID file :param is_cancel: flag showing whether this termination is caused by instance's cancel_flag. ''' try: main_proc = psutil.Process(pid=pid) child_procs = main_proc.children(recursive=True) for child_proc in child_procs: try: os.kill(child_proc.pid, signal.SIGKILL) except (TypeError, OSError): pass os.kill(main_proc.pid, signal.SIGKILL) try: os.remove(pidfile) except (OSError): pass except (TypeError, psutil.Error, OSError): try: os.kill(pid, signal.SIGKILL) except (OSError): pass
[docs] def get_fact_cache(self, host): ''' Get the entire fact cache only if the fact_cache_type is 'jsonfile' ''' if self.config.fact_cache_type != 'jsonfile': raise Exception('Unsupported fact cache type. Only "jsonfile" is supported for reading and writing facts from ansible-runner') fact_cache = os.path.join(self.config.fact_cache, host) if os.path.exists(fact_cache): with open(fact_cache) as f: return json.loads(f.read()) return {}
[docs] def set_fact_cache(self, host, data): ''' Set the entire fact cache data only if the fact_cache_type is 'jsonfile' ''' if self.config.fact_cache_type != 'jsonfile': raise Exception('Unsupported fact cache type. Only "jsonfile" is supported for reading and writing facts from ansible-runner') fact_cache = os.path.join(self.config.fact_cache, host) if not os.path.exists(os.path.dirname(fact_cache)): os.makedirs(os.path.dirname(fact_cache), mode=0o700) with open(fact_cache, 'w') as f: return f.write(json.dumps(data))