Documentation
¶
Overview ¶
Package control carries commands down and telemetry up between the tiers of a LoadWave cluster.
Both tiers speak the same protocol: an agent joining a coordinator over TCP and a worker process joining its agent over a Unix socket use the identical client and server here. The node always dials, and holds one long-lived bidirectional stream open for as long as it is alive. That direction is a deliberate choice — it lets agents live behind NAT with no inbound ports, and it makes a broken stream an unambiguous liveness signal instead of something that has to be inferred from timeouts.
Index ¶
- Constants
- Variables
- type Client
- func (c *Client) Connected() bool
- func (c *Client) Dropped() uint64
- func (c *Client) Run(ctx context.Context) error
- func (c *Client) Send(msg *loadwavev1.NodeUp) bool
- func (c *Client) SendLog(event *loadwavev1.LogEvent) bool
- func (c *Client) SendMetrics(batch *loadwavev1.MetricBatch) bool
- func (c *Client) SendRunStatus(update *loadwavev1.RunStatusUpdate) bool
- type ClientConfig
- type Handler
- type Server
- type ServerConfig
- type Session
- type SessionHandler
- type SessionRegistry
- func (r *SessionRegistry) Add(session *Session) *Session
- func (r *SessionRegistry) All() []*Session
- func (r *SessionRegistry) Broadcast(msg *loadwavev1.NodeDown) error
- func (r *SessionRegistry) Get(id string) (*Session, bool)
- func (r *SessionRegistry) Len() int
- func (r *SessionRegistry) Remove(session *Session) bool
Constants ¶
const ( DefaultQueueSize = 1024 DefaultReconnectMin = 250 * time.Millisecond DefaultReconnectMax = 15 * time.Second // DefaultHeartbeatInterval is used until the supervisor says otherwise. DefaultHeartbeatInterval = 2 * time.Second )
Defaults applied to a zero ClientConfig.
const ( DefaultServerHeartbeatInterval = 2 * time.Second DefaultMetricsInterval = time.Second DefaultServerQueueSize = 64 )
Defaults applied to a zero ServerConfig.
Variables ¶
var ErrSessionClosed = errors.New("control session is closed")
ErrSessionClosed is returned by Session.Send once the node has gone.
Functions ¶
This section is empty.
Types ¶
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client maintains a node's control stream, reconnecting as needed.
func NewClient ¶
func NewClient(cfg ClientConfig) (*Client, error)
NewClient validates the configuration and prepares a client. It does not connect; Run does that.
func (*Client) Dropped ¶
Dropped reports how many upstream messages were discarded because the queue was full.
func (*Client) Run ¶
Run connects and keeps the stream alive until ctx is cancelled.
It returns nil on cancellation. Any other error means the client gave up, which currently only happens if the target cannot be parsed at all.
func (*Client) Send ¶
func (c *Client) Send(msg *loadwavev1.NodeUp) bool
Send queues a message for the supervisor, reporting whether it was accepted.
It never blocks. Telemetry is worth less than throughput, so a full queue discards the message and increments the drop counter, which is reported in the next batch and surfaced in the dashboard. Nothing here is load-bearing for correctness — the supervisor tolerates gaps by design.
func (*Client) SendLog ¶
func (c *Client) SendLog(event *loadwavev1.LogEvent) bool
SendLog queues a log event.
func (*Client) SendMetrics ¶
func (c *Client) SendMetrics(batch *loadwavev1.MetricBatch) bool
SendMetrics queues a metric batch.
func (*Client) SendRunStatus ¶
func (c *Client) SendRunStatus(update *loadwavev1.RunStatusUpdate) bool
SendRunStatus queues a run status update.
type ClientConfig ¶
type ClientConfig struct {
// Target is a gRPC target: "host:port" for a coordinator, or
// "unix:///path/to.sock" for a local agent.
Target string
// Hello advertises this node's identity and capacity. It is re-sent on
// every reconnection, so a supervisor always has current information.
Hello *loadwavev1.NodeHello
// Handler receives downstream commands.
Handler Handler
// Heartbeat supplies the current statistics each time a heartbeat is due.
// Optional; without it, heartbeats carry only a sequence number.
Heartbeat func() *loadwavev1.NodeHeartbeat
Logger *slog.Logger
// QueueSize bounds the upstream buffer. When it fills, telemetry is
// dropped rather than allowed to block the caller: a load generator must
// not slow down because its reporting channel is congested. Zero applies
// DefaultQueueSize.
QueueSize int
// ReconnectMin and ReconnectMax bound the backoff between attempts.
ReconnectMin time.Duration
ReconnectMax time.Duration
// DialOptions are appended to the defaults, for TLS and interceptors.
DialOptions []grpc.DialOption
}
ClientConfig configures a node's connection to its supervisor.
type Handler ¶
type Handler interface {
OnAccepted(ctx context.Context, msg *loadwavev1.Accepted) error
OnStartRun(ctx context.Context, msg *loadwavev1.StartRun) error
OnSetQuota(ctx context.Context, msg *loadwavev1.SetQuota) error
OnStopRun(ctx context.Context, msg *loadwavev1.StopRun) error
}
Handler receives the commands a supervisor sends down the stream.
Calls are made from the client's receive goroutine, one at a time and in order. An implementation that needs to do something slow should hand off to its own goroutine rather than block, since a stalled handler stops the node noticing anything else — including a stop command.
type Server ¶
type Server struct {
loadwavev1.UnimplementedControlServiceServer
// contains filtered or unexported fields
}
Server implements the ControlService for one supervisor tier.
func NewServer ¶
func NewServer(cfg ServerConfig) (*Server, error)
NewServer prepares a supervisor endpoint.
func (*Server) Join ¶
func (s *Server) Join(stream loadwavev1.ControlService_JoinServer) error
Join implements loadwavev1.ControlServiceServer.
type ServerConfig ¶
type ServerConfig struct {
Handler SessionHandler
Logger *slog.Logger
// Version is reported to nodes so they can warn about a mismatch.
Version string
// HeartbeatInterval is how often nodes should report liveness.
HeartbeatInterval time.Duration
// MetricsInterval is how often nodes should flush metric batches. It
// doubles as the bucket width the coordinator's store expects.
MetricsInterval time.Duration
// QueueSize bounds each session's outbound command buffer.
QueueSize int
}
ServerConfig configures a supervisor's end of the protocol.
type Session ¶
type Session struct {
// ID is the node's self-declared identifier. Reconnections reuse it,
// which is how a supervisor recognises a returning node rather than
// treating it as a new one.
ID string
// Hello is what the node advertised on its most recent connection.
Hello *loadwavev1.NodeHello
// RemoteAddr is the peer address, for operator-facing display.
RemoteAddr string
// JoinedAt is when this particular connection was established.
JoinedAt time.Time
// contains filtered or unexported fields
}
Session is a supervisor's handle on one connected node.
It is safe for concurrent use, and remains usable until the node disconnects, at which point every Send fails with ErrSessionClosed.
func (*Session) Send ¶
func (s *Session) Send(msg *loadwavev1.NodeDown) error
Send queues a command for the node.
Commands, unlike telemetry, matter: dropping a StopRun would leave a node generating load nobody asked for. So a full queue is an error the caller must deal with, rather than something silently swallowed. In practice the queue only fills when a node has stopped reading, which the stream will shortly report anyway.
type SessionHandler ¶
type SessionHandler interface {
// OnJoin is called once the node has identified itself. Returning an
// error rejects the connection, and the node will retry.
OnJoin(ctx context.Context, session *Session) error
// OnLeave is called exactly once per accepted session, when it ends.
OnLeave(session *Session)
OnHeartbeat(session *Session, beat *loadwavev1.NodeHeartbeat)
OnMetrics(session *Session, batch *loadwavev1.MetricBatch)
OnRunStatus(session *Session, update *loadwavev1.RunStatusUpdate)
OnLog(session *Session, event *loadwavev1.LogEvent)
}
SessionHandler receives everything a connected node does.
Every method is called from that node's own receive goroutine, so implementations must be safe for concurrent use across sessions, and must not block for long.
type SessionRegistry ¶
type SessionRegistry struct {
// contains filtered or unexported fields
}
SessionRegistry tracks the nodes currently connected to a supervisor.
It is the piece that makes reconnection work: a node that drops and comes back reuses its id, and the registry replaces the stale session rather than accumulating ghosts. Both the coordinator and the agent embed one.
func NewSessionRegistry ¶
func NewSessionRegistry() *SessionRegistry
NewSessionRegistry returns an empty registry.
func (*SessionRegistry) Add ¶
func (r *SessionRegistry) Add(session *Session) *Session
Add registers a session, displacing and closing any previous one with the same node id. It reports the displaced session, if there was one.
func (*SessionRegistry) All ¶
func (r *SessionRegistry) All() []*Session
All returns every live session.
func (*SessionRegistry) Broadcast ¶
func (r *SessionRegistry) Broadcast(msg *loadwavev1.NodeDown) error
Broadcast sends a command to every connected node, returning the first error. Delivery to the remaining nodes is attempted regardless: a run must still stop on nine agents when the tenth is unreachable.
func (*SessionRegistry) Get ¶
func (r *SessionRegistry) Get(id string) (*Session, bool)
Get returns the current session for a node id.
func (*SessionRegistry) Len ¶
func (r *SessionRegistry) Len() int
Len reports how many nodes are connected.
func (*SessionRegistry) Remove ¶
func (r *SessionRegistry) Remove(session *Session) bool
Remove deregisters a session, but only if it is still the current one for its node id.
The guard matters during a reconnection race: the new session can be registered before the old one's goroutine has finished unwinding, and an unguarded delete would evict the live session on behalf of the dead one.
Source Files