feat(strategy): 支持策略历史安全清理并完善批量操作记录

bkmonitor/bkmonitor/management/commands/clean_strategy_history.py

@@ -0,0 +1,84 @@
+# -*- coding: utf-8 -*-  # noqa: UP009
+"""
+Tencent is pleased to support the open source community by making 蓝鲸智云 - 监控平台 (BlueKing - Monitor) available.
+Copyright (C) 2017-2025 Tencent. 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.
+"""
+
+from django.core.management.base import BaseCommand
+from django.core.management.base import CommandError
+
+from bkmonitor.strategy.history import MIN_CLEAN_STRATEGY_HISTORY_DAYS
+from bkmonitor.strategy.history import CleanStrategyHistoryParams
+from bkmonitor.strategy.history import clean_strategy_history
+
+
+class Command(BaseCommand):
+    """清理过期的策略变更历史。"""
+
+    help = (
+        "清理指定天数以前的策略历史，并保留最近成功快照和必要的删除记录。"
+        f"默认 dry-run；真正删除需加 --execute。--days 不得小于 {MIN_CLEAN_STRATEGY_HISTORY_DAYS}。"
+    )
+    command_name = "clean strategy history"
+    deprecation_warning = ""
+
+    @staticmethod
+    def cleanup(params: CleanStrategyHistoryParams) -> int:
+        return clean_strategy_history(params)
+
+    def add_arguments(self, parser) -> None:
+        parser.add_argument(
+            "--days",
+            type=int,
+            required=True,
+            help=f"保留最近 N 天的全部策略历史，最小 {MIN_CLEAN_STRATEGY_HISTORY_DAYS}",
+        )
+        parser.add_argument(
+            "--batch_size",
+            type=int,
+            default=1000,
+            help="单批删除数量，默认 1000",
+        )
+        parser.add_argument(
+            "--keep_latest_snapshots",
+            type=int,
+            default=1,
+            help="每个策略额外保留的最近成功快照数量，默认 1",
+        )
+        parser.add_argument(
+            "--execute",
+            action="store_true",
+            help="真正执行删除；省略时仅 dry-run 统计预计删除数量",
+        )
+
+    def handle(self, *_args, **options) -> None:
+        try:
+            params = CleanStrategyHistoryParams(
+                days=options["days"],
+                batch_size=options["batch_size"],
+                keep_latest_snapshots=options["keep_latest_snapshots"],
+                dry_run=not options["execute"],
+            )
+        except ValueError as exc:
+            raise CommandError(str(exc)) from exc
+
+        if params.days < MIN_CLEAN_STRATEGY_HISTORY_DAYS:
+            raise CommandError(f"days must be >= {MIN_CLEAN_STRATEGY_HISTORY_DAYS}, got {params.days}")
+
+        if self.deprecation_warning:
+            self.stdout.write(self.style.WARNING(self.deprecation_warning))
+        self.stdout.write(
+            f"{self.command_name} start: "
+            f"days={params.days}, batch_size={params.batch_size}, "
+            f"keep_latest_snapshots={params.keep_latest_snapshots}, dry_run={params.dry_run}"
+        )
+        matched = self.cleanup(params)
+        if params.dry_run:
+            self.stdout.write(self.style.WARNING(f"{self.command_name} dry-run done: would delete {matched} records"))
+        else:
+            self.stdout.write(self.style.SUCCESS(f"{self.command_name} done: deleted {matched} records"))

bkmonitor/bkmonitor/management/commands/clean_strategy_history_compat.py

@@ -0,0 +1,33 @@
+# -*- coding: utf-8 -*-  # noqa: UP009
+"""
+Tencent is pleased to support the open source community by making 蓝鲸智云 - 监控平台 (BlueKing - Monitor) available.
+Copyright (C) 2017-2025 Tencent. 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.
+"""
+
+from bkmonitor.management.commands.clean_strategy_history import Command as CleanStrategyHistoryCommand
+from bkmonitor.strategy.history import MIN_CLEAN_STRATEGY_HISTORY_DAYS
+from bkmonitor.strategy.history import CleanStrategyHistoryParams
+from bkmonitor.strategy.history import clean_strategy_history_compat
+
+
+class Command(CleanStrategyHistoryCommand):
+    """兼容旧批量操作历史状态的临时清理命令。"""
+
+    help = (
+        "临时兼容清理命令：为窗口外旧版批量 update 历史额外保留最近快照，"
+        "并为已删除策略保留最新 delete 记录；其余规则与 clean_strategy_history 一致。"
+        f"默认 dry-run；真正删除需加 --execute。--days 不得小于 {MIN_CLEAN_STRATEGY_HISTORY_DAYS}。"
+    )
+    command_name = "clean strategy history compatibility"
+    deprecation_warning = (
+        "deprecated compatibility command: only use while legacy bulk update/delete status records need cleanup"
+    )
+
+    @staticmethod
+    def cleanup(params: CleanStrategyHistoryParams) -> int:
+        return clean_strategy_history_compat(params)

