Merge branch 'feature-warm_shutdown_celery' into 'master'

Feature warm shutdown celery

See merge request !234

circle/manager/mancelery.py

@@ -16,12 +16,15 @@
 # with CIRCLE.  If not, see <http://www.gnu.org/licenses/>.
 
 from celery import Celery
+from celery.signals import worker_ready
 from datetime import timedelta
 from kombu import Queue, Exchange
 from os import getenv
 
 HOSTNAME = "localhost"
 CACHE_URI = getenv("CACHE_URI", "pylibmc://127.0.0.1:11211/")
+QUEUE_NAME = HOSTNAME + '.man'
+
 
 celery = Celery('manager',
                 broker=getenv("AMQP_URI"),
@@ -57,3 +60,10 @@ celery.conf.update(
     }
 
 )
+
+
+@worker_ready.connect()
+def cleanup_tasks(conf=None, **kwargs):
+    '''Discard all task and clean up activity.'''
+    from vm.models.activity import cleanup
+    cleanup(queue_name=QUEUE_NAME)

circle/manager/moncelery.py

@@ -16,12 +16,14 @@
 # with CIRCLE.  If not, see <http://www.gnu.org/licenses/>.
 
 from celery import Celery
+from celery.signals import worker_ready
 from datetime import timedelta
 from kombu import Queue, Exchange
 from os import getenv
 
 HOSTNAME = "localhost"
 CACHE_URI = getenv("CACHE_URI", "pylibmc://127.0.0.1:11211/")
+QUEUE_NAME = HOSTNAME + '.monitor'
 
 celery = Celery('monitor',
                 broker=getenv("AMQP_URI"),
@@ -34,7 +36,7 @@ celery.conf.update(
     CELERY_CACHE_BACKEND=CACHE_URI,
     CELERY_TASK_RESULT_EXPIRES=300,
     CELERY_QUEUES=(
-        Queue(HOSTNAME + '.monitor', Exchange('monitor', type='direct'),
+        Queue(QUEUE_NAME, Exchange('monitor', type='direct'),
               routing_key="monitor"),
     ),
     CELERYBEAT_SCHEDULE={
@@ -70,3 +72,10 @@ celery.conf.update(
     }
 
 )
+
+
+@worker_ready.connect()
+def cleanup_tasks(conf=None, **kwargs):
+    '''Discard all task and clean up activity.'''
+    from vm.models.activity import cleanup
+    cleanup(queue_name=QUEUE_NAME)

circle/manager/slowcelery.py

@@ -16,12 +16,14 @@
 # with CIRCLE.  If not, see <http://www.gnu.org/licenses/>.
 
 from celery import Celery
+from celery.signals import worker_ready
 from datetime import timedelta
 from kombu import Queue, Exchange
 from os import getenv
 
 HOSTNAME = "localhost"
 CACHE_URI = getenv("CACHE_URI", "pylibmc://127.0.0.1:11211/")
+QUEUE_NAME = HOSTNAME + '.man.slow'
 
 celery = Celery('manager.slow',
                 broker=getenv("AMQP_URI"),
@@ -36,7 +38,7 @@ celery.conf.update(
     CELERY_CACHE_BACKEND=CACHE_URI,
     CELERY_TASK_RESULT_EXPIRES=300,
     CELERY_QUEUES=(
-        Queue(HOSTNAME + '.man.slow', Exchange('manager.slow', type='direct'),
+        Queue(QUEUE_NAME, Exchange('manager.slow', type='direct'),
               routing_key="manager.slow"),
     ),
     CELERYBEAT_SCHEDULE={
@@ -48,3 +50,10 @@ celery.conf.update(
     }
 
 )
+
+
+@worker_ready.connect()
+def cleanup_tasks(conf=None, **kwargs):
+    '''Discard all task and clean up activity.'''
+    from vm.models.activity import cleanup
+    cleanup(queue_name=QUEUE_NAME)

circle/vm/models/activity.py

@@ -20,7 +20,6 @@ from contextlib import contextmanager
 from logging import getLogger
 from warnings import warn
 
-from celery.signals import worker_ready
 from celery.contrib.abortable import AbortableAsyncResult
 
 from django.core.urlresolvers import reverse
@@ -278,17 +277,17 @@ def node_activity(code_suffix, node, task_uuid=None, user=None,
     return activitycontextimpl(act)
 
 
-@worker_ready.connect()
 def cleanup(conf=None, **kwargs):
     # TODO check if other manager workers are running
-    from celery.task.control import discard_all
-    discard_all()
     msg_txt = ugettext_noop("Manager is restarted, activity is cleaned up. "
                             "You can try again now.")
     message = create_readable(msg_txt, msg_txt)
+    queue_name = kwargs.get('queue_name', None)
     for i in InstanceActivity.objects.filter(finished__isnull=True):
-        i.finish(False, result=message)
-        logger.error('Forced finishing stale activity %s', i)
+        op = i.get_operation()
+        if op and op.async_queue == queue_name:
+            i.finish(False, result=message)
+            logger.error('Forced finishing stale activity %s', i)
     for i in NodeActivity.objects.filter(finished__isnull=True):
         i.finish(False, result=message)
         logger.error('Forced finishing stale activity %s', i)

miscellaneous/mancelery.conf

@@ -6,9 +6,14 @@ respawn limit 30 30
 setgid cloud
 setuid cloud
 
+kill timeout 360
+kill signal SIGTERM
+
 script
     cd /home/cloud/circle/circle
     . /home/cloud/.virtualenvs/circle/bin/activate
     . /home/cloud/.virtualenvs/circle/bin/postactivate
-    exec ./manage.py celery --app=manager.mancelery worker --autoreload --loglevel=info --hostname=mancelery -B -c 10
+    ./manage.py celery -f --app=manager.mancelery purge
+    exec ./manage.py celery --app=manager.mancelery worker --autoreload --loglevel=info --hostname=mancelery -B -c 3
 end script
+

miscellaneous/moncelery.conf

@@ -3,6 +3,7 @@ description     "CIRCLE moncelery for monitoring jobs"
 respawn
 respawn limit 30 30
 
+
 setgid cloud
 setuid cloud
 
@@ -10,5 +11,7 @@ script
     cd /home/cloud/circle/circle
     . /home/cloud/.virtualenvs/circle/bin/activate
     . /home/cloud/.virtualenvs/circle/bin/postactivate
-    exec ./manage.py celery --app=manager.moncelery worker --autoreload --loglevel=info --hostname=moncelery -B -c 3
+    ./manage.py celery -f --app=manager.moncelery purge
+    exec ./manage.py celery --app=manager.moncelery worker --autoreload --loglevel=info --hostname=moncelery -B -c 2
 end script
+

miscellaneous/slowcelery.conf

-description     "CIRCLE mancelery for slow jobs"
+description     "CIRCLE slowcelery for resource intensive or long  jobs"
 
 respawn
 respawn limit 30 30
@@ -6,9 +6,15 @@ respawn limit 30 30
 setgid cloud
 setuid cloud
 
+kill timeout 360                                                                                      
+kill signal INT 
+                                          
+
 script
     cd /home/cloud/circle/circle
     . /home/cloud/.virtualenvs/circle/bin/activate
     . /home/cloud/.virtualenvs/circle/bin/postactivate
-    exec ./manage.py celery --app=manager.slowcelery worker --autoreload --loglevel=info --hostname=slowcelery -B -c 5
+    ./manage.py celery -f --app=manager.slowcelery purge
+    exec ./manage.py celery --app=manager.slowcelery worker --autoreload --loglevel=info --hostname=slowcelery -B -c 1
 end script
+