153 lines
6.4 KiB
Python
153 lines
6.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""数据保留 / 归档 / 降采样策略配置 —— issue #33 / PRD 5.2。
|
||
|
||
数据生命周期管理，模板配置驱动（换行业只改 `config/retention.template.yaml`）：
|
||
|
||
- **保留（retention）**：时序原始数据 TDengine `KEEP` 天数（raw_keep_days）；
|
||
- **降采样（downsampling）**：按点位匹配规则把高精度数据聚合为低频长期
|
||
保留（如 1Hz → 1m AVG 保留 730 天），生成 TDengine 降采样 SQL 片段
|
||
（`SELECT _wstart, {agg}(value) ... INTERVAL({interval})`）；
|
||
- **归档（archive）**：超保留期历史归档到 MinIO 对象存储
|
||
（归档桶 + 对象键前缀，联动 templating 命名），对象保留期可独立配置。
|
||
|
||
本模块不依赖 TDengine/MinIO SDK：仅产出策略配置与可执行 SQL/路径，
|
||
供模板配置台预览 / 运维执行；与 #31（MinIO 生命周期）策略联动。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from dataclasses import dataclass, field
|
||
from typing import Dict, List, Optional
|
||
|
||
from data_bus.templating import TemplateNaming, sanitize_sql
|
||
|
||
#: 默认配置资产路径（相对本模块）
|
||
DEFAULT_CONFIG_PATH = os.path.join(
|
||
os.path.dirname(os.path.abspath(__file__)), "config", "retention.template.yaml")
|
||
|
||
#: 允许的降采样聚合函数（TDengine 支持）
|
||
ALLOWED_AGGS = {"avg", "max", "min", "last", "sum"}
|
||
|
||
|
||
@dataclass
|
||
class DownsampleRule:
|
||
"""一条降采样规则（按点位匹配）。"""
|
||
|
||
name: str
|
||
interval: str # TDengine INTERVAL（如 1m / 1h）
|
||
agg: str # avg / max / min / last / sum
|
||
keep_days: int # 降采样后保留天数
|
||
match: dict = field(default_factory=dict) # 点位匹配条件（protocol/前缀）
|
||
|
||
def sql_fragment(self, table: str, value_col: str = "value") -> str:
|
||
"""生成该规则作用于某子表的降采样查询（示例，供调度/配置台预览）。"""
|
||
return (
|
||
f"SELECT _wstart AS ts, {self.agg}({value_col}) AS {value_col}_agg "
|
||
f"FROM {table} WHERE ts >= now - {self.keep_days}d "
|
||
f"INTERVAL({self.interval})"
|
||
)
|
||
|
||
|
||
class RetentionPolicy:
|
||
"""数据保留 / 归档 / 降采样策略（模板配置驱动）。"""
|
||
|
||
def __init__(
|
||
self,
|
||
template: str,
|
||
raw_keep_days: int = 90,
|
||
downsampling: Optional[List[dict]] = None,
|
||
archive: Optional[dict] = None,
|
||
) -> None:
|
||
self.template = template
|
||
self.naming = TemplateNaming(template=template)
|
||
self.raw_keep_days = int(raw_keep_days)
|
||
self.downsampling = [self._parse_rule(r) for r in (downsampling or [])]
|
||
self.archive = dict(archive or {})
|
||
self._validate()
|
||
|
||
# ------------------------------------------------------------------
|
||
@classmethod
|
||
def from_template_config(cls, path: str = DEFAULT_CONFIG_PATH) -> "RetentionPolicy":
|
||
"""从模板配置资产加载（config/retention.template.yaml）。"""
|
||
import yaml
|
||
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
raw = yaml.safe_load(fh) or {}
|
||
r = raw.get("retention", {}) or {}
|
||
return cls(
|
||
template=str(raw.get("template", "default")),
|
||
raw_keep_days=int(r.get("raw_keep_days", 90)),
|
||
downsampling=r.get("downsampling", []),
|
||
archive=r.get("archive", {}),
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
@staticmethod
|
||
def _parse_rule(rule: dict) -> DownsampleRule:
|
||
return DownsampleRule(
|
||
name=str(rule.get("name", "rule")),
|
||
interval=str(rule.get("interval", "1m")),
|
||
agg=str(rule.get("agg", "avg")).lower(),
|
||
keep_days=int(rule.get("keep_days", 730)),
|
||
match=dict(rule.get("match", {}) or {}),
|
||
)
|
||
|
||
def _validate(self) -> None:
|
||
if self.raw_keep_days <= 0:
|
||
raise ValueError("raw_keep_days 必须 > 0")
|
||
for r in self.downsampling:
|
||
if r.keep_days <= 0:
|
||
raise ValueError(f"降采样规则 {r.name!r} keep_days 必须 > 0")
|
||
if r.agg not in ALLOWED_AGGS:
|
||
raise ValueError(
|
||
f"降采样规则 {r.name!r} agg={r.agg!r} 非法，允许 {sorted(ALLOWED_AGGS)}")
|
||
if self.archive.get("enabled") and int(self.archive.get("retention_days", 0)) <= 0:
|
||
raise ValueError("归档 retention_days 必须 > 0")
|
||
|
||
# ------------------------------------------------------------------
|
||
def tdengine_keep_days(self) -> int:
|
||
"""时序原始数据 TDengine KEEP 天数（供 generate_schema 使用）。"""
|
||
return self.raw_keep_days
|
||
|
||
def downsampling_rules(self) -> List[DownsampleRule]:
|
||
"""全部降采样规则。"""
|
||
return list(self.downsampling)
|
||
|
||
def matching_rules(self, point: dict) -> List[DownsampleRule]:
|
||
"""命中某点位的降采样规则（match：protocol 相等 / 前缀匹配）。"""
|
||
hits = []
|
||
for r in self.downsampling:
|
||
if not r.match:
|
||
hits.append(r)
|
||
continue
|
||
if r.match.get("protocol") and r.match["protocol"] != point.get("protocol"):
|
||
continue
|
||
prefix = r.match.get("point_id_prefix")
|
||
if prefix and not str(point.get("point_id", "")).startswith(prefix):
|
||
continue
|
||
hits.append(r)
|
||
return hits
|
||
|
||
def archive_config(self) -> dict:
|
||
"""归档配置：桶名（联动 templating 命名）+ 对象键前缀 + 对象保留天数。"""
|
||
enabled = bool(self.archive.get("enabled"))
|
||
# 桶名对齐 TemplateNaming.bucket 语义：{template}-{suffix}
|
||
suffix = sanitize_sql(str(self.archive.get("bucket_suffix", "archive")))
|
||
return {
|
||
"enabled": enabled,
|
||
"bucket": f"{self.template}-{suffix}",
|
||
"object_prefix": str(self.archive.get("object_prefix", "history")),
|
||
"retention_days": int(self.archive.get("retention_days", 3650)),
|
||
}
|
||
|
||
def brief(self) -> dict:
|
||
"""策略摘要（部署/巡检）。"""
|
||
return {
|
||
"template": self.template,
|
||
"raw_keep_days": self.raw_keep_days,
|
||
"downsampling": [{"name": r.name, "interval": r.interval,
|
||
"agg": r.agg, "keep_days": r.keep_days}
|
||
for r in self.downsampling],
|
||
"archive": self.archive_config(),
|
||
}
|