Celery update

bk-plugin-framework/bk_plugin_framework/runtime/callback/celery/tasks.py

@@ -14,7 +14,7 @@
 import random
 import time
 
-from celery import task, current_app
+from celery import shared_task, current_app
 
 from bk_plugin_framework.kit import State
 from bk_plugin_framework.metrics import BK_PLUGIN_CALLBACK_EXCEPTION_COUNT, HOSTNAME, BK_PLUGIN_CALLBACK_TIME
@@ -35,7 +35,7 @@ def _set_schedule_state(trace_id: str, state: State):
         logger.exception("[execute] set schedule state error")
 
 
-@task(ignore_result=True)
+@shared_task(ignore_result=True)
 def callback(trace_id: str, callback_id: str, callback_data: str):
     with get_schedule_lock(trace_id) as lock:
         if not lock.locked:

bk-plugin-framework/bk_plugin_framework/runtime/schedule/celery/tasks.py

@@ -12,7 +12,7 @@
 
 import logging
 
-from celery import task
+from celery import shared_task
 
 from bk_plugin_framework.kit import State
 from bk_plugin_framework.utils import local
@@ -31,7 +31,7 @@ def _set_schedule_state(trace_id: str, state: State):
         logger.exception("[execute] set schedule state error")
 
 
-@task(ignore_result=True)
+@shared_task(ignore_result=True)
 def schedule(trace_id: str):
     local.set_trace_id(trace_id)
 
@@ -56,7 +56,7 @@ def schedule(trace_id: str):
         _set_schedule_state(trace_id=trace_id, state=State.FAIL)
 
 
-@task(ignore_result=True)
+@shared_task(ignore_result=True)
 def delete_expired_schedule():
     logger.info("[delete_expired_schedule] start to delete expire schedule")
     rows = Schedule.objects.delete_expired_schedule(settings.SCHEDULE_PERSISTENT_DAYS)