fix: 对齐技术架构补齐传输 mTLS 与结构化 JSON 日志（NFR 9 章）
架构核对发现 2 处差距，本次补齐： - Kafka 上行支持 SSL/SASL_SSL 双向 mTLS（8.2 服务间 mTLS 边缘网关↔总线） - 网关日志支持结构化 JSON 输出（NFR 可维护：统一日志规范） - 配置示例与 README 验收口径同步更新
This commit is contained in:
@@ -0,0 +1,9 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""iAOP-Core · 数据总线 + 时序库（Data Bus）—— 模板化封装。
|
||||
|
||||
对应 PRD 5.2「② 数据总线 + 时序库」与 Issue #4（EPIC）：
|
||||
复用 Kafka + TDengine + PostgreSQL + MinIO，全部改为**按模板配置**
|
||||
topic / 时序库表 / 关系表 / 对象桶，换行业只改模板资产（YAML + 点位字典），
|
||||
内核代码零改动；批量写入缓冲 + 幂等去重，保障「数据不丢不重」。
|
||||
"""
|
||||
__version__ = "0.1.0"
|
||||
@@ -0,0 +1,116 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""PostgreSQL 关系表 schema（模板 / 模型 / 用户 / 权限）—— 子任务 #30。
|
||||
|
||||
按模板自动生成隔离 schema（`tpl_{tpl}`），内含四张基础关系表：
|
||||
- templates 模板注册表（行业模板配置资产）；
|
||||
- models 模型注册表（模型版本 / 算法 / 超参包）；
|
||||
- users 用户表（角色 + 细粒度权限 JSONB）；
|
||||
- permissions 角色-资源-动作权限矩阵。
|
||||
|
||||
另生成角色授权语句（GRANT），按模板配置声明 databus_ro / databus_rw 等角色。
|
||||
不依赖 psycopg2：仅产出 DDL 文本，供模板配置台预览 / 运维执行。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import List, Optional
|
||||
|
||||
from .templating import TemplateNaming
|
||||
|
||||
# 默认基础表清单（对应子任务 #30「模板/模型/用户/权限」）
|
||||
DEFAULT_TABLES: List[str] = ["templates", "models", "users", "permissions"]
|
||||
|
||||
# 表 → 列定义（可被模板 YAML 的 schema_tables 覆盖为子集/扩展）
|
||||
TABLE_COLUMNS: dict = {
|
||||
"templates": [
|
||||
"id BIGSERIAL PRIMARY KEY",
|
||||
"template VARCHAR(64) NOT NULL UNIQUE",
|
||||
"version VARCHAR(32) NOT NULL DEFAULT '0.1.0'",
|
||||
"config JSONB NOT NULL DEFAULT '{}'::jsonb",
|
||||
"status VARCHAR(16) NOT NULL DEFAULT 'draft'",
|
||||
"created_at TIMESTAMPTZ NOT NULL DEFAULT now()",
|
||||
],
|
||||
"models": [
|
||||
"id BIGSERIAL PRIMARY KEY",
|
||||
"model_id VARCHAR(128) NOT NULL UNIQUE",
|
||||
"template VARCHAR(64) REFERENCES {schema}.templates(template) ON DELETE CASCADE",
|
||||
"algorithm VARCHAR(32) NOT NULL DEFAULT 'xgboost'",
|
||||
"hyperparams JSONB NOT NULL DEFAULT '{}'::jsonb",
|
||||
"version VARCHAR(32) NOT NULL DEFAULT '0.1.0'",
|
||||
"status VARCHAR(16) NOT NULL DEFAULT 'staging'",
|
||||
"created_at TIMESTAMPTZ NOT NULL DEFAULT now()",
|
||||
],
|
||||
"users": [
|
||||
"id BIGSERIAL PRIMARY KEY",
|
||||
"username VARCHAR(64) NOT NULL UNIQUE",
|
||||
"role VARCHAR(32) NOT NULL DEFAULT 'viewer'",
|
||||
"permissions JSONB NOT NULL DEFAULT '[]'::jsonb",
|
||||
"created_at TIMESTAMPTZ NOT NULL DEFAULT now()",
|
||||
],
|
||||
"permissions": [
|
||||
"id BIGSERIAL PRIMARY KEY",
|
||||
"role VARCHAR(32) NOT NULL",
|
||||
"resource VARCHAR(64) NOT NULL",
|
||||
"action VARCHAR(16) NOT NULL",
|
||||
"UNIQUE (role, resource, action)",
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def generate_schema_ddl(
|
||||
naming: TemplateNaming,
|
||||
tables: Optional[List[str]] = None,
|
||||
table_columns: Optional[dict] = None,
|
||||
) -> str:
|
||||
"""生成 schema 与基础关系表的完整 DDL（幂等 IF NOT EXISTS）。
|
||||
|
||||
Args:
|
||||
naming: 模板命名器（决定 schema 名）。
|
||||
tables: 表清单（默认 DEFAULT_TABLES）。
|
||||
table_columns: 表→列定义覆盖（默认 TABLE_COLUMNS）。
|
||||
|
||||
Returns:
|
||||
多语句 DDL 文本（含 CREATE SCHEMA / CREATE TABLE）。
|
||||
"""
|
||||
names = tables or DEFAULT_TABLES
|
||||
cols = table_columns or TABLE_COLUMNS
|
||||
schema = naming.pg_schema()
|
||||
|
||||
lines = [f"CREATE SCHEMA IF NOT EXISTS {schema};", ""]
|
||||
for table in names:
|
||||
columns = cols.get(table)
|
||||
if not columns:
|
||||
continue
|
||||
lines.append(f"CREATE TABLE IF NOT EXISTS {schema}.{table} (")
|
||||
lines.append(" " + ",\n ".join(c.format(schema=schema) for c in columns))
|
||||
lines.append(");")
|
||||
lines.append("")
|
||||
return "\n".join(lines).rstrip() + "\n"
|
||||
|
||||
|
||||
def generate_grant_ddl(
|
||||
naming: TemplateNaming,
|
||||
roles: Optional[List[str]] = None,
|
||||
) -> List[str]:
|
||||
"""按模板声明的关系角色生成授权语句。
|
||||
|
||||
角色约定：`{role}_ro` 只读、`{role}_rw` 读写（可扩展）。
|
||||
Args:
|
||||
naming: 模板命名器。
|
||||
roles: 角色清单（如 ["databus"] → databus_ro / databus_rw）。
|
||||
Returns:
|
||||
GRANT 语句列表。
|
||||
"""
|
||||
schema = naming.pg_schema()
|
||||
grants: List[str] = []
|
||||
for role in (roles or ["databus"]):
|
||||
for suffix, privileges in (("ro", "SELECT"), ("rw", "SELECT, INSERT, UPDATE, DELETE")):
|
||||
role_name = f"{role}_{suffix}"
|
||||
grants.append(f"GRANT USAGE ON SCHEMA {schema} TO {role_name};")
|
||||
grants.append(
|
||||
f"GRANT {privileges} ON ALL TABLES IN SCHEMA {schema} TO {role_name};"
|
||||
)
|
||||
if suffix == "rw":
|
||||
grants.append(
|
||||
f"GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA {schema} TO {role_name};"
|
||||
)
|
||||
return grants
|
||||
@@ -0,0 +1,156 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""TDengine 超级表 schema 自动生成 + 批量 INSERT —— 依据点位字典（子任务 #29）。
|
||||
|
||||
模型（TDengine 2.0/3.0 通用）：
|
||||
- 一张超级表 `{tpl}_points`：列 `ts / value / quality`，标签
|
||||
`device_id / unit / data_type`；
|
||||
- 点位字典每个测点自动生成一张子表 `{tpl}_pt_{point_id}`（点位维度，
|
||||
字典 CSV 变更即重建子表集，内核零改动）；
|
||||
- 批量写入：按子表聚合多行 `INSERT INTO {sub} VALUES (...),(...);`，
|
||||
单条样本约 512B 时对齐 PRD 5.2「5k 条/100ms」基线。
|
||||
|
||||
不依赖 taospy：本模块只负责生成 DDL / 批量 SQL；真实落库由
|
||||
`batch_writer.StoreSink` 的客户端实现（缺失时降级内存 sink 联调）。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
from .templating import TemplateNaming, sanitize_sql, _sql_str
|
||||
|
||||
# dataType → TDengine 列类型（点位字典 schema.VALID_DATA_TYPES 子集）
|
||||
TD_VALUE_TYPE: Dict[str, str] = {
|
||||
"float": "DOUBLE",
|
||||
"int": "BIGINT",
|
||||
"bool": "BOOL",
|
||||
}
|
||||
DEFAULT_VALUE_TYPE = "DOUBLE"
|
||||
|
||||
# 样本记录字段（与 edge-gateway spool / kafka_sink 保持一致）
|
||||
ROW_KEYS = ("device_id", "point_id", "value", "ts", "quality")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PointSpec:
|
||||
"""点位字典中的最小维度信息（schema 生成所需）。"""
|
||||
|
||||
device_id: str
|
||||
point_id: str
|
||||
unit: str = ""
|
||||
data_type: str = "float"
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, raw: dict) -> "PointSpec":
|
||||
return cls(
|
||||
device_id=str(raw.get("device_id") or ""),
|
||||
point_id=str(raw.get("point_id") or ""),
|
||||
unit=str(raw.get("unit") or ""),
|
||||
data_type=str(raw.get("data_type") or raw.get("dataType") or "float"),
|
||||
)
|
||||
|
||||
|
||||
def value_type(data_type: str) -> str:
|
||||
return TD_VALUE_TYPE.get((data_type or "").lower(), DEFAULT_VALUE_TYPE)
|
||||
|
||||
|
||||
def generate_supertable_ddl(
|
||||
naming: TemplateNaming,
|
||||
retention_days: Optional[int] = None,
|
||||
) -> str:
|
||||
"""生成超级表 DDL（自动建表，幂等 IF NOT EXISTS）。
|
||||
|
||||
Args:
|
||||
naming: 模板命名器（决定超级表名）。
|
||||
retention_days: 数据保留天数（TDengine KEEP 表选项，可选）。
|
||||
|
||||
Returns:
|
||||
CREATE STABLE 语句。
|
||||
"""
|
||||
ddl = (
|
||||
f"CREATE STABLE IF NOT EXISTS {naming.stable()} (\n"
|
||||
" ts TIMESTAMP,\n"
|
||||
" value DOUBLE,\n"
|
||||
" quality TINYINT\n"
|
||||
") TAGS (\n"
|
||||
" device_id NCHAR(64),\n"
|
||||
" unit NCHAR(16),\n"
|
||||
" data_type NCHAR(16)\n"
|
||||
")"
|
||||
)
|
||||
if retention_days and int(retention_days) > 0:
|
||||
ddl += f" KEEP({int(retention_days)})"
|
||||
return ddl + ";"
|
||||
|
||||
|
||||
def generate_subtable_ddls(
|
||||
naming: TemplateNaming,
|
||||
points: List[PointSpec],
|
||||
use_typed_value: bool = True,
|
||||
) -> List[str]:
|
||||
"""按点位字典为每个测点生成子表 DDL（依据点位字典自动生成）。
|
||||
|
||||
Args:
|
||||
naming: 模板命名器。
|
||||
points: 点位字典（测点维度集合）。
|
||||
use_typed_value: 是否按 dataType 派生 value 列类型（TDengine 3.x
|
||||
支持列级类型；2.x 超级表统一 DOUBLE，传 False 时忽略）。
|
||||
|
||||
Returns:
|
||||
CREATE TABLE ... USING ... TAGS(...) 语句列表（顺序与 points 一致）。
|
||||
"""
|
||||
if use_typed_value:
|
||||
# 说明：超级表 value 列按最宽类型 DOUBLE 建（TDengine 列类型在
|
||||
# CREATE STABLE 时固定），子表继承，这里保留 data_type 标签供聚合。
|
||||
pass
|
||||
out: List[str] = []
|
||||
seen: set = set()
|
||||
for p in points:
|
||||
sub = naming.subtable(p.point_id)
|
||||
if sub in seen:
|
||||
continue
|
||||
seen.add(sub)
|
||||
tags = ", ".join(
|
||||
_sql_str(v) for v in (p.device_id, p.unit or "", (p.data_type or "float").lower())
|
||||
)
|
||||
out.append(
|
||||
f"CREATE TABLE IF NOT EXISTS {sub} USING {naming.stable()} "
|
||||
f"TAGS ({tags});"
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def build_batch_insert(
|
||||
naming: TemplateNaming,
|
||||
rows: List[dict],
|
||||
) -> List[str]:
|
||||
"""把样本批聚合为按子表分组的批量 INSERT 语句。
|
||||
|
||||
Args:
|
||||
naming: 模板命名器（决定子表名）。
|
||||
rows: 样本列表，每条含 device_id/point_id/value/ts(/quality)。
|
||||
|
||||
Returns:
|
||||
SQL 语句列表：每子表一条 `INSERT INTO {sub} VALUES (...),(...);`。
|
||||
空输入返回空列表。
|
||||
"""
|
||||
grouped: Dict[str, List[dict]] = {}
|
||||
for row in rows:
|
||||
point_id = str(row.get("point_id") or "")
|
||||
sub = naming.subtable(point_id)
|
||||
grouped.setdefault(sub, []).append(row)
|
||||
|
||||
statements: List[str] = []
|
||||
for sub in sorted(grouped):
|
||||
tuples = []
|
||||
for row in grouped[sub]:
|
||||
ts = row.get("ts")
|
||||
ts_ms = int(ts * 1000) if isinstance(ts, float) else int(ts)
|
||||
value = row.get("value")
|
||||
value_sql = "NULL" if value is None else repr(float(value))
|
||||
quality = int(row.get("quality", 1))
|
||||
tuples.append(f"({ts_ms}, {value_sql}, {quality})")
|
||||
statements.append(
|
||||
f"INSERT INTO {sub} VALUES " + ", ".join(tuples) + ";"
|
||||
)
|
||||
return statements
|
||||
@@ -0,0 +1,123 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""数据总线命名与分区模板化 —— 换行业只改配置，内核零改动（PRD 5.2 / Issue #4）。
|
||||
|
||||
模板（template）是行业复制的唯一粒度，所有外部资源命名均由
|
||||
「模板名 + 点位维度」推导：
|
||||
|
||||
- Kafka topic ：`{topic_prefix}.{device_id}.points`（与 edge-gateway 上行一致）；
|
||||
分区 = hash(device_id) % num_partitions，保证单设备分区内有序；
|
||||
- TDengine ：超级表 `{tpl}_points`、每测点子表 `{tpl}_pt_{point_id}`；
|
||||
- PostgreSQL ：schema `tpl_{tpl}`，表名 `{schema}.{table}`；
|
||||
- MinIO ：桶 `{template}-artifacts`，对象键 `features/{model_id}/{...}`。
|
||||
|
||||
命名清洗规则：
|
||||
- Kafka topic / MinIO 桶：小写 + 保留 `-` / `_` / `.` 之外替换为 `_`；
|
||||
- SQL 标识符（TDengine / PostgreSQL）：额外把 `-` 替换为 `_`，避免引号转义。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import re
|
||||
from typing import List, Optional
|
||||
|
||||
# 保留字符集（Kafka topic / MinIO 桶名均允许小写字母、数字、- _ .）
|
||||
_KEEP = re.compile(r"[^a-z0-9_.-]+")
|
||||
# SQL 标识符额外排除 `-`（TDengine / PostgreSQL 不加引号时不允许）
|
||||
_SQL_KEEP = re.compile(r"[^a-z0-9_.]+")
|
||||
|
||||
|
||||
def sanitize(name: str) -> str:
|
||||
"""通用命名清洗：小写 + 非安全字符替换为 `_`。"""
|
||||
s = (name or "").strip().lower()
|
||||
s = _KEEP.sub("_", s)
|
||||
s = s.strip("._") or "tpl"
|
||||
return s
|
||||
|
||||
|
||||
def sanitize_sql(name: str) -> str:
|
||||
"""SQL 标识符清洗：`-` 与其余非安全字符替换为 `_`。"""
|
||||
s = sanitize(name)
|
||||
s = _SQL_KEEP.sub("_", s)
|
||||
return s.strip("._") or "tpl"
|
||||
|
||||
|
||||
def _sql_str(value: str) -> str:
|
||||
"""SQL 字符串字面量转义（单引号加倍）。"""
|
||||
return "'" + str(value).replace("'", "''") + "'"
|
||||
|
||||
|
||||
class TemplateNaming:
|
||||
"""按模板推导全部外部资源命名（Kafka / TDengine / PostgreSQL / MinIO）。"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
template: str,
|
||||
topic_prefix: Optional[str] = None,
|
||||
num_partitions: int = 12,
|
||||
bucket_suffix: str = "artifacts",
|
||||
):
|
||||
self.template = sanitize(template)
|
||||
# SQL 标识符用下划线形态（避免 `-` 需加引号）
|
||||
self.tpl_sql = sanitize_sql(self.template)
|
||||
self.topic_prefix = sanitize(topic_prefix) if topic_prefix else self.template
|
||||
self.num_partitions = max(1, int(num_partitions))
|
||||
self.bucket_suffix = sanitize_sql(bucket_suffix)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Kafka：topic 命名 / 分区策略（子任务 #28）
|
||||
# ------------------------------------------------------------------
|
||||
def topic(self, device_id: str) -> str:
|
||||
"""上行 topic：`{topic_prefix}.{device_id}.points`（对齐 edge-gateway）。"""
|
||||
return f"{self.topic_prefix}.{sanitize(device_id)}.points"
|
||||
|
||||
def partition(self, device_id: str, num_partitions: Optional[int] = None) -> int:
|
||||
"""分区策略：按 device_id 一致性哈希 → 单设备分区内严格有序。"""
|
||||
n = num_partitions or self.num_partitions
|
||||
digest = hashlib.md5(sanitize(device_id).encode("utf-8")).hexdigest()
|
||||
return int(digest[:8], 16) % n
|
||||
|
||||
def partitions(self, device_ids: List[str], num_partitions: Optional[int] = None) -> dict:
|
||||
"""设备 → 分区映射（模板配置台预览用）。"""
|
||||
n = num_partitions or self.num_partitions
|
||||
return {d: self.partition(d, n) for d in device_ids}
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# TDengine：超级表 + 每测点子表（子任务 #29）
|
||||
# ------------------------------------------------------------------
|
||||
def stable(self) -> str:
|
||||
"""时序超级表：`{tpl}_points`。"""
|
||||
return f"{self.tpl_sql}_points"
|
||||
|
||||
def subtable(self, point_id: str) -> str:
|
||||
"""测点子表：`{tpl}_pt_{point_id}`（点位维度，字典驱动自动生成）。"""
|
||||
return f"{self.tpl_sql}_pt_{sanitize_sql(point_id)}"
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# PostgreSQL：schema 与表（子任务 #30）
|
||||
# ------------------------------------------------------------------
|
||||
def pg_schema(self) -> str:
|
||||
"""关系 schema：`tpl_{tpl}`。"""
|
||||
return f"tpl_{self.tpl_sql}"
|
||||
|
||||
def pg_table(self, table: str) -> str:
|
||||
"""`{schema}.{table}` 限定名。"""
|
||||
return f"{self.pg_schema()}.{sanitize_sql(table)}"
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# MinIO：对象桶与对象键（子任务 #31）
|
||||
# ------------------------------------------------------------------
|
||||
def bucket(self) -> str:
|
||||
"""对象存储桶：`{template}-artifacts`（S3 桶名允许 `-`）。"""
|
||||
return f"{self.template}-{self.bucket_suffix}"
|
||||
|
||||
def snapshot_key(self, model_id: str, date: str, seq: int) -> str:
|
||||
"""特征快照对象键：`features/{model_id}/{date}/{seq:06d}.jsonl`。"""
|
||||
return f"features/{sanitize_sql(model_id)}/{date}/{int(seq):06d}.jsonl"
|
||||
|
||||
def model_artifact_key(self, model_id: str, version: str) -> str:
|
||||
"""模型 artifact 对象键：`models/{model_id}/{version}/model.bin`。"""
|
||||
return f"models/{sanitize_sql(model_id)}/{sanitize_sql(version)}/model.bin"
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
def _sql_str(self, value: str) -> str:
|
||||
return _sql_str(value)
|
||||
@@ -12,7 +12,8 @@
|
||||
| 采集 P99 延迟 | ≤ 1.8s（600 点位 1Hz） | `collector/metrics.py` 统计，`engine.py` 调度 |
|
||||
| 丢失率 | ≤ 0.02% | `collector/spool.py` 断点续传 + 背压丢弃计数 |
|
||||
| 可用性 | ≥ 99.8% | `collector/metrics.py` 轮次成功率 |
|
||||
| 安全 | 零控制指令下发 | 驱动抽象仅暴露只读读接口（`drivers/base.py`） |
|
||||
| 安全-网络 | 严格只读、零控制指令；传输 mTLS | 驱动仅只读接口；`upstream/kafka_sink.py` 支持 SSL/SASL_SSL 双向 mTLS |
|
||||
| 可维护 | 统一日志规范（结构化 JSON） | `main.py --log-json` 输出结构化 JSON 日志 |
|
||||
|
||||
## 目录结构
|
||||
|
||||
|
||||
@@ -51,3 +51,14 @@ kafka:
|
||||
bootstrap_servers: "10.20.0.10:9092"
|
||||
topic_prefix: "iaop.ti-cl4" # 模板化 topic：{prefix}.{device_id}.points
|
||||
batch_size: 500
|
||||
# 传输安全（NFR 9 章：服务间 mTLS，边缘网关↔总线双向认证）。
|
||||
# 现场部署必须开启；本地联调可保持 PLAINTEXT。
|
||||
security:
|
||||
protocol: SSL # PLAINTEXT | SSL | SASL_SSL
|
||||
ca_location: "/etc/iaop/certs/ca.crt"
|
||||
cert_location: "/etc/iaop/certs/gateway.crt"
|
||||
key_location: "/etc/iaop/certs/gateway.key"
|
||||
# protocol: SASL_SSL 时追加：
|
||||
# sasl_mechanism: "PLAIN"
|
||||
# sasl_username: "edge-gateway"
|
||||
# sasl_password: "..."
|
||||
|
||||
@@ -35,6 +35,8 @@ def parse_args(argv: List[str]) -> argparse.Namespace:
|
||||
parser.add_argument("--rounds", type=int, default=0,
|
||||
help="采集轮数上限（0=无限，调试用）")
|
||||
parser.add_argument("--verbose", action="store_true", help="输出调试日志")
|
||||
parser.add_argument("--log-json", action="store_true",
|
||||
help="结构化 JSON 日志（NFR 9 章统一日志规范；生产模式建议开启）")
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
@@ -78,7 +80,7 @@ def build_engine(config: dict, point_dict: PointDict):
|
||||
|
||||
|
||||
def build_sink(config: dict, spool) -> object:
|
||||
"""按模板配置组装 Kafka 上行通道。"""
|
||||
"""按模板配置组装 Kafka 上行通道（含 mTLS 传输安全，NFR 9 章）。"""
|
||||
from upstream import KafkaSink
|
||||
|
||||
kafka_cfg = config.get("kafka", {})
|
||||
@@ -87,11 +89,33 @@ def build_sink(config: dict, spool) -> object:
|
||||
topic_prefix=kafka_cfg.get("topic_prefix", "iaop"),
|
||||
spool=spool,
|
||||
batch_size=int(kafka_cfg.get("batch_size", 500)),
|
||||
security=kafka_cfg.get("security") or {},
|
||||
)
|
||||
|
||||
|
||||
def main(argv: List[str]) -> int:
|
||||
args = parse_args(argv)
|
||||
|
||||
if args.log_json:
|
||||
# NFR 9 章「可维护」：统一日志规范 —— 结构化 JSON，便于采集到统一日志平台
|
||||
import json as _json
|
||||
|
||||
class JsonFormatter(logging.Formatter):
|
||||
def format(self, record):
|
||||
payload = {
|
||||
"ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
|
||||
"level": record.levelname,
|
||||
"logger": record.name,
|
||||
"msg": record.getMessage(),
|
||||
}
|
||||
if record.exc_info:
|
||||
payload["exc"] = self.formatException(record.exc_info)
|
||||
return _json.dumps(payload, ensure_ascii=False)
|
||||
|
||||
handler = logging.StreamHandler()
|
||||
handler.setFormatter(JsonFormatter())
|
||||
logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, handlers=[handler])
|
||||
else:
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG if args.verbose else logging.INFO,
|
||||
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
|
||||
|
||||
@@ -29,11 +29,24 @@ class KafkaSink:
|
||||
topic_prefix: str,
|
||||
spool: SpoolStore,
|
||||
batch_size: int = 500,
|
||||
security: Optional[dict] = None,
|
||||
):
|
||||
"""
|
||||
Args:
|
||||
bootstrap_servers: Kafka 地址列表（逗号分隔）；
|
||||
topic_prefix: 上行 topic 前缀（模板化 `{prefix}.{device}.points`）；
|
||||
spool: 本地缓存（断点续传）；
|
||||
batch_size: 单批上限；
|
||||
security: 传输安全配置（NFR 9 章：服务间 mTLS）。
|
||||
约定字段：protocol(PLAINTEXT|SSL|SASL_SSL)、ca_location、
|
||||
cert_location、key_location、sasl_username、sasl_password。
|
||||
其中 SSL/SASL_SSL + ca/cert/key = 边缘网关↔总线双向 mTLS。
|
||||
"""
|
||||
self.bootstrap_servers = bootstrap_servers
|
||||
self.topic_prefix = topic_prefix
|
||||
self.spool = spool
|
||||
self.batch_size = max(1, batch_size)
|
||||
self.security = security or {}
|
||||
self._producer = None
|
||||
self._degraded = False # True = Kafka 不可用，仅保留 spool
|
||||
self._lock = threading.Lock()
|
||||
@@ -42,6 +55,27 @@ class KafkaSink:
|
||||
self._connect()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
def _producer_config(self) -> dict:
|
||||
"""组装 confluent-kafka producer 配置（含 mTLS/SASL 透传）。"""
|
||||
conf = {"bootstrap.servers": self.bootstrap_servers}
|
||||
protocol = self.security.get("protocol", "PLAINTEXT").upper()
|
||||
conf["security.protocol"] = protocol
|
||||
if protocol in ("SSL", "SASL_SSL"):
|
||||
for key, kafka_key in (
|
||||
("ca_location", "ssl.ca.location"),
|
||||
("cert_location", "ssl.certificate.location"),
|
||||
("key_location", "ssl.key.location"),
|
||||
("key_password", "ssl.key.password"),
|
||||
):
|
||||
if self.security.get(key):
|
||||
conf[kafka_key] = self.security[key]
|
||||
if protocol == "SASL_SSL":
|
||||
conf["sasl.mechanism"] = self.security.get("sasl_mechanism", "PLAIN")
|
||||
if self.security.get("sasl_username"):
|
||||
conf["sasl.username"] = self.security["sasl_username"]
|
||||
conf["sasl.password"] = self.security.get("sasl_password", "")
|
||||
return conf
|
||||
|
||||
def _connect(self) -> None:
|
||||
"""尝试连接 Kafka；失败则降级（不阻断采集）。"""
|
||||
try:
|
||||
@@ -51,7 +85,7 @@ class KafkaSink:
|
||||
self._degraded = True
|
||||
return
|
||||
try:
|
||||
self._producer = Producer({"bootstrap.servers": self.bootstrap_servers})
|
||||
self._producer = Producer(self._producer_config())
|
||||
except Exception as exc:
|
||||
logger.warning("Kafka 初始化失败(%s)，降级为 spool-only 模式", exc)
|
||||
self._degraded = True
|
||||
|
||||
Reference in New Issue
Block a user