bkmonitor/bkmonitor/models/strategy.py

@@ -439,6 +439,7 @@ class StrategyHistoryModel(Model):
             ("delete", _lazy("删除")),
             ("create", _lazy("创建")),
             ("update", _lazy("更新")),
+            ("bulk_update", _lazy("批量更新")),
         ),
         db_index=True,
         max_length=12,

bkmonitor/bkmonitor/strategy/history.py

@@ -0,0 +1,254 @@
+# -*- coding: utf-8 -*-  # noqa: UP009
+"""
+Tencent is pleased to support the open source community by making 蓝鲸智云 - 监控平台 (BlueKing - Monitor) available.
+Copyright (C) 2017-2025 Tencent. 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.
+"""
+
+from collections.abc import Iterator
+from datetime import datetime
+from datetime import timedelta
+
+from django.db.models import OuterRef
+from django.db.models import QuerySet
+from django.db.models import Subquery
+from django.utils import timezone
+
+from bkmonitor.models import StrategyHistoryModel
+from bkmonitor.models import StrategyModel
+
+SNAPSHOT_OPERATIONS = ("create", "update", "bulk_update")
+DELETE_OPERATIONS = ("delete",)
+LEGACY_SNAPSHOT_OPERATION = "update"
+STRATEGY_ID_CHUNK_SIZE = 500
+# 管理命令允许的最小保留天数；业务层 CleanStrategyHistoryParams 仍只要求正整数，便于单测构造窗口。
+MIN_CLEAN_STRATEGY_HISTORY_DAYS = 30
+
+
+class CleanStrategyHistoryParams:
+    """策略历史清理参数。
+
+    校验逻辑集中在此类中：days、batch_size、keep_latest_snapshots 均须为正整数。
+    """
+
+    def __init__(
+        self,
+        days: int,
+        batch_size: int = 1000,
+        keep_latest_snapshots: int = 1,
+        dry_run: bool = False,
+    ):
+        self.days = self._require_positive_int("days", days)
+        self.batch_size = self._require_positive_int("batch_size", batch_size)
+        self.keep_latest_snapshots = self._require_positive_int("keep_latest_snapshots", keep_latest_snapshots)
+        if not isinstance(dry_run, bool):
+            raise ValueError("dry_run must be a boolean")
+        self.dry_run = dry_run
+
+    @staticmethod
+    def _require_positive_int(name: str, value: object) -> int:
+        if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
+            raise ValueError(f"{name} must be a positive integer")
+        return value
+
+
+def _collect_latest_history_ids(queryset: QuerySet, limit: int = 1) -> set[int]:
+    """获取查询集中每个策略按时间排序后的最近若干条历史 ID。
+
+    Args:
+        queryset: 已包含业务过滤条件的策略历史查询集。
+        limit: 每个策略保留的最近记录条数。
+
+    Returns:
+        每个策略最近 limit 条历史记录的 ID 集合。
+    """
+    keep_history_ids: set[int] = set()
+    remaining = queryset.order_by()
+
+    for _ in range(limit):
+        latest_id = (
+            remaining.filter(strategy_id=OuterRef("strategy_id")).order_by("-create_time", "-id").values("id")[:1]
+        )
+        batch_ids = set(remaining.filter(id=Subquery(latest_id)).values_list("id", flat=True))
+        if not batch_ids:
+            break
+        keep_history_ids.update(batch_ids)
+        remaining = remaining.exclude(id__in=batch_ids)
+
+    return keep_history_ids
+
+
+def _collect_keep_history_ids(
+    strategy_ids: list[int],
+    keep_latest_snapshots: int = 1,
+    legacy_status_before: datetime | None = None,
+) -> set[int]:
+    """计算一批策略必须保留的历史记录 ID。
+
+    Args:
+        strategy_ids: 本批需要处理的策略 ID。
+        keep_latest_snapshots: 每个策略保留的最近成功快照条数。
+        legacy_status_before: 额外保留此时间之前状态未回写的旧批量更新快照。
+
+    Returns:
+        最近成功快照，以及已删除策略的最新删除记录 ID。
+    """
+    if not strategy_ids:
+        return set()
+
+    existing_strategy_ids = set(StrategyModel.objects.filter(id__in=strategy_ids).values_list("id", flat=True))
+    deleted_strategy_ids = set(strategy_ids) - existing_strategy_ids
+
+    recoverable_snapshots = (
+        StrategyHistoryModel.objects.filter(
+            strategy_id__in=strategy_ids,
+            operate__in=SNAPSHOT_OPERATIONS,
+            status=True,
+        )
+        .exclude(content={})
+        .exclude(content__isnull=True)
+    )
+    keep_history_ids = _collect_latest_history_ids(recoverable_snapshots, limit=keep_latest_snapshots)
+
+    if legacy_status_before is not None:
+        legacy_snapshots = (
+            StrategyHistoryModel.objects.filter(
+                strategy_id__in=strategy_ids,
+                operate=LEGACY_SNAPSHOT_OPERATION,
+                status=False,
+                message="",
+                create_time__lt=legacy_status_before,
+            )
+            .exclude(content={})
+            .exclude(content__isnull=True)
+        )
+        keep_history_ids.update(_collect_latest_history_ids(legacy_snapshots, limit=keep_latest_snapshots))
+
+    if deleted_strategy_ids:
+        delete_histories = StrategyHistoryModel.objects.filter(
+            strategy_id__in=deleted_strategy_ids,
+            operate__in=DELETE_OPERATIONS,
+        )
+        keep_history_ids.update(_collect_latest_history_ids(delete_histories))
+
+    return keep_history_ids
+
+
+def _iter_expired_strategy_id_chunks(
+    before: datetime,
+    chunk_size: int = STRATEGY_ID_CHUNK_SIZE,
+) -> Iterator[list[int]]:
+    """按 strategy_id 分批产出存在过期历史的策略。
+
+    Args:
+        before: 清理截止时间。
+        chunk_size: 每批策略数量。
+
+    Yields:
+        按 strategy_id 升序排列的策略 ID 列表。
+    """
+    last_strategy_id = None
+    while True:
+        queryset = StrategyHistoryModel.objects.filter(create_time__lt=before)
+        if last_strategy_id is not None:
+            queryset = queryset.filter(strategy_id__gt=last_strategy_id)
+
+        strategy_ids = list(
+            queryset.order_by("strategy_id").values_list("strategy_id", flat=True).distinct()[:chunk_size]
+        )
+        if not strategy_ids:
+            return
+
+        yield strategy_ids
+        last_strategy_id = strategy_ids[-1]
+
+
+def _delete_queryset_in_batches(queryset: QuerySet, batch_size: int) -> int:
+    """按主键分批删除查询集中的记录。
+
+    Args:
+        queryset: 待删除的策略历史查询集。
+        batch_size: 单批删除数量。
+
+    Returns:
+        实际删除的记录数量。
+    """
+    model = queryset.model
+    last_pk = None
+    deleted = 0
+
+    while True:
+        batch = queryset.order_by("pk")
+        if last_pk is not None:
+            batch = batch.filter(pk__gt=last_pk)
+
+        history_ids = list(batch.values_list("pk", flat=True)[:batch_size])
+        if not history_ids:
+            return deleted
+
+        deleted_count, _ = model._default_manager.using(queryset.db).filter(pk__in=history_ids).delete()
+        deleted += deleted_count
+        last_pk = history_ids[-1]
+
+
+def _clean_strategy_history(
+    params: CleanStrategyHistoryParams,
+    legacy_status_compat: bool = False,
+) -> int:
+    """清理指定天数以前的冗余策略历史。
+
+    保留最近 ``params.days`` 天的全部历史；更早的记录中，每个策略额外保留全局最近
+    ``params.keep_latest_snapshots`` 条成功快照。已删除策略额外保留全局最新删除记录。
+
+    Args:
+        params: 清理参数，校验在 ``CleanStrategyHistoryParams`` 中完成。
+        legacy_status_compat: 是否额外保留状态未回写的旧批量更新快照。
+
+    Returns:
+        实际删除的历史记录数量。
+
+    Raises:
+        ValueError: 由 ``CleanStrategyHistoryParams`` 在构造时抛出。
+    """
+    before = timezone.now() - timedelta(days=params.days)
+    deleted = 0
+
+    for strategy_ids in _iter_expired_strategy_id_chunks(before):
+        keep_history_ids = _collect_keep_history_ids(
+            strategy_ids,
+            keep_latest_snapshots=params.keep_latest_snapshots,
+            legacy_status_before=before if legacy_status_compat else None,
+        )
+        queryset = StrategyHistoryModel.objects.filter(
+            strategy_id__in=strategy_ids,
+            create_time__lt=before,
+        )
+        if keep_history_ids:
+            queryset = queryset.exclude(id__in=keep_history_ids)
+        if params.dry_run:
+            deleted += queryset.count()
+        else:
+            deleted += _delete_queryset_in_batches(queryset, params.batch_size)
+
+    return deleted
+
+
+def clean_strategy_history(params: CleanStrategyHistoryParams) -> int:
+    """按标准规则清理策略历史。"""
+    return _clean_strategy_history(params)
+
+
+def clean_strategy_history_compat(params: CleanStrategyHistoryParams) -> int:
+    """兼容旧批量操作状态的临时清理入口。
+
+    旧批量更新和删除与普通操作共用 ``update/delete`` 类型，且成功状态未回写。
+    兼容模式为清理窗口外 ``status=False``、空错误消息的旧 ``update`` 记录
+    提供独立快照保留名额，避免其挤掉已确认成功快照；删除记录仍只为已删除策略
+    保留最新一条。该入口不修改历史状态，待旧数据不再需要兼容后，应弃用并恢复
+    标准清理命令。
+    """
+    return _clean_strategy_history(params, legacy_status_compat=True)