import os import stat import time import json import errno import signal from subprocess import Popen, PIPE, CalledProcessError, TimeoutExpired, run as run_subprocess import shutil import codecs import collections import datetime import logging import traceback import pexpect import ansible_runner.plugins from ansible_runner.output import debug from .utils import OutputEventFilter, cleanup_artifact_dir, ensure_str, collect_new_events from .exceptions import CallbackError, AnsibleRunnerException logger = logging.getLogger('ansible-runner') class Runner: def __init__(self, config, cancel_callback=None, remove_partials=True, event_handler=None, artifacts_handler=None, finished_callback=None, status_handler=None): self.config = config self.cancel_callback = cancel_callback self.event_handler = event_handler self.artifacts_handler = artifacts_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 self.last_stdout_update = 0.0 # default runner mode to pexpect self.runner_mode = self.config.runner_mode if hasattr(self.config, 'runner_mode') else 'pexpect' self.directory_isolation_path = self.config.directory_isolation_path if hasattr(self.config, 'directory_isolation_path') else None self.directory_isolation_cleanup = self.config.directory_isolation_cleanup if hasattr(self.config, 'directory_isolation_cleanup') else None self.process_isolation = self.config.process_isolation if hasattr(self.config, 'process_isolation') else None self.process_isolation_path_actual = self.config.process_isolation_path_actual if hasattr(self.config, 'process_isolation_path_actual') else None 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() if 'uuid' in event_data: filename = f"{event_data['uuid']}-partial.json" partial_filename = os.path.join(self.config.artifact_dir, 'job_events', filename) full_filename = os.path.join(self.config.artifact_dir, 'job_events', f"{event_data['counter']}-{event_data['uuid']}.json" ) try: event_data.update({'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 as e: msg = "Failed to open ansible stdout callback plugin partial data" \ f" file {partial_filename} with error {str(e)}" debug(msg) if self.config.check_job_event_data: raise AnsibleRunnerException(msg) from e # prefer 'created' from partial data, but verbose events set time here if 'created' not in event_data: event_data['created'] = datetime.datetime.now(datetime.timezone.utc).isoformat() 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: temporary_filename = full_filename + '.tmp' with codecs.open(temporary_filename, 'w', encoding='utf-8') as write_file: os.chmod(temporary_filename, stat.S_IRUSR | stat.S_IWUSR) json.dump(event_data, write_file) os.rename(temporary_filename, full_filename) except IOError as e: debug(f"Failed writing event data: {e}") def status_callback(self, status): self.status = status status_data = {'status': status, 'runner_ident': str(self.config.ident)} if status == 'starting': status_data.update({'command': self.config.command, 'env': self.config.env, 'cwd': self.config.cwd}) 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) def run(self): ''' Launch the Ansible task configured in self.config (A RunnerConfig object), returns once the invocation is complete ''' # pylint: disable=R1732 password_patterns = [] password_values = [] self.status_callback('starting') 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 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) 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) if hasattr(self.config, 'suppress_ansible_output'): suppress_ansible_output = self.config.suppress_ansible_output else: suppress_ansible_output = False if not self.config.suppress_output_file: stdout_filename = os.path.join(self.config.artifact_dir, 'stdout') stderr_filename = os.path.join(self.config.artifact_dir, 'stderr') os.close(os.open(stdout_filename, os.O_CREAT, stat.S_IRUSR | stat.S_IWUSR)) stdout_handle = codecs.open(stdout_filename, 'w', encoding='utf-8') stderr_handle = codecs.open(stderr_filename, 'w', encoding='utf-8') else: stdout_handle = None stderr_handle = None stdout_handle = OutputEventFilter(stdout_handle, self.event_callback, suppress_ansible_output, output_json=self.config.json_mode) stderr_handle = OutputEventFilter(stderr_handle, self.event_callback, suppress_ansible_output, output_json=self.config.json_mode) if self.runner_mode == 'pexpect' and 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. if self.config.containerized: # We call the actual docker or podman executable right where we are cwd = os.getcwd() # If this is containerized, the shell environment calling podman has little # to do with the actual job environment, but still needs PATH, auth, etc. pexpect_env = os.environ.copy() # But we still rely on env vars to pass secrets pexpect_env.update(self.config.env) # Write the keys to pass into container to expected file in artifacts dir # option expecting should have already been written in ansible_runner.config.runner env_file_host = os.path.join(self.config.artifact_dir, 'env.list') with open(env_file_host, 'w') as f: f.write( '\n'.join( [f"{key}={value}" for key, value in self.config.env.items()] ) ) else: cwd = self.config.cwd pexpect_env = self.config.env env = { ensure_str(k): ensure_str(v) if k != 'PATH' and isinstance(v, str) else v for k, v in pexpect_env.items() } self.status_callback('running') self.last_stdout_update = time.time() # The subprocess runner interface provides stdin/stdout/stderr with streaming capability # to the caller if input_fd/output_fd/error_fd is passed to config class. # Alsp, provides an workaround for known issue in pexpect for long running non-interactive process # https://pexpect.readthedocs.io/en/stable/commonissues.html#truncated-output-just-before-child-exits if self.runner_mode == 'subprocess': if hasattr(self.config, 'input_fd') and self.config.input_fd: input_fd = self.config.input_fd else: input_fd = None if hasattr(self.config, 'output_fd') and self.config.output_fd: output_fd = self.config.output_fd else: output_fd = PIPE if hasattr(self.config, 'error_fd') and self.config.error_fd: error_fd = self.config.error_fd else: error_fd = PIPE subprocess_timeout = self.config.subprocess_timeout if hasattr(self.config, 'subprocess_timeout') else None try: stdout_response = '' stderr_response = '' kwargs = { 'cwd': cwd, 'env': env, 'stdin': input_fd, 'stdout': output_fd, 'stderr': error_fd, 'universal_newlines': True, } if subprocess_timeout is not None: kwargs.update({'timeout': subprocess_timeout}) proc_out = run_subprocess(command, check=True, **kwargs) stdout_response = proc_out.stdout stderr_response = proc_out.stderr self.rc = proc_out.returncode except CalledProcessError as exc: logger.debug("%s execution failed, returncode: %s, output: %s, stdout: %s, stderr: %s", exc.cmd, exc.returncode, exc.output, exc.stdout, exc.stderr) self.rc = exc.returncode self.errored = True stdout_response = exc.stdout stderr_response = exc.stderr except TimeoutExpired as exc: logger.debug("%s execution timedout, timeout: %s, output: %s, stdout: %s, stderr: %s", exc.cmd, exc.timeout, exc.output, exc.stdout, exc.stderr) self.rc = 254 stdout_response = exc.stdout stderr_response = exc.stderr self.timed_out = True except Exception as exc: stderr_response = traceback.format_exc() self.rc = 254 self.errored = True logger.debug("received exception: %s", exc) if self.timed_out or self.errored: self.kill_container() if stdout_response is not None: if isinstance(stdout_response, bytes): stdout_response = stdout_response.decode() stdout_handle.write(stdout_response) if stderr_response is not None: if isinstance(stderr_response, bytes): stderr_response = stderr_response.decode() stderr_handle.write(stderr_response) stdout_handle.close() stderr_handle.close() else: try: child = pexpect.spawn( command[0], command[1:], cwd=cwd, env=env, ignore_sighup=True, encoding='utf-8', codec_errors='replace', 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 expect close' )( exitstatus=127, isalive=lambda: False, expect=lambda *args, **kwargs: None, close=lambda: None, ) # 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(str(e)) stdout_handle.write('\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: stdout_handle.close() stderr_handle.close() # 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(f"Exception in Cancel Callback: {e}") from 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: self.kill_container() Runner.handle_termination(child.pid) if self.config.idle_timeout and (time.time() - self.last_stdout_update) > self.config.idle_timeout: self.kill_container() Runner.handle_termination(child.pid) self.timed_out = True # fix for https://github.com/ansible/ansible-runner/issues/1330 # Since we're (ab)using pexpect's logging callback as our source of stdout data, we need to pump the stream one last # time, in case any new output was written by the child between the last return from expect and its termination. Ideally # this would have an arbitrarily large timeout value as well, in case a ridiculous amount of data was written, but just # invoking one last pump should cover the vast majority of real-world cases. child.expect(pexpect.EOF, timeout=5) # close the child to ensure no more output will be written before we close the stream interposers child.close() stdout_handle.close() stderr_handle.close() self.rc = child.exitstatus if not (self.timed_out or self.canceled) else 254 if self.canceled: self.status_callback('canceled') elif self.rc == 0 and not self.timed_out: self.status_callback('successful') elif self.timed_out: self.status_callback('timeout') else: self.status_callback('failed') 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.directory_isolation_path and self.directory_isolation_cleanup: shutil.rmtree(self.directory_isolation_path) if self.process_isolation and self.process_isolation_path_actual: def _delete(retries=15): try: shutil.rmtree(self.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.artifacts_handler is not None: try: self.artifacts_handler(self.config.artifact_dir) except Exception as e: raise CallbackError(f"Exception in Artifact Callback: {e}") from e if self.finished_callback is not None: try: self.finished_callback(self) except Exception as e: raise CallbackError(f"Exception in Finished Callback: {e}") from 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 stderr(self): ''' Returns an open file handle to the stderr representing the Ansible run ''' stderr_path = os.path.join(self.config.artifact_dir, 'stderr') if not os.path.exists(stderr_path): raise AnsibleRunnerException("stderr missing") return open(os.path.join(self.config.artifact_dir, 'stderr'), 'r') @property def events(self): ''' A generator that will return all ansible job events in the order that they were emitted from Ansible :Example: .. code-block:: { "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(f"events directory is missing: {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 {'skipped': last_event.get('skipped', {}), 'ok': last_event.get('ok', {}), 'dark': last_event.get('dark', {}), 'failures': last_event.get('failures', {}), 'ignored': last_event.get('ignored', {}), 'rescued': last_event.get('rescued', {}), 'processed': last_event.get('processed', {}), 'changed': last_event.get('changed', {}) } 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 def kill_container(self): ''' Internal method to terminate a container being used for job isolation ''' container_name = self.config.container_name if container_name: container_cli = self.config.process_isolation_executable cmd = [container_cli, 'kill', container_name] with Popen(cmd, stdout=PIPE, stderr=PIPE) as proc: _, stderr = proc.communicate() if proc.returncode: logger.info("Error from %s kill %s command:\n%s", container_cli, container_name, stderr) else: logger.info("Killed container %s", container_name) @classmethod def handle_termination(cls, pid, pidfile=None): ''' 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 ''' try: pgroup = os.getpgid(pid) os.killpg(pgroup, signal.SIGKILL) except (OSError, ProcessLookupError): pass try: os.remove(pidfile) except (TypeError, OSError): pass 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 {} 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))