testing new celery configs

Author: landon-greylabs

…anagement/commands/check_celery_tasks.py …/management/commands/terminate_worker.pyapp/api/management/commands/check_celery_tasks.py renamed to app/api/management/commands/terminate_worker.py app/api/management/commands/check_celery_tasks.py renamed to app/api/management/commands/terminate_worker.py

@@ -8,17 +8,25 @@ class Command(BaseCommand):
 
     def handle(self, *args, **options):
 
-        this_pod = f"celery@{str(os.environ.get('THIS_POD_NAME'))}"
+        # get celery worker
+        this_worker = f"celery@{str(os.environ.get('THIS_POD_NAME'))}"
+        
+        # sending initial SIGTERM to celery worker for warm-shutdown
+        celery.app.control.broadcast('shutdown', destination=[this_worker])
 
         def get_task_list():
             # Inspect all nodes.
             i = celery.app.control.inspect()
+            
             # Tasks received, but are still waiting to be executed.
-            reserved = i.reserved()[this_pod]
+            reserved = i.reserved()[this_worker]
             print(f'Reserved tasks -> {str(reserved)}')
+            
             # Active tasks
-            active = i.active()[this_pod]
+            active = i.active()[this_worker]
             print(f'Active tasks -> {str(reserved)}')
+            
+            # Sum all tasks
             tasks = len(active) + len(reserved)
             return int(tasks)
 

app/scanerr/celery.py

@@ -1,5 +1,6 @@
 from __future__ import absolute_import, unicode_literals
 from celery import Celery
+from celery.signals import worker_shutdown
 from django.conf import settings
 import scanerr, os
 
@@ -23,6 +24,7 @@
     worker_prefetch_multiplier=1,
     worker_hijack_root_logger=False,
     task_always_eager=False,
+    worker_cancel_long_running_tasks_on_connection_loss=True,
 )
 
 # setting tasks to auto-discover
@@ -31,5 +33,12 @@
 # setting debug
 @app.task(bind=False)
 def debug_task(self):
-    print('Request: {0!r}'.format(self.request))
+    print(f'Request: {self.request}')
+
+# notify of SIGTERM
+@worker_shutdown.connect
+def on_worker_shutdown(**kwargs):
+    print(f'- WORKER SHUTTING DOWN - \n\n{kwargs}')
+    
+
 

k8s/prod/celery-deployment.yaml

@@ -14,12 +14,12 @@ spec:
       labels:
         app: celery-deployment
     spec:
-      terminationGracePeriodSeconds: 30
+      terminationGracePeriodSeconds: 300
       imagePullSecrets:
         - name: regcred
       containers:
         - name: celery
-          image: <IMAGE> # scanerr/server:facff34 # 
+          image: <IMAGE> # scanerr/server:99e2252 # 
           imagePullPolicy: Always
           command: ["/entrypoint.sh", "celery"]
           envFrom:
@@ -33,11 +33,11 @@ spec:
           resources:
             limits:
               cpu: "1"
-              memory: "5Gi"
+              memory: "3.5Gi"
             requests:
               cpu: "1"
               memory: "2Gi"
           lifecycle:
             preStop:
               exec:
-                command: ["python3", "manage.py", "check_celery_tasks"]
+                command: ["python3", "manage.py", "terminate_worker"]

notes/Kubernetes.md

@@ -191,7 +191,7 @@ kubectl apply -f ./k8s/prod/kubeip-daemon.yaml
    - `for p in $(kubectl get pods | grep Terminating | awk '{print $1}'); do kubectl delete pod $p --grace-period=0 --force;done`
  - Stream Logs for all celery-deployments:
    - `kubectl logs -f --all-containers deployment/celery-deployment`
-   - `kubectl logs -f --selector=app=celery-deployment --all-containers`
+   - `kubectl logs -f --selector=app=celery-deployment --all-containers --max-log-requests=7`