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

@@ -1,63 +1,63 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/
 
 """
 
 RhodeCode, a web based repository management software
 versioning implementation: http://www.python.org/dev/peps/pep-0386/
 """
 
 import os
 import sys
 import platform
 
 VERSION = tuple(open(os.path.join(
     os.path.dirname(__file__), 'VERSION')).read().split('.'))
 
 BACKENDS = {
     'hg': 'Mercurial repository',
     'git': 'Git repository',
     'svn': 'Subversion repository',
 }
 
 CELERY_ENABLED = False
 CELERY_EAGER = False
 
 # link to config for pyramid
 CONFIG = {}
 
 # Populated with the settings dictionary from application init in
 # rhodecode.conf.environment.load_pyramid_environment
 PYRAMID_SETTINGS = {}
 
 # Linked module for extensions
 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'
 __url__ = 'https://code.rhodecode.com'
 
 is_windows = __platform__ in ['Windows']
 is_unix = not is_windows
 is_test = False
 disable_error_handler = False

rhodecode/lib/celerylib/loader.py

@@ -1,257 +1,265 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/
 """
 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
 
 from celery import Celery
 from celery import signals
 from celery import Task
 from celery import exceptions  # noqa
 from kombu.serialization import register
 from pyramid.threadlocal import get_current_request
 
 import rhodecode
 
 from rhodecode.lib.auth import AuthUser
 from rhodecode.lib.celerylib.utils import get_ini_config, parse_ini_vars
 from rhodecode.lib.ext_json import json
 from rhodecode.lib.pyramid_utils import bootstrap, setup_logging, prepare_request
 from rhodecode.lib.utils2 import str2bool
 from rhodecode.model import meta
 
 
 register('json_ext', json.dumps, json.loads,
          content_type='application/x-json-ext',
          content_encoding='utf-8')
 
 log = logging.getLogger('celery.rhodecode.loader')
 
 
 def add_preload_arguments(parser):
     parser.add_argument(
         '--ini', default=None,
         help='Path to ini configuration file.'
     )
     parser.add_argument(
         '--ini-var', default=None,
         help='Comma separated list of key=value to pass to ini.'
     )
 
 
 def get_logger(obj):
     custom_log = logging.getLogger(
         'rhodecode.task.{}'.format(obj.__class__.__name__))
 
     if rhodecode.CELERY_ENABLED:
         try:
             custom_log = obj.get_logger()
         except Exception:
             pass
 
     return custom_log
 
 
 base_celery_config = {
     'result_backend': 'rpc://',
     'result_expires': 60 * 60 * 24,
     'result_persistent': True,
     'imports': [],
     'worker_max_tasks_per_child': 100,
     'accept_content': ['json_ext'],
     '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()
 celery_app.user_options['preload'].add(add_preload_arguments)
 ini_file_glob = None
 
 
 @signals.setup_logging.connect
 def setup_logging_callback(**kwargs):
     setup_logging(ini_file_glob)
 
 
 @signals.user_preload_options.connect
 def on_preload_parsed(options, **kwargs):
     ini_location = options['ini']
     ini_vars = options['ini_var']
     celery_app.conf['INI_PYRAMID'] = options['ini']
 
     if ini_location is None:
         print('You must provide the paste --ini argument')
         exit(-1)
 
     options = None
     if ini_vars is not None:
         options = parse_ini_vars(ini_vars)
 
     global ini_file_glob
     ini_file_glob = ini_location
 
     log.debug('Bootstrapping RhodeCode application...')
     env = bootstrap(ini_location, options=options)
 
     setup_celery_app(
         app=env['app'], root=env['root'], request=env['request'],
         registry=env['registry'], closer=env['closer'],
         ini_location=ini_location)
 
     # fix the global flag even if it's disabled via .ini file because this
     # is a worker code that doesn't need this to be disabled.
     rhodecode.CELERY_ENABLED = True
 
 
 @signals.task_success.connect
 def task_success_signal(result, **kwargs):
     meta.Session.commit()
     celery_app.conf['PYRAMID_CLOSER']()
 
 
 @signals.task_retry.connect
 def task_retry_signal(
         request, reason, einfo, **kwargs):
     meta.Session.remove()
     celery_app.conf['PYRAMID_CLOSER']()
 
 
 @signals.task_failure.connect
 def task_failure_signal(
         task_id, exception, args, kwargs, traceback, einfo, **kargs):
     meta.Session.remove()
     celery_app.conf['PYRAMID_CLOSER']()
 
 
 @signals.task_revoked.connect
 def task_revoked_signal(
         request, terminated, signum, expired, **kwargs):
     celery_app.conf['PYRAMID_CLOSER']()
 
 
 def setup_celery_app(app, root, request, registry, closer, ini_location):
     ini_dir = os.path.dirname(os.path.abspath(ini_location))
     celery_config = base_celery_config
     celery_config.update({
         # store celerybeat scheduler db where the .ini file is
         'beat_schedule_filename': os.path.join(ini_dir, 'celerybeat-schedule'),
     })
     ini_settings = get_ini_config(ini_location)
     log.debug('Got custom celery conf: %s', ini_settings)
 
     celery_config.update(ini_settings)
     celery_app.config_from_object(celery_config)
 
     celery_app.conf.update({'PYRAMID_APP': app})
     celery_app.conf.update({'PYRAMID_ROOT': root})
     celery_app.conf.update({'PYRAMID_REQUEST': request})
     celery_app.conf.update({'PYRAMID_REGISTRY': registry})
     celery_app.conf.update({'PYRAMID_CLOSER': closer})
 
 
 def configure_celery(config, ini_location):
     """
     Helper that is called from our application creation logic. It gives
     connection info into running webapp and allows execution of tasks from
     RhodeCode itself
     """
     # store some globals into rhodecode
     rhodecode.CELERY_ENABLED = str2bool(
         config.registry.settings.get('use_celery'))
     if rhodecode.CELERY_ENABLED:
         log.info('Configuring celery based on `%s` file', ini_location)
         setup_celery_app(
             app=None, root=None, request=None, registry=config.registry,
             closer=None, ini_location=ini_location)
 
 
 class RequestContextTask(Task):
     """
     This is a celery task which will create a rhodecode app instance context
     for the task, patch pyramid with the original request
     that created the task and also add the user to the context.
     """
 
     def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
                     link=None, link_error=None, shadow=None, **options):
         """ queue the job to run (we are in web request context here) """
 
         req = get_current_request()
 
         # web case
         if hasattr(req, 'user'):
             ip_addr = req.user.ip_addr
             user_id = req.user.user_id
 
         # api case
         elif hasattr(req, 'rpc_user'):
             ip_addr = req.rpc_user.ip_addr
             user_id = req.rpc_user.user_id
         else:
             raise Exception(
                 'Unable to fetch required data from request: {}. \n'
                 'This task is required to be executed from context of '
                 'request in a webapp'.format(repr(req)))
 
         if req:
             # we hook into kwargs since it is the only way to pass our data to
             # the celery worker
             options['headers'] = options.get('headers', {})
             options['headers'].update({
                 'rhodecode_proxy_data': {
                     'environ': {
                         'PATH_INFO': req.environ['PATH_INFO'],
                         'SCRIPT_NAME': req.environ['SCRIPT_NAME'],
                         'HTTP_HOST': req.environ.get('HTTP_HOST',
                             req.environ['SERVER_NAME']),
                         'SERVER_NAME': req.environ['SERVER_NAME'],
                         'SERVER_PORT': req.environ['SERVER_PORT'],
                         'wsgi.url_scheme': req.environ['wsgi.url_scheme'],
                     },
                     'auth_user': {
                         'ip_addr': ip_addr,
                         'user_id': user_id
                     },
                 }
             })
 
         return super(RequestContextTask, self).apply_async(
             args, kwargs, task_id, producer, link, link_error, shadow, **options)
 
     def __call__(self, *args, **kwargs):
         """ rebuild the context and then run task on celery worker """
 
         proxy_data = getattr(self.request, 'rhodecode_proxy_data', None)
         if not proxy_data:
             return super(RequestContextTask, self).__call__(*args, **kwargs)
 
         log.debug('using celery proxy data to run task: %r', proxy_data)
         # re-inject and register threadlocals for proper routing support
         request = prepare_request(proxy_data['environ'])
         request.user = AuthUser(user_id=proxy_data['auth_user']['user_id'],
                                 ip_addr=proxy_data['auth_user']['ip_addr'])
 
         return super(RequestContextTask, self).__call__(*args, **kwargs)
 

rhodecode/lib/celerylib/tasks.py

@@ -1,275 +1,285 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2012-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/
 
 """
 RhodeCode task modules, containing all task that suppose to be run
 by celery daemon
 """
 
 import os
+import time
 
 import rhodecode
 from rhodecode.lib import audit_logger
 from rhodecode.lib.celerylib import get_logger, async_task, RequestContextTask
 from rhodecode.lib.hooks_base import log_create_repository
 from rhodecode.lib.rcmail.smtp_mailer import SmtpMailer
 from rhodecode.lib.utils2 import safe_int, str2bool
 from rhodecode.model.db import Session, Repository, User
 
 
 @async_task(ignore_result=True, base=RequestContextTask)
 def send_email(recipients, subject, body='', html_body='', email_config=None):
     """
     Sends an email with defined parameters from the .ini files.
 
     :param recipients: list of recipients, it this is empty the defined email
         address from field 'email_to' is used instead
     :param subject: subject of the mail
     :param body: body of the mail
     :param html_body: html version of body
     """
     log = get_logger(send_email)
 
     email_config = email_config or rhodecode.CONFIG
     subject = "%s %s" % (email_config.get('email_prefix', ''), subject)
     if not recipients:
         # if recipients are not defined we send to email_config + all admins
         admins = [
             u.email for u in User.query().filter(User.admin == True).all()]
         recipients = [email_config.get('email_to')] + admins
 
     mail_server = email_config.get('smtp_server') or None
     if mail_server is None:
         log.error("SMTP server information missing. Sending email failed. "
                   "Make sure that `smtp_server` variable is configured "
                   "inside the .ini file")
         return False
 
     mail_from = email_config.get('app_email_from', 'RhodeCode')
     user = email_config.get('smtp_username')
     passwd = email_config.get('smtp_password')
     mail_port = email_config.get('smtp_port')
     tls = str2bool(email_config.get('smtp_use_tls'))
     ssl = str2bool(email_config.get('smtp_use_ssl'))
     debug = str2bool(email_config.get('debug'))
     smtp_auth = email_config.get('smtp_auth')
 
     try:
         m = SmtpMailer(mail_from, user, passwd, mail_server, smtp_auth,
                        mail_port, ssl, tls, debug=debug)
         m.send(recipients, subject, body, html_body)
     except Exception:
         log.exception('Mail sending failed')
         return False
     return True
 
 
 @async_task(ignore_result=True, base=RequestContextTask)
 def create_repo(form_data, cur_user):
     from rhodecode.model.repo import RepoModel
     from rhodecode.model.user import UserModel
     from rhodecode.model.settings import SettingsModel
 
     log = get_logger(create_repo)
 
     cur_user = UserModel()._get_user(cur_user)
     owner = cur_user
 
     repo_name = form_data['repo_name']
     repo_name_full = form_data['repo_name_full']
     repo_type = form_data['repo_type']
     description = form_data['repo_description']
     private = form_data['repo_private']
     clone_uri = form_data.get('clone_uri')
     repo_group = safe_int(form_data['repo_group'])
     landing_rev = form_data['repo_landing_rev']
     copy_fork_permissions = form_data.get('copy_permissions')
     copy_group_permissions = form_data.get('repo_copy_permissions')
     fork_of = form_data.get('fork_parent_id')
     state = form_data.get('repo_state', Repository.STATE_PENDING)
 
     # repo creation defaults, private and repo_type are filled in form
     defs = SettingsModel().get_default_repo_settings(strip_prefix=True)
     enable_statistics = form_data.get(
         'enable_statistics', defs.get('repo_enable_statistics'))
     enable_locking = form_data.get(
         'enable_locking', defs.get('repo_enable_locking'))
     enable_downloads = form_data.get(
         'enable_downloads', defs.get('repo_enable_downloads'))
 
     try:
         repo = RepoModel()._create_repo(
             repo_name=repo_name_full,
             repo_type=repo_type,
             description=description,
             owner=owner,
             private=private,
             clone_uri=clone_uri,
             repo_group=repo_group,
             landing_rev=landing_rev,
             fork_of=fork_of,
             copy_fork_permissions=copy_fork_permissions,
             copy_group_permissions=copy_group_permissions,
             enable_statistics=enable_statistics,
             enable_locking=enable_locking,
             enable_downloads=enable_downloads,
             state=state
         )
         Session().commit()
 
         # now create this repo on Filesystem
         RepoModel()._create_filesystem_repo(
             repo_name=repo_name,
             repo_type=repo_type,
             repo_group=RepoModel()._get_repo_group(repo_group),
             clone_uri=clone_uri,
         )
         repo = Repository.get_by_repo_name(repo_name_full)
         log_create_repository(created_by=owner.username, **repo.get_dict())
 
         # update repo commit caches initially
         repo.update_commit_cache()
 
         # set new created state
         repo.set_state(Repository.STATE_CREATED)
         repo_id = repo.repo_id
         repo_data = repo.get_api_data()
 
         audit_logger.store(
             'repo.create', action_data={'data': repo_data},
             user=cur_user,
             repo=audit_logger.RepoWrap(repo_name=repo_name, repo_id=repo_id))
 
         Session().commit()
     except Exception:
         log.warning('Exception occurred when creating repository, '
                     'doing cleanup...', exc_info=True)
         # rollback things manually !
         repo = Repository.get_by_repo_name(repo_name_full)
         if repo:
             Repository.delete(repo.repo_id)
             Session().commit()
             RepoModel()._delete_filesystem_repo(repo)
         raise
 
     # it's an odd fix to make celery fail task when exception occurs
     def on_failure(self, *args, **kwargs):
         pass
 
     return True
 
 
 @async_task(ignore_result=True, base=RequestContextTask)
 def create_repo_fork(form_data, cur_user):
     """
     Creates a fork of repository using internal VCS methods
     """
     from rhodecode.model.repo import RepoModel
     from rhodecode.model.user import UserModel
 
     log = get_logger(create_repo_fork)
 
     cur_user = UserModel()._get_user(cur_user)
     owner = cur_user
 
     repo_name = form_data['repo_name']  # fork in this case
     repo_name_full = form_data['repo_name_full']
     repo_type = form_data['repo_type']
     description = form_data['description']
     private = form_data['private']
     clone_uri = form_data.get('clone_uri')
     repo_group = safe_int(form_data['repo_group'])
     landing_rev = form_data['landing_rev']
     copy_fork_permissions = form_data.get('copy_permissions')
     fork_id = safe_int(form_data.get('fork_parent_id'))
 
     try:
         fork_of = RepoModel()._get_repo(fork_id)
         RepoModel()._create_repo(
             repo_name=repo_name_full,
             repo_type=repo_type,
             description=description,
             owner=owner,
             private=private,
             clone_uri=clone_uri,
             repo_group=repo_group,
             landing_rev=landing_rev,
             fork_of=fork_of,
             copy_fork_permissions=copy_fork_permissions
         )
 
         Session().commit()
 
         base_path = Repository.base_path()
         source_repo_path = os.path.join(base_path, fork_of.repo_name)
 
         # now create this repo on Filesystem
         RepoModel()._create_filesystem_repo(
             repo_name=repo_name,
             repo_type=repo_type,
             repo_group=RepoModel()._get_repo_group(repo_group),
             clone_uri=source_repo_path,
         )
         repo = Repository.get_by_repo_name(repo_name_full)
         log_create_repository(created_by=owner.username, **repo.get_dict())
 
         # update repo commit caches initially
         config = repo._config
         config.set('extensions', 'largefiles', '')
         repo.update_commit_cache(config=config)
 
         # set new created state
         repo.set_state(Repository.STATE_CREATED)
 
         repo_id = repo.repo_id
         repo_data = repo.get_api_data()
         audit_logger.store(
             'repo.fork', action_data={'data': repo_data},
             user=cur_user,
             repo=audit_logger.RepoWrap(repo_name=repo_name, repo_id=repo_id))
 
         Session().commit()
     except Exception as e:
         log.warning('Exception %s occurred when forking repository, '
                     'doing cleanup...', e)
         # rollback things manually !
         repo = Repository.get_by_repo_name(repo_name_full)
         if repo:
             Repository.delete(repo.repo_id)
             Session().commit()
             RepoModel()._delete_filesystem_repo(repo)
         raise
 
     # it's an odd fix to make celery fail task when exception occurs
     def on_failure(self, *args, **kwargs):
         pass
 
     return True
 
 
 @async_task(ignore_result=True)
 def sync_repo(*args, **kwargs):
     from rhodecode.model.scm import ScmModel
     log = get_logger(sync_repo)
-
-    log.info('Pulling from %s', kwargs['repo_name'])
-    ScmModel().pull_changes(kwargs['repo_name'], kwargs['username'])
+    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

@@ -1,156 +1,169 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 os
 import json
 import logging
 import datetime
 
 from functools import partial
 
 from pyramid.compat import configparser
 from celery.result import AsyncResult
 import celery.loaders.base
 import celery.schedules
 
 
 log = logging.getLogger(__name__)
 
 
 def get_task_id(task):
     task_id = None
     if isinstance(task, AsyncResult):
         task_id = task.task_id
 
     return task_id
 
 
 def crontab(value):
     return celery.schedules.crontab(**value)
 
 
 def timedelta(value):
     return datetime.timedelta(**value)
 
 
 def safe_json(get, section, key):
     value = ''
     try:
         value = get(key)
         json_value = json.loads(value)
     except ValueError:
         msg = 'The %s=%s is not valid json in section %s' % (
             key, value, section
         )
         raise ValueError(msg)
 
     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')
 
     if has_option('kwargs'):
         config['kwargs'] = safe_json(get, section, 'kwargs')
 
+    if has_option('force_update'):
+        config['force_update'] = get('force_update')
+
     return config
 
 
 def get_ini_config(ini_location):
     """
     Converts basic ini configuration into celery 4.X options
     """
     def key_converter(key_name):
         pref = 'celery.'
         if key_name.startswith(pref):
             return key_name[len(pref):].replace('.', '_').lower()
 
     def type_converter(parsed_key, value):
         # cast to int
         if value.isdigit():
             return int(value)
 
         # cast to bool
         if value.lower() in ['true', 'false', 'True', 'False']:
             return value.lower() == 'true'
         return value
 
     parser = configparser.SafeConfigParser(
         defaults={'here': os.path.abspath(ini_location)})
     parser.read(ini_location)
 
     ini_config = {}
     for k, v in parser.items('app:main'):
         pref = 'celery.'
         if k.startswith(pref):
             ini_config[key_converter(k)] = type_converter(key_converter(k), v)
 
     beat_config = {}
     for section in parser.sections():
         if section.startswith('celerybeat:'):
             name = section.split(':', 1)[1]
             beat_config[name] = get_beat_config(parser, section)
 
     # final compose of settings
     celery_settings = {}
 
     if ini_config:
         celery_settings.update(ini_config)
     if beat_config:
         celery_settings.update({'beat_schedule': beat_config})
 
     return celery_settings
 
 
 def parse_ini_vars(ini_vars):
     options = {}
     for pairs in ini_vars.split(','):
         key, value = pairs.split('=')
         options[key] = value
     return options

rhodecode/templates/forms/textinput.pt

@@ -1,23 +1,26 b''
 <span tal:define="name name|field.name;
                   css_class css_class|field.widget.css_class;
                   oid oid|field.oid;
                   mask mask|field.widget.mask;
                   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}"
            tal:attributes="class string: form-control ${css_class or ''};
                            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}',
          function (oid) {
             $("#" + oid).mask("${mask}",
                  {placeholder:"${mask_placeholder}"});
          });
     </script>
 </span>
  No newline at end of file