feat: 规则审计-策略新增/编辑-后端接口 --story=121513458

【新增】
1. 策略模型新增 sql 字段用于存储规则审计生成的 sql
2. 新增数据平台SQL构造器
3. 策略新增/编辑/删除/启停/重试/获取待更新策略接口支持规则审计
4. 规则审计 bkbase flow 配置
5. RuleAuditSQLFormatter将前端规则审计配置转换为 SQLGenerator 可识别的 SqlConfig。
【优化】
1. 优化控件代码结构
2. 优化 sql 生成器联表和主表引用

src/backend/config/default.py

@@ -363,6 +363,11 @@
     *os.getenv("BKAPP_CORS_ALLOW_HEADERS", "").split(","),
 ]
 
+# 队列存储时长(天)
+DEFAULT_QUEUE_STORAGE_EXPIRES = int(os.getenv("BKAPP_DEFAULT_QUEUE_STORAGE_EXPIRES", 1))
+# HDFS存储时长(天) -1 表示不限制
+DEFAULT_HDFS_STORAGE_EXPIRES = int(os.getenv("BKAPP_DEFAULT_HDFS_STORAGE_EXPIRES", -1))
+
 """
 以下为框架代码 请勿修改
 """

src/backend/core/lock.py

@@ -46,7 +46,7 @@ def release(self) -> bool:
         return cache.delete(self.lock_name)
 
 
-def lock(lock_name: str, *, load_lock_name: callable = None, timeout: int = settings.DEFAULT_CACHE_LOCK_TIMEOUT):
+def lock(lock_name: str = "", *, load_lock_name: callable = None, timeout: int = settings.DEFAULT_CACHE_LOCK_TIMEOUT):
     """
     并发锁装饰器
     """

src/backend/core/sql/builder.py

@@ -0,0 +1,26 @@
+# -*- coding: utf-8 -*-
+"""
+TencentBlueKing is pleased to support the open source community by making
+蓝鲸智云 - 审计中心 (BlueKing - Audit Center) available.
+Copyright (C) 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 http://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.
+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.
+"""
+from pypika.queries import QueryBuilder
+
+
+class BKBaseQueryBuilder(QueryBuilder):
+    """
+    数据平台SQL构造器
+    """
+
+    QUOTE_CHAR = "`"

src/backend/core/sql/model.py

@@ -39,7 +39,7 @@ class Field(BaseModel):
     raw_name: str  # 原始字段名称
     display_name: str  # 作为 sql 的列名，唯一
     field_type: FieldType  # 字段类型
-    aggregate: AggregateType = None  # 聚合函数
+    aggregate: Optional[AggregateType] = None  # 聚合函数
 
 
 class LinkField(BaseModel):
@@ -107,7 +107,7 @@ class SqlConfig(BaseModel):
     """
 
     select_fields: List[Field]  # 作为 sql 的列
-    from_table: str  # 主表
+    from_table: str = ""  # 主表
     join_tables: Optional[List[JoinTable]] = None  # 联表
     where: Optional[WhereCondition] = None  # 筛选条件
     group_by: List[Field] = PydanticField(default_factory=list)  # 分组条件；如果未指定但有聚合函数，则会自动添加 group by 条件

src/backend/core/sql/sql_builder.py

@@ -15,7 +15,7 @@
 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.
 """
-from typing import Dict
+from typing import Dict, Optional
 
 from pypika import Field as PypikaField
 from pypika import Table
@@ -88,17 +88,19 @@ def _build_from(self, query: QueryBuilder) -> QueryBuilder:
         if self.config.from_table:
             query = query.from_(self._get_table(self.config.from_table))
         if self.config.join_tables:
-            query = self._build_join(query)
+            query = self._build_join(self.config.from_table, query)
         return query
 
-    def _build_join(self, query: QueryBuilder) -> QueryBuilder:
+    def _build_join(self, from_table: Optional[str], query: QueryBuilder) -> QueryBuilder:
         """添加 JOIN 子句"""
         for join_table in self.config.join_tables:
             left_table = self._get_table(join_table.left_table)
