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) {