Documentation
¶
Overview ¶
Package nodes is the in-memory registry of agents currently connected to the admin server. It tracks NodeRegistration metadata, the live frame send channel, and the timestamps used for inactivity expiry.
Registry concurrency model:
- The agent service handler creates an Entry on stream open and removes it on stream close. The handler owns the entry; nothing outside ever closes Send.
- Other components (ControlService, UI) read the registry through read-only methods: List, Lookup. Mutations are funneled through Add / Remove.
- The frame send channel is the only path from the server to an agent. The agent service's writer goroutine drains it and writes to the bidi stream. Producers MUST use TryEnqueue (non-blocking).
Index ¶
- func TryEnqueue(e *Entry, f *adminv1.Frame) bool
- type Entry
- type NodeChange
- type NodeInfo
- type Registry
- func (r *Registry) Add(ctx context.Context, cancel context.CancelFunc, info NodeInfo, sendBuffer int) (*Entry, func())
- func (r *Registry) AggregateModels() []string
- func (r *Registry) AnyWithModel(modelName string) (*Entry, bool)
- func (r *Registry) ForEach(fn func(*Entry))
- func (r *Registry) Inactivity(now time.Time, timeout time.Duration) []NodeInfo
- func (r *Registry) List() []NodeInfo
- func (r *Registry) Lookup(nodeID string) (*Entry, bool)
- func (r *Registry) MarkStale(nodeID string) bool
- func (r *Registry) SetHostMetrics(nodeID string, m *adminv1.HostMetrics)
- func (r *Registry) Touch(nodeID string, at time.Time)
- func (r *Registry) Watch() (<-chan NodeChange, func())
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type Entry ¶
type Entry struct {
// NodeID is the stable identifier the agent reported. Used as the
// registry key.
NodeID string
// Info captures NodeRegistration fields plus runtime stats. Never
// mutate these fields directly; go through Touch / UpdateInfo on the
// Registry instead so the Connected/LastSeenAt counters update
// consistently.
Info NodeInfo
// Send is the per-agent frame queue the server writes to. The agent
// stream's writer goroutine drains it in order. Buffer size is
// configured at registry construction time.
Send chan *adminv1.Frame
// CtxDone fires when the agent stream handler is exiting (server-
// initiated close, server shutdown, or client disconnect). Producers
// of frames check this before blocking on Send.
CtxDone <-chan struct{}
// contains filtered or unexported fields
}
Entry is a single connected agent.
type NodeChange ¶
NodeChange is published whenever an agent connects or disconnects, or on inactivity timeout.
type NodeInfo ¶
type NodeInfo struct {
NodeID string
Version string
Labels map[string]string
StartedAt time.Time
LastSeenAt time.Time
Connected bool
RegisteredModels []string
// HostMetrics is the latest sample the agent shipped via Heartbeat
// (nil until the first one arrives). Stored as the wire message; the
// control service forwards it verbatim to the UI.
HostMetrics *adminv1.HostMetrics
}
NodeInfo is the snapshot the UI sees via ControlService.ListNodes.
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry maintains the live set of connected agents.
func (*Registry) Add ¶
func (r *Registry) Add(ctx context.Context, cancel context.CancelFunc, info NodeInfo, sendBuffer int) (*Entry, func())
Add registers a new agent. Returns the entry plus a deregister function the caller (the AgentService handler) MUST call when its stream ends.
cancel, when non-nil, is the cancel function of the stream context the caller passed as ctx. When an agent reconnects under a NodeID that is still registered (its previous stream has not noticed the disconnect yet), Add evicts the old entry AND cancels its stream, so the server never keeps two live streams — and two copies of every event — for one node. Pass nil to opt out of that cancellation.
The cleanup function is idempotent.
func (*Registry) AggregateModels ¶
AggregateModels returns the union of every connected agent's RegisteredModels. The list is sorted alphabetically and de-duplicated case-insensitively so the UI gets a stable view regardless of how many agents reported it.
func (*Registry) AnyWithModel ¶
AnyWithModel returns any connected entry that registered the named model. Returns (nil, false) when no such agent is connected.
The implementation is deliberately first-match (no scoring): models are typically homogeneous across the fleet, and any agent that has the model can answer.
func (*Registry) ForEach ¶
ForEach calls fn with every live entry. fn MUST NOT block on the entry's Send channel; it is meant for fan-out enumeration only. The registry holds the read lock during iteration.
func (*Registry) Inactivity ¶
Inactivity inspects every entry and returns those whose LastSeenAt is older than `now - timeout`. The caller decides whether to evict them.
func (*Registry) Lookup ¶
Lookup returns the entry for nodeID, or (nil, false). The Entry's fields (especially Info) may be racy with concurrent Touch / UpdateInfo calls; consume them only as a snapshot.
func (*Registry) MarkStale ¶
MarkStale flips a node to disconnected without evicting its entry — the inactivity janitor's action on a peer whose stream is silent past the timeout. Returns true when the node existed and was connected (i.e. this call actually changed state and notified watchers). The stream itself is left alone: if it turns out to be alive, the next frame's Touch revives the node.
func (*Registry) SetHostMetrics ¶
func (r *Registry) SetHostMetrics(nodeID string, m *adminv1.HostMetrics)
SetHostMetrics records the latest heartbeat host-metrics sample for a node.
func (*Registry) Touch ¶
Touch updates the last-seen timestamp on every event/heartbeat the AgentService handler receives. Idempotent. A node previously marked stale (MarkStale) is revived: Connected flips back to true and watchers get the reconnect notification.
func (*Registry) Watch ¶
func (r *Registry) Watch() (<-chan NodeChange, func())
Watch subscribes to node-change notifications. The returned channel receives NodeChange values; the cancel function MUST be called to release resources.
The channel is unbuffered: if a watcher does not drain promptly, the publisher will skip notifications for that watcher (we never block the fan-out path on a slow watcher).
Source Files