Documentation
¶
Overview ¶
Package stream owns the bidi stream lifecycle between the agent and an admin server: the registration handshake, heartbeats, command dispatch (Subscribe / Unsubscribe / SnapshotRequest / Goodbye), event egress from the in-process bus, and graceful drain on shutdown.
Stream is intentionally an inner detail of the agent. The agent's top- level New/Run constructs streams in a loop; each Stream represents one bidi attempt. A stream's lifetime ends when:
- the server sends Goodbye, or
- the underlying transport closes, or
- the parent context is cancelled.
On any of these the stream returns from Run with a typed error class. The agent inspects the error to decide whether to reconnect (with backoff) or exit.
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ErrServerGoodbye = errors.New("admin agent: server sent Goodbye")
ErrServerGoodbye is returned by Run when the admin server initiated a graceful shutdown of this stream. The agent reconnects.
var ErrTransport = errors.New("admin agent: transport closed")
ErrTransport indicates the underlying HTTP/2 stream broke. The agent reconnects with backoff.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
NodeID string
Version string
Labels map[string]string
StartedAt time.Time
Bus *observability.Bus
Buffer *buffer.PerKind
Metrics *metrics.Metrics
// Host, when non-nil, contributes a HostMetrics sample to every
// heartbeat frame (see agent/hostmetrics).
Host interface{ Collect() *adminv1.HostMetrics }
Logger *slog.Logger
Heartbeat time.Duration
DrainTimeout time.Duration
// DataStudio, when non-nil, enables the agent-side Data Studio
// dispatcher. The stream forwards DataStudioRequest commands to it
// and ships the resulting DataStudioResponse back over the bidi
// stream. Models reported by RegisteredModels() are included in
// NodeRegistration so the admin server can route requests to the
// right agent.
DataStudio DataStudioDispatcher
// Rbac, when non-nil, enables the agent-side RBAC snapshot
// dispatcher (agent/rbac.Handler). The stream forwards RbacRequest
// commands to it and ships the RbacResponse back.
Rbac RbacDispatcher
// OnAccepted, when non-nil, is invoked at most once per Stream, on
// the first frame successfully received FROM the server. That first
// frame is the earliest hard evidence that the server authenticated
// and accepted this stream (a Send success is not: the client may
// buffer frames locally while the server is already rejecting the
// call with 401). The agent uses it to reset the dial backoff and
// log the honest "connected" line (OR5-2).
OnAccepted func()
}
Config bundles the dependencies a Stream needs.
type DataStudioDispatcher ¶
type DataStudioDispatcher interface {
Dispatch(ctx context.Context, req *adminv1.DataStudioRequest) *adminv1.DataStudioResponse
RegisteredModels() []string
}
DataStudioDispatcher executes a Data Studio request and returns the matching response. Concrete implementation lives in admin/agent/datastudio.Handler; the interface lets stream.go stay independent of pkg/model imports (which would otherwise create a dependency loop with pkg/app).
type RbacDispatcher ¶
type RbacDispatcher interface {
Dispatch(req *adminv1.RbacRequest) *adminv1.RbacResponse
}
RbacDispatcher answers an RBAC snapshot request. Concrete implementation lives in agent/rbac.Handler; the interface keeps stream.go free of authz imports.
type Stream ¶
type Stream struct {
// contains filtered or unexported fields
}
Stream owns one bidi connection. Construct via New, drive via Run.
func New ¶
func New(client adminv1connect.AgentServiceClient, cfg Config) *Stream
New constructs a Stream wrapping the given AgentService client. Run must be called to drive it; the constructor performs no network IO.
Source Files