diff --git a/README.md b/README.md index f6fcc9d..5395f96 100644 --- a/README.md +++ b/README.md @@ -68,6 +68,25 @@ class MyPlugin(Plugin): # url = "postgresql+asyncpg://user:pwd@127.0.0.1:5432/mydb" # generate_schemas = true +# 对象存储接入(可选):在插件 config.toml 配置 storage 段后, +# 基类 init_storage() 自动初始化 self.storage(S3/Minio,直连不经网关):: +# +# # config/workers/my_plugin/config.toml +# [workers.config.storage] +# endpoint = "192.168.1.10:9000" # Minio 地址 host:port(不带协议前缀) +# access_key = "minioadmin" +# secret_key = "minioadmin" +# bucket = "chongming-files" +# secure = false +# # region = "" # 可选 +# # auto_create_bucket = true # 默认 true:启动时确保 bucket 存在 +# +# 语义:未配置 / enabled=false → self.storage 为 None(插件自行降级); +# 缺必填键(endpoint/access_key/secret_key/bucket)→ 启动硬失败(ValueError); +# 连接失败 → 告警降级为 None。API 见 chongming_worker.storage.StorageClient +# (put_file/put_bytes/get_file/get_bytes/delete/presigned_put_url/presigned_get_url +# 及对应 async a* 版本,handler 中直接 await)。 + # 暴露给 app lifespan 的模块级入口(bound method,签名与旧写法一致) plugin = MyPlugin() create_workers = plugin.create_workers @@ -89,6 +108,7 @@ cleanup = plugin.cleanup - **幂等**:重复 `create_workers` 只初始化一次(`asyncio.Lock` 守卫,并发调用安全) - **统一日志**:`logger` 自动带 `[type]` 前缀 - **数据库**:`init_database()` 统一建库建表、记录,`cleanup` 自动 `dispose()`(SQLite 按 `db_name`;config 含 `database.url` 时走 PostgreSQL) +- **对象存储**:`init_storage()` 按 config 的 `storage` 段初始化 `self.storage`(S3/Minio,可选),`cleanup` 自动关闭 - **global_var**:`set_var()` / `get_var()` 统一读写插件 `global_var` 模块;`cleanup` 时自动将已发布的变量重置为 `None`,避免业务代码拿到已释放的资源 - **后台任务**:`spawn_task()` 跟踪,`cleanup` 时自动取消并 `gather` - **清理钩子**:`register_cleanup()` 注册,`cleanup` 时逆序执行 diff --git a/pyproject.toml b/pyproject.toml index c1486cf..1c3b5b5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ requires = ["hatchling"] [project] name = "chongming-worker" -version = "0.2.0" +version = "0.3.0" description = "chongming worker" readme = "README.md" requires-python = ">=3.12" @@ -12,6 +12,7 @@ dependencies = [ "eclipse-zenoh>=1.9.0", "pydantic>=2.0", "tomli>=2.4.1", + "minio>=7.2.0", ] [project.scripts] diff --git a/src/chongming_worker/__init__.py b/src/chongming_worker/__init__.py index 2ab7cc7..8c97e4b 100644 --- a/src/chongming_worker/__init__.py +++ b/src/chongming_worker/__init__.py @@ -25,6 +25,9 @@ from chongming_worker.runtime import ( from chongming_worker.plugin import ( Plugin, ) +from chongming_worker.storage import ( + StorageClient, +) __all__ = [ "RPCMethod", @@ -41,4 +44,5 @@ __all__ = [ "create_worker", "create_workers", "Plugin", + "StorageClient", ] diff --git a/src/chongming_worker/plugin.py b/src/chongming_worker/plugin.py index bc49fe9..7b14dcf 100644 --- a/src/chongming_worker/plugin.py +++ b/src/chongming_worker/plugin.py @@ -50,6 +50,8 @@ Plugin 基类 — 统一 Worker 插件初始化与清理生命周期 - 统一日志:``logger`` 自动带 ``[type]`` 前缀 - 数据库:``_bootstrap()`` / ``init_database()`` 统一建库建表、记录,``cleanup`` 自动 dispose (SQLite 按 ``db_name`` 建库;config 含 ``database.url`` 时优先走 PostgreSQL) + - 对象存储:``_bootstrap()`` / ``init_storage()`` 按 config 的 storage 段初始化 + (S3/Minio,``[workers.config.storage]``),``cleanup`` 自动关闭 - global_var:``set_var()`` 写入插件自己的 ``global_var`` 模块(白名单由插件决定) - 后台任务:``spawn_task()`` 跟踪,``cleanup`` 时自动取消 - 清理钩子:``register_cleanup()`` 注册,``cleanup`` 时逆序执行 @@ -90,6 +92,9 @@ class Plugin: self.config: dict = {} #: Database 实例(由 init_database 填充;类型标注为 Any 以避免强依赖 chongming_database) self.db: Any = None + #: StorageClient 实例(由 init_storage 按 config 的 storage 段填充;类型标注为 Any + #: 以避免强依赖 minio) + self.storage: Any = None self._initialized = False #: 初始化互斥锁——同一进程内并发 create_workers(如测试模拟多 worker 启动)串行化 self._init_lock = asyncio.Lock() @@ -104,12 +109,14 @@ class Plugin: # ------------------------------------------------------------------ async def _bootstrap(self) -> None: """ - 默认资源引导:按 db_name(SQLite)或 database.url(PostgreSQL)初始化数据库。 + 默认资源引导:按 db_name(SQLite)或 database.url(PostgreSQL)初始化数据库, + 并按 config 的 storage 段初始化对象存储(S3/Minio,可选)。 - 由基类在 on_init 之前自动调用;子类需要完全自定义建库逻辑时覆写此方法。 + 由基类在 on_init 之前自动调用;子类需要完全自定义引导逻辑时覆写此方法。 """ if self.db_name or self._database_config().get("url"): await self.init_database(self.db_name) + await self.init_storage() async def on_init(self) -> None: """首次 create_workers 时调用一次,初始化业务资源(数据库已由 _bootstrap 引导)。""" @@ -183,6 +190,13 @@ class Plugin: if self.db is not None: await self.db.dispose() self.db = None + # 4.5 关闭对象存储客户端(S3/Minio,可选) + if self.storage is not None: + try: + self.storage.close() + except Exception as e: # noqa: BLE001 + self.logger.warning("[%s] 关闭对象存储失败: %s", self.type, e) + self.storage = None # 5. 清空通过 set_var 发布的全局变量,避免 global_var 继续持有已释放资源 for name in list(self._vars): try: @@ -247,6 +261,33 @@ class Plugin: self.logger.info("[%s] 数据库已初始化: %s", self.type, db_path) return db + async def init_storage(self) -> Any: + """ + 初始化对象存储客户端并存入 self.storage(S3/Minio,配置驱动,可选)。 + + 读取运行时配置中的 storage 段(插件 config.toml 的 ``[workers.config.storage]``: + endpoint/access_key/secret_key/bucket/secure/region/auto_create_bucket)。 + + 语义(详见 ``chongming_worker.storage.StorageClient.from_config``): + - 未配置或 ``enabled = false`` → self.storage 保持 None(插件自行降级) + - 配置段存在但缺必填键 → 抛 ValueError(启动硬失败,快速反馈配置错误) + - 连接失败 / bucket 不可用 → 告警并降级为 None + + 返回 StorageClient 或 None。 + """ + from chongming_worker.storage import StorageClient + + storage_conf = self.config.get("storage") if isinstance(self.config, dict) else None + self.storage = StorageClient.from_config( + storage_conf if isinstance(storage_conf, dict) else None + ) + if self.storage is not None: + self.logger.info( + "[%s] 对象存储已就绪: %s bucket=%s", + self.type, self.storage.endpoint, self.storage.bucket, + ) + return self.storage + def set_var(self, name: str, value: Any) -> None: """ 写入运行时变量,并同步到插件 global_var 模块。 diff --git a/src/chongming_worker/storage.py b/src/chongming_worker/storage.py new file mode 100644 index 0000000..40d556f --- /dev/null +++ b/src/chongming_worker/storage.py @@ -0,0 +1,210 @@ +""" +对象存储客户端(S3/Minio)— 配置驱动,进程插件直连,不依赖网关 + +进程插件(external worker)在 ``config/workers//config.toml`` 配置 +``[workers.config.storage]`` 段后,由 Plugin 基类 ``_bootstrap`` 自动初始化 +``self.storage``(StorageClient)。上传 / 下载 / 预签名 URL 直接与 Minio +交互,数据路径完全不经过网关(预签名 URL 模式下网关甚至不参与数据传输)。 + +配置示例:: + + # config/workers//config.toml + [workers.config.storage] + endpoint = "192.168.1.10:9000" # Minio 地址 host:port(不带协议前缀) + access_key = "minioadmin" + secret_key = "minioadmin" + bucket = "chongming-files" + secure = false # false=http(默认),true=https + region = "" # 可选 + auto_create_bucket = true # 可选,默认 true:启动时确保 bucket 存在 + +``StorageClient.from_config`` 语义: + - 无配置段 / ``enabled = false`` → 返回 None(静默关闭) + - 配置段存在但缺必填键(endpoint/access_key/secret_key/bucket)→ 抛 ValueError + (启动硬失败,与契约校验哲学一致,快速反馈配置错误) + - 连接失败 / bucket 不可用 → warning 日志 + 返回 None(降级,与 mqtt_forward 一致) +""" +from __future__ import annotations + +import asyncio +import io +import logging +from datetime import timedelta +from pathlib import Path +from typing import Any + +logger = logging.getLogger("chongming_worker.storage") + +#: 必填配置键(缺失时启动硬失败) +_REQUIRED_KEYS = ("endpoint", "access_key", "secret_key", "bucket") + + +class StorageClient: + """Minio(S3) 客户端包装:同步核心 + async 便捷方法。 + + minio SDK 为 urllib3 同步实现,async 方法用 ``asyncio.to_thread`` 包装, + 供 RPC handler 直接 ``await``;同步方法供后台任务 / 脚本直接使用。 + """ + + def __init__(self, client: Any, bucket: str, endpoint: str) -> None: + self._client = client + self.bucket = bucket + self.endpoint = endpoint + + # ------------------------------------------------------------------ + # 构造 + # ------------------------------------------------------------------ + @classmethod + def from_config(cls, storage_conf: dict | None) -> "StorageClient | None": + """按配置构造客户端;未配置 / 禁用返回 None,配置错误抛 ValueError。 + + Args: + storage_conf: ``[workers.config.storage]`` 段展平后的 dict, + None / 空 dict 表示未启用(静默关闭) + + Returns: + StorageClient;禁用或连接失败(降级)时为 None + + Raises: + ValueError: 配置段存在但缺少必填键(endpoint/access_key/secret_key/bucket) + """ + if not storage_conf or not isinstance(storage_conf, dict): + return None + if not storage_conf.get("enabled", True): + logger.info("storage 已禁用(enabled=false),跳过初始化") + return None + + missing = [k for k in _REQUIRED_KEYS if not storage_conf.get(k)] + if missing: + raise ValueError(f"storage 配置缺少必填项: {', '.join(missing)}") + + try: + from minio import Minio + import urllib3 + except ImportError as e: # pragma: no cover - 依赖缺失属打包错误 + raise RuntimeError( + "缺少 minio 依赖:请确认 chongming-worker 的 minio 依赖已安装" + ) from e + + endpoint = storage_conf["endpoint"] + # 自定义连接池:连接超时 3s(Minio 不可达时快速降级,避免 SDK 默认 + # 长重试拖慢 worker 启动),读超时 60s(大文件传输留足余量) + http_client = urllib3.PoolManager( + timeout=urllib3.Timeout(connect=3.0, read=60.0) + ) + client = Minio( + endpoint, + access_key=storage_conf["access_key"], + secret_key=storage_conf["secret_key"], + secure=bool(storage_conf.get("secure", False)), + region=storage_conf.get("region") or None, + http_client=http_client, + ) + return cls._connect(client, endpoint, storage_conf) + + @classmethod + def _connect( + cls, client: Any, endpoint: str, storage_conf: dict + ) -> "StorageClient | None": + """连接校验 + bucket 确保;失败告警并降级为 None。""" + bucket = storage_conf["bucket"] + try: + exists = client.bucket_exists(bucket) + if not exists: + if storage_conf.get("auto_create_bucket", True): + client.make_bucket(bucket, location=storage_conf.get("region") or None) + else: + raise RuntimeError(f"bucket 不存在: {bucket}") + return cls(client, bucket, endpoint) + except Exception as e: # noqa: BLE001 + logger.warning( + "storage 连接失败,降级为不可用: %s (endpoint=%s, bucket=%s)", + e, endpoint, bucket, + ) + return None + + # ------------------------------------------------------------------ + # 同步核心(minio SDK 为同步实现) + # ------------------------------------------------------------------ + def put_file( + self, object_key: str, local_path: str | Path, content_type: str | None = None + ) -> str: + """上传本地文件,返回 object_key。""" + self._client.fput_object( + self.bucket, object_key, str(local_path), content_type=content_type + ) + return object_key + + def put_bytes( + self, object_key: str, data: bytes, content_type: str | None = None + ) -> str: + """上传内存字节,返回 object_key。""" + self._client.put_object( + self.bucket, object_key, io.BytesIO(data), len(data), + content_type=content_type, + ) + return object_key + + def get_file(self, object_key: str, local_path: str | Path) -> Path: + """下载到本地文件,返回目标路径。""" + self._client.fget_object(self.bucket, object_key, str(local_path)) + return Path(local_path) + + def get_bytes(self, object_key: str) -> bytes: + """下载为内存字节。""" + resp = self._client.get_object(self.bucket, object_key) + try: + return resp.read() + finally: + resp.close() + resp.release_conn() + + def delete(self, object_key: str) -> None: + """删除对象(不存在时由服务端决定,一般静默成功)。""" + self._client.remove_object(self.bucket, object_key) + + def presigned_put_url(self, object_key: str, expires: int = 3600) -> str: + """预签名 PUT URL:拿到 URL 者可直传 Minio(数据不经网关)。""" + return self._client.presigned_put_object( + self.bucket, object_key, expires=timedelta(seconds=expires) + ) + + def presigned_get_url(self, object_key: str, expires: int = 3600) -> str: + """预签名 GET URL:拿到 URL 者可直接下载(数据不经网关)。""" + return self._client.presigned_get_object( + self.bucket, object_key, expires=timedelta(seconds=expires) + ) + + def close(self) -> None: + """兼容清理钩子(minio SDK 无显式关闭;urllib3 连接池随解释器回收)。""" + + # ------------------------------------------------------------------ + # async 便捷方法(to_thread 包装同步 SDK,RPC handler 直接 await) + # ------------------------------------------------------------------ + async def aput_file( + self, object_key: str, local_path: str | Path, content_type: str | None = None + ) -> str: + return await asyncio.to_thread(self.put_file, object_key, local_path, content_type) + + async def aput_bytes( + self, object_key: str, data: bytes, content_type: str | None = None + ) -> str: + return await asyncio.to_thread(self.put_bytes, object_key, data, content_type) + + async def aget_file(self, object_key: str, local_path: str | Path) -> Path: + return await asyncio.to_thread(self.get_file, object_key, local_path) + + async def aget_bytes(self, object_key: str) -> bytes: + return await asyncio.to_thread(self.get_bytes, object_key) + + async def adelete(self, object_key: str) -> None: + await asyncio.to_thread(self.delete, object_key) + + async def apresigned_put_url(self, object_key: str, expires: int = 3600) -> str: + return await asyncio.to_thread(self.presigned_put_url, object_key, expires) + + async def apresigned_get_url(self, object_key: str, expires: int = 3600) -> str: + return await asyncio.to_thread(self.presigned_get_url, object_key, expires) + + +__all__ = ["StorageClient"]