Changeset - 894a662b12b3
[Not reviewed]
default
0 6 0
Mads Kiilerich - 6 years ago 2020-02-13 16:41:51
mads@kiilerich.com
Grafted from: 9162e637d4e7
celery: refactor initialization - replace global CELERY_ON flag with CELERY_APP with the actual celery app that it indicates

Prepare for fixing how 193138922d56 broke celery due to magic dependencies on
initialization of global state at import time.
6 files changed with 16 insertions and 16 deletions:
0 comments (0 inline, 0 general)
kallithea/__init__.py
Show inline comments
 
@@ -19,41 +19,41 @@ Kallithea, a web based repository manage
 

	
 
Versioning implementation: http://www.python.org/dev/peps/pep-0386/
 

	
 
This file was forked by the Kallithea project in July 2014.
 
Original author and date, and relevant copyright and licensing information is below:
 
:created_on: Apr 9, 2010
 
:author: marcink
 
:copyright: (c) 2013 RhodeCode GmbH, (C) 2014 Bradley M. Kuhn, and others.
 
:license: GPLv3, see LICENSE.md for more details.
 
"""
 

	
 
import platform
 
import sys
 

	
 

	
 
if sys.version_info < (3, 6):
 
    raise Exception('Kallithea requires python 3.6 or later')
 

	
 
VERSION = (0, 5, 99)
 
BACKENDS = {
 
    'hg': 'Mercurial repository',
 
    'git': 'Git repository',
 
}
 

	
 
CELERY_ON = False
 
CELERY_APP = None  # set to Celery app instance if using Celery
 
CELERY_EAGER = False
 

	
 
CONFIG = {}
 

	
 
# Linked module for extensions
 
EXTENSIONS = {}
 

	
 
__version__ = '.'.join(str(each) for each in VERSION)
 
__platform__ = platform.system()
 
__license__ = 'GPLv3'
 
__py_version__ = sys.version_info
 
__author__ = "Various Authors"
 
__url__ = 'https://kallithea-scm.org/'
 

	
 
is_windows = __platform__ in ['Windows']
 
is_unix = not is_windows
kallithea/bin/kallithea_cli_celery.py
Show inline comments
 
@@ -11,31 +11,30 @@
 
#
 
# You should have received a copy of the GNU General Public License
 
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 

	
 
import click
 

	
 
import kallithea
 
import kallithea.bin.kallithea_cli_base as cli_base
 
from kallithea.lib import celerypylons
 

	
 

	
 
@cli_base.register_command(config_file_initialize_app=True)
 
@click.argument('celery_args', nargs=-1)
 
def celery_run(celery_args):
 
    """Start Celery worker(s) for asynchronous tasks.
 

	
 
    This commands starts the Celery daemon which will spawn workers to handle
 
    certain asynchronous tasks for Kallithea.
 

	
 
    Any extra arguments you pass to this command will be passed through to
 
    Celery. Use '--' before such extra arguments to avoid options to be parsed
 
    by this CLI command.
 
    """
 

	
 
    if not kallithea.CELERY_ON:
 
    if not kallithea.CELERY_APP:
 
        raise Exception('Please set use_celery = true in .ini config '
 
                        'file before running this command')
 

	
 
    app = celerypylons.make_app()
 
    cmd = celerypylons.worker.worker(app)
 
    cmd = celerypylons.worker.worker(kallithea.CELERY_APP)
 
    return cmd.run_from_argv(None, command='celery-run -c CONFIG_FILE --', argv=list(celery_args))
kallithea/config/app_cfg.py
Show inline comments
 
@@ -13,48 +13,49 @@
 
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
"""
 
Global configuration file for TurboGears2 specific settings in Kallithea.
 

	
 
This file complements the .ini file.
 
