From 0ef88e1a4cdd3a496a078737e6e26e463f0f56d0 Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 01:49:02 +0800 Subject: [PATCH] test: add kafka production smoke --- docs/operations/current-ecs-deployment.md | 8 + .../go-vehicle-gateway-verification.md | 10 ++ tools/go_kafka_prod_smoke.py | 162 ++++++++++++++++++ tools/test_go_kafka_prod_smoke.py | 62 +++++++ 4 files changed, 242 insertions(+) create mode 100755 tools/go_kafka_prod_smoke.py create mode 100644 tools/test_go_kafka_prod_smoke.py diff --git a/docs/operations/current-ecs-deployment.md b/docs/operations/current-ecs-deployment.md index 2a432b7c..3b834984 100644 --- a/docs/operations/current-ecs-deployment.md +++ b/docs/operations/current-ecs-deployment.md @@ -75,6 +75,14 @@ Kafka 消费组： | `go-stat-writer` | `vehicle.raw.go.gb32960.v1`、`vehicle.raw.go.jt808.v1` | 2026-07-02 验证 lag 为 `0` | | `go-realtime-api` | `vehicle.event.go.unified.v1` | 2026-07-02 验证 lag 为 `0` | +可重复 smoke： + +```bash +python3 tools/go_kafka_prod_smoke.py --host 114.55.58.251 --user root --max-lag 100 +``` + +该脚本通过 SSH 登录 Kafka ECS，检查 Go 生产 topic 是否存在，并校验 `go-history-writer`、`go-stat-writer`、`go-realtime-api` 的消费组 lag。运行环境需要已配置 SSH 免密或已建立可用的 SSH 认证方式。 + ## TDengine 数据层 默认数据库：`lingniu_vehicle_ts` diff --git a/docs/operations/go-vehicle-gateway-verification.md b/docs/operations/go-vehicle-gateway-verification.md index ef88e4ce..b2f1b842 100644 --- a/docs/operations/go-vehicle-gateway-verification.md +++ b/docs/operations/go-vehicle-gateway-verification.md @@ -76,6 +76,16 @@ ss -lntp | egrep ':(808|32960|20210)\b' ## Kafka 验证 +优先使用 smoke 脚本做可重复检查： + +```bash +python3 tools/go_kafka_prod_smoke.py --host 114.55.58.251 --user root --max-lag 100 +``` + +脚本会检查 `vehicle.raw.go.gb32960.v1`、`vehicle.raw.go.jt808.v1`、`vehicle.raw.go.yutong-mqtt.v1`、`vehicle.event.go.unified.v1` 是否存在，并汇总 `go-history-writer`、`go-stat-writer`、`go-realtime-api` 的消费 lag。运行环境需要 SSH 免密或已建立可用的 SSH 认证方式。 + +手工核对命令： + ```bash kafka-console-consumer --bootstrap-server 172.17.111.56:9092 \ --topic vehicle.raw.go.jt808.v1 --max-messages 1 --timeout-ms 10000 diff --git a/tools/go_kafka_prod_smoke.py b/tools/go_kafka_prod_smoke.py new file mode 100755 index 00000000..06e0c12a --- /dev/null +++ b/tools/go_kafka_prod_smoke.py @@ -0,0 +1,162 @@ +#!/usr/bin/env python3 +"""Kafka topic and consumer lag smoke checks for the Go native production stack.""" + +from __future__ import annotations + +import argparse +import json +import subprocess +import sys +from dataclasses import asdict, dataclass + + +DEFAULT_HOST = "114.55.58.251" +DEFAULT_USER = "root" +DEFAULT_BOOTSTRAP = "127.0.0.1:9092" +DEFAULT_KAFKA_BIN = "/opt/kafka/current/bin" +DEFAULT_TOPICS = [ + "vehicle.raw.go.gb32960.v1", + "vehicle.raw.go.jt808.v1", + "vehicle.raw.go.yutong-mqtt.v1", + "vehicle.event.go.unified.v1", +] +DEFAULT_GROUPS = [ + "go-history-writer", + "go-stat-writer", + "go-realtime-api", +] + + +@dataclass(frozen=True) +class Check: + name: str + status: str + message: str + + +@dataclass(frozen=True) +class ConsumerLag: + group: str + topic: str + partition: int + current_offset: int | None + log_end_offset: int + lag: int | None + + +def parse_topic_list(output: str) -> set[str]: + return {line.strip() for line in output.splitlines() if line.strip()} + + +def topic_checks(existing: set[str], required: list[str]) -> list[Check]: + checks: list[Check] = [] + for topic in required: + if topic in existing: + checks.append(Check("topic." + topic, "pass", "present")) + else: + checks.append(Check("topic." + topic, "fail", "missing")) + return checks + + +def parse_consumer_group_describe(output: str) -> list[ConsumerLag]: + rows: list[ConsumerLag] = [] + for line in output.splitlines(): + parts = line.split() + if len(parts) < 6 or parts[0] == "GROUP": + continue + group, topic = parts[0], parts[1] + try: + partition = int(parts[2]) + current = parse_optional_int(parts[3]) + log_end = int(parts[4]) + lag = parse_optional_int(parts[5]) + except ValueError: + continue + rows.append(ConsumerLag(group, topic, partition, current, log_end, lag)) + return rows + + +def parse_optional_int(value: str) -> int | None: + value = value.strip() + if value == "-": + return None + return int(value) + + +def group_lag_check(group: str, rows: list[ConsumerLag], max_lag: int) -> Check: + group_rows = [row for row in rows if row.group == group] + if not group_rows: + return Check("group." + group, "fail", "no rows") + concrete_lags = [row.lag for row in group_rows if row.lag is not None] + if not concrete_lags: + return Check("group." + group, "fail", "no concrete lag rows") + max_seen = max(concrete_lags) + topics = sorted({row.topic for row in group_rows}) + status = "pass" if max_seen <= max_lag else "fail" + return Check( + "group." + group, + status, + "max_lag=" + str(max_seen) + "; threshold=" + str(max_lag) + "; topics=" + ",".join(topics), + ) + + +def ssh(host: str, user: str, command: str, timeout: float) -> str: + target = user + "@" + host if user else host + completed = subprocess.run( + ["ssh", "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null", target, command], + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + timeout=timeout, + ) + return completed.stdout + + +def remote_topic_command(kafka_bin: str, bootstrap: str) -> str: + return kafka_bin.rstrip("/") + "/kafka-topics.sh --bootstrap-server " + bootstrap + " --list" + + +def remote_group_command(kafka_bin: str, bootstrap: str, group: str) -> str: + return kafka_bin.rstrip("/") + "/kafka-consumer-groups.sh --bootstrap-server " + bootstrap + " --describe --group " + group + + +def run(args: argparse.Namespace) -> tuple[str, list[Check]]: + topic_output = ssh(args.host, args.user, remote_topic_command(args.kafka_bin, args.bootstrap), args.timeout) + checks = topic_checks(parse_topic_list(topic_output), DEFAULT_TOPICS) + for group in DEFAULT_GROUPS: + group_output = ssh(args.host, args.user, remote_group_command(args.kafka_bin, args.bootstrap, group), args.timeout) + checks.append(group_lag_check(group, parse_consumer_group_describe(group_output), args.max_lag)) + status = "fail" if any(check.status == "fail" for check in checks) else "pass" + return status, checks + + +def parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--host", default=DEFAULT_HOST) + parser.add_argument("--user", default=DEFAULT_USER) + parser.add_argument("--bootstrap", default=DEFAULT_BOOTSTRAP) + parser.add_argument("--kafka-bin", default=DEFAULT_KAFKA_BIN) + parser.add_argument("--max-lag", type=int, default=100) + parser.add_argument("--timeout", type=float, default=20.0) + return parser.parse_args(argv) + + +def main(argv: list[str]) -> int: + args = parse_args(argv) + try: + status, checks = run(args) + except Exception as exc: + status = "fail" + checks = [Check("kafka.ssh", "fail", str(exc))] + print(json.dumps({ + "status": status, + "host": args.host, + "bootstrap": args.bootstrap, + "checks": [asdict(check) for check in checks], + }, ensure_ascii=False, indent=2)) + return 0 if status == "pass" else 1 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/tools/test_go_kafka_prod_smoke.py b/tools/test_go_kafka_prod_smoke.py new file mode 100644 index 00000000..9b6d5b74 --- /dev/null +++ b/tools/test_go_kafka_prod_smoke.py @@ -0,0 +1,62 @@ +import unittest + +from tools import go_kafka_prod_smoke as smoke + + +class GoKafkaProdSmokeTest(unittest.TestCase): + def test_parse_topic_list_marks_required_topics_present(self): + topics = smoke.parse_topic_list(""" +vehicle.raw.go.gb32960.v1 +vehicle.raw.go.jt808.v1 +vehicle.raw.go.yutong-mqtt.v1 +vehicle.event.go.unified.v1 +""") + + checks = smoke.topic_checks(topics, smoke.DEFAULT_TOPICS) + + self.assertTrue(all(check.status == "pass" for check in checks)) + + def test_topic_checks_fail_when_required_topic_missing(self): + topics = {"vehicle.raw.go.gb32960.v1"} + + checks = smoke.topic_checks(topics, ["vehicle.raw.go.gb32960.v1", "vehicle.event.go.unified.v1"]) + + by_name = {check.name: check for check in checks} + self.assertEqual(by_name["topic.vehicle.event.go.unified.v1"].status, "fail") + + def test_parse_consumer_group_describe_reads_lag_rows(self): + rows = smoke.parse_consumer_group_describe(""" +GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG +go-realtime-api vehicle.event.go.unified.v1 0 15992 15992 0 +go-realtime-api vehicle.event.go.unified.v1 1 2036 2036 3 +""") + + self.assertEqual(rows[0].group, "go-realtime-api") + self.assertEqual(rows[1].topic, "vehicle.event.go.unified.v1") + self.assertEqual(rows[1].lag, 3) + + def test_lag_checks_fail_when_lag_exceeds_threshold(self): + rows = [ + smoke.ConsumerLag("go-history-writer", "vehicle.raw.go.jt808.v1", 0, 100, 100, 0), + smoke.ConsumerLag("go-history-writer", "vehicle.raw.go.jt808.v1", 1, 100, 110, 10), + ] + + check = smoke.group_lag_check("go-history-writer", rows, max_lag=5) + + self.assertEqual(check.status, "fail") + self.assertIn("max_lag=10", check.message) + + def test_lag_checks_pass_when_group_has_rows_and_lag_is_small(self): + rows = [ + smoke.ConsumerLag("go-stat-writer", "vehicle.raw.go.gb32960.v1", 0, 100, 100, 0), + smoke.ConsumerLag("go-stat-writer", "vehicle.raw.go.jt808.v1", 1, 100, 101, 1), + ] + + check = smoke.group_lag_check("go-stat-writer", rows, max_lag=5) + + self.assertEqual(check.status, "pass") + self.assertIn("max_lag=1", check.message) + + +if __name__ == "__main__": + unittest.main()