feat(backend) 增加查询数据库表接口 #1022

dbm-ui/backend/db_services/mysql/constants.py

@@ -17,11 +17,14 @@
 DEFAULT_ORIGIN_PROXY_PORT = 10000
 DEFAULT_ORIGIN_MYSQL_PORT = 20000
 
-# 查询库表sql语句
+# 闪回查询库表sql语句
 QUERY_SCHEMA_DBS_SQL = (
     "SELECT SCHEMA_NAME from information_schema.SCHEMATA WHERE {db_sts} and SCHEMA_NAME not in {sys_db_list}"
 )
 QUERY_SCHEMA_TABLES_SQL = (
     "SELECT TABLE_SCHEMA,TABLE_NAME FROM information_schema.TABLES "
     "WHERE TABLE_TYPE='BASE TABLE' AND (TABLE_SCHEMA IN {db_list}) AND {table_sts}"
 )
+
+# 根据库名查询表名的sql语句
+QUERY_TABLES_FROM_DB_SQL = "select table_schema, table_name from information_schema.tables where {db_sts}"

dbm-ui/backend/db_services/mysql/remote_service/handlers.py

@@ -8,17 +8,16 @@
 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 itertools
 from collections import defaultdict
 from itertools import chain
 from typing import Any, Dict, List, Union
 
 from django.utils.translation import ugettext as _
 
 from backend.components import DRSApi
-from backend.constants import IP_PORT_DIVIDER
 from backend.db_meta.api.cluster.base.handler import ClusterHandler
-from backend.db_services.mysql.constants import QUERY_SCHEMA_DBS_SQL, QUERY_SCHEMA_TABLES_SQL
-from backend.db_services.mysql.remote_service.exceptions import RemoteServiceBaseException
+from backend.db_services.mysql.constants import QUERY_SCHEMA_DBS_SQL, QUERY_SCHEMA_TABLES_SQL, QUERY_TABLES_FROM_DB_SQL
 from backend.flow.consts import SYSTEM_DBS
 
 
@@ -79,6 +78,36 @@ def show_databases(
                 )
         return cluster_databases
 
