Help us shape the future of Go! Take the 2026 Go Developer Survey

stats

package
v1.3.102 Latest Latest
Warning

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

 Go to latest Published: Oct 23, 2026 License: AGPL-3.0 Imports: 17 Imported by: 0

Documentation

Overview

Package stats pkg/visor/stats/bitmap.go c3-vis-core

pkg/visor/stats/bitmap.go: 288-bit-per-day uptime bitmap helpers.

Each UTC day is encoded as 288 bits (one per 5-minute slot) in 36 raw bytes. Bit set = the tier or service was online during at least one sampler tick within the slot. The encoding is bit-identical to the format pkg/deployment/tpd/store/redis_uptime.go uses for integrated uptime tracking, so renderers and dashboards can consume either source unchanged.

Package stats pkg/visor/stats/schema.go c3-vis-core

Package stats pkg/visor/stats/sink.go c3-vis-core for mirroring telemetry into an external destination, typically a CXO publisher.

The bbolt store is the source of truth for the visor's local telemetry. The Sink is a write-through hook that lets a publisher (CXO, or any future destination) receive a per-leaf-update view of the same data. The Tracker calls Sink methods after each successful bbolt write so the two views stay in sync without requiring a separate poll loop.

Path conventions (matching §07 of skywire-specs): 

transports/telemetry/<sh>                 → compact sharded binary
                                            telemetry (see pkg/telemetrywire;
                                            <sh> = 2-hex shard 00..0f, the
                                            only per-transport telemetry now
                                            mirrored to the CXO/TPD feed)
transports/<uuid>/<YYYY-MM-DD>/rollup     → that day's rollup (JSON, bbolt-only)
transports/<uuid>/<YYYY-MM-DD>/timeline   → 36-byte uptime bitmap (bbolt-only)
tiers/<tier>/<YYYY-MM-DD>                 → 36-byte bitmap (bbolt-only)
services/<slug>/<YYYY-MM-DD>              → 36-byte bitmap (bbolt-only)

The rollup and timeline are nested under <YYYY-MM-DD> as siblings. Putting the daily JSON at the bare-date leaf ("transports/<uuid>/<date>") would collide with the timeline write ("transports/<uuid>/<date>/timeline") because the treestore is strictly a tree — a node is leaf XOR sub-tree, never both. The /rollup suffix keeps <date> unambiguously a branch.

Sinks are expected to be non-blocking; the Tracker invokes them from the sampler goroutine and does not wait for completion. Implementations that need to do I/O should buffer internally.

Package stats pkg/visor/stats/store.go c3-vis-core

Package stats pkg/visor/stats/tracker.go c3-vis-core

The Tracker owns the periodic sampling loop, holds the in-memory day-start baselines for per-transport bandwidth deltas, and exposes probes that the visor wires into the transport manager, the dmsg init code, and the app discovery manager.

Probes are pull-style: the Tracker queries them on each tick rather than receiving push events. This avoids any callback / event bus machinery and keeps the wiring trivial — the visor just provides closures over its existing fields.

Index

Constants

View Source 
const BitmapSize = SlotsPerDay / 8 // 36

BitmapSize is the byte length of a one-day bitmap.

View Source 
const SchemaVersion = 1

SchemaVersion is the current bbolt layout version. Bumped on any breaking change to bucket layout or record shape.

View Source 
const SlotsPerDay = 24 * 60 / 5 // 288

SlotsPerDay is the number of 5-minute slots in a UTC day.

Variables

View Source 
var ErrSchemaMismatch = errors.New("stats: schema version mismatch")

ErrSchemaMismatch is returned by OpenStore when the persisted schema version does not match the constant SchemaVersion compiled into the binary.

Functions

func And 

func And(a, b []byte) []byte

And returns the bitwise AND of two equal-length bitmaps. Returns nil if the inputs are not both BitmapSize bytes. Used to derive composite views like "process AND dmsg AND skynet" for callers that want a strict online definition.

func GetSlot 

func GetSlot(bm []byte, slot int) bool

GetSlot reports whether the bit for slot is set in bm.

func HydrateSink 

func HydrateSink(store *Store, sink Sink, publishWindowDays int, now time.Time, isLive func(id uuid.UUID) bool) (map[uint8][32]byte, error)

