Merge pull request #7982 from guohelu/feature_template_webhook

feat: 任务模板支持webhook --story=126514126

Author: dengyh

bin/pre_release

@@ -4,4 +4,5 @@ python manage.py createcachetable django_cache
 python manage.py update_component_models
 python manage.py update_variable_models
 python manage.py sync_saas_apigw
-python manage.py register_bksops_notice
+python manage.py register_bksops_notice
+python manage.py sync_webhook_events ./config webhook_resources.yaml

config/default.py

@@ -117,6 +117,7 @@
     "apigw_manager.apigw",
     "bk_notice_sdk",
     "bk_audit.contrib.bk_audit",
+    "webhook",
 )
 
 # 这里是默认的中间件，大部分情况下，不需要改动

config/webhook_resources.yaml

@@ -0,0 +1,6 @@
+version: 1
+events:
+  - code: task_failed
+    name: 任务失败
+  - code: task_finished
+    name: 任务完成

dist/bkflow_django_webhook-1.2.0rc0-py2.py3-none-any.whl

@@ -259,5 +259,14 @@ class GseAgentStatus(Enum):
     ONlINE = 1
 
 
-def get_default_scope():
-    return {PROJECT: ["*"]}
+class WebhookScopeType(Enum):
+    """webhook作用域类型"""
+
+    TEMPLATE = "template"
+
+
+class WebhookEventType(Enum):
+    """webhook事件类型"""
+
+    TASK_FAILED = "task_failed"
+    TASK_FINISHED = "task_finished"

gcloud/constants.py

@@ -70,6 +70,7 @@ class CreateTaskTemplateSerializer(BaseTaskTemplateSerializer):
     pipeline_tree = serializers.CharField()
     project = serializers.IntegerField(write_only=True)
     template_id = serializers.CharField(help_text="模板ID", source="id", read_only=True)
+    webhook_configs = serializers.JSONField(help_text="webhook配置", required=False)
 
     def validate_project(self, value):
         try:
@@ -113,6 +114,7 @@ class Meta:
             "pipeline_tree",
             "project",
             "template_id",
+            "webhook_configs",
         ]
 
 

gcloud/core/apis/drf/serilaziers/task_template.py

@@ -51,6 +51,8 @@
 from gcloud.tasktmpl3.signals import post_template_save_commit
 from gcloud.template_base.domains.template_manager import TemplateManager
 from gcloud.user_custom_config.constants import TASKTMPL_ORDERBY_OPTIONS
+from gcloud.utils.webhook import apply_webhook_configs, get_webhook_configs, clear_scope_webhooks
+from gcloud.constants import WebhookScopeType
 
 logger = logging.getLogger("root")
 manager = TemplateManager(template_model_cls=TaskTemplate)
@@ -166,10 +168,12 @@ def list(self, request, *args, **kwargs):
             user_model.objects.get(username=request.user.username).tasktemplate_set.all().values_list("id", flat=True)
         )
         template_ids = [obj["id"] for obj in data]
+        webhook_configs = get_webhook_configs(scope_code=template_ids)
         templates_labels = TemplateLabelRelation.objects.fetch_templates_labels(template_ids)
         for obj in data:
             obj["is_add"] = 1 if obj["id"] in collected_templates else 0
             obj["template_labels"] = templates_labels.get(obj["id"], [])
