fix(backend): 单据todo优化和问题修复 #8661

dbm-ui/backend/db_meta/models/cluster.py

@@ -449,14 +449,14 @@ def get_cluster_id__primary_address_map(cls, cluster_ids: List[int]) -> Dict[int
                     continue
 
                 # 取一个状态正常的 spider-master 接入层
-                spider_instance = cluster.proxyinstance_set.filter(
+                spider_inst = cluster.proxyinstance_set.filter(
                     tendbclusterspiderext__spider_role=TenDBClusterSpiderRole.SPIDER_MASTER,
                     status=InstanceStatus.RUNNING.value,
                 ).first()
-
-                ctl_address = "{}{}{}".format(spider_instance.machine.ip, IP_PORT_DIVIDER, spider_instance.port + 1000)
-                addresses.append(ctl_address)
-                ctl_address__cluster_id_map[ctl_address] = cluster.id
+                if spider_inst:
+                    ctl_address = "{}{}{}".format(spider_inst.machine.ip, IP_PORT_DIVIDER, spider_inst.port + 1000)
+                    addresses.append(ctl_address)
+                    ctl_address__cluster_id_map[ctl_address] = cluster.id
 
             logger.info("addresses: {}".format(addresses))
 

dbm-ui/backend/ticket/constants.py

@@ -127,9 +127,11 @@ class TicketFlowStatus(str, StructuredEnum):
     SKIPPED = EnumField("SKIPPED", _("跳过"))
 
 
+# 流程成功状态
 FLOW_FINISHED_STATUS = [TicketFlowStatus.SKIPPED, TicketFlowStatus.SUCCEEDED]
+# 流程未执行状态
 FLOW_NOT_EXECUTE_STATUS = [TicketFlowStatus.SKIPPED, TicketFlowStatus.PENDING]
-
+# 任务树和流程状态映射
 BAMBOO_STATE__TICKET_STATE_MAP = {
     StateType.FINISHED.value: TicketFlowStatus.SUCCEEDED.value,
     StateType.FAILED.value: TicketFlowStatus.FAILED.value,
@@ -140,6 +142,13 @@ class TicketFlowStatus(str, StructuredEnum):
     StateType.CREATED.value: TicketFlowStatus.RUNNING.value,
     StateType.READY.value: TicketFlowStatus.RUNNING.value,
 }
+# 任务流程与todo的状态映射
+INNER_FLOW_TODO_STATUS_MAP = {
+    TicketFlowStatus.TERMINATED: TodoStatus.DONE_FAILED,
+    TicketFlowStatus.FAILED: TodoStatus.TODO,
+    TicketFlowStatus.SUCCEEDED: TodoStatus.DONE_SUCCESS,
+    TicketFlowStatus.RUNNING: TodoStatus.DONE_SUCCESS,
+}
 
 EXCLUSIVE_TICKET_EXCEL_PATH = "backend/ticket/exclusive_ticket.xlsx"
 
@@ -296,9 +305,9 @@ def get_approve_mode_by_ticket(cls, ticket_type):
     TENDBCLUSTER_AUTHORIZE_RULES = TicketEnumField("TENDBCLUSTER_AUTHORIZE_RULES", _("TenDB Cluster 授权"), _("权限管理"))
     TENDBCLUSTER_EXCEL_AUTHORIZE_RULES = TicketEnumField("TENDBCLUSTER_EXCEL_AUTHORIZE_RULES", _("TenDB Cluster EXCEL授权"), _("权限管理"))  # noqa
     TENDBCLUSTER_STANDARDIZE = TicketEnumField("TENDBCLUSTER_STANDARDIZE", _("TenDB Cluster 集群标准化"), register_iam=False)
-    TENDBCLUSTER_METADATA_IMPORT = TicketEnumField("TENDBCLUSTER_METADATA_IMPORT", _("TenDB Cluster 元数据导入"),register_iam=False)  # noqa
+    TENDBCLUSTER_METADATA_IMPORT = TicketEnumField("TENDBCLUSTER_METADATA_IMPORT", _("TenDB Cluster 元数据导入"), register_iam=False)  # noqa
     TENDBCLUSTER_APPEND_DEPLOY_CTL = TicketEnumField("TENDBCLUSTER_APPEND_DEPLOY_CTL", _("TenDB Cluster 追加部署中控"), register_iam=False)  # noqa
-    TENDBSINGLE_METADATA_IMPORT = TicketEnumField("TENDBSINGLE_METADATA_IMPORT", _("TenDB Single 元数据导入"), register_iam=False)  # noqa
+    TENDBSINGLE_METADATA_IMPORT = TicketEnumField("TENDBSINGLE_METADATA_IMPORT", _("TenDB Single 元数据导入"),  register_iam=False)  # noqa
     TENDBSINGLE_STANDARDIZE = TicketEnumField("TENDBSINGLE_STANDARDIZE", _("TenDB Single 集群标准化"), register_iam=False)  # noqa
     TENDBCLUSTER_DATA_MIGRATE = TicketEnumField("TENDBCLUSTER_DATA_MIGRATE", _("TenDB Cluster DB克隆"), _("数据处理"))
     TENDBCLUSTER_DUMP_DATA = TicketEnumField("TENDBCLUSTER_DUMP_DATA", _("TenDB Cluster 数据导出"), _("数据处理"))

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

@@ -26,6 +26,7 @@
 from backend.ticket.builders.common.base import fetch_cluster_ids
 from backend.ticket.constants import (
     BAMBOO_STATE__TICKET_STATE_MAP,
+    INNER_FLOW_TODO_STATUS_MAP,
     FlowCallbackType,
     FlowErrCode,
     FlowMsgType,
@@ -106,7 +107,8 @@ def _status(self) -> str:
         else:
             status = BAMBOO_STATE__TICKET_STATE_MAP.get(self.flow_tree.status, constants.TicketFlowStatus.RUNNING)
 
-        todo_status = TodoStatus.TODO if status == TicketFlowStatus.FAILED else TodoStatus.DONE_SUCCESS
+        # 根据流程状态映射todo的状态
+        todo_status = INNER_FLOW_TODO_STATUS_MAP.get(status, TodoStatus.TODO)
         fail_todo = self.flow_obj.todo_of_flow.filter(type=TodoType.INNER_FAILED).first()
         # 如果任务失败，且不存在todo，则创建一条
         if not fail_todo and todo_status == TodoStatus.TODO:

dbm-ui/backend/ticket/handler.py

@@ -68,7 +68,8 @@ def add_related_object(cls, ticket_data: List[Dict]) -> List[Dict]:
                 {int(cluster_id): info["immute_domain"] for cluster_id, info in clusters.items()}
             )
             instances = ticket.details.get("instances", {})
-            instance_id_ip_port_map.update({int(inst_id): info["instance"] for inst_id, info in instances.items()})
+            if isinstance(instances, dict):
+                instance_id_ip_port_map.update({int(inst_id): info["instance"] for inst_id, info in instances.items()})
             ticket_id_obj_ids_map[ticket.id] = {
                 "cluster_ids": fetch_cluster_ids(ticket.details),
                 "instance_ids": fetch_instance_ids(ticket.details),
@@ -419,7 +420,6 @@ def ticket_status_standardization(cls):
         """
         旧单据状态标准化。TODO: 迁移后此段代码可删除
         """
-        batch = 50
 
         # 标准化只针对running的单据，其他状态单据不影响
         running_tickets = list(Ticket.objects.filter(status=TicketStatus.RUNNING))
@@ -430,36 +430,31 @@ def ticket_status_standardization(cls):
             print(f"ticket[{ticket.id}] status {raw_status} ---> {ticket.status}")
 
         # 失败的单据要增加一条todo关联
-        failed_tickets = Ticket.objects.prefetch_related("flows").filter(status=TicketStatus.FAILED)
-        todos = []
+        failed_tickets = Ticket.objects.prefetch_related("flows__todo_of_flow").filter(status=TicketStatus.FAILED)
         for ticket in failed_tickets:
             inner_flow = ticket.flows.filter(flow_type=FlowType.INNER_FLOW, status=TicketFlowStatus.FAILED).first()
             if not inner_flow or inner_flow.todo_of_flow.exists():
                 continue
-            todo = Todo(
+            Todo.objects.create(
                 name=_("【{}】单据任务执行失败，待处理").format(ticket.get_ticket_type_display()),
                 flow=inner_flow,
                 ticket=ticket,
                 type=TodoType.INNER_FAILED,
                 context=BaseTodoContext(inner_flow.id, ticket.id).to_dict(),
             )
-            todos.append(todo)
             print(f"ticket[{ticket.id}] add a failed todo")
 
         # 待审批的单据要增加一条todo关联
-        itsm_tickets = Ticket.objects.prefetch_related("flows").filter(status=TicketStatus.FAILED)
+        itsm_tickets = Ticket.objects.prefetch_related("flows").filter(status=TicketStatus.APPROVE)
         for ticket in itsm_tickets:
             itsm_flow = ticket.flows.filter(flow_type=FlowType.BK_ITSM, status=TicketFlowStatus.RUNNING).first()
             if not itsm_flow or itsm_flow.todo_of_flow.exists():
                 continue
-            todo = Todo(
+            Todo.objects.create(
                 name=_("【{}】单据等待审批").format(ticket.get_ticket_type_display()),
                 flow=itsm_flow,
                 ticket=ticket,
                 type=TodoType.ITSM,
                 context=ItsmTodoContext(itsm_flow.id, ticket.id).to_dict(),
             )
-            todos.append(todo)
             print(f"ticket[{ticket.id}] add a itsm todo")
-
-        Todo.objects.bulk_create(todos, batch_size=batch)

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

@@ -44,17 +44,18 @@ def get_operators(self, todo_type, ticket, operators):
         # 构造单据状态与处理人之间的对应关系
         # - 审批中：提单人可撤销，dba可处理
         # - 待执行：提单人 + 单据协助人
-        # - 待继续：提单人 + dba + 单据协助人
-        # - 待补货：提单人 + dba + 单据协助人
-        # - 已失败：提单人 + dba + 单据协助人
+        # - 待继续：提单人 + 单据协助人 + dba
+        # - 待补货：提单人 + 单据协助人 + dba
+        # - 已失败：提单人 + 单据协助人 + dba
         todo_operators_map = {
             TodoType.ITSM: dba,
             TodoType.APPROVE: creator + biz_helpers,
-            TodoType.INNER_APPROVE: creator + dba + biz_helpers,
-            TodoType.RESOURCE_REPLENISH: creator + dba + biz_helpers,
-            TodoType.INNER_FAILED: creator + dba + biz_helpers,
+            TodoType.INNER_APPROVE: creator + biz_helpers + dba,
+            TodoType.RESOURCE_REPLENISH: creator + biz_helpers + dba,
+            TodoType.INNER_FAILED: creator + biz_helpers + dba,
         }
-        operators = list(set(operators + todo_operators_map.get(todo_type, [])))
+        # 按照顺序去重
+        operators = list(dict.fromkeys(operators + todo_operators_map.get(todo_type, [])))
         return operators
 
     def create(self, **kwargs):

dbm-ui/backend/ticket/serializers.py

@@ -131,7 +131,7 @@ def get_status(self, obj):
         return obj.status
 
     def get_ticket_type_display(self, obj):
-        return obj.get_ticket_type_display()
+        return TicketType.get_choice_label(obj.ticket_type)
 
     def get_status_display(self, obj):
         return TicketStatus.get_choice_label(obj.status)

dbm-ui/backend/ticket/todos/pause_todo.py

@@ -12,8 +12,7 @@
 
 from backend.db_meta.models.sqlserver_dts import DtsStatus, SqlserverDtsInfo
 from backend.ticket import todos
-from backend.ticket.constants import TicketFlowStatus, TicketType, TodoType
-from backend.ticket.flow_manager import manager
+from backend.ticket.constants import TicketFlowStatus, TicketType, TodoStatus, TodoType
 from backend.ticket.flow_manager.manager import TicketFlowManager
 from backend.ticket.todos import ActionType, BaseTodoContext
 
@@ -43,7 +42,7 @@ def _process(self, username, action, params):
 
         # 所有待办完成后，执行后面的flow
         if not self.todo.ticket.todo_of_ticket.exist_unfinished():
-            manager.TicketFlowManager(ticket=self.todo.ticket).run_next_flow()
+            TicketFlowManager(ticket=self.todo.ticket).run_next_flow()
 
         # 如果是数据迁移单据，更改迁移记录状态信息
         if self.todo.ticket.ticket_type in [TicketType.SQLSERVER_INCR_MIGRATE, TicketType.SQLSERVER_FULL_MIGRATE]:
@@ -73,3 +72,17 @@ def _process(self, username, action, params):
         self.todo.refresh_from_db(fields=["flow"])
         if self.todo.flow.status == TicketFlowStatus.SUCCEEDED:
             self.todo.set_success(username, action)
+
+
+@todos.TodoActorFactory.register(TodoType.INNER_FAILED)
+class FailedTodo(todos.TodoActor):
+    """来自主流程-失败后待确认"""
+
+    def _process(self, username, action, params):
+        # 终止-仅将todo进行终止(任务流程的终止)，确认-关联flow进行重试
+        if action == ActionType.TERMINATE:
+            self.todo.set_status(username, TodoStatus.DONE_FAILED)
+        else:
+            manager = TicketFlowManager(ticket=self.todo.ticket)
+            fail_inner_flow = manager.get_ticket_flow_cls(self.todo.flow.flow_type)(self.todo.flow)
+            fail_inner_flow.retry()

dbm-ui/backend/ticket/views.py

@@ -92,7 +92,7 @@ class TicketViewSet(viewsets.AuditedModelViewSet):
     单据视图
     """
 
-    queryset = Ticket.objects.all().prefetch_related("todo_of_ticket")
+    queryset = Ticket.objects.all()
     serializer_class = TicketSerializer
     filter_class = TicketListFilter
     pagination_class = AuditedLimitOffsetPagination
@@ -158,6 +158,11 @@ def _get_self_manage_tickets(cls, user):
         ticket_filter = Q(creator=user.username) | reduce(operator.or_, manage_filters or [Q()])
         return Ticket.objects.filter(ticket_filter)
 
+    def filter_queryset(self, queryset):
+        """filter_class可能导致预取的todo失效，这里重新取一次"""
+        queryset = super().filter_queryset(queryset)
+        return queryset.prefetch_related("todo_of_ticket")
+
     def get_queryset(self):
         """
         单据queryset规则--针对list：