agent

package
v0.2.0-beta.10 Latest Latest
Warning

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

 Go to latest Published: Sep 23, 2026 License: Apache-2.0 Imports: 35 Imported by: 0

Documentation

Overview

Package agent is the 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 separate work, 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 GRPCTransport (grpc_transport.go).

Index

Constants

This section is empty.

Variables

View Source 
var ErrBuildWindowViolated = errors.New("agent: build peer exceeded its flow-control window")

ErrBuildWindowViolated ends a dispatched build whose peer sent more unacknowledged data frames than buildWindowFrames allows, for the same reason ErrExecWindowViolated ends an exec.

View Source 
var ErrCAFingerprintMismatch = errors.New("agent: control plane CA does not match the pinned fingerprint")

ErrCAFingerprintMismatch is returned by DialEnroll when the control plane's CA does not match the pinned fingerprint.

View Source 
var ErrExecWindowViolated = errors.New("agent: exec peer exceeded its flow-control window")

ErrExecWindowViolated ends an exec stream whose peer sent more unacknowledged data frames than execWindowFrames allows. Failing the one exec is the only safe response: blocking recvLoop would stall every other call sharing this session, and dropping the frame would corrupt the stream silently.

View Source 
var ErrExitStateUnsupported = errors.New("agent: this node's container runtime cannot inspect container exit state")

ErrExitStateUnsupported is what an InspectExitState request gets when this node's runtime cannot inspect a container's exit state. Callers treat any error here as "can't tell" and fall back to waiting out the full readiness budget.

View Source 
var ErrMeshUnavailable = errors.New("agent: this node has no mesh networking configured")

ErrMeshUnavailable is what an ApplyMesh/RotateMeshKey request gets when this node's Session was started with no MeshApplier wired up (RunSession's own WithMesh option never called): mesh networking not enabled for this node, or this agent process predates mesh support. The control plane's Coordinator treats this the same as any other per-node ApplyMesh failure (NodeResult.Err), never a fleet-wide error.

View Source 
var ErrNodeNotMeshCapable = fmt.Errorf("network: node's transport does not support mesh operations")

ErrNodeNotMeshCapable is returned when nodeID's registered transport exists (the node is connected) but does not support mesh operations, which in practice means it isn't really a *GRPCTransport at all (a test double registered directly, or a future transport kind that never implements mesh). Never expected against a real agent connection, since every real one is built by newGRPCTransport.

View Source 
var ErrNodeNotRegistered = errors.New("agent: node not registered in this transport registry")

ErrNodeNotRegistered is Registry.Get's failure mode for an unknown node ID.

View Source 
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.

View Source 
var ErrTTYUnsupported = errors.New("agent: this node's container runtime does not support interactive exec")

ErrTTYUnsupported is what an ExecRequest asking for a PTY gets when this agent's runtime has no interactive exec to offer.

Functions

func CertFingerprint 

func CertFingerprint(der []byte) string

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 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 this package'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, opts ...SessionOption) 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 (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 BuildDispatcher added in v0.2.0 

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

BuildDispatcher resolves a node ID to that node's session and runs a build there. Satisfies internal/build.NodeBuilder, which is how build.Router reaches a node without knowing anything about transports.

func NewBuildDispatcher added in v0.2.0 

func NewBuildDispatcher(registry *Registry) *BuildDispatcher

NewBuildDispatcher builds a dispatcher over registry.

func (*BuildDispatcher) BuildOnNode added in v0.2.0 

func (d *BuildDispatcher) BuildOnNode(ctx context.Context, nodeID string, req build.RemoteRequest, image io.Writer, progress func(build.ProgressEvent)) (*build.Result, error)

BuildOnNode implements internal/build.NodeBuilder.

type BuildRelay added in v0.2.0 

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

BuildRelay owns the agent side of every in-flight dispatched build on one Session, the same per-session state ExecRelay holds for exec.

func NewBuildRelay added in v0.2.0 

func NewBuildRelay(runner BuildRunner, send func(*agentpb.AgentMessage)) *BuildRelay

NewBuildRelay builds a relay running builds through runner and writing its frames through send, which must serialize writes to the Session stream. A nil runner is valid: every dispatched build is then rejected with a clear error.

func (*BuildRelay) Cancel added in v0.2.0 

func (r *BuildRelay) Cancel(buildID string)

