routing

package
v0.11.6 Latest Latest
Warning

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

 Go to latest Published: Sep 12, 2026 License: Apache-2.0 Imports: 9 Imported by: 0

Documentation

Overview

Package routing implements the server-side fanout: it receives proto events from agents (via the AgentService handler) and republishes them to UI subscribers (via the ControlService.StreamEvents handler).

The fanout is intentionally simpler than pkg/observability.Bus:

  • Producers send copies, not pooled refcounted events. Once an event reaches the server, it's already off the agent's hot path; the allocation cost is negligible compared to the network IO.
  • Drop policy: per-subscription channel, drop-newest on overflow. Counted in EventBus.Stats.
  • Filter and per-kind sampling rate are evaluated at fanout time (each subscription has its own filter). This keeps the AGENT-side subscription a single union of all UI demands; the server narrows per UI.

Index

Constants

This section is empty.

Variables

View Source 
var ErrDataStudioInflightOverflow = errors.New("admin server: too many in-flight data studio requests")

ErrDataStudioInflightOverflow is returned when too many simultaneous data studio requests are pending.

View Source 
var ErrDataStudioTimeout = errors.New("admin server: data studio request timed out")

ErrDataStudioTimeout is returned when an agent does not answer a DataStudioRequest within the configured timeout.

View Source 
var ErrRbacInflightOverflow = errors.New("admin server: too many in-flight rbac requests")

ErrRbacInflightOverflow is returned when too many simultaneous rbac requests are pending.

View Source 
var ErrRbacTimeout = errors.New("admin server: rbac request timed out")

ErrRbacTimeout is returned when an agent does not answer an RbacRequest within the configured timeout.

View Source 
var ErrSnapshotInflightOverflow = errors.New("admin server: too many in-flight snapshot requests")

ErrSnapshotInflightOverflow is returned when too many simultaneous snapshot requests are pending. This guards against a misbehaving UI flooding the request channel.

View Source 
var ErrSnapshotTimeout = errors.New("admin server: snapshot request timed out")

ErrSnapshotTimeout is returned when an agent does not answer a SnapshotRequest within the configured timeout.

Functions

func Wait 

func Wait(ch <-chan *adminv1.SnapshotResponse, cancel func(), timeout time.Duration) (*adminv1.SnapshotResponse, error)

Wait blocks for a response on ch up to the deadline. Cancel is the cleanup func from Begin (always run, even on success).

func WaitDataStudio 

func WaitDataStudio(ch <-chan *adminv1.DataStudioResponse, cancel func(), timeout time.Duration) (*adminv1.DataStudioResponse, error)

WaitDataStudio blocks for a response on ch up to the deadline. Cancel is the cleanup func from Begin (always run, even on success).

func WaitRbac 

func WaitRbac(ch <-chan *adminv1.RbacResponse, cancel func(), timeout time.Duration) (*adminv1.RbacResponse, error)

WaitRbac blocks for a response on ch up to the deadline. Cancel is the cleanup func from Begin (always run, even on success).

Types

type AuditEntry 

type AuditEntry struct {
	Time   time.Time
	Actor  string
	Action string // e.g. "datastudio.create"
	Target string // human-readable, e.g. `Article #42 ("default")`
	NodeID string
}

AuditEntry is one fleet-plane action: something an operator did THROUGH the admin server (Data Studio mutations, and future manage actions). Per-app admin actions stay in each node's in-process Orbit audit ring; this ring covers the fleet plane the server itself routes.

type AuditRing 

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

AuditRing is a bounded, in-memory, newest-wins ring of fleet-plane audit entries. Same discipline as event replay: never persisted, drop-oldest on overflow.

func NewAuditRing 

func NewAuditRing(capacity int) *AuditRing

NewAuditRing constructs a ring. capacity <= 0 defaults to 2048.

func (*AuditRing) Append 

func (r *AuditRing) Append(e AuditEntry)

Append records one entry, evicting the oldest when full.

func (*AuditRing) Len 

func (r *AuditRing) Len() int

