PythonExecutor.py 8.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234
  1. #!/usr/bin/env python
  2. '''
  3. Licensed to the Apache Software Foundation (ASF) under one
  4. or more contributor license agreements. See the NOTICE file
  5. distributed with this work for additional information
  6. regarding copyright ownership. The ASF licenses this file
  7. to you under the Apache License, Version 2.0 (the
  8. "License"); you may not use this file except in compliance
  9. with the License. You may obtain a copy of the License at
  10. http://www.apache.org/licenses/LICENSE-2.0
  11. Unless required by applicable law or agreed to in writing, software
  12. distributed under the License is distributed on an "AS IS" BASIS,
  13. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14. See the License for the specific language governing permissions and
  15. limitations under the License.
  16. '''
  17. import json
  18. import logging
  19. import os
  20. import subprocess
  21. import pprint
  22. import threading
  23. import platform
  24. from threading import Thread
  25. import time
  26. from BackgroundCommandExecutionHandle import BackgroundCommandExecutionHandle
  27. from ambari_commons.os_check import OSConst, OSCheck
  28. from Grep import Grep
  29. import sys
  30. from ambari_commons import shell
  31. from ambari_commons.shell import shellRunner
  32. logger = logging.getLogger()
  33. class PythonExecutor:
  34. """
  35. Performs functionality for executing python scripts.
  36. Warning: class maintains internal state. As a result, instances should not be
  37. used as a singleton for a concurrent execution of python scripts
  38. """
  39. NO_ERROR = "none"
  40. grep = Grep()
  41. event = threading.Event()
  42. python_process_has_been_killed = False
  43. def __init__(self, tmpDir, config):
  44. self.tmpDir = tmpDir
  45. self.config = config
  46. pass
  47. def open_subprocess_files(self, tmpoutfile, tmperrfile, override_output_files):
  48. if override_output_files: # Recreate files
  49. tmpout = open(tmpoutfile, 'w')
  50. tmperr = open(tmperrfile, 'w')
  51. else: # Append to files
  52. tmpout = open(tmpoutfile, 'a')
  53. tmperr = open(tmperrfile, 'a')
  54. return tmpout, tmperr
  55. def run_file(self, script, script_params, tmp_dir, tmpoutfile, tmperrfile,
  56. timeout, tmpstructedoutfile, logger_level, callback, task_id,
  57. override_output_files = True, handle = None, log_info_on_failure=True):
  58. """
  59. Executes the specified python file in a separate subprocess.
  60. Method returns only when the subprocess is finished.
  61. Params arg is a list of script parameters
  62. Timeout meaning: how many seconds should pass before script execution
  63. is forcibly terminated
  64. override_output_files option defines whether stdout/stderr files will be
  65. recreated or appended.
  66. The structured out file, however, is preserved during multiple invocations that use the same file.
  67. """
  68. script_params += [tmpstructedoutfile, logger_level, tmp_dir]
  69. pythonCommand = self.python_command(script, script_params)
  70. logger.debug("Running command " + pprint.pformat(pythonCommand))
  71. if handle is None:
  72. tmpout, tmperr = self.open_subprocess_files(tmpoutfile, tmperrfile, override_output_files)
  73. process = self.launch_python_subprocess(pythonCommand, tmpout, tmperr)
  74. # map task_id to pid
  75. callback(task_id, process.pid)
  76. logger.debug("Launching watchdog thread")
  77. self.event.clear()
  78. self.python_process_has_been_killed = False
  79. thread = Thread(target = self.python_watchdog_func, args = (process, timeout))
  80. thread.start()
  81. # Waiting for the process to be either finished or killed
  82. process.communicate()
  83. self.event.set()
  84. thread.join()
  85. result = self.prepare_process_result(process, tmpoutfile, tmperrfile, tmpstructedoutfile, timeout=timeout)
  86. if log_info_on_failure and result['exitcode']:
  87. self.on_failure(pythonCommand, result)
  88. return result
  89. else:
  90. holder = Holder(pythonCommand, tmpoutfile, tmperrfile, tmpstructedoutfile, handle)
  91. background = BackgroundThread(holder, self)
  92. background.start()
  93. return {"exitcode": 777}
  94. def on_failure(self, pythonCommand, result):
  95. """
  96. Log some useful information after task failure.
  97. """
  98. logger.info("Command " + pprint.pformat(pythonCommand) + " failed with exitcode=" + str(result['exitcode']))
  99. cmd_list = ["ps faux", "netstat -tulpn"]
  100. shell_runner = shellRunner()
  101. for cmd in cmd_list:
  102. ret = shell_runner.run(cmd)
  103. logger.info("Command '{0}' returned {1}. {2}{3}".format(cmd, ret["exitCode"], ret["error"], ret["output"]))
  104. def prepare_process_result(self, process, tmpoutfile, tmperrfile, tmpstructedoutfile, timeout=None):
  105. out, error, structured_out = self.read_result_from_files(tmpoutfile, tmperrfile, tmpstructedoutfile)
  106. # Building results
  107. returncode = process.returncode
  108. if self.python_process_has_been_killed:
  109. error = str(error) + "\n Python script has been killed due to timeout" + \
  110. (" after waiting %s secs" % str(timeout) if timeout else "")
  111. returncode = 999
  112. result = self.condenseOutput(out, error, returncode, structured_out)
  113. logger.debug("Result: %s" % result)
  114. return result
  115. def read_result_from_files(self, out_path, err_path, structured_out_path):
  116. out = open(out_path, 'r').read()
  117. error = open(err_path, 'r').read()
  118. try:
  119. with open(structured_out_path, 'r') as fp:
  120. structured_out = json.load(fp)
  121. except Exception:
  122. if os.path.exists(structured_out_path):
  123. errMsg = 'Unable to read structured output from ' + structured_out_path
  124. structured_out = {
  125. 'msg' : errMsg
  126. }
  127. logger.warn(structured_out)
  128. else:
  129. structured_out = {}
  130. return out, error, structured_out
  131. def launch_python_subprocess(self, command, tmpout, tmperr):
  132. """
  133. Creates subprocess with given parameters. This functionality was moved to separate method
  134. to make possible unit testing
  135. """
  136. close_fds = None if OSCheck.get_os_family() == OSConst.WINSRV_FAMILY else True
  137. if OSCheck.get_os_family() == OSConst.WINSRV_FAMILY:
  138. command_env = dict(os.environ)
  139. command_env["PYTHONPATH"] = os.pathsep.join(sys.path)
  140. for k, v in command_env.iteritems():
  141. command_env[k] = str(v)
  142. else:
  143. command_env = None
  144. return subprocess.Popen(command,
  145. stdout=tmpout,
  146. stderr=tmperr, close_fds=close_fds, env=command_env)
  147. def isSuccessfull(self, returncode):
  148. return not self.python_process_has_been_killed and returncode == 0
  149. def python_command(self, script, script_params):
  150. #we need manually pass python executable on windows because sys.executable will return service wrapper
  151. python_binary = os.environ['PYTHON_EXE'] if 'PYTHON_EXE' in os.environ else sys.executable
  152. python_command = [python_binary, script] + script_params
  153. return python_command
  154. def condenseOutput(self, stdout, stderr, retcode, structured_out):
  155. log_lines_count = self.config.get('heartbeat', 'log_lines_count')
  156. grep = self.grep
  157. result = {
  158. "exitcode": retcode,
  159. "stdout": grep.tail(stdout, log_lines_count) if log_lines_count else stdout,
  160. "stderr": grep.tail(stderr, log_lines_count) if log_lines_count else stderr,
  161. "structuredOut" : structured_out
  162. }
  163. return result
  164. def python_watchdog_func(self, python, timeout):
  165. self.event.wait(timeout)
  166. if python.returncode is None:
  167. logger.error("Subprocess timed out and will be killed")
  168. shell.kill_process_with_children(python.pid)
  169. self.python_process_has_been_killed = True
  170. pass
  171. class Holder:
  172. def __init__(self, command, out_file, err_file, structured_out_file, handle):
  173. self.command = command
  174. self.out_file = out_file
  175. self.err_file = err_file
  176. self.structured_out_file = structured_out_file
  177. self.handle = handle
  178. class BackgroundThread(threading.Thread):
  179. def __init__(self, holder, pythonExecutor):
  180. threading.Thread.__init__(self)
  181. self.holder = holder
  182. self.pythonExecutor = pythonExecutor
  183. def run(self):
  184. process_out, process_err = self.pythonExecutor.open_subprocess_files(self.holder.out_file, self.holder.err_file, True)
  185. logger.debug("Starting process command %s" % self.holder.command)
  186. process = self.pythonExecutor.launch_python_subprocess(self.holder.command, process_out, process_err)
  187. logger.debug("Process has been started. Pid = %s" % process.pid)
  188. self.holder.handle.pid = process.pid
  189. self.holder.handle.status = BackgroundCommandExecutionHandle.RUNNING_STATUS
  190. self.holder.handle.on_background_command_started(self.holder.handle.command['taskId'], process.pid)
  191. process.communicate()
  192. self.holder.handle.exitCode = process.returncode
  193. process_condensed_result = self.pythonExecutor.prepare_process_result(process, self.holder.out_file, self.holder.err_file, self.holder.structured_out_file)
  194. logger.debug("Calling callback with args %s" % process_condensed_result)
  195. self.holder.handle.on_background_command_complete_callback(process_condensed_result, self.holder.handle)
  196. logger.debug("Exiting from thread for holder pid %s" % self.holder.handle.pid)