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

fix(ops): recover stale history error health

Browse Source
This commit is contained in:
lingniu
2026-07-16 06:48:47 +08:00
parent e99087f329
commit ec3071d59b
5 changed files with 72 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/ops/go-service-observability.md
View File

@@ -47,7 +47,7 @@ Gateway 的 NATS 接受边界由 WAL outbox 提供。`vehicle_durable_outbox_bac
批处理内部 in-flight pending 默认允许 `100` 条窗口：`-fast-writer-batch-pending-max=100`、`-history-batch-pending-max=100`、`-history-rows-pending-max=100`、`-stat-batch-pending-max=100`。这类指标用于发现批写卡住，不再因为高频流量下几十条正在 flush/ack 的瞬时批次直接降级；NATS consumer pending、ack pending 和 Kafka lag 仍按独立阈值判断真实积压。
topic 流转检查默认只抓完全断层：history/realtime 必须消费三类 raw topic，stat 必须消费三类 fields topic；bridge 未配置 subject 的 `route_error` 默认超过 `0` 条就降级，可用 `-bridge-route-error-max=-1` 临时关闭；fast-writer `invalid_json` 默认超过 `0` 条就降级，可用 `-fast-writer-invalid-json-max=-1` 临时关闭；history-writer `invalid_json` 默认超过 `0` 条就降级，可用 `-history-invalid-json-max=-1` 临时关闭；realtime projector `invalid_json` 默认超过 `0` 条就降级，可用 `-realtime-invalid-json-max=-1` 临时关闭；stat-writer `invalid_json` 默认超过 `0` 条就降级，可用 `-stat-invalid-json-max=-1` 临时关闭；某协议 gateway raw publish 达到 `100` 条，扣除 `skipped_non_realtime` 后仍然没有 fields publish 时降级；实时 fields 缺失或 publish error 比例超过 `20%` 时也会降级；某协议 gateway raw/fields publish 达到 `1000` 条但 bridge 没有写对应 Kafka topic 时降级；某协议 gateway raw publish 达到 `1000` 条但 fast-writer 没有消费同 raw subject 并成功写 Redis/ack 时降级；某 raw topic bridge 写 Kafka 达到 `1000` 条但 history 未收到/写入或 realtime 未收到/更新时降级；某 fields topic bridge 写 Kafka 达到 `100` 条但 stat-writer 未配置或未收到该 topic 时降级。解析质量检查默认在单协议 gateway frame 样本达到 `100` 条后启用，`OK` 以外的 `PARTIAL/BAD_FRAME/other` 比例超过 `5%` 时降级，用于发现协议解析器或上游报文质量突然恶化。响应质量检查默认在单协议 gateway response 尝试样本达到 `100` 条后启用，`build_error/write_error` 比例超过 `5%` 时降级；`skipped` 表示该消息类型无需响应，不计入分母。身份解析质量检查默认在单协议 identity 样本达到 `100` 条后启用，`resolved` 以外的 `unresolved/error/timeout` 比例超过 `20%` 时降级，用于发现 VIN/车牌/phone 映射突然失效。history-writer 会对实时数据帧检查 raw envelope 是否携带 `parsed_fields`，默认单 topic/protocol 达到 `100` 帧后，缺失比例超过 `5%` 降级，用于发现历史证据层丢失扁平化解析字段。Redis 快路径字段质量检查默认在单 raw subject 写入字段达到 `1000` 个后启用，`skipped_stale/seen` 超过 `20%` 时降级，用于发现大面积旧帧补发或设备时间漂移。stat-writer 还会在单个 fields topic 成功 append 达到 `100` 条但未抽取到任何里程样本时降级，或 VIN 缺失、里程缺失、非正里程、source 缺失这类可行动跳过比例超过 `60%` 时降级，用于发现字段映射或身份/source 关联断裂；单 topic 找到 `100` 条里程样本但没有 source tracking，或 source endpoint 缺失比例超过 `60%` 时也会降级，用于发现 `vehicle_data_source` 来源管理断层；单 topic 已写入 `100` 条里程样本但最终 `vehicle_daily_mileage` 投影写入仍为 `0` 时也会降级，用于发现 source 候选层到最终查询层的断层；`skipped_same_mileage` 是重复总里程去重，不计入可行动跳过比例。已出现过的成功/received `*_last_*_unix_seconds` 如果超过 `300s` 未刷新，也会降级；刚启动尚未出现过该指标时不会因缺失 last activity 降级。`capacity-check` JSON 会额外派生 `*_last_*_age_seconds`，便于直接读取距今秒数。特殊环境可用 `-required-consumer-topics=''` 临时关闭消费 topic 合同检查。
topic 流转检查默认只抓完全断层：history/realtime 必须消费三类 raw topic，stat 必须消费三类 fields topic；bridge 未配置 subject 的 `route_error` 默认超过 `0` 条就降级，可用 `-bridge-route-error-max=-1` 临时关闭；fast-writer `invalid_json` 默认超过 `0` 条就降级，可用 `-fast-writer-invalid-json-max=-1` 临时关闭；history-writer `invalid_json` 默认超过 `0` 条就降级，可用 `-history-invalid-json-max=-1` 临时关闭；realtime projector `invalid_json` 默认超过 `0` 条就降级，可用 `-realtime-invalid-json-max=-1` 临时关闭；stat-writer `invalid_json` 默认超过 `0` 条就降级，可用 `-stat-invalid-json-max=-1` 临时关闭；某协议 gateway raw publish 达到 `100` 条，扣除 `skipped_non_realtime` 后仍然没有 fields publish 时降级；实时 fields 缺失或 publish error 比例超过 `20%` 时也会降级；某协议 gateway raw/fields publish 达到 `1000` 条但 bridge 没有写对应 Kafka topic 时降级；某协议 gateway raw publish 达到 `1000` 条但 fast-writer 没有消费同 raw subject 并成功写 Redis/ack 时降级；某 raw topic bridge 写 Kafka 达到 `1000` 条但 history 未收到/写入或 realtime 未收到/更新时降级；某 fields topic bridge 写 Kafka 达到 `100` 条但 stat-writer 未配置或未收到该 topic 时降级。解析质量检查默认在单协议 gateway frame 样本达到 `100` 条后启用，`OK` 以外的 `PARTIAL/BAD_FRAME/other` 比例超过 `5%` 时降级，用于发现协议解析器或上游报文质量突然恶化。响应质量检查默认在单协议 gateway response 尝试样本达到 `100` 条后启用，`build_error/write_error` 比例超过 `5%` 时降级；`skipped` 表示该消息类型无需响应，不计入分母。身份解析质量检查默认在单协议 identity 样本达到 `100` 条后启用，`resolved` 以外的 `unresolved/error/timeout` 比例超过 `20%` 时降级，用于发现 VIN/车牌/phone 映射突然失效。history-writer 会对实时数据帧检查 raw envelope 是否携带 `parsed_fields`，默认单 topic/protocol 达到 `100` 帧后，缺失比例超过 `5%` 降级，用于发现历史证据层丢失扁平化解析字段。Redis 快路径字段质量检查默认在单 raw subject 写入字段达到 `1000` 个后启用，`skipped_stale/seen` 超过 `20%` 时降级，用于发现大面积旧帧补发或设备时间漂移。stat-writer 还会在单个 fields topic 成功 append 达到 `100` 条但未抽取到任何里程样本时降级，或 VIN 缺失、里程缺失、非正里程、source 缺失这类可行动跳过比例超过 `60%` 时降级，用于发现字段映射或身份/source 关联断裂；单 topic 找到 `100` 条里程样本但没有 source tracking，或 source endpoint 缺失比例超过 `60%` 时也会降级，用于发现 `vehicle_data_source` 来源管理断层；单 topic 已写入 `100` 条里程样本但最终 `vehicle_daily_mileage` 投影写入仍为 `0` 时也会降级，用于发现 source 候选层到最终查询层的断层；`skipped_same_mileage` 是重复总里程去重，不计入可行动跳过比例。已出现过的成功/received `*_last_*_unix_seconds` 如果超过 `300s` 未刷新，也会降级；刚启动尚未出现过该指标时不会因缺失 last activity 降级。`capacity-check` JSON 会额外派生 `*_last_*_age_seconds`，便于直接读取距今秒数。历史位置派生写入错误使用 `vehicle_history_last_location_write_unix_seconds{status="error"}` 判断是否仍在发生：默认只在最近 `300s` 内出现错误时降级，进程生命周期累计 counter 继续保留用于审计；可用 `-history-location-error-recent-seconds` 调整窗口，设为 `0` 时恢复累计值门禁。特殊环境可用 `-required-consumer-topics=''` 临时关闭消费 topic 合同检查。
`capacity-check` 还会默认请求 Realtime API 的 `/api/stats/daily-metrics/diagnostics/reasons`，把当天每日里程诊断聚合放到 JSON 的 `daily_mileage_diagnostics`。这项默认只展示 `vehicle_total`、`actionable_issue_total` 和原因分布，不会因为存在业务缺口直接退出 `2`；需要把缺口纳入发布/巡检阻断时，显式传 `-daily-mileage-diagnostics-max-actionable=0` 或其他阈值。诊断 API 不可访问会被视为观测能力故障并降级；特殊环境可用 `-daily-mileage-diagnostics-url=''` 关闭。

1
docs/ops/vehicle-ingest-runbook.md
View File

@@ -290,6 +290,7 @@ export MYSQL_DSN="$(sed -n 's/^MYSQL_DSN=//p' /opt/lingniu-go-native/env/base.en
| `vehicle_async_sink_publish_duration_ms_histogram_bucket` | p99 超过 `capacity-check -async-sink-p99-ms` 或连续 5 分钟上升 | Gateway async worker 写 NATS/Kafka 变慢，通常会带动 queue depth 增长。 |
| `vehicle_history_batch_flush_total{status="error"}` | 任意增长 | TDengine 批写失败；Kafka offset 不会提交，应先查 TDengine 和 SQL 错误。 |
| `vehicle_history_parsed_fields_total{status="missing"}` | 单 topic/protocol 实时帧样本达到 `capacity-check -history-parsed-field-min-frames` 后，缺失比例超过 `capacity-check -history-parsed-field-max-missing-ratio` | history-writer 收到的实时 raw envelope 没有携带预计算 `parsed_fields`，TDengine RAW 证据层会缺少扁平化解析字段。优先查 gateway `BuildFieldsEnvelope` / raw envelope 构造、协议字段映射和最近部署。 |
| `vehicle_history_location_writes_total{status="error"}` / `vehicle_history_last_location_write_unix_seconds{status="error"}` | 累计错误超过 `-history-location-error-max`，且最后一次错误距今不超过 `-history-location-error-recent-seconds`（默认 `300s`） | TDengine 位置派生表最近仍有写入失败。优先查异常设备时间、坐标投影和 TDengine 时间范围；错误窗口恢复后健康状态自动恢复，累计 counter 仍用于审计。缺少最后错误时间时保持降级，避免观测缺失被误判为恢复。 |
| `vehicle_history_batch_pending_messages` | 超过 `capacity-check -history-batch-pending-max` 或 burst 后不回落 | history-writer 已拉取但未完成写入/提交，可能卡在 TDengine 或 Kafka commit。 |
| `vehicle_history_batch_pending_rows` | 超过 `capacity-check -history-rows-pending-max` 或 burst 后不回落 | TDengine 有批量写入积压，通常早于 Kafka lag 放大。 |
| `vehicle_history_batch_flush_duration_ms_histogram_bucket{status="ok"}` | p99 超过 `capacity-check -history-flush-p99-ms` 或持续上升 | TDengine 写入延迟增加，可能需要降低 batch size 或扩容 TDengine。 |

1
go/vehicle-gateway/cmd/capacity-check/main.go
View File

@@ -121,6 +121,7 @@ func main() {
flag.Float64Var(&thresholds.HistoryWriteRecentE2EP99MS, "history-write-recent-e2e-p99-ms", thresholds.HistoryWriteRecentE2EP99MS, "degrade when recent gateway received_at to history TDengine write p99 exceeds this many milliseconds; <=0 disables")
flag.Float64Var(&thresholds.HistoryWorkersMin, "history-workers-min", thresholds.HistoryWorkersMin, "degrade when history-writer runtime worker count is below this value; <=0 disables")
flag.Float64Var(&thresholds.HistoryLocationErrorMax, "history-location-error-max", thresholds.HistoryLocationErrorMax, "degrade when history-writer location derived write errors exceed this value; <0 disables")
flag.Float64Var(&thresholds.HistoryLocationErrorRecentSec, "history-location-error-recent-seconds", thresholds.HistoryLocationErrorRecentSec, "only degrade on history location write errors observed within this many seconds; <=0 evaluates the lifetime counter")
flag.Float64Var(&thresholds.HistoryLocationAccountingMaxRatio, "history-location-accounting-max-mismatch-ratio", thresholds.HistoryLocationAccountingMaxRatio, "degrade when location derived status counters do not account for successful history raw writes by more than this ratio; <=0 disables")
flag.Float64Var(&thresholds.HistoryFallbackErrorMax, "history-fallback-error-max", thresholds.HistoryFallbackErrorMax, "degrade when history-writer batch fallback single writes still fail; <0 disables")
flag.Float64Var(&thresholds.HistoryRetryMax, "history-retry-max", thresholds.HistoryRetryMax, "degrade when history-writer transient write retries exceed this value; <0 disables")

36
go/vehicle-gateway/internal/capacity/check.go
View File

@@ -563,6 +563,7 @@ type Thresholds struct {
HistoryWorkersMin float64
HistoryInvalidJSONMax float64
HistoryLocationErrorMax float64
HistoryLocationErrorRecentSec float64
HistoryLocationAccountingMaxRatio float64
HistoryFallbackErrorMax float64
HistoryRetryMax float64
@@ -690,6 +691,7 @@ func DefaultThresholds() Thresholds {
HistoryWorkersMin: 3,
HistoryInvalidJSONMax: 0,
HistoryLocationErrorMax: 0,
HistoryLocationErrorRecentSec: 300,
HistoryLocationAccountingMaxRatio: 0.02,
HistoryFallbackErrorMax: 0,
HistoryRetryMax: 100,
@@ -870,7 +872,7 @@ func EvaluateWithThresholds(metricsByService map[string]string, thresholds Thres
report.Findings = append(report.Findings, findingsForBridgeMessageCompletion(parsed, thresholds)...)
report.Findings = append(report.Findings, findingsForFieldsProjector(parsed, thresholds)...)
report.Findings = append(report.Findings, findingsForHistoryInvalidJSON(parsed, thresholds)...)
report.Findings = append(report.Findings, findingsForHistoryLocationErrors(parsed, thresholds)...)
report.Findings = append(report.Findings, findingsForHistoryLocationErrors(parsed, thresholds, report.CheckedAt)...)
report.Findings = append(report.Findings, findingsForHistoryLocationAccounting(parsed, thresholds)...)
report.Findings = append(report.Findings, findingsForHistoryFallbackErrors(parsed, thresholds)...)
report.Findings = append(report.Findings, findingsForWriteRetries(parsed, thresholds)...)
@@ -4079,10 +4081,22 @@ func findingsForFastWriterRedisFields(parsed parsedMetricSet, thresholds Thresho
return findings
}
func findingsForHistoryLocationErrors(parsed parsedMetricSet, thresholds Thresholds) []string {
func findingsForHistoryLocationErrors(parsed parsedMetricSet, thresholds Thresholds, checkedAt time.Time) []string {
if thresholds.HistoryLocationErrorMax < 0 {
return nil
}
lastErrorByTopic := map[string]float64{}
if thresholds.HistoryLocationErrorRecentSec > 0 {
for _, sample := range parsed.samples {
if sample.name != "vehicle_history_last_location_write_unix_seconds" || sample.labels["status"] != "error" || sample.value <= 0 {
continue
}
topic := strings.TrimSpace(sample.labels["topic"])
if sample.value > lastErrorByTopic[topic] {
lastErrorByTopic[topic] = sample.value
}
}
}
var findings []string
for _, sample := range parsed.samples {
if sample.name != "vehicle_history_location_writes_total" || sample.labels["status"] != "error" {
@@ -4095,11 +4109,27 @@ func findingsForHistoryLocationErrors(parsed parsedMetricSet, thresholds Thresho
if topic == "" {
topic = "unknown"
}
recency := ""
if thresholds.HistoryLocationErrorRecentSec > 0 {
if lastError, ok := lastErrorByTopic[topic]; ok {
ageSec := checkedAt.Sub(time.Unix(int64(lastError), 0).UTC()).Seconds()
if ageSec < 0 {
ageSec = 0
}
if ageSec > thresholds.HistoryLocationErrorRecentSec {
continue
}
recency = fmt.Sprintf(", last error %.0fs ago", ageSec)
} else {
recency = ", last error activity unavailable"
}
}
findings = append(findings, fmt.Sprintf(
"history location writes error %.0f exceeds %.0f for topic %s",
"history location writes error %.0f exceeds %.0f for topic %s%s",
sample.value,
thresholds.HistoryLocationErrorMax,
topic,
recency,
))
}
sort.Strings(findings)

38
go/vehicle-gateway/internal/capacity/check_test.go
View File

@@ -1173,12 +1173,14 @@ vehicle_history_kafka_messages_total{status="invalid_json",topic="vehicle.raw.go
}
func TestEvaluateReportsHistoryLocationWriteErrors(t *testing.T) {
recentError := time.Now().Add(-time.Minute).Unix()
report := EvaluateWithThresholds(map[string]string{
"history": `vehicle_service_info{service="vehicle-history-writer"} 1
vehicle_history_config{setting="workers"} 3
vehicle_kafka_consumer_info{service="vehicle-history-writer",group="go-history-writer",topic="vehicle.raw.go.jt808.v1"} 1
vehicle_history_location_writes_total{status="error",topic="vehicle.raw.go.jt808.v1"} 2`,
}, Thresholds{HistoryLocationErrorMax: 0})
vehicle_history_location_writes_total{status="error",topic="vehicle.raw.go.jt808.v1"} 2
vehicle_history_last_location_write_unix_seconds{status="error",topic="vehicle.raw.go.jt808.v1"} ` + strconv.FormatInt(recentError, 10),
}, Thresholds{HistoryLocationErrorMax: 0, HistoryLocationErrorRecentSec: 300})
if report.Status != StatusDegraded {
t.Fatalf("status = %s, want degraded: %#v", report.Status, report)
@@ -1187,6 +1189,7 @@ vehicle_history_location_writes_total{status="error",topic="vehicle.raw.go.jt808
for _, want := range []string{
"history location writes error 2 exceeds 0",
"vehicle.raw.go.jt808.v1",
"last error",
} {
if !strings.Contains(joined, want) {
t.Fatalf("finding missing %q in:\n%s", want, joined)
@@ -1194,6 +1197,37 @@ vehicle_history_location_writes_total{status="error",topic="vehicle.raw.go.jt808
}
}
func TestEvaluateIgnoresRecoveredHistoryLocationWriteErrors(t *testing.T) {
staleError := time.Now().Add(-10 * time.Minute).Unix()
report := EvaluateWithThresholds(map[string]string{
"history": `vehicle_service_info{service="vehicle-history-writer"} 1
vehicle_history_config{setting="workers"} 3
vehicle_kafka_consumer_info{service="vehicle-history-writer",group="go-history-writer",topic="vehicle.raw.go.jt808.v1"} 1
vehicle_history_location_writes_total{status="error",topic="vehicle.raw.go.jt808.v1"} 2
vehicle_history_last_location_write_unix_seconds{status="error",topic="vehicle.raw.go.jt808.v1"} ` + strconv.FormatInt(staleError, 10),
}, Thresholds{HistoryLocationErrorMax: 0, HistoryLocationErrorRecentSec: 300})
if report.Status != StatusOK {
t.Fatalf("status = %s, want ok after the error window recovered: %#v", report.Status, report)
}
if len(report.Findings) != 0 {
t.Fatalf("findings = %#v, want none after the error window recovered", report.Findings)
}
}
func TestEvaluateKeepsHistoryLocationErrorWhenLastActivityIsUnavailable(t *testing.T) {
report := EvaluateWithThresholds(map[string]string{
"history": `vehicle_service_info{service="vehicle-history-writer"} 1
vehicle_history_config{setting="workers"} 3
vehicle_kafka_consumer_info{service="vehicle-history-writer",group="go-history-writer",topic="vehicle.raw.go.jt808.v1"} 1
vehicle_history_location_writes_total{status="error",topic="vehicle.raw.go.jt808.v1"} 2`,
}, Thresholds{HistoryLocationErrorMax: 0, HistoryLocationErrorRecentSec: 300})
if report.Status != StatusDegraded || !strings.Contains(strings.Join(report.Findings, "\n"), "last error activity unavailable") {
t.Fatalf("missing timestamp must remain degraded, got %#v", report)
}
}
func TestEvaluateCanDisableHistoryLocationWriteErrorLimit(t *testing.T) {
report := EvaluateWithThresholds(map[string]string{
"history": `vehicle_service_info{service="vehicle-history-writer"} 1