Documentation
¶
Overview ¶
Package observability is the in-process EVENT BUS — not the logging package. For structured logging, request IDs and the helpers you use on every app, see the sibling package observe; this one is what admin panels subscribe to. The bus exposes runtime activity (HTTP requests, SQL statements, session changes, custom application events) to optional observers — most importantly, the orbit module (github.com/jcsvwinston/orbit), which consumes this bus via Runtime.Observability() to power its live admin feed.
The package is the "core" Phase-2 deliverable of the admin refactor. It is owned by the framework's hot path; correctness, lock-discipline, and idle cost matter more than ergonomics.
Architecture invariants ¶
The Bus is a process-wide fan-out: one publisher (Emit) → N subscribers (chan Event). It does NOT spawn goroutines. Each subscriber drains its own channel from its own goroutine.
The hot path on the publisher side has a hard "zero-cost when nobody is watching" requirement. Bus.HasSubscribers(kind) is a single atomic load (target: < 5 ns on x86). Hooks MUST gate all event construction on it before allocating, copying request/query bodies, or doing any other instrumentation work.
Events are pooled via sync.Pool to keep allocations off the hot path when an operator IS watching. Concrete event types embed a refcount; the bus increments the refcount per delivery target and decrements on drop or successful subscriber Release(). When the refcount reaches zero, the event resets and returns to its pool.
The bus uses non-blocking sends (select with default) so a slow subscriber cannot stall the framework. When a subscriber's channel is full, the event for that subscriber is dropped and counted in Bus.Stats(kind).Dropped. The publisher never blocks.
Events are immutable once Emit returns ownership to the bus. Both the publisher and the subscriber MUST treat the event as read-only after that point. The only legal mutation is Release(), which is safe to call once per acquired reference.
Ownership rules (read carefully before writing producer code) ¶
The producer calls AcquireXxxEvent() to obtain an event with refcount=1 from the pool. The producer fills exported fields and calls Bus.Emit(e). At that point the producer transfers ownership to the bus. The producer MUST NOT touch the event after Emit returns.
The bus increments the refcount once per delivery target (subscriber whose Filter matches). It then attempts a non-blocking send for each target. On a successful send the consumer owns one reference. On drop the bus calls Release immediately on behalf of that target.
The original reference the producer transferred is released by the bus after fanout. So the bus never holds a long-term reference; it merely forwards ownership.
Each consumer reads its event from the channel, processes it, and calls Release() exactly once. Calling Release more times than references are held panics — that is by design, to surface lifecycle bugs in tests.
When there are no subscribers, the bus calls Release on the producer's reference and returns immediately. The producer never observes that "no one is listening"; it only observes that Emit took ownership.
Hook responsibilities ¶
Hooks (HTTP middleware, SQL observer, session activity recorder) live in the hooks subpackage. Each hook starts every code path with:
if !bus.HasSubscribers(observability.KindHTTPRequest) {
next.ServeHTTP(w, r)
return
}
and only allocates the event after that gate is open. Hooks are also responsible for sanitizing user-supplied data: query strings are redacted at the source, SQL argument values become "type(len):***" markers rather than raw values, etc. The bus carries pre-sanitized strings to keep the admin server stateless about what is sensitive.
Threading model summary ¶
Producer goroutine: AcquireXxx → fill → Bus.Emit → Bus.Emit returns
[Bus internal: read subscribers under RLock, refcount++ per target,
non-blocking send, release on drop, release original ref]
Subscriber goroutine: <-sub.Ch() → process → Event.Release()
No goroutine is spawned by the bus or by Subscribe. Cancellation removes the subscription from the bus map; the subscriber's channel is leaked (drained by the runtime) so that an in-flight Emit racing with cancel cannot panic by sending on a closed channel.
Index ¶
Constants ¶
const DefaultSubscriberChannelSize = 256
DefaultSubscriberChannelSize is the default per-subscription channel capacity. Subscribers that fall behind by more than this many events see their excess events dropped (counted in Bus.Stats(kind).Dropped).
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Bus ¶
type Bus struct {
// contains filtered or unexported fields
}
Bus is the in-process observability fan-out. It is safe for concurrent use and never spawns goroutines. See doc.go for the full ownership and threading model.
func NewBus ¶
NewBus returns a fresh Bus. The logger is used only for unexpected non-fatal conditions (a Subscribe with a closed bus, etc.). Pass nil to silence; the bus will fall back to slog.Default on the rare occasions it logs.
func (*Bus) Emit ¶
Emit publishes the event to every matching subscriber. Ownership of e transfers to the bus on entry; the caller MUST NOT touch the event after Emit returns. See doc.go for the full lifecycle.
Emit is safe to call from any goroutine.
func (*Bus) HasSubscribers ¶
HasSubscribers returns true if at least one subscription matches events of the given kind. This is the hot-path gate that hooks call BEFORE constructing an event. It is one atomic load on the read side and is designed to be inlined by the compiler.
HasSubscribers is conservative: it answers "would this kind have any subscriber" without applying NodeID filters. That is fine — false positives only cost the producer the construction of an event that the bus then drops cheaply via Filter.Matches.
func (*Bus) Subscribe ¶
func (b *Bus) Subscribe(filter Filter, opts *SubscribeOptions) (*Subscription, func())
Subscribe returns a new Subscription. The caller drains sub.Ch() from a dedicated goroutine and calls sub.Cancel() (or its returned cancel function) exactly once when done.
SubscribeOptions tune buffer size and other knobs. Pass nil for defaults.
func (*Bus) SubscriberCount ¶
SubscriberCount returns the total number of live subscriptions across all kinds. Use Stats(kind) for per-kind counts.
type CustomEvent ¶
type CustomEvent struct {
Name string
Labels map[string]string
Payload []byte
ContentType string
// contains filtered or unexported fields
}
CustomEvent is the application-defined extension point. Use it for domain-specific signals that do not fit one of the framework-managed kinds. Payload is opaque bytes; ContentType helps the UI render it.
func AcquireCustomEvent ¶
func AcquireCustomEvent(emittedAt time.Time, nodeID string) *CustomEvent
AcquireCustomEvent returns a zero-valued event with refcount=1 from the pool. The Payload slice has spare capacity for append.
func (*CustomEvent) EmittedAt ¶
EmittedAt and NodeID are shared across all events through the embedded baseEvent. Concrete types do not redefine these.
func (*CustomEvent) Kind ¶
func (e *CustomEvent) Kind() EventKind
func (*CustomEvent) Release ¶
func (e *CustomEvent) Release()
type Event ¶
type Event interface {
// Kind returns the type of this event. The publisher and subscribers may
// switch on Kind to type-assert to the concrete struct.
Kind() EventKind
// EmittedAt returns the wall-clock timestamp the producer set on the
// event before calling Bus.Emit. Subscribers MUST NOT mutate this.
EmittedAt() time.Time
// NodeID returns the framework process identifier the event came from.
// May be empty during local development.
NodeID() string
// Release decrements the refcount by one. When the refcount reaches
// zero, the event is reset to its zero value and returned to its
// sync.Pool. Calling Release more times than references are held
// panics; this is an intentional crash-on-bug.
Release()
// contains filtered or unexported methods
}
Event is the sealed interface every observability event satisfies. It is sealed (private method) so the framework can guarantee that every event passing through Bus.Emit was produced via Acquire* and is therefore part of the refcount + sync.Pool lifecycle. External packages construct events with the exported AcquireXxxEvent helpers.
type EventKind ¶
type EventKind uint8
EventKind classifies events for routing, sampling, and per-type metrics. Values are stable: never reorder, never reuse. New kinds append at the end and bump numEventKinds. KindUnknown is the zero value to surface accidents.
type Filter ¶
type Filter struct {
// Kinds restricts to these event kinds. Empty = all kinds.
Kinds []EventKind
// NodeIDs restricts to events from these node identifiers. Empty = all
// nodes. Useful for the admin server when it has aggregated events from
// many agents and the UI only wants one node's view.
NodeIDs []string
}
Filter narrows what a subscriber wants to receive from the bus. Zero value (Filter{}) means "everything". Filter values are immutable once a subscription has been created — copy-on-write if you need to change them.
type HTTPRequestEvent ¶
type HTTPRequestEvent struct {
Method string
Path string
Status int
Duration time.Duration
RequestID string
TraceID string
UserID string
RemoteIP string
UserAgent string
PayloadPreview string
// contains filtered or unexported fields
}
HTTPRequestEvent describes a completed HTTP request handled by the framework's router. Fields are pre-sanitized: PayloadPreview is a redacted summary of the request body or query string; sensitive query keys (key, secret, password, token) are starred at the producer.
func AcquireHTTPRequestEvent ¶
func AcquireHTTPRequestEvent(emittedAt time.Time, nodeID string) *HTTPRequestEvent
AcquireHTTPRequestEvent returns a zero-valued event with refcount=1 from the pool. The caller fills exported fields and the embedded baseEvent's timestamp + node id, then transfers ownership to Bus.Emit.
func (*HTTPRequestEvent) EmittedAt ¶
EmittedAt and NodeID are shared across all events through the embedded baseEvent. Concrete types do not redefine these.
func (*HTTPRequestEvent) Release ¶
func (e *HTTPRequestEvent) Release()
Release implements Event. It returns the event to httpPool when the last reference is released.
type RingBuffer ¶
type RingBuffer[T any] struct { // contains filtered or unexported fields }
RingBuffer is a bounded, drop-oldest buffer for retained event copies. It is the per-kind "replay" store the agent and admin server use so a freshly opened panel sees recent activity instead of an empty stream.
RingBuffer holds COPIES of event payloads, not references to pooled events. It therefore does NOT participate in the refcount lifecycle: a caller who wants to retain an event past Bus.Emit must copy the relevant fields out into a RingBufferEntry-shaped value and pass that to Push.
RingBuffer is goroutine-safe. Push and Snapshot may be called from different goroutines.
func NewRingBuffer ¶
func NewRingBuffer[T any](capacity int) *RingBuffer[T]
NewRingBuffer constructs an empty ring buffer with the given capacity. Capacity must be > 0; non-positive values default to 64.
func (*RingBuffer[T]) Capacity ¶
func (r *RingBuffer[T]) Capacity() int
Capacity returns the configured maximum number of entries.
func (*RingBuffer[T]) Dropped ¶
func (r *RingBuffer[T]) Dropped() uint64
Dropped returns the cumulative number of entries that were overwritten because the buffer was full at the time of Push.
func (*RingBuffer[T]) Len ¶
func (r *RingBuffer[T]) Len() int
Len returns the current number of buffered entries.
func (*RingBuffer[T]) Push ¶
func (r *RingBuffer[T]) Push(at time.Time, data T)
Push records the data at the given timestamp. If the buffer is full the oldest entry is dropped (drop-oldest), and the dropped counter is bumped. Push never blocks.
func (*RingBuffer[T]) Snapshot ¶
func (r *RingBuffer[T]) Snapshot(limit int) []T
Snapshot returns up to limit items, newest first. It allocates a fresh slice; callers may retain the result without further synchronization. limit <= 0 returns an empty slice.
type SQLStatementEvent ¶
type SQLStatementEvent struct {
ModelName string
Operation string
Query string
Args []string
Duration time.Duration
Err string
// RowsAffected is the driver-reported row count for exec-style
// operations; 0 means "not reported" (SELECTs, unsupported drivers).
RowsAffected int64
RequestID string
TraceID string
UserID string
// contains filtered or unexported fields
}
SQLStatementEvent describes a SQL query executed by the framework's CRUD layer. Args is pre-sanitized: each entry is a "type(len):***" marker for strings/bytes; primitives are formatted as "type:value".
func AcquireSQLStatementEvent ¶
func AcquireSQLStatementEvent(emittedAt time.Time, nodeID string) *SQLStatementEvent
AcquireSQLStatementEvent returns a zero-valued event with refcount=1 from the pool. The Args slice has spare capacity ready for append.
func (*SQLStatementEvent) EmittedAt ¶
EmittedAt and NodeID are shared across all events through the embedded baseEvent. Concrete types do not redefine these.
func (*SQLStatementEvent) Kind ¶
func (e *SQLStatementEvent) Kind() EventKind
func (*SQLStatementEvent) Release ¶
func (e *SQLStatementEvent) Release()
type SessionChangeEvent ¶
type SessionChangeEvent struct {
Change SessionChangeKind
TokenShort string
UserID string
IP string
UserAgent string
LastRoute string
TraceID string
// contains filtered or unexported fields
}
SessionChangeEvent describes a lifecycle event on a user session. The session token never travels in cleartext; only TokenShort (first 8 chars) goes on the wire.
func AcquireSessionChangeEvent ¶
func AcquireSessionChangeEvent(emittedAt time.Time, nodeID string) *SessionChangeEvent
AcquireSessionChangeEvent returns a zero-valued event with refcount=1 from the pool.
func (*SessionChangeEvent) EmittedAt ¶
EmittedAt and NodeID are shared across all events through the embedded baseEvent. Concrete types do not redefine these.
func (*SessionChangeEvent) Kind ¶
func (e *SessionChangeEvent) Kind() EventKind
func (*SessionChangeEvent) Release ¶
func (e *SessionChangeEvent) Release()
type SessionChangeKind ¶
type SessionChangeKind uint8
SessionChangeKind narrows a session change to one of three discrete events. The zero value SessionChangeUnspecified is illegal in published events.
const ( SessionChangeUnspecified SessionChangeKind = 0 SessionChangeCreated SessionChangeKind = 1 SessionChangeTouched SessionChangeKind = 2 SessionChangeDestroyed SessionChangeKind = 3 )
func (SessionChangeKind) String ¶
func (k SessionChangeKind) String() string
type SubscribeOptions ¶
type SubscribeOptions struct {
// ChannelSize is the per-subscriber channel capacity. Larger buffers
// tolerate slower consumers but trade memory and worst-case latency.
// Default: DefaultSubscriberChannelSize.
ChannelSize int
}
SubscribeOptions tunes how a subscription buffers and behaves under pressure. Zero value is safe (means defaults).
type Subscription ¶
type Subscription struct {
// contains filtered or unexported fields
}
Subscription is a handle to a live subscription. It carries the channel the consumer drains and a Cancel that removes the subscription from the bus. Cancel is safe to call multiple times; subsequent calls are no-ops.
func (*Subscription) Cancel ¶
func (s *Subscription) Cancel()
Cancel removes the subscription from the bus and decrements the per-kind counters. It does NOT close the channel: closing while Bus.Emit may be holding a reference would race; instead the channel is leaked and GC'd once nothing holds it.
Pending events still buffered in the channel can be drained or ignored at the consumer's discretion. If they are drained, each one MUST still be Released to return the underlying memory to its pool.
func (*Subscription) Ch ¶
func (s *Subscription) Ch() <-chan Event
Ch returns the channel the subscriber drains. Each event read from Ch MUST be Released by the consumer exactly once.
func (*Subscription) Filter ¶
func (s *Subscription) Filter() Filter
Filter returns a copy of the filter this subscription was created with.
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
Package hooks plugs the framework's existing instrumentation points (HTTP middleware, SQL observer, session manager) into the observability.Bus so the agent (and direct subscribers) can receive strongly typed events.
|
Package hooks plugs the framework's existing instrumentation points (HTTP middleware, SQL observer, session manager) into the observability.Bus so the agent (and direct subscribers) can receive strongly typed events. |