feat(backend): 任务完成后进行通知 #2588

.github/workflows/auto_create_release.yml

@@ -84,7 +84,7 @@ jobs:
         cat ${filename}
         git config --global user.email [email protected]
         git config --global user.name github-actions
-        git remote set-url origin https://x-access-token:${{ secrets.WORKFLOW_ACTION_TOKEN }}@github.com/$GITHUB_REPOSITORY
+        git remote set-url origin https://x-access-token:${{ secrets.WORKFLOW_ACTION_TOKEN }}@github.com/TencentBlueKing/blueking-dbm
         git add .
         git commit -m "docs: release version ${{ steps.yaml-data.outputs.data }} #1"
         git push origin ${{ steps.get-current-branch.outputs.current_branch }}

dbm-ui/backend/configuration/constants.py

@@ -63,6 +63,7 @@ class SystemSettingsEnum(str, StructuredEnum):
     DBM_MIGRATE_USER = EnumField("DBM_MIGRATE_USER", _("具备迁移权限的人员名单"))
     BIZ_CONFIG = EnumField("BIZ_CONFIG", _("全业务通用配置信息"))
     AFFINITY = EnumField("AFFINITY", _("容灾要求(各个环境可能不同，比如SG为空)"))
+    SYSTEM_MSG_TYPE = EnumField("SYSTEM_MSG_TYPE", _("系统消息通知方式"))
 
 
 class BizSettingsEnum(str, StructuredEnum):
@@ -155,6 +156,7 @@ class BizSettingsEnum(str, StructuredEnum):
     [SystemSettingsEnum.DBM_MIGRATE_USER, "list", DBM_DEFAULT_MIGRATE_USER, _("具备迁移权限的人员名单")],
     [SystemSettingsEnum.BIZ_CONFIG, "dict", BIZ_CONFIG_INFO, _("默认的全业务配置信息")],
     [SystemSettingsEnum.AFFINITY, "list", [], _("环境的容灾要求")],
+    [SystemSettingsEnum.SYSTEM_MSG_TYPE, "list", ["weixin", "mail"], _("系统消息通知方式")],
 ]
 
 # 环境配置项 是否支持DNS解析 pulsar flow used

dbm-ui/backend/dbm_init/medium/BaseDockerfile

@@ -24,5 +24,6 @@ RUN set -ex && \
     wget -qO dbbackup-go-deps-community.tar.gz ${DOWNLOAD_URL}/${VERSION}/dbbackup-go-deps-community.tar.gz && \
     wget -qO bind.tar.gz ${DOWNLOAD_URL}/${VERSION}/bind.tar.gz && \
     wget -qO nginx-portable.tgz ${DOWNLOAD_URL}/${VERSION}/nginx-portable.tgz && \
+    wget -qO dba-toolkit.tar.gz ${DOWNLOAD_URL}/${VERSION}/dba-toolkit.tar.gz && \
     wget -qO tmysqlparse ${DOWNLOAD_URL}/${VERSION}/tmysqlparse && \
     wget -qO redis_dts.tar.gz ${DOWNLOAD_URL}/${VERSION}/redis_dts.tar.gz

dbm-ui/backend/dbm_init/medium/medium.lock

@@ -63,6 +63,10 @@ mysql:
     commitDate: 200001010101
     name: mysql-monitor.tar.gz
     version: 1.0.3
+- mysql-dba-toolkit:
+    buildPath: /toolkit/dba-toolkit.tar.gz
+    name: dba-toolkit.tar.gz
+    version: 1.0.0
 pulsar:
 - actuator:
     buildPath: /blueking-dbm/dbm-services/bigdata/db-tools/dbactuator/build/dbactuator

dbm-ui/backend/flow/signal/handlers.py

@@ -8,21 +8,27 @@
 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.config
+import logging
 
+from celery.task import task
 from django.dispatch import receiver
 from django.utils import timezone
 from django.utils.translation import ugettext as _
 from pipeline.eri.signals import post_set_state
 
