shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: verify raw parsed json in production smoke

Browse Source
This commit is contained in:
lingniu
2026-07-02 02:00:50 +08:00
parent 601150b364
commit 27f4ad3468
4 changed files with 82 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
docs/operations/current-ecs-deployment.md
View File

@@ -169,7 +169,7 @@ python3 tools/go_native_prod_smoke.py --date 2026-07-02 --timeout 8
`go_systemd_prod_smoke.py` 通过 SSH 检查四个 Go systemd 服务是否 `active` 且 `enabled`，并确认 `808`、`32960` 由 `gateway` 监听、`20210` 由 `realtime-api` 监听，同时要求 `/opt/lingniu-go-native/spool/gateway` 没有待回放文件，避免旧 Java/Docker 进程占用生产端口或 Kafka spool 静默堆积。运行环境需要已配置 SSH 免密或已建立可用的 SSH 认证方式。
`go_native_prod_smoke.py` 通过 `20210` HTTP API 验证 GB32960、JT808、宇通 MQTT 的 RAW 查询，三类协议的位置和里程点查询，GB32960/JT808 的 `daily_mileage_km`、`daily_total_mileage_km` 统计，以及三类协议的 Redis realtime snapshot、online、protocol 查询。默认检查今天东八区数据时，还会要求三类 RAW 最新样本不超过 15 分钟，避免旧数据误判为接收正常；任一检查不达标时退出码为非 0。
`go_native_prod_smoke.py` 通过 `20210` HTTP API 验证 GB32960、JT808、宇通 MQTT 的 RAW 查询及结构化 `parsed_json`，三类协议的位置和里程点查询，GB32960/JT808 的 `daily_mileage_km`、`daily_total_mileage_km` 统计，以及三类协议的 Redis realtime snapshot、online、protocol 查询。默认检查今天东八区数据时，还会要求三类 RAW 最新样本不超过 15 分钟，避免旧数据误判为接收正常；任一检查不达标时退出码为非 0。
## 部署命令
@@ -216,7 +216,7 @@ journalctl -u lingniu-go-realtime-api --since '5 minutes ago' --no-pager
- `gateway` 正在监听 `32960` 和 `808`。
- `realtime-api` 正在监听 `20210`。
- gateway Kafka spool 当前无待回放文件。
- GB32960、JT808、YUTONG_MQTT 的东八区 RAW 查询均可命中生产数据。
- GB32960、JT808、YUTONG_MQTT 的东八区 RAW 查询均可命中生产数据，且最新 RAW 样本包含结构化 `parsed_json`。
- GB32960、JT808、YUTONG_MQTT 的最新 RAW 样本均在 15 分钟内。
- GB32960、JT808、YUTONG_MQTT 的 `vehicle_locations`、`vehicle_mileage_points` 查询均可命中生产数据。
- `vehicle_daily_metric` 可查到 `daily_mileage_km` 和 `daily_total_mileage_km`。

2
docs/operations/go-vehicle-gateway-verification.md
View File

@@ -28,7 +28,7 @@ python3 tools/go_prod_acceptance.py --date 2026-07-02
- `go_systemd_prod_smoke.py`：应用 ECS systemd 服务 active/enabled、端口归属和 gateway spool
- `go_kafka_prod_smoke.py`：Kafka topic 和消费组 lag
- `go_native_prod_smoke.py`：HTTP API、TDengine/MySQL/Redis 数据链路
- `go_native_prod_smoke.py`：HTTP API、RAW `parsed_json`、TDengine/MySQL/Redis 数据链路
如需单独检查 systemd：

48
tools/go_native_prod_smoke.py
View File

