Changeset - 82662f9faaf4
[Not reviewed]
default
0 8 0
Mads Kiilerich - 9 years ago 2016-09-06 00:51:18
madski@unity3d.com
celeryd: annotate tasks so they can be run directly without run_task

This also makes the system less forgiving about celery configuration problems
and thus easier to debug. I like that.
8 files changed with 52 insertions and 56 deletions:
0 comments (0 inline, 0 general)
kallithea/controllers/admin/settings.py
Show inline comments
 
@@ -37,7 +37,7 @@ from webob.exc import HTTPFound
 
from kallithea.lib import helpers as h
 
from kallithea.lib.auth import LoginRequired, HasPermissionAnyDecorator
 
from kallithea.lib.base import BaseController, render
 
from kallithea.lib.celerylib import tasks, run_task
 
from kallithea.lib.celerylib import tasks
 
from kallithea.lib.exceptions import HgsubversionImportError
 
from kallithea.lib.utils import repo2db_mapper, set_app_settings
 
from kallithea.model.db import Ui, Repository, Setting
 
@@ -325,7 +325,7 @@ class SettingsController(BaseController)
 

	
 
            recipients = [test_email] if test_email else None
 

	
 
            run_task(tasks.send_email, recipients, test_email_subj,
 
            tasks.send_email(recipients, test_email_subj,
 
                     test_email_txt_body, test_email_html_body)
 

	
 
            h.flash(_('Send email task created'), category='success')
 
@@ -398,7 +398,7 @@ class SettingsController(BaseController)
 
        if request.POST:
 
            repo_location = self._get_hg_ui_settings()['paths_root_path']
 
            full_index = request.POST.get('full_index', False)
 
            run_task(tasks.whoosh_index, repo_location, full_index)
 
            tasks.whoosh_index(repo_location, full_index)
 
            h.flash(_('Whoosh reindex task scheduled'), category='success')
 
            raise HTTPFound(location=url('admin_settings_search'))
 

	
kallithea/controllers/summary.py
Show inline comments
 
@@ -49,7 +49,6 @@ from kallithea.lib.auth import LoginRequ
 
from kallithea.lib.base import BaseRepoController, render
 
from kallithea.lib.vcs.backends.base import EmptyChangeset
 
from kallithea.lib.markup_renderer import MarkupRenderer
 
from kallithea.lib.celerylib import run_task
 
from kallithea.lib.celerylib.tasks import get_commits_stats
 
from kallithea.lib.compat import json
 
from kallithea.lib.vcs.nodes import FileNode
 
@@ -225,6 +224,5 @@ class SummaryController(BaseRepoControll
 
            c.no_data = True
 

	
 
        recurse_limit = 500  # don't recurse more than 500 times when parsing
 
        run_task(get_commits_stats, c.db_repo.repo_name, ts_min_y,
 
                 ts_max_y, recurse_limit)
 
        get_commits_stats(c.db_repo.repo_name, ts_min_y, ts_max_y, recurse_limit)
 
        return render('summary/statistics.html')
kallithea/lib/celerylib/__init__.py
Show inline comments
 
@@ -27,8 +27,6 @@ Original author and date, and relevant c
 

	
 

	
 
import os
 
import socket
 
import traceback
 
import logging
 

	
 
from pylons import config
 
@@ -62,33 +60,36 @@ class FakeTask(object):
 
    task_id = None
 

	
 

	
 
def run_task(task, *args, **kwargs):
 
    global CELERY_ON
 
    if CELERY_ON:
 
        try:
 
            t = task.apply_async(args=args, kwargs=kwargs)
 
            log.info('running task %s:%s', t.task_id, task)
 
            return t
 
def task(f_org):
 
    """Wrapper of celery.task.task, running async if CELERY_ON
 
    """
 

	
 
        except socket.error as e:
 
            if isinstance(e, IOError) and e.errno == 111:
 
                log.debug('Unable to connect to celeryd. Sync execution')
 
                CELERY_ON = False
 
    if CELERY_ON:
 
        def f_async(*args, **kwargs):
 
            log.info('executing %s task', f_org.__name__)
 
            try:
 
                f_org(*args, **kwargs)
 
            finally:
 
                log.info('executed %s task', f_org.__name__)
 
        f_async.__name__ = f_org.__name__
 
        import celery.task
 
        runner = celery.task.task(ignore_result=True)(f_async)
 
        def f_wrapped(*args, **kwargs):
 
            t = runner.apply_async(args=args, kwargs=kwargs)
 
            log.info('executing task %s in async mode - id %s', f_org, t.task_id)
 
            return t
 
            else:
 
                log.error(traceback.format_exc())
 
        except KeyError as e:
 
                log.debug('Unable to connect to celeryd. Sync execution')
 
        def f_wrapped(*args, **kwargs):
 
            log.info('executing task %s in sync', f_org.__name__)
 
            try:
 
                result = f_org(*args, **kwargs)
 
        except Exception as e:
 
            log.error(traceback.format_exc())
 

	
 
    log.debug('executing task %s in sync mode', task)
 
    try:
 
        result = task(*args, **kwargs)
 
    except Exception as e:
 
        log.error('exception running sync task %s: %s', task, e)
 
                log.error('exception executing sync task %s in sync', f_org.__name__, e)
 
        raise # TODO: return this in FakeTask as with async tasks?
 
    return FakeTask(result)
 

	
 
    return f_wrapped
 

	
 

	
 
def __get_lockkey(func, *fargs, **fkwargs):
 
    params = list(fargs)
kallithea/lib/celerylib/tasks.py
Show inline comments
 
@@ -26,8 +26,6 @@ Original author and date, and relevant c
 
:license: GPLv3, see LICENSE.md for more details.
 
"""
 

	
 
from celery.task import task
 

	
 
import os
 
import traceback
 
import logging
 
@@ -40,7 +38,8 @@ from string import lower
 
from pylons import config
 

	
 
from kallithea import CELERY_ON
 
from kallithea.lib.celerylib import run_task, locked_task, dbsession, \
 
from kallithea.lib import celerylib
 
from kallithea.lib.celerylib import locked_task, dbsession, \
 
    str2bool, __get_lockkey, LockHeld, DaemonLock, get_session
 
from kallithea.lib.helpers import person
 
from kallithea.lib.rcmail.smtp_mailer import SmtpMailer
 
@@ -60,7 +59,7 @@ __all__ = ['whoosh_index', 'get_commits_
 
log = logging.getLogger(__name__)
 

	
 

	
 
@task(ignore_result=True)
 
@celerylib.task
 
@locked_task
 
@dbsession
 
def whoosh_index(repo_location, full_index):
 
@@ -73,7 +72,7 @@ def whoosh_index(repo_location, full_ind
 
                         .run(full_index=full_index)
 

	
 

	
 
@task(ignore_result=True)
 
@celerylib.task
 
@dbsession
 
def get_commits_stats(repo_name, ts_min_y, ts_max_y, recurse_limit=100):
 
    DBS = get_session()
 
@@ -228,18 +227,17 @@ def get_commits_stats(repo_name, ts_min_
 

	
 
        # execute another task if celery is enabled
 
        if len(repo.revisions) > 1 and CELERY_ON and recurse_limit > 0:
 
            recurse_limit -= 1
 
            run_task(get_commits_stats, repo_name, ts_min_y, ts_max_y,
 
                     recurse_limit)
 
        if recurse_limit <= 0:
 
            log.debug('Breaking recursive mode due to reach of recurse limit')
 
        return True
 
            get_commits_stats(repo_name, ts_min_y, ts_max_y, recurse_limit - 1)
 
        elif recurse_limit <= 0:
 
            log.debug('Not recursing - limit has been reached')
 
        else:
 
            log.debug('Not recursing')
 
    except LockHeld:
 
        log.info('Task with key %s already running', lockkey)
 
        return 'Task with key %s already running' % lockkey
 

	
 

	
 
@task(ignore_result=True)
 
@celerylib.task
 
@dbsession
 
def send_email(recipients, subject, body='', html_body='', headers=None, author=None):
 
    """
 
@@ -326,7 +324,7 @@ def send_email(recipients, subject, body
 
        return False
 
    return True
 

	
 
@task(ignore_result=False)
 
@celerylib.task
 
@dbsession
 
def create_repo(form_data, cur_user):
 
    from kallithea.model.repo import RepoModel
 
@@ -410,7 +408,7 @@ def create_repo(form_data, cur_user):
 
    return True
 

	
 

	
 
@task(ignore_result=False)
 
@celerylib.task
 
@dbsession
 
def create_repo_fork(form_data, cur_user):
 
    """
kallithea/model/notification.py
Show inline comments
 
@@ -62,7 +62,7 @@ class NotificationModel(BaseModel):
 
        :param with_email: send email with this notification
 
        :param email_kwargs: additional dict to pass as args to email template
 
        """
 
        from kallithea.lib.celerylib import tasks, run_task
 
        from kallithea.lib.celerylib import tasks
 
        email_kwargs = email_kwargs or {}
 
        if recipients and not getattr(recipients, '__iter__', False):
 
            raise Exception('recipients must be a list or iterable')
 
@@ -132,7 +132,7 @@ class NotificationModel(BaseModel):
 
            email_html_body = EmailNotificationModel() \
 
                                .get_email_tmpl(type_, 'html', **html_kwargs)
 

	
 
            run_task(tasks.send_email, [rec.email], email_subject, email_txt_body,
 
            tasks.send_email([rec.email], email_subject, email_txt_body,
 
                     email_html_body, headers, author=created_by_obj)
 

	
 
        return notif
kallithea/model/repo.py
Show inline comments
 
@@ -473,8 +473,8 @@ class RepoModel(BaseModel):
 
        :param form_data:
 
        :param cur_user:
 
        """
 
        from kallithea.lib.celerylib import tasks, run_task
 
        return run_task(tasks.create_repo, form_data, cur_user)
 
        from kallithea.lib.celerylib import tasks
 
        return tasks.create_repo(form_data, cur_user)
 

	
 
    def _update_permissions(self, repo, perms_new=None, perms_updates=None,
 
                            check_perms=True):
 
@@ -522,8 +522,8 @@ class RepoModel(BaseModel):
 
        :param form_data:
 
        :param cur_user:
 
        """
 
        from kallithea.lib.celerylib import tasks, run_task
 
        return run_task(tasks.create_repo_fork, form_data, cur_user)
 
        from kallithea.lib.celerylib import tasks
 
        return tasks.create_repo_fork(form_data, cur_user)
 

	
 
    def delete(self, repo, forks=None, fs_remove=True, cur_user=None):
 
        """
kallithea/model/user.py
Show inline comments
 
@@ -329,7 +329,7 @@ class UserModel(BaseModel):
 
        allowing users to copy-paste or manually enter the token from the
 
        email.
 
        """
 
        from kallithea.lib.celerylib import tasks, run_task
 
        from kallithea.lib.celerylib import tasks
 
        from kallithea.model.notification import EmailNotificationModel
 
        import kallithea.lib.helpers as h
 

	
 
@@ -364,8 +364,7 @@ class UserModel(BaseModel):
 
                reset_token=token,
 
                reset_url=link)
 
            log.debug('sending email')
 
            run_task(tasks.send_email, [user_email],
 
                     _("Password reset link"), body, html_body)
 
            tasks.send_email([user_email], _("Password reset link"), body, html_body)
 
            log.info('send new password mail to %s', user_email)
 
        else:
 
            log.debug("password reset email %s not found", user_email)
 
@@ -375,7 +374,7 @@ class UserModel(BaseModel):
 
                     timestamp=timestamp)
 

	
 
    def verify_reset_password_token(self, email, timestamp, token):
 
        from kallithea.lib.celerylib import tasks, run_task
 
        from kallithea.lib.celerylib import tasks
 
        from kallithea.lib import auth
 
        import kallithea.lib.helpers as h
 
        user = User.get_by_email(email)
 
@@ -401,7 +400,7 @@ class UserModel(BaseModel):
 
        return expected_token == token
 

	
 
    def reset_password(self, user_email, new_passwd):
 
        from kallithea.lib.celerylib import tasks, run_task
 
        from kallithea.lib.celerylib import tasks
 
        from kallithea.lib import auth
 
        user = User.get_by_email(user_email)
 
        if user is not None:
 
@@ -414,7 +413,7 @@ class UserModel(BaseModel):
 
        if new_passwd is None:
 
            raise Exception('unable to set new password')
 

	
 
        run_task(tasks.send_email, [user_email],
 
        tasks.send_email([user_email],
 
                 _('Password reset notification'),
 
                 _('The password to your account %s has been changed using password reset form.') % (user.username,))
 
        log.info('send password reset mail to %s', user_email)
kallithea/tests/models/test_notifications.py
Show inline comments
 
@@ -265,7 +265,7 @@ class TestNotifications(TestController):
 
            # Email type TYPE_PASSWORD_RESET has no corresponding notification type - test it directly:
 
            desc = 'TYPE_PASSWORD_RESET'
 
            kwargs = dict(user='John Doe', reset_token='decbf64715098db5b0bd23eab44bd792670ab746', reset_url='http://reset.com/decbf64715098db5b0bd23eab44bd792670ab746')
 
            kallithea.lib.celerylib.run_task(kallithea.lib.celerylib.tasks.send_email, ['john@doe.com'],
 
            kallithea.lib.celerylib.tasks.send_email(['john@doe.com'],
 
                "Password reset link",
 
                EmailNotificationModel().get_email_tmpl(EmailNotificationModel.TYPE_PASSWORD_RESET, 'txt', **kwargs),
 
                EmailNotificationModel().get_email_tmpl(EmailNotificationModel.TYPE_PASSWORD_RESET, 'html', **kwargs),
0 comments (0 inline, 0 general)