feat: 同步图表分析和主分支代码

bklog/apps/log_measure/handlers/metric_collectors/log_clustering.py

@@ -25,23 +25,22 @@
 from apps.log_clustering.models import ClusteringConfig
 from apps.log_measure.utils.metric import MetricUtils
 from bk_monitor.constants import TimeFilterEnum
-from bk_monitor.utils.metric import register_metric, Metric
+from bk_monitor.utils.metric import Metric, register_metric
 
 
 class ClusteringMetricCollector(object):
     @staticmethod
     @register_metric("clustering", description=_("聚类计数"), data_name="metric", time_filter=TimeFilterEnum.MINUTE5)
     def clustering_count():
         clustering_query_set = (
-            ClusteringConfig.objects.filter(modify_flow__isnull=False)
+            ClusteringConfig.objects.filter(signature_enable=True, access_finished=True)
             .values("bk_biz_id")
             .annotate(total=Count("bk_biz_id"))
             .order_by()
         )
         metrics = []
         total = 0
         for clustering_obj in clustering_query_set:
-
             metrics.append(
                 Metric(
                     metric_name="count",

bklog/apps/log_search/constants.py

@@ -1426,6 +1426,17 @@ class SearchMode(ChoicesEnum):
     _choices_labels = ((UI, _("UI模式")), (SQL, _("SQL模式")))
 
 
+class QueryMode(ChoicesEnum):
+    """
+    查询模式
+    """
+
+    UI = "ui"
+    SQL = "sql"
+
+    _choices_labels = ((UI, _("UI模式")), (SQL, _("SQL模式")))
+
+
 # 索引集无数据检查缓存前缀
 INDEX_SET_NO_DATA_CHECK_PREFIX = "index_set_no_data_check_prefix"
 
@@ -1496,6 +1507,17 @@ class FavoriteListOrderType(ChoicesEnum):
     )
 
 
+class FavoriteType(ChoicesEnum):
+    """
+    收藏类型
+    """
+
+    SEARCH = "search"
+    CHART = "chart"
+
+    _choices_labels = ((SEARCH, _("检索")), (CHART, _("图表")))
+
+
 # 用户指引步骤
 USER_GUIDE_STEP_LIST = [
     {"title": _("业务选择框"), "target": "#bizSelectorGuide", "content": _("业务选择框的位置全部换到左侧导航")},
@@ -1627,3 +1649,29 @@ class OperatorEnum:
 
 # 检索选项历史记录API返回数据数量大小
 SEARCH_OPTION_HISTORY_NUM = 20
+
+# SQL模板
+SQL_PREFIX = "SELECT DATE_TRUNC(MAX(dtEventTime), 'minute') AS dtEventTime, COUNT(*) AS log_count"
+SQL_SUFFIX = "GROUP BY minute1 ORDER BY minute1 DESC LIMIT 10"
+
+# 日志检索条件到sql操作符的映射
+SQL_CONDITION_MAPPINGS = {
+    ">": ">",
+    ">=": ">=",
+    "<": "<",
+    "<=": "<=",
+    "=": "=",
+    "!=": "!=",
+    "=~": "LIKE",
+    "&=~": "LIKE",
+    "!=~": "NOT LIKE",
+    "&!=~": "NOT LIKE",
+    "contains": "LIKE",
+    "not contains": "NOT LIKE",
+    "contains match phrase": "MATCH_PHRASE",
+    "not contains match phrase": "NOT MATCH_PHRASE",
+    "all contains match phrase": "MATCH_PHRASE",
+    "all not contains match phrase": "NOT MATCH_PHRASE",
+    "is true": "IS TRUE",
+    "is false": "IS FALSE",
+}

bklog/apps/log_search/exceptions.py

@@ -536,3 +536,18 @@ class SettingMenuException(BasePermException):
 class FunctionGuideException(BaseException):
     ERROR_CODE = "1002"
     MESSAGE = _("不存在该功能引导")
+
+
+# =================================================
+# 图表分析
+# =================================================
+
+
+class IndexSetDorisQueryException(BaseException):
+    ERROR_CODE = "1004"
+    MESSAGE = _("该索引集不支持doris查询")
+
+
+class SQLQueryException(BaseException):
+    ERROR_CODE = "1005"
+    MESSAGE = _("SQL查询异常: {name}")

bklog/apps/log_search/handlers/index_set.py

@@ -22,7 +22,7 @@
 import json
 import re
 from collections import defaultdict
-from typing import Optional, List
+from typing import List, Optional
 
 from django.conf import settings
 from django.db import transaction
@@ -88,8 +88,8 @@
     Scenario,
     Space,
     StorageClusterRecord,
-    UserIndexSetFieldsConfig,
     UserIndexSetCustomConfig,
+    UserIndexSetFieldsConfig,
 )
 from apps.log_search.tasks.mapping import sync_single_index_set_mapping_snapshot
 from apps.log_search.tasks.sync_index_set_archive import sync_index_set_archive
@@ -1727,10 +1727,10 @@ def delete(self):
 
 class UserIndexSetConfigHandler(object):
     def __init__(
-            self,
-            index_set_id: int = None,
-            index_set_ids: List[int] = None,
-            index_set_type: str = IndexSetType.SINGLE.value,
+        self,
+        index_set_id: int = None,
+        index_set_ids: List[int] = None,
+        index_set_type: str = IndexSetType.SINGLE.value,
     ):
         self.index_set_id = index_set_id
         self.index_set_ids = index_set_ids

bklog/apps/log_search/handlers/search/chart_handlers.py

@@ -0,0 +1,207 @@
+# -*- coding: utf-8 -*-
+"""
+Tencent is pleased to support the open source community by making BK-LOG 蓝鲸日志平台 available.
+Copyright (C) 2021 THL A29 Limited, a Tencent company.  All rights reserved.
+BK-LOG 蓝鲸日志平台 is licensed under the MIT License.
+License for BK-LOG 蓝鲸日志平台:
+--------------------------------------------------------------------
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+documentation files (the "Software"), to deal in the Software without restriction, including without limitation
+the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software,
+and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+The above copyright notice and this permission notice shall be included in all copies or substantial
+portions of the Software.
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
+LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+We undertake not to change the open source license (MIT license) applicable to the current version of
+the project delivered to anyone in the future.
+"""
+import re
+
+from django.utils.module_loading import import_string
+from django.utils.translation import ugettext as _
+
+from apps.api import BkDataQueryApi
+from apps.log_search.constants import (
+    SQL_CONDITION_MAPPINGS,
+    SQL_PREFIX,
+    SQL_SUFFIX,
+    SearchMode,
+)
+from apps.log_search.exceptions import (
+    BaseSearchIndexSetException,
+    IndexSetDorisQueryException,
+    SQLQueryException,
+)
+from apps.log_search.models import LogIndexSet
+from apps.utils.log import logger
+
+
+class ChartHandler(object):
+    def __init__(self, index_set_id):
+        self.index_set_id = index_set_id
+        try:
+            self.data = LogIndexSet.objects.get(index_set_id=self.index_set_id)
+        except LogIndexSet.DoesNotExist:
+            raise BaseSearchIndexSetException(
+                BaseSearchIndexSetException.MESSAGE.format(index_set_id=self.index_set_id)
+            )
+
+    @classmethod
+    def get_instance(cls, index_set_id, mode):
+        mapping = {
+            SearchMode.UI.value: "UIChartHandler",
+            SearchMode.SQL.value: "SQLChartHandler",
+        }
+        try:
+            chart_instance = import_string(
+                "apps.log_search.handlers.search.chart_handlers.{}".format(mapping.get(mode))
+            )
+            return chart_instance(index_set_id=index_set_id)
+        except ImportError as error:
+            raise NotImplementedError(f"{mode} class not implement, error: {error}")
+
+    def get_chart_data(self, params: dict) -> dict:
+        """
+        获取图表相关信息
+        :param params: 图表参数
+        :return: 图表数据 dict
+        """
+        raise NotImplementedError(_("功能暂未实现"))
+
+    @staticmethod
+    def generate_sql(params: dict) -> str:
+        """
+        根据过滤条件生成sql
+        :param params: 过滤条件
+        """
+        start_time = params["start_time"] * 1000
+        end_time = params["end_time"] * 1000
+
+        sql = f"WHERE dtEventTimeStamp>={start_time} AND dtEventTimeStamp<={end_time}"
+        addition = params["addition"]
+
+        for condition in addition:
+            sql += " AND "
+            field_name = condition["field"]
+            operator = condition["operator"]
+            values = condition["value"]
+            # 获取sql操作符
+            sql_operator = SQL_CONDITION_MAPPINGS.get(operator)
+            # 异常情况,跳过
+            if not sql_operator or field_name in ["*", "query_string"]:
+                continue
+            # IS TRUE和IS FALSE的逻辑
+            if operator in ["is true", "is false"]:
+                sql += f"{field_name} {sql_operator}"
+                continue
+
+            # values 不为空时才走后面的逻辑
+            if not values:
+                continue
+
+            # 组内条件的与或关系
+            condition_type = "OR"
+            if operator in ["&=~", "&!=~", "all contains match phrase", "all not contains match phrase"]:
+                condition_type = "AND"
+
+            tmp_sql = ""
+            for index, value in enumerate(values):
+                if operator in ["=~", "&=~", "!=~", "&!=~"]:
+                    # 替换通配符
+                    value = value.replace("*", "%")
+                    value = value.replace("?", "_")
+                elif operator in ["contains", "not contains"]:
+                    # 添加通配符
+                    value = f"%{value}%"
+
+                if index > 0:
+                    tmp_sql += f" {condition_type} "
+                if isinstance(value, str):
+                    value = value.replace("'", "''")
+                    value = f"\'{value}\'"
+                tmp_sql += f"{field_name} {sql_operator} {value}"
+
+            # 有两个以上的值时加括号
+            sql += tmp_sql if len(values) == 1 else ("(" + tmp_sql + ")")
+        return f"{SQL_PREFIX} {sql} {SQL_SUFFIX}"
+
+
+class UIChartHandler(ChartHandler):
+    def get_chart_data(self, params: dict) -> dict:
+        """
+        UI模式获取图表相关信息
+        :param params: 图表参数
+        :return: 图表数据 dict
+        """
+        # TODO 待实现
+        return {}
+
+
+class SQLChartHandler(ChartHandler):
+    def get_chart_data(self, params) -> dict:
+        """
+        Sql模式获取图表相关信息
+        :param params: 图表参数
+        :return: 图表数据 dict
+        """
+        if not self.data.support_doris:
+            raise IndexSetDorisQueryException()
+        parsed_sql = self.parse_sql_syntax(self.data.doris_table_id, params["sql"])
+        data = self.fetch_query_data(parsed_sql)
+        return data
+
+    @staticmethod
+    def parse_sql_syntax(doris_table_id: str, raw_sql: str):
+        """
+        解析sql语法
+        """
+        # 如果不存在FROM则添加,存在则覆盖
+        pattern = (
+            r"^\s*?(SELECT\s+?.+?)"
+            r"(?:\bFROM\b.+?)?"
+            r"(\bWHERE\b.*|\bGROUP\s+?BY\b.*|\bHAVING\b.*|\bORDER\s+?BY\b.*|\bLIMIT\b.*|\bINTO\s+?OUTFILE\b.*)?$"
+        )
+        matches = re.match(pattern, raw_sql, re.DOTALL | re.IGNORECASE)
+        if not matches:
+            raise SQLQueryException(SQLQueryException.MESSAGE.format(name=_("缺少SQL查询的关键字")))
+        parsed_sql = matches.group(1) + f" FROM {doris_table_id} "
+        if matches.group(2):
+            parsed_sql += matches.group(2)
+        return parsed_sql
+
+    @staticmethod
+    def fetch_query_data(sql: str) -> dict:
+        """
+        获取查询结果
+        :param sql: 查询sql
+        :return: 查询结果 dict
+        """
+        result_data = BkDataQueryApi.query({"sql": sql}, raw=True)
+        result = result_data.get("result")
+        if not result:
+            # SQL查询失败, 抛出异常
+            errors_message = result_data.get("message", {})
+            errors = result_data.get("errors", {}).get("error")
+            if errors:
+                errors_message = errors_message + ":" + errors
+            logger.info("SQL query exception [%s]", errors_message)
+            raise SQLQueryException(
+                SQLQueryException.MESSAGE.format(name=errors_message),
+                errors={"sql": sql},
+            )
+
+        data_list = result_data["data"]["list"]
+        result_schema = result_data["data"].get("result_schema", [])
+
+        data = {
+            "total_records": result_data["data"]["totalRecords"],
+            "time_taken": result_data["data"]["timetaken"],
+            "list": data_list,
+            "select_fields_order": result_data["data"]["select_fields_order"],
+            "result_schema": result_schema,
+        }
+        return data

bklog/apps/log_search/handlers/search/favorite_handlers.py

@@ -29,6 +29,7 @@
     INDEX_SET_NOT_EXISTED,
     FavoriteGroupType,
     FavoriteListOrderType,
+    FavoriteType,
     FavoriteVisibleType,
     IndexSetType,
 )
@@ -184,14 +185,23 @@ def create_or_update(
         search_fields: list,
         is_enable_display_fields: bool,
         display_fields: list,
+        chart_params: dict = None,
         search_mode: str = None,
         index_set_id: int = None,
         group_id: int = None,
         index_set_ids: list = None,
         index_set_type: str = IndexSetType.SINGLE.value,
+        favorite_type: str = FavoriteType.SEARCH.value,
     ) -> dict:
+        chart_params = chart_params or {}
         # 构建params
-        params = {"ip_chooser": ip_chooser, "addition": addition, "keyword": keyword, "search_fields": search_fields}
+        params = {
+            "ip_chooser": ip_chooser,
+            "addition": addition,
+            "keyword": keyword,
+            "search_fields": search_fields,
+            "chart_params": chart_params,
+        }
         space_uid = self.space_uid if self.space_uid else self.data.space_uid
         search_mode = search_mode if search_mode else self.data.search_mode
 
@@ -245,6 +255,7 @@ def create_or_update(
                 display_fields=display_fields,
                 index_set_ids=index_set_ids,
                 index_set_type=index_set_type,
+                favorite_type=favorite_type,
             )
 
         return model_to_dict(self.data)