scheduler: added DB models and db parsers for the RhodeCode scheduler.

- migration schema
- small fixes for form templates
- migrated celeryloader utils for the scheduler

rhodecode/lib/celerylib/scheduler.py

@@ -0,0 +1,60 b''
+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2010-2017 RhodeCode GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# (only), as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# This program is dual-licensed. If you wish to learn more about the
+# RhodeCode Enterprise Edition, including its added features, Support services,
+# and proprietary license terms, please see https://rhodecode.com/licenses/
+import logging
+import importlib
+
+from celery.beat import (
+    PersistentScheduler, ScheduleEntry as CeleryScheduleEntry)
+
+log = logging.getLogger(__name__)
+
+
+class FileScheduleEntry(CeleryScheduleEntry):
+    def __init__(self, name=None, task=None, last_run_at=None,
+                 total_run_count=None, schedule=None, args=(), kwargs=None,
+                 options=None, relative=False, app=None, **_kwargs):
+        kwargs = kwargs or {}
+        options = options or {}
+
+        # because our custom loader passes in some variables that the original
+        # function doesn't expect, we have this thin wrapper
+
+        super(FileScheduleEntry, self).__init__(
+            name=name, task=task, last_run_at=last_run_at,
+            total_run_count=total_run_count, schedule=schedule, args=args,
+            kwargs=kwargs, options=options, relative=relative, app=app)
+
+
+class FileScheduler(PersistentScheduler):
+    """CE base scheduler"""
+    Entry = FileScheduleEntry
+
+    def setup_schedule(self):
+        log.info("setup_schedule called")
+        super(FileScheduler, self).setup_schedule()
+
+
+try:
+    # try if we have EE scheduler available
+    module = importlib.import_module('rc_ee.lib.celerylib.scheduler')
+    RcScheduler = module.RcScheduler
+except ImportError:
+    # fallback to CE scheduler
+    RcScheduler = FileScheduler

rhodecode/lib/dbmigrate/versions/082_version_4_11_0.py

@@ -0,0 +1,32 b''
+import logging
+
+from sqlalchemy import *
+
+from rhodecode.model import meta
+from rhodecode.lib.dbmigrate.versions import _reset_base, notify
+
+log = logging.getLogger(__name__)
+
+
+def upgrade(migrate_engine):
+    """
+    Upgrade operations go here.
+    Don't create your own engine; bind migrate_engine to your metadata
+    """
+    _reset_base(migrate_engine)
+    from rhodecode.lib.dbmigrate.schema import db_4_11_0_0
+    db_4_11_0_0.ScheduleEntry().__table__.create()
+
+    # issue fixups
+    fixups(db_4_11_0_0, meta.Session)
+
+
+def downgrade(migrate_engine):
+    meta = MetaData()
+    meta.bind = migrate_engine
+
+
+def fixups(models, _SESSION):
+    pass
+
+

rhodecode/__init__.py

@@ -51,7 +51,7 b' PYRAMID_SETTINGS = {}'
 EXTENSIONS = {}
 
 __version__ = ('.'.join((str(each) for each in VERSION[:3])))
-__dbversion__ = 81  # defines current db version for migrations
+__dbversion__ = 82  # defines current db version for migrations
 __platform__ = platform.system()
 __license__ = 'AGPLv3, and Commercial License'
 __author__ = 'RhodeCode GmbH'

rhodecode/lib/celerylib/loader.py

@@ -20,7 +20,11 b''
 """
 Celery loader, run with::
 
-    celery worker --beat --app rhodecode.lib.celerylib.loader --loglevel DEBUG --ini=._dev/dev.ini
+    celery worker \
+        --beat \
+        --app rhodecode.lib.celerylib.loader \
+        --scheduler rhodecode.lib.celerylib.scheduler.RcScheduler \
+        --loglevel DEBUG --ini=._dev/dev.ini
 """
 import os
 import logging
@@ -83,6 +87,10 b' base_celery_config = {'
     'task_serializer': 'json_ext',
     'result_serializer': 'json_ext',
     'worker_hijack_root_logger': False,
+    'database_table_names': {
+        'task': 'beat_taskmeta',
+        'group': 'beat_groupmeta',
+    }
 }
 # init main celery app
 celery_app = Celery()

rhodecode/lib/celerylib/tasks.py

