server

package
v0.0.0-...-6b8ee43 Latest Latest
Warning

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

 Go to latest Published: Jul 25, 2026 License: AGPL-3.0 Imports: 49 Imported by: 0

Documentation

Overview

Package server manages plugin process connections, the 5-stage startup protocol, and command dispatch.

Index

Constants

View Source 
const (
	DefaultCommandTimeout = 30 * time.Second
	CompletionTimeout     = 500 * time.Millisecond
)

Default timeouts for plugin commands.

View Source 
const (
	ReloadOutcomeApplied = "applied"
	ReloadOutcomeFailed  = "failed"
	// ReloadOutcomeNone is reported before the first reload is processed,
	// while Generation is still 0.
	ReloadOutcomeNone = "none"
)

Reload outcome strings reported by ReloadStatus. A reload that ran to completion is "applied"; one that returned an error is "failed". Both advance the generation counter: the counter answers "was a reload PROCESSED", not "did it change anything".

View Source 
const APIVersion = "0.1.0"

APIVersion is the IPC protocol version.

View Source 
const MaxPendingPerProcess = 100

MaxPendingPerProcess limits pending requests to prevent memory exhaustion.

Variables

View Source 
var (
	ErrSchemaModuleEmpty      = errors.New("schema module name is empty")
	ErrSchemaModuleDuplicate  = errors.New("schema module already registered")
	ErrSchemaHandlerDuplicate = errors.New("schema handler already registered")
	ErrSchemaNotFound         = errors.New("schema not found")
	ErrRPCNotFound            = errors.New("RPC not found")
	ErrRPCDuplicate           = errors.New("RPC wire method already registered")
	ErrNotificationDuplicate  = errors.New("notification wire method already registered")
)

Errors for schema registration.

View Source 
var ErrClientConfigNotFound = errors.New("client config not found")

ErrClientConfigNotFound is returned when no config exists for a client name.

View Source 
var ErrDuplicateClient = errors.New("duplicate client name")

ErrDuplicateClient is returned when a client with the same name is already connected.

View Source 
var ErrEmptyCommand = errors.New("empty command")

ErrEmptyCommand is returned when the command is empty.

View Source 
var ErrPluginConnectionClosed = errors.New("plugin connection closed")

ErrPluginConnectionClosed is returned when the plugin's connection is no longer available.

View Source 
var ErrPluginProcessNotRunning = errors.New("plugin process not running")

ErrPluginProcessNotRunning is returned when a plugin command targets a non-running process.

View Source 
var ErrReloadInProgress = errors.New("config reload already in progress")

ErrReloadInProgress is returned when a config reload is attempted while another is already running. Callers can check this with errors.Is to decide whether to queue the reload (SIGHUP) or reject it (CLI/API).

View Source 
var ErrSilent = errors.New("silent")

ErrSilent is returned when a command should produce no response.

View Source 
var ErrSubsystemConnectionClosed = errors.New("subsystem connection closed")

ErrSubsystemConnectionClosed is returned when the subsystem's connection is no longer available.

View Source 
var ErrSubsystemNotRunning = errors.New("subsystem not running")

ErrSubsystemNotRunning is returned when a command targets a non-running subsystem.

View Source 
var ErrUnauthorized = errors.New(plugin.UnauthorizedMessage)

ErrUnauthorized is returned when a command is denied by authorization. Its text is plugin.UnauthorizedMessage because operators read it directly: the ssh exec handler prints this error, not the Response.Error below.

View Source 
var ErrUnknownCommand = errors.New("unknown command")

ErrUnknownCommand is returned when a command is not recognized.

Functions

func BuiltinCount 

func BuiltinCount() int

BuiltinCount returns the number of registered builtin handlers.

func ClientConfigKey 

func ClientConfigKey(name string) string

ClientConfigKey returns the blob key (relative; storage adds the file/active/ namespace) for a managed client's config.

func ClientNameFromConfigKey 

func ClientNameFromConfigKey(key string) (string, bool)

ClientNameFromConfigKey extracts the managed client name from a written blob key, reporting false when the key is not a per-client config key. It tolerates the storage namespace prefix (file/active/) by matching on the key's basename, so it works whether the observer reports the relative or the resolved key.

func ExtractConfigSubtree 

func ExtractConfigSubtree(configTree map[string]any, path string) any

ExtractConfigSubtree extracts a subtree from the config based on path. Always returns data wrapped in its full path structure from root. Supports:

  • "*" -> entire tree
  • "bgp" -> {"bgp": configTree["bgp"]}
  • "bgp/peer" -> {"bgp": {"peer": configTree["bgp"]["peer"]}}

func GetVersion 

func GetVersion() (string, string)

GetVersion returns the current version and build date.

func IsReadOnlyPath 

func IsReadOnlyPath(path string) bool

IsReadOnlyPath returns true if the command path starts with a read-only verb. With verb-first grammar, "show", "monitor", and "resolve" are read-only; "clear", "set", "request", "commit", "update" are not.

func IsStreamingCommand 

func IsStreamingCommand(input string) bool

IsStreamingCommand returns true if the input matches any registered streaming prefix.

func LoadBuiltins 

func LoadBuiltins(d *Dispatcher, wireToPath, pathToDesc map[string]string, pathToArgDefs map[string][]command.ArgDef)

LoadBuiltins registers all builtin handlers with the dispatcher. The wireToPath map provides the dispatch key for each handler, derived from the YANG command tree (WireMethod -> CLI path). pathToDesc provides YANG descriptions for help text. Handlers without a YANG entry are skipped.

func LoadBuiltinsWithAliases 

func LoadBuiltinsWithAliases(d *Dispatcher, wireToPaths map[string][]string, pathToDesc map[string]string, pathToArgDefs map[string][]command.ArgDef, cmdTree *command.Node)

LoadBuiltinsWithAliases registers all builtin handlers with the dispatcher, including all YANG command aliases for each wire method. When cmdTree is non-nil, commands whose YANG path passes through a ze:ensure-exists node are wrapped to auto-ensure the parent resource and rollback on failure.

func LookupCommandHelp 

func LookupCommandHelp(ctx *CommandContext, name, kind string) (*plugin.Response, error)

LookupCommandHelp looks up a command by name in builtins then plugins. The kind parameter is used in error messages (e.g., "command", "bgp rib command").

func MonitorEventFormatter 

func MonitorEventFormatter() func(string) string

MonitorEventFormatter returns the registered event formatter, or nil if none is registered.

func PeerSubcommandKeywords 

func PeerSubcommandKeywords(wireToPath map[string]string) map[string]bool

PeerSubcommandKeywords returns the set of words that immediately follow `peer` in BGP peer command paths. Used by config validation to reject peer names that would collide with subcommand dispatch. The wireToPath map is typically built via yang.WireMethodToPath(loader).

func RegisterDefaultHandlers 

func RegisterDefaultHandlers(d *Dispatcher, wireToPath, pathToDesc map[string]string, pathToArgDefs map[string][]command.ArgDef)

RegisterDefaultHandlers registers all builtin handlers with the dispatcher.

func RegisterMonitorEventFormatter 

func RegisterMonitorEventFormatter(fn func(string) string)

RegisterMonitorEventFormatter registers the function that formats raw JSON event lines into compact one-liners for monitor streaming output (both CLI and TUI). Called from the monitor plugin's init().

func RegisterMonitorProvider 

func RegisterMonitorProvider(p MonitorProvider)

RegisterMonitorProvider registers a TUI monitor provider for a streaming prefix.

func RegisterProcessCleanup 

func RegisterProcessCleanup(fn ProcessCleanupFunc)

RegisterProcessCleanup registers a callback invoked during cleanupProcess. Called from init() to avoid import cycles between server and command packages.

func RegisterRPCs 

func RegisterRPCs(rpcs ...RPCRegistration)

RegisterRPCs adds RPCs to the package-level registry. Called from init() in register.go files.

func RegisterStreamingHandler 

func RegisterStreamingHandler(prefix string, h StreamingHandler)

RegisterStreamingHandler registers a streaming command handler for a prefix. The prefix is matched case-insensitively against command input. Called from plugin init() functions.

func RequireReactor 

func RequireReactor(ctx *CommandContext) (plugin.ReactorLifecycle, *plugin.Response, error)

RequireReactor returns the reactor or an error response if not available.

func SetVersion 

func SetVersion(v, d string)

SetVersion sets the application version and build date (called from main).

func StreamEventMonitor 

func StreamEventMonitor(ctx context.Context, s *Server, w io.Writer, _ string, args []string) error

