diff --git a/core/data-bus/config/retention.template.yaml b/core/data-bus/config/retention.template.yaml new file mode 100644 index 0000000..793f839 --- /dev/null +++ b/core/data-bus/config/retention.template.yaml @@ -0,0 +1,40 @@ +# -*- coding: utf-8 -*- +# 模板「数据保留/归档/降采样策略」配置资产示例：ti-cl4（Template-Ti 一期）。 +# +# 说明（issue #33 / PRD 5.2「② 数据总线 + 时序库」）： +# - raw_keep_days：时序原始数据 TDengine KEEP 天数（保留期后由降采样/归档接续）； +# - downsampling：按点位匹配（protocol / point_id_prefix）把高精度数据降采样 +# 为低频长期保留（agg ∈ avg/max/min/last/sum，interval 为 TDengine INTERVAL）； +# - archive：超保留期历史归档到 MinIO（桶 = {template}-archive，对象保留期独立）； +# - 与 #31（MinIO 生命周期）策略联动：归档对象生命周期由 MinIO ILM 配置承载。 +# - 换行业只改本文件，内核零改动。 +template: ti-cl4 +version: 1.0.0 + +retention: + # 时序原始数据保留（TDengine KEEP 天数） + raw_keep_days: 90 + + # 降采样规则（按点位匹配；多条规则按声明顺序，首个命中生效） + downsampling: + - name: raw_1m_avg + description: 常规工艺点 1Hz → 1m AVG 长期保留 + match: + protocol: opcua + interval: 1m + agg: avg + keep_days: 730 # 降采样后保留 2 年 + - name: energy_1h_sum + description: 能源点 1Hz → 1h SUM 保留 + match: + protocol: energy + interval: 1h + agg: sum + keep_days: 3650 # 能耗长期保留 10 年 + + # 归档：超保留期历史 → MinIO（对象保留期独立，配合 #31 ILM） + archive: + enabled: true + bucket_suffix: archive + object_prefix: history + retention_days: 3650 # 归档对象保留 10 年 diff --git a/core/data-bus/retention_policy.py b/core/data-bus/retention_policy.py new file mode 100644 index 0000000..9b60145 --- /dev/null +++ b/core/data-bus/retention_policy.py @@ -0,0 +1,152 @@ +# -*- 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(), + } diff --git a/core/data-bus/tests/test_retention_policy.py b/core/data-bus/tests/test_retention_policy.py new file mode 100644 index 0000000..55c85e5 --- /dev/null +++ b/core/data-bus/tests/test_retention_policy.py @@ -0,0 +1,127 @@ +# -*- coding: utf-8 -*- +"""数据保留/归档/降采样策略配置测试（issue #33）。 + +覆盖： +1. 模板配置资产加载（保留天数 / 降采样规则 / 归档配置）； +2. tdengine_keep_days 与 generate_schema 联动（KEEP 参数）； +3. 降采样 SQL 片段生成 + 点位匹配（protocol / 前缀）； +4. 归档配置（桶名联动模板命名、对象保留期）； +5. 非法配置拒绝（keep_days≤0 / 非法 agg / 归档保留期≤0）。 +""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: F401 + +from data_bus.retention_policy import ( # noqa: E402 + ALLOWED_AGGS, + DEFAULT_CONFIG_PATH, + DownsampleRule, + RetentionPolicy, +) +from data_bus.tdengine_schema import ( # noqa: E402 + generate_schema, + specs_from_point_dict_rows, +) +from data_bus.templating import TemplateNaming # noqa: E402 + +CONFIG = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "config", "retention.template.yaml", +) + + +class TestConfigLoad(unittest.TestCase): + """配置资产加载。""" + + def setUp(self): + self.policy = RetentionPolicy.from_template_config(CONFIG) + + def test_template_and_keep(self): + self.assertEqual(self.policy.template, "ti-cl4") + self.assertEqual(self.policy.tdengine_keep_days(), 90) + + def test_downsampling_rules(self): + rules = self.policy.downsampling_rules() + self.assertEqual(len(rules), 2) + self.assertEqual(rules[0].interval, "1m") + self.assertEqual(rules[0].agg, "avg") + self.assertEqual(rules[0].keep_days, 730) + + def test_archive_config(self): + cfg = self.policy.archive_config() + self.assertTrue(cfg["enabled"]) + self.assertEqual(cfg["bucket"], "ti-cl4-archive") + self.assertEqual(cfg["retention_days"], 3650) + + def test_default_config_path_exists(self): + self.assertTrue(os.path.isfile(DEFAULT_CONFIG_PATH)) + + +class TestKeepLinkedToSchema(unittest.TestCase): + """保留天数与 TDengine schema 联动（KEEP）。""" + + def test_generate_schema_uses_keep_days(self): + policy = RetentionPolicy.from_template_config(CONFIG) + naming = TemplateNaming("ti-cl4") + rows = [{"device_id": "CLF-01", "point_id": "CLF-01.TEMP", + "unit": "℃", "dataType": "float"}] + ddl = generate_schema( + naming, specs_from_point_dict_rows(rows), + retention_days=policy.tdengine_keep_days())[0] + self.assertIn(f"KEEP({policy.tdengine_keep_days()})", ddl) + + +class TestDownsampleSql(unittest.TestCase): + """降采样 SQL 生成与点位匹配。""" + + def setUp(self): + self.policy = RetentionPolicy.from_template_config(CONFIG) + + def test_sql_fragment(self): + rule = DownsampleRule(name="r", interval="1m", agg="avg", keep_days=730) + sql = rule.sql_fragment("tpl_ti_cl4.CLF_01_TEMP") + self.assertIn("SELECT _wstart AS ts", sql) + self.assertIn("avg(value) AS value_agg", sql) + self.assertIn("INTERVAL(1m)", sql) + + def test_matching_rules_by_protocol(self): + hits = self.policy.matching_rules( + {"point_id": "CLF-01.TEMP", "protocol": "opcua"}) + self.assertEqual([h.name for h in hits], ["raw_1m_avg"]) + + def test_matching_rules_by_prefix(self): + # 无匹配规则时返回空（energy 规则不命中 opcua 点位） + hits = self.policy.matching_rules( + {"point_id": "CLF-01.TEMP", "protocol": "opcua"}) + self.assertNotIn("energy_1h_sum", [h.name for h in hits]) + + def test_allowed_aggs(self): + self.assertTrue({"avg", "max", "min", "last", "sum"} <= ALLOWED_AGGS) + + +class TestValidation(unittest.TestCase): + """非法配置拒绝。""" + + def test_keep_days_positive(self): + with self.assertRaises(ValueError): + RetentionPolicy(template="t", raw_keep_days=0) + + def test_invalid_agg(self): + with self.assertRaises(ValueError): + RetentionPolicy( + template="t", + downsampling=[{"name": "r", "interval": "1m", + "agg": "median", "keep_days": 30}]) + + def test_archive_retention_positive(self): + with self.assertRaises(ValueError): + RetentionPolicy( + template="t", + archive={"enabled": True, "retention_days": 0}) + + +if __name__ == "__main__": + unittest.main()