agentserver

package
v1.7.0 Latest Latest
Warning

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

 Go to latest Published: Sep 18, 2026 License: AGPL-3.0 Imports: 36 Imported by: 0

Documentation

Overview

Package agentserver implements the gRPC Ingest service that remote agents connect to for enrollment and event streaming (Pro only).

Index

Constants

View Source 
const CapabilityLogs = "logs"

CapabilityLogs is advertised by agents able to serve container logs on demand.

View Source 
const OfflineReportGrace = 2 * time.Minute

OfflineReportGrace is how long after startup the stale watcher waits before reporting agents that are absent with no live stream. An agent reconnects with a backoff capped at 60s ±25% jitter, so a shorter delay would page for every agent still on its way back after a server restart.

Variables

View Source 
var (
	ErrBadSignature = errors.New("invalid ed25519 signature")
	ErrClockSkew    = errors.New("client clock skew exceeds 300s")
	ErrAgentRevoked = errors.New("agent is revoked")
	ErrAgentUnknown = errors.New("agent not found")
)
View Source 
var (
	ErrSessionRevoked  = errors.New("agent_revoked")
	ErrSessionReplaced = errors.New("session_replaced")
	ErrSessionStale    = errors.New("session_stale")
	ErrSessionClosed   = errors.New("session_closed")
)

Reasons a session is torn down. They travel to the Push handler as the stream context's cancellation cause, so it can tell the agent whether to give up or reconnect — reporting every teardown as a revocation would make an agent that was merely reaped as stale exit for good.

View Source 
var (
	ErrAgentNotConnected = errors.New("agent not connected")
	ErrAgentCannotServe  = errors.New("agent does not support this command")
	ErrTooManyRequests   = errors.New("too many in-flight commands for this agent")
)

Errors returned by the command path, mapped to HTTP status by the API layer.

Functions

func CauseForReason added in v1.3.8 

func CauseForReason(reason string) error

CauseForReason maps a Close reason to the cause handed to the agent.

func GenerateChallenge 

func GenerateChallenge() (*agentpb.AuthChallenge, error)

GenerateChallenge creates a 32-byte random nonce for the auth handshake.

func LoadOrGenerateTLS 

func LoadOrGenerateTLS(certFile, keyFile string, hosts []string, logger *slog.Logger) (*tls.Config, error)

LoadOrGenerateTLS returns a TLS config for the agent gRPC server.

If both certFile and keyFile are non-empty, the keypair is loaded from disk. Otherwise a self-signed dev certificate is generated in-memory and a loud warning is logged — agents must then connect with --grpc-insecure-skip-tls-verify. Intended for local development only.

hosts seeds the certificate SAN (Subject Alternative Names) so a strict client can still verify it. Empty/unknown hosts default to 127.0.0.1 + localhost.

func LogsCommand added in v1.3.8 

func LogsCommand(externalID string, lines int, timestamps, follow bool) *agentpb.AgentCommand

LogsCommand builds a logs command with a fresh request id. Exported because the SSE follow path drives SendCommand directly to stream chunks as they arrive.

func ResolvePublicURL 

func ResolvePublicURL(req *http.Request, cfg PublicURLConfig) (string, []string)

ResolvePublicURL returns the grpcs:// URL that remote agents should use plus a list of warnings. Resolution priority: explicit config > X-Forwarded-Host+Proto headers from the HTTP request > request Host header.

A "public_url_appears_local" warning is appended whenever the resolved host is localhost, 127.x.x.x, ::1, or an RFC-1918 / link-local address.

func Verify 

func Verify(req *agentpb.AuthResponse, nonce []byte, ag *agent.Agent, serverNow time.Time) error

Verify checks the AuthResponse signature and returns nil on success. Payload (byte-exact): nonce(32) || agent_uuid_bytes(16) || timestamp_be64(8).

Types

type CertificateHandler 

type CertificateHandler interface {
	HandleAgentEvent(ctx context.Context, agentID string, ev *agentpb.CertificateInfo) error
}

CertificateHandler processes a certificate scan result from a remote agent.

type ContainerHandler 

type ContainerHandler interface {
	HandleAgentEvent(ctx context.Context, agentID string, ev *agentpb.ContainerEvent, meta EventMeta) error
}

