control

Documentation

Overview

Package control implements the loom.v1.Control gRPC service — the controller<->agent control plane (DESIGN.md §8). This is the skeleton: the coordination RPCs (Health/Register/Capabilities/TimeSync) work; the flow lifecycle RPCs are inherited as Unimplemented until the agent fills them in.

Index

• Constants
• func Dial(addr string, opts ...DialOption) (loomv1.ControlClient, *grpc.ClientConn, error)
• func NewGRPCServer(s *Server) *grpc.Server
• func Sync(ctx context.Context, c loomv1.ControlClient) (timesync.Sample, error)
• type DialOption
  • func WithToken(token string) DialOption
• type Option
  • func WithAuthToken(token string) Option
  • func WithComponents(c *components.Components) Option
  • func WithMaxFlows(n int) Option
  • func WithTelemetryInterval(d time.Duration) Option
• type Server
  • func NewServer(version string, opts ...Option) *Server
  • func (s *Server) Arm(_ context.Context, req *loomv1.ArmRequest) (*loomv1.ArmResponse, error)
  • func (s *Server) AuthEnabled() bool
  • func (s *Server) Capabilities(context.Context, *loomv1.CapabilitiesRequest) (*loomv1.CapabilitiesResponse, error)
  • func (s *Server) Configure(_ context.Context, req *loomv1.ConfigureRequest) (*loomv1.ConfigureResponse, error)
  • func (s *Server) Destroy(_ context.Context, req *loomv1.DestroyRequest) (*loomv1.DestroyResponse, error)
  • func (s *Server) Health(context.Context, *loomv1.HealthRequest) (*loomv1.HealthResponse, error)
  • func (s *Server) Register(_ context.Context, req *loomv1.RegisterRequest) (*loomv1.RegisterResponse, error)
  • func (s *Server) Start(_ context.Context, req *loomv1.StartRequest) (*loomv1.StartResponse, error)
  • func (s *Server) Stop(_ context.Context, req *loomv1.StopRequest) (*loomv1.StopResponse, error)
  • func (s *Server) StreamTelemetry(req *loomv1.TelemetryRequest, stream loomv1.Control_StreamTelemetryServer) error
  • func (s *Server) TimeSync(_ context.Context, req *loomv1.TimeSyncRequest) (*loomv1.TimeSyncResponse, error)

Constants

const APIVersion = 1

APIVersion is the control-plane wire version this build speaks (ADR-0021). Bump it on a breaking proto change so peers can detect a mismatch.

Functions

func Dial

func Dial(addr string, opts ...DialOption) (loomv1.ControlClient, *grpc.ClientConn, error)

Dial connects a control client to addr (host:port). It uses insecure transport for now; optional mTLS (ADR-0014) is added later. Pass WithToken to authenticate against an agent that requires a control-plane token. The returned connection must be closed by the caller.

func NewGRPCServer

func NewGRPCServer(s *Server) *grpc.Server

NewGRPCServer builds a *grpc.Server with the control service registered. When the Server has an auth token set, token-checking unary/stream interceptors are installed.

func Sync

func Sync(ctx context.Context, c loomv1.ControlClient) (timesync.Sample, error)

Sync performs one four-timestamp time-sync exchange against an agent and returns the estimated clock offset and round-trip delay. The caller stamps t1 before the RPC and t4 after it; the agent stamps t2/t3 (see Server.TimeSync and the timesync package).

Types

type DialOption

type DialOption func(*dialConfig)

DialOption configures a control-plane dial.

func WithToken

func WithToken(token string) DialOption

WithToken attaches a shared bearer token (ADR-0014) to every RPC on the connection. An empty token is a no-op, so callers can pass it unconditionally.

type Option

type Option func(*Server)

Option configures a Server (ADR-0022). Apply at construction via NewServer.

func WithAuthToken

func WithAuthToken(token string) Option

