supervisor.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407
  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 daemon.pidlockfile import PIDLockFile
  29. import daemon
  30. import exceptions
  31. import grp
  32. import logging
  33. import optparse
  34. import os
  35. import pkg_resources
  36. import pwd
  37. import signal
  38. import subprocess
  39. import sys
  40. import threading
  41. import time
  42. import desktop.lib.daemon_utils
  43. import desktop.lib.paths
  44. import desktop.log
  45. # BaseException not available on python 2.4
  46. try:
  47. MyBaseException = exceptions.BaseException
  48. except AttributeError:
  49. MyBaseException = exceptions.Exception
  50. PROC_NAME = 'supervisor'
  51. LOG = logging.getLogger()
  52. # If a process restarts mre than MAX_RESTARTS_IN_WINDOW times
  53. # within TIME_WINDOW number of seconds, the supervisor shuts down
  54. TIME_WINDOW = 120
  55. MAX_RESTARTS_IN_WINDOW = 3
  56. # User and group to setuid/setgid down to for any supervisees that don't have
  57. # the drop_root option set to False
  58. SETUID_USER = "hue"
  59. SETGID_GROUP = "hue"
  60. g_user_uid = None # We figure out the numeric uid/gid later
  61. g_user_gid = None
  62. # The entry point group in which to find processes to supervise.
  63. ENTRY_POINT_GROUP = "desktop.supervisor.specs"
  64. # How long to wait while trying to acquire the supervisor pid lock
  65. # file. We shouldn't spin long here - we'd rather fail to start up.
  66. LOCKFILE_TIMEOUT = 2
  67. # The hue program
  68. HUE_BIN = os.path.join(desktop.lib.paths.get_run_root(),
  69. 'build', 'env', 'bin', 'hue')
  70. ######
  71. CHILD_PIDS = []
  72. SHOULD_STOP = False
  73. # How long to wait for children processes to die gracefully before
  74. # killing them forceably (seconds)
  75. WAIT_FOR_DEATH = 5
  76. class SuperviseeSpec(object):
  77. """
  78. A specification of something that should be supervised.
  79. Instances should have a .cmdv property which returns a list
  80. which will be passed through to subprocess.call.
  81. """
  82. def __init__(self, drop_root=True):
  83. """
  84. @param drop_root: if True, the supervisor will drop root privileges
  85. before calling the subprocess.
  86. """
  87. self.drop_root = drop_root
  88. class DjangoCommandSupervisee(SuperviseeSpec):
  89. """A supervisee which is simply a desktop django management command."""
  90. def __init__(self, django_command, **kwargs):
  91. SuperviseeSpec.__init__(self, **kwargs)
  92. self.django_command = django_command
  93. @property
  94. def cmdv(self):
  95. return [ HUE_BIN, self.django_command ]
  96. class TimeOutPIDLockFile(PIDLockFile):
  97. """A PIDLockFile subclass that passes through a timeout on acquisition."""
  98. def __init__(self, lockfile, timeout, **kwargs):
  99. PIDLockFile.__init__(self, lockfile, **kwargs)
  100. self.timeout = timeout
  101. def __enter__(self):
  102. super(TimeOutPIDLockFile, self).acquire(timeout=self.timeout)
  103. return self
  104. class Supervisor(threading.Thread):
  105. """A thread responsible for keeping the supervised subprocess running"""
  106. # States of the subprocess
  107. STATES = (PENDING, RUNNING, FINISHED, ERROR) = range(4)
  108. def __init__(self, cmdv, **kwargs):
  109. super(Supervisor, self).__init__()
  110. self.cmdv = cmdv
  111. self.popen_kwargs = kwargs
  112. self.state = Supervisor.PENDING
  113. def run(self):
  114. global CHILD_PIDS
  115. try:
  116. restart_timestamps = []
  117. proc_str = " ".join(self.cmdv)
  118. while True:
  119. self.state = Supervisor.RUNNING
  120. LOG.info("Starting process %s" % proc_str)
  121. pipe = subprocess.Popen(self.cmdv, close_fds=True,
  122. stdin=file("/dev/null"),
  123. **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.warn("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. self.state = Supervisor.ERROR
  148. return
  149. LOG.error("Process %s exited abnormally. Restarting it." % (proc_str,))
  150. except MyBaseException, ex:
  151. LOG.exception("Uncaught exception. Supervisor exiting.")
  152. self.state = Supervisor.ERROR
  153. def shutdown(sups):
  154. global SHOULD_STOP
  155. SHOULD_STOP = True
  156. LOG.warn("Supervisor shutting down!")
  157. for pid in CHILD_PIDS:
  158. try:
  159. os.kill(pid, signal.SIGINT)
  160. except OSError:
  161. pass
  162. LOG.warn("Waiting for children to exit for %d seconds..." % WAIT_FOR_DEATH)
  163. t = time.time()
  164. still_alive = False
  165. while time.time() < t + WAIT_FOR_DEATH:
  166. still_alive = False
  167. for sup in sups:
  168. sup.join(0.2)
  169. still_alive = still_alive or sup.isAlive()
  170. if not still_alive:
  171. break
  172. if still_alive:
  173. LOG.warn("Children have not exited after %d seconds. Killing them with SIGKILL." %
  174. WAIT_FOR_DEATH)
  175. for pid in CHILD_PIDS:
  176. try:
  177. os.kill(pid, signal.SIGKILL)
  178. except OSError:
  179. pass
  180. sys.exit(1)
  181. def sig_handler(signum, frame):
  182. raise SystemExit("Signal %d received. Exiting" % signum)
  183. def parse_args():
  184. parser = optparse.OptionParser()
  185. parser.add_option("-d", "--daemon", dest="daemonize",
  186. action="store_true", default=False)
  187. parser.add_option("-p", "--pid-file", dest="pid_file",
  188. metavar="PID_FILE", default="supervisor.pid")
  189. parser.add_option("-l", "--log-dir", dest="log_dir",
  190. metavar="DIR", default="logs")
  191. parser.add_option('-e', '--exclude', dest='supervisee_exclusions',
  192. metavar='EXCLUSIONS', default=[], action='append',
  193. help='Command NOT to run from supervisor. May be included more than once.')
  194. parser.add_option('-s', '--show', dest='show_supervisees',
  195. action='store_true', default=False)
  196. parser.add_option('-u', '--user', dest='user',
  197. action='store', default=SETUID_USER)
  198. parser.add_option('-g', '--group', dest='group',
  199. action='store', default=SETGID_GROUP)
  200. (options, args) = parser.parse_args()
  201. return options
  202. def get_pid_cmdline(pid):
  203. return subprocess.Popen(["ps", "-p", str(pid), "-o", "cmd", "h"],
  204. stdout=subprocess.PIPE, close_fds=True).communicate()[0]
  205. def get_supervisees():
  206. """Pull the supervisor specifications out of the entry point."""
  207. eps = list(pkg_resources.iter_entry_points(ENTRY_POINT_GROUP))
  208. return dict((ep.name, ep.load()) for ep in eps)
  209. def setup_user_info():
  210. """Translate the user/group info into uid/gid."""
  211. if os.geteuid() != 0:
  212. return
  213. global g_user_uid, g_user_gid
  214. g_user_uid, g_user_gid = \
  215. desktop.lib.daemon_utils.get_uid_gid(SETUID_USER, SETGID_GROUP)
  216. def drop_privileges():
  217. """Drop root privileges down to the specified SETUID_USER.
  218. N.B. DO NOT USE THE logging MODULE FROM WITHIN THIS FUNCTION.
  219. This function is run in forked processes right before it calls
  220. exec, but the fork may have occured while a different thread
  221. had locked the log. Since it's a forked process, the log will
  222. be locked forever in the subprocess and thus a logging.X may
  223. block forever.
  224. """
  225. we_are_root = os.getuid() == 0
  226. if not we_are_root:
  227. print >>sys.stdout, "[INFO] Not running as root, skipping privilege drop"
  228. return
  229. try:
  230. pw = pwd.getpwnam(SETUID_USER)
  231. except KeyError:
  232. print >>sys.stderr, "[ERROR] Couldn't get user information for user " + SETUID_USER
  233. raise
  234. try:
  235. gr = grp.getgrnam(SETGID_GROUP)
  236. except KeyError:
  237. print >>sys.stderr, "[ERROR] Couldn't get group information for group " + SETGID_GROUP
  238. raise
  239. # gid has to be set first
  240. os.setgid(gr.gr_gid)
  241. os.setuid(pw.pw_uid)
  242. def _init_log(log_dir):
  243. """Initialize logging configuration"""
  244. desktop.log.basic_logging(PROC_NAME, log_dir)
  245. if os.geteuid() == 0:
  246. desktop.log.chown_log_dir(g_user_uid, g_user_gid)
  247. def main():
  248. global SETUID_USER, SETGID_GROUP
  249. options = parse_args()
  250. SETUID_USER = options.user
  251. SETGID_GROUP = options.group
  252. root = desktop.lib.paths.get_run_root()
  253. log_dir = os.path.join(root, options.log_dir)
  254. if options.show_supervisees:
  255. for name, supervisee in get_supervisees().iteritems():
  256. if name not in options.supervisee_exclusions:
  257. print(name)
  258. sys.exit(0)
  259. # Let our children know
  260. os.environ['DESKTOP_LOG_DIR'] = log_dir
  261. if not os.path.exists(log_dir):
  262. os.makedirs(log_dir)
  263. setup_user_info()
  264. pid_file = os.path.abspath(os.path.join(root, options.pid_file))
  265. pidfile_context = TimeOutPIDLockFile(pid_file, LOCKFILE_TIMEOUT)
  266. existing_pid = pidfile_context.read_pid()
  267. if existing_pid:
  268. cmdline = get_pid_cmdline(existing_pid)
  269. if not cmdline.strip():
  270. # pid is not actually running
  271. pidfile_context.break_lock()
  272. else:
  273. LOG.error("Pid file %s indicates that Hue is already running (pid %d)" %
  274. (pid_file, existing_pid))
  275. sys.exit(1)
  276. elif pidfile_context.is_locked():
  277. # If there's no pidfile but there is a lock, it's a strange situation,
  278. # but we should break the lock because it doesn't seem to be actually running
  279. logging.warn("No existing pid file, but lock exists. Breaking lock.")
  280. pidfile_context.break_lock()
  281. if options.daemonize:
  282. outfile = file(os.path.join(log_dir, 'supervisor.out'), 'a+', 0)
  283. context = daemon.DaemonContext(
  284. working_directory=root,
  285. pidfile=pidfile_context,
  286. stdout=outfile,
  287. stderr=outfile,
  288. )
  289. context.signal_map = {
  290. signal.SIGTERM: sig_handler,
  291. }
  292. context.open()
  293. os.umask(022)
  294. # Log initialization must come after daemonization, which closes all open files.
  295. # Log statements before this point goes to stderr.
  296. _init_log(log_dir)
  297. sups = []
  298. try:
  299. for name, supervisee in get_supervisees().iteritems():
  300. if name in options.supervisee_exclusions:
  301. continue
  302. if supervisee.drop_root:
  303. preexec_fn = drop_privileges
  304. else:
  305. preexec_fn = None
  306. if options.daemonize:
  307. log_stdout = file(os.path.join(log_dir, name + '.out'), 'a+', 0)
  308. log_stderr = log_stdout
  309. else:
  310. # Passing None to subprocess.Popen later makes the subprocess inherit the
  311. # standard fds from the supervisor
  312. log_stdout = None
  313. log_stderr = None
  314. sup = Supervisor(supervisee.cmdv,
  315. stdout=log_stdout, stderr=log_stderr,
  316. preexec_fn=preexec_fn)
  317. sup.start()
  318. sups.append(sup)
  319. wait_loop(sups, options)
  320. except MyBaseException, ex:
  321. LOG.exception("Exception in supervisor main loop")
  322. shutdown(sups) # shutdown() exits the process
  323. return 0
  324. def wait_loop(sups, options):
  325. while True:
  326. time.sleep(1)
  327. for sup in sups:
  328. sup.join(0.1)
  329. if not sup.isAlive():
  330. if sup.state == Supervisor.FINISHED:
  331. sups.remove(sup)
  332. else:
  333. shutdown(sups) # shutdown() exits the process
  334. if __name__ == "__main__":
  335. sys.exit(main())