Documentation
¶
Overview ¶
Package agent is TASKS.md 3.1's transport boundary: the interface ADR 003 describes ("the reconciler and everything above the transport boundary never knows whether it's talking to a local in-process agent or a remote one over mTLS") but that Phase 1 never actually built, confirmed directly against this repo before writing this package: every reconcile controller (internal/reconcile/application, internal/reconcile/database) takes a bare docker.Runtime, and cmd/levelrail/main.go's dynamicSource hands every controller the same single local docker.Runtime with no node concept anywhere. There is exactly one implicit node today.
This package makes that boundary explicit without moving it: Transport has exactly docker.Runtime's method set (see the type below for why), so nothing above this package changes shape yet. Wiring internal/reconcile controllers to actually route through a Transport selected per node, instead of the single shared docker.Runtime dynamicSource still hands out today, is TASKS.md 3.3's job, once placement (which service runs on which node) exists to select by. The real gRPC implementation of Transport, reached over the reverse- dialed mTLS connection ADR 003 describes, is TASKS.md 3.2.
Index ¶
- Variables
- func CertFingerprint(der []byte) string
- func Execute(ctx context.Context, rt docker.Runtime, req *agentpb.AgentRequest, ...) *agentpb.AgentResponse
- func NewServerCredentials(ca *CA, hosts []string, validFor time.Duration) (credentials.TransportCredentials, error)
- func RunSession(ctx context.Context, addr string, id *Identity, rt docker.Runtime, ...) error
- type CA
- type EnrollStore
- type GRPCTransport
- func (t *GRPCTransport) Create(ctx context.Context, spec docker.ContainerSpec) (string, error)
- func (t *GRPCTransport) EnsureNetwork(_ context.Context, name string) (string, error)
- func (t *GRPCTransport) EnsureVolume(ctx context.Context, name string) error
- func (t *GRPCTransport) Events(ctx context.Context) (<-chan docker.Event, <-chan error)
- func (t *GRPCTransport) Exec(_ context.Context, containerID string, cmd []string) (io.ReadCloser, error)
- func (t *GRPCTransport) ExecWithInput(_ context.Context, containerID string, cmd []string, _ io.Reader) (io.ReadCloser, error)
- func (t *GRPCTransport) InspectByName(ctx context.Context, name string) (*docker.ContainerState, error)
- func (t *GRPCTransport) ListByPrefix(ctx context.Context, prefix string) ([]docker.ContainerState, error)
- func (t *GRPCTransport) ListImages(ctx context.Context, repo string) ([]docker.ImageInfo, error)
- func (t *GRPCTransport) ListNetworksByPrefix(_ context.Context, prefix string) ([]docker.NetworkInfo, error)
- func (t *GRPCTransport) Remove(ctx context.Context, id string, force bool) error
- func (t *GRPCTransport) RemoveNetwork(_ context.Context, name string) error
- func (t *GRPCTransport) Start(ctx context.Context, id string) error
- func (t *GRPCTransport) Stop(ctx context.Context, id string, timeout time.Duration) error
- func (t *GRPCTransport) UpdateResources(ctx context.Context, id string, resources docker.Resources) error
- type Identity
- type Local
- type Option
- type Registry
- type Server
- type Transport
Constants ¶
This section is empty.
Variables ¶
var ErrNodeNotRegistered = errors.New("agent: node not registered in this transport registry")
ErrNodeNotRegistered is Registry.Get's failure mode for an unknown node ID.
var ErrSessionClosed = errors.New("agent: session closed")
ErrSessionClosed is returned by mux.Call once the underlying stream has ended, for any pending or future call.
Functions ¶
func CertFingerprint ¶
CertFingerprint returns a stable, hex-encoded SHA-256 digest of a DER-encoded certificate: store.Node.CertFingerprint's own value, and what the control plane's Session handler (TASKS.md 3.2) recomputes from an incoming mTLS connection's actual peer certificate to confirm it matches the fingerprint recorded at enrollment, rather than trusting the certificate's CommonName field alone (which a differently-issued certificate could also claim).
func Execute ¶
func Execute(ctx context.Context, rt docker.Runtime, req *agentpb.AgentRequest, emitEvent func(*agentpb.ProxiedEvent)) *agentpb.AgentResponse
Execute runs one AgentRequest against rt and returns the matching AgentResponse, always carrying req.RequestId. A docker.Runtime failure becomes AgentResponse.Error (a plain string over the wire, deliberately: this is a control-plane-to-agent RPC boundary, not a package boundary within one Go process, so wrapping with %w has nothing left to attach to on the receiving end; GRPCTransport reconstructs a plain error from this string instead), never a Go error return of Execute's own: this function's contract is "always produce exactly one response for exactly one request," success or failure alike, so the caller always has something to send back.
ctx is the whole Session's own lifetime, not a fresh per-call one: WatchEvents in particular needs to keep running past this one call returning, for as long as the session itself stays open, and using the same ctx for every other operation keeps "when does this stop" answered identically everywhere in this function rather than inventing a second, shorter-lived context needing its own justification. emitEvent is only ever called from the WatchEvents path, unprompted, for as long as ctx stays alive.
func NewServerCredentials ¶
func NewServerCredentials(ca *CA, hosts []string, validFor time.Duration) (credentials.TransportCredentials, error)
NewServerCredentials issues a fresh server certificate from ca (covering hosts) and returns grpc TransportCredentials configured for TASKS.md 3.2's mTLS model: a client certificate is verified when presented but not required (VerifyClientCertIfGiven), because Enroll (ADR 003's join-token exchange) is called with no client certificate at all, by design (DialEnroll's own doc comment explains why), while Session always presents one, verified against ca when it does.
Must be passed to grpc.NewServer via grpc.Creds, not used to build a raw tls.Listener handed to Serve directly: grpc-go's own peer-info extraction (what Server.Session's peerIdentity relies on to read the client certificate back out of a request's context) only populates correctly when grpc's credentials layer performs the handshake itself. A raw tls.Listener does the TLS handshake transparently at the net.Conn level, which works for plain data transfer but leaves grpc with no TLS state to attach to the connection's context, confirmed the hard way: an earlier version of this wiring, tested against a real listener, failed with "no peer TLS info on this connection" on every real mTLS connection until this file switched to grpc.Creds.
func RunSession ¶
func RunSession(ctx context.Context, addr string, id *Identity, rt docker.Runtime, logger *slog.Logger) error
RunSession dials addr with id's mTLS credentials (verified against id.CACertPEM, closing DialEnroll's own TOFU window), opens the one persistent Session stream ADR 003 describes, and serves every incoming AgentRequest against rt until ctx is cancelled or the connection fails. Returns the error that ended the session (nil only if ctx itself was the cause); never retries or reconnects on its own, that's cmd/levelrail-agent's own reconnect loop's job (TASKS.md 3.2's remaining wiring, and ADR 003's Consequences section's own "real, tested" reconnection/backpressure/version-negotiation requirement), kept out of this function so it stays a single, directly testable connection attempt rather than a policy about how many times or how fast to retry.
Types ¶
type CA ¶
type CA struct {
// contains filtered or unexported fields
}
CA is the control plane's own certificate authority: one self-signed root, generated once (or loaded from disk across restarts) and used to sign both the control plane's own gRPC server certificate and every enrolling agent's client certificate.
func GenerateCA ¶
GenerateCA creates a new self-signed CA. ed25519 throughout this package (CA and every leaf certificate): fast key generation, small keys, and full support in Go's crypto/tls for both signing and TLS handshakes, with none of RSA's parameter-size decisions to make.
func LoadCA ¶
LoadCA parses a CA previously persisted via CertPEM/KeyPEM, the counterpart GenerateCA's caller needs across a control plane restart: generating a fresh CA on every startup would invalidate every already-enrolled agent's certificate for no reason.
func (*CA) CertPEM ¶
CertPEM returns the CA's own certificate, PEM-encoded: what every agent needs to verify the control plane's server certificate (returned to an enrolling agent as EnrollResponse.CaCertPem), and what this process needs to persist to disk to survive a restart via LoadCA.
func (*CA) IssueClientCert ¶
func (ca *CA) IssueClientCert(commonName string, validFor time.Duration) (certPEM, keyPEM []byte, err error)
IssueClientCert issues a client certificate identifying commonName (a node's store.Node.ID), signed by ca, for TASKS.md 3.2's enrollment flow. validFor has no auto-renewal built in here: certificate rotation on a schedule is real, named Phase 3 scope (ADR 003's Consequences section), a caller-level concern layered on top of this primitive, not solved by this one function alone.
func (*CA) IssueServerCert ¶
func (ca *CA) IssueServerCert(hosts []string, validFor time.Duration) (certPEM, keyPEM []byte, err error)
IssueServerCert issues the control plane's own gRPC listener TLS certificate, signed by ca, valid for the given hostnames/IPs agents will dial. Not agent-facing in the enrollment response the way IssueClientCert's output is: the control plane loads this cert locally for its own gRPC server, per cmd/levelrail's own startup wiring, not built here.
type EnrollStore ¶
type EnrollStore interface {
GetNodeJoinTokenByHash(ctx context.Context, hash string) (*store.NodeJoinToken, error)
MarkNodeJoinTokenUsed(ctx context.Context, id string) error
SaveNode(ctx context.Context, n store.Node) error
GetNode(ctx context.Context, id string) (*store.Node, error)
UpdateNodeStatus(ctx context.Context, id string, status store.NodeStatus) error
TouchNodeLastSeen(ctx context.Context, id string) error
}
EnrollStore is the narrow store surface Server needs: validate and consume a join token (TASKS.md 3.1), persist the newly enrolled node, and track its connection state. *store.DB satisfies this structurally.
type GRPCTransport ¶
type GRPCTransport struct {
// contains filtered or unexported fields
}
GRPCTransport implements Transport by dispatching every call over an established Session stream to a real, remote agent. Built from a *mux, not a raw stream: the request/response multiplexing this type depends on is already handled underneath it. Unexported constructor (newGRPCTransport): callers get one of these from the control plane's Session handler once an agent's stream is accepted and authenticated (server.go, TASKS.md 3.2's remaining wiring), never constructed directly.
func (*GRPCTransport) Create ¶
func (t *GRPCTransport) Create(ctx context.Context, spec docker.ContainerSpec) (string, error)
Create implements Transport (docker.Runtime).
func (*GRPCTransport) EnsureNetwork ¶
EnsureNetwork implements Transport (docker.Runtime). Deliberately not wired to the remote agent yet, the same documented gap Exec's own doc comment describes: a new agentpb op and proto regeneration, not a small extension of the existing dispatch switch. Per-app Docker networking (this method's own caller, internal/reconcile/application's Controller) is single-node scope today; a service placed on a remote node fails loudly here rather than silently reconciling without one.
func (*GRPCTransport) EnsureVolume ¶
func (t *GRPCTransport) EnsureVolume(ctx context.Context, name string) error
EnsureVolume implements Transport (docker.Runtime).
func (*GRPCTransport) Events ¶
Events implements Transport (docker.Runtime). The one method that doesn't fit mux.Call's request/response shape: docker.Runtime.Events returns two channels streaming until ctx is cancelled, not one response. This issues a single WatchEvents request (acked immediately like any other Empty-result call, then relays every ProxiedEvent mux.subscribe's channel receives onto a fresh pair of channels this method returns, translating docker.Event's wire shape back and closing both channels once ctx is done or the subscription itself ends (session closed, agent-side stream error). ADR 003's push-not- poll event design, real end to end.
Backpressure: mux's eventChanBuffer (64) bounds how far this can fall behind the agent's own event rate before frames start being dropped rather than blocking the shared recvLoop; a caller needing a guarantee stronger than "recent events, best effort" would need a different mechanism, not built here.
func (*GRPCTransport) Exec ¶
func (t *GRPCTransport) Exec(_ context.Context, containerID string, cmd []string) (io.ReadCloser, error)
Exec implements Transport (docker.Runtime). Deliberately not wired to the remote agent yet: this is a known, documented gap, not an oversight. docker.Runtime.Exec was added for internal/backup's Dumper, and every other Transport method here fits mux.Call's single request/response shape (or, for Events, its own explicit streaming workaround, see Events' doc comment above); Exec's ReadCloser return needs the same kind of dedicated streaming path Events required, plus a new AgentRequest/AgentResponse op in internal/agent/agentpb, which means a proto change and regeneration, not a small extension of the existing dispatch switch in execute.go. That is real, separate work. Rather than fake it with a call that would silently produce no data or the wrong data, a database backup for a service placed on a remote node fails loudly with this error today; only Local's in-process Transport (this control plane's own node) supports Exec until the wiring above lands.
func (*GRPCTransport) ExecWithInput ¶
func (t *GRPCTransport) ExecWithInput(_ context.Context, containerID string, cmd []string, _ io.Reader) (io.ReadCloser, error)
ExecWithInput implements Transport (docker.Runtime). Exec's own doc comment applies identically here, and then some: ExecWithInput needs everything Exec would (a dedicated streaming path, a new agentpb op) plus a second stream direction for stdin, so it inherits the same documented gap rather than a new one. A database restore for a service placed on a remote node fails loudly with this error today, the same "fail loudly, never fake it" posture Exec's own doc comment describes, until the wiring above lands; only Local's in-process Transport (this control plane's own node) supports ExecWithInput until then.
func (*GRPCTransport) InspectByName ¶
func (t *GRPCTransport) InspectByName(ctx context.Context, name string) (*docker.ContainerState, error)
InspectByName implements Transport (docker.Runtime).
func (*GRPCTransport) ListByPrefix ¶
func (t *GRPCTransport) ListByPrefix(ctx context.Context, prefix string) ([]docker.ContainerState, error)
ListByPrefix implements Transport (docker.Runtime).
func (*GRPCTransport) ListImages ¶
ListImages implements Transport (docker.Runtime).
func (*GRPCTransport) ListNetworksByPrefix ¶
func (t *GRPCTransport) ListNetworksByPrefix(_ context.Context, prefix string) ([]docker.NetworkInfo, error)
ListNetworksByPrefix implements Transport (docker.Runtime). Same documented gap as EnsureNetwork above.
func (*GRPCTransport) RemoveNetwork ¶
func (t *GRPCTransport) RemoveNetwork(_ context.Context, name string) error
RemoveNetwork implements Transport (docker.Runtime). Same documented gap as EnsureNetwork above.
func (*GRPCTransport) Start ¶
func (t *GRPCTransport) Start(ctx context.Context, id string) error
Start implements Transport (docker.Runtime).
func (*GRPCTransport) Stop ¶
Stop implements Transport (docker.Runtime). timeout < 0 (wait indefinitely, docker.Runtime.Stop's own convention) is carried across the wire as TimeoutMs: -1, not clamped to 0, so the agent applies the identical "wait indefinitely" semantics rather than a silently different one.
func (*GRPCTransport) UpdateResources ¶
func (t *GRPCTransport) UpdateResources(ctx context.Context, id string, resources docker.Resources) error
UpdateResources implements Transport (docker.Runtime).
type Identity ¶
Identity is what an enrolled node needs to reconnect: its own client certificate/key and the control plane's CA certificate, all PEM, exactly EnrollResponse's three credential fields. cmd/levelrail-agent owns persisting/loading this to/from disk across restarts; this package only consumes it.
func DialEnroll ¶
DialEnroll connects to addr and exchanges joinToken for an Identity.
The connection for this one call is not verified against any CA (InsecureSkipVerify): there is no CA certificate to verify against yet, obtaining one is what this call is for. Trust here rests entirely on joinToken's own secrecy, a trust-on-first-use model (the same one k3s and Nomad's own join-token bootstrapping use), not on TLS server verification: a real, deliberate tradeoff, not an oversight. An attacker able to both intercept this one connection and obtain a valid, unexpired, not-yet-used join token could complete a fraudulent enrollment; the join token being a genuine secret (minted server-side, shown once, single-use, TASKS.md 3.1) is what actually carries the security weight here, not this connection's transport. Every connection after this one (RunSession below, and any future re-enrollment once an Identity already exists) verifies the server certificate against the CA this call returns, closing that window to this one bootstrap step only.
type Local ¶
Local adapts a docker.Runtime this process already has open (this machine's own Docker socket, via docker.NewClient) into a Transport for this process's own node. This is the node-communication design's "single-node mode... in-memory transport that implements the same interface the gRPC transport implements," made real: every method call happens in-process, no network, no serialization, and (because Transport is exactly docker.Runtime's shape) any docker.Runtime value already satisfies Transport structurally without needing this wrapper type at all. Local exists anyway, as a named, documented adapter, so call sites read "this is a node transport" rather than relying on Go's structural typing to make that intent legible.
type Option ¶
type Option func(*Server)
Option configures optional Server behavior.
func WithHeartbeatInterval ¶
WithHeartbeatInterval overrides how often Session touches last_seen_at for a connected node. Without one configured, defaultHeartbeatInterval applies. cmd/levelrail's own main.go reads APP_NODE_HEARTBEAT_INTERVAL and passes the parsed duration here, following the project's "no hardcoded thresholds, use env vars" rule; this package itself never reads the environment directly, the same "constructor args only" convention api.WithSessionTTL already establishes.
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry resolves a node ID to the Transport that reaches it. A single-node deployment registers exactly one entry (its own Local transport, dynamicSource's own local docker.Runtime wrapped per Local's doc comment); a multi-node deployment (TASKS.md 3.2/3.3) registers one entry per enrolled node, most of them real gRPC transports. Registry itself has no opinion on how a node's transport got built or how long it should live, only "given an ID, hand back the Transport for it," the same "pure lookup, not a decision" shape GetAPITokenByHash establishes for a different resource in internal/store.
func (*Registry) Register ¶
Register associates nodeID with t, replacing any existing entry for that ID. Safe for concurrent use.
func (*Registry) Unregister ¶
Unregister removes nodeID's entry, if any. Not an error if nodeID was never registered, matching this codebase's established idempotent- delete convention (store.DeleteDesiredService, store.DeleteNode).
type Server ¶
type Server struct {
agentpb.UnimplementedAgentServiceServer
// contains filtered or unexported fields
}
Server implements agentpb.AgentServiceServer.
func NewServer ¶
func NewServer(ca *CA, st EnrollStore, registry *Registry, logger *slog.Logger, opts ...Option) *Server
NewServer builds a Server. logger defaults to slog.Default() if nil.
func (*Server) Enroll ¶
func (s *Server) Enroll(ctx context.Context, req *agentpb.EnrollRequest) (*agentpb.EnrollResponse, error)
Enroll implements agentpb.AgentServiceServer.
func (*Server) Session ¶
func (s *Server) Session(stream agentpb.AgentService_SessionServer) error
Session implements agentpb.AgentServiceServer: accepts an mTLS-authenticated agent's persistent stream, confirms its certificate actually matches the node it claims to be (not just that *some* certificate this CA issued was presented: a compromised or misconfigured node presenting a different, still-CA-issued certificate for another node's ID must not be trusted as that other node), and wires up a GRPCTransport into Registry for the rest of the control plane to use until the stream ends.
type Transport ¶
Transport is what a reconcile controller needs from whichever node a resource is placed on. Deliberately identical in shape to docker.Runtime (embedded, not duplicated field by field, so the two can never silently drift apart) rather than a new, narrower interface: docker.Runtime is already the exact narrow surface TASKS.md 1.2/1.3 scoped down to "what a reconcile controller needs from Docker," and Transport's whole point is "the same thing, now reachable on a specific node," not a different capability set. A future RPC surface an agent exposes beyond container operations (build dispatch for 3.5, telemetry collection already covered by ADR 008's separate federated design) extends Transport then, not now: this pass only closes the gap ADR 003 already described as existing.
Source Files
Directories