diff --git a/README.md b/README.md index 2b1de4f..42d51c3 100644 --- a/README.md +++ b/README.md @@ -309,13 +309,13 @@ take away the control surface used to repair it. |---|---|---| | Dashboard | `:2112/ui` | Route graph, pending/retained state, bus diagnostics, device commissioning | | Manual | `:2112/ui/docs` | Offline getting started, CAN setup, concepts, filters, API, troubleshooting | -| MCP endpoint | `:2112/mcp` | Streamable HTTP tools for agent configuration, health, and delivery statistics | +| MCP endpoint | `:2112/mcp` | Streamable HTTP tools for agent configuration, health, delivery, and source PGN statistics | | MCP reference | `:2112/ui/mcp` | Embedded connection guide, tool catalog, and call examples | | REST API | `:2112/api/v1/...` | Configuration, live state, PGN catalog, inventory, commissioning | | API reference | `:2112/api/docs` | Interactive, embedded OpenAPI 3.1 documentation | | OpenAPI document | `:2112/api/openapi.json` | Machine-readable discovery for SDKs, scripts, and agents | | Health | `:2112/health` | Rolled-up component health; mirrored at `/api/v1/health` | -| Metrics | `:2112/metrics` | Prometheus exposition | +| Metrics | `:2112/metrics` | Prometheus exposition, including per-source/sender PGN traffic and value distributions | | SSE / WebSocket sinks | `:8080/` | Data and replay endpoints | | TCP sinks | Configured listener address | Live-only NDJSON stream | @@ -329,12 +329,24 @@ An MCP client can connect without a cloud relay or companion process: } ``` -The MCP server exposes nine tools to read the complete configuration, create or +The MCP server exposes twelve tools to read the complete configuration, create or update sources, sinks, and connector routes, delete each entity type, and read -health or delivery statistics. It uses the same validation, SQLite persistence, -and hot reconciliation as the UI and REST API. The server, tool schemas, and -reference page are all embedded in the Beacon binary; no internet connection, -remote schema, CDN, or hosted MCP service is required. +health, delivery statistics, or per-source PGN traffic metrics. Operators and +agents can approve or clear persistent expected-traffic baselines for each +source. It uses the same validation, SQLite persistence, and hot reconciliation as the UI and REST API. +The server, tool schemas, and reference page are all embedded in the Beacon +binary; no internet connection, remote schema, CDN, or hosted MCP service is +required. + +Each source overview groups traffic by source address and PGN, including PGNs +Beacon cannot decode. It reports learned frequency and jitter, gaps and bursts, +last-seen age, addressing and decode outcomes, traffic share and estimated CAN +load, payload lengths, decoded-field quantiles/availability/rate-of-change, and +bounded raw-wire fingerprints, entropy, byte ranges, and change masks. Approved +baselines make missing streams, frequency drift, payload/decode changes, address +moves, and out-of-range values visible after a restart. The scrape-safe subset +is exported as `beacon_source_pgn_*` at `/metrics`; raw payloads and fingerprint +identifiers stay in the UI/MCP response to avoid unbounded Prometheus labels. The admin API also exposes the complete PGN and field catalog, best-effort CAN/USB hardware discovery, stable Device NAME inventory, commissioning diff --git a/internal/app/app.go b/internal/app/app.go index 448d27b..d8f693e 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -95,12 +95,12 @@ func Run(ctx context.Context, opts Options) (*App, error) { } } - met, promHandler, err := metrics.New() + reg := stats.NewRegistry() + met, promHandler, err := metrics.New(reg) if err != nil { _ = st.Close() return nil, fmt.Errorf("init metrics: %w", err) } - reg := stats.NewRegistry() appliance, err := identity.LoadOrCreate(ctx, st) if err != nil { _ = st.Close() @@ -122,11 +122,19 @@ func Run(ctx context.Context, opts Options) (*App, error) { _ = st.Close() return nil, fmt.Errorf("start data server: %w", err) } + // Load approved expectations before sources start so even their first + // observed stream/change event is compared and persisted. + if err := reg.AttachSourceMetricPersistence(ctx, st.DB()); err != nil { + _ = ds.Stop(ctx) + _ = st.Close() + return nil, fmt.Errorf("load source metric history: %w", err) + } sup := supervisor.New(st, busMgr, ds, log, met, reg) if err := sup.Reconcile(ctx); err != nil { sup.Stop() _ = ds.Stop(ctx) + _ = reg.CloseSourceMetricPersistence(ctx) _ = st.Close() return nil, fmt.Errorf("initial reconcile: %w", err) } @@ -135,7 +143,6 @@ func Run(ctx context.Context, opts Options) (*App, error) { if err := inv.Refresh(ctx); err != nil { log.Warn("load N2K inventory", "err", err) } - a := &App{log: log, st: st, ds: ds, sup: sup, reg: reg, cfgSvc: cfgSvc, identity: appliance, inv: inv} invCtx, invCancel := context.WithCancel(context.Background()) a.invCancel = invCancel @@ -293,5 +300,7 @@ func (a *App) Close(ctx context.Context) error { } a.sup.Stop() // connectors flush final checkpoints _ = a.ds.Stop(ctx) - return a.st.Close() + persistenceErr := a.reg.CloseSourceMetricPersistence(ctx) + storeErr := a.st.Close() + return errors.Join(persistenceErr, storeErr) } diff --git a/internal/mcpserver/server.go b/internal/mcpserver/server.go index 13b84fa..67efa8d 100644 --- a/internal/mcpserver/server.go +++ b/internal/mcpserver/server.go @@ -42,6 +42,9 @@ var toolCatalog = []ToolInfo{ {Name: "delete_connector", Access: "delete", Description: "Delete a connector route and its route-owned queue."}, {Name: "get_health", Access: "read", Description: "Read rolled-up health and live source, sink, and connector states."}, {Name: "get_delivery_statistics", Access: "read", Description: "Read delivery rates, totals, pending delivery, retained history, limits, drops, and errors."}, + {Name: "get_source_metrics", Access: "read", Description: "Inspect all PGNs by source and sender, including unknown raw payloads, timing, load, decode quality, addressing, gaps, anomalies, field distributions, approved baselines, and lifecycle events."}, + {Name: "commit_source_traffic_baseline", Access: "write", Description: "Replace one source's persistent expected-traffic baseline with every PGN stream observed since Beacon started."}, + {Name: "clear_source_traffic_baseline", Access: "delete", Description: "Clear one source's approved traffic baseline without deleting its observations or event history."}, } // Catalog returns a copy so callers cannot mutate the registered tool list. @@ -288,6 +291,29 @@ type deliveryStatisticsInput struct { ConnectorID string `json:"connector_id,omitempty" jsonschema:"Optional connector route id. Omit to return every configured connector."` } +type sourceMetricsInput struct { + SourceID string `json:"source_id,omitempty" jsonschema:"Optional configured source id. Omit to inspect every source."` + PGN *uint32 `json:"pgn,omitempty" jsonschema:"Optional PGN number filter."` + SourceAddress *uint8 `json:"source_address,omitempty" jsonschema:"Optional NMEA 2000 source-address filter."` + EventLimit int `json:"event_limit,omitempty" jsonschema:"Recent lifecycle events per source to include; zero defaults to 50."` +} + +type sourceMetricsOutput struct { + GeneratedAt time.Time `json:"generated_at"` + Sources map[string][]stats.SourcePGNMetric `json:"sources"` + Baselines map[string][]stats.SourceTrafficBaseline `json:"baselines"` + Events map[string][]stats.SourceMetricEvent `json:"events"` +} + +type sourceTrafficBaselineInput struct { + SourceID string `json:"source_id" jsonschema:"Configured source id."` +} + +type sourceTrafficBaselineOutput struct { + SourceID string `json:"source_id"` + Baselines []stats.SourceTrafficBaseline `json:"baselines"` +} + type deliveryStatisticsOutput struct { Connectors map[string]deliveryStatistics `json:"connectors"` } @@ -462,6 +488,105 @@ func registerTools(server *sdkmcp.Server, svc *config.Service, reg *stats.Regist } return nil, out, nil }) + + sdkmcp.AddTool(server, tool("get_source_metrics", "Get source PGN metrics", readAnnotations), + func(ctx context.Context, _ *sdkmcp.CallToolRequest, in sourceMetricsInput) (*sdkmcp.CallToolResult, sourceMetricsOutput, error) { + out := sourceMetricsOutput{ + GeneratedAt: time.Now().UTC(), + Sources: map[string][]stats.SourcePGNMetric{}, + Baselines: map[string][]stats.SourceTrafficBaseline{}, + Events: map[string][]stats.SourceMetricEvent{}, + } + eventLimit := in.EventLimit + if eventLimit <= 0 || eventLimit > 200 { + eventLimit = 50 + } + if in.SourceID != "" { + if _, err := svc.GetSource(ctx, in.SourceID); err != nil { + return nil, sourceMetricsOutput{}, publicError(log, err) + } + out.Sources[in.SourceID] = filterSourceMetrics(reg.SourcePGNMetrics(in.SourceID), in) + out.Baselines[in.SourceID] = filterSourceBaselines(reg.SourceTrafficBaselines(in.SourceID), in) + out.Events[in.SourceID] = filterSourceMetricEvents(reg.SourceMetricEvents(in.SourceID, eventLimit), in) + return nil, out, nil + } + sources, err := svc.ListSources(ctx) + if err != nil { + return nil, sourceMetricsOutput{}, publicError(log, err) + } + all := reg.AllSourcePGNMetrics() + for _, source := range sources { + out.Sources[source.ID] = filterSourceMetrics(all[source.ID], in) + out.Baselines[source.ID] = filterSourceBaselines(reg.SourceTrafficBaselines(source.ID), in) + out.Events[source.ID] = filterSourceMetricEvents(reg.SourceMetricEvents(source.ID, eventLimit), in) + } + return nil, out, nil + }) + + sdkmcp.AddTool(server, tool("commit_source_traffic_baseline", "Set expected traffic baseline", writeAnnotations), + func(ctx context.Context, _ *sdkmcp.CallToolRequest, in sourceTrafficBaselineInput) (*sdkmcp.CallToolResult, sourceTrafficBaselineOutput, error) { + if _, err := svc.GetSource(ctx, in.SourceID); err != nil { + return nil, sourceTrafficBaselineOutput{}, publicError(log, err) + } + baselines, err := reg.CommitSourceTrafficBaseline(ctx, in.SourceID) + if err != nil { + return nil, sourceTrafficBaselineOutput{}, publicError(log, err) + } + return nil, sourceTrafficBaselineOutput{SourceID: in.SourceID, Baselines: baselines}, nil + }) + + sdkmcp.AddTool(server, tool("clear_source_traffic_baseline", "Clear source traffic baseline", deleteAnnotations), + func(ctx context.Context, _ *sdkmcp.CallToolRequest, in sourceTrafficBaselineInput) (*sdkmcp.CallToolResult, sourceTrafficBaselineOutput, error) { + if _, err := svc.GetSource(ctx, in.SourceID); err != nil { + return nil, sourceTrafficBaselineOutput{}, publicError(log, err) + } + if err := reg.ClearSourceTrafficBaseline(ctx, in.SourceID); err != nil { + return nil, sourceTrafficBaselineOutput{}, publicError(log, err) + } + return nil, sourceTrafficBaselineOutput{SourceID: in.SourceID, Baselines: []stats.SourceTrafficBaseline{}}, nil + }) +} + +func filterSourceMetrics(metrics []stats.SourcePGNMetric, in sourceMetricsInput) []stats.SourcePGNMetric { + out := make([]stats.SourcePGNMetric, 0, len(metrics)) + for _, metric := range metrics { + if in.PGN != nil && metric.PGN != *in.PGN { + continue + } + if in.SourceAddress != nil && metric.SourceAddress != *in.SourceAddress { + continue + } + out = append(out, metric) + } + return out +} + +func filterSourceBaselines(baselines []stats.SourceTrafficBaseline, in sourceMetricsInput) []stats.SourceTrafficBaseline { + out := make([]stats.SourceTrafficBaseline, 0, len(baselines)) + for _, baseline := range baselines { + if in.PGN != nil && baseline.PGN != *in.PGN { + continue + } + if in.SourceAddress != nil && baseline.SourceAddress != *in.SourceAddress { + continue + } + out = append(out, baseline) + } + return out +} + +func filterSourceMetricEvents(events []stats.SourceMetricEvent, in sourceMetricsInput) []stats.SourceMetricEvent { + out := make([]stats.SourceMetricEvent, 0, len(events)) + for _, event := range events { + if in.PGN != nil && event.PGN != *in.PGN { + continue + } + if in.SourceAddress != nil && event.SourceAddress != *in.SourceAddress { + continue + } + out = append(out, event) + } + return out } func tool(name, title string, annotations *sdkmcp.ToolAnnotations) *sdkmcp.Tool { diff --git a/internal/mcpserver/server_test.go b/internal/mcpserver/server_test.go index 152d406..0fd7c81 100644 --- a/internal/mcpserver/server_test.go +++ b/internal/mcpserver/server_test.go @@ -17,6 +17,7 @@ import ( "github.com/open-ships/beacon/internal/config" "github.com/open-ships/beacon/internal/model" + "github.com/open-ships/beacon/internal/msg" "github.com/open-ships/beacon/internal/stats" "github.com/open-ships/beacon/internal/store" "github.com/open-ships/beacon/internal/supervisor" @@ -65,6 +66,10 @@ func newTestMCP(t *testing.T) testMCP { }} svc := config.NewService(st, rec, slog.New(slog.NewTextHandler(io.Discard, nil))) reg := stats.NewRegistry() + if err := reg.AttachSourceMetricPersistence(context.Background(), st.DB()); err != nil { + _ = st.Close() + t.Fatal(err) + } server := NewServer(svc, reg, "test", slog.New(slog.NewTextHandler(io.Discard, nil))) clientTransport, serverTransport := sdkmcp.NewInMemoryTransports() serverSession, err := server.Connect(context.Background(), serverTransport, nil) @@ -82,6 +87,7 @@ func newTestMCP(t *testing.T) testMCP { t.Cleanup(func() { _ = clientSession.Close() _ = serverSession.Close() + _ = reg.CloseSourceMetricPersistence(context.Background()) _ = st.Close() }) return testMCP{svc: svc, stats: reg, client: clientSession, server: serverSession, store: st, rec: rec} @@ -235,6 +241,87 @@ func TestConfigureInspectAndDeleteThroughTools(t *testing.T) { } } +func TestGetSourceMetricsReturnsAndFiltersSharedPGNStore(t *testing.T) { + tm := newTestMCP(t) + mustPutSource := model.Source{ID: "can0", Name: "CAN", Type: model.SourceSocketCAN, Interface: "can0"} + if err := tm.svc.PutSource(context.Background(), mustPutSource, true); err != nil { + t.Fatal(err) + } + tm.stats.RecordSource("can0", &msg.Envelope{ + PGN: 127250, PGNName: "Vessel Heading", Source: 12, + Raw: []byte{1, 2, 3, 4, 5, 6, 7, 8}, Payload: json.RawMessage(`{"heading":1.5}`), + }) + tm.stats.RecordSource("can0", &msg.Envelope{PGN: 128259, Source: 44, Raw: []byte{1, 2}}) + + result := callTool(t, tm.client, "get_source_metrics", map[string]any{ + "source_id": "can0", "pgn": 127250, "source_address": 12, + }) + if result.IsError { + t.Fatalf("get_source_metrics: %s", toolErrorText(result)) + } + out := decodeStructured[sourceMetricsOutput](t, result) + streams, ok := out.Sources["can0"] + if !ok || len(streams) != 1 { + t.Fatalf("source metrics = %+v", out) + } + stream := streams[0] + if stream.PGN != 127250 || stream.SourceAddress != 12 || stream.PayloadBytesLast != 8 || stream.Messages != 1 { + t.Fatalf("filtered stream = %+v", stream) + } + if out.GeneratedAt.IsZero() { + t.Fatal("generated_at was not populated") + } + + result = callTool(t, tm.client, "get_source_metrics", map[string]any{"source_id": "missing"}) + if !result.IsError || !strings.Contains(toolErrorText(result), config.ErrNotFound.Error()) { + t.Fatalf("unknown source result = isError %v, content %q", result.IsError, toolErrorText(result)) + } +} + +func TestSourceTrafficBaselineToolsPersistAndReportEvents(t *testing.T) { + tm := newTestMCP(t) + if err := tm.svc.PutSource(context.Background(), model.Source{ + ID: "can0", Name: "CAN", Type: model.SourceSocketCAN, Interface: "can0", + }, true); err != nil { + t.Fatal(err) + } + for i := 0; i < 4; i++ { + tm.stats.RecordSource("can0", &msg.Envelope{ + PGN: 127250, PGNName: "Vessel Heading", Source: 12, Dest: 255, Priority: 2, + Raw: []byte{1, 2, 3, byte(i)}, Decode: msg.DecodeInfo{Status: "decoded", Complete: true}, + }) + } + + result := callTool(t, tm.client, "commit_source_traffic_baseline", map[string]any{"source_id": "can0"}) + if result.IsError { + t.Fatalf("commit baseline: %s", toolErrorText(result)) + } + committed := decodeStructured[sourceTrafficBaselineOutput](t, result) + if len(committed.Baselines) != 1 || committed.Baselines[0].PGN != 127250 { + t.Fatalf("committed baseline = %+v", committed) + } + + result = callTool(t, tm.client, "get_source_metrics", map[string]any{"source_id": "can0", "event_limit": 10}) + if result.IsError { + t.Fatalf("get metrics after baseline: %s", toolErrorText(result)) + } + out := decodeStructured[sourceMetricsOutput](t, result) + if len(out.Baselines["can0"]) != 1 || len(out.Events["can0"]) == 0 { + t.Fatalf("baseline/event output = %+v", out) + } + if got := out.Sources["can0"][0].BaselineStatus; got != "matching" { + t.Fatalf("baseline status = %q, want matching", got) + } + + result = callTool(t, tm.client, "clear_source_traffic_baseline", map[string]any{"source_id": "can0"}) + if result.IsError { + t.Fatalf("clear baseline: %s", toolErrorText(result)) + } + if baselines := tm.stats.SourceTrafficBaselines("can0"); len(baselines) != 0 { + t.Fatalf("baselines after clear = %+v", baselines) + } +} + func TestValidationAndDependencyFailuresAreToolErrors(t *testing.T) { tm := newTestMCP(t) diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 66d8537..549a454 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -5,6 +5,7 @@ package metrics import ( "context" "net/http" + "strconv" "sync" "github.com/prometheus/client_golang/prometheus" @@ -13,26 +14,58 @@ import ( otelprom "go.opentelemetry.io/otel/exporters/prometheus" api "go.opentelemetry.io/otel/metric" sdkmetric "go.opentelemetry.io/otel/sdk/metric" + + "github.com/open-ships/beacon/internal/stats" ) type gaugeKey struct{ kind, id string } type Set struct { - connectorMessages api.Int64Counter - connectorBytes api.Int64Counter - sourceMessages api.Int64Counter - subscriberDrops api.Int64Counter - queueDepth api.Int64ObservableGauge - queueBytes api.Int64ObservableGauge - componentState api.Int64ObservableGauge - sinkClients api.Int64UpDownCounter + connectorMessages api.Int64Counter + connectorBytes api.Int64Counter + sourceMessages api.Int64Counter + subscriberDrops api.Int64Counter + queueDepth api.Int64ObservableGauge + queueBytes api.Int64ObservableGauge + componentState api.Int64ObservableGauge + sinkClients api.Int64UpDownCounter + sourcePGNMessages api.Int64ObservableCounter + sourcePGNFrequency api.Float64ObservableGauge + sourcePGNPeriod api.Float64ObservableGauge + sourcePGNLastSeen api.Float64ObservableGauge + sourcePGNPayload api.Float64ObservableGauge + sourcePGNGap api.Int64ObservableGauge + sourcePGNGapRatio api.Float64ObservableGauge + sourcePGNGaps api.Int64ObservableCounter + sourcePGNAnomaly api.Int64ObservableGauge + sourcePGNAnomalies api.Int64ObservableCounter + sourcePGNTiming api.Float64ObservableGauge + sourcePGNTraffic api.Float64ObservableGauge + sourcePGNDecode api.Int64ObservableCounter + sourcePGNDecodeOutcome api.Int64ObservableCounter + sourcePGNDecodeMissing api.Int64ObservableCounter + sourcePGNBursts api.Int64ObservableCounter + sourcePGNPayloadLength api.Int64ObservableCounter + sourcePGNDestination api.Int64ObservableCounter + sourcePGNPriority api.Int64ObservableCounter + sourcePGNIdentity api.Int64ObservableCounter + sourcePGNStatus api.Int64ObservableGauge + sourcePGNBaseline api.Int64ObservableGauge + sourcePGNBaselineValue api.Float64ObservableGauge + sourcePGNRaw api.Float64ObservableGauge + sourcePGNRawByte api.Float64ObservableGauge + sourcePGNField api.Float64ObservableGauge + sourcePGNFieldState api.Float64ObservableGauge + sourcePGNFieldQuality api.Int64ObservableCounter + sourcePGNFieldAnomalies api.Int64ObservableCounter + sourcePGNCategory api.Float64ObservableGauge mu sync.Mutex depths map[string][2]int64 // connector -> {depth, bytes} states map[gaugeKey]int64 } -func New() (*Set, http.Handler, error) { +func New(registries ...*stats.Registry) (*Set, http.Handler, error) { reg := prometheus.NewRegistry() exporter, err := otelprom.New(otelprom.WithRegisterer(reg)) if err != nil { @@ -50,6 +83,36 @@ func New() (*Set, http.Handler, error) { s.queueDepth, _ = meter.Int64ObservableGauge("beacon.connector.queue.depth") s.queueBytes, _ = meter.Int64ObservableGauge("beacon.connector.queue.bytes") s.componentState, _ = meter.Int64ObservableGauge("beacon.component.state") + s.sourcePGNMessages, _ = meter.Int64ObservableCounter("beacon.source.pgn.messages") + s.sourcePGNFrequency, _ = meter.Float64ObservableGauge("beacon.source.pgn.frequency_hz") + s.sourcePGNPeriod, _ = meter.Float64ObservableGauge("beacon.source.pgn.expected_period_seconds") + s.sourcePGNLastSeen, _ = meter.Float64ObservableGauge("beacon.source.pgn.last_seen_unixtime") + s.sourcePGNPayload, _ = meter.Float64ObservableGauge("beacon.source.pgn.payload_bytes") + s.sourcePGNGap, _ = meter.Int64ObservableGauge("beacon.source.pgn.gap_active") + s.sourcePGNGapRatio, _ = meter.Float64ObservableGauge("beacon.source.pgn.gap_ratio") + s.sourcePGNGaps, _ = meter.Int64ObservableCounter("beacon.source.pgn.gaps") + s.sourcePGNAnomaly, _ = meter.Int64ObservableGauge("beacon.source.pgn.anomaly_active") + s.sourcePGNAnomalies, _ = meter.Int64ObservableCounter("beacon.source.pgn.anomalies") + s.sourcePGNTiming, _ = meter.Float64ObservableGauge("beacon.source.pgn.timing_seconds") + s.sourcePGNTraffic, _ = meter.Float64ObservableGauge("beacon.source.pgn.traffic") + s.sourcePGNDecode, _ = meter.Int64ObservableCounter("beacon.source.pgn.decode.messages") + s.sourcePGNDecodeOutcome, _ = meter.Int64ObservableCounter("beacon.source.pgn.decode.outcomes") + s.sourcePGNDecodeMissing, _ = meter.Int64ObservableCounter("beacon.source.pgn.decode.missing_fields") + s.sourcePGNBursts, _ = meter.Int64ObservableCounter("beacon.source.pgn.bursts") + s.sourcePGNPayloadLength, _ = meter.Int64ObservableCounter("beacon.source.pgn.payload_length.messages") + s.sourcePGNDestination, _ = meter.Int64ObservableCounter("beacon.source.pgn.destination.messages") + s.sourcePGNPriority, _ = meter.Int64ObservableCounter("beacon.source.pgn.priority.messages") + s.sourcePGNIdentity, _ = meter.Int64ObservableCounter("beacon.source.pgn.identity_changes") + s.sourcePGNStatus, _ = meter.Int64ObservableGauge("beacon.source.pgn.status") + s.sourcePGNBaseline, _ = meter.Int64ObservableGauge("beacon.source.pgn.baseline_state") + s.sourcePGNBaselineValue, _ = meter.Float64ObservableGauge("beacon.source.pgn.baseline_value") + s.sourcePGNRaw, _ = meter.Float64ObservableGauge("beacon.source.pgn.raw_payload") + s.sourcePGNRawByte, _ = meter.Float64ObservableGauge("beacon.source.pgn.raw_byte") + s.sourcePGNField, _ = meter.Float64ObservableGauge("beacon.source.pgn.field.value") + s.sourcePGNFieldState, _ = meter.Float64ObservableGauge("beacon.source.pgn.field.state") + s.sourcePGNFieldQuality, _ = meter.Int64ObservableCounter("beacon.source.pgn.field.quality") + s.sourcePGNFieldAnomalies, _ = meter.Int64ObservableCounter("beacon.source.pgn.field.anomalies") + s.sourcePGNCategory, _ = meter.Float64ObservableGauge("beacon.source.pgn.field.category_summary") _, err = meter.RegisterCallback(func(_ context.Context, o api.Observer) error { s.mu.Lock() defer s.mu.Unlock() @@ -66,9 +129,244 @@ func New() (*Set, http.Handler, error) { if err != nil { return nil, nil, err } + if len(registries) > 0 && registries[0] != nil { + reg := registries[0] + _, err = meter.RegisterCallback(func(_ context.Context, o api.Observer) error { + observeSourcePGNMetrics(o, s, reg.AllSourcePGNMetrics()) + return nil + }, s.sourcePGNMessages, s.sourcePGNFrequency, s.sourcePGNPeriod, + s.sourcePGNLastSeen, s.sourcePGNPayload, s.sourcePGNGap, + s.sourcePGNGapRatio, s.sourcePGNGaps, s.sourcePGNAnomaly, + s.sourcePGNAnomalies, s.sourcePGNTiming, s.sourcePGNTraffic, + s.sourcePGNDecode, s.sourcePGNDecodeOutcome, s.sourcePGNDecodeMissing, + s.sourcePGNBursts, s.sourcePGNPayloadLength, s.sourcePGNDestination, + s.sourcePGNPriority, s.sourcePGNIdentity, s.sourcePGNStatus, s.sourcePGNBaseline, + s.sourcePGNBaselineValue, s.sourcePGNRaw, s.sourcePGNRawByte, + s.sourcePGNField, s.sourcePGNFieldState, s.sourcePGNFieldQuality, + s.sourcePGNFieldAnomalies, s.sourcePGNCategory) + if err != nil { + return nil, nil, err + } + } return s, promhttp.HandlerFor(reg, promhttp.HandlerOpts{}), nil } +func observeSourcePGNMetrics(o api.Observer, set *Set, all map[string][]stats.SourcePGNMetric) { + for _, streams := range all { + for _, stream := range streams { + manufacturerCode := "" + if stream.ManufacturerCode != nil { + manufacturerCode = strconv.Itoa(int(*stream.ManufacturerCode)) + } + base := []attribute.KeyValue{ + attribute.String("source", stream.SourceID), + attribute.Int64("pgn", int64(stream.PGN)), + attribute.String("pgn_name", stream.PGNName), + attribute.Int64("source_address", int64(stream.SourceAddress)), + attribute.String("device_name", stream.DeviceNameHex), + attribute.String("variant", stream.Variant), + attribute.String("transport", stream.Transport), + attribute.String("manufacturer_code", manufacturerCode), + } + baselineAttrs := appendMetricAttribute(base, attribute.String("status", stream.BaselineStatus)) + o.ObserveInt64(set.sourcePGNBaseline, boolMetric(stream.Expected), api.WithAttributes(baselineAttrs...)) + statusAttrs := appendMetricAttribute(base, attribute.String("status", stream.Status)) + o.ObserveInt64(set.sourcePGNStatus, 1, api.WithAttributes(statusAttrs...)) + if stream.Expected { + for statistic, value := range map[string]float64{ + "expected_frequency_hz": stream.BaselineFrequencyHz, + "frequency_tolerance_percent": stream.BaselineTolerancePercent, + "frequency_drift_percent": stream.FrequencyDriftPercent, + } { + attrs := appendMetricAttribute(base, attribute.String("statistic", statistic)) + o.ObserveFloat64(set.sourcePGNBaselineValue, value, api.WithAttributes(attrs...)) + } + } + if !stream.Observed { + continue + } + o.ObserveInt64(set.sourcePGNMessages, stream.Messages, api.WithAttributes(base...)) + o.ObserveFloat64(set.sourcePGNFrequency, stream.FrequencyHz, api.WithAttributes(base...)) + o.ObserveFloat64(set.sourcePGNPeriod, stream.ExpectedPeriodSeconds, api.WithAttributes(base...)) + o.ObserveFloat64(set.sourcePGNLastSeen, float64(stream.LastSeen.UnixNano())/1e9, api.WithAttributes(base...)) + for statistic, value := range map[string]float64{ + "shortest": stream.ShortestPeriodSeconds, "median": stream.ExpectedPeriodSeconds, + "p95": stream.PeriodP95Seconds, "p99": stream.PeriodP99Seconds, + "mad": stream.JitterMADSeconds, + } { + attrs := appendMetricAttribute(base, attribute.String("statistic", statistic)) + o.ObserveFloat64(set.sourcePGNTiming, value, api.WithAttributes(attrs...)) + } + for statistic, value := range map[string]float64{ + "messages_per_second": stream.RecentMessagesPerSec, + "bytes_per_second": stream.RecentBytesPerSec, + "estimated_bus_load_percent": stream.EstimatedBusLoadPercent, + "source_share_percent": stream.TrafficSharePercent, + } { + attrs := appendMetricAttribute(base, attribute.String("statistic", statistic)) + o.ObserveFloat64(set.sourcePGNTraffic, value, api.WithAttributes(attrs...)) + } + for status, count := range stream.DecodeStatuses { + attrs := appendMetricAttribute(base, attribute.String("status", status)) + o.ObserveInt64(set.sourcePGNDecode, count, api.WithAttributes(attrs...)) + } + for outcome, count := range map[string]int64{ + "complete": stream.DecodeComplete, "incomplete": stream.DecodeIncomplete, + "fallback": stream.DecodeFallback, "unknown": stream.UnknownMessages, + } { + attrs := appendMetricAttribute(base, attribute.String("outcome", outcome)) + o.ObserveInt64(set.sourcePGNDecodeOutcome, count, api.WithAttributes(attrs...)) + } + for field, count := range stream.MissingDecodedFields { + attrs := appendMetricAttribute(base, attribute.String("field", field)) + o.ObserveInt64(set.sourcePGNDecodeMissing, count, api.WithAttributes(attrs...)) + } + o.ObserveInt64(set.sourcePGNBursts, stream.BurstCount, api.WithAttributes(base...)) + o.ObserveInt64(set.sourcePGNIdentity, stream.IdentityChanges, api.WithAttributes(base...)) + for destination, count := range stream.DestinationCounts { + attrs := appendMetricAttribute(base, attribute.String("destination", destination)) + o.ObserveInt64(set.sourcePGNDestination, count, api.WithAttributes(attrs...)) + } + for priority, count := range stream.PriorityCounts { + attrs := appendMetricAttribute(base, attribute.String("priority", priority)) + o.ObserveInt64(set.sourcePGNPriority, count, api.WithAttributes(attrs...)) + } + for statistic, value := range map[string]float64{ + "last": float64(stream.PayloadBytesLast), "min": float64(stream.PayloadBytesMin), + "max": float64(stream.PayloadBytesMax), "mean": stream.PayloadBytesMean, + } { + attrs := appendMetricAttribute(base, attribute.String("statistic", statistic)) + o.ObserveFloat64(set.sourcePGNPayload, value, api.WithAttributes(attrs...)) + } + gap := int64(0) + if stream.GapActive { + gap = 1 + } + o.ObserveInt64(set.sourcePGNGap, gap, api.WithAttributes(base...)) + o.ObserveFloat64(set.sourcePGNGapRatio, stream.GapRatio, api.WithAttributes(base...)) + o.ObserveInt64(set.sourcePGNGaps, stream.GapCount, api.WithAttributes(base...)) + anomaly := int64(0) + if stream.AnomalyActive { + anomaly = 1 + } + o.ObserveInt64(set.sourcePGNAnomaly, anomaly, api.WithAttributes(base...)) + o.ObserveInt64(set.sourcePGNAnomalies, stream.AnomalyCount, api.WithAttributes(base...)) + if stream.Raw != nil { + for length, count := range stream.Raw.LengthCounts { + attrs := appendMetricAttribute(base, attribute.String("length_bytes", length)) + o.ObserveInt64(set.sourcePGNPayloadLength, count, api.WithAttributes(attrs...)) + } + for statistic, value := range map[string]float64{ + "distinct_payloads": float64(stream.Raw.DistinctPayloads), + "fingerprint_overflow": float64(stream.Raw.DistinctPayloadOverflow), + "unchanged_seconds": stream.Raw.UnchangedSeconds, + "hamming_distance_mean_bits": stream.Raw.HammingDistanceMean, + "hamming_distance_p95_bits": stream.Raw.HammingDistanceP95, + } { + attrs := appendMetricAttribute(base, attribute.String("statistic", statistic)) + o.ObserveFloat64(set.sourcePGNRaw, value, api.WithAttributes(attrs...)) + } + for _, rawByte := range stream.Raw.Bytes { + byteBase := appendMetricAttribute(base, attribute.Int("offset", rawByte.Offset)) + for statistic, value := range map[string]float64{ + "minimum": float64(rawByte.Minimum), "maximum": float64(rawByte.Maximum), + "mode": float64(rawByte.MostCommon), "mode_share": rawByte.MostCommonShare, + "entropy_bits": rawByte.EntropyBits, "changed_share": rawByte.ChangedShare, + } { + attrs := appendMetricAttribute(byteBase, attribute.String("statistic", statistic)) + o.ObserveFloat64(set.sourcePGNRawByte, value, api.WithAttributes(attrs...)) + } + if mask, err := strconv.ParseUint(rawByte.ChangedBitMaskHex, 16, 8); err == nil { + attrs := appendMetricAttribute(byteBase, attribute.String("statistic", "changed_bit_mask")) + o.ObserveFloat64(set.sourcePGNRawByte, float64(mask), api.WithAttributes(attrs...)) + } + } + } + + for _, field := range stream.Fields { + fieldBase := appendMetricAttribute(base, + attribute.String("field", field.Field), attribute.String("unit", field.Unit)) + if field.Kind == "number" { + values := map[string]*float64{ + "last": field.LastNumeric, "min": field.Minimum, "max": field.Maximum, + "mean": field.Mean, "stddev": field.StdDev, "last_change": field.LastChange, + "p05": field.P05, "p50": field.P50, "p95": field.P95, "p99": field.P99, + "rate_of_change_per_second": field.LastRateOfChange, + "catalog_minimum": field.CatalogMinimum, "catalog_maximum": field.CatalogMaximum, + } + for statistic, value := range values { + if value == nil { + continue + } + attrs := appendMetricAttribute(fieldBase, attribute.String("statistic", statistic)) + o.ObserveFloat64(set.sourcePGNField, *value, api.WithAttributes(attrs...)) + } + o.ObserveInt64(set.sourcePGNFieldAnomalies, field.AnomalyCount, api.WithAttributes(fieldBase...)) + } + for statistic, value := range map[string]float64{ + "availability_percent": field.AvailabilityPercent, + "stuck_seconds": field.StuckSeconds, + "anomaly_score": field.AnomalyScore, + } { + attrs := appendMetricAttribute(fieldBase, attribute.String("statistic", statistic)) + o.ObserveFloat64(set.sourcePGNFieldState, value, api.WithAttributes(attrs...)) + } + for kind, count := range map[string]int64{ + "present": field.PresentMessages, "missing": field.MissingMessages, + "invalid": field.InvalidCount, "out_of_range": field.OutOfRangeCount, + "novel_category": field.NovelValueCount, + } { + attrs := appendMetricAttribute(fieldBase, attribute.String("kind", kind)) + o.ObserveInt64(set.sourcePGNFieldQuality, count, api.WithAttributes(attrs...)) + } + if field.Kind == "category" { + distinct, total, top := categorySummary(field) + for statistic, value := range map[string]float64{ + "distinct_values": float64(distinct), "samples": float64(total), + "top_value_share": top, "overflow_samples": float64(field.Other), + } { + attrs := appendMetricAttribute(fieldBase, attribute.String("statistic", statistic)) + o.ObserveFloat64(set.sourcePGNCategory, value, api.WithAttributes(attrs...)) + } + } + } + } + } +} + +func boolMetric(value bool) int64 { + if value { + return 1 + } + return 0 +} + +func categorySummary(field stats.FieldDistribution) (distinct int, total int64, topShare float64) { + var top int64 + for _, count := range field.Values { + distinct++ + total += count + if count > top { + top = count + } + } + total += field.Other + if field.Other > 0 { + distinct++ + } + if total > 0 { + topShare = float64(top) / float64(total) + } + return distinct, total, topShare +} + +func appendMetricAttribute(base []attribute.KeyValue, values ...attribute.KeyValue) []attribute.KeyValue { + out := make([]attribute.KeyValue, 0, len(base)+len(values)) + out = append(out, base...) + out = append(out, values...) + return out +} + func (s *Set) ConnectorMessages(ctx context.Context, connector, stage string, n int64) { if s == nil { return diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go index 1ad9193..4f78d46 100644 --- a/internal/metrics/metrics_test.go +++ b/internal/metrics/metrics_test.go @@ -2,10 +2,15 @@ package metrics import ( "context" + "encoding/json" "io" "net/http/httptest" "strings" "testing" + + "github.com/open-ships/beacon/internal/msg" + "github.com/open-ships/beacon/internal/n2kcatalog" + "github.com/open-ships/beacon/internal/stats" ) func TestNilSetIsSafe(t *testing.T) { @@ -72,3 +77,54 @@ func TestPrometheusExposition(t *testing.T) { } } } + +func TestPrometheusExposesSharedSourcePGNMetrics(t *testing.T) { + reg := stats.NewRegistry() + reg.RecordSource("can0", &msg.Envelope{ + PGN: 127250, PGNName: "Vessel Heading", Source: 12, + Raw: []byte{1, 2, 3, 4, 5, 6, 7, 8}, + Payload: json.RawMessage(`{"heading":1.5,"reference":"magnetic"}`), + Physical: map[string]n2kcatalog.PhysicalField{ + "heading": {Value: 1.5, Unit: "rad"}, + }, + }) + _, handler, err := New(reg) + if err != nil { + t.Fatal(err) + } + + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, httptest.NewRequest("GET", "/metrics", nil)) + body := rec.Body.String() + for _, want := range []string{ + "beacon_source_pgn_messages_total", + "beacon_source_pgn_frequency_hz", + "beacon_source_pgn_expected_period_seconds", + "beacon_source_pgn_last_seen_unixtime", + "beacon_source_pgn_payload_bytes", + "beacon_source_pgn_gap_active", + "beacon_source_pgn_anomaly_active", + "beacon_source_pgn_timing_seconds", + "beacon_source_pgn_traffic", + "beacon_source_pgn_decode_messages_total", + "beacon_source_pgn_decode_outcomes_total", + "beacon_source_pgn_payload_length_messages_total", + "beacon_source_pgn_destination_messages_total", + "beacon_source_pgn_priority_messages_total", + "beacon_source_pgn_baseline_state", + "beacon_source_pgn_status", + "beacon_source_pgn_raw_payload", + "beacon_source_pgn_raw_byte", + "beacon_source_pgn_field_value", + "beacon_source_pgn_field_state", + "beacon_source_pgn_field_quality_total", + "beacon_source_pgn_field_category_summary", + `pgn="127250"`, + `source="can0"`, + `source_address="12"`, + } { + if !strings.Contains(body, want) { + t.Fatalf("source PGN exposition missing %q:\n%s", want, body) + } + } +} diff --git a/internal/stats/source_baseline.go b/internal/stats/source_baseline.go new file mode 100644 index 0000000..5d0ea35 --- /dev/null +++ b/internal/stats/source_baseline.go @@ -0,0 +1,191 @@ +package stats + +import ( + "fmt" + "sort" + "strconv" + "strings" + "time" +) + +func (r *Registry) sourceBaselinesFor(source string) []SourceTrafficBaseline { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]SourceTrafficBaseline, 0) + for key, baseline := range r.sourceBaselines { + if key.source == source { + out = append(out, baseline) + } + } + return out +} + +func applySourceBaselines(metrics []SourcePGNMetric, baselines []SourceTrafficBaseline, now, startedAt time.Time) []SourcePGNMetric { + matched := make(map[string]bool) + byKey := make(map[string]SourceTrafficBaseline, len(baselines)) + for _, baseline := range baselines { + byKey[baseline.Identity+fmt.Sprintf(":%d", baseline.PGN)] = baseline + } + for i := range metrics { + identity := sourceBaselineIdentity(metrics[i].DeviceNameHex, metrics[i].SourceAddress) + key := identity + fmt.Sprintf(":%d", metrics[i].PGN) + baseline, ok := byKey[key] + if !ok && metrics[i].DeviceNameHex != "" { + fallback := sourceBaselineIdentity("", metrics[i].SourceAddress) + fmt.Sprintf(":%d", metrics[i].PGN) + baseline, ok = byKey[fallback] + key = fallback + } + if !ok { + metrics[i].BaselineStatus = "not_baselined" + continue + } + matched[key] = true + compareSourceBaseline(&metrics[i], baseline) + } + for _, baseline := range baselines { + key := baseline.Identity + fmt.Sprintf(":%d", baseline.PGN) + if matched[key] { + continue + } + grace := 10 * time.Second + if baseline.ExpectedFrequencyHz > 0 { + candidate := time.Duration(3 / baseline.ExpectedFrequencyHz * float64(time.Second)) + if candidate > grace { + grace = candidate + } + } + status := "awaiting" + gap := false + if now.Sub(startedAt) > grace { + status = "missing" + gap = true + } + approved := baseline.ApprovedAt + metrics = append(metrics, SourcePGNMetric{ + Observed: false, SourceID: baseline.SourceID, PGN: baseline.PGN, PGNName: baseline.PGNName, + Variant: baseline.Variant, Transport: baseline.Transport, DecodeStatus: baseline.DecodeStatus, + SourceAddress: baseline.SourceAddress, DeviceNameHex: baseline.DeviceNameHex, + FrequencyHz: baseline.ExpectedFrequencyHz, + ExpectedPeriodSeconds: baselinePeriodSeconds(baseline.ExpectedFrequencyHz), + Expected: true, BaselineStatus: status, + BaselineFrequencyHz: baseline.ExpectedFrequencyHz, + BaselineTolerancePercent: baseline.FrequencyTolerancePercent, + BaselineApprovedAt: &approved, BaselineIssues: []string{"expected stream has not been observed"}, + GapActive: gap, Status: status, + }) + } + return metrics +} + +func compareSourceBaseline(metric *SourcePGNMetric, baseline SourceTrafficBaseline) { + metric.Expected = true + metric.BaselineStatus = "matching" + metric.BaselineFrequencyHz = baseline.ExpectedFrequencyHz + metric.BaselineTolerancePercent = baseline.FrequencyTolerancePercent + approved := baseline.ApprovedAt + metric.BaselineApprovedAt = &approved + issues := make([]string, 0) + if baseline.ExpectedFrequencyHz > 0 && metric.FrequencyHz > 0 { + metric.FrequencyDriftPercent = (metric.FrequencyHz - baseline.ExpectedFrequencyHz) / baseline.ExpectedFrequencyHz * 100 + if mathAbs(metric.FrequencyDriftPercent) > baseline.FrequencyTolerancePercent { + issues = append(issues, fmt.Sprintf("frequency drift %.1f%%", metric.FrequencyDriftPercent)) + } + } + if metric.PayloadBytesLast > 0 && !containsInt(baseline.PayloadLengths, int(metric.PayloadBytesLast)) { + issues = append(issues, fmt.Sprintf("unexpected payload length %d B", metric.PayloadBytesLast)) + } + if baseline.DecodeStatus != "" && metric.DecodeStatus != baseline.DecodeStatus { + issues = append(issues, "decode status changed") + } + if baseline.Variant != "" && metric.Variant != baseline.Variant { + issues = append(issues, "decode variant changed") + } + if baseline.Transport != "" && metric.Transport != baseline.Transport { + issues = append(issues, "transport changed") + } + for destination := range metric.DestinationCounts { + if !containsUint8String(baseline.Destinations, destination) { + issues = append(issues, fmt.Sprintf("new destination %s", destination)) + } + } + for priority := range metric.PriorityCounts { + if !containsUint8String(baseline.Priorities, priority) { + issues = append(issues, fmt.Sprintf("new priority %s", priority)) + } + } + if baseline.DeviceNameHex != "" && metric.SourceAddress != baseline.SourceAddress { + issues = append(issues, fmt.Sprintf("source address changed from %d", baseline.SourceAddress)) + } + for _, field := range metric.Fields { + expected, ok := baseline.Fields[field.Field] + if !ok || field.LastNumeric == nil { + continue + } + if *field.LastNumeric < expected.Minimum || *field.LastNumeric > expected.Maximum { + issues = append(issues, fmt.Sprintf("%s outside approved range", field.Field)) + } + } + compareRawBaseline(metric, baseline, &issues) + sort.Strings(issues) + metric.BaselineIssues = issues + if len(issues) > 0 { + metric.BaselineStatus = "changed" + if metric.Status != "gap" && metric.Status != "anomaly" { + metric.Status = "changed" + } + } +} + +func compareRawBaseline(metric *SourcePGNMetric, baseline SourceTrafficBaseline, issues *[]string) { + if metric.Raw == nil || len(baseline.RawBytes) == 0 { + return + } + expected := make(map[int]BaselineRawByte, len(baseline.RawBytes)) + for _, rawByte := range baseline.RawBytes { + expected[rawByte.Offset] = rawByte + } + outside := make([]string, 0) + newChangeBits := make([]string, 0) + for _, observed := range metric.Raw.Bytes { + approved, ok := expected[observed.Offset] + if !ok || observed.Minimum < approved.Minimum || observed.Maximum > approved.Maximum { + outside = append(outside, strconv.Itoa(observed.Offset)) + } + observedMask, observedErr := strconv.ParseUint(observed.ChangedBitMaskHex, 16, 8) + approvedMask, approvedErr := strconv.ParseUint(approved.ChangedBitMaskHex, 16, 8) + if ok && observedErr == nil && approvedErr == nil && observedMask&^approvedMask != 0 { + newChangeBits = append(newChangeBits, strconv.Itoa(observed.Offset)) + } + } + if len(outside) > 0 { + *issues = append(*issues, "raw byte outside approved range at offset "+strings.Join(outside, ", ")) + } + if len(newChangeBits) > 0 { + *issues = append(*issues, "new raw bit changes at offset "+strings.Join(newChangeBits, ", ")) + } +} + +func baselinePeriodSeconds(frequencyHz float64) float64 { + if frequencyHz <= 0 { + return 0 + } + return 1 / frequencyHz +} + +func containsInt(values []int, want int) bool { + for _, value := range values { + if value == want { + return true + } + } + return false +} + +func containsUint8String(values []int, want string) bool { + for _, value := range values { + if fmt.Sprintf("%d", value) == want { + return true + } + } + return false +} diff --git a/internal/stats/source_metrics.go b/internal/stats/source_metrics.go new file mode 100644 index 0000000..9725f06 --- /dev/null +++ b/internal/stats/source_metrics.go @@ -0,0 +1,1029 @@ +package stats + +import ( + "encoding/json" + "fmt" + "math" + "sort" + "strconv" + "sync" + "time" + + "github.com/open-ships/beacon/internal/msg" +) + +const ( + sourceIntervalSamples = 64 + maxSourceFields = 128 + maxCategoryValues = 16 + anomalyZThreshold = 6.0 +) + +// FieldDistribution is the process-local distribution of one decoded field +// on a source/PGN/sender stream. Numeric fields expose descriptive statistics; +// category fields expose bounded value counts (overflow is counted in Other). +type FieldDistribution struct { + Field string `json:"field"` + Kind string `json:"kind"` + Unit string `json:"unit,omitempty"` + Samples int64 `json:"samples"` + Last string `json:"last"` + Minimum *float64 `json:"minimum,omitempty"` + Maximum *float64 `json:"maximum,omitempty"` + Mean *float64 `json:"mean,omitempty"` + StdDev *float64 `json:"stddev,omitempty"` + LastNumeric *float64 `json:"last_numeric,omitempty"` + LastChange *float64 `json:"last_change,omitempty"` + P05 *float64 `json:"p05,omitempty"` + P50 *float64 `json:"p50,omitempty"` + P95 *float64 `json:"p95,omitempty"` + P99 *float64 `json:"p99,omitempty"` + LastRateOfChange *float64 `json:"last_rate_of_change,omitempty"` + StuckSeconds float64 `json:"stuck_seconds,omitempty"` + PresentMessages int64 `json:"present_messages"` + MissingMessages int64 `json:"missing_messages"` + AvailabilityPercent float64 `json:"availability_percent"` + InvalidCount int64 `json:"invalid_count,omitempty"` + OutOfRangeCount int64 `json:"out_of_range_count,omitempty"` + NovelValueCount int64 `json:"novel_value_count,omitempty"` + CatalogMinimum *float64 `json:"catalog_minimum,omitempty"` + CatalogMaximum *float64 `json:"catalog_maximum,omitempty"` + Values map[string]int64 `json:"values,omitempty"` + Other int64 `json:"other,omitempty"` + Anomalous bool `json:"anomalous"` + AnomalyScore float64 `json:"anomaly_score,omitempty"` + AnomalyCount int64 `json:"anomaly_count,omitempty"` + LastAnomalyAt *time.Time `json:"last_anomaly_at,omitempty"` + AnomalyReason string `json:"anomaly_reason,omitempty"` +} + +// SourcePGNMetric describes one distinct stream on a configured source. The +// CAN source address is part of the identity so two devices sending the same +// PGN remain independently observable when one stops transmitting. +type SourcePGNMetric struct { + Observed bool `json:"observed"` + SourceID string `json:"source_id"` + PGN uint32 `json:"pgn"` + PGNName string `json:"pgn_name,omitempty"` + Variant string `json:"variant,omitempty"` + Transport string `json:"transport,omitempty"` + ManufacturerCode *uint16 `json:"manufacturer_code,omitempty"` + DecodeStatus string `json:"decode_status"` + DecodeStatuses map[string]int64 `json:"decode_statuses"` + DecodeComplete int64 `json:"decode_complete"` + DecodeIncomplete int64 `json:"decode_incomplete"` + DecodeFallback int64 `json:"decode_fallback"` + UnknownMessages int64 `json:"unknown_messages"` + MissingDecodedFields map[string]int64 `json:"missing_decoded_fields,omitempty"` + SourceAddress uint8 `json:"source_address"` + DeviceName *uint64 `json:"device_name,omitempty"` + DeviceNameHex string `json:"device_name_hex,omitempty"` + Messages int64 `json:"messages"` + FirstSeen time.Time `json:"first_seen"` + LastSeen time.Time `json:"last_seen"` + AgeSeconds float64 `json:"age_seconds"` + FrequencyHz float64 `json:"frequency_hz"` + ExpectedPeriodSeconds float64 `json:"expected_period_seconds"` + ShortestPeriodSeconds float64 `json:"shortest_period_seconds,omitempty"` + LongestPeriodSeconds float64 `json:"longest_period_seconds,omitempty"` + PeriodP95Seconds float64 `json:"period_p95_seconds,omitempty"` + PeriodP99Seconds float64 `json:"period_p99_seconds,omitempty"` + JitterMADSeconds float64 `json:"jitter_mad_seconds,omitempty"` + JitterPercent float64 `json:"jitter_percent,omitempty"` + BurstCount int64 `json:"burst_count"` + RecentMessagesPerSec float64 `json:"recent_messages_per_sec"` + RecentBytesPerSec float64 `json:"recent_bytes_per_sec"` + EstimatedBusLoadPercent float64 `json:"estimated_bus_load_percent"` + TrafficSharePercent float64 `json:"traffic_share_percent"` + Expected bool `json:"expected"` + BaselineStatus string `json:"baseline_status"` + BaselineIssues []string `json:"baseline_issues,omitempty"` + BaselineApprovedAt *time.Time `json:"baseline_approved_at,omitempty"` + BaselineFrequencyHz float64 `json:"baseline_frequency_hz,omitempty"` + BaselineTolerancePercent float64 `json:"baseline_tolerance_percent,omitempty"` + FrequencyDriftPercent float64 `json:"frequency_drift_percent,omitempty"` + PayloadBytesLast int64 `json:"payload_bytes_last"` + PayloadBytesMin int64 `json:"payload_bytes_min"` + PayloadBytesMax int64 `json:"payload_bytes_max"` + PayloadBytesMean float64 `json:"payload_bytes_mean"` + GapActive bool `json:"gap_active"` + GapRatio float64 `json:"gap_ratio,omitempty"` + GapCount int64 `json:"gap_count"` + LastGapAt *time.Time `json:"last_gap_at,omitempty"` + LongestGapSeconds float64 `json:"longest_gap_seconds,omitempty"` + AnomalyActive bool `json:"anomaly_active"` + RecentAnomaly bool `json:"recent_anomaly"` + AnomalyCount int64 `json:"anomaly_count"` + LastAnomalyAt *time.Time `json:"last_anomaly_at,omitempty"` + AnomalyScore float64 `json:"anomaly_score,omitempty"` + AnomalyField string `json:"anomaly_field,omitempty"` + AnomalyReason string `json:"anomaly_reason,omitempty"` + Status string `json:"status"` + DestinationCounts map[string]int64 `json:"destination_counts"` + PriorityCounts map[string]int64 `json:"priority_counts"` + IdentityChanges int64 `json:"identity_changes"` + Raw *RawPayloadDiagnostics `json:"raw,omitempty"` + Fields []FieldDistribution `json:"fields,omitempty"` +} + +type sourceStreamKey struct { + source string + pgn uint32 + address uint8 +} + +type sourceAddressKey struct { + source string + address uint8 +} + +type sourceDeviceKey struct { + source string + name uint64 +} + +type sourceEventState struct { + messages int64 + gapCount int64 + anomalyCount int64 + identityChanges int64 + decodeStatus string + payloadLengths int + novelValues int64 +} + +type runningDistribution struct { + count int64 + last float64 + min float64 + max float64 + mean float64 + m2 float64 +} + +func (d *runningDistribution) add(value float64) { + d.last = value + if d.count == 0 { + d.count = 1 + d.min = value + d.max = value + d.mean = value + return + } + d.count++ + if value < d.min { + d.min = value + } + if value > d.max { + d.max = value + } + delta := value - d.mean + d.mean += delta / float64(d.count) + d.m2 += delta * (value - d.mean) +} + +func (d *runningDistribution) stddev() float64 { + if d.count < 2 { + return 0 + } + return math.Sqrt(d.m2 / float64(d.count-1)) +} + +type sourceField struct { + kind string + unit string + numeric runningDistribution + changes runningDistribution + values map[string]int64 + lastCategory string + other int64 + anomalous bool + score float64 + anomalies int64 + lastAt time.Time + reason string + lowerBound *float64 + upperBound *float64 + numericSamples [sourceIntervalSamples]float64 + numericHead int + numericLen int + presentMessages int64 + missingMessages int64 + invalidCount int64 + outOfRangeCount int64 + novelValueCount int64 + lastSeen time.Time + lastChanged time.Time + lastRateOfChange float64 +} + +type sourceStream struct { + mu sync.Mutex + + key sourceStreamKey + pgnName string + variant string + transport string + manufacturerCode *uint16 + deviceName *uint64 + identityChanges int64 + messages int64 + firstSeen time.Time + lastSeen time.Time + intervals [sourceIntervalSamples]time.Duration + intHead int + intLen int + payload runningDistribution + fields map[string]*sourceField + destinations map[uint8]int64 + priorities map[uint8]int64 + decodeStatuses map[string]int64 + lastDecodeStatus string + decodeComplete int64 + decodeIncomplete int64 + decodeFallback int64 + unknownMessages int64 + missingDecodedFields map[string]int64 + burstCount int64 + wire sourceWireStats + rate sourceRateStats + + gapCount int64 + lastGap time.Time + longestGap time.Duration + + anomalyActive bool + anomalyCount int64 + lastAnomaly time.Time + currentAnomalyScore float64 + lastAnomalyScore float64 + lastAnomalyField string + lastAnomalyReason string +} + +func (s *sourceStream) addInterval(value time.Duration) { + s.intervals[s.intHead] = value + s.intHead = (s.intHead + 1) % sourceIntervalSamples + if s.intLen < sourceIntervalSamples { + s.intLen++ + } +} + +func (s *sourceStream) sourceEventState() sourceEventState { + s.mu.Lock() + defer s.mu.Unlock() + state := sourceEventState{ + messages: s.messages, gapCount: s.gapCount, anomalyCount: s.anomalyCount, + identityChanges: s.identityChanges, decodeStatus: s.lastDecodeStatus, + payloadLengths: len(s.wire.lengths), + } + for _, field := range s.fields { + state.novelValues += field.novelValueCount + } + return state +} + +func (s *sourceStream) intervalValues() []time.Duration { + out := make([]time.Duration, s.intLen) + start := (s.intHead - s.intLen + sourceIntervalSamples) % sourceIntervalSamples + for i := range out { + out[i] = s.intervals[(start+i)%sourceIntervalSamples] + } + return out +} + +func medianInterval(values []time.Duration) time.Duration { + if len(values) == 0 { + return 0 + } + copyValues := append([]time.Duration(nil), values...) + sort.Slice(copyValues, func(i, j int) bool { return copyValues[i] < copyValues[j] }) + mid := len(copyValues) / 2 + if len(copyValues)%2 == 1 { + return copyValues[mid] + } + return copyValues[mid-1]/2 + copyValues[mid]/2 +} + +func durationPercentile(values []time.Duration, percentile float64) time.Duration { + if len(values) == 0 { + return 0 + } + copyValues := append([]time.Duration(nil), values...) + sort.Slice(copyValues, func(i, j int) bool { return copyValues[i] < copyValues[j] }) + index := int(math.Ceil(percentile*float64(len(copyValues)))) - 1 + if index < 0 { + index = 0 + } + if index >= len(copyValues) { + index = len(copyValues) - 1 + } + return copyValues[index] +} + +func intervalMAD(values []time.Duration, median time.Duration) time.Duration { + if len(values) == 0 { + return 0 + } + deviations := make([]time.Duration, 0, len(values)) + for _, value := range values { + delta := value - median + if delta < 0 { + delta = -delta + } + deviations = append(deviations, delta) + } + return medianInterval(deviations) +} + +func gapThreshold(expected time.Duration) time.Duration { + if expected <= 0 { + return 0 + } + threshold := 3 * expected + if floor := expected + 500*time.Millisecond; threshold < floor { + threshold = floor + } + return threshold +} + +func sourcePayloadSize(e *msg.Envelope) int { + if len(e.Raw) > 0 { + return len(e.Raw) + } + return len(e.Payload) +} + +func (s *sourceStream) record(now time.Time, e *msg.Envelope) { + s.mu.Lock() + defer s.mu.Unlock() + + if s.messages == 0 { + s.firstSeen = now + } + if !s.lastSeen.IsZero() && now.After(s.lastSeen) { + interval := now.Sub(s.lastSeen) + expectedBefore := medianInterval(s.intervalValues()) + if s.intLen >= 3 && interval > gapThreshold(expectedBefore) { + s.gapCount++ + s.lastGap = now + if interval > s.longestGap { + s.longestGap = interval + } + } + if s.intLen >= 3 && expectedBefore > 0 && interval < expectedBefore/3 { + s.burstCount++ + } + s.addInterval(interval) + } + s.messages++ + s.lastSeen = now + if e.PGNName != "" { + s.pgnName = e.PGNName + } + if e.Variant != "" { + s.variant = e.Variant + } + if e.Transport != "" { + s.transport = e.Transport + } + if e.ManufacturerCode != nil { + manufacturer := *e.ManufacturerCode + s.manufacturerCode = &manufacturer + } + if e.DeviceName != nil { + name := *e.DeviceName + if s.deviceName != nil && *s.deviceName != name { + s.identityChanges++ + } + s.deviceName = &name + } + payloadSize := sourcePayloadSize(e) + s.payload.add(float64(payloadSize)) + s.rate.record(now, payloadSize, e.Transport == "Fast" || e.Transport == "fast") + s.wire.record(now, e.Raw) + if s.destinations == nil { + s.destinations = make(map[uint8]int64) + s.priorities = make(map[uint8]int64) + s.decodeStatuses = make(map[string]int64) + s.missingDecodedFields = make(map[string]int64) + } + s.destinations[e.Dest]++ + s.priorities[e.Priority]++ + decodeStatus := normalizeDecodeStatus(e.Decode.Status) + s.decodeStatuses[decodeStatus]++ + s.lastDecodeStatus = decodeStatus + if e.Decode.Complete { + s.decodeComplete++ + } else { + s.decodeIncomplete++ + } + if e.Decode.Fallback { + s.decodeFallback++ + } + if decodeStatus == "unknown" { + s.unknownMessages++ + } + for _, missing := range e.Decode.Missing { + s.missingDecodedFields[missing]++ + } + s.anomalyActive = false + s.currentAnomalyScore = 0 + for _, field := range s.fields { + field.anomalous = false + } + + observedFields := make(map[string]bool) + physicalNames := make([]string, 0, len(e.Physical)) + for name := range e.Physical { + physicalNames = append(physicalNames, name) + } + sort.Strings(physicalNames) + for _, name := range physicalNames { + value := e.Physical[name] + field := s.field(name, "number", value.Unit) + if field == nil { + continue + } + field.lowerBound = cloneFloat(value.Minimum) + field.upperBound = cloneFloat(value.Maximum) + observedFields[name] = true + if field.recordNumeric(now, value.Value, true) { + s.noteAnomaly(now, name, field.score, field.reason) + } + } + + values := map[string]any{} + if len(e.Payload) > 0 && string(e.Payload) != "null" { + _ = json.Unmarshal(e.Payload, &values) + } + flat := make(map[string]any) + flattenScalars("", values, flat) + fieldNames := make([]string, 0, len(flat)) + for name := range flat { + fieldNames = append(fieldNames, name) + } + sort.Strings(fieldNames) + for _, name := range fieldNames { + if _, described := e.Physical[name]; described { + continue + } + switch value := flat[name].(type) { + case float64: + field := s.field(name, "number", "") + if field != nil { + observedFields[name] = true + if field.recordNumeric(now, value, true) { + s.noteAnomaly(now, name, field.score, field.reason) + } + } + case string: + if field := s.field(name, "category", ""); field != nil { + observedFields[name] = true + field.recordCategory(now, value) + } + case bool: + if field := s.field(name, "category", ""); field != nil { + observedFields[name] = true + field.recordCategory(now, strconv.FormatBool(value)) + } + } + } + for name, field := range s.fields { + if !observedFields[name] { + field.missingMessages++ + } + } +} + +func normalizeDecodeStatus(status string) string { + switch status { + case "decoded", "unknown", "partial", "fallback", "error": + return status + case "": + return "unspecified" + default: + return "other" + } +} + +func cloneFloat(value *float64) *float64 { + if value == nil { + return nil + } + out := *value + return &out +} + +func flattenScalars(prefix string, value any, out map[string]any) { + switch typed := value.(type) { + case map[string]any: + keys := make([]string, 0, len(typed)) + for key := range typed { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + name := key + if prefix != "" { + name = prefix + "." + key + } + flattenScalars(name, typed[key], out) + } + case []any: + for i, item := range typed { + flattenScalars(prefix+"["+strconv.Itoa(i)+"]", item, out) + } + case float64, string, bool: + if prefix != "" { + out[prefix] = typed + } + } +} + +func (s *sourceStream) field(name, kind, unit string) *sourceField { + if field := s.fields[name]; field != nil { + if unit != "" { + field.unit = unit + } + return field + } + if len(s.fields) >= maxSourceFields { + return nil + } + field := &sourceField{kind: kind, unit: unit} + if kind == "category" { + field.values = make(map[string]int64) + } + s.fields[name] = field + return field +} + +func (f *sourceField) recordNumeric(now time.Time, value float64, detect bool) bool { + f.anomalous = false + f.score = 0 + f.reason = "" + f.presentMessages++ + if math.IsNaN(value) || math.IsInf(value, 0) { + f.invalidCount++ + f.lastSeen = now + return false + } + if detect { + f.detectAnomaly(value) + } + if f.numeric.count > 0 { + change := math.Abs(value - f.numeric.last) + f.changes.add(change) + if now.After(f.lastSeen) { + f.lastRateOfChange = (value - f.numeric.last) / now.Sub(f.lastSeen).Seconds() + } + if change > math.Max(math.Abs(f.numeric.last)*1e-9, 1e-12) { + f.lastChanged = now + } + } else { + f.lastChanged = now + } + f.numeric.add(value) + f.numericSamples[f.numericHead] = value + f.numericHead = (f.numericHead + 1) % sourceIntervalSamples + if f.numericLen < sourceIntervalSamples { + f.numericLen++ + } + f.lastSeen = now + if f.anomalous { + f.anomalies++ + f.lastAt = now + if f.reason == "below catalog minimum" || f.reason == "above catalog maximum" { + f.outOfRangeCount++ + } + } + return f.anomalous +} + +func (f *sourceField) detectAnomaly(value float64) { + if f.lowerBound != nil && value < *f.lowerBound { + f.anomalous = true + f.reason = "below catalog minimum" + f.score = 1e9 + return + } + if f.upperBound != nil && value > *f.upperBound { + f.anomalous = true + f.reason = "above catalog maximum" + f.score = 1e9 + return + } + if f.numeric.count < 5 { + return + } + sd := f.numeric.stddev() + if sd > 1e-12 { + f.score = math.Abs(value-f.numeric.mean) / sd + if f.score >= anomalyZThreshold { + f.anomalous = true + f.reason = fmt.Sprintf("value changed %.1f standard deviations from baseline", f.score) + return + } + } + change := math.Abs(value - f.numeric.last) + if f.changes.count >= 5 { + changeSD := f.changes.stddev() + if changeSD > 1e-12 { + changeScore := (change - f.changes.mean) / changeSD + if changeScore > f.score { + f.score = changeScore + } + if changeScore >= anomalyZThreshold { + f.anomalous = true + f.reason = fmt.Sprintf("step change was %.1f standard deviations above normal", changeScore) + return + } + } + } + if sd <= 1e-12 && change > math.Max(math.Abs(f.numeric.mean)*0.5, 1) { + f.anomalous = true + f.score = anomalyZThreshold + f.reason = "large change from a stable baseline" + } +} + +func (f *sourceField) recordCategory(now time.Time, value string) { + if len(value) > 64 { + value = value[:64] + "…" + } + _, known := f.values[value] + if known || len(f.values) < maxCategoryValues { + f.values[value]++ + } else { + f.other++ + } + if f.presentMessages > 0 && !known { + f.novelValueCount++ + } + if f.presentMessages == 0 || f.lastCategory != value { + f.lastChanged = now + } + f.presentMessages++ + f.lastCategory = value + f.lastSeen = now +} + +func (s *sourceStream) noteAnomaly(now time.Time, field string, score float64, reason string) { + if !s.anomalyActive { + s.anomalyCount++ + } + s.anomalyActive = true + s.lastAnomaly = now + if score >= s.currentAnomalyScore { + s.currentAnomalyScore = score + s.lastAnomalyScore = score + s.lastAnomalyField = field + s.lastAnomalyReason = reason + } +} + +func (s *sourceStream) snapshot(now time.Time) SourcePGNMetric { + s.mu.Lock() + defer s.mu.Unlock() + + intervals := s.intervalValues() + expected := medianInterval(intervals) + periodP95 := durationPercentile(intervals, 0.95) + periodP99 := durationPercentile(intervals, 0.99) + jitterMAD := intervalMAD(intervals, expected) + var shortest, longest time.Duration + for _, interval := range intervals { + if shortest == 0 || interval < shortest { + shortest = interval + } + if interval > longest { + longest = interval + } + } + age := now.Sub(s.lastSeen) + if age < 0 { + age = 0 + } + gapActive := len(intervals) >= 3 && age > gapThreshold(expected) + gapRatio := 0.0 + if expected > 0 { + gapRatio = float64(age) / float64(expected) + } + recentWindow := time.Minute + if candidate := 3 * expected; candidate > recentWindow { + recentWindow = candidate + } + if recentWindow > 10*time.Minute { + recentWindow = 10 * time.Minute + } + recentAnomaly := !s.lastAnomaly.IsZero() && now.Sub(s.lastAnomaly) <= recentWindow + recentMessages, recentBytes, busLoad := s.rate.snapshot(now) + status := "active" + if len(intervals) < 3 { + status = "warming" + } + if recentAnomaly { + status = "anomaly" + } + if gapActive { + status = "gap" + } + + out := SourcePGNMetric{ + Observed: true, SourceID: s.key.source, PGN: s.key.pgn, PGNName: s.pgnName, + Variant: s.variant, Transport: s.transport, DecodeStatus: s.lastDecodeStatus, + DecodeStatuses: cloneStringTotals(s.decodeStatuses), DecodeComplete: s.decodeComplete, + DecodeIncomplete: s.decodeIncomplete, DecodeFallback: s.decodeFallback, + UnknownMessages: s.unknownMessages, MissingDecodedFields: cloneStringTotals(s.missingDecodedFields), + SourceAddress: s.key.address, Messages: s.messages, + FirstSeen: s.firstSeen, LastSeen: s.lastSeen, AgeSeconds: age.Seconds(), + ExpectedPeriodSeconds: expected.Seconds(), ShortestPeriodSeconds: shortest.Seconds(), + LongestPeriodSeconds: longest.Seconds(), PeriodP95Seconds: periodP95.Seconds(), + PeriodP99Seconds: periodP99.Seconds(), JitterMADSeconds: jitterMAD.Seconds(), + BurstCount: s.burstCount, RecentMessagesPerSec: recentMessages, + RecentBytesPerSec: recentBytes, EstimatedBusLoadPercent: busLoad, + GapActive: gapActive, GapRatio: gapRatio, + GapCount: s.gapCount, LastGapAt: timePtr(s.lastGap), LongestGapSeconds: s.longestGap.Seconds(), + AnomalyActive: s.anomalyActive, RecentAnomaly: recentAnomaly, + AnomalyCount: s.anomalyCount, LastAnomalyAt: timePtr(s.lastAnomaly), + AnomalyScore: s.lastAnomalyScore, AnomalyField: s.lastAnomalyField, AnomalyReason: s.lastAnomalyReason, + Status: status, DestinationCounts: cloneUint8Totals(s.destinations), + PriorityCounts: cloneUint8Totals(s.priorities), IdentityChanges: s.identityChanges, + Raw: s.wire.snapshot(now), + } + if expected > 0 { + out.JitterPercent = float64(jitterMAD) / float64(expected) * 100 + } + if expected > 0 { + out.FrequencyHz = 1 / expected.Seconds() + } + if s.payload.count > 0 { + out.PayloadBytesLast = int64(s.payload.last) + out.PayloadBytesMin = int64(s.payload.min) + out.PayloadBytesMax = int64(s.payload.max) + out.PayloadBytesMean = s.payload.mean + } + if s.deviceName != nil { + name := *s.deviceName + out.DeviceName = &name + out.DeviceNameHex = fmt.Sprintf("%016x", name) + } + if s.manufacturerCode != nil { + manufacturer := *s.manufacturerCode + out.ManufacturerCode = &manufacturer + } + + fieldNames := make([]string, 0, len(s.fields)) + for name := range s.fields { + fieldNames = append(fieldNames, name) + } + sort.Strings(fieldNames) + for _, name := range fieldNames { + out.Fields = append(out.Fields, fieldSnapshot(name, s.fields[name], now, s.messages)) + } + return out +} + +func fieldSnapshot(name string, field *sourceField, now time.Time, messages int64) FieldDistribution { + out := FieldDistribution{ + Field: name, Kind: field.kind, Unit: field.unit, + Anomalous: field.anomalous, AnomalyScore: field.score, + AnomalyCount: field.anomalies, LastAnomalyAt: timePtr(field.lastAt), + AnomalyReason: field.reason, PresentMessages: field.presentMessages, + MissingMessages: field.missingMessages, InvalidCount: field.invalidCount, + OutOfRangeCount: field.outOfRangeCount, NovelValueCount: field.novelValueCount, + CatalogMinimum: cloneFloat(field.lowerBound), CatalogMaximum: cloneFloat(field.upperBound), + } + if messages > 0 { + out.AvailabilityPercent = float64(field.presentMessages) / float64(messages) * 100 + } + if !field.lastChanged.IsZero() { + out.StuckSeconds = max(0, now.Sub(field.lastChanged).Seconds()) + } + if field.kind == "number" && field.numeric.count > 0 { + last, minValue, maxValue := field.numeric.last, field.numeric.min, field.numeric.max + mean, stddev := field.numeric.mean, field.numeric.stddev() + out.Samples = field.numeric.count + out.Last = formatMetricNumber(last) + out.LastNumeric = &last + out.Minimum = &minValue + out.Maximum = &maxValue + out.Mean = &mean + out.StdDev = &stddev + values := make([]float64, field.numericLen) + start := (field.numericHead - field.numericLen + sourceIntervalSamples) % sourceIntervalSamples + for i := range values { + values[i] = field.numericSamples[(start+i)%sourceIntervalSamples] + } + p05, p50 := floatPercentile(values, 0.05), floatPercentile(values, 0.50) + p95, p99 := floatPercentile(values, 0.95), floatPercentile(values, 0.99) + out.P05, out.P50, out.P95, out.P99 = &p05, &p50, &p95, &p99 + if field.changes.count > 0 { + change := field.changes.last + out.LastChange = &change + rate := field.lastRateOfChange + out.LastRateOfChange = &rate + } + return out + } + out.Values = make(map[string]int64, len(field.values)) + for value, count := range field.values { + out.Values[value] = count + out.Samples += count + } + out.Other = field.other + out.Samples += field.other + out.Last = field.lastCategory + return out +} + +func cloneStringTotals(in map[string]int64) map[string]int64 { + out := make(map[string]int64, len(in)) + for key, value := range in { + out[key] = value + } + return out +} + +func cloneUint8Totals(in map[uint8]int64) map[string]int64 { + out := make(map[string]int64, len(in)) + for key, value := range in { + out[strconv.Itoa(int(key))] = value + } + return out +} + +func formatMetricNumber(value float64) string { + return strconv.FormatFloat(value, 'g', 6, 64) +} + +// SourcePGNMetrics returns every observed PGN/sender stream for one source. +// Results are sorted with active problems first, then by PGN and address. +func (r *Registry) SourcePGNMetrics(source string) []SourcePGNMetric { + if r == nil { + return nil + } + r.mu.Lock() + streams := make([]*sourceStream, 0) + for key, stream := range r.sourceStreams { + if key.source == source { + streams = append(streams, stream) + } + } + r.mu.Unlock() + now := r.now() + out := make([]SourcePGNMetric, 0, len(streams)) + for _, stream := range streams { + out = append(out, stream.snapshot(now)) + } + out = applySourceBaselines(out, r.sourceBaselinesFor(source), now, r.startedAt) + applyTrafficShares(out) + sortSourcePGNMetrics(out) + return out +} + +// AllSourcePGNMetrics returns source metrics keyed by configured source id. +func (r *Registry) AllSourcePGNMetrics() map[string][]SourcePGNMetric { + out := map[string][]SourcePGNMetric{} + if r == nil { + return out + } + r.mu.Lock() + streams := make([]*sourceStream, 0, len(r.sourceStreams)) + for _, stream := range r.sourceStreams { + streams = append(streams, stream) + } + r.mu.Unlock() + now := r.now() + for _, stream := range streams { + metric := stream.snapshot(now) + out[metric.SourceID] = append(out[metric.SourceID], metric) + } + for _, baseline := range r.SourceTrafficBaselines("") { + if _, ok := out[baseline.SourceID]; !ok { + out[baseline.SourceID] = nil + } + } + for source := range out { + out[source] = applySourceBaselines(out[source], r.sourceBaselinesFor(source), now, r.startedAt) + applyTrafficShares(out[source]) + sortSourcePGNMetrics(out[source]) + } + return out +} + +func applyTrafficShares(metrics []SourcePGNMetric) { + total := 0.0 + for _, metric := range metrics { + total += metric.RecentBytesPerSec + } + if total <= 0 { + return + } + for i := range metrics { + metrics[i].TrafficSharePercent = metrics[i].RecentBytesPerSec / total * 100 + } +} + +func sortSourcePGNMetrics(metrics []SourcePGNMetric) { + priority := map[string]int{"missing": 0, "gap": 1, "anomaly": 2, "changed": 3, "awaiting": 4, "warming": 5, "active": 6} + sort.Slice(metrics, func(i, j int) bool { + left, right := priority[metrics[i].Status], priority[metrics[j].Status] + if left != right { + return left < right + } + if metrics[i].PGN != metrics[j].PGN { + return metrics[i].PGN < metrics[j].PGN + } + return metrics[i].SourceAddress < metrics[j].SourceAddress + }) +} + +func (r *Registry) getSourceStream(source string, e *msg.Envelope) (*sourceStream, bool) { + key := sourceStreamKey{source: source, pgn: e.PGN, address: e.Source} + r.mu.Lock() + defer r.mu.Unlock() + stream := r.sourceStreams[key] + created := false + if stream == nil { + stream = &sourceStream{key: key, fields: make(map[string]*sourceField)} + r.sourceStreams[key] = stream + created = true + } + return stream, created +} + +func (r *Registry) sourceLifecycleEvents(now time.Time, source string, e *msg.Envelope, created bool, before, after sourceEventState) []SourceMetricEvent { + base := SourceMetricEvent{Time: now.UTC(), SourceID: source, PGN: e.PGN, + SourceAddress: e.Source, DeviceNameHex: e.DeviceNameHex} + if base.DeviceNameHex == "" && e.DeviceName != nil { + base.DeviceNameHex = fmt.Sprintf("%016x", *e.DeviceName) + } + var events []SourceMetricEvent + add := func(kind, severity, summary string, details map[string]string) { + event := base + event.Kind, event.Severity, event.Summary, event.Details = kind, severity, summary, details + events = append(events, event) + } + if created { + kind, severity := "new_stream", "info" + if normalizeDecodeStatus(e.Decode.Status) == "unknown" { + kind, severity = "unknown_stream", "warning" + } + add(kind, severity, fmt.Sprintf("First observed PGN %d from address %d", e.PGN, e.Source), nil) + } + if after.gapCount > before.gapCount { + add("gap_recovered", "warning", fmt.Sprintf("PGN %d resumed after a sending gap", e.PGN), nil) + } + if after.anomalyCount > before.anomalyCount { + add("value_anomaly", "warning", fmt.Sprintf("PGN %d contained an anomalous decoded value", e.PGN), nil) + } + if before.messages > 0 && after.decodeStatus != before.decodeStatus { + add("decode_status_changed", "warning", fmt.Sprintf("PGN %d decode status changed from %s to %s", e.PGN, before.decodeStatus, after.decodeStatus), + map[string]string{"before": before.decodeStatus, "after": after.decodeStatus}) + } + if before.messages > 0 && after.payloadLengths > before.payloadLengths { + add("payload_length_changed", "warning", fmt.Sprintf("PGN %d introduced a new payload length", e.PGN), + map[string]string{"length_bytes": strconv.Itoa(sourcePayloadSize(e))}) + } + if after.identityChanges > before.identityChanges { + add("device_name_changed", "error", fmt.Sprintf("Address %d changed Device NAME while sending PGN %d", e.Source, e.PGN), nil) + } + if after.novelValues > before.novelValues { + add("new_categorical_value", "info", fmt.Sprintf("PGN %d introduced a new decoded categorical value", e.PGN), nil) + } + return events +} + +func (r *Registry) sourceIdentityEvents(now time.Time, source string, e *msg.Envelope) []SourceMetricEvent { + if e.DeviceName == nil { + return nil + } + name := *e.DeviceName + addressKey := sourceAddressKey{source: source, address: e.Source} + deviceKey := sourceDeviceKey{source: source, name: name} + r.mu.Lock() + previousName, addressKnown := r.sourceAddressNames[addressKey] + previousAddress, deviceKnown := r.sourceDeviceAddresses[deviceKey] + r.sourceAddressNames[addressKey] = name + r.sourceDeviceAddresses[deviceKey] = e.Source + r.mu.Unlock() + base := SourceMetricEvent{Time: now.UTC(), SourceID: source, PGN: e.PGN, + SourceAddress: e.Source, DeviceNameHex: fmt.Sprintf("%016x", name)} + var events []SourceMetricEvent + if addressKnown && previousName != name { + event := base + event.Kind, event.Severity = "address_conflict", "error" + event.Summary = fmt.Sprintf("Source address %d changed from Device NAME %016x to %016x", e.Source, previousName, name) + events = append(events, event) + } + if deviceKnown && previousAddress != e.Source { + event := base + event.Kind, event.Severity = "source_address_changed", "warning" + event.Summary = fmt.Sprintf("Device NAME %016x moved from source address %d to %d", name, previousAddress, e.Source) + event.Details = map[string]string{"before": strconv.Itoa(int(previousAddress)), "after": strconv.Itoa(int(e.Source))} + events = append(events, event) + } + return events +} diff --git a/internal/stats/source_metrics_test.go b/internal/stats/source_metrics_test.go new file mode 100644 index 0000000..e78c8f6 --- /dev/null +++ b/internal/stats/source_metrics_test.go @@ -0,0 +1,264 @@ +package stats + +import ( + "encoding/json" + "math" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/open-ships/beacon/internal/msg" + "github.com/open-ships/beacon/internal/n2kcatalog" + "github.com/open-ships/beacon/internal/store" +) + +func TestSourcePGNMetricsTrackSenderFrequencyPayloadAndValues(t *testing.T) { + now := time.Unix(1_700_000_000, 0).UTC() + reg := newRegistryAt(func() time.Time { return now }) + deviceName := uint64(0x1122334455667788) + + for i := 0; i < 6; i++ { + reg.RecordSource("can0", &msg.Envelope{ + PGN: 127250, PGNName: "Vessel Heading", Source: 12, DeviceName: &deviceName, + Raw: []byte{1, 2, 3, 4, 5, 6, 7, 8}, + Payload: json.RawMessage(`{"heading":10,"mode":"magnetic"}`), + Physical: map[string]n2kcatalog.PhysicalField{ + "heading": {Value: 10, Unit: "rad"}, + }, + }) + now = now.Add(time.Second) + } + reg.RecordSource("can0", &msg.Envelope{PGN: 127250, Source: 44, Raw: []byte{1}}) + + metrics := reg.SourcePGNMetrics("can0") + if len(metrics) != 2 { + t.Fatalf("streams = %d, want two sender-specific rows: %+v", len(metrics), metrics) + } + var stream SourcePGNMetric + for _, metric := range metrics { + if metric.SourceAddress == 12 { + stream = metric + } + } + if stream.Messages != 6 || stream.PGN != 127250 || stream.PGNName != "Vessel Heading" { + t.Fatalf("stream identity/totals = %+v", stream) + } + if stream.DeviceNameHex != "1122334455667788" { + t.Fatalf("device name = %q", stream.DeviceNameHex) + } + if math.Abs(stream.FrequencyHz-1) > 0.001 || math.Abs(stream.ExpectedPeriodSeconds-1) > 0.001 { + t.Fatalf("frequency = %v Hz period = %v s, want 1", stream.FrequencyHz, stream.ExpectedPeriodSeconds) + } + if stream.PayloadBytesLast != 8 || stream.PayloadBytesMin != 8 || stream.PayloadBytesMax != 8 || stream.PayloadBytesMean != 8 { + t.Fatalf("payload distribution = %+v", stream) + } + heading := findField(t, stream.Fields, "heading") + if heading.Kind != "number" || heading.Unit != "rad" || heading.Samples != 6 || heading.Mean == nil || *heading.Mean != 10 { + t.Fatalf("heading distribution = %+v", heading) + } + mode := findField(t, stream.Fields, "mode") + if mode.Kind != "category" || mode.Values["magnetic"] != 6 || mode.Last != "magnetic" { + t.Fatalf("mode distribution = %+v", mode) + } +} + +func TestSourcePGNMetricsExposeWireDecodeAddressingAndFieldQuality(t *testing.T) { + now := time.Unix(1_700_000_000, 0).UTC() + reg := newRegistryAt(func() time.Time { return now }) + minimum, maximum := 0.0, 100.0 + record := func(raw []byte, payload string, physical map[string]n2kcatalog.PhysicalField) { + reg.RecordSource("can0", &msg.Envelope{ + PGN: 130999, Source: 9, Dest: 255, Priority: 3, Raw: raw, + Payload: json.RawMessage(payload), Physical: physical, + Decode: msg.DecodeInfo{Status: "unknown"}, + }) + now = now.Add(time.Second) + } + record([]byte{0, 10}, `{"temperature":10}`, map[string]n2kcatalog.PhysicalField{ + "temperature": {Value: 10, Unit: "K", Minimum: &minimum, Maximum: &maximum}, + }) + record([]byte{0, 10}, `{}`, nil) + record([]byte{1, 14}, `{"temperature":1000}`, map[string]n2kcatalog.PhysicalField{ + "temperature": {Value: 1000, Unit: "K", Minimum: &minimum, Maximum: &maximum}, + }) + + stream := reg.SourcePGNMetrics("can0")[0] + if stream.DecodeStatus != "unknown" || stream.UnknownMessages != 3 || stream.DecodeStatuses["unknown"] != 3 { + t.Fatalf("decode diagnostics = %+v", stream) + } + if stream.DestinationCounts["255"] != 3 || stream.PriorityCounts["3"] != 3 { + t.Fatalf("addressing diagnostics = dest %+v priority %+v", stream.DestinationCounts, stream.PriorityCounts) + } + if stream.Raw == nil || stream.Raw.DistinctPayloads != 2 || stream.Raw.LengthCounts["2"] != 3 || len(stream.Raw.Samples) != 2 { + t.Fatalf("raw payload diagnostics = %+v", stream.Raw) + } + if len(stream.Raw.Bytes) != 2 || math.Abs(stream.Raw.Bytes[0].ChangedShare-0.5) > 0.001 || stream.Raw.HammingDistanceP95 != 2 { + t.Fatalf("raw byte distributions = %+v", stream.Raw) + } + field := findField(t, stream.Fields, "temperature") + if field.PresentMessages != 2 || field.MissingMessages != 1 || math.Abs(field.AvailabilityPercent-66.6667) > 0.01 { + t.Fatalf("field availability = %+v", field) + } + if field.OutOfRangeCount != 1 || !field.Anomalous || field.LastRateOfChange == nil { + t.Fatalf("field quality = %+v", field) + } +} + +func TestSourceTrafficBaselineAndEventsSurviveRestart(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "beacon.db") + now := time.Unix(1_700_000_000, 0).UTC() + st, err := store.Open(dbPath) + if err != nil { + t.Fatal(err) + } + reg := newRegistryAt(func() time.Time { return now }) + if err := reg.AttachSourceMetricPersistence(t.Context(), st.DB()); err != nil { + t.Fatal(err) + } + for i := 0; i < 5; i++ { + reg.RecordSource("can0", &msg.Envelope{ + PGN: 127250, PGNName: "Vessel Heading", Source: 12, Dest: 255, Priority: 2, + Raw: []byte{1, 2, 3, 4}, Decode: msg.DecodeInfo{Status: "decoded", Complete: true}, + }) + now = now.Add(time.Second) + } + baselines, err := reg.CommitSourceTrafficBaseline(t.Context(), "can0") + if err != nil { + t.Fatal(err) + } + if len(baselines) != 1 || math.Abs(baselines[0].ExpectedFrequencyHz-1) > 0.001 { + t.Fatalf("committed baselines = %+v", baselines) + } + reg.RecordSource("can0", &msg.Envelope{ + PGN: 127250, PGNName: "Vessel Heading", Source: 12, Dest: 255, Priority: 2, + Raw: []byte{1, 2, 3, 255}, Decode: msg.DecodeInfo{Status: "decoded", Complete: true}, + }) + changed := reg.SourcePGNMetrics("can0")[0] + if changed.BaselineStatus != "changed" || changed.Status != "changed" || + !strings.Contains(strings.Join(changed.BaselineIssues, " "), "raw byte") { + t.Fatalf("raw baseline change = %+v", changed) + } + if err := reg.CloseSourceMetricPersistence(t.Context()); err != nil { + t.Fatal(err) + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + + st, err = store.Open(dbPath) + if err != nil { + t.Fatal(err) + } + defer func() { _ = st.Close() }() + restarted := newRegistryAt(func() time.Time { return now }) + if err := restarted.AttachSourceMetricPersistence(t.Context(), st.DB()); err != nil { + t.Fatal(err) + } + defer func() { _ = restarted.CloseSourceMetricPersistence(t.Context()) }() + if loaded := restarted.SourceTrafficBaselines("can0"); len(loaded) != 1 || loaded[0].PGN != 127250 { + t.Fatalf("loaded baselines = %+v", loaded) + } + if events := restarted.SourceMetricEvents("can0", 20); len(events) < 2 { + t.Fatalf("loaded events = %+v, want stream and baseline events", events) + } + now = now.Add(11 * time.Second) + metrics := restarted.SourcePGNMetrics("can0") + if len(metrics) != 1 || metrics[0].Observed || metrics[0].Status != "missing" || !metrics[0].GapActive { + t.Fatalf("missing expected stream = %+v", metrics) + } +} + +func TestSourcePGNMetricsDetectCurrentAndRecoveredGaps(t *testing.T) { + now := time.Unix(1_700_000_000, 0).UTC() + reg := newRegistryAt(func() time.Time { return now }) + envelope := &msg.Envelope{PGN: 128259, Source: 7, Raw: []byte{1, 2}} + for i := 0; i < 5; i++ { + reg.RecordSource("can0", envelope) + now = now.Add(time.Second) + } + + now = now.Add(3100 * time.Millisecond) + stream := reg.SourcePGNMetrics("can0")[0] + if !stream.GapActive || stream.Status != "gap" || stream.GapRatio < 4 { + t.Fatalf("active gap not detected: %+v", stream) + } + + reg.RecordSource("can0", envelope) + stream = reg.SourcePGNMetrics("can0")[0] + if stream.GapActive || stream.GapCount != 1 || stream.LastGapAt == nil || stream.LongestGapSeconds < 4 { + t.Fatalf("recovered gap not retained: %+v", stream) + } +} + +func TestSourcePGNMetricsFlagLargePhysicalValueChange(t *testing.T) { + now := time.Unix(1_700_000_000, 0).UTC() + reg := newRegistryAt(func() time.Time { return now }) + record := func(value float64) { + reg.RecordSource("can0", &msg.Envelope{ + PGN: 130312, Source: 9, Payload: json.RawMessage(`{"temperature":10}`), + Physical: map[string]n2kcatalog.PhysicalField{"temperature": {Value: value, Unit: "K"}}, + }) + now = now.Add(time.Second) + } + for i := 0; i < 6; i++ { + record(10) + } + record(100) + + stream := reg.SourcePGNMetrics("can0")[0] + if !stream.AnomalyActive || !stream.RecentAnomaly || stream.Status != "anomaly" || stream.AnomalyCount != 1 { + t.Fatalf("stream anomaly = %+v", stream) + } + if stream.AnomalyField != "temperature" || stream.AnomalyReason == "" { + t.Fatalf("anomaly explanation = %+v", stream) + } + field := findField(t, stream.Fields, "temperature") + if !field.Anomalous || field.AnomalyCount != 1 || field.Maximum == nil || *field.Maximum != 100 { + t.Fatalf("field anomaly = %+v", field) + } + + record(10) + stream = reg.SourcePGNMetrics("can0")[0] + if stream.AnomalyActive || !stream.RecentAnomaly || stream.AnomalyField != "temperature" { + t.Fatalf("recent anomaly context should survive a normal sample: %+v", stream) + } +} + +func TestSourcePGNMetricsFlagLargeGenericDecodedValueChange(t *testing.T) { + now := time.Unix(1_700_000_000, 0).UTC() + reg := newRegistryAt(func() time.Time { return now }) + for i := 0; i < 6; i++ { + reg.RecordSource("gateway", &msg.Envelope{ + PGN: 99999, Source: 3, Payload: json.RawMessage(`{"sensor":10}`), + }) + now = now.Add(time.Second) + } + reg.RecordSource("gateway", &msg.Envelope{ + PGN: 99999, Source: 3, Payload: json.RawMessage(`{"sensor":1000}`), + }) + stream := reg.SourcePGNMetrics("gateway")[0] + if !stream.AnomalyActive || stream.AnomalyField != "sensor" { + t.Fatalf("generic decoded anomaly = %+v", stream) + } +} + +func TestRemoveSourceDropsPGNMetrics(t *testing.T) { + reg := NewRegistry() + reg.RecordSource("can0", &msg.Envelope{PGN: 127250, Source: 1}) + reg.RemoveSource("can0") + if metrics := reg.SourcePGNMetrics("can0"); len(metrics) != 0 { + t.Fatalf("removed source metrics = %+v", metrics) + } +} + +func findField(t *testing.T, fields []FieldDistribution, name string) FieldDistribution { + t.Helper() + for _, field := range fields { + if field.Field == name { + return field + } + } + t.Fatalf("field %q not found in %+v", name, fields) + return FieldDistribution{} +} diff --git a/internal/stats/source_persistence.go b/internal/stats/source_persistence.go new file mode 100644 index 0000000..7d059f3 --- /dev/null +++ b/internal/stats/source_persistence.go @@ -0,0 +1,468 @@ +package stats + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "sort" + "strconv" + "sync" + "time" +) + +const ( + maxSourceMetricEvents = 200 + maxPersistedMetricEvents = 2000 + sourceMetricEventBufferSize = 256 +) + +type BaselineField struct { + Minimum float64 `json:"minimum"` + Maximum float64 `json:"maximum"` + Unit string `json:"unit,omitempty"` +} + +type BaselineRawByte struct { + Offset int `json:"offset"` + Minimum uint8 `json:"minimum"` + Maximum uint8 `json:"maximum"` + ChangedBitMaskHex string `json:"changed_bit_mask_hex"` +} + +// SourceTrafficBaseline is an operator-approved expectation for one PGN from +// one stable Device NAME (or source address when no NAME is available). +type SourceTrafficBaseline struct { + SourceID string `json:"source_id"` + Identity string `json:"identity"` + PGN uint32 `json:"pgn"` + PGNName string `json:"pgn_name,omitempty"` + SourceAddress uint8 `json:"source_address"` + DeviceNameHex string `json:"device_name_hex,omitempty"` + ExpectedFrequencyHz float64 `json:"expected_frequency_hz"` + FrequencyTolerancePercent float64 `json:"frequency_tolerance_percent"` + PayloadLengths []int `json:"payload_lengths"` + DecodeStatus string `json:"decode_status"` + Variant string `json:"variant,omitempty"` + Transport string `json:"transport,omitempty"` + Destinations []int `json:"destinations"` + Priorities []int `json:"priorities"` + Fields map[string]BaselineField `json:"fields,omitempty"` + RawBytes []BaselineRawByte `json:"raw_bytes,omitempty"` + ApprovedAt time.Time `json:"approved_at"` +} + +type SourceMetricEvent struct { + ID int64 `json:"id,omitempty"` + Time time.Time `json:"time"` + SourceID string `json:"source_id"` + PGN uint32 `json:"pgn"` + SourceAddress uint8 `json:"source_address"` + DeviceNameHex string `json:"device_name_hex,omitempty"` + Kind string `json:"kind"` + Severity string `json:"severity"` + Summary string `json:"summary"` + Details map[string]string `json:"details,omitempty"` +} + +type sourceBaselineKey struct { + source, identity string + pgn uint32 +} + +type sourceMetricEventRing struct { + items []SourceMetricEvent +} + +func (r *sourceMetricEventRing) add(event SourceMetricEvent) { + if len(r.items) >= maxSourceMetricEvents { + copy(r.items, r.items[len(r.items)-maxSourceMetricEvents+1:]) + r.items = r.items[:maxSourceMetricEvents-1] + } + r.items = append(r.items, event) +} + +type sourceMetricPersistence struct { + db *sql.DB + events chan SourceMetricEvent + done chan struct{} + mu sync.Mutex + errors int64 + closed bool +} + +// AttachSourceMetricPersistence loads baselines and recent change events, then +// starts a non-blocking event writer. Call this before serving baseline APIs. +func (r *Registry) AttachSourceMetricPersistence(ctx context.Context, db *sql.DB) error { + if r == nil || db == nil { + return nil + } + baselines, err := loadSourceBaselines(ctx, db) + if err != nil { + return err + } + events, err := loadSourceMetricEvents(ctx, db) + if err != nil { + return err + } + persistence := &sourceMetricPersistence{ + db: db, events: make(chan SourceMetricEvent, sourceMetricEventBufferSize), done: make(chan struct{}), + } + r.mu.Lock() + if r.sourcePersistence != nil { + r.mu.Unlock() + return errors.New("source metric persistence already attached") + } + r.sourceBaselines = baselines + for _, event := range events { + ring := r.sourceMetricEvents[event.SourceID] + if ring == nil { + ring = &sourceMetricEventRing{} + r.sourceMetricEvents[event.SourceID] = ring + } + ring.add(event) + } + r.sourcePersistence = persistence + r.mu.Unlock() + go persistence.run() + return nil +} + +func loadSourceBaselines(ctx context.Context, db *sql.DB) (map[sourceBaselineKey]SourceTrafficBaseline, error) { + rows, err := db.QueryContext(ctx, `SELECT doc FROM source_metric_baselines`) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + out := make(map[sourceBaselineKey]SourceTrafficBaseline) + for rows.Next() { + var doc string + if err := rows.Scan(&doc); err != nil { + return nil, err + } + var baseline SourceTrafficBaseline + if err := json.Unmarshal([]byte(doc), &baseline); err != nil { + return nil, err + } + out[sourceBaselineKey{baseline.SourceID, baseline.Identity, baseline.PGN}] = baseline + } + return out, rows.Err() +} + +func loadSourceMetricEvents(ctx context.Context, db *sql.DB) ([]SourceMetricEvent, error) { + rows, err := db.QueryContext(ctx, `SELECT id, doc FROM ( + SELECT id, doc FROM source_metric_events ORDER BY id DESC LIMIT ? + ) ORDER BY id`, maxPersistedMetricEvents) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + out := make([]SourceMetricEvent, 0) + for rows.Next() { + var id int64 + var doc string + if err := rows.Scan(&id, &doc); err != nil { + return nil, err + } + var event SourceMetricEvent + if err := json.Unmarshal([]byte(doc), &event); err != nil { + return nil, err + } + event.ID = id + out = append(out, event) + } + return out, rows.Err() +} + +func (p *sourceMetricPersistence) run() { + defer close(p.done) + for event := range p.events { + doc, err := json.Marshal(event) + if err == nil { + _, err = p.db.Exec(`INSERT INTO source_metric_events + (ts, source_id, pgn, source_address, kind, severity, doc) + VALUES (?, ?, ?, ?, ?, ?, ?)`, event.Time.UnixNano(), event.SourceID, + event.PGN, event.SourceAddress, event.Kind, event.Severity, string(doc)) + } + if err == nil { + _, err = p.db.Exec(`DELETE FROM source_metric_events WHERE id NOT IN ( + SELECT id FROM source_metric_events ORDER BY id DESC LIMIT ? + )`, maxPersistedMetricEvents) + } + if err != nil { + p.mu.Lock() + p.errors++ + p.mu.Unlock() + } + } +} + +func (r *Registry) recordSourceMetricEvent(event SourceMetricEvent) { + if r == nil { + return + } + if event.Time.IsZero() { + event.Time = r.now().UTC() + } + r.mu.Lock() + ring := r.sourceMetricEvents[event.SourceID] + if ring == nil { + ring = &sourceMetricEventRing{} + r.sourceMetricEvents[event.SourceID] = ring + } + ring.add(event) + persistence := r.sourcePersistence + r.mu.Unlock() + if persistence != nil { + persistence.enqueue(event) + } +} + +func (p *sourceMetricPersistence) enqueue(event SourceMetricEvent) { + p.mu.Lock() + defer p.mu.Unlock() + if p.closed { + return + } + select { + case p.events <- event: + default: + p.errors++ + } +} + +func (r *Registry) SourceMetricEvents(source string, limit int) []SourceMetricEvent { + if r == nil { + return []SourceMetricEvent{} + } + r.mu.Lock() + ring := r.sourceMetricEvents[source] + if ring == nil { + r.mu.Unlock() + return []SourceMetricEvent{} + } + items := append([]SourceMetricEvent(nil), ring.items...) + r.mu.Unlock() + if limit <= 0 || limit > len(items) { + limit = len(items) + } + out := append([]SourceMetricEvent(nil), items[len(items)-limit:]...) + sort.Slice(out, func(i, j int) bool { return out[i].Time.After(out[j].Time) }) + return out +} + +func sourceBaselineIdentity(deviceNameHex string, address uint8) string { + if deviceNameHex != "" { + return "name:" + deviceNameHex + } + return fmt.Sprintf("address:%d", address) +} + +func baselineFromMetric(metric SourcePGNMetric, approvedAt time.Time) SourceTrafficBaseline { + lengths := make([]int, 0) + if metric.Raw != nil { + for length := range metric.Raw.LengthCounts { + parsed, err := strconv.Atoi(length) + if err == nil { + lengths = append(lengths, parsed) + } + } + } + if len(lengths) == 0 { + for value := metric.PayloadBytesMin; value <= metric.PayloadBytesMax && value-metric.PayloadBytesMin < 256; value++ { + lengths = append(lengths, int(value)) + } + } + sort.Ints(lengths) + destinations := sortedIntKeys(metric.DestinationCounts) + priorities := sortedIntKeys(metric.PriorityCounts) + fields := make(map[string]BaselineField) + for _, field := range metric.Fields { + if field.P05 == nil || field.P95 == nil { + continue + } + span := *field.P95 - *field.P05 + margin := mathMax(span*0.25, mathMax(mathAbs(*field.P50)*0.02, 1e-9)) + fields[field.Field] = BaselineField{Minimum: *field.P05 - margin, Maximum: *field.P95 + margin, Unit: field.Unit} + } + rawBytes := make([]BaselineRawByte, 0) + if metric.Raw != nil { + for _, rawByte := range metric.Raw.Bytes { + rawBytes = append(rawBytes, BaselineRawByte{ + Offset: rawByte.Offset, Minimum: rawByte.Minimum, Maximum: rawByte.Maximum, + ChangedBitMaskHex: rawByte.ChangedBitMaskHex, + }) + } + } + return SourceTrafficBaseline{ + SourceID: metric.SourceID, Identity: sourceBaselineIdentity(metric.DeviceNameHex, metric.SourceAddress), + PGN: metric.PGN, PGNName: metric.PGNName, SourceAddress: metric.SourceAddress, + DeviceNameHex: metric.DeviceNameHex, ExpectedFrequencyHz: metric.FrequencyHz, + FrequencyTolerancePercent: 25, PayloadLengths: lengths, DecodeStatus: metric.DecodeStatus, + Variant: metric.Variant, Transport: metric.Transport, Destinations: destinations, + Priorities: priorities, Fields: fields, RawBytes: rawBytes, ApprovedAt: approvedAt, + } +} + +func sortedIntKeys(values map[string]int64) []int { + out := make([]int, 0, len(values)) + for value := range values { + parsed, err := strconv.ParseUint(value, 10, 8) + if err == nil { + out = append(out, int(parsed)) + } + } + sort.Slice(out, func(i, j int) bool { return out[i] < out[j] }) + return out +} + +func mathAbs(value float64) float64 { + if value < 0 { + return -value + } + return value +} + +func mathMax(left, right float64) float64 { + if left > right { + return left + } + return right +} + +func (r *Registry) CommitSourceTrafficBaseline(ctx context.Context, source string) ([]SourceTrafficBaseline, error) { + if r == nil { + return nil, errors.New("source metrics unavailable") + } + metrics := r.SourcePGNMetrics(source) + approvedAt := r.now().UTC() + baselines := make([]SourceTrafficBaseline, 0, len(metrics)) + for _, metric := range metrics { + if metric.Observed { + baselines = append(baselines, baselineFromMetric(metric, approvedAt)) + } + } + r.mu.Lock() + persistence := r.sourcePersistence + r.mu.Unlock() + if persistence == nil { + return nil, errors.New("source metric persistence unavailable") + } + tx, err := persistence.db.BeginTx(ctx, nil) + if err != nil { + return nil, err + } + defer func() { _ = tx.Rollback() }() + if _, err := tx.ExecContext(ctx, `DELETE FROM source_metric_baselines WHERE source_id = ?`, source); err != nil { + return nil, err + } + for _, baseline := range baselines { + doc, err := json.Marshal(baseline) + if err != nil { + return nil, err + } + if _, err := tx.ExecContext(ctx, `INSERT INTO source_metric_baselines + (source_id, identity, pgn, approved_at, doc) VALUES (?, ?, ?, ?, ?)`, + baseline.SourceID, baseline.Identity, baseline.PGN, baseline.ApprovedAt.UnixNano(), string(doc)); err != nil { + return nil, err + } + } + if err := tx.Commit(); err != nil { + return nil, err + } + r.mu.Lock() + for key := range r.sourceBaselines { + if key.source == source { + delete(r.sourceBaselines, key) + } + } + for _, baseline := range baselines { + r.sourceBaselines[sourceBaselineKey{baseline.SourceID, baseline.Identity, baseline.PGN}] = baseline + } + r.mu.Unlock() + r.recordSourceMetricEvent(SourceMetricEvent{Time: approvedAt, SourceID: source, + Kind: "baseline_committed", Severity: "info", + Summary: fmt.Sprintf("Set expected traffic baseline from %d observed streams", len(baselines))}) + return baselines, nil +} + +func (r *Registry) ClearSourceTrafficBaseline(ctx context.Context, source string) error { + if r == nil { + return nil + } + r.mu.Lock() + persistence := r.sourcePersistence + r.mu.Unlock() + if persistence == nil { + return errors.New("source metric persistence unavailable") + } + if _, err := persistence.db.ExecContext(ctx, `DELETE FROM source_metric_baselines WHERE source_id = ?`, source); err != nil { + return err + } + r.mu.Lock() + for key := range r.sourceBaselines { + if key.source == source { + delete(r.sourceBaselines, key) + } + } + r.mu.Unlock() + r.recordSourceMetricEvent(SourceMetricEvent{Time: r.now().UTC(), SourceID: source, + Kind: "baseline_cleared", Severity: "info", Summary: "Cleared source traffic baseline"}) + return nil +} + +func (r *Registry) SourceTrafficBaselines(source string) []SourceTrafficBaseline { + if r == nil { + return []SourceTrafficBaseline{} + } + r.mu.Lock() + out := make([]SourceTrafficBaseline, 0) + for key, baseline := range r.sourceBaselines { + if source == "" || key.source == source { + out = append(out, baseline) + } + } + r.mu.Unlock() + sort.Slice(out, func(i, j int) bool { + if out[i].SourceID != out[j].SourceID { + return out[i].SourceID < out[j].SourceID + } + if out[i].PGN != out[j].PGN { + return out[i].PGN < out[j].PGN + } + return out[i].Identity < out[j].Identity + }) + return out +} + +func (r *Registry) CloseSourceMetricPersistence(ctx context.Context) error { + if r == nil { + return nil + } + r.mu.Lock() + persistence := r.sourcePersistence + r.sourcePersistence = nil + r.mu.Unlock() + if persistence == nil { + return nil + } + persistence.mu.Lock() + if !persistence.closed { + persistence.closed = true + close(persistence.events) + } + persistence.mu.Unlock() + select { + case <-persistence.done: + persistence.mu.Lock() + errorsCount := persistence.errors + persistence.mu.Unlock() + if errorsCount > 0 { + return fmt.Errorf("source metric persistence encountered %d write errors", errorsCount) + } + return nil + case <-ctx.Done(): + return ctx.Err() + } +} diff --git a/internal/stats/source_wire_metrics.go b/internal/stats/source_wire_metrics.go new file mode 100644 index 0000000..da87569 --- /dev/null +++ b/internal/stats/source_wire_metrics.go @@ -0,0 +1,352 @@ +package stats + +import ( + "crypto/sha256" + "encoding/hex" + "math" + "math/bits" + "sort" + "strconv" + "time" +) + +const ( + maxAnalyzedPayloadBytes = 64 + maxRawSamples = 8 + maxTrackedFingerprints = 256 + maxReportedFingerprints = 8 + sourceRateBuckets = 60 +) + +// RawPayloadSample is a bounded recent wire-payload example. It is exposed in +// the UI and MCP, but never emitted as a Prometheus label. +type RawPayloadSample struct { + ObservedAt time.Time `json:"observed_at"` + Hex string `json:"hex"` + Fingerprint string `json:"fingerprint"` + Length int `json:"length"` +} + +type PayloadFingerprint struct { + Fingerprint string `json:"fingerprint"` + Count int64 `json:"count"` + Share float64 `json:"share"` + Length int `json:"length"` + LastSeen time.Time `json:"last_seen"` +} + +// ByteDistribution describes one raw payload byte position. Counts stay +// internal; only bounded descriptive values leave the registry. +type ByteDistribution struct { + Offset int `json:"offset"` + Samples int64 `json:"samples"` + Minimum uint8 `json:"minimum"` + Maximum uint8 `json:"maximum"` + MostCommon uint8 `json:"most_common"` + MostCommonShare float64 `json:"most_common_share"` + EntropyBits float64 `json:"entropy_bits"` + ChangedShare float64 `json:"changed_share"` + ChangedBitMaskHex string `json:"changed_bit_mask_hex"` +} + +type RawPayloadDiagnostics struct { + LastHex string `json:"last_hex,omitempty"` + LastFingerprint string `json:"last_fingerprint,omitempty"` + LengthCounts map[string]int64 `json:"length_counts,omitempty"` + DistinctPayloads int `json:"distinct_payloads"` + DistinctPayloadOverflow int64 `json:"distinct_payload_overflow,omitempty"` + UnchangedSeconds float64 `json:"unchanged_seconds,omitempty"` + HammingDistanceMean float64 `json:"hamming_distance_mean,omitempty"` + HammingDistanceP95 float64 `json:"hamming_distance_p95,omitempty"` + LastChangedBytes []int `json:"last_changed_bytes,omitempty"` + Fingerprints []PayloadFingerprint `json:"top_fingerprints,omitempty"` + Bytes []ByteDistribution `json:"byte_distributions,omitempty"` + Samples []RawPayloadSample `json:"recent_samples,omitempty"` +} + +type sourceByteDistribution struct { + samples int64 + counts map[uint8]uint32 + minimum uint8 + maximum uint8 + comparisons int64 + changed int64 + bitChanges [8]int64 +} + +type trackedFingerprint struct { + digest string + count int64 + length int + lastSeen time.Time +} + +type sourceWireStats struct { + previous []byte + lastChanged time.Time + lastChangedBytes []int + lengths map[int]int64 + fingerprints map[string]*trackedFingerprint + fingerprintOverflow int64 + bytes []sourceByteDistribution + hamming runningDistribution + hammingSamples [sourceIntervalSamples]float64 + hammingHead int + hammingLen int + samples [maxRawSamples]RawPayloadSample + sampleHead int + sampleLen int + lastHex string + lastFingerprint string +} + +func payloadFingerprint(raw []byte) string { + digest := sha256.Sum256(raw) + return hex.EncodeToString(digest[:8]) +} + +func (w *sourceWireStats) record(now time.Time, raw []byte) { + if len(raw) == 0 { + return + } + if w.lengths == nil { + w.lengths = make(map[int]int64) + } + w.lengths[len(raw)]++ + fingerprint := payloadFingerprint(raw) + w.lastHex = hex.EncodeToString(raw) + w.lastFingerprint = fingerprint + if w.fingerprints == nil { + w.fingerprints = make(map[string]*trackedFingerprint) + } + if item := w.fingerprints[fingerprint]; item != nil { + item.count++ + item.lastSeen = now + } else if len(w.fingerprints) < maxTrackedFingerprints { + w.fingerprints[fingerprint] = &trackedFingerprint{ + digest: fingerprint, count: 1, length: len(raw), lastSeen: now, + } + } else { + w.fingerprintOverflow++ + } + + changedBytes := make([]int, 0) + if len(w.previous) > 0 { + maxLength := max(len(w.previous), len(raw)) + distance := 0 + for i := 0; i < maxLength; i++ { + var before, after byte + if i < len(w.previous) { + before = w.previous[i] + } + if i < len(raw) { + after = raw[i] + } + if before != after || i >= len(w.previous) || i >= len(raw) { + changedBytes = append(changedBytes, i) + } + distance += bits.OnesCount8(before ^ after) + } + w.hamming.add(float64(distance)) + w.hammingSamples[w.hammingHead] = float64(distance) + w.hammingHead = (w.hammingHead + 1) % sourceIntervalSamples + if w.hammingLen < sourceIntervalSamples { + w.hammingLen++ + } + } + changed := len(w.previous) == 0 || len(changedBytes) > 0 + if changed { + w.lastChanged = now + w.lastChangedBytes = append([]int(nil), changedBytes...) + sample := RawPayloadSample{ObservedAt: now, Hex: w.lastHex, Fingerprint: fingerprint, Length: len(raw)} + w.samples[w.sampleHead] = sample + w.sampleHead = (w.sampleHead + 1) % maxRawSamples + if w.sampleLen < maxRawSamples { + w.sampleLen++ + } + } + + analyzed := min(len(raw), maxAnalyzedPayloadBytes) + for len(w.bytes) < analyzed { + w.bytes = append(w.bytes, sourceByteDistribution{counts: make(map[uint8]uint32)}) + } + for i := 0; i < analyzed; i++ { + value := raw[i] + byteStats := &w.bytes[i] + if byteStats.samples == 0 { + byteStats.minimum, byteStats.maximum = value, value + } else { + if value < byteStats.minimum { + byteStats.minimum = value + } + if value > byteStats.maximum { + byteStats.maximum = value + } + } + byteStats.samples++ + byteStats.counts[value]++ + if i < len(w.previous) { + byteStats.comparisons++ + delta := w.previous[i] ^ value + if delta != 0 { + byteStats.changed++ + } + for bit := 0; bit < 8; bit++ { + if delta&(1< 0 { + share = float64(item.count) / float64(total) + } + fingerprints = append(fingerprints, PayloadFingerprint{ + Fingerprint: item.digest, Count: item.count, Share: share, + Length: item.length, LastSeen: item.lastSeen, + }) + } + sort.Slice(fingerprints, func(i, j int) bool { + if fingerprints[i].Count != fingerprints[j].Count { + return fingerprints[i].Count > fingerprints[j].Count + } + return fingerprints[i].Fingerprint < fingerprints[j].Fingerprint + }) + if len(fingerprints) > maxReportedFingerprints { + fingerprints = fingerprints[:maxReportedFingerprints] + } + out.Fingerprints = fingerprints + + for offset, value := range w.bytes { + if value.samples == 0 { + continue + } + var mode uint8 + var modeCount uint32 + entropy := 0.0 + for candidate, count := range value.counts { + if count > modeCount { + mode, modeCount = candidate, count + } + probability := float64(count) / float64(value.samples) + entropy -= probability * math.Log2(probability) + } + changedShare := 0.0 + var changedMask uint8 + if value.comparisons > 0 { + changedShare = float64(value.changed) / float64(value.comparisons) + for bit, count := range value.bitChanges { + if count > 0 { + changedMask |= 1 << bit + } + } + } + out.Bytes = append(out.Bytes, ByteDistribution{ + Offset: offset, Samples: value.samples, Minimum: value.minimum, Maximum: value.maximum, + MostCommon: mode, MostCommonShare: float64(modeCount) / float64(value.samples), + EntropyBits: entropy, ChangedShare: changedShare, + ChangedBitMaskHex: hex.EncodeToString([]byte{changedMask}), + }) + } + start = (w.sampleHead - w.sampleLen + maxRawSamples) % maxRawSamples + for i := 0; i < w.sampleLen; i++ { + out.Samples = append(out.Samples, w.samples[(start+i)%maxRawSamples]) + } + return out +} + +func floatPercentile(values []float64, percentile float64) float64 { + if len(values) == 0 { + return 0 + } + copyValues := append([]float64(nil), values...) + sort.Float64s(copyValues) + index := int(math.Ceil(percentile*float64(len(copyValues)))) - 1 + if index < 0 { + index = 0 + } + if index >= len(copyValues) { + index = len(copyValues) - 1 + } + return copyValues[index] +} + +type sourceRateBucket struct { + second int64 + messages int64 + bytes int64 + wireBits int64 +} + +type sourceRateStats struct { + buckets [sourceRateBuckets]sourceRateBucket +} + +func (r *sourceRateStats) record(now time.Time, payloadBytes int, fastPacket bool) { + second := now.Unix() + index := int(second % sourceRateBuckets) + bucket := &r.buckets[index] + if bucket.second != second { + *bucket = sourceRateBucket{second: second} + } + bucket.messages++ + bucket.bytes += int64(payloadBytes) + frames := 1 + if fastPacket || payloadBytes > 8 { + remaining := max(payloadBytes-6, 0) + frames += (remaining + 6) / 7 + } + bucket.wireBits += int64(frames * 130) // extended CAN frame including framing/stuffing allowance +} + +func (r *sourceRateStats) snapshot(now time.Time) (messagesPerSecond, bytesPerSecond, busLoadPercent float64) { + cutoff := now.Unix() - sourceRateBuckets + 1 + var messages, bytesCount, wireBits int64 + for _, bucket := range r.buckets { + if bucket.second < cutoff || bucket.second > now.Unix() { + continue + } + messages += bucket.messages + bytesCount += bucket.bytes + wireBits += bucket.wireBits + } + messagesPerSecond = float64(messages) / sourceRateBuckets + bytesPerSecond = float64(bytesCount) / sourceRateBuckets + busLoadPercent = float64(wireBits) / (250_000 * sourceRateBuckets) * 100 + return +} diff --git a/internal/stats/stats.go b/internal/stats/stats.go index dbf40e5..adee033 100644 --- a/internal/stats/stats.go +++ b/internal/stats/stats.go @@ -354,13 +354,20 @@ func (c *counters) depthHistoryLocked() []int64 { // Snapshot/All, the same nil-safe convention as metrics.Set, so callers // never need a nil check around instrumentation. type Registry struct { - now func() time.Time - - mu sync.Mutex - conn map[string]*counters - source map[string]*counters - sink map[string]*counters - events map[string]*eventRing + now func() time.Time + startedAt time.Time + + mu sync.Mutex + conn map[string]*counters + source map[string]*counters + sink map[string]*counters + events map[string]*eventRing + sourceStreams map[sourceStreamKey]*sourceStream + sourceBaselines map[sourceBaselineKey]SourceTrafficBaseline + sourceMetricEvents map[string]*sourceMetricEventRing + sourcePersistence *sourceMetricPersistence + sourceAddressNames map[sourceAddressKey]uint64 + sourceDeviceAddresses map[sourceDeviceKey]uint8 } // NewRegistry returns an empty Registry using the real wall clock. @@ -372,11 +379,17 @@ func NewRegistry() *Registry { // (via r.now) to exercise rate decay deterministically. func newRegistryAt(now func() time.Time) *Registry { return &Registry{ - now: now, - conn: map[string]*counters{}, - source: map[string]*counters{}, - sink: map[string]*counters{}, - events: map[string]*eventRing{}, + now: now, + startedAt: now(), + conn: map[string]*counters{}, + source: map[string]*counters{}, + sink: map[string]*counters{}, + events: map[string]*eventRing{}, + sourceStreams: map[sourceStreamKey]*sourceStream{}, + sourceBaselines: map[sourceBaselineKey]SourceTrafficBaseline{}, + sourceMetricEvents: map[string]*sourceMetricEventRing{}, + sourceAddressNames: map[sourceAddressKey]uint64{}, + sourceDeviceAddresses: map[sourceDeviceKey]uint8{}, } } @@ -440,12 +453,22 @@ func (r *Registry) recordEvent(kind, id string, e Event) { } func (r *Registry) RecordSource(source string, e *msg.Envelope) { - if r == nil { + if r == nil || e == nil { return } now := r.now() ev := eventFromEnvelope(now, "received", "", e) r.getSource(source).record(now, 1, int64(ev.SizeBytes)) + stream, created := r.getSourceStream(source, e) + before := stream.sourceEventState() + stream.record(now, e) + after := stream.sourceEventState() + for _, event := range r.sourceLifecycleEvents(now, source, e, created, before, after) { + r.recordSourceMetricEvent(event) + } + for _, event := range r.sourceIdentityEvents(now, source, e) { + r.recordSourceMetricEvent(event) + } r.recordEvent("source", source, ev) } @@ -614,6 +637,11 @@ func (r *Registry) RemoveSource(source string) { r.mu.Lock() delete(r.source, source) delete(r.events, "source:"+source) + for key := range r.sourceStreams { + if key.source == source { + delete(r.sourceStreams, key) + } + } r.mu.Unlock() } diff --git a/internal/store/store.go b/internal/store/store.go index 774619d..6b1e627 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -48,6 +48,26 @@ var migrations = []string{ current_doc TEXT NOT NULL, PRIMARY KEY(endpoint, device_name) );`, + `CREATE TABLE IF NOT EXISTS source_metric_baselines ( + source_id TEXT NOT NULL, + identity TEXT NOT NULL, + pgn INTEGER NOT NULL, + approved_at INTEGER NOT NULL, + doc TEXT NOT NULL, + PRIMARY KEY(source_id, identity, pgn) + ); + CREATE INDEX IF NOT EXISTS source_metric_baselines_source ON source_metric_baselines(source_id); + CREATE TABLE IF NOT EXISTS source_metric_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts INTEGER NOT NULL, + source_id TEXT NOT NULL, + pgn INTEGER NOT NULL, + source_address INTEGER NOT NULL, + kind TEXT NOT NULL, + severity TEXT NOT NULL, + doc TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS source_metric_events_source_ts ON source_metric_events(source_id, ts DESC);`, } func Open(path string) (*Store, error) { diff --git a/internal/ui/assets/app.css b/internal/ui/assets/app.css index bb2cda0..3a74761 100644 --- a/internal/ui/assets/app.css +++ b/internal/ui/assets/app.css @@ -370,17 +370,15 @@ code { .dag-node-connector { border-color: var(--border-strong); - background: var(--black); - color: #fff; } .dag-node-connector:hover { - color: #fff; + color: var(--gray-dark); } .dag-node-connector code { - background: rgba(255, 255, 255, 0.12); - color: #fff; + background: rgba(255, 255, 255, 0.65); + color: var(--black); } .dag-node-sink { @@ -417,7 +415,7 @@ code { .dag-metrics { justify-content: space-between; - border-top: 1px solid rgba(255, 255, 255, 0.14); + border-top: 1px solid rgba(96, 115, 159, 0.22); padding-top: var(--space-3); } @@ -426,10 +424,27 @@ code { } .dag-link { + --dag-link-color: var(--border-strong); position: relative; min-width: 48px; } +.dag-link.state-up { + --dag-link-color: var(--success); +} + +.dag-link.state-degraded { + --dag-link-color: var(--warning); +} + +.dag-link.state-error { + --dag-link-color: var(--error); +} + +.dag-link.state-restarting { + --dag-link-color: var(--accent); +} + .dag-link::before { content: ""; position: absolute; @@ -437,7 +452,7 @@ code { right: var(--space-2); left: var(--space-2); height: 2px; - background: var(--border-strong); + background: var(--dag-link-color); transform: translateY(-50%); } @@ -448,8 +463,8 @@ code { right: var(--space-2); width: 8px; height: 8px; - border-top: 2px solid var(--border-strong); - border-right: 2px solid var(--border-strong); + border-top: 2px solid var(--dag-link-color); + border-right: 2px solid var(--dag-link-color); transform: translateY(-50%) rotate(45deg); } @@ -635,6 +650,164 @@ code { min-width: 920px; } +.source-metrics-heading { + align-items: start; +} + +.source-baseline-actions { + display: flex; + flex-wrap: wrap; + justify-content: flex-end; + gap: 0.45rem; +} + +.source-metrics-heading > div { + display: grid; + gap: 0.25rem; +} + +.source-metrics-heading p { + margin: 0; + color: #666; + font-size: 0.84rem; +} + +.source-metrics-table { + min-width: 1280px; +} + +.source-metrics-table td { + vertical-align: top; +} + +.source-stream-gap td { + background: rgba(180, 35, 24, 0.055); +} + +.source-stream-anomaly td { + background: rgba(166, 95, 0, 0.06); +} + +.source-stream-changed td, +.source-stream-awaiting td { + background: rgba(166, 95, 0, 0.04); +} + +.source-stream-missing td { + background: rgba(180, 35, 24, 0.075); +} + +.source-gap-text { + color: var(--error); + font-weight: 800; +} + +.source-anomaly-note { + display: block; + max-width: 11rem; + margin-top: 0.35rem; + overflow: hidden; + color: var(--warning); + font-size: 0.75rem; + font-weight: 750; + text-overflow: ellipsis; + white-space: nowrap; +} + +.source-baseline-state { + display: block; + margin-top: 0.32rem; + color: #666; + font-size: 0.72rem; + font-weight: 750; +} + +.source-values { + min-width: 19rem; +} + +.source-values summary { + color: var(--accent-dark); + cursor: pointer; + font-weight: 800; +} + +.source-value-list { + display: grid; + gap: 0.55rem; + width: min(42rem, 70vw); + margin-top: 0.65rem; +} + +.source-value-list h4 { + margin: 0.65rem 0 0; + color: var(--black); + font-size: 0.86rem; +} + +.source-raw-summary { + display: grid; + grid-template-columns: max-content minmax(0, 1fr); + gap: 0.3rem 0.7rem; + margin: 0; +} + +.source-raw-summary dt { + color: #666; + font-size: 0.74rem; + font-weight: 800; + text-transform: uppercase; +} + +.source-raw-summary dd { + min-width: 0; + margin: 0; + overflow-wrap: anywhere; +} + +.source-byte-table { + min-width: 640px; + font-size: 0.78rem; +} + +.source-fingerprint-list, +.source-raw-samples { + color: #555; + font-size: 0.8rem; + line-height: 1.6; +} + +.source-raw-samples { + display: grid; + gap: 0.35rem; +} + +.source-raw-samples div { + min-width: 0; + overflow-wrap: anywhere; +} + +.source-event-table { + min-width: 900px; +} + +.source-value-row { + display: grid; + gap: 0.15rem; + border-left: 3px solid var(--border); + padding-left: 0.55rem; +} + +.source-value-row > span, +.source-value-row > small { + color: #666; + line-height: 1.35; +} + +.source-value-anomaly { + border-left-color: var(--warning); +} + .event-table td:last-child code { white-space: normal; } @@ -807,13 +980,72 @@ code { box-shadow: 0 0 0 4px var(--warning-bg); } -.state-error { - border-color: rgba(180, 35, 24, 0.45); +/* Component health surfaces share one visual language across the flow DAG, + route arrows, metadata/config tables, and detail status card. Badges retain + the explicit state text, so color is never the sole status signal. */ +.component-status-surface, +.component-status-row { + --component-state-accent: var(--gray); + --component-state-bg: var(--surface-muted); + --component-state-hover: var(--surface-strong); + --component-state-border: var(--border-strong); +} + +.component-status-surface.state-up, +.component-status-row.state-up { + --component-state-accent: var(--success); + --component-state-bg: var(--success-bg); + --component-state-hover: #ddf1e6; + --component-state-border: rgba(15, 122, 79, 0.42); +} + +.component-status-surface.state-degraded, +.component-status-row.state-degraded { + --component-state-accent: var(--warning); + --component-state-bg: var(--warning-bg); + --component-state-hover: #fbe9c8; + --component-state-border: rgba(166, 95, 0, 0.42); +} + +.component-status-surface.state-error, +.component-status-row.state-error { + --component-state-accent: var(--error); + --component-state-bg: var(--error-bg); + --component-state-hover: #f9e2df; + --component-state-border: rgba(180, 35, 24, 0.45); +} + +.component-status-surface.state-restarting, +.component-status-row.state-restarting { + --component-state-accent: var(--accent); + --component-state-bg: #edf0ff; + --component-state-hover: #e1e6ff; + --component-state-border: rgba(35, 55, 255, 0.35); +} + +.component-status-surface { + border-color: var(--component-state-border); + background: var(--component-state-bg); } -.state-degraded, -.state-restarting { - border-color: rgba(166, 95, 0, 0.42); +.dag-node.component-status-surface { + box-shadow: inset 4px 0 0 var(--component-state-accent); +} + +.overview-card.component-status-surface { + box-shadow: inset 4px 0 0 var(--component-state-accent), var(--shadow); +} + +.component-status-row { + background: var(--component-state-bg); +} + +.component-status-row > td:first-child { + box-shadow: inset 4px 0 0 var(--component-state-accent); +} + +.table tbody tr.component-status-row[data-href]:hover { + background: var(--component-state-hover); } .btn { @@ -1821,18 +2053,22 @@ code { } .metadata-section[aria-label="Connectors metadata"] td:nth-child(5)::before { - content: "Filters"; + content: "State"; } .metadata-section[aria-label="Connectors metadata"] td:nth-child(6)::before { - content: "Msg/s"; + content: "Filters"; } .metadata-section[aria-label="Connectors metadata"] td:nth-child(7)::before { - content: "Queue"; + content: "Msg/s"; } .metadata-section[aria-label="Connectors metadata"] td:nth-child(8)::before { + content: "Queue"; + } + + .metadata-section[aria-label="Connectors metadata"] td:nth-child(9)::before { content: "Action"; } } diff --git a/internal/ui/dashboard.go b/internal/ui/dashboard.go index fb631ec..5d16788 100644 --- a/internal/ui/dashboard.go +++ b/internal/ui/dashboard.go @@ -53,7 +53,7 @@ type dashboardEndpointNode struct { ConnectorCount int } -// endpointState resolves one configured component's node state: +// componentState resolves one configured component's displayed state: // // - its live supervisor status, if statuses (the reconciler's current // snapshot) currently reports one for (kind, id) — "up"/"degraded"/ @@ -64,9 +64,9 @@ type dashboardEndpointNode struct { // - "restarting", if it's enabled but transiently missing from statuses // anyway — e.g. between a hot-apply's stop of the old instance and the // new one's start landing in the supervisor's state map. This is the -// dashboard's documented tolerance for that window: it renders a -// neutral node, never a crash and never a silently dropped component. -func endpointState(statuses []supervisor.Status, kind, id string, enabled bool) string { +// UI's documented tolerance for that window: it renders an explicit +// restarting state, never a crash and never a silently dropped component. +func componentState(statuses []supervisor.Status, kind, id string, enabled bool) string { for _, s := range statuses { if s.Kind == kind && s.ID == id { return s.State @@ -95,7 +95,7 @@ func sourceEndpointNodes(sources []model.Source, statuses []supervisor.Status, c Type: string(s.Type), Detail: sourceDetail(s), Enabled: s.Enabled, - State: endpointState(statuses, "source", s.ID, s.Enabled), + State: componentState(statuses, "source", s.ID, s.Enabled), ConnectorCount: connectorCounts[s.ID], } } @@ -112,7 +112,7 @@ func sinkEndpointNodes(sinks []model.Sink, statuses []supervisor.Status, connect Type: string(s.Type), Detail: sinkDetail(s), Enabled: s.Enabled, - State: endpointState(statuses, "sink", s.ID, s.Enabled), + State: componentState(statuses, "sink", s.ID, s.Enabled), ConnectorCount: connectorCounts[s.ID], } } @@ -176,7 +176,7 @@ func dashboardFlows(connectors []model.Connector, reg *stats.Registry, statuses flows[i] = dashboardFlow{ Connector: c, Snapshot: snap, - State: stateFor(statuses, "connector", c.ID), + State: componentState(statuses, "connector", c.ID, c.Enabled), BytesPerSecText: humanizeBytes(snap.BytesPerSec, "/s"), Source: src, Sink: sink, diff --git a/internal/ui/dashboard_test.go b/internal/ui/dashboard_test.go index bf26b2f..acfaa0d 100644 --- a/internal/ui/dashboard_test.go +++ b/internal/ui/dashboard_test.go @@ -77,7 +77,7 @@ func markerSnippet(t *testing.T, body, marker string) string { if idx < 0 { t.Fatalf("dashboard fragment has no marker %q:\n%s", marker, body) } - start := idx - 250 + start := idx - 450 if start < 0 { start = 0 } @@ -225,6 +225,15 @@ func TestDashboardFragConnectorErrorBadge(t *testing.T) { if !strings.Contains(body, "badge-success\">enabled") { t.Fatalf("dashboard fragment lost the enabled badge alongside the error badge:\n%s", body) } + for _, want := range []string{ + `dag-node-connector component-status-surface state-error`, + `dag-link state-error`, + ``, + } { + if !strings.Contains(body, want) { + t.Errorf("dashboard fragment missing connector error colorization %q:\n%s", want, body) + } + } } // --- Endpoint node states + transient-absence tolerance --- @@ -239,7 +248,7 @@ func TestDashboardFragEndpointNodeStates(t *testing.T) { must(t, svc.PutSink(ctx, model.Sink{ID: "err-sink", Name: "Err Sink", Type: model.SinkTCP, Enabled: true, Address: "127.0.0.1:9001"}, true)) for _, id := range []string{"up", "degraded", "restarting", "off"} { must(t, svc.PutConnector(ctx, model.Connector{ - ID: id + "-conn", Name: id + " connector", SourceID: id + "-src", SinkID: "err-sink", Enabled: true, + ID: id + "-conn", Name: id + " connector", SourceID: id + "-src", SinkID: "err-sink", Enabled: id != "off", }, true)) } @@ -250,18 +259,24 @@ func TestDashboardFragEndpointNodeStates(t *testing.T) { {Kind: "source", ID: "up-src", State: "up"}, {Kind: "source", ID: "degraded-src", State: "degraded"}, {Kind: "sink", ID: "err-sink", State: "error", Err: "boom"}, + {Kind: "connector", ID: "up-conn", State: "up"}, + {Kind: "connector", ID: "degraded-conn", State: "degraded"}, }) body := dashboardFrag(t, srv) cases := []struct { - name, wantBadgeClass, wantText string + name, wantBadgeClass, wantText, wantStateClass string }{ - {"Up Src", "badge-success", "up"}, - {"Degraded Src", "badge-warning", "degraded"}, - {"Err Sink", "badge-error", "error"}, - {"Restarting Src", "badge-ghost", "restarting"}, - {"Off Src", "badge-ghost", "disabled"}, + {"Up Src", "badge-success", "up", "component-status-surface state-up"}, + {"Degraded Src", "badge-warning", "degraded", "component-status-surface state-degraded"}, + {"Err Sink", "badge-error", "error", "component-status-surface state-error"}, + {"Restarting Src", "badge-ghost", "restarting", "component-status-surface state-restarting"}, + {"Off Src", "badge-ghost", "disabled", "component-status-surface state-disabled"}, + {"up connector", "badge-success", "up", "component-status-surface state-up"}, + {"degraded connector", "badge-warning", "degraded", "component-status-surface state-degraded"}, + {"restarting connector", "badge-ghost", "restarting", "component-status-surface state-restarting"}, + {"off connector", "badge-ghost", "disabled", "component-status-surface state-disabled"}, } for _, c := range cases { snip := markerSnippet(t, body, c.name) @@ -271,6 +286,48 @@ func TestDashboardFragEndpointNodeStates(t *testing.T) { if !strings.Contains(snip, ">"+c.wantText+"<") { t.Errorf("node for %q = %q, want state text %q", c.name, snip, c.wantText) } + if !strings.Contains(snip, c.wantStateClass) { + t.Errorf("node for %q = %q, want status surface class %q", c.name, snip, c.wantStateClass) + } + } + + for _, want := range []string{ + ``, + ``, + ``, + ``, + ``, + ``, + ``, + ``, + ``, + ``, + ``, + } { + if !strings.Contains(body, want) { + t.Errorf("dashboard metadata missing status-colored row %q", want) + } + } + + for _, tc := range []struct { + path, want string + }{ + {"/ui/sources", ``}, + {"/ui/sinks", ``}, + {"/ui/connectors", ``}, + {"/ui/frag/sources/up-src/overview", `
`}, + {"/ui/frag/sinks/err-sink/overview", `
`}, + {"/ui/frag/connectors/up-conn/overview", `
`}, + } { + resp, err := http.Get(srv.URL + tc.path) + if err != nil { + t.Fatal(err) + } + mustStatus(t, resp, http.StatusOK) + page := mustBody(t, resp) + if !strings.Contains(page, tc.want) { + t.Errorf("%s missing status background class %q:\n%s", tc.path, tc.want, page) + } } } diff --git a/internal/ui/docs/05-api.md b/internal/ui/docs/05-api.md index 1147ad3..94a17c7 100644 --- a/internal/ui/docs/05-api.md +++ b/internal/ui/docs/05-api.md @@ -24,9 +24,16 @@ session headers: The tools are `get_config`, `put_source`, `put_sink`, `put_connector`, `delete_source`, `delete_sink`, `delete_connector`, `get_health`, and -`get_delivery_statistics`. Their input and output JSON Schemas are returned by -MCP `tools/list`. Configuration writes use the same validation, SQLite -persistence, and immediate supervisor reconciliation as the REST calls below. +`get_delivery_statistics`, plus `get_source_metrics`, +`commit_source_traffic_baseline`, and `clear_source_traffic_baseline`. Source +metrics can be filtered by configured source id, PGN, and NMEA 2000 source +address. They include learned timing/jitter, rates and estimated bus load, +addressing, decode quality, payload-size and raw-byte distributions, last-seen +age, gaps/bursts, anomalies, decoded-field quantiles and availability, approved +expectations, and recent lifecycle events. Input and output JSON Schemas are +returned by MCP `tools/list`. Configuration and baseline writes persist to +SQLite; configuration changes also reconcile immediately through the +supervisor. MCP needs no cloud relay, separate process, remote schema, or internet access. It can change live configuration, so bind the admin listener to localhost or a @@ -143,6 +150,16 @@ A known-but-idle connector reports a zero snapshot rather than 404 — exposition at `/metrics` (admin port) for the same data in a form built for scraping/alerting rather than point queries. +Source overview pages show one row for each `(source, source address, PGN)` +stream. The same process-local store is available to agents through +`get_source_metrics` and at `/metrics` under the +`beacon_source_pgn_*` metric families. Frequency and gap thresholds are learned +from recent arrival intervals. Live observation windows reset when Beacon +restarts; operator-approved expected-traffic baselines and bounded lifecycle +events persist in SQLite. Prometheus exports bounded numeric summaries and +finite labels, while raw hexdumps and payload fingerprints remain available in +the UI and MCP response rather than becoming high-cardinality time series. + ## Health and system info ``` diff --git a/internal/ui/docs/06-troubleshooting.md b/internal/ui/docs/06-troubleshooting.md index 9f9ad19..2d65194 100644 --- a/internal/ui/docs/06-troubleshooting.md +++ b/internal/ui/docs/06-troubleshooting.md @@ -20,6 +20,22 @@ Work from the source outward: everything, or no connector wired to this source at all). If it's flat at zero, nothing is reaching beacon from the interface. +For one missing sensor or PGN on an otherwise busy bus, open that source's +overview and check **PGN traffic**. Gap rows are sorted first and show last-seen +age versus the stream's learned period. The equivalent Prometheus signal is +`beacon_source_pgn_gap_active`; agents can query the same store with the MCP +`get_source_metrics` tool. + +After the source has run under normal conditions, use **Set expected traffic +baseline** to save the observed senders, PGNs, rates, payload shapes, decode +metadata, addressing, and field ranges as the expected baseline. The overview +will then surface a PGN that never returns after startup, frequency drift, new +payload lengths or destinations, Device NAME/address changes, and values outside +the approved range. Baselines can also be managed with the MCP +`commit_source_traffic_baseline` and `clear_source_traffic_baseline` tools. +Setting the baseline replaces the previous one with streams observed since +Beacon started; do not set it while an expected device is absent. + ## CAN write failures A `socketcan`/`usbcan` sink confirms every push; on failure the connector diff --git a/internal/ui/forms.go b/internal/ui/forms.go index bcd7082..b45733f 100644 --- a/internal/ui/forms.go +++ b/internal/ui/forms.go @@ -47,20 +47,6 @@ type alertData struct { Message string } -// stateFor returns the State of the status in statuses matching kind/id -// ("source"/"sink", per internal/supervisor.Status.Kind's literal values — -// see internal/supervisor/supervisor.go), or "unknown" if none is reported -// yet (e.g. a disabled entity the supervisor never started, or a request -// racing just after a config write's reconcile). -func stateFor(statuses []supervisor.Status, kind, id string) string { - for _, s := range statuses { - if s.Kind == kind && s.ID == id { - return s.State - } - } - return "unknown" -} - // referencingConnectorNames returns the ids of every connector referencing // id as its source (kind="source") or sink (kind="sink"), for composing the // delete-in-use error message: config.ErrInUse alone doesn't carry enough @@ -210,7 +196,7 @@ type sourceRow struct { func sourceRows(sources []model.Source, statuses []supervisor.Status) []sourceRow { rows := make([]sourceRow, len(sources)) for i, s := range sources { - rows[i] = sourceRow{Source: s, State: stateFor(statuses, "source", s.ID), Detail: sourceDetail(s)} + rows[i] = sourceRow{Source: s, State: componentState(statuses, "source", s.ID, s.Enabled), Detail: sourceDetail(s)} } return rows } @@ -636,7 +622,7 @@ type sinkRow struct { func sinkRows(sinks []model.Sink, statuses []supervisor.Status) []sinkRow { rows := make([]sinkRow, len(sinks)) for i, s := range sinks { - rows[i] = sinkRow{Sink: s, State: stateFor(statuses, "sink", s.ID), Detail: sinkDetail(s)} + rows[i] = sinkRow{Sink: s, State: componentState(statuses, "sink", s.ID, s.Enabled), Detail: sinkDetail(s)} } return rows } @@ -1047,8 +1033,8 @@ func handleSinkDelete(svc *config.Service, log *slog.Logger) http.HandlerFunc { // connectorRow is one row of the connectors table // (frag_connector_table.html): model.Connector plus its live stats -// snapshot and its source/sink NAMES (nameOrID — falls back to the raw id -// when the source/sink has no name set). reg.Snapshot returns a zero +// snapshot, supervisor state, and source/sink NAMES (nameOrID — falls back +// to the raw id when the source/sink has no name set). reg.Snapshot returns a zero // Snapshot for a connector it has never recorded (not yet started, or // started but idle since boot) — so a row for a connector with no traffic // yet renders zero queue depth/msg-per-second rather than needing a @@ -1059,6 +1045,7 @@ type connectorRow struct { Snapshot stats.Snapshot SourceName string SinkName string + State string } // connectorRows builds connectorTableData's rows. sourceNames/sinkNames are @@ -1067,7 +1054,7 @@ type connectorRow struct { // handler already has to fetch for other reasons (the connector form's //