feat(backend): mysql备份巡检 close #1449

dbm-ui/backend/db_periodic_task/local_tasks/mysql_backup/__init__.py

@@ -0,0 +1,11 @@
+# -*- coding: utf-8 -*-
+"""
+TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-DB管理系统(BlueKing-BK-DBM) available.
+Copyright (C) 2017-2023 THL A29 Limited, a Tencent company. All rights reserved.
+Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at https://opensource.org/licenses/MIT
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+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.
+"""
+from .task import mysql_backup_check_task

dbm-ui/backend/db_periodic_task/local_tasks/mysql_backup/bklog_query.py

@@ -0,0 +1,127 @@
+# -*- coding: utf-8 -*-
+"""
+TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-DB管理系统(BlueKing-BK-DBM) available.
+Copyright (C) 2017-2023 THL A29 Limited, a Tencent company. All rights reserved.
+Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at https://opensource.org/licenses/MIT
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+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 datetime
+import json
+from typing import Dict, List
+
+from backend import env
+from backend.components.bklog.client import BKLogApi
+from backend.utils.string import pascal_to_snake
+
+
+def _get_log_from_bklog(collector, start_time, end_time, query_string="*") -> List[Dict]:
+    """
+    从日志平台获取对应采集项的日志
+    @param collector: 采集项名称
+    @param start_time: 开始时间
+    @param end_time: 结束时间
+    @param query_string: 过滤条件
+    """
+    resp = BKLogApi.esquery_search(
+        {
+            "indices": f"{env.DBA_APP_BK_BIZ_ID}_bklog.{collector}",
+            "start_time": start_time,
+            "end_time": end_time,
+            # 这里需要精确查询集群域名，所以可以通过log: "key: \"value\""的格式查询
+            "query_string": query_string,
+            "start": 0,
+            "size": 6000,
+            "sort_list": [["dtEventTimeStamp", "asc"], ["gseIndex", "asc"], ["iterationIndex", "asc"]],
+        },
+        use_admin=True,
+    )
+    backup_logs = []
+    for hit in resp["hits"]["hits"]:
+        raw_log = json.loads(hit["_source"]["log"])
+        backup_logs.append({pascal_to_snake(key): value for key, value in raw_log.items()})
+
+    return backup_logs
+
+
+class ClusterBackup:
+    """
+    集群前一天备份信息，包括全备和binlog
+    """
+
+    def __init__(self, cluster_id: int, cluster_domain: str):
+        self.cluster_id = cluster_id
+        self.cluster_domain = cluster_domain
+        self.backups = {}
+        self.success = False
+
+    def query_backup_log_from_bklog(self, start_time: datetime.datetime, end_time: datetime.datetime) -> List[Dict]:
+        """
+        通过日志平台查询集群的时间范围内的全备备份记录
+        :param start_time: 开始时间
+        :param end_time: 结束时间
+        """
+        backup_files = []
+        backup_logs = _get_log_from_bklog(
+            collector="mysql_backup_result",
+            start_time=start_time.strftime("%Y-%m-%d %H:%M:%S"),
+            end_time=end_time.strftime("%Y-%m-%d %H:%M:%S"),
+            query_string=f'log: "cluster_id: {self.cluster_id}"',
+            # query_string=f'log: "cluster_address: \\"{self.cluster_domain}\\""',
+        )
+        for log in backup_logs:
+            bf = {
+                "backup_id": log["backup_id"],
+                "cluster_domain": log["cluster_address"],
+                "cluster_id": log["cluster_id"],
+                "task_id": log["task_id"],
+                "file_name": log["file_name"],
+                "file_size": log["file_size"],
+                "file_type": log["file_type"],
+                "mysql_host": log["mysql_host"],
+                "mysql_port": log["mysql_port"],
+                "mysql_role": log["mysql_role"],
+                "backup_type": log["backup_type"],
+                "data_schema_grant": log["data_schema_grant"],
+                "backup_begin_time": log["backup_begin_time"],
+                "backup_end_time": log["backup_end_time"],
+                "consistent_backup_time": log["consistent_backup_time"],
+                "shard_value": log["shard_value"],
+            }
+            backup_files.append(bf)
+        return backup_files
+
+    def query_binlog_from_bklog(self, start_time: datetime.datetime, end_time: datetime.datetime) -> List[Dict]:
+        """
+        通过日志平台查询集群的时间范围内的binlog 备份记录
+        :param cluster_domain: 集群
+        :param start_time: 开始时间
+        :param end_time: 结束时间
+        """
+        binlogs = []
+        backup_logs = _get_log_from_bklog(
+            collector="mysql_binlog_result",
+            start_time=start_time.strftime("%Y-%m-%d %H:%M:%S"),
+            end_time=end_time.strftime("%Y-%m-%d %H:%M:%S"),
+            query_string=f'log: "cluster_id: {self.cluster_id}"',
+            # query_string=f'log: "cluster_address: \\"{self.cluster_domain}\\""',
+        )
+        for log in backup_logs:
+            bl = {
+                "cluster_domain": log["cluster_domain"],
+                "cluster_id": log["cluster_id"],
+                "task_id": log["task_id"],
+                "file_name": log["filename"],  # file_name
+                "file_size": log["size"],
+                "file_mtime": log["file_mtime"],
+                "file_type": "binlog",
+                "mysql_host": log["host"],
+                "mysql_port": log["port"],
+                "mysql_role": log["db_role"],
+                "backup_status": log["backup_status"],
+                "backup_status_info": log["backup_status_info"],
+            }
+            binlogs.append(bl)
+        return binlogs

dbm-ui/backend/db_periodic_task/local_tasks/mysql_backup/check_binlog_backup.py

@@ -0,0 +1,95 @@
+# -*- coding: utf-8 -*-
+"""
+TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-DB管理系统(BlueKing-BK-DBM) available.
+Copyright (C) 2017-2023 THL A29 Limited, a Tencent company. All rights reserved.
+Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at https://opensource.org/licenses/MIT
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+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 datetime
+from collections import defaultdict
+
+from django.utils.translation import ugettext as _
+
+from backend.db_meta.enums import ClusterType, InstanceInnerRole, InstanceRole, MachineType
+from backend.db_meta.models import Cluster
+from backend.db_report.enums import MysqlBackupCheckSubType
+from backend.db_report.models import MysqlBackupCheckReport
+
+from .bklog_query import ClusterBackup
+
+
+def check_binlog_backup():
+    _check_tendbha_binlog_backup()
+    _check_tendbcluster_binlog_backup()
+
+
+def _check_tendbha_binlog_backup():
+    """
+    master 实例必须要有备份binlog
+    且binlog序号要连续
+    """
+    return _check_binlog_backup(ClusterType.TenDBHA)
+
+
+def _check_tendbcluster_binlog_backup():
+    """
+    master 实例必须要有备份binlog
+    且binlog序号要连续
+    """
+    return _check_binlog_backup(ClusterType.TenDBCluster)
+
+
+def _check_binlog_backup(cluster_type):
+    """
+    master 实例必须要有备份binlog
+    且binlog序号要连续
+    """
+    for c in Cluster.objects.filter(cluster_type=cluster_type):
+        backup = ClusterBackup(c.id, c.immute_domain)
+        now_time = datetime.datetime.now().date()
+        start_time = datetime.datetime.combine(now_time, datetime.time())
+        end_time = start_time + datetime.timedelta(hours=23, minutes=59, seconds=59)
+        items = backup.query_binlog_from_bklog(start_time, end_time)
+        instance_binlogs = defaultdict(list)
+        shard_binlog_stat = {}
+        for i in items:
+            instance = "{}:{}".format(i.get("mysql_host"), i.get("mysql_port"))
+            if i.get("mysql_role") == "master" and i.get("backup_status") == 4:
+                instance_binlogs[instance].append(i.get("file_name"))
+        backup.success = True
+
+        for inst, binlogs in instance_binlogs.items():
+            suffixes = [f.split(".", 1)[1] for f in binlogs]
+            shard_binlog_stat[inst] = is_consecutive_strings(suffixes)
+            if not shard_binlog_stat[inst]:
+                backup.success = False
+        if not backup.success:
+            MysqlBackupCheckReport.objects.create(
+                bk_biz_id=c.bk_biz_id,
+                bk_cloud_id=c.bk_cloud_id,
+                cluster=c.immute_domain,
+                cluster_type=ClusterType.TenDBCluster,
+                status=False,
+                msg="binlog is not consecutive:{}".format(shard_binlog_stat),
+                subtype=MysqlBackupCheckSubType.BinlogSeq.value,
+            )
+
+
+def is_consecutive_strings(str_list: list):
+    """
+    判断字符串数字是否连续
+    """
+    int_list = [int(s) for s in str_list]
+    int_sorted = list(set(int_list))
+    int_sorted.sort()
+    if len(int_sorted) == 0:
+        return False
+    if len(int_sorted) == 1:
+        return True
+    if len(int_sorted) == int_sorted[-1] - int_sorted[0] + 1:
+        return True
+    else:
+        return False

