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) SourceMetricEvents(source string, limit int) []SourceMetricEvent
- func (r *Registry) SourcePGNMetrics(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 SourceMetricEvent
- type SourcePGNMetric
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"`
}
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"`
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 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"`
}
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) SourceMetricEvents ¶ added in v1.0.2
func (r *Registry) SourceMetricEvents(source string, limit int) []SourceMetricEvent
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) 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 a preview copy 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"`
// 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 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 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"`
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"`
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.
Source Files