Compare commits

2 Commits
3 changed files with 103 additions and 18 deletions
+65
View File
@@ -0,0 +1,65 @@
# chongming.service — systemd 管理打包后的 chongming 应用(Nuitka 单二进制 + 模块库)
#
# 安装步骤(以 /opt/chongming 为例,目录名可改,需同步修改下方 WorkingDirectory/ExecStart):
# sudo mkdir -p /opt/chongming
# sudo cp -r build/. /opt/chongming/ # 打包产物:chongming 二进制 + config.toml + config/ + resources/ + static.svfs
# sudo useradd --system --home /opt/chongming --shell /usr/sbin/nologin chongming
# sudo chown -R chongming:chongming /opt/chongming
# sudo cp deploy/chongming.service /etc/systemd/system/
# sudo systemctl daemon-reload
# sudo systemctl enable --now chongming
#
# 常用操作:
# systemctl status chongming # 状态
# journalctl -u chongming -f # 实时日志(应用自身日志另见 /opt/chongming/logs/)
# systemctl restart chongming # 重启(SIGTERM 优雅排空,最长等待 30s)
#
# 可选:/etc/chongming.env 中写 APP_WORKERS=<N> 覆盖 worker 数(默认取
# config.toml [production.server].workers,打包版通常为 4)。
[Unit]
Description=Chongming Zenoh Gateway (FastAPI + Zenoh + MQTT)
Wants=network-online.target
After=network-online.target
# 60s 内失败 5 次则放弃(避免配置错误导致崩溃循环)
StartLimitIntervalSec=60
StartLimitBurst=5
[Service]
Type=simple
# 打包应用入口:Nuitka 编译产物,前台运行 gunicorn master(Linux 分支)
ExecStart=/opt/chongming/chongming
# 所有相对路径(config.toml / config/ / resources/*.mbank / static.svfs /
# data/ / logs/ / uploads/)均相对工作目录解析,必须指向安装目录
WorkingDirectory=/opt/chongming
# 非特权用户运行(HTTP 8000 / MQTT 1883 / Zenoh 7447 等端口均无需 root)
User=chongming
Group=chongming
UMask=0027
# 每机可选覆盖(如 APP_WORKERS),文件不存在时忽略(- 前缀)
EnvironmentFile=-/etc/chongming.env
Environment=PYTHONUNBUFFERED=1
# 优雅停机:SIGTERM 由 gunicorn 排空 worker,等在途请求/告警上传完成
KillSignal=SIGTERM
KillMode=control-group
TimeoutStopSec=30
# 异常退出自动拉起
Restart=on-failure
RestartSec=3
# Zenoh / MQTT / 并发连接放开文件描述符上限
LimitNOFILE=65535
LimitNPROC=65535
# 轻量加固(应用无 setuid/特权端口需求,不影响运行)
PrivateTmp=true
NoNewPrivileges=true
[Install]
WantedBy=multi-user.target
+22 -11
View File
@@ -12,6 +12,9 @@ logger = get_logger("mqtt")
MQTT_TOPIC_PREFIX = "chongming"
MQTT_SUBSCRIBE_TOPIC = f"{MQTT_TOPIC_PREFIX}/+/+"
# 已按"非信封载荷"路由过告警的主题(每主题只告警一次,避免设备 5s 心跳刷屏)
_warned_legacy_topics: set = set()
def _compile_topic(topic: str) -> re.Pattern:
"""
@@ -154,18 +157,26 @@ class MQTTBridge:
except json.JSONDecodeError:
return
# 信封协议(严格): {"meta": {...}, "data": {...}}
# 非信封格式载荷直接丢弃并告警,请设备侧按新协议发布
# 信封协议: {"meta": {...}, "data": {...}}
# 兼容非信封载荷(如第三方设备原始报文): 整包视为业务数据、meta 置空,
# 与 worker 运行时 runtime.py 的容错语义一致(method 默认 GET)——
# 若在此丢弃,盒子心跳等设备报文将永远到不了 Worker。
if not isinstance(pl, dict) or ("meta" not in pl and "data" not in pl):
logger.warning(
"丢弃非信封格式载荷 [%s]: 请使用 {'meta': {...}, 'data': {...}} 格式", t
)
return
meta = pl.get("meta") or {}
data = pl.get("data") or {}
if not isinstance(data, dict):
data = {}
pl = {"meta": meta, "data": data}
data = pl if isinstance(pl, dict) else {}
pl = {"meta": {}, "data": data}
if t not in _warned_legacy_topics:
_warned_legacy_topics.add(t)
logger.warning(
"非信封格式载荷(缺少 meta/data)按原始报文路由 [%s]: "
"整包视为 data,建议设备侧改用 {'meta': {...}, 'data': {...}} 格式",
t,
)
else:
meta = pl.get("meta") or {}
data = pl.get("data") or {}
if not isinstance(data, dict):
data = {}
pl = {"meta": meta, "data": data}
# 1. 优先匹配显式配置的自定义订阅主题(顺序即优先级)
for ct, compiled in self._compiled:
+16 -7
View File
@@ -139,16 +139,18 @@ async def test_envelope_meta_fields():
print("✅ 信封 meta 字段 (method/timeout/cacheable) 透传通过")
async def test_flat_payload_dropped():
"""严格信封协议:扁平载荷(无 meta/data)应被丢弃且不调用 worker"""
async def test_flat_payload_routed():
"""非信封(扁平)载荷兼容:整包视为 data、meta 置空照常路由(与 worker 运行时容错一致)"""
wm = FakeWM()
bridge, published = make_bridge(
wm, [{"topic": "devices/+/telemetry", "worker": "device"}]
)
await bridge._handle(make_msg("devices/d1/telemetry", {"_method": "POST", "v": 1}))
assert wm.calls == []
assert published == []
print("✅ 非信封(扁平)载荷被丢弃通过")
assert wm.calls, "扁平载荷不应被丢弃,应照常路由到 worker"
assert wm.calls[-1]["worker_type"] == "device"
assert wm.calls[-1]["payload"] == {"meta": {}, "data": {"_method": "POST", "v": 1}}
assert published[-1][0] == "devices/d1/telemetry/response"
print("✅ 非信封(扁平)载荷按原始报文路由通过")
async def test_non_json_dropped():
@@ -182,7 +184,14 @@ async def test_board_ping_topic():
assert wm.calls[-1]["worker_type"] == "aibox_sophon_adapter"
assert wm.calls[-1]["subject"] == "aibox_sophon_adapter.board_ping"
assert published[-1][0] == "/board_ping/response"
print("✅ /board_ping 自定义主题路由通过")
# 回归:盒子心跳为原始报文(无信封),不得被丢弃,应整包作为 data 路由
wm.calls.clear(); published.clear()
await bridge._handle(make_msg("/board_ping", {"BoardId": "B001", "Status": 1}))
assert wm.calls and wm.calls[-1]["payload"] == {
"meta": {}, "data": {"BoardId": "B001", "Status": 1}
}, wm.calls
assert published[-1][0] == "/board_ping/response"
print("✅ /board_ping 自定义主题路由通过(含原始报文兼容)")
def test_plugin_config_driven_collection():
@@ -209,7 +218,7 @@ async def main():
test_compile_topic()
await test_custom_routing()
await test_envelope_meta_fields()
await test_flat_payload_dropped()
await test_flat_payload_routed()
await test_non_json_dropped()
await test_board_ping_topic()
test_plugin_config_driven_collection()