Len is intended for tests.

func (*AuditRing) List 

func (r *AuditRing) List(limit int) []AuditEntry

List returns up to limit entries, newest first. limit <= 0 means all.

type DataStudioRouter 

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

DataStudioRouter correlates server-issued DataStudioRequest frames with the DataStudioResponse frames the agent sends back over the same bidi stream. Mirrors the SnapshotRouter pattern.

func NewDataStudioRouter 

func NewDataStudioRouter(maxAlive int) *DataStudioRouter

NewDataStudioRouter constructs a router. maxAlive caps the total number of pending requests across all agents; values <= 0 default to 256.

func (*DataStudioRouter) Begin 

func (r *DataStudioRouter) Begin() (id string, ch chan *adminv1.DataStudioResponse, cancel func(), err error)

Begin allocates a fresh request_id and registers a channel. The returned cancel MUST be called even on success to release the slot.

func (*DataStudioRouter) PendingCount 

func (r *DataStudioRouter) PendingCount() int

PendingCount is intended for tests / metrics.

func (*DataStudioRouter) Resolve 

func (r *DataStudioRouter) Resolve(resp *adminv1.DataStudioResponse) bool

Resolve delivers a DataStudioResponse to the awaiting Begin caller (if any). Returns true when a pending request consumed the response.

type EventBus 

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

EventBus is the server-side fanout from agents to UIs.

func NewEventBus 

func NewEventBus() *EventBus

NewEventBus constructs an EventBus.

func (*EventBus) AggregateFilter 

func (b *EventBus) AggregateFilter() *adminv1.Filter

AggregateFilter computes the union of every live subscription's Filter (Types and SqlModels are unioned). The server pushes this to each agent as a single Subscribe with id = "server-aggregate", so the agent only has to maintain one bus subscription regardless of how many UIs are open.

HTTP/SQL-specific filters and NodeIDs are NOT aggregated server-side; the per-UI filter is enforced inside Publish via subscription.matches. The agent only needs to know "what kinds to ship and from which models", not how to route to specific UIs.

func (*EventBus) AggregateSampling 

func (b *EventBus) AggregateSampling() map[string]float32

AggregateSampling computes the per-kind sampling rate the agent-side aggregate Subscribe should apply: the MAX rate any live subscription wants for that kind (a subscription without an entry wants 1.0 — the proto default). Kinds whose aggregate is 1.0 are omitted. Returns nil when there are no subscribers or nothing samples below 1.0.

Publish compensates for this shared agent-side rate per subscription (see sampleResidual), so a 0.1-rate panel and a 1.0-rate panel can coexist: the agent ships at 1.0 and the server thins the 0.1 panel.

func (*EventBus) HasDemand 

func (b *EventBus) HasDemand(t adminv1.EventType) bool

HasDemand reports whether at least one subscription matches the given event kind. The agent-service writer uses this to decide whether to shut off ingress when no UI is watching.

func (*EventBus) Publish 

func (b *EventBus) Publish(e *adminv1.Event) int

Publish fans an event out to every matching subscription. Returns the number of subscribers the event was delivered to (excluding drops).

func (*EventBus) Stats 

func (b *EventBus) Stats() Stats

Stats returns publish counters.

func (*EventBus) Subscribe 

func (b *EventBus) Subscribe(filter *adminv1.Filter, samplingRate map[string]float32, channelSize int) (*EventSubscription, func())

Subscribe registers a new subscription. The caller drains sub.Ch and MUST call cancel exactly once to release resources.

channelSize is the per-subscription buffer. Pass 0 for default 256.

func (*EventBus) SubscriberCount 

func (b *EventBus) SubscriberCount() int

SubscriberCount returns the total number of live subscriptions.

type EventSubscription 

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

EventSubscription is one UI-side subscription. The server creates one per ControlService.StreamEvents call.

func (*EventSubscription) Cancel 

func (s *EventSubscription) Cancel()

Cancel removes the subscription from the bus. Idempotent. Does NOT close the channel (per pkg/observability convention) — pending events are GC'd once the consumer stops reading.