WithAuthToken sets the shared control-plane token (ADR-0014). When non-empty, NewGRPCServer installs interceptors that reject any RPC lacking a matching bearer token; empty leaves the plane open.

func WithComponents

func WithComponents(c *components.Components) Option

WithComponents sets the datapath/generator/scheduler/payload registries this agent builds flows from and advertises via Capabilities. Defaults to components.Default().

func WithMaxFlows

func WithMaxFlows(n int) Option

WithMaxFlows caps concurrently configured flows; Configure returns ResourceExhausted past the cap. n <= 0 removes the limit (not recommended on a reachable agent).

func WithTelemetryInterval

func WithTelemetryInterval(d time.Duration) Option

WithTelemetryInterval sets how often StreamTelemetry emits samples (0 = 1s).

type Server

type Server struct {
	loomv1.UnimplementedControlServer
	// contains filtered or unexported fields
}

Server implements loomv1.ControlServer. Flow-lifecycle RPCs (Configure/Arm/Start/Stop/Destroy/StreamTelemetry) are inherited from UnimplementedControlServer and return codes.Unimplemented for now.

func NewServer

func NewServer(version string, opts ...Option) *Server

NewServer returns a control Server reporting the given version, configured by opts.

func (*Server) Arm

func (s *Server) Arm(_ context.Context, req *loomv1.ArmRequest) (*loomv1.ArmResponse, error)

Arm is a no-op for now (receivers/ephemeral ports arrive later).

func (*Server) AuthEnabled

func (s *Server) AuthEnabled() bool

AuthEnabled reports whether a control-plane token is configured.

func (*Server) Capabilities

func (s *Server) Capabilities(context.Context, *loomv1.CapabilitiesRequest) (*loomv1.CapabilitiesResponse, error)

Capabilities reports what this agent can do, from its configured components.

func (*Server) Configure

func (s *Server) Configure(_ context.Context, req *loomv1.ConfigureRequest) (*loomv1.ConfigureResponse, error)

Configure builds and stores a flow, returning its id. Ephemeral data-port assignment is a later step; data_port is 0 for now.

func (*Server) Destroy

func (s *Server) Destroy(_ context.Context, req *loomv1.DestroyRequest) (*loomv1.DestroyResponse, error)

Destroy tears a flow down and releases its datapath.

func (*Server) Health

func (s *Server) Health(context.Context, *loomv1.HealthRequest) (*loomv1.HealthResponse, error)

Health reports liveness, build version, and the wire API version.

func (*Server) Register

func (s *Server) Register(_ context.Context, req *loomv1.RegisterRequest) (*loomv1.RegisterResponse, error)

Register enrolls an agent. Minimal for now: the agent id is echoed as the session. Token/mTLS enrollment (ADR-0014) arrives with the controller/agent.

func (*Server) Start

func (s *Server) Start(_ context.Context, req *loomv1.StartRequest) (*loomv1.StartResponse, error)

Start runs a configured flow, optionally at a scheduled time (start_at) on this agent's clock so flows across agents begin in lockstep.

func (*Server) Stop

func (s *Server) Stop(_ context.Context, req *loomv1.StopRequest) (*loomv1.StopResponse, error)

Stop cancels a running flow.

func (*Server) StreamTelemetry

func (s *Server) StreamTelemetry(req *loomv1.TelemetryRequest, stream loomv1.Control_StreamTelemetryServer) error

StreamTelemetry streams a flow's telemetry until the flow finishes or the client disconnects. When the flow was started with a report interval I, samples are anchored to the gate: one per boundary at startAt + k*I, each carrying that interval's delta (the controller is then a pure summer, with no second clock). With no interval (legacy controllers, report_interval_nanos == 0) it falls back to a free-running cadence.

func (*Server) TimeSync

func (s *Server) TimeSync(_ context.Context, req *loomv1.TimeSyncRequest) (*loomv1.TimeSyncResponse, error)

TimeSync stamps t2 on receipt and t3 on send for the four-timestamp exchange.