agent

package
v0.8.0 Latest Latest
Warning

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

 Go to latest Published: Oct 21, 2026 License: Apache-2.0 Imports: 53 Imported by: 0

Documentation ¶

Overview ¶

Package agent is the transport boundary between reconcile controllers and the node they run on. Local is the in-process form; GRPCTransport (grpc_transport.go) is the remote form.

Index ¶

Constants ¶

View Source 
const (
	DefaultSecurityCoalesceWindow = 10 * time.Minute
	DefaultSecurityPerNodeCap     = 500
	DefaultSecurityRetention      = 30 * 24 * time.Hour
)

Defaults for SecurityConfig.

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 fails one exec whose peer exceeded its window; blocking recvLoop would stall the session.

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

ErrExitStateUnsupported means the runtime cannot inspect exit state; callers wait out the full readiness budget.

View Source 
var ErrFilePathEscapes = errors.New("agent: file path escapes the storage directory")

ErrFilePathEscapes is returned for any path that would resolve outside the storage root.

View Source 
var ErrFileStoreDisabled = errors.New("agent: file storage is not enabled on this node (no storage directory configured)")

ErrFileStoreDisabled means the agent has no storage directory configured, so it refuses every file operation.

View Source 
var ErrImageLoadUnsupported = errors.New("agent: this node's container runtime cannot load an image archive")

ErrImageLoadUnsupported is returned when the node runtime cannot load an image archive.

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 returned when no MeshApplier was wired in. The control plane treats it as a per-node failure, never a fleet-wide one.

View Source 
var ErrNodeNotConnected = errors.New("agent: node is not connected")

ErrNodeNotConnected is returned by RequestRenew when the node has no live session to carry the request.

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

ErrNodeNotMeshCapable is returned when a node's registered transport does not support mesh operations. Real agent connections never hit it.

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 the hex SHA-256 of a DER-encoded certificate. Sessions compare against this rather than trusting CommonName, which another certificate could also claim.

func Execute ¶ 

func Execute(ctx context.Context, rt docker.Runtime, req *agentpb.AgentRequest, emitEvent func(*agentpb.ProxiedEvent)) *agentpb.AgentResponse

Execute returns exactly one AgentResponse per request, carrying req.RequestId, even on failure. Errors travel as plain strings because the gRPC boundary cannot carry %w wrapping.

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, opts ...CredentialOption) (credentials.TransportCredentials, error)

NewServerCredentials returns mTLS credentials that verify a client certificate when presented but do not require one. Must go to grpc.Creds, not a raw tls.Listener: grpc only sees the peer TLS state when it performs the handshake.

func RunSession ¶ 

func RunSession(ctx context.Context, addr string, id *Identity, rt docker.Runtime, logger *slog.Logger, opts ...SessionOption) error

RunSession serves one persistent Session stream until ctx is cancelled or the connection fails. It never retries; the caller owns reconnection.

func WithFileRoot ¶ 

func WithFileRoot(ctx context.Context, dir string) context.Context

WithFileRoot enables file operations confined to dir for sessions served under ctx.

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 returns a relay that runs builds through runner and writes frames through send; a nil runner rejects every build.

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 without blocking 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 runs dispatched builds; *build.Client satisfies it.

type CA ¶ 

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

CA is the control plane's self-signed root, signing its server cert and every agent's client cert.

func GenerateCA ¶ 

func GenerateCA() (*CA, error)

GenerateCA creates a new self-signed CA using ed25519 keys.

func LoadCA ¶ 

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

LoadCA restores a CA persisted via CertPEM/KeyPEM; regenerating it would invalidate every enrolled agent.

func (*CA) CertPEM ¶ 

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

CertPEM returns the CA certificate, PEM-encoded, for agents to verify the control plane and for persistence.

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) IssueServerCert ¶ 

func (ca *CA) IssueServerCert(hosts []string, validFor time.Duration) (certPEM, keyPEM []byte, err error)