"""
 

	
 
import logging
 
import os
 
import platform
 
import sys
 

	
 
import alembic.config
 
import mercurial
 
import tg
 
from alembic.migration import MigrationContext
 
from alembic.script.base import ScriptDirectory
 
from sqlalchemy import create_engine
 
from tg.configuration import AppConfig
 
from tg.support.converters import asbool
 

	
 
import kallithea.lib.locale
 
import kallithea.model.base
 
import kallithea.model.meta
 
from kallithea.lib import celerypylons
 
from kallithea.lib.middleware.https_fixup import HttpsFixup
 
from kallithea.lib.middleware.permanent_repo_url import PermanentRepoUrl
 
from kallithea.lib.middleware.simplegit import SimpleGit
 
from kallithea.lib.middleware.simplehg import SimpleHg
 
from kallithea.lib.middleware.wrapper import RequestWrapper
 
from kallithea.lib.utils import check_git_version, load_rcextensions, make_ui, set_app_settings, set_indexer_config, set_vcs_config
 
from kallithea.lib.utils2 import safe_str, str2bool
 

	
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
class KallitheaAppConfig(AppConfig):
 
    # Note: AppConfig has a misleading name, as it's not the application
 
    # configuration, but the application configurator. The AppConfig values are
 
    # used as a template to create the actual configuration, which might
 
    # overwrite or extend the one provided by the configurator template.
 

	
 
    # To make it clear, AppConfig creates the config and sets into it the same
 
    # values that AppConfig itself has. Then the values from the config file and
 
    # gearbox options are loaded and merged into the configuration. Then an
 
    # after_init_config(conf) method of AppConfig is called for any change that
 
    # might depend on options provided by configuration files.
 

	
 
@@ -137,49 +138,50 @@ def setup_configuration(app):
 
        log.warn('database alembic revision checking is disabled')
 
    else:
 
        dbconf = config['sqlalchemy.url']
 
        alembic_cfg = alembic.config.Config()
 
        alembic_cfg.set_main_option('script_location', 'kallithea:alembic')
 
        alembic_cfg.set_main_option('sqlalchemy.url', dbconf)
 
        script_dir = ScriptDirectory.from_config(alembic_cfg)
 
        available_heads = sorted(script_dir.get_heads())
 

	
 
        engine = create_engine(dbconf)
 
        with engine.connect() as conn:
 
            context = MigrationContext.configure(conn)
 
            current_heads = sorted(str(s) for s in context.get_current_heads())
 
        if current_heads != available_heads:
 
            log.error('Failed to run Kallithea:\n\n'
 
                      'The database version does not match the Kallithea version.\n'
 
                      'Please read the documentation on how to upgrade or downgrade the database.\n'
 
                      'Current database version id(s): %s\n'
 
                      'Expected database version id(s): %s\n'
 
                      'If you are a developer and you know what you are doing, you can add `ignore_alembic_revision = True` '
 
                      'to your .ini file to skip the check.\n' % (' '.join(current_heads), ' '.join(available_heads)))
 
            sys.exit(1)
 

	
 
    # store some globals into kallithea
 
    kallithea.CELERY_ON = str2bool(config.get('use_celery'))
 
    if str2bool(config.get('use_celery')):
 
        kallithea.CELERY_APP = celerypylons.make_app()
 
    kallithea.CELERY_EAGER = str2bool(config.get('celery.always.eager'))
 
    kallithea.CONFIG = config
 

	
 
    load_rcextensions(root_path=config['here'])
 

	
 
    repos_path = safe_str(make_ui().configitems(b'paths')[0][1])
 
    config['base_path'] = repos_path
 
    set_app_settings(config)
 

	
 
    instance_id = kallithea.CONFIG.get('instance_id', '*')
 
    if instance_id == '*':
 
        instance_id = '%s-%s' % (platform.uname()[1], os.getpid())
 
        kallithea.CONFIG['instance_id'] = instance_id
 

	
 
    # update kallithea.CONFIG with the meanwhile changed 'config'
 
    kallithea.CONFIG.update(config)
 

	
 
    # configure vcs and indexer libraries (they are supposed to be independent
 
    # as much as possible and thus avoid importing tg.config or
 
    # kallithea.CONFIG).
 
    set_vcs_config(kallithea.CONFIG)
 
    set_indexer_config(kallithea.CONFIG)
 

	
 
    check_git_version()
kallithea/controllers/admin/repos.py
Show inline comments
 
@@ -160,54 +160,53 @@ class ReposController(BaseRepoController
 

	
 
        return htmlfill.render(
 
            render('admin/repos/repo_add.html'),
 
            defaults=defaults,
 
            errors={},
 
            prefix_error=False,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @LoginRequired()
 
    def repo_creating(self, repo_name):
 
        c.repo = repo_name
 
        c.task_id = request.GET.get('task_id')
 
        if not c.repo:
 
            raise HTTPNotFound()
 
        return render('admin/repos/repo_creating.html')
 

	
 
    @LoginRequired()
 
    @jsonify
 
    def repo_check(self, repo_name):
 
        c.repo = repo_name
 
        task_id = request.GET.get('task_id')
 

	
 
        if task_id and task_id not in ['None']:
 
            from kallithea import CELERY_ON
 
            from kallithea.lib import celerypylons
 
            if CELERY_ON:
 
                task = celerypylons.result.AsyncResult(task_id)
 
                if task.failed():
 
                    raise HTTPInternalServerError(task.traceback)
 
            if kallithea.CELERY_APP:
 
                task_result = celerypylons.result.AsyncResult(task_id)
 
                if task_result.failed():
 
                    raise HTTPInternalServerError(task_result.traceback)
 

	
 
        repo = Repository.get_by_repo_name(repo_name)
 
        if repo and repo.repo_state == Repository.STATE_CREATED:
 
            if repo.clone_uri:
 
                h.flash(_('Created repository %s from %s')
 
                        % (repo.repo_name, repo.clone_uri_hidden), category='success')
 
            else:
 
                repo_url = h.link_to(repo.repo_name,
 
                                     h.url('summary_home',
 
                                           repo_name=repo.repo_name))
 
                fork = repo.fork
 
                if fork is not None:
 
                    fork_name = fork.repo_name
 
                    h.flash(h.HTML(_('Forked repository %s as %s'))
 
                            % (fork_name, repo_url), category='success')
 
                else:
 
                    h.flash(h.HTML(_('Created repository %s')) % repo_url,
 
                            category='success')
 
            return {'result': True}
 
        return {'result': False}
 

	
 
    @HasRepoPermissionLevelDecorator('admin')
 
    def update(self, repo_name):
 
        c.repo_info = self._load_repo()
kallithea/lib/celerylib/__init__.py
Show inline comments
 
@@ -12,76 +12,76 @@
 
# You should have received a copy of the GNU General Public License
 
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
"""
 