+from backend import env
+from backend.components import CmsiApi
+from backend.configuration.constants import SystemSettingsEnum
+from backend.configuration.models import SystemSettings
 from backend.db_dirty.handlers import DBDirtyMachineHandler
+from backend.exceptions import ApiError
 from backend.flow.consts import StateType
 from backend.flow.engine.bamboo.engine import BambooEngine
 from backend.flow.models import FlowNode, FlowTree
 from backend.ticket.constants import FlowCallbackType, FlowType, TicketFlowStatus
 from backend.ticket.flow_manager.inner import InnerFlow
 from backend.ticket.flow_manager.manager import TicketFlowManager
-from backend.ticket.models import Ticket
+from backend.ticket.models import Flow, Ticket
 
 logger = logging.getLogger("flow")
 
@@ -90,6 +96,7 @@ def callback_ticket(ticket_id, root_id):
     # 在认为inner flow执行结束情况下，执行inner flow的后继动作
     inner_flow_obj = InnerFlow(flow_obj=current_flow)
     if inner_flow_obj.status not in [TicketFlowStatus.PENDING, TicketFlowStatus.RUNNING]:
+        send_msg_for_flow.apply_async(args=[current_flow.id])
         inner_flow_obj.callback(callback_type=FlowCallbackType.POST_CALLBACK.value)
 
     # 如果flow type的类型为快速任务，则跳过callback
@@ -99,3 +106,39 @@ def callback_ticket(ticket_id, root_id):
     if current_flow and current_flow.flow_obj_id == root_id:
         manager = TicketFlowManager(ticket=ticket)
         manager.run_next_flow()
+
+
+@task(ignore_result=True)
+def send_msg_for_flow(flow_id: int):
+    """
+    发送消息
+    """
+    flow = Flow.objects.get(id=flow_id)
+    inner_flow_obj = InnerFlow(flow_obj=flow)
+    msg_types = CmsiApi.get_msg_type()
+    ticket = flow.ticket
+    ticket_type = ticket.get_ticket_type_display()
+    for msg_type in msg_types:
+        if msg_type["type"] not in SystemSettings.get_setting_value(
+            key=SystemSettingsEnum.SYSTEM_MSG_TYPE.value, default=["weixin", "mail"]
+        ):
+            continue
+        try:
+            CmsiApi.send_msg(
+                {
+                    "msg_type": msg_type["type"],
+                    "receiver__username": ticket.creator,
+                    "title": _("DBM数据库管理  {ticket_type} 执行结果").format(ticket_type=ticket_type),
+                    "content": _(
+                        "{ticket_type} {flow_alias} 执行{flow_status}。\n" "单据详情：{ticket_url}\n" "任务详情：{flow_url}\n"
+                    ).format(
+                        ticket_type=ticket_type,
+                        flow_alias=flow.flow_alias,
+                        flow_status=flow.get_status_display(),
+                        ticket_url=ticket.url,
+                        flow_url=inner_flow_obj.url,
+                    ),
+                }
+            )
+        except ApiError as err:
+            logger.error(f"send message error, ticket_id:{ticket.id}, root_id:{flow.flow_obj_id}, err:{err}")

dbm-ui/backend/flow/utils/mysql/mysql_act_playload.py

@@ -48,6 +48,7 @@
 from backend.flow.utils.mysql.proxy_act_payload import ProxyActPayload
 from backend.flow.utils.tbinlogdumper.tbinlogdumper_act_payload import TBinlogDumperActPayload
 from backend.ticket.constants import TicketType
+from backend.ticket.models import Ticket
 
 apply_list = [TicketType.MYSQL_SINGLE_APPLY.value, TicketType.MYSQL_HA_APPLY.value]
 
@@ -951,6 +952,7 @@ def get_partition_payload(self, **kwargs) -> dict:
         """
         表分区
         """
+        ticket = Ticket.objects.get(id=self.ticket_data["uid"])
         if self.ticket_data["ticket_type"] == TicketType.MYSQL_PARTITION:
             shard_name = ""
         else:
