From 2947dae4349f1a6b432f8ed4c8863f33f4ee6078 Mon Sep 17 00:00:00 2001 From: yunmei Date: Wed, 5 Aug 2026 02:13:55 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=8C=E6=88=90=20issue=20#31=20?= =?UTF-8?q?=E2=91=A1=20MinIO=20=E5=AF=B9=E8=B1=A1=E5=AD=98=E5=82=A8?= =?UTF-8?q?=E6=A1=B6=E4=B8=8E=E7=94=9F=E5=91=BD=E5=91=A8=E6=9C=9F=E7=AD=96?= =?UTF-8?q?=E7=95=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/data-bus/README.md | 1 + core/data-bus/__init__.py | 3 + core/data-bus/config/minio.template.yaml | 27 ++++ core/data-bus/minio_lifecycle.py | 160 ++++++++++++++++++++ core/data-bus/tests/test_minio_lifecycle.py | 151 ++++++++++++++++++ 5 files changed, 342 insertions(+) create mode 100644 core/data-bus/config/minio.template.yaml create mode 100644 core/data-bus/minio_lifecycle.py create mode 100644 core/data-bus/tests/test_minio_lifecycle.py diff --git a/core/data-bus/README.md b/core/data-bus/README.md index 101821b..e4ec3b2 100644 --- a/core/data-bus/README.md +++ b/core/data-bus/README.md @@ -11,6 +11,7 @@ topic / 时序库表 / 关系表 / 对象桶：换行业只改模板资产（模 |------|------| | `templating.py` | 模板命名推导：Kafka topic + 分区策略、TDengine 超级表/子表、PostgreSQL schema/表、MinIO 桶/对象键 | | `kafka_naming.py` | Kafka topic 命名/分区模板化组件（issue #28）：配置外置、点位维度覆盖、按 (topic,partition) 生产路由 | +| `minio_lifecycle.py` | MinIO 对象桶 + 生命周期策略模板化（issue #31）：桶/对象键命名 + `mc mb`/`mc ilm` 落地命令 + S3 兼容生命周期策略 JSON | | `tdengine_schema.py` | 时序库 schema 自动生成（超级表 + 每测点子表，**依据点位字典**）+ 批量 INSERT SQL | | `postgres_schema.py` | 关系库 schema（模板 / 模型 / 用户 / 权限）+ 角色授权语句 | | `batch_writer.py` | 批量写入缓冲：批量聚合（默认 5000 条/0.1s，对齐 PRD 5.2「5k/100ms」基线）、幂等去重、失败重试 —— **数据不丢不重** | diff --git a/core/data-bus/__init__.py b/core/data-bus/__init__.py index 7d51237..874c90a 100644 --- a/core/data-bus/__init__.py +++ b/core/data-bus/__init__.py @@ -11,6 +11,7 @@ topic / 时序库表 / 关系表 / 对象桶，换行业只改模板资产（YAM - kafka_naming Kafka topic 命名/分区模板化组件（配置外置 + 点位维度覆盖 + 生产路由）； - tdengine_schema 时序超级表/子表 DDL + 批量 INSERT SQL 生成； - postgres_schema 关系库 schema（模板/模型/用户/权限）+ 授权语句； +- minio_lifecycle MinIO 对象桶 + 生命周期策略模板化（mc mb/ilm + S3 兼容策略 JSON）； - batch_writer 批量写入缓冲（StoreSink 抽象 / MemorySink / TdengineSink）， 批量聚合 + 幂等去重 + 失败重试（不丢不重）。 @@ -26,6 +27,7 @@ from .batch_writer import ( TdengineSink, ) from .kafka_naming import KafkaTopicNaming +from .minio_lifecycle import MinioLifecycle from .postgres_schema import generate_grant_ddl, generate_schema_ddl from .tdengine_schema import ( PointSpec, @@ -42,6 +44,7 @@ __all__ = [ "sanitize", "sanitize_sql", "KafkaTopicNaming", + "MinioLifecycle", "PointSpec", "generate_supertable_ddl", "generate_subtable_ddls", diff --git a/core/data-bus/config/minio.template.yaml b/core/data-bus/config/minio.template.yaml new file mode 100644 index 0000000..f3689a5 --- /dev/null +++ b/core/data-bus/config/minio.template.yaml @@ -0,0 +1,27 @@ +# -*- coding: utf-8 -*- +# 模板「MinIO 对象存储桶 + 生命周期策略」配置资产示例：ti-cl4（氯化车间/海绵钛，Template-Ti 一期）。 +# +# 说明（issue #31 / PRD 5.2「② 数据总线 + 时序库」）： +# - 桶命名：`{template}-{bucket_suffix}`（S3 桶名允许 `-`，与 templating.bucket() 一致）； +# 对象键前缀：特征快照 `features/...`、模型 artifact `models/...`（templating 派生）； +# - lifecycle：按前缀声明对象过期天数（S3 LifecycleConfiguration 兼容，MinIO 支持）， +# 与数据保留/归档策略（#33）联动，避免对象无限堆积； +# - 换行业只改本文件（template / bucket_suffix / lifecycle），内核零改动。 +template: ti-cl4 +version: 1.0.0 + +minio: + # 桶后缀（缺省 artifacts → 桶名 ti-cl4-artifacts） + bucket_suffix: artifacts + # mc 命令远端别名（联调 local / 生产 cluster；mc alias set 时定义） + mc_alias: local + # 生命周期规则（prefix 必填；expiration_days 过期天数，≥ 1） + lifecycle: + # 特征快照：供训练/回溯，保留 180 天自动过期 + - id: features-expire + prefix: features/ + expiration_days: 180 + # 模型 artifact：随版本长期保留（3 年），出清历史版本前不自动删 + - id: models-retain + prefix: models/ + expiration_days: 1095 diff --git a/core/data-bus/minio_lifecycle.py b/core/data-bus/minio_lifecycle.py new file mode 100644 index 0000000..4db1d08 --- /dev/null +++ b/core/data-bus/minio_lifecycle.py @@ -0,0 +1,160 @@ +# -*- coding: utf-8 -*- +"""MinIO 对象存储桶 + 生命周期策略模板化 —— issue #31 / PRD 5.2。 + +在 #4（EPIC）templating 的 MinIO 命名（桶/对象键）之上，交付**完整的 +MinIO 对象存储组件**： + +- **桶**：`{template}-artifacts`（S3 桶名允许 `-`，无需 SQL 清洗）， + 对象键沿用 templating：特征快照 `features/{model_id}/{date}/{seq}.jsonl`、 + 模型 artifact `models/{model_id}/{version}/model.bin`； +- **生命周期策略**：按前缀配置对象过期天数（S3 LifecycleConfiguration 兼容， + MinIO 支持 JSON 形式），如特征快照保留 180 天、模型 artifact 长期保留， + 与数据保留/归档策略（#33）联动，避免对象无限堆积； +- **落地方式**：只产出**配置文本**（`mc mb` / `mc ilm` 命令 + 生命周期 + 策略 JSON），不依赖 minio SDK，供模板配置台预览 / 运维执行； +- **换行业只改配置**：模板 YAML（`config/minio.template.yaml`）驱动， + 内核代码零改动。 + +验收口径（issue #31 / EPIC #4）： +- 桶命名模板化（`{template}-{suffix}`），配置可覆盖； +- 生命周期规则按前缀声明（ID / 前缀 / 过期天数），生成可执行命令与策略 JSON； +- 缺省配置覆盖特征快照 + 模型 artifact 两类对象。 +""" +from __future__ import annotations + +import json +import os +from typing import Dict, List, Optional + +from data_bus.templating import TemplateNaming + +#: 默认配置资产路径（相对本模块） +DEFAULT_CONFIG_PATH = os.path.join( + os.path.dirname(os.path.abspath(__file__)), "config", "minio.template.yaml") + + +def _normalize_rules(lifecycle: Optional[List[dict]]) -> List[dict]: + """规范化生命周期规则：过滤非法行，缺省 ID 自动生成。 + + 每条规则形如 ``{"id": ..., "prefix": ..., "expiration_days": ...}``； + - prefix 必填（生命周期按对象键前缀匹配）； + - expiration_days 为过期天数（≥ 1）； + - 缺省 id 用前缀派生（保证 S3 Rule ID 唯一且稳定）。 + """ + rules: List[dict] = [] + for r in (lifecycle or []): + prefix = str(r.get("prefix") or "").strip() + days = int(r.get("expiration_days") or 0) + if not prefix or days < 1: + continue + rule_id = str(r.get("id") or "").strip() or f"{prefix}expire" + rules.append({ + "id": rule_id, + "prefix": prefix, + "expiration_days": days, + }) + return rules + + +class MinioLifecycle: + """MinIO 桶 + 生命周期策略模板化（模板 + 前缀维度）。""" + + def __init__( + self, + template: str, + bucket_suffix: str = "artifacts", + lifecycle: Optional[List[dict]] = None, + mc_alias: str = "local", + ) -> None: + """ + Args: + template: 行业模板名（如 ti-cl4 / resin）； + bucket_suffix: 桶后缀（缺省 artifacts → `{template}-artifacts`）； + lifecycle: 生命周期规则 `[{"prefix": "features/", "expiration_days": 180}]`； + mc_alias: `mc` 命令的远端别名（联调 local / 生产 cluster）。 + """ + self._tpl = TemplateNaming(template, bucket_suffix=bucket_suffix) + self.bucket_suffix = self._tpl.bucket_suffix + self.mc_alias = mc_alias + self._rules = _normalize_rules(lifecycle) + + # ------------------------------------------------------------------ + @classmethod + def from_template_config(cls, path: str = DEFAULT_CONFIG_PATH) -> "MinioLifecycle": + """从模板配置资产加载（config/minio.template.yaml）。""" + import yaml + with open(path, "r", encoding="utf-8") as fh: + raw = yaml.safe_load(fh) or {} + m = raw.get("minio", {}) or {} + return cls( + template=str(raw.get("template", "default")), + bucket_suffix=str(m.get("bucket_suffix", "artifacts")), + lifecycle=m.get("lifecycle", []), + mc_alias=str(m.get("mc_alias", "local")), + ) + + # ------------------------------------------------------------------ + # 命名：桶 / 对象键（复用 templating，与 #4 命名一致） + # ------------------------------------------------------------------ + def bucket(self) -> str: + """对象存储桶：`{template}-{bucket_suffix}`（S3 桶名允许 `-`）。""" + return self._tpl.bucket() + + def snapshot_key(self, model_id: str, date: str, seq: int) -> str: + """特征快照对象键：`features/{model_id}/{date}/{seq:06d}.jsonl`。""" + return self._tpl.snapshot_key(model_id, date, seq) + + def model_artifact_key(self, model_id: str, version: str) -> str: + """模型 artifact 对象键：`models/{model_id}/{version}/model.bin`。""" + return self._tpl.model_artifact_key(model_id, version) + + # ------------------------------------------------------------------ + # 落地：mc 命令 + 生命周期策略 JSON（只产配置文本，不依赖 SDK） + # ------------------------------------------------------------------ + def bucket_create_statements(self) -> List[str]: + """桶创建命令（`mc mb`，--ignore-existing 幂等）。""" + return [f"mc mb --ignore-existing {self.mc_alias}/{self.bucket()}"] + + def lifecycle_cli_statements(self) -> List[str]: + """生命周期规则落地命令（`mc ilm add`，幂等可重跑）。 + + 示例：`mc ilm add --expiry-days 180 --prefix "features/" local/ti-cl4-artifacts` + """ + bucket = self.bucket() + return [ + f'mc ilm add --expiry-days {r["expiration_days"]} ' + f'--prefix "{r["prefix"]}" {self.mc_alias}/{bucket}' + for r in self._rules + ] + + def lifecycle_policy_json(self, indent: int = 2) -> str: + """S3 LifecycleConfiguration 兼容的 JSON 策略（MinIO 支持）。 + + 返回结构：``{"Rules": [{"ID", "Status", "Filter": {"Prefix"}, + "Expiration": {"Days"}}]}``，可直接用于 + ``PutBucketLifecycleConfiguration`` 或配置台预览。 + """ + rules = [ + { + "ID": r["id"], + "Status": "Enabled", + "Filter": {"Prefix": r["prefix"]}, + "Expiration": {"Days": r["expiration_days"]}, + } + for r in self._rules + ] + return json.dumps({"Rules": rules}, indent=indent, ensure_ascii=False) + + def brief(self) -> dict: + """配置摘要（部署/巡检用）。""" + return { + "template": self._tpl.template, + "bucket": self.bucket(), + "mc_alias": self.mc_alias, + "lifecycle_rules": list(self._rules), + } + + @property + def rules(self) -> List[dict]: + """规范化后的生命周期规则（只读视图）。""" + return list(self._rules) diff --git a/core/data-bus/tests/test_minio_lifecycle.py b/core/data-bus/tests/test_minio_lifecycle.py new file mode 100644 index 0000000..5f563ea --- /dev/null +++ b/core/data-bus/tests/test_minio_lifecycle.py @@ -0,0 +1,151 @@ +# -*- coding: utf-8 -*- +"""MinIO 对象存储桶 + 生命周期策略模板化测试（issue #31）。 + +覆盖： +1. 模板配置资产加载（config/minio.template.yaml，含生命周期规则）； +2. 桶命名：`{template}-{bucket_suffix}`（S3 桶名允许 `-`）； +3. 对象键：特征快照 / 模型 artifact（与 templating 命名一致）； +4. 落地命令：`mc mb` 幂等建桶、`mc ilm add` 生命周期规则命令； +5. 生命周期策略 JSON（S3 LifecycleConfiguration 兼容，MinIO 支持）； +6. 规则规范化：缺省 id 派生、非法行过滤。 +""" +import json +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: F401 + +from data_bus.minio_lifecycle import ( # noqa: E402 + DEFAULT_CONFIG_PATH, + MinioLifecycle, +) + +CONFIG = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "config", "minio.template.yaml", +) + + +class TestConfigLoad(unittest.TestCase): + """模板配置资产加载与摘要。""" + + def test_from_template_config(self): + life = MinioLifecycle.from_template_config(CONFIG) + self.assertEqual(life._tpl.template, "ti-cl4") + self.assertEqual(life.bucket(), "ti-cl4-artifacts") + self.assertEqual(life.mc_alias, "local") + brief = life.brief() + self.assertEqual(brief["bucket"], "ti-cl4-artifacts") + prefixes = [r["prefix"] for r in brief["lifecycle_rules"]] + self.assertIn("features/", prefixes) + self.assertIn("models/", prefixes) + + def test_default_config_path_exists(self): + self.assertTrue(os.path.isfile(DEFAULT_CONFIG_PATH)) + + +class TestBucketNaming(unittest.TestCase): + """桶命名与对象键（复用 templating 命名，与 #4 一致）。""" + + def setUp(self): + self.life = MinioLifecycle(template="ti-cl4") + + def test_bucket_format(self): + self.assertEqual(self.life.bucket(), "ti-cl4-artifacts") + + def test_bucket_suffix_custom(self): + self.assertEqual( + MinioLifecycle(template="resin", bucket_suffix="store").bucket(), + "resin-store") + + def test_snapshot_key(self): + # model_id 经 sanitize_sql 清洗（`-` → `_`，与 templating 一致） + self.assertEqual( + self.life.snapshot_key("quality-forecast", "2026-08-04", 3), + "features/quality_forecast/2026-08-04/000003.jsonl") + + def test_model_artifact_key(self): + self.assertEqual( + self.life.model_artifact_key("quality-forecast", "v1.2"), + "models/quality_forecast/v1.2/model.bin") + + +class TestBucketCreateStatements(unittest.TestCase): + """mc mb 幂等建桶命令。""" + + def test_create_statement(self): + life = MinioLifecycle(template="ti-cl4") + self.assertEqual( + life.bucket_create_statements(), + ["mc mb --ignore-existing local/ti-cl4-artifacts"]) + + def test_create_statement_custom_alias(self): + life = MinioLifecycle(template="ti-cl4", mc_alias="cluster") + self.assertEqual( + life.bucket_create_statements(), + ["mc mb --ignore-existing cluster/ti-cl4-artifacts"]) + + +class TestLifecycleCli(unittest.TestCase): + """mc ilm add 生命周期命令（按前缀 + 过期天数）。""" + + def test_cli_statements(self): + life = MinioLifecycle(template="ti-cl4", lifecycle=[ + {"prefix": "features/", "expiration_days": 180}, + {"prefix": "models/", "expiration_days": 1095}, + ]) + cmds = life.lifecycle_cli_statements() + self.assertEqual(len(cmds), 2) + self.assertIn('mc ilm add --expiry-days 180 --prefix "features/" local/ti-cl4-artifacts', cmds) + self.assertIn('mc ilm add --expiry-days 1095 --prefix "models/" local/ti-cl4-artifacts', cmds) + + def test_no_rules_no_commands(self): + life = MinioLifecycle(template="ti-cl4", lifecycle=[]) + self.assertEqual(life.lifecycle_cli_statements(), []) + + +class TestLifecyclePolicyJson(unittest.TestCase): + """S3 LifecycleConfiguration 兼容 JSON 策略（MinIO 支持）。""" + + def test_policy_json_structure(self): + life = MinioLifecycle(template="ti-cl4", lifecycle=[ + {"id": "features-expire", "prefix": "features/", "expiration_days": 180}, + ]) + policy = json.loads(life.lifecycle_policy_json()) + rules = policy["Rules"] + self.assertEqual(len(rules), 1) + rule = rules[0] + self.assertEqual(rule["ID"], "features-expire") + self.assertEqual(rule["Status"], "Enabled") + self.assertEqual(rule["Filter"]["Prefix"], "features/") + self.assertEqual(rule["Expiration"]["Days"], 180) + + def test_policy_json_roundtrip(self): + life = MinioLifecycle.from_template_config(CONFIG) + policy = json.loads(life.lifecycle_policy_json()) + self.assertEqual(len(policy["Rules"]), 2) + + +class TestRuleNormalization(unittest.TestCase): + """规则规范化：缺省 id 派生、非法行过滤。""" + + def test_default_id_from_prefix(self): + life = MinioLifecycle(template="ti-cl4", lifecycle=[ + {"prefix": "features/", "expiration_days": 180}, + ]) + self.assertEqual(life.rules[0]["id"], "features/expire") + + def test_invalid_rows_dropped(self): + life = MinioLifecycle(template="ti-cl4", lifecycle=[ + {"prefix": "", "expiration_days": 180}, # 无前缀 → 丢弃 + {"prefix": "models/", "expiration_days": 0}, # 过期天数非法 → 丢弃 + {"prefix": "logs/", "expiration_days": 30}, # 合法 + ]) + self.assertEqual(len(life.rules), 1) + self.assertEqual(life.rules[0]["prefix"], "logs/") + + +if __name__ == "__main__": + unittest.main()