kallithea.lib.celerylib
 
~~~~~~~~~~~~~~~~~~~~~~~
 

	
 
celery libs for Kallithea
 

	
 
This file was forked by the Kallithea project in July 2014.
 
Original author and date, and relevant copyright and licensing information is below:
 
:created_on: Nov 27, 2010
 
:author: marcink
 
:copyright: (c) 2013 RhodeCode GmbH, and others.
 
:license: GPLv3, see LICENSE.md for more details.
 
"""
 

	
 

	
 
import logging
 
import os
 
from hashlib import md5
 

	
 
from decorator import decorator
 
from tg import config
 

	
 
from kallithea import CELERY_EAGER, CELERY_ON
 
import kallithea
 
from kallithea.lib.pidlock import DaemonLock, LockHeld
 
from kallithea.lib.utils2 import safe_bytes
 
from kallithea.model import meta
 

	
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
class FakeTask(object):
 
    """Fake a sync result to make it look like a finished task"""
 

	
 
    def __init__(self, result):
 
        self.result = result
 

	
 
    def failed(self):
 
        return False
 

	
 
    traceback = None # if failed
 

	
 
    task_id = None
 

	
 

	
 
def task(f_org):
 
    """Wrapper of celery.task.task, running async if CELERY_ON
 
    """Wrapper of celery.task.task, running async if CELERY_APP
 
    """
 

	
 
    if CELERY_ON:
 
    if kallithea.CELERY_APP:
 
        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__
 
        from kallithea.lib import celerypylons
 
        runner = celerypylons.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:
 
        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('exception executing sync task %s in sync: %r', f_org.__name__, e)
 
                raise # TODO: return this in FakeTask as with async tasks?
 
            return FakeTask(result)
 

	
 
@@ -107,28 +107,28 @@ def locked_task(func):
 
        log.info('running task with lockkey %s', lockkey)
 
        try:
 
            l = DaemonLock(os.path.join(lockkey_path, lockkey))
 
            ret = func(*fargs, **fkwargs)
 
            l.release()
 
            return ret
 
        except LockHeld:
 
            log.info('LockHeld')
 
            return 'Task with key %s already running' % lockkey
 

	
 
    return decorator(__wrapper, func)
 

	
 

	
 