@@ -25,6 +25,7 @@ class CheckSpec:
minimum: int
kind: str = "total"
max_age_minutes: float | None = None
require_parsed_json: bool = False
@dataclass(frozen=True)
@@ -66,12 +67,14 @@ def check_total(
minimum: int,
max_age_minutes: float | None = None,
now: dt.datetime | None = None,
require_parsed_json: bool = False,
) -> Check:
count = int(payload.get("total") or 0)
items = payload.get("items") or []
sample = sample_summary(items)
age_message = latest_age_message(items, now)
if count >= minimum:
parsed_message = ""
if max_age_minutes is not None:
latest_age = latest_age_minutes(items, now)
if latest_age is None:
@@ -85,7 +88,14 @@ def check_total(
f"count={count}; latest_age_minutes={latest_age:.1f}; "
f"expected <= {max_age_minutes:.1f}; sample={sample}",
)
return Check(name, "pass", count, minimum, f"count={count}; {age_message}; sample={sample}")
if require_parsed_json:
parsed_message = parsed_json_message(items)
if not parsed_message.startswith("parsed_json=ok"):
return Check(name, "fail", count, minimum, f"count={count}; {parsed_message}; sample={sample}")
details = [f"count={count}", age_message]
if parsed_message:
details.append(parsed_message)
return Check(name, "pass", count, minimum, "; ".join(details) + f"; sample={sample}")
return Check(name, "fail", count, minimum, f"count={count}; expected >= {minimum}; {age_message}; sample={sample}")
@@ -116,6 +126,23 @@ def latest_age_message(items: list[Any], now: dt.datetime | None = None) -> str:
return f"latest_age_minutes={latest_age:.1f}"
def parsed_json_message(items: list[Any]) -> str:
if not items or not isinstance(items[0], dict):
return "missing parsed_json"
raw = items[0].get("parsed_json")
if isinstance(raw, dict) and raw:
return "parsed_json=ok"
if not isinstance(raw, str) or not raw.strip() or raw.strip().lower() == "null":
return "missing parsed_json"
try:
parsed = json.loads(raw)
except json.JSONDecodeError:
return "invalid parsed_json"
if isinstance(parsed, dict) and parsed:
return "parsed_json=ok"
return "invalid parsed_json"
def check_realtime(name: str, payload: dict[str, Any]) -> Check:
if payload.get("online") is False:
return Check(name, "fail", 0, 1, "online=false")
@@ -192,6 +219,7 @@ def build_check_specs(
{**raw_params, "protocol": "GB32960"},
min_raw,
max_age_minutes=max_raw_age_minutes,
require_parsed_json=True,
),
CheckSpec(
"jt808.raw",
@@ -199,6 +227,7 @@ def build_check_specs(
{**raw_params, "protocol": "JT808"},
min_raw,
max_age_minutes=max_raw_age_minutes,
require_parsed_json=True,
),
CheckSpec(
"yutong_mqtt.raw",
@@ -206,6 +235,7 @@ def build_check_specs(
{**raw_params, "protocol": "YUTONG_MQTT"},
min_raw,
max_age_minutes=max_raw_age_minutes,
require_parsed_json=True,
),
CheckSpec("gb32960.locations", "/api/history/locations", gb32960_history_params, min_history),
CheckSpec("gb32960.mileage_points", "/api/history/mileage-points", gb32960_history_params, min_history),
@@ -304,7 +334,13 @@ def run_checks(base_url: str, specs: list[CheckSpec], timeout: float) -> list[Ch
if spec.kind == "realtime":
checks.append(check_realtime(spec.name, payload))
else:
checks.append(check_total(spec.name, payload, spec.minimum, spec.max_age_minutes))
checks.append(check_total(
spec.name,
payload,
spec.minimum,
spec.max_age_minutes,
require_parsed_json=spec.require_parsed_json,
))
except Exception as exc: # pragma: no cover - exercised by live failures.
checks.append(Check(spec.name, "fail", 0, spec.minimum, f"request failed: {exc}"))
return checks
@@ -317,7 +353,13 @@ def query_payloads(base_url: str, specs: list[CheckSpec], timeout: float) -> tup
try:
payload = query_json(base_url, spec.path, spec.params, timeout)
payloads[spec.name] = payload
checks.append(check_total(spec.name, payload, spec.minimum, spec.max_age_minutes))
checks.append(check_total(
spec.name,
payload,
spec.minimum,
spec.max_age_minutes,
require_parsed_json=spec.require_parsed_json,
))
except Exception as exc:
checks.append(Check(spec.name, "fail", 0, spec.minimum, f"request failed: {exc}"))
return checks, payloads

34
tools/test_go_native_prod_smoke.py
View File

@@ -173,6 +173,40 @@ class GoNativeProdSmokeTest(unittest.TestCase):
self.assertEqual(check.status, "pass")
self.assertIn("latest_age_minutes=15.0", check.message)
def test_raw_check_requires_structured_parsed_json(self):
check = smoke.check_total(
"jt808.raw",
{
"total": 1,
"items": [{
"ts": "2026-07-01 17:45:00",
"parsed_json": "",
}],
},
minimum=1,
require_parsed_json=True,
)
self.assertEqual(check.status, "fail")
self.assertIn("missing parsed_json", check.message)
def test_raw_check_accepts_structured_parsed_json(self):
check = smoke.check_total(
"gb32960.raw",
{
"total": 1,
"items": [{
"ts": "2026-07-01 17:45:00",
"parsed_json": "{\"header\":{\"vin\":\"LTEST\"}}",
}],
},
minimum=1,
require_parsed_json=True,
)
self.assertEqual(check.status, "pass")
self.assertIn("parsed_json=ok", check.message)
if __name__ == "__main__":
unittest.main()