feat(backend): 支持日志采集实时下发配置 #10313

# Reviewed, transaction id: 41316

Author: iSecloud

.gtmproject.yaml

@@ -6,7 +6,7 @@ github:
   repo_name: "blueking-dbm"
 
   # 指定里程碑ID，
-  milestone_id: "12"
+  milestone_id: "13"
 
 project:
   # 主分支

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

@@ -43,15 +43,18 @@ def __init__(self):
             description=_("创建采集项的前置检查"),
         )
         self.list_collectors = self.generate_data_api(
-            method="GET",
-            url="databus_collectors/",
-            description=_("获取采集项列表"),
+            method="GET", url="databus_collectors/", description=_("获取采集项列表"), cache_time=60 * 60
         )
         self.search_index_set = self.generate_data_api(
             method="GET",
             url="search_index_set/",
             description=_("查询索引集列表"),
         )
+        self.run_databus_collectors = self.generate_data_api(
+            method="POST",
+            url="databus_collectors/{collector_config_id}/run/",
+            description=_("日志平台-订阅IP下发"),
+        )
 
 
 BKLogApi = _BKLogApi()

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

@@ -114,6 +114,24 @@
 # mysql和tendbcluster采集器一致
 INSTANCE_MONITOR_PLUGINS[DBType.TenDBCluster] = INSTANCE_MONITOR_PLUGINS[DBType.MySQL]
 
