File Content: async_wrapper.py # -*- coding: utf-8 -*- # Copyright: (c) 2012, Michael DeHaan <michael.dehaan@gmail.com>, and others # GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) from __future__ import absolute_import, division, print_function __metaclass__ = type import errno import json import shlex import shutil import os import subprocess import sys import traceback import signal import time import syslog import multiprocessing from ansible.module_utils._text import to_text, to_bytes PY3 = sys.version_info[0] == 3 syslog.openlog('ansible-%s' % os.path.basename(__file__)) syslog.syslog(syslog.LOG_NOTICE, 'Invoked with %s' % " ".join(sys.argv[1:])) # pipe for communication between forked process and parent ipc_watcher, ipc_notifier = multiprocessing.Pipe() job_path = '' def notice(msg): syslog.syslog(syslog.LOG_NOTICE, msg) def end(res=None, exit_msg=0): if res is not None: print(json.dumps(res)) sys.stdout.flush() sys.exit(exit_msg) def daemonize_self(): # daemonizing code: http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/66012 try: pid = os.fork() if pid > 0: # exit first parent end() except OSError: e = sys.exc_info()[1] end({'msg': "fork #1 failed: %d (%s)\n" % (e.errno, e.strerror), 'failed': True}, 1) # decouple from parent environment (does not chdir / to keep the directory context the same as for non async tasks) os.setsid() os.umask(int('022', 8)) # do second fork try: pid = os.fork() if pid > 0: # TODO: print 'async_wrapper_pid': pid, but careful as it will pollute expected output. end() except OSError: e = sys.exc_info()[1] end({'msg': "fork #2 failed: %d (%s)\n" % (e.errno, e.strerror), 'failed': True}, 1) dev_null = open('/dev/null', 'w') os.dup2(dev_null.fileno(), sys.stdin.fileno()) os.dup2(dev_null.fileno(), sys.stdout.fileno()) os.dup2(dev_null.fileno(), sys.stderr.fileno()) # NB: this function copied from module_utils/json_utils.py. Ensure any changes are propagated there. # FUTURE: AnsibleModule-ify this module so it's Ansiballz-compatible and can use the module_utils copy of this function. def _filter_non_json_lines(data): ''' Used to filter unrelated output around module JSON output, like messages from tcagetattr, or where dropbear spews MOTD on every single command (which is nuts). Filters leading lines before first line-starting occurrence of '{', and filter all trailing lines after matching close character (working from the bottom of output). ''' warnings = [] # Filter initial junk lines = data.splitlines() for start, line in enumerate(lines): line = line.strip() if line.startswith(u'{'): break else: raise ValueError('No start of json char found') # Filter trailing junk lines = lines[start:] for reverse_end_offset, line in enumerate(reversed(lines)): if line.strip().endswith(u'}'): break else: raise ValueError('No end of json char found') if reverse_end_offset > 0: # Trailing junk is uncommon and can point to things the user might # want to change. So print a warning if we find any trailing_junk = lines[len(lines) - reverse_end_offset:] warnings.append('Module invocation had junk after the JSON data: %s' % '\n'.join(trailing_junk)) lines = lines[:(len(lines) - reverse_end_offset)] return ('\n'.join(lines), warnings) def _get_interpreter(module_path): with open(module_path, 'rb') as module_fd: head = module_fd.read(1024) if head[0:2] != b'#!': return None return head[2:head.index(b'\n')].strip().split(b' ') def _make_temp_dir(path): # TODO: Add checks for permissions on path. try: os.makedirs(path) except OSError as e: if e.errno != errno.EEXIST: raise def jwrite(info): jobfile = job_path + ".tmp" tjob = open(jobfile, "w") try: tjob.write(json.dumps(info)) except (IOError, OSError) as e: notice('failed to write to %s: %s' % (jobfile, str(e))) raise e finally: tjob.close() os.rename(jobfile, job_path) def _run_module(wrapped_cmd, jid): jwrite({"started": 1, "finished": 0, "ansible_job_id": jid}) result = {} # signal grandchild process started and isolated from being terminated # by the connection being closed sending a signal to the job group ipc_notifier.send(True) ipc_notifier.close() outdata = '' filtered_outdata = '' stderr = '' try: cmd = [to_bytes(c, errors='surrogate_or_strict') for c in shlex.split(wrapped_cmd)] # call the module interpreter directly (for non-binary modules) # this permits use of a script for an interpreter on non-Linux platforms interpreter = _get_interpreter(cmd[0]) if interpreter: cmd = interpreter + cmd script = subprocess.Popen(cmd, shell=False, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) (outdata, stderr) = script.communicate() if PY3: outdata = outdata.decode('utf-8', 'surrogateescape') stderr = stderr.decode('utf-8', 'surrogateescape') (filtered_outdata, json_warnings) = _filter_non_json_lines(outdata) result = json.loads(filtered_outdata) if json_warnings: # merge JSON junk warnings with any existing module warnings module_warnings = result.get('warnings', []) if not isinstance(module_warnings, list): module_warnings = [module_warnings] module_warnings.extend(json_warnings) result['warnings'] = module_warnings if stderr: result['stderr'] = stderr jwrite(result) except (OSError, IOError): e = sys.exc_info()[1] result = { "failed": 1, "cmd": wrapped_cmd, "msg": to_text(e), "outdata": outdata, # temporary notice only "stderr": stderr } result['ansible_job_id'] = jid jwrite(result) except (ValueError, Exception): result = { "failed": 1, "cmd": wrapped_cmd, "data": outdata, # temporary notice only "stderr": stderr, "msg": traceback.format_exc() } result['ansible_job_id'] = jid jwrite(result) def main(): if len(sys.argv) < 5: end({ "failed": True, "msg": "usage: async_wrapper <jid> <time_limit> <modulescript> <argsfile> [-preserve_tmp] " "Humans, do not call directly!" }, 1) jid = "%s.%d" % (sys.argv[1], os.getpid()) time_limit = sys.argv[2] wrapped_module = sys.argv[3] argsfile = sys.argv[4] if '-tmp-' not in os.path.dirname(wrapped_module): preserve_tmp = True elif len(sys.argv) > 5: preserve_tmp = sys.argv[5] == '-preserve_tmp' else: preserve_tmp = False # consider underscore as no argsfile so we can support passing of additional positional parameters if argsfile != '_': cmd = "%s %s" % (wrapped_module, argsfile) else: cmd = wrapped_module step = 5 async_dir = os.environ.get('ANSIBLE_ASYNC_DIR', '~/.ansible_async') # setup job output directory jobdir = os.path.expanduser(async_dir) global job_path job_path = os.path.join(jobdir, jid) try: _make_temp_dir(jobdir) except Exception as e: end({ "failed": 1, "msg": "could not create directory: %s - %s" % (jobdir, to_text(e)), "exception": to_text(traceback.format_exc()), }, 1) # immediately exit this process, leaving an orphaned process # running which immediately forks a supervisory timing process try: pid = os.fork() if pid: # Notify the overlord that the async process started # we need to not return immediately such that the launched command has an attempt # to initialize PRIOR to ansible trying to clean up the launch directory (and argsfile) # this probably could be done with some IPC later. Modules should always read # the argsfile at the very first start of their execution anyway # close off notifier handle in grandparent, probably unnecessary as # this process doesn't hang around long enough ipc_notifier.close() # allow waiting up to 2.5 seconds in total should be long enough for worst # loaded environment in practice. retries = 25 while retries > 0: if ipc_watcher.poll(0.1): break else: retries = retries - 1 continue notice("Return async_wrapper task started.") end({"failed": 0, "started": 1, "finished": 0, "ansible_job_id": jid, "results_file": job_path, "_ansible_suppress_tmpdir_delete": (not preserve_tmp)}, 0) else: # The actual wrapper process # close off the receiving end of the pipe from child process ipc_watcher.close() # Daemonize, so we keep on running daemonize_self() # we are now daemonized, create a supervisory process notice("Starting module and watcher") sub_pid = os.fork() if sub_pid: # close off inherited pipe handles ipc_watcher.close() ipc_notifier.close() # the parent stops the process after the time limit remaining = int(time_limit) # set the child process group id to kill all children os.setpgid(sub_pid, sub_pid) notice("Start watching %s (%s)" % (sub_pid, remaining)) time.sleep(step) while os.waitpid(sub_pid, os.WNOHANG) == (0, 0): notice("%s still running (%s)" % (sub_pid, remaining)) time.sleep(step) remaining = remaining - step if remaining <= 0: # ensure we leave response in poll location res = {'msg': 'Timeout exceeded', 'failed': True, 'child_pid': sub_pid} jwrite(res) # actually kill it notice("Timeout reached, now killing %s" % (sub_pid)) os.killpg(sub_pid, signal.SIGKILL) notice("Sent kill to group %s " % sub_pid) time.sleep(1) if not preserve_tmp: shutil.rmtree(os.path.dirname(wrapped_module), True) end(res) notice("Done in kid B.") if not preserve_tmp: shutil.rmtree(os.path.dirname(wrapped_module), True) end() else: # the child process runs the actual module notice("Start module (%s)" % os.getpid()) _run_module(cmd, jid) notice("Module complete (%s)" % os.getpid()) except Exception as e: notice("error: %s" % e) end({"failed": True, "msg": "FATAL ERROR: %s" % e}, "async_wrapper exited prematurely") if __name__ == '__main__': main() Submit