celery: improve fetching of repo_check tasks.

- add refresh link to fail message
- introduce proper timeouts so we don't block forever.

rhodecode/apps/repository/views/repo_checks.py

@@ -1,109 +1,113 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2011-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
 
 from pyramid.view import view_config
 from pyramid.httpexceptions import HTTPFound, HTTPNotFound
 
 from rhodecode.apps._base import BaseAppView
 from rhodecode.lib import helpers as h
 from rhodecode.lib.auth import (NotAnonymous, HasRepoPermissionAny)
 from rhodecode.model.db import Repository
 
 log = logging.getLogger(__name__)
 
 
 class RepoChecksView(BaseAppView):
     def load_default_context(self):
         c = self._get_local_tmpl_context()
 
         return c
 
     @NotAnonymous()
     @view_config(
         route_name='repo_creating', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_creating.mako')
     def repo_creating(self):
         c = self.load_default_context()
 
         repo_name = self.request.matchdict['repo_name']
         db_repo = Repository.get_by_repo_name(repo_name)
 
         # check if maybe repo is already created
         if db_repo and db_repo.repo_state in [Repository.STATE_CREATED]:
             # re-check permissions before redirecting to prevent resource
             # discovery by checking the 302 code
             perm_set = ['repository.read', 'repository.write', 'repository.admin']
             has_perm = HasRepoPermissionAny(*perm_set)(
                 db_repo.repo_name, 'Repo Creating check')
             if not has_perm:
                 raise HTTPNotFound()
 
             raise HTTPFound(h.route_path(
                 'repo_summary', repo_name=db_repo.repo_name))
 
         c.task_id = self.request.GET.get('task_id')
         c.repo_name = repo_name
 
         return self._get_template_context(c)
 
     @NotAnonymous()
     @view_config(
         route_name='repo_creating_check', request_method='GET',
         renderer='json_ext')
     def repo_creating_check(self):
         _ = self.request.translate
         task_id = self.request.GET.get('task_id')
         self.load_default_context()
 
         repo_name = self.request.matchdict['repo_name']
 
         if task_id and task_id not in ['None']:
             import rhodecode
-            from rhodecode.lib.celerylib.loader import celery_app
+            from rhodecode.lib.celerylib.loader import celery_app, exceptions
             if rhodecode.CELERY_ENABLED:
+                log.debug('celery: checking result for task:%s', task_id)
                 task = celery_app.AsyncResult(task_id)
-                task.get()
-                if task.failed():
+                try:
+                    task.get(timeout=10)
+                except exceptions.TimeoutError:
+                    task = None
+                if task and task.failed():
                     msg = self._log_creation_exception(task.result, repo_name)
                     h.flash(msg, category='error')
                     raise HTTPFound(h.route_path('home'), code=501)
 
         db_repo = Repository.get_by_repo_name(repo_name)
         if db_repo and db_repo.repo_state == Repository.STATE_CREATED:
             if db_repo.clone_uri:
                 clone_uri = db_repo.clone_uri_hidden
                 h.flash(_('Created repository %s from %s')
                         % (db_repo.repo_name, clone_uri), category='success')
             else:
                 repo_url = h.link_to(
                     db_repo.repo_name,
                     h.route_path('repo_summary', repo_name=db_repo.repo_name))
                 fork = db_repo.fork
                 if fork:
                     fork_name = fork.repo_name
                     h.flash(h.literal(_('Forked repository %s as %s')
                                       % (fork_name, repo_url)), category='success')
                 else:
                     h.flash(h.literal(_('Created repository %s') % repo_url),
                             category='success')
             return {'result': True}
         return {'result': False}

rhodecode/lib/celerylib/loader.py

@@ -1,256 +1,257 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
 """
 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,
 }
 # 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/templates/admin/repos/repo_creating.mako

@@ -1,77 +1,79 b''
 ## -*- coding: utf-8 -*-
 <%inherit file="/base/base.mako"/>
 
 <%def name="title()">
     ${_('{} Creating repository').format(c.repo_name)}
     %if c.rhodecode_name:
         &middot; ${h.branding(c.rhodecode_name)}
     %endif
 </%def>
 
 <%def name="breadcrumbs_links()">
     ${_('Creating repository')} ${c.repo_name}
 </%def>
 
 <%def name="menu_bar_nav()">
     ${self.menu_items(active='repositories')}
 </%def>
 <%def name="main()">
 <div class="box">
     <!-- box / title -->
     <div class="title">
         ${self.breadcrumbs()}
     </div>
 
     <div id="progress-message">
         ${_('Repository "%(repo_name)s" is being created, you will be redirected when this process is finished.' % {'repo_name':c.repo_name})}
     </div>
 
     <div id="progress">
         <div class="progress progress-striped active">
           <div class="progress-bar progress-bar" role="progressbar"
                aria-valuenow="100" aria-valuemin="0" aria-valuemax="100">
           </div>
         </div>
     </div>
 </div>
 
 <script>
 (function worker() {
   var skipCheck = false;
   var url = "${h.route_path('repo_creating_check', repo_name=c.repo_name, _query=dict(task_id=c.task_id))}";
   $.ajax({
     url: url,
+    timeout: 60*1000, // sets timeout to 60 seconds
     complete: function(resp) {
-        if (resp.status == 200) {
+        if (resp.status === 200) {
             var jsonResponse = resp.responseJSON;
 
             if (jsonResponse === undefined) {
                 setTimeout(function () {
                     // we might have a backend problem, try dashboard again
                     window.location = "${h.route_path('repo_summary', repo_name = c.repo_name)}";
                 }, 3000);
             } else {
                 if (skipCheck || jsonResponse.result === true) {
                     // success, means go to dashboard
                     window.location = "${h.route_path('repo_summary', repo_name = c.repo_name)}";
                 } else {
                     // Schedule the next request when the current one's complete
                     setTimeout(worker, 1000);
                 }
             }
         }
         else {
+            var message = _gettext('Fetching repository state failed. Error code: {0} {1}. Try <a href="{2}">refreshing</a> this page.').format(resp.status, resp.statusText, url);
             var payload = {
                 message: {
-                    message: _gettext('Fetching repository state failed. Error code: {0} {1}. Try refreshing this page.').format(resp.status, resp.statusText),
+                    message: message,
                     level: 'error',
                     force: true
                 }
             };
             $.Topic('/notifications').publish(payload);
         }
     }
   });
 })();
 </script>
 </%def>