celeryd_detach.py 5.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168
  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('--fake',
  23. default=False, action='store_true', dest='fake',
  24. help="Don't fork (for debugging purposes)"),
  25. )
  26. def detach(path, argv, logfile=None, pidfile=None, uid=None,
  27. gid=None, umask=0, working_directory=None, fake=False, app=None):
  28. fake = 1 if C_FAKEFORK else fake
  29. with detached(logfile, pidfile, uid, gid, umask, working_directory, fake):
  30. try:
  31. os.execv(path, [path] + argv)
  32. except Exception:
  33. if app is None:
  34. from celery import current_app
  35. app = current_app
  36. app.log.setup_logging_subsystem('ERROR', logfile)
  37. logger.critical("Can't exec %r", ' '.join([path] + argv),
  38. exc_info=True)
  39. return EX_FAILURE
  40. class PartialOptionParser(OptionParser):
  41. def __init__(self, *args, **kwargs):
  42. self.leftovers = []
  43. OptionParser.__init__(self, *args, **kwargs)
  44. def _process_long_opt(self, rargs, values):
  45. arg = rargs.pop(0)
  46. if '=' in arg:
  47. opt, next_arg = arg.split('=', 1)
  48. rargs.insert(0, next_arg)
  49. had_explicit_value = True
  50. else:
  51. opt = arg
  52. had_explicit_value = False
  53. try:
  54. opt = self._match_long_opt(opt)
  55. option = self._long_opt.get(opt)
  56. except BadOptionError:
  57. option = None
  58. if option:
  59. if option.takes_value():
  60. nargs = option.nargs
  61. if len(rargs) < nargs:
  62. if nargs == 1:
  63. self.error('{0} requires an argument'.format(opt))
  64. else:
  65. self.error('{0} requires {1} arguments'.format(
  66. opt, nargs))
  67. elif nargs == 1:
  68. value = rargs.pop(0)
  69. else:
  70. value = tuple(rargs[0:nargs])
  71. del rargs[0:nargs]
  72. elif had_explicit_value:
  73. self.error('{0} option does not take a value'.format(opt))
  74. else:
  75. value = None
  76. option.process(opt, value, values, self)
  77. else:
  78. self.leftovers.append(arg)
  79. def _process_short_opts(self, rargs, values):
  80. arg = rargs[0]
  81. try:
  82. OptionParser._process_short_opts(self, rargs, values)
  83. except BadOptionError:
  84. self.leftovers.append(arg)
  85. if rargs and not rargs[0][0] == '-':
  86. self.leftovers.append(rargs.pop(0))
  87. class detached_celeryd(object):
  88. option_list = OPTION_LIST
  89. usage = '%prog [options] [celeryd options]'
  90. version = celery.VERSION_BANNER
  91. description = ('Detaches Celery worker nodes. See `celery worker --help` '
  92. 'for the list of supported worker arguments.')
  93. command = sys.executable
  94. execv_path = sys.executable
  95. if sys.version_info < (2, 7): # does not support pkg/__main__.py
  96. execv_argv = ['-m', 'celery.__main__', 'worker']
  97. else:
  98. execv_argv = ['-m', 'celery', 'worker']
  99. def __init__(self, app=None):
  100. self.app = app
  101. def Parser(self, prog_name):
  102. return PartialOptionParser(prog=prog_name,
  103. option_list=self.option_list,
  104. usage=self.usage,
  105. description=self.description,
  106. version=self.version)
  107. def parse_options(self, prog_name, argv):
  108. parser = self.Parser(prog_name)
  109. options, values = parser.parse_args(argv)
  110. if options.logfile:
  111. parser.leftovers.append('--logfile={0}'.format(options.logfile))
  112. if options.pidfile:
  113. parser.leftovers.append('--pidfile={0}'.format(options.pidfile))
  114. return options, values, parser.leftovers
  115. def execute_from_commandline(self, argv=None):
  116. if argv is None:
  117. argv = sys.argv
  118. config = []
  119. seen_cargs = 0
  120. for arg in argv:
  121. if seen_cargs:
  122. config.append(arg)
  123. else:
  124. if arg == '--':
  125. seen_cargs = 1
  126. config.append(arg)
  127. prog_name = os.path.basename(argv[0])
  128. options, values, leftovers = self.parse_options(prog_name, argv[1:])
  129. sys.exit(detach(
  130. app=self.app, path=self.execv_path,
  131. argv=self.execv_argv + leftovers + config,
  132. **vars(options)
  133. ))
  134. def main(app=None):
  135. detached_celeryd(app).execute_from_commandline()
  136. if __name__ == '__main__': # pragma: no cover
  137. main()