ContainerHandler processes a container event from a remote agent.

type ContainerInventoryHandler added in v1.3.8 

type ContainerInventoryHandler interface {
	HandleAgentInventory(ctx context.Context, agentID string, ev *agentpb.ContainerInventory, meta EventMeta) error
}

ContainerInventoryHandler reconciles a full container snapshot from an agent.

type Deps 

type Deps struct {
	AgentStore  *store.AgentStore
	Broadcaster EventBroadcaster
	Sessions    *Sessions
	Limiter     *Limiter
	Dispatcher  *Dispatcher
	Logger      *slog.Logger
}

Deps groups the dependencies required by the agent gRPC server.

type DispatchDeps 

type DispatchDeps struct {
	Container   ContainerHandler
	Inventory   ContainerInventoryHandler
	Endpoint    EndpointHandler
	Heartbeat   HeartbeatHandler
	Resource    ResourceHandler
	Certificate CertificateHandler
	Swarm       SwarmTopologyHandler
	Kubernetes  KubernetesTopologyHandler
	HostOS      HostOSHandler
	// LabelSync, if set, provisions endpoint/cert monitors from a container's
	// labels after each container event. Optional (nil = no label discovery).
	LabelSync LabelSyncFunc
}

DispatchDeps groups the optional per-domain handlers the dispatcher calls. A nil handler means that event type is silently ignored.

type Dispatcher 

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

Dispatcher routes AgentEvents to the appropriate domain handler.

func NewDispatcher 

func NewDispatcher(deps DispatchDeps) *Dispatcher

NewDispatcher creates a Dispatcher with the given handler set.

func (*Dispatcher) Dispatch 

func (d *Dispatcher) Dispatch(ctx context.Context, agentID string, evt *agentpb.AgentEvent) error

Dispatch routes evt to the handler matching its body type, attributing every event to agentID — the identity proven by the auth handshake, NOT the client-controlled agent_id carried on the wire. Returns an error if a handler is wired and returns an error. Silently ignores events whose handler is nil.

func (*Dispatcher) RejectedEvents added in v1.6.0 

func (d *Dispatcher) RejectedEvents(agentID string) uint64

RejectedEvents returns how many events the dispatcher refused for an out of range observation time since startup, for agentID.

type EndpointHandler 

type EndpointHandler interface {
	HandleAgentEvent(ctx context.Context, agentID string, ev *agentpb.EndpointEvent, meta EventMeta) error
}

EndpointHandler processes an endpoint probe result from a remote agent.

type EventBroadcaster 

type EventBroadcaster interface {
	BroadcastEvent(eventType string, data any)
}

EventBroadcaster is the minimal interface required to publish SSE events to connected clients.

type EventMeta added in v1.6.0 

type EventMeta = agentevent.Meta

EventMeta is the observation time and the replay flag of an agent event.

type HeartbeatHandler 

type HeartbeatHandler interface {
	HandleAgentEvent(ctx context.Context, agentID string, ev *agentpb.HeartbeatEvent) error
}

HeartbeatHandler processes a heartbeat ping from a remote agent.

type HostOSHandler added in v1.7.0 

type HostOSHandler interface {
	HandleAgentHostOS(ctx context.Context, agentID string, ev *agentpb.HostOSMsg) error
}

HostOSHandler records the operating system identity an agent reports for its host.

type KubernetesTopologyHandler 

type KubernetesTopologyHandler interface {
	HandleAgentEvent(ctx context.Context, agentID string, ev *agentpb.KubernetesTopology) error
}

KubernetesTopologyHandler processes a full Kubernetes topology snapshot from an agent.

type LabelSyncFunc 

type LabelSyncFunc func(ctx context.Context, agentID, containerName, externalID string, labels map[string]string)

LabelSyncFunc provisions label-discovered endpoint/cert monitors for a remote agent's container. Invoked for every container event so monitors track label changes: created on first sight, deprovisioned when a label is removed. The agent itself probes them and pushes the results.

type Limiter 

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

Limiter manages per-agent token-bucket rate limiters.

func NewLimiter 

func NewLimiter(eventsPerSecond int) *Limiter