HydrateSink walks the entire bbolt store and pushes the live-transport telemetry within the CXO publish window (the trailing publishWindowDays from now) to the sink as compact sharded binary leaves. Used at startup to seed the publisher's in-memory tree from on-disk state so cold subscribers see the current telemetry immediately, not just samples taken after restart.

isLive gates which transports are broadcast: only transports still live at hydrate time are packed into the shards. The bbolt store retains records for recently-closed transports (inside the retention window) so the visor's own /stats history stays complete, but those dead records must NOT be mirrored to the CXO/TPD telemetry feed — packing them would bloat the Root that TPD fills over a short-lived announce conn. A nil predicate means "all records are live" (used by tests that don't exercise the live gate).

Returns the shard-signature map actually pushed (keyed by shard 0..15), which the caller uses to seed the Tracker's publishedShards so the first sample tick doesn't redundantly re-Put identical shards. The number of leaves pushed is len(the returned map).

func Render 

func Render(bm []byte) string

Render converts a 36-byte bitmap to its 288-character ASCII form, '.' for set bits and ' ' for unset, matching TPD's v3 timeline shape. A zero-length or nil input renders as 288 spaces.

func SetSlot 

func SetSlot(bm []byte, slot int)

SetSlot sets the bit for the given slot index in bm. bm must be at least BitmapSize bytes; the slot index must be in [0, SlotsPerDay). Out-of-range slots are ignored to make the helper safe to call with computed indices at day boundaries.

func SlotForTime 

func SlotForTime(t time.Time) int

SlotForTime returns the 0-based 5-minute slot index for t (in UTC).

Types

type Config 

type Config struct {
	SampleInterval time.Duration // default 1m
	RetentionDays  int           // default 30
	// PublishWindowDays caps the number of trailing days mirrored
	// to the Sink. 0 means "match RetentionDays" — the Sink sees
	// every day the bbolt store keeps. Typical: 7 (publisher
	// exposes one week, store retains 30 days).
	PublishWindowDays int
	Logger            *logging.Logger
}

Config bundles tracker tunables. Zero values mean "use defaults".

type DailyRollup 

type DailyRollup struct {
	Date         string  `json:"date"` // YYYY-MM-DD UTC
	SentBytes    uint64  `json:"sent_bytes"`
	RecvBytes    uint64  `json:"recv_bytes"`
	LatencyMinMS float64 `json:"latency_min_ms,omitempty"`
	LatencyMaxMS float64 `json:"latency_max_ms,omitempty"`
	LatencyAvgMS float64 `json:"latency_avg_ms,omitempty"`
	Samples      uint32  `json:"samples"`
}

DailyRollup is the sealed per-day view. Bandwidth values are deltas computed from the day-start baseline; latency stats are the accumulation of all samples taken within the day.

type LatencyTriple 

type LatencyTriple struct {
	Min, Max, Avg float64
}

LatencyTriple carries the live min/max/avg snapshot. Tracker merges these into today's daily row and overwrites Current.

type LiveSnapshot 

type LiveSnapshot struct {
	SentBytes uint64 `json:"sent_bytes"`
	RecvBytes uint64 `json:"recv_bytes"`
	// ThroughputBps is the transport's passively-observed PEAK goodput
	// (a tracked EWMA of sent+recv bytes/sec), sourced from
	// transport.ManagedTransport.GetThroughputBps(). It is a real
	// capacity metric NOT derivable from the cumulative SentBytes/RecvBytes
	// counters (those give average-over-interval; this is the tracked
	// peak), so it is carried alongside them on the telemetry feed.
	ThroughputBps float64   `json:"throughput_bps,omitempty"`
	LatencyMinMS  float64   `json:"latency_min_ms,omitempty"`
	LatencyMaxMS  float64   `json:"latency_max_ms,omitempty"`
	LatencyAvgMS  float64   `json:"latency_avg_ms,omitempty"`
	SampledAt     time.Time `json:"sampled_at"`
	Type          string    `json:"type,omitempty"`
	// SameNetwork marks a transport to a visor on this one's own network;
	// its bytes are not paid for. Sent as telemetrywire.TypeFlagSameNetwork.
	SameNetwork bool `json:"same_network,omitempty"`
}

LiveSnapshot is the most recent observation for a transport.

Type carries the wire-protocol type ("stcpr"/"sudph"/"stcp"/"dmsg") so TPD's CXO aggregator can record per-transport uptime heartbeats without a redis lookup. Empty on snapshots produced before the field was added — TPD treats that as "skip uptime, but bandwidth and latency still apply" so old visors keep working.

