diff --git a/confs/README.md b/confs/README.md index 1a393d0..7a6a4be 100644 --- a/confs/README.md +++ b/confs/README.md @@ -227,7 +227,7 @@ except RpcError as e: ## 集成关系 -- 由 `core`（`PMC`）通过 `init.txt` 中的 `bash /opt/micro_servers/confs/_run.sh` 拉起并守护（confs 最先启动，所有依赖 confs 的服务必须在 confs 就绪后启动；confs 日志推送 db 为异步 fail-safe，db 不可用时静默丢弃，不阻塞 confs 启动） +- 由 `core`（`PMC`）通过 `init.txt` 中的 `bash /opt/micro_servers/confs/_run.sh` 拉起并守护（confs 最先启动即可；各业务服务通过 pymsg `retry_on_refused=True` 在拉取配置时自动阻塞等待 confs 就绪，不严格要求启动顺序；confs 日志推送 db 为异步 fail-safe，db 不可用时静默丢弃，不阻塞 confs 启动） - 各内部服务通过 pymsg `RpcClient` 调 `get` 路由按需获取配置，替代本地散落的 ini/json/yaml - 自身日志通过 `log_client.py` 异步推送至 `db:8005` 的 Logs 表（与其他服务一致） @@ -235,33 +235,32 @@ except RpcError as e: | 服务 | 配置文件 | 接入方式 | |------|---------|---------| -| [static](../static/) | [static.ini](configs/static.ini) | confs 强依赖（confs 不可用时 static 直接退出）→ 环境变量覆盖（最高优先级，Docker 兼容） | -| [dnfs](../dnfs/) | [dnfs.ini](configs/dnfs.ini) | confs 强依赖（confs 不可用时 dnfs 直接退出）→ 环境变量覆盖（最高优先级，Docker 兼容） | -| [db](../db/) | [db.ini](configs/db.ini) | confs 强依赖（confs 不可用时 db 直接退出）→ 环境变量覆盖（最高优先级，Docker 兼容） | +| [static](../static/) | [static.ini](configs/static.ini) | confs 强依赖（pymsg 自动阻塞等待 confs 就绪）→ 环境变量覆盖（最高优先级，Docker 兼容） | +| [dnfs](../dnfs/) | [dnfs.ini](configs/dnfs.ini) | confs 强依赖（pymsg 自动阻塞等待 confs 就绪）→ 环境变量覆盖（最高优先级，Docker 兼容） | +| [db](../db/) | [db.ini](configs/db.ini) | confs 强依赖（pymsg 自动阻塞等待 confs 就绪）→ 环境变量覆盖（最高优先级，Docker 兼容） | static 的接入实现见 [static/fsys.py](../static/fsys.py)： - 两层加载顺序：confs（static.ini，强依赖）→ 环境变量覆盖（优先级最高） -- confs 连接失败或 static.ini 缺失必需配置项时，static 直接 `sys.exit(1)` 终止，不使用默认值或回退本地文件 +- pymsg `retry_on_refused=True` 自动阻塞等待 confs 就绪（每秒重试，每 5 秒打印等待提示），不会因启动顺序导致进程退出 +- static.ini 缺失必需配置项时仍然报错退出（配置错误，非启动时序问题） - 配置项涵盖：`[server]` host/port、`[directory]` root、`[ssl]` enabled/cert_file/key_file（ssl 有默认值）、`[cors]` allowed_origins - confs 连接地址为引导配置：环境变量 `CONFS_HOST`/`CONFS_PORT` 优先，默认 `127.0.0.1:8010` - 支持的环境变量：`LISTEN_ADDR`/`LISTEN_PORT`/`STATIC_ROOT`/`CONFS_HOST`/`CONFS_PORT` -- 启动顺序：confs 必须先于 static 启动；已删除 `--config` 命令行参数和本地 `config.ini` 回退逻辑 +- 已删除 `--config` 命令行参数和本地 `config.ini` 回退逻辑 dnfs 的接入实现见 [dnfs/dnfs.py](../dnfs/dnfs.py)： - 两层加载顺序：confs（dnfs.ini，强依赖）→ 环境变量覆盖（优先级最高） -- confs 连接失败或 dnfs.ini 缺失必需配置项时，dnfs 直接 `sys.exit(1)` 终止，不使用默认值 +- pymsg `retry_on_refused=True` 自动阻塞等待 confs 就绪，不严格要求启动顺序 - 配置项涵盖：`[server]` host/port、`[database]` host/port/timeout、`[ai_review]` url/timeout - confs 连接地址为引导配置：环境变量 `CONFS_HOST`/`CONFS_PORT` 优先，默认 `127.0.0.1:8010` - 支持的环境变量：`LISTEN_ADDR`/`LISTEN_PORT`/`DB_HOST`/`DB_PORT`/`REVIEW_URL`/`CONFS_HOST`/`CONFS_PORT` -- 启动顺序：confs → db → dnfs → ai_review/...（dnfs 同时依赖 confs 和 db） db 的接入实现见 [db/dbv2.py](../db/dbv2.py)： - 两层加载顺序：confs（db.ini，强依赖）→ 环境变量覆盖（优先级最高） -- confs 连接失败或 db.ini 缺失必需配置项时，db 直接 `sys.exit(1)` 终止，不使用默认值 +- pymsg `retry_on_refused=True` 自动阻塞等待 confs 就绪，不严格要求启动顺序 - 配置项涵盖：`[server]` host/port/workers、`[database]` url、`[logs]` retention_days/max_records/cleanup_interval - confs 连接地址为引导配置：环境变量 `CONFS_HOST`/`CONFS_PORT` 优先，默认 `127.0.0.1:8010` - 支持的环境变量：`LISTEN_ADDR`/`LISTEN_PORT`/`WORKERS`/`DATABASE_URL`/`CONFS_HOST`/`CONFS_PORT` -- 启动顺序：confs 必须先于 db 启动（confs → db → ai_review/...） - 独立运维工具 `db_tools.py`（backup/export/import 等）仍直接读 `DATABASE_URL` 环境变量，不依赖 confs，避免运维脚本依赖服务就绪 其他服务（gateway / waf / ai_review / smtp）尚未接入，仍读各自本地配置。 diff --git a/db/README.md b/db/README.md index cc3da2d..4cd8898 100644 --- a/db/README.md +++ b/db/README.md @@ -72,13 +72,13 @@ confs 自身的连接地址不通过 confs 获取（避免循环依赖），通 | `CONFS_HOST` | confs 服务地址（引导配置） | — | | `CONFS_PORT` | confs 服务端口（引导配置） | — | -### 启动顺序要求 +### 启动顺序 -**confs 必须先于 db 启动**（启动顺序调整为：confs → db → ai_review/...）。若 confs 不可用： -- db 打印致命错误到 stderr（包含 confs 地址和具体错误原因），以 exit code 1 终止 -- 不会静默使用默认值启动 -- 配置完整性校验：若 db.ini 中缺少必需的 section/key 且无对应环境变量覆盖，同样报错退出 -- 启动成功时控制台打印：`配置来源: confs 统一配置服务 (db.ini @ host:port)` +db 通过 pymsg `retry_on_refused=True` 自动阻塞等待 confs 就绪，**不严格要求启动顺序**： +- confs 先启动 → db 立即拉取配置，正常启动 +- db 先启动 → pymsg 自动重试（每秒一次，每 5 秒打印等待提示），confs 启动后自动连接成功 +- 配置完整性校验：若 db.ini 中缺少必需的 section/key 且无对应环境变量覆盖，仍然报错退出 +- 启动成功时控制台打印：`配置来源: confs (db.ini @ host:port)` > 独立运维工具 `db_tools.py`（backup/export/import/vacuum 等）仍直接读 `DATABASE_URL` 环境变量，不依赖 confs 服务就绪。 > Docker 环境通过 `docker-entrypoint.sh` 传入环境变量（见 [Dockerfile](Dockerfile)），需确保 `CONFS_HOST`/`CONFS_PORT` 指向 confs 服务地址。 diff --git a/db/dbv2.py b/db/dbv2.py index b0123af..efd124f 100644 --- a/db/dbv2.py +++ b/db/dbv2.py @@ -82,26 +82,22 @@ def _merge_ini(text): def _fetch_from_confs(): - """从 confs 统一配置服务拉取 db.ini 全文，失败直接终止进程。""" + """从 confs 统一配置服务拉取 db.ini 全文。 + + 使用 retry_on_refused=True：confs 未就绪时 pymsg 自动阻塞重试（每秒一次， + 每 5 秒打印一次等待提示），直到 confs 启动后自动连接成功。 + RpcError（如配置文件不存在）仍直接报错退出——这是配置错误而非启动时序问题。 + """ from pymsg import RpcClient, RpcError # noqa: E402 + client = RpcClient(_confs_host, _confs_port, 'db', timeout=5, + retry_on_refused=True) try: - client = RpcClient(_confs_host, _confs_port, 'db', timeout=5) data = client.call('get', {'name': 'db.ini'}) - return data.get('content') - except (ConnectionRefusedError, ConnectionError, OSError) as e: - print(f'[dbv2] 致命错误: 无法连接 confs 服务 ({_confs_host}:{_confs_port}): {e}', - file=_sys.stderr) - print(f'[dbv2] 请确认 confs 服务已启动并监听 {_confs_host}:{_confs_port}', - file=_sys.stderr) - _sys.exit(1) except RpcError as e: print(f'[dbv2] 致命错误: 从 confs 获取 db.ini 失败: {e}', file=_sys.stderr) _sys.exit(1) - except Exception as e: - print(f'[dbv2] 致命错误: confs 配置拉取异常: {e}', - file=_sys.stderr) - _sys.exit(1) + return data.get('content') _confs_text = _fetch_from_confs() diff --git a/dnfs/README.md b/dnfs/README.md index 2201130..c7de774 100644 --- a/dnfs/README.md +++ b/dnfs/README.md @@ -254,14 +254,16 @@ confs 自身的连接地址不通过 confs 获取（避免循环依赖），通 | `CONFS_HOST` | `confs` | confs 服务主机名（引导配置） | | `CONFS_PORT` | `8010` | confs 服务端口（引导配置） | -### 启动顺序要求 +### 启动顺序 -**confs 必须先于 dnfs 启动**（启动顺序：confs → db → dnfs → ai_review/...）。若 confs 不可用： -- dnfs 打印致命错误到 stderr（包含 confs 地址和具体错误原因），以 exit code 1 终止 -- 不会静默使用默认值启动 -- 配置完整性校验：若 dnfs.ini 中缺少必需的 section/key 且无对应环境变量覆盖，同样报错退出 +dnfs 通过 pymsg `retry_on_refused=True` 自动阻塞等待 confs 就绪，**拉取配置阶段不严格要求启动顺序**： +- confs 先启动 → dnfs 立即拉取配置 +- dnfs 先启动 → pymsg 自动重试（每秒一次，每 5 秒打印等待提示），confs 启动后自动连接成功 +- 配置完整性校验：若 dnfs.ini 中缺少必需的 section/key 且无对应环境变量覆盖，仍然报错退出 - 启动成功时控制台打印：`配置来源: confs (dnfs.ini @ host:port)` +> 注意：dnfs 对 db/ai_review 的 RPC 调用是运行时按需连接（首次请求时才建立 TCP），db/ai_review 未就绪时相关 API 返回 502 但进程不退出；待依赖服务启动后自动恢复。 + ## 前端 三文件架构： diff --git a/dnfs/dnfs.py b/dnfs/dnfs.py index 177e271..9d19daa 100755 --- a/dnfs/dnfs.py +++ b/dnfs/dnfs.py @@ -63,25 +63,20 @@ def _merge_ini(text): def _fetch_from_confs(): - """从 confs 统一配置服务拉取 dnfs.ini 全文，失败直接终止进程。""" + """从 confs 统一配置服务拉取 dnfs.ini 全文。 + + 使用 retry_on_refused=True：confs 未就绪时 pymsg 自动阻塞重试（每秒一次， + 每 5 秒打印一次等待提示），直到 confs 启动后自动连接成功。 + RpcError（如配置文件不存在）仍直接报错退出——这是配置错误而非启动时序问题。 + """ try: - data = RpcClient(_confs_host, _confs_port, 'dnfs', timeout=5).call( - 'get', {'name': 'dnfs.ini'}) - return data.get('content') - except (ConnectionRefusedError, ConnectionError, OSError) as e: - print(f'[dnfs] 致命错误: 无法连接 confs 服务 ({_confs_host}:{_confs_port}): {e}', - file=sys.stderr) - print(f'[dnfs] 请确认 confs 服务已启动并监听 {_confs_host}:{_confs_port}', - file=sys.stderr) - sys.exit(1) + data = RpcClient(_confs_host, _confs_port, 'dnfs', timeout=5, + retry_on_refused=True).call('get', {'name': 'dnfs.ini'}) except RpcError as e: print(f'[dnfs] 致命错误: 从 confs 获取 dnfs.ini 失败: {e}', file=sys.stderr) sys.exit(1) - except Exception as e: - print(f'[dnfs] 致命错误: confs 配置拉取异常: {e}', - file=sys.stderr) - sys.exit(1) + return data.get('content') _confs_text = _fetch_from_confs() diff --git a/pymsg/README.md b/pymsg/README.md index ad76e58..3e84706 100644 --- a/pymsg/README.md +++ b/pymsg/README.md @@ -110,6 +110,20 @@ resp = client.call('{"action": "select", "table": "Users"}', timeout=10.0) 目标不可达时抛出 `ConnectionRefusedError`，超时无回复时抛出 `TimeoutError`。 +**启动等待模式（retry_on_refused）**： + +```python +# 服务启动时阻塞等待下游服务就绪，连接被拒绝时自动重试 +resp = client.call('{"_route": "get", "name": "db.ini"}', + timeout=5.0, + retry_on_refused=True, # 连接拒绝时阻塞重试 + retry_interval=1.0) # 重试间隔 1 秒（默认） +``` + +`retry_on_refused=True` 时，`ConnectionRefusedError` 不会抛出，而是按 `retry_interval`（默认 1.0 秒）间隔不断重试，直到连接成功。重试期间每 5 秒向 stderr 打印一次等待提示（如 `[pymsg] 等待服务 127.0.0.1:8010 就绪... (已重试 N 次)`），避免日志刷屏。仅对 `ConnectionRefusedError` 重试；`TimeoutError`、`ConnectionError`、业务错误等仍立即抛出。 + +适用场景：微服务启动顺序不确定时，服务进程可在不依赖外部进程管理器的情况下自动等待下游依赖就绪，无需手动 sys.exit + 外部重启。`log_client` 等 fail-safe 场景应保持默认 `retry_on_refused=False`，避免后台线程阻塞。 + **3. 长连接监听（回调模式）** ```python @@ -226,7 +240,23 @@ except RpcError as e: ... # 服务端业务错误（success=false） ``` -`call(route, payload=None, *, timeout=None)`：自动注入 `_route`、JSON 编解码、检查 `success`，返回 `data` 字段；业务错误抛 `RpcError`，网络故障仍抛 `ConnectionRefusedError` / `TimeoutError`（来自 `MsgClient`）。 +| 构造参数 | 含义 | +|----------|------| +| `addr` / `port` | 服务端地址与端口 | +| `name` | 客户端标识（透传 `MsgClient`） | +| `timeout` | 默认调用超时秒数（默认 `10.0`） | +| `retry_on_refused` | 连接被拒绝时是否阻塞重试（关键字参数，默认 `False`） | +| `retry_interval` | 重试间隔秒数（关键字参数，默认 `1.0`） | + +`call(route, payload=None, *, timeout=None, retry_on_refused=None, retry_interval=None)`：自动注入 `_route`、JSON 编解码、检查 `success`，返回 `data` 字段；业务错误抛 `RpcError`，网络故障仍抛 `ConnectionRefusedError` / `TimeoutError`（来自 `MsgClient`）。`retry_on_refused` 和 `retry_interval` 参数可在单次调用时覆盖构造时的设置。 + +**启动等待示例**（解决微服务启动顺序问题）： + +```python +# confs 客户端：启动时阻塞等待 confs 就绪，不依赖启动顺序 +rpc = RpcClient("127.0.0.1", 8010, "db", timeout=5, retry_on_refused=True) +data = rpc.call("get", {"name": "db.ini"}) # confs 未启动时自动阻塞等待 +``` ## 实际用例 diff --git a/pymsg/pymsg/__init__.py b/pymsg/pymsg/__init__.py index 48d4e7e..6a9830e 100644 --- a/pymsg/pymsg/__init__.py +++ b/pymsg/pymsg/__init__.py @@ -22,6 +22,7 @@ import select import socket as _socket import struct import threading +import time import traceback from concurrent.futures import ThreadPoolExecutor @@ -200,34 +201,56 @@ class MsgClient: ) # ---- 请求-应答（短连接）------------------------------------------------ - def call(self, msg: str, *, timeout: "float | None" = None) -> str: + def call(self, msg: str, *, timeout: "float | None" = None, + retry_on_refused: bool = False, + retry_interval: float = 1.0, + _log_every: float = 5.0) -> str: """发送一条消息并阻塞等待单条回复（短连接：connect → send → recv → close）。 适合请求-应答模式，如 RPC 调用。不会给消息添加 name 前缀。 Args: - msg: 要发送的原始字符串。 - timeout: 超时秒数，默认 10 秒。 + msg: 要发送的原始字符串。 + timeout: 超时秒数，默认 10 秒。 + retry_on_refused: 目标服务未启动（ConnectionRefusedError）时是否阻塞重试， + 直到连接成功。适用于启动顺序不确定的服务发现场景。 + 默认 False（连接拒绝立即抛异常）。 + retry_interval: 重试间隔秒数，默认 1.0。 + _log_every: 重试期间打印等待提示的间隔（秒），默认每 5 秒打印一次，避免刷屏。 Returns: str: 服务器返回的字符串。 Raises: - ConnectionRefusedError: 目标服务不可达。 + ConnectionRefusedError: 目标服务不可达且 ``retry_on_refused=False``。 TimeoutError: 等待响应超时。 ConnectionError: 连接在收全数据前断开。 """ _tout = timeout if timeout is not None else 10.0 - try: - with _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM) as s: - s.settimeout(_tout) - s.connect((self.addr, self.port)) - send(s, msg) - return recv(s, timeout=_tout) - except ConnectionRefusedError: - raise ConnectionRefusedError( - f"Could not connect to {self.addr}:{self.port}" - ) + _last_log = 0.0 + _attempt = 0 + while True: + try: + with _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM) as s: + s.settimeout(_tout) + s.connect((self.addr, self.port)) + send(s, msg) + return recv(s, timeout=_tout) + except ConnectionRefusedError: + if not retry_on_refused: + raise ConnectionRefusedError( + f"Could not connect to {self.addr}:{self.port}" + ) + _attempt += 1 + now = time.monotonic() + if now - _last_log >= _log_every or _attempt == 1: + print( + f"[pymsg] 等待服务 {self.addr}:{self.port} 就绪... " + f"(已重试 {_attempt} 次)", + file=__import__('sys').stderr, + ) + _last_log = now + time.sleep(retry_interval) # ---- 持续监听（阻塞当前线程，回调模式）------------------------------- def listen(self, on_message: "callable") -> None: diff --git a/pymsg/pymsg/rpc.py b/pymsg/pymsg/rpc.py index 74b2977..127e76a 100644 --- a/pymsg/pymsg/rpc.py +++ b/pymsg/pymsg/rpc.py @@ -36,21 +36,33 @@ class RpcClient: rpc = RpcClient("127.0.0.1", 8005, "dnfs") data = rpc.call("execute", {"action": "select", "table": "Users"}) + + # 启动时阻塞等待服务就绪（连接被拒绝时自动重试） + rpc = RpcClient("127.0.0.1", 8010, "static", retry_on_refused=True) """ def __init__(self, addr: str, port: int, name: str = "rpc", - timeout: float = 10.0) -> None: + timeout: float = 10.0, + *, retry_on_refused: bool = False, + retry_interval: float = 1.0) -> None: self._client = MsgClient(addr, port, name) self._timeout = timeout + self._retry_on_refused = retry_on_refused + self._retry_interval = retry_interval def call(self, route: str, payload: "dict | None" = None, *, - timeout: "float | None" = None): + timeout: "float | None" = None, + retry_on_refused: "bool | None" = None, + retry_interval: "float | None" = None): """调用指定路由，返回响应中的 ``data`` 字段。 Args: route: 路由名（注入为 ``_route`` 字段）。 payload: 业务字段 dict（``_route`` 键会被覆盖）。 timeout: 本次调用的超时秒数，缺省用构造时的值。 + retry_on_refused: 本次调用是否在连接被拒绝时阻塞重试， + None 表示使用构造时的值（默认 False）。 + retry_interval: 本次调用的重试间隔，None 表示使用构造时的值。 Returns: 服务端返回的 ``data``（可能为 None）。 @@ -58,13 +70,17 @@ class RpcClient: Raises: RpcError: 服务端返回 success=false，或响应非法。 pymsg.TimeoutError: 等待响应超时。 - ConnectionRefusedError: 目标服务不可达。 + ConnectionRefusedError: 目标服务不可达且未开启重试。 """ req = dict(payload or {}) req["_route"] = route + _retry = self._retry_on_refused if retry_on_refused is None else retry_on_refused + _interval = self._retry_interval if retry_interval is None else retry_interval resp = self._client.call( json.dumps(req, ensure_ascii=False), timeout=self._timeout if timeout is None else timeout, + retry_on_refused=_retry, + retry_interval=_interval, ) try: obj = json.loads(resp) diff --git a/static/README.md b/static/README.md index ec49623..162c6ad 100644 --- a/static/README.md +++ b/static/README.md @@ -125,12 +125,12 @@ confs 自身的连接地址不通过 confs 获取（避免循环依赖），通 | `CONFS_HOST` | confs 服务主机名（引导配置，Docker 环境使用服务名） | | `CONFS_PORT` | confs 服务端口（引导配置） | -### 启动顺序要求 +### 启动顺序 -**confs 必须先于 static 启动**。若 confs 不可用： -- static 打印致命错误到 stderr（包含 confs 地址和具体错误原因），以 exit code 1 终止 -- 不会静默使用默认值或回退本地配置文件启动 -- 配置完整性校验：若 static.ini 中缺少必需的 section/key（server.host/port、directory.root、cors.allowed_origins）且无对应环境变量覆盖，同样报错退出 +static 通过 pymsg `retry_on_refused=True` 自动阻塞等待 confs 就绪，**拉取配置阶段不严格要求启动顺序**： +- confs 先启动 → static 立即拉取配置 +- static 先启动 → pymsg 自动重试（每秒一次，每 5 秒打印等待提示），confs 启动后自动连接成功 +- 配置完整性校验：若 static.ini 中缺少必需的 section/key（server.host/port、directory.root、cors.allowed_origins）且无对应环境变量覆盖，仍然报错退出 - ssl 段为可选项，未配置时默认 `enabled=false`、cert/key 使用默认文件名 ### 历史遗留：本地 config.ini 已删除 diff --git a/static/fsys.py b/static/fsys.py index ae56e99..9a9b201 100644 --- a/static/fsys.py +++ b/static/fsys.py @@ -72,26 +72,22 @@ def _merge_ini(text): def _fetch_from_confs(): - """从 confs 统一配置服务拉取 static.ini 全文，失败直接终止进程。""" + """从 confs 统一配置服务拉取 static.ini 全文。 + + 使用 retry_on_refused=True：confs 未就绪时 pymsg 自动阻塞重试（每秒一次， + 每 5 秒打印一次等待提示），直到 confs 启动后自动连接成功。 + RpcError（如配置文件不存在）仍直接报错退出——这是配置错误而非启动时序问题。 + """ from pymsg import RpcClient, RpcError # noqa: E402 + client = RpcClient(_confs_host, _confs_port, 'static', timeout=5, + retry_on_refused=True) try: - client = RpcClient(_confs_host, _confs_port, 'static', timeout=5) data = client.call('get', {'name': 'static.ini'}) - return data.get('content') - except (ConnectionRefusedError, ConnectionError, OSError) as e: - print(f'[static] 致命错误: 无法连接 confs 服务 ({_confs_host}:{_confs_port}): {e}', - file=sys.stderr) - print(f'[static] 请确认 confs 服务已启动并监听 {_confs_host}:{_confs_port}', - file=sys.stderr) - sys.exit(1) except RpcError as e: print(f'[static] 致命错误: 从 confs 获取 static.ini 失败: {e}', file=sys.stderr) sys.exit(1) - except Exception as e: - print(f'[static] 致命错误: confs 配置拉取异常: {e}', - file=sys.stderr) - sys.exit(1) + return data.get('content') _confs_text = _fetch_from_confs()