diff --git a/docs/ops/go-service-observability.md b/docs/ops/go-service-observability.md index a1e3f06f..85a06c04 100644 --- a/docs/ops/go-service-observability.md +++ b/docs/ops/go-service-observability.md @@ -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=''` 关闭。 diff --git a/docs/ops/vehicle-ingest-runbook.md b/docs/ops/vehicle-ingest-runbook.md index 8cb7b885..03eacf48 100644 --- a/docs/ops/vehicle-ingest-runbook.md +++ b/docs/ops/vehicle-ingest-runbook.md @@ -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。 | diff --git a/go/vehicle-gateway/cmd/capacity-check/main.go b/go/vehicle-gateway/cmd/capacity-check/main.go index b9f9ec3f..133ee6cb 100644 --- a/go/vehicle-gateway/cmd/capacity-check/main.go +++ b/go/vehicle-gateway/cmd/capacity-check/main.go @@ -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") diff --git a/go/vehicle-gateway/internal/capacity/check.go b/go/vehicle-gateway/internal/capacity/check.go index a9b29f0a..cc001fde 100644 --- a/go/vehicle-gateway/internal/capacity/check.go +++ b/go/vehicle-gateway/internal/capacity/check.go @@ -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) diff --git a/go/vehicle-gateway/internal/capacity/check_test.go b/go/vehicle-gateway/internal/capacity/check_test.go index 8bbdaf7d..8fb136b1 100644 --- a/go/vehicle-gateway/internal/capacity/check_test.go +++ b/go/vehicle-gateway/internal/capacity/check_test.go @@ -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