From 5539c5e0c2108d84abfdb351fdb1db0a6b15d499 Mon Sep 17 00:00:00 2001 From: lingniu Date: Sun, 19 Jul 2026 21:00:56 +0800 Subject: [PATCH] fix(stats): recover durable daily mileage projections --- docs/ops/go-service-observability.md | 2 + docs/ops/vehicle-ingest-runbook.md | 2 + go/vehicle-gateway/cmd/stat-writer/main.go | 30 +++++++ .../cmd/stat-writer/main_test.go | 14 ++++ .../internal/stats/daily_metric.go | 81 +++++++++++++++++++ .../internal/stats/daily_metric_test.go | 45 +++++++++++ 6 files changed, 174 insertions(+) diff --git a/docs/ops/go-service-observability.md b/docs/ops/go-service-observability.md index 470c23e4..5d404a5e 100644 --- a/docs/ops/go-service-observability.md +++ b/docs/ops/go-service-observability.md @@ -151,6 +151,8 @@ systemctl start lingniu-go-capacity-check.service | `vehicle_stat_sources_total` | Stat writer source tracking results for `vehicle_data_source`. Labels: `topic`, `protocol`, `status`; status includes `attempted`, `written`, `skipped_throttled`, `skipped_missing_endpoint`. This intentionally avoids source IP labels to keep metrics low-cardinality. | | `vehicle_stat_projections_total` | Final daily mileage projection results from `vehicle_daily_mileage_source` to `vehicle_daily_mileage`. Labels: `topic`, `protocol`, `status`; status includes `attempted`, `written`, `skipped_throttled`. | | `vehicle_stat_pending_projections_total` / `vehicle_stat_pending_projection_entries` | Eventual projection tail flush. `attempted` and `written` should grow together; `error` indicates a pending final projection remains queued for retry. The entries gauge should stay bounded by active vehicle/protocol/day keys and drain after the projection interval. | +| `vehicle_stat_startup_projection_reconcile_targets` / `vehicle_stat_startup_projection_reconcile_written` / `vehicle_stat_startup_projection_reconcile_failed` | Startup recovery for durable daily-mileage candidates that are newer than, or missing from, the final projection. Healthy startup has `written = targets` and `failed = 0`. | +| `vehicle_stat_last_startup_projection_reconcile_unix_seconds` | Last startup recovery result, labeled `status="ok"` or `status="error"`. An error preserves the gap as observable evidence and must be followed by a targeted backfill or another successful restart reconciliation. | | `vehicle_realtime_updates_total` | Redis/MySQL realtime projector updates. Labels: `topic`, `status`. | | `vehicle_realtime_kafka_messages_total` | Realtime projector Kafka message results by raw topic. Labels: `topic`, `status`; `invalid_json` messages are committed and isolated before Redis/MySQL projection. | | `vehicle_realtime_last_message_unix_seconds` | Last realtime Kafka message by topic/status. Labels: `topic`, `status`. | diff --git a/docs/ops/vehicle-ingest-runbook.md b/docs/ops/vehicle-ingest-runbook.md index 4395b443..affc491c 100644 --- a/docs/ops/vehicle-ingest-runbook.md +++ b/docs/ops/vehicle-ingest-runbook.md @@ -47,6 +47,8 @@ Gateway 在 NATS 模式下设置 `FIELDS_DERIVE_FROM_RAW_ENABLED=true`，每帧 如果 stat-writer 少消费某个 fields topic，对应协议的每日里程不会进入 `vehicle_daily_mileage`。stat-writer 启动时会拒绝 `vehicle.raw.*` 这类 RAW topic 配置，避免统计链路重新从原始报文解析并破坏 RAW/fields 解耦。修正后可能出现短时间 Kafka lag，这是在追补历史 backlog；`vehicle_stat_writes_total{status="ok"}` 只表示 append 调用成功，真正证明里程样本入库的是 `vehicle_stat_samples_total{status="written"}` 持续增长且 lag 下降。 +stat-writer 的节流尾部投影队列保存在内存中，但来源候选事实已经落在 MySQL。服务启动时默认执行 `STATS_RECONCILE_PROJECTIONS_ON_START=true` 的当天持久化候选扫描，把“候选比最终日表新”或“最终日表缺失”的车辆/协议重新投影；这用于关闭候选提交后进程重启造成的漏算窗口，尤其避免静止车辆因没有后续里程变化而一直漏到次日回补。默认超时由 `STATS_RECONCILE_PROJECTIONS_TIMEOUT_SECONDS=30` 控制。核验 `vehicle_stat_startup_projection_reconcile_targets`、`vehicle_stat_startup_projection_reconcile_written`、`vehicle_stat_startup_projection_reconcile_failed` 和 `vehicle_stat_last_startup_projection_reconcile_unix_seconds{status}`；失败不会伪装成功，仍由日志、指标和次日 backfill 继续暴露。 + stat-writer 默认 `STATS_WORKERS=3`，每个 worker 是同一 consumer group 的独立 Kafka reader。Kafka 按车辆 key 固定分区，因此单车累计里程仍按分区顺序执行，不同分区并行。调整 worker 后同时核对 `vehicle_stat_config{setting="workers"}`、每个 `vehicle_stat_worker_active{worker}`、MySQL 连接数、write p99 和 stat lag；worker 不应超过 fields topic 的有效分区并行度。 stat-writer 对死锁、锁等待和网络中断等瞬时 MySQL 故障按 `STATS_RETRY_ATTEMPTS` 有限重试，重试耗尽后保持失败关闭并等待存储恢复。只有 `NULL`、数值越界、类型转换、字段过长和 CHECK 约束这类明确的单消息数据错误，才会在同样的有限重试耗尽后写入 `STATS_QUARANTINE_DIR` 并提交该 Kafka 偏移量，避免一条坏消息阻塞整个分区；表结构、权限和未知程序错误不会被隔离。隔离文件包含原 topic、partition、offset、key、value 和错误，默认权限为 `0600`，处理前不得删除。 diff --git a/go/vehicle-gateway/cmd/stat-writer/main.go b/go/vehicle-gateway/cmd/stat-writer/main.go index c64e38c9..a8775d73 100644 --- a/go/vehicle-gateway/cmd/stat-writer/main.go +++ b/go/vehicle-gateway/cmd/stat-writer/main.go @@ -86,6 +86,18 @@ func main() { logger.Info("platform source startup normalization finished", "stat_date", statDate, "protocol", envelope.ProtocolJT808, "normalized", normalized) } } + if cfg.ReconcileProjectionsOnStart { + statDate := time.Now().In(cfg.Location).Format("2006-01-02") + reconcileCtx, cancel := context.WithTimeout(ctx, cfg.ReconcileProjectionsTimeout) + result, err := writer.ReconcileDateProjections(reconcileCtx, statDate) + cancel() + recordStartupProjectionReconcile(registry, result, err) + if err != nil { + logger.Warn("daily mileage startup projection reconciliation incomplete", "stat_date", statDate, "targets", result.Targets, "attempted", result.Attempted, "written", result.Written, "failed", result.Failed, "error", err) + } else { + logger.Info("daily mileage startup projection reconciliation finished", "stat_date", statDate, "targets", result.Targets, "written", result.Written) + } + } var appender statAppender = retryStatAppender{ delegate: writer, attempts: cfg.RetryAttempts, @@ -124,6 +136,20 @@ type pendingProjectionFlusher interface { FlushPendingProjections(context.Context, time.Time) (stats.ProjectionFlushResult, error) } +func recordStartupProjectionReconcile(registry *metrics.Registry, result stats.ProjectionReconcileResult, err error) { + if registry == nil { + return + } + registry.SetGauge("vehicle_stat_startup_projection_reconcile_targets", nil, float64(result.Targets)) + registry.SetGauge("vehicle_stat_startup_projection_reconcile_written", nil, float64(result.Written)) + registry.SetGauge("vehicle_stat_startup_projection_reconcile_failed", nil, float64(result.Failed)) + status := "ok" + if err != nil || result.Failed > 0 { + status = "error" + } + metrics.RecordLastActivity(registry, "vehicle_stat_last_startup_projection_reconcile_unix_seconds", metrics.Labels{"status": status}) +} + func runPendingProjectionFlusher(ctx context.Context, logger interface { Error(string, ...any) Warn(string, ...any) @@ -919,6 +945,8 @@ type config struct { RetryDelay time.Duration NormalizePlatformSourcesOnStart bool NormalizePlatformSourcesTimeout time.Duration + ReconcileProjectionsOnStart bool + ReconcileProjectionsTimeout time.Duration QuarantineDir string } @@ -963,6 +991,8 @@ func loadConfig() config { RetryDelay: time.Duration(envInt("STATS_RETRY_DELAY_MS", 20)) * time.Millisecond, NormalizePlatformSourcesOnStart: env("STATS_NORMALIZE_PLATFORM_SOURCES_ON_START", "true") != "false", NormalizePlatformSourcesTimeout: time.Duration(envInt("STATS_NORMALIZE_PLATFORM_SOURCES_TIMEOUT_SECONDS", 30)) * time.Second, + ReconcileProjectionsOnStart: env("STATS_RECONCILE_PROJECTIONS_ON_START", "true") != "false", + ReconcileProjectionsTimeout: time.Duration(envInt("STATS_RECONCILE_PROJECTIONS_TIMEOUT_SECONDS", 30)) * time.Second, QuarantineDir: env("STATS_QUARANTINE_DIR", "/var/lib/lingniu-go-native/stat-writer-quarantine"), } } diff --git a/go/vehicle-gateway/cmd/stat-writer/main_test.go b/go/vehicle-gateway/cmd/stat-writer/main_test.go index 66563a6d..91403358 100644 --- a/go/vehicle-gateway/cmd/stat-writer/main_test.go +++ b/go/vehicle-gateway/cmd/stat-writer/main_test.go @@ -1086,6 +1086,12 @@ func TestLoadConfigDefaultsToGoFieldsTopics(t *testing.T) { if cfg.NormalizePlatformSourcesTimeout != 30*time.Second { t.Fatalf("NormalizePlatformSourcesTimeout = %s, want 30s", cfg.NormalizePlatformSourcesTimeout) } + if !cfg.ReconcileProjectionsOnStart { + t.Fatal("ReconcileProjectionsOnStart = false, want default true") + } + if cfg.ReconcileProjectionsTimeout != 30*time.Second { + t.Fatalf("ReconcileProjectionsTimeout = %s, want 30s", cfg.ReconcileProjectionsTimeout) + } } func TestConfigValidateRejectsRawTopics(t *testing.T) { @@ -1172,6 +1178,8 @@ func TestLoadConfigReadsStatBatchSettings(t *testing.T) { t.Setenv("STATS_RETRY_DELAY_MS", "33") t.Setenv("STATS_NORMALIZE_PLATFORM_SOURCES_ON_START", "false") t.Setenv("STATS_NORMALIZE_PLATFORM_SOURCES_TIMEOUT_SECONDS", "17") + t.Setenv("STATS_RECONCILE_PROJECTIONS_ON_START", "false") + t.Setenv("STATS_RECONCILE_PROJECTIONS_TIMEOUT_SECONDS", "19") t.Setenv("STATS_QUARANTINE_DIR", "/tmp/stat-writer-quarantine-test") cfg := loadConfig() @@ -1197,6 +1205,12 @@ func TestLoadConfigReadsStatBatchSettings(t *testing.T) { if cfg.NormalizePlatformSourcesTimeout != 17*time.Second { t.Fatalf("NormalizePlatformSourcesTimeout = %s, want 17s", cfg.NormalizePlatformSourcesTimeout) } + if cfg.ReconcileProjectionsOnStart { + t.Fatal("ReconcileProjectionsOnStart = true, want env override false") + } + if cfg.ReconcileProjectionsTimeout != 19*time.Second { + t.Fatalf("ReconcileProjectionsTimeout = %s, want 19s", cfg.ReconcileProjectionsTimeout) + } if cfg.QuarantineDir != "/tmp/stat-writer-quarantine-test" { t.Fatalf("QuarantineDir = %q, want env override", cfg.QuarantineDir) } diff --git a/go/vehicle-gateway/internal/stats/daily_metric.go b/go/vehicle-gateway/internal/stats/daily_metric.go index e34300a7..6f7f6b5e 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric.go +++ b/go/vehicle-gateway/internal/stats/daily_metric.go @@ -95,6 +95,13 @@ type ProjectionFlushResult struct { Failed int } +type ProjectionReconcileResult struct { + Targets int + Attempted int + Written int + Failed int +} + type CacheStats struct { LastTotalMileageEntries int LastSourceSeenEntries int @@ -553,6 +560,62 @@ func (w *Writer) FlushPendingProjections(ctx context.Context, readyBefore time.T return result, flushErr } +// ReconcileDateProjections rebuilds final daily rows whose durable source +// candidates are newer than the projection (or whose projection is missing). +// The pending queue is intentionally in-memory for the hot path; this durable +// sweep closes the crash window between a committed source candidate and its +// throttled final projection. +func (w *Writer) ReconcileDateProjections(ctx context.Context, statDate string) (ProjectionReconcileResult, error) { + var result ProjectionReconcileResult + if w.query == nil { + return result, errors.New("daily mileage projection reconciliation requires query support") + } + statDate = strings.TrimSpace(statDate) + if statDate == "" { + return result, errors.New("daily mileage projection reconciliation requires stat date") + } + rows, err := w.query.QueryContext(ctx, staleDailyMileageProjectionTargetsSQL, statDate) + if err != nil { + return result, err + } + type target struct { + vin string + protocol envelope.Protocol + } + var targets []target + for rows.Next() { + var item target + if err := rows.Scan(&item.vin, &item.protocol); err != nil { + _ = rows.Close() + return result, err + } + if strings.TrimSpace(item.vin) != "" && strings.TrimSpace(string(item.protocol)) != "" { + targets = append(targets, item) + } + } + if err := rows.Close(); err != nil { + return result, err + } + if err := rows.Err(); err != nil { + return result, err + } + result.Targets = len(targets) + var reconcileErr error + for _, item := range targets { + if err := ctx.Err(); err != nil { + return result, errors.Join(reconcileErr, err) + } + result.Attempted++ + if err := ProjectDailyMileage(ctx, w.exec, item.vin, statDate, item.protocol); err != nil { + result.Failed++ + reconcileErr = errors.Join(reconcileErr, fmt.Errorf("reconcile daily mileage %s/%s/%s: %w", item.vin, statDate, item.protocol, err)) + continue + } + result.Written++ + } + return result, reconcileErr +} + func (w *Writer) projectPendingMileage(ctx context.Context, sample MetricSample) error { project := func(exec Execer) error { if strings.Contains(sample.SourceKey, platformSourceKeyPrefix) { @@ -576,6 +639,24 @@ func (w *Writer) projectPendingMileage(ctx context.Context, sample MetricSample) return project(w.exec) } +const staleDailyMileageProjectionTargetsSQL = ` +SELECT s.vin, s.protocol +FROM vehicle_daily_mileage_source s +LEFT JOIN vehicle_daily_mileage m + ON m.vin = s.vin + AND m.stat_date = s.stat_date + AND m.protocol = s.protocol +WHERE s.stat_date = ? + AND s.quality_status = '` + QualityOK + `' + AND ( + m.vin IS NULL + OR s.updated_at > m.updated_at + OR s.latest_event_time > m.updated_at + ) +GROUP BY s.vin, s.protocol +ORDER BY s.protocol ASC, s.vin ASC +` + func (w *Writer) seenSameMileage(sample MetricSample) bool { key := mileageCacheKey(sample) w.mu.Lock() diff --git a/go/vehicle-gateway/internal/stats/daily_metric_test.go b/go/vehicle-gateway/internal/stats/daily_metric_test.go index 38f11689..29d27d6e 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric_test.go +++ b/go/vehicle-gateway/internal/stats/daily_metric_test.go @@ -888,6 +888,51 @@ func TestWriterPendingProjectionFailureDoesNotBlockOtherVehicles(t *testing.T) { } } +func TestWriterReconcilesDurableProjectionAfterPendingQueueIsLost(t *testing.T) { + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual)) + if err != nil { + t.Fatalf("sqlmock.New() error = %v", err) + } + defer db.Close() + + writer := NewWriter(db, time.FixedZone("Asia/Shanghai", 8*3600)) + vin := "LMRKH9AC7R1004120" + statDate := "2026-07-19" + protocol := envelope.ProtocolJT808 + mock.ExpectQuery(staleDailyMileageProjectionTargetsSQL). + WithArgs(statDate). + WillReturnRows(sqlmock.NewRows([]string{"vin", "protocol"}).AddRow(vin, string(protocol))) + expectProjectDailyMileageSQL(mock, vin, statDate, protocol) + mock.ExpectCommit() + + result, err := writer.ReconcileDateProjections(context.Background(), statDate) + if err != nil { + t.Fatalf("ReconcileDateProjections() error = %v", err) + } + if result.Targets != 1 || result.Attempted != 1 || result.Written != 1 || result.Failed != 0 { + t.Fatalf("reconcile result = %+v, want one recovered projection", result) + } + if got := writer.CacheStats().PendingProjectionEntries; got != 0 { + t.Fatalf("startup reconciliation should not depend on pending memory, got %d entries", got) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + +func TestWriterProjectionReconciliationRequiresQueryableStore(t *testing.T) { + writer := NewWriter(&recordingExec{}, time.FixedZone("Asia/Shanghai", 8*3600)) + + result, err := writer.ReconcileDateProjections(context.Background(), "2026-07-19") + + if err == nil || !strings.Contains(err.Error(), "query support") { + t.Fatalf("ReconcileDateProjections() error = %v, want query support failure", err) + } + if result != (ProjectionReconcileResult{}) { + t.Fatalf("reconcile result = %+v, want empty", result) + } +} + func TestWriterAppendProjectsImmediatelyForNewSource(t *testing.T) { exec := &recordingExec{} loc := time.FixedZone("Asia/Shanghai", 8*3600)