Documentation
¶
Overview ¶
Package agentserver implements the gRPC Ingest service that remote agents connect to for enrollment and event streaming (Pro only).
Index ¶
- Variables
- func GenerateChallenge() (*agentpb.AuthChallenge, error)
- func LoadOrGenerateTLS(certFile, keyFile string, hosts []string, logger *slog.Logger) (*tls.Config, error)
- func ResolvePublicURL(req *http.Request, cfg PublicURLConfig) (string, []string)
- func Verify(req *agentpb.AuthResponse, nonce []byte, ag *agent.Agent, serverNow time.Time) error
- type CertificateHandler
- type ContainerHandler
- type Deps
- type DispatchDeps
- type Dispatcher
- type EndpointHandler
- type EventBroadcaster
- type HeartbeatHandler
- type KubernetesTopologyHandler
- type LabelSyncFunc
- type Limiter
- type PublicURLConfig
- type ResourceHandler
- type Server
- type Sessions
- func (s *Sessions) Close(agentID, reason string)
- func (s *Sessions) EventsPerSecond5m() float64
- func (s *Sessions) EventsSeen(agentID string) int64
- func (s *Sessions) IncrEvents(agentID string)
- func (s *Sessions) IsConnected(agentID string) bool
- func (s *Sessions) ListConnected() []string
- func (s *Sessions) Open(agentID string, cancel context.CancelFunc, addr string)
- func (s *Sessions) SetLifecycleAlertHook(fn func(agentID, reason string, connected bool))
- func (s *Sessions) StartRingAdvancer(ctx context.Context)
- func (s *Sessions) StartStaleWatcher(ctx context.Context, interval, threshold time.Duration, ...)
- type StaleAgentsFn
- type SwarmTopologyHandler
Constants ¶
This section is empty.
Variables ¶
Functions ¶
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 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.
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) error
}
ContainerHandler processes a container event from a remote agent.
type Deps ¶
type Deps struct {
AgentStore *sqlite.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
Endpoint EndpointHandler
Heartbeat HeartbeatHandler
Resource ResourceHandler
Certificate CertificateHandler
Swarm SwarmTopologyHandler
Kubernetes KubernetesTopologyHandler
// 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, evt *agentpb.AgentEvent) error
Dispatch routes evt to the handler matching its body type. Returns an error if a handler is wired and returns an error. Silently ignores events whose handler is nil.
type EndpointHandler ¶
type EndpointHandler interface {
HandleAgentEvent(ctx context.Context, agentID string, ev *agentpb.EndpointEvent) error
}
EndpointHandler processes an endpoint probe result from a remote agent.
type EventBroadcaster ¶
EventBroadcaster is the minimal interface required to publish SSE events to connected clients.
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 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 ¶
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.
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
}
PublicURLConfig holds the inputs for resolving the gRPC public URL.
type ResourceHandler ¶
type ResourceHandler interface {
HandleAgentEvent(ctx context.Context, agentID string, ev *agentpb.ResourceSample) 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 (*Server) Start ¶
Start binds to listen, registers the Ingest service, and begins serving. It blocks until the context is cancelled or an error occurs. 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 ¶
StartTokenGC launches a background goroutine that purges unconsumed enrollment tokens older than 7 days. It ticks every hour until ctx is cancelled.
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) EventsPerSecond5m ¶
EventsPerSecond5m returns the average events/s over the last 5 minutes.
func (*Sessions) EventsSeen ¶
EventsSeen returns the events_seen counter for agentID (0 if not connected).
func (*Sessions) IncrEvents ¶
IncrEvents increments the events_seen counter for agentID and the ring buffer.
func (*Sessions) IsConnected ¶
IsConnected reports whether agentID currently has an active stream.
func (*Sessions) ListConnected ¶
ListConnected returns all currently connected agent IDs.
func (*Sessions) Open ¶
func (s *Sessions) Open(agentID string, cancel context.CancelFunc, addr string)
Open registers an active stream for agentID and cancels any pre-existing one.
func (*Sessions) SetLifecycleAlertHook ¶
SetLifecycleAlertHook registers a callback fired on connect/disconnect transitions. Must be called once at wiring time, before any agent connects.
func (*Sessions) StartRingAdvancer ¶
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 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). Runs until ctx is done.
type StaleAgentsFn ¶
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.