+            if not from_table:
+                from_table = left_table
+                query = query.from_(from_table)
             right_table = self._get_table(join_table.right_table)
-            join_type_str = join_table.join_type.value.upper()
             try:
-                join_function = getattr(query, join_type_str.lower())
+                join_function = getattr(query, join_table.join_type.value.lower())
             except AttributeError:
                 raise UnsupportedJoinTypeError(join_table.join_type)
             if not join_function:

src/backend/locale/en/LC_MESSAGES/django.po

@@ -8,7 +8,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: PACKAGE VERSION\n"
 "Report-Msgid-Bugs-To: \n"
-"POT-Creation-Date: 2024-12-26 16:22+0800\n"
+"POT-Creation-Date: 2025-01-03 16:11+0800\n"
 "PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
 "Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
 "Language-Team: LANGUAGE <[email protected]>\n"
@@ -2069,15 +2069,19 @@ msgstr "Include"
 msgid "Exclude"
 msgstr "Exclude"
 
+#, python-brace-format
 msgid "表 '{table_name}' 未在配置中声明。"
 msgstr "Table '{table_name}' is not declared in the configuration."
 
+#, python-brace-format
 msgid "不支持的 JOIN 类型：'{join_type}'。"
 msgstr "Unsupported JOIN type: '{join_type}'."
 
+#, python-brace-format
 msgid "不支持的操作符：'{operator}'。"
 msgstr "Unsupported operator: '{operator}'."
 
+#, python-brace-format
 msgid "不支持的聚合类型：'{aggregate_type}'。"
 msgstr "Unsupported aggregate type: '{aggregate_type}'."
 
@@ -2201,6 +2205,12 @@ msgstr "Data Source"
 msgid "数据开发"
 msgstr "Data Flow"
 
+msgid "规则审计"
+msgstr "Rule Audit"
+
+msgid "控件"
+msgstr "Control"
+
 msgid "Create or Update Flow Failed"
 msgstr "Create or Update Flow Failed"
 
@@ -2214,6 +2224,9 @@ msgstr "Cluster Not Exists"
 msgid "Control Not Exists"
 msgstr "Cluster Not Exists"
 
+msgid "Not Support DataSource"
+msgstr "Not Support DataSource"
+
 msgid "操作行为审计"
 msgstr "Operation Audit"
 
@@ -3493,6 +3506,9 @@ msgstr "Policy type not supported"
 msgid "策略类型不可修改"
 msgstr "Policy type cannot be modified"
 
+msgid "联表配置错误"
+msgstr "Link table configuration error"
+
 msgid "Strategy Name"
 msgstr "Strategy Name"
 
@@ -3632,6 +3648,12 @@ msgstr "Is Priority"
 msgid "Field Description"
 msgstr "Field Description"
 
+msgid "Map Field"
+msgstr "Map Field"
+
+msgid "Field Value"
+msgstr "Field Value"
+
 msgid "control_id and control_version_id are required when strategy_type is model"
 msgstr "control_id and control_version_id are required when strategy_type is model"
 
@@ -3843,6 +3865,9 @@ msgstr "Links"
 msgid "Link Table Config"
 msgstr "Link Table Configuration"
 
+msgid "逗号分隔的标签ID列表"
+msgstr "Comma-separated tag ID list"
+
 msgid "Link Table Count"
 msgstr "Link Table Count"
 

src/backend/locale/zh_CN/LC_MESSAGES/django.po

@@ -8,7 +8,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: PACKAGE VERSION\n"
 "Report-Msgid-Bugs-To: \n"
-"POT-Creation-Date: 2024-12-26 16:22+0800\n"
+"POT-Creation-Date: 2025-01-03 16:11+0800\n"
 "PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
 "Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
 "Language-Team: LANGUAGE <[email protected]>\n"
