Documentation
¶
Overview ¶
Package stats tracks live per-connector counters and rolling boundary-acceptance rates for the config API and the future UI dashboard. It is independent of OTel (which serves Prometheus via internal/metrics): this registry exists for cheap, synchronous, in-process reads from HTTP handlers.
Index ¶
- type ByteDistribution
- type Event
- type FieldDistribution
- type PayloadFingerprint
- type RawPayloadDiagnostics
- type RawPayloadSample
- type Registry
- func (r *Registry) All() map[string]Snapshot
- func (r *Registry) AllSourcePGNMetrics() map[string][]SourcePGNMetric
- func (r *Registry) AttachSourceMetricPersistence(ctx context.Context, db *sql.DB) error
- func (r *Registry) CloseSourceMetricPersistence(ctx context.Context) error
- func (r *Registry) Recent(kind, id string, limit int) []Event
- func (r *Registry) Record(connector string, msgs, bytes int64)
- func (r *Registry) RecordConnectorEvent(connector, stage string, e *msg.Envelope)
- func (r *Registry) RecordSink(sink, connector string, e *msg.Envelope)
- func (r *Registry) RecordSource(source string, e *msg.Envelope)
- func (r *Registry) RecordSourceDrops(source string, n int64)
- func (r *Registry) RecordStage(connector, stage string, n int64)
- func (r *Registry) Remove(connector string)
- func (r *Registry) RemoveSink(sink string)
- func (r *Registry) RemoveSource(source string)
- func (r *Registry) SetQueue(connector string, depth, bytes int64)
- func (r *Registry) SetQueueStats(connector string, s queue.Stats)
- func (r *Registry) SetRuntime(connector, deliveryClass, state string, err error)
- func (r *Registry) SinkSnapshot(sink string) (Snapshot, bool)
- func (r *Registry) Snapshot(connector string) (Snapshot, bool)
- func (r *Registry) SourceMetricCapacity(source string) SourceMetricCapacity
- func (r *Registry) SourceMetricEvents(source string, limit int) []SourceMetricEvent
- func (r *Registry) SourcePGNLastPayloadsFiltered(source string, filter SourcePGNMetricFilter) []SourcePGNLastPayload
- func (r *Registry) SourcePGNLastPayloadsFilteredLimit(source string, filter SourcePGNMetricFilter, limit int) ([]SourcePGNLastPayload, bool)
- func (r *Registry) SourcePGNMetrics(source string) []SourcePGNMetric
- func (r *Registry) SourcePGNMetricsFiltered(source string, filter SourcePGNMetricFilter) []SourcePGNMetric
- func (r *Registry) SourcePGNMetricsFilteredLimit(source string, filter SourcePGNMetricFilter, limit int) ([]SourcePGNMetric, bool)
- func (r *Registry) SourcePGNMetricsForAddress(source string, address uint8) []SourcePGNMetric
- func (r *Registry) SourcePGNSummaries(source string) []SourcePGNMetric
- func (r *Registry) SourceSnapshot(source string) (Snapshot, bool)
- func (r *Registry) SubscribeStream(kind, id string, buffer int) (<-chan []byte, func())
- func (r *Registry) Touch(connector string)
- type Snapshot
- type SourceMetricCapacity
- type SourceMetricEvent
- type SourcePGNLastPayload
- type SourcePGNMetric
- type SourcePGNMetricFilter
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type ByteDistribution ¶ added in v1.0.2
type ByteDistribution struct {
Offset int `json:"offset"`
Samples int64 `json:"samples"`
Minimum uint8 `json:"minimum"`
Maximum uint8 `json:"maximum"`
MostCommon uint8 `json:"most_common"`
EntropyBits float64 `json:"entropy_bits"`
ChangedBitMaskHex string `json:"changed_bit_mask_hex"`
OtherSamples int64 `json:"other_samples,omitempty"`
}
ByteDistribution describes one raw payload byte position. Counts stay internal; only bounded descriptive values leave the registry.
type Event ¶
type Event struct {
Time time.Time `json:"time"`
Stage string `json:"stage"`
ConnectorID string `json:"connector_id,omitempty"`
PGN uint32 `json:"pgn"`
PGNName string `json:"pgn_name,omitempty"`
Source uint8 `json:"source"`
Dest uint8 `json:"dest"`
Priority uint8 `json:"priority"`
Timestamp time.Time `json:"timestamp"`
Payload string `json:"payload,omitempty"`
SizeBytes int `json:"size_bytes"`
}
type FieldDistribution ¶ added in v1.0.2
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"`
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"`
}
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 PayloadFingerprint ¶ added in v1.0.2
type RawPayloadDiagnostics ¶ added in v1.0.2
type RawPayloadDiagnostics struct {
LastHex string `json:"last_hex,omitempty"`
LastFingerprint string `json:"last_fingerprint,omitempty"`
LastLength int `json:"last_length,omitempty"`
LastHexTruncated bool `json:"last_hex_truncated,omitempty"`
RetainedByteLimit int `json:"retained_byte_limit"`
TruncatedSamples int64 `json:"truncated_samples,omitempty"`
LengthCounts map[string]int64 `json:"length_counts,omitempty"`
LengthCountOverflow int64 `json:"length_count_overflow,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"`
LastChangeOutsidePrefix bool `json:"last_change_outside_prefix,omitempty"`
Fingerprints []PayloadFingerprint `json:"top_fingerprints,omitempty"`
Bytes []ByteDistribution `json:"byte_distributions,omitempty"`
Samples []RawPayloadSample `json:"recent_samples,omitempty"`
}
type RawPayloadSample ¶ added in v1.0.2
type RawPayloadSample struct {
ObservedAt time.Time `json:"observed_at"`
Hex string `json:"hex"`
Fingerprint string `json:"fingerprint"`
Length int `json:"length"`
HexTruncated bool `json:"hex_truncated,omitempty"`
}
RawPayloadSample is a bounded recent wire-payload example. It is exposed in the UI and MCP, but never emitted as a Prometheus label.
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry tracks live per-connector counters and rolling rates for the API and UI. It is independent of OTel (which serves Prometheus). A nil *Registry no-ops Record/SetQueue/Touch/Remove and returns empty from Snapshot/All, the same nil-safe convention as metrics.Set, so callers never need a nil check around instrumentation.
func NewRegistry ¶
func NewRegistry() *Registry
NewRegistry returns an empty Registry using the real wall clock.
func (*Registry) AllSourcePGNMetrics ¶ added in v1.0.2
func (r *Registry) AllSourcePGNMetrics() map[string][]SourcePGNMetric
AllSourcePGNMetrics returns source metrics keyed by configured source id.
func (*Registry) AttachSourceMetricPersistence ¶ added in v1.0.2
AttachSourceMetricPersistence loads recent source lifecycle events and starts their non-blocking persistence writer.
func (*Registry) CloseSourceMetricPersistence ¶ added in v1.0.2
func (*Registry) RecordConnectorEvent ¶
func (*Registry) RecordSink ¶
func (*Registry) RecordSourceDrops ¶
func (*Registry) RecordStage ¶
RecordStage counts a route-runtime transition without assuming that every terminal transition is confirmed delivery.
func (*Registry) Remove ¶
Remove drops a connector's stats and recent events (deleted connectors), including its queue-depth history ring — the whole *counters entry (totals, rate buckets, and depth ring alike) is deleted as one unit, so there's no separate ring-eviction step to keep in sync with this method.
Ordering contract: callers must ensure the connector's pipeline is fully stopped (Connector.Stop is synchronous — it blocks on the connector's internal WaitGroup) before calling Remove. Record does not distinguish a never-seen id from a just-removed one: get lazily (re)creates a fresh, zeroed *counters entry (empty depth ring included) for any id not currently in the map. So a Record (or SetQueue) call that lands after Remove — from a pipeline that is somehow still running or racing the removal — will silently resurrect the entry under the same id, just reset to zero rather than holding the pre-removal totals/history. It won't reappear with stale data, but it will reappear. Removing only after the pipeline has fully stopped is what prevents that resurrection.
func (*Registry) RemoveSink ¶
func (*Registry) RemoveSource ¶
RemoveSource and RemoveSink drop process-local counters and recent payload events for an entity deleted from config. Disabled or hot-restarted entities deliberately keep their history; the supervisor calls these only after it observes the configured id disappear entirely.
func (*Registry) SetQueue ¶
SetQueue records current queue depth/bytes (from the prune loop) and appends depth to the connector's depth-history ring (see depthRingSize), which Snapshot surfaces as DepthHistory for the UI sparkline. Because every call appends a history sample, SetQueue is for genuine periodic measurements only — presence registration belongs to Touch.
func (*Registry) SetRuntime ¶
func (*Registry) Snapshot ¶
Snapshot returns a connector's current counters/rates and whether it has ever been recorded.
func (*Registry) SourceMetricCapacity ¶ added in v1.2.6
func (r *Registry) SourceMetricCapacity(source string) SourceMetricCapacity
func (*Registry) SourceMetricEvents ¶ added in v1.0.2
func (r *Registry) SourceMetricEvents(source string, limit int) []SourceMetricEvent
func (*Registry) SourcePGNLastPayloadsFiltered ¶ added in v1.2.3
func (r *Registry) SourcePGNLastPayloadsFiltered(source string, filter SourcePGNMetricFilter) []SourcePGNLastPayload
SourcePGNLastPayloadsFiltered returns one retained latest payload per matching sensor/PGN stream, sorted by sender address and PGN for stable MCP output. Oversized omitted bodies are reported by SourcePGNMetric's LatestPayloadTruncated/LatestPayloadTruncations fields instead.
func (*Registry) SourcePGNLastPayloadsFilteredLimit ¶ added in v1.2.6
func (r *Registry) SourcePGNLastPayloadsFilteredLimit(source string, filter SourcePGNMetricFilter, limit int) ([]SourcePGNLastPayload, bool)
SourcePGNLastPayloadsFilteredLimit bounds payload cloning before it occurs. A non-positive limit is unbounded for internal callers.
func (*Registry) SourcePGNMetrics ¶ added in v1.0.2
func (r *Registry) SourcePGNMetrics(source string) []SourcePGNMetric
SourcePGNMetrics returns every observed PGN/sender stream for one source. Results are sorted with active problems first, then by PGN and address.
func (*Registry) SourcePGNMetricsFiltered ¶ added in v1.2.3
func (r *Registry) SourcePGNMetricsFiltered(source string, filter SourcePGNMetricFilter) []SourcePGNMetric
SourcePGNMetricsFiltered returns rich metrics for matching PGN/sender streams. Filtering occurs before snapshot generation so narrow MCP and UI reads do not rebuild diagnostics for unrelated sensors.
func (*Registry) SourcePGNMetricsFilteredLimit ¶ added in v1.2.6
func (r *Registry) SourcePGNMetricsFilteredLimit(source string, filter SourcePGNMetricFilter, limit int) ([]SourcePGNMetric, bool)
SourcePGNMetricsFilteredLimit applies a response bound before deep-copying raw and decoded-field diagnostics. Problems retain the same ordering as the full snapshot, followed by PGN and sender address. A non-positive limit is unbounded for internal callers.
func (*Registry) SourcePGNMetricsForAddress ¶ added in v1.2.3
func (r *Registry) SourcePGNMetricsForAddress(source string, address uint8) []SourcePGNMetric
SourcePGNMetricsForAddress returns rich metrics for one opened device only, avoiding full diagnostic snapshots for unrelated devices.
func (*Registry) SourcePGNSummaries ¶ added in v1.2.3
func (r *Registry) SourcePGNSummaries(source string) []SourcePGNMetric
SourcePGNSummaries returns compact per-stream state for high-frequency UI refreshes. Rich decoded-field and raw-payload diagnostics remain available through SourcePGNMetrics when explicitly requested.
func (*Registry) SubscribeStream ¶ added in v1.1.0
SubscribeStream subscribes to future source-received or sink-sent envelopes. It is an observability tap: publishers never block and drop older preview data when a subscriber cannot keep up. Each message is the canonical three-key consumer envelope JSON.
func (*Registry) Touch ¶
Touch ensures a connector's entry exists (so it shows up in All()/ Snapshot immediately) and zeroes its depth/bytes gauges, WITHOUT appending to the depth-history ring. It exists for Connector.Start's synchronous presence registration: on a hot-apply restart (config edit → supervisor Stop + new Start) the registry entry survives — Remove only fires on delete — so Start seeding via SetQueue(id, 0, 0) would append a genuine 0 mid-history, drawing a dip-to-zero notch in the sparkline that looks like the queue drained and refilled when it did no such thing. History samples must only come from the prune loop's real periodic measurements (SetQueue); Touch covers the "make the connector visible now, real numbers follow within milliseconds" path.
type Snapshot ¶
type Snapshot struct {
TotalMessages int64 `json:"total_messages"`
TotalBytes int64 `json:"total_bytes"`
MsgPerSec float64 `json:"msg_per_sec"` // over last 10s window
BytesPerSec float64 `json:"bytes_per_sec"` // over last 10s window
QueueDepth int64 `json:"queue_depth"`
QueueBytes int64 `json:"queue_bytes"`
RetainedDepth int64 `json:"retained_depth"`
RetainedBytes int64 `json:"retained_bytes"`
QueueCursor int64 `json:"queue_cursor"`
QueueTail int64 `json:"queue_tail"`
OldestPending *time.Time `json:"oldest_pending,omitempty"`
OldestRetained *time.Time `json:"oldest_retained,omitempty"`
LimitMessages int64 `json:"limit_messages,omitempty"`
LimitBytes int64 `json:"limit_bytes,omitempty"`
HeadroomMessages int64 `json:"headroom_messages,omitempty"`
HeadroomBytes int64 `json:"headroom_bytes,omitempty"`
DeliveryClass string `json:"delivery_class,omitempty"`
State string `json:"state,omitempty"`
LastError string `json:"last_error,omitempty"`
Drops int64 `json:"drops"`
StageTotals map[string]int64 `json:"stage_totals,omitempty"`
StageTotalsOverflow int64 `json:"stage_totals_overflow,omitempty"`
PreviewDocumentsOmitted int64 `json:"preview_documents_omitted,omitempty"`
// Source-only diagnostic capacity accounting. Exact source traffic totals
// above continue increasing when a novel PGN/sender cannot be admitted to
// the bounded rich-diagnostics cache.
SourceMetricStreams int `json:"source_metric_streams,omitempty"`
SourceMetricStreamLimit int `json:"source_metric_stream_limit,omitempty"`
SourceMetricGlobalStreams int `json:"source_metric_global_streams,omitempty"`
SourceMetricGlobalLimit int `json:"source_metric_global_limit,omitempty"`
SourceMetricMessagesOmitted int64 `json:"source_metric_messages_omitted,omitempty"`
SourceMetricStreamsExpired int64 `json:"source_metric_streams_expired,omitempty"`
SourceDeviceNamesOmitted int64 `json:"source_device_names_omitted,omitempty"`
// DepthHistory is the last depthRingSize QueueDepth readings, oldest
// first, as recorded by successive SetQueue calls. Absent (nil/omitted
// from JSON) for a connector that has never had SetQueue called on it —
// additive/non-breaking: existing Snapshot consumers (the config API's
// metrics endpoints, the connector detail/dashboard UI fragments) that
// don't know this field simply ignore it.
DepthHistory []int64 `json:"depth_history,omitempty"`
}
Snapshot is the point-in-time view of one connector's counters, returned by Snapshot and All. Rates are computed over the trailing 10-second window; totals never decay.
type SourceMetricCapacity ¶ added in v1.2.6
type SourceMetricCapacity struct {
TrackedStreams int `json:"tracked_streams"`
StreamLimit int `json:"stream_limit"`
GlobalTrackedStreams int `json:"global_tracked_streams"`
GlobalStreamLimit int `json:"global_stream_limit"`
MessagesOmitted int64 `json:"messages_omitted"`
StreamsExpired int64 `json:"streams_expired"`
DeviceNamesOmitted int64 `json:"device_names_omitted"`
PreviewDocsOmitted int64 `json:"preview_documents_omitted"`
}
SourceMetricCapacity describes the bounded rich-diagnostics cache for one configured source. MessagesOmitted does not mean traffic loss: canonical Envelopes and exact source totals continue while only a novel stream's per-PGN diagnostic state is skipped.
type SourceMetricEvent ¶ added in v1.0.2
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 SourcePGNLastPayload ¶ added in v1.2.3
type SourcePGNLastPayload struct {
SourceID string
PGN uint32
PGNName string
SourceAddress uint8
DeviceNameHex string
LastSeen time.Time
PayloadBytes int
Truncated bool
Payload json.RawMessage `json:"-"`
}
SourcePGNLastPayload is the single most recently observed decoded payload for one configured-source/PGN/sender stream. The registry stores only this one payload per bounded stream and omits bodies above maxRetainedDecodedPayloadBytes rather than retaining invalid JSON prefixes. Payload is intentionally excluded from generic JSON/schema reflection so an MCP seam can decode the raw JSON into its true object/array/scalar type.
type SourcePGNMetric ¶ added in v1.0.2
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"`
MissingDecodedOverflow int64 `json:"missing_decoded_fields_overflow,omitempty"`
SourceAddress uint8 `json:"source_address"`
DeviceName *uint64 `json:"device_name,omitempty"`
DeviceNameHex string `json:"device_name_hex,omitempty"`
Messages int64 `json:"messages"`
DiagnosticSamples int64 `json:"diagnostic_samples"`
DiagnosticTruncations int64 `json:"diagnostic_truncations,omitempty"`
LatestPayloadBytes int `json:"latest_payload_bytes,omitempty"`
LatestPayloadTruncated bool `json:"latest_payload_truncated,omitempty"`
LatestPayloadTruncations int64 `json:"latest_payload_truncations,omitempty"`
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"`
PeriodP90Seconds float64 `json:"period_p90_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"`
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"`
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"`
}
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 SourcePGNMetricFilter ¶ added in v1.2.3
SourcePGNMetricFilter limits rich metric snapshots before their field and raw diagnostics are copied and sorted. DeviceNameHex is the stable NMEA 2000 Device NAME without regard to an optional 0x prefix or hex letter case.
Source Files