celery: ping db connection before task execution to recycle db connections.

rhodecode/lib/celerylib/loader.py

@@ -40,7 +40,7 @@ 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.celerylib.utils import get_ini_config, parse_ini_vars, ping_db
 from rhodecode.lib.ext_json import json
 from rhodecode.lib.pyramid_utils import bootstrap, setup_logging, prepare_request
 from rhodecode.lib.utils2 import str2bool
@@ -144,6 +144,11 @@ def on_preload_parsed(options, **kwargs):
     rhodecode.CELERY_ENABLED = True
 
 
+@signals.task_prerun.connect
+def task_prerun_signal(task_id, task, args, **kwargs):
+    ping_db()
+
+
 @signals.task_success.connect
 def task_success_signal(result, **kwargs):
     meta.Session.commit()
@@ -227,8 +232,7 @@ def maybe_prepare_env(req):
         environ.update({
             'PATH_INFO': req.environ['PATH_INFO'],
             'SCRIPT_NAME': req.environ['SCRIPT_NAME'],
-            'HTTP_HOST':
-                req.environ.get('HTTP_HOST', req.environ['SERVER_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'],

rhodecode/lib/celerylib/utils.py

@@ -22,6 +22,7 @@ import os
 import json
 import logging
 import datetime
+import time
 
 from functools import partial
 
@@ -30,7 +31,6 @@ from celery.result import AsyncResult
 import celery.loaders.base
 import celery.schedules
 
-
 log = logging.getLogger(__name__)
 
 
@@ -167,3 +167,21 @@ def parse_ini_vars(ini_vars):
         key, value = pairs.split('=')
         options[key] = value
     return options
+
+
+def ping_db():
+    from rhodecode.model import meta
+    from rhodecode.model.db import DbMigrateVersion
+    log.info('Testing DB connection...')
+
+    for test in range(10):
+        try:
+            scalar = DbMigrateVersion.query().scalar()
+            log.debug('DB PING %s@%s', scalar, scalar.version)
+            break
+        except Exception:
+            retry = 1
+            log.debug('DB not ready, next try in %ss', retry)
+            time.sleep(retry)
+        finally:
+            meta.Session.remove()