agent

package
v0.2.0-beta.17 Latest Latest
Warning

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

 Go to latest Published: Oct 5, 2026 License: Apache-2.0 Imports: 46 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

View Source 
const (
	DefaultClientCertValidity = 90 * 24 * time.Hour
	DefaultCertRenewGrace     = 24 * time.Hour
)

Defaults for the env-tunable certificate settings (ADR 021).

View Source 
const (
	DefaultRenewFraction = 2.0 / 3.0
)

Defaults for RenewConfig. Renewing at two thirds of the lifetime leaves the last third to retry through outages (the step-ca and cert-manager default).

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 ErrInvalidCSR = errors.New("agent: invalid certificate signing request")

ErrInvalidCSR is returned by SignClientCSR for a malformed request, a bad signature, or a key type this CA does not issue for.

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 ErrStatsUnsupported = errors.New("agent: this node's container runtime cannot report container stats")

ErrStatsUnsupported is what a Stats request gets when this node's runtime cannot report container resource usage.

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

func IsAuthRejection(err error) bool

IsAuthRejection is isAuthRejection for callers outside this package.

func LocalAgentInfo added in v0.2.0 

func LocalAgentInfo() *agentpb.AgentInfo

LocalAgentInfo describes this agent binary and the machine it runs on.

func NewKeyAndCSR added in v0.2.0 

func NewKeyAndCSR(commonName string) (keyPEM, csrDER []byte, err error)

NewKeyAndCSR generates a fresh ed25519 private key on this machine and a CSR for it. The key never leaves the caller; only the CSR is sent.

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.

func (*CA) SignClientCSR added in v0.2.0 

func (ca *CA) SignClientCSR(commonName string, csrDER []byte, validFor time.Duration, now time.Time) (IssuedCert, error)

SignClientCSR issues a client certificate for commonName over the CSR's public key. The CSR's own subject, SANs and extensions are ignored: the control plane alone decides what the certificate asserts.

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.

func WithTrustedCA added in v0.2.0 

func WithTrustedCA(caPEM []byte) EnrollOption

WithTrustedCA verifies the control plane against caPEM, the CA an already-enrolled node has on disk, instead of a fingerprint pin.

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
	RotateNodeCert(ctx context.Context, id, presentedFingerprint string, c store.NodeCert, grace time.Duration, now time.Time) error
	ReenrollNodeCert(ctx context.Context, id string, c store.NodeCert, now time.Time) error
	SyncNodeCertDetails(ctx context.Context, id, fingerprint, serial string, notAfter time.Time) error
	UpdateNodeAgentInfo(ctx context.Context, id string, info store.NodeAgentInfo, now time.Time) error
}

EnrollStore is the store surface Server needs. *store.DB satisfies it.

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

type GPUProbe func(ctx context.Context) gpu.Info

GPUProbe returns the node's current GPU snapshot.

type GPUSink added in v0.2.0 

type GPUSink interface {
	SetNodeGPU(ctx context.Context, nodeID string, info gpu.Info) error
}

GPUSink receives a node's GPU report on the control plane.

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

func (t *GRPCTransport) NetworkConnect(ctx context.Context, network, containerID string) error

NetworkConnect implements Transport (docker.Runtime).

func (*GRPCTransport) NetworkDisconnect added in v0.2.0 

func (t *GRPCTransport) NetworkDisconnect(ctx context.Context, network, containerID string, force bool) error

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

func (t *GRPCTransport) Stats(ctx context.Context, containerID string) (docker.ContainerStats, error)

Stats implements docker.StatsInspector over the wire: the remote-node half of the control plane's own telemetry.Collector, which otherwise only ever sees this process's own local Docker socket. An agent too old to know this op answers with ErrStatsUnsupported's own string, which the collector's caller treats as "skip this target, try again next tick," the same degrade InspectExitState's doc comment describes for an old agent.

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. The private key is generated here and only a CSR is sent (ADR 021); a control plane too old to sign CSRs returns a key of its own instead, which is used as is. 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.

func DialReenroll added in v0.2.0 

func DialReenroll(ctx context.Context, addr, token, nodeID string, opts ...EnrollOption) (*Identity, error)

DialReenroll exchanges a re-enrollment token for a new certificate for an existing node, keeping its identity. nodeID may be empty when the old identity file is gone; the token alone names the node.

func (*Identity) Expired added in v0.2.0 

func (id *Identity) Expired(now time.Time) bool

Expired reports whether the identity's certificate is past its NotAfter, treating an unparseable certificate as expired.

func (*Identity) Leaf added in v0.2.0 

func (id *Identity) Leaf() (*x509.Certificate, error)

Leaf parses the identity's client certificate.

type IdentityChecker added in v0.2.0 

type IdentityChecker func(ctx context.Context, id *Identity) error

IdentityChecker proves an identity is accepted by the control plane without opening a Session.

func CheckIdentityAt added in v0.2.0 

func CheckIdentityAt(addr string) IdentityChecker