+    def show_tables(
+        self, cluster_db_infos: List[Dict], cluster_id__role_map: Dict[int, str] = None
+    ) -> List[Dict[str, Union[str, List]]]:
+        """
+        批量查询集群的数据库列表
+        @param cluster_db_infos: 集群DB信息
+        @param cluster_id__role_map: (可选)集群ID和对应查询库表角色的映射表
+        """
+        cluster_id__role_map = cluster_id__role_map or {}
+
+        cluster_table_infos: List[Dict[str, Union[str, List]]] = []
+        for info in cluster_db_infos:
+            cluster_handler, address = self._get_cluster_address(cluster_id__role_map, info["cluster_id"])
+
+            # 构造数据表查询语句
+            db_sts = " or ".join([f"table_schema='{db}'" for db in info["dbs"]])
+            query_table_sql = QUERY_TABLES_FROM_DB_SQL.format(db_sts=db_sts)
+
+            # 执行DRS，并聚合库所包含的表数据
+            bk_cloud_id = cluster_handler.cluster.bk_cloud_id
+            rpc_results = DRSApi.rpc({"bk_cloud_id": bk_cloud_id, "addresses": [address], "cmds": [query_table_sql]})
+            table_data = rpc_results[0]["cmd_results"][0]["table_data"]
+            aggregate_table_data: Dict[str, List[str]] = {db: [] for db in info["dbs"]}
+            for data in table_data:
+                aggregate_table_data[data["table_schema"]].append(data["table_name"])
+
+            cluster_table_infos.append({"cluster_id": cluster_handler.cluster_id, "table_data": aggregate_table_data})
+
+        return cluster_table_infos
+
     def check_cluster_database(self, check_infos: List[Dict[str, Any]]) -> List[Dict[str, Dict]]:
         """
         批量校验集群下的DB是否存在

dbm-ui/backend/db_services/mysql/remote_service/mock_data.py

@@ -16,6 +16,8 @@
     {"cluster_id": 2, "databases": ["db2", "db3"]},
 ]
 
+SHOW_TABLES_RESPONSE_DATA = [{"cluster_id": 1, "table_data": {"db1": [], "db2": [], "db3": ["test1"]}}]
+
 CHECK_CLUSTER_DATABASE_REQUEST_DATA = {"infos": [{"cluster_id": 1, "db_names": ["test1", "test2"]}]}
 
 CHECK_CLUSTER_DATABASE_RESPONSE_DATA = [
@@ -29,15 +31,15 @@
         "databases_ignore": [],
         "tables": [],
         "tables_ignore": [],
-        "message": "不存在可用于闪回的库",
+        "message": "this is a error message",
     },
     {
         "cluster_id": 63,
         "databases": [],
         "databases_ignore": [],
         "tables": ["iijkk"],
         "tables_ignore": [],
-        "message": "不存在可用于闪回的表",
+        "message": "this is a error message",
     },
     {"cluster_id": 63, "databases": [], "databases_ignore": [], "tables": [], "tables_ignore": [], "message": ""},
 ]

dbm-ui/backend/db_services/mysql/remote_service/serializers.py

@@ -18,8 +18,10 @@
     FLASHBACK_CHECK_DATA,
     SHOW_DATABASES_REQUEST_DATA,
     SHOW_DATABASES_RESPONSE_DATA,
+    SHOW_TABLES_RESPONSE_DATA,
 )
 from backend.flow.consts import TenDBBackUpLocation
+from backend.ticket.builders.mysql.base import DBTableField
 
 
 class ShowDatabasesRequestSerializer(serializers.Serializer):
@@ -39,6 +41,19 @@ class Meta:
         swagger_schema_fields = {"example": SHOW_DATABASES_REQUEST_DATA}
 
 
+class ShowTablesRequestSerializer(serializers.Serializer):
+    class DbInfoSerializer(serializers.Serializer):
+        cluster_id = serializers.IntegerField(help_text=_("集群ID列表"))
+        dbs = serializers.ListField(help_text=_("查询的DB列表"), child=DBTableField(db_field=True))
+
+    cluster_db_infos = serializers.ListSerializer(help_text=_("集群数据库信息"), child=DbInfoSerializer())
+
+
+class ShowTablesResponseSerializer(serializers.Serializer):
+    class Meta:
+        swagger_schema_fields = {"example": SHOW_TABLES_RESPONSE_DATA}
+
+
 class ShowDatabasesResponseSerializer(serializers.Serializer):
     class Meta:
         swagger_schema_fields = {"example": SHOW_DATABASES_RESPONSE_DATA}
@@ -63,10 +78,10 @@ class Meta:
 class CheckFlashbackInfoSerializer(serializers.Serializer):
     class FlashbackSerializer(serializers.Serializer):
         cluster_id = serializers.IntegerField(help_text=_("集群ID"))
-        databases = serializers.ListField(help_text=_("目标库列表"), child=serializers.CharField())
-        databases_ignore = serializers.ListField(help_text=_("忽略库列表"), child=serializers.CharField())
-        tables = serializers.ListField(help_text=_("目标table列表"), child=serializers.CharField())
-        tables_ignore = serializers.ListField(help_text=_("忽略table列表"), child=serializers.CharField())
+        databases = serializers.ListField(help_text=_("目标库列表"), child=DBTableField(db_field=True))
+        databases_ignore = serializers.ListField(help_text=_("忽略库列表"), child=DBTableField(db_field=True))
+        tables = serializers.ListField(help_text=_("目标table列表"), child=DBTableField())
+        tables_ignore = serializers.ListField(help_text=_("忽略table列表"), child=DBTableField())
 
     infos = serializers.ListSerializer(help_text=_("flashback信息"), child=FlashbackSerializer(), allow_empty=False)
 

dbm-ui/backend/db_services/mysql/remote_service/views.py

@@ -23,6 +23,8 @@
     CheckFlashbackInfoSerializer,
     ShowDatabasesRequestSerializer,
     ShowDatabasesResponseSerializer,
+    ShowTablesRequestSerializer,
+    ShowTablesResponseSerializer,
 )
 from backend.iam_app.handlers.drf_perm import DBManageIAMPermission
 
@@ -59,6 +61,19 @@ def show_cluster_databases(self, request, bk_biz_id):
             ),
         )
 
+    @common_swagger_auto_schema(
+        operation_summary=_("查询集群数据表列表"),
+        request_body=ShowTablesRequestSerializer(),
+        tags=[SWAGGER_TAG],
+        responses={status.HTTP_200_OK: ShowTablesResponseSerializer()},
+    )
+    @action(methods=["POST"], detail=False, serializer_class=ShowTablesRequestSerializer)
+    def show_cluster_tables(self, request, bk_biz_id):
+        validated_data = self.params_validate(self.get_serializer_class())
+        return Response(
+            RemoteServiceHandler(bk_biz_id=bk_biz_id).show_tables(cluster_db_infos=validated_data["cluster_db_infos"])
+        )
+
     @common_swagger_auto_schema(
         operation_summary=_("校验DB是否在集群内"),
         request_body=CheckClusterDatabaseSerializer(),