+INSTANCE_BKLOG_PLUGINS = {
+    DBType.Redis: {
+        MachineType.PREDIXY: {"name": "predixy", "plugin_id": "redis_slowlog"},
+        MachineType.TWEMPROXY: {"name": "twemproxy", "plugin_id": "redis_slowlog"},
+        MachineType.REDIS: {"name": "redis", "plugin_id": "redis_slowlog"},
+        MachineType.TENDISCACHE: {"name": "tendiscache", "plugin_id": "redis_slowlog"},
+        MachineType.TENDISSSD: {"name": "tendisssd", "plugin_id": "redis_slowlog"},
+        MachineType.TENDISPLUS: {"name": "tendisplus", "plugin_id": "redis_slowlog"},
+    },
+    DBType.MySQL: {
+        MachineType.BACKEND: {"name": "mysql", "plugin_id": "mysql_slowlog"},
+    },
+    DBType.TenDBCluster: {
+        MachineType.SPIDER: {"name": "spider", "plugin_id": "mysql_slowlog"},
+        MachineType.REMOTE: {"name": "mysql", "plugin_id": "mysql_slowlog"},
+    },
+}
+
 SET_NAME_TEMPLATE = "db.{db_type}.{monitor_plugin_name}"
 
 SHORT_NAMES = list(

dbm-ui/backend/db_periodic_task/local_tasks/db_meta/constants.py

@@ -8,7 +8,7 @@
 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 backend.configuration.constants import DBType
 from backend.db_meta.enums import ClusterType
 
 UNIFY_QUERY_PARAMS = {
@@ -33,6 +33,15 @@
     "type": "instant",
 }
 
+EXPORTER_UP_QUERY_TEMPLATE = {
+    DBType.Redis: {
+        "range": 5,
+        "dbm_redis_exporter": """count by (cluster_domain) (
+            bkmonitor:exporter_dbm_redis_exporter:redis_up{instance_role='redis_master'}
+        )""",
+    }
+}
+
 QUERY_TEMPLATE = {
     ClusterType.TendisTwemproxyRedisInstance: {
         "range": 5,

dbm-ui/backend/db_periodic_task/local_tasks/db_meta/sync_cluster_stat.py

@@ -12,6 +12,7 @@
 import datetime
 import json
 import logging
+import time
 from collections import defaultdict
 
 from celery import current_app
@@ -26,6 +27,7 @@
 from backend.db_meta.models import Cluster
 from backend.db_periodic_task.local_tasks import register_periodic_task, start_new_span
 from backend.db_periodic_task.local_tasks.db_meta.constants import (
+    EXPORTER_UP_QUERY_TEMPLATE,
     QUERY_TEMPLATE,
     SAME_QUERY_TEMPLATE_CLUSTER_TYPE_MAP,
     UNIFY_QUERY_PARAMS,
@@ -35,6 +37,29 @@
 logger = logging.getLogger("celery")
 
 
+def query_cluster_exporter_up(db_type, exporter):
+    """查询某类集群的 exporter 是否正常"""
+    # 获取查询模板
+    query_template = EXPORTER_UP_QUERY_TEMPLATE.get(db_type)
+    if not query_template:
+        logger.error("No query template for cluster type: %s and exporter: %s", db_type, exporter)
+        return {}
+
+    # 查询业务固定为DBA，查询时间取模板range
+    params = copy.deepcopy(UNIFY_QUERY_PARAMS)
+    params["bk_biz_id"] = env.DBA_APP_BK_BIZ_ID
+    params["end_time"] = int(time.time())
+    params["start_time"] = params["end_time"] - int(query_template["range"]) * 60
+    params["query_configs"][0]["promql"] = query_template[exporter]
+
+    # 查询exporter up指标
+    series = BKMonitorV3Api.unify_query(params)["series"]
+    cluster_exporter_up_map = {
+        data["dimensions"]["cluster_domain"]: data["datapoints"][0][0] for data in series if data["datapoints"]
+    }
+    return cluster_exporter_up_map
+
+
 def query_cap(bk_biz_id, cluster_type, cap_key="used"):
     """查询某类集群的某种容量: used/total"""
 

dbm-ui/backend/db_report/views/exporter.py

@@ -0,0 +1,53 @@
+# -*- 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 logging
+
+from django.db.models import Count
+from django.utils.translation import gettext as _
+from rest_framework.decorators import action
+from rest_framework.response import Response
+
+from backend.bk_web.swagger import common_swagger_auto_schema
+from backend.bk_web.viewsets import SystemViewSet
+from backend.configuration.constants import DBType
+from backend.db_meta.enums import InstanceRole
+from backend.db_meta.models import StorageInstance
+from backend.db_report.enums import SWAGGER_TAG
+
+logger = logging.getLogger("root")
+
+
+class ClusterExporterUpViewSet(SystemViewSet):
+    @common_swagger_auto_schema(
+        operation_summary=_("获取redis集群的exporter数与分片数不一致的报表"),
+        tags=[SWAGGER_TAG],
+    )
+    @action(methods=["POST"], detail=False)
+    def get_redis_exporter_mismatch(self, request):
+        """获取redis集群的exporter数与分片数不一致的报表"""
+        # 获取exporter数与集群的映射
+        from backend.db_periodic_task.local_tasks.db_meta.sync_cluster_stat import query_cluster_exporter_up
+
+        exporter_map = query_cluster_exporter_up(DBType.Redis, "dbm_redis_exporter")
+        # 获取元数据的集群分片映射
+        redis_masters = StorageInstance.objects.filter(instance_role=InstanceRole.REDIS_MASTER)
+        shard_map = {
+            item["cluster__immute_domain"]: item["total"]
+            for item in redis_masters.values("cluster__immute_domain").annotate(total=Count("id"))
+        }
+        # 过滤exporter与元数据不一致的集群
+        mismatch_clusters = [
+            {"domain": domain, "shard": shard, "exporter_up": exporter_map.get(domain, 0)}
+            for domain, shard in shard_map.items()
+            if shard != exporter_map.get(domain, 0)
+        ]
+        return Response(mismatch_clusters)

dbm-ui/backend/db_services/redis/toolbox/handlers.py

@@ -26,6 +26,7 @@
 from backend.db_services.redis.resources.redis_cluster.query import RedisListRetrieveResource
 from backend.exceptions import ApiResultError
 from backend.flow.utils.base.payload_handler import PayloadHandler
+from backend.flow.utils.redis import redis_util
 from backend.flow.utils.redis.redis_proxy_util import (
     get_cluster_proxy_version,
     get_cluster_proxy_version_for_upgrade,
@@ -138,6 +139,11 @@ def get_update_cluster_versions(cls, cluster_id: int, node_type: str):
         else:
             return get_cluster_proxy_version_for_upgrade(cluster_id)
 
+    @classmethod
+    def list_cluster_big_version(cls, cluster_id: int):
+        """查询集群可更新的大版本"""
+        return redis_util.get_cluster_update_version(cluster_id)
+
     @classmethod
     def webconsole_rpc(cls, cluster_id: int, cmd: str, db_num: int = 0, raw: bool = True, **kwargs):
         """

dbm-ui/backend/db_services/redis/toolbox/serializers.py

@@ -220,6 +220,10 @@ class GetClusterVersionSerializer(serializers.Serializer):
     type = serializers.ChoiceField(help_text=_("请求版本类型"), choices=RedisVersionQueryType.get_choices())
 
 
+class ListClusterBigVersionSerializer(serializers.Serializer):
+    cluster_id = serializers.IntegerField(help_text=_("集群ID"))
+
+
 class GetClusterCapacityInfoSerializer(serializers.Serializer):
     cluster_id = serializers.IntegerField(help_text=_("集群ID"))
     new_storage_version = serializers.CharField(help_text=_("存储版本"))

dbm-ui/backend/db_services/redis/toolbox/views.py

@@ -25,6 +25,7 @@
     GetClusterCapacityInfoSerializer,
     GetClusterModuleInfoSerializer,
     GetClusterVersionSerializer,
+    ListClusterBigVersionSerializer,
     QueryByOneClusterSerializer,
     QueryClusterIpsSerializer,
 )
@@ -85,6 +86,16 @@ def get_cluster_versions(self, request, bk_biz_id, **kwargs):
         else:
             return Response(ToolboxHandler.get_update_cluster_versions(cluster_id, node_type))
 
+    @common_swagger_auto_schema(
+        operation_summary=_("查询集群可更新大版本"),
+        query_serializer=ListClusterBigVersionSerializer(),
+        tags=[SWAGGER_TAG],
+    )
+    @action(methods=["GET"], detail=False, serializer_class=ListClusterBigVersionSerializer, pagination_class=None)
+    def list_cluster_big_version(self, request, bk_biz_id, **kwargs):
+        data = self.params_validate(self.get_serializer_class())
+        return Response(ToolboxHandler.list_cluster_big_version(data["cluster_id"]))
+
     @common_swagger_auto_schema(
         operation_summary=_("获取集群容量变更所需信息"),
         query_serializer=GetClusterCapacityInfoSerializer(),

dbm-ui/backend/env/bklog.py

@@ -19,7 +19,7 @@
 BKLOG_CLUSTER_SUPPORT_HOT_COLD = get_type_env(key="BKLOG_CLUSTER_SUPPORT_HOT_COLD", _type=bool, default=False)
 
 # 日志默认存储天数
-BKLOG_DEFAULT_RETENTION = get_type_env(key="BKLOG_DEFAULT_RETENTION", _type=int, default=7)
+BKLOG_DEFAULT_RETENTION = get_type_env(key="BKLOG_DEFAULT_RETENTION", _type=int, default=30)
 
 # 自定义日志保留天数
 BKLOG_MYSQL_DBBACKUP_RESULT_RETENTION = get_type_env(key="BKLOG_MYSQL_DBBACKUP_RESULT_RETENTION", _type=int)

dbm-ui/backend/flow/consts.py

@@ -202,7 +202,7 @@ class StateType(str, StructuredEnum):
 
 
 class OperateCollectorActionEnum(str, StructuredEnum):
-    INSTALL = EnumField("install", _("安装"))
+    INSTALL = EnumField("INSTALL", _("安装"))
     UNINSTALL = EnumField("UNINSTALL", _("卸载"))
 
 

dbm-ui/backend/flow/plugins/components/collections/common/base_service.py

@@ -275,7 +275,7 @@ def __log__(
         return JobApi.get_job_instance_ip_log({**payload, **ip_dict}, raw=True)
 
     @staticmethod
-    def __url__(job_instance_id: int, link: bool = True):
+    def __url__(job_instance_id: int, link: bool = False):
         """
         获取job任务链接
         """

dbm-ui/backend/flow/utils/cc_manage.py

@@ -17,11 +17,11 @@
 from django.db import transaction
 
 from backend import env
-from backend.components import BKMonitorV3Api, CCApi
+from backend.components import BKLogApi, BKMonitorV3Api, CCApi
 from backend.configuration.models import BizSettings, DBAdministrator
 from backend.db_meta.enums import ClusterType, ClusterTypeMachineTypeDefine
 from backend.db_meta.models import AppMonitorTopo, Cluster, ClusterMonitorTopo, Machine, ProxyInstance, StorageInstance
-from backend.db_meta.models.cluster_monitor import INSTANCE_MONITOR_PLUGINS, SET_NAME_TEMPLATE
+from backend.db_meta.models.cluster_monitor import INSTANCE_BKLOG_PLUGINS, INSTANCE_MONITOR_PLUGINS, SET_NAME_TEMPLATE
 from backend.db_monitor.models import CollectInstance, MonitorPolicy
 from backend.db_monitor.utils import create_bklog_collector
 from backend.db_services.cmdb.biz import get_or_create_cmdb_module_with_name, get_or_create_set_with_name
@@ -581,12 +581,17 @@ def operate_collector(bk_biz_id: int, db_type: str, machine_type: str, bk_instan
     if not bk_instance_ids:
         return
 
-    logger.error(
+    logger.info(
         "operate_collector: {db_type} {machine_type} {bk_instance_ids} {action}".format(
             db_type=db_type, machine_type=machine_type, bk_instance_ids=bk_instance_ids, action=action
         )
     )
 
+    # 获取下发的实例和采集范围
+    nodes = [{"id": bk_instance_id, "bk_biz_id": bk_biz_id} for bk_instance_id in bk_instance_ids]
+    scope = {"bk_biz_id": bk_biz_id, "object_type": "SERVICE", "node_type": "INSTANCE", "nodes": nodes}
+
+    # --- 下发监控采集器 ---
     plugin_id = INSTANCE_MONITOR_PLUGINS[db_type][machine_type]["plugin_id"]
     collect_instances = CollectInstance.objects.filter(db_type=db_type, plugin_id=plugin_id)
     for collect_ins in collect_instances:
@@ -595,22 +600,31 @@ def operate_collector(bk_biz_id: int, db_type: str, machine_type: str, bk_instan
             continue
         # 下发采集器
         try:
-            nodes = [{"id": bk_instance_id, "bk_biz_id": bk_biz_id} for bk_instance_id in bk_instance_ids]
             BKMonitorV3Api.run_collect_config(
-                {
-                    "bk_biz_id": env.DBA_APP_BK_BIZ_ID,
-                    "id": collect_ins.collect_id,
-                    "scope": {
-                        "bk_biz_id": bk_biz_id,
-                        "object_type": "SERVICE",
-                        "node_type": "INSTANCE",
-                        "nodes": nodes,
-                    },
-                    "action": action,
-                }
+                {"bk_biz_id": env.DBA_APP_BK_BIZ_ID, "id": collect_ins.collect_id, "scope": scope, "action": action},
+                use_admin=True,
             )
         except ApiError as err:
-            logger.error(f"[run_collect_config] id:{collect_ins.collect_id} error: {err}")
+            logger.error(f"[monitor] id:{collect_ins.collect_id} error: {err}")
+
+    # --- 下发日志采集器 ---
+    plugin_id = INSTANCE_BKLOG_PLUGINS[db_type][machine_type]["plugin_id"]
+    # 获取当前采集项的列表
+    data = BKLogApi.list_collectors({"bk_biz_id": env.DBA_APP_BK_BIZ_ID, "pagesize": 500, "page": 1}, use_admin=True)
+    collectors_name__info_map = {collector["collector_config_name_en"]: collector for collector in data["list"]}
+    collect = collectors_name__info_map.get(plugin_id)
+    # 忽略不存在的采集项
+    if not collect:
+        return
+    # 下发采集器
+    collect_id = collect["collector_config_id"]
+    try:
+        BKLogApi.run_databus_collectors(
+            {"bk_biz_id": env.DBA_APP_BK_BIZ_ID, "collector_config_id": collect_id, "scope": scope, "action": action},
+            use_admin=True,
+        )
+    except ApiError as err:
+        logger.error(f"[bklog] id:{collect_id} error: {err}")
 
 
 def trigger_operate_collector(