/usr/lib/python2.7/dist-packages/djcelery/snapshot.py is in python-django-celery 3.1.17-3.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | from __future__ import absolute_import, unicode_literals
from collections import defaultdict
from datetime import datetime, timedelta
from django.conf import settings
from celery import states
from celery.events.state import Task
from celery.events.snapshot import Polaroid
from celery.five import monotonic
from celery.utils.log import get_logger
from celery.utils.timeutils import maybe_iso8601
from .models import WorkerState, TaskState
from .utils import fromtimestamp, maybe_make_aware
WORKER_UPDATE_FREQ = 60 # limit worker timestamp write freq.
SUCCESS_STATES = frozenset([states.SUCCESS])
# Expiry can be timedelta or None for never expire.
EXPIRE_SUCCESS = getattr(settings, 'CELERYCAM_EXPIRE_SUCCESS',
timedelta(days=1))
EXPIRE_ERROR = getattr(settings, 'CELERYCAM_EXPIRE_ERROR',
timedelta(days=3))
EXPIRE_PENDING = getattr(settings, 'CELERYCAM_EXPIRE_PENDING',
timedelta(days=5))
NOT_SAVED_ATTRIBUTES = frozenset(['name', 'args', 'kwargs', 'eta'])
logger = get_logger(__name__)
debug = logger.debug
def aware_tstamp(secs):
"""Event timestamps uses the local timezone."""
return maybe_make_aware(fromtimestamp(secs))
class Camera(Polaroid):
TaskState = TaskState
WorkerState = WorkerState
clear_after = True
worker_update_freq = WORKER_UPDATE_FREQ
expire_states = {
SUCCESS_STATES: EXPIRE_SUCCESS,
states.EXCEPTION_STATES: EXPIRE_ERROR,
states.UNREADY_STATES: EXPIRE_PENDING,
}
def __init__(self, *args, **kwargs):
super(Camera, self).__init__(*args, **kwargs)
self._last_worker_write = defaultdict(lambda: (None, None))
def get_heartbeat(self, worker):
try:
heartbeat = worker.heartbeats[-1]
except IndexError:
return
# Check for timezone settings
if getattr(settings, "USE_TZ", False):
return aware_tstamp(heartbeat)
return datetime.fromtimestamp(heartbeat)
def handle_worker(self, hostname_worker):
(hostname, worker) = hostname_worker
last_write, obj = self._last_worker_write[hostname]
if not last_write or \
monotonic() - last_write > self.worker_update_freq:
obj = self.WorkerState.objects.update_or_create(
hostname=hostname,
defaults={'last_heartbeat': self.get_heartbeat(worker)},
)
self._last_worker_write[hostname] = (monotonic(), obj)
return obj
def handle_task(self, uuid_task, worker=None):
"""Handle snapshotted event."""
uuid, task = uuid_task
if task.worker and task.worker.hostname:
worker = self.handle_worker(
(task.worker.hostname, task.worker),
)
defaults = {
'name': task.name,
'args': task.args,
'kwargs': task.kwargs,
'eta': maybe_make_aware(maybe_iso8601(task.eta)),
'expires': maybe_make_aware(maybe_iso8601(task.expires)),
'state': task.state,
'tstamp': aware_tstamp(task.timestamp),
'result': task.result or task.exception,
'traceback': task.traceback,
'runtime': task.runtime,
'worker': worker
}
# Some fields are only stored in the RECEIVED event,
# so we should remove these from default values,
# so that they are not overwritten by subsequent states.
[defaults.pop(attr, None) for attr in NOT_SAVED_ATTRIBUTES
if defaults[attr] is None]
return self.update_task(task.state,
task_id=uuid, defaults=defaults)
def update_task(self, state, **kwargs):
objects = self.TaskState.objects
defaults = kwargs.pop('defaults', None) or {}
if not defaults.get('name'):
return
obj, created = objects.get_or_create(defaults=defaults, **kwargs)
if created:
return obj
else:
if states.state(state) < states.state(obj.state):
keep = Task.merge_rules[states.RECEIVED]
defaults = dict(
(k, v) for k, v in defaults.items()
if k not in keep
)
for k, v in defaults.items():
setattr(obj, k, v)
for datefield in ('eta', 'expires', 'tstamp'):
# Brute force trying to fix #183
setattr(obj, datefield, maybe_make_aware(getattr(obj, datefield)))
obj.save()
return obj
def on_shutter(self, state, commit_every=100):
def _handle_tasks():
for i, task in enumerate(state.tasks.items()):
self.handle_task(task)
for worker in state.workers.items():
self.handle_worker(worker)
_handle_tasks()
def on_cleanup(self):
expired = (self.TaskState.objects.expire_by_states(states, expires)
for states, expires in self.expire_states.items())
dirty = sum(item for item in expired if item is not None)
if dirty:
debug('Cleanup: Marked %s objects as dirty.', dirty)
self.TaskState.objects.purge()
debug('Cleanup: %s objects purged.', dirty)
return dirty
return 0
|