chore(celery): use quorum queye

Author: duanhongyi

charts/controller/templates/_helpers.tpl

@@ -170,7 +170,7 @@ env:
       name: rabbitmq-creds
       key: password
 - name: "DRYCC_RABBITMQ_URL"
-  value: "amqp://$(DRYCC_RABBITMQ_USERNAME):$(DRYCC_RABBITMQ_PASSWORD)@drycc-rabbitmq.{{$.Release.Namespace}}.svc.{{$.Values.global.clusterDomain}}:5672/drycc"
+  value: "amqp://$(DRYCC_RABBITMQ_USERNAME):$(DRYCC_RABBITMQ_PASSWORD)@drycc-rabbitmq.{{$.Release.Namespace}}.svc.{{$.Values.global.clusterDomain}}:5672/controller"
 {{- end }}
 {{- if eq .Values.global.passportLocation "on-cluster"}}
 - name: "DRYCC_PASSPORT_URL"

rootfs/api/settings/celery.py

@@ -33,53 +33,41 @@ class Config(object):
     task_routes={
         'api.tasks.scale_app': {
             'queue': 'controller.high',
-            'exchange': 'controller.priority',
-            'routing_key': 'controller.priority.high',
+            'exchange': 'controller.priority', 'routing_key': 'controller.priority.high',
         },
         'api.tasks.mount_app': {
             'queue': 'controller.high',
-            'exchange': 'controller.priority',
-            'routing_key': 'controller.priority.high',
+            'exchange': 'controller.priority', 'routing_key': 'controller.priority.high',
         },
         'api.tasks.restart_app': {
             'queue': 'controller.high',
-            'exchange': 'controller.priority',
-            'routing_key': 'controller.priority.high',
+            'exchange': 'controller.priority', 'routing_key': 'controller.priority.high',
         },
         'api.tasks.run_pipeline': {
             'queue': 'controller.high',
-            'exchange': 'controller.priority',
-            'routing_key': 'controller.priority.high',
+            'exchange': 'controller.priority', 'routing_key': 'controller.priority.high',
         },
         'api.tasks.downstream_model_owner': {
             'queue': 'controller.high',
-            'exchange': 'controller.priority',
-            'routing_key': 'controller.priority.high',
+            'exchange': 'controller.priority', 'routing_key': 'controller.priority.high',
         },
         'api.tasks.send_measurements': {
             'queue': 'controller.middle',
-            'exchange': 'controller.priority',
-            'routing_key': 'controller.priority.middle',
+            'exchange': 'controller.priority', 'routing_key': 'controller.priority.middle',
         },
     },
     task_queues=(
         Queue(
-            'controller.low',
-            exchange=Exchange('controller.priority', type="direct"),
-            routing_key='controller.priority.low',
-            queue_arguments={'x-max-priority': 16},
+            'controller.low', exchange=Exchange('controller.priority', type="direct"),
+            routing_key='controller.priority.low', queue_arguments={'x-queue-type': 'quorum'},
         ),
         Queue(
-            'controller.high',
-            exchange=Exchange('controller.priority', type="direct"),
-            routing_key='controller.priority.high',
-            queue_arguments={'x-max-priority': 64},
+            'controller.high', exchange=Exchange('controller.priority', type="direct"),
+            routing_key='controller.priority.high', queue_arguments={'x-queue-type': 'quorum'},
         ),
         Queue(
-            'controller.middle',
-            exchange=Exchange('controller.priority', type="direct"),
-            routing_key='controller.priority.middle',
-            queue_arguments={'x-max-priority': 32},
+            'controller.middle', exchange=Exchange('controller.priority', type="direct"),
+            routing_key='controller.priority.middle', queue_arguments={'x-queue-type': 'quorum'},
         ),
     ),
 )

rootfs/requirements.txt

@@ -22,7 +22,7 @@ ndg-httpsclient==0.5.1
 pytz==2024.1
 requests==2.32.0
 requests-toolbelt==1.0.0
-celery==5.3.6
+celery==5.5.0b3
 hiredis==2.3.2
 django_redis==5.4.0
 dj-database-url==2.1.0