server

Documentation

Overview

Package server implements the agent-specific Envoy xDS server. It builds xDS snapshots from local pod storage and the service registry, generating Envoy listeners, clusters, endpoints, and routes.

Index

• func AssertWatchFilter(c *cache.SnapshotCache, reg registry.Registry)
• type AgentXdsServer
  • func NewAgentXdsServer(ctx context.Context, clusterName string, nodeName string, trustDomain string, ...) (*AgentXdsServer, error)
  • func (s *AgentXdsServer) NeedLeaderElection() bool
  • func (s *AgentXdsServer) PreListen(ctx context.Context) error
• type RegistryRefresher
  • func NewRegistryRefresher(clusterName, nodeName string, snapshotCache *cache.SnapshotCache, ...) *RegistryRefresher
  • func (r *RegistryRefresher) NeedLeaderElection() bool
  • func (r *RegistryRefresher) Start(ctx context.Context) error

Functions

func AssertWatchFilter

func AssertWatchFilter(c *cache.SnapshotCache, reg registry.Registry)

AssertWatchFilter pushes the cache's current dependency set to the registry as the watch service filter (registry.WatchScoper capability), so the registrar fans out endpoint changes for this node's dependencies only. No-op for registries without watch scoping or when the filter is unchanged.

Types

type AgentXdsServer

type AgentXdsServer struct {
	xds.XdsServer
	// contains filtered or unexported fields
}

AgentXdsServer is an xDS server that generates Envoy configuration from local pod storage and a service registry. It embeds xds.XdsServer and implements the ServerCallback interface to generate an initial snapshot before starting to accept connections.

The server maintains versioned snapshots of Envoy resources (listeners, clusters, endpoints, routes) and serves them to local Envoy proxy instances via the xDS protocol.

func NewAgentXdsServer

func NewAgentXdsServer(ctx context.Context, clusterName string, nodeName string, trustDomain string, registry registry.Registry, storage storage.Storage[*cniv1.CNIPod], snapshotCache *cache.SnapshotCache, callbacks serverv3.Callbacks, log *slog.Logger) (*AgentXdsServer, error)

NewAgentXdsServer creates a new AgentXdsServer. It initializes an xDS server with a snapshot cache and registers itself as a callback to generate the initial Envoy snapshot before listening for client connections. The server listens on a Unix domain socket at the default xDS socket path. callbacks (optional, may be nil) observe the discovery streams — the agent passes the ACK tracker's callbacks so pod lifecycle can await Envoy ACKs.

func (*AgentXdsServer) NeedLeaderElection

func (s *AgentXdsServer) NeedLeaderElection() bool

NeedLeaderElection returns false so the xDS server runs on EVERY replica, not just the leader. Each edge/agent pod serves xDS to its own co-located Envoy over a node-local UDS; leader-gating it would leave all non-leader proxies without a control plane and break data-plane HA. (The embedded xds.Server already declares this; AgentXdsServer states it explicitly so the per-pod-runnable contract is visible at this type.) On the node agent (leader election off) this method is a no-op.

func (*AgentXdsServer) PreListen

func (s *AgentXdsServer) PreListen(ctx context.Context) error

PreListen generates the initial Envoy snapshot from local pod storage and the service registry. It creates listeners, clusters, endpoints, and routes, then sets the snapshot in the cache before the server starts accepting xDS client connections.

type RegistryRefresher

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

RegistryRefresher is a controller-runtime runnable that rebuilds the xDS cluster/endpoint/route snapshot whenever the registry reports endpoint changes. It bridges a registry.ChangeNotifier to the snapshot cache's LoadClustersFromRegistry, so services registered after the agent started become routable without restarting the agent.

If the registry does not implement registry.ChangeNotifier, the refresher is a no-op for the lifetime of the process (the initial snapshot built during the xDS server's PreListen still applies).

func NewRegistryRefresher

func NewRegistryRefresher(clusterName, nodeName string, snapshotCache *cache.SnapshotCache, reg registry.Registry, log *slog.Logger) *RegistryRefresher

NewRegistryRefresher creates a RegistryRefresher.

func (*RegistryRefresher) NeedLeaderElection

func (r *RegistryRefresher) NeedLeaderElection() bool

NeedLeaderElection returns false so the refresher runs on EVERY replica, not just the leader. Each edge/agent pod feeds its own co-located Envoy via the snapshot cache; leader-gating this runnable would starve all non-leader proxies of cluster/endpoint updates and break data-plane HA. The reconciler (status writer) is the only component that must be a singleton — it stays leader-election-aware via the controller-runtime builder. On the node agent (leader election off) this method is a no-op.

func (*RegistryRefresher) Start

func (r *RegistryRefresher) Start(ctx context.Context) error

Start blocks until ctx is cancelled, rebuilding the cluster snapshot from the registry on each (debounced) change notification — registry endpoint changes and node dependency-set changes (pod add/remove or changed declared upstreams) both funnel into the same debounced reload. It implements controller-runtime's manager.Runnable.