feat(backend): 密码修改支持异步模式 #7598

Author: iSecloud

dbm-ui/backend/components/mysql_priv_manager/client.py

@@ -128,6 +128,7 @@ def __init__(self):
             method="POST",
             url="/priv/modify_admin_password",
             description=_("新增或者修改实例中管理用户的密码"),
+            default_timeout=self.TIMEOUT,
         )
         self.get_password = self.generate_data_api(
             method="POST",

dbm-ui/backend/configuration/handlers/password.py

@@ -24,7 +24,14 @@
 from backend.db_meta.models import Machine
 from backend.db_periodic_task.models import DBPeriodicTask
 from backend.db_services.ipchooser.query.resource import ResourceQueryHelper
-from backend.flow.consts import DEFAULT_INSTANCE
+from backend.db_services.taskflow.handlers import TaskFlowHandler
+from backend.flow.consts import DEFAULT_INSTANCE, FAILED_STATES, SUCCEED_STATES
+from backend.flow.engine.bamboo.engine import BambooEngine
+from backend.flow.engine.controller.mysql import MySQLController
+from backend.flow.models import FlowTree
+from backend.flow.plugins.components.collections.common.external_service import ExternalServiceComponent
+from backend.ticket.constants import TicketType
+from backend.utils.basic import generate_root_id
 from backend.utils.string import base64_decode, base64_encode
 
 logger = logging.getLogger("root")
@@ -132,13 +139,21 @@ def query_admin_password(
         return admin_password_data
 
     @classmethod
-    def modify_admin_password(cls, operator: str, password: str, lock_hour: int, instance_list: List[Dict]):
+    def modify_admin_password(
+        cls,
+        operator: str,
+        password: str,
+        lock_hour: int,
+        instance_list: List[Dict],
+        is_async: bool = False,
+    ):
         """
         修改db的admin密码
         @param operator: 操作人
         @param password: 修改密码
         @param lock_hour: 锁定时长
         @param instance_list: 修改的实例列表
+        @param is_async: 是否异步执行
         """
         # 获取业务信息，任取一台machine查询
         machine = Machine.objects.get(bk_cloud_id=instance_list[0]["bk_cloud_id"], ip=instance_list[0]["ip"])
@@ -172,7 +187,7 @@ def modify_admin_password(cls, operator: str, password: str, lock_hour: int, ins
 
         # 填充参数，修改admin的密码
         db_type = db_type.pop()
-        modify_password_params = {
+        params = {
             # username固定是ADMIN，与DBM_MYSQL_ADMIN_USER保持一致
             "username": DB_ADMIN_USER_MAP[db_type],
             "component": db_type,
@@ -183,11 +198,37 @@ def modify_admin_password(cls, operator: str, password: str, lock_hour: int, ins
             "security_rule_name": DBM_PASSWORD_SECURITY_NAME,
             "async": False,
         }
-        data = DBPrivManagerApi.modify_admin_password(
-            params=modify_password_params, raw=True, timeout=DBPrivManagerApi.TIMEOUT
-        )["data"]
+
+        # 同步执行直接调用接口，异步执行则返回任务ID
+        if not is_async:
+            resp = DBPrivManagerApi.modify_admin_password(params=params, timeout=DBPrivManagerApi.TIMEOUT, raw=True)
+            data = resp["data"]
+        else:
+            data = root_id = generate_root_id()
+            params.update(ticket_type=TicketType.ADMIN_PASSWORD_MODIFY, bk_biz_id=bk_biz_id, created_by=operator)
+            MySQLController(root_id=root_id, ticket_data=params).mysql_randomize_password()
         return data
 
+    @classmethod
+    def query_async_modify_result(cls, root_id: str):
+        """
+        查询异步密码修改结果
+        @param root_id: 任务ID
+        """
+        flow_tree = FlowTree.objects.get(root_id=root_id)
+        # 任务未完成，退出
+        if flow_tree.status not in [*FAILED_STATES, *SUCCEED_STATES]:
+            return {"status": flow_tree.status, "data": ""}
+
+        # 查询修改密码的节点id
+        task_handler = TaskFlowHandler(root_id)
+        node_id = task_handler.get_node_id_by_component(flow_tree.tree, component_code=ExternalServiceComponent.code)[
+            0
+        ]
+        # 查询输出数据
+        resp = BambooEngine(root_id).get_node_output_data(node_id).data["resp"]
+        return {"status": flow_tree.status, "data": resp["data"]}
+
     @classmethod
     def _get_password_role(cls, cluster_type, role):
         """获取实例对应的密码角色"""

dbm-ui/backend/configuration/serializers.py

@@ -118,6 +118,7 @@ class InstanceInfoSerializer(serializers.Serializer):
     lock_hour = serializers.IntegerField(help_text=_("密码到期小时"))
     password = serializers.CharField(help_text=_("密码"))
     instance_list = serializers.ListSerializer(help_text=_("实例信息"), child=InstanceInfoSerializer())
+    is_async = serializers.BooleanField(help_text=_("是否异步执行"), required=False, default=False)
 
     def validate(self, attrs):
         # 校验密码中的特殊字符
@@ -133,6 +134,10 @@ def validate(self, attrs):
         return attrs
 
 
+class QueryAsyncModifyResultSerializer(serializers.Serializer):
+    root_id = serializers.CharField(help_text=_("任务ID"))
+
+
 class PasswordPolicySerializer(serializers.Serializer):
     class PolicySerializer(serializers.Serializer):
         class IncludeRuleSerializer(serializers.Serializer):

dbm-ui/backend/configuration/views/password_policy.py

@@ -28,6 +28,7 @@
     ModifyAdminPasswordSerializer,
     ModifyMySQLPasswordRandomCycleSerializer,
     PasswordPolicySerializer,
+    QueryAsyncModifyResultSerializer,
     VerifyPasswordResponseSerializer,
     VerifyPasswordSerializer,
 )
@@ -157,3 +158,13 @@ def modify_admin_password(self, request, *args, **kwargs):
         validated_data = self.params_validate(self.get_serializer_class())
         validated_data["operator"] = request.user.username
         return Response(DBPasswordHandler.modify_admin_password(**validated_data))
+
+    @common_swagger_auto_schema(
+        operation_summary=_("查询异步密码修改执行结果"),
+        request_body=QueryAsyncModifyResultSerializer(),
+        tags=[SWAGGER_TAG],
+    )
+    @action(methods=["POST"], detail=False, serializer_class=QueryAsyncModifyResultSerializer)
+    def query_async_modify_result(self, request, *args, **kwargs):
+        validated_data = self.params_validate(self.get_serializer_class())
+        return Response(DBPasswordHandler.query_async_modify_result(**validated_data))

dbm-ui/backend/flow/engine/bamboo/engine.py

@@ -101,6 +101,10 @@ def get_node_input_data(self, node_id: str) -> EngineAPIResult:
         result = api.get_execution_data_inputs(runtime=BambooDjangoRuntime(), node_id=node_id)
         return result
 
+    def get_node_output_data(self, node_id: str) -> EngineAPIResult:
+        result = api.get_execution_data_outputs(runtime=BambooDjangoRuntime(), node_id=node_id)
+        return result
+
     def get_node_histories(self, node_id: str) -> EngineAPIResult:
         result = api.get_node_histories(runtime=BambooDjangoRuntime(), node_id=node_id)
         return result

dbm-ui/backend/flow/engine/bamboo/scene/mysql/mysql_random_password.py

@@ -24,6 +24,8 @@
 
 def random_password_callback(params, data, kwargs, global_data):
     """密码随机化成功后的回调函数, 记录随机化成功和失败IP"""
+    if not global_data.get("uid"):
+        return
     flow = Ticket.objects.get(id=global_data["uid"]).current_flow()
     ticket_data = flow.details["ticket_data"]
     ticket_data.update(random_results=data)
@@ -55,6 +57,8 @@ def mysql_randomize_password(self):
                 "api_import_module": "DBPrivManagerApi",
                 "api_call_func": "modify_admin_password",
                 "success_callback_path": f"{random_password_callback.__module__}.{random_password_callback.__name__}",
+                "output": True,
+                "raw": True,
             },
         )
         mysql_authorize_rules.run_pipeline()

dbm-ui/backend/flow/plugins/components/collections/common/external_service.py

@@ -30,12 +30,14 @@ def _execute(self, data, parent_data):
         api_import_module: str = kwargs.get("api_import_module")
         api_call_func: str = kwargs.get("api_call_func")
         params: Dict[str, Any] = kwargs.get("params")
+        # 是否返回原始请求
+        raw = kwargs.get("raw", False)
 
         external_service: Callable = getattr(
             getattr(importlib.import_module(api_import_path), api_import_module), api_call_func
         )
         try:
-            resp = external_service(params)
+            resp = external_service(params, raw=raw)
             self.log_info(_("第三方接口: {} 请求成功! 返回参数为: {}").format(f"{api_import_path}.{api_call_func}", resp))
         except (ApiResultError, ApiRequestError) as e:
             self.log_info(_("第三方接口:{} 调用失败！错误信息为: {}").format(f"{api_import_path}.{api_call_func}", e))
@@ -47,11 +49,15 @@ def _execute(self, data, parent_data):
         # 成功时调用回调函数
         success_callback = kwargs.get("success_callback_path")
         if success_callback:
-            func_module, func_name = success_callback.rsplit(",", 1)
+            func_module, func_name = success_callback.rsplit(".", 1)
             getattr(importlib.import_module(func_module), func_name)(
                 params=params, data=resp, kwargs=kwargs, global_data=global_data
             )
 
+        # 存入返回响应data
+        if kwargs.get("output"):
+            data.outputs.resp = resp
+
         return True
 
 

dbm-ui/backend/ticket/builders/redis/redis_toolbox_cut_off.py

@@ -18,15 +18,19 @@
 from backend.db_services.dbbase.constants import IpSource
 from backend.flow.engine.controller.redis import RedisController
 from backend.ticket import builders
-from backend.ticket.builders.common.base import BaseOperateResourceParamBuilder, SkipToRepresentationMixin
+from backend.ticket.builders.common.base import (
+    BaseOperateResourceParamBuilder,
+    DisplayInfoSerializer,
+    SkipToRepresentationMixin,
+)
 from backend.ticket.builders.redis.base import BaseRedisTicketFlowBuilder, ClusterValidateMixin
 from backend.ticket.constants import TicketType
 
 
 class RedisClusterCutOffDetailSerializer(SkipToRepresentationMixin, ClusterValidateMixin, serializers.Serializer):
     """整机替换"""
 
-    class InfoSerializer(serializers.Serializer):
+    class InfoSerializer(DisplayInfoSerializer):
         class HostInfoSerializer(serializers.Serializer):
             ip = serializers.IPAddressField()
             spec_id = serializers.IntegerField()
@@ -97,13 +101,10 @@ class RedisClusterCutOffFlowBuilder(BaseRedisTicketFlowBuilder):
 
     def patch_ticket_detail(self):
         """redis_master -> backend_group"""
-
-        super().patch_ticket_detail()
-
-        resource_spec = {}
         cluster_ids = list(itertools.chain(*[infos["cluster_ids"] for infos in self.ticket.details["infos"]]))
         id__cluster = {cluster.id: cluster for cluster in Cluster.objects.filter(id__in=cluster_ids)}
         for info in self.ticket.details["infos"]:
+            resource_spec = {}
             # 取第一个cluster即可，即使是多集群，也是单机多实例的情况
             cluster = id__cluster[info["cluster_ids"][0]]
             for role in [
@@ -149,3 +150,4 @@ def patch_ticket_detail(self):
             info["resource_spec"] = resource_spec
 
         self.ticket.save(update_fields=["details"])
+        super().patch_ticket_detail()

dbm-ui/backend/ticket/constants.py

@@ -459,6 +459,7 @@ def get_cluster_type_by_ticket(cls, ticket_type):
 
     # 资源池
     RESOURCE_IMPORT = EnumField("RESOURCE_IMPORT", _("资源池导入"))
+    ADMIN_PASSWORD_MODIFY = EnumField("ADMIN_PASSWORD_MODIFY", _("临时密码修改"))
     # fmt: on
 
     # VM