@@ -2069,15 +2069,19 @@ msgstr "包含"
 msgid "Exclude"
 msgstr "不包含"
 
+#, python-brace-format
 msgid "表 '{table_name}' 未在配置中声明。"
 msgstr "表 '{table_name}' 未在配置中声明。"
 
+#, python-brace-format
 msgid "不支持的 JOIN 类型：'{join_type}'。"
 msgstr "不支持的 JOIN 类型：'{join_type}'。"
 
+#, python-brace-format
 msgid "不支持的操作符：'{operator}'。"
 msgstr "不支持的操作符：'{operator}'。"
 
+#, python-brace-format
 msgid "不支持的聚合类型：'{aggregate_type}'。"
 msgstr "不支持的聚合类型：'{aggregate_type}'。"
 
@@ -2201,6 +2205,12 @@ msgstr "数据源"
 msgid "数据开发"
 msgstr "数据开发"
 
+msgid "规则审计"
+msgstr "规则审计"
+
+msgid "控件"
+msgstr "控件"
+
 msgid "Create or Update Flow Failed"
 msgstr "创建或更新任务失败"
 
@@ -2214,6 +2224,9 @@ msgstr "集群不存在"
 msgid "Control Not Exists"
 msgstr "控件不存在"
 
+msgid "Not Support DataSource"
+msgstr "不支持的数据源"
+
 msgid "操作行为审计"
 msgstr "操作行为审计"
 
@@ -3493,6 +3506,9 @@ msgstr "策略类型不支持"
 msgid "策略类型不可修改"
 msgstr "策略类型不可修改"
 
+msgid "联表配置错误"
+msgstr "联表配置错误"
+
 msgid "Strategy Name"
 msgstr "策略名称"
 
@@ -3632,6 +3648,12 @@ msgstr "是否优先"
 msgid "Field Description"
 msgstr "字段描述"
 
+msgid "Map Field"
+msgstr "映射字段"
+
+msgid "Field Value"
+msgstr "字段值"
+
 msgid "control_id and control_version_id are required when strategy_type is model"
 msgstr "当 strategy_type 为 model 时，必须提供 control_id 和 control_version_id"
 
@@ -3827,6 +3849,9 @@ msgstr "连接"
 msgid "Link Table Config"
 msgstr "联表配置"
 
+msgid "逗号分隔的标签ID列表"
+msgstr "逗号分隔的标签ID列表"
+
 msgid "Link Table Count"
 msgstr "联表数"
 

src/backend/services/web/analyze/constants.py

@@ -38,6 +38,8 @@
 BKBASE_ERROR_LOG_LEVEL = "ERROR"
 BKBASE_ORIGIN_DATA_FIELD = "origin_data"
 BKBASE_STRATEGY_ID_FIELD = "strategy_id"
+DEFAULT_QUEUE_STORAGE_CLUSTER_KEY = "default_queue_storage_cluster"
+DEFAULT_HDFS_STORAGE_CLUSTER_KEY = "default_hdfs_storage_cluster"
 
 
 class ControlTypeChoices(TextChoices):
@@ -176,3 +178,12 @@ class ObjectType(TextChoices):
 
     RAW_DATA = "rawdata", gettext_lazy("数据源")
     DATAFLOW = "dataflow", gettext_lazy("数据开发")
+
+
+class BaseControlTypeChoices(TextChoices):
+    """
+    基础控件类型
+    """
+
+    RULE_AUDIT = "rule_audit", gettext_lazy("规则审计")
+    CONTROL = "control", gettext_lazy("控件")

src/backend/services/web/analyze/controls/aiops.py

@@ -169,6 +169,11 @@ def disabled(self) -> None:
         if flow_status in [FlowNodeStatusChoices.RUNNING, FlowNodeStatusChoices.FAILED]:
             self._toggle_strategy(FlowStatusToggleChoices.STOP.value)
 
