stats

package
v1.0.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 19, 2026 License: Apache-2.0 Imports: 16 Imported by: 0

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

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type BaselineField added in v1.0.2 

type BaselineField struct {
	Minimum float64 `json:"minimum"`
	Maximum float64 `json:"maximum"`
	Unit    string  `json:"unit,omitempty"`
}

type BaselineRawByte added in v1.0.2 

type BaselineRawByte struct {
	Offset            int    `json:"offset"`
	Minimum           uint8  `json:"minimum"`
	Maximum           uint8  `json:"maximum"`
	ChangedBitMaskHex string `json:"changed_bit_mask_hex"`
}

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"`
	MostCommonShare   float64 `json:"most_common_share"`
	EntropyBits       float64 `json:"entropy_bits"`
	ChangedShare      float64 `json:"changed_share"`
	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"`
	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"`
}

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 PayloadFingerprint struct {
	Fingerprint string    `json:"fingerprint"`
	Count       int64     `json:"count"`
	Share       float64   `json:"share"`
	Length      int       `json:"length"`
	LastSeen    time.Time `json:"last_seen"`
}

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

func (r *Registry) All() map[string]Snapshot

All returns every tracked connector's current snapshot, keyed by id.

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 

func (r *Registry) AttachSourceMetricPersistence(ctx context.Context, db *sql.DB) error

AttachSourceMetricPersistence loads baselines and recent change events, then starts a non-blocking event writer. Call this before serving baseline APIs.

func (*Registry) ClearSourceTrafficBaseline added in v1.0.2 

func (r *Registry) ClearSourceTrafficBaseline(ctx context.Context, source string) error

func (*Registry) CloseSourceMetricPersistence added in v1.0.2 

func (r *Registry) CloseSourceMetricPersistence(ctx context.Context) error

func (*Registry) CommitSourceTrafficBaseline added in v1.0.2 

func (r *Registry) CommitSourceTrafficBaseline(ctx context.Context, source string) ([]SourceTrafficBaseline, error)

func (*Registry) Recent 

func (r *Registry) Recent(kind, id string, limit int) []Event

func (*Registry) Record 

func (r *Registry) Record(connector string, msgs, bytes int64)

Record adds boundary-accepted messages/bytes for a connector at time now.

func (*Registry) RecordConnectorEvent 

func (r *Registry) RecordConnectorEvent(connector, stage string, e *msg.Envelope)

func (*Registry) RecordSink 

func (r *Registry) RecordSink(sink, connector string, e *msg.Envelope)

func (*Registry) RecordSource 

func (r *Registry) RecordSource(source string, e *msg.Envelope)

func (*Registry) RecordSourceDrops 

func (r *Registry) RecordSourceDrops(source string, n int64)

func (*Registry) RecordStage 

func (r *Registry) RecordStage(connector, stage string, n int64)

RecordStage counts a route-runtime transition without assuming that every terminal transition is confirmed delivery.

func (*Registry) Remove 

func (r *Registry) Remove(connector string)

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 (r *Registry) RemoveSink(sink string)

func (*Registry) RemoveSource 

func (r *Registry) RemoveSource(source string)

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 

func (r *Registry) SetQueue(connector string, depth, bytes int64)

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

func (r *Registry) SetQueueStats(connector string, s queue.Stats)

func (*Registry) SetRuntime 

func (r *Registry) SetRuntime(connector, deliveryClass, state string, err error)

func (*Registry) SinkSnapshot 

func (r *Registry) SinkSnapshot(sink string) (Snapshot, bool)

func (*Registry) Snapshot 

func (r *Registry) Snapshot(connector string) (Snapshot, bool)

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

func (r *Registry) SourceSnapshot(source string) (Snapshot, bool)

func (*Registry) SourceTrafficBaselines added in v1.0.2 

func (r *Registry) SourceTrafficBaselines(source string) []SourceTrafficBaseline

func (*Registry) Touch 

func (r *Registry) Touch(connector string)

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"`
	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"`
}

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 SourceTrafficBaseline added in v1.0.2 

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"`
}

SourceTrafficBaseline is an operator-approved expectation for one PGN from one stable Device NAME (or source address when no NAME is available).

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.