IssueServerCert issues the control plane's gRPC listener certificate, signed by ca.

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 CredentialOption ¶ 

type CredentialOption func(*credentialConfig)

CredentialOption configures NewServerCredentials.

func WithCredentialClock ¶ 

func WithCredentialClock(now func() time.Time) CredentialOption

WithCredentialClock overrides the time source for certificate validity.

func WithRejectObserver ¶ 

func WithRejectObserver(f func(RejectedHandshake)) CredentialOption

WithRejectObserver calls f for every client certificate the handshake refuses. f runs on the handshake path and must not block.

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.

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 and returns immediately. Attaching is a Docker round trip that must not stall the session receive loop.

type FileEntry ¶ 

type FileEntry struct {
	Path string
	Size int64
}

FileEntry is one object found by ListFiles.

type FileStore ¶ 

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

FileStore performs object operations strictly inside root.

func NewFileStore ¶ 

func NewFileStore(dir string) (*FileStore, error)

NewFileStore returns a store rooted at dir, creating it when missing.

func (*FileStore) Delete ¶ 

func (s *FileStore) Delete(rel string) error

Delete removes rel; a missing object is success.

func (*FileStore) List ¶ 

func (s *FileStore) List(prefix string) (io.ReadCloser, error)

List returns "<size>\t<path>" lines for every file whose path starts with prefix.

func (*FileStore) Open ¶ 

func (s *FileStore) Open(rel string) (io.ReadCloser, error)

Open returns the object at rel for reading.

func (*FileStore) Put ¶ 

func (s *FileStore) Put(rel string, r io.Reader, maxBytes int64) error

Put writes r to rel through a temp file so a failed upload never leaves a partial object visible.

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 node transports through registry.

func (*GRPCSink) ApplyMesh ¶ added in v0.2.0 

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

ApplyMesh forwards cfg to nodeID's live agent session, failing only that node's pass if it has no registered transport.

func (*GRPCSink) MeshStatus ¶ added in v0.2.0 

func (s *GRPCSink) MeshStatus(ctx context.Context, nodeID string) (network.Status, error)

MeshStatus returns nodeID's live WireGuard view over its agent session.

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 is the control-plane side of a reverse-dialed agent connection, built on the agent's session mux.

func (*GRPCTransport) ApplyMesh ¶ added in v0.2.0 

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

ApplyMesh lives on meshTransport, not Transport, so container-only callers never see mesh dispatch.

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 dispatches a build to this transport's node and writes the image tar to image. The image is never loaded on the build node; the requesting control plane loads it.

func (*GRPCTransport) CopyFromContainer ¶ 

func (t *GRPCTransport) CopyFromContainer(ctx context.Context, containerID, path string, maxBytes int64) (io.ReadCloser, docker.PathStat, error)

CopyFromContainer implements docker.ArchiveRuntime over the exec frame stream: the same window, credit and cancel as any exec, no new RPC shape.

func (*GRPCTransport) CopyToContainer ¶ 

func (t *GRPCTransport) CopyToContainer(ctx context.Context, containerID, dir string, content io.Reader, maxBytes int64) error

CopyToContainer implements docker.ArchiveRuntime. It returns once the agent's archive API has consumed the whole stream, or with its error.

func (*GRPCTransport) Create ¶ 

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

Create implements Transport (docker.Runtime).

func (*GRPCTransport) DeleteFile ¶ 

func (t *GRPCTransport) DeleteFile(ctx context.Context, path string) error

DeleteFile removes path; a missing object is success.

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 relays WatchEvents over mux and drops events under backpressure rather than blocking the shared recv loop.

func (*GRPCTransport) Exec ¶ 

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

Exec runs its initial round trip synchronously, so a failed attach errors here rather than on a later Read.

func (*GRPCTransport) ExecTTY ¶ added in v0.2.0 

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

ExecTTY runs over the usual exec stream; the PTY lives on the agent and this end only adds resize frames.