StreamEventMonitor is the streaming handler for the "event monitor" command. It parses arguments, creates subscriptions, registers a MonitorClient, and streams events until the context is canceled. Registration: called from internal/component/bgp/plugins/cmd/monitor/monitor.go init() via RegisterStreamingHandler("event monitor", StreamEventMonitor).

func StreamingPrefixes 

func StreamingPrefixes() []string

StreamingPrefixes returns the registered streaming command prefixes, sorted.

func UnregisterStreamingHandler 

func UnregisterStreamingHandler(prefix string)

UnregisterStreamingHandler removes a previously registered streaming handler.

Types

type Command 

type Command struct {
	Name             string
	Handler          Handler
	Help             string
	ReadOnly         bool             // True if command only reads state (safe for "ze show")
	RequiresSelector bool             // True if command requires an explicit selector instead of implicit/all scope
	ArgDefs          []command.ArgDef // Typed argument definitions from YANG leaves.
}

Command represents a registered command with metadata.

type CommandContext 

type CommandContext struct {
	Server         *Server           // Gateway to all server state (reactor, dispatcher, etc.)
	Process        *process.Process  // The API process (for session state)
	RequestContext context.Context   // Request-scoped context from the trusted transport.
	Peer           string            // Peer selector: "*" for all, or specific peer selector value. Empty = "*"
	Username       string            // Authenticated username (empty = no auth, full access)
	RemoteAddr     string            // Remote address of the client (e.g., SSH peer IP:port)
	Surface        string            // Trusted caller surface for audit attribution.
	Meta           map[string]any    // Route metadata from UpdateRoute RPC; nil if not set.
	Selectors      map[string]string // Extracted typed selector values, keyed by selector keyword.
}

CommandContext provides access to reactor and session state. Dependencies are accessed through Server; per-request state is stored directly.

func (*CommandContext) CommitManager 

func (c *CommandContext) CommitManager() any

CommitManager returns the commit manager via Server. Nil-safe: returns nil if Server is nil.

func (*CommandContext) Context 

func (c *CommandContext) Context() context.Context

Context returns the request context for this command. Nil-safe: falls back from request -> server -> background.

func (*CommandContext) Dispatcher 

func (c *CommandContext) Dispatcher() *Dispatcher

Dispatcher returns the command dispatcher via Server. Nil-safe: returns nil if Server is nil.

func (*CommandContext) PeerSelector 

func (c *CommandContext) PeerSelector() string

PeerSelector returns the effective neighbor selector. Returns "*" if no neighbor was specified.

func (*CommandContext) ProtocolReactor 

func (c *CommandContext) ProtocolReactor(name string) any

ProtocolReactor returns a named protocol reactor from the Coordinator. Callers type-assert to the protocol-specific interface they need. Nil-safe: returns nil if Server is nil or protocol not registered.

func (*CommandContext) Reactor 

func (c *CommandContext) Reactor() plugin.ReactorLifecycle

Reactor returns the BGP reactor lifecycle interface via Server. Nil-safe: returns nil if Server is nil.

func (*CommandContext) Selector 

func (c *CommandContext) Selector(name string) string

Selector returns the extracted typed selector value for keyword `name`. Nil-safe. Lookup is case-insensitive.

func (*CommandContext) Subscriptions 

func (c *CommandContext) Subscriptions() *SubscriptionManager

Subscriptions returns the subscription manager via Server. Nil-safe: returns nil if Server is nil.

type CommandDef 

type CommandDef struct {
	Name        string        // Command name (e.g., "myapp status")
	Description string        // Help text
	Args        string        // Usage hint (e.g., "<component>")
	Completable bool          // Process handles arg completion
	Hidden      bool          // Hidden from completion and help (works when typed in full)
	Timeout     time.Duration // Per-command timeout (0 = default 30s)
}

CommandDef describes a command to register. Passed from process to registry during registration.

type CommandRegistry 

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

CommandRegistry manages plugin commands. Thread-safe for concurrent registration and lookup.

func NewCommandRegistry 

func NewCommandRegistry() *CommandRegistry

NewCommandRegistry creates a new command registry.

func (*CommandRegistry) AddBuiltin 

func (r *CommandRegistry) AddBuiltin(name string)

AddBuiltin marks a command name as builtin (cannot be shadowed). Called during dispatcher initialization.

func (*CommandRegistry) All 

func (r *CommandRegistry) All() []*RegisteredCommand

All returns all registered commands.

func (*CommandRegistry) Complete 

func (r *CommandRegistry) Complete(partial string) []Completion

Complete returns commands matching the partial input. Used for CLI command completion.

func (*CommandRegistry) Freeze 

func (r *CommandRegistry) Freeze()

Freeze creates an immutable snapshot of the commands and deprecated maps. After Freeze(), Lookup uses atomic.Load instead of RLock. MUST be called after all Register calls complete (after startup barrier). Safe to call multiple times (each call overwrites the previous snapshot).

func (*CommandRegistry) IsBuiltin 

func (r *CommandRegistry) IsBuiltin(name string) bool

IsBuiltin returns true if the command name is a builtin.

func (*CommandRegistry) Lookup 

func (r *CommandRegistry) Lookup(name string) *RegisteredCommand

Lookup finds a command by exact name (case-insensitive). If no primary match is found, checks deprecated aliases and returns the canonical command (logging a deprecation warning once per session). After Freeze(), uses lock-free atomic.Load on the frozen snapshot.

func (*CommandRegistry) LookupDeprecatedPrefix 

func (r *CommandRegistry) LookupDeprecatedPrefix(lowerInput string) (*RegisteredCommand, int)

LookupDeprecatedPrefix finds the longest deprecated alias that is a prefix of lowerInput (already lowercased by the caller). Returns the canonical RegisteredCommand and the matched prefix length, or (nil, 0) if no deprecated alias matches. Logs a deprecation warning on first match.

func (*CommandRegistry) Register 

func (r *CommandRegistry) Register(proc *process.Process, defs []CommandDef) []RegisterResult

Register adds commands for a process. Returns results for each command (success or failure reason).

func (*CommandRegistry) RegisterDeprecated 

func (r *CommandRegistry) RegisterDeprecated(proc *process.Process, oldName, newName string) error

RegisterDeprecated adds a deprecated alias that maps oldName to the canonical command registered under newName. When the old name is looked up, the canonical RegisteredCommand is returned and a deprecation warning is logged once per session.

The alias name is validated with the same parser as a real command, and the alias is rejected if it conflicts with a builtin, an already-registered command, or an existing alias, or if the canonical command is not registered. This makes an unreachable or shadowing alias impossible to register.

Requiring the canonical to be already registered is safe because a plugin's deprecated aliases (CommandDeprecatedNames) reference that same plugin's commands, which startup registers immediately before the aliases.

func (*CommandRegistry) Unregister 

func (r *CommandRegistry) Unregister(proc *process.Process, names []string)

Unregister removes commands owned by the process. Only the owning process can unregister a command. Unknown commands are silently ignored. If frozen, publishes a new snapshot reflecting the removal.

func (*CommandRegistry) UnregisterAll 

func (r *CommandRegistry) UnregisterAll(proc *process.Process)

UnregisterAll removes all commands and deprecated aliases owned by the process. Called when a process dies. If frozen, publishes a new snapshot reflecting the removal.

func (*CommandRegistry) VisibleCommandEntries 

func (r *CommandRegistry) VisibleCommandEntries() []command.CommandEntry

VisibleCommandEntries returns completion-tree entries for every non-hidden registered command. Hidden commands are excluded so they never surface in tab-completion or help (they still dispatch when typed in full via Lookup). Used to inject plugin-registered commands into the operational command tree (command.MergeCommandPaths) so interactive tab-completion offers them, matching the shell-completion path that already reads Complete().

type Completion 

type Completion struct {
	Value  string `json:"value"`            // The completion text
	Help   string `json:"help,omitempty"`   // Optional description
	Source string `json:"source,omitempty"` // "builtin" or process name (verbose mode)
	Hidden bool   `json:"hidden,omitempty"` // Hidden from completion tree (works when typed in full)
}

Completion represents a single completion suggestion. Used for both command and argument completion.

type ConfigBlock 

type ConfigBlock struct {
	Handler string // Handler path (e.g., "bgp/peer")
	Action  string // create, modify, delete
	Path    string // Full path (e.g., "bgp/peer[address=192.0.2.1]")
	Data    string // JSON data
}

ConfigBlock represents a config command to send to a plugin.

func ParseCommand 

func ParseCommand(line string) (*ConfigBlock, error)

ParseCommand parses a namespace command. Format: <namespace> <path> <action> {json}. Or: <namespace> <action> {json} (for namespace-level config). Or: <namespace> commit|rollback|diff.

type ConfigEventGateway 

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

