celeryd_detach.py 4.9 KB

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