From 9c43c85bef7594412c5ac11a2ab9d4f8bea32d6a Mon Sep 17 00:00:00 2001 From: Bharat Kathi Date: Sat, 29 Aug 2026 13:33:50 -0700 Subject: [PATCH] feat(p987): report why frames fail to decode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A frame that produces no signals never reaches the live path — it is stored with a status in its metadata and dropped. That is the right behaviour, but it was entirely silent, so "frames are arriving and nothing shows up live" could only be diagnosed by querying ClickHouse. Count every decode outcome and report it. Each distinct (bus, can id, reason) logs once when first seen, so a new unknown id is visible immediately; repeats roll up every 30s into one line carrying the decoded and undecodable totals plus the worst offenders. Frames arrive at a few hundred per second, so a line per failure would bury the log and out-write the disk. The totals matter as much as the reasons: "0 decoded, 4000 undecodable" and "nothing arriving at all" look identical from the live path, and now they don't. Also replaces the per-frame invalid-timestamp warning with the same counter, and pulls the status strings into constants so the metadata blob and the reporter cannot drift apart. --- p987/main.go | 3 + p987/service/decode_report.go | 128 ++++++++++++++++++++++++++++++++++ p987/service/message.go | 37 +++++----- 3 files changed, 152 insertions(+), 16 deletions(-) create mode 100644 p987/service/decode_report.go diff --git a/p987/main.go b/p987/main.go index 765998d6..5f604acd 100644 --- a/p987/main.go +++ b/p987/main.go @@ -23,6 +23,9 @@ func main() { database.Init() } + // Started before the subscription so the first frames are counted. + service.InitDecodeReporting() + mqtt.SetMessageHandler(service.HandleInboundMessage) if err := mqtt.Init(context.Background()); err != nil { logger.SugarLogger.Fatalf("Failed to initialize MQTT: %v", err) diff --git a/p987/service/decode_report.go b/p987/service/decode_report.go new file mode 100644 index 00000000..073b511a --- /dev/null +++ b/p987/service/decode_report.go @@ -0,0 +1,128 @@ +package service + +import ( + "fmt" + "sort" + "strings" + "sync" + "time" + + "github.com/gaucho-racing/mapache/p987/pkg/logger" +) + +// A frame that produces no signals never reaches the live path — it is +// stored with a status in its metadata and dropped. That is correct +// behaviour but it made "frames arrive, nothing shows up live" impossible +// to diagnose without querying ClickHouse, so decode outcomes are counted +// here and reported. +// +// Frames arrive at a few hundred per second, so a line per failure would +// bury everything else and out-write the disk. Each distinct +// (bus, can id, reason) logs once when first seen — a new unknown id is +// visible immediately — and repeats are rolled up on an interval. + +const decodeReportInterval = 30 * time.Second + +// maxReportedKinds bounds the summary line. A bus carrying dozens of +// undecodable ids should not produce an unbounded log entry. +const maxReportedKinds = 12 + +type decodeKey struct { + bus string + canID int + status string +} + +type decodeStat struct { + total int64 + sinceReport int64 +} + +var ( + decodeMu sync.Mutex + decodeStats = map[decodeKey]*decodeStat{} + decodedOK int64 + decodedFailed int64 +) + +// NoteDecodeOutcome records one frame's decode result. +func NoteDecodeOutcome(bus string, canID int, status, note string) { + decodeMu.Lock() + if status == statusOK { + decodedOK++ + decodeMu.Unlock() + return + } + decodedFailed++ + + key := decodeKey{bus: bus, canID: canID, status: status} + stat, ok := decodeStats[key] + if !ok { + stat = &decodeStat{} + decodeStats[key] = stat + } + first := stat.total == 0 + stat.total++ + stat.sinceReport++ + decodeMu.Unlock() + + if first { + logger.SugarLogger.Warnf("[DECODE] %s 0x%X %s: %s", bus, canID, status, note) + } +} + +// InitDecodeReporting starts the rollup. Runs for the life of the process. +func InitDecodeReporting() { + go func() { + ticker := time.NewTicker(decodeReportInterval) + defer ticker.Stop() + for range ticker.C { + reportDecodeOutcomes() + } + }() +} + +func reportDecodeOutcomes() { + decodeMu.Lock() + ok, failed := decodedOK, decodedFailed + decodedOK, decodedFailed = 0, 0 + + type entry struct { + key decodeKey + count int64 + } + entries := make([]entry, 0, len(decodeStats)) + for k, s := range decodeStats { + if s.sinceReport > 0 { + entries = append(entries, entry{k, s.sinceReport}) + s.sinceReport = 0 + } + } + decodeMu.Unlock() + + if ok == 0 && failed == 0 { + return + } + + // Loudest offenders first — that is what you act on. + sort.Slice(entries, func(i, j int) bool { return entries[i].count > entries[j].count }) + + var b strings.Builder + fmt.Fprintf(&b, "[DECODE] %s: %d decoded, %d undecodable", decodeReportInterval, ok, failed) + if len(entries) > 0 { + b.WriteString(" —") + for i, e := range entries { + if i == maxReportedKinds { + fmt.Fprintf(&b, " (+%d more kinds)", len(entries)-i) + break + } + fmt.Fprintf(&b, " %s/0x%X %s ×%d", e.key.bus, e.key.canID, e.key.status, e.count) + } + } + + if failed > 0 { + logger.SugarLogger.Warnln(b.String()) + return + } + logger.SugarLogger.Infoln(b.String()) +} diff --git a/p987/service/message.go b/p987/service/message.go index 7e623e51..9223eff4 100644 --- a/p987/service/message.go +++ b/p987/service/message.go @@ -16,6 +16,15 @@ import ( mapache "github.com/gaucho-racing/mapache/mapache-go/v3" ) +// Decode outcome statuses, recorded in the stored frame's metadata and +// counted by the decode reporter. +const ( + statusOK = "ok" + statusUnknownCANID = "unknown_can_id" + statusDecodeError = "decode_error" + statusInvalidTimestamp = "invalid_timestamp" +) + // headerSize is the relay's wire format: u64 BE microsecond timestamp // followed by a u16 BE upload key, then the raw CAN payload. const headerSize = 10 @@ -86,13 +95,10 @@ func ProcessFrame(vehicleID, bus string, canID, timestamp int, data []byte) (mod switch { case !IsValidProducedAt(timestamp): - logger.SugarLogger.Warnf("Frame with invalid timestamp: vehicle=%s bus=%s can_id=0x%X ts=%d decoded=%s", - vehicleID, bus, canID, timestamp, producedAt.UTC().Format(time.RFC3339Nano)) - meta = MustJSON(map[string]any{ - "status": "invalid_timestamp", - "note": fmt.Sprintf("ts=%d (%s) is before %s", timestamp, - producedAt.UTC().Format(time.RFC3339Nano), minValidProducedAt.UTC().Format(time.RFC3339Nano)), - }) + note := fmt.Sprintf("ts=%d (%s) is before %s", timestamp, + producedAt.UTC().Format(time.RFC3339Nano), minValidProducedAt.UTC().Format(time.RFC3339Nano)) + NoteDecodeOutcome(bus, canID, statusInvalidTimestamp, note) + meta = MustJSON(map[string]any{"status": statusInvalidTimestamp, "note": note}) default: decoded, meta = decodeFrame(bus, canID, data) } @@ -130,18 +136,17 @@ func ProcessFrame(vehicleID, bus string, canID, timestamp int, data []byte) (mod func decodeFrame(bus string, canID int, data []byte) ([]mapache.Signal, []byte) { messageStruct := model.GetMessage(bus, canID) if messageStruct == nil { - return nil, MustJSON(map[string]any{ - "status": "unknown_can_id", - "note": fmt.Sprintf("no decoder registered for can id 0x%X on bus %s", canID, bus), - }) + note := fmt.Sprintf("no decoder registered for can id 0x%X on bus %s", canID, bus) + NoteDecodeOutcome(bus, canID, statusUnknownCANID, note) + return nil, MustJSON(map[string]any{"status": statusUnknownCANID, "note": note}) } if err := messageStruct.FillFromBytes(data); err != nil { - return nil, MustJSON(map[string]any{ - "status": "decode_error", - "note": err.Error(), - }) + note := err.Error() + NoteDecodeOutcome(bus, canID, statusDecodeError, note) + return nil, MustJSON(map[string]any{"status": statusDecodeError, "note": note}) } - return messageStruct.ExportSignals(), MustJSON(map[string]any{"status": "ok"}) + NoteDecodeOutcome(bus, canID, statusOK, "") + return messageStruct.ExportSignals(), MustJSON(map[string]any{"status": statusOK}) } func HandleMessage(vehicleID string, bus string, canID int, message []byte) {