ConfigEventGateway adapts Server to the internal/component/config/transaction.EventGateway interface used by the config transaction orchestrator.

The adapter hides the namespace parameter (always "config") and converts between the orchestrator's []byte payloads and Server's string event payloads.

Performance note: each emit/dispatch round-trips through []byte -> string -> []byte (one copy in EmitConfigEvent, one copy in the SubscribeConfigEvent handler bridge). This is acceptable for small config transaction payloads (~hundreds of bytes per ack) and trades two small allocations against the simpler string-based deliverEvent path. If this ever becomes a hot path, the right fix is to add a []byte-native variant to deliverEvent rather than complicating the adapter.

func NewConfigEventGateway 

func NewConfigEventGateway(s *Server) *ConfigEventGateway

NewConfigEventGateway creates a new adapter wrapping the given Server. The Server must outlive the gateway; the gateway holds a reference but does not manage Server lifecycle.

func (*ConfigEventGateway) EmitConfigEvent 

func (g *ConfigEventGateway) EmitConfigEvent(eventType string, payload []byte) (int, error)

EmitConfigEvent publishes a stream event in the config namespace. Returns the number of plugin processes that received the event. Empty payloads are rejected — config acks always carry at least a transaction-id envelope, so an empty []byte is a programmer error.

func (*ConfigEventGateway) SubscribeConfigEvent 

func (g *ConfigEventGateway) SubscribeConfigEvent(eventType string, handler func(payload []byte)) func()

SubscribeConfigEvent registers a handler for a config namespace event type. The handler is invoked synchronously from deliverEvent. Returns an unsubscribe function; nil handler returns a no-op unsubscribe.

func (*ConfigEventGateway) SubscribeEvent 

func (g *ConfigEventGateway) SubscribeEvent(namespace, eventType string, handler func(payload any)) func()

SubscribeEvent registers a handler for any namespace/event pair. Operation settlement uses this to wait for side-effect events outside config.

type ConfigLoader 

type ConfigLoader func() (map[string]any, error)

ConfigLoader loads a new config tree from disk or other source. Returns the parsed config tree or an error. Set on Server.configLoader before calling ReloadFromDisk.

type ConfigReader 

type ConfigReader func(name string) ([]byte, error)

ConfigReader reads a client's config by name from the hub's blob store. Returns the raw config bytes, or ErrClientConfigNotFound if the client has no config entry.

type Dispatcher 

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

Dispatcher routes commands to handlers.

func NewDispatcher 

func NewDispatcher() *Dispatcher

NewDispatcher creates a new command dispatcher.

func (*Dispatcher) BeginAccounting 

func (d *Dispatcher) BeginAccounting(ctx *CommandContext, input string) func()

BeginAccounting records command START and returns a function that records STOP. It is exported for command paths such as streaming handlers that intentionally bypass Dispatch but must still share the same AAA accounting hook.

func (*Dispatcher) Commands 

func (d *Dispatcher) Commands() []*Command

Commands returns all registered commands.

func (*Dispatcher) Dispatch 

func (d *Dispatcher) Dispatch(ctx *CommandContext, input string) (*plugin.Response, error)

Dispatch parses and executes a command. Supports inline selector extraction both for typed forms like `show demo name <name> detail` and for positional selector slots that appear before a later action token.

func (*Dispatcher) ForwardToPlugin 

func (d *Dispatcher) ForwardToPlugin(cmdCtx *CommandContext, command string, args []string, peerSelector string) (*plugin.Response, error)

ForwardToPlugin routes a command to a plugin process by exact name lookup. Used by proxy handlers that bridge CLI builtins to plugin commands. Returns ErrUnknownCommand if the command is not registered (plugin may not be running).

func (*Dispatcher) IsAuthorized 

func (d *Dispatcher) IsAuthorized(ctx *CommandContext, input string, readOnly bool) bool

IsAuthorized checks if the user is allowed to execute the command. Exported for use by streaming handlers (e.g., monitor) that bypass the normal dispatch path.

func (*Dispatcher) Lookup 

func (d *Dispatcher) Lookup(name string) *Command

Lookup finds a command by exact name.

func (*Dispatcher) Pending 

func (d *Dispatcher) Pending() *PendingRequests

Pending returns the pending requests tracker.

func (*Dispatcher) Register 

func (d *Dispatcher) Register(name string, handler Handler, help string)

Register adds a builtin command handler. Also marks the command as builtin in the registry to prevent shadowing.

func (*Dispatcher) RegisterWithOptions 

func (d *Dispatcher) RegisterWithOptions(name string, handler Handler, help string, opts RegisterOptions)

RegisterWithOptions adds a builtin command handler with additional options.

func (*Dispatcher) Registry 

func (d *Dispatcher) Registry() *CommandRegistry

Registry returns the plugin command registry.

func (*Dispatcher) SetAccountingHook 

func (d *Dispatcher) SetAccountingHook(h aaa.Accountant)

SetAccountingHook sets the accounting recorder for the dispatcher. When set, command START/STOP records are sent for every dispatched command. Accounting failures never block command execution.

func (*Dispatcher) SetAuditRecorder 

func (d *Dispatcher) SetAuditRecorder(recorder audit.Recorder)

SetAuditRecorder sets the structured audit recorder for mutation commands.

func (*Dispatcher) SetAuthorizer 

func (d *Dispatcher) SetAuthorizer(a aaa.Authorizer)

SetAuthorizer sets the authorization checker for the dispatcher. When set, all commands are checked against the authorizer before execution.

func (*Dispatcher) SetSubsystems 

func (d *Dispatcher) SetSubsystems(sm *SubsystemManager)

SetSubsystems sets the subsystem manager.

func (*Dispatcher) Subsystems 

func (d *Dispatcher) Subsystems() *SubsystemManager

Subsystems returns the subsystem manager.

type EngineEventHandler 

type EngineEventHandler func(payload any)

EngineEventHandler is invoked when a stream event matches an engine subscription. The payload is the publisher's typed Go value, passed as `any`. Consumers type-assert to the documented payload type for the (namespace, eventType) pair. Handlers are called synchronously from deliverEvent; they MUST NOT block on external I/O.

A handler that panics is recovered by the dispatch loop, logged, and the remaining handlers for the same event still fire. The panic does NOT propagate to the emitter.

type EnsureStep 

type EnsureStep struct {
	Handler         Handler // Creation handler (idempotent: succeeds if resource exists)
	RollbackHandler Handler // Deletion handler for undo on descendant failure
}

EnsureStep describes one ancestor resource that must exist before a descendant command can execute. Built at registration time from the YANG command tree's ze:ensure-exists annotations.

type EventMonitorOpts 

type EventMonitorOpts struct {
	IncludeTypes []string
	ExcludeTypes []string
	Peer         string
	Direction    string
}

EventMonitorOpts holds parsed arguments for the event monitor command.

func ParseEventMonitorArgs 

func ParseEventMonitorArgs(args []string) (*EventMonitorOpts, error)

ParseEventMonitorArgs parses keyword arguments for the event monitor command.

Syntax: [include|exclude <types>] [peer <selector>] [direction received|sent] Keywords may appear in any order. Include and exclude are mutually exclusive.

type EventRecord 

type EventRecord struct {
	Timestamp time.Time
	Namespace string
	EventType string
}

EventRecord is one entry in the global event ring.

type EventRing 

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

EventRing is a fixed-size circular buffer of EventRecord values. Safe for concurrent use. Append is O(1) with no allocation.

func NewEventRing 

func NewEventRing(capacity int) *EventRing

NewEventRing creates a ring with the given capacity.

func (*EventRing) Append 

func (r *EventRing) Append(namespace, eventType string)

Append adds a record to the ring, overwriting the oldest if full.

func (*EventRing) Count 

func (r *EventRing) Count() int

Count returns the number of records currently in the ring.

func (*EventRing) NamespaceCounts 

func (r *EventRing) NamespaceCounts() map[string]int

NamespaceCounts returns a map of namespace -> event count.

func (*EventRing) SetOnAppend 

func (r *EventRing) SetOnAppend(fn func(EventRecord))

SetOnAppend registers a callback invoked after each Append, outside the lock. Used by the web SSE broker to push live log entries. The callback must not block; a slow callback delays the Append caller.

func (*EventRing) Snapshot 

func (r *EventRing) Snapshot(limit int, namespace string) []EventRecord

Snapshot returns up to limit records, newest first. If limit <= 0, returns all records. If namespace is non-empty, only matching records are returned.

type FullReloadFunc 

type FullReloadFunc func(context.Context) error

FullReloadFunc runs the hub-level reload path for commits triggered through RPC. It includes plugin transactions plus ConfigProvider, engine, and subsystem refresh.

