supervisor.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406
  1. #!/usr/bin/env python
  2. # Licensed to Cloudera, Inc. under one
  3. # or more contributor license agreements. See the NOTICE file
  4. # distributed with this work for additional information
  5. # regarding copyright ownership. Cloudera, Inc. licenses this file
  6. # to you under the Apache License, Version 2.0 (the
  7. # "License"); you may not use this file except in compliance
  8. # with the License. You may obtain a copy of the License at
  9. #
  10. # http://www.apache.org/licenses/LICENSE-2.0
  11. #
  12. # Unless required by applicable law or agreed to in writing, software
  13. # distributed under the License is distributed on an "AS IS" BASIS,
  14. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. # See the License for the specific language governing permissions and
  16. # limitations under the License.
  17. """
  18. Simple supervisor application that watches subprocesses and restarts them.
  19. If a process appears to be continuously failing, it will kill the entire
  20. supervisor, hopefully triggering some kind of external monitoring system.
  21. This is usually preferable compared to a partially-up system.
  22. This is heavily modeled after the one_for_one supervision policy in Erlang/OTP:
  23. http://erlang.org/documentation/doc-4.8.2/doc/design_principles/sup_princ.html
  24. In order to have your application managed by supervisor, you need to add
  25. an entry_point to your application's egg with the name 'desktop.supervisor.specs'.
  26. This entry point should point to a SuperviseeSpec instance in your module.
  27. """
  28. from __future__ import print_function
  29. from builtins import range
  30. from builtins import object
  31. import daemon
  32. import grp
  33. import logging
  34. import optparse
  35. import os
  36. import pkg_resources
  37. import pwd
  38. import signal
  39. import subprocess
  40. import sys
  41. import threading
  42. import time
  43. import desktop.lib.daemon_utils
  44. import desktop.lib.paths
  45. import desktop.log
  46. if sys.version_info[0] > 2:
  47. from daemon.pidfile import TimeoutPIDLockFile
  48. from daemon.daemon import DaemonContext
  49. open_file = open
  50. else:
  51. from daemon.pidlockfile import PIDLockFile
  52. open_file = file
  53. class TimeoutPIDLockFile(PIDLockFile):
  54. """A PIDLockFile subclass that passes through a timeout on acquisition."""
  55. def __init__(self, lockfile, timeout, **kwargs):
  56. PIDLockFile.__init__(self, lockfile, **kwargs)
  57. self.timeout = timeout
  58. def __enter__(self):
  59. super(TimeoutPIDLockFile, self).acquire(timeout=self.timeout)
  60. return self
  61. PROC_NAME = 'supervisor'
  62. LOG = logging.getLogger()
  63. # If a process restarts mre than MAX_RESTARTS_IN_WINDOW times
  64. # within TIME_WINDOW number of seconds, the supervisor shuts down
  65. TIME_WINDOW = 120
  66. MAX_RESTARTS_IN_WINDOW = 3
  67. # User and group to setuid/setgid down to for any supervisees that don't have
  68. # the drop_root option set to False
  69. SETUID_USER = "hue"
  70. SETGID_GROUP = "hue"
  71. g_user_uid = None # We figure out the numeric uid/gid later
  72. g_user_gid = None
  73. # The entry point group in which to find processes to supervise.
  74. ENTRY_POINT_GROUP = "desktop.supervisor.specs"
  75. # How long to wait while trying to acquire the supervisor pid lock
  76. # file. We shouldn't spin long here - we'd rather fail to start up.
  77. LOCKFILE_TIMEOUT = 2
  78. # The hue program
  79. HUE_BIN = os.path.join(desktop.lib.paths.get_run_root(), 'build', 'env', 'bin', 'hue')
  80. ######
  81. CHILD_PIDS = []
  82. SHOULD_STOP = False
  83. # How long to wait for children processes to die gracefully before
  84. # killing them forceably (seconds)
  85. WAIT_FOR_DEATH = 5
  86. class SuperviseeSpec(object):
  87. """
  88. A specification of something that should be supervised.
  89. Instances should have a .cmdv property which returns a list
  90. which will be passed through to subprocess.call.
  91. """
  92. def __init__(self, drop_root=True):
  93. """
  94. @param drop_root: if True, the supervisor will drop root privileges
  95. before calling the subprocess.
  96. """
  97. self.drop_root = drop_root
  98. class DjangoCommandSupervisee(SuperviseeSpec):
  99. """A supervisee which is simply a desktop django management command."""
  100. def __init__(self, django_command, **kwargs):
  101. SuperviseeSpec.__init__(self, **kwargs)
  102. self.django_command = django_command
  103. @property
  104. def cmdv(self):
  105. return [HUE_BIN, self.django_command]
  106. class Supervisor(threading.Thread):
  107. """A thread responsible for keeping the supervised subprocess running"""
  108. # States of the subprocess
  109. STATES = (PENDING, RUNNING, FINISHED, ERROR) = list(range(4))
  110. def __init__(self, cmdv, **kwargs):
  111. super(Supervisor, self).__init__()
  112. self.cmdv = cmdv
  113. self.popen_kwargs = kwargs
  114. self.state = Supervisor.PENDING
  115. def run(self):
  116. global CHILD_PIDS
  117. try:
  118. restart_timestamps = []
  119. proc_str = " ".join(self.cmdv)
  120. while True:
  121. self.state = Supervisor.RUNNING
  122. LOG.info("Starting process %s" % proc_str)
  123. pipe = subprocess.Popen(self.cmdv, close_fds=True, stdin=open_file("/dev/null"), **self.popen_kwargs)
  124. LOG.info("Started proceses (pid %s) %s" % (pipe.pid, proc_str))
  125. CHILD_PIDS.append(pipe.pid)
  126. exitcode = pipe.wait()
  127. if exitcode == 0:
  128. LOG.info('Command "%s" exited normally.' % (proc_str,))
  129. self.state = Supervisor.FINISHED
  130. return
  131. if exitcode != 0:
  132. LOG.warning("Exit code for %s: %d" % (proc_str, exitcode))
  133. self.state = Supervisor.ERROR
  134. et = time.time()
  135. if SHOULD_STOP:
  136. LOG.info("Stopping %s because supervisor exiting" % proc_str)
  137. self.state = Supervisor.FINISHED
  138. return
  139. restart_timestamps.append(et)
  140. restart_timestamps = [t for t in restart_timestamps if t > et - TIME_WINDOW]
  141. if len(restart_timestamps) > MAX_RESTARTS_IN_WINDOW:
  142. earliest_restart = min(restart_timestamps)
  143. ago = et - earliest_restart
  144. LOG.error(
  145. "Process %s has restarted more than %d times in the last %d seconds" % (
  146. proc_str, MAX_RESTARTS_IN_WINDOW, int(ago))
  147. )
  148. self.state = Supervisor.ERROR
  149. return
  150. LOG.error("Process %s exited abnormally. Restarting it." % (proc_str,))
  151. except BaseException as ex:
  152. LOG.exception("Uncaught exception. Supervisor exiting.")
  153. self.state = Supervisor.ERROR
  154. def shutdown(sups):
  155. global SHOULD_STOP
  156. SHOULD_STOP = True
  157. LOG.warning("Supervisor shutting down!")
  158. for pid in CHILD_PIDS:
  159. try:
  160. os.kill(pid, signal.SIGINT)
  161. except OSError:
  162. pass
  163. LOG.warning("Waiting for children to exit for %d seconds..." % WAIT_FOR_DEATH)
  164. t = time.time()
  165. still_alive = False
  166. while time.time() < t + WAIT_FOR_DEATH:
  167. still_alive = False
  168. for sup in sups:
  169. sup.join(0.2)
  170. still_alive = still_alive or sup.is_alive()
  171. if not still_alive:
  172. break
  173. if still_alive:
  174. LOG.warning("Children have not exited after %d seconds. Killing them with SIGKILL." %
  175. WAIT_FOR_DEATH)
  176. for pid in CHILD_PIDS:
  177. try:
  178. os.kill(pid, signal.SIGKILL)
  179. except OSError:
  180. pass
  181. sys.exit(1)
  182. def sig_handler(signum, frame):
  183. raise SystemExit("Signal %d received. Exiting" % signum)
  184. def parse_args():
  185. parser = optparse.OptionParser()
  186. parser.add_option("-d", "--daemon", dest="daemonize", action="store_true", default=False)
  187. parser.add_option("-p", "--pid-file", dest="pid_file", metavar="PID_FILE", default="supervisor.pid")
  188. parser.add_option("-l", "--log-dir", dest="log_dir", metavar="DIR", default="logs")
  189. parser.add_option(
  190. '-e', '--exclude', dest='supervisee_exclusions', metavar='EXCLUSIONS', default=[], action='append',
  191. help='Command NOT to run from supervisor. May be included more than once.'
  192. )
  193. parser.add_option('-s', '--show', dest='show_supervisees', action='store_true', default=False)
  194. parser.add_option('-u', '--user', dest='user', action='store', default=SETUID_USER)
  195. parser.add_option('-g', '--group', dest='group', action='store', default=SETGID_GROUP)
  196. (options, args) = parser.parse_args()
  197. return options
  198. def get_pid_cmdline(pid):
  199. return subprocess.Popen(["ps", "-p", str(pid), "-o", "cmd", "h"], stdout=subprocess.PIPE, close_fds=True).communicate()[0]
  200. def get_supervisees():
  201. """Pull the supervisor specifications out of the entry point."""
  202. eps = list(pkg_resources.iter_entry_points(ENTRY_POINT_GROUP))
  203. return dict((ep.name, ep.load()) for ep in eps)
  204. def setup_user_info():
  205. """Translate the user/group info into uid/gid."""
  206. if os.geteuid() != 0:
  207. return
  208. global g_user_uid, g_user_gid
  209. g_user_uid, g_user_gid = desktop.lib.daemon_utils.get_uid_gid(SETUID_USER, SETGID_GROUP)
  210. def drop_privileges():
  211. """Drop root privileges down to the specified SETUID_USER.
  212. N.B. DO NOT USE THE logging MODULE FROM WITHIN THIS FUNCTION.
  213. This function is run in forked processes right before it calls
  214. exec, but the fork may have occured while a different thread
  215. had locked the log. Since it's a forked process, the log will
  216. be locked forever in the subprocess and thus a logging.X may
  217. block forever.
  218. """
  219. we_are_root = os.getuid() == 0
  220. if not we_are_root:
  221. print("[INFO] Not running as root, skipping privilege drop", file=sys.stdout)
  222. return
  223. try:
  224. pw = pwd.getpwnam(SETUID_USER)
  225. except KeyError:
  226. print("[ERROR] Couldn't get user information for user " + SETUID_USER, file=sys.stderr)
  227. raise
  228. try:
  229. gr = grp.getgrnam(SETGID_GROUP)
  230. except KeyError:
  231. print("[ERROR] Couldn't get group information for group " + SETGID_GROUP, file=sys.stderr)
  232. raise
  233. # gid has to be set first
  234. os.setgid(gr.gr_gid)
  235. os.setuid(pw.pw_uid)
  236. def _init_log(log_dir):
  237. """Initialize logging configuration"""
  238. desktop.log.basic_logging(PROC_NAME, log_dir)
  239. if os.geteuid() == 0:
  240. desktop.log.chown_log_dir(g_user_uid, g_user_gid)
  241. def main():
  242. global SETUID_USER, SETGID_GROUP
  243. options = parse_args()
  244. SETUID_USER = options.user
  245. SETGID_GROUP = options.group
  246. root = desktop.lib.paths.get_run_root()
  247. log_dir = os.path.join(root, options.log_dir)
  248. if options.show_supervisees:
  249. for name, supervisee in get_supervisees().items():
  250. if name not in options.supervisee_exclusions:
  251. print(name)
  252. sys.exit(0)
  253. # Let our children know
  254. os.environ['DESKTOP_LOG_DIR'] = log_dir
  255. if not os.path.exists(log_dir):
  256. os.makedirs(log_dir)
  257. setup_user_info()
  258. pid_file = os.path.abspath(os.path.join(root, options.pid_file))
  259. pidfile_context = TimeoutPIDLockFile(pid_file, LOCKFILE_TIMEOUT)
  260. existing_pid = pidfile_context.read_pid()
  261. if existing_pid:
  262. cmdline = get_pid_cmdline(existing_pid)
  263. if not cmdline.strip():
  264. # pid is not actually running
  265. pidfile_context.break_lock()
  266. else:
  267. LOG.error("Pid file %s indicates that Hue is already running (pid %d)" % (pid_file, existing_pid))
  268. sys.exit(1)
  269. elif pidfile_context.is_locked():
  270. # If there's no pidfile but there is a lock, it's a strange situation,
  271. # but we should break the lock because it doesn't seem to be actually running
  272. logging.warn("No existing pid file, but lock exists. Breaking lock.")
  273. pidfile_context.break_lock()
  274. if options.daemonize:
  275. if sys.version_info[0] > 2:
  276. outfile = open_file(os.path.join(log_dir, 'supervisor.out'), 'ba+', 0)
  277. else:
  278. outfile = open_file(os.path.join(log_dir, 'supervisor.out'), 'a+', 0)
  279. context = daemon.DaemonContext(
  280. working_directory=root,
  281. pidfile=pidfile_context,
  282. stdout=outfile,
  283. stderr=outfile,
  284. )
  285. context.signal_map = {
  286. signal.SIGTERM: sig_handler,
  287. }
  288. context.open()
  289. os.umask(0o22)
  290. # Log initialization must come after daemonization, which closes all open files.
  291. # Log statements before this point goes to stderr.
  292. _init_log(log_dir)
  293. sups = []
  294. try:
  295. for name, supervisee in get_supervisees().items():
  296. if name in options.supervisee_exclusions:
  297. continue
  298. if supervisee.drop_root:
  299. preexec_fn = drop_privileges
  300. else:
  301. preexec_fn = None
  302. if options.daemonize:
  303. if sys.version_info[0] > 2:
  304. log_stdout = open_file(os.path.join(log_dir, name + '.out'), 'ba+', 0)
  305. else:
  306. log_stdout = open_file(os.path.join(log_dir, name + '.out'), 'a+', 0)
  307. log_stderr = log_stdout
  308. else:
  309. # Passing None to subprocess.Popen later makes the subprocess inherit the
  310. # standard fds from the supervisor
  311. log_stdout = None
  312. log_stderr = None
  313. sup = Supervisor(supervisee.cmdv, stdout=log_stdout, stderr=log_stderr, preexec_fn=preexec_fn)
  314. sup.start()
  315. sups.append(sup)
  316. wait_loop(sups, options)
  317. except BaseException as ex:
  318. LOG.exception("Exception in supervisor main loop")
  319. shutdown(sups) # shutdown() exits the process
  320. return 0
  321. def wait_loop(sups, options):
  322. while True:
  323. time.sleep(1)
  324. for sup in sups:
  325. sup.join(0.1)
  326. if not sup.is_alive():
  327. if sup.state == Supervisor.FINISHED:
  328. sups.remove(sup)
  329. else:
  330. shutdown(sups) # shutdown() exits the process
  331. if __name__ == "__main__":
  332. sys.exit(main())