Cancel stops a build the control plane no longer wants. No terminal frame follows: the caller that cancelled is not reading anymore.

func (*BuildRelay) CloseAll added in v0.2.0 

func (r *BuildRelay) CloseAll()

CloseAll stops every in-flight build, for a session that is ending.

func (*BuildRelay) Credit added in v0.2.0 

func (r *BuildRelay) Credit(c *agentpb.BuildCredit)

Credit records the control plane's refund of this build's output window.

func (*BuildRelay) Input added in v0.2.0 

func (r *BuildRelay) Input(in *agentpb.BuildInput)

Input routes one build-context frame to its build.

func (*BuildRelay) Start added in v0.2.0 

func (r *BuildRelay) Start(ctx context.Context, buildID string, req *agentpb.BuildRequest)

Start begins the build identified by buildID (the request's own RequestId, which also tags every frame of this build). It returns immediately: creating the context directory and running the build both happen off the session's receive loop.

type BuildRunner added in v0.2.0 

type BuildRunner interface {
	SolveRemote(ctx context.Context, req build.RemoteRequest, out io.Writer, progress func(build.ProgressEvent)) (*build.Result, error)
}

BuildRunner is what BuildRelay needs from internal/build to actually run a dispatched build. *build.Client satisfies it; an agent with no reachable BuildKit is constructed without one and rejects dispatched builds explicitly rather than failing halfway through one.

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 

func GenerateCA() (*CA, error)

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 

func LoadCA(certPEM, keyPEM []byte) (*CA, error)

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 

func (ca *CA) CertPEM() []byte

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) Fingerprint added in v0.2.0 

func (ca *CA) Fingerprint() string

Fingerprint returns the hex SHA-256 of the CA certificate's DER, the value an agent pins at enrollment (APP_CA_FINGERPRINT).

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 the 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.

func (*CA) KeyPEM 

func (ca *CA) KeyPEM() []byte

KeyPEM returns the CA's own private key, PEM-encoded, PKCS#8: the other half LoadCA needs. Never sent over the wire to an agent, only ever persisted locally by the control plane's own startup code.

type EnrollOption added in v0.2.0 

type EnrollOption func(*enrollConfig)

EnrollOption configures DialEnroll.

func WithPinnedCAFingerprint added in v0.2.0 

func WithPinnedCAFingerprint(fingerprint string) EnrollOption

WithPinnedCAFingerprint makes DialEnroll refuse any control plane whose CA certificate's SHA-256 (hex, colons optional) is not fingerprint.

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, persist the newly enrolled node, and track its connection state. *store.DB satisfies this structurally.

type ExecRelay added in v0.2.0 

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

ExecRelay owns the agent side of every in-flight remote exec on one Session: it runs the command against the local docker.Runtime, streams its output back frame by frame under the control plane's flow-control window, feeds arriving stdin frames into the command as they land, and stops the local exec when the control plane cancels or the session ends. Unlike every other operation, an exec spans many frames in both directions, so it needs per-session state Execute's stateless dispatch cannot hold.

func NewExecRelay added in v0.2.0 

func NewExecRelay(rt docker.Runtime, send func(*agentpb.AgentMessage)) *ExecRelay

NewExecRelay builds a relay running commands against rt and writing its frames through send, which must serialize writes to the Session stream.

func (*ExecRelay) Cancel added in v0.2.0 

func (r *ExecRelay) Cancel(execID string)

Cancel stops an exec the control plane no longer wants. No terminal frame follows: the caller that cancelled is not reading anymore.

func (*ExecRelay) CloseAll added in v0.2.0 

func (r *ExecRelay) CloseAll()

CloseAll stops every in-flight exec, for a session that is ending.

func (*ExecRelay) Credit added in v0.2.0 

func (r *ExecRelay) Credit(c *agentpb.ExecCredit)

Credit records the control plane's refund of this exec's output window.

func (*ExecRelay) Input added in v0.2.0 

func (r *ExecRelay) Input(in *agentpb.ExecInput)

Input routes one stdin frame to its exec.

func (*ExecRelay) Resize added in v0.2.0 

func (r *ExecRelay) Resize(rz *agentpb.ExecResize)

Resize records the newest terminal size for a PTY exec. A no-op for an exec that never asked for one, or that has already ended.

func (*ExecRelay) Start added in v0.2.0 