dbm-ui/backend/db_periodic_task/local_tasks/mysql_backup/check_full_backup.py

@@ -0,0 +1,152 @@
+# -*- coding: utf-8 -*-
+"""
+TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-DB管理系统(BlueKing-BK-DBM) available.
+Copyright (C) 2017-2023 THL A29 Limited, a Tencent company. All rights reserved.
+Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at https://opensource.org/licenses/MIT
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+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 datetime
+from collections import defaultdict
+
+from django.utils.translation import ugettext as _
+
+from backend.db_meta.enums import ClusterType, InstanceInnerRole, InstanceRole, MachineType
+from backend.db_meta.models import Cluster
+from backend.db_report.enums import MysqlBackupCheckSubType
+from backend.db_report.models import MysqlBackupCheckReport
+
+from .bklog_query import ClusterBackup
+
+
+def check_full_backup():
+    _check_tendbha_full_backup()
+    _check_tendbcluster_full_backup()
+
+
+class BackupFile:
+    def __init__(self, file_name: str, file_size: int, file_type=""):
+        self.file_name = file_name
+        self.file_size = file_size
+        self.file_type = file_type
+
+
+class MysqlBackup:
+    def __init__(self, cluster_id: int, cluster_domain: str, backup_id=""):
+        self.cluster_id = cluster_id
+        self.cluster_domain = cluster_domain
+        self.backup_id = backup_id
+        self.backup_type = ""
+        self.backup_role = ""
+        self.data_schema_grant = ""
+        self.consistent_backup_time = ""
+        self.shard_value = -1
+        # self.file_index = BackupFile()
+        # self.file_priv = BackupFile()
+        self.file_tar = []
+
+
+def _check_tendbha_full_backup():
+    """
+    tendbha 必须有一份完整的备份
+    """
+    for c in Cluster.objects.filter(cluster_type=ClusterType.TenDBHA):
+        backup = ClusterBackup(c.id, c.immute_domain)
+        now_time = datetime.datetime.now().date()
+        start_time = datetime.datetime.combine(now_time, datetime.time())
+        end_time = start_time + datetime.timedelta(hours=23, minutes=59, seconds=59)
+        items = backup.query_backup_log_from_bklog(start_time, end_time)
+        # print("cluster={} backup_items:{}".format(c.immute_domain, items))
+
+        for i in items:
+            if i.get("data_schema_grant", "") == "all":
+                bid = i.get("backup_id")
+                if bid not in backup.backups:
+                    backup.backups[bid] = MysqlBackup(i["cluster_id"], i["cluster_domain"], i["backup_id"])
+
+                bf = BackupFile(i.get("file_name"), i.get("file_size"))
+                backup.backups[bid].data_schema_grant = i.get("data_schema_grant")
+                if i.get("file_type") == "index":
+                    backup.backups[bid].file_index = bf
+                elif i.get("file_type") == "priv":
+                    backup.backups[bid].file_priv = bf
+                elif i.get("file_type") == "tar":
+                    backup.backups[bid].file_tar.append(bf)
+                else:
+                    pass
+        for bid, bk in backup.backups.items():
+            if bk.data_schema_grant == "all":
+                if bk.file_index and bk.file_priv and bk.file_tar:
+                    backup.success = True
+                    break
+        if not backup.success:
+            MysqlBackupCheckReport.objects.create(
+                bk_biz_id=c.bk_biz_id,
+                bk_cloud_id=c.bk_cloud_id,
+                cluster=c.immute_domain,
+                cluster_type=ClusterType.TenDBHA,
+                status=False,
+                msg="no success full backup found",
+                subtype=MysqlBackupCheckSubType.FullBackup.value,
+            )
+
+
+def _check_tendbcluster_full_backup():
+    """
+    tendbcluster 集群必须有完整的备份
+    """
+    for c in Cluster.objects.filter(cluster_type=ClusterType.TenDBCluster):
+        backup = ClusterBackup(c.id, c.immute_domain)
+        now_time = datetime.datetime.now().date()
+        start_time = datetime.datetime.combine(now_time, datetime.time())
+        end_time = start_time + datetime.timedelta(hours=23, minutes=59, seconds=59)
+        items = backup.query_backup_log_from_bklog(start_time, end_time)
+        # print("cluster={} backup_items:{}".format(c.immute_domain, items))
+
+        for i in items:
+            if i.get("data_schema_grant", "") == "all":
+                bid = "{}#{}".format(i.get("backup_id"), i.get("shard_value"))
+                if bid not in backup.backups:
+                    backup.backups[bid] = MysqlBackup(i["cluster_id"], i["cluster_domain"], i["backup_id"])
+
+                bf = BackupFile(i.get("file_name"), i.get("file_size"))
+                backup.backups[bid].data_schema_grant = i.get("data_schema_grant")
+                if i.get("file_type") == "index":
+                    backup.backups[bid].file_index = bf
+                elif i.get("file_type") == "priv":
+                    backup.backups[bid].file_priv = bf
+                elif i.get("file_type") == "tar":
+                    backup.backups[bid].file_tar.append(bf)
+                else:
+                    pass
+        backup_id_stat = defaultdict(list)
+        backup_id_invalid = {}
+        for bid, bk in backup.backups.items():
+            backup_id, shard_id = bid.split("#", 1)
+            if bk.data_schema_grant == "all":
+                if bk.file_index and bk.file_priv and bk.file_tar:
+                    #  这一个 shard ok
+                    backup_id_stat[backup_id].append({shard_id: True})
+                else:
+                    # 这一个 shard 不ok，整个backup_id 无效
+                    backup_id_invalid[backup_id] = True
+                    backup_id_stat[backup_id].append({shard_id: False})
+        message = ""
+        for backup_id, stat in backup_id_stat.items():
+            if backup_id not in backup_id_invalid:
+                backup.success = True
+                message = "shard_id:{}".format(backup_id_stat[backup_id])
+                break
+
+        if not backup.success:
+            MysqlBackupCheckReport.objects.create(
+                bk_biz_id=c.bk_biz_id,
+                bk_cloud_id=c.bk_cloud_id,
+                cluster=c.immute_domain,
+                cluster_type=ClusterType.TenDBCluster,
+                status=False,
+                msg="no success full backup found:{}".format(message),
+                subtype=MysqlBackupCheckSubType.FullBackup.value,
+            )