test: verify realtime snapshots are fresh

2026-07-02 08:43:27 +08:00
parent c211447747
commit 8def635bf5
4 changed files with 81 additions and 7 deletions
--- 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 快照最近刷新。
--- 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：

--- 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,
--- 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")