agent

package
v0.1.0-alpha.8 Latest Latest
Warning

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

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

Documentation

Overview

Package agent is the node-side runtime that keeps a node in sync with the control plane. It maintains a single AgentSync bidi stream: a hello on every (re)connect, periodic heartbeats carrying live host metrics, and applies the full AssignmentSet the control plane pushes. Reconciling those assignments against the local container runtime is the executor's job (T-15); this skeleton establishes the stream, heartbeats, and reconnection.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func GatewayIP 

func GatewayIP(cidr string) (string, error)

GatewayIP returns the gateway (.1) address of a /24 subnet — where the per-network internal DNS resolver binds (T-47).

func NetworkName 

func NetworkName(projectID, envID string) string

NetworkName is the node-local Docker network name for an environment. It is deterministic (idempotent EnsureNetwork) and valid per Docker's naming rules.

func NextFreeSubnet 

func NextFreeSubnet(used []string) (string, error)

NextFreeSubnet returns the lowest 10.201.X.0/24 not present in used, or an error when the pool is exhausted.

Types

type Agent 

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

Agent maintains the node's AgentSync stream to the control plane.

func New 

func New(cfg Config) *Agent

New builds an Agent, filling defaults.

func (*Agent) Executor 

func (a *Agent) Executor() *Executor

Executor returns the node's executor (nil when the agent runs without a container runtime), exposing its assignment view for log stream resolution.

func (*Agent) Run 

func (a *Agent) Run(ctx context.Context) error

Run maintains the sync stream until ctx is canceled, reconnecting with exponential backoff + jitter after any session ends.

type BuildServer 

type BuildServer struct {
	clusterv1.UnimplementedAgentLocalServiceServer
	// contains filtered or unexported fields
}

BuildServer implements the build methods of AgentLocalService on builder nodes (T-35). Other AgentLocalService methods are added by later tasks (T-41 logs, T-49 exec, T-65 volumes) via the same embedded server.

func NewBuildServer 

func NewBuildServer(cfg BuildServerConfig) *BuildServer

NewBuildServer constructs the build server.

func (*BuildServer) CancelBuild 

func (s *BuildServer) CancelBuild(_ context.Context, req *clusterv1.CancelBuildRequest) (*clusterv1.CancelBuildResponse, error)

CancelBuild aborts an in-flight build on this node.

func (*BuildServer) RunBuild 

func (s *BuildServer) RunBuild(req *clusterv1.RunBuildRequest, stream grpc.ServerStreamingServer[clusterv1.BuildEvent]) error

RunBuild fetches the source, builds it (Dockerfile or nixpacks), pushes the image to the registry, and streams progress back to the control plane.

type BuildServerConfig 

type BuildServerConfig struct {
	Builder          *builder.Builder
	Fetch            SourceFetcher
	RegistryAuth     builder.RegistryAuth
	RegistryInsecure bool
	// LocalLoad loads the built image into the local Docker store instead of
	// pushing to a registry (single-node/dev, T-54).
	LocalLoad bool
	Logger    *slog.Logger
}

BuildServerConfig configures the build server.

type Config 

type Config struct {
	NodeID  string
	Version string
	Clock   clock.Clock
	Logger  *slog.Logger

	// Runtime is reconciled against pushed assignments by the executor (T-15).
	// When nil the agent maintains the stream/heartbeats but runs no executor
	// (e.g. a control-only node without Docker).
	Runtime runtime.ContainerRuntime
	// HostIP is where the executor publishes container ports (mesh IP, or
	// 127.0.0.1 in single-node/dev).
	HostIP string
	// RegistryAuth pulls private images (join credential, T-17); nil for public.
	RegistryAuth *runtime.RegistryAuth

	// Dial opens a fresh connection to a control node's AgentSyncService. The
	// agent calls it on every (re)connect and closes the returned Conn when the
	// session ends.
	Dial func(ctx context.Context) (*Conn, error)

	// Sample returns host metrics for heartbeats. Defaults to a gopsutil probe.
	Sample SampleFunc
	// DiskPath is the filesystem the default sampler probes for disk usage.
	DiskPath string

	// HeartbeatInterval overrides the 10s default (tests shorten it via a fake
	// clock).
	HeartbeatInterval time.Duration

	// Store, when set, enables the metrics sampler (T-59): a 15s loop recording
	// node, per-instance and proxy series into the local ring TSDB.
	Store tsdb.Store
	// ProxyStats supplies per-env proxy samples for the metrics sampler (nil on
	// nodes without a proxy).
	ProxyStats ProxyMetricsFunc
	// MetricsInterval overrides the 15s sampler cadence (tests use a fake clock).
	MetricsInterval time.Duration

	// OnAssignments is invoked with every AssignmentSet the control plane
	// pushes. The executor (T-15) reconciles Docker to it; the skeleton records
	// the version and forwards the set here.
	OnAssignments func(*clusterv1.AssignmentSet)
}

Config configures an Agent.

type Conn 

type Conn struct {
	grpc.ClientConnInterface
	Close func() error
}

Conn is a control-plane connection paired with a closer. The agent redials on every (re)connect and closes the connection when a session ends.

type ExecServer 

type ExecServer struct {
	clusterv1.UnimplementedAgentLocalServiceServer
	// contains filtered or unexported fields
}

ExecServer implements the interactive methods of AgentLocalService (T-49): Exec (bidi TTY), Top (process snapshot) and ProxyTCP (port-forward leg). It runs on every node against the local container runtime; the control plane's ExecService relays user streams to it over the mesh.

It embeds UnimplementedAgentLocalServiceServer so it can stand alone in tests; the production node composes these methods with the build/log/volume methods into one AgentLocalService when the :8444 mTLS listener is wired.

func NewExecServer 

func NewExecServer(rt crt.ContainerRuntime, log *slog.Logger) *ExecServer

NewExecServer builds the exec server over a container runtime.

func (*ExecServer) Exec 

func (s *ExecServer) Exec(stream grpc.BidiStreamingServer[clusterv1.AgentExecInput, clusterv1.AgentExecOutput]) error

Exec runs a command inside a running container and relays its TTY/pipes over the bidi stream. The first client message must carry AgentExecStart.

func (*ExecServer) ProxyTCP 

func (s *ExecServer) ProxyTCP(stream grpc.BidiStreamingServer[clusterv1.TCPChunk, clusterv1.TCPChunk]) error

ProxyTCP splices the bidi stream to a TCP target. The first chunk carries the dial address ("containerIP:port" or "meshIP:port"); its data (if any) is forwarded after the dial.

func (*ExecServer) Top 

func (s *ExecServer) Top(ctx context.Context, req *clusterv1.AgentTopRequest) (*clusterv1.AgentTopResponse, error)

Top returns a one-shot process listing from inside the container.

type Executor 

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

Executor converges the local container runtime to the assignment set the control plane pushes. It is the agent's reconcile loop: idempotent, crash-safe (adopts existing containers by label on restart), and never touches containers it does not own.

func NewExecutor 

func NewExecutor(cfg ExecutorConfig) *Executor

NewExecutor builds an Executor, filling defaults.

func (*Executor) InstanceStats 

func (e *Executor) InstanceStats(ctx context.Context) []InstanceMetrics

InstanceStats samples resource usage for every service container this node owns, keyed by instance (assignment) id. It powers the metrics sampler (T-59); a container whose one-shot stat probe fails is skipped rather than failing the whole sweep.

func (*Executor) MatchingStreams 

func (e *Executor) MatchingStreams(sel *zatterav1.LogSelector) []logstore.StreamID

MatchingStreams returns the assignment ids on this node whose metadata match the log selector (used by the agent log query server, T-54).

func (*Executor) NetworkScopes 

func (e *Executor) NetworkScopes() []NetScope

NetworkScopes returns the distinct (project, env) bridge scopes this node currently has assignments for — one per bridge gateway. It is the input to the internal DNS resolver's Reconcile (F26): the resolver serves <svc>.internal on each gateway within that gateway's project+env scope.

func (*Executor) Run 

func (e *Executor) Run(ctx context.Context)

Run drives reconciliation: on every submitted set and on a periodic tick (which retries failed assignments and re-checks liveness) until ctx ends.

func (*Executor) Submit 

func (e *Executor) Submit(set *clusterv1.AssignmentSet)

Submit hands the executor the latest assignment set (coalescing: only the newest pending set is kept) and wakes the Run loop.

type ExecutorConfig 

type ExecutorConfig struct {
	NodeID string
	// HostIP is where published ports bind (the node's mesh IP, or 127.0.0.1 in
	// single-node/dev).
	HostIP  string
	Runtime runtime.ContainerRuntime
	Clock   clock.Clock
	Logger  *slog.Logger
	// Report emits status transitions. Required in production; tests capture it.
	Report StatusSink
	// RegistryAuth pulls private images (from the join credential, T-17). May be
	// nil for public images.
	RegistryAuth *runtime.RegistryAuth
	// PollInterval drives the liveness re-check + retry tick (default 5s).
	PollInterval time.Duration
}

ExecutorConfig configures the reconciler.

type HTTPSourceFetcher 

type HTTPSourceFetcher struct {
	Client *http.Client
}

HTTPSourceFetcher fetches source tarballs over HTTP(S) with the node's mTLS client (the control blob endpoint is node-cert-authenticated).

func (HTTPSourceFetcher) Fetch 

func (h HTTPSourceFetcher) Fetch(ctx context.Context, url string) (io.ReadCloser, error)

Fetch GETs url and returns the body stream.

type HealthConfig 

type HealthConfig struct {
	Interval  time.Duration
	Timeout   time.Duration
	Grace     time.Duration
	Threshold int
}

HealthConfig is the resolved probe schedule for one instance.

type HostSample 

type HostSample struct {
	CPUPercent       float64
	MemoryUsedBytes  uint64
	MemoryTotalBytes uint64
	DiskUsedBytes    uint64
	DiskTotalBytes   uint64
}

HostSample is a point-in-time snapshot of node resource usage carried by a heartbeat.

type InstanceMetrics 

type InstanceMetrics struct {
	InstanceID  string
	CPUPercent  float64
	MemoryBytes float64
	NetRxBytes  float64
	NetTxBytes  float64
}

InstanceMetrics is one instance's resource sample keyed by its instance id (the assignment id).

type InstanceMetricsFunc 

type InstanceMetricsFunc func(ctx context.Context) []InstanceMetrics

InstanceMetricsFunc returns per-instance samples for every instance running on this node.

type LocalServer 

type LocalServer struct {
	clusterv1.UnimplementedAgentLocalServiceServer
	// contains filtered or unexported fields
}

LocalServer composes the node-local AgentLocalService methods that the control plane dials over the mesh (T-54): builds (BuildServer) and interactive exec/top/port-forward (ExecServer). QueryLogs/Stats/volume methods remain Unimplemented until their agent-side backends land.

A single gRPC server can register only one AgentLocalServiceServer, so this type forwards each method to the right sub-server instead of embedding both.

func NewLocalServer 

func NewLocalServer(build *BuildServer, exec *ExecServer, logs *LogServer, stats *StatsServer, rt runtime.ContainerRuntime) *LocalServer

NewLocalServer builds the composite. Any sub-server may be nil (e.g. a non-builder node passes build=nil); the corresponding methods then report Unimplemented via the embedded base. rt backs the volume data-path ops.

func (*LocalServer) CancelBuild 

func (s *LocalServer) CancelBuild(ctx context.Context, req *clusterv1.CancelBuildRequest) (*clusterv1.CancelBuildResponse, error)

CancelBuild dispatches to the build sub-server.

func (*LocalServer) EnableSelfUpgrade 

func (s *LocalServer) EnableSelfUpgrade(cfg *UpgradeConfig)

EnableSelfUpgrade turns on UpgradeBinary for this node.

func (*LocalServer) Exec 

func (s *LocalServer) Exec(stream grpc.BidiStreamingServer[clusterv1.AgentExecInput, clusterv1.AgentExecOutput]) error

Exec dispatches to the exec sub-server.

func (*LocalServer) ListVolumeFiles 

func (s *LocalServer) ListVolumeFiles(ctx context.Context, req *clusterv1.AgentListFilesRequest) (*clusterv1.AgentListFilesResponse, error)

ListVolumeFiles lists one directory inside a volume (T-77). Read-only: it never follows a path out of the volume, and never writes.

func (*LocalServer) ProxyTCP 

func (s *LocalServer) ProxyTCP(stream grpc.BidiStreamingServer[clusterv1.TCPChunk, clusterv1.TCPChunk]) error

ProxyTCP dispatches to the exec sub-server.

func (*LocalServer) QueryLogs 

func (s *LocalServer) QueryLogs(q *zatterav1.LogQuery, stream grpc.ServerStreamingServer[zatterav1.LogLine]) error

QueryLogs dispatches to the log sub-server.

func (*LocalServer) ReadVolumeFile 

func (s *LocalServer) ReadVolumeFile(req *clusterv1.AgentReadFileRequest, stream grpc.ServerStreamingServer[clusterv1.TCPChunk]) error

ReadVolumeFile streams one regular file out of a volume (T-77).

func (*LocalServer) RemoveVolume 

func (s *LocalServer) RemoveVolume(ctx context.Context, req *clusterv1.AgentRemoveVolumeRequest) (*clusterv1.AgentRemoveVolumeResponse, error)

RemoveVolume deletes the node-local docker volume for a deleted Volume (T-62).

func (*LocalServer) RestoreVolume 

func (s *LocalServer) RestoreVolume(req *clusterv1.AgentRestoreVolumeRequest, stream grpc.ServerStreamingServer[clusterv1.VolumeOpProgress]) error

RestoreVolume restores a snapshot into the volume's host path. The control plane guarantees the service is stopped before dispatching (single-writer).

func (*LocalServer) RunBuild 

func (s *LocalServer) RunBuild(req *clusterv1.RunBuildRequest, stream grpc.ServerStreamingServer[clusterv1.BuildEvent]) error

RunBuild dispatches to the build sub-server.

func (*LocalServer) SnapshotVolume 

func (s *LocalServer) SnapshotVolume(req *clusterv1.AgentSnapshotVolumeRequest, stream grpc.ServerStreamingServer[clusterv1.VolumeOpProgress]) error

SnapshotVolume snapshots the named volume's host path to S3 (T-65). It optionally runs a pre-hook command inside the mounting container (e.g. pg_dump) first, then chunks + dedups + encrypts via the snapshot engine, streaming progress. The final message carries the manifest key.

func (*LocalServer) Stats 

func (s *LocalServer) Stats(ctx context.Context, q *zatterav1.StatsQuery) (*zatterav1.StatsResponse, error)

Stats dispatches to the stats sub-server (local ring TSDB, T-60).

func (*LocalServer) Top 

func (s *LocalServer) Top(ctx context.Context, req *clusterv1.AgentTopRequest) (*clusterv1.AgentTopResponse, error)

Top dispatches to the exec sub-server.

func (*LocalServer) UpgradeBinary 

func (s *LocalServer) UpgradeBinary(ctx context.Context, req *clusterv1.AgentUpgradeRequest) (*clusterv1.AgentUpgradeResponse, error)

UpgradeBinary downloads, verifies and installs a new zattera binary, then restarts the daemon (T-95).

Workload containers are docker-managed and outlive this process — the executor returns on context cancel without stopping anything — so the restart costs this node's ingress and agent stream for a few seconds, not its workloads.

type LogCapture 

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

LogCapture follows the logs of every managed service container and appends them to the node-local logstore keyed by assignment id (T-54). The control plane reads them back via AgentLocalService.QueryLogs.

func NewLogCapture 

func NewLogCapture(rt runtime.ContainerRuntime, store *logstore.Segmented, log *slog.Logger) *LogCapture

NewLogCapture builds a capture over a runtime + logstore.

func (*LogCapture) Run 

func (c *LogCapture) Run(ctx context.Context)

Run polls the owned service containers and keeps one follower per assignment, starting followers for new containers and stopping them for gone ones.

type LogServer 

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

LogServer implements AgentLocalService.QueryLogs from the node-local logstore (T-54). The control plane's LogService fans this out across nodes and merges.

func NewLogServer 

func NewLogServer(store *logstore.Segmented, resolve StreamResolver) *LogServer

NewLogServer builds the agent log query server.

func (*LogServer) QueryLogs 

func (s *LogServer) QueryLogs(q *zatterav1.LogQuery, stream grpc.ServerStreamingServer[zatterav1.LogLine]) error

QueryLogs streams stored (and, when follow is set, live) log lines for the selector's local streams.

type Manager 

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

Manager owns one Monitor per running assignment. The executor calls Ensure when a container reaches RUNNING and Remove when it is torn down.

func NewManager 

func NewManager(base context.Context, cfg ManagerConfig) *Manager

NewManager builds a health manager. base bounds every monitor's lifetime.

func (*Manager) Ensure 

func (m *Manager) Ensure(ctx context.Context, a *zatterav1.Assignment, spec *zatterav1.ServiceSpec, containerID string)

Ensure starts a monitor for the assignment if one is not already running (idempotent, so callers may invoke it on every reconcile). A service without a health check is reported HEALTHY immediately. For HTTP/TCP probes the container is inspected to resolve the probe address; the inspect happens only when a monitor is actually created.

func (*Manager) Reconcile 

func (m *Manager) Reconcile(live map[string]bool)

Reconcile stops monitors whose assignment is no longer live.

func (*Manager) Remove 

func (m *Manager) Remove(assignID string)

Remove stops the monitor for an assignment (idempotent).

type ManagerConfig 

type ManagerConfig struct {
	Clock   clock.Clock
	Report  StatusSink
	Runtime crt.ContainerRuntime
	HostIP  string
	Logger  *slog.Logger
}

ManagerConfig configures the health manager.

type Monitor 

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

Monitor probes one instance on the clock and reports state transitions (only on change) via the sink.

func (*Monitor) Run 

func (m *Monitor) Run(ctx context.Context)

Run probes until ctx is canceled.

type NetScope 

type NetScope struct {
	Gateway   string
	ProjectID string
	EnvID     string
}

NetScope is one (project, env) bridge network this node runs containers on, with the gateway IP where the per-network internal DNS resolver binds.

type NodeMetrics 

type NodeMetrics struct {
	CPUPercent float64
	MemUsed    float64
	MemTotal   float64
	DiskUsed   float64
	DiskTotal  float64
	NetRxBytes float64
	NetTxBytes float64
}

NodeMetrics is a point-in-time node resource sample recorded to the TSDB. Network counters are cumulative byte totals (rates are derived downstream).

type NodeMetricsFunc 

type NodeMetricsFunc func() NodeMetrics

NodeMetricsFunc returns the current node sample. Production uses gopsutil.

type Probe 

type Probe func(ctx context.Context) error

Probe performs one health check. A nil error means the instance passed.

type ProxyMetricsFunc 

type ProxyMetricsFunc func() map[string]*clusterv1.ProxySample

ProxyMetricsFunc returns per-environment proxy samples (nil when this node runs no proxy).

type SampleFunc 

type SampleFunc func() HostSample

SampleFunc returns the current host sample. Production uses gopsutil; tests inject a deterministic stub.

type SourceFetcher 

type SourceFetcher interface {
	Fetch(ctx context.Context, url string) (io.ReadCloser, error)
}

SourceFetcher fetches a source tarball from a control-plane blob URL. The production implementation dials over node mTLS; tests inject a fake.

type StatsServer 

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

StatsServer serves AgentLocalService.Stats from this node's local ring TSDB (T-60). The control plane fans a StatsQuery out to each relevant node and merges the returned series; this side just applies the scope/metric/time filter against local series.

func NewStatsServer 

func NewStatsServer(store tsdb.Store, nodeID string, clk clock.Clock) *StatsServer

NewStatsServer builds the agent-side historical stats server.

func (*StatsServer) Stats 

func (s *StatsServer) Stats(_ context.Context, q *zatterav1.StatsQuery) (*zatterav1.StatsResponse, error)

Stats returns local series matching the query's scope, metric filter and time range at the resolution nearest the requested step.

type StatusSink 

type StatusSink func(observed map[string]*zatterav1.AssignmentObserved)

StatusSink receives observed instance-status transitions. The agent wires it to enqueue a StatusBatch on the sync stream.

type StreamResolver 

type StreamResolver func(sel *zatterav1.LogSelector) []logstore.StreamID

StreamResolver maps a log selector to the node-local stream ids (assignment ids) it should read, using the metadata the agent already holds.

type UpgradeConfig 

type UpgradeConfig struct {
	// AllowedBaseURL is the only prefix this node will download from. The
	// control plane picks the URL, but a node that blindly executed whatever
	// URL it was handed would turn any control-plane compromise into arbitrary
	// code on every node; this bounds that.
	AllowedBaseURL string
	// Client fetches the binary (nil = a default with a long timeout).
	Client *http.Client
	// Restart replaces the running process. Nil uses restartSelf.
	Restart func(context.Context) error
	// ExecPath overrides the binary location (tests).
	ExecPath string
	Logger   *slog.Logger
}

UpgradeConfig configures self-upgrade on a node.

Source Files

View all Source files

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.