func (r *ExecRelay) Start(ctx context.Context, execID string, req *agentpb.ExecRequest)

Start begins the exec identified by execID (the request's own RequestId, which also tags every frame of this exec). It returns immediately: attaching to the container is a Docker round trip, and doing it inline would stall the session's whole receive loop. The acknowledgment, or the error that stopped the exec from attaching, is sent once that round trip finishes.

type GRPCSink added in v0.2.0 

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

GRPCSink dispatches mesh config to remote nodes over their own agent Session stream, resolved per call through registry.

func NewGRPCSink added in v0.2.0 

func NewGRPCSink(registry *Registry) *GRPCSink

NewGRPCSink builds a GRPCSink that resolves each node's transport through registry, the same *Registry Server.Session already populates (server.go) and every other remote-dispatch path in this package (GRPCTransport itself, build dispatch) already shares.

func (*GRPCSink) ApplyMesh added in v0.2.0 

func (s *GRPCSink) ApplyMesh(ctx context.Context, nodeID string, cfg network.DeviceConfig) (network.NodeIdentity, error)

ApplyMesh implements network.ConfigSink by forwarding cfg to nodeID's live agent Session, if it has one. A node with no registered transport (never enrolled, or disconnected) fails this one node's pass with agent.ErrNodeNotRegistered wrapped, exactly the "one node's problem, never the fleet's" contract ConfigSink.ApplyMesh's own doc comment requires: Distribute records it in that node's NodeResult.Err and keeps distributing to every other node.

func (*GRPCSink) RotateKey added in v0.2.0 

func (s *GRPCSink) RotateKey(ctx context.Context, nodeID string) (network.RotationResult, error)

RotateKey implements network.KeyRotator by asking nodeID's live agent Session to rotate its own key. Same not-registered/not-mesh-capable failure modes as ApplyMesh.

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's remaining wiring), never constructed directly.

func (*GRPCTransport) ApplyMesh added in v0.2.0 

func (t *GRPCTransport) ApplyMesh(ctx context.Context, cfg network.DeviceConfig) (network.NodeIdentity, error)

ApplyMesh sends cfg to this transport's own node over its Session stream and returns what the node reported back about itself. Part of meshTransport (mesh_sink.go), not Transport itself: mesh operations are not a docker.Runtime concern, so keeping them off Transport's own interface means a caller that only wants container operations never has to know mesh dispatch exists.

func (*GRPCTransport) BuildOnNode added in v0.2.0 

func (t *GRPCTransport) BuildOnNode(ctx context.Context, req build.RemoteRequest, image io.Writer, progress func(build.ProgressEvent)) (*build.Result, error)

BuildOnNode implements RemoteBuilder: it dispatches req to this transport's node, streams req.ContextDir up as a tar, relays progress through progress, and writes the docker-save image tar the build produced into image. The image is never loaded on the build node: the control plane that asked for the build is the one that wants it.

func (*GRPCTransport) Create 

func (t *GRPCTransport) Create(ctx context.Context, spec docker.ContainerSpec) (string, error)

Create implements Transport (docker.Runtime).

func (*GRPCTransport) EnsureNetwork 

func (t *GRPCTransport) EnsureNetwork(ctx context.Context, name string) (string, error)

EnsureNetwork implements Transport (docker.Runtime).

func (*GRPCTransport) EnsureVolume 

func (t *GRPCTransport) EnsureVolume(ctx context.Context, name string) error

EnsureVolume implements Transport (docker.Runtime).

func (*GRPCTransport) Events 

func (t *GRPCTransport) Events(ctx context.Context) (<-chan docker.Event, <-chan error)

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(ctx context.Context, containerID string, cmd []string) (io.ReadCloser, error)

Exec implements Transport (docker.Runtime). The initial round trip is synchronous, matching docker.Client.Exec's own behavior: an exec that cannot attach (no such container, daemon refused) fails here rather than on a later Read. Output arrives afterwards as ExecOutput frames, reassembled by the returned stream.

func (*GRPCTransport) ExecTTY added in v0.2.0 

func (t *GRPCTransport) ExecTTY(ctx context.Context, containerID string, opts docker.ExecTTYOptions) (docker.ExecSession, error)

ExecTTY implements docker.TTYRuntime. The PTY lives on the agent's side; this end is the same windowed frame stream every other exec uses, plus a resize frame, so a remote terminal and a local one differ only in how far the bytes travel.

func (*GRPCTransport) ExecWithInput 

func (t *GRPCTransport) ExecWithInput(ctx context.Context, containerID string, cmd []string, stdin io.Reader) (io.ReadCloser, error)

ExecWithInput implements Transport (docker.Runtime). Exec plus a stdin direction: a goroutine drains stdin into ExecInput frames while output frames stream back, the same concurrency docker.Client.ExecWithInput needs locally to keep a command that writes while it reads from deadlocking.

func (*GRPCTransport) InspectByName 

func (t *GRPCTransport) InspectByName(ctx context.Context, name string) (*docker.ContainerState, error)

InspectByName implements Transport (docker.Runtime).

func (*GRPCTransport) InspectExitState added in v0.2.0 

func (t *GRPCTransport) InspectExitState(ctx context.Context, name string) (*docker.ExitState, error)

InspectExitState implements docker.ExitStateInspector over the wire, so a service placed on a remote node gets the same fail-fast-on-crash readiness behavior a locally placed one already had: without it, a container that OOM-killed itself mid-readiness looked merely "not ready yet" until the whole readiness budget expired.

An agent too old to know this op answers with an error, which every caller of docker.ExitStateInspector already treats as "can't tell" and degrades to waiting out the full budget.

func (*GRPCTransport) ListByPrefix 

func (t *GRPCTransport) ListByPrefix(ctx context.Context, prefix string) ([]docker.ContainerState, error)

ListByPrefix implements Transport (docker.Runtime).

func (*GRPCTransport) ListImages 

func (t *GRPCTransport) ListImages(ctx context.Context, repo string) ([]docker.ImageInfo, error)

ListImages implements Transport (docker.Runtime).

func (*GRPCTransport) ListNetworksByPrefix 

func (t *GRPCTransport) ListNetworksByPrefix(ctx context.Context, prefix string) ([]docker.NetworkInfo, error)

ListNetworksByPrefix implements Transport (docker.Runtime).

func (*GRPCTransport) Remove 

func (t *GRPCTransport) Remove(ctx context.Context, id string, force bool) error

Remove implements Transport (docker.Runtime).

func (*GRPCTransport) RemoveNetwork 

func (t *GRPCTransport) RemoveNetwork(ctx context.Context, name string) error

RemoveNetwork implements Transport (docker.Runtime).

func (*GRPCTransport) RotateMeshKey added in v0.2.0 

func (t *GRPCTransport) RotateMeshKey(ctx context.Context) (network.RotationResult, error)

RotateMeshKey asks this transport's own node to rotate its mesh key and returns the change. Part of meshTransport, same reasoning as ApplyMesh above.

func (*GRPCTransport) Start 

func (t *GRPCTransport) Start(ctx context.Context, id string) error

Start implements Transport (docker.Runtime).

func (*GRPCTransport) Stop 

func (t *GRPCTransport) Stop(ctx context.Context, id string, timeout time.Duration) error

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 

type Identity struct {
	NodeID        string
	ClientCertPEM []byte
	ClientKeyPEM  []byte
	CACertPEM     []byte
}

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 

func DialEnroll(ctx context.Context, addr, joinToken, nodeName string, opts ...EnrollOption) (*Identity, error)

DialEnroll connects to addr and exchanges joinToken for an Identity. There is no CA on disk yet, so the server is verified against the pinned CA fingerprint when one is given (see ADR 003), and trusted on first use otherwise. Every later connection verifies against the returned CA.

type Local 

type Local struct {
	docker.Runtime
}

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.

func NewLocal 

func NewLocal(rt docker.Runtime) Local

NewLocal wraps rt as this process's own node Transport.

func (Local) ExecTTY added in v0.2.0 

func (l Local) ExecTTY(ctx context.Context, containerID string, opts docker.ExecTTYOptions) (docker.ExecSession, error)

ExecTTY implements docker.TTYRuntime by forwarding to the wrapped runtime, which the embedded docker.Runtime interface would otherwise hide even when the concrete value behind it supports interactive exec.

func (Local) InspectExitState added in v0.2.0 

func (l Local) InspectExitState(ctx context.Context, name string) (*docker.ExitState, error)

InspectExitState implements docker.ExitStateInspector by forwarding to the wrapped runtime, which the embedded docker.Runtime interface would otherwise hide even when the concrete value behind it can inspect exit state.

type MeshApplier added in v0.2.0 

type MeshApplier interface {
	ApplyMesh(ctx context.Context, nodeID string, cfg network.DeviceConfig) (network.NodeIdentity, error)
	RotateKey(ctx context.Context, nodeID string) (network.RotationResult, error)
}

MeshApplier is what serveSession needs to apply this node's own mesh config and rotate its own mesh key. Exactly *network.LocalSink's own two methods (ApplyMesh, RotateKey): a real WireGuard device wired up by cmd/levelrail-agent satisfies this by constructing a *network.LocalSink around its own network.Mesh and private key, with zero adapter code needed, the identical reuse the control plane's own setupMesh (cmd/levelrail/mesh.go) already gets from LocalSink for its own local node. See RunSession's WithMesh option for how this and this node's own ID reach serveSession.

type Option 

type Option func(*Server)

Option configures optional Server behavior.

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 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 NewRegistry 

func NewRegistry() *Registry

NewRegistry builds an empty Registry.

func (*Registry) Get 

func (r *Registry) Get(nodeID string) (Transport, error)

Get returns the Transport registered for nodeID, or ErrNodeNotRegistered.

func (*Registry) Register 

func (r *Registry) Register(nodeID string, t Transport)

Register associates nodeID with t, replacing any existing entry for that ID. Safe for concurrent use.

func (*Registry) Unregister 

func (r *Registry) Unregister(nodeID string)

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 RemoteBuilder added in v0.2.0 

type RemoteBuilder interface {
	BuildOnNode(ctx context.Context, req build.RemoteRequest, image io.Writer, progress func(build.ProgressEvent)) (*build.Result, error)
}

RemoteBuilder is the optional capability a Transport advertises when it can run a build dispatched to its node. Only *GRPCTransport does: a Local transport is this process's own Docker socket, which is what "build locally" already means, not somewhere to dispatch to.

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 SessionOption added in v0.2.0 

type SessionOption func(*sessionConfig)

SessionOption configures optional RunSession behavior.

func WithBuildRunner added in v0.2.0 

func WithBuildRunner(runner BuildRunner) SessionOption

WithBuildRunner lets this node accept builds dispatched to it by the control plane. Without one, a dispatched build is rejected with a clear error instead of failing partway through: an agent whose local BuildKit is unreachable still serves every container operation normally.

func WithHeartbeatInterval 

func WithHeartbeatInterval(d time.Duration) SessionOption

WithHeartbeatInterval overrides how often RunSession sends an unprompted Heartbeat frame up the Session stream. Without one configured, defaultHeartbeatInterval applies. cmd/levelrail-agent's own main.go reads APP_NODE_HEARTBEAT_INTERVAL and passes the parsed duration here, the project's "no hardcoded thresholds, use env vars" rule; this package itself never reads the environment directly.

func WithKeepalive added in v0.2.0 

func WithKeepalive(pingTime, timeout time.Duration) SessionOption

WithKeepalive overrides this connection's HTTP/2 PING keepalive timing. Without one configured, defaultKeepaliveTime/ defaultKeepaliveTimeout apply. cmd/levelrail-agent's own main.go reads APP_NODE_KEEPALIVE_TIME/APP_NODE_KEEPALIVE_TIMEOUT and passes the parsed durations here, the same env-var-with-default convention WithHeartbeatInterval above follows.

func WithMesh added in v0.2.0 

func WithMesh(nodeID string, mesh MeshApplier) SessionOption

WithMesh lets this node accept and apply mesh config (and rotate its own mesh key) dispatched by the control plane, over ApplyMesh/ RotateMeshKey requests on this same Session stream. nodeID is this node's own ID, since RotateMeshKeyRequest carries no fields of its own (this file's own header on why: it always means "rotate the node holding this stream," never another). Without this option, both request kinds answer with ErrMeshUnavailable: a node whose Docker operations work fine but has no mesh device configured, not a broken connection.

type Transport 

type Transport interface {
	docker.Runtime
}

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 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. An RPC surface an agent exposes beyond container operations stays off this interface and is advertised as an optional capability instead (see RemoteBuilder, build_dispatch.go), so a Local transport does not have to pretend to implement something only a remote agent can do.

Source Files

View all Source files

Directories

Path Synopsis

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.