@@ -967,7 +969,7 @@ def get_partition_payload(self, **kwargs) -> dict:
                     "master_ip": self.cluster["ip"],
                     "master_port": self.cluster["port"],
                     "shard_name": shard_name,
-                    "ticket": "{}/self-service/my-tickets?id={}".format(env.BK_SAAS_HOST, self.ticket_data["uid"]),
+                    "ticket": ticket.url,
                     "file_path": self.cluster["file_path"],
                 },
             },

dbm-ui/backend/ticket/admin.py

@@ -15,20 +15,30 @@
 
 @admin.register(models.Ticket)
 class TicketAdmin(admin.ModelAdmin):
-    list_display = ("bk_biz_id", "ticket_type", "status", "creator", "update_at")
+    list_display = ("id", "bk_biz_id", "ticket_type", "status", "creator", "update_at")
     list_filter = ("bk_biz_id", "ticket_type", "status")
-    search_fields = ("id",)
+    search_fields = ("id", "creator")
 
 
 @admin.register(models.Flow)
 class TicketFlowAdmin(admin.ModelAdmin):
-    list_display = ("ticket_id", "flow_type", "flow_alias", "flow_obj_id", "status", "update_at")
-    search_fields = ("ticket_id", "flow_obj_id")
-    list_filter = ("flow_type", "status")
+    list_display = ("id", "ticket_id", "flow_type", "flow_alias", "flow_obj_id", "status", "update_at")
+    search_fields = ("flow_obj_id",)
+    list_filter = ("flow_type", "flow_alias", "status")
 
 
 @admin.register(models.TicketFlowConfig)
 class TicketFlowConfigAdmin(admin.ModelAdmin):
     list_display = ("ticket_type", "group", "editable")
     search_fields = ("ticket_type", "group", "editable")
     list_filter = ("group",)
+
+
+@admin.register(models.Todo)
+class TodoAdmin(admin.ModelAdmin):
+    list_display = ("id", "name", "flow_id", "ticket_id", "operators", "type", "done_by", "done_at")
+    list_filter = (
+        "type",
+        "status",
+    )
+    search_fields = ("id", "name", "done_by")

dbm-ui/backend/ticket/builders/__init__.py

@@ -152,13 +152,12 @@ def get_params(self):
                 {
                     "name": _("单据链接"),
                     "type": "LINK",
-                    "value": f"{env.BK_SAAS_HOST}/{self.ticket.bk_biz_id}/my-tickets?id={self.ticket.id}",
+                    "value": self.ticket.url,
                 },
                 {
                     "name": _("需求信息"),
                     "type": "LINK",
-                    "value": f"{env.BK_SAAS_HOST}/{self.ticket.bk_biz_id}/"
-                    f"my-tickets?id={self.ticket.id}&isFullscreen=true",
+                    "value": f"{self.ticket.url}&isFullscreen=true",
                 },
             ],
             "meta": {

dbm-ui/backend/ticket/builders/mysql/mysql_import_sqlfile.py

@@ -55,7 +55,7 @@ def get_params(self):
 
         # 添加语义执行结果的链接
         root_id = self.ticket.details["root_id"]
-        semantic_url = f"{env.BK_SAAS_HOST}/database/{self.ticket.bk_biz_id}/mission-details/{root_id}/"
+        semantic_url = f"{env.BK_SAAS_HOST}/{self.ticket.bk_biz_id}/task-history/detail/{root_id}"
         params["dynamic_fields"].append({"name": _("模拟执行链接"), "type": "LINK", "value": semantic_url})
 
         return params

dbm-ui/backend/ticket/flow_manager/inner.py

@@ -16,6 +16,7 @@
 
 from django.utils.translation import gettext as _
 
+from backend import env
 from backend.db_dirty.handlers import DBDirtyMachineHandler
 from backend.db_meta.exceptions import ClusterExclusiveOperateException
 from backend.db_meta.models import Cluster
@@ -96,7 +97,7 @@ def _status(self) -> str:
 
     @property
     def _url(self) -> str:
-        return f"/{self.ticket.bk_biz_id}/task-history/detail/{self.root_id}"
+        return f"{env.BK_SAAS_HOST}/{self.ticket.bk_biz_id}/task-history/detail/{self.root_id}"
 
     def check_exclusive_operations(self):
         """判断执行互斥"""

dbm-ui/backend/ticket/models/ticket.py

@@ -113,6 +113,10 @@ class Meta:
             models.Index(fields=["status"]),
         ]
 
