celeryd_detach.py 5.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173
  1. # -*- coding: utf-8 -*-
  2. """
  3. celery.bin.celeryd_detach
  4. ~~~~~~~~~~~~~~~~~~~~~~~~~
  5. Program used to daemonize the worker
  6. Using :func:`os.execv` because forking and multiprocessing
  7. leads to weird issues (it was a long time ago now, but it
  8. could have something to do with the threading mutex bug)
  9. """
  10. from __future__ import absolute_import
  11. import celery
  12. import os
  13. import sys
  14. from optparse import OptionParser, BadOptionError
  15. from celery.platforms import EX_FAILURE, detached
  16. from celery.utils.log import get_logger
  17. from celery.bin.base import daemon_options, Option
  18. __all__ = ['detached_celeryd', 'detach']
  19. logger = get_logger(__name__)
  20. C_FAKEFORK = os.environ.get('C_FAKEFORK')
  21. OPTION_LIST = daemon_options(default_pidfile='celeryd.pid') + (
  22. Option('--workdir', default=None, dest='working_directory'),
  23. Option('--fake',
  24. default=False, action='store_true', dest='fake',
  25. help="Don't fork (for debugging purposes)"),
  26. )
  27. def detach(path, argv, logfile=None, pidfile=None, uid=None,
  28. gid=None, umask=None, working_directory=None, fake=False, app=None,
  29. executable=None):
  30. fake = 1 if C_FAKEFORK else fake
  31. with detached(logfile, pidfile, uid, gid, umask, working_directory, fake,
  32. after_forkers=False):
  33. try:
  34. if executable is not None:
  35. path = executable
  36. os.execv(path, [path] + argv)
  37. except Exception:
  38. if app is None:
  39. from celery import current_app
  40. app = current_app
  41. app.log.setup_logging_subsystem('ERROR', logfile)
  42. logger.critical("Can't exec %r", ' '.join([path] + argv),
  43. exc_info=True)
  44. return EX_FAILURE
  45. class PartialOptionParser(OptionParser):
  46. def __init__(self, *args, **kwargs):
  47. self.leftovers = []
  48. OptionParser.__init__(self, *args, **kwargs)
  49. def _process_long_opt(self, rargs, values):
  50. arg = rargs.pop(0)
  51. if '=' in arg:
  52. opt, next_arg = arg.split('=', 1)
  53. rargs.insert(0, next_arg)
  54. had_explicit_value = True
  55. else:
  56. opt = arg
  57. had_explicit_value = False
  58. try:
  59. opt = self._match_long_opt(opt)
  60. option = self._long_opt.get(opt)
  61. except BadOptionError:
  62. option = None
  63. if option:
  64. if option.takes_value():
  65. nargs = option.nargs
  66. if len(rargs) < nargs:
  67. if nargs == 1:
  68. self.error('{0} requires an argument'.format(opt))
  69. else:
  70. self.error('{0} requires {1} arguments'.format(
  71. opt, nargs))
  72. elif nargs == 1:
  73. value = rargs.pop(0)
  74. else:
  75. value = tuple(rargs[0:nargs])
  76. del rargs[0:nargs]
  77. elif had_explicit_value:
  78. self.error('{0} option does not take a value'.format(opt))
  79. else:
  80. value = None
  81. option.process(opt, value, values, self)
  82. else:
  83. self.leftovers.append(arg)
  84. def _process_short_opts(self, rargs, values):
  85. arg = rargs[0]
  86. try:
  87. OptionParser._process_short_opts(self, rargs, values)
  88. except BadOptionError:
  89. self.leftovers.append(arg)
  90. if rargs and not rargs[0][0] == '-':
  91. self.leftovers.append(rargs.pop(0))
  92. class detached_celeryd(object):
  93. option_list = OPTION_LIST
  94. usage = '%prog [options] [celeryd options]'
  95. version = celery.VERSION_BANNER
  96. description = ('Detaches Celery worker nodes. See `celery worker --help` '
  97. 'for the list of supported worker arguments.')
  98. command = sys.executable
  99. execv_path = sys.executable
  100. if sys.version_info < (2, 7): # does not support pkg/__main__.py
  101. execv_argv = ['-m', 'celery.__main__', 'worker']
  102. else:
  103. execv_argv = ['-m', 'celery', 'worker']
  104. def __init__(self, app=None):
  105. self.app = app
  106. def Parser(self, prog_name):
  107. return PartialOptionParser(prog=prog_name,
  108. option_list=self.option_list,
  109. usage=self.usage,
  110. description=self.description,
  111. version=self.version)
  112. def parse_options(self, prog_name, argv):
  113. parser = self.Parser(prog_name)
  114. options, values = parser.parse_args(argv)
  115. if options.logfile:
  116. parser.leftovers.append('--logfile={0}'.format(options.logfile))
  117. if options.pidfile:
  118. parser.leftovers.append('--pidfile={0}'.format(options.pidfile))
  119. return options, values, parser.leftovers
  120. def execute_from_commandline(self, argv=None):
  121. if argv is None:
  122. argv = sys.argv
  123. config = []
  124. seen_cargs = 0
  125. for arg in argv:
  126. if seen_cargs:
  127. config.append(arg)
  128. else:
  129. if arg == '--':
  130. seen_cargs = 1
  131. config.append(arg)
  132. prog_name = os.path.basename(argv[0])
  133. options, values, leftovers = self.parse_options(prog_name, argv[1:])
  134. sys.exit(detach(
  135. app=self.app, path=self.execv_path,
  136. argv=self.execv_argv + leftovers + config,
  137. **vars(options)
  138. ))
  139. def main(app=None):
  140. detached_celeryd(app).execute_from_commandline()
  141. if __name__ == '__main__': # pragma: no cover
  142. main()