control.py 8.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282
  1. # -*- coding: utf-8 -*-
  2. """
  3. celery.worker.control
  4. ~~~~~~~~~~~~~~~~~~~~~
  5. Remote control commands.
  6. """
  7. from __future__ import absolute_import
  8. import os
  9. from kombu.utils.encoding import safe_repr
  10. from itertools import imap
  11. from celery.platforms import signals as _signals
  12. from celery.utils import timeutils
  13. from celery.utils.compat import UserDict
  14. from celery.utils.log import get_logger
  15. from celery.utils import jsonify
  16. from . import state
  17. from .state import revoked
  18. DEFAULT_TASK_INFO_ITEMS = ('exchange', 'routing_key', 'rate_limit')
  19. logger = get_logger(__name__)
  20. class Panel(UserDict):
  21. data = dict() # Global registry.
  22. @classmethod
  23. def register(cls, method, name=None):
  24. cls.data[name or method.__name__] = method
  25. return method
  26. @Panel.register
  27. def revoke(panel, task_id, terminate=False, signal=None, **kwargs):
  28. """Revoke task by task id."""
  29. revoked.add(task_id)
  30. if terminate:
  31. signum = _signals.signum(signal or 'TERM')
  32. for request in state.reserved_requests:
  33. if request.id == task_id:
  34. logger.info('Terminating %s (%s)', task_id, signum)
  35. request.terminate(panel.consumer.pool, signal=signum)
  36. break
  37. else:
  38. return {'ok': 'terminate: task {0} not found'.format(task_id)}
  39. return {'ok': 'terminating {0} ({1})'.format(task_id, signal)}
  40. logger.info('Revoking task %s', task_id)
  41. return {'ok': 'revoking task {0}'.format(task_id)}
  42. @Panel.register
  43. def report(panel):
  44. return {'ok': panel.app.bugreport()}
  45. @Panel.register
  46. def enable_events(panel):
  47. dispatcher = panel.consumer.event_dispatcher
  48. if not dispatcher.enabled:
  49. dispatcher.enable()
  50. dispatcher.send('worker-online')
  51. logger.info('Events enabled by remote.')
  52. return {'ok': 'events enabled'}
  53. return {'ok': 'events already enabled'}
  54. @Panel.register
  55. def disable_events(panel):
  56. dispatcher = panel.consumer.event_dispatcher
  57. if dispatcher.enabled:
  58. dispatcher.send('worker-offline')
  59. dispatcher.disable()
  60. logger.info('Events disabled by remote.')
  61. return {'ok': 'events disabled'}
  62. return {'ok': 'events already disabled'}
  63. @Panel.register
  64. def heartbeat(panel):
  65. logger.debug('Heartbeat requested by remote.')
  66. dispatcher = panel.consumer.event_dispatcher
  67. dispatcher.send('worker-heartbeat', freq=5, **state.SOFTWARE_INFO)
  68. @Panel.register
  69. def rate_limit(panel, task_name, rate_limit, **kwargs):
  70. """Set new rate limit for a task type.
  71. See :attr:`celery.task.base.Task.rate_limit`.
  72. :param task_name: Type of task.
  73. :param rate_limit: New rate limit.
  74. """
  75. try:
  76. timeutils.rate(rate_limit)
  77. except ValueError as exc:
  78. return {'error': 'Invalid rate limit string: {0!r}'.format(exc)}
  79. try:
  80. panel.app.tasks[task_name].rate_limit = rate_limit
  81. except KeyError:
  82. logger.error('Rate limit attempt for unknown task %s',
  83. task_name, exc_info=True)
  84. return {'error': 'unknown task'}
  85. if not hasattr(panel.consumer.ready_queue, 'refresh'):
  86. logger.error('Rate limit attempt, but rate limits disabled.')
  87. return {'error': 'rate limits disabled'}
  88. panel.consumer.ready_queue.refresh()
  89. if not rate_limit:
  90. logger.info('Rate limits disabled for tasks of type %s', task_name)
  91. return {'ok': 'rate limit disabled successfully'}
  92. logger.info('New rate limit for tasks of type %s: %s.',
  93. task_name, rate_limit)
  94. return {'ok': 'new rate limit set successfully'}
  95. @Panel.register
  96. def time_limit(panel, task_name=None, hard=None, soft=None, **kwargs):
  97. try:
  98. task = panel.app.tasks[task_name]
  99. except KeyError:
  100. logger.error('Change time limit attempt for unknown task %s',
  101. task_name, exc_info=True)
  102. return {'error': 'unknown task'}
  103. task.soft_time_limit = soft
  104. task.time_limit = hard
  105. logger.info('New time limits for tasks of type %s: soft=%s hard=%s',
  106. task_name, soft, hard)
  107. return {'ok': 'time limits set successfully'}
  108. @Panel.register
  109. def dump_schedule(panel, safe=False, **kwargs):
  110. from celery.worker.job import Request
  111. schedule = panel.consumer.timer.schedule
  112. if not schedule.queue:
  113. return []
  114. def prepare_entries():
  115. for entry in schedule.info():
  116. item = entry['item']
  117. if item.args and isinstance(item.args[0], Request):
  118. yield {'eta': entry['eta'],
  119. 'priority': entry['priority'],
  120. 'request': item.args[0].info(safe=safe)}
  121. return list(prepare_entries())
  122. @Panel.register
  123. def dump_reserved(panel, safe=False, **kwargs):
  124. reserved = state.reserved_requests
  125. if not reserved:
  126. logger.debug('--Empty queue--')
  127. return []
  128. logger.debug('* Dump of currently reserved tasks:\n%s',
  129. '\n'.join(imap(safe_repr, reserved)))
  130. return [request.info(safe=safe)
  131. for request in reserved]
  132. @Panel.register
  133. def dump_active(panel, safe=False, **kwargs):
  134. return [request.info(safe=safe)
  135. for request in state.active_requests]
  136. @Panel.register
  137. def stats(panel, **kwargs):
  138. asinfo = {}
  139. if panel.consumer.controller.autoscaler:
  140. asinfo = panel.consumer.controller.autoscaler.info()
  141. return {'total': state.total_count,
  142. 'consumer': panel.consumer.info,
  143. 'pool': panel.consumer.pool.info,
  144. 'autoscaler': asinfo,
  145. 'pid': os.getpid()}
  146. @Panel.register
  147. def dump_revoked(panel, **kwargs):
  148. return list(state.revoked)
  149. @Panel.register
  150. def dump_tasks(panel, taskinfoitems=None, **kwargs):
  151. tasks = panel.app.tasks
  152. taskinfoitems = taskinfoitems or DEFAULT_TASK_INFO_ITEMS
  153. def _extract_info(task):
  154. fields = dict((field, str(getattr(task, field, None)))
  155. for field in taskinfoitems
  156. if getattr(task, field, None) is not None)
  157. if fields:
  158. info = imap('='.join, fields.iteritems())
  159. return '{0} [{1}]'.format(task.name, ' '.join(info))
  160. return task.name
  161. return [_extract_info(tasks[task]) for task in sorted(tasks)]
  162. @Panel.register
  163. def ping(panel, **kwargs):
  164. return 'pong'
  165. @Panel.register
  166. def pool_grow(panel, n=1, **kwargs):
  167. if panel.consumer.controller.autoscaler:
  168. panel.consumer.controller.autoscaler.force_scale_up(n)
  169. else:
  170. panel.consumer.pool.grow(n)
  171. return {'ok': 'spawned worker processes'}
  172. @Panel.register
  173. def pool_shrink(panel, n=1, **kwargs):
  174. if panel.consumer.controller.autoscaler:
  175. panel.consumer.controller.autoscaler.force_scale_down(n)
  176. else:
  177. panel.consumer.pool.shrink(n)
  178. return {'ok': 'terminated worker processes'}
  179. @Panel.register
  180. def pool_restart(panel, modules=None, reload=False, reloader=None, **kwargs):
  181. panel.consumer.controller.reload(modules, reload, reloader=reloader)
  182. return {'ok': 'reload started'}
  183. @Panel.register
  184. def autoscale(panel, max=None, min=None):
  185. autoscaler = panel.consumer.controller.autoscaler
  186. if autoscaler:
  187. max_, min_ = autoscaler.update(max, min)
  188. return {'ok': 'autoscale now min={0} max={1}'.format(max_, min_)}
  189. raise ValueError('Autoscale not enabled')
  190. @Panel.register
  191. def shutdown(panel, msg='Got shutdown from remote', **kwargs):
  192. logger.warning(msg)
  193. raise SystemExit(msg)
  194. @Panel.register
  195. def add_consumer(panel, queue, exchange=None, exchange_type=None,
  196. routing_key=None, **options):
  197. panel.consumer.add_task_queue(queue, exchange, exchange_type,
  198. routing_key, **options)
  199. return {'ok': 'add consumer {0}'.format(queue)}
  200. @Panel.register
  201. def cancel_consumer(panel, queue=None, **_):
  202. panel.consumer.cancel_task_queue(queue)
  203. return {'ok': 'no longer consuming from {0}'.format(queue)}
  204. @Panel.register
  205. def active_queues(panel):
  206. """Returns the queues associated with each worker."""
  207. return [dict(queue.as_dict(recurse=True))
  208. for queue in panel.consumer.task_consumer.queues]
  209. @Panel.register
  210. def dump_conf(panel, **kwargs):
  211. return jsonify(dict(panel.app.conf))