type Probes 

type Probes struct {
	// Transports returns a snapshot of the visor's currently live
	// transports. The Tracker iterates the result and reads bandwidth
	// + latency from each.
	Transports func() []TransportProbe

	// TierStates returns the current state of each named tier:
	// process / dmsg / skynet. Tiers absent from the map are treated
	// as offline. Returning nil disables tier sampling for the tick.
	TierStates func() map[string]bool

	// ServiceStates returns the current state of each registered
	// service slug. Returning nil disables service sampling for the
	// tick.
	ServiceStates func() map[string]bool
}

Probes is the set of pull functions the Tracker calls on each tick. All probes are expected to be non-blocking and safe to call from a dedicated goroutine.

type SampleTx 

type SampleTx struct {
	// contains filtered or unexported fields
}

SampleTx is the per-tick batched store interface handed to the Tracker's sample callback. It wraps a *bbolt.Tx so all of a sample's mutations (PutTransportRecord, MarkTierSlot, MarkServiceSlot, MarkTransportSlot) and reads (GetTransportRecord, TierBitmap, ...) commit in one transaction → one fdatasync per sample instead of one per call. Methods mirror the Store API but without the implicit db.View/Update wrappers.

Lifetime: valid only inside the callback passed to Store.UpdateSample. Reading or writing outside that scope is undefined behavior.

func (*SampleTx) GetTransportRecord 

func (s *SampleTx) GetTransportRecord(id uuid.UUID) (*TransportRecord, error)

GetTransportRecord looks up the persisted record under the open tx. Returns nil + nil if absent (consistent with the standalone Store.GetTransportRecord).

func (*SampleTx) MarkServiceSlot 

func (s *SampleTx) MarkServiceSlot(svc string, date time.Time, slot int) error

MarkServiceSlot is the tx-scoped MarkServiceSlot.

func (*SampleTx) MarkTierSlot 

func (s *SampleTx) MarkTierSlot(tier string, date time.Time, slot int) error

MarkTierSlot is the tx-scoped MarkTierSlot.

func (*SampleTx) MarkTransportSlot 

func (s *SampleTx) MarkTransportSlot(tpID string, date time.Time, slot int) error

MarkTransportSlot is the tx-scoped MarkTransportSlot.

func (*SampleTx) PutTransportRecord 

func (s *SampleTx) PutTransportRecord(rec *TransportRecord) error

PutTransportRecord overwrites the persisted record. Mirrors Store.PutTransportRecord but uses the open tx.

func (*SampleTx) ServiceBitmap 

func (s *SampleTx) ServiceBitmap(svc string, date time.Time) []byte

ServiceBitmap is the tx-scoped ServiceBitmap.

func (*SampleTx) TierBitmap 

func (s *SampleTx) TierBitmap(tier string, date time.Time) []byte

TierBitmap returns a copy of the tier bitmap under the open tx.

func (*SampleTx) TransportBitmap 

func (s *SampleTx) TransportBitmap(tpID string, date time.Time) []byte

TransportBitmap is the tx-scoped TransportBitmap.

type Sink 

type Sink interface {
	Put(path string, value []byte)
	Delete(path string)
	PutBatch(ops []SinkOp)
}

Sink receives per-path writes mirroring the bbolt store. Methods are called from the Tracker's sampler goroutine; implementations that want to do network I/O should defer it to their own loop.

The Sink is best-effort: a returning error is not propagated to the caller. Implementations should log internally if needed.

PutBatch collapses a slice of (path, value) ops into one sink call. Sinks that proxy to a contended downstream (the CXO publisher, whose mutex is also taken by transport-manager re-registers and other publishers) MUST implement this in a way that acquires the downstream's lock once for the entire batch — otherwise sampler ticks with many mirrors (e.g. one transport tier × N services) serialize against every other writer.

type SinkOp 

type SinkOp struct {
	Path  string
	Value []byte // nil = delete
}

SinkOp is a single (path, value) pair for a batched Put. A nil Value means "delete this path" — folded into PutBatch so the sampler can hand the sink a single fan-out operation per tick instead of N round-trips.

type Store 

type Store struct {
	// contains filtered or unexported fields
}

Store is the bbolt-backed persistence layer for the visor's local telemetry. Safe for concurrent use; bbolt serializes writes internally and the Store wraps reads in its own transactions.