def get_session():
 
    sa = meta.Session()
 
    return sa
 

	
 

	
 
def dbsession(func):
 
    def __wrapper(func, *fargs, **fkwargs):
 
        try:
 
            ret = func(*fargs, **fkwargs)
 
            return ret
 
        finally:
 
            if CELERY_ON and not CELERY_EAGER:
 
            if kallithea.CELERY_APP and not kallithea.CELERY_EAGER:
 
                meta.Session.remove()
 

	
 
    return decorator(__wrapper, func)
kallithea/lib/celerylib/tasks.py
Show inline comments
 
@@ -15,49 +15,49 @@
 
kallithea.lib.celerylib.tasks
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

	
 
Kallithea task modules, containing all task that suppose to be run
 
by celery daemon
 

	
 
This file was forked by the Kallithea project in July 2014.
 
Original author and date, and relevant copyright and licensing information is below:
 
:created_on: Oct 6, 2010
 
:author: marcink
 
:copyright: (c) 2013 RhodeCode GmbH, and others.
 
:license: GPLv3, see LICENSE.md for more details.
 
"""
 

	
 
import email.utils
 
import logging
 
import os
 
import traceback
 
from collections import OrderedDict
 
from operator import itemgetter
 
from time import mktime
 

	
 
from tg import config
 

	
 
from kallithea import CELERY_ON
 
import kallithea
 
from kallithea.lib import celerylib, ext_json
 
from kallithea.lib.helpers import person
 
from kallithea.lib.hooks import log_create_repository
 
from kallithea.lib.rcmail.smtp_mailer import SmtpMailer
 
from kallithea.lib.utils import action_logger
 
from kallithea.lib.utils2 import ascii_bytes, str2bool
 
from kallithea.lib.vcs.utils import author_email
 
from kallithea.model.db import RepoGroup, Repository, Statistics, User
 

	
 

	
 
__all__ = ['whoosh_index', 'get_commits_stats', 'send_email']
 

	
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
@celerylib.task
 
@celerylib.locked_task
 
@celerylib.dbsession
 
def whoosh_index(repo_location, full_index):
 
    from kallithea.lib.indexers.daemon import WhooshIndexingDaemon
 
    celerylib.get_session() # initialize database connection
 

	
 
    index_location = config['index_dir']
 
@@ -197,49 +197,49 @@ def get_commits_stats(repo_name, ts_min_
 

	
 
        log.debug('last revision %s', last_rev)
 
        leftovers = len(repo.revisions[last_rev:])
 
        log.debug('revisions to parse %s', leftovers)
 

	
 
        if last_rev == 0 or leftovers < parse_limit:
 
            log.debug('getting code trending stats')
 
            stats.languages = ascii_bytes(ext_json.dumps(__get_codes_stats(repo_name)))
 

	
 
        try:
 
            stats.repository = dbrepo
 
            stats.stat_on_revision = last_cs.revision if last_cs else 0
 
            DBS.add(stats)
 
            DBS.commit()
 
        except:
 
            log.error(traceback.format_exc())
 
            DBS.rollback()
 
            lock.release()
 
            return False
 

	
 
        # final release
 
        lock.release()
 

	
 
        # execute another task if celery is enabled
 
        if len(repo.revisions) > 1 and CELERY_ON and recurse_limit > 0:
 
        if len(repo.revisions) > 1 and kallithea.CELERY_APP and recurse_limit > 0:
 
            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 celerylib.LockHeld:
 
        log.info('Task with key %s already running', lockkey)
 
        return 'Task with key %s already running' % lockkey
 

	
 

	
 
@celerylib.task
 
@celerylib.dbsession
 
def send_email(recipients, subject, body='', html_body='', headers=None, author=None):
 
    """
 
    Sends an email with defined parameters from the .ini files.
 

	
 
    :param recipients: list of recipients, if this is None, the defined email
 
        address from field 'email_to' and all admins is used instead
 
    :param subject: subject of the mail
 
    :param body: body of the mail
 
    :param html_body: html version of body
 
    :param headers: dictionary of prepopulated e-mail headers
 
    :param author: User object of the author of this mail, if known and relevant
 
    """
0 comments (0 inline, 0 general)