NewLimiter creates a per-agent Limiter allowing eventsPerSecond events per second for each agent, with an equal burst. A value <= 0 falls back to the default (1000/s). Driven by MAINTENANT_AGENT_RATE_LIMIT_PER_SECOND.

func (*Limiter) Allow 

func (l *Limiter) Allow(agentID string) (bool, time.Duration)

Allow reports whether agentID may process one more event right now. Returns (allowed, retryAfter). retryAfter is only meaningful when !allowed.

func (*Limiter) Cleanup 

func (l *Limiter) Cleanup()

Cleanup removes limiters that have been idle for more than limiterIdleTimeout.

type PublicURLConfig 

type PublicURLConfig struct {
	// Explicit override from MAINTENANT_GRPC_URL env or --grpc-url flag.
	// If non-empty, it is used as-is (after ensuring the grpcs:// scheme).
	Explicit string

	// ListenAddr is the address the gRPC server is bound to (e.g. "127.0.0.1:8443").
	// Used as fallback when neither Explicit nor request headers are available.
	ListenAddr string

	// TrustedProxies lists the peers whose X-Forwarded-* headers are believed.
	TrustedProxies []netip.Prefix
}

PublicURLConfig holds the inputs for resolving the gRPC public URL.

type ResourceHandler 

type ResourceHandler interface {
	HandleAgentEvent(ctx context.Context, agentID string, ev *agentpb.ResourceSample, meta EventMeta) error
}

ResourceHandler processes a resource sample from a remote agent.

type Server 

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

Server wraps the gRPC server lifecycle for the Ingest service.

func New 

func New(deps Deps) *Server

New creates a Server. Call Start to begin accepting connections.

func (*Server) Start 

func (s *Server) Start(ctx context.Context, listen string, tlsCfg *tls.Config) error

Start binds to listen and registers the Ingest service, then serves in the background. It returns as soon as the bind succeeds or fails, so call it synchronously and check the error; the server then runs and stops on its own as ctx allows. If tlsCfg is nil the server listens in h2c (plaintext) — only safe behind a trusted reverse proxy (MAINTENANT_GRPC_TLS_INSECURE=true).

func (*Server) StartTokenGC 

func (s *Server) StartTokenGC(ctx context.Context)

StartTokenGC launches a background goroutine that purges unconsumed enrollment tokens older than 7 days. It ticks every hour until ctx is cancelled.

func (*Server) Stop 

func (s *Server) Stop()

Stop performs a graceful shutdown of the gRPC server.

type Sessions 

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

Sessions tracks all currently connected agent streams.

func NewSessions 

func NewSessions(logger *slog.Logger, broadcaster EventBroadcaster) *Sessions

NewSessions creates an empty Sessions registry. broadcaster may be nil (SSE events will not be emitted).

func (*Sessions) Close 

func (s *Sessions) Close(agentID, reason string)

Close removes and cancels whatever active stream currently exists for agentID, regardless of which one opened it. For administrative teardown (revoke, delete) where any live session must go.

func (*Sessions) CloseStream added in v1.5.0 

func (s *Sessions) CloseStream(agentID string, tok Token, reason string)

CloseStream removes and cancels the active stream for agentID only if it is still the one identified by tok. A stale tok (superseded by a reconnect) or a zero Token is a no-op, so a handler unwinding after being replaced can never tear down the session that replaced it.

func (*Sessions) DeliverResult added in v1.3.8 

func (s *Sessions) DeliverResult(agentID string, res *agentpb.CommandResult)

DeliverResult routes a CommandResult from agentID to whoever awaits it.

func (*Sessions) EventsPerSecond5m 

func (s *Sessions) EventsPerSecond5m() float64

EventsPerSecond5m returns the average events/s over the last 5 minutes.

func (*Sessions) EventsSeen 

func (s *Sessions) EventsSeen(agentID string) int64

EventsSeen returns the events_seen counter for agentID (0 if not connected).

func (*Sessions) FetchLogs added in v1.3.8 

func (s *Sessions) FetchLogs(ctx context.Context, agentID, externalID string, lines int, timestamps bool) ([]string, error)