func OpenStore 

func OpenStore(path string) (*Store, error)

OpenStore opens (or creates) the bbolt database at path and ensures the four top-level buckets exist. Schema version is stamped on first run; on subsequent opens the version is verified — a mismatch returns ErrSchemaMismatch so the caller can decide whether to migrate or refuse.

Before the open OpenStore runs an integrity check on the existing file (if any). bbolt's own consistency-check (tx.Check) walks pages + freelist + bucket inodes and surfaces problems like the "circular dependency" assertion that bbolt would otherwise panic on during a later commit — a panic that fires in bbolt's internal batch goroutine and is therefore not recoverable from caller code. Treating a corrupt stats DB as "delete + recreate" is safe because the store is local telemetry; everything it holds is regenerated from runtime samples.

func (*Store) AllTransportRecords 

func (s *Store) AllTransportRecords() ([]*TransportRecord, error)

AllTransportRecords returns every persisted transport record. Used by the HTTP /stats/transports/history endpoint and by the CXO publisher for initial-state hydration.

func (*Store) Close 

func (s *Store) Close() error

func (*Store) DB 

func (s *Store) DB() *bbolt.DB

Close releases the bbolt file handle. DB is the store's database, for other local telemetry kept beside it.

func (*Store) DeleteTransportRecord 

func (s *Store) DeleteTransportRecord(id uuid.UUID) error

DeleteTransportRecord removes a transport entry. Called by the retention sweep for closed transports with no recent activity.

func (*Store) GetTransportRecord 

func (s *Store) GetTransportRecord(id uuid.UUID) (*TransportRecord, error)

GetTransportRecord returns the persisted record for a transport, or nil if no record has been written yet.

func (*Store) MarkServiceSlot 

func (s *Store) MarkServiceSlot(service string, date time.Time, slot int) error

MarkServiceSlot is the per-service analog of MarkTierSlot.

func (*Store) MarkTierSlot 

func (s *Store) MarkTierSlot(tier string, date time.Time, slot int) error

MarkTierSlot ensures the bitmap key for tier/date exists and sets the bit for slot. The read-modify-write happens inside a single bbolt transaction, so concurrent ticks for the same tier/date can safely race — the OR is idempotent.

func (*Store) MarkTransportSlot 

func (s *Store) MarkTransportSlot(tpID string, date time.Time, slot int) error

MarkTransportSlot is the per-transport analog of MarkTierSlot. The transport ID (UUID string) is the sub-bucket key. Same wire format as the tier/service bitmaps — 36 raw bytes, MSB-first, 288 5-min slots — and bit-identical to TPD's tp-uptime:<id>:<date>:timeline redis bitmap so the two views can be ORed without translation.

func (*Store) PruneBitmaps 

func (s *Store) PruneBitmaps(cutoff time.Time) (int, error)

PruneBitmaps deletes tier and service bitmap keys whose date is strictly older than cutoff. Returns the number of keys removed. Called from the retention sweep at UTC-midnight.

func (*Store) PutTransportRecord 

func (s *Store) PutTransportRecord(rec *TransportRecord) error

PutTransportRecord overwrites the persisted record for a transport. Callers (the Tracker) read-modify-write under their own mutex; the store does not merge.

func (*Store) ServiceBitmap 

func (s *Store) ServiceBitmap(service string, date time.Time) ([]byte, error)

ServiceBitmap is the per-service analog of TierBitmap.

func (*Store) ServiceDates 

func (s *Store) ServiceDates(service string) ([]string, error)

ServiceDates is the per-service analog of TierDates.

func (*Store) ServiceNames 

func (s *Store) ServiceNames() ([]string, error)

ServiceNames is the per-service analog of TierNames.

func (*Store) TierBitmap 

func (s *Store) TierBitmap(tier string, date time.Time) ([]byte, error)

TierBitmap returns the persisted bitmap for tier on the given UTC date, or a zero-filled bitmap if none exists. The result is a copy; callers may mutate it freely.

func (*Store) TierDates 

func (s *Store) TierDates(tier string) ([]string, error)

TierDates lists the UTC dates that have a stored bitmap for tier, sorted ascending. Used by query endpoints that need a date range.

func (*Store) TierNames 

func (s *Store) TierNames() ([]string, error)

TierNames lists every tier that has any persisted data. Used by listing endpoints (`GET /stats/uptime` with no filter returns all).

