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 Event
- type Registry
- func (r *Registry) All() map[string]Snapshot
- 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) SourceSnapshot(source string) (Snapshot, bool)
- func (r *Registry) Touch(connector string)
- type Snapshot
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
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 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) 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) 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.
Source Files