FetchLogs performs a one-shot log tail on agentID, collecting the agent's chunks into a single slice. Shared by the REST and MCP read paths.

func (*Sessions) HasCapability added in v1.3.8 

func (s *Sessions) HasCapability(agentID, capability string) bool

HasCapability reports whether the agent's live stream advertised capability.

func (*Sessions) IncrEvents 

func (s *Sessions) IncrEvents(agentID string)

IncrEvents increments the events_seen counter for agentID and the ring buffer.

func (*Sessions) IsConnected 

func (s *Sessions) IsConnected(agentID string) bool

IsConnected reports whether agentID currently has an active stream.

func (*Sessions) ListConnected 

func (s *Sessions) ListConnected() []string

ListConnected returns all currently connected agent IDs.

func (*Sessions) Open 

func (s *Sessions) Open(agentID string, cancel context.CancelCauseFunc, addr string, caps []string, send chan *agentpb.ServerMessage) Token

Open registers an active stream for agentID and cancels any pre-existing one. caps are the command families the agent advertised; send is the queue its Push goroutine drains to write to the stream (both may be nil for a telemetry-only stream, in which case no command can be issued to this agent). The returned Token identifies this stream for CloseStream.

func (*Sessions) RecordSpoolStatus added in v1.6.0 

func (s *Sessions) RecordSpoolStatus(agentID string, st *agentpb.SpoolStatus)

RecordSpoolStatus stores what agentID declared about its spool.

func (*Sessions) SendCommand added in v1.3.8 

func (s *Sessions) SendCommand(ctx context.Context, agentID, capability string, cmd *agentpb.AgentCommand) (<-chan *agentpb.CommandResult, func(), error)

SendCommand issues cmd to agentID and returns a channel carrying its replies, closed once the terminal result arrives or the stream dies. The returned release func MUST be called by the caller (defer): it drops the pending entry and, when the command may still be running, tells the agent to stop.

func (*Sessions) SetLifecycleAlertHook 

func (s *Sessions) SetLifecycleAlertHook(fn func(agentID, reason string, connected bool))

SetLifecycleAlertHook registers a callback fired on connect/disconnect transitions. Must be called once at wiring time, before any agent connects.

func (*Sessions) SpoolStatus added in v1.6.0 

func (s *Sessions) SpoolStatus(agentID string) *SpoolState

SpoolStatus returns what agentID last declared, or nil when it is not connected or has never reported (an agent older than the spool).

func (*Sessions) StartRingAdvancer 

func (s *Sessions) StartRingAdvancer(ctx context.Context)

StartRingAdvancer runs the ring buffer advance tick every 5s until ctx is done.

func (*Sessions) StartStaleWatcher 

func (s *Sessions) StartStaleWatcher(ctx context.Context, interval, threshold, grace time.Duration, staleAgents StaleAgentsFn)

StartStaleWatcher emits agent.disconnected SSE for agents that have not been seen within threshold but still appear in the sessions map (dead stream), and reports the outage of stale agents that have no session at all.

That second half is what makes an absent agent visible across a restart: the connect/disconnect hook only fires on a transition, so an agent that went down while the server was stopped (or during the shutdown storm, where alerts are suppressed on purpose) would otherwise stay silently disconnected forever. grace holds that reporting back after startup, giving healthy agents time to reconnect; each outage is announced once, until the agent is back. Runs until ctx is done.

type SpoolState added in v1.6.0 

type SpoolState struct {
	Queued              int64
	Draining            bool
	DroppedSinceConnect int64
	ReportedAt          time.Time
}

SpoolState is what an agent last said about its outbound queue.

type StaleAgentsFn 

type StaleAgentsFn func(ctx context.Context, threshold time.Duration) ([]string, error)

StaleAgentsFn returns agent IDs whose last_seen_at is older than threshold. Passed to StartStaleWatcher so sessions doesn't import the sqlite package.

type SwarmTopologyHandler 

type SwarmTopologyHandler interface {
	HandleAgentEvent(ctx context.Context, agentID string, ev *agentpb.SwarmTopology) error
}

SwarmTopologyHandler processes a full swarm topology snapshot from an agent.

type Token added in v1.5.0 

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

Token identifies one stream, so a handler closes only the session it opened.

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.