type Handler 

type Handler func(ctx *CommandContext, args []string) (*plugin.Response, error)

Handler processes a command and returns a response.

type Hub 

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

Hub orchestrates plugin communication and command routing.

func NewHub 

func NewHub(registry *SchemaRegistry, subsystems *SubsystemManager) *Hub

NewHub creates a new Hub with the given registry and subsystem manager.

func (*Hub) ProcessConfig 

func (h *Hub) ProcessConfig(ctx context.Context, blocks []ConfigBlock) error

ProcessConfig processes a configuration transaction. Sends all commands to plugins, then commits each affected namespace.

func (*Hub) RouteCommand 

func (h *Hub) RouteCommand(ctx context.Context, block *ConfigBlock) error

RouteCommand routes a command to the appropriate plugin. Format: <namespace> <path> <action> {json}.

func (*Hub) RouteCommit 

func (h *Hub) RouteCommit(ctx context.Context, namespace string) error

RouteCommit sends a commit command to a plugin. Format: <namespace> commit.

func (*Hub) RouteRollback 

func (h *Hub) RouteRollback(ctx context.Context, namespace string) error

RouteRollback sends a rollback command to a plugin. Format: <namespace> rollback.

type ManagedConfigService 

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

ManagedConfigService handles hub-side config-fetch and config-changed operations for managed clients. It reads client configs via a ConfigReader and computes version hashes for change detection. Tracks connected clients and rejects duplicate names. Safe for concurrent use.

func NewManagedConfigService 

func NewManagedConfigService(reader ConfigReader) *ManagedConfigService

NewManagedConfigService creates a service that reads client configs via reader.

func (*ManagedConfigService) BuildConfigChanged 

func (s *ManagedConfigService) BuildConfigChanged(clientName string) (fleet.ConfigChanged, error)

BuildConfigChanged creates a config-changed notification for a client. Reads the client's current config and computes its version hash.

func (*ManagedConfigService) HandleConfigFetch 

func (s *ManagedConfigService) HandleConfigFetch(clientName string, req fleet.ConfigFetchRequest) (fleet.ConfigFetchResponse, error)

HandleConfigFetch processes a config-fetch request from a managed client. If the client's version matches the current config, returns status "current". Otherwise returns the full config as base64 with the new version hash.

func (*ManagedConfigService) RegisterClient 

func (s *ManagedConfigService) RegisterClient(name string) error

RegisterClient marks a client as connected. Returns ErrDuplicateClient if a client with the same name is already connected. Caller MUST call UnregisterClient when the client disconnects.

func (*ManagedConfigService) UnregisterClient 

func (s *ManagedConfigService) UnregisterClient(name string)

UnregisterClient removes a client from the connected set.

type ManagedServer 

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

ManagedServer authenticates managed fleet clients (per-client secret), answers config-fetch/config-ack/ping over MuxConn, and pushes config-changed to connected clients. It owns its listeners and one goroutine per connection.

func NewManagedServer 

func NewManagedServer(cfg ManagedServerConfig) (*ManagedServer, error)

NewManagedServer builds a managed server. ReadConfig is required. A self-signed TLS certificate is generated: the transport is encrypted, but a remote managed client cannot verify a self-signed cert against a CA, so today it must connect with tls-insecure. Verifiable server-cert distribution (CA cert or pinned fingerprint in the client config) is tracked in plan/spec-managed-server-hardening.md.

func (*ManagedServer) Addrs 

func (s *ManagedServer) Addrs() []net.Addr

Addrs returns the bound listener addresses (useful when a port-0 was requested, and for tests to discover the ephemeral port).

func (*ManagedServer) NotifyConfigChanged 

func (s *ManagedServer) NotifyConfigChanged(name string)

NotifyConfigChanged enqueues a config-changed push for the named client. It is non-blocking and safe to call from the storage write path: the round-trip runs on notifyWorker. On a full queue the notify is dropped (the client picks up the change on its next fetch/reconnect).

func (*ManagedServer) Start 

func (s *ManagedServer) Start(ctx context.Context) error

Start binds the listeners and begins serving in background goroutines. It returns once binding succeeds; serving continues until Stop or ctx cancellation.

func (*ManagedServer) Stop 

func (s *ManagedServer) Stop()

Stop cancels serving and waits for all goroutines to drain. Listeners are closed by closeOnDone when the context is canceled.

type ManagedServerConfig 

type ManagedServerConfig struct {
	Addrs         []string          // Listen addresses (server blocks that declare client entries).
	ClientSecrets map[string]string // Per-client name -> secret (authoritative for managed clients).
	ReadConfig    ConfigReader      // Reads a client's config by name (over the hub blob store).
	Metrics       metrics.Registry  // Optional; nil installs no-op counters.
}

ManagedServerConfig configures the dedicated managed-config TLS server.

type MonitorClient 

type MonitorClient struct {
	EventChan chan string     // Buffered channel for formatted events.
	Ctx       context.Context // Client-scoped context for cancellation.
	Dropped   atomic.Uint64   // Count of events dropped due to full channel.
	// contains filtered or unexported fields
}

MonitorClient represents an active monitor session.

func NewMonitorClient 

func NewMonitorClient(ctx context.Context, id string, subs []*Subscription, bufSize int) *MonitorClient

NewMonitorClient creates a monitor client with the given subscriptions and buffer size. Caller MUST call MonitorManager.Remove(id) when done to release resources.

type MonitorManager 

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

MonitorManager manages active monitor clients. Parallel to SubscriptionManager (which manages plugin process subscriptions).

func NewMonitorManager 

func NewMonitorManager() *MonitorManager

NewMonitorManager creates a new monitor manager.

func (*MonitorManager) Add 

func (mm *MonitorManager) Add(mc *MonitorClient)

Add registers a monitor client.

func (*MonitorManager) Count 

func (mm *MonitorManager) Count() int

Count returns the number of active monitors.

func (*MonitorManager) Deliver 

func (mm *MonitorManager) Deliver(namespace, eventType, direction, peerAddr, peerName, output string)

Deliver sends a formatted event to all matching monitors. Uses non-blocking send: if a monitor's channel is full, the event is dropped and the dropped counter is incremented (backpressure). peerName is the configured peer name (may be empty).

func (*MonitorManager) DeliverLazy 

func (mm *MonitorManager) DeliverLazy(namespace, eventType, direction, peerAddr, peerName string, build func() string)

DeliverLazy sends an event to matching monitors, invoking build only when at least one monitor matches. This avoids formatting cost for events that no monitor subscribes to (the common case when structured plugin consumers are present but no CLI monitor is attached). build is called outside the manager lock so JSON formatting does not block monitor registration. peerName is the configured peer name (may be empty).

Race note: GetMatching releases mm.mu before build() and enqueue() run, so a concurrent Remove(id) may drop a monitor between matching and delivery. That is safe: enqueue uses a non-blocking send on a buffered channel that the removed client's reader will simply stop consuming on Ctx cancellation. The dropped counter on the removed client may tick up, which is harmless.

func (*MonitorManager) DeliverLazyTyped 

func (mm *MonitorManager) DeliverLazyTyped(ns events.NamespaceID, et events.EventTypeID, dir events.Direction, peerAddr, peerName string, build func() string)

DeliverLazyTyped is the hot-path variant of DeliverLazy. It accepts pre-resolved typed IDs, skipping the string-to-ID lookups and their associated global event registry RLock acquisitions. Returns immediately via an atomic load when no monitors are registered (the common production case).

func (*MonitorManager) GetMatching 

func (mm *MonitorManager) GetMatching(namespace, eventType, direction, peerAddr, peerName string) []*MonitorClient

GetMatching returns monitors with subscriptions matching the event. A monitor matches if any of its subscriptions match. peerName is the configured peer name (may be empty).

func (*MonitorManager) GetMatchingTyped 

func (mm *MonitorManager) GetMatchingTyped(ns events.NamespaceID, et events.EventTypeID, dir events.Direction, peerAddr, peerName string) []*MonitorClient

GetMatchingTyped returns monitors with subscriptions matching the event, using pre-resolved typed IDs. Avoids the string-to-ID lookups in GetMatching.

func (*MonitorManager) HasMonitors 

func (mm *MonitorManager) HasMonitors() bool

HasMonitors returns true if any monitor clients are registered. Uses an atomic load instead of acquiring the mutex, making it suitable for hot-path early-exit checks.

func (*MonitorManager) Remove 

func (mm *MonitorManager) Remove(id string)

Remove unregisters a monitor client by ID.

type MonitorProvider 

type MonitorProvider struct {
	Prefix   string
	CreateFn func(ctx context.Context, args []string) (eventCh <-chan string, renderFn func(w, h int) string, cancel func(), err error)
}