func (*GRPCTransport) ExecWithInput ¶ 

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

ExecWithInput drains stdin concurrently with output, so a command that writes while reading does not deadlock.

func (*GRPCTransport) GetFile ¶ 

func (t *GRPCTransport) GetFile(ctx context.Context, path string) (io.ReadCloser, error)

GetFile streams the object at path; the caller closes it.

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; an agent too old to know it is treated as can't-tell.

func (*GRPCTransport) ListByPrefix ¶ 

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

ListByPrefix implements Transport (docker.Runtime).

func (*GRPCTransport) ListFiles ¶ 

func (t *GRPCTransport) ListFiles(ctx context.Context, prefix string) ([]FileEntry, error)

ListFiles returns every object whose path starts with prefix.

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

func (t *GRPCTransport) MeshStatus(ctx context.Context) (network.Status, error)

MeshStatus asks this transport's node for its live WireGuard view.

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) PutFile ¶ 

func (t *GRPCTransport) PutFile(ctx context.Context, path string, content io.Reader, maxBytes int64) error

PutFile streams content to path inside the node's confined storage directory.

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; an agent too old for it makes the collector skip this target.

func (*GRPCTransport) Stop ¶ 

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

Stop sends timeout < 0 as TimeoutMs -1, so an indefinite wait is not silently clamped to zero.

func (*GRPCTransport) UpdateResources ¶ 

func (t *GRPCTransport) UpdateResources(ctx context.Context, id string, resources docker.Resources) error

UpdateResources implements Transport (docker.Runtime).

type HostProbe ¶ added in v0.2.2 

type HostProbe func(ctx context.Context) HostReport

HostProbe reads the node's current host report.

func NewHostProbe ¶ added in v0.2.2 

func NewHostProbe(rt docker.Runtime, diskPath string) HostProbe

NewHostProbe reads host memory, this process's RSS and the unmanaged running containers rt can see. A runtime without stats support yields no container breakdown.

type HostReport ¶ added in v0.2.2 

type HostReport struct {
	MemoryTotalBytes           uint64
	MemoryAvailableBytes       uint64
	AgentRSSBytes              uint64
	OtherContainersMemoryBytes uint64
	OtherContainers            []UnmanagedContainer
	CPUCores                   uint32
	DiskTotalBytes             uint64
	DiskFreeBytes              uint64
	DiskInodesTotal            uint64
	DiskInodesFree             uint64
	SwapTotalBytes             uint64
	SwapFreeBytes              uint64
}

HostReport is one node's host memory split as the agent reads it. MemoryTotalBytes, DiskTotalBytes and DiskFreeBytes are zero when unreadable.

type HostReportSink ¶ added in v0.2.2 

type HostReportSink interface {
	SetNodeHostReport(nodeID string, r HostReport)
}

HostReportSink receives a node's host report on the control plane.

type HostReportStore ¶ added in v0.2.2 

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

HostReportStore keeps each node's latest host report in memory only: it is a live gauge, and a restart simply waits for the next agent tick.

func NewHostReportStore ¶ added in v0.2.2 

func NewHostReportStore() *HostReportStore

NewHostReportStore returns an empty store.

func (*HostReportStore) Get ¶ added in v0.2.2 

func (s *HostReportStore) Get(nodeID string) (HostReport, bool)

Get returns nodeID's latest report, false when none arrived or it is stale.

func (*HostReportStore) SetNodeHostReport ¶ added in v0.2.2 

func (s *HostReportStore) SetNodeHostReport(nodeID string, r HostReport)

SetNodeHostReport implements HostReportSink.

type Identity ¶ 

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

Identity is the PEM cert, key and CA that an enrolled node reconnects with; cmd/levelrail-agent persists them.

func DialEnroll ¶ 

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

DialEnroll exchanges joinToken for an Identity over addr; the private key never leaves the agent.

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 is the on-disk agent identity; writes are atomic, so a crash never leaves a torn one. The previous identity is kept as <path>.prev until a renewed certificate is 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
	NotBefore   time.Time
	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 this process's own docker.Runtime into a Transport for the local node, with no network hop.

