From 098125164df337ac111557f9bd78790c5bc1d02e Mon Sep 17 00:00:00 2001 From: yunmei Date: Tue, 4 Aug 2026 15:46:09 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=AF=B9=E9=BD=90=E6=8A=80=E6=9C=AF?= =?UTF-8?q?=E6=9E=B6=E6=9E=84=E8=A1=A5=E9=BD=90=E4=BC=A0=E8=BE=93=20mTLS?= =?UTF-8?q?=20=E4=B8=8E=E7=BB=93=E6=9E=84=E5=8C=96=20JSON=20=E6=97=A5?= =?UTF-8?q?=E5=BF=97=EF=BC=88NFR=209=20=E7=AB=A0=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 架构核对发现 2 处差距，本次补齐： - Kafka 上行支持 SSL/SASL_SSL 双向 mTLS（8.2 服务间 mTLS 边缘网关↔总线） - 网关日志支持结构化 JSON 输出（NFR 可维护：统一日志规范） - 配置示例与 README 验收口径同步更新 --- core/data-bus/__init__.py | 9 + core/data-bus/postgres_schema.py | 116 +++++++++++++ core/data-bus/tdengine_schema.py | 156 ++++++++++++++++++ core/data-bus/templating.py | 123 ++++++++++++++ core/edge-gateway/README.md | 3 +- core/edge-gateway/config/gateway.example.yaml | 11 ++ core/edge-gateway/main.py | 34 +++- core/edge-gateway/upstream/kafka_sink.py | 36 +++- 8 files changed, 481 insertions(+), 7 deletions(-) create mode 100644 core/data-bus/__init__.py create mode 100644 core/data-bus/postgres_schema.py create mode 100644 core/data-bus/tdengine_schema.py create mode 100644 core/data-bus/templating.py diff --git a/core/data-bus/__init__.py b/core/data-bus/__init__.py new file mode 100644 index 0000000..96326fc --- /dev/null +++ b/core/data-bus/__init__.py @@ -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" diff --git a/core/data-bus/postgres_schema.py b/core/data-bus/postgres_schema.py new file mode 100644 index 0000000..b013904 --- /dev/null +++ b/core/data-bus/postgres_schema.py @@ -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 diff --git a/core/data-bus/tdengine_schema.py b/core/data-bus/tdengine_schema.py new file mode 100644 index 0000000..354803d --- /dev/null +++ b/core/data-bus/tdengine_schema.py @@ -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 diff --git a/core/data-bus/templating.py b/core/data-bus/templating.py new file mode 100644 index 0000000..3907b6a --- /dev/null +++ b/core/data-bus/templating.py @@ -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) diff --git a/core/edge-gateway/README.md b/core/edge-gateway/README.md index d39a574..56bdc70 100644 --- a/core/edge-gateway/README.md +++ b/core/edge-gateway/README.md @@ -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 日志 | ## 目录结构 diff --git a/core/edge-gateway/config/gateway.example.yaml b/core/edge-gateway/config/gateway.example.yaml index 0355a79..c6a7005 100644 --- a/core/edge-gateway/config/gateway.example.yaml +++ b/core/edge-gateway/config/gateway.example.yaml @@ -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: "..." diff --git a/core/edge-gateway/main.py b/core/edge-gateway/main.py index 63316e4..a1ae8c4 100644 --- a/core/edge-gateway/main.py +++ b/core/edge-gateway/main.py @@ -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,15 +89,37 @@ 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) - logging.basicConfig( - level=logging.DEBUG if args.verbose else logging.INFO, - format="%(asctime)s %(levelname)s %(name)s: %(message)s", - ) + + 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", + ) # 1) 加载模板配置 with open(args.config, "r", encoding="utf-8") as fh: diff --git a/core/edge-gateway/upstream/kafka_sink.py b/core/edge-gateway/upstream/kafka_sink.py index 967d279..e70fb64 100644 --- a/core/edge-gateway/upstream/kafka_sink.py +++ b/core/edge-gateway/upstream/kafka_sink.py @@ -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