feat: pipeline dependencies version update (#227)

* feat: pipeline dependencies version update

* minor: flake8 fix

* minor: unit test fix

* Update pr_check.yml

* minor: git ci fix

* minor: git ci fix

---------

Co-authored-by: NULL <[email protected]>

.github/workflows/pr_check.yml

@@ -6,23 +6,23 @@ on:
 
 jobs:
   engine-lint:
-    uses: TencentBlueKing/bamboo-engine/.github/workflows/engine_flake8_and_black.yml@master
+    uses: TencentBlueKing/bamboo-engine/.github/workflows/engine_flake8_and_black.yml@bamboo_pipeline_3.24_lts
 
   pipeline-lint:
-    uses: TencentBlueKing/bamboo-engine/.github/workflows/runtime_pipeline_flake8_and_black.yml@master
+    uses: TencentBlueKing/bamboo-engine/.github/workflows/runtime_pipeline_flake8_and_black.yml@bamboo_pipeline_3.24_lts
 
   engine-unittest:
     needs: engine-lint
-    uses: TencentBlueKing/bamboo-engine/.github/workflows/engine_unittest.yml@master
+    uses: TencentBlueKing/bamboo-engine/.github/workflows/engine_unittest.yml@bamboo_pipeline_3.24_lts
 
   pipeline-unittest:
     needs: pipeline-lint
-    uses: TencentBlueKing/bamboo-engine/.github/workflows/runtime_pipeline_unittest.yml@master
+    uses: TencentBlueKing/bamboo-engine/.github/workflows/runtime_pipeline_unittest.yml@bamboo_pipeline_3.24_lts
 
   runtime-pipeline-intergration-test:
     needs: [engine-unittest, pipeline-unittest]
-    uses: TencentBlueKing/bamboo-engine/.github/workflows/runtime_pipeline_end_to_end_test.yml@master
+    uses: TencentBlueKing/bamboo-engine/.github/workflows/runtime_pipeline_end_to_end_test.yml@bamboo_pipeline_3.24_lts
 
   pipeline-intergration-test:
     needs: pipeline-unittest
-    uses: TencentBlueKing/bamboo-engine/.github/workflows/pipeline_end_to_end_test.yml@master
+    uses: TencentBlueKing/bamboo-engine/.github/workflows/pipeline_end_to_end_test.yml@bamboo_pipeline_3.24_lts

runtime/bamboo-pipeline/pipeline/apps.py

@@ -19,7 +19,10 @@
 from django.apps import AppConfig
 from django.conf import settings
 from redis.sentinel import Sentinel
-from rediscluster import RedisCluster
+try:
+    from redis.cluster import RedisCluster
+except ImportError:
+    from rediscluster import RedisCluster
 
 logger = logging.getLogger("root")
 

runtime/bamboo-pipeline/pipeline/contrib/celery_tools/__init__.py

@@ -0,0 +1,12 @@
+# -*- coding: utf-8 -*-
+"""
+Tencent is pleased to support the open source community by making 蓝鲸智云PaaS平台社区版 (BlueKing PaaS Community
+Edition) available.
+Copyright (C) 2017 THL A29 Limited, a Tencent company. All rights reserved.
+Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+http://opensource.org/licenses/MIT
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+specific language governing permissions and limitations under the License.
+"""

runtime/bamboo-pipeline/pipeline/contrib/celery_tools/periodic.py

@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+"""
+Tencent is pleased to support the open source community by making 蓝鲸智云PaaS平台社区版 (BlueKing PaaS Community
+Edition) available.
+Copyright (C) 2017 THL A29 Limited, a Tencent company. All rights reserved.
+Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+http://opensource.org/licenses/MIT
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+specific language governing permissions and limitations under the License.
+"""
+from celery import Task, current_app
+from celery.schedules import maybe_schedule
+
+
+class PipelinePeriodicTask(Task):
+    """A task that adds itself to the :setting:`beat_schedule` setting."""
+
+    abstract = True
+    ignore_result = True
+    relative = False
+    options = None
+    compat = True
+
+    def __init__(self):
+        if not hasattr(self, 'run_every'):
+            raise NotImplementedError(
+                'Periodic tasks must have a run_every attribute')
+        self.run_every = maybe_schedule(self.run_every, self.relative)
+        super(PipelinePeriodicTask, self).__init__()
+
+    @classmethod
+    def on_bound(cls, app):
+        app.conf.beat_schedule[cls.name] = {
+            'task': cls.name,
+            'schedule': cls.run_every,
+            'args': (),
+            'kwargs': {},
+            'options': cls.options or {},
+            'relative': cls.relative,
+        }
+
+
+def periodic_task(*args, **options):
+    """Deprecated decorator, please use :setting:`beat_schedule`."""
+    return current_app.task(**dict({'base': PipelinePeriodicTask}, **options))

runtime/bamboo-pipeline/pipeline/contrib/periodic_task/tasks.py

@@ -16,7 +16,7 @@
 import traceback
 
 import pytz
-from celery import task
+from celery import current_app
 from django.utils import timezone
 from django.utils.module_loading import import_string
 from bamboo_engine import api as bamboo_engine_api
@@ -35,7 +35,7 @@
 logger = logging.getLogger("celery")
 
 
-@task(ignore_result=True)
+@current_app.task(ignore_result=True)
 def periodic_task_start(*args, **kwargs):
     try:
         periodic_task = PeriodicTask.objects.get(id=kwargs["period_task_id"])
@@ -69,19 +69,28 @@ def periodic_task_start(*args, **kwargs):
         )
 
         result = instance.start(
-            periodic_task.creator, check_workers=False, priority=periodic_task.priority, queue=periodic_task.queue,
+            periodic_task.creator,
+            check_workers=False,
+            priority=periodic_task.priority,
+            queue=periodic_task.queue,
         )
     except Exception:
         et = traceback.format_exc()
         logger.error(et)
         PeriodicTaskHistory.objects.record_schedule(
-            periodic_task=periodic_task, pipeline_instance=None, ex_data=et, start_success=False,
+            periodic_task=periodic_task,
+            pipeline_instance=None,
+            ex_data=et,
+            start_success=False,
         )
         return
 
     if not result.result:
         PeriodicTaskHistory.objects.record_schedule(
-            periodic_task=periodic_task, pipeline_instance=None, ex_data=result.message, start_success=False,
+            periodic_task=periodic_task,
+            pipeline_instance=None,
+            ex_data=result.message,
+            start_success=False,
         )
         return
 
@@ -93,7 +102,7 @@ def periodic_task_start(*args, **kwargs):
     PeriodicTaskHistory.objects.record_schedule(periodic_task=periodic_task, pipeline_instance=instance, ex_data="")
 
 
-@task(ignore_result=True)
+@current_app.task(ignore_result=True)
 def bamboo_engine_periodic_task_start(*args, **kwargs):
     try:
         periodic_task = PeriodicTask.objects.get(id=kwargs["period_task_id"])
@@ -147,16 +156,24 @@ def bamboo_engine_periodic_task_start(*args, **kwargs):
         et = traceback.format_exc()
         logger.error(et)
         PeriodicTaskHistory.objects.record_schedule(
-            periodic_task=periodic_task, pipeline_instance=None, ex_data=et, start_success=False,
+            periodic_task=periodic_task,
+            pipeline_instance=None,
+            ex_data=et,
+            start_success=False,
         )
         return
 
     if not result.result:
         PipelineInstance.objects.filter(id=instance.instance_id).update(
-            start_time=None, is_started=False, executor="",
+            start_time=None,
+            is_started=False,
+            executor="",
         )
         PeriodicTaskHistory.objects.record_schedule(
-            periodic_task=periodic_task, pipeline_instance=None, ex_data=result.message, start_success=False,
+            periodic_task=periodic_task,
+            pipeline_instance=None,
+            ex_data=result.message,
+            start_success=False,
         )
         return
 

runtime/bamboo-pipeline/pipeline/contrib/statistics/tasks.py

@@ -16,7 +16,7 @@
 import ujson as json
 from copy import deepcopy
 
-from celery import task
+from celery import current_app
 from bamboo_engine import api as bamboo_engine_api
 
 from pipeline.component_framework.constants import LEGACY_PLUGINS_VERSION
@@ -112,7 +112,7 @@ def recursive_collect_components(activities, status_tree, instance_id, stack=Non
     return component_list
 
 
-@task
+@current_app.task
 def pipeline_post_save_statistics_task(instance_id):
     instance = PipelineInstance.objects.get(instance_id=instance_id)
     # 统计流程标准插件个数，子流程个数，网关个数
@@ -134,7 +134,7 @@ def pipeline_post_save_statistics_task(instance_id):
         )
 
 
-@task
+@current_app.task
 def pipeline_archive_statistics_task(instance_id):
     instance = PipelineInstance.objects.get(instance_id=instance_id)
     engine_ver = 1