func NewLocal ¶ 

func NewLocal(rt docker.Runtime) Local

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

func (Local) CopyFromContainer ¶ 

func (l Local) CopyFromContainer(ctx context.Context, containerID, path string, maxBytes int64) (io.ReadCloser, docker.PathStat, error)

CopyFromContainer implements docker.ArchiveRuntime by forwarding to the wrapped runtime, which the embedded interface would otherwise hide.

func (Local) CopyToContainer ¶ 

func (l Local) CopyToContainer(ctx context.Context, containerID, dir string, content io.Reader, maxBytes int64) error

CopyToContainer implements docker.ArchiveRuntime by forwarding.

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 forwards to the wrapped runtime, since embedding would hide the optional ExitStateInspector.

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

func (l Local) InspectImageLayers(ctx context.Context, ref string) ([]string, error)

InspectImageLayers implements docker.ImageLayerInspector by forwarding.

func (Local) LoadImage ¶ added in v0.2.0 

func (l Local) LoadImage(ctx context.Context, r io.Reader) error

LoadImage implements docker.ImageLoader by forwarding to the wrapped runtime.

func (Local) RunToCompletion ¶ added in v0.2.0 

func (l Local) RunToCompletion(ctx context.Context, spec docker.ContainerSpec, maxOutput int) (docker.RunOutcome, error)

RunToCompletion implements docker.RunToCompletionRuntime by forwarding to the wrapped runtime, which the embedded docker.Runtime would hide.

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 satisfied by *network.LocalSink's ApplyMesh and RotateKey methods, with no adapter.

type MeshStatusReader ¶ added in v0.2.0 

type MeshStatusReader interface {
	Status(ctx context.Context) (network.Status, error)
}

MeshStatusReader is the optional half of a MeshApplier that can report the node's live WireGuard view, for node-pair diagnostics.

type NodeFiles ¶ 

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

NodeFiles resolves a node ID to its session and runs file operations there.

func NewNodeFiles ¶ 

func NewNodeFiles(registry *Registry) *NodeFiles

NewNodeFiles builds the control plane side of agent file destinations over registry.

func (*NodeFiles) Delete ¶ 

func (n *NodeFiles) Delete(ctx context.Context, nodeID, path string) error

Delete implements backup.NodeFiles.

func (*NodeFiles) Get ¶ 

func (n *NodeFiles) Get(ctx context.Context, nodeID, path string) (io.ReadCloser, error)

Get implements backup.NodeFiles.

func (*NodeFiles) List ¶ 

func (n *NodeFiles) List(ctx context.Context, nodeID, prefix string) ([]FileEntry, error)

List implements backup.NodeFiles.

func (*NodeFiles) Put ¶ 

func (n *NodeFiles) Put(ctx context.Context, nodeID, path string, r io.Reader) error

Put implements backup.NodeFiles.

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 WithHostReportSink ¶ added in v0.2.2 

func WithHostReportSink(sink HostReportSink) Option

WithHostReportSink records the host 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 WithSecurityLog ¶ 

func WithSecurityLog(st SecurityStore, cfg SecurityConfig) Option

WithSecurityLog turns on the per-node security event log, stored in st.

func WithSessionObserver ¶ added in v0.2.0 

func WithSessionObserver(f func(nodeID, remoteIP string)) Option

WithSessionObserver calls f with each authenticated node's remote IP as its session opens, so the mesh can record where to reach that node.

type Registry ¶ 

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

Registry resolves a node ID to the Transport that reaches it.

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 RejectedHandshake ¶ 

type RejectedHandshake struct {
	Subject, SourceIP, Reason, Fingerprint string
}

RejectedHandshake describes a client certificate refused during the TLS handshake. Subject is what the certificate claims, not a verified identity.

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 implemented by transports that can run a build on their node.

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 RenewalStatus ¶ 

