fix: pulsar/doris shrink ip choose TencentBlueKing#7469

dbm-ui/backend/flow/engine/bamboo/scene/doris/doris_base_flow.py

@@ -384,8 +384,16 @@ def get_all_node_ips_in_ticket(data: dict) -> list:
 
 
 def fe_exists_in_ticket(data: dict) -> bool:
-    return DorisRoleEnum.FOLLOWER in data["nodes"] or DorisRoleEnum.OBSERVER in data["nodes"]
+    return role_exists_in_ticket(data=data, role=DorisRoleEnum.FOLLOWER) or role_exists_in_ticket(
+        data=data, role=DorisRoleEnum.OBSERVER
+    )
 
 
 def be_exists_in_ticket(data: dict) -> bool:
-    return DorisRoleEnum.HOT in data["nodes"] or DorisRoleEnum.COLD in data["nodes"]
+    return role_exists_in_ticket(data=data, role=DorisRoleEnum.HOT) or role_exists_in_ticket(
+        data=data, role=DorisRoleEnum.COLD
+    )
+
+
+def role_exists_in_ticket(data: dict, role: DorisRoleEnum) -> bool:
+    return role in data["nodes"] and data["nodes"][role]

dbm-ui/backend/flow/engine/bamboo/scene/doris/doris_shrink_flow.py

@@ -27,7 +27,9 @@
     get_node_ips_in_ticket_by_role,
 )
 from backend.flow.engine.bamboo.scene.doris.exceptions import (
+    BeMachineCountException,
     FollowerScaleUpUnsupportedException,
+    NoShrinkMachineException,
     RoleMachineCountException,
     RoleMachineCountMustException,
 )
@@ -39,7 +41,11 @@
 from backend.flow.plugins.components.collections.doris.get_doris_payload import GetDorisActPayloadComponent
 from backend.flow.plugins.components.collections.doris.get_doris_resource import GetDorisResourceComponent
 from backend.flow.plugins.components.collections.es.trans_files import TransFileComponent
-from backend.flow.utils.doris.consts import DORIS_FOLLOWER_MUST_COUNT, DORIS_OBSERVER_NOT_COUNT
+from backend.flow.utils.doris.consts import (
+    DORIS_BACKEND_NOT_COUNT,
+    DORIS_FOLLOWER_MUST_COUNT,
+    DORIS_OBSERVER_NOT_COUNT,
+)
 from backend.flow.utils.doris.doris_act_payload import DorisActPayload
 from backend.flow.utils.doris.doris_context_dataclass import DnsKwargs, DorisActKwargs, DorisApplyContext
 
@@ -151,12 +157,26 @@ def shrink_doris_flow(self):
         doris_pipeline.run_pipeline()
 
     def check_shrink_role_ip_count(self, data: dict):
-        # 扩容无需检查数据节点数量
+        # 缩容 检查各个角色节点传参
+        if not fe_exists_in_ticket(data=data) and not be_exists_in_ticket(data=data):
+            logger.error(_("Doris缩容未选择机器"))
+            raise NoShrinkMachineException()
+
+        former_cold_cnt = len(self.get_role_ips_in_dbmeta(InstanceRole.DORIS_BACKEND_COLD))
+        del_cold_cnt = len(get_node_ips_in_ticket_by_role(data, DorisRoleEnum.COLD))
+
+        former_hot_cnt = len(self.get_role_ips_in_dbmeta(InstanceRole.DORIS_BACKEND_HOT))
+        del_hot_cnt = len(get_node_ips_in_ticket_by_role(data, DorisRoleEnum.HOT))
+
+        # 检查 所有数据节点 剩余数量不能为0
+        if former_cold_cnt + former_hot_cnt - del_cold_cnt - del_hot_cnt == DORIS_BACKEND_NOT_COUNT:
+            logger.error(_("Doris 缩容后数据节点数量不能为{}".format(DORIS_BACKEND_NOT_COUNT)))
+            raise BeMachineCountException(must_count=DORIS_BACKEND_NOT_COUNT)
 
         # 检查 follower 数量
         follower_count = len(get_node_ips_in_ticket_by_role(data, DorisRoleEnum.FOLLOWER))
         if follower_count > 0:
