fix(backend): cmdb dbm_meta 字段支持压缩，解决多实例监控问题 #3529

dbm-ui/backend/configuration/handlers/password.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 base64
 from collections import defaultdict
 from typing import Any, Dict, List
 
@@ -24,6 +23,7 @@
 from backend.db_periodic_task.models import DBPeriodicTask
 from backend.db_services.ipchooser.query.resource import ResourceQueryHelper
 from backend.flow.consts import MySQLPasswordRole
+from backend.utils.string import base64_decode, base64_encode
 
 
 class DBPasswordHandler(object):
@@ -40,7 +40,7 @@ def verify_password_strength(cls, password: str, echo: bool = False):
             name=AsymmetricCipherConfigType.PASSWORD.value, content=password, salted=False
         )
         # 密码需要用base64加密后传输
-        b64_plain_password = base64.b64encode(plain_password.encode("utf-8")).decode("utf-8")
+        b64_plain_password = base64_encode(plain_password)
         check_result = MySQLPrivManagerApi.check_password(
             {"password": b64_plain_password, "security_rule_name": DBM_PASSWORD_SECURITY_NAME}
         )
@@ -84,7 +84,7 @@ def query_mysql_admin_password(
         mysql_admin_password_data["results"] = mysql_admin_password_data.pop("items")
         cloud_info = ResourceQueryHelper.search_cc_cloud(get_cache=True)
         for data in mysql_admin_password_data["results"]:
-            data["password"] = base64.b64decode(data["password"]).decode("utf-8")
+            data["password"] = base64_decode(data["password"])
             data["bk_cloud_name"] = cloud_info[str(data["bk_cloud_id"])]["bk_cloud_name"]
 
         return mysql_admin_password_data
@@ -120,7 +120,7 @@ def modify_mysql_admin_password(cls, operator: str, password: str, lock_hour: in
         modify_password_params = {
             "username": DBM_MYSQL_ADMIN_USER,
             "component": DBType.MySQL.value,
-            "password": base64.b64encode(password.encode("utf-8")).decode("utf-8"),
+            "password": base64_encode(password),
             "lock_hour": lock_hour,
             "operator": operator,
             "clusters": cluster_infos,
@@ -176,4 +176,4 @@ def query_proxy_password(cls):
         }
         data = MySQLPrivManagerApi.get_password(params)["items"][0]
         # 注意要用base64解密
-        return base64.b64decode(data["password"]).decode("utf8")
+        return base64_decode(data["password"])

dbm-ui/backend/core/encrypt/aes.py

@@ -13,6 +13,7 @@
 from Crypto.Cipher import AES
 
 from backend.core.encrypt.constants import AES_BLOCK_SIZE, AES_PADDING
+from backend.utils.string import base64_encode
 
 
 def pad_it(data):
@@ -38,7 +39,7 @@ def encrypt(data: str, aes_key: str) -> str:
     aes_key = aes_key.encode("utf-8")
     cipher = AES.new(aes_key, AES.MODE_CBC, aes_key)
     data = cipher.encrypt(pad_it(data).encode("utf-8"))
-    return base64.b64encode(data).decode("utf-8")
+    return base64_encode(data)
 
 
 def decrypt(data: str, aes_key: str) -> str:

dbm-ui/backend/db_meta/api/proxy_instance/apis.py

@@ -77,41 +77,3 @@ def update(proxies):
 
         proxy_obj.status = new_status
         proxy_obj.save()
-
-
-# @transaction.atomic
-# def decommission(instances: List[Dict]):
-#     """
-#     TODO：没使用到？待删除
-#     1. 仅支持 下架实例不在任何一个集群
-#     必要条件：
-#         1. 不属于任何一个集群 ;属于集群的实例，需要走集群内下架接口
-#
-#     场景：
-#         1. 上架了，但未添加到集群
-#         2. 从集群内清理掉了 ；调用了 delete_proxies()
-#     """
-#     logger.info("user request decmmission instances {}".format(instances))
-#     proxy_objs = common.filter_out_instance_obj(instances, ProxyInstance.objects.all())
-#
-#     _t = common.in_another_cluster(proxy_objs)
-#     if _t:
-#         raise Exception(_("proxy {} 在集群里边").format(_t))
-#
-#     _t = common.not_exists(instances, ProxyInstance.objects.all())
-#     if _t:
-#         raise Exception(_("proxy {} 不存在").format(_t))
-#
-#     for proxy_obj in proxy_objs:
-#         logger.info("remove proxy {} ".format(proxy_obj))
-#         CcManage(proxy_obj.bk_biz_id).delete_service_instance(bk_instance_ids=[proxy_obj.bk_instance_id])
-#
-#         # 需要检查， 是否该机器上所有实例都已经清理干净，
-#         if len(ProxyInstance.objects.filter(machine__ip=proxy_obj.machine.ip).all()) > 0:
-#             logger.info("ignore storage machine {} , another instance existed.".format(proxy_obj.machine))
-#         else:
-#             logger.info("proxy machine {}".format(proxy_obj.machine))
-#             CcManage(
-#                 proxy_obj.bk_biz_id,
-#             ).recycle_host([proxy_obj.machine.bk_host_id])
-#             proxy_obj.machine.delete()

dbm-ui/backend/db_meta/models/machine.py

@@ -8,7 +8,9 @@
 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 copy
+import gzip
+import io
+import json
 from dataclasses import asdict
 
 from django.db import models
@@ -21,6 +23,7 @@
 from backend.db_meta.enums import AccessLayer, ClusterType, MachineType
 from backend.db_meta.exceptions import HostDoseNotExistInCmdbException
 from backend.db_meta.models import AppCache, BKCity
+from backend.utils.string import base64_encode
 
 
 class Machine(AuditedModel):
@@ -56,95 +59,81 @@ def __str__(self):
         return self.ip
 
     @property
-    def dbm_meta(self) -> list:
+    def dbm_meta(self) -> dict:
         proxies = self.proxyinstance_set.all()
         storages = self.storageinstance_set.all()
 
         host_labels = []
 
-        def shrink_dbm_meta(dbm_meta):
-            """数据裁剪"""
-
-            if not dbm_meta:
-                return []
-
-            # 剔除实例属性，仅保留集群属性
-            first_one = copy.deepcopy(dbm_meta[0])
-            for custom_attr in ["instance_role", "instance_port"]:
-                first_one.pop(custom_attr)
-
-            return {
-                "version": "v1",
-                "common": first_one,
-                "custom": list(
-                    map(lambda x: {"instance_role": x["instance_role"], "instance_port": x["instance_port"]}, dbm_meta)
-                ),
-            }
-
-        def remove_duplicates(seq):
-            unique = set()
-            for d in seq:
-                t = tuple(d.items())
-                unique.add(t)
-
-            return shrink_dbm_meta([dict(x) for x in unique])
-
-        if proxies:
-            for proxy in proxies:
-                for cluster in proxy.cluster.all():
-                    tendb_cluster_spider_ext = getattr(proxy, "tendbclusterspiderext", None)
-                    host_labels.append(
-                        asdict(
-                            CommonHostDBMeta(
-                                app=AppCache.get_app_attr(cluster.bk_biz_id, default=cluster.bk_biz_id),
-                                appid=str(cluster.bk_biz_id),
-                                cluster_type=cluster.cluster_type,
-                                cluster_domain=cluster.immute_domain,
-                                db_type=ClusterType.cluster_type_to_db_type(cluster.cluster_type),
-                                # tendbcluster中扩展了proxy的类型，需要特殊处理
-                                instance_role=tendb_cluster_spider_ext.spider_role
-                                if tendb_cluster_spider_ext
-                                else "proxy",
-                                instance_port=str(proxy.port),
-                            )
+        def compress_dbm_meta_content(dbm_meta: dict) -> str:
+            """
+            压缩 dbm_meta
+            """
+            # 使用gzip压缩
+            # python3.6 gzip 不支持 mtime 参数，python3.10 可以直接使用 gzip.compress 压缩
+            buf = io.BytesIO()
+            with gzip.GzipFile(fileobj=buf, mode="wb", mtime=0) as f:
+                f.write(json.dumps(dbm_meta).encode("utf-8"))
+            compressed_data = buf.getvalue()
+
+            # 将压缩后的字节转换为Base64编码的字符串
+            base64_encoded_str = base64_encode(compressed_data)
+            return base64_encoded_str
+
+        for proxy in proxies:
+            for cluster in proxy.cluster.all():
+                tendb_cluster_spider_ext = getattr(proxy, "tendbclusterspiderext", None)
+                host_labels.append(
+                    asdict(
+                        CommonHostDBMeta(
+                            app=AppCache.get_app_attr(cluster.bk_biz_id, default=cluster.bk_biz_id),
+                            appid=str(cluster.bk_biz_id),
+                            cluster_type=cluster.cluster_type,
+                            cluster_domain=cluster.immute_domain,
+                            db_type=ClusterType.cluster_type_to_db_type(cluster.cluster_type),
+                            # tendbcluster中扩展了proxy的类型，需要特殊处理
+                            instance_role=tendb_cluster_spider_ext.spider_role
+                            if tendb_cluster_spider_ext
+                            else "proxy",
+                            instance_port=str(proxy.port),
                         )
                     )
-
-        if storages:
-            for storage in storages:
-                # influxdb需要单独处理
-                if storage.cluster_type == ClusterType.Influxdb.value:
-                    host_labels.append(
-                        asdict(
-                            CommonHostDBMeta(
-                                app=AppCache.get_app_attr(storage.bk_biz_id, default=storage.bk_biz_id),
-                                appid=str(storage.bk_biz_id),
-                                cluster_domain=storage.machine.ip,
-                                cluster_type=storage.cluster_type,
-                                db_type=ClusterType.cluster_type_to_db_type(storage.cluster_type),
-                                instance_role=storage.instance_role,
-                                instance_port=str(storage.port),
-                            )
+                )
+
+        for storage in storages:
+            # influxdb需要单独处理
+            if storage.cluster_type == ClusterType.Influxdb.value:
+                host_labels.append(
+                    asdict(
+                        CommonHostDBMeta(
+                            app=AppCache.get_app_attr(storage.bk_biz_id, default=storage.bk_biz_id),
+                            appid=str(storage.bk_biz_id),
+                            cluster_domain=storage.machine.ip,
+                            cluster_type=storage.cluster_type,
+                            db_type=ClusterType.cluster_type_to_db_type(storage.cluster_type),
+                            instance_role=storage.instance_role,
+                            instance_port=str(storage.port),
                         )
                     )
-                    continue
-
-                for cluster in storage.cluster.all():
-                    host_labels.append(
-                        asdict(
-                            CommonHostDBMeta(
-                                app=AppCache.get_app_attr(cluster.bk_biz_id, default=cluster.bk_biz_id),
-                                appid=str(cluster.bk_biz_id),
-                                cluster_domain=cluster.immute_domain,
-                                cluster_type=cluster.cluster_type,
-                                db_type=ClusterType.cluster_type_to_db_type(cluster.cluster_type),
-                                instance_role=storage.instance_role,
-                                instance_port=str(storage.port),
-                            )
+                )
+                continue
+
+            for cluster in storage.cluster.all():
+                host_labels.append(
+                    asdict(
+                        CommonHostDBMeta(
+                            app=AppCache.get_app_attr(cluster.bk_biz_id, default=cluster.bk_biz_id),
+                            appid=str(cluster.bk_biz_id),
+                            cluster_domain=cluster.immute_domain,
+                            cluster_type=cluster.cluster_type,
+                            db_type=ClusterType.cluster_type_to_db_type(cluster.cluster_type),
+                            instance_role=storage.instance_role,
+                            instance_port=str(storage.port),
                         )
                     )
+                )
 
-        return remove_duplicates(host_labels)
+        return {"version": "v2", "content": compress_dbm_meta_content({"common": {}, "custom": host_labels})}
 
     @classmethod
     def get_host_info_from_cmdb(cls, bk_host_id: int) -> dict:

dbm-ui/backend/db_meta/utils.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 base64
 import logging
 import os.path
 from collections import defaultdict
@@ -34,6 +33,7 @@
 from backend.db_services.ipchooser.constants import DB_MANAGE_SET
 from backend.db_services.ipchooser.query import resource
 from backend.flow.utils.cc_manage import CcManage
+from backend.utils.string import base64_encode
 
 logger = logging.getLogger("root")
 
@@ -69,7 +69,7 @@ def remove_cluster(cluster_id, job_clean=True, cc_clean=True):
             JobApi.fast_execute_script(
                 {
                     "bk_biz_id": env.JOB_BLUEKING_BIZ_ID,
-                    "script_content": str(base64.b64encode(script_content.encode("utf-8")), "utf-8"),
+                    "script_content": base64_encode(script_content),
                     "task_name": _("清理集群"),
                     "account_alias": "root",
                     "script_language": 1,
@@ -141,7 +141,7 @@ def remove_cluster_ips(bk_host_ids, job_clean=True, cc_clean=True):
                 JobApi.fast_execute_script(
                     {
                         "bk_biz_id": env.JOB_BLUEKING_BIZ_ID,
-                        "script_content": str(base64.b64encode(script_content.encode("utf-8")), "utf-8"),
+                        "script_content": base64_encode(script_content),
                         "task_name": _("清理集群"),
                         "account_alias": "root",
                         "script_language": 1,

dbm-ui/backend/db_periodic_task/local_tasks/db_proxy.py

@@ -9,7 +9,6 @@
 specific language governing permissions and limitations under the License.
 """
 
-import base64
 import copy
 import logging
 from collections import defaultdict
@@ -31,6 +30,7 @@
 from backend.db_proxy.models import ClusterExtension, DBCloudProxy, DBExtension
 from backend.db_services.ipchooser.query.resource import ResourceQueryHelper
 from backend.utils.redis import RedisConn
+from backend.utils.string import base64_encode
 
 logger = logging.getLogger("celery")
 
@@ -110,7 +110,7 @@ def _job_push_config_file(_cloud_id, _file_list, _nginx_list):
                     "service_url": f"http://{extension.ip}:{extension.port}",
                 }
                 file_name = f"{extension.bk_biz_id}_{extension.db_type}_{extension.cluster_name}_nginx.conf"
-                file_content = str(base64.b64encode(template.render(conf_payload).encode("utf-8")), "utf-8")
+                file_content = base64_encode(template.render(conf_payload))
                 file_list.append({"file_name": file_name, "content": file_content})
 
                 # 这里先提前写入access url，至于是否执行成功根据is_flush

dbm-ui/backend/db_proxy/views/job_callback/views.py

@@ -9,7 +9,6 @@
 specific language governing permissions and limitations under the License.
 """
 
-import base64
 import json
 import logging
 
@@ -29,6 +28,7 @@
 from backend.flow.consts import SUCCESS_LIST
 from backend.flow.utils.script_template import fast_execute_script_common_kwargs
 from backend.utils.redis import RedisConn
+from backend.utils.string import base64_encode
 
 logger = logging.getLogger("root")
 
@@ -73,7 +73,7 @@ def push_conf_callback(self, request):
         job_payload = {
             "bk_biz_id": env.JOB_BLUEKING_BIZ_ID,
             "task_name": "restart_nginx",
-            "script_content": str(base64.b64encode(restart_nginx_tpl.encode("utf-8")), "utf-8"),
+            "script_content": base64_encode(restart_nginx_tpl),
             "script_language": 1,
             "target_server": {
                 "ip_list": [

dbm-ui/backend/db_proxy/views/jobapi/views.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 base64
 from typing import Any, Dict
 
 from django.utils.translation import ugettext as _
@@ -31,6 +30,7 @@
     TransferFileSerializer,
 )
 from backend.db_proxy.views.views import BaseProxyPassViewSet
+from backend.utils.string import base64_encode
 
 
 class JobApiProxyPassViewSet(BaseProxyPassViewSet):
@@ -55,7 +55,7 @@ def fast_execute_script(self, request):
         job_payloads: Dict[str, Any] = {
             "bk_biz_id": env.JOB_BLUEKING_BIZ_ID,
             "task_name": _("DBM 快速脚本执行"),
-            "script_content": str(base64.b64encode(validated_data["script_content"].encode("utf-8")), "utf-8"),
+            "script_content": base64_encode(validated_data["script_content"]),
             "script_language": validated_data["script_language"],
             "target_server": {"ip_list": validated_data["ip_list"]},
             "timeout": validated_data["timeout"],