+    def check_flow_status(self, strategy_id: int, success_status: str, failed_status: str, other_status: str):
+        if not AiopsFeature(help_text="check_flow_status").available:
+            return
+        return check_flow_status.delay(strategy_id, success_status, failed_status, other_status)
+
     def _toggle_strategy(self, status: str, force: bool = False) -> None:
         # update flow
         params = {
@@ -196,7 +201,7 @@ def _toggle_strategy(self, status: str, force: bool = False) -> None:
                 self.strategy.status = StrategyStatusChoices.STARTING
                 self.strategy.save(update_fields=["status"])
                 toggle_monitor.delay(strategy_id=self.strategy.strategy_id, is_active=True)
-                check_flow_status.delay(
+                self.check_flow_status(
                     strategy_id=self.strategy.strategy_id,
                     success_status=StrategyStatusChoices.RUNNING,
                     failed_status=StrategyStatusChoices.START_FAILED,
@@ -207,7 +212,7 @@ def _toggle_strategy(self, status: str, force: bool = False) -> None:
                 self.strategy.status = StrategyStatusChoices.UPDATING
                 self.strategy.save(update_fields=["status"])
                 toggle_monitor.delay(strategy_id=self.strategy.strategy_id, is_active=True)
-                check_flow_status.delay(
+                self.check_flow_status(
                     strategy_id=self.strategy.strategy_id,
                     success_status=StrategyStatusChoices.RUNNING,
                     failed_status=StrategyStatusChoices.UPDATE_FAILED,
@@ -218,9 +223,9 @@ def _toggle_strategy(self, status: str, force: bool = False) -> None:
                 self.strategy.status = StrategyStatusChoices.STOPPING
                 self.strategy.save(update_fields=["status"])
                 toggle_monitor.delay(strategy_id=self.strategy.strategy_id, is_active=False)
-                check_flow_status.delay(
+                self.check_flow_status(
                     strategy_id=self.strategy.strategy_id,
-                    success_status=StrategyStatusChoices.DISABLED.value,
+                    success_status=StrategyStatusChoices.DISABLED,
                     failed_status=StrategyStatusChoices.STOP_FAILED,
                     other_status=StrategyStatusChoices.STOPPING,
                 )

src/backend/services/web/analyze/controls/base.py

@@ -24,26 +24,7 @@
 from services.web.strategy_v2.models import Strategy
 
 
-class Controller:
-    """
-    Control Strategy and Event
-    """
-
-    def __new__(cls, strategy_id: int, *args, **kwargs) -> "Controller":
-        # check child
-        if cls.__name__ != Controller.__name__:
-            return super().__new__(cls)
-        # load control
-        control_id = Strategy.objects.get(strategy_id=strategy_id).control_id
-        control_type_id = Control.objects.get(control_id=control_id).control_type_id
-        # import controller
-        controller_path = "services.web.analyze.controls.{}.{}Controller".format(
-            control_type_id.lower(), control_type_id
-        )
-        controller_class = import_string(controller_path)
-        # init controller
-        return controller_class(strategy_id)
-
+class BaseControl(abc.ABC):
     def __init__(self, strategy_id: int):
         self.strategy: Strategy = Strategy.objects.get(strategy_id=strategy_id)
 
@@ -86,3 +67,24 @@ def disabled(self) -> None:
         """
 
         raise NotImplementedError()
+
+
+class Controller(BaseControl, abc.ABC):
+    """
+    Control Strategy and Event
+    """
+
+    def __new__(cls, strategy_id: int, *args, **kwargs) -> "Controller":
+        # check child
+        if cls.__name__ != Controller.__name__:
+            return super().__new__(cls)
+        # load control
+        control_id = Strategy.objects.get(strategy_id=strategy_id).control_id
+        control_type_id = Control.objects.get(control_id=control_id).control_type_id
+        # import controller
+        controller_path = "services.web.analyze.controls.{}.{}Controller".format(
+            control_type_id.lower(), control_type_id
+        )
+        controller_class = import_string(controller_path)
+        # init controller
+        return controller_class(strategy_id)