func (*EventSubscription) Ch 

func (s *EventSubscription) Ch() <-chan *adminv1.Event

Ch returns the subscription's event channel. Each event is owned by the consumer (no Release semantics; it's a proto value).

type RbacRouter 

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

RbacRouter correlates server-issued RbacRequest frames with the RbacResponse frames the agent sends back over the same bidi stream. Mirrors the DataStudioRouter/SnapshotRouter pattern.

func NewRbacRouter 

func NewRbacRouter(maxAlive int) *RbacRouter

NewRbacRouter constructs a router. maxAlive caps the total number of pending requests across all agents; values <= 0 default to 64.

func (*RbacRouter) Begin 

func (r *RbacRouter) Begin() (id string, ch chan *adminv1.RbacResponse, cancel func(), err error)

Begin allocates a fresh request_id and registers a channel. The returned cancel MUST be called even on success to release the slot.

func (*RbacRouter) PendingCount 

func (r *RbacRouter) PendingCount() int

PendingCount is intended for tests / metrics.

func (*RbacRouter) Resolve 

func (r *RbacRouter) Resolve(resp *adminv1.RbacResponse) bool

Resolve delivers an RbacResponse to the awaiting Begin caller (if any). Returns true when a pending request consumed the response.

type Replay 

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

Replay is the per-kind drop-oldest ring buffer the server keeps so a freshly opened UI panel sees a few seconds of recent activity instead of an empty stream.

The buffer holds proto Event copies — they are already off the agent's hot path by the time they reach here, so reuse / pooling is not worth the complexity.

func NewReplay 

func NewReplay(c ReplayCapacities) *Replay

NewReplay constructs a Replay with the given capacities.

func (*Replay) LenSnapshot 

func (r *Replay) LenSnapshot() map[adminv1.EventType]int

LenSnapshot returns the per-kind sizes; useful for /metrics gauges.

func (*Replay) Push 

func (r *Replay) Push(e *adminv1.Event)

Push records the event at the appropriate per-kind bucket. Events of unknown kind are dropped silently.

func (*Replay) Snapshot 

func (r *Replay) Snapshot(filter *adminv1.Filter, limit int) []*adminv1.Event

Snapshot returns up to limit events that match the filter, oldest first. limit <= 0 returns everything currently buffered.

type ReplayCapacities 

type ReplayCapacities struct {
	HTTP    int
	SQL     int
	Session int
	Custom  int
}

ReplayCapacities tunes per-kind capacity. Pass zero on a kind to disable replay for that kind.

type SnapshotRouter 

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

SnapshotRouter correlates server-issued SnapshotRequest frames with the SnapshotResponse frames the agent sends back over the same bidi stream. Identifiers are short opaque strings allocated at request time.

func NewSnapshotRouter 

func NewSnapshotRouter(maxAlive int) *SnapshotRouter

NewSnapshotRouter constructs a router. maxAlive caps the total number of pending requests across all agents; values <= 0 default to 256.

func (*SnapshotRouter) Begin 

func (r *SnapshotRouter) Begin() (id string, ch chan *adminv1.SnapshotResponse, cancel func(), err error)

Begin allocates a fresh request_id and registers a channel the caller will block on. The returned cancel MUST be called even on success (after consuming the response) to release the slot.

func (*SnapshotRouter) PendingCount 

func (r *SnapshotRouter) PendingCount() int

PendingCount is intended for tests / metrics.

func (*SnapshotRouter) Resolve 

func (r *SnapshotRouter) Resolve(resp *adminv1.SnapshotResponse) bool

Resolve delivers a SnapshotResponse to the awaiting Begin caller (if any). Returns true when a pending request consumed the response.

type Stats 

type Stats struct {
	Published uint64
	Dropped   uint64
	// Sampled counts deliveries skipped by a subscription's sampling
	// rate (by design, unlike Dropped which signals backpressure).
	Sampled uint64
}

Stats returns published / dropped / sampled totals.

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.