MonitorProvider creates a TUI monitor session for a streaming command.

func GetMonitorProvider 

func GetMonitorProvider(input string) (*MonitorProvider, []string)

GetMonitorProvider returns a provider for the given command input, or nil.

type PeerFilter 

type PeerFilter struct {
	Selector string // "*", "10.0.0.1", "!10.0.0.1", "my-peer", "!my-peer"
}

PeerFilter specifies which peers to filter.

func (*PeerFilter) Matches 

func (pf *PeerFilter) Matches(peerAddr, peerName string) bool

Matches returns true if the peer matches this filter. Matches against both the peer address (IP) and peer name.

type PendingRequest 

type PendingRequest struct {
	Serial   string                // Alpha serial (a, b, bcd, ...)
	Command  string                // Matched command name
	Process  *process.Process      // Target process
	Timeout  time.Duration         // Timeout for response
	RespChan chan *plugin.Response // Channel to deliver response
	// contains filtered or unexported fields
}

PendingRequest represents an in-flight plugin command request.

type PendingRequests 

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

PendingRequests tracks in-flight plugin command requests. Thread-safe for concurrent access.

func NewPendingRequests 

func NewPendingRequests() *PendingRequests

NewPendingRequests creates a new pending requests tracker.

func (*PendingRequests) Add 

func (p *PendingRequests) Add(req *PendingRequest) string

Add registers a new pending request and starts the timeout timer. Returns the assigned alpha serial, or empty string if limit exceeded. If limit exceeded, sends error response to RespChan.

func (*PendingRequests) CancelAll 

func (p *PendingRequests) CancelAll(proc *process.Process)

CancelAll cancels all pending requests for a process (process death). Sends error response to all waiting clients.

func (*PendingRequests) Complete 

func (p *PendingRequests) Complete(serial string, resp *plugin.Response) bool

Complete delivers a final response and removes the request. Returns true if the serial was found (response delivered).

func (*PendingRequests) Count 

func (p *PendingRequests) Count(proc *process.Process) int

Count returns the number of pending requests for a process.

func (*PendingRequests) Partial 

func (p *PendingRequests) Partial(serial string, resp *plugin.Response) bool

Partial delivers a partial response (streaming) and resets the timeout. Returns true if the serial was found.

func (*PendingRequests) Total 

func (p *PendingRequests) Total() int

Total returns the total number of pending requests.

type ProcessCleanupFunc 

type ProcessCleanupFunc func(processName string)

ProcessCleanupFunc is called when a plugin process exits. Receives the process name for scoped cleanup.

type QuiesceFunc 

type QuiesceFunc func(ctx context.Context) error

QuiesceFunc drains a subsystem's pending asynchronous work, returning when the subsystem has settled or ctx is canceled (deadline reached).

A QuiesceFunc MUST honor ctx cancellation. quiesceAll bounds each drain with a per-quiescer deadline by canceling ctx, but it cannot force-return a drain that ignores ctx — such a quiescer would hang the barrier (and the caller) despite the timeout. The only current registrant, the reactor's FlushForwardPool, selects on ctx.Done() (forward_pool_barrier.go).

type Quiescer 

type Quiescer struct {
	Name    string
	Quiesce QuiesceFunc
}

Quiescer is a named subsystem drain invoked by `request quiesce`.

type QuiescerRegistry 

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

QuiescerRegistry holds subsystem drains invoked by `request quiesce`. Registration is at runtime (a subsystem needs a live reference such as the reactor), so this is a lock-guarded registry rather than an init() table.

func (*QuiescerRegistry) All 

func (r *QuiescerRegistry) All() []Quiescer

All returns a snapshot of the registered quiescers.

func (*QuiescerRegistry) Register 

func (r *QuiescerRegistry) Register(name string, fn QuiesceFunc)

Register adds a named subsystem drain. Safe for concurrent use.

type RPCParams 

type RPCParams struct {
	Selector string   `json:"selector,omitempty"` // Peer selector (optional)
	Args     []string `json:"args,omitempty"`     // Command arguments (optional)
}

RPCParams is the standard params format for JSON RPC requests from socket clients. Handlers receive Args as positional arguments and Selector as the peer filter. Identity (Username) is never accepted from client JSON -- it MUST be injected by the transport layer (SSH session, plugin process manager, TLS auth).

type RPCRegistration 

type RPCRegistration struct {
	WireMethod       string  // "module:rpc-name" format (e.g., "ze-bgp:peer-list")
	Handler          Handler // Handler function
	RequiresSelector bool    // True if peer commands must have explicit selector (not default "*")
	PluginCommand    string  // If set, this builtin proxies to a runtime plugin command (e.g., "bgp rib show")
}

RPCRegistration maps a YANG RPC wire method to its handler function. The CLI command name is derived from the YANG command tree (-cmd.yang modules) via yang.WireMethodToPath(). It is not stored in the registration. Help text comes from YANG descriptions. Read-only classification comes from the verb position in the command tree (show/validate/monitor = read-only).

func AllBuiltinRPCs 

func AllBuiltinRPCs() []RPCRegistration

AllBuiltinRPCs returns all RPCs registered via init() + RegisterRPCs(). Includes server, handler, and editor RPCs (when their packages are imported).

type RegisterOptions 

type RegisterOptions struct {
	ReadOnly         bool             // True if command only reads state
	RequiresSelector bool             // True if the command requires an explicit selector value
	PluginProxy      bool             // True if this builtin proxies to a plugin command (allows plugin to register same name)
	ArgDefs          []command.ArgDef // Typed argument definitions from YANG leaves
}

RegisterOptions holds optional settings for command registration.

type RegisterResult 

type RegisterResult struct {
	Name  string // Command that was registered
	OK    bool   // True if registration succeeded
	Error string // Error message if failed
}

RegisterResult holds the result of a single command registration.

type RegisteredCommand 

type RegisteredCommand struct {
	Name         string
	LowerName    string // Pre-lowercased at registration for dispatch matching (zero alloc per lookup)
	Description  string
	Args         string           // Usage hint (e.g., "<component>")
	Completable  bool             // Process handles arg completion
	Hidden       bool             // Hidden from completion and help (works when typed in full)
	Timeout      time.Duration    // Per-command timeout
	Process      *process.Process // Owning process
	RegisteredAt time.Time
}

RegisteredCommand represents a plugin command in the registry.

type RegisteredNotification 

type RegisteredNotification struct {
	Module      string          // YANG module name
	Name        string          // Notification name in kebab-case
	WireMethod  string          // Wire format "module:notification-name"
	Description string          // From YANG description
	Leaves      []yang.LeafMeta // Notification data leaves
}

RegisteredNotification represents a notification indexed in the schema registry.

type RegisteredRPC 

type RegisteredRPC struct {
	Module      string          // YANG module name (e.g., "ze-bgp-api")
	Name        string          // RPC name in kebab-case (e.g., "peer-list")
	WireMethod  string          // Wire format "module:rpc-name" (e.g., "ze-bgp:peer-list")
	CLICommand  string          // CLI text command (e.g., "bgp peer list")
	Description string          // From YANG description
	Input       []yang.LeafMeta // Input parameter leaves
	Output      []yang.LeafMeta // Output parameter leaves
	Handler     Handler         // Handler function (set during registration)
}

RegisteredRPC represents an RPC indexed in the schema registry.

type Schema 

type Schema struct {
	Module      string   // YANG module name
	Namespace   string   // YANG namespace URI
	Yang        string   // Full YANG module text
	Imports     []string // Imported module names (from YANG import statements)
	Handlers    []string // Handler paths (e.g., "bgp", "bgp/peer")
	Plugin      string   // Plugin that registered this schema
	Priority    int      // Config ordering (lower = processed first, default 1000)
	WantsConfig []string // Config roots plugin wants (from "declare wants config <root>")
}

Schema represents a YANG schema registered by a plugin.

type SchemaRegistry 

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

SchemaRegistry stores and manages schemas from all plugins.

func NewSchemaRegistry 

func NewSchemaRegistry() *SchemaRegistry

NewSchemaRegistry creates a new schema registry.

func (*SchemaRegistry) Count 

func (r *SchemaRegistry) Count() int

Count returns the number of registered schemas.

func (*SchemaRegistry) FindHandler 

func (r *SchemaRegistry) FindHandler(path string) (*Schema, string)

FindHandler returns the schema for a handler path using longest prefix match. For example, if "bgp" and "bgp/peer" are registered, FindHandler("bgp/peer/timers") returns the schema for "bgp/peer". Predicates like [address=192.0.2.1] are stripped before matching. After Freeze(), uses lock-free atomic.Load on the frozen snapshot.