+            obj["webhook_configs"] = webhook_configs.get(str(obj["id"]), {})
         return self.get_paginated_response(data) if page is not None else Response(data)
 
     @swagger_auto_schema(
@@ -196,11 +200,13 @@ def list_with_top_collection(self, request, *args, **kwargs):
         # 注入权限
         data = self.injection_auth_actions(request, serializer.data, serializer.instance)
         template_ids = [obj["id"] for obj in data]
+        webhook_configs = get_webhook_configs(scope_code=template_ids)
         templates_labels = TemplateLabelRelation.objects.fetch_templates_labels(template_ids)
         for obj in data:
             obj["template_labels"] = templates_labels.get(obj["id"], [])
             obj["is_collected"] = 1 if obj["id"] in collection_template_ids else 0
             obj["collection_id"] = collection_template_map.get(obj["id"], -1)
+            obj["webhook_configs"] = webhook_configs.get(str(obj["id"]), {})
         return self.get_paginated_response(data) if page is not None else Response(data)
 
     def retrieve(self, request, *args, **kwargs):
@@ -209,6 +215,8 @@ def retrieve(self, request, *args, **kwargs):
         data = self.injection_auth_actions(request, serializer.data, instance)
         labels = TemplateLabelRelation.objects.fetch_templates_labels([instance.id]).get(instance.id, [])
         data["template_labels"] = [label["label_id"] for label in labels]
+        webhook_configs = get_webhook_configs(scope_code=[str(instance.id)])
+        data["webhook_configs"] = webhook_configs.get(str(instance.id), {})
         bk_audit_add_event(
             username=request.user.username,
             action_id=IAMMeta.FLOW_VIEW_ACTION,
@@ -224,6 +232,7 @@ def create(self, request, *args, **kwargs):
         creator = request.user.username
         pipeline_tree = json.loads(serializer.validated_data.pop("pipeline_tree"))
         description = serializer.validated_data.pop("description", "")
+        webhook_configs = serializer.validated_data.pop("webhook_configs", {})
         with transaction.atomic():
             result = manager.create_pipeline(
                 name=name, creator=creator, pipeline_tree=pipeline_tree, description=description
@@ -237,6 +246,14 @@ def create(self, request, *args, **kwargs):
             serializer.validated_data["pipeline_template_id"] = result["data"].template_id
             template_labels = serializer.validated_data.pop("template_labels")
             self.perform_create(serializer)
+            if webhook_configs:
+                apply_result = apply_webhook_configs(webhook_configs, str(serializer.instance.id))
+                if not apply_result["result"]:
+                    message = apply_result["message"]
+                    logger.error(message)
+                    return Response(
+                        {"detail": ErrorDetail(message, err_code.REQUEST_PARAM_INVALID.code)}, exception=True
+                    )
             self._sync_template_lables(serializer.instance.id, template_labels)
             headers = self.get_success_headers(serializer.data)
         # 发送信号
@@ -277,6 +294,7 @@ def update(self, request, *args, **kwargs):
         editor = request.user.username
         pipeline_tree = json.loads(serializer.validated_data.pop("pipeline_tree"))
         description = serializer.validated_data.pop("description", "")
+        webhook_config = serializer.validated_data.pop("webhook_config", {})
         with transaction.atomic():
             result = manager.update_pipeline(
                 pipeline_template=template.pipeline_template,
@@ -293,6 +311,17 @@ def update(self, request, *args, **kwargs):
 
             serializer.validated_data["pipeline_template"] = template.pipeline_template
             template_labels = serializer.validated_data.pop("template_labels")
+            if webhook_config:
+                apply_result = apply_webhook_configs(webhook_config, str(serializer.instance.id))
+                if not apply_result["result"]:
+                    message = apply_result["message"]
+                    logger.error(message)
+                    return Response(
+                        {"detail": ErrorDetail(message, err_code.REQUEST_PARAM_INVALID.code)}, exception=True
+                    )
+            elif not webhook_config and get_webhook_configs([str(serializer.instance.id)]):
+                clear_scope_webhooks(WebhookScopeType.TEMPLATE.value, [str(serializer.instance.id)])
+
             self.perform_update(serializer)
             self._sync_template_lables(serializer.instance.id, template_labels)
         # 发送信号
@@ -332,6 +361,11 @@ def destroy(self, request, *args, **kwargs):
         relation_queryset = TemplateRelationship.objects.filter(ancestor_template_id=pipeline_template_id)
         for relation in relation_queryset:
             relation.templatescheme_set.clear()
+        clear_result = clear_scope_webhooks(WebhookScopeType.TEMPLATE.value, [template.id])
+        if not clear_result["result"]:
+            message = clear_result["message"]
+            logger.error(message)
+            return Response({"detail": ErrorDetail(message, err_code.REQUEST_PARAM_INVALID.code)}, exception=True)
         # 删除流程模板
         template.is_deleted = True
         template.save()

gcloud/core/apis/drf/viewsets/task_template.py

@@ -42,6 +42,8 @@
     TimeoutNodeConfig,
 )
 from gcloud.taskflow3.signals import taskflow_finished, taskflow_revoked
+from gcloud.constants import WebhookScopeType, WebhookEventType
+from webhook.signals import event_broadcast_signal
 
 logger = logging.getLogger("celery")
 
@@ -83,6 +85,22 @@ def _send_node_fail_message(node_id, pipeline_id):
             logger.exception("pipeline_fail_handler[taskflow_id=%s] task delay error: %s" % (taskflow.id, e))
 
 
+def send_task_message(pipeline_id, msg_type):
+    try:
+        taskflow = TaskFlowInstance.objects.get(pipeline_instance__instance_id=pipeline_id)
+        # broadcast events through webhooks
+        event = WebhookEventType.TASK_FAILED.value if msg_type == ATOM_FAILED else WebhookEventType.TASK_FINISHED.value
+        scopes = [(WebhookScopeType.TEMPLATE.value, str(taskflow.template_id))]
+        event_broadcast_signal.send(
+            sender=event, scopes=scopes, extra_info={"taskflow_id": taskflow.id, "event": event}
+        )
+
+    except Exception as e:
+        logger.exception(f"[send_task_message] task() send message({msg_type}) error: {e}")
+    else:
+        logger.info(f"[send_task_message] task() send message({msg_type}) success")
+
+
 def _check_and_callback(task, *args, **kwargs):
     record = TaskCallBackRecord.objects.filter(task_id=task.id).first()
     if not record:
@@ -185,6 +203,7 @@ def bamboo_engine_eri_post_set_state_handler(sender, node_id, to_state, version,
             return
 
         _send_node_fail_message(node_id=node_id, pipeline_id=root_id)
+        send_task_message(pipeline_id=root_id, msg_type=ATOM_FAILED)
 
     elif to_state == bamboo_engine_states.REVOKED and node_id == root_id:
         try:
@@ -198,6 +217,7 @@ def bamboo_engine_eri_post_set_state_handler(sender, node_id, to_state, version,
             pipeline_end.send(sender=Pipeline, root_pipeline_id=root_id)
         except Exception:
             logger.exception("pipeline_end send error")
+        send_task_message(pipeline_id=root_id, msg_type=TASK_FINISHED)
 
     try:
         _node_timeout_info_update(settings.redis_inst, to_state, node_id, version)

gcloud/taskflow3/signals/handlers.py

@@ -24,6 +24,8 @@
 from gcloud.template_base.apis.drf.permission import CommonTemplatePermission, ProjectTemplatePermission
 from gcloud.template_base.apis.drf.serilaziers.template import BatchDeleteSerialzer, TemplateIdsSerializer
 from gcloud.template_base.domains.template_manager import TemplateManager
+from gcloud.utils.webhook import clear_scope_webhooks
+from gcloud.constants import WebhookScopeType
 
 logger = logging.getLogger("root")
 
@@ -39,6 +41,9 @@ def batch_delete(self, request, *args, **kwargs):
         body_serializer = self.template_ids_serializer(data=data)
         body_serializer.is_valid(raise_exception=True)
         template_ids = body_serializer.validated_data.get("template_ids")
+        clear_result = clear_scope_webhooks(WebhookScopeType.TEMPLATE.value, template_ids)
+        if not clear_result["result"]:
+            raise APIException(f'[batch_delete] clear_webhooks False: {clear_result["message"]}')
 
         manager = TemplateManager(template_model_cls=self.tmpl_model)
         result = manager.batch_delete(template_ids)

gcloud/template_base/apis/drf/viewsets/template.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 logging
+
+from webhook.api import apply_scope_subscriptions, apply_scope_webhooks
+from gcloud.constants import WebhookScopeType, WebhookEventType
+from webhook.api import clear_webhooks, get_scope_webhook_retry_policy
+
+logger = logging.getLogger("root")
+
+
+def get_webhook_configs(scope_code: list):
+    try:
+        webhooks = get_scope_webhook_retry_policy(scope_type=WebhookScopeType.TEMPLATE.value, scope_code=scope_code)
+    except Exception as e:
+        logger.exception(f"get_scope_webhooks error: {e}")
+        return {"result": False, "message": f"Failed to get webhook configs: {e}", "data": {}, "code": "500"}
+
+    return webhooks
+
+
+def clear_scope_webhooks(scope_type: str, scope_code: list):
+    try:
+        clear_webhooks(scope_type=scope_type, scope_code=scope_code)
+    except Exception as e:
+        logger.exception(f"clear_webhooks error: {e}")
+        return {"result": False, "message": f"Failed to clear webhooks: {e}", "data": {}, "code": "500"}
+
+    return {"result": True, "message": "success", "data": {}, "code": "0"}
+
+
+def apply_webhook_configs(webhook_configs, scope_code):
+    webhook_code = f"template_{scope_code}_webhook"
+    webhook_name = f"template_{scope_code}_webhook"
+    webhook_configs.update({"code": webhook_code, "name": webhook_name})
+    scope_type = WebhookScopeType.TEMPLATE.value
+    subscription_configs = {webhook_code: [WebhookEventType.TASK_FAILED.value, WebhookEventType.TASK_FINISHED.value]}
+    try:
+        apply_scope_webhooks(scope_type=scope_type, scope_code=scope_code, webhooks=[webhook_configs])
+        apply_scope_subscriptions(
+            scope_type=scope_type, scope_code=scope_code, subscription_configs=subscription_configs
+        )
+    except Exception as e:
+        logger.exception(f"apply_webhook_configs error: {e}")
+        return {"result": False, "message": f"Failed to apply webhook configs: {e}", "data": {}, "code": "500"}
+
+    return {"result": True, "message": "success", "data": {}, "code": "0"}