@@ -24,6 +24,7 b' by celery daemon'
 """
 
 import os
+import time
 
 import rhodecode
 from rhodecode.lib import audit_logger
@@ -268,8 +269,17 b' def create_repo_fork(form_data, cur_user'
 def sync_repo(*args, **kwargs):
     from rhodecode.model.scm import ScmModel
     log = get_logger(sync_repo)
-
-    log.info('Pulling from %s', kwargs['repo_name'])
+    repo_name = kwargs['repo_name']
+    log.info('Pulling from %s', repo_name)
+    dbrepo = Repository.get_by_repo_name(repo_name)
+    if dbrepo and dbrepo.clone_uri:
     ScmModel().pull_changes(kwargs['repo_name'], kwargs['username'])
+    else:
+        log.debug('Repo `%s` not found or without a clone_url', repo_name)
 
 
+@async_task(ignore_result=False)
+def beat_check(*args, **kwargs):
+    log = get_logger(beat_check)
+    log.info('Got args: %r and kwargs %r', args, kwargs)
+    return time.time()

rhodecode/lib/celerylib/utils.py

@@ -64,34 +64,44 b' def safe_json(get, section, key):'
     return json_value
 
 
-def get_beat_config(parser, section):
-    SCHEDULE_TYPE_MAP = {
+def raw_2_schedule(schedule_value, schedule_type):
+    schedule_type_map = {
         'crontab': crontab,
         'timedelta': timedelta,
         'integer': int
     }
+    scheduler_cls = schedule_type_map.get(schedule_type)
+
+    if scheduler_cls is None:
+        raise ValueError(
+            'schedule type %s in section is invalid' % (
+                schedule_type,
+            )
+        )
+    try:
+        schedule = scheduler_cls(schedule_value)
+    except TypeError:
+        log.exception('Failed to compose a schedule from value: %r', schedule_value)
+        schedule = None
+    return schedule
+
+
+def get_beat_config(parser, section):
+
     get = partial(parser.get, section)
     has_option = partial(parser.has_option, section)
 
     schedule_type = get('type')
     schedule_value = safe_json(get, section, 'schedule')
 
-    scheduler_cls = SCHEDULE_TYPE_MAP.get(schedule_type)
-
-    if scheduler_cls is None:
-        raise ValueError(
-            'schedule type %s in section %s is invalid' % (
-                schedule_type,
-                section
-            )
-        )
-
-    schedule = scheduler_cls(schedule_value)
-
     config = {
+        'schedule_type': schedule_type,
+        'schedule_value': schedule_value,
         'task': get('task'),
-        'schedule': schedule,
     }
+    schedule = raw_2_schedule(schedule_value, schedule_type)
+    if schedule:
+        config['schedule'] = schedule
 
     if has_option('args'):
         config['args'] = safe_json(get, section, 'args')
@@ -99,6 +109,9 b' def get_beat_config(parser, section):'
     if has_option('kwargs'):
         config['kwargs'] = safe_json(get, section, 'kwargs')
 
+    if has_option('force_update'):
+        config['force_update'] = get('force_update')
+
     return config
 
 

rhodecode/model/db.py

@@ -36,7 +36,7 b' import collections'
 
 from sqlalchemy import (
     or_, and_, not_, func, TypeDecorator, event,
-    Index, UniqueConstraint, ForeignKey, CheckConstraint, Column,
+    Index, Sequence, UniqueConstraint, ForeignKey, CheckConstraint, Column,
     Boolean, String, Unicode, UnicodeText, DateTime, Integer, LargeBinary,
     Text, Float, PickleType)
 from sqlalchemy.sql.expression import true, false
@@ -225,10 +225,10 b' class BaseModel(object):'
         """return list with keys and values tuples corresponding
         to this model data """
 
-        l = []
+        lst = []
         for k in self._get_keys():
-            l.append((k, getattr(self, k),))
-        return l
+            lst.append((k, getattr(self, k),))
+        return lst
 
     def populate_obj(self, populate_dict):
         """populate model with data from given populate_dict"""
@@ -4208,6 +4208,136 b' class RepoReviewRule(Base, BaseModel):'
             self.repo_review_rule_id, self.repo)
 
 
+class ScheduleEntry(Base, BaseModel):
+    __tablename__ = 'schedule_entries'
+    __table_args__ = (
+        UniqueConstraint('schedule_name', name='s_schedule_name_idx'),
+        UniqueConstraint('task_uid', name='s_task_uid_idx'),
+        {'extend_existing': True, 'mysql_engine': 'InnoDB',
+         'mysql_charset': 'utf8', 'sqlite_autoincrement': True},
+    )
+    schedule_types = ['crontab', 'timedelta', 'integer']
+    schedule_entry_id = Column('schedule_entry_id', Integer(), primary_key=True)
+
+    schedule_name = Column("schedule_name", String(255), nullable=False, unique=None, default=None)
+    schedule_description = Column("schedule_description", String(10000), nullable=True, unique=None, default=None)
+    schedule_enabled = Column("schedule_enabled", Boolean(), nullable=False, unique=None, default=True)
+
+    _schedule_type = Column("schedule_type", String(255), nullable=False, unique=None, default=None)
+    schedule_definition = Column('schedule_definition_json', MutationObj.as_mutable(JsonType(default=lambda: "", dialect_map=dict(mysql=LONGTEXT()))))
+
+    schedule_last_run = Column('schedule_last_run', DateTime(timezone=False), nullable=True, unique=None, default=None)
+    schedule_total_run_count = Column('schedule_total_run_count', Integer(), nullable=True, unique=None, default=0)
+
+    # task
+    task_uid = Column("task_uid", String(255), nullable=False, unique=None, default=None)
+    task_dot_notation = Column("task_dot_notation", String(4096), nullable=False, unique=None, default=None)
+    task_args = Column('task_args_json', MutationObj.as_mutable(JsonType(default=list, dialect_map=dict(mysql=LONGTEXT()))))
+    task_kwargs = Column('task_kwargs_json', MutationObj.as_mutable(JsonType(default=dict, dialect_map=dict(mysql=LONGTEXT()))))
+
+    created_on = Column('created_on', DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
+    updated_on = Column('updated_on', DateTime(timezone=False), nullable=True, unique=None, default=None)
+
+    @hybrid_property
+    def schedule_type(self):
+        return self._schedule_type
+
+    @schedule_type.setter
+    def schedule_type(self, val):
+        if val not in self.schedule_types:
+            raise ValueError('Value must be on of `{}` and got `{}`'.format(
+                val, self.schedule_type))
+
+        self._schedule_type = val
+
+    @classmethod
+    def get_uid(cls, obj):
+        args = obj.task_args
+        kwargs = obj.task_kwargs
+        if isinstance(args, JsonRaw):
+            try:
+                args = json.loads(args)
+            except ValueError:
+                args = tuple()
+
+        if isinstance(kwargs, JsonRaw):
+            try:
+                kwargs = json.loads(kwargs)
+            except ValueError:
+                kwargs = dict()
+
+        dot_notation = obj.task_dot_notation
+        val = '.'.join(map(safe_str, [
+            sorted(dot_notation), args, sorted(kwargs.items())]))
+        return hashlib.sha1(val).hexdigest()
+
+    @classmethod
+    def get_by_schedule_name(cls, schedule_name):
+        return cls.query().filter(cls.schedule_name == schedule_name).scalar()
+
+    @classmethod
+    def get_by_schedule_id(cls, schedule_id):
+        return cls.query().filter(cls.schedule_entry_id == schedule_id).scalar()
+
+    @property
+    def task(self):
+        return self.task_dot_notation
+
+    @property
+    def schedule(self):
+        from rhodecode.lib.celerylib.utils import raw_2_schedule
+        schedule = raw_2_schedule(self.schedule_definition, self.schedule_type)
+        return schedule
+
+    @property
+    def args(self):
+        try:
+            return list(self.task_args or [])
+        except ValueError:
+            return list()
+
+    @property
+    def kwargs(self):
+        try:
+            return dict(self.task_kwargs or {})
+        except ValueError:
+            return dict()
+
+    def _as_raw(self, val):
+        if hasattr(val, 'de_coerce'):
+            val = val.de_coerce()
+            if val:
+                val = json.dumps(val)
+
+        return val
+
+    @property
+    def schedule_definition_raw(self):
+        return self._as_raw(self.schedule_definition)
+
+    @property
+    def args_raw(self):
+        return self._as_raw(self.task_args)
+
+    @property
+    def kwargs_raw(self):
+        return self._as_raw(self.task_kwargs)
+
+    def __repr__(self):
+        return '<DB:ScheduleEntry({}:{})>'.format(
+            self.schedule_entry_id, self.schedule_name)
+
+
[e-mail]_for(ScheduleEntry, 'before_update')
+def update_task_uid(mapper, connection, target):
+    target.task_uid = ScheduleEntry.get_uid(target)
+
+
[e-mail]_for(ScheduleEntry, 'before_insert')
+def set_task_uid(mapper, connection, target):
+    target.task_uid = ScheduleEntry.get_uid(target)
+
+
 class DbMigrateVersion(Base, BaseModel):
     __tablename__ = 'db_migrate_version'
     __table_args__ = (

rhodecode/templates/forms/textinput.pt

@@ -5,6 +5,7 b''
                   placeholder placeholder|field.widget.placeholder|field.placeholder|'';
                   mask_placeholder mask_placeholder|field.widget.mask_placeholder;
                   style style|field.widget.style;
+                  help_block help_block|field.widget.help_block|'';
 "
       tal:omit-tag="">
     <input type="text" name="${name}" value="${cstruct}"
@@ -12,6 +13,8 b''
                            style style"
             placeholder="${placeholder}"
            id="${oid}"/>
+
+    <p tal:condition="help_block" class="help-block">${help_block}</p>
     <script tal:condition="mask" type="text/javascript">
       deform.addCallback(
          '${oid}',