+    @property
+    def url(self):
+        return f"{env.BK_SAAS_HOST}/{self.bk_biz_id}/my-tickets?id={self.id}"
+
     def set_failed(self):
         self.status = TicketStatus.FAILED
         self.save()

dbm-ui/backend/ticket/models/todo.py

@@ -8,19 +8,55 @@
 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 django.db import models
 from django.utils import timezone
 from django.utils.translation import ugettext_lazy as _
 
+from backend import env
 from backend.bk_web.constants import LEN_MIDDLE, LEN_SHORT
 from backend.bk_web.models import AuditedModel
+from backend.components import CmsiApi
+from backend.configuration.constants import SystemSettingsEnum
+from backend.configuration.models import SystemSettings
+from backend.exceptions import ApiError
 from backend.ticket.constants import TicketFlowStatus, TodoStatus, TodoType
 
+logger = logging.getLogger("root")
+
 
 class TodoManager(models.Manager):
     def exist_unfinished(self):
         return self.filter(status__in=[TodoStatus.TODO, TodoStatus.RUNNING]).exists()
 
+    def create(self, **kwargs):
+        todo = super().create(**kwargs)
+        msg_types = CmsiApi.get_msg_type()
+        ticket = todo.ticket
+        ticket_type = ticket.get_ticket_type_display()
+        for msg_type in msg_types:
+            if msg_type["type"] not in SystemSettings.get_setting_value(
+                key=SystemSettingsEnum.SYSTEM_MSG_TYPE.value, default=["weixin", "mail"]
+            ):
+                continue
+            try:
+                CmsiApi.send_msg(
+                    {
+                        "msg_type": msg_type["type"],
+                        "receiver__username": ",".join(todo.operators),
+                        "title": _("DBM数据库管理 待办通知").format(ticket_type=ticket_type),
+                        "content": _("有一条[{ticket_type}]待办需要您处理\n" "待办详情：{todo_url}\n").format(
+                            ticket_type=ticket_type,
+                            todo_url=todo.url,
+                        ),
+                    }
+                )
+            except ApiError as err:
+                logger.error(f"send message error, ticket_id:{ticket.id}, todo_id:{todo.id}, err:{err}")
+
+        return todo
+
 
 class Todo(AuditedModel):
     """
@@ -49,6 +85,10 @@ class Todo(AuditedModel):
 
     objects = TodoManager()
 
+    @property
+    def url(self):
+        return f"{env.BK_SAAS_HOST}/{self.ticket.bk_biz_id}/my-todos?id={self.id}"
+
     def set_status(self, username, status):
         self.status = status
 

helm-charts/bk-dbm/Chart.yaml

@@ -79,5 +79,5 @@ dependencies:
 description: A Helm chart for bkdbm
 name: bk-dbm
 type: application
-version: 1.3.0-alpha.21
-appVersion: 1.3.0-alpha.21
+version: 1.3.0-alpha.22
+appVersion: 1.3.0-alpha.22

helm-charts/bk-dbm/charts/db-resource/Chart.yaml

@@ -1,5 +1,5 @@
 apiVersion: v2
-appVersion: 0.0.1-alpha.66
+appVersion: 0.0.1-alpha.68
 description: A Helm chart for Kubernetes
 name: db-resource
 type: application

helm-charts/bk-dbm/charts/dbm/Chart.yaml

@@ -1,5 +1,5 @@
 apiVersion: v2
-appVersion: 1.3.0-alpha.159
+appVersion: 1.3.0-alpha.177
 description: A Helm chart for dbm
 name: dbm
 type: application