agent

type Agent ¶

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

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

func New(cfg Config) *Agent

func (a *Agent) Executor() *Executor

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

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(cfg BuildServerConfig) *BuildServer

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

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

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

	// 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(rt crt.ContainerRuntime, log *slog.Logger) *ExecServer

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

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

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

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(cfg ExecutorConfig) *Executor

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

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

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

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

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 (h HTTPSourceFetcher) Fetch(ctx context.Context, url string) (io.ReadCloser, error)

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 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(build *BuildServer, exec *ExecServer, logs *LogServer) *LocalServer

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

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

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

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

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

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

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(rt runtime.ContainerRuntime, store *logstore.Segmented, log *slog.Logger) *LogCapture

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

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(store *logstore.Segmented, resolve StreamResolver) *LogServer

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

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(base context.Context, cfg ManagerConfig) *Manager

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

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

func (m *Manager) Remove(assignID string)

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 (m *Monitor) Run(ctx context.Context)

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

type Probe func(ctx context.Context) error

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

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