CheckIdentityAt returns an IdentityChecker that dials addr presenting the identity and calls CheckIdentity, requiring it to be current.

type IdentityFile added in v0.2.0 

type IdentityFile struct {
	Path string
	// contains filtered or unexported fields
}

IdentityFile persists an agent Identity as JSON (0600). Writes go to a temp file that is fsynced and renamed over the target, so a crash leaves either the old or the new identity on disk, never a torn one. During a renewal the previous identity is kept beside it as <path>.prev until the new certificate has been proven to work.

func NewIdentityFile added in v0.2.0 

func NewIdentityFile(path string) *IdentityFile

NewIdentityFile returns an IdentityFile for path.

func (*IdentityFile) CheckWritable added in v0.2.0 

func (f *IdentityFile) CheckWritable() error

CheckWritable proves the identity file's directory accepts new files by creating and removing a probe, so enrollment can fail before it spends the one-time join token on an identity that could never be saved.

func (*IdentityFile) Confirm added in v0.2.0 

func (f *IdentityFile) Confirm() error

Confirm drops the pre-renewal backup once the new identity works.

func (*IdentityFile) HasStaged added in v0.2.0 

func (f *IdentityFile) HasStaged() bool

HasStaged reports whether a renewal was staged but never confirmed or rolled back, meaning the agent stopped in between.

func (*IdentityFile) Load added in v0.2.0 

func (f *IdentityFile) Load() (*Identity, error)

Load reads the identity and removes temp files a crashed write left behind. It returns an error wrapping os.ErrNotExist when none exists.

func (*IdentityFile) LoadPrev added in v0.2.0 

func (f *IdentityFile) LoadPrev() (*Identity, error)

LoadPrev reads the pre-renewal identity kept at PrevPath.

func (*IdentityFile) PrevPath added in v0.2.0 

func (f *IdentityFile) PrevPath() string

PrevPath is where the pre-renewal identity is kept.

func (*IdentityFile) ResolveStaged added in v0.2.0 

func (f *IdentityFile) ResolveStaged(ctx context.Context, check func(context.Context, *Identity) error) (*Identity, error)

ResolveStaged settles a renewal the agent stopped in the middle of: the staged identity is kept if check accepts it, otherwise the backup is restored. It returns the identity to use.

func (*IdentityFile) Rollback added in v0.2.0 

func (f *IdentityFile) Rollback() error

Rollback restores the pre-renewal identity over the current one.

func (*IdentityFile) Save added in v0.2.0 

func (f *IdentityFile) Save(id *Identity) error

Save atomically replaces the identity on disk with id.

func (*IdentityFile) Stage added in v0.2.0 

func (f *IdentityFile) Stage(next *Identity) error

Stage keeps the current identity file as PrevPath, then saves next. Confirm or Rollback completes it.

type IdentityHolder added in v0.2.0 

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

IdentityHolder is the agent's in-memory current identity, swapped after a renewal so the next connection presents the new certificate without a restart.

func NewIdentityHolder added in v0.2.0 

func NewIdentityHolder(id *Identity) *IdentityHolder

NewIdentityHolder wraps id.

func (*IdentityHolder) Current added in v0.2.0 

func (h *IdentityHolder) Current() *Identity

Current returns the identity in use.

func (*IdentityHolder) Set added in v0.2.0 

func (h *IdentityHolder) Set(id *Identity)

Set replaces the identity in use.

type IdentityPersister added in v0.2.0 

type IdentityPersister interface {
	Stage(next *Identity) error
	Confirm() error
	Rollback() error
}

IdentityPersister stages a renewed identity durably and then confirms or rolls it back. *IdentityFile implements it.

type IssuedCert added in v0.2.0 

type IssuedCert struct {
	PEM         []byte
	Fingerprint string
	Serial      string
	NotAfter    time.Time
}

IssuedCert is a signed client certificate plus the fields the control plane records about it.

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.

func (Local) InspectImageID added in v0.2.0 

func (l Local) InspectImageID(ctx context.Context, ref string) (string, error)

InspectImageID implements docker.ImageInspector by forwarding; an unsupporting runtime reports "" so digest verification is skipped.

func (Local) Stats added in v0.2.0 

func (l Local) Stats(ctx context.Context, containerID string) (docker.ContainerStats, error)

Stats implements docker.StatsInspector by forwarding to the wrapped runtime, the same forwarding InspectExitState above already does for a different optional capability.

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.

func WithCertValidity added in v0.2.0 

func WithCertValidity(d time.Duration) Option

WithCertValidity sets how long issued agent certificates stay valid.

func WithClock added in v0.2.0 

func WithClock(now func() time.Time) Option

WithClock overrides the server's time source, for tests.

func WithGPUSink added in v0.2.0 

func WithGPUSink(sink GPUSink) Option

WithGPUSink persists GPU reports agents send up their Session stream.

func WithRenewGrace added in v0.2.0 

func WithRenewGrace(d time.Duration) Option

