refactor(backend): 整包更新 #10440

# Reviewed, transaction id: 41910

Author: iSecloud

dbm-ui/backend/components/bklog/handler.py

@@ -55,7 +55,7 @@ def query_logs(
                     "end_time": datetime2str(end_time),
                     "query_string": query_string,
                     "start": 0,
-                    "size": 1000,
+                    "size": 10000,
                     "sort_list": [
                         ["dtEventTimeStamp", sorting_rule],
                         ["gseIndex", sorting_rule],

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

@@ -278,9 +278,9 @@ def _retry(self) -> Any:
         self.run()
 
     def _revoke(self, operator) -> Any:
-        # 停止相关联的todo
         from backend.ticket.todos import TodoActionType, TodoActorFactory
 
+        # 停止相关联的todo
         todos = Todo.objects.filter(ticket=self.ticket, flow=self.flow_obj, status=TodoStatus.TODO)
         for todo in todos:
             TodoActorFactory.actor(todo).process(operator, TodoActionType.TERMINATE, params={})

dbm-ui/backend/ticket/tasks/ticket_tasks.py

@@ -8,7 +8,6 @@
 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 json
 import logging
 import operator
 from collections import defaultdict
@@ -22,12 +21,11 @@
 from django.utils import timezone
 from django.utils.translation import gettext as _
 
-from backend import env
-from backend.components import BKLogApi
+from backend.components.bklog.handler import BKLogHandler
 from backend.configuration.constants import PLAT_BIZ_ID, DBType
 from backend.constants import DEFAULT_SYSTEM_USER
 from backend.core import notify
-from backend.db_meta.enums import ClusterType, InstanceInnerRole
+from backend.db_meta.enums import ClusterType
 from backend.db_meta.models import Cluster, StorageInstance
 from backend.ticket.builders.common.constants import MYSQL_CHECKSUM_TABLE, MySQLDataRepairTriggerMode
 from backend.ticket.constants import (
@@ -43,7 +41,7 @@
 )
 from backend.ticket.exceptions import TicketTaskTriggerException
 from backend.ticket.models.ticket import Flow, Ticket, TicketFlowsConfig
-from backend.utils.time import date2str, datetime2str
+from backend.utils.time import date2str
 
 logger = logging.getLogger("root")
 
@@ -86,24 +84,16 @@ def auto_create_data_repair_ticket(cls):
         # 例行时间校验默认间隔一天
         now = datetime.now(timezone.utc).astimezone()
         start_time, end_time = now - timedelta(days=1), now
-        # TODO: 目前这个esquery_search最多支持10000条查询，后续可以改造成scroll进行查询
-        resp = BKLogApi.esquery_search(
-            {
-                "indices": f"{env.DBA_APP_BK_BIZ_ID}_bklog.mysql_checksum_result",
-                "start_time": datetime2str(start_time),
-                "end_time": datetime2str(end_time),
-                "query_string": "*",
-                "start": 0,
-                "size": 10000,
-                "sort_list": [["dtEventTimeStamp", "asc"], ["gseIndex", "asc"], ["iterationIndex", "asc"]],
-            }
-        )
 
-        # 根据集群ID聚合日志
+        total_checksum_logs = BKLogHandler.query_logs(
+            collector="mysql_checksum_result", start_time=start_time, end_time=end_time, query_string="*", size=-1
+        )
+        # 根据集群ID聚合日志，提前跳过校验一致的log
         cluster__checksum_logs_map: Dict[int, List[Dict]] = defaultdict(list)
-        for hit in resp["hits"]["hits"]:
-            checksum_log = json.loads(hit["_source"]["log"])
-            cluster__checksum_logs_map[checksum_log["cluster_id"]].append(checksum_log)
+        for log in total_checksum_logs:
+            if log["master_crc"] == log["this_crc"] and log["master_cnt"] == log["this_cnt"]:
+                continue
+            cluster__checksum_logs_map[log["cluster_id"]].append(log)
 
         cluster_map = {c.id: c for c in Cluster.objects.filter(id__in=list(cluster__checksum_logs_map.keys()))}
         biz__db_type__repair_infos: Dict[int, Dict[DBType, List]] = defaultdict(lambda: defaultdict(list))
@@ -116,29 +106,18 @@ def auto_create_data_repair_ticket(cls):
                 continue
 
             cluster = cluster_map[cluster_id]
+            logger.info(_("为集群{}生成修复单据信息".format(cluster.immute_domain)))
 
-            # 根据logs获取ip:port和实例的映射
-            inst_filter_list = [
-                (
-                    Q(
-                        cluster=cluster,
-                        machine__ip=log["ip"],
-                        port=log["port"],
-                        instance_inner_role=InstanceInnerRole.SLAVE,
-                    )
-                    | Q(
-                        cluster=cluster,
-                        machine__ip=log["master_ip"],
-                        port=log["master_port"],
-                        instance_inner_role=InstanceInnerRole.MASTER,
-                    )
-                )
-                for log in checksum_logs
-            ]
-            inst_filters = reduce(operator.or_, inst_filter_list)
+            # 获取logs中的ip:port实例
+            inst_filter_list = []
+            for log in checksum_logs:
+                inst_filter_list.append(f"{log['ip']}:{log['port']}")
+                inst_filter_list.append(f"{log['master_ip']}:{log['master_port']}")
+            # 过滤需要进行修复的实例
             ip_port__instance_id_map: Dict[str, StorageInstance] = {
                 f"{inst.machine.ip}:{inst.port}": inst
-                for inst in StorageInstance.objects.select_related("machine").filter(inst_filters)
+                for inst in StorageInstance.objects.select_related("machine").filter(cluster=cluster_id)
+                if f"{inst.machine.ip}:{inst.port}" in inst_filter_list
             }
 
             data_repair_infos: List[Dict[str, Any]] = []

dbm-ui/backend/utils/batch_request.py

@@ -10,7 +10,6 @@
 """
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from copy import deepcopy
-from multiprocessing.pool import ThreadPool
 from typing import Callable
 
 import wrapt
@@ -99,28 +98,22 @@ def batch_request(
         final_request_params = format_params(params, get_count, func, start_key, limit_key)
 
     data = []
-
-    # 根据请求总数并发请求
-    pool = ThreadPool(settings.CONCURRENT_NUMBER)
     futures = []
 
-    for req in final_request_params:
-        start = 0
-        while start < req["count"]:
-            request_params = {"page": {limit_key: limit, start_key: start}}
-            if sort:
-                request_params["page"]["sort"] = sort
-            request_params.update(req["params"])
-            futures.append(pool.apply_async(inject_request(func), args=(request_params,), kwds=kwargs))
-
-            start += limit
-
-    pool.close()
-    pool.join()
+    with ThreadPoolExecutor(max_workers=settings.CONCURRENT_NUMBER) as ex:
+        for req in final_request_params:
+            start = 0
+            while start < req["count"]:
+                request_params = {"page": {limit_key: limit, start_key: start}}
+                if sort:
+                    request_params["page"]["sort"] = sort
+                request_params.update(req["params"])
+                # 提交任务到线程池
+                futures.append(ex.submit(inject_request(func), params=request_params, **kwargs))
+                start += limit
 
-    # 取值
     for future in futures:
-        data.extend(get_data(future.get()))
+        data.extend(get_data(future.result()))
 
     return data
 

helm-charts/bk-dbm/Chart.yaml

@@ -99,5 +99,5 @@ dependencies:
 description: A Helm chart for bk-dbm
 name: bk-dbm
 type: application
-version: 1.5.0-alpha.51
-appVersion: 1.5.0-alpha.51
+version: 1.5.0-alpha.52
+appVersion: 1.5.0-alpha.52

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

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

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

@@ -1,5 +1,5 @@
 apiVersion: v2
-appVersion: 0.0.1-alpha.102
+appVersion: 0.0.1-alpha.104
 description: A Helm chart for Kubernetes
 name: db-simulation
 type: application

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

@@ -1,5 +1,5 @@
 apiVersion: v2
-appVersion: 1.5.0-alpha.2185
+appVersion: 1.5.0-alpha.2262
 description: A Helm chart for dbm
 name: dbm
 type: application

helm-charts/bk-dbm/templates/configmaps/dbm-configmap.yaml

@@ -92,5 +92,5 @@ data:
   BK_APIGW_RESOURCE_DOCS_ARCHIVE_FILE: "{{ .Values.bk.bkApigwResourceDocsArchiveFile }}"
   BK_APIGATEWAY_DOMAIN : "{{ .Values.bk.bkApigatewayDomain | default .Values.bk.bkComponentApiUrl }}"
   DBM_STATSDB_URL: "{{ $dbmStatsDB.host }}:{{ $dbmStatsDB.port }}"
-  DBM_STATSDB_USER: "{{ $dbmStatsDB.username }}"
+  DBM_STATSDB_USER: "{{ $dbmStatsDB.user }}"
   DBM_STATSDB_PASSWORD: "{{ $dbmStatsDB.password }}"