func (*SchemaRegistry) FindRPC 

func (r *SchemaRegistry) FindRPC(wireMethod string) (*RegisteredRPC, error)

FindRPC returns the registered RPC for an exact wire method match.

func (*SchemaRegistry) FindRPCByCommand 

func (r *SchemaRegistry) FindRPCByCommand(cliCommand string) (*RegisteredRPC, error)

FindRPCByCommand returns the registered RPC for a CLI text command.

func (*SchemaRegistry) Freeze 

func (r *SchemaRegistry) Freeze()

Freeze creates an immutable snapshot of the handler and module maps. After Freeze(), FindHandler uses atomic.Load instead of RLock. MUST be called after all Register calls complete (after startup barrier). Safe to call multiple times (each call overwrites the previous snapshot).

func (*SchemaRegistry) GetByHandler 

func (r *SchemaRegistry) GetByHandler(path string) (*Schema, error)

GetByHandler returns a schema by exact handler path.

func (*SchemaRegistry) GetByModule 

func (r *SchemaRegistry) GetByModule(name string) (*Schema, error)

GetByModule returns a schema by module name.

func (*SchemaRegistry) ListHandlers 

func (r *SchemaRegistry) ListHandlers() map[string]string

ListHandlers returns all registered handler paths with their modules.

func (*SchemaRegistry) ListModules 

func (r *SchemaRegistry) ListModules() []string

ListModules returns all registered module names.

func (*SchemaRegistry) ListNotifications 

func (r *SchemaRegistry) ListNotifications(module string) []*RegisteredNotification

ListNotifications returns all registered notifications, optionally filtered by YANG module name. Pass empty string to list all notifications.

func (*SchemaRegistry) ListRPCs 

func (r *SchemaRegistry) ListRPCs(module string) []*RegisteredRPC

ListRPCs returns all registered RPCs, optionally filtered by YANG module name. Pass empty string to list all RPCs.

func (*SchemaRegistry) Register 

func (r *SchemaRegistry) Register(schema *Schema) error

Register adds a schema to the registry. Returns error if module name or handler paths conflict with existing registrations.

func (*SchemaRegistry) RegisterCLICommand 

func (r *SchemaRegistry) RegisterCLICommand(cliCommand, wireMethod string) error

RegisterCLICommand associates a CLI text command with a wire method. The wire method must already be registered via RegisterRPCs.

func (*SchemaRegistry) RegisterNotifications 

func (r *SchemaRegistry) RegisterNotifications(module string, notifs []yang.NotificationMeta) error

RegisterNotifications indexes notifications extracted from a YANG module.

func (*SchemaRegistry) RegisterRPCs 

func (r *SchemaRegistry) RegisterRPCs(module string, rpcs []yang.RPCMeta) error

RegisterRPCs indexes RPCs extracted from a YANG module. Wire methods use the stripped module prefix (e.g., "ze-bgp-api" → "ze-bgp:peer-list").

type Server 

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

Server manages API connections and command dispatch.

func NewServer 

func NewServer(config *ServerConfig, reactor plugin.ReactorLifecycle) (*Server, error)

NewServer creates a new API server.

func (*Server) AllPluginCapabilities 

func (s *Server) AllPluginCapabilities() []plugin.InjectedCapability

AllPluginCapabilities returns all stored capabilities (global + all per-peer). Used by the restart handler to compute max restart-time for the GR marker.

func (*Server) CallDoctorCheck 

func (s *Server) CallDoctorCheck(ctx context.Context, pluginName, checkName string) (*rpc.DoctorCheckOutput, error)

CallDoctorCheck invokes a plugin's doctor check callback and returns diagnostics.

func (*Server) CallFilterUpdate 

func (s *Server) CallFilterUpdate(ctx context.Context, pluginName string, input *rpc.FilterUpdateInput) (*rpc.FilterUpdateOutput, error)

CallFilterUpdate sends a filter-update RPC to a named plugin and returns the response. Returns an error if the plugin is not found, not connected, or the RPC fails.

func (*Server) CommitManager 

func (s *Server) CommitManager() any

CommitManager returns the commit manager.

func (*Server) ConfigPath 

func (s *Server) ConfigPath() string

ConfigPath returns the path to the config file. Empty if not set.

func (*Server) Context 

func (s *Server) Context() context.Context

Context returns the server's context. Used by RPC handlers that need a cancellable context tied to the server's lifetime (e.g., coordinator reload).

func (*Server) DecodeNLRI 

func (s *Server) DecodeNLRI(family, hexData string) (string, error)

DecodeNLRI decodes NLRI by routing to the appropriate family plugin via RPC. Returns the JSON representation of the decoded NLRI. Returns error if no plugin registered or plugin not running.

func (*Server) Dispatcher 

func (s *Server) Dispatcher() *Dispatcher

Dispatcher returns the command dispatcher.

func (*Server) DoctorCheckPlugins 

func (s *Server) DoctorCheckPlugins() map[string][]plugin.DoctorCheckRegistration

DoctorCheckPlugins returns plugin names and their doctor check registrations.

func (*Server) DrainSIGHUP 

func (s *Server) DrainSIGHUP() bool

DrainSIGHUP returns true if a SIGHUP was queued and clears the flag.

func (*Server) Emit 

func (s *Server) Emit(namespace, eventType string, payload any) (int, error)

Emit satisfies the pkg/ze.EventBus interface. It is a thin alias for EmitEngineEvent so engine components can depend on the public ze.EventBus type without importing this package directly.

func (*Server) EmitEngineEvent 

func (s *Server) EmitEngineEvent(namespace, eventType string, payload any) (int, error)

EmitEngineEvent publishes an event from the engine to the stream system. Both engine subscribers and plugin process subscribers receive it. Returns the number of plugin processes that received the event (engine handler count is intentionally not reported because engine subscribers are in-process and always receive synchronously when matching).

The event must use a registered (namespace, eventType) per events.IsValidEvent; unknown pairs return an error and deliver to nobody (neither engine handlers nor plugin subscribers).

payload is the publisher's typed Go value. In-process subscribers receive it directly; plugin-process subscribers receive JSON bytes marshaled once per Emit (only when at least one plugin-process subscriber exists).

func (*Server) EncodeNLRI 

func (s *Server) EncodeNLRI(family string, args []string) ([]byte, error)

EncodeNLRI encodes NLRI by routing to the appropriate family plugin via RPC. Returns error if no plugin registered or plugin not running.

func (*Server) EventRing 

func (s *Server) EventRing() *EventRing

EventRing returns the global event history ring for CLI queries.

func (*Server) FilterInfo 

func (s *Server) FilterInfo(pluginName, filterName string) (declaredAttrs []string, raw bool)

FilterInfo returns declaration info for a named filter: declared attributes and raw flag. Returns nil attributes and false if the plugin or filter is not found.

func (*Server) FilterOnError 

func (s *Server) FilterOnError(pluginName, filterName string) rpc.OnErrorPolicy

FilterOnError returns the declared on-error mode for a named filter. Returns rpc.OnErrorReject (fail-closed) if the plugin or filter is not found.

func (*Server) GetDecodeFamilies 

func (s *Server) GetDecodeFamilies() []string

GetDecodeFamilies returns all families that have decode plugins registered. Used by Session to auto-add Multiprotocol capabilities in OPEN. Plugins that can decode a family should advertise that family to peers.

func (*Server) GetPluginCapabilitiesForPeer 

func (s *Server) GetPluginCapabilitiesForPeer(peerAddr string) []plugin.InjectedCapability

GetPluginCapabilitiesForPeer returns plugin-declared capabilities for a specific peer. Returns global capabilities plus any peer-specific capabilities (per-peer takes precedence).

func (*Server) GetSchemaDeclarations 

func (s *Server) GetSchemaDeclarations() []plugin.SchemaDeclaration

GetSchemaDeclarations returns all schema declarations from registered plugins. Used for two-phase config parsing to extend the schema before parsing peer config. Should be called after Stage 1 (Registration) completes for all plugins.

func (*Server) HandleAdHocPluginSession 

func (s *Server) HandleAdHocPluginSession(reader io.ReadCloser, writer io.WriteCloser) error

HandleAdHocPluginSession runs the 5-stage plugin handshake and runtime command loop over an arbitrary bidirectional stream (e.g., an SSH channel). The session uses coordinator == nil, so all stage barriers are skipped. Blocks until the connection closes or the server shuts down. Caller MUST close reader and writer after this returns.

func (*Server) HasConfigLoader 

func (s *Server) HasConfigLoader() bool

HasConfigLoader reports whether a config loader has been set. Used by SIGHUP handler to decide between coordinator path and direct reload.