-            logger.error(_("DorisFollower不支持缩容,当前选择缩容机器数量为{}".format(follower_count)))
+            logger.error(_("Doris Follower不支持缩容,当前选择缩容机器数量为{}".format(follower_count)))
             raise FollowerScaleUpUnsupportedException(machine_count=follower_count)
 
         # 检查 observer 数量

dbm-ui/backend/flow/engine/bamboo/scene/doris/exceptions.py

@@ -52,3 +52,9 @@ class ReplaceMachineCountException(DorisFlowBaseException):
     ERROR_CODE = "006"
     MESSAGE = _("Doris替换机器数量不一致")
     MESSAGE_TPL = _("Doris集群替换角色{doris_role}数量不一致,已选旧节点个数{old_count},新节点个数{new_count}")
+
+
+class NoShrinkMachineException(DorisFlowBaseException):
+    ERROR_CODE = "007"
+    MESSAGE = _("机器数量异常")
+    MESSAGE_TPL = _("Doris缩容选择机器不能为空")

dbm-ui/backend/flow/engine/bamboo/scene/pulsar/pulsar_sub_flow.py

@@ -268,6 +268,8 @@ def del_bookkeeper_sub_flow(self, act_kwargs: PulsarActKwargs, data: dict, cur_b
 
         # 停止BookKeeper服务 + 退役BookKeeper
         for bookie_ip in data["shrink_bookie_ips"]:
+            # 每个bookie 串行 进行停服和退役操作
+            act_kwargs.exec_ip = bookie_ip
             act_kwargs.get_pulsar_payload_func = PulsarActPayload.get_stop_process_payload.__name__
             act_kwargs.role = PulsarRoleEnum.BookKeeper
             sub_pipeline.add_act(

dbm-ui/backend/flow/urls.py

@@ -160,6 +160,7 @@
 from backend.flow.views.pulsar_reboot import RebootPulsarSceneApiView
 from backend.flow.views.pulsar_replace import ReplacePulsarSceneApiView
 from backend.flow.views.pulsar_scale_up import ScaleUpPulsarSceneApiView
+from backend.flow.views.pulsar_shrink import ShrinkPulsarSceneApiView
 from backend.flow.views.redis_cluster import (
     InstallPredixyClusterSceneApiView,
     InstallRedisInstanceSceneApiView,
@@ -442,6 +443,7 @@
     url(r"^scene/destroy_pulsar$", DestroyPulsarSceneApiView.as_view()),
     url(r"^scene/reboot_pulsar$", RebootPulsarSceneApiView.as_view()),
     url(r"^scene/replace_pulsar$", ReplacePulsarSceneApiView.as_view()),
+    url(r"^scene/shrink_pulsar$", ShrinkPulsarSceneApiView.as_view()),
     url(r"^scene/fake_install_pulsar$", FakeInstallPulsarSceneApiView.as_view()),
     url(r"^scene/import_resource_init$", ImportResourceInitStepApiView.as_view()),
     url("^scene/mysql_data_migrate$", MysqlDataMigrateSceneApiView.as_view()),

dbm-ui/backend/flow/views/pulsar_shrink.py

@@ -12,15 +12,15 @@
 from django.utils.translation import ugettext as _
 from rest_framework.permissions import AllowAny
 from rest_framework.response import Response
-from rest_framework.views import APIView
 
 from backend.flow.engine.controller.pulsar import PulsarController
+from backend.flow.views.base import FlowTestView
 from backend.utils.basic import generate_root_id
 
 logger = logging.getLogger("root")
 
 
-class ShrinkPulsarSceneApiView(APIView):
+class ShrinkPulsarSceneApiView(FlowTestView):
 
     permission_classes = [AllowAny]