agent

Documentation

Overview

Package agent provides the node agent implementation.

Index

• Variables
• func ResolveFacts(params map[string]any, fr *job.FactsRegistration, hostname string) (map[string]any, error)
• func SeedSystemTemplates(ctx context.Context, logger *slog.Logger, objStore jetstream.ObjectStore) error
• type Agent
  • func New(appFs avfs.VFS, appConfig config.Config, logger *slog.Logger, ...) *Agent
  • func (a *Agent) GetFacts() map[string]any
  • func (a *Agent) IsReady() error
  • func (a *Agent) LastHeartbeatTime() time.Time
  • func (a *Agent) SetMeterProvider(mp *sdkmetric.MeterProvider)
  • func (a *Agent) SetSubComponents(scs map[string]job.SubComponentInfo)
  • func (a *Agent) Start()
  • func (a *Agent) State() string
  • func (a *Agent) Stop(ctx context.Context)
• type JobContext
• type NATSPublisher
• type PreflightResult
  • func RunPreflight(ctx context.Context, logger *slog.Logger, execManager exec.Manager) ([]PreflightResult, bool)
• type ProcessorFunc
  • func NewCertificateProcessor(certProvider certificate.Provider, logger *slog.Logger) ProcessorFunc
  • func NewCommandProcessor(commandProvider command.Provider, _ *slog.Logger) ProcessorFunc
  • func NewDockerProcessor(dockerProvider dockerProv.Provider, _ *slog.Logger) ProcessorFunc
  • func NewFileProcessor(fileProvider fileProv.Provider, _ *slog.Logger) ProcessorFunc
  • func NewNetworkProcessor(dnsProvider dns.Provider, pingProvider ping.Provider, ...) ProcessorFunc
  • func NewNodeProcessor(hostProvider nodeHost.Provider, diskProvider disk.Provider, ...) ProcessorFunc
  • func NewScheduleProcessor(cronProvider cron.Provider, logger *slog.Logger) ProcessorFunc
• type ProviderRegistry
  • func NewProviderRegistry() *ProviderRegistry
  • func (r *ProviderRegistry) AllProviders() []any
  • func (r *ProviderRegistry) Dispatch(ctx context.Context, req job.Request) (json.RawMessage, error)
  • func (r *ProviderRegistry) Register(category string, processFn ProcessorFunc, providers ...any)

Variables

var (
	// ErrJobEnvelopeMissing means PKI is enabled but the job data is not a
	// signed envelope at all, or is missing a required envelope field.
	ErrJobEnvelopeMissing = errors.New("job data is not a signed envelope")
	// ErrControllerKeyUnknown means PKI is enabled but this agent has not
	// completed enrollment, so it holds no controller public key to verify
	// against.
	ErrControllerKeyUnknown = errors.New("no cached controller public key: agent not enrolled")
	// ErrJobSignatureInvalid means the envelope signature did not verify
	// against the cached controller key (or the previous key, during a
	// rotation grace period).
	ErrJobSignatureInvalid = errors.New("invalid controller signature on job data")
)

Errors returned by unwrapJobEnvelope. Each names a distinct rejection reason so an operator reading agent logs or job termination reasons can tell an unenrolled agent apart from a forged or malformed job.

Functions

func ResolveFacts

func ResolveFacts(
	params map[string]any,
	fr *job.FactsRegistration,
	hostname string,
) (map[string]any, error)

ResolveFacts walks all string values in params and replaces @fact.X references with values from the agent's cached facts. Returns a new map with resolved values, or an error if any reference cannot be resolved.

func SeedSystemTemplates

func SeedSystemTemplates(
	ctx context.Context,
	logger *slog.Logger,
	objStore jetstream.ObjectStore,
) error

SeedSystemTemplates uploads embedded osapi templates to the NATS object store with the "osapi/" prefix. Templates are updated if the embedded content differs from what's in the store (SHA comparison). These templates are protected from user deletion.

Types

type Agent

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

Agent implements job processing with clean lifecycle management.

func New

func New(
	appFs avfs.VFS,
	appConfig config.Config,
	logger *slog.Logger,
	jobClient client.JobClient,
	streamName string,
	hostProvider nodeHost.Provider,
	diskProvider disk.Provider,
	memoryProvider memory.Provider,
	loadProvider load.Provider,
	netinfoProvider netinfo.Provider,
	processProvider process.Provider,
	registry *ProviderRegistry,
	registryKV jetstream.KeyValue,
	factsKV jetstream.KeyValue,
	execManager exec.Manager,
	natsClient NATSPublisher,
) *Agent

New creates a new agent instance.

func (*Agent) GetFacts

func (a *Agent) GetFacts() map[string]any

GetFacts returns the agent's current facts as a flat map suitable for template rendering. Returns nil if facts haven't been collected yet. Uses JSON round-trip so the map automatically includes all fields from FactsRegistration without hardcoding field names.

func (*Agent) IsReady

func (a *Agent) IsReady() error

IsReady returns nil when the agent is ready to process jobs, or an error describing why it is not.

func (*Agent) LastHeartbeatTime

func (a *Agent) LastHeartbeatTime() time.Time

LastHeartbeatTime returns the timestamp of the last successful heartbeat write. Returns zero time if no heartbeat has been written.

