feat: 完成 issue #33 ② 数据保留/归档/降采样策略配置
This commit is contained in:
@@ -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 年
|
||||
@@ -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(),
|
||||
}
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user