celery: fixed bug where celery was not running
This commit is contained in:
parent
fa69408b87
commit
38df61623b
3 changed files with 14 additions and 16 deletions
|
|
@ -248,9 +248,9 @@ class ReposController(BaseRepoController):
|
|||
task_id = request.GET.get('task_id')
|
||||
|
||||
if task_id and task_id not in ['None']:
|
||||
from rhodecode import CELERY_ENABLED
|
||||
import rhodecode
|
||||
from celery.result import AsyncResult
|
||||
if CELERY_ENABLED:
|
||||
if rhodecode.CELERY_ENABLED:
|
||||
task = AsyncResult(task_id)
|
||||
if task.failed():
|
||||
msg = self._log_creation_exception(task.result, c.repo)
|
||||
|
|
|
|||
|
|
@ -34,7 +34,6 @@ from decorator import decorator
|
|||
|
||||
from zope.cachedescriptors.property import Lazy as LazyProperty
|
||||
|
||||
from rhodecode import CELERY_ENABLED, CELERY_EAGER
|
||||
from rhodecode.config import utils
|
||||
from rhodecode.lib.utils2 import safe_str, md5_safe, aslist
|
||||
from rhodecode.lib.pidlock import DaemonLock, LockHeld
|
||||
|
|
@ -54,8 +53,7 @@ class ResultWrapper(object):
|
|||
|
||||
|
||||
def run_task(task, *args, **kwargs):
|
||||
global CELERY_ENABLED
|
||||
if CELERY_ENABLED:
|
||||
if rhodecode.CELERY_ENABLED:
|
||||
try:
|
||||
t = task.apply_async(args=args, kwargs=kwargs)
|
||||
log.info('running task %s:%s', t.task_id, task)
|
||||
|
|
@ -63,18 +61,18 @@ def run_task(task, *args, **kwargs):
|
|||
|
||||
except socket.error as e:
|
||||
if isinstance(e, IOError) and e.errno == 111:
|
||||
log.debug('Unable to connect to celeryd. Sync execution')
|
||||
CELERY_ENABLED = False
|
||||
log.error('Unable to connect to celeryd. Sync execution')
|
||||
rhodecode.CELERY_ENABLED = False
|
||||
else:
|
||||
log.exception("Exception while connecting to celeryd.")
|
||||
log.error("Exception while connecting to celeryd.")
|
||||
except KeyError as e:
|
||||
log.debug('Unable to connect to celeryd. Sync execution')
|
||||
log.error('Unable to connect to celeryd. Sync execution')
|
||||
except Exception as e:
|
||||
log.exception(
|
||||
"Exception while trying to run task asynchronous. "
|
||||
"Fallback to sync execution.")
|
||||
|
||||
log.debug('executing task %s in sync mode', task)
|
||||
else:
|
||||
log.debug('executing task %s in sync mode', task)
|
||||
return ResultWrapper(task(*args, **kwargs))
|
||||
|
||||
|
||||
|
|
@ -106,7 +104,7 @@ def locked_task(func):
|
|||
|
||||
|
||||
def get_session():
|
||||
if CELERY_ENABLED:
|
||||
if rhodecode.CELERY_ENABLED:
|
||||
utils.initialize_database(config)
|
||||
sa = meta.Session()
|
||||
return sa
|
||||
|
|
@ -118,7 +116,7 @@ def dbsession(func):
|
|||
ret = func(*fargs, **fkwargs)
|
||||
return ret
|
||||
finally:
|
||||
if CELERY_ENABLED and not CELERY_EAGER:
|
||||
if rhodecode.CELERY_ENABLED and not rhodecode.CELERY_EAGER:
|
||||
meta.Session.remove()
|
||||
|
||||
return decorator(__wrapper, func)
|
||||
|
|
@ -126,7 +124,7 @@ def dbsession(func):
|
|||
|
||||
def vcsconnection(func):
|
||||
def __wrapper(func, *fargs, **fkwargs):
|
||||
if CELERY_ENABLED and not CELERY_EAGER:
|
||||
if rhodecode.CELERY_ENABLED and not rhodecode.CELERY_EAGER:
|
||||
backends = config['vcs.backends'] = aslist(
|
||||
config.get('vcs.backends', 'hg,git'), sep=',')
|
||||
for alias in rhodecode.BACKENDS.keys():
|
||||
|
|
|
|||
|
|
@ -30,7 +30,7 @@ import logging
|
|||
from celery.task import task
|
||||
from pylons import config
|
||||
|
||||
from rhodecode import CELERY_ENABLED
|
||||
import rhodecode
|
||||
from rhodecode.lib.celerylib import (
|
||||
run_task, dbsession, __get_lockkey, LockHeld, DaemonLock,
|
||||
get_session, vcsconnection)
|
||||
|
|
@ -45,7 +45,7 @@ add_cache(config) # pragma: no cover
|
|||
|
||||
|
||||
def get_logger(cls):
|
||||
if CELERY_ENABLED:
|
||||
if rhodecode.CELERY_ENABLED:
|
||||
try:
|
||||
log = cls.get_logger()
|
||||
except Exception:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue