feat: pipeline管理工具集成 --story=119892878

pipeline/contrib/engine_admin/__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.
+"""

pipeline/contrib/engine_admin/apps.py

@@ -0,0 +1,18 @@
+# -*- 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 django.apps import AppConfig
+
+
+class EngineAdminConfig(AppConfig):
+    name = 'pipeline.contrib.engine_admin'
+    verbose_name = 'PipelineEngineAdmin'

pipeline/contrib/engine_admin/handlers/__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.
+"""

pipeline/contrib/engine_admin/handlers/base.py

@@ -0,0 +1,30 @@
+# -*- 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 enum import Enum
+
+
+class HandlerType(Enum):
+    PIPELINE_ENGINE = "pipeline_engine"
+
+
+class ActionRequestHandler:
+    handler_type = None
+
+    def __init__(self, request, action, instance_id):
+        self.request = request
+        self.instance_id = instance_id
+        self.action = action
+
+    def execute(self, *args, **kwargs):
+        action = getattr(self, self.action, None)
+        return action(*args, **kwargs)

pipeline/contrib/engine_admin/handlers/pipeline_engine.py

@@ -0,0 +1,49 @@
+# -*- 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 pipeline.service import task_service
+from .base import ActionRequestHandler, HandlerType
+
+
+class PipelineEngineRequestHandler(ActionRequestHandler):
+    handler_type = HandlerType.PIPELINE_ENGINE
+
+    def task_pause(self):
+        return task_service.pause_pipeline(pipeline_id=self.instance_id)
+
+    def task_resume(self, **kwargs):
+        return task_service.resume_pipeline(pipeline_id=self.instance_id)
+
+    def task_revoke(self, **kwargs):
+        return task_service.revoke_pipeline(pipeline_id=self.instance_id)
+
+    def node_retry(self, **kwargs):
+        return task_service.retry_activity(act_id=self.instance_id, inputs=kwargs.get("inputs"))
+
+    def node_skip(self, **kwargs):
+        return task_service.skip_activity(act_id=self.instance_id)
+
+    def node_callback(self, **kwargs):
+        return task_service.callback(act_id=self.instance_id, data=kwargs.get("data"))
+
+    def node_skip_exg(self, **kwargs):
+        return task_service.skip_exclusive_gateway(gateway_id=self.instance_id, flow_id=kwargs["flow_id"])
+
+    def node_skip_cpg(self, **kwargs):
+        return task_service.skip_conditional_parallel_gateway(
+            gateway_id=self.instance_id, flow_ids=kwargs["flow_ids"], converge_gateway_id=kwargs["converge_gateway_id"]
+        )
+
+    def node_forced_fail(self, **kwargs):
+        return task_service.forced_fail(
+            act_id=self.instance_id, ex_data=f"force fail by {kwargs.get('operator', 'engine_admin')}"
+        )

pipeline/contrib/engine_admin/tests.py

@@ -0,0 +1,58 @@
+# -*- 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.
+"""
+import json
+
+from django.http import JsonResponse
+from django.test import TestCase
+
+from pipeline.contrib.engine_admin.views import _ensure_return_json_response, _check_api_permission
+from pipeline.engine.utils import ActionResult
+
+
+def check_permission_fail(request, *args, **kwargs):
+    return False
+
+
+def check_permission_success(request, *args, **kwargs):
+    return True
+
+
+class EngineAdminTestCase(TestCase):
+
+    def test_ensure_return_json_response(self):
+        def func_return_json_response(*args, **kwargs):
+            return JsonResponse({"result": True})
+
+        def func_return_action_result(*args, **kwargs):
+            return ActionResult(result=True, message="success")
+
+        self.assertIsInstance(_ensure_return_json_response(func_return_json_response)(), JsonResponse)
+        self.assertIsInstance(_ensure_return_json_response(func_return_action_result)(), JsonResponse)
+
+    def test_check_api_permission(self):
+        def func_return_json_response(request, *args, **kwargs):
+            return JsonResponse({"result": True})
+
+        with self.settings(
+            PIPELINE_ENGINE_ADMIN_API_PERMISSION="pipeline.contrib.engine_admin.tests.check_permission_fail"
+        ):
+            fail_response = _check_api_permission(func_return_json_response)(request=None)
+            fail_result = json.loads(fail_response.content).get("result")
+            self.assertFalse(fail_result)
+
+        with self.settings(
+            PIPELINE_ENGINE_ADMIN_API_PERMISSION="pipeline.contrib.engine_admin.tests.check_permission_success"
+        ):
+            success_response = _check_api_permission(func_return_json_response)(request=None)
+            success_result = json.loads(success_response.content).get("result")
+            self.assertTrue(success_result)

pipeline/contrib/engine_admin/urls.py

@@ -0,0 +1,39 @@
+# -*- 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 django.urls import path, register_converter
+from django.urls.converters import StringConverter
+
+from . import views
+
+API_VERSION = "v1"
+
+
+class EngineConverter(StringConverter):
+    regex = "pipeline_engine"
+
+
+register_converter(EngineConverter, "engine")
+
+
+urlpatterns = [
+    path("", views.render_index),
+    path(f"api/{API_VERSION}/<engine:engine_type>/task_pause/<str:instance_id>/", views.task_pause),
+    path(f"api/{API_VERSION}/<engine:engine_type>/task_resume/<str:instance_id>/", views.task_resume),
+    path(f"api/{API_VERSION}/<engine:engine_type>/task_revoke/<str:instance_id>/", views.task_revoke),
+    path(f"api/{API_VERSION}/<engine:engine_type>/node_retry/<str:instance_id>/", views.node_retry),
+    path(f"api/{API_VERSION}/<engine:engine_type>/node_skip/<str:instance_id>/", views.node_skip),
+    path(f"api/{API_VERSION}/<engine:engine_type>/node_callback/<str:instance_id>/", views.node_callback),
+    path(f"api/{API_VERSION}/<engine:engine_type>/node_skip_exg/<str:instance_id>/", views.node_skip_exg),
+    path(f"api/{API_VERSION}/<engine:engine_type>/node_skip_cpg/<str:instance_id>/", views.node_skip_cpg),
+    path(f"api/{API_VERSION}/<engine:engine_type>/node_forced_fail/<str:instance_id>/", views.node_forced_fail),
+]

pipeline/contrib/engine_admin/views.py

@@ -0,0 +1,155 @@
+# -*- 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.
+"""
+import functools
+import json
+
+from django.conf import settings
+from django.http import JsonResponse
+from django.shortcuts import render
+from django.utils.module_loading import import_string
+from django.views.decorators.http import require_POST
+
+from pipeline.engine.utils import ActionResult
+
+from .handlers.pipeline_engine import PipelineEngineRequestHandler
+
+ENGINE_REQUEST_HANDLERS = {"pipeline_engine": PipelineEngineRequestHandler}
+
+
+def _ensure_return_json_response(func):
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+        response = func(*args, **kwargs)
+        if isinstance(response, dict):
+            return JsonResponse(response)
+        if isinstance(response, ActionResult):
+            return JsonResponse(
+                {"result": response.result, "message": response.message, "data": None})
+        return response
+
+    return wrapper
+
+
+def _check_api_permission(func):
+    @functools.wraps(func)
+    def wrapper(request, *args, **kwargs):
+        if getattr(settings, "PIPELINE_ENGINE_ADMIN_API_PERMISSION", None):
+            try:
+                perm_func = import_string(settings.PIPELINE_ENGINE_ADMIN_API_PERMISSION)
+            except ImportError:
+                return JsonResponse(
+                    {
+                        "result": False,
+                        "message": "Pipeline engine admin permission function import error.",
+                        "data": None,
+                    }
+                )
+            if not perm_func(request, *args, **kwargs):
+                return JsonResponse(
+                    {
+                        "result": False,
+                        "message": "You have no permission to call pipeline engine admin api.",
+                        "data": None,
+                    }
+                )
+        return func(request, *args, **kwargs)
+
+    return wrapper
+
+
+@_check_api_permission
+def render_index(request, *args, **kwargs):
+    return render(request, "engine_admin/index.html",
+                  {"BKAPP_CSRF_COOKIE_NAME": settings.CSRF_COOKIE_NAME})
+
+
+@require_POST
+@_check_api_permission
+@_ensure_return_json_response
+def task_pause(request, engine_type, instance_id):
+    """
+    暂停任务
+    """
+    handler = ENGINE_REQUEST_HANDLERS[engine_type](request, "task_pause", instance_id)
+    return handler.execute()
+
+
+@require_POST
+@_check_api_permission
+@_ensure_return_json_response
+def task_resume(request, engine_type, instance_id):
+    handler = ENGINE_REQUEST_HANDLERS[engine_type](request, "task_resume", instance_id)
+    return handler.execute()
+
+
+@require_POST
+@_check_api_permission
+@_ensure_return_json_response
+def task_revoke(request, engine_type, instance_id):
+    handler = ENGINE_REQUEST_HANDLERS[engine_type](request, "task_revoke", instance_id)
+    return handler.execute()
+
+
+@require_POST
+@_check_api_permission
+@_ensure_return_json_response
+def node_retry(request, engine_type, instance_id):
+    inputs = json.loads(request.body).get("inputs")
+    handler = ENGINE_REQUEST_HANDLERS[engine_type](request, "node_retry", instance_id)
+    return handler.execute(inputs=inputs)
+
+
+@require_POST
+@_check_api_permission
+@_ensure_return_json_response
+def node_skip(request, engine_type, instance_id):
+    handler = ENGINE_REQUEST_HANDLERS[engine_type](request, "node_skip", instance_id)
+    return handler.execute()
+
+
+@require_POST
+@_check_api_permission
+@_ensure_return_json_response
+def node_callback(request, engine_type, instance_id):
+    body = json.loads(request.body)
+    data, version = body.get("data"), body.get("version")
+    handler = ENGINE_REQUEST_HANDLERS[engine_type](request, "node_callback", instance_id)
+    return handler.execute(data=data, version=version)
+
+
+@require_POST
+@_check_api_permission
+@_ensure_return_json_response
+def node_skip_exg(request, engine_type, instance_id):
+    flow_id = json.loads(request.body).get("flow_id")
+    handler = ENGINE_REQUEST_HANDLERS[engine_type](request, "node_skip_exg", instance_id)
+    return handler.execute(flow_id=flow_id)
+
+
+@require_POST
+@_check_api_permission
+@_ensure_return_json_response
+def node_skip_cpg(request, engine_type, instance_id):
+    body = json.loads(request.body)
+    converge_gateway_id = body.get("converge_gateway_id")
+    flow_ids = body.get("flow_ids")
+    handler = ENGINE_REQUEST_HANDLERS[engine_type](request, "node_skip_cpg", instance_id)
+    return handler.execute(flow_ids=flow_ids, converge_gateway_id=converge_gateway_id)
+
+
+@require_POST
+@_check_api_permission
+@_ensure_return_json_response
+def node_forced_fail(request, engine_type, instance_id):
+    handler = ENGINE_REQUEST_HANDLERS[engine_type](request, "node_forced_fail", instance_id)
+    return handler.execute()