type RenewalStatus struct {
	NextRenewAt         time.Time
	LastError           string
	LastFailedAt        time.Time
	NextRetryAt         time.Time
	ConsecutiveFailures int
}

RenewalStatus is the renewer's schedule and last failure, reported to the control plane so the node page can show it.

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) RenewNow ¶ 

func (r *Renewer) RenewNow()

RenewNow makes Run attempt a renewal immediately, skipping any wait.

func (*Renewer) RenewOnce ¶ added in v0.2.0 

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

RenewOnce keeps the current certificate on failure, which the control plane still accepts during the 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. A RenewNow call skips the wait. The schedule comes from the certificate, so a restart resumes it.

func (*Renewer) Status ¶ 

func (r *Renewer) Status() RenewalStatus

Status returns the current schedule and failure state.

func (*Renewer) Updates ¶ 

func (r *Renewer) Updates() <-chan struct{}

Updates signals (coalesced) that Status changed.

type SecurityConfig ¶ 

type SecurityConfig struct {
	CoalesceWindow time.Duration
	PerNodeCap     int
	Retention      time.Duration
}

SecurityConfig bounds the log. Zero fields take the defaults.

type SecurityStore ¶ 

type SecurityStore interface {
	AddNodeSecurityEvent(ctx context.Context, ev store.NodeSecurityEvent, p store.SecurityEventPolicy, now time.Time) error
	PruneNodeSecurityEvents(ctx context.Context, cutoff time.Time) (int64, error)
	SetNodeCAPin(ctx context.Context, nodeID, state string, now time.Time) error
	SetNodeRenewalReport(ctx context.Context, s store.NodeTrustState, now time.Time) error
}

SecurityStore is the persistence the security event log needs. *store.DB satisfies it.

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 GoAway to every agent, best-effort; agents that miss it fall back to the GracefulStop timeout.

func (*Server) ObserveRejected ¶ 

func (s *Server) ObserveRejected(r RejectedHandshake)

ObserveRejected records a client certificate the TLS handshake refused.

func (*Server) RecordSecurityEvent ¶ 

func (s *Server) RecordSecurityEvent(ev store.NodeSecurityEvent)

RecordSecurityEvent appends an event without coalescing, for operator actions such as revoking a certificate.

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) RequestRenew ¶ 

func (s *Server) RequestRenew(nodeID string) error

RequestRenew asks nodeID's agent, over its live session, to renew its certificate now.

func (*Server) Session ¶ 

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

Session implements agentpb.AgentServiceServer for an mTLS-authenticated agent stream.

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 sets the build runner; without one, dispatched builds are rejected up front.

func WithFileStorage ¶ 

func WithFileStorage(dir string) SessionOption

WithFileStorage lets the control plane read and write objects under dir, and nowhere else, over this session.

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 the Heartbeat cadence; the env var is read by cmd/levelrail-agent, not here.

func WithHostProbe ¶ added in v0.2.2 

func WithHostProbe(p HostProbe) SessionOption

WithHostProbe makes RunSession report this node's host memory split to the control plane periodically.

func WithKeepalive ¶ added in v0.2.0 

func WithKeepalive(pingTime, timeout time.Duration) SessionOption

WithKeepalive overrides the HTTP/2 PING keepalive timing; the caller supplies env-derived durations.

func WithMesh ¶ added in v0.2.0 

func WithMesh(nodeID string, mesh MeshApplier) SessionOption

WithMesh enables ApplyMesh and RotateMeshKey on this stream; without it both answer ErrMeshUnavailable.

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 the node a resource is placed on. It embeds docker.Runtime so the two cannot drift; optional agent capabilities go in separate interfaces like RemoteBuilder.

type UnmanagedContainer ¶ added in v0.2.2 

type UnmanagedContainer struct {
	Name        string
	MemoryBytes uint64
}

UnmanagedContainer is one running container on a node that Levelrail does not manage.

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.