WithRenewGrace sets how long a renewed node's previous certificate stays accepted, so a renewal whose response was lost cannot lock the node out.

func WithRequireCSR added in v0.2.0 

func WithRequireCSR(require bool) Option

WithRequireCSR refuses enrollments from agents that do not send a CSR, closing the legacy path where the control plane generates the key.

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

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

Resolve is Get plus the local node's own ID, for callers that just need a runtime for whichever node a resource is placed on.

func (*Registry) SetLocal added in v0.2.0 

func (r *Registry) SetLocal(localID string, t Transport)

SetLocal makes Resolve answer localID with t, this process's own runtime. It is deliberately not in byID: Get callers (build dispatch, mesh sink) must keep treating the local node as "not a remote agent".

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

type RenewConfig struct {
	// Fraction of the certificate's lifetime after which renewal starts.
	Fraction float64
	// Jitter is the fraction of lifetime added at random so nodes enrolled
	// together do not all renew at once.
	Jitter             float64
	RetryMin, RetryMax time.Duration
}

RenewConfig tunes when and how often the agent renews its certificate.

type Renewer added in v0.2.0 

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

Renewer keeps an agent's certificate fresh over its authenticated connection (ADR 021).

func NewRenewer added in v0.2.0 

func NewRenewer(holder *IdentityHolder, persist IdentityPersister, check IdentityChecker, cfg RenewConfig, logger *slog.Logger) *Renewer

NewRenewer builds a Renewer. check defaults to CheckIdentityAt(addr) when nil.

func (*Renewer) Holder added in v0.2.0 

func (r *Renewer) Holder() *IdentityHolder

Holder returns the identity holder the renewer updates.

func (*Renewer) RenewAt added in v0.2.0 

func (r *Renewer) RenewAt(id *Identity) (time.Time, error)

RenewAt is when id should be renewed: Fraction of its lifetime plus up to Jitter more.

func (*Renewer) RenewOnce added in v0.2.0 

func (r *Renewer) RenewOnce(ctx context.Context, client agentpb.AgentServiceClient) error

RenewOnce generates a fresh key, has the control plane sign it, persists the result, and proves it works before switching to it. On any failure the agent keeps using its current certificate, which the control plane still accepts inside the renewal overlap window.

func (*Renewer) Run added in v0.2.0 

func (r *Renewer) Run(ctx context.Context, client agentpb.AgentServiceClient)

Run renews whenever the current certificate reaches its renewal time, retrying with capped backoff, until ctx is done.

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

func (s *Server) CheckIdentity(ctx context.Context, _ *agentpb.CheckIdentityRequest) (*agentpb.CheckIdentityResponse, error)

CheckIdentity implements agentpb.AgentServiceServer.

func (*Server) Disconnect added in v0.2.0 

func (s *Server) Disconnect(nodeID string)

Disconnect ends nodeID's live Session, if any, so a node whose certificate was just revoked stops acting on an already-open stream.

func (*Server) Enroll 

func (s *Server) Enroll(ctx context.Context, req *agentpb.EnrollRequest) (*agentpb.EnrollResponse, error)

Enroll implements agentpb.AgentServiceServer.

func (*Server) NotifyShutdown added in v0.2.0 

func (s *Server) NotifyShutdown()

NotifyShutdown sends a GoAway frame to every connected agent, telling each to close its own end of the Session stream now rather than wait to be cut off by the control plane's own bounded GracefulStop timeout (cmd/levelrail's stopAgentGRPCServer). Best-effort: a send failure here just means that agent falls back to the timeout, same as before this existed.

func (*Server) Reenroll added in v0.2.0 

func (s *Server) Reenroll(ctx context.Context, req *agentpb.ReenrollRequest) (*agentpb.ReenrollResponse, error)

Reenroll implements agentpb.AgentServiceServer: a token bound to an existing node buys that node a new certificate. Minting the token was the operator's authorization, so a revocation is cleared.

func (*Server) Renew added in v0.2.0 

func (s *Server) Renew(ctx context.Context, req *agentpb.RenewRequest) (*agentpb.RenewResponse, error)

Renew implements agentpb.AgentServiceServer. The CSR is signed before the store is updated: if the update fails, the new certificate was never recorded and so is never trusted.

func (*Server) Session 

func (s *Server) Session(stream agentpb.AgentService_SessionServer) error

Session implements agentpb.AgentServiceServer: accepts an mTLS-authenticated agent's persistent stream once its certificate is confirmed to be one this node may use, and wires a GRPCTransport into Registry 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 WithGPUProbe added in v0.2.0 

func WithGPUProbe(p GPUProbe) SessionOption

WithGPUProbe makes RunSession report this node's GPUs to the control plane on connect and periodically after.

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.

func WithRenewer added in v0.2.0 

func WithRenewer(r *Renewer) SessionOption

WithRenewer keeps the session's certificate fresh with r while the session is open. The identity r holds replaces the one RunSession was given, so each reconnect presents the newest certificate.

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.