func (*Store) TransportBitmap 

func (s *Store) TransportBitmap(tpID string, date time.Time) ([]byte, error)

TransportBitmap is the per-transport analog of TierBitmap.

func (*Store) TransportBitmapDates 

func (s *Store) TransportBitmapDates(tpID string) ([]string, error)

TransportBitmapDates is the per-transport analog of TierDates.

func (*Store) TransportBitmapIDs 

func (s *Store) TransportBitmapIDs() ([]string, error)

TransportBitmapIDs lists every transport ID with a stored bitmap. Returned values are the UUID strings, suitable for re-keying the sink path (transports/<id>/<date>/timeline).

func (*Store) UpdateSample 

func (s *Store) UpdateSample(fn func(*SampleTx) error) error

UpdateSample runs fn inside a single bbolt write transaction. Returns the commit error or the fn's error (whichever fires first).

Used by the Tracker to coalesce a full sample tick (N transports + M tiers + S services worth of writes) into one commit. Replaces what used to be 2N + M + S separate Update transactions per sample — each of which fsynced — with one.

type Tracker 

type Tracker struct {
	// contains filtered or unexported fields
}

Tracker periodically samples transport bandwidth/latency and tier and service uptime, persisting rollups into a Store.

func NewTracker 

func NewTracker(store *Store, probes Probes, conf Config) *Tracker

NewTracker constructs a Tracker but does not start the sample loop; call Run to begin. The store is owned by the Tracker for the lifetime of Run; Close releases it.

func (*Tracker) Close 

func (t *Tracker) Close() error

Close stops the sampler and closes the store. Safe to call multiple times.

func (*Tracker) Run 

func (t *Tracker) Run(ctx context.Context)

Run starts the sampler. Returns immediately; sampling continues until ctx is canceled or Close is called. Idempotent: a second Run on the same tracker is a no-op.

func (*Tracker) SeedPublishedShards 

func (t *Tracker) SeedPublishedShards(sigs map[uint8][32]byte)

SeedPublishedShards records the shard signatures the sink was already primed with before the sampler started — the live set hydrated by HydrateSink at startup (see the visor's seedSinkFromStore). Seeding them means the first sample tick does NOT redundantly re-Put every shard that hydrate already published with identical content; a shard only re-Puts once one of its transports actually changes. Call before Run. The passed map is adopted directly.

func (*Tracker) SetSink 

func (t *Tracker) SetSink(s Sink)

SetSink replaces the Tracker's mirror sink. Pass nil to detach (resets to a no-op sink). Safe to call before or after Run.

func (*Tracker) Store 

func (t *Tracker) Store() *Store

Store returns the underlying bbolt-backed store. Exposed so HTTP handlers and the CXO publisher can read snapshots without going through the tracker (which is sample-loop-only).

type TransportProbe 

type TransportProbe struct {
	ID            uuid.UUID
	Edges         []cipher.PubKey
	Type          string
	Label         string
	SentBytes     uint64
	RecvBytes     uint64
	ThroughputBps float64
	LatencyMS     LatencyTriple
	// SameNetwork marks a transport to a visor on this one's own network
	// (transport.ManagedTransport.SameNetwork).
	SameNetwork bool
	// Sent and Recv are the stream digest checkpoints of the current conn.
	Sent, Recv telemetrywire.DirCheckpoints
}

TransportProbe is the per-transport view the Tracker needs each tick. The visor adapter constructs these by walking the transport manager — keeping the type local insulates the package from the wider transport API surface.

type TransportRecord 

type TransportRecord struct {
	ID        uuid.UUID       `json:"id"`
	Edges     []cipher.PubKey `json:"edges"`
	Type      string          `json:"type"`
	Label     string          `json:"label,omitempty"`
	FirstSeen time.Time       `json:"first_seen"`
	LastSeen  time.Time       `json:"last_seen"`
	Current   *LiveSnapshot   `json:"current,omitempty"`
	Daily     []DailyRollup   `json:"daily,omitempty"`
}

TransportRecord is the persisted per-transport rollup state. Stored in the `transports` bucket keyed by the transport ID (UUID, 16 bytes).

Current is overwritten on every sample with the live snapshot; Daily is append-only with one entry per UTC day. The day-start cumulative baseline is held in memory by the Tracker, not persisted — on restart the next sample seeds a fresh baseline and the partial day's loss is bounded by StatsSampleInterval.

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.