state.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451
  1. # -*- coding: utf-8 -*-
  2. """
  3. celery.events.state
  4. ~~~~~~~~~~~~~~~~~~~
  5. This module implements a datastructure used to keep
  6. track of the state of a cluster of workers and the tasks
  7. it is working on (by consuming events).
  8. For every event consumed the state is updated,
  9. so the state represents the state of the cluster
  10. at the time of the last event.
  11. Snapshots (:mod:`celery.events.snapshot`) can be used to
  12. take "pictures" of this state at regular intervals
  13. to e.g. store that in a database.
  14. """
  15. from __future__ import absolute_import
  16. import threading
  17. from datetime import datetime
  18. from heapq import heappush, heappop
  19. from itertools import islice
  20. from kombu.clocks import timetuple
  21. from kombu.utils import kwdict
  22. from celery import states
  23. from celery.datastructures import AttributeDict
  24. from celery.five import items, values, monotonic
  25. from celery.utils.functional import LRUCache
  26. from celery.utils.log import get_logger
  27. # The window (in percentage) is added to the workers heartbeat
  28. # frequency. If the time between updates exceeds this window,
  29. # then the worker is considered to be offline.
  30. HEARTBEAT_EXPIRE_WINDOW = 200
  31. # Max drift between event timestamp and time of event received
  32. # before we alert that clocks may be unsynchronized.
  33. HEARTBEAT_DRIFT_MAX = 16
  34. DRIFT_WARNING = """\
  35. Substantial drift from %s may mean clocks are out of sync. Current drift is
  36. %s seconds. [orig: %s recv: %s]
  37. """
  38. logger = get_logger(__name__)
  39. warn = logger.warning
  40. R_STATE = '<State: events={0.event_count} tasks={0.task_count}>'
  41. R_WORKER = '<Worker: {0.hostname} ({0.status_string})'
  42. R_TASK = '<Task: {0.name}({0.uuid}) {0.state}>'
  43. __all__ = ['Worker', 'Task', 'State', 'heartbeat_expires']
  44. def heartbeat_expires(timestamp, freq=60,
  45. expire_window=HEARTBEAT_EXPIRE_WINDOW):
  46. return timestamp + freq * (expire_window / 1e2)
  47. def with_unique_field(attr):
  48. def _decorate_cls(cls):
  49. def __eq__(this, other):
  50. if isinstance(other, this.__class__):
  51. return getattr(this, attr) == getattr(other, attr)
  52. return NotImplemented
  53. cls.__eq__ = __eq__
  54. def __ne__(this, other):
  55. return not this.__eq__(other)
  56. cls.__ne__ = __ne__
  57. def __hash__(this):
  58. return hash(getattr(this, attr))
  59. cls.__hash__ = __hash__
  60. return cls
  61. return _decorate_cls
  62. @with_unique_field('hostname')
  63. class Worker(AttributeDict):
  64. """Worker State."""
  65. heartbeat_max = 4
  66. expire_window = HEARTBEAT_EXPIRE_WINDOW
  67. pid = None
  68. _defaults = {'hostname': None, 'pid': None, 'freq': 60}
  69. def __init__(self, **fields):
  70. dict.__init__(self, self._defaults, **fields)
  71. self.heartbeats = []
  72. def on_online(self, timestamp=None, local_received=None, **kwargs):
  73. """Callback for the :event:`worker-online` event."""
  74. self.update(**kwargs)
  75. self.update_heartbeat(local_received, timestamp)
  76. def on_offline(self, **kwargs):
  77. """Callback for the :event:`worker-offline` event."""
  78. self.update(**kwargs)
  79. self.heartbeats = []
  80. def on_heartbeat(self, timestamp=None, local_received=None, **kwargs):
  81. """Callback for the :event:`worker-heartbeat` event."""
  82. self.update(**kwargs)
  83. self.update_heartbeat(local_received, timestamp)
  84. def update_heartbeat(self, received, timestamp):
  85. if not received or not timestamp:
  86. return
  87. drift = abs(int(received) - int(timestamp))
  88. if drift > HEARTBEAT_DRIFT_MAX:
  89. warn(DRIFT_WARNING, self.hostname, drift,
  90. datetime.fromtimestamp(received),
  91. datetime.fromtimestamp(timestamp))
  92. heartbeats, hbmax = self.heartbeats, self.heartbeat_max
  93. if not heartbeats or (received and received > heartbeats[-1]):
  94. heappush(heartbeats, received)
  95. if len(heartbeats) > hbmax:
  96. heartbeats[:] = heartbeats[hbmax:]
  97. def __repr__(self):
  98. return R_WORKER.format(self)
  99. @property
  100. def status_string(self):
  101. return 'ONLINE' if self.alive else 'OFFLINE'
  102. @property
  103. def heartbeat_expires(self):
  104. return heartbeat_expires(self.heartbeats[-1],
  105. self.freq, self.expire_window)
  106. @property
  107. def alive(self):
  108. return bool(self.heartbeats and monotonic() < self.heartbeat_expires)
  109. @property
  110. def id(self):
  111. return '{0.hostname}.{0.pid}'.format(self)
  112. @with_unique_field('uuid')
  113. class Task(AttributeDict):
  114. """Task State."""
  115. #: How to merge out of order events.
  116. #: Disorder is detected by logical ordering (e.g. :event:`task-received`
  117. #: must have happened before a :event:`task-failed` event).
  118. #:
  119. #: A merge rule consists of a state and a list of fields to keep from
  120. #: that state. ``(RECEIVED, ('name', 'args')``, means the name and args
  121. #: fields are always taken from the RECEIVED state, and any values for
  122. #: these fields received before or after is simply ignored.
  123. merge_rules = {states.RECEIVED: ('name', 'args', 'kwargs',
  124. 'retries', 'eta', 'expires')}
  125. #: meth:`info` displays these fields by default.
  126. _info_fields = ('args', 'kwargs', 'retries', 'result',
  127. 'eta', 'runtime', 'expires', 'exception',
  128. 'exchange', 'routing_key')
  129. #: Default values.
  130. _defaults = dict(uuid=None, name=None, state=states.PENDING,
  131. received=False, sent=False, started=False,
  132. succeeded=False, failed=False, retried=False,
  133. revoked=False, args=None, kwargs=None, eta=None,
  134. expires=None, retries=None, worker=None, result=None,
  135. exception=None, timestamp=None, runtime=None,
  136. traceback=None, exchange=None, routing_key=None,
  137. clock=0)
  138. def __init__(self, **fields):
  139. dict.__init__(self, self._defaults, **fields)
  140. def update(self, state, timestamp, fields, _state=states.state):
  141. """Update state from new event.
  142. :param state: State from event.
  143. :param timestamp: Timestamp from event.
  144. :param fields: Event data.
  145. """
  146. time_received = fields.get('local_received') or 0
  147. if self.worker and time_received:
  148. self.worker.update_heartbeat(time_received, timestamp)
  149. if state != states.RETRY and self.state != states.RETRY and \
  150. _state(state) < _state(self.state):
  151. # this state logically happens-before the current state, so merge.
  152. self.merge(state, timestamp, fields)
  153. else:
  154. self.state = state
  155. self.timestamp = timestamp
  156. super(Task, self).update(fields)
  157. def merge(self, state, timestamp, fields):
  158. """Merge with out of order event."""
  159. keep = self.merge_rules.get(state)
  160. if keep is not None:
  161. fields = dict((key, fields.get(key)) for key in keep)
  162. super(Task, self).update(fields)
  163. def on_sent(self, timestamp=None, **fields):
  164. """Callback for the :event:`task-sent` event."""
  165. self.sent = timestamp
  166. self.update(states.PENDING, timestamp, fields)
  167. def on_received(self, timestamp=None, **fields):
  168. """Callback for the :event:`task-received` event."""
  169. self.received = timestamp
  170. self.update(states.RECEIVED, timestamp, fields)
  171. def on_started(self, timestamp=None, **fields):
  172. """Callback for the :event:`task-started` event."""
  173. self.started = timestamp
  174. self.update(states.STARTED, timestamp, fields)
  175. def on_failed(self, timestamp=None, **fields):
  176. """Callback for the :event:`task-failed` event."""
  177. self.failed = timestamp
  178. self.update(states.FAILURE, timestamp, fields)
  179. def on_retried(self, timestamp=None, **fields):
  180. """Callback for the :event:`task-retried` event."""
  181. self.retried = timestamp
  182. self.update(states.RETRY, timestamp, fields)
  183. def on_succeeded(self, timestamp=None, **fields):
  184. """Callback for the :event:`task-succeeded` event."""
  185. self.succeeded = timestamp
  186. self.update(states.SUCCESS, timestamp, fields)
  187. def on_revoked(self, timestamp=None, **fields):
  188. """Callback for the :event:`task-revoked` event."""
  189. self.revoked = timestamp
  190. self.update(states.REVOKED, timestamp, fields)
  191. def on_unknown_event(self, shortype, timestamp=None, **fields):
  192. self.update(shortype.upper(), timestamp, fields)
  193. def info(self, fields=None, extra=[]):
  194. """Information about this task suitable for on-screen display."""
  195. fields = self._info_fields if fields is None else fields
  196. def _keys():
  197. for key in list(fields) + list(extra):
  198. value = getattr(self, key, None)
  199. if value is not None:
  200. yield key, value
  201. return dict(_keys())
  202. def __repr__(self):
  203. return R_TASK.format(self)
  204. @property
  205. def ready(self):
  206. return self.state in states.READY_STATES
  207. class State(object):
  208. """Records clusters state."""
  209. event_count = 0
  210. task_count = 0
  211. def __init__(self, callback=None,
  212. workers=None, tasks=None, taskheap=None,
  213. max_workers_in_memory=5000, max_tasks_in_memory=10000):
  214. self.event_callback = callback
  215. self.workers = (LRUCache(max_workers_in_memory)
  216. if workers is None else workers)
  217. self.tasks = (LRUCache(max_tasks_in_memory)
  218. if tasks is None else tasks)
  219. self._taskheap = [] if taskheap is None else taskheap
  220. self.max_workers_in_memory = max_workers_in_memory
  221. self.max_tasks_in_memory = max_tasks_in_memory
  222. self._mutex = threading.Lock()
  223. self.handlers = {'task': self.task_event,
  224. 'worker': self.worker_event}
  225. self._get_handler = self.handlers.__getitem__
  226. def freeze_while(self, fun, *args, **kwargs):
  227. clear_after = kwargs.pop('clear_after', False)
  228. with self._mutex:
  229. try:
  230. return fun(*args, **kwargs)
  231. finally:
  232. if clear_after:
  233. self._clear()
  234. def clear_tasks(self, ready=True):
  235. with self._mutex:
  236. return self._clear_tasks(ready)
  237. def _clear_tasks(self, ready=True):
  238. if ready:
  239. in_progress = dict(
  240. (uuid, task) for uuid, task in self.itertasks()
  241. if task.state not in states.READY_STATES)
  242. self.tasks.clear()
  243. self.tasks.update(in_progress)
  244. else:
  245. self.tasks.clear()
  246. self._taskheap[:] = []
  247. def _clear(self, ready=True):
  248. self.workers.clear()
  249. self._clear_tasks(ready)
  250. self.event_count = 0
  251. self.task_count = 0
  252. def clear(self, ready=True):
  253. with self._mutex:
  254. return self._clear(ready)
  255. def get_or_create_worker(self, hostname, **kwargs):
  256. """Get or create worker by hostname.
  257. Return tuple of ``(worker, was_created)``.
  258. """
  259. try:
  260. worker = self.workers[hostname]
  261. worker.update(kwargs)
  262. return worker, False
  263. except KeyError:
  264. worker = self.workers[hostname] = Worker(
  265. hostname=hostname, **kwargs)
  266. return worker, True
  267. def get_or_create_task(self, uuid):
  268. """Get or create task by uuid."""
  269. try:
  270. return self.tasks[uuid], False
  271. except KeyError:
  272. task = self.tasks[uuid] = Task(uuid=uuid)
  273. return task, True
  274. def worker_event(self, type, fields):
  275. """Process worker event."""
  276. try:
  277. hostname = fields['hostname']
  278. except KeyError:
  279. pass
  280. else:
  281. worker, created = self.get_or_create_worker(hostname)
  282. handler = getattr(worker, 'on_' + type, None)
  283. if handler:
  284. handler(**fields)
  285. return worker, created
  286. def task_event(self, type, fields, timetuple=timetuple):
  287. """Process task event."""
  288. uuid = fields['uuid']
  289. hostname = fields['hostname']
  290. worker, _ = self.get_or_create_worker(hostname)
  291. task, created = self.get_or_create_task(uuid)
  292. task.worker = worker
  293. maxtasks = self.max_tasks_in_memory * 2
  294. taskheap = self._taskheap
  295. timestamp = fields.get('timestamp') or 0
  296. clock = 0 if type == 'sent' else fields.get('clock')
  297. heappush(taskheap, timetuple(clock, timestamp, worker.id, task))
  298. if len(taskheap) > maxtasks:
  299. heappop(taskheap)
  300. handler = getattr(task, 'on_' + type, None)
  301. if type == 'received':
  302. self.task_count += 1
  303. if handler:
  304. handler(**fields)
  305. else:
  306. task.on_unknown_event(type, **fields)
  307. return created
  308. def event(self, event):
  309. with self._mutex:
  310. return self._dispatch_event(event)
  311. def _dispatch_event(self, event, kwdict=kwdict):
  312. self.event_count += 1
  313. event = kwdict(event)
  314. group, _, subject = event['type'].partition('-')
  315. try:
  316. self._get_handler(group)(subject, event)
  317. except KeyError:
  318. pass
  319. if self.event_callback:
  320. self.event_callback(self, event)
  321. def itertasks(self, limit=None):
  322. for index, row in enumerate(items(self.tasks)):
  323. yield row
  324. if limit and index + 1 >= limit:
  325. break
  326. def tasks_by_time(self, limit=None):
  327. """Generator giving tasks ordered by time,
  328. in ``(uuid, Task)`` tuples."""
  329. seen = set()
  330. for evtup in islice(reversed(self._taskheap), 0, limit):
  331. uuid = evtup[3].uuid
  332. if uuid not in seen:
  333. yield uuid, evtup[3]
  334. seen.add(uuid)
  335. tasks_by_timestamp = tasks_by_time
  336. def tasks_by_type(self, name, limit=None):
  337. """Get all tasks by type.
  338. Return a list of ``(uuid, Task)`` tuples.
  339. """
  340. return islice(
  341. ((uuid, task) for uuid, task in self.tasks_by_time()
  342. if task.name == name),
  343. 0, limit,
  344. )
  345. def tasks_by_worker(self, hostname, limit=None):
  346. """Get all tasks by worker.
  347. """
  348. return islice(
  349. ((uuid, task) for uuid, task in self.tasks_by_time()
  350. if task.worker.hostname == hostname),
  351. 0, limit,
  352. )
  353. def task_types(self):
  354. """Return a list of all seen task types."""
  355. return list(sorted(set(task.name for task in values(self.tasks))))
  356. def alive_workers(self):
  357. """Return a list of (seemingly) alive workers."""
  358. return [w for w in values(self.workers) if w.alive]
  359. def __repr__(self):
  360. return R_STATE.format(self)
  361. def __reduce__(self):
  362. return self.__class__, (
  363. self.event_callback, self.workers, self.tasks, self._taskheap,
  364. self.max_workers_in_memory, self.max_tasks_in_memory,
  365. )