func (*Server) HasFullReloadFunc 

func (s *Server) HasFullReloadFunc() bool

HasFullReloadFunc reports whether a hub-level reload hook has been set.

func (*Server) HasProcesses 

func (s *Server) HasProcesses() bool

HasProcesses returns true if any plugin processes were loaded during startup. Used by the main loop to decide whether to listen for server-done (all processes exited). Without this, configs with no plugins cause immediate daemon exit.

func (*Server) MarkReloadProcessed 

func (s *Server) MarkReloadProcessed(applied bool)

MarkReloadProcessed records that a reload sequence completed and advances the generation counter. `applied` is false when the reload returned an error.

MUST be called only once the ENTIRE reload sequence has run, not merely the plugin-server half: the config knobs a reload rejects are diffed by the subsystems that engine.Reload fans out to, which runs AFTER Server.ReloadConfig. Marking any earlier would advance the fence before the rejection it is meant to fence had happened, and an observer could then read state the reload had not yet touched. cmd/ze/hub/main_reload.go doReload is the correct (and only) caller.

A reload refused with ErrReloadInProgress was never processed -- it is queued and replayed -- so its caller must not mark it.

func (*Server) Monitors 

func (s *Server) Monitors() *MonitorManager

Monitors returns the monitor manager for CLI monitor sessions.

func (*Server) ProcessManager 

func (s *Server) ProcessManager() *process.ProcessManager

ProcessManager returns the process manager. Used by BGP hook implementations to iterate plugin processes.

func (*Server) QueueSIGHUP 

func (s *Server) QueueSIGHUP()

QueueSIGHUP queues a SIGHUP for later processing if a transaction is active.

func (*Server) Quiescers 

func (s *Server) Quiescers() []Quiescer

Quiescers returns a snapshot of the registered subsystem drains.

func (*Server) Reactor 

func (s *Server) Reactor() plugin.ReactorLifecycle

func (*Server) ReactorAny 

func (s *Server) ReactorAny() any

ReactorAny returns the reactor as any, satisfying registry.PluginServerAccessor.

func (*Server) ReactorFor 

func (s *Server) ReactorFor(name string) any

ReactorFor returns a named protocol reactor from the Coordinator, or nil. This allows plugins to access non-BGP reactors (e.g., OSPF, IS-IS) by name.

func (*Server) RegisterQuiescer 

func (s *Server) RegisterQuiescer(name string, fn QuiesceFunc)

RegisterQuiescer registers a named subsystem drain, invoked by `request quiesce`. Called at wiring time (e.g. when the reactor is attached to the server), not from init(), because the drain closes over a live reference.

func (*Server) ReloadConfig 

func (s *Server) ReloadConfig(ctx context.Context, newTree map[string]any) error

ReloadConfig orchestrates config reload across all config-interested plugins. Follows verify→apply protocol: all plugins must verify before any apply. Returns nil if there are no changes, or if verify→apply succeeds. Returns error if verify fails for any plugin, or if a reload is already in progress.

func (*Server) ReloadFromDisk 

func (s *Server) ReloadFromDisk(ctx context.Context) error

ReloadFromDisk loads the config from the configured loader and triggers reload. Returns error if the loader is not set, parsing fails, or reload fails.

func (*Server) ReloadFull 

func (s *Server) ReloadFull(ctx context.Context) error

ReloadFull runs the hub-level reload hook.

func (*Server) ReloadStatus 

func (s *Server) ReloadStatus() (generation uint64, outcome string, at time.Time)

ReloadStatus returns the number of reloads processed since daemon start, the outcome of the most recent one, and when it finished. Before the first reload: (0, ReloadOutcomeNone, zero time).

An observer fences on the generation: read it, trigger a reload, then poll until it advances. At that point every reload step has run, so the resulting state (or the deliberate absence of a change) is safe to assert.

func (*Server) Running 

func (s *Server) Running() bool

Running returns true if the server is running.

func (*Server) SetCommitManager 

func (s *Server) SetCommitManager(cm any)

SetCommitManager sets the commit manager. Called by the BGP plugin during configuration to inject a CommitManager created with BGP-specific types. MUST be called before any RPC dispatch (i.e., during init-time registration). NOT safe for concurrent use with CommitManager().

func (*Server) SetConfigLoader 

func (s *Server) SetConfigLoader(loader ConfigLoader)

SetConfigLoader sets the function used by ReloadFromDisk to load the config tree.

func (*Server) SetFullReloadFunc 

func (s *Server) SetFullReloadFunc(fn FullReloadFunc)

SetFullReloadFunc sets the function used by daemon-reload RPC when the hub is wired.

func (*Server) SetProcessSpawner 

func (s *Server) SetProcessSpawner(sp plugin.ProcessSpawner)

SetProcessSpawner sets the PluginManager as the process spawner. When set, runPluginPhase delegates process creation to the spawner instead of creating ProcessManager directly. Must be called before Start. If the spawner supports SetMetricsRegistry (e.g., PluginManager), the server's metrics registry is forwarded for plugin health metrics.

func (*Server) SetRebootFunc 

func (s *Server) SetRebootFunc(fn func())

SetRebootFunc sets the function called for "daemon reboot" commands. Called by the daemon to wire graceful shutdown + OS reboot.

func (*Server) SetShutdownFunc 

func (s *Server) SetShutdownFunc(fn func())

SetShutdownFunc sets a reactor-independent daemon-shutdown callback. The daemon wires it (ungated by BGP) to the same signal-based teardown SIGTERM triggers, so `request shutdown` stops a reactorless daemon (OSPF-only, etc.).

func (*Server) Start 

func (s *Server) Start() error

Start begins accepting connections.

func (*Server) StartWithContext 

func (s *Server) StartWithContext(ctx context.Context) error

StartWithContext begins accepting connections with the given context. External access is via SSH; the plugin server handles only in-process dispatch.

func (*Server) Stop 

func (s *Server) Stop()

Stop signals the server to stop and cleans up resources.

func (*Server) Subscribe 

func (s *Server) Subscribe(namespace, eventType string, handler func(payload any)) func()

Subscribe satisfies the pkg/ze.EventBus interface. It is a thin alias for SubscribeEngineEvent that adapts the handler signature from EngineEventHandler (a named type) to a plain func, which is what ze.EventBus declares.

func (*Server) SubscribeEngineEvent 

func (s *Server) SubscribeEngineEvent(namespace, eventType string, handler EngineEventHandler) func()

SubscribeEngineEvent registers an engine-side handler for stream events matching the given namespace and event type. The returned function unregisters the handler when called; safe to call multiple times.

Handlers fire synchronously from deliverEvent. They must not block on external I/O. The handler receives the publisher's typed payload via `any`; consumers type-assert to the canonical type documented next to the event constant in the publishing package.

Engine subscriptions are parallel to plugin process subscriptions managed by SubscriptionManager. Both fire on the same deliverEvent call.

Subscriptions are NOT validated against the event registry: subscribing to an unknown (namespace, eventType) pair, or to a per-plugin event type that is not yet registered, succeeds silently. Such a subscription is dead until the matching emit arrives. This avoids races with per-plugin event types like "verify-bgp" that are registered dynamically when the plugin starts.

Engine subscribers receive ALL events for the given (namespace, eventType) regardless of direction or peer address. Plugin process subscribers can filter on direction and peer; engine subscribers cannot. This is intended for engine-internal coordination (e.g. config transactions) where direction has no meaning.

A nil handler is rejected: the call returns a no-op unsubscribe function without registering anything. This catches programmer errors loudly via "the handler I just registered never fires" rather than via a nil-pointer panic at first dispatch.

func (*Server) Subscriptions 

func (s *Server) Subscriptions() *SubscriptionManager

Subscriptions returns the subscription manager.

func (*Server) TxLocked 

func (s *Server) TxLocked() bool

TxLocked reports whether a config transaction is in progress.

func (*Server) UpdateProtocolConfig 

func (s *Server) UpdateProtocolConfig(families, customEvents, customSendTypes []string)

UpdateProtocolConfig sets protocol-specific auto-load configuration after the reactor has parsed settings. Called by the protocol plugin's RunEngine after creating the reactor, so that family/event/send auto-load phases have the data.

func (*Server) Wait 

func (s *Server) Wait(ctx context.Context) error

Wait waits for the server to stop.

func (*Server) WaitForStartupComplete 

func (s *Server) WaitForStartupComplete(ctx context.Context) error

WaitForStartupComplete blocks until all plugin startup phases are done. Returns a non-nil error if a config-path plugin failed during startup (e.g., invalid BGP config) or if the context deadline is exceeded.

type ServerConfig 

type ServerConfig struct {
	ConfigPath                string                // Path to config file (for peer save)
	Plugins                   []plugin.PluginConfig // External plugins to spawn
	ConfiguredFamilies        []string              // Families configured on peers (for deferred auto-load)
	ConfiguredCustomEvents    []string              // Custom event types in peer receive config (for auto-load)
	ConfiguredCustomSendTypes []string              // Custom send types in peer send config (for auto-load)
	ConfiguredPaths           []string              // Top-level config sections present (for config-driven auto-load)
	Hub                       *plugin.HubConfig     // TLS transport config (nil = no TLS listener)
	MetricsRegistry           metrics.Registry      // Prometheus metrics registry (nil = metrics disabled)
}

ServerConfig holds API server configuration.

type StreamingHandler 

type StreamingHandler func(ctx context.Context, s *Server, w io.Writer, username string, args []string) error

StreamingHandler handles streaming commands (e.g., monitor). ctx is the session context, s is the plugin server, w is the output writer, username is the authenticated SSH user (for authorization), args are command arguments.

func GetStreamingHandlerForCommand 

func GetStreamingHandlerForCommand(input string) (StreamingHandler, []string)

GetStreamingHandlerForCommand returns the handler and extracted args for a command. Matches the longest registered prefix. Returns (nil, nil) if no prefix matches.

type Subscription 

type Subscription struct {
	Namespace    events.NamespaceID // compact ID assigned at registration time
	EventType    events.EventTypeID // compact ID assigned at registration time
	Direction    events.Direction   // typed enum (DirReceived, DirSent, DirBoth)
	PeerFilter   *PeerFilter        // nil = all peers
	PluginFilter string             // plugin name filter (empty = all)
}

Subscription represents an event subscription.

func BuildEventMonitorSubscriptions 

func BuildEventMonitorSubscriptions(opts *EventMonitorOpts) []*Subscription

BuildEventMonitorSubscriptions creates subscriptions from parsed options. With no include/exclude filter, subscribes to all event types across all namespaces. With include, subscribes only to those types (in whichever namespaces they exist). With exclude, subscribes to all types except those listed.

func ParseSubscription 

func ParseSubscription(args []string) (*Subscription, error)

ParseSubscription parses a subscribe/unsubscribe command. Format: [peer <sel> | plugin <name>] [<namespace>] event <type> [direction received|sent|both]. Namespace defaults to "bgp" when peer is set.

func (*Subscription) Equals 

func (s *Subscription) Equals(other *Subscription) bool

Equals returns true if two subscriptions are identical.

func (*Subscription) Matches 

func (s *Subscription) Matches(ns events.NamespaceID, et events.EventTypeID, dir events.Direction, peerAddr, peerName string) bool

Matches returns true if this subscription matches the event. peerAddr is the peer's IP address; peerName is the configured peer name (may be empty).

type SubscriptionManager 

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

SubscriptionManager tracks subscriptions per process.

func NewSubscriptionManager 

func NewSubscriptionManager() *SubscriptionManager

NewSubscriptionManager creates a new subscription manager.

func (*SubscriptionManager) Add 

func (sm *SubscriptionManager) Add(proc *process.Process, sub *Subscription)

Add adds a subscription for a process.

func (*SubscriptionManager) ClearProcess 

func (sm *SubscriptionManager) ClearProcess(proc *process.Process)

ClearProcess removes all subscriptions for a process.

func (*SubscriptionManager) Count 

func (sm *SubscriptionManager) Count(proc *process.Process) int

Count returns the number of subscriptions for a process.

func (*SubscriptionManager) GetMatching 

func (sm *SubscriptionManager) GetMatching(ns events.NamespaceID, et events.EventTypeID, dir events.Direction, peerAddr, peerName string) []*process.Process

GetMatching returns all processes with subscriptions matching the event. peerName is the configured peer name (may be empty for non-BGP events or emit-event RPCs).

func (*SubscriptionManager) GetSubscriptions 

func (sm *SubscriptionManager) GetSubscriptions(proc *process.Process) []*Subscription

GetSubscriptions returns all subscriptions for a process.

func (*SubscriptionManager) Remove 

func (sm *SubscriptionManager) Remove(proc *process.Process, sub *Subscription) bool

Remove removes a subscription for a process. Returns true if the subscription was found and removed.

type SubsystemConfig 

type SubsystemConfig struct {
	Name       string   // Subsystem name (cache, route, session)
	Binary     string   // Path to binary or full command
	Commands   []string // Commands this subsystem handles (for pre-registration)
	ConfigPath string   // Config file path (passed to child process)
}

SubsystemConfig describes a forked subsystem process.

type SubsystemHandler 

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

SubsystemHandler wraps a forked process that handles a subset of commands. It spawns the subprocess, completes the 5-stage protocol, and routes commands to it via pipes.

func NewSubsystemHandler 

func NewSubsystemHandler(config SubsystemConfig) *SubsystemHandler

NewSubsystemHandler creates a handler backed by a forked process.

func (*SubsystemHandler) Commands 

func (h *SubsystemHandler) Commands() []string

Commands returns the commands this subsystem handles.

func (*SubsystemHandler) Handle 

func (h *SubsystemHandler) Handle(ctx context.Context, command string) (*plugin.Response, error)

Handle sends a command to the subsystem via RPC and returns the response.

func (*SubsystemHandler) Name 

func (h *SubsystemHandler) Name() string

Name returns the subsystem name.

func (*SubsystemHandler) Running 

func (h *SubsystemHandler) Running() bool

Running returns true if the subsystem process is running.

func (*SubsystemHandler) Schema 

func (h *SubsystemHandler) Schema() *plugin.PluginSchemaDecl

Schema returns the YANG schema declared by this subsystem, or nil if none.

func (*SubsystemHandler) Signal 

func (h *SubsystemHandler) Signal(sig os.Signal) error

Signal sends an OS signal to the subsystem's external process. Returns an error if the process is not running or is internal (goroutine).

func (*SubsystemHandler) Start 

func (h *SubsystemHandler) Start(ctx context.Context) error

Start spawns the subsystem process and completes the 5-stage protocol.

func (*SubsystemHandler) Stop 

func (h *SubsystemHandler) Stop()

Stop terminates the subsystem process.

type SubsystemManager 

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

SubsystemManager manages multiple subsystem handlers.

func NewSubsystemManager 

func NewSubsystemManager() *SubsystemManager

NewSubsystemManager creates a new subsystem manager.

func (*SubsystemManager) AllCommands 

func (m *SubsystemManager) AllCommands() []string

AllCommands returns all commands from all subsystems.

func (*SubsystemManager) AllSchemas 

func (m *SubsystemManager) AllSchemas() []*Schema

AllSchemas returns all schemas from all subsystems.

func (*SubsystemManager) FindHandler 

func (m *SubsystemManager) FindHandler(command string) *SubsystemHandler

FindHandler returns the handler for a given command, or nil if not found. After Freeze(), uses lock-free atomic.Load on the frozen snapshot.

func (*SubsystemManager) Freeze 

func (m *SubsystemManager) Freeze()

Freeze creates an immutable snapshot of the handler map. After Freeze(), Get and FindHandler use atomic.Load instead of RLock. MUST be called after all Register calls complete (after startup barrier). Safe to call multiple times (each call overwrites the previous snapshot).

func (*SubsystemManager) Get 

func (m *SubsystemManager) Get(name string) *SubsystemHandler

Get returns a subsystem handler by name. After Freeze(), uses lock-free atomic.Load on the frozen snapshot.

func (*SubsystemManager) Names 

func (m *SubsystemManager) Names() []string

Names returns the names of all registered subsystems.

func (*SubsystemManager) Register 

func (m *SubsystemManager) Register(config SubsystemConfig)

Register adds a subsystem configuration.

func (*SubsystemManager) RegisterSchemas 

func (m *SubsystemManager) RegisterSchemas(registry *SchemaRegistry) error

RegisterSchemas registers all subsystem schemas with the given registry.

func (*SubsystemManager) Replace 

func (m *SubsystemManager) Replace(name string, handler *SubsystemHandler)

Replace swaps in an already-constructed handler and stops the previous one after publishing the replacement. Used by reload paths that pre-start a replacement before disrupting the old subsystem.

func (*SubsystemManager) StartAll 

func (m *SubsystemManager) StartAll(ctx context.Context) error

StartAll starts all registered subsystems.

func (*SubsystemManager) StopAll 

func (m *SubsystemManager) StopAll()

StopAll stops all subsystems.

func (*SubsystemManager) Unregister 

func (m *SubsystemManager) Unregister(name string)

Unregister stops and removes a subsystem by name. No-op if the name is not registered. If frozen, publishes a new snapshot reflecting the removal.

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.