diff --git a/docs/operations/current-ecs-deployment.md b/docs/operations/current-ecs-deployment.md index 8e8e63cd..4dcfeac0 100644 --- a/docs/operations/current-ecs-deployment.md +++ b/docs/operations/current-ecs-deployment.md @@ -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 查询及结构化 `parsed_json`，三类协议的位置和里程点查询及 `frame_id` 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`，三类协议的位置和里程点查询及 `frame_id` RAW 回链，GB32960/JT808 的 `daily_mileage_km`、`daily_total_mileage_km` 统计公式，以及三类协议的 Redis realtime snapshot、online、protocol 查询。实时 snapshot/protocol 查询会校验 `updated_at_ms` 新鲜度；默认检查今天东八区数据时，还会要求三类 RAW 最新样本不超过 15 分钟，避免旧数据误判为接收正常；任一检查不达标时退出码为非 0。 ## 部署命令 @@ -220,4 +220,4 @@ journalctl -u lingniu-go-realtime-api --since '5 minutes ago' --no-pager - GB32960、JT808、YUTONG_MQTT 的最新 RAW 样本均在 15 分钟内。 - GB32960、JT808、YUTONG_MQTT 的 `vehicle_locations`、`vehicle_mileage_points` 查询均可命中生产数据，且样本包含 `frame_id` 回链 RAW。 - `vehicle_daily_metric` 可查到 `daily_mileage_km` 和 `daily_total_mileage_km`，且公式满足 `daily_mileage_km = latest_total_mileage_km - first_total_mileage_km`、`daily_total_mileage_km = latest_total_mileage_km`。 -- Redis realtime 可查到 GB32960、JT808、YUTONG_MQTT 的在线状态和协议快照。 +- Redis realtime 可查到 GB32960、JT808、YUTONG_MQTT 的在线状态和协议快照，且 snapshot/protocol 快照最近刷新。 diff --git a/docs/operations/go-vehicle-gateway-verification.md b/docs/operations/go-vehicle-gateway-verification.md index d661f3bf..0646e840 100644 --- a/docs/operations/go-vehicle-gateway-verification.md +++ b/docs/operations/go-vehicle-gateway-verification.md @@ -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、RAW `parsed_json`、历史核心表 `frame_id` 回链、每日指标公式、TDengine/MySQL/Redis 数据链路 +- `go_native_prod_smoke.py`：HTTP API、RAW `parsed_json`、历史核心表 `frame_id` 回链、每日指标公式、Redis 实时新鲜度、TDengine/MySQL/Redis 数据链路 如需单独检查 systemd： diff --git a/tools/go_native_prod_smoke.py b/tools/go_native_prod_smoke.py index cc737170..b6675cfa 100755 --- a/tools/go_native_prod_smoke.py +++ b/tools/go_native_prod_smoke.py @@ -15,6 +15,7 @@ from typing import Any SHANGHAI = dt.timezone(dt.timedelta(hours=8)) DEFAULT_BASE_URL = "http://115.29.187.205:20210" +DEFAULT_REALTIME_MAX_AGE_MINUTES = 15.0 @dataclass(frozen=True) @@ -192,9 +193,28 @@ def metric_formula_message(items: list[Any]) -> str: return f"metric_formula mismatch value={metric_value:.3f} expected={expected:.3f}" -def check_realtime(name: str, payload: dict[str, Any]) -> Check: +def check_realtime( + name: str, + payload: dict[str, Any], + max_age_minutes: float | None = None, + now: dt.datetime | None = None, +) -> Check: if payload.get("online") is False: return Check(name, "fail", 0, 1, "online=false") + updated_message = "" + if max_age_minutes is not None: + updated_age = realtime_updated_age_minutes(payload, now) + if updated_age is None: + return Check(name, "fail", 0, 1, "missing updated_at_ms") + updated_message = f"updated_age_minutes={updated_age:.1f}" + if updated_age > max_age_minutes: + return Check( + name, + "fail", + 0, + 1, + updated_message + f"; expected <= {max_age_minutes:.1f}", + ) identifier = payload.get("vin") or payload.get("vehicle_key") or "" protocols = payload.get("protocols") or [] protocol = payload.get("protocol") or "" @@ -214,9 +234,23 @@ def check_realtime(name: str, payload: dict[str, Any]) -> Check: ensure_ascii=False, sort_keys=True, ) + if updated_message: + message = updated_message + "; " + message return Check(name, "pass", 1, 1, message) +def realtime_updated_age_minutes(payload: dict[str, Any], now: dt.datetime | None = None) -> float | None: + try: + updated_ms = int(payload.get("updated_at_ms")) + except (TypeError, ValueError): + return None + current = now or dt.datetime.now(tz=dt.timezone.utc) + if current.tzinfo is None: + current = current.replace(tzinfo=dt.timezone.utc) + updated = dt.datetime.fromtimestamp(updated_ms / 1000, tz=dt.timezone.utc) + return max(0.0, (current.astimezone(dt.timezone.utc) - updated).total_seconds() / 60) + + def sample_summary(items: list[Any]) -> str: if not items: return "{}" @@ -372,9 +406,23 @@ def build_realtime_specs(payloads: dict[str, dict[str, Any]]) -> list[CheckSpec] encoded = urllib.parse.quote(identifier, safe="") base_path = "/api/realtime/vehicles/" + encoded specs.extend([ - CheckSpec(prefix + ".realtime", base_path, {}, 1, "realtime"), + CheckSpec( + prefix + ".realtime", + base_path, + {}, + 1, + "realtime", + max_age_minutes=DEFAULT_REALTIME_MAX_AGE_MINUTES, + ), CheckSpec(prefix + ".realtime_online", base_path + "/online", {}, 1, "realtime"), - CheckSpec(prefix + ".realtime_protocol", base_path + "/protocols/" + protocol, {}, 1, "realtime"), + CheckSpec( + prefix + ".realtime_protocol", + base_path + "/protocols/" + protocol, + {}, + 1, + "realtime", + max_age_minutes=DEFAULT_REALTIME_MAX_AGE_MINUTES, + ), ]) return specs @@ -385,7 +433,7 @@ def run_checks(base_url: str, specs: list[CheckSpec], timeout: float) -> list[Ch try: payload = query_json(base_url, spec.path, spec.params, timeout) if spec.kind == "realtime": - checks.append(check_realtime(spec.name, payload)) + checks.append(check_realtime(spec.name, payload, spec.max_age_minutes)) else: checks.append(check_total( spec.name, diff --git a/tools/test_go_native_prod_smoke.py b/tools/test_go_native_prod_smoke.py index 36a566ca..ee65fdd9 100644 --- a/tools/test_go_native_prod_smoke.py +++ b/tools/test_go_native_prod_smoke.py @@ -140,6 +140,32 @@ class GoNativeProdSmokeTest(unittest.TestCase): self.assertEqual(check.status, "fail") self.assertIn("online=false", check.message) + def test_realtime_check_fails_when_snapshot_is_stale(self): + now = dt.datetime(2026, 7, 2, 0, 30, 0, tzinfo=dt.timezone.utc) + + check = smoke.check_realtime( + "jt808.realtime", + {"vehicle_key": "JT808:013307811170", "updated_at_ms": 1782950400000}, + max_age_minutes=15, + now=now, + ) + + self.assertEqual(check.status, "fail") + self.assertIn("updated_age_minutes=30.0", check.message) + + def test_realtime_check_passes_when_snapshot_is_fresh(self): + now = dt.datetime(2026, 7, 2, 0, 30, 0, tzinfo=dt.timezone.utc) + + check = smoke.check_realtime( + "jt808.realtime", + {"vehicle_key": "JT808:013307811170", "updated_at_ms": 1782951600000}, + max_age_minutes=15, + now=now, + ) + + self.assertEqual(check.status, "pass") + self.assertIn("updated_age_minutes=10.0", check.message) + def test_parse_tdengine_utc_timestamp_as_aware_utc(self): parsed = smoke.parse_tdengine_utc_timestamp("2026-07-01 17:36:02")