func (*Agent) SetMeterProvider

func (a *Agent) SetMeterProvider(
	mp *sdkmetric.MeterProvider,
)

SetMeterProvider creates OTEL instruments for job metrics.

func (*Agent) SetSubComponents

func (a *Agent) SetSubComponents(
	scs map[string]job.SubComponentInfo,
)

SetSubComponents sets the sub-component info published in heartbeats.

func (*Agent) Start

func (a *Agent) Start()

Start starts the agent without blocking. Call Stop to shut down.

func (*Agent) State

func (a *Agent) State() string

State returns the agent's scheduling state. Read by the heartbeat on its own goroutine while a drain or an enrollment acceptance writes it.

func (*Agent) Stop

func (a *Agent) Stop(
	ctx context.Context,
)

Stop gracefully shuts down the agent, waiting for in-flight jobs to finish or the context deadline to expire.

type JobContext

type JobContext struct {
	// JobID from the original job request
	JobID string
	// AgentHostname identifies which agent is processing this job
	AgentHostname string
	// JobData contains the raw job request data
	JobData []byte
	// ResponseKV is the key-value bucket for storing responses
	ResponseKV string
}

JobContext contains the context and data for a single job execution.

type NATSPublisher

type NATSPublisher interface {
	PublishCore(subject string, data []byte) error
	Subscribe(subject string, handler nats.MsgHandler) (*nats.Subscription, error)
}

NATSPublisher provides core NATS pub/sub for enrollment. Satisfied by the nats-client's Client type.

type PreflightResult

type PreflightResult struct {
	Name   string
	Passed bool
	Error  string
}

PreflightResult holds the outcome of a single preflight check.

func RunPreflight

func RunPreflight(
	ctx context.Context,
	logger *slog.Logger,
	execManager exec.Manager,
) ([]PreflightResult, bool)

RunPreflight runs sudo and capability checks. When called, it always checks both sudo access and Linux capabilities. Returns the combined results and whether all checks passed.

type ProcessorFunc

type ProcessorFunc func(context.Context, job.Request) (json.RawMessage, error)

ProcessorFunc is a function that handles a job request and returns a JSON-encoded result.

func NewCertificateProcessor

func NewCertificateProcessor(
	certProvider certificate.Provider,
	logger *slog.Logger,
) ProcessorFunc

NewCertificateProcessor returns a ProcessorFunc that handles certificate-related operations.

func NewCommandProcessor

func NewCommandProcessor(
	commandProvider command.Provider,
	_ *slog.Logger,
) ProcessorFunc

NewCommandProcessor returns a ProcessorFunc that handles command-related operations.

func NewDockerProcessor

func NewDockerProcessor(
	dockerProvider dockerProv.Provider,
	_ *slog.Logger,
) ProcessorFunc

NewDockerProcessor returns a ProcessorFunc that handles docker-related operations.

func NewFileProcessor

func NewFileProcessor(
	fileProvider fileProv.Provider,
	_ *slog.Logger,
) ProcessorFunc

NewFileProcessor returns a ProcessorFunc that handles file-related operations.

func NewNetworkProcessor

func NewNetworkProcessor(
	dnsProvider dns.Provider,
	pingProvider ping.Provider,
	interfaceProvider iface.Provider,
	routeProvider route.Provider,
	logger *slog.Logger,
) ProcessorFunc

NewNetworkProcessor returns a ProcessorFunc that handles network-related operations.

func NewNodeProcessor

func NewNodeProcessor(
	hostProvider nodeHost.Provider,
	diskProvider disk.Provider,
	memoryProvider memory.Provider,
	loadProvider load.Provider,
	sysctlProvider sysctl.Provider,
	ntpProvider ntp.Provider,
	timezoneProvider timezone.Provider,
	powerProvider power.Provider,
	processProvider processProv.Provider,
	userProvider user.Provider,
	packageProvider apt.Provider,
	logProvider logProv.Provider,
	serviceProvider serviceProv.Provider,
	appConfig config.Config,
	logger *slog.Logger,
) ProcessorFunc

NewNodeProcessor returns a ProcessorFunc that handles node-related operations.

func NewScheduleProcessor

func NewScheduleProcessor(
	cronProvider cron.Provider,
	logger *slog.Logger,
) ProcessorFunc

NewScheduleProcessor returns a ProcessorFunc that handles schedule-related operations.

type ProviderRegistry

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

ProviderRegistry maps job categories to processor functions and tracks all registered providers for lifecycle wiring (e.g., facts).

func NewProviderRegistry

func NewProviderRegistry() *ProviderRegistry

NewProviderRegistry creates an empty ProviderRegistry.

func (*ProviderRegistry) AllProviders

func (r *ProviderRegistry) AllProviders() []any

AllProviders returns all providers registered across all categories.

func (*ProviderRegistry) Dispatch

func (r *ProviderRegistry) Dispatch(
	ctx context.Context,
	req job.Request,
) (json.RawMessage, error)

Dispatch routes a job request to the registered processor for its category.

func (*ProviderRegistry) Register

func (r *ProviderRegistry) Register(
	category string,
	processFn ProcessorFunc,
	providers ...any,